Checkpoint commands: recording a paused request, and reading it back.
The two metadata keys below are frozen compatibility contracts. A token minted
by the pre-extraction engine points at ask_user_request:<tool_call_id> or
interrupt_payload:<interrupt_id>, and a resume that looked anywhere else
would answer "possible tampering" for every pause in flight at cutover.
Which key a claim uses is derived from the claim, not from which subsystem is
running — that is the whole of the ask_user / generic-interrupt split, reduced
to one expression.
PausePayloadStore
PausePayloadStore(*, commands: Any = None, binder: ScopeBinder | None = None)
Records and recovers the paused request through one checkpoint port.
A deployment with no checkpointer is supported for recording — the write
is best-effort, exactly as it was, because not every saver takes pending
writes and a failed metadata write must never take down a healthy pause.
Recovery is not best-effort: without the recorded payload there is nothing
to check the request hash against, and guessing is the failure SEC-PTK-5
exists to prevent.
Source code in src/symfonic/capabilities/human/checkpoints.py
| def __init__(self, *, commands: Any = None, binder: ScopeBinder | None = None) -> None:
self._commands = commands
self._binder = binder or ScopeBinder()
|
load
async
load(claims: PauseClaims) -> Any
Recover the paused request and check it against the token's hash.
Source code in src/symfonic/capabilities/human/checkpoints.py
| async def load(self, claims: PauseClaims) -> Any:
"""Recover the paused request and check it against the token's hash."""
if self._commands is None:
raise InteractionConfigurationError(
"no checkpoint commands are wired, so the paused request cannot "
"be recovered; a resume with nothing to verify against would be "
"a resume of whatever the caller sent"
)
raw = await self._commands.read_metadata(
thread_id=claims.thread_id,
checkpoint_id=claims.checkpoint_id,
key=payload_key(claims),
)
if raw is None:
self._refuse_missing(claims)
try:
payload = json.loads(raw)
except (ValueError, TypeError) as exc:
raise PayloadBindingError(
f"the recorded payload for {claims.jti!r} is unreadable; it is "
"refused rather than resumed from a partial parse"
) from exc
# The recorded text is checked first, and it is the text the mint
# hashed — for a model payload, and for every token the pre-extraction
# engine minted from ``request.model_dump_json()``. Re-hashing the
# parsed value is the fallback for a store that normalises what it was
# handed, and it is what refuses when neither form matches.
if not self._binder.payload_body_matches(claims, body=raw):
self._binder.check_payload(claims, payload=payload)
return payload
|
record
async
record(claims: PauseClaims, payload: Any, *, task_id: str) -> bool
Best effort, and it reports which effort it made.
Source code in src/symfonic/capabilities/human/checkpoints.py
| async def record(self, claims: PauseClaims, payload: Any, *, task_id: str) -> bool:
"""Best effort, and it reports which effort it made."""
if self._commands is None:
return False
# The bytes that were hashed, not a second serialisation of the same
# object: the resume path checks the recorded text against the token's
# request_hash, and only identical text can match it.
body = self._binder.request_body(payload)
try:
await self._commands.write_metadata(
thread_id=claims.thread_id,
checkpoint_id=claims.checkpoint_id,
key=payload_key(claims),
value=body,
task_id=task_id,
)
except PayloadBindingError:
raise
except Exception:
# Some savers do not implement pending writes. The resume path
# still has the registered schema, so this is a reported
# degradation rather than a failed pause.
return False
return True
|
payload_key
payload_key(claims: PauseClaims) -> str
Where this pause's payload lives in checkpoint metadata.
Source code in src/symfonic/capabilities/human/checkpoints.py
| def payload_key(claims: PauseClaims) -> str:
"""Where this pause's payload lives in checkpoint metadata."""
if claims.name == ASK_USER:
if not claims.tool_call_id:
raise PayloadBindingError(_NO_TOOL_CALL_ID)
return f"{ASK_USER_KEY_PREFIX}{claims.tool_call_id}"
if not claims.interrupt_id:
raise PayloadBindingError(
f"the {claims.name!r} pause carries no interrupt_id, so its payload "
"has no metadata key; it cannot be recorded or recovered"
)
return f"{INTERRUPT_KEY_PREFIX}{claims.interrupt_id}"
|
require_payload_key_inputs(name: str, *, tool_call_id: str = '') -> None
The half of :func:payload_key that is checkable before a token exists.
A pause that mints first and discovers only afterwards that its payload has
nowhere to live leaves an issued, unconsumable row in the issuance ledger —
one that counts against the legacy drain proof until it expires. The generic
half is not checkable here: the interrupt id is minted with the claims.
Source code in src/symfonic/capabilities/human/checkpoints.py
| def require_payload_key_inputs(name: str, *, tool_call_id: str = "") -> None:
"""The half of :func:`payload_key` that is checkable before a token exists.
A pause that mints first and discovers only afterwards that its payload has
nowhere to live leaves an issued, unconsumable row in the issuance ledger —
one that counts against the legacy drain proof until it expires. The generic
half is not checkable here: the interrupt id is minted with the claims.
"""
if name == ASK_USER and not tool_call_id:
raise PayloadBindingError(_NO_TOOL_CALL_ID)
|