symfonic.services.conversation.refs

refs

Checkpoint-shaped value types: safe boundaries and checkpoint refs.

Split out of :mod:symfonic.services.conversation.values so each module holds one vocabulary — session/transcript values there, the checkpoint registry's vocabulary here — and re-exported from values so every existing import site keeps working.

Both types encode a governance rule rather than a data shape: :class:SafeBoundaryMarker derives its id, so marking one boundary twice yields one marker; :class:CheckpointRef records expiry instead of deleting, and refuses to attribute a checkpoint whose thread key cannot name its tenant.

CheckpointRef dataclass

CheckpointRef(thread_id: str, checkpoint_id: str, writer_line: WriterLine, format_version: int, created_at: datetime, package_version: str | None = None, safe_boundary: bool = False, finalized: bool = True, expired: bool = False, expiry_reason: str | None = None, tenant_id: str | None = None)

One checkpoint, as the authoritative registry knows it.

writer_line and format_version are what make a rollback decidable; expired plus expiry_reason are what make an expiry explicit rather than an absence.

attribution_is_certain property

attribution_is_certain: bool

Whether this ref can name its tenant without guessing.

Two ways to be certain: the tenant was recorded on the ref (the migrated writers do this), or the thread key carries exactly the two separators the derivation introduces, so its first segment is provably the whole tenant id.

owning_tenant property

owning_tenant: str

The tenant this checkpoint belongs to. Refuses to guess.

:class:SessionIdentity refuses to mint a key from a separator- bearing tenant, but that guard never applied to a key read back out of a backend — the legacy engine derived keys without validating either id. So for a key with more separators than the derivation introduces (acme:eu:_:s1), the first segment may be a prefix of the tenant rather than the tenant, and this raises instead of answering.

That matters because this value addresses expiry notices (TenantNotificationPolicy): answering 'acme' here would deliver tenant acme:eu's thread id and checkpoint id to a different tenant. A recorded tenant_id is the authoritative escape hatch and is used whenever present.

owning_tenant_or_none property

owning_tenant_or_none: str | None

:attr:owning_tenant, or None where it would refuse.

For callers that must partition a mixed set — notify what can be addressed, quarantine the rest — rather than abort the whole batch.

expire

expire(reason: str) -> CheckpointRef

Return an expired copy. Expiry is recorded, never a deletion.

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

def expire(self, reason: str) -> CheckpointRef:
    """Return an expired copy. Expiry is recorded, never a deletion."""
    return replace(self, expired=True, expiry_reason=reason)

SafeBoundaryMarker dataclass

SafeBoundaryMarker(thread_id: str, boundary_id: str, sequence: int, writer_line: WriterLine, created_at: datetime)

A contract-tested point a thread may be replayed from.

boundary_id is derived from the thread, sequence, and state digest, so marking the same boundary twice — from a retry, a second process, or a replayed migration — produces the same identifier and therefore one marker.

create classmethod

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

Build a marker, deriving the id unless the writer's is supplied.

boundary_id is for rehydration only: it re-adopts a boundary under the id the writer recorded, rather than re-deriving one from inputs (sequence, digest) that a restarted process cannot recover.

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

@classmethod
def create(
    cls,
    *,
    thread_id: str,
    sequence: int,
    digest: str,
    writer_line: WriterLine,
    created_at: datetime,
    boundary_id: str | None = None,
) -> SafeBoundaryMarker:
    """Build a marker, deriving the id unless the writer's is supplied.

    ``boundary_id`` is for rehydration only: it re-adopts a boundary under
    the id the writer recorded, rather than re-deriving one from inputs
    (sequence, digest) that a restarted process cannot recover.
    """
    return cls(
        thread_id=thread_id,
        boundary_id=boundary_id or cls.derive_id(thread_id, sequence, digest),
        sequence=sequence,
        writer_line=writer_line,
        created_at=created_at,
    )