symfonic.agent.cutover.kernel_resume

kernel_resume

Redeeming a pause on the kernel route, and continuing the turn (HK2, TA8.35).

HK1 left a token nobody could redeem and said so in the token itself. This is the other half: one preparation, three entry points, and an ordering that is enforced by the shapes rather than by a rule somebody has to remember.

The one resume contract, stated once. For run, stream and stream_typed alike:

1. the token authenticates, decodes and is inside its window;
2. it binds all four axes -- tenant scope, session, run and call -- and any one of them mismatching refuses, by its own name, fail-closed;
3. the paused request is recovered and hashed against the token's claim;
4. the answer validates against the interaction's registered response schema;
5. the paused turn is rebuilt from the checkpointer, never from process memory, and an absent one is refused by name rather than continued from nothing;
6. only then is the token consumed, exactly once, by one atomic claim;
7. the rehydrated messages become the turn's history before the kernel is entered, so the model's first sight of the resumed run already contains the answer.

Steps 1-5 can all fail, and each of them failing leaves the token unconsumed: consume is step 6 for exactly that reason. Step 7 cannot be reordered because :func:prepare returns the rehydrated turn and the three entry points take one -- there is no way to spell "enter the kernel and rehydrate afterwards".

Three entry points, one preparation. Three resume semantics would be three chances to disagree about what a token means, so the three functions below differ only in which projection of the same kernel invocation they return. resume_stream_typed is asserted against the kernel implementation even though SymfonicAgent.stream_typed still dispatches _stream_typed_impl: the flip is ST3's, and building the route is not flipping it.

ResumedTurn dataclass

ResumedTurn(outcome: ResumeOutcome, history: tuple[BaseMessage, ...], prompt: str = RESUME_PROMPT)

A redeemed token and the turn it continues, rehydrated and ready.

Holding both is what makes the ordering structural. The entry points take this value, so a caller cannot reach the kernel with an unredeemed token or with an un-rehydrated history: there is no other constructor for the thing they accept.

answered_event

answered_event(outcome: ResumeOutcome) -> Any | None

The AskUserAnsweredEvent the legacy body yields before continuing.

Parity, and it is only owed on the typed surface: SymfonicAgent.resume is a typed generator and emits this before it re-enters the graph, so a human-in-the-loop consumer written against stream_typed today receives it and would stop receiving it on a flipped route. run and stream have no legacy resume to differ from -- the shipped API has exactly one resume method -- so nothing is invented for them.

None for a deployment's own registered interaction: this event names selected_labels and a question set, which a generic interrupt does not have. The legacy body agrees -- resume_interrupt emits no equivalent -- so a fabricated one would be a difference rather than parity.

Source code in src/symfonic/agent/cutover/kernel_resume.py

def answered_event(outcome: ResumeOutcome) -> Any | None:
    """The ``AskUserAnsweredEvent`` the legacy body yields before continuing.

    Parity, and it is only owed on the typed surface: ``SymfonicAgent.resume``
    is a typed generator and emits this before it re-enters the graph, so a
    human-in-the-loop consumer written against ``stream_typed`` today receives
    it and would stop receiving it on a flipped route. ``run`` and ``stream``
    have no legacy resume to differ from -- the shipped API has exactly one
    resume method -- so nothing is invented for them.

    ``None`` for a deployment's own registered interaction: this event names
    ``selected_labels`` and a question set, which a generic interrupt does not
    have. The legacy body agrees -- ``resume_interrupt`` emits no equivalent --
    so a fabricated one would be a difference rather than parity.
    """
    from symfonic.capabilities.human.registration import ASK_USER
    from symfonic.core.contracts.types import AskUserAnsweredEvent

    answers = getattr(outcome.response, "answers", None)
    if outcome.name != ASK_USER or not answers:
        return None
    labels: list[str] = []
    for answer in answers:
        labels.extend(getattr(answer, "selections", ()) or ())
    return AskUserAnsweredEvent(
        run_id=outcome.run_id,
        session_id=outcome.session_id,
        tool_call_id=outcome.tool_call_id,
        selected_labels=labels,
        has_other=any(getattr(a, "other", None) is not None for a in answers),
        time_to_answer_seconds=outcome.time_to_resolve_seconds,
    )

