symfonic.services.conversation.registry

registry

The authoritative checkpoint registry.

"Authoritative" in the strong sense: a checkpoint the registry has never seen is not resumable, however healthy it looks in the backend. That inversion is the point — without it, "which checkpoints exist?" is answered by whichever storage engine happens to be attached, and no drain, freeze, or horizon can ever reach a fixed point.

Three governance verbs live here and nowhere else:

• issuance freeze — closes new issuance while leaving reads and idempotent replays open, so a retirement drain can finish;
• idempotent safe-boundary markers — derived ids, so marking the same boundary from a retry or a second process yields one marker;
• crash-expiry reconciliation — state that was issued but never finalized is expired explicitly, with a reason, rather than lingering as a resumable frame nobody can vouch for.

CheckpointRegistry

CheckpointRegistry(*, clock: object | None = None)

The single source of truth for which checkpoints exist and may resume.

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

def __init__(self, *, clock: object | None = None) -> None:
    self._clock = clock if clock is not None else _SystemClock()
    self._refs: dict[tuple[str, str], CheckpointRef] = {}
    self._boundaries: dict[str, dict[str, SafeBoundaryMarker]] = {}
    self._migrations: dict[tuple[str, str], MigrationLink] = {}
    self._migration_locks: dict[tuple[str, str], asyncio.Lock] = {}
    self._freeze: IssuanceFreeze | None = None

expire

expire(thread_id: str, checkpoint_id: str, reason: str) -> CheckpointRef

Mark a ref expired. Recorded, never deleted.

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

def expire(self, thread_id: str, checkpoint_id: str, reason: str) -> CheckpointRef:
    """Mark a ref expired. Recorded, never deleted."""
    key = (thread_id, checkpoint_id)
    ref = self._refs[key]
    expired = ref.expire(reason)
    self._refs[key] = expired
    return expired

finalize

finalize(thread_id: str, checkpoint_id: str) -> CheckpointRef

Close the crash window on a ref. Unknown refs are refused.

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

def finalize(self, thread_id: str, checkpoint_id: str) -> CheckpointRef:
    """Close the crash window on a ref. Unknown refs are refused."""
    key = (thread_id, checkpoint_id)
    ref = self._refs.get(key)
    if ref is None:
        raise KeyError(f"no registered checkpoint {checkpoint_id!r} on {thread_id!r}")
    finalized = ref.finalize()
    self._refs[key] = finalized
    return finalized

freeze_issuance

freeze_issuance(*, reason: str) -> IssuanceFreeze

Close new issuance. The first reason is the one that is kept.

A second freeze does not overwrite the first: the operator who declared the drain is the one whose reason belongs in the record.

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

def freeze_issuance(self, *, reason: str) -> IssuanceFreeze:
    """Close new issuance. The first reason is the one that is kept.

    A second freeze does not overwrite the first: the operator who
    declared the drain is the one whose reason belongs in the record.
    """
    if self._freeze is None:
        self._freeze = IssuanceFreeze(reason=reason, at=self._clock.now())
    return self._freeze

mark_safe_boundary

mark_safe_boundary(thread_id: str, *, sequence: int, digest: str, writer_line: WriterLine, boundary_id: str | None = None) -> SafeBoundaryMarker

Record a replayable boundary. Idempotent by derived id.

boundary_id re-adopts a boundary under the id its writer derived, which is what makes the marker idempotent across a restart as well as within one process: the (sequence, digest) it was hashed from are not recoverable from a bare listing, so re-deriving mints a second id.

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

def mark_safe_boundary(
    self,
    thread_id: str,
    *,
    sequence: int,
    digest: str,
    writer_line: WriterLine,
    boundary_id: str | None = None,
) -> SafeBoundaryMarker:
    """Record a replayable boundary. Idempotent by derived id.

    ``boundary_id`` re-adopts a boundary under the id its *writer* derived,
    which is what makes the marker idempotent across a restart as well as
    within one process: the (sequence, digest) it was hashed from are not
    recoverable from a bare listing, so re-deriving mints a second id.
    """
    marker = SafeBoundaryMarker.create(
        thread_id=thread_id,
        sequence=sequence,
        digest=digest,
        writer_line=writer_line,
        created_at=self._clock.now(),
        boundary_id=boundary_id,
    )
    bucket = self._boundaries.setdefault(thread_id, {})
    return bucket.setdefault(marker.boundary_id, marker)

migration_lock

migration_lock(thread_id: str, legacy_checkpoint_id: str) -> asyncio.Lock

