def create_native_resume_router(
host: Any, *, prefix: str = "/api/v1", app: Any = None,
conversation_owner: Any = None,
) -> APIRouter:
"""Mount native tokens on the compatibility class-name SSE wire.
The host must expose authenticated ticket recovery and scoped typed
continuation. This does not adapt legacy tokens or switch an existing
router. The body is raw JSON for the native ticket's registered response
schema (for example, {"answer": "yes"}), not a ResumeRequest wrapper.
Budgeted deployments admit the actual issuer agent before redemption and
guard its raw usage before typed projection.
"""
if not all(callable(getattr(host, name, None)) for name in ("ticket", "resume_stream")):
raise TypeError("native resume host must provide ticket and resume_stream")
if conversation_owner is not None:
from symfonic.platform.agent_host import AgentHost
if not callable(getattr(conversation_owner, "authorize_working", None)):
raise TypeError("native resume owner must provide authorize_working")
# Only the reviewed human service derives the session from signed claims.
if type(host) is not AgentHost or not host._supports_resume_admission():
raise TypeError("owned native resume requires the supported human host")
_check_production_auth_gate(app)
router = APIRouter(prefix=prefix, tags=["agent"])
@router.post("/resume/{pause_token}")
async def resume(
pause_token: str,
raw_request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> Any:
from sse_starlette.sse import EventSourceResponse
def error(frame: ErrorFrame) -> Any:
return EventSourceResponse(single_error_stream(frame))
budgeted = get_budget_tracker() is not None
if budgeted:
from symfonic.platform.agent_host import AgentHost
if type(host) is not AgentHost or not host._supports_resume_admission():
return error(ErrorFrame(
status=503, detail="native resume cannot attest budget usage",
))
principal = None
if conversation_owner is not None:
from symfonic.agent.fastapi.identity import request_principal
from symfonic.platform.values import AuthenticatedPrincipal
from symfonic.platform.working_access import WorkingOwnerGrant
principal = request_principal(raw_request)
if (not isinstance(principal, AuthenticatedPrincipal)
or principal.derivation.get("verified") is not True
or principal.derivation.get("principal_from_verifier") is not True):
return error(ErrorFrame(
status=401, detail="resume requires a verified principal",
))
if (principal.scope.tenant_id != scope.tenant_id
or principal.scope.sub_tenant_id != scope.sub_tenant_id
or principal.scope.namespace != scope.namespace):
return error(ErrorFrame(
status=403, detail="resume principal scope mismatch",
))
try:
ticket = await host.ticket(pause_token)
# Full materialised identity preserves nested scope boundaries.
if (scope.scope_path != ticket.scope.scope_path and (
conversation_owner is not None
or not bool(getattr(ticket, "cross_scope_allowed", False)))):
return error(ErrorFrame(
status=403, detail="continuation belongs to another authenticated scope",
))
if conversation_owner is not None:
owner_principal = cast(AuthenticatedPrincipal, principal)
session_id = getattr(ticket, "session_id", None)
claim_id = getattr(ticket, "claim_id", None)
if not isinstance(session_id, str) or not session_id:
return error(ErrorFrame(
status=503, detail="continuation owner identity unavailable",
))
if not isinstance(claim_id, str) or not claim_id:
return error(ErrorFrame(
status=503, detail="continuation claim identity unavailable",
))
try:
async with asyncio.timeout(5.0):
grant = await conversation_owner.authorize_working(
owner_principal, session_id, "resume",
)
except Exception:
return error(ErrorFrame(
status=503, detail="continuation owner unavailable",
))
expected = WorkingOwnerGrant(
owner_principal.principal_id, owner_principal.scope.tenant_id,
session_id, "resume",
)
if type(grant) is not WorkingOwnerGrant or grant != expected:
return error(ErrorFrame(
status=403, detail="continuation belongs to another conversation",
))
except Exception as exc:
return error(ERROR_MAPPER.map(_native_error(exc), RESUME_POLICY))
try:
body = await raw_request.json()
except Exception:
return error(ErrorFrame(status=400, detail="Invalid JSON body"))
validate = getattr(getattr(ticket, "response_schema", None), "model_validate", None)
if not callable(validate):
return error(ErrorFrame(
status=400, detail="continuation has no registered response schema",
))
try:
answer = validate(body)
except ValidationError:
return error(ErrorFrame(status=400, detail="Invalid resume payload"))
except Exception:
return error(ErrorFrame(status=500, detail="Resume payload validation failed"))
# An allowed cross-scope answer still resumes and bills the issuer.
if budgeted:
from symfonic.platform.telemetry import _tenant_of
await check_budget_for_tenant(raw_request, _tenant_of(ticket.scope))
owner_guard = None
if conversation_owner is not None:
async def owner_guard(
issuer_scope: Any, signed_session: str, signed_claim_id: str,
claiming: bool,
) -> None:
if (issuer_scope.scope_path != scope.scope_path
or signed_session != session_id or signed_claim_id != claim_id):
raise SymfonicAgentError(
"continuation belongs to another conversation", code="forbidden",
)
if not claiming:
return
try:
async with asyncio.timeout(5.0):
fresh_grant = await conversation_owner.authorize_working(
owner_principal, signed_session, "resume",
)
except Exception as exc:
raise _NativeResumeUnavailable(
"continuation owner unavailable",
) from exc
if type(fresh_grant) is not WorkingOwnerGrant or fresh_grant != expected:
raise SymfonicAgentError(
"continuation belongs to another conversation", code="forbidden",
)
guard_arg = {"owner_guard": owner_guard} if owner_guard is not None else {}
async def events() -> Any:
if budgeted:
from symfonic.platform.resume_budget import admit_resume_budget
def admission(agent: Any, issuer_scope: Any) -> Any:
try:
return admit_resume_budget(agent, issuer_scope)
except Exception as exc:
if getattr(exc, "code", None) == "native_stream_unavailable":
raise _NativeResumeUnavailable(
"native resume cannot attest budget usage",
) from exc
raise _NativeResumeFailure("resume preparation failed") from exc
source = host.resume_stream(
pause_token, answer, scope=scope, admission=admission, **guard_arg,
)
else:
source = host.resume_stream(pause_token, answer, scope=scope, **guard_arg)
try:
async with aclosing(source):
async for event in source:
yield event
except Exception as exc:
raise _native_error(exc) from exc
return EventSourceResponse(
SseFrameAdapter(RESUME_POLICY).frames(_visible_events(events()), encode_typed_event),
)
return router