symfonic.kernel.contracts.checkpoints

checkpoints

What a paused turn has to leave behind to be continued (HK2, TA8.35).

HK1 gave the kernel a way to stop. Stopping is only half of a pause: a run that stopped and kept its state in the process that stopped it is a cache, and a cache does not survive the deployment rolling, the worker recycling, or the answer arriving forty minutes later on a different pod. TA8.19 recorded the checkpointer thread/key role and the rehydration role as absent on the kernel route, with evidence. This module is the kernel-contracts half of making them present.

:class:TurnCheckpoint is the state: which thread and checkpoint the pause is bound to, which call it stopped on, and the messages the turn had produced when it stopped. It is a stdlib value with a JSON body, for the same reason :class:~symfonic.kernel.contracts.interrupts.PendingInterrupt is one -- the capability that writes one and the route that reads one both name it without naming each other, and what is written has to survive a process boundary rather than a function call.

Messages travel as mappings, not as facade objects. The kernel does not know what an adopter's Message is and must not learn; a checkpoint that pickled one would bind a durable record to an importable class, which is the shape that turns a library upgrade into an unreadable backlog of paused runs. The mapping form is the wire, and the layer that owns the message vocabulary converts in both directions.

TurnCheckpoint dataclass

TurnCheckpoint(run_id: str = '', session_id: str = '', thread_id: str = '', checkpoint_id: str = '', scope_hash: str = '', tool_call_id: str = '', interrupt_id: str = '', name: str = '', prompt: str = '', messages: tuple[Mapping[str, Any], ...] = tuple(), version: int = TURN_STATE_VERSION)

One paused turn's continuable state, keyed by thread and checkpoint.

Every identity field is here as well as on the pause token's claims, and that duplication is deliberate: the resume path checks them against each other. A checkpoint whose run_id disagrees with the token's is a record the token does not describe, and continuing from it would answer one run's question into another run's transcript.

decode classmethod

decode(raw: str) -> TurnCheckpoint

Read a recorded body, refusing a version this build cannot account for.

ValueError rather than a capability error: this module is kernel contracts and names no taxonomy. The caller that has one wraps it.

Source code in src/symfonic/kernel/contracts/checkpoints.py

@classmethod
def decode(cls, raw: str) -> TurnCheckpoint:
    """Read a recorded body, refusing a version this build cannot account for.

    ``ValueError`` rather than a capability error: this module is kernel
    contracts and names no taxonomy. The caller that has one wraps it.
    """
    body = json.loads(raw)
    if not isinstance(body, dict):
        raise ValueError(
            f"a recorded turn state must be a mapping, got {type(body).__name__}"
        )
    version = body.get("version")
    if version != TURN_STATE_VERSION:
        raise ValueError(
            f"this recorded turn state is version {version!r} and this build "
            f"reads version {TURN_STATE_VERSION}; refusing to continue a turn "
            "from a state whose field meanings it cannot account for"
        )
    messages = body.get("messages") or []
    if not isinstance(messages, list):
        raise ValueError("a recorded turn state's messages must be a list")
    return cls(
        run_id=str(body.get("run_id", "")),
        session_id=str(body.get("session_id", "")),
        thread_id=str(body.get("thread_id", "")),
        checkpoint_id=str(body.get("checkpoint_id", "")),
        scope_hash=str(body.get("scope_hash", "")),
        tool_call_id=str(body.get("tool_call_id", "")),
        interrupt_id=str(body.get("interrupt_id", "")),
        name=str(body.get("name", "")),
        prompt=str(body.get("prompt", "")),
        messages=tuple(dict(message) for message in messages),
        version=TURN_STATE_VERSION,
    )

encode

encode() -> str

The canonical body a checkpoint store keeps. Deterministic.

Source code in src/symfonic/kernel/contracts/checkpoints.py

def encode(self) -> str:
    """The canonical body a checkpoint store keeps. Deterministic."""
    return json.dumps(
        {
            "version": self.version,
            "run_id": self.run_id,
            "session_id": self.session_id,
            "thread_id": self.thread_id,
            "checkpoint_id": self.checkpoint_id,
            "scope_hash": self.scope_hash,
            "tool_call_id": self.tool_call_id,
            "interrupt_id": self.interrupt_id,
            "name": self.name,
            "prompt": self.prompt,
            "messages": [dict(message) for message in self.messages],
        },
        separators=(",", ":"),
        sort_keys=True,
    )

captured_messages

captured_messages(transcript: Any, turn: Any = None, requests: Sequence[Any] = ()) -> tuple[dict[str, Any], ...]

The turn's messages at the moment it paused, including the asking round.

Two sources, and the second is the one that is easy to miss. The transcript holds every closed round; the round that reached pre-tool is not closed yet -- ConversationPort.close_round runs after the tools it is waiting on. So the assistant message carrying the very call the pause is bound to exists nowhere but in turn and requests, and a checkpoint taken from the transcript alone would rehydrate a conversation in which the model never asked the question being answered. The provider then receives a tool result for a call it cannot see, which every strict provider rejects.

requests are the reserved calls, so the ids here are the ids the resumed transcript joins on (RES-3).

Source code in src/symfonic/kernel/contracts/checkpoints.py

def captured_messages(
    transcript: Any, turn: Any = None, requests: Sequence[Any] = ()
) -> tuple[dict[str, Any], ...]:
    """The turn's messages at the moment it paused, including the asking round.

    Two sources, and the second is the one that is easy to miss. The transcript
    holds every *closed* round; the round that reached ``pre-tool`` is not
    closed yet -- ``ConversationPort.close_round`` runs after the tools it is
    waiting on. So the assistant message carrying the very call the pause is
    bound to exists nowhere but in ``turn`` and ``requests``, and a checkpoint
    taken from the transcript alone would rehydrate a conversation in which the
    model never asked the question being answered. The provider then receives a
    tool result for a call it cannot see, which every strict provider rejects.

    ``requests`` are the *reserved* calls, so the ids here are the ids the
    resumed transcript joins on (RES-3).
    """
    messages = [_message_dict(message) for message in getattr(transcript, "typed", ())]
    if requests:
        messages.append(
            {
                "role": "assistant",
                "content": str(getattr(turn, "text", "") or ""),
                "tool_calls": [_tool_call_dict(request) for request in requests],
            }
        )
    return tuple(messages)