Skip to content

symfonic.agent.fastapi.shared_typed_stream

shared_typed_stream

Selected-root /stream/typed over the shared native conversation authority.

The frames are the frozen typed wire: SSE event is the StreamEvent class name and data its JSON payload (encode_typed_event, the encoder the legacy route uses), projected from kernel events by typed_stream. The turn is the selected /stream turn (shared_turn): the same pre-model refusals, X-Session-ID issuance, publication of a usable final before the ResponseCompleteEvent frame, and discard of an issued empty session. No SymfonicAgent is constructed or called here.

Declared divergences from the legacy root, the same ones the selected /stream declares:

  • the issued session id travels only in the X-Session-ID header;
  • a turn that completes with an empty answer, or stops at the tool limit, ends in the stream policy's error frame and publishes nothing, where legacy emits ResponseCompleteEvent with text=''.

By default a pause terminal (ask_user/interrupt) is refused with the error frame instead of forwarding its token, because the default root refuses /resume and the token could not be redeemed; admission already refuses the capabilities that pause, so that is a fail-closed guard. With guarded durable resume (shared_resume_guard.GuardedResume) this route alone admits the configured human capability, records the pause durably before its frame, and the selected /resume redeems it on any worker. A pause minted with the built-in one-field ask_user registration is framed as the generic InterruptEvent (native_pause_events), as a recovered continuation frames it; a multi-question AskUserRequest registration frames as AskUserQuestionEvent. Event classes and order come from the kernel projection; full wire parity is not claimed.

create_shared_typed_stream_router 

create_shared_typed_stream_router(host: Any, conversation: Any, *, guarded: Any = None) -> APIRouter

Mount /stream/typed for the shared root; the caller owns prefix and app.

guarded (a verified GuardedResume) admits its human capability and forwards a pause only after recording it for the selected /resume.

Source code in src/symfonic/agent/fastapi/shared_typed_stream.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def create_shared_typed_stream_router(
    host: Any, conversation: Any, *, guarded: Any = None
) -> APIRouter:
    """Mount ``/stream/typed`` for the shared root; the caller owns prefix and app.

    ``guarded`` (a verified ``GuardedResume``) admits its human capability and
    forwards a pause only after recording it for the selected ``/resume``.
    """
    router = APIRouter()

    @router.post("/stream/typed")
    async def stream_typed(
        request: AgentRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> Any:
        if guarded is None:
            turn, agent, admission = await open_shared_turn(
                host, conversation, request, raw_request, scope
            )
            events = _without_pause(turn.events(agent, admission))
        else:
            turn, agent, admission = await open_shared_turn(
                host, conversation, request, raw_request, scope, admit=guarded.admit
            )
            # The built-in one-field ask_user frames as the generic interrupt,
            # exactly as a recovered continuation frames it.
            raw = guarded.resumable_only(native_pause_events(turn.events(agent, admission)))
            typed = guarded.recording_pauses(typed_stream(raw), turn)
        if guarded is None:
            typed = typed_stream(events)
        return TurnResponse(
            turn,
            SseFrameAdapter(STREAM_POLICY).frames(typed, encode_typed_event),
            headers=turn.headers,
        )

    return router