symfonic.services.conversation.compat

compat

Bidirectional checkpoint / session / transcript format compatibility.

The migration-window assumption, in both directions: state written by the legacy path is readable and resumable by the migrated services, and state written by the migrated services is readable and resumable by the legacy path. Wave rollback depends on the second half as much as the first.

The mechanism is deliberately the least clever one available: an additive envelope under a single reserved key. A legacy reader that ignores keys it does not know sees byte-identical metadata; a migrated reader that finds no envelope treats the state as legacy. Nothing is rewritten, nothing is moved, and there is no version of this file that mutates a legacy field.

Session rows use the same discipline at a different level: the migrated row renders exactly the legacy five keys, and any key the legacy path added rides along in SessionRecord.extra so a rollback loses nothing.

CompatibilityReport dataclass

CompatibilityReport(direction: str, writer_line: WriterLine, format_version: int, readable_by_legacy: bool, readable_by_migrated: bool, reason: str)

Which direction a piece of state can cross, and why.

Carries no state content — only the verdict — so it is safe to log next to a tenant identifier.

StateEnvelope dataclass

StateEnvelope(format_version: int, writer_line: WriterLine, package_version: str | None = None, safe_boundary: bool = False, boundary_id: str | None = None, boundary_sequence: int | None = None, finalized: bool | None = None)

The provenance block attached to migrated-written state.

to_metadata_value

to_metadata_value() -> dict[str, Any]

Render as JSON-native scalars only.

Checkpoint metadata crosses a serializer this package does not own, so anything richer than a scalar is a portability bet on somebody else's codec.

boundary_id and boundary_sequence are written here, with the state, because they are the only place a boundary's identity survives a restart. The id is a hash of (thread, sequence, digest); a process that comes up against a bare listing has none of those, so a boundary whose id is not persisted alongside its state is a boundary the next process can only guess at — and a guessed boundary id is one no writer ever marked.

finalized is here for the same reason and answers the same class of question: a crash that ends the process takes the registry's in-memory finalization state with it, so a restart that could not read finalization back would have to assume every durable row was a completed write — and crash-expiry could then only ever see a crash that left the process alive.

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

def to_metadata_value(self) -> dict[str, Any]:
    """Render as JSON-native scalars only.

    Checkpoint metadata crosses a serializer this package does not own, so
    anything richer than a scalar is a portability bet on somebody else's
    codec.

    ``boundary_id`` and ``boundary_sequence`` are written *here*, with the
    state, because they are the only place a boundary's identity survives a
    restart. The id is a hash of (thread, sequence, digest); a process that
    comes up against a bare listing has none of those, so a boundary whose
    id is not persisted alongside its state is a boundary the next process
    can only guess at — and a guessed boundary id is one no writer ever
    marked.

    ``finalized`` is here for the same reason and answers the same class of
    question: a crash that ends the *process* takes the registry's
    in-memory finalization state with it, so a restart that could not read
    finalization back would have to assume every durable row was a
    completed write — and crash-expiry could then only ever see a crash
    that left the process alive.
    """
    return {
        "format_version": self.format_version,
        "writer_line": self.writer_line,
        "package_version": self.package_version,
        "safe_boundary": self.safe_boundary,
        "boundary_id": self.boundary_id,
        "boundary_sequence": self.boundary_sequence,
        "finalized": self.finalized,
    }

decode_envelope

decode_envelope(metadata: Mapping[str, Any] | None) -> StateEnvelope

Read the envelope, or state positively that this is legacy state.

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

def decode_envelope(metadata: Mapping[str, Any] | None) -> StateEnvelope:
    """Read the envelope, or state positively that this is legacy state."""
    if not metadata or RESERVED_METADATA_KEY not in metadata:
        return StateEnvelope(
            format_version=LEGACY_FORMAT_VERSION, writer_line="legacy"
        )
    body = metadata[RESERVED_METADATA_KEY]
    if not isinstance(body, Mapping):
        raise CheckpointFormatError(
            f"{RESERVED_METADATA_KEY!r} is not a mapping; the envelope is corrupt"
        )
    version = body.get("format_version")
    if not isinstance(version, int) or version < 1:
        raise CheckpointFormatError(
            f"envelope carries no usable format_version ({version!r})"
        )
    if version > CURRENT_FORMAT_VERSION:
        raise CheckpointFormatError(
            f"envelope format_version {version} is newer than this package "
            f"understands ({CURRENT_FORMAT_VERSION}); refusing to resume"
        )
    writer_line = body.get("writer_line")
    if writer_line not in _WRITER_LINES:
        raise CheckpointFormatError(
            f"envelope writer_line {writer_line!r} is not one of {sorted(_WRITER_LINES)}"
        )
    boundary_sequence = body.get("boundary_sequence")
    finalized = body.get("finalized")
    return StateEnvelope(
        format_version=version,
        writer_line=writer_line,  # type: ignore[arg-type]
        package_version=body.get("package_version"),
        safe_boundary=bool(body.get("safe_boundary", False)),
        boundary_id=body.get("boundary_id"),
        # An envelope written before this field existed simply omits it; a
        # non-integer is treated as absent rather than coerced, because a
        # sequence guessed from a malformed value orders boundaries wrongly.
        boundary_sequence=boundary_sequence if isinstance(boundary_sequence, int) else None,
        # Same discipline: anything that is not a bool is "the writer did not
        # say", never a coerced truthiness. ``bool("false")`` is ``True``, and
        # guessing that direction resumes a half-written frame.
        finalized=finalized if isinstance(finalized, bool) else None,
    )

