symfonic.services.conversation.migration

migration

Safe-boundary replay migration.

The governing clause: a legacy checkpoint may migrate only by replay from a contract-tested safe boundary; arbitrary mid-frame state is never translated.

That is enforced structurally, not by review. There is no function in this module that reads a legacy frame and writes a migrated one — the only verbs are plan and migrate, and migrate calls a replay port. A frame in the middle of a node's execution carries invariants that node established and nobody else re-validated; translating it would mean asserting those invariants on somebody else's behalf.

Migration is idempotent by outcome, not by luck: the replayed checkpoint is linked to the legacy ref in the registry, so a second call — from this migrator, from another instance, or from another worker sharing the registry — returns the first result and never replays twice. A lock keyed by the legacy ref closes the window between "no link yet" and "link recorded", which spans an awaited replay whose side effects the port has already committed — and that lock is handed out by the registry (migration_lock), not held on the migrator, because a per-instance lock is no lock at all once two migrators share one registry. The legacy ref is never removed, because a rollback has to find it there.

MigrationOutcome dataclass

MigrationOutcome(thread_id: str, checkpoint_id: str, migrated: bool, already_migrated: bool = False, boundary_id: str | None = None)

What migrating actually did.

MigrationPlan dataclass

MigrationPlan(strategy: str, ref: CheckpointRef, boundary: SafeBoundaryMarker | None, decision: HorizonDecision)

What migrating this artifact would involve, before anything runs.

ReplayPort

Bases: Protocol

Re-executes a thread from a safe boundary and returns the new id.

A port because replay belongs to whoever owns the graph, not to the registry. This package decides whether and from where; it never decides how.

replay async

replay(thread_id: str, boundary_id: str) -> str

Replay thread_id from boundary_id; return the new id.

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

async def replay(self, thread_id: str, boundary_id: str) -> str:
    """Replay ``thread_id`` from ``boundary_id``; return the new id."""

SafeBoundaryMigrator

SafeBoundaryMigrator(*, registry: CheckpointRegistry, horizon: Any, clock: object | None = None)

Plans and performs replay migrations under a resumability horizon.

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

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

horizon property

horizon: Any

The horizon this migrator decides under.

Readable because a caller that requires a particular horizon kind (library mode requires a package-version horizon) must be able to check the wiring when it is built, not discover it mid-replay.

migrate async

migrate(ref: CheckpointRef, *, replay: ReplayPort, **context: Any) -> MigrationOutcome

Replay from the boundary and register the result as migrated.

Everything from the "already migrated?" question to recording the link happens under one lock, because the question is only answered correctly while nobody else can be mid-replay of the same legacy ref.

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

async def migrate(
    self, ref: CheckpointRef, *, replay: ReplayPort, **context: Any
) -> MigrationOutcome:
    """Replay from the boundary and register the result as migrated.

    Everything from the "already migrated?" question to recording the link
    happens under one lock, because the question is only answered correctly
    while nobody else can be mid-replay of the same legacy ref.
    """
    async with self._lock_for(ref):
        return await self._migrate_locked(ref, replay=replay, **context)

plan

plan(ref: CheckpointRef, **context: Any) -> MigrationPlan

Decide the strategy without touching the thread.

Order is load-bearing: the horizon is consulted before the boundary lookup, so an expired artifact reports expiry rather than reporting that its (irrelevant) boundary is missing.

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

def plan(self, ref: CheckpointRef, **context: Any) -> MigrationPlan:
    """Decide the strategy without touching the thread.

    Order is load-bearing: the horizon is consulted *before* the boundary
    lookup, so an expired artifact reports expiry rather than reporting
    that its (irrelevant) boundary is missing.
    """
    decision = self._horizon.decide(ref, **context)
    if decision.verdict == "expired":
        raise ResumabilityHorizonExpiredError(
            f"checkpoint {ref.checkpoint_id!r} on thread {ref.thread_id!r} "
            f"cannot be migrated: {decision.reason}.",
            support_route=decision.support_route,
        )
    if decision.verdict == "resumable":
        return MigrationPlan(
            strategy="no_migration_needed",
            ref=ref,
            boundary=None,
            decision=decision,
        )
    boundary = self._registry.latest_safe_boundary(ref.thread_id)
    if boundary is None:
        raise UnsafeBoundaryError(
            f"thread {ref.thread_id!r} has no contract-tested safe boundary; "
            "arbitrary mid-frame state is never translated"
        )
    return MigrationPlan(
        strategy="replay_from_safe_boundary",
        ref=ref,
        boundary=boundary,
        decision=decision,
    )