symfonic.agent.fastapi.shared_chat

shared_chat

Explicit compatibility composition with one selected native chat handler.

create_shared_chat_router

create_shared_chat_router(agent: Any, host: Any, conversation: Any, *, prefix: str = '/api/v1', app: Any = None, typed_resume: bool = False, metrics_collector: Any = None) -> Any

Select native chat, durable legacy chats listing, and explicit refusals.

Pass conversation.transcript_reader() as agent: the authority's own reader, with no facade constructed. compatibility_reader(reader_factory) still composes the facade's public transcript read instead. Mount this root once, in place of the compatibility root. /stream is served by the same Agent host and authority with the compatibility chunk frames, an issued session id in the X-Session-ID header, and an error frame for an empty or tool-limit final (legacy sends done('') for empty). /stream/typed runs the same turn with the frozen typed frames, and refuses a pause terminal; resume refuses with 409 before calling the facade. No pause token or native host ticket is translated or assumed interchangeable. typed_resume=True opts into guarded durable resume: it requires the PostgreSQL authority and the supported human continuation host (refused at mount otherwise), lets /stream/typed alone pause, and serves /resume for those pauses on any worker. /chat and /stream still refuse the human capability. metrics_collector mounts /metrics/conversations and its detail over the collector the host's telemetry writes, answered only for the conversations this authority answers the principal's transcript for. /metrics/summary stays unmounted: it is a tenant aggregate. Only turns whose Agent composes telemetry over that collector are recorded; nothing here verifies that pairing per turn. Pass the outermost FastAPI app and include this router directly in it; mounting the selected app beneath another app is not governed.

Source code in src/symfonic/agent/fastapi/shared_chat.py

def create_shared_chat_router(
    agent: Any,
    host: Any,
    conversation: Any,
    *,
    prefix: str = "/api/v1",
    app: Any = None,
    typed_resume: bool = False,
    metrics_collector: Any = None,
) -> Any:
    """Select native chat, durable legacy chats listing, and explicit refusals.

    Pass ``conversation.transcript_reader()`` as ``agent``: the authority's own
    reader, with no facade constructed. ``compatibility_reader(reader_factory)``
    still composes the facade's public transcript read instead. Mount
    this root once, in place of the compatibility root. ``/stream`` is served
    by the same Agent host and authority with the compatibility chunk frames,
    an issued session id in the ``X-Session-ID`` header, and an error frame
    for an empty or tool-limit final (legacy sends ``done('')`` for empty).
    ``/stream/typed`` runs the same turn with the frozen typed frames, and
    refuses a pause terminal; resume refuses with 409 before calling the
    facade. No pause token or native host ticket is translated or assumed
    interchangeable.
    ``typed_resume=True`` opts into guarded durable resume: it requires the
    PostgreSQL authority and the supported human continuation host (refused
    at mount otherwise), lets ``/stream/typed`` alone pause, and serves
    ``/resume`` for those pauses on any worker. ``/chat`` and ``/stream``
    still refuse the human capability.
    ``metrics_collector`` mounts ``/metrics/conversations`` and its detail
    over the collector the host's telemetry writes, answered only for the
    conversations this authority answers the principal's transcript for.
    ``/metrics/summary`` stays unmounted: it is a tenant aggregate. Only turns
    whose Agent composes ``telemetry`` over that collector are recorded;
    nothing here verifies that pairing per turn.
    Pass the outermost FastAPI app and include this router directly in it;
    mounting the selected app beneath another app is not governed.
    """
    validate_authority(conversation)
    guarded = None
    if typed_resume is not False:
        if typed_resume is not True:
            raise TypeError("typed_resume must be a bool")
        from symfonic.agent.fastapi.shared_resume_guard import GuardedResume

        guarded = GuardedResume(host, conversation)
    compose = agent is _COMPOSE_AUTHORITY_READER
    if not compose and not conversation.owns_reader(agent):
        raise ValueError("shared chat requires the authority's compatibility reader")
    if app is None:
        raise ValueError("shared chat owner requires an explicit FastAPI app")
    if metrics_collector is not None:
        require_metrics_collector(metrics_collector)
    router = APIRouter(prefix=prefix, tags=["agent"])

    @router.get("/chats")
    async def list_chats(
        request: Request,
        scope: Any = Depends(get_tenant_scope),  # noqa: B008
    ) -> list[dict[str, Any]]:
        try:
            return await conversation.list_sessions(principal_of(request, scope))
        except Exception as exc:
            raise raise_mapped(exc, CRUD_POLICY) from exc

    async def refused(
        request: Request,
        scope: Any = Depends(get_tenant_scope),  # noqa: B008
    ) -> None:
        try:
            principal_of(request, scope)
            refuse_legacy_invocation()
        except Exception as exc:
            raise raise_mapped(exc, CHAT_POLICY) from exc

    @router.get("/chats/{session_id}/transcript")
    async def transcript(
        session_id: str,
        request: Request,
        scope: Any = Depends(get_tenant_scope),  # noqa: B008
    ) -> list[Any]:
        try:
            return await conversation.transcript(principal_of(request, scope), session_id)
        except Exception as exc:
            raise raise_mapped(exc, CRUD_POLICY) from exc

    if guarded is None:
        router.add_api_route("/resume/{pause_token}", refused, methods=["POST"])
    else:
        from symfonic.agent.fastapi.shared_resume import create_shared_resume_router

        router.include_router(create_shared_resume_router(guarded))
    if metrics_collector is not None:
        router.include_router(create_owned_metrics_router(conversation, metrics_collector))
    router.include_router(create_shared_stream_router(host, conversation))
    router.include_router(create_shared_typed_stream_router(host, conversation, guarded=guarded))
    router.include_router(
        create_native_chat_router(
            host,
            sessions=None,
            conversation=conversation,
            prefix="",
            app=app,
            _owner_claimed=True,
        )
    )
    if compose:
        # The authority holds one reader; a refused mount must not consume it.
        assert_chat_owner_available(app, prefix)
        assert_chat_family_routes_available(app, router)
        conversation.transcript_reader()
    claim_http_chat_owner(app, prefix, "shared")
    return router

