Skip to content

symfonic.agent.fastapi.shared_resume

shared_resume

Selected-root POST /resume/{pause_token} over guarded durable resume.

Mounted only when GuardedResume verified the composition. Refusals happen before the token is spent and in this order: the verified principal must match the request scope; the token must authenticate (unknown 404, tampered 401, expired 410); its signed scope must be the request scope exactly; and the conversation authority must hold a pending pause this root recorded for this principal, tenant, session and claim, still at its turn's revision. The claim is re-checked inside the host's guarded redemption, so a pause that went stale between the check and the claim refuses too.

The body is the ticket's registered response schema as raw JSON (for example {"answer": "yes"}). Frames are the frozen typed wire. A completion is published to the paused conversation before its ResponseCompleteEvent frame; a second pause is recorded against the same turn before its frame. A token is single-use: a resumed turn that then fails is an at-most-once outcome and publishes nothing, and its pause row is settled as abandoned (shared_resume_turn). The final is read from the kernel result, so a resumed turn at the tool limit or with an empty answer is refused as the typed stream refuses it.

Deltas are framed as the turn streams, before publication. If publication then refuses -- for example a 409 because another worker published to the conversation after the claim-time check -- the client has already received text that is not, and will not be, in the transcript.

create_shared_resume_router 

create_shared_resume_router(guarded: Any) -> APIRouter

Mount /resume/{pause_token}; the caller owns prefix and app.

Source code in src/symfonic/agent/fastapi/shared_resume.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def create_shared_resume_router(guarded: Any) -> APIRouter:
    """Mount ``/resume/{pause_token}``; the caller owns prefix and app."""
    host, conversation = guarded.host, guarded.conversation
    router = APIRouter()

    @router.post("/resume/{pause_token}")
    async def resume(
        pause_token: str,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> Any:
        try:
            principal = principal_of(raw_request, scope)
            ticket = await host.ticket(pause_token)
            session_id = await _pending_session(guarded, principal, pause_token, scope, ticket)
        except Exception as exc:
            return _error(ERROR_MAPPER.map(_native_error(exc), RESUME_POLICY))
        claim_id = ticket.claim_id
        try:
            body = await raw_request.json()
        except Exception:
            return _error(ErrorFrame(status=400, detail="Invalid JSON body"))
        try:
            answer = ticket.response_schema.model_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"))
        budgeted = get_budget_tracker() is not None
        if budgeted:
            await check_budget_for_tenant(raw_request, str(scope.tenant_id))
        resumed = ResumedTurn(guarded, principal, session_id, scope, claim_id)

        async def owner_guard(
            issuer_scope: Any, signed_session: str, signed_claim: str, claiming: bool
        ) -> None:
            if (
                issuer_scope.scope_path != scope.scope_path
                or signed_session != session_id
                or signed_claim != claim_id
            ):
                raise SymfonicAgentError(
                    "continuation belongs to another conversation", code="forbidden"
                )
            if claiming:
                await conversation.pauses.pending(principal, session_id, claim_id)

        async def events() -> Any:
            source = host.resume_stream(
                pause_token,
                answer,
                scope=scope,
                owner_guard=owner_guard,
                admission=resumed.admission(budgeted=budgeted),
                on_claimed=resumed.mark_claimed,
            )
            try:
                async with aclosing(source):
                    with bind_conversation_turn(ConversationTurn(session_id, scope)):
                        async for event in source:
                            await resumed.settle(event)
                            yield event
            except Exception as exc:
                raise _native_error(exc) from exc
            finally:
                await resumed.release()

        return EventSourceResponse(
            SseFrameAdapter(RESUME_POLICY).frames(events(), encode_typed_event),
        )

    return router