symfonic.capabilities.human

human

Human-interaction and pause-token capability (T3.4.3).

Pausing a run for a human used to be spread across seven places: a tool module, a graph node, two contract modules, a token facade with a process-wide default manager, three consumption stores, and two nearly-identical resume methods on the agent class — each with its own scope check and its own metadata key. The rules were only readable by reading all seven, and two of them could only be reached by constructing an agent.

Everything a pause decides now lives here:

• Registrations — :mod:.registration. ask_user is a registration with three facts on it, not a second subsystem.
• Binding — :mod:.binding. The scope, session, and request checks, and the two legacy hash formulas they are checked against.
• Tokens — :mod:.tokens. Mint, authenticate, bind, consume — one ordering, in one place, with the claim last.
• Consumption — :mod:.consumption. Single-winner redemption as one atomic conditional write on the adopter's backend (LIB-TL-2), plus the in-process reference implementation and the SEC-PTK-7 durability check.
• The ledger — :mod:.ledger. The operated platform's authoritative issuance/consumption record, which is also its linearization point, over the issuance tables of :mod:.issuance and with the verdict of :mod:.drain.
• Checkpoint commands — :mod:.checkpoints. Recording and recovering the paused request under the two legacy metadata keys, unchanged.
• The checkpointer role — :mod:.turnstate and :mod:.threads (HK2). The paused turn -- its prompt and its messages -- recorded under a third, additive key on the same port, and the tenant:sub:session thread key both routes now derive from one place.
• Pausing — :mod:.pausing. The pre-tool handler that stops a run, and the ordering that makes resumable earned rather than announced.
• Resume — :mod:.resume. One command contract for both families.

