typed_route(delegate: Any, query: str, *, run_id: str, session_id: str = '', response_model: type[Any] | None = None, scope: Any = None, history: Sequence[Any] | None = None, attachments: Sequence[Any] | None = None, tenant_id: str | None = None, agent_depth: int | None = None) -> AsyncIterator[Any]
The typed projection of one invocation -- ST2's kernel route.
A third projection of one invocation, not a facade over the second. It
reaches :meth:InvocationKernel.stream_typed directly and never touches
stream: ST1 measured a facade over the StreamChunk producer and
found it loses event order, tool and message attribution and the usage
surface outright, and a wrapper cannot restore what the projection beneath
it already collapsed.
Independently governed, and deliberately not yet dispatched. This is the
typed route ST1's no_typed_cutover_dispatch row asks for, with its own
observability entry_point so a typed turn is attributable as one. It is
not registered in CAPABILITY_SWITCHES and no engine dispatch reads it:
a name earns a switch by owning an atomic segment of a turn that dispatch
actually consults, and adding one that dispatch does not read is precisely
the register-counted-as-switch mistake routes.py exists to prevent.
Building the route and flipping to it are separate, separately revertible
steps, and the flip is ST3's.
entry_point="stream_typed" rather than "stream": the two projections
are separately admitted, separately abandoned and separately measured, and
filing them under one name would make a typed regression invisible in the
streaming route's numbers.
delegate is the :class:~symfonic.agent.cutover.delegate.KernelDelegate
whose plan factory, observability suite and delegation scope this turn uses.
Passed rather than reconstructed, because a second plan factory would be a
second answer to what this agent compiles.
Source code in src/symfonic/agent/cutover/typed_route.py
| async def typed_route(
delegate: Any,
query: str,
*,
run_id: str,
session_id: str = "",
response_model: type[Any] | None = None,
scope: Any = None,
history: Sequence[Any] | None = None,
attachments: Sequence[Any] | None = None,
tenant_id: str | None = None,
agent_depth: int | None = None,
) -> AsyncIterator[Any]:
"""The typed projection of one invocation -- ST2's kernel route.
**A third projection of one invocation, not a facade over the second.** It
reaches :meth:`InvocationKernel.stream_typed` directly and never touches
``stream``: ST1 measured a facade over the ``StreamChunk`` producer and
found it loses event order, tool and message attribution and the usage
surface outright, and a wrapper cannot restore what the projection beneath
it already collapsed.
**Independently governed, and deliberately not yet dispatched.** This is the
typed route ST1's ``no_typed_cutover_dispatch`` row asks for, with its own
observability ``entry_point`` so a typed turn is attributable as one. It is
*not* registered in ``CAPABILITY_SWITCHES`` and no engine dispatch reads it:
a name earns a switch by owning an atomic segment of a turn that dispatch
actually consults, and adding one that dispatch does not read is precisely
the register-counted-as-switch mistake ``routes.py`` exists to prevent.
Building the route and flipping to it are separate, separately revertible
steps, and the flip is ST3's.
``entry_point="stream_typed"`` rather than ``"stream"``: the two projections
are separately admitted, separately abandoned and separately measured, and
filing them under one name would make a typed regression invisible in the
streaming route's numbers.
``delegate`` is the :class:`~symfonic.agent.cutover.delegate.KernelDelegate`
whose plan factory, observability suite and delegation scope this turn uses.
Passed rather than reconstructed, because a second plan factory would be a
second answer to what this agent compiles.
"""
request = turn_request(
query,
scope,
history,
attachments,
cap=delegate._history_cap, # noqa: SLF001
run_id=run_id,
)
observed = for_turn(
delegate._observability, # noqa: SLF001 - the delegate's own body, next door
run_id=run_id,
session_id=session_id,
tenant_id=tenant_id,
entry_point="stream_typed",
prompt=query,
request=request,
)
plan = delegate._plans.compile(response_model, sink_factory=observed) # noqa: SLF001
# Through the delegate's own kernel door, never a second one: IPL-1
# declares which modules may enter the invocation kernel, and adding this
# one would make two entry points into the single invocation path.
source = cast(
AsyncIterator[AgentEvent],
delegate.kernel_typed_stream(plan, request),
)
projected = typed_stream(source)
try:
# Held open for the whole drain: a hand-off happens mid-stream, and a
# scope closed after the first event would read depth ``0``.
async with delegate._delegation_scope(agent_depth): # noqa: SLF001
async for event in projected:
yield event
finally:
# Both, and in this order. ``projected`` owns closing ``source``, but a
# consumer that walked away leaves the projection itself suspended, and
# the release below is the abandoned run's only terminal at the
# observability seam.
await closing(projected)
if observed is not None:
await observed.release()
|