prepare async

prepare(capability: Any, *, envelope: Any, response: Any, scope: Any, session_id: str, run_id: str, call_id: str) -> ResumedTurn

Redeem one token and rebuild the turn it paused. Steps 1-7, in order.

All four axes are required, not defaulted. :class:~symfonic.capabilities.human.values.ResumeCommand lets each be None for a transport that genuinely has no such fact, and an unstated axis is not checked. That default must never be how this route opts out: a resume endpoint that forgot to thread run_id would silently stop checking which turn an answer belongs to, and nothing would fail. So a missing value is refused here, before anything is redeemed.

Source code in src/symfonic/agent/cutover/kernel_resume.py

async def prepare(
    capability: Any,
    *,
    envelope: Any,
    response: Any,
    scope: Any,
    session_id: str,
    run_id: str,
    call_id: str,
) -> ResumedTurn:
    """Redeem one token and rebuild the turn it paused. Steps 1-7, in order.

    **All four axes are required, not defaulted.**
    :class:`~symfonic.capabilities.human.values.ResumeCommand` lets each be
    ``None`` for a transport that genuinely has no such fact, and an unstated
    axis is not checked. That default must never be how *this* route opts out:
    a resume endpoint that forgot to thread ``run_id`` would silently stop
    checking which turn an answer belongs to, and nothing would fail. So a
    missing value is refused here, before anything is redeemed.
    """
    for name, value in (
        ("session_id", session_id),
        ("run_id", run_id),
        ("call_id", call_id),
    ):
        if not value:
            raise InteractionConfigurationError(
                f"a kernel-route resume must state {name}; a redemption that "
                "leaves an axis unstated is not checked on that axis, and three "
                "of four checks is how a token gets redeemed against the wrong "
                "turn, session or question"
            )
    outcome = await capability.resume(
        ResumeCommand(
            envelope=envelope,
            response=response,
            scope=scope,
            session_id=session_id,
            run_id=run_id,
            call_id=call_id,
        ),
        # The route that continues the run is the route that must have the
        # state. See ``ResumeService.resume``: the legacy body must not.
        require_turn=True,
    )
    return ResumedTurn(outcome=outcome, history=rehydrate(outcome))

rehydrate

rehydrate(outcome: ResumeOutcome) -> tuple[BaseMessage, ...]

The paused turn's messages, with the answer folded in as its observation.

From the checkpoint, and only from the checkpoint. Every message here came out of outcome.turn, which :class:~symfonic.capabilities.human.turnstate.TurnCheckpointStore read from the checkpoint port on this call. Nothing is recovered from a cache keyed by run id, and nothing is remembered between processes -- which is what makes a worker that never saw the pause able to continue it.

The answer arrives as a tool message joined on the reserved call id, because that is what the paused round was waiting for: the model asked for a call, the kernel stopped before dispatching it, and the person's answer is that call's observation. Anything else -- a fresh user turn, say -- would leave the assistant's tool call unanswered in the transcript, which is the orphaned-call state every strict provider rejects on the next request.

Source code in src/symfonic/agent/cutover/rehydration.py

