Architecture

The dataflow shape

Three streams feed one keyed transform, and four streams come out of it.

Two things about that diagram are load-bearing.

First, the tool-results and approvals topics are inputs. They are not a side channel or a callback — they are ordinary elements, keyed the same way events are, flattened into the same stream. A resumed activation is a new element on the same key, indistinguishable in the runner's eyes from a fresh event.

Second, the loop closes outside the DAG. Beam DAGs are acyclic. An agent that calls a tool, gets a result, and calls another tool cannot be expressed as a cycle in the graph, and unrolling the loop into N copies of the transform bounds the iteration count at pipeline-construction time. So iteration goes through the message bus instead: out to the outbox topic, through the effector, back in on the results topic. The DAG stays acyclic and the loop stays unbounded.

Two paths through RunAgent

An activation either finishes or suspends. Both tracks below start with one element on one key and end with the same .output; what differs is how many activations it took and what had to survive in between.

Fast path

The agent runs to completion inside one process() call. Read-only tools execute inline; model calls go through the async client. One element in, one decision out. Nothing is persisted except working memory and the sequence counter.

async def triage(ctx: ActivationContext) -> Complete:
    """Decide what to do about one event for one entity.

    Module-level, not a closure: the DoFn holding the agent is serialized for
    the runner, so the agent has to pickle by reference.
    """
    ctx.memory.append("recent", ctx.event, max_items=32)
    seen = len(ctx.memory.ring("recent"))

    response = await ctx.call_model(
        LlmRequest(
            model_id="fake-1",
            messages=[f"event={ctx.event.decode()} seen={seen}"],
            tools_schema=None,
            sampling_params=None,
        )
    )
    return Complete(output=b"%s:%d" % (response.response, seen))

Re-injection path

For a side-effectful tool or a human approval, the agent stages a ToolIntent, persists a Continuation in keyed state, and yields. The activation is over — the worker moves on. When the ToolResult or Approval arrives on the same key, the continuation is rehydrated and the agent is invoked again, this time with ctx.is_resume true.

async def refund(ctx: ActivationContext) -> Complete | Suspend:
    """Request a refund, then report what the effector did.

    Two activations, one logical unit of work. `ctx.act` stages the intent and
    returns its deterministic id; `Suspend` persists the continuation and arms
    the fail-closed timeout. Nothing has been written to the outside world when
    this function returns — the intent is a request, not an effect.
    """
    if not ctx.is_resume:
        ctx.act("payments.refund", '{"amount": 4200}', ttl_ms=INTENT_TTL_MS)
        return Suspend(snapshot=b"awaiting-refund", adapter="example", timeout_ms=30_000)

    # On resume the same activation continues: same key, same seq, and the
    # snapshot it persisted is available as ctx.snapshot.
    assert ctx.resume_result is not None
    return Complete(output=b"refunded:" + ctx.resume_result.payload)

The suspended activation keeps its seq. That matters more than it looks: the sequence number scopes both the replay-cache keys and the deterministic intent ids, so a resumed activation staging a second intent produces an id that continues the first one's step index rather than colliding with it.

Inside the stateful DoFn

The runtime is one Beam stateful DoFn. Its state and timers are:

The Python SDK has no MapState, so bounded maps live inside single-value protobuf blobs with explicit LRU eviction. Every blob is capped at 100 KiB and working memory carries a 1 MiB soft cap per key with a compaction hook.

State is protobuf, never pickle. That is what makes pipeline --update compatibility tractable: additive proto changes only, and a breaking change requires a state_schema_version bump with lazy migration and a golden-blob compatibility test.

The async bridge

Beam's Python SDK has no portable async DoFn, and the runtime is async internally — model calls, tool execution, the whole activation. The bridge resolves that: setup() starts one background thread per DoFn instance with a dedicated asyncio loop and shared httpx pools. process() submits the activation coroutine and blocks with activation_timeout. On timeout the coroutine is cancelled and the element routes to .errors with no state mutation.

One thread per DoFn instance, not per element, is why the connection pools are shared and why nothing in the activation path may block the loop — the ASYNC lint rules exist to enforce that.

What comes out

RunAgent returns a RunAgentOutputs, and a complete pipeline consumes every field:

outputs = keyed | "Agent" >> RunAgent(
    route_by_event, config=AgentConfig(provider_factory=make_provider)
)

assert_that(outputs.output, equal_to([b"assessed", b"notified"]), label="output")

intents = outputs.intents | "IntentNames" >> beam.Map(
    lambda intent: (intent.entity_key, intent.tool_name)
)
assert_that(intents, equal_to([(b"k-notify", "slack.post")]), label="intents")

# An ActivationError names the key, why it failed, and the element's
# event time — never a wall clock.
errors = outputs.errors | "ErrorShape" >> beam.Map(
    lambda error: (error.entity_key, error.reason, error.event_time_ms)
)
assert_that(errors, equal_to([(b"k-broken", "activation_error", 1_000)]), label="errors")

There is a fifth field, dead_letter, which is populated only when intents_to resolved to a WriteIntents outbox writer. It carries intents that could not be serialized. When errors_to is also configured the two streams merge before the sink, so the errors topic carries exactly one record schema.

Next

• Correctness invariants — the seven rules this architecture exists to hold.
• State and memory — what is stored per key and how it is bounded.
• The errors output — the operational reference for .errors.