mount_selected_chat_root

mount_selected_chat_root(app: Any, *, selection: str = 'agent', legacy_agent: Any = None, compatibility_reader: Any = None, host: Any = None, conversation: Any = None, prefix: str = '/api/v1', typed_resume: bool = False, metrics_collector: Any = None) -> Any

Compose exactly one chat family in the outermost served FastAPI app.

This is the supported native HTTP boundary. The caller must serve app directly; a selected child mounted below an external parent cannot inspect an earlier parent route and is outside this contract. Only selection="agent" is supported; legacy selection or legacy_agent raises the removed factory's versioned migration error without invoking it. The native default requires host and conversation and shares one authenticated conversation authority between /chat and /chats. With compatibility_reader omitted it composes conversation.transcript_reader(), the authority's own reader, so no SymfonicAgent is constructed. That reader is composed after the host, resume, metrics, production-auth, owner and route-conflict refusals, so a refused mount leaves the authority's single reader slot for a corrected mount on a fresh app; an explicitly supplied reader the authority composed (including compatibility_reader(factory)) is still accepted. Any other authority type is refused before a reader is composed.

Source code in src/symfonic/agent/fastapi/shared_chat.py

def mount_selected_chat_root(
    app: Any,
    *,
    selection: str = "agent",
    legacy_agent: Any = None,
    compatibility_reader: Any = None,
    host: Any = None,
    conversation: Any = None,
    prefix: str = "/api/v1",
    typed_resume: bool = False,
    metrics_collector: Any = None,
) -> Any:
    """Compose exactly one chat family in the outermost served FastAPI app.

    This is the supported native HTTP boundary. The caller must serve ``app``
    directly; a selected child mounted below an external parent cannot inspect
    an earlier parent route and is outside this contract. Only ``selection="agent"``
    is supported; legacy selection or ``legacy_agent`` raises the removed factory's
    versioned migration error without invoking it. The native default requires
    ``host`` and ``conversation`` and shares one authenticated conversation authority
    between ``/chat`` and ``/chats``. With ``compatibility_reader`` omitted it
    composes ``conversation.transcript_reader()``, the authority's own reader,
    so no ``SymfonicAgent`` is constructed. That reader is composed after the
    host, resume, metrics, production-auth, owner and route-conflict
    refusals, so a refused mount leaves the authority's single reader slot for
    a corrected mount on a fresh app; an explicitly supplied reader the
    authority composed (including ``compatibility_reader(factory)``) is still
    accepted. Any other authority type is refused before a reader is composed.
    """
    if not hasattr(app, "state") or not hasattr(app, "include_router"):
        raise TypeError("chat root requires a FastAPI app")
    if getattr(app.state, "symfonic_http_root_selection", None) is not None:
        raise ValueError("chat root already selected")
    if selection == "legacy" or legacy_agent is not None:
        from symfonic.agent.entry_point_removal import EntryPointRemovedError

        raise EntryPointRemovedError("create_agent_router")
    if selection != "agent":
        raise ValueError("chat selection must be 'agent'; legacy selection was removed")
    if any(item is None for item in (host, conversation)):
        raise TypeError("agent selection requires host and conversation")
    validate_authority(conversation)
    router = create_shared_chat_router(
        _COMPOSE_AUTHORITY_READER if compatibility_reader is None else compatibility_reader,
        host,
        conversation,
        prefix=prefix,
        app=app,
        typed_resume=typed_resume,
        metrics_collector=metrics_collector,
    )

    selected = assert_chat_family_routes_available(app, router)
    app.include_router(router)
    register_chat_family_routes(app, selected)
    app.state.symfonic_http_root_selection = selection
    return app