The harness is the graph, not the agent

POST /expand {topic}
      │
      ▼
  api-worker ──publish──► expand.plan
                              │
  planner-worker   (LLM: topic → research angles) ──► expand.research
                              │
  research-worker  (agent loop: tools::web_search ×N) ──► expand.neighbors
                              │
  neighbors-worker (LLM: adjacent disciplines) ──► expand.synthesis
                              │
  synthesis-worker (LLM: related topics + how they relate) ──► state: done
                              │
POST /result {job_id} ◄── api-worker reads state

Notice how little of this is agentic. Four of the five model-calling nodes are a single constrained LLM call with a fixed input and a fixed output shape. Only research-worker runs a real loop, because only research genuinely 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 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, 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. The pipeline nodes are deliberately not declared here, because they need ANTHROPIC_API_KEY and this file is in git. They self-connect as local processes and read the key from the host environment. When you want a node running as an engine-managed microVM instead, it ships an iii.worker.yaml and the key is injected through the manifest's env: map.

"""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. Writing it is state::set. Handing off is iii::durable::publish. 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. That is what makes a node independently restartable, independently deployable, and independently rewritable in a different language, which is exactly why one of these six is TypeScript and the other five are Python without that being a problem.

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 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, and a job sitting at a stage whose subscriber never started tells you why nothing is happening. Without it, 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; what actually happens is a message to tools-worker, which owns the search capability for the whole system. Swapping the search provider is a change in one node that no other node notices, and 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 it, a capable model reads "research this topic" and delivers a polished writeup, which then becomes the input to three more stages that each want to write their own. The instruction that keeps a pipeline stage in its lane is usually a prohibition, not a description, and each node's prompt in this repo 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. It is this:

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

Any node's function can be invoked by id with a synthetic payload, without the HTTP layer and without the stages before it. That means a broken synthesis step is reproduced by triggering synthesis against a job id that already has findings, instead of re-running four minutes of research to get back to the failure. 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.
• Steps deserve different budgets. A planner is a cheap call; research is expensive. 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. This is the disqualifying condition.
• 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. 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 actually designing.
3. The factory A lights-on factory is this shape applied to software delivery, with governance gates and provenance on a queue that takes a tracker card and returns a merged pull request.