The four outputs — beam-agents

RunAgent does not return a PCollection. It returns a RunAgentOutputs carrying four of them, and a pipeline that consumes only the first is incomplete in a way nothing will tell you about at runtime.

This example drives three keys down three different paths in one bounded pipeline and asserts on every stream, so each output is demonstrated rather than described.

The four streams

.output — bytes. Terminal agent outputs; what the pipeline is for.
.intents — ToolIntent. Side-effect requests, bound for the outbox topic and the effector.
.traces — TraceEvent. Observability spans for the activation and everything inside it.
.errors — ActivationError. Dead letters: element-level failures, never a failed bundle.

There is a fifth field, dead_letter, populated only when intents_to resolves to a WriteIntents outbox writer; it carries intents that could not be serialized. Architecture covers where it fits.

.errors is the one people forget. An element-level failure never fails the bundle — it is routed here and the pipeline moves on. Leave the stream unconsumed and the failure is not "handled by default", it is invisible.

One agent, three paths

Rather than three pipelines, the agent branches on the event payload, so a single bounded run exercises every output.

async def route_by_event(ctx: ActivationContext) -> Complete:
    """Take a different path per event so one pipeline exercises each output."""
    if ctx.event == b"BROKEN":
        # Routed to `.errors` as `activation_error`. Nothing this activation
        # staged — including this memory write — reaches durable state.
        ctx.memory.set("scratch", b"never-persisted")
        raise RuntimeError("downstream schema changed")

    if ctx.event == b"NOTIFY":
        ctx.act("slack.post", '{"channel": "#ops"}', ttl_ms=INTENT_TTL_MS)
        return Complete(output=b"notified")

    response = await ctx.call_model(
        LlmRequest(
            model_id="fake-1",
            messages=[ctx.event.decode()],
            tools_schema=None,
            sampling_params=None,
        )
    )
    return Complete(output=response.response)

website/examples/four_outputs.py (region: agent) — executed by the repository’s offline test tier.

review takes the ordinary path: a model call, then Complete. It produces an .output element and a full span set on .traces.
NOTIFY stages an intent with ctx.act and completes. It produces both an .output and an .intents element.
BROKEN writes to memory and then raises. That ordering is the point of the example, and the next section is about what becomes of the write.

Asserting on every stream

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")

website/examples/four_outputs.py (region: outputs) — executed by the repository’s offline test tier.

Three assertions, one per stream, each with an explicit label — assert_that needs distinct step names inside a single pipeline.

The .errors assertion is worth reading slowly. An ActivationError names the key, a reason drawn from a closed taxonomy — here activation_error — and the element's event time, never a wall clock. That last detail is what makes a dead letter replayable, and what makes two runs over the same input produce byte-identical error records. The rest of the taxonomy (activation_timeout, orphaned_result, hitl_timeout, intent_dead_letter) is in the errors output.

What a dead letter means

It means the activation committed nothing.

Not "partially applied", not "applied but unreported". The ctx.memory.set call on the BROKEN path is staged in the activation context and discarded with everything else when the activation raises: no memory write, no intent, no output, and the key's sequence counter does not advance. The record on .errors is the only evidence the key was touched at all.

That is why consuming this stream is not optional housekeeping. It is how you learn a key produced nothing, and no other channel will tell you.

Traces show the commit boundary from outside

# Traces show the atomic-commit rule from the outside. A committed
# activation emits its whole span set — START, whatever happened in the
# middle, END. The failed one emits a single ERROR event: the traces it
# staged before raising were discarded with the rest of its effects,
# exactly like its memory write.
traced = outputs.traces | "TraceShape" >> beam.Map(
    lambda event: (event.entity_key, event.event_type)
)
assert_that(
    traced,
    equal_to(
        [
            (b"k-model", TraceEvent.ACTIVATION_START),
            (b"k-model", TraceEvent.LLM_CALL),
            (b"k-model", TraceEvent.ACTIVATION_END),
            (b"k-notify", TraceEvent.ACTIVATION_START),
            (b"k-notify", TraceEvent.INTENT_EMITTED),
            (b"k-notify", TraceEvent.ACTIVATION_END),
            (b"k-broken", TraceEvent.ERROR),
        ]
    ),
    label="traces",
)

website/examples/four_outputs.py (region: traces) — executed by the repository’s offline test tier.

Compare the two committed keys with the failed one. k-model and k-notify each emit ACTIVATION_START, an event for whatever happened in the middle, and ACTIVATION_END. k-broken emits a single ERROR.

It emits one event because traces are staged like every other effect: the ACTIVATION_START span that activation produced before raising was discarded along with its memory write. So .traces is not a log written as things happen — it is the committed record of what actually took effect, which means correctness invariant 1 can be audited from outside the runtime by reading it.

Traces has the attribute conventions and the OTLP and BigQuery sinks.

Wiring the streams to sinks

The example asserts on the streams in-process because it is a test. A deployed pipeline attaches sinks instead, through AgentConfig:

