symfonic.agent.fastapi.transport.codec

codec

InvocationCodec (port registry row 17) — TRN-7.

/chat, /stream, /stream/typed, and /resume are four projections of one invocation path (TRN-4). They share this codec because four handlers — and, the moment a second host exists, more than four — map the same shape. The CRUD routes deliberately do not: for those the pydantic model already is the codec, and interposing a second one would be ceremony with no consumer.

Nothing here performs I/O or reads ambient state: decode_invocation is a pure projection of a validated request plus an already-derived scope, which is what lets the same values feed a CLI or a test recorder unchanged.

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,
    }