symfonic.services.conversation.checkpoint

checkpoint

Checkpointer readiness, flush, and close — the conversation-side owner.

This is the object T2.3.4's RunLifecycle registers as its CheckpointerPort. Three of its rules come straight from that contract and are the reason it exists as a service rather than as three flags on an engine:

• ensure_ready is lazy and idempotent per instance, because readiness can open a pool or run a migration, and the call site that happened to trigger it is not the call site that should own its teardown;
• a failed readiness does not latch — the next call retries, because a transient backend outage that permanently poisoned a run would be worse than the outage;
• flush reports failure and never raises, so a flush failure can never block close. A checkpointer that refuses to close because it could not write is a leaked pool.

A deployment with no checkpointer at all is a supported configuration, not an error: when neither durable driver is enabled, nothing is wired, and every verb here is a no-op.

CheckpointAdapterPort

Bases: Protocol

The durable-state handle this service governs.

Structurally a superset of the kernel's CheckpointerPort (readiness, flush, close) plus the two reads restart recovery needs. Adopters implement it with their own saver; nothing here knows what a checkpoint physically is.

latest async

latest(thread_id: str) -> Any

Newest (checkpoint_id, metadata) for a thread, or None.

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

async def latest(self, thread_id: str) -> Any:
    """Newest ``(checkpoint_id, metadata)`` for a thread, or ``None``."""

list_checkpoints async

list_checkpoints(thread_id: str) -> Any

Every (checkpoint_id, metadata) for a thread, oldest first.

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

async def list_checkpoints(self, thread_id: str) -> Any:
    """Every ``(checkpoint_id, metadata)`` for a thread, oldest first."""

CheckpointService

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

Owns readiness and teardown for one run's durable conversation state.

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

def __init__(
    self,
    *,
    adapter: CheckpointAdapterPort | None,
    registry: CheckpointRegistry | None = None,
    clock: object | None = None,
) -> None:
    self._adapter = adapter
    self._registry = registry
    self._clock = clock if clock is not None else _SystemClock()
    self._ready = False
    self._closed = False
    self._flush_failures: tuple[str, ...] = ()
    self._ready_lock = asyncio.Lock()

durable property

durable: bool

Whether any durable state exists at all for this deployment.

flush_failures property

flush_failures: tuple[str, ...]

Flush failures, as safe-to-log text. Never exception objects.

close async

close() -> None

Release the durable handle, flushed or not. Idempotent.

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

async def close(self) -> None:
    """Release the durable handle, flushed or not. Idempotent."""
    if self._adapter is None or self._closed:
        return
    self._closed = True
    self._ready = False
    await self._adapter.close()

ensure_ready async

ensure_ready() -> None

Open durable state once. Idempotent on success, retried on failure.

The lock is what makes "once" true under concurrency: readiness may open a pool or run a migration, and two coroutines of the same run that both observed _ready is False would otherwise do it twice. The second waiter re-checks after acquiring, so it costs one flag read on the hot path once readiness is established.

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

async def ensure_ready(self) -> None:
    """Open durable state once. Idempotent on success, retried on failure.

    The lock is what makes "once" true under concurrency: readiness may
    open a pool or run a migration, and two coroutines of the same run that
    both observed ``_ready is False`` would otherwise do it twice. The
    second waiter re-checks after acquiring, so it costs one flag read on
    the hot path once readiness is established.
    """
    if self._adapter is None or self._ready:
        return
    async with self._ready_lock:
        if self._ready:
            return
        await self._adapter.ensure_ready()
        self._ready = True

flush async

flush() -> None

Push buffered writes. Failure is recorded, never raised.

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

async def flush(self) -> None:
    """Push buffered writes. Failure is recorded, never raised."""
    if self._adapter is None or not self._ready:
        return
    try:
        await self._adapter.flush()
    except Exception as exc:  # noqa: BLE001 - the contract is "never raises"
        self._flush_failures = (*self._flush_failures, str(exc))

record_write async

record_write(thread_id: str, checkpoint_id: str, *, safe_boundary: bool = False, sequence: int | None = None, digest: str | None = None, package_version: str | None = None, tenant_id: str | None = None, finalized: bool = True) -> CheckpointRef

Register a checkpoint this service wrote as authoritative.

A safe-boundary write also marks the boundary, because a boundary nobody recorded is a boundary no future migration can replay from — and the moment a boundary is written is the only moment its sequence and digest are known for free.

That marker lives in a process-local registry, so the durable half is the caller's: stamp registry.latest_safe_boundary(thread_id)'s boundary_id and sequence into the envelope (:func:~symfonic.services.conversation.compat.encode_envelope) that goes out with the state. Restart adoption re-adopts the boundary under exactly that id and never re-derives one, so the id a migration replays from is the id a writer marked — before and after a restart alike.

finalized=False opens the crash window on this ref. Its durable half is the caller's in the same way: stamp encode_envelope(..., finalized=False) into the metadata that goes out with the write, and a finalized=True envelope when the write is closed. A restart adopts what the envelope says, so an unfinalized row left behind by a crash that ended the process is expired by reconcile_crash_expiry with a reason rather than resumed mid-frame.

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

async def record_write(
    self,
    thread_id: str,
    checkpoint_id: str,
    *,
    safe_boundary: bool = False,
    sequence: int | None = None,
    digest: str | None = None,
    package_version: str | None = None,
    tenant_id: str | None = None,
    finalized: bool = True,
) -> CheckpointRef:
    """Register a checkpoint this service wrote as authoritative.

    A safe-boundary write also marks the boundary, because a boundary
    nobody recorded is a boundary no future migration can replay from —
    and the moment a boundary is written is the only moment its sequence
    and digest are known for free.

    That marker lives in a process-local registry, so the *durable* half
    is the caller's: stamp ``registry.latest_safe_boundary(thread_id)``'s
    ``boundary_id`` and ``sequence`` into the envelope
    (:func:`~symfonic.services.conversation.compat.encode_envelope`) that
    goes out with the state. Restart adoption re-adopts the boundary under
    exactly that id and never re-derives one, so the id a migration replays
    from is the id a writer marked — before and after a restart alike.

    ``finalized=False`` opens the crash window on this ref. Its durable
    half is the caller's in the same way: stamp
    ``encode_envelope(..., finalized=False)`` into the metadata that goes
    out with the write, and a ``finalized=True`` envelope when the write is
    closed. A restart adopts what the envelope says, so an unfinalized row
    left behind by a crash that ended the process is expired by
    ``reconcile_crash_expiry`` with a reason rather than resumed mid-frame.
    """
    ref = CheckpointRef(
        thread_id=thread_id,
        checkpoint_id=checkpoint_id,
        writer_line="migrated",
        format_version=1,
        created_at=self._clock.now(),
        package_version=package_version,
        safe_boundary=safe_boundary,
        finalized=finalized,
        tenant_id=tenant_id,
    )
    if self._registry is None:
        return ref
    registered = self._registry.register(ref, finalized=finalized)
    if safe_boundary and sequence is not None:
        self._registry.mark_safe_boundary(
            thread_id,
            sequence=sequence,
            digest=digest or checkpoint_id,
            writer_line="migrated",
        )
    return registered