symfonic.agent.cutover.typed_projection

typed_projection

Projecting the kernel's own event stream onto the typed StreamEvent vocabulary.

ST2 (TA8.29) builds the shape ST1 chose: a dedicated typed projection over the kernel, not a facade over the public stream() producer. ST1 measured that facade and found it loses event order, tool and message attribution, and the usage surface outright, because those are exactly what the StreamChunk projection collapses. Routing the typed path through that producer would reintroduce every one of those losses one layer down, so this module reads the kernel's events and nothing else.

The seam is the same one :mod:~symfonic.agent.cutover.projection sits on -- read a finished kernel value, write the shape a caller already holds -- and the two are deliberately siblings rather than one module with a mode: projection writes StreamChunk and this one writes StreamEvent, and they are not projections of each other.

Translation is not interpretation, here too. Where the kernel reported nothing, nothing is written: a run whose provider reported no tokens emits no UsageEvent rather than one carrying zeros, because a zero that looks measured is a fabrication that survives into a bill.

as_typed_events

as_typed_events(event: AgentEvent) -> tuple[StreamEvent, ...]

Project one kernel event onto zero or more typed events, in order.

Order within the returned tuple is part of the contract, not an accident: a tool call announces itself before its arguments, and a tool result arrives before the timing that describes it. A caller that yields the tuple in order preserves the kernel's total order, because the kernel assigns a dense monotonic index and nothing here reorders or renumbers.

Terminal kinds (done, error, cancelled) return (): what a terminal means is the caller's decision -- an error propagates, a cancellation emits no normal terminal at all -- and burying either in a projection would put a failure on the success path.

ask_user and interrupt are the exception, and they are terminal too (TA8.34). They return their structured event, because a pause is the one terminal that carries something the consumer has to act on rather than something the caller has to decide about: the questions, and the token they are answered with. Leaving that to the caller would mean a paused run reaching a human-in-the-loop consumer as a stream that simply ended, which is ST1's typed_interrupt_translation row failing on the new route the same way it never worked on it before.

Source code in src/symfonic/agent/cutover/typed_projection.py

def as_typed_events(event: AgentEvent) -> tuple[StreamEvent, ...]:
    """Project one kernel event onto zero or more typed events, in order.

    Order within the returned tuple is part of the contract, not an accident:
    a tool call announces itself before its arguments, and a tool result
    arrives before the timing that describes it. A caller that yields the tuple
    in order preserves the kernel's total order, because the kernel assigns a
    dense monotonic index and nothing here reorders or renumbers.

    Terminal kinds (``done``, ``error``, ``cancelled``) return ``()``: what a
    terminal means is the caller's decision -- an error propagates, a
    cancellation emits no normal terminal at all -- and burying either in a
    projection would put a failure on the success path.

    ``ask_user`` and ``interrupt`` are the exception, and they are terminal too
    (TA8.34). They return their structured event, because a pause is the one
    terminal that carries something the consumer has to *act on* rather than
    something the caller has to decide about: the questions, and the token they
    are answered with. Leaving that to the caller would mean a paused run
    reaching a human-in-the-loop consumer as a stream that simply ended, which
    is ST1's ``typed_interrupt_translation`` row failing on the new route the
    same way it never worked on it before.
    """
    if event.interrupt is not None:
        return interrupt_events(event)
    if event.kind == "thinking":
        return (ThinkingDeltaEvent(text=event.text or ""),) if event.text else ()
    if event.kind == "text_delta":
        return (TextDeltaEvent(text=event.text or ""),) if event.text else ()
    if event.kind == "tool_call" and event.tool_call is not None:
        call = event.tool_call
        started: list[StreamEvent] = [
            ToolCallStartEvent(
                tool_name=call.name,
                tool_call_id=call.id,
                # ``executing`` and not ``announced``: the kernel emits this
                # when the run reserves and dispatches the call, which is the
                # runtime phase, not the model's own generation revealing it.
                phase="executing",
            )
        ]
        if call.arguments:
            started.append(
                ToolCallDeltaEvent(
                    tool_call_id=call.id, args_delta=_encode(call.arguments)
                )
            )
        return tuple(started)
    if event.kind == "tool_result" and event.tool_call is not None:
        call = event.tool_call
        return (
            ToolResultEvent(
                tool_call_id=call.id,
                result=call.result if call.result is not None else (call.error or ""),
            ),
            ExtensionEvent(
                type=TOOL_TIMING_EXTENSION,
                payload={
                    "tool_call_id": call.id,
                    "tool_name": call.name,
                    "duration_ms": call.duration_ms,
                    "is_error": call.error is not None,
                },
                event_id=call.id,
            ),
        )
    return ()