The lock every migrator must hold while replaying this legacy ref.

It lives here, not on the migrator, for the same reason :class:MigrationLink does: the window it closes spans an awaited replay whose side effects the port commits before any link exists, so a lock scoped to one migrator instance serialises nothing once two migrators share this registry. record_migration is first-wins, but by then the loser has already replayed. Cross-process exclusion is out of scope for an in-memory registry; a durable implementation supplies it by making this lock durable.

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

def migration_lock(self, thread_id: str, legacy_checkpoint_id: str) -> asyncio.Lock:
    """The lock every migrator must hold while replaying this legacy ref.

    It lives here, not on the migrator, for the same reason
    :class:`MigrationLink` does: the window it closes spans an *awaited*
    replay whose side effects the port commits before any link exists, so
    a lock scoped to one migrator instance serialises nothing once two
    migrators share this registry. ``record_migration`` is first-wins, but
    by then the loser has already replayed. Cross-*process* exclusion is
    out of scope for an in-memory registry; a durable implementation
    supplies it by making this lock durable.
    """
    key = (thread_id, legacy_checkpoint_id)
    lock = self._migration_locks.get(key)
    if lock is None:
        lock = asyncio.Lock()
        self._migration_locks[key] = lock
    return lock

migration_of

migration_of(thread_id: str, checkpoint_id: str) -> MigrationLink | None

The replay this legacy ref already produced, if any.

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

def migration_of(self, thread_id: str, checkpoint_id: str) -> MigrationLink | None:
    """The replay this legacy ref already produced, if any."""
    return self._migrations.get((thread_id, checkpoint_id))

reconcile_crash_expiry

reconcile_crash_expiry(*, grace: timedelta) -> ReconciliationReport

Expire issued-but-never-finalized state older than grace.

Idempotent: an already-expired ref is not reported a second time, so a reconciler on a timer does not manufacture a rising expiry count.

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

def reconcile_crash_expiry(self, *, grace: timedelta) -> ReconciliationReport:
    """Expire issued-but-never-finalized state older than ``grace``.

    Idempotent: an already-expired ref is not reported a second time, so a
    reconciler on a timer does not manufacture a rising expiry count.
    """
    now = self._clock.now()
    expired: list[CheckpointRef] = []
    inspected = 0
    for key, ref in list(self._refs.items()):
        if ref.finalized or ref.expired:
            continue
        inspected += 1
        if now - ref.created_at < grace:
            continue
        marked = ref.expire("crash-expiry: issued but never finalized")
        self._refs[key] = marked
        expired.append(marked)
    return ReconciliationReport(
        at=now,
        grace_seconds=grace.total_seconds(),
        expired=tuple(expired),
        inspected=inspected,
    )

record_migration

record_migration(thread_id: str, legacy_checkpoint_id: str, *, checkpoint_id: str, boundary_id: str) -> MigrationLink

Link a legacy ref to the checkpoint its replay produced.

Idempotent, and the first link wins: if two racing replays somehow both landed, the one already recorded is the one every later reader resolves to, so a legacy ref never resolves to two different migrated checkpoints depending on who asks.

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

def record_migration(
    self,
    thread_id: str,
    legacy_checkpoint_id: str,
    *,
    checkpoint_id: str,
    boundary_id: str,
) -> MigrationLink:
    """Link a legacy ref to the checkpoint its replay produced.

    Idempotent, and the *first* link wins: if two racing replays somehow
    both landed, the one already recorded is the one every later reader
    resolves to, so a legacy ref never resolves to two different migrated
    checkpoints depending on who asks.
    """
    key = (thread_id, legacy_checkpoint_id)
    existing = self._migrations.get(key)
    if existing is not None:
        return existing
    link = MigrationLink(
        thread_id=thread_id,
        legacy_checkpoint_id=legacy_checkpoint_id,
        checkpoint_id=checkpoint_id,
        boundary_id=boundary_id,
        at=self._clock.now(),
    )
    self._migrations[key] = link
    return link

refs_for

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

Every known ref for a thread, oldest first.

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

def refs_for(self, thread_id: str) -> tuple[CheckpointRef, ...]:
    """Every known ref for a thread, oldest first."""
    rows = [ref for (tid, _), ref in self._refs.items() if tid == thread_id]
    return tuple(sorted(rows, key=lambda r: (r.created_at, r.checkpoint_id)))

register

register(ref: CheckpointRef, *, finalized: bool = True, issuance: bool = True) -> CheckpointRef

