The harness is the graph, not the agent

flowchart TD
  req["POST /expand {topic}"] --> api["api-worker"]
  api -- expand.plan --> planner["planner-worker · LLM: topic → research angles"]
  planner -- expand.research --> researcher["research-worker · agent loop: tools::web_search ×N"]
  researcher -- expand.neighbors --> neighbors["neighbors-worker · LLM: adjacent disciplines"]
  neighbors -- expand.synthesis --> synth["synthesis-worker · LLM: related topics + how they relate"]
  synth -- "state: done" --> read["api-worker reads state"]
  read --> res["POST /result {job_id}"]
  class api,planner,neighbors,synth,read warn
  class researcher pass
  class req,res dim

Notice how little of this is agentic. Four of the five model-calling nodes are one constrained LLM call with a fixed input and output shape. Only research-worker runs a real loop, because only research needs one: the number of searches is not knowable in advance. Everything else is a transformation that happens to use a model.

# Only managed infrastructure workers live here. The 6 custom pipeline nodes
# (api, planner, research, neighbors, synthesis, tools) self-connect as local
# processes so their ANTHROPIC_API_KEY comes from the host env, never git.
workers:
  - name: iii-observability
  - name: configuration
  - name: state
  - name: queue
  - name: http

Five lines. Each one expands into a real subsystem configured in config/. The HTTP server binds 0.0.0.0:3111 with a request timeout and a concurrency limit, and the state store enables trigger fan-out. The queue picks a persistence adapter, and observability turns on traces, metrics, and logs under the service name concept-expander.

The comment matters as much as the config. This file deliberately leaves the pipeline nodes out, because they need ANTHROPIC_API_KEY and git tracks the file. They self-connect as local processes and read the key from the host environment. To run a node as an engine-managed microVM instead, ship an iii.worker.yaml: the manifest's env: map injects the key.

"""Node 1: break the incoming topic into concrete research angles."""
import os

import anthropic
from iii import register_worker, InitOptions
from iii_helpers.observability import Logger

worker = register_worker(
    os.environ.get("III_URL", "ws://localhost:49134"),
    InitOptions(worker_name="planner-worker"),
)
logger = Logger()
client = anthropic.Anthropic()      # ANTHROPIC_API_KEY from host env


def get_job(job_id: str) -> dict:
    return worker.trigger(
        {"function_id": "state::get", "payload": {"scope": "jobs", "key": job_id}}
    ) or {}


def publish(topic: str, job_id: str) -> None:
    worker.trigger(
        {"function_id": "iii::durable::publish",
         "payload": {"topic": topic, "data": {"job_id": job_id}}}
    )

Everything a node can do to the outside world goes through worker.trigger with a function_id. Reading state is state::get, and writing it is state::set. Handing off is iii::durable::publish, and calling another node's capability is that node's function id. One verb, and the id says which subsystem you are talking to.

The consequence is that nodes have no imports of each other. planner-worker does not know research-worker exists: it publishes to expand.research and stops caring. Because of that decoupling, you can restart, redeploy, or rewrite one node on its own, in a different language. One of these six is TypeScript and the other five are Python for that reason.

const jobId = randomUUID();
logger.info('accepted expand job', { jobId, topic });

await worker.trigger({
  function_id: 'state::set',
  payload: {
    scope: 'jobs', key: jobId,
    value: { status: 'pending', topic, stage: 'plan' },
  },
});

// Hand off to the first pipeline node via the durable queue.
await worker.trigger({
  function_id: 'iii::durable::publish',
  payload: { topic: 'expand.plan', data: { job_id: jobId } },
});

return {
  status_code: 202,
  body: { job_id: jobId, status: 'pending' },
  headers: { 'Content-Type': 'application/json' },
};

The 202 is the design. The request does not wait for research to finish, and it could not: this pipeline runs for minutes. The caller gets an id, the queue drives the rest, and /result reports where it has got to.

Why the payload is a pointer and not the data

1. Message size Research findings are long. Putting them on the queue makes every message large, and every retry re-delivers all of it.
2. One writer The job object has a single owner at a time, the node currently handling that stage. Reading and writing it is a get and a set, not a merge.
3. Inspectable The whole state of a run is one key in one scope. Polling /result is just reading it, and so is debugging.
4. The catch Two nodes handling the same job concurrently would clobber each other. This pipeline is strictly sequential, so that clash does not arise. A fan-out stage would need the merge this design avoids.

