symfonic.capabilities.human.turnstate

turnstate

The checkpointer role, on the kernel route (HK2, TA8.35).

TA8.19 §2.2 recorded this role as absent with evidence: the migrated path poured no _thread_id, held no checkpointer, and had no LangGraph configurable mapping to put one in. This module is the role, made present -- and made present on the port that already existed rather than on a new one.

Why the existing checkpoint port and not a second one. :class:~symfonic.capabilities.human.contracts.CheckpointCommandPort is already the surface a pause writes its request through, already keyed by (thread_id, checkpoint_id, key), and already the thing a deployment wires when it wants pauses to survive a restart. A second port would mean a deployment could wire durable requests and volatile turn state, or the reverse, and would then discover the mismatch only when somebody answered a question -- which is precisely the class of failure require_durable_consumption (SEC-PTK-7) exists to refuse at construction.

A third key, declared as additive. ask_user_request:<tool_call_id> and interrupt_payload:<interrupt_id> are frozen compatibility contracts (:mod:~symfonic.capabilities.human.checkpoints) and neither is touched here. The turn state lives under a new prefix, so a token minted by the pre-extraction engine still finds its request exactly where it left it. What such a token does not find is a recorded turn state, because the build that minted it did not write one -- and this module says exactly that rather than reporting a missing checkpoint or, worse, "possible tampering". A pause in flight across the upgrade is answerable as far as its own build could make it answerable, and the one thing it cannot do is named.

Recording is not best-effort here, and that is the difference from the payload store. A request that failed to record still leaves a pause a human can be shown; a turn state that failed to record leaves a token that cannot continue anything. So the write reports whether it landed, and the pause reads that report to decide whether it may call itself resumable -- earned after the write, never announced before it.

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

stamped_turn

stamped_turn(state: TurnCheckpoint, claims: Any) -> TurnCheckpoint

Stamp the checkpoint with the identity the token actually carries.

The caller supplies the messages and the prompt; it does not get to supply the ids. thread_id and checkpoint_id in particular are resolved during the mint -- a caller that guessed them would file the record under a key the resume path does not look in, and the pause would be unresumable with nothing having reported a failure.

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

def stamped_turn(state: TurnCheckpoint, claims: Any) -> TurnCheckpoint:
    """Stamp the checkpoint with the identity the *token* actually carries.

    The caller supplies the messages and the prompt; it does not get to supply
    the ids. ``thread_id`` and ``checkpoint_id`` in particular are resolved
    during the mint -- a caller that guessed them would file the record under a
    key the resume path does not look in, and the pause would be unresumable
    with nothing having reported a failure.
    """
    return TurnCheckpoint(
        run_id=claims.run_id,
        session_id=claims.session_id,
        thread_id=claims.thread_id,
        checkpoint_id=claims.checkpoint_id,
        scope_hash=claims.scope_hash,
        tool_call_id=claims.tool_call_id,
        interrupt_id=claims.interrupt_id,
        name=claims.name,
        prompt=state.prompt,
        messages=state.messages,
    )

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