symfonic.services.conversation.recovery

recovery

Restart recovery and state overrides.

Two things the legacy engine did inline, extracted because both are decisions rather than plumbing:

Restart recovery answers "what should happen to this thread after a restart?" — and answers it against the registry, not against the backend. A checkpoint the registry cannot account for is quarantined rather than resumed: unaccounted-for durable state is exactly what an authoritative registry exists to notice.

"Cannot account for" is not the same as "has not yet heard of", and the difference is :meth:RestartRecoveryService.adopt. The registry is process-local and starts empty, and the legacy engine never knew it existed — so without a rehydrate step every restart would quarantine both the legacy path's checkpoints and this package's own. Adoption enumerates the thread's durable rows and derives each ref's provenance from its envelope (absent envelope → legacy, format version 0), which is the one place where the backend informs the registry. Rows it cannot read, and rows the backend reports as latest but does not list, stay unaccounted for and still quarantine.

Finalization is adopted the same way — read off the envelope, never assumed — so a write that was registered unfinalized and interrupted by a crash that ended the process comes back unfinalized and is visible to crash-expiry.

State overrides keep the run-config keys (_thread_id, _checkpoint_id) out of graph state. They look like state, they are passed alongside state, and putting them in state silently changes what the graph sees on every resume.

RecoveryDecision dataclass

RecoveryDecision(action: RecoveryAction, thread_id: str, reason: str, checkpoint_id: str | None = None, safe_boundary: SafeBoundaryMarker | None = None)

What to do with a thread's durable state after a restart.

RestartRecoveryService

RestartRecoveryService(*, adapter: CheckpointAdapterPort | None, registry: CheckpointRegistry, clock: object | None = None)

Decides resume / migrate / quarantine / fresh for a restarted thread.

Source code in src/symfonic/services/conversation/recovery.py

def __init__(
    self,
    *,
    adapter: CheckpointAdapterPort | None,
    registry: CheckpointRegistry,
    clock: object | None = None,
) -> None:
    self._adapter = adapter
    self._registry = registry
    self._clock = clock if clock is not None else _SystemClock()

adopt async

adopt(thread_id: str) -> tuple[CheckpointRef, ...]

Rehydrate the registry from a thread's durable state.

Provenance is read, never assumed: each row's envelope says which line wrote it, and a row with no envelope is legacy at format version 0 — the positive statement :func:decode_envelope makes, not a guess. A row whose envelope is corrupt or from a newer format is skipped, so it stays unaccounted for and quarantines rather than resuming under an invented provenance. Already-registered rows are left exactly as they are: adoption never overwrites the registry's own record.

Safe boundaries are adopted only under the id the writer recorded in the envelope. A row that claims to be a boundary but names none leaves nothing to replay from, and inventing an id from its listing position would be worse than having none — the migrator would hand a replay port a boundary that never existed.

Returns the refs this call adopted, so a caller can log what a restart took ownership of.

Source code in src/symfonic/services/conversation/recovery.py

async def adopt(self, thread_id: str) -> tuple[CheckpointRef, ...]:
    """Rehydrate the registry from a thread's durable state.

    Provenance is *read*, never assumed: each row's envelope says which
    line wrote it, and a row with no envelope is legacy at format version
    0 — the positive statement :func:`decode_envelope` makes, not a guess.
    A row whose envelope is corrupt or from a newer format is skipped, so
    it stays unaccounted for and quarantines rather than resuming under an
    invented provenance. Already-registered rows are left exactly as they
    are: adoption never overwrites the registry's own record.

    Safe boundaries are adopted only under the id the writer recorded in
    the envelope. A row that claims to be a boundary but names none leaves
    nothing to replay *from*, and inventing an id from its listing position
    would be worse than having none — the migrator would hand a replay port
    a boundary that never existed.

    Returns the refs this call adopted, so a caller can log what a restart
    took ownership of.
    """
    if self._adapter is None:
        return ()
    rows = await self._adapter.list_checkpoints(thread_id)
    adopted: list[CheckpointRef] = []
    for sequence, row in enumerate(rows or ()):
        checkpoint_id, metadata = row
        if self._registry.get(thread_id, checkpoint_id) is not None:
            continue
        try:
            envelope = decode_envelope(metadata)
        except CheckpointFormatError:
            continue
        finalized = _adopted_finalization(envelope)
        ref = self._registry.register(
            CheckpointRef(
                thread_id=thread_id,
                checkpoint_id=checkpoint_id,
                writer_line=envelope.writer_line,
                format_version=envelope.format_version,
                created_at=self._clock.now(),
                package_version=envelope.package_version,
                safe_boundary=envelope.safe_boundary,
                finalized=finalized,
            ),
            finalized=finalized,
            # Adoption is not issuance: every row here was read *out of*
            # the backend and already exists durably. Refusing it during a
            # freeze would not prevent any state from existing — it would
            # only leave the thread unaccounted for, so a restart mid-drain
            # would quarantine every thread and none could be migrated out,
            # ending the drain the freeze was declared for.
            issuance=False,
        )
        adopted.append(ref)
        if envelope.safe_boundary and envelope.boundary_id is not None:
            # Use the id the *writer* recorded, never a re-derived one.
            # ``derive_id`` hashes (thread, sequence, digest), and after a
            # restart neither the sequence nor the digest the writer used
            # is recoverable from a listing — so re-deriving produced a
            # second id for one physical boundary and handed ReplayPort a
            # boundary the writer never marked.
            self._registry.mark_safe_boundary(
                thread_id,
                sequence=(
                    envelope.boundary_sequence
                    if envelope.boundary_sequence is not None
                    else sequence
                ),
                digest=checkpoint_id,
                writer_line=envelope.writer_line,
                boundary_id=envelope.boundary_id,
            )
    return tuple(adopted)

StateOverrides

Splits run-config keys out of graph-state overrides.

split classmethod

split(overrides: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]

Return (configurable, graph_state) without mutating the input.

A None for a configurable key is refused rather than dropped: the caller meant to pass a thread id and computed nothing, and silently continuing starts a brand-new thread under the same session.

Source code in src/symfonic/services/conversation/recovery.py

@classmethod
def split(cls, overrides: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    """Return ``(configurable, graph_state)`` without mutating the input.

    A ``None`` for a configurable key is refused rather than dropped: the
    caller meant to pass a thread id and computed nothing, and silently
    continuing starts a brand-new thread under the same session.
    """
    configurable: dict[str, Any] = {}
    state: dict[str, Any] = {}
    for key, value in overrides.items():
        target = cls.CONFIGURABLE_KEYS.get(key)
        if target is None:
            state[key] = value
            continue
        if value is None:
            raise ConversationServiceError(
                f"state override {key!r} is None; refusing to start an "
                "unrelated thread in place of the one you meant"
            )
        configurable[target] = value
    return configurable, state