config = AgentConfig(
    provider_factory=make_provider,
    intents_to="kafka://broker:9092/agent.intents",
    traces_to="otlp://collector:4317",
    errors_to="kafka://broker:9092/agent.errors",
)

Sink URIs are validated when AgentConfig is constructed, before a pipeline exists. The otlp:// scheme is accepted for traces_to alone: it is a best-effort tap that drops on delivery failure by contract, and intents and errors are correctness-bearing streams that need a lossless sink.

Run it

uv run python website/examples/four_outputs.py

The complete program

"""The four outputs: `.output`, `.intents`, `.traces`, and `.errors`.

`RunAgent` returns a `RunAgentOutputs` with four named `PCollection`s, and a
pipeline is expected to consume all of them:

    .output   terminal agent outputs (bytes)
    .intents  ToolIntent side-effect requests, bound for the outbox topic
    .traces   TraceEvent observability records
    .errors   ActivationError dead letters

`.errors` is the one people forget. Element-level failures never fail the
bundle — they are routed here — and a dead letter means the activation
committed *nothing*: no memory write, no intent, no output. The record is the
only evidence the key was touched at all.

This example drives three keys down three different paths in one bounded
pipeline, and asserts on every stream.

Run it:  python website/examples/four_outputs.py
"""

from __future__ import annotations

import apache_beam as beam
from apache_beam.testing.util import assert_that, equal_to

from beam_agents import AgentConfig, RunAgent
from beam_agents._protos import AgentEnvelope, TraceEvent
from beam_agents.core.agent import Complete
from beam_agents.core.context import ActivationContext
from beam_agents.model.client import LlmRequest
from beam_agents.model.fake import FakeLLM, match_any, respond_with

INTENT_TTL_MS = 60_000


def make_provider() -> FakeLLM:
    return FakeLLM([(match_any(), respond_with(b"assessed"))])


# region: agent
async def route_by_event(ctx: ActivationContext) -> Complete:
    """Take a different path per event so one pipeline exercises each output."""
    if ctx.event == b"BROKEN":
        # Routed to `.errors` as `activation_error`. Nothing this activation
        # staged — including this memory write — reaches durable state.
        ctx.memory.set("scratch", b"never-persisted")
        raise RuntimeError("downstream schema changed")

    if ctx.event == b"NOTIFY":
        ctx.act("slack.post", '{"channel": "#ops"}', ttl_ms=INTENT_TTL_MS)
        return Complete(output=b"notified")

    response = await ctx.call_model(
        LlmRequest(
            model_id="fake-1",
            messages=[ctx.event.decode()],
            tools_schema=None,
            sampling_params=None,
        )
    )
    return Complete(output=response.response)


# endregion: agent


def _event(key: bytes, payload: bytes) -> AgentEnvelope:
    return AgentEnvelope(entity_key=key, event_time_ms=1_000, external_event=payload)


def main() -> None:
    with beam.Pipeline() as pipeline:
        keyed = (
            pipeline
            | "Events"
            >> beam.Create(
                [
                    _event(b"k-model", b"review"),
                    _event(b"k-notify", b"NOTIFY"),
                    _event(b"k-broken", b"BROKEN"),
                ]
            )
            | "Key"
            >> beam.WithKeys(lambda e: e.entity_key).with_output_types(tuple[bytes, AgentEnvelope])
        )

        # region: outputs
        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")
        # endregion: outputs

        # region: traces
        # Traces show the atomic-commit rule from the outside. A committed
        # activation emits its whole span set — START, whatever happened in the
        # middle, END. The failed one emits a single ERROR event: the traces it
        # staged before raising were discarded with the rest of its effects,
        # exactly like its memory write.
        traced = outputs.traces | "TraceShape" >> beam.Map(
            lambda event: (event.entity_key, event.event_type)
        )
        assert_that(
            traced,
            equal_to(
                [
                    (b"k-model", TraceEvent.ACTIVATION_START),
                    (b"k-model", TraceEvent.LLM_CALL),
                    (b"k-model", TraceEvent.ACTIVATION_END),
                    (b"k-notify", TraceEvent.ACTIVATION_START),
                    (b"k-notify", TraceEvent.INTENT_EMITTED),
                    (b"k-notify", TraceEvent.ACTIVATION_END),
                    (b"k-broken", TraceEvent.ERROR),
                ]
            ),
            label="traces",
        )
        # endregion: traces

    print("four_outputs: ok")


if __name__ == "__main__":
    main()

website/examples/four_outputs.py — executed by the repository’s offline test tier.

The errors output — the operational reference for .errors, its reason taxonomy, and how to configure a dead-letter sink.
Traces — TraceEvent, the OpenTelemetry GenAI attribute conventions, and the exporters.
The effector — what happens to an element on .intents after it leaves the pipeline.
Correctness invariants — atomic commit, stated and tested.
RunAgentOutputs in the API reference.