`symfonic.agent.fastapi.transport`¶

transport ¶

Transport ports for the FastAPI/SSE host (T4.1.1 transport-ports.md).

Rows 17–19 of the port registry live here: the invocation codec, the SSE frame adapter, and the error mapper. A handler that imports from this package and calls one facade or service method is a mapper (TRN-1); a handler that decides a status, assembles a scope, or re-implements the error table is not.

There is deliberately no PlatformServices object to hand these out: a registry that resolves collaborators on demand passes every layering check and rebuilds the monolith one indirection down.

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.

DetailMode ¶

Bases: Enum

Where a frame's detail string comes from (EMAP-7).

Server-side faults never echo their own message: an operator-side configuration error names hosts, DSNs, and occasionally credentials, and the person who needs that text is reading the log, not the socket.

ErrorFrame `dataclass` ¶

ErrorFrame(status: int, detail: str, code: str | None = None, headers: Mapping[str, str] | None = None)

The mapped result: status, detail, machine-readable code, headers.

payload ¶

payload() -> dict[str, object]

The SSE/JSON body shape (STR-3).

status is always present; code appears only when the error carried one, so a client that reads detail sees exactly what it saw before and a client that wants to branch finally can.

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

def payload(self) -> dict[str, object]:
    """The SSE/JSON body shape (STR-3).

    ``status`` is always present; ``code`` appears only when the error
    carried one, so a client that reads ``detail`` sees exactly what it saw
    before and a client that wants to branch finally can.
    """
    body: dict[str, object] = {"detail": self.detail, "status": self.status}
    if self.code is not None:
        body["code"] = self.code
    return body

ErrorPolicy `dataclass` ¶

ErrorPolicy(name: str, fallback_status: int, opaque_detail: str, collapse_server_errors: bool = False)

An entry point's declared difference from every other one (EMAP-3).

The four invocation entry points differ in exactly one way — what an unmapped failure becomes — and that difference is a value passed to the mapper, not a fifth copy of the mapping.

HttpErrorMapper ¶

ErrorMapper (port row 19): map(error, policy) -> ErrorFrame.

Sync, total, and pure: no I/O, no ambient state, no logging. A mapper that logged would make the same failure appear twice in different words.

map ¶

map(error: BaseException, policy: ErrorPolicy) -> ErrorFrame

Map any exception onto the frame the entry point should emit.

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

def map(self, error: BaseException, policy: ErrorPolicy) -> ErrorFrame:
    """Map any exception onto the frame the entry point should emit."""
    # The two scope branches are class-level decisions with verbatim
    # details; they precede code lookup because ``SecurityScopeError`` is
    # a ``SymfonicAgentError`` and would otherwise fall through uncoded.
    if isinstance(error, SecurityScopeError):
        return ErrorFrame(status=403, detail=SCOPE_DENIED_DETAIL, code="forbidden")
    if isinstance(error, ScopeValidationError):
        return ErrorFrame(status=400, detail=str(error), code="bad_request")

    # The platform resolver's refusals, mapped here rather than at the
    # dependency that catches them. Same reason as the two branches above:
    # a status code written in a handler is one the mapper cannot keep
    # consistent, which is what ``STATUS_LITERAL_CEILING`` ratchets down.
    auth_frame = _platform_auth_frame(error)
    if auth_frame is not None:
        return auth_frame

    if not isinstance(error, _TAXONOMY):
        return ErrorFrame(status=500, detail=OPAQUE_SERVER_DETAIL, code=None)

    code = self._code_of(error)
    rule = CODE_TO_STATUS.get(code) if code is not None else None
    if rule is None:
        return self._fallback(error, policy, code)

    status = rule.active
    if (
        status >= 500
        and policy.collapse_server_errors
        and isinstance(error, SymfonicAgentError)
    ):
        # Shipped ``/resume`` behaviour: a server-side status becomes the
        # entry point's 400 and the message is echoed.
        return ErrorFrame(
            status=policy.fallback_status, detail=str(error), code=code,
        )
    return ErrorFrame(
        status=status,
        detail=self._detail(error, policy, rule),
        code=code,
        headers=rule.headers,
    )

InvocationCodec ¶

Decode the wire input; encode the result. Sync, pure, total.

decode_invocation ¶

decode_invocation(payload: Any, *, scope: Any, is_admin: bool = False, principal_id: str | None = None) -> InvocationRequest

Project a validated AgentRequest onto an invocation value.