def rehydrate(outcome: ResumeOutcome) -> tuple[BaseMessage, ...]:
    """The paused turn's messages, with the answer folded in as its observation.

    **From the checkpoint, and only from the checkpoint.** Every message here
    came out of ``outcome.turn``, which
    :class:`~symfonic.capabilities.human.turnstate.TurnCheckpointStore` read
    from the checkpoint port on this call. Nothing is recovered from a cache
    keyed by run id, and nothing is remembered between processes -- which is
    what makes a worker that never saw the pause able to continue it.

    The answer arrives as a ``tool`` message joined on the reserved call id,
    because that is what the paused round was waiting for: the model asked for
    a call, the kernel stopped before dispatching it, and the person's answer is
    that call's observation. Anything else -- a fresh user turn, say -- would
    leave the assistant's tool call unanswered in the transcript, which is the
    orphaned-call state every strict provider rejects on the next request.
    """
    checkpoint = outcome.turn
    if checkpoint is None:
        raise InteractionConfigurationError(
            "this resume has no recorded turn state to rebuild from; a "
            "continuation assembled from nothing would answer the person's "
            "question into a conversation the model has never seen"
        )
    messages = [_message(dict(body)) for body in checkpoint.messages]
    reserved = _reserved_call_ids(messages)
    answered = outcome.tool_call_id or outcome.interrupt_id
    messages.append(
        ToolMessage(content=_answer_text(outcome.response), tool_call_id=answered)
    )
    # Every other call the round reserved gets an observation too. The pause
    # stopped the whole round at the first registered interaction, so the rest
    # never ran -- and a provider handed an assistant message with three tool
    # calls and one tool result rejects the request outright.
    messages.extend(
        ToolMessage(content=NOT_DISPATCHED, tool_call_id=call_id)
        for call_id in reserved
        if call_id and call_id != answered
    )
    return tuple(messages)

resume_run async

resume_run(delegate: Any, turn: ResumedTurn, **options: Any) -> AgentResponse

Continue a paused turn and collect its blocking projection.

Source code in src/symfonic/agent/cutover/kernel_resume.py

async def resume_run(delegate: Any, turn: ResumedTurn, **options: Any) -> AgentResponse:
    """Continue a paused turn and collect its blocking projection."""
    return await delegate.run(
        turn.prompt,
        run_id=turn.run_id,
        session_id=turn.session_id,
        history=turn.history,
        **options,
    )

resume_stream

resume_stream(delegate: Any, turn: ResumedTurn, **options: Any) -> AsyncIterator[StreamChunk]

Continue a paused turn and project it onto StreamChunk.

Source code in src/symfonic/agent/cutover/kernel_resume.py

def resume_stream(
    delegate: Any, turn: ResumedTurn, **options: Any
) -> AsyncIterator[StreamChunk]:
    """Continue a paused turn and project it onto ``StreamChunk``."""
    return delegate.stream(
        turn.prompt,
        run_id=turn.run_id,
        session_id=turn.session_id,
        history=turn.history,
        **options,
    )

resume_stream_typed async

resume_stream_typed(delegate: Any, turn: ResumedTurn, **options: Any) -> AsyncIterator[Any]

Continue a paused turn and project it onto StreamEvent.

Reached through :func:~symfonic.agent.cutover.typed_route.typed_route, the same body KernelDelegate.stream_typed reaches, so this is the kernel implementation of the typed route rather than a fourth one. That the public SymfonicAgent.stream_typed still dispatches _stream_typed_impl is ST3's business; the contract this function meets is the same one the other two meet, and it is asserted here rather than deferred to the flip.

The answered event comes first, as it does on the legacy body, and it comes from :func:answered_event rather than from a second construction site.

Source code in src/symfonic/agent/cutover/kernel_resume.py

async def resume_stream_typed(
    delegate: Any, turn: ResumedTurn, **options: Any
) -> AsyncIterator[Any]:
    """Continue a paused turn and project it onto ``StreamEvent``.

    Reached through :func:`~symfonic.agent.cutover.typed_route.typed_route`, the
    same body ``KernelDelegate.stream_typed`` reaches, so this is the kernel
    implementation of the typed route rather than a fourth one. That the public
    ``SymfonicAgent.stream_typed`` still dispatches ``_stream_typed_impl`` is
    ST3's business; the contract this function meets is the same one the other
    two meet, and it is asserted here rather than deferred to the flip.

    The answered event comes first, as it does on the legacy body, and it comes
    from :func:`answered_event` rather than from a second construction site.
    """
    answered = answered_event(turn.outcome)
    if answered is not None:
        yield answered
    stream = typed_route(
        delegate,
        turn.prompt,
        run_id=turn.run_id,
        session_id=turn.session_id,
        history=turn.history,
        **options,
    )
    try:
        async for event in stream:
            yield event
    finally:
        # The projection owns closing what it reads; this owns closing the
        # projection, because a consumer that walked away leaves it suspended.
        await closing(stream)