The stage field is what makes a stalled run diagnosable. A job sitting at research for ten minutes tells you which node to look at. It also names the stage whose subscriber never started, so you learn why nothing is happening. Without that field, a pipeline that stops looks exactly like a pipeline that is slow.

@beta_tool
def web_search(query: str) -> str:
    """Search the web for current information about a query.

    Args:
        query: A focused search query.
    """
    logger.info(f"web_search -> tools-worker: {query!r}")
    return worker.trigger(
        {"function_id": "tools::web_search", "payload": {"query": query},
         "timeout_ms": 170_000}
    ) or ""


def research_handler(payload: dict) -> None:
    ...
    runner = client.beta.messages.tool_runner(
        model=MODEL,
        max_tokens=8000,
        output_config={"effort": "high"},
        tools=[web_search],
        system=(
            "You are a research agent. Investigate the topic along the given angles, "
            "using the web_search tool as needed to ground yourself in current, factual "
            "information. Produce concise notes capturing key concepts, findings, and any "
            "fields or disciplines the topic touches. Do not write a final essay yet."
        ),
        messages=[{"role": "user", "content": f"Topic: {topic}\n\nResearch angles:\n{angle_list}"}],
    )
    final = runner.until_done()
    findings = "".join(b.text for b in final.content if b.type == "text")

    job["findings"] = findings
    job["stage"] = "neighbors"
    set_job(job_id, job)
    publish("expand.neighbors", job_id)

The tool function body is one worker.trigger. The model thinks it is calling web_search. The real event is a message to tools-worker, which owns the search capability for the whole system. Swapping the search provider then changes one node, and no other node notices. The generous timeout_ms is there because a cross-node tool call is a network round trip, not a function call.

The sentence doing the most work

"Do not write a final essay yet." Without that line, a capable model reads "research this topic" and delivers a polished writeup. That writeup then becomes the input to three more stages, each of which wants to write its own. The instruction that holds a stage inside its scope is usually a prohibition, not a description, and every node's prompt here carries one.

export ANTHROPIC_API_KEY=sk-ant-...

# terminal 1 — start the engine + managed infra workers
iii --config config.yaml

# terminal 2 — start the 6 pipeline nodes (installs deps on first run)
./scripts/dev.sh

# one-shot: kick off a topic, poll to completion, print the summary
./scripts/expand.sh "cellular automata"

The interesting entry point is not the HTTP one. Try this instead:

iii trigger http::expand --json '{"body":{"topic":"cellular automata"}}'
iii trigger http::result --json '{"body":{"job_id":"<id>"}}'

You can invoke any node's function by id with a synthetic payload, skipping the HTTP layer and earlier stages. So a broken synthesis stage is reproducible. Trigger synthesis against a job id that already has findings, and skip the four minutes of research. In a monolithic agent the equivalent is replaying a transcript and hoping.

What the observability config buys

1. Traces OpenTelemetry spans at full sampling, exported to memory for local work. A run is a tree across six processes rather than six unrelated log streams.
2. Logs Every node uses the same Logger helper, so logger.info("research complete; handing to neighbors node") lands correlated with the span that emitted it.
3. The default Exporter is memory and sampling is 1.0. Correct for development, wrong for production on both counts, and worth changing before this ever handles real volume.

Reach for it when

• The stages are stable. You can name them before you start, and they do not change per run.
• The work outlives a session. Minutes to hours, where "keep an agent loop open" stops being an answer.
• Stages deserve different budgets. A planner is a cheap call; research is expensive. So one model and one effort level for both is waste at one end and a constraint at the other.
• Partial progress is worth keeping. If losing four minutes of research to a crash in synthesis is unacceptable, the durable handoff is the whole reason to be here.

Do not reach for it when

• The structure varies per run. A fixed graph cannot decide it needs a different shape today, and that gap is disqualifying.
• The whole thing takes ten seconds. The queue, the state store, and six processes are pure overhead against one call that finishes before the infrastructure finishes starting.
• You have not proved the pipeline yet. Write it as one agent first. Then split it when a stage starts deserving its own budget, its own retry, or its own language.

Where this sits

1. pi Building a harness with pi is the opposite construction: take one existing agent and subtract until it fits a new domain. One process, one loop, six extension surfaces.
2. This page Compose from parts instead. Many processes, one loop total, and the wiring is the artifact you are designing.
3. The factory A lights-on factory is this shape applied to software delivery. Governance gates and provenance sit on the same queue, which takes a tracker card and returns a merged pull request.