describe_compatibility

describe_compatibility(metadata: Mapping[str, Any] | None) -> CompatibilityReport

Answer both directions for one piece of persisted state.

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

def describe_compatibility(metadata: Mapping[str, Any] | None) -> CompatibilityReport:
    """Answer both directions for one piece of persisted state."""
    envelope = decode_envelope(metadata)
    if envelope.written_by_legacy:
        return CompatibilityReport(
            direction="legacy->migrated",
            writer_line="legacy",
            format_version=envelope.format_version,
            readable_by_legacy=True,
            readable_by_migrated=True,
            reason="no envelope; migrated readers treat absent provenance as legacy",
        )
    return CompatibilityReport(
        direction="migrated->legacy",
        writer_line="migrated",
        format_version=envelope.format_version,
        readable_by_legacy=True,
        readable_by_migrated=True,
        reason=(
            f"envelope is additive under {RESERVED_METADATA_KEY!r}; every legacy "
            "field is unchanged"
        ),
    )

encode_envelope

encode_envelope(*, package_version: str, safe_boundary: bool = False, boundary_id: str | None = None, boundary_sequence: int | None = None, finalized: bool = True) -> StateEnvelope

Build the envelope for state this package is about to write.

A safe_boundary=True write should carry the boundary_id and boundary_sequence that CheckpointService.record_write derived (readable back as registry.latest_safe_boundary(thread_id)); without them the boundary is recorded in this process only and does not survive a restart.

finalized=False opens the crash window durably: stamp it on the envelope that goes out with a write registered as unfinalized, and stamp a finalized=True envelope when the write is closed. A row that is still False when a later process adopts it is a write interrupted by a crash, and the grace window — not an assumption — decides its fate.

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

def encode_envelope(
    *,
    package_version: str,
    safe_boundary: bool = False,
    boundary_id: str | None = None,
    boundary_sequence: int | None = None,
    finalized: bool = True,
) -> StateEnvelope:
    """Build the envelope for state this package is about to write.

    A ``safe_boundary=True`` write should carry the ``boundary_id`` and
    ``boundary_sequence`` that ``CheckpointService.record_write`` derived
    (readable back as ``registry.latest_safe_boundary(thread_id)``); without
    them the boundary is recorded in this process only and does not survive a
    restart.

    ``finalized=False`` opens the crash window durably: stamp it on the
    envelope that goes out with a write registered as unfinalized, and stamp a
    ``finalized=True`` envelope when the write is closed. A row that is still
    ``False`` when a later process adopts it is a write interrupted by a crash,
    and the grace window — not an assumption — decides its fate.
    """
    return StateEnvelope(
        format_version=CURRENT_FORMAT_VERSION,
        writer_line="migrated",
        package_version=package_version,
        safe_boundary=safe_boundary,
        boundary_id=boundary_id,
        boundary_sequence=boundary_sequence,
        finalized=finalized,
    )

merge_metadata

merge_metadata(metadata: Mapping[str, Any] | None, envelope: StateEnvelope) -> dict[str, Any]

Attach the envelope additively, refusing to shadow anything.

Two refusals, and they are the same contract read from both ends: if something already occupies the reserved key, this package does not know whose it is, and overwriting it would corrupt state belonging to a writer nobody has identified; and if the key holds an envelope from a newer format than this package writes, stamping the older version over it would destroy provenance a newer node depends on. :func:decode_envelope already refuses to read future-format state — writing over it would make the refusal cosmetic, and a downgrade is never silent here.

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

def merge_metadata(
    metadata: Mapping[str, Any] | None, envelope: StateEnvelope
) -> dict[str, Any]:
    """Attach the envelope additively, refusing to shadow anything.

    Two refusals, and they are the same contract read from both ends: if
    something already occupies the reserved key, this package does not know
    whose it is, and overwriting it would corrupt state belonging to a writer
    nobody has identified; and if the key holds an envelope from a *newer*
    format than this package writes, stamping the older version over it would
    destroy provenance a newer node depends on. :func:`decode_envelope` already
    refuses to *read* future-format state — writing over it would make the
    refusal cosmetic, and a downgrade is never silent here.
    """
    merged = dict(metadata or {})
    existing = merged.get(RESERVED_METADATA_KEY)
    if existing is not None:
        if not _is_envelope_body(existing):
            raise CheckpointFormatError(
                f"metadata key {RESERVED_METADATA_KEY!r} is already occupied by a "
                "value this package did not write; refusing to overwrite it"
            )
        existing_version = existing.get("format_version")
        if isinstance(existing_version, int) and existing_version > CURRENT_FORMAT_VERSION:
            raise CheckpointFormatError(
                f"existing envelope format_version {existing_version} is newer than "
                f"this package writes ({CURRENT_FORMAT_VERSION}); refusing to "
                "downgrade another writer's provenance"
            )
    merged[RESERVED_METADATA_KEY] = envelope.to_metadata_value()
    return merged