interrupt_events

interrupt_events(event: AgentEvent) -> tuple[StreamEvent, ...]

Project one paused kernel event onto the typed pause vocabulary.

Returns () for anything that is not a pause, so the caller can hand every event through without branching on the kind twice.

Source code in src/symfonic/agent/cutover/typed_interrupts.py

def interrupt_events(event: AgentEvent) -> tuple[StreamEvent, ...]:
    """Project one paused kernel event onto the typed pause vocabulary.

    Returns ``()`` for anything that is not a pause, so the caller can hand
    every event through without branching on the kind twice.
    """
    pending = event.interrupt
    if pending is None:
        return ()
    if event.kind == "ask_user":
        return (_ask_user(event, pending),)
    return (_generic(event, pending),)

response_complete

response_complete(text: str, duration_ms: float) -> ResponseCompleteEvent

The typed terminal, built from what the turn actually produced.

Source code in src/symfonic/agent/cutover/typed_projection.py

def response_complete(text: str, duration_ms: float) -> ResponseCompleteEvent:
    """The typed terminal, built from what the turn actually produced."""
    return ResponseCompleteEvent(text=text, duration_ms=round(duration_ms, 1))

typed_stream async

typed_stream(source: AsyncIterator[AgentEvent]) -> AsyncIterator[StreamEvent]

Drain one kernel event stream into typed events, in the kernel's order.

The whole typed route in one loop, and everything it does not do is as load-bearing as what it does.

• It never renumbers or reorders. The kernel assigns a dense monotonic index; a projection that sorted or batched would break the ordering guarantee every consumer joins on.
• On an error terminal it emits nothing and lets the run's own exception propagate on the next step. Synthesising a typed error event would put a failure on the success path of every existing consumer, and wrapping the exception would hide the class an adopter already catches.
• On a cancelled terminal it emits nothing -- specifically, no ResponseCompleteEvent. An incomplete turn may not emit the terminal a completed turn emits, and inventing a final response for a run that was walked away from is fabrication, not cleanup.
• On an ask_user or interrupt terminal this loop emits the pause and no ResponseCompleteEvent, for the same reason: a paused run produced no final answer. The pause itself is emitted rather than swallowed because it is the whole point of the terminal -- the loop above yields whatever as_typed_events returns, so the structured event lands in the kernel's own order, ahead of nothing and behind every delta the run produced before it stopped.

That is this function's behaviour, not the public surface's: one layer up, SymfonicAgent._stream_typed_kernel_impl yields a ResponseCompleteEvent on every non-GeneratorExit exit, a pause included, because the legacy body does and the pause path's parity is the acceptance criterion. So a consumer of stream_typed does see a terminal after a pause. The difference is deliberate and recorded in .agent/team/framework-refactor/evidence/RET-PREP/hk1-pause-surface.md §6 ("One difference is stated rather than discovered"). * The source is closed on every exit path, including the one where the consumer walks away mid-stream. An abandoned async generator otherwise finalises whenever the garbage collector reaches it, in nobody's task.

Source code in src/symfonic/agent/cutover/typed_projection.py

