Skip to content

symfonic.kernel.streaming

streaming

Compatibility entry point for the shared runner's typed projection.

A projection, not a second pipeline: same plan, same ports, same context, same round bookkeeping. Streaming and non-streaming drifting apart is the classic way a framework acquires two answers to "what did the run produce", so the differences here are confined to which model method is called and which events come out.

Cancellation is left strictly alone. asyncio.CancelledError is a BaseException, nothing here catches it, and generator cleanup runs — which is why kind="cancelled" is not emitted: inventing a yield during teardown would claim an observation the kernel did not make (EVT-9/EVT-10).

That last point is a constraint on finalizers, not only on this module. When a consumer abandons the stream, aclose() throws GeneratorExit at the suspended yield and ctx.teardown() runs while it unwinds. A finalizer may await there, but a finalizer that caused this generator to yield again would raise RuntimeError: async generator ignored GeneratorExit. Finalizers clean up; they do not emit.

stream_events async 

stream_events(kernel: Any, plan: InvocationPlan, request: TurnRequest) -> AsyncIterator[KernelEvent]

Yield the public typed projection, retained for import compatibility.

Source code in src/symfonic/kernel/streaming.py 
33
34
35
36
37
38
39
40
41
async def stream_events(
    kernel: Any, plan: InvocationPlan, request: TurnRequest
) -> AsyncIterator[KernelEvent]:
    """Yield the public typed projection, retained for import compatibility."""
    from symfonic.kernel.runner import InvocationRunner

    source = InvocationRunner().events(kernel, plan, request, stream_model=True)
    async for event in TypedStreamAdapter.attach(plan).project(source):
        yield event