Record a checkpoint. Idempotent; refuses a changed writer line.

Re-registering an identical ref while frozen is explicitly allowed: an idempotent replay of state that was already issued is not issuance, and refusing it would make a retry during a drain look like a new checkpoint.

issuance=False says "this ref records durable state that already exists; registering it is bookkeeping catch-up", so the freeze does not apply. Two callers may say it: :meth:register_replay (which keys the claim to a legacy ref already accounted for) and restart adoption (whose rows were read out of the backend, so refusing them prevents no state from existing — it only leaves the thread quarantined).

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

def register(
    self, ref: CheckpointRef, *, finalized: bool = True, issuance: bool = True
) -> CheckpointRef:
    """Record a checkpoint. Idempotent; refuses a changed writer line.

    Re-registering an identical ref while frozen is explicitly allowed: an
    idempotent replay of state that was already issued is not issuance,
    and refusing it would make a retry during a drain look like a new
    checkpoint.

    ``issuance=False`` says "this ref records durable state that already
    exists; registering it is bookkeeping catch-up", so the freeze does not
    apply. Two callers may say it: :meth:`register_replay` (which keys the
    claim to a legacy ref already accounted for) and restart adoption
    (whose rows were read *out of* the backend, so refusing them prevents
    no state from existing — it only leaves the thread quarantined).
    """
    key = (ref.thread_id, ref.checkpoint_id)
    existing = self._refs.get(key)
    if existing is not None:
        if existing.writer_line != ref.writer_line:
            raise ValueError(
                f"checkpoint {ref.checkpoint_id!r} is already registered as "
                f"{existing.writer_line!r}; a writer line never changes"
            )
        return existing
    if issuance and self._freeze is not None:
        raise IssuanceFrozenError(
            f"checkpoint issuance is frozen ({self._freeze.reason}); "
            f"{ref.checkpoint_id!r} cannot be registered"
        )
    stored = ref if finalized else replace(ref, finalized=False)
    self._refs[key] = stored
    return stored

register_replay

register_replay(ref: CheckpointRef, *, legacy_checkpoint_id: str) -> CheckpointRef

Register the checkpoint a safe-boundary replay produced.

This is the write a retirement drain is made of, so it survives an issuance freeze — otherwise freezing issuance to drain legacy state would guarantee nothing could ever be drained, and the freeze's stated purpose ("leaving reads and idempotent replays open, so a retirement drain can finish") would be unreachable.

The exemption is keyed, not blanket: while frozen, the legacy ref must already be authoritative here. A replay of something this registry has never seen is new state wearing the word "migration".

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

def register_replay(
    self, ref: CheckpointRef, *, legacy_checkpoint_id: str
) -> CheckpointRef:
    """Register the checkpoint a safe-boundary replay produced.

    This is the write a retirement drain is *made of*, so it survives an
    issuance freeze — otherwise freezing issuance to drain legacy state
    would guarantee nothing could ever be drained, and the freeze's stated
    purpose ("leaving reads and idempotent replays open, so a retirement
    drain can finish") would be unreachable.

    The exemption is keyed, not blanket: while frozen, the legacy ref must
    already be authoritative here. A replay of something this registry has
    never seen is new state wearing the word "migration".
    """
    if self._freeze is not None and not self.is_authoritative(
        ref.thread_id, legacy_checkpoint_id
    ):
        raise IssuanceFrozenError(
            f"checkpoint issuance is frozen ({self._freeze.reason}); the replay "
            f"of {legacy_checkpoint_id!r} on {ref.thread_id!r} cannot be "
            "registered because that legacy ref is not authoritative here — "
            "only state the registry already accounts for may be drained"
        )
    return self.register(ref, issuance=False)

IssuanceFreeze dataclass

IssuanceFreeze(reason: str, at: datetime)

Why and when new issuance was closed.

MigrationLink dataclass

MigrationLink(thread_id: str, legacy_checkpoint_id: str, checkpoint_id: str, boundary_id: str, at: datetime)

Which migrated checkpoint a legacy ref was replayed into, and from where.

The link lives in the registry rather than on the migrator because the question "has this legacy frame already been replayed?" outlives any one migrator instance. Answering it from an instance attribute means a second migrator — a second worker, a retry after a restart — replays a thread the first one already replayed, with whatever side effects the replay port committed.

ReconciliationReport dataclass

ReconciliationReport(at: datetime, grace_seconds: float, expired: tuple[CheckpointRef, ...] = (), inspected: int = 0)

What a crash-expiry pass expired, and on what grounds.