async def typed_stream(
    source: AsyncIterator[AgentEvent],
) -> AsyncIterator[StreamEvent]:
    """Drain one kernel event stream into typed events, in the kernel's order.

    The whole typed route in one loop, and everything it does *not* do is as
    load-bearing as what it does.

    * It never renumbers or reorders. The kernel assigns a dense monotonic
      index; a projection that sorted or batched would break the ordering
      guarantee every consumer joins on.
    * On an ``error`` terminal it emits **nothing** and lets the run's own
      exception propagate on the next step. Synthesising a typed error event
      would put a failure on the success path of every existing consumer, and
      wrapping the exception would hide the class an adopter already catches.
    * On a ``cancelled`` terminal it emits **nothing** -- specifically, no
      ``ResponseCompleteEvent``. An incomplete turn may not emit the terminal a
      completed turn emits, and inventing a final response for a run that was
      walked away from is fabrication, not cleanup.
    * On an ``ask_user`` or ``interrupt`` terminal *this loop* emits the
      **pause** and no ``ResponseCompleteEvent``, for the same reason: a paused
      run produced no final answer. The pause itself is emitted rather than
      swallowed because it is the whole point of the terminal -- the loop above
      yields whatever ``as_typed_events`` returns, so the structured event lands
      in the kernel's own order, ahead of nothing and behind every delta the run
      produced before it stopped.

      That is this function's behaviour, **not** the public surface's: one layer
      up, ``SymfonicAgent._stream_typed_kernel_impl`` yields a
      ``ResponseCompleteEvent`` on every non-``GeneratorExit`` exit, a pause
      included, because the legacy body does and the pause path's parity is the
      acceptance criterion. So a consumer of ``stream_typed`` *does* see a
      terminal after a pause. The difference is deliberate and recorded in
      ``.agent/team/framework-refactor/evidence/RET-PREP/hk1-pause-surface.md``
      §6 ("One difference is stated rather than discovered").
    * The source is closed on every exit path, including the one where the
      consumer walks away mid-stream. An abandoned async generator otherwise
      finalises whenever the garbage collector reaches it, in nobody's task.
    """
    try:
        async for event in source:
            for projected in as_typed_events(event):
                yield projected
            if event.kind == "done":
                usage = usage_event(event.result)
                if usage is not None:
                    yield usage
                result = event.result
                yield response_complete(
                    result.text if result is not None else "",
                    result.duration_ms if result is not None else 0.0,
                )
    finally:
        aclose = getattr(source, "aclose", None)
        if aclose is not None:
            await aclose()

usage_event

usage_event(result: AgentResult[Any] | None) -> UsageEvent | None

The run's token usage, or None when the provider reported none.

None is the load-bearing case. TokenUsage defaults every count to 0 and documents 0 as unreported, so emitting a UsageEvent for an all-zero report would publish "this run cost nothing" with the same confidence as a real measurement. Cost and usage are attributed only where real evidence exists.

The guard reads input_tokens and output_tokens only, and deliberately not total_tokens: UsageEvent carries just those two fields, so a provider reporting TokenUsage(total_tokens=1200) with no split would produce UsageEvent(input_tokens=0, output_tokens=0) -- two zeros that look measured, published because real evidence existed somewhere this event cannot carry. An unsplit total is dropped here rather than rendered as zeros; carrying it would need a channel that can say "total, unsplit", which this event has no field for.

Source code in src/symfonic/agent/cutover/typed_projection.py

def usage_event(result: AgentResult[Any] | None) -> UsageEvent | None:
    """The run's token usage, or ``None`` when the provider reported none.

    ``None`` is the load-bearing case. ``TokenUsage`` defaults every count to
    ``0`` and documents ``0`` as *unreported*, so emitting a ``UsageEvent`` for
    an all-zero report would publish "this run cost nothing" with the same
    confidence as a real measurement. Cost and usage are attributed only where
    real evidence exists.

    The guard reads ``input_tokens`` and ``output_tokens`` **only**, and
    deliberately not ``total_tokens``: ``UsageEvent`` carries just those two
    fields, so a provider reporting ``TokenUsage(total_tokens=1200)`` with no
    split would produce ``UsageEvent(input_tokens=0, output_tokens=0)`` -- two
    zeros that look measured, published *because* real evidence existed
    somewhere this event cannot carry. An unsplit total is dropped here rather
    than rendered as zeros; carrying it would need a channel that can say
    "total, unsplit", which this event has no field for.
    """
    if result is None:
        return None
    usage = result.usage
    if not (usage.input_tokens or usage.output_tokens):
        return None
    return UsageEvent(
        input_tokens=usage.input_tokens, output_tokens=usage.output_tokens
    )