Identity fields present on the shipped AgentRequest model (tenant_id) are ignored, not merged: the scope is whatever the resolver derived. They are ignored rather than rejected because the field is part of the frozen request model and the shipped clients send it (TRN-5 outranks TRN-8's "rejected" wording here; the security property — identity never comes from the body — is identical).

principal_id falls back to the resolved scope's tenant when the host authenticated nobody by name. That is the convention :meth:~symfonic.platform.scope.HeaderScopeResolver.resolve already ships — facts.get("principal_id") or tenant_id — reused rather than re-decided, so an admin claim always names some resolved subject and never the empty string. It is still never read from the body.

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

def decode_invocation(
    self,
    payload: Any,
    *,
    scope: Any,
    is_admin: bool = False,
    principal_id: str | None = None,
) -> InvocationRequest:
    """Project a validated ``AgentRequest`` onto an invocation value.

    Identity fields present on the shipped ``AgentRequest`` model
    (``tenant_id``) are **ignored**, not merged: the scope is whatever the
    resolver derived. They are ignored rather than rejected because the
    field is part of the frozen request model and the shipped clients send
    it (TRN-5 outranks TRN-8's "rejected" wording here; the security
    property — identity never comes from the body — is identical).

    ``principal_id`` falls back to the resolved scope's tenant when the host
    authenticated nobody by name. That is the convention
    :meth:`~symfonic.platform.scope.HeaderScopeResolver.resolve` already
    ships — ``facts.get("principal_id") or tenant_id`` — reused rather than
    re-decided, so an admin claim always names *some* resolved subject and
    never the empty string. It is still never read from the body.
    """
    return InvocationRequest(
        query=payload.query,
        scope=scope,
        is_admin=is_admin,
        session_id=getattr(payload, "session_id", None),
        attachments=getattr(payload, "attachments", None),
        principal_id=str(
            principal_id or getattr(scope, "tenant_id", "") or ""
        ),
    )

encode_result ¶

encode_result(result: Any) -> Mapping[str, Any] | Any

Encode a facade result for the wire.

/chat declares response_model=AgentResponse, so FastAPI already owns the serialization and the codec must not double-encode; returning the value unchanged is the encoding. The method exists so a second host has one place to put its own.

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

def encode_result(self, result: Any) -> Mapping[str, Any] | Any:
    """Encode a facade result for the wire.

    ``/chat`` declares ``response_model=AgentResponse``, so FastAPI already
    owns the serialization and the codec must not double-encode; returning
    the value unchanged is the encoding. The method exists so a second host
    has one place to put its own.
    """
    return result

InvocationRequest `dataclass` ¶

InvocationRequest(query: str, scope: Any, is_admin: bool = False, session_id: str | None = None, attachments: Sequence[Any] | None = None, principal_id: str = '')

The transport-neutral invocation value the facade is called with.

scope and is_admin come from the resolved principal and never from the body (TRN-8). session_id is an opaque string that round-trips verbatim, colons and all (TRN-9) — transport does not parse or re-key it.

admin_claim ¶

admin_claim() -> Any

The claim a host binds around the call, or None for no claim.

Built here rather than in each handler so the three invocation routes cannot bind three different things. None when the request is not an administrator's, because binding "not an admin" and binding nothing are the same fact and the second one costs no object.

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

def admin_claim(self) -> Any:
    """The claim a host binds around the call, or ``None`` for no claim.

    Built here rather than in each handler so the three invocation routes
    cannot bind three different things. ``None`` when the request is not an
    administrator's, because binding "not an admin" and binding nothing are
    the same fact and the second one costs no object.
    """
    if not self.is_admin:
        return None
    from symfonic.agent.cutover.authority import (  # noqa: PLC0415
        AdminAuthority,
    )

    return AdminAuthority(
        principal_id=self.principal_id, is_admin=True
    )

as_kwargs ¶

as_kwargs() -> dict[str, Any]

Keyword arguments for the facade's invocation methods.

is_admin is deliberately not among them (TA8.43, C1-K). The keyword was replaced on the 11.0 line: the facade refuses is_admin=True by name, because a bit the calling code passes is a bit the calling code chose, and the authority that gates the tenant budget breaker has to come from an authenticated principal instead. The field stays on this value because it is the resolved principal's fact (TRN-8) and it is what the host binds -- see :func:~symfonic.agent.cutover.authority.bind_admin_authority and the handlers in symfonic.agent.fastapi.router. Dropping the field would have lost the fact; passing it on would have kept the retired keyword.

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

def as_kwargs(self) -> dict[str, Any]:
    """Keyword arguments for the facade's invocation methods.

    ``is_admin`` is deliberately **not** among them (TA8.43, C1-K). The
    keyword was replaced on the 11.0 line: the facade refuses
    ``is_admin=True`` by name, because a bit the calling code passes is a
    bit the calling code chose, and the authority that gates the tenant
    budget breaker has to come from an authenticated principal instead. The
    field stays on this value because it *is* the resolved principal's fact
    (TRN-8) and it is what the host binds -- see
    :func:`~symfonic.agent.cutover.authority.bind_admin_authority` and the
    handlers in ``symfonic.agent.fastapi.router``. Dropping the field would
    have lost the fact; passing it on would have kept the retired keyword.
    """
    return {
        "query": self.query,
        "attachments": self.attachments,
        "scope": self.scope,
        "session_id": self.session_id,
    }

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)

StatusRule `dataclass` ¶

StatusRule(active: int, target: int | None = None, detail_mode: DetailMode = DetailMode.MESSAGE, detail_override: str | None = None, headers: Mapping[str, str] | None = None)

One row of the EMAP-4 table.

target is the status the row should carry and does not yet: three classes deserve 409/503/504 and surface as 500 today, and flipping them is a tier-1 behaviour change that COMPAT-POL puts at MAJOR (EMAP-5). Carrying the target in the table is what makes the eventual flip a scheduled change rather than a surprise, and what lets a test assert the intent today.

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)

symfonic.agent.fastapi.transport¶

transport ¶

AdapterDeclaration dataclass ¶

DetailMode ¶

ErrorFrame dataclass ¶

payload ¶

ErrorPolicy dataclass ¶

HttpErrorMapper ¶

map ¶

InvocationCodec ¶

decode_invocation ¶

encode_result ¶

InvocationRequest dataclass ¶

admin_claim ¶

as_kwargs ¶

SseFrameAdapter ¶

terminated property ¶

frames async ¶

StatusRule dataclass ¶

error_frame ¶

single_error_stream async ¶

`symfonic.agent.fastapi.transport`¶

AdapterDeclaration `dataclass` ¶

ErrorFrame `dataclass` ¶

ErrorPolicy `dataclass` ¶

InvocationRequest `dataclass` ¶

terminated `property` ¶

frames `async` ¶

StatusRule `dataclass` ¶

single_error_stream `async` ¶