symfonic.agent.fastapi.transport.sse

sse

STR — the SSE frame adapter (port registry row 18).

The buffering contract is T2.3.1's (BP-1..14) and is consumed unchanged; what lives here is only what happens at the HTTP boundary: how an invocation event becomes a wire frame, what happens when the run raises, and what happens when the client walks away.

Three properties this module exists to make true in one place rather than in four generators:

• At most one adapter-authored error frame, and nothing after it (STR-4). The engine's own terminal event (done/cancelled) still comes from the event stream — the adapter never invents a frame the wire did not have, so the frozen frame shapes (STR-2, TRN-5) are byte-compatible.
• Disconnect is cancellation, not an error (STR-5, EMAP-8). A client that hangs up gets no error frame and produces no 5xx, and the source stream is closed rather than left to a garbage collector.
• Pre-stream failures still speak SSE (STR-6): a body that will not parse is one error frame and a closed stream, never an opaque connection close.

AdapterDeclaration dataclass

AdapterDeclaration(full_policy: str = 'shed', terminal_policy: str = 'reserve', max_frames: int = 256, max_bytes: int = 8 * 1024 * 1024, text_reconstruction: str = 'terminal-only', heartbeat_seconds: float | None = None)

BP-1: a buffer that is not declared is a contract violation.

These are the values T2.3.1 §5 fixes for the shipped SSE host. They are data rather than comments so that T4.1.4's fault suite and T4.4.1's rule checks can read them instead of trusting a docstring.

SseFrameAdapter

SseFrameAdapter(policy: ErrorPolicy, *, mapper: HttpErrorMapper = ERROR_MAPPER, declaration: AdapterDeclaration = SSE_ADAPTER_DECLARATION)

EventFrameAdapter (row 18): invocation events → transport frames.

One instance per connection (STR-1, STR-8): the adapter holds per-stream state, so sharing one between connections would let a slow consumer decide another tenant's terminal frame.

Source code in src/symfonic/agent/fastapi/transport/sse.py

def __init__(
    self,
    policy: ErrorPolicy,
    *,
    mapper: HttpErrorMapper = ERROR_MAPPER,
    declaration: AdapterDeclaration = SSE_ADAPTER_DECLARATION,
) -> None:
    self._policy = policy
    self._mapper = mapper
    self._declaration = declaration
    self._terminated = False

terminated property

terminated: bool

True once this adapter has authored its terminal error frame.

frames async

frames(source: AsyncIterator[Any], encode: Callable[[Any], dict[str, str]]) -> AsyncIterator[dict[str, str]]

Project source onto wire frames; never re-order, never re-emit.

encode is the route's frozen frame shape (STR-2) — the adapter owns the lifecycle, the route owns the shape, and neither owns both.

Source code in src/symfonic/agent/fastapi/transport/sse.py

async def frames(
    self,
    source: AsyncIterator[Any],
    encode: Callable[[Any], dict[str, str]],
) -> AsyncIterator[dict[str, str]]:
    """Project ``source`` onto wire frames; never re-order, never re-emit.

    ``encode`` is the route's frozen frame shape (STR-2) — the adapter owns
    the lifecycle, the route owns the shape, and neither owns both.
    """
    try:
        async for event in source:
            yield encode(event)
    except (asyncio.CancelledError, GeneratorExit):
        # EMAP-8: a cancellation is not an error. Reporting one as a 5xx is
        # how availability dashboards start lying.
        raise
    except BaseException as exc:  # noqa: BLE001 - mapped, then re-shaped
        frame = self._mapper.map(exc, self._policy)
        if frame.status >= 500:
            logger.exception(
                "SSE stream failed (entry_point=%s)", self._policy.name,
            )
        self._terminated = True
        yield error_frame(frame)
    finally:
        await self._aclose(source)

error_frame

error_frame(frame: ErrorFrame) -> dict[str, str]

Render an :class:ErrorFrame as the one SSE error shape (STR-3).

Source code in src/symfonic/agent/fastapi/transport/sse.py

def error_frame(frame: ErrorFrame) -> dict[str, str]:
    """Render an :class:`ErrorFrame` as the one SSE error shape (STR-3)."""
    return {"event": "error", "data": json.dumps(frame.payload())}

single_error_stream async

single_error_stream(frame: ErrorFrame) -> AsyncIterator[dict[str, str]]

One error frame, then end of stream (STR-6).

Source code in src/symfonic/agent/fastapi/transport/sse.py

async def single_error_stream(frame: ErrorFrame) -> AsyncIterator[dict[str, str]]:
    """One ``error`` frame, then end of stream (STR-6)."""
    yield error_frame(frame)