Skip to content

symfonic.agent.fastapi.shared_resume_turn

shared_resume_turn

One selected-root resume: what it publishes, and what it settles on failure.

The final answer is read from the kernel's raw done result through the same answer_of_result the selected typed stream publishes with, never from the projected ResponseCompleteEvent text. A resumed turn that stops at the tool limit or with an empty answer is therefore refused exactly as that stream refuses it, and publishes nothing.

Once the host reports that this request won the token's single-use claim, a failure that published nothing -- an error, a late revision conflict, a client disconnect -- moves the pause row to abandoned under a shielded cancel scope. The token is already spent, so this never makes the pause redeemable again; it only stops a dead row from staying pending.

ResumedTurn 

ResumedTurn(guarded: Any, principal: Any, session_id: str, scope: Any, claim_id: str)

State shared by the guarded admission, the frames, and the cleanup.

Source code in src/symfonic/agent/fastapi/shared_resume_turn.py 
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self, guarded: Any, principal: Any, session_id: str, scope: Any, claim_id: str
) -> None:
    self._guarded = guarded
    self._principal = principal
    self.session_id = session_id
    self._scope = scope
    self.claim_id = claim_id
    self._observer: _FinalObserver | None = None
    self.claimed = False
    self.settled = False

admission 

admission(*, budgeted: bool) -> Any

The host's pre-claim admission hook, observing every raw event.

Source code in src/symfonic/agent/fastapi/shared_resume_turn.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
def admission(self, *, budgeted: bool) -> Any:
    """The host's pre-claim admission hook, observing every raw event."""

    def admit(agent: Any, issuer_scope: Any) -> _Admitted:
        budget = None
        if budgeted:
            from symfonic.platform.resume_budget import admit_resume_budget

            budget = admit_resume_budget(agent, issuer_scope)
        self._observer = _FinalObserver(None if budget is None else budget.event_guard)
        return _Admitted(None if budget is None else budget.prepared_plan, self._observer)

    return admit

release async 

release() -> None

Abandon a claimed pause that published nothing; idempotent, shielded.

Source code in src/symfonic/agent/fastapi/shared_resume_turn.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def release(self) -> None:
    """Abandon a claimed pause that published nothing; idempotent, shielded."""
    if not self.claimed or self.settled:
        return
    self.settled = True
    with anyio.CancelScope(shield=True):
        try:
            await self._guarded.conversation.pauses.abandon(
                self._principal, self.session_id, self.claim_id
            )
        except Exception:
            logging.getLogger(__name__).warning(
                "could not abandon a claimed pause whose resume published nothing",
                exc_info=True,
            )

settle async 

settle(event: Any) -> None

Publish a completion, or record a second pause, before its frame.

Source code in src/symfonic/agent/fastapi/shared_resume_turn.py 
 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
async def settle(self, event: Any) -> None:
    """Publish a completion, or record a second pause, before its frame."""
    from symfonic.core.contracts.types import ResponseCompleteEvent

    pauses = self._guarded.conversation.pauses
    if isinstance(event, ResponseCompleteEvent):
        from symfonic.platform.transport_conversation import UnusableFinal, answer_of_result

        if self._observer is None or not self._observer.done:
            raise UnusableFinal("resumed turn completed without a kernel result")
        answer = answer_of_result(self._observer.result)
        await pauses.publish(self._principal, self.session_id, self.claim_id, answer)
        self.settled = True
        return
    token = getattr(event, "pause_token", None)
    if token is not None:
        pause = await pauses.pending(self._principal, self.session_id, self.claim_id)
        await self._guarded.record_pause(
            self._principal,
            self.session_id,
            self._scope,
            token,
            pause.query,
            pause.base_revision,
            supersedes=self.claim_id,
        )
        self.settled = True