symfonic.agent.fastapi.chat_owner

chat_owner

Per-app ownership guard for compatibility-shaped HTTP chat roots.

assert_chat_family_routes_available

assert_chat_family_routes_available(app: Any, router: Any) -> tuple[Any, ...]

Refuse existing root routes that would win selected family requests.

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

def assert_chat_family_routes_available(
    app: Any, router: Any
) -> tuple[Any, ...]:
    """Refuse existing root routes that would win selected family requests."""
    selected = _concrete_routes(router.routes)
    for route in selected:
        for method in getattr(route, "methods", ()) or ():
            for path in _probes(route.path):
                if _first_endpoint(app, _scope(method, path))[0]:
                    raise ValueError(f"chat family route conflict: {method} {path}")
    return selected

assert_chat_owner_available

assert_chat_owner_available(app: Any, prefix: str) -> str

Refuse an already claimed or served POST {prefix}/chat; claim nothing.

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

def assert_chat_owner_available(app: Any, prefix: str) -> str:
    """Refuse an already claimed or served ``POST {prefix}/chat``; claim nothing."""
    if not hasattr(app, "state") or not hasattr(app, "router"):
        raise TypeError("chat owner requires a FastAPI app")
    path = f"{prefix.rstrip('/')}/chat"
    if path in (getattr(app.state, _STATE_KEY, None) or {}) or path in _post_chat_paths(app):
        raise ValueError(f"chat owner already selected for {path}")
    return path

claim_http_chat_owner

claim_http_chat_owner(app: Any, prefix: str, owner: str) -> None

Reserve an explicit root and fail closed on a later mixed co-mount.

A factory with an explicit app rejects claims for the same direct path. Opt-in roots additionally check the final composed router at startup and request time, covering outer include prefixes and app-less legacy routers mounted after the reservation. The passed app must be the outermost ASGI root: a selected app mounted beneath another router cannot inspect parent routes and is unsupported. Historical app-less legacy use is unchanged.

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

def claim_http_chat_owner(app: Any, prefix: str, owner: str) -> None:
    """Reserve an explicit root and fail closed on a later mixed co-mount.

    A factory with an explicit app rejects claims for the same direct path.
    Opt-in roots additionally check the final composed router at startup and
    request time, covering outer include prefixes and app-less legacy routers
    mounted after the reservation. The passed app must be the outermost ASGI
    root: a selected app mounted beneath another router cannot inspect parent
    routes and is unsupported. Historical app-less legacy use is unchanged.
    """
    if app is None:
        if owner == "shared":
            raise ValueError("shared chat owner requires an explicit FastAPI app")
        return
    path = assert_chat_owner_available(app, prefix)
    claims = getattr(app.state, _STATE_KEY, None)
    if claims is None:
        claims = {}
    if owner != "legacy":
        _install_composed_guard(app)
    claims[path] = owner
    setattr(app.state, _STATE_KEY, claims)

register_chat_family_routes

register_chat_family_routes(app: Any, selected: tuple[Any, ...]) -> None

Guard the selected route family against later sibling shadowing.

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

def register_chat_family_routes(app: Any, selected: tuple[Any, ...]) -> None:
    """Guard the selected route family against later sibling shadowing."""
    setattr(app.state, _SELECTED_KEY, selected)
    _assert_selected_routes_unique(app)
    _install_composed_guard(app)