Authenticity is consumed, never redefined. The signed envelope, the signing key lifecycle (including library mode's locally derived key), and the pin-less legacy-artifact policy are T1.2.6/T2.3.6's, reached through :class:~.contracts.EnvelopeSignerPort. This package imports nothing but itself — the T1.2.1 matrix puts capability → runtime-service at no — and a suite asserts it contains no hmac, no secrets, and no key material.

One winner, in both deployment modes. Operated platforms bind the issuance ledger, whose atomic claim is the linearization point; library deployments bind a conditional-write store over the adopter's persistence backend. Nothing above :mod:.consumption can tell which it got, and the operated-only purposes are reachable only through the object that has them.

AskUserPayload

Bases: BaseModel

What the agent asks. One field, because one question is the whole tool.

AskUserResponse

Bases: BaseModel

What the person answers.

BackendIssuanceRecords

BackendIssuanceRecords(backend: Any, *, durable: bool | None = None)

The same three tables on the operator's persistence backend.

Source code in src/symfonic/capabilities/human/issuance.py

def __init__(self, backend: Any, *, durable: bool | None = None) -> None:
    missing = [
        verb for verb in BACKEND_VERBS if not callable(getattr(backend, verb, None))
    ]
    if missing:
        raise InteractionConfigurationError(
            f"{type(backend).__name__} cannot hold the issuance table: it is "
            f"missing {', '.join(missing)}. Redemption needs one atomic "
            "conditional write and nothing more, but the issuance record and "
            "the retirement horizon have to be read back — by the next worker, "
            "and by this one after a restart — so this is the wider port"
        )
    self._backend = backend
    self._durable = declared_durability(backend) if durable is None else durable

CallBindingError

Bases: PauseTokenError

The token was minted for another call of the same run (HK2).

ask_user correlates on tool_call_id and a registered interrupt on interrupt_id; this is the refusal for both. A run that paused twice -- routine for an onboarding agent still working a checklist -- has two live tokens whose run, session and scope are all identical, so the call id is the only axis that separates them. Verifying the other three and not this one would let the second question's answer be recorded against the first.

CheckpointCommandPort

Bases: Protocol

The four checkpoint verbs a pause needs, separated from any saver.

The legacy engine reached into a LangGraph saver from three call sites with four differently-shaped config dicts. This is that surface, named.

CheckpointLostError

Bases: HumanInteractionError

The checkpoint is gone and the backend never promised to keep it.

ConditionalWriteConsumption

ConditionalWriteConsumption(backend: Any, *, durable: bool | None = None, clock: Callable[[], float] = time.time)

LIB-TL-2 — one atomic conditional write on the adopter's backend.

The port has exactly one verb because a port that also offered a read would invite the read-then-write sequence the contract forbids. claim is a single await: there is no branch in this method that could become two round-trips later.

Source code in src/symfonic/capabilities/human/consumption.py

def __init__(
    self,
    backend: Any,
    *,
    durable: bool | None = None,
    clock: Callable[[], float] = time.time,
) -> None:
    if not callable(getattr(backend, "insert_if_absent", None)):
        raise InteractionConfigurationError(
            f"{type(backend).__name__} does not offer insert_if_absent(key, "
            "record); single-use consumption needs one atomic conditional "
            "write, and a read-then-write sequence cannot provide it"
        )
    self._backend = backend
    self._clock = clock
    # A backend that does not say, and an adopter who does not say either,
    # is treated as volatile — see :func:`declared_durability`.
    self._durable = declared_durability(backend) if durable is None else durable

ConditionalWritePort

Bases: Protocol

The adopter persistence backend's atomic conditional insert (LIB-TL-2).

INSERT ... ON CONFLICT DO NOTHING RETURNING jti in SQL, a unique-_id insert in Mongo, a single-lock section in the in-memory reference. One method on purpose: a port that also offered a read would invite the read-then-write sequence the contract forbids.

ConsumptionDurabilityError

Bases: InteractionConfigurationError

Durable checkpoints with a volatile jti store (SEC-PTK-7).

ConsumptionPort

Bases: Protocol

Single-winner redemption of one jti (SEC-PTK-3).

One method, and its contract is the whole of the acceptance criterion: concurrent callers competing for one jti see exactly one True, and the decision is reached by a single atomic operation rather than by a read followed by a write.

ConsumptionRecord dataclass

ConsumptionRecord(jti: str, scope_hash: str, name: str = ASK_USER, consumed_at: float = 0.0)

What a redemption writes. Ids only — no payload, no answer.

CrossScopeRedemption dataclass

CrossScopeRedemption(name: str, interrupt_id: str, jti: str, expected_scope_hash: str, presented_scope_hash: str, at: float)

An audit record for the one relaxation the contract permits.

DrainProof dataclass

DrainProof(drained: bool, outstanding: int, reason: str, horizon: float | None = None, deadline: float | None = None, scope: str = PROCESS)

Whether every legacy-pinned token has drained, and why not if not.

DuplicateInteractionError

Bases: InteractionRegistrationError

Two registrations claim one name; there is no defensible winner.

EnvelopeSignerPort

Bases: Protocol

T2.3.6's EnvelopeSigner: mint an authenticated envelope, verify one.

verify returns the invocation pin and raises on tampering, on an unsupported layout, and on an unavailable keyset. This package never inspects a signature, holds key material, or decides what "authentic" means; it only decides what a token binds.

HumanInteractionCapability

HumanInteractionCapability(*, registry: InteractionRegistry, tokens: PauseTokenService, payloads: PausePayloadStore, resumes: ResumeService, binding: Callable[[], Any] | None = None, encode_token: Callable[[Any], str] | None = None, decode_token: Callable[[str], Any] | None = None, turns: TurnCheckpointStore | None = None)

Registrations, tokens, checkpoint commands, and resume, wired once.

Source code in src/symfonic/capabilities/human/capability.py

def __init__(
    self,
    *,
    registry: InteractionRegistry,
    tokens: PauseTokenService,
    payloads: PausePayloadStore,
    resumes: ResumeService,
    binding: Callable[[], Any] | None = None,
    encode_token: Callable[[Any], str] | None = None,
    decode_token: Callable[[str], Any] | None = None,
    turns: TurnCheckpointStore | None = None,
) -> None:
    self.registry = registry
    self.tokens = tokens
    self.payloads = payloads
    self.resumes = resumes
    self.binding = binding
    self.encode_token = encode_token
    #: The inverse of :attr:`encode_token`. Without it a pause taken on the
    #: kernel route has no public reader -- see ``agent.cutover.continuation``.
    self.decode_token = decode_token
    #: The checkpointer role (HK2). ``None`` is a deployment that can ask a
    #: question and never continue the run that asked it, which is what
    #: every deployment was before TA8.35 -- so it is a state, not a bug.
    self.turns = turns or TurnCheckpointStore()

active property

active: bool

True when at least one interaction is registered.

A deployment that registered none must not advertise ask_user: a tool whose only possible outcome is a refusal is an invitation to pause a run that can never be resumed.

compile classmethod

compile(registrations: Iterable[InteractionRegistration], *, signer: Any, ttl: TTLPolicy, consumption: Any = None, ledger: Any = None, checkpoints: Any = None, clock: Callable[[], float] = time.time, pinless_policy: Any = None, audit: Callable[[CrossScopeRedemption], None] | None = None, require_durable: bool = True, binding: Callable[[], Any] | None = None, encode_token: Callable[[Any], str] | None = None, decode_token: Callable[[str], Any] | None = None) -> HumanInteractionCapability

Wire the capability. Every argument is a port or a policy.

Source code in src/symfonic/capabilities/human/capability.py

@classmethod
def compile(
    cls,
    registrations: Iterable[InteractionRegistration],
    *,
    signer: Any,
    ttl: TTLPolicy,
    consumption: Any = None,
    ledger: Any = None,
    checkpoints: Any = None,
    clock: Callable[[], float] = time.time,
    pinless_policy: Any = None,
    audit: Callable[[CrossScopeRedemption], None] | None = None,
    require_durable: bool = True,
    binding: Callable[[], Any] | None = None,
    encode_token: Callable[[Any], str] | None = None,
    decode_token: Callable[[str], Any] | None = None,
) -> HumanInteractionCapability:
    """Wire the capability. Every argument is a port or a policy."""
    registry = InteractionRegistry()
    for registration in registrations:
        registry.register(registration)
    tokens = PauseTokenService(
        signer=signer,
        ttl=ttl,
        consumption=consumption,
        ledger=ledger,
        clock=clock,
        pinless_policy=pinless_policy,
    )
    payloads = PausePayloadStore(commands=checkpoints)
    # One port, two records. A deployment cannot end up with durable
    # requests and volatile turn state, or the reverse, and then find out
    # only when somebody answers a question.
    turns = TurnCheckpointStore(commands=checkpoints)
    if require_durable:
        require_durable_consumption(
            consumption if consumption is not None else ledger,
            checkpoints_durable=payloads.durable,
        )
    return cls(
        registry=registry,
        tokens=tokens,
        payloads=payloads,
        resumes=ResumeService(
            tokens=tokens,
            registry=registry,
            payloads=payloads,
            clock=clock,
            audit=audit,
            turns=turns,
        ),
        binding=binding,
        encode_token=encode_token,
        decode_token=decode_token,
        turns=turns,
    )

contribute

contribute(request: CapabilityRequest) -> CapabilityContribution

Offer the pause point to the turn being compiled (HK1, TA8.34).

A stage and a tool, and :mod:~symfonic.capabilities.human.contribution holds the reasoning for why both: the tool is how the model asks, the pre-tool stage is where the run actually stops, and only the stage sees the reserved call id a pause has to be bound to.

Contributing nothing is a real answer, the same one DelegationCapability gives for a parent with no children. Three states produce it, and each is a deployment that cannot serve a pause: nothing registered, no per-run :class:PauseBinding resolver, or no way to render a minted envelope as the opaque token a consumer answers with. Advertising ask_user in any of them would invite the model to stop a run that could never be resumed -- and a run stopped by a pause nobody can answer is strictly worse than one that never stopped. The contribution still carries the capability's name, so a plan records that human interaction was folded and found nothing rather than that it was never folded.

request is read for its grants and found to need none. Minting a token, recording its payload and writing the issuance row all go through this capability's own ports, which the composition root wired; none of them is an :class:~symfonic.kernel.contracts.effects.EffectFamily the plan grants, and declaring one it was not granted is refused at fold rather than at the point of the effect (STG-8).

Source code in src/symfonic/capabilities/human/capability.py

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Offer the pause point to the turn being compiled (HK1, TA8.34).

    **A stage and a tool**, and :mod:`~symfonic.capabilities.human.contribution`
    holds the reasoning for why both: the tool is how the model asks, the
    ``pre-tool`` stage is where the run actually stops, and only the stage
    sees the reserved call id a pause has to be bound to.

    **Contributing nothing is a real answer**, the same one
    ``DelegationCapability`` gives for a parent with no children. Three
    states produce it, and each is a deployment that cannot serve a pause:
    nothing registered, no per-run :class:`PauseBinding` resolver, or no
    way to render a minted envelope as the opaque token a consumer answers
    with. Advertising ``ask_user`` in any of them would invite the model to
    stop a run that could never be resumed -- and a run stopped by a pause
    nobody can answer is strictly worse than one that never stopped. The
    contribution still carries the capability's name, so a plan records
    that human interaction was folded and found nothing rather than that it
    was never folded.

    ``request`` is read for its grants and found to need none. Minting a
    token, recording its payload and writing the issuance row all go
    through this capability's own ports, which the composition root wired;
    none of them is an :class:`~symfonic.kernel.contracts.effects.EffectFamily`
    the plan grants, and declaring one it was not granted is refused at
    fold rather than at the point of the effect (STG-8).
    """
    return build_contribution(
        self,
        request,
        binding=self.binding,
        encode_token=self.encode_token,
    )

pause async

pause(*, pin: Any, scope: Any, run_id: str, session_id: str, thread_id: str, payload: Any, root_run_id: str = '', name: str = ASK_USER, tool_call_id: str = '', checkpoint_id: str | None = None, ttl_seconds: float | None = None, legacy_pinned: bool = False, turn_state: TurnCheckpoint | None = None) -> InteractionEvent

Mint a pause, record its payload, and record the turn it stopped.

turn_state is what makes the pause redeemable (HK2). It is optional because a caller may have nothing continuable to record -- a transport minting a pause outside a kernel turn, for instance -- and the returned event then says resumable=False rather than pretending otherwise.

Source code in src/symfonic/capabilities/human/capability.py

async def pause(
    self,
    *,
    pin: Any,
    scope: Any,
    run_id: str,
    session_id: str,
    thread_id: str,
    payload: Any,
    root_run_id: str = "",
    name: str = ASK_USER,
    tool_call_id: str = "",
    checkpoint_id: str | None = None,
    ttl_seconds: float | None = None,
    legacy_pinned: bool = False,
    turn_state: TurnCheckpoint | None = None,
) -> InteractionEvent:
    """Mint a pause, record its payload, and record the turn it stopped.

    ``turn_state`` is what makes the pause redeemable (HK2). It is optional
    because a caller may have nothing continuable to record -- a transport
    minting a pause outside a kernel turn, for instance -- and the returned
    event then says ``resumable=False`` rather than pretending otherwise.
    """
    registration = self.registry.get(name)
    try:
        registration.payload_schema.model_validate(payload)
    except Exception as exc:
        # The shipped minter logged this and returned ``None``, producing a
        # pause with no token and therefore no way to resume. A refusal
        # stops the run where the bug is.
        raise PayloadBindingError(
            f"the {name!r} payload does not fit its registered schema, so no "
            f"pause token is minted for it: {exc}"
        ) from exc
    # Everything a pause can be refused for is refused before the mint, so
    # a failed pause leaves no issued row the ledger will count as
    # outstanding and nothing can ever consume.
    require_payload_key_inputs(name, tool_call_id=tool_call_id)
    resolved = checkpoint_id or await self.payloads.resolve_checkpoint_id(thread_id)
    if not resolved:
        raise PauseCheckpointNotFoundError(
            f"thread {thread_id!r} has no checkpoint to pause at; a token "
            "bound to no checkpoint could never recover its own request"
        )
    minted = await self.tokens.mint(
        pin=pin,
        scope=scope,
        run_id=run_id,
        root_run_id=root_run_id,
        session_id=session_id,
        thread_id=thread_id,
        checkpoint_id=resolved,
        payload=payload,
        name=name,
        tool_call_id=tool_call_id,
        ttl_seconds=ttl_seconds,
        legacy_pinned=legacy_pinned,
    )
    await self.payloads.record(minted.claims, payload, task_id=run_id)
    # Earned, in this order and no other. The write is attempted, the write
    # reports, and only a report of success makes the pause say it can be
    # resumed. Anything that read ``resumable`` before this line would be
    # reading an intention.
    resumable = False
    if turn_state is not None:
        resumable = await self.turns.record(
            minted.claims, stamped_turn(turn_state, minted.claims)
        )
    return InteractionEvent(
        name=name,
        payload=payload,
        pause=minted,
        run_id=run_id,
        session_id=session_id,
        interrupt_id=minted.claims.interrupt_id,
        tool_call_id=minted.claims.tool_call_id,
        resumable=resumable,
    )

HumanInteractionError

Bases: Exception

Base for everything this capability raises.

InMemoryConsumption

InMemoryConsumption(*, clock: Callable[[], float] = time.time)

The in-process reference implementation (LIB-TL-2's single-lock case).

Correct within one process and honest about the rest: durable is False, and :func:require_durable_consumption is what stops a deployment with durable checkpoints from quietly using it.

Source code in src/symfonic/capabilities/human/consumption.py

def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
    self._consumed: dict[str, ConsumptionRecord] = {}
    self._lock = asyncio.Lock()
    self._clock = clock

clear

clear() -> None

Reset the set — a fixture helper, not a production verb.

Source code in src/symfonic/capabilities/human/consumption.py

def clear(self) -> None:
    """Reset the set — a fixture helper, not a production verb."""
    self._consumed.clear()

InProcessIssuanceRecords

InProcessIssuanceRecords()

Three dicts. Correct for one process, and unwilling to claim more.

Source code in src/symfonic/capabilities/human/issuance.py

def __init__(self) -> None:
    self._issued: dict[str, IssuedToken] = {}
    self._consumed: dict[str, TokenConsumption] = {}
    self._horizon: RetirementHorizon | None = None

InteractionConfigurationError

Bases: HumanInteractionError, ValueError

The deployment wired something that cannot work, before any run starts.

InteractionEvent dataclass

InteractionEvent(name: str, payload: Any, pause: MintedPause, run_id: str, session_id: str, interrupt_id: str = '', tool_call_id: str = '', resumable: bool = False)

The pause a transport serialises: it owns the wire format, this the facts.

InteractionRegistration dataclass

InteractionRegistration(name: str, payload_schema: Any, response_schema: Any, validate_response: Callable[[Any, Any], None] | None = None, cross_scope_allowed: bool = False, built_in: bool = False, metadata: Mapping[str, Any] = dict())

One registered pause point, and everything both ends of it need.

ask_user classmethod

ask_user(*, payload_schema: Any, response_schema: Any, validate_response: Callable[[Any, Any], None] | None = None, cross_scope_allowed: bool = False, metadata: Mapping[str, Any] | None = None) -> InteractionRegistration

The built-in, in the one shape it is allowed to have.

Source code in src/symfonic/capabilities/human/registration.py

@classmethod
def ask_user(
    cls,
    *,
    payload_schema: Any,
    response_schema: Any,
    validate_response: Callable[[Any, Any], None] | None = None,
    cross_scope_allowed: bool = False,
    metadata: Mapping[str, Any] | None = None,
) -> InteractionRegistration:
    """The built-in, in the one shape it is allowed to have."""
    return cls(
        name=ASK_USER,
        payload_schema=payload_schema,
        response_schema=response_schema,
        validate_response=validate_response,
        cross_scope_allowed=cross_scope_allowed,
        built_in=True,
        metadata=metadata or {},
    )

InteractionRegistrationError

Bases: HumanInteractionError, ValueError

A named interaction is not describable, so nothing may register it.

InteractionRegistry

InteractionRegistry()

The one place a name is resolved, on both ends of a pause.

Source code in src/symfonic/capabilities/human/registration.py

def __init__(self) -> None:
    self._registrations: dict[str, InteractionRegistration] = {}

names

names() -> tuple[str, ...]

Sorted, so a refusal message reads the same on every worker.

Source code in src/symfonic/capabilities/human/registration.py

def names(self) -> tuple[str, ...]:
    """Sorted, so a refusal message reads the same on every worker."""
    return tuple(sorted(self._registrations))

InteractionSchema

Bases: Protocol

A payload or response schema: anything that validates and returns.

InteractionToolSpec dataclass

InteractionToolSpec(name: str, description: str, coroutine: Callable[..., Any], parameters: tuple[str, ...] = ())

One interaction, described as a tool the composition root can bind.

A description and not a runtime tool object, for the reason :class:~symfonic.capabilities.delegation.contracts.DelegationToolSpec is one: a capability that constructed a StructuredTool would put a third-party tool library on the import path of a package with no other use for it. symfonic.agent.cutover.delegation.bind_contributed_tool is the step that wraps it, at the composition root, which is where the runtime's tool type is known.

IssuanceLedgerPort

Bases: ConsumptionPort, Protocol

The operated platform's authoritative record, which is also the winner.

A ledger is a consumption port: making issuance and consumption the same object is what gives operated mode one linearization point rather than two places that each believe they decide. The extra verbs — issuance, drain proof, retirement horizon — are the operated-only purposes (LIB-TL-4), and a library deployment reaches them by not having this object at all.

IssuanceRecordBackendPort

Bases: ConditionalWritePort, Protocol

Where an operated platform keeps its issuance table (SEC-PTK-7, SCP-FRZ-2).

Wider than :class:ConditionalWritePort, and for a reason that does not apply there: redemption must never read before it writes, but the issuance record and the retirement horizon exist precisely to be read back — by the next worker, and by this one after a restart. A ledger given only the conditional write keeps them in memory, and says so rather than reporting a reach it does not have.

IssuedToken dataclass

IssuedToken(jti: str, scope_hash: str, name: str, issued_at: float, expires_at: float, legacy_pinned: bool = False, vector_hash: str = '')

One issuance row. Ids, times, and the one bit the drain gate reads.

MintedPause dataclass

MintedPause(envelope: Any, claims: PauseClaims)

A signed envelope and the claims inside it, together.

OperatedPlatformOnlyError

Bases: HumanInteractionError

A drain proof or retirement horizon was asked of a library deployment.

PauseBinding dataclass

PauseBinding(pin: Any = None, scope: Any = None, run_id: str = '', root_run_id: str = '', session_id: str = '', thread_id: str = '', checkpoint_id: str | None = None)

What a pause has to be bound to, resolved per run by the composition root.

Every field is a fact about this run and none of them is known when the plan is compiled, which is why this arrives through a callable rather than on the capability: a bundle folded once at construction would bind every turn to the first turn's tenant, run and thread. That is the same defect TurnRequest.scope exists to correct, one capability over.

checkpoint_id may be None, in which case the capability resolves it through its checkpoint port. A deployment with neither cannot pause, and :meth:HumanInteractionCapability.pause refuses rather than minting a token bound to no checkpoint.

PauseCheckpointNotFoundError

Bases: HumanInteractionError

A durable backend has no such checkpoint or no such pause payload.

PauseClaims dataclass

PauseClaims(run_id: str, session_id: str, scope_hash: str, thread_id: str, checkpoint_id: str, request_hash: str, exp: int, jti: str, root_run_id: str = '', tool_call_id: str = '', name: str = 'ask_user', interrupt_id: str = '', issued_at: float = 0.0, legacy_pinned: bool = False)

What the token binds: who, which run, which thread, which request.

as_legacy_dict

as_legacy_dict() -> dict[str, Any]

Exactly the legacy claim names, for a reader that predates this.

Source code in src/symfonic/capabilities/human/values.py

def as_legacy_dict(self) -> dict[str, Any]:
    """Exactly the legacy claim names, for a reader that predates this."""
    return {key: getattr(self, key) for key in LEGACY_CLAIM_KEYS}

decode classmethod

decode(raw: bytes) -> PauseClaims

Read a verified body. Called only after the envelope verified.

Source code in src/symfonic/capabilities/human/values.py

@classmethod
def decode(cls, raw: bytes) -> PauseClaims:
    """Read a verified body. Called only *after* the envelope verified."""
    try:
        body = json.loads(raw)
    except (ValueError, TypeError) as exc:
        raise PauseTokenUnauthorizedError(
            "pause-token claims are unreadable; a body that cannot be parsed "
            "is refused rather than partially honoured"
        ) from exc
    return cls.from_legacy_dict(body)

encode

encode() -> bytes

The canonical wire body the envelope signs over.

Source code in src/symfonic/capabilities/human/values.py

def encode(self) -> bytes:
    """The canonical wire body the envelope signs over."""
    body = self.as_legacy_dict()
    body.update({key: getattr(self, key) for key in sorted(_ADDED_CLAIM_KEYS)})
    return json.dumps(body, separators=(",", ":"), sort_keys=True).encode("utf-8")

expired

expired(now: float) -> bool

exp is inclusive: a token is live through its expiry second.

Source code in src/symfonic/capabilities/human/values.py

def expired(self, now: float) -> bool:
    """``exp`` is inclusive: a token is live *through* its expiry second."""
    return now > self.exp

from_legacy_dict classmethod

from_legacy_dict(body: Any) -> PauseClaims

Build from a claim mapping, refusing anything it cannot account for.

Source code in src/symfonic/capabilities/human/values.py

@classmethod
def from_legacy_dict(cls, body: Any) -> PauseClaims:
    """Build from a claim mapping, refusing anything it cannot account for."""
    if not isinstance(body, dict):
        raise PauseTokenUnauthorizedError(
            f"pause-token claims must be a mapping, got {type(body).__name__}"
        )
    known = set(LEGACY_CLAIM_KEYS) | _ADDED_CLAIM_KEYS
    unknown = sorted(set(body) - known)
    if unknown:
        # A verified envelope means the producer is trusted, not that this
        # build understands everything it wrote. Ignoring an unknown claim
        # would silently drop a binding a newer producer added on purpose.
        raise PauseTokenUnauthorizedError(
            f"pause-token claims carry {unknown}, which this build does not "
            "understand; it refuses rather than dropping a binding"
        )
    missing = sorted(_REQUIRED_CLAIM_KEYS - set(body))
    if missing:
        raise PauseTokenUnauthorizedError(
            f"pause-token claims are missing {missing}; refusing to guess"
        )
    return cls(**{key: body[key] for key in body})

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

PauseTokenError

Bases: HumanInteractionError

Base for every refusal that names the token itself.

PauseTokenExpiredError

Bases: PauseTokenError

The token was real and its window closed. Ask the question again.

PauseTokenReplayedError

Bases: PauseTokenError

The token was real and somebody already redeemed it.

PauseTokenService

PauseTokenService(*, signer: Any, ttl: TTLPolicy, consumption: Any = None, ledger: Any = None, clock: Callable[[], float] = time.time, pinless_policy: Any = None, binder: ScopeBinder | None = None)

Mint, validate, and consume pause tokens against exactly one winner-seam.

Source code in src/symfonic/capabilities/human/tokens.py

def __init__(
    self,
    *,
    signer: Any,
    ttl: TTLPolicy,
    consumption: Any = None,
    ledger: Any = None,
    clock: Callable[[], float] = time.time,
    pinless_policy: Any = None,
    binder: ScopeBinder | None = None,
) -> None:
    for verb in ("mint", "verify"):
        if not callable(getattr(signer, verb, None)):
            raise InteractionConfigurationError(
                f"the envelope signer must offer {verb}(...); got "
                f"{type(signer).__name__}. Authenticity is consumed, not remade"
            )
    if (consumption is None) == (ledger is None):
        raise InteractionConfigurationError(
            "exactly one consumption seam is required: an issuance ledger "
            "(operated platform) or a conditional-write store (library mode). "
            "Two seams are two places that each believe they decide the "
            "winner; none is no single-use enforcement at all"
        )
    self._signer = signer
    self._ttl = ttl
    self._ledger = ledger
    self._consumption = ledger if ledger is not None else consumption
    self._clock = clock
    self._pinless = pinless_policy
    self._binder = binder or ScopeBinder()

operated property

operated: bool

Whether an authoritative issuance ledger backs this service.

authenticate

authenticate(envelope: Any, *, at: float | None = None) -> ValidatedPause

Steps 1–3: verify, decode, expire. No state is touched.

Verification precedes decoding so a forged envelope cannot choose how it is read; expiry is checked after decoding because the expiry claim is part of what the signature covers, so trusting it earlier would let a forgery declare itself fresh.

Source code in src/symfonic/capabilities/human/tokens.py

def authenticate(self, envelope: Any, *, at: float | None = None) -> ValidatedPause:
    """Steps 1–3: verify, decode, expire. No state is touched.

    Verification precedes decoding so a forged envelope cannot choose how it
    is read; expiry is checked after decoding because the expiry claim is
    part of what the signature covers, so trusting it earlier would let a
    forgery declare itself fresh.
    """
    try:
        pin = self._signer.verify(
            envelope, pinless_policy=self._pinless, schema_id=self.SCHEMA_ID
        )
    except Exception as exc:  # every refusal denies; ``refusals.py`` says why
        raise PauseTokenUnauthorizedError.from_signer_refusal(exc) from exc
    # The signer only consults schema_id on its pin-less branch, so the
    # family is this reader's: same keyset, another schema, still not a pause.
    presented = getattr(envelope, "schema_id", "")
    if presented != self.SCHEMA_ID:
        raise PauseTokenUnauthorizedError(
            f"this envelope travels under schema {presented!r}, not "
            f"{self.SCHEMA_ID!r}; a token from another payload family is not "
            "redeemable here however well it verifies"
        )
    claims = PauseClaims.decode(envelope.payload)
    now = self._clock() if at is None else at
    if claims.expired(now):
        raise PauseTokenExpiredError(
            "this pause token has expired; the question was not answered "
            "inside its lifetime, so ask it again rather than resuming"
        )
    return ValidatedPause(claims=claims, pin=pin)

bind

bind(claims: PauseClaims, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, registration: Any = None) -> bool

Step 4 — the four axes (HK2). Returns whether scopes were crossed.

None is "not stated" for the last three; the binder says why each is separately refusable. The registration is the only thing that can open cross-scope redemption, and it is passed in -- rather than looked up, or reduced to a boolean a caller could pass -- so a caller cannot bind against a different posture than the one it validates the answer with.

Source code in src/symfonic/capabilities/human/tokens.py

def bind(
    self,
    claims: PauseClaims,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    registration: Any = None,
) -> bool:
    """Step 4 — the four axes (HK2). Returns whether scopes were crossed.

    ``None`` is "not stated" for the last three; the binder says why each is
    separately refusable. The registration is the *only* thing that can open
    cross-scope redemption, and it is passed in -- rather than looked up, or
    reduced to a boolean a caller could pass -- so a caller cannot bind
    against a different posture than the one it validates the answer with.
    """
    return self._binder.check(
        claims,
        scope=scope,
        session_id=session_id,
        run_id=run_id,
        call_id=call_id,
        cross_scope_allowed=bool(
            getattr(registration, "cross_scope_allowed", False)
        ),
    )

consume async

consume(claims: PauseClaims) -> None

Step 5 — the single atomic claim. Losing it is a replay, not a fault.

Source code in src/symfonic/capabilities/human/tokens.py

async def consume(self, claims: PauseClaims) -> None:
    """Step 5 — the single atomic claim. Losing it is a replay, not a fault."""
    won = await self._consumption.claim(
        claims.jti, scope_hash=claims.scope_hash, name=claims.name
    )
    if not won:
        raise PauseTokenReplayedError(
            "this pause token has already been redeemed; single use is enforced "
            "by one atomic operation, so exactly one caller wins"
        )

drain_proof async

drain_proof(now: float | None = None) -> Any

CUT-AIR-3 — proof that legacy-pinned tokens have drained.

Source code in src/symfonic/capabilities/human/tokens.py

async def drain_proof(self, now: float | None = None) -> Any:
    """CUT-AIR-3 — proof that legacy-pinned tokens have drained."""
    return await self._require_ledger("a drain proof").drain_proof(now)

mint async

mint(*, pin: Any, scope: Any, run_id: str, session_id: str, thread_id: str, checkpoint_id: str, payload: Any, root_run_id: str = '', name: str = ASK_USER, tool_call_id: str = '', interrupt_id: str | None = None, ttl_seconds: float | None = None, legacy_pinned: bool = False) -> MintedPause

Bind a pause to this scope, session, and request, and sign it.

Source code in src/symfonic/capabilities/human/tokens.py

async def mint(
    self,
    *,
    pin: Any,
    scope: Any,
    run_id: str,
    session_id: str,
    thread_id: str,
    checkpoint_id: str,
    payload: Any,
    root_run_id: str = "",
    name: str = ASK_USER,
    tool_call_id: str = "",
    interrupt_id: str | None = None,
    ttl_seconds: float | None = None,
    legacy_pinned: bool = False,
) -> MintedPause:
    """Bind a pause to this scope, session, and request, and sign it."""
    now = self._clock()
    lifetime = self._ttl.resolve(ttl_seconds)
    if interrupt_id is None:
        # ``i-<12 hex>`` -- the legacy correlation-id shape consumers match on.
        interrupt_id = "" if name == ASK_USER else f"i-{uuid.uuid4().hex[:12]}"
    claims = PauseClaims(
        run_id=run_id,
        root_run_id=root_run_id or run_id,
        session_id=session_id,
        scope_hash=self._binder.scope_hash(scope),
        thread_id=thread_id,
        checkpoint_id=checkpoint_id,
        request_hash=self._binder.request_hash(payload),
        exp=int(now + lifetime),
        jti=uuid.uuid4().hex,
        tool_call_id=tool_call_id,
        name=name,
        interrupt_id=interrupt_id,
        issued_at=now,
        legacy_pinned=legacy_pinned,
    )
    envelope = self._signer.mint(
        pin=pin, payload=claims.encode(), schema_id=self.SCHEMA_ID
    )
    if self._ledger is not None:
        await self._ledger.record_issuance(
            IssuedToken(
                jti=claims.jti,
                scope_hash=claims.scope_hash,
                name=claims.name,
                issued_at=now,
                expires_at=float(claims.exp),
                legacy_pinned=legacy_pinned,
                vector_hash=getattr(pin, "vector_hash", ""),
            )
        )
    return MintedPause(envelope=envelope, claims=claims)

record_retirement_horizon async

record_retirement_horizon(at: float, *, reason: str) -> None

SCP-FRZ-2 — the date past which nothing may be extended.

Source code in src/symfonic/capabilities/human/tokens.py

async def record_retirement_horizon(self, at: float, *, reason: str) -> None:
    """SCP-FRZ-2 — the date past which nothing may be extended."""
    await self._require_ledger("a retirement horizon").record_retirement_horizon(
        at, reason=reason
    )

scope_hash

scope_hash(scope: Any) -> str

The binding hash, so a caller can record what was presented.

Source code in src/symfonic/capabilities/human/tokens.py

def scope_hash(self, scope: Any) -> str:
    """The binding hash, so a caller can record what was presented."""
    return self._binder.scope_hash(scope)

validate async

validate(envelope: Any, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, at: float | None = None, registration: Any = None) -> ValidatedPause

Authenticate, then bind. Still consumes nothing.

Source code in src/symfonic/capabilities/human/tokens.py

async def validate(
    self,
    envelope: Any,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    at: float | None = None,
    registration: Any = None,
) -> ValidatedPause:
    """Authenticate, then bind. Still consumes nothing."""
    validated = self.authenticate(envelope, at=at)
    crossed = self.bind(
        validated.claims,
        scope=scope,
        session_id=session_id,
        run_id=run_id,
        call_id=call_id,
        registration=registration,
    )
    return ValidatedPause(validated.claims, validated.pin, crossed)

validate_and_consume async

validate_and_consume(envelope: Any, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, at: float | None = None, registration: Any = None) -> ValidatedPause

The whole ordering, for a caller with nothing to check in between.

Source code in src/symfonic/capabilities/human/tokens.py

async def validate_and_consume(
    self,
    envelope: Any,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    at: float | None = None,
    registration: Any = None,
) -> ValidatedPause:
    """The whole ordering, for a caller with nothing to check in between."""
    validated = await self.validate(
        envelope,
        scope=scope,
        session_id=session_id,
        run_id=run_id,
        call_id=call_id,
        at=at,
        registration=registration,
    )
    await self.consume(validated.claims)
    return validated

PauseTokenUnauthorizedError

PauseTokenUnauthorizedError(*args: object, code: str | None = None)

Bases: PauseTokenError

The token does not authenticate: forged, tampered, or unreadable.

One refusal that reaches here is not about the token at all: when the verification keyset is unavailable the signer must deny, and denying is all it can do — but telling the holder "your token is bad" while the secret manager is down is a lie, and one their retry logic acts on. So the reason may override the code (EMAP-6): the transport reads code and answers 503 + Retry-After instead of 401. The class is unchanged because the security decision is unchanged; only the explanation is now honest.

Source code in src/symfonic/capabilities/human/errors.py

def __init__(self, *args: object, code: str | None = None) -> None:
    super().__init__(*args)
    if code is not None:
        self.code = code

from_signer_refusal classmethod

from_signer_refusal(cause: BaseException) -> PauseTokenUnauthorizedError

The denial a failed verify becomes, carrying the reason's code.

Lives beside the class rather than at the call site so there is one place where "which refusals are about the token?" is answered, and so the token service does not have to know that a signer's dependency has a taxonomy at all.

Source code in src/symfonic/capabilities/human/errors.py

@classmethod
def from_signer_refusal(cls, cause: BaseException) -> PauseTokenUnauthorizedError:
    """The denial a failed ``verify`` becomes, carrying the reason's code.

    Lives beside the class rather than at the call site so there is one
    place where "which refusals are about the token?" is answered, and so
    the token service does not have to know that a signer's dependency has
    a taxonomy at all.
    """
    return cls(f"pause token is not accepted: {cause}", code=availability_code(cause))

PayloadBindingError

Bases: PauseTokenError

The paused request is not the one the token was minted for (SEC-PTK-5).

ResponseValidationError

Bases: HumanInteractionError, ValueError

The answer does not fit the registered response schema.

ResumeCommand dataclass

ResumeCommand(envelope: Any, response: Any, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None)

One redemption attempt, in the shape a transport can build from a request.

session_id, run_id and call_id are the three axes beside scope that a redemption is checked against (HK2). Each defaults to None, meaning "this transport did not state it", and an unstated axis is not checked -- which is why :mod:~symfonic.agent.cutover.kernel_resume builds this value with all four and refuses to build one without them. The default belongs to a caller that genuinely has no such fact; it must not be how a caller that should have had one silently opts out of the check.

ResumeOutcome dataclass

ResumeOutcome(name: str, thread_id: str, checkpoint_id: str, payload: Any, response: Any, run_id: str, session_id: str, tool_call_id: str = '', interrupt_id: str = '', cross_scope: bool = False, time_to_resolve_seconds: float = 0.0, turn: Any = None)

What a graph runner needs to continue, and what telemetry needs to record.

as_configurable

as_configurable() -> dict[str, Any]

The run-config shape the resume executes against.

Source code in src/symfonic/capabilities/human/values.py

def as_configurable(self) -> dict[str, Any]:
    """The run-config shape the resume executes against."""
    return {
        "configurable": {
            "thread_id": self.thread_id,
            "checkpoint_id": self.checkpoint_id,
        }
    }

ResumeService

ResumeService(*, tokens: PauseTokenService, registry: InteractionRegistry, payloads: PausePayloadStore, clock: Callable[[], float] = time.time, audit: Callable[[CrossScopeRedemption], None] | None = None, turns: TurnCheckpointStore | None = None)

Turns one redemption attempt into either an outcome or a named refusal.

Source code in src/symfonic/capabilities/human/resume.py

def __init__(
    self,
    *,
    tokens: PauseTokenService,
    registry: InteractionRegistry,
    payloads: PausePayloadStore,
    clock: Callable[[], float] = time.time,
    audit: Callable[[CrossScopeRedemption], None] | None = None,
    turns: TurnCheckpointStore | None = None,
) -> None:
    self._tokens = tokens
    self._registry = registry
    self._payloads = payloads
    self._clock = clock
    self._audit = audit
    self._turns = turns

resume async

resume(command: ResumeCommand, *, require_turn: bool = False) -> ResumeOutcome

Redeem one token. require_turn is the caller's demand, not the request's.

A route that will continue the run needs the recorded turn state and must be refused, by name, when there is none. A route that only needs the answer validated -- the legacy body, which continues through its own saver -- must not be, or every token minted before this build stops resolving the moment the deployment upgrades. Same six checks, same order, one decision about what counts as missing.

Source code in src/symfonic/capabilities/human/resume.py

async def resume(
    self, command: ResumeCommand, *, require_turn: bool = False
) -> ResumeOutcome:
    """Redeem one token. ``require_turn`` is the caller's demand, not the
    request's.

    A route that will *continue the run* needs the recorded turn state and
    must be refused, by name, when there is none. A route that only needs
    the answer validated -- the legacy body, which continues through its own
    saver -- must not be, or every token minted before this build stops
    resolving the moment the deployment upgrades. Same six checks, same
    order, one decision about what counts as missing.
    """
    now = self._clock()
    # 1-3. Authenticity, then the claims, then expiry. Nothing stateful yet.
    authenticated = self._tokens.authenticate(command.envelope, at=now)
    claims = authenticated.claims

    # 4. The registration decides the scope posture, so it is resolved
    #    before the binding check rather than after it. Four axes since
    #    HK2 -- scope, session, run and call -- each separately refusable.
    registration = self._registry.get(claims.name)
    cross_scope = self._tokens.bind(
        claims,
        scope=command.scope,
        session_id=command.session_id,
        run_id=command.run_id,
        call_id=command.call_id,
        registration=registration,
    )
    if cross_scope:
        self._record_cross_scope(claims, command, at=now)

    # 5. The paused request, checked against the hash the token carries.
    payload = await self._payloads.load(claims)
    payload = self._validate_payload(registration, payload)

    # 6. The answer, against its own schema and then against the question.
    response = self._validate_response(registration, command.response, payload)

    # 7. The paused turn's state, rebuilt from the checkpointer (HK2).
    #    Before the claim, with everything else that can fail: a token burnt
    #    on a checkpoint read that timed out is a question the person has
    #    already answered and can no longer answer again.
    turn = await self._load_turn(claims, require_turn)

    # 8. The single-use claim. Last, and exactly once.
    await self._tokens.consume(claims)

    return ResumeOutcome(
        name=claims.name,
        thread_id=claims.thread_id,
        checkpoint_id=claims.checkpoint_id,
        payload=payload,
        response=response,
        run_id=claims.run_id,
        session_id=claims.session_id,
        tool_call_id=claims.tool_call_id,
        interrupt_id=claims.interrupt_id,
        cross_scope=cross_scope,
        time_to_resolve_seconds=max(0.0, now - claims.issued_at)
        if claims.issued_at
        else 0.0,
        turn=turn,
    )

RetirementHorizon dataclass

RetirementHorizon(at: float, reason: str)

SCP-FRZ-2: the date, and the operator's reason for it.

RetirementHorizonError

Bases: HumanInteractionError

A recorded retirement horizon forbids this, and re-recording it too.

RunBindingError

Bases: PauseTokenError

The token was minted for another run of the same session (HK2).

Its own class rather than a SessionBindingError, because the two say different things to an operator. A session mismatch is a caller answering from the wrong conversation; a run mismatch is a caller answering the right conversation's previous question -- a stale browser tab, a retried request, a queue that replayed. Both refuse; only one of them suggests somebody is looking at an old page.

ScopeBinder

Checks the bindings a pause token carries. Holds no state.

call_id_of staticmethod

call_id_of(claims: PauseClaims) -> str

Which id this pause correlates on -- the split, in one expression.

ask_user joins on the reserved tool_call_id and a registered interrupt on its own interrupt_id; a claim carries whichever its family minted. Derived from the claims rather than from which subsystem is asking, exactly as :func:~symfonic.capabilities.human.checkpoints.payload_key derives the metadata key.

Source code in src/symfonic/capabilities/human/binding.py

@staticmethod
def call_id_of(claims: PauseClaims) -> str:
    """Which id this pause correlates on -- the split, in one expression.

    ``ask_user`` joins on the reserved ``tool_call_id`` and a registered
    interrupt on its own ``interrupt_id``; a claim carries whichever its
    family minted. Derived from the claims rather than from which subsystem
    is asking, exactly as
    :func:`~symfonic.capabilities.human.checkpoints.payload_key` derives the
    metadata key.
    """
    return claims.tool_call_id or claims.interrupt_id

check

check(claims: PauseClaims, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, cross_scope_allowed: bool = False) -> bool

Return whether this redemption crossed scopes. Raise when it may not.

Four axes, checked one at a time (HK2). Scope says which tenant, session says which conversation, run says which turn of it, and the call id says which of that turn's questions. They are four independent facts and each is separately refusable, because a redemption that satisfies three of them is a real, reachable mistake rather than a hypothetical: an operator with two paused runs open answers the wrong tab (run), an onboarding agent that asked twice gets the second answer filed against the first question (call), and a shared-inbox admin answers a colleague's session (session). Only the tenancy axis has ever had a bypass.

None means "not stated" for the last three, and a caller that does not state them gets no check for them. That is why the resume route states all four rather than trusting this default: the default is what a transport left out, and a transport that forgets an axis must not be the thing that decides the axis does not matter.

The session, run and call checks run regardless of cross_scope_allowed: crossing scopes is about who redeems (a tenant admin approving a sub-tenant's action), never about which run gets resumed. Relaxing them all at once would let one opt-in reach every paused run in the deployment.

Source code in src/symfonic/capabilities/human/binding.py

def check(
    self,
    claims: PauseClaims,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    cross_scope_allowed: bool = False,
) -> bool:
    """Return whether this redemption crossed scopes. Raise when it may not.

    **Four axes, checked one at a time (HK2).** Scope says which tenant,
    session says which conversation, run says which turn of it, and the call
    id says which of that turn's questions. They are four independent facts
    and each is separately refusable, because a redemption that satisfies
    three of them is a real, reachable mistake rather than a hypothetical:
    an operator with two paused runs open answers the wrong tab (run), an
    onboarding agent that asked twice gets the second answer filed against
    the first question (call), and a shared-inbox admin answers a colleague's
    session (session). Only the tenancy axis has ever had a bypass.

    ``None`` means "not stated" for the last three, and a caller that does
    not state them gets no check for them. That is why the resume route
    states all four rather than trusting this default: the default is what a
    *transport* left out, and a transport that forgets an axis must not be
    the thing that decides the axis does not matter.

    The session, run and call checks run regardless of
    ``cross_scope_allowed``: crossing scopes is about *who* redeems (a
    tenant admin approving a sub-tenant's action), never about which run
    gets resumed. Relaxing them all at once would let one opt-in reach every
    paused run in the deployment.
    """
    presented = self.scope_hash(scope)
    cross_scope = presented != claims.scope_hash
    if cross_scope and not cross_scope_allowed:
        raise ScopeBindingError(
            "this pause token was minted under a different tenant scope; "
            "redeeming it here would cross a tenancy boundary the "
            "registration did not open"
        )
    if session_id is not None and session_id != claims.session_id:
        raise SessionBindingError(
            f"this pause token belongs to session {claims.session_id!r}, "
            f"not {session_id!r}"
        )
    if run_id is not None and run_id != claims.run_id:
        raise RunBindingError(
            f"this pause token belongs to run {claims.run_id!r}, not "
            f"{run_id!r}; it answers a question a different turn asked"
        )
    if call_id is not None and call_id != self.call_id_of(claims):
        raise CallBindingError(
            f"this pause token belongs to call {self.call_id_of(claims)!r}, "
            f"not {call_id!r}; a run that paused more than once has one live "
            "token per question and they are not interchangeable"
        )
    return cross_scope

check_payload

check_payload(claims: PauseClaims, *, payload: Any) -> None

SEC-PTK-5 — the paused request is the one the token was minted for.

Source code in src/symfonic/capabilities/human/binding.py

def check_payload(self, claims: PauseClaims, *, payload: Any) -> None:
    """SEC-PTK-5 — the paused request is the one the token was minted for."""
    if self.request_hash(payload) != claims.request_hash:
        raise PayloadBindingError(
            "the recovered request does not hash to the token's "
            "request_hash; the paused request was altered between mint and "
            "resume, and this refuses rather than resuming the new one"
        )

payload_body_matches

payload_body_matches(claims: PauseClaims, *, body: str) -> bool

Whether a recorded body hashes to the token's request_hash.

Source code in src/symfonic/capabilities/human/binding.py

def payload_body_matches(self, claims: PauseClaims, *, body: str) -> bool:
    """Whether a recorded body hashes to the token's ``request_hash``."""
    return hash_payload_body(body) == claims.request_hash

request_body

request_body(payload: Any) -> str

What to record, so that what is recorded is what was hashed.

Source code in src/symfonic/capabilities/human/binding.py

def request_body(self, payload: Any) -> str:
    """What to record, so that what is recorded is what was hashed."""
    return payload_body(payload)

ScopeBindingError

Bases: PauseTokenError

The redeeming scope is not the minting scope (SEC-PTK-4).

SessionBindingError

Bases: PauseTokenError

The token belongs to another session of the same tenant.

TTLPolicy dataclass

TTLPolicy(default_seconds: float, maximum_seconds: float)

The lifetime a pause may have, and the bound nothing may exceed.

The maximum is the same number the operated ledger uses to bound its drain proof, which is why refusing is the only correct answer to a request past it: a clamped token would tell the caller they have a window they do not, and would make the drain deadline a guess.

TokenConsumption dataclass

TokenConsumption(jti: str, scope_hash: str, name: str, consumed_at: float)

One redemption row: who won, and when. Losers are not recorded here.

TokenIssuanceLedger

TokenIssuanceLedger(*, maximum_ttl_seconds: float, clock: Callable[[], float] = time.time, store: Any = None, records: Any = None)

Issuance, consumption, the maximum-TTL bound, and the retirement horizon.

Source code in src/symfonic/capabilities/human/ledger.py

def __init__(
    self,
    *,
    maximum_ttl_seconds: float,
    clock: Callable[[], float] = time.time,
    store: Any = None,
    records: Any = None,
) -> None:
    if maximum_ttl_seconds <= 0:
        raise TokenTTLError(
            "the maximum pause-token lifetime must be positive; it is the "
            "bound the drain deadline is computed from"
        )
    self._max_ttl = float(maximum_ttl_seconds)
    self._clock = clock
    self._lock = asyncio.Lock()
    # The library-mode conditional write, reused: a second implementation
    # here would be a second place that believes it decides the winner.
    self._writes = (
        None if store is None else ConditionalWriteConsumption(store, clock=clock)
    )
    self._records = _records_for(records)

deployment_wide property

deployment_wide: bool

Whether both tables are shared, rather than this worker's memory.

durable property

durable: bool

As durable as the weaker of the two tables a redemption reads.

drain_deadline async

drain_deadline() -> float | None

The horizon plus the maximum TTL: the last moment anything can live.

Source code in src/symfonic/capabilities/human/ledger.py

async def drain_deadline(self) -> float | None:
    """The horizon plus the maximum TTL: the last moment anything can live."""
    recorded = await self._records.horizon()
    return None if recorded is None else recorded.at + self._max_ttl

outstanding async

outstanding(now: float | None = None) -> tuple[str, ...]

Issued, unconsumed, and not yet expired.

Source code in src/symfonic/capabilities/human/ledger.py

async def outstanding(self, now: float | None = None) -> tuple[str, ...]:
    """Issued, unconsumed, and not yet expired."""
    moment = self._clock() if now is None else now
    return tuple(sorted(row.jti for row in await self._live(moment)))

record_retirement_horizon async

record_retirement_horizon(at: float, *, reason: str) -> None

SCP-FRZ-2 — the date after which nothing legacy-pinned may outlive.

Source code in src/symfonic/capabilities/human/ledger.py

async def record_retirement_horizon(self, at: float, *, reason: str) -> None:
    """SCP-FRZ-2 — the date after which nothing legacy-pinned may outlive."""
    if at <= self._clock():
        raise RetirementHorizonError(
            "a retirement horizon must be in the future; a backdated one "
            "would retire tokens that nothing had stopped supporting"
        )
    async with self._lock:
        standing, recorded = await self._records.claim_horizon(float(at), reason)
    if not recorded:
        raise RetirementHorizonError(
            f"a retirement horizon is already recorded at {standing.at} "
            f"({standing.reason!r}); the operator who declared the "
            "retirement is the one whose date and reason stand"
        )

TokenLedgerError

Bases: HumanInteractionError

The issuance ledger refuses a record that contradicts what it holds.

TokenTTLError

Bases: HumanInteractionError, ValueError

A lifetime outside the configured bound. Never clamped, always refused.

TurnCheckpointStore

TurnCheckpointStore(*, commands: Any = None)

Records a paused turn's continuable state, and reads it back.

The whole of the checkpointer role, in one object with two verbs. It holds no state of its own: everything it knows it asks the port for, which is what makes a different process asking the same port get the same answer.

Source code in src/symfonic/capabilities/human/turnstate.py

def __init__(self, *, commands: Any = None) -> None:
    self._commands = commands

wired property

wired: bool

Whether this deployment has a checkpointer at all.

find async

find(claims: PauseClaims) -> TurnCheckpoint | None

The recorded state, or None when there simply is not one.

The half of :meth:load a route-agnostic resume needs. The legacy route continues a paused run through LangGraph's own saver and never wanted this record, so a token minted for it has none -- and answering that with a refusal would break the one thing HK2 must not break, which is that a pause in flight when a deployment upgrades still resolves.

Only the three absences are folded into None. A recorded state that is unreadable, or that describes another turn, still raises: those are not "there is nothing here", they are "there is something here and it is wrong", and continuing past either is how a resume answers into a transcript nobody checked.

Source code in src/symfonic/capabilities/human/turnstate.py

async def find(self, claims: PauseClaims) -> TurnCheckpoint | None:
    """The recorded state, or ``None`` when there simply is not one.

    The half of :meth:`load` a *route-agnostic* resume needs. The legacy
    route continues a paused run through LangGraph's own saver and never
    wanted this record, so a token minted for it has none -- and answering
    that with a refusal would break the one thing HK2 must not break, which
    is that a pause in flight when a deployment upgrades still resolves.

    Only the three *absences* are folded into ``None``. A recorded state
    that is unreadable, or that describes another turn, still raises: those
    are not "there is nothing here", they are "there is something here and
    it is wrong", and continuing past either is how a resume answers into a
    transcript nobody checked.
    """
    try:
        return await self.load(claims)
    except (
        CheckpointLostError,
        InteractionConfigurationError,
        PauseCheckpointNotFoundError,
    ):
        return None

load async

load(claims: PauseClaims) -> TurnCheckpoint

Rebuild the paused turn's state from the port. Never from memory.

Every fact returned came out of the checkpoint store on this call, which is what makes a process that did not pause the run able to continue it. Nothing is cached here and nothing is remembered between calls; a store that lost the row answers the same way for the process that wrote it as for any other.

Source code in src/symfonic/capabilities/human/turnstate.py

async def load(self, claims: PauseClaims) -> TurnCheckpoint:
    """Rebuild the paused turn's state from the port. Never from memory.

    Every fact returned came out of the checkpoint store on this call, which
    is what makes a process that did not pause the run able to continue it.
    Nothing is cached here and nothing is remembered between calls; a store
    that lost the row answers the same way for the process that wrote it as
    for any other.
    """
    if self._commands is None:
        raise InteractionConfigurationError(
            "no checkpoint commands are wired, so a paused turn's state was "
            "never recorded and cannot be rebuilt; this deployment can ask a "
            "question but not continue the run that asked it"
        )
    raw = await self._commands.read_metadata(
        thread_id=claims.thread_id,
        checkpoint_id=claims.checkpoint_id,
        key=turn_state_key(claims),
    )
    if raw is None:
        self._refuse_missing(claims)
    try:
        checkpoint = TurnCheckpoint.decode(raw)
    except ValueError as exc:
        raise PayloadBindingError(
            f"the recorded turn state for {claims.jti!r} cannot be read "
            f"({exc}); it is refused rather than continued from a partial "
            "or half-understood transcript"
        ) from exc
    self._require_same_turn(claims, checkpoint)
    return checkpoint

record async

record(claims: PauseClaims, checkpoint: TurnCheckpoint) -> bool

Write the turn state. Reports whether it landed; never raises for it.

False is a real answer with a caller that acts on it: the pause is still minted, still published and still shows the person the question -- it simply declares resumable=False, because nothing on the other end could rebuild the turn. Raising instead would end a turn that a human can still usefully be asked, and returning True regardless is the lie HK1 shipped resumable=False to prevent, with the sign flipped.

Source code in src/symfonic/capabilities/human/turnstate.py

async def record(self, claims: PauseClaims, checkpoint: TurnCheckpoint) -> bool:
    """Write the turn state. Reports whether it landed; never raises for it.

    ``False`` is a real answer with a caller that acts on it: the pause is
    still minted, still published and still shows the person the question --
    it simply declares ``resumable=False``, because nothing on the other end
    could rebuild the turn. Raising instead would end a turn that a human
    can still usefully be asked, and returning ``True`` regardless is the
    lie HK1 shipped ``resumable=False`` to prevent, with the sign flipped.
    """
    if self._commands is None:
        return False
    try:
        await self._commands.write_metadata(
            thread_id=claims.thread_id,
            checkpoint_id=claims.checkpoint_id,
            key=turn_state_key(claims),
            value=checkpoint.encode(),
            task_id=claims.run_id,
        )
    except Exception:  # noqa: BLE001 - reported, and the caller acts on it
        return False
    return True

TurnStateNotRecordedError

Bases: HumanInteractionError

A recorded turn state does not describe the turn the token binds (HK2).

UnissuedTokenError

Bases: PauseTokenError

The authoritative ledger never issued this token, so it does not exist.

A separate class from :class:PauseTokenUnauthorizedError even though both deny: this one means the envelope was fine and the ledger still says no, which is a deployment/routing question, not a forgery.

UnknownInteractionError

Bases: HumanInteractionError, KeyError

Nothing is registered under that name.

Also a :class:KeyError because the shipped registry was a dict and an adopter's except KeyError around a lookup is a reasonable thing to have.

ValidatedPause dataclass

ValidatedPause(claims: PauseClaims, pin: Any, cross_scope: bool = False)

What survived authentication, expiry, and binding.

ask_user_registration

ask_user_registration(*, cross_scope_allowed: bool = False) -> InteractionRegistration

The ask_user interaction, with schemas that actually validate.

cross_scope_allowed stays false: a token minted for one scope being redeemable in another is the cross-scope redemption the ledger exists to record, and it is not something a default should hand out.

Source code in src/symfonic/capabilities/human/factory.py

def ask_user_registration(
    *, cross_scope_allowed: bool = False
) -> InteractionRegistration:
    """The ``ask_user`` interaction, with schemas that actually validate.

    ``cross_scope_allowed`` stays false: a token minted for one scope being
    redeemable in another is the cross-scope redemption the ledger exists to
    record, and it is not something a default should hand out.
    """
    return InteractionRegistration.ask_user(
        payload_schema=AskUserPayload,
        response_schema=AskUserResponse,
        cross_scope_allowed=cross_scope_allowed,
    )

hash_payload

hash_payload(payload: Any) -> str

The legacy PauseToken.hash_payload formula, unchanged.

Source code in src/symfonic/capabilities/human/binding.py

def hash_payload(payload: Any) -> str:
    """The legacy ``PauseToken.hash_payload`` formula, unchanged."""
    return hash_payload_body(payload_body(payload))

hash_scope

hash_scope(scope: Any) -> str

The legacy PauseToken.hash_scope formula, unchanged.

Read structurally — the tenancy scope type lives outside this layer and must not be imported into it. An object with no tenant_id is refused rather than hashed as the empty tenant, which would make every unauthenticated caller share one scope.

Source code in src/symfonic/capabilities/human/binding.py

def hash_scope(scope: Any) -> str:
    """The legacy ``PauseToken.hash_scope`` formula, unchanged.

    Read structurally — the tenancy scope type lives outside this layer and must
    not be imported into it. An object with no ``tenant_id`` is refused rather
    than hashed as the empty tenant, which would make every unauthenticated
    caller share one scope.
    """
    tenant_id = getattr(scope, "tenant_id", None)
    if not tenant_id:
        raise ScopeBindingError(
            "cannot bind a pause token to a scope with no tenant_id; an "
            "unattributed token is one every caller could redeem"
        )
    sub_tenant_id = getattr(scope, "sub_tenant_id", None) or _ABSENT
    namespace = getattr(scope, "namespace", None) or _ABSENT
    raw = f"{tenant_id}:{sub_tenant_id}:{namespace}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

human_interaction

human_interaction(*, signer: Any, binding: Callable[[], Any], encode_token: Callable[[Any], str], decode_token: Callable[[str], Any] | None = None, registrations: Iterable[InteractionRegistration] | None = None, ttl: TTLPolicy | None = None, consumption: Any = None, **options: Any) -> HumanInteractionCapability

Build the capability that lets a turn pause and be resumed.

Agent(provider, capabilities=[human_interaction(
    signer=my_signer,
    binding=lambda: PauseBinding(run_id=..., session_id=...),
    encode_token=my_encoder,
)])

Three arguments are required because only the deployment can supply them, and because the capability contributes nothing without them.

Parameters:

Name	Type	Description	Default
signer	Any	mints and verifies pause tokens. No default: a shared one would make every deployment honour every other's tokens.	required
binding	Callable[[], Any]	resolves what this run is, per run. No default: a PauseBinding() with empty fields is a token bound to nothing.	required
encode_token	Callable[[Any], str]	renders a minted envelope as the opaque string a person answers with. No default: the token format is a deployment's choice, not the framework's.	required
decode_token	Callable[[str], Any] | None	the inverse. Optional, and its absence has a cost -- without it a pause taken on the kernel route has no public reader, so the run can be stopped and not continued.	None
registrations	Iterable[InteractionRegistration] | None	which interactions this agent offers. Defaults to ask_user alone -- the one every deployment wants and the only one whose schema is obvious.	None
ttl	TTLPolicy | None	how long a pause may stay open. Defaults to :data:DEFAULT_TTL.	None
consumption	Any	where spending a token is recorded. Defaults to a single-process store; see :func:redemption_is_shared.	None
**options	Any	forwarded to HumanInteractionCapability.compile -- ledger, checkpoints, clock and audit.	{}

Raises:

Type	Description
InteractionConfigurationError	if the composed capability would contribute nothing, naming what is missing.

Source code in src/symfonic/capabilities/human/factory.py

def human_interaction(
    *,
    signer: Any,
    binding: Callable[[], Any],
    encode_token: Callable[[Any], str],
    decode_token: Callable[[str], Any] | None = None,
    registrations: Iterable[InteractionRegistration] | None = None,
    ttl: TTLPolicy | None = None,
    consumption: Any = None,
    **options: Any,
) -> HumanInteractionCapability:
    """Build the capability that lets a turn pause and be resumed.

        Agent(provider, capabilities=[human_interaction(
            signer=my_signer,
            binding=lambda: PauseBinding(run_id=..., session_id=...),
            encode_token=my_encoder,
        )])

    Three arguments are required because only the deployment can supply them,
    and because the capability contributes nothing without them.

    Args:
        signer: mints and verifies pause tokens. No default: a shared one would
            make every deployment honour every other's tokens.
        binding: resolves what *this run* is, per run. No default: a
            ``PauseBinding()`` with empty fields is a token bound to nothing.
        encode_token: renders a minted envelope as the opaque string a person
            answers with. No default: the token format is a deployment's
            choice, not the framework's.
        decode_token: the inverse. Optional, and its absence has a cost --
            without it a pause taken on the kernel route has no public reader,
            so the run can be stopped and not continued.
        registrations: which interactions this agent offers. Defaults to
            ``ask_user`` alone -- the one every deployment wants and the only
            one whose schema is obvious.
        ttl: how long a pause may stay open. Defaults to :data:`DEFAULT_TTL`.
        consumption: where spending a token is recorded. Defaults to a
            single-process store; see :func:`redemption_is_shared`.
        **options: forwarded to ``HumanInteractionCapability.compile`` --
            ledger, checkpoints, clock and audit.

    Raises:
        InteractionConfigurationError: if the composed capability would
            contribute nothing, naming what is missing.
    """
    declared = tuple(registrations) if registrations is not None else (
        ask_user_registration(),
    )
    if not declared:
        raise InteractionConfigurationError(
            "no interaction is registered, so this capability would advertise "
            "no tool and never pause. Pass registrations, or leave the argument "
            "out to get ask_user."
        )
    seam = consumption if consumption is not None else InMemoryConsumption()
    return HumanInteractionCapability.compile(
        declared,
        signer=signer,
        ttl=ttl if ttl is not None else DEFAULT_TTL,
        consumption=seam,
        binding=binding,
        encode_token=encode_token,
        decode_token=decode_token,
        **options,
    )

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}"

redemption_is_shared

redemption_is_shared(consumption: Any) -> bool

Whether two replicas would agree that a token has been spent.

False means each replica holds its own record, so a token redeemed on one can be redeemed again on another. That is a legitimate choice for a single-process deployment and an unpleasant surprise for any other, which is the reason it is reported rather than assumed.

Source code in src/symfonic/capabilities/human/factory.py

def redemption_is_shared(consumption: Any) -> bool:
    """Whether two replicas would agree that a token has been spent.

    ``False`` means each replica holds its own record, so a token redeemed on
    one can be redeemed again on another. That is a legitimate choice for a
    single-process deployment and an unpleasant surprise for any other, which
    is the reason it is reported rather than assumed.
    """
    return not isinstance(consumption, InMemoryConsumption)

require_durable_consumption

require_durable_consumption(consumption: Any, *, checkpoints_durable: bool) -> None

SEC-PTK-7 — durable checkpoints must not get a volatile jti store.

A deployment whose checkpoints survive a restart and whose consumed set does not has single-use enforcement only until the next deploy: every token minted before the restart becomes redeemable a second time, and nothing in the logs says so.

A fully volatile deployment is supported, not an error — that is the developer laptop, and its tokens die with the process anyway.

Source code in src/symfonic/capabilities/human/consumption.py

def require_durable_consumption(consumption: Any, *, checkpoints_durable: bool) -> None:
    """SEC-PTK-7 — durable checkpoints must not get a volatile jti store.

    A deployment whose checkpoints survive a restart and whose consumed set does
    not has single-use enforcement only until the next deploy: every token
    minted before the restart becomes redeemable a second time, and nothing in
    the logs says so.

    A fully volatile deployment is supported, not an error — that is the
    developer laptop, and its tokens die with the process anyway.
    """
    if checkpoints_durable and not declared_durability(consumption):
        raise ConsumptionDurabilityError(
            "this deployment has durable checkpoints and a volatile pause-token "
            "consumption store, so single use would not survive a restart. Back "
            "consumption with the same persistence backend as the checkpointer, "
            "or state explicitly that this deployment does not need it"
        )

thread_id_for

thread_id_for(scope: Any, session_id: str) -> str

tenant:sub:session -- the legacy formula, character for character.

Read structurally, like every other scope reader in this package: the tenancy scope type lives outside this layer and importing it would make the capability depend on the engine it is being extracted from.

A missing tenant_id or session_id is refused rather than folded into an empty string. ":_:" and "acme:_:" are perfectly valid-looking keys that every unattributed caller would share, which is the same failure :func:~symfonic.capabilities.human.binding.hash_scope refuses one layer over -- and here it would be a durable one, since the checkpoints filed under such a key outlive the run that wrote them.

Source code in src/symfonic/capabilities/human/threads.py

def thread_id_for(scope: Any, session_id: str) -> str:
    """``tenant:sub:session`` -- the legacy formula, character for character.

    Read structurally, like every other scope reader in this package: the
    tenancy scope type lives outside this layer and importing it would make the
    capability depend on the engine it is being extracted from.

    A missing ``tenant_id`` or ``session_id`` is refused rather than folded into
    an empty string. ``":_:"`` and ``"acme:_:"`` are perfectly valid-looking
    keys that every unattributed caller would share, which is the same failure
    :func:`~symfonic.capabilities.human.binding.hash_scope` refuses one layer
    over -- and here it would be a *durable* one, since the checkpoints filed
    under such a key outlive the run that wrote them.
    """
    tenant_id = getattr(scope, "tenant_id", None)
    if not tenant_id:
        raise InteractionConfigurationError(
            "cannot derive a checkpoint thread id from a scope with no "
            "tenant_id; every unattributed run would share one thread and read "
            "back each other's paused state"
        )
    if not session_id:
        raise InteractionConfigurationError(
            "cannot derive a checkpoint thread id with no session_id; the "
            "session is what separates one conversation's paused state from "
            "the next one's within the same tenant"
        )
    sub = getattr(scope, "sub_tenant_id", None) or ABSENT_SUB_TENANT
    return f"{tenant_id}:{sub}:{session_id}"

turn_state_key

turn_state_key(claims: PauseClaims) -> str

Where this pause's turn state lives.

Keyed by jti rather than by tool_call_id or interrupt_id, which is the one place this module deliberately departs from :func:~symfonic.capabilities.human.checkpoints.payload_key. Those two keys are frozen because tokens in flight point at them; this key is new, so it is free to be keyed by the thing that is unique per mint. A round that paused twice on the same reserved call -- a resumed run that pauses again on a retry of the same call id -- would otherwise overwrite the first pause's state with the second's, and the first token would then continue from a transcript that already contains its own answer.

Source code in src/symfonic/capabilities/human/turnstate.py

def turn_state_key(claims: PauseClaims) -> str:
    """Where this pause's turn state lives.

    Keyed by ``jti`` rather than by ``tool_call_id`` or ``interrupt_id``, which
    is the one place this module deliberately departs from
    :func:`~symfonic.capabilities.human.checkpoints.payload_key`. Those two keys
    are frozen because tokens in flight point at them; this key is new, so it is
    free to be keyed by the thing that is unique per *mint*. A round that paused
    twice on the same reserved call -- a resumed run that pauses again on a
    retry of the same call id -- would otherwise overwrite the first pause's
    state with the second's, and the first token would then continue from a
    transcript that already contains its own answer.
    """
    if not claims.jti:
        raise PayloadBindingError(
            "these pause claims carry no jti, so the paused turn's state has no "
            "key; nothing could record or recover it"
        )
    return f"{TURN_STATE_KEY_PREFIX}{claims.jti}"