Skip to content

symfonic.agent.fastapi.shared_resume_guard

shared_resume_guard

Mount-time proof that a selected-root pause can be resumed on any worker.

The selected root admits the human capability only when every store that a cross-worker resume reads is shared and durable: the PostgreSQL conversation authority (revision compare-and-set and pause records), the reviewed AgentHost + HumanContinuationService pair (signed session/claim recovery and the guarded claim), and a pause-token redemption store and checkpoint port that both declare durability, with redemption shared across replicas. Anything less refuses at mount, never at the first resume.

GuardedResume 

GuardedResume(host: Any, conversation: Any)

The verified composition the selected typed stream and /resume share.

Source code in src/symfonic/agent/fastapi/shared_resume_guard.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(self, host: Any, conversation: Any) -> None:
    from symfonic.platform.agent_host import AgentHost
    from symfonic.platform.conversation_pg import PostgresConversationAuthority

    if type(conversation) is not PostgresConversationAuthority:
        raise TypeError("selected-root resume requires the PostgreSQL conversation authority")
    if type(host) is not AgentHost or not host._supports_resume_admission():
        raise TypeError("selected-root resume requires the supported human continuation host")
    capability = host._continuation._capability
    if not callable(getattr(capability, "decode_token", None)) or not (
        _redemption_is_deployment_wide(capability)
    ):
        raise TypeError(
            "selected-root resume requires durable, deployment-wide pause-token "
            "redemption and durable checkpoints"
        )
    self.host = host
    self.conversation = conversation
    self.capability = capability

admit 

admit(agent: Any, tenant_id: str, query: str) -> Any

Admit the configured human capability on the typed stream only.

Source code in src/symfonic/agent/fastapi/shared_resume_guard.py 
66
67
68
def admit(self, agent: Any, tenant_id: str, query: str) -> Any:
    """Admit the configured human capability on the typed stream only."""
    return admit_native_stream(agent, tenant_id, query, human=self.capability)

bound_ticket async 

bound_ticket(token: Any, session_id: str, scope: Any) -> Any

Authenticate a token and require its signed session and exact scope.

Source code in src/symfonic/agent/fastapi/shared_resume_guard.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
async def bound_ticket(self, token: Any, session_id: str, scope: Any) -> Any:
    """Authenticate a token and require its signed session and exact scope."""
    if not isinstance(token, str) or not token:
        raise _forbidden("pause carries no redeemable token")
    ticket = await self.host.ticket(token)
    claim = getattr(ticket, "claim_id", None)
    if (
        ticket.session_id != session_id
        or not isinstance(claim, str)
        or not claim
        or ticket.scope.scope_path != scope.scope_path
    ):
        raise _forbidden("pause token is bound to another conversation")
    return ticket

record_pause async 

record_pause(principal: Any, session_id: str, scope: Any, token: Any, query: str, revision: Any, *, supersedes: str | None = None) -> None

File the pause before its token can reach the client.

Source code in src/symfonic/agent/fastapi/shared_resume_guard.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
async def record_pause(
    self,
    principal: Any,
    session_id: str,
    scope: Any,
    token: Any,
    query: str,
    revision: Any,
    *,
    supersedes: str | None = None,
) -> None:
    """File the pause before its token can reach the client."""
    if type(revision) is not int:
        raise _forbidden("paused turn has no checked conversation revision")
    ticket = await self.bound_ticket(token, session_id, scope)
    await self.conversation.pauses.record(
        principal,
        session_id,
        ticket.claim_id,
        query,
        revision,
        expires_at=self._signed_expiry(token),
        supersedes=supersedes,
    )

recording_pauses async 

recording_pauses(typed: Any, turn: Any) -> Any

Record each projected pause durably before its frame is yielded.

Source code in src/symfonic/agent/fastapi/shared_resume_guard.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def recording_pauses(self, typed: Any, turn: Any) -> Any:
    """Record each projected pause durably before its frame is yielded."""
    from contextlib import aclosing

    async with aclosing(typed):
        async for event in typed:
            token = getattr(event, "pause_token", None)
            if token is not None:
                await self.record_pause(
                    turn.principal,
                    turn.session_id,
                    turn.scope,
                    token,
                    turn.query,
                    turn.revision,
                )
                turn.settle()
            yield event

resumable_only async 

resumable_only(events: Any) -> Any

Refuse a raw pause the capability did not report as resumable.

Source code in src/symfonic/agent/fastapi/shared_resume_guard.py 
117
118
119
120
121
122
123
124
125
126
async def resumable_only(self, events: Any) -> Any:
    """Refuse a raw pause the capability did not report as resumable."""
    from contextlib import aclosing

    async with aclosing(events):
        async for event in events:
            paused = event.kind in _PAUSE_KINDS or event.interrupt is not None
            if paused and (event.interrupt is None or event.interrupt.resumable is not True):
                raise _forbidden("the paused turn cannot be resumed")
            yield event