symfonic.services.conversation

conversation

Conversation, session, transcript, and checkpoint services (T3.4.1).

History strategies, session identity, transcript persistence and query, checkpointer readiness, restart recovery, and state overrides — separated from agent orchestration, each behind its own port.

Two contracts run through the whole package and are the reason it is one package rather than four:

Bidirectional format compatibility. During the migration window, state written by the legacy path is readable and resumable by these services, and state written by these services is readable and resumable by the legacy path. The mechanism is an additive envelope under a single reserved metadata key (:mod:~symfonic.services.conversation.compat) plus a session row that renders exactly the legacy five keys. Wave rollback depends on both halves.

Replay-only migration under an explicit horizon. The registry (:mod:~symfonic.services.conversation.registry) is the single authority on which checkpoints exist; the horizon (:mod:~symfonic.services.conversation.horizon) says how long legacy-format state stays migratable; and the only crossing is replay from a contract-tested safe boundary (:mod:~symfonic.services.conversation.migration). Arbitrary mid-frame state is never translated, and expiry is always explicit and always carries a support route.

Library mode gets its own horizon and its own journey (:mod:~symfonic.services.conversation.library): the replay path ships in the package and survives the T4.4.6 retirement of the legacy engine, and adopter- local artifacts are keyed by package version rather than by the operated platform's calendar.

AmbiguousThreadKeyError

Bases: SessionIdentityError

A thread key's tenant segment cannot be reconstructed unambiguously.

Raised in both directions, because the ambiguity is symmetric:

• forward — a tenant_id / sub_tenant_id carrying the thread-key separator is refused, so no new key can be minted that parses back as a different tenant, and
• backward — a key read out of a backend whose tenant segment is not provably whole (more separators than the derivation puts there) is quarantined for attribution: it may still be parsed positionally and resumed, but it may not be used to name a tenant.

The migrated path never re-attributes such state. Pre-existing state written by the legacy path under a separator-bearing tenant stays readable at its key and stays quarantined for tenant attribution until an operator records the owning tenant explicitly (CheckpointRef.tenant_id), which is the one attribution this package treats as authoritative.

CalendarHorizon dataclass

CalendarHorizon(cutoff: datetime, notice_period: timedelta, support_route: str)

Operated-platform horizon: a published cutoff and a notice window.

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."""

CheckpointFormatError

Bases: ConversationServiceError

Persisted state does not match a format this package can vouch for.

Covers an envelope from a future format version, a corrupt envelope, and a reserved key already occupied by something else. Every one of those is a refusal, never a silent downgrade to "assume legacy".

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)

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)

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

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.

ConversationCapability

ConversationCapability(*, history: HistoryStrategy | None = None, clock: object | None = None)

Resolves a session identity and a history policy into one plan.

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

def __init__(
    self, *, history: HistoryStrategy | None = None, clock: object | None = None
) -> None:
    self._history = history
    self._clock = clock if clock is not None else _SystemClock()

plan_for

plan_for(identity: SessionIdentity) -> ConversationPlan

Deterministic for a given identity: no clock, no counters, no ids.

Reproducibility is the property that lets a parity harness compare the legacy and migrated paths turn for turn.

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

def plan_for(self, identity: SessionIdentity) -> ConversationPlan:
    """Deterministic for a given identity: no clock, no counters, no ids.

    Reproducibility is the property that lets a parity harness compare the
    legacy and migrated paths turn for turn.
    """
    return ConversationPlan(
        thread_id=identity.thread_id,
        identity=identity,
        history=resolve_history(self._history),
    )

ConversationPlan dataclass

ConversationPlan(thread_id: str, identity: SessionIdentity, history: HistoryDirective)

What one session's conversation looks like before any turn runs.

to_legacy_overrides

to_legacy_overrides() -> dict[str, Any]

Config overrides plus the thread key, for the legacy adapter.

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

def to_legacy_overrides(self) -> dict[str, Any]:
    """Config overrides plus the thread key, for the legacy adapter."""
    overrides = self.history.to_legacy_overrides()
    overrides["_thread_id"] = self.thread_id
    return overrides

ConversationServiceError

Bases: SymfonicError

Root of the conversation/session/transcript/checkpoint taxonomy.

ExpiryNotice dataclass

ExpiryNotice(tenant_id: str, thread_id: str, checkpoint_id: str, deadline: datetime, support_route: str)

What one tenant is told, carrying no conversation content.

ExportReceipt dataclass

ExportReceipt(tenant_id: str, thread_id: str, checkpoint_id: str, receipt: str, at: datetime)

Proof that a tenant's expiring state was handed back before expiry.

HistoryDirective dataclass

HistoryDirective(kind: HistoryKind, summarize: bool = False, window_messages: int | None = None, trigger_chars: int | None = None, keep_recent: int | None = None)

What a strategy decided, as an inert value.

default classmethod

default() -> HistoryDirective

The framework's documented default: summarize overflow.

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

@classmethod
def default(cls) -> HistoryDirective:
    """The framework's documented default: summarize overflow."""
    return cls(
        kind="summarizing",
        summarize=True,
        trigger_chars=_DEFAULT_TRIGGER_CHARS,
        keep_recent=_DEFAULT_KEEP_RECENT,
    )

to_legacy_overrides

to_legacy_overrides() -> dict[str, Any]

Render as the legacy config fields, carrying no engine types.

Both mechanisms are always named. Leaving one unset would let whatever the adopter's config already held govern alongside this directive, which is precisely the "two limiters, unclear winner" ambiguity the strategy objects exist to remove.

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

def to_legacy_overrides(self) -> dict[str, Any]:
    """Render as the legacy config fields, carrying no engine types.

    Both mechanisms are always named. Leaving one unset would let whatever
    the adopter's config already held govern alongside this directive,
    which is precisely the "two limiters, unclear winner" ambiguity the
    strategy objects exist to remove.
    """
    return {
        "max_conversation_messages": self.window_messages or UNBOUNDED,
        "compaction_trigger_chars": self.trigger_chars or UNBOUNDED,
        "compaction_keep_recent": self.keep_recent or _DEFAULT_KEEP_RECENT,
        "compaction_enabled": self.summarize,
    }

HistoryStrategy

Bases: Protocol

A named, swappable policy for keeping a conversation in the window.

directive

directive() -> HistoryDirective

Return the inert decision. Must not touch configuration.

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

def directive(self) -> HistoryDirective:
    """Return the inert decision. Must not touch configuration."""

HorizonDecision dataclass

HorizonDecision(verdict: HorizonVerdict, reason: str, support_route: str)

The horizon's answer for one artifact.

InMemorySessionStore

InMemorySessionStore()

The default store: partitioned by tenant, with a reverse owner index.

The reverse index is what makes a cross-tenant collision detectable without scanning every tenant — the legacy manager needed the same thing and it is the reason the storage layout is two maps rather than one.

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

def __init__(self) -> None:
    self._rows: dict[str, dict[str, SessionRecord]] = {}
    self._owners: dict[str, str] = {}

IssuanceFreeze dataclass

IssuanceFreeze(reason: str, at: datetime)

Why and when new issuance was closed.

IssuanceFrozenError

Bases: ConversationServiceError

New checkpoint issuance is frozen; only reads and replays remain.

The freeze is how a retirement drain reaches a fixed point: no new state of the retiring shape can appear while the existing state is migrated out.

So the drain itself is not what this refuses. Registering the output of a safe-boundary replay, and adopting rows read back out of the backend after a restart, both stay open while frozen — a freeze that closed them would guarantee that nothing could ever be drained. Raised while frozen only for genuinely new state, including a "migration" of a legacy ref the registry never accounted for, which is new state under another name.

LibraryModeUpgrade

LibraryModeUpgrade(*, registry: CheckpointRegistry, migrator: SafeBoundaryMigrator, horizon: PackageVersionHorizon, package_version: str | None = None)

Guides an adopter across the legacy-engine retirement boundary.

Holds no state of its own: the registry is the authority, the migrator performs the replay, and the horizon decides support. This class is the adopter-facing sequencing of those three.

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

def __init__(
    self,
    *,
    registry: CheckpointRegistry,
    migrator: SafeBoundaryMigrator,
    horizon: PackageVersionHorizon,
    package_version: str | None = None,
) -> None:
    if not isinstance(horizon, PackageVersionHorizon):
        raise ValueError(
            "library mode requires a PackageVersionHorizon: adopter-local "
            "artifacts are keyed by package version, not by the operated "
            "platform's calendar"
        )
    # The migrator's horizon is the one ``upgrade()`` actually decides
    # under, so checking only this class's own horizon would let a
    # mismatched wiring pass construction, report guidance happily from the
    # validated horizon, and then fail inside the replay with a raw
    # TypeError (a calendar horizon's ``decide`` needs ``now``, which
    # ``upgrade`` does not pass). Refuse it where it is wired instead.
    if not isinstance(migrator.horizon, PackageVersionHorizon):
        raise ValueError(
            "the migrator wired into library mode decides under "
            f"{type(migrator.horizon).__name__}, not a PackageVersionHorizon; "
            "guidance and upgrade would answer under different horizons"
        )
    self._registry = registry
    self._migrator = migrator
    self._horizon = horizon
    self.package_version = package_version or installed_package_version()

guidance

guidance(ref: CheckpointRef) -> UpgradeGuidance

Report, never raise. Guidance is what an adopter reads first.

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

def guidance(self, ref: CheckpointRef) -> UpgradeGuidance:
    """Report, never raise. Guidance is what an adopter reads first."""
    decision = self._horizon.decide(ref, package_version=self.package_version)
    if decision.verdict == "expired":
        return self._guidance(
            "expired",
            ref,
            decision.reason,
            next_step=(
                "Downgrade to a supported package version to migrate this "
                "state, or discard it."
            ),
        )
    if decision.verdict == "resumable":
        return self._guidance(
            "resumable", ref, decision.reason, next_step="Nothing to do."
        )
    boundary = self._registry.latest_safe_boundary(ref.thread_id)
    if boundary is None:
        return self._guidance(
            "unsafe",
            ref,
            "no contract-tested safe boundary exists for this thread, and "
            "arbitrary mid-frame state is never translated",
            next_step="Start a new thread; this state cannot be replayed.",
        )
    return self._guidance(
        "migration_required",
        ref,
        decision.reason,
        boundary=boundary,
        next_step=(
            f"Replay thread {ref.thread_id} from safe boundary "
            f"{boundary.boundary_id} (sequence {boundary.sequence})."
        ),
    )

require_resumable

require_resumable(ref: CheckpointRef) -> CheckpointRef

Assert the artifact can be used as-is, or raise the reason it cannot.

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

def require_resumable(self, ref: CheckpointRef) -> CheckpointRef:
    """Assert the artifact can be used as-is, or raise the reason it cannot."""
    guidance = self.guidance(ref)
    if guidance.verdict == "resumable":
        return ref
    if guidance.verdict == "expired":
        raise ResumabilityHorizonExpiredError(
            f"{ref.checkpoint_id!r} is beyond the resumability horizon: "
            f"{guidance.reason}.",
            support_route=guidance.support_route,
        )
    raise UnsafeBoundaryError(
        f"{ref.checkpoint_id!r} is not resumable as-is: {guidance.reason}"
    )

upgrade async

upgrade(ref: CheckpointRef, *, replay: ReplayPort) -> MigrationOutcome

Perform the guided migration. Raises where guidance reported a stop.

The two refusals are distinct exception types on purpose: :class:UnsafeBoundaryError is permanent for that artifact, while :class:ResumabilityHorizonExpiredError is a statement about the support window and names the route out of it.

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

async def upgrade(
    self, ref: CheckpointRef, *, replay: ReplayPort
) -> MigrationOutcome:
    """Perform the guided migration. Raises where guidance reported a stop.

    The two refusals are distinct exception types on purpose:
    :class:`UnsafeBoundaryError` is permanent for that artifact, while
    :class:`ResumabilityHorizonExpiredError` is a statement about the
    support window and names the route out of it.
    """
    return await self._migrator.migrate(
        ref, replay=replay, package_version=self.package_version
    )

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.

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.

NullHistoryStrategy dataclass

NullHistoryStrategy()

No management: history grows to the model's own limit.

PackageVersionHorizon dataclass

PackageVersionHorizon(supported_through: str, support_route: str)

Library-mode horizon: keyed by package version, never by the calendar.

ReconciliationReport dataclass

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

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

RecoveryDecision dataclass

RecoveryDecision(action: RecoveryAction, thread_id: str, reason: str, checkpoint_id: str | None = None, safe_boundary: SafeBoundaryMarker | None = None)

What to do with a thread's durable state after a restart.

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."""

RestartRecoveryService

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

Decides resume / migrate / quarantine / fresh for a restarted thread.

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

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

adopt async

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

Rehydrate the registry from a thread's durable state.

Provenance is read, never assumed: each row's envelope says which line wrote it, and a row with no envelope is legacy at format version 0 — the positive statement :func:decode_envelope makes, not a guess. A row whose envelope is corrupt or from a newer format is skipped, so it stays unaccounted for and quarantines rather than resuming under an invented provenance. Already-registered rows are left exactly as they are: adoption never overwrites the registry's own record.

Safe boundaries are adopted only under the id the writer recorded in the envelope. A row that claims to be a boundary but names none leaves nothing to replay from, and inventing an id from its listing position would be worse than having none — the migrator would hand a replay port a boundary that never existed.

Returns the refs this call adopted, so a caller can log what a restart took ownership of.

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

async def adopt(self, thread_id: str) -> tuple[CheckpointRef, ...]:
    """Rehydrate the registry from a thread's durable state.

    Provenance is *read*, never assumed: each row's envelope says which
    line wrote it, and a row with no envelope is legacy at format version
    0 — the positive statement :func:`decode_envelope` makes, not a guess.
    A row whose envelope is corrupt or from a newer format is skipped, so
    it stays unaccounted for and quarantines rather than resuming under an
    invented provenance. Already-registered rows are left exactly as they
    are: adoption never overwrites the registry's own record.

    Safe boundaries are adopted only under the id the writer recorded in
    the envelope. A row that claims to be a boundary but names none leaves
    nothing to replay *from*, and inventing an id from its listing position
    would be worse than having none — the migrator would hand a replay port
    a boundary that never existed.

    Returns the refs this call adopted, so a caller can log what a restart
    took ownership of.
    """
    if self._adapter is None:
        return ()
    rows = await self._adapter.list_checkpoints(thread_id)
    adopted: list[CheckpointRef] = []
    for sequence, row in enumerate(rows or ()):
        checkpoint_id, metadata = row
        if self._registry.get(thread_id, checkpoint_id) is not None:
            continue
        try:
            envelope = decode_envelope(metadata)
        except CheckpointFormatError:
            continue
        finalized = _adopted_finalization(envelope)
        ref = self._registry.register(
            CheckpointRef(
                thread_id=thread_id,
                checkpoint_id=checkpoint_id,
                writer_line=envelope.writer_line,
                format_version=envelope.format_version,
                created_at=self._clock.now(),
                package_version=envelope.package_version,
                safe_boundary=envelope.safe_boundary,
                finalized=finalized,
            ),
            finalized=finalized,
            # Adoption is not issuance: every row here was read *out of*
            # the backend and already exists durably. Refusing it during a
            # freeze would not prevent any state from existing — it would
            # only leave the thread unaccounted for, so a restart mid-drain
            # would quarantine every thread and none could be migrated out,
            # ending the drain the freeze was declared for.
            issuance=False,
        )
        adopted.append(ref)
        if envelope.safe_boundary and envelope.boundary_id is not None:
            # Use the id the *writer* recorded, never a re-derived one.
            # ``derive_id`` hashes (thread, sequence, digest), and after a
            # restart neither the sequence nor the digest the writer used
            # is recoverable from a listing — so re-deriving produced a
            # second id for one physical boundary and handed ReplayPort a
            # boundary the writer never marked.
            self._registry.mark_safe_boundary(
                thread_id,
                sequence=(
                    envelope.boundary_sequence
                    if envelope.boundary_sequence is not None
                    else sequence
                ),
                digest=checkpoint_id,
                writer_line=envelope.writer_line,
                boundary_id=envelope.boundary_id,
            )
    return tuple(adopted)

ResumabilityHorizon

Bases: Protocol

The shape both horizons share, so callers bind to neither.

decide

decide(ref: CheckpointRef, **context: object) -> HorizonDecision

Answer resumable / migration_required / expired for one artifact.

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

def decide(self, ref: CheckpointRef, **context: object) -> HorizonDecision:
    """Answer resumable / migration_required / expired for one artifact."""

ResumabilityHorizonExpiredError

ResumabilityHorizonExpiredError(message: str, *, support_route: str)

Bases: ConversationServiceError

The artifact is past its published support horizon.

Always carries the support route in its message: an expiry an adopter cannot act on is indistinguishable from a bug.

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

def __init__(self, message: str, *, support_route: str) -> None:
    super().__init__(f"{message} Support route: {support_route}")
    self.support_route = support_route

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,
    )

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,
    )

SessionIdentity dataclass

SessionIdentity(tenant_id: str, session_id: str, sub_tenant_id: str | None = None)

Tenant + sub-tenant + session, and the thread key they derive.

thread_id property

thread_id: str

The legacy derivation, character for character.

as_configurable

as_configurable(*, checkpoint_id: str | None = None) -> dict[str, Any]

The graph-runner config shape. checkpoint_id only when resuming.

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

def as_configurable(self, *, checkpoint_id: str | None = None) -> dict[str, Any]:
    """The graph-runner config shape. ``checkpoint_id`` only when resuming."""
    configurable: dict[str, Any] = {"thread_id": self.thread_id}
    if checkpoint_id is not None:
        configurable["checkpoint_id"] = checkpoint_id
    return {"configurable": configurable}

SessionIdentityError

Bases: ConversationServiceError

A session identity could not be derived, parsed, or trusted.

Raised rather than defaulted: a guessed tenant is a cross-tenant read, and a guessed thread id silently forks one conversation into two.

SessionIdentityService

Derives and parses session identities. No state, by design.

for_scope

for_scope(scope: Any, session_id: str) -> SessionIdentity

Derive from an authenticated scope object, read by attribute.

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

def for_scope(self, scope: Any, session_id: str) -> SessionIdentity:
    """Derive from an authenticated scope object, read by attribute."""
    tenant_id = getattr(scope, "tenant_id", None)
    if not tenant_id:
        raise SessionIdentityError("scope carries no tenant_id")
    return self.identify(
        tenant_id,
        session_id,
        sub_tenant_id=getattr(scope, "sub_tenant_id", None),
    )

from_thread_id

from_thread_id(thread_id: str, *, tenant_id: str | None = None) -> SessionIdentity

Parse a thread key back into its identity.

split(":", 2) on purpose: a session id may legitimately contain colons (adopters use URLs and composite keys), and only the first two separators are structural. Splitting greedily would corrupt exactly the ids an adopter cannot change.

This is a positional parse, not a tenant attribution. For a key the legacy path wrote under a separator-bearing tenant (which it never validated), the leading segment is a prefix of the tenant rather than the tenant, and no parse can tell that key apart from an exempt colon-bearing session id. Reading and resuming such a thread is unaffected — its key is unchanged — but naming its tenant is refused: see :attr:CheckpointRef.owning_tenant and :func:~symfonic.services.conversation.values.tenant_segment_is_provable.

Pass tenant_id when the caller already knows the tenant (from an authenticated scope, or from a ref that recorded it). The prefix is then verified rather than inferred, which is the one way a key with extra separators can be attributed. A separator-bearing tenant_id is still refused: such state is quarantined, never re-attributed.

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

def from_thread_id(
    self, thread_id: str, *, tenant_id: str | None = None
) -> SessionIdentity:
    """Parse a thread key back into its identity.

    ``split(":", 2)`` on purpose: a session id may legitimately contain
    colons (adopters use URLs and composite keys), and only the first two
    separators are structural. Splitting greedily would corrupt exactly
    the ids an adopter cannot change.

    This is a *positional* parse, not a tenant attribution. For a key the
    legacy path wrote under a separator-bearing tenant (which it never
    validated), the leading segment is a prefix of the tenant rather than
    the tenant, and no parse can tell that key apart from an exempt
    colon-bearing session id. Reading and resuming such a thread is
    unaffected — its key is unchanged — but naming its tenant is refused:
    see :attr:`CheckpointRef.owning_tenant` and
    :func:`~symfonic.services.conversation.values.tenant_segment_is_provable`.

    Pass ``tenant_id`` when the caller already knows the tenant (from an
    authenticated scope, or from a ref that recorded it). The prefix is
    then *verified* rather than inferred, which is the one way a key with
    extra separators can be attributed. A separator-bearing ``tenant_id``
    is still refused: such state is quarantined, never re-attributed.
    """
    parts = thread_id.split(":", 2)
    if len(parts) != 3:
        raise SessionIdentityError(
            f"thread id {thread_id!r} is not tenant:sub:session"
        )
    if tenant_id is not None:
        return self._verified(thread_id, tenant_id)
    tenant_id_part, sub, session_id = parts
    return SessionIdentity(
        tenant_id=tenant_id_part,
        session_id=session_id,
        sub_tenant_id=None if sub == "_" else sub,
    )

SessionRecord dataclass

SessionRecord(session_id: str, tenant_id: str, created_at: datetime, last_active: datetime, message_count: int = 0, extra: tuple[tuple[str, Any], ...] = ())

One session row, in the migrated shape, with a legacy projection.

extra carries any key the legacy path wrote that this package does not model. Dropping it would make a rollback lossy, which is the one thing the bidirectional assumption forbids.

to_legacy_dict

to_legacy_dict() -> dict[str, Any]

Exactly the keys the legacy SessionManager wrote, same types.

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

def to_legacy_dict(self) -> dict[str, Any]:
    """Exactly the keys the legacy ``SessionManager`` wrote, same types."""
    row: dict[str, Any] = {
        "session_id": self.session_id,
        "tenant_id": self.tenant_id,
        "created_at": self.created_at.isoformat(),
        "last_active": self.last_active.isoformat(),
        "message_count": self.message_count,
    }
    row.update(dict(self.extra))
    return row

SessionService

SessionService(*, store: SessionStorePort | None = None, clock: object | None = None, max_per_tenant: int = MAX_SESSIONS_PER_TENANT)

Creates, finds, and ages session rows for one deployment.

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

def __init__(
    self,
    *,
    store: SessionStorePort | None = None,
    clock: object | None = None,
    max_per_tenant: int = MAX_SESSIONS_PER_TENANT,
) -> None:
    self._store = store if store is not None else InMemorySessionStore()
    self._clock = clock if clock is not None else _SystemClock()
    self._max_per_tenant = max_per_tenant

ensure

ensure(tenant_id: str, session_id: str | None) -> str

Return the caller's session, or issue one.

A session id already owned by a different tenant is never joined: the asking tenant gets a fresh id and the owner's row is untouched. Guessing another tenant's id must not be a way into their session.

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

def ensure(self, tenant_id: str, session_id: str | None) -> str:
    """Return the caller's session, or issue one.

    A session id already owned by a *different* tenant is never joined:
    the asking tenant gets a fresh id and the owner's row is untouched.
    Guessing another tenant's id must not be a way into their session.
    """
    if session_id is None:
        return self.create(tenant_id)
    owner = self._store.owner_of(session_id)
    if owner == tenant_id:
        return session_id
    if owner is None:
        self._register(tenant_id, session_id)
        return session_id
    logger.warning(
        "session id collision refused (requested tenant=%s, owner=%s)",
        tenant_id,
        owner,
    )
    return self.create(tenant_id)

list

list(tenant_id: str) -> tuple[SessionRecord, ...]

Newest activity first, matching the legacy listing order.

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

def list(self, tenant_id: str) -> tuple[SessionRecord, ...]:
    """Newest activity first, matching the legacy listing order."""
    return tuple(
        sorted(
            self._store.list(tenant_id),
            key=lambda record: record.last_active,
            reverse=True,
        )
    )

touch

touch(tenant_id: str, session_id: str) -> None

Advance activity and count a message. A foreign id is a no-op.

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

def touch(self, tenant_id: str, session_id: str) -> None:
    """Advance activity and count a message. A foreign id is a no-op."""
    record = self._store.get(tenant_id, session_id)
    if record is None:
        return
    self._store.put(record.touched(self._clock.now()))

SessionStorePort

Bases: Protocol

Where session rows live. One tenant-scoped map, four verbs.

owner_of

owner_of(session_id: str) -> str | None

Which tenant holds this id, for cross-tenant collision refusal.

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

def owner_of(self, session_id: str) -> str | None:
    """Which tenant holds this id, for cross-tenant collision refusal."""

SlidingWindowHistoryStrategy dataclass

SlidingWindowHistoryStrategy(window_size: int)

Keep the most recent window_size messages; drop older ones.

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,
    }

StateOverrides

Splits run-config keys out of graph-state overrides.

split classmethod

split(overrides: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]

Return (configurable, graph_state) without mutating the input.

A None for a configurable key is refused rather than dropped: the caller meant to pass a thread id and computed nothing, and silently continuing starts a brand-new thread under the same session.

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

@classmethod
def split(cls, overrides: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    """Return ``(configurable, graph_state)`` without mutating the input.

    A ``None`` for a configurable key is refused rather than dropped: the
    caller meant to pass a thread id and computed nothing, and silently
    continuing starts a brand-new thread under the same session.
    """
    configurable: dict[str, Any] = {}
    state: dict[str, Any] = {}
    for key, value in overrides.items():
        target = cls.CONFIGURABLE_KEYS.get(key)
        if target is None:
            state[key] = value
            continue
        if value is None:
            raise ConversationServiceError(
                f"state override {key!r} is None; refusing to start an "
                "unrelated thread in place of the one you meant"
            )
        configurable[target] = value
    return configurable, state

SummarizingHistoryStrategy dataclass

SummarizingHistoryStrategy(trigger_chars: int = _DEFAULT_TRIGGER_CHARS, keep_recent: int = _DEFAULT_KEEP_RECENT)

Summarize overflow into a running summary, keeping recent turns raw.

TenantNotificationPolicy

TenantNotificationPolicy(*, registry: CheckpointRegistry, horizon: CalendarHorizon, export: Callable[[CheckpointRef], str] | None = None)

Decides who is notified, and what may still be exported.

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

def __init__(
    self,
    *,
    registry: CheckpointRegistry,
    horizon: CalendarHorizon,
    export: Callable[[CheckpointRef], str] | None = None,
) -> None:
    self._registry = registry
    self._horizon = horizon
    self._export = export

expiring

expiring(*, now: datetime) -> tuple[CheckpointRef, ...]

Refs the horizon would refuse once the cutoff passes.

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

def expiring(self, *, now: datetime) -> tuple[CheckpointRef, ...]:
    """Refs the horizon would refuse once the cutoff passes."""
    return tuple(
        ref
        for ref in self._registry.all_refs()
        if self._horizon.decide(ref, now=now).verdict == "migration_required"
    )

export_expiring

export_expiring(*, now: datetime) -> tuple[ExportReceipt, ...]

Export everything still inside the window. Nothing after it.

Same attribution rule as :meth:notices, for a stronger reason: a receipt carries the state itself, so handing one to a guessed tenant is a cross-tenant data release rather than a mis-addressed warning.

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

def export_expiring(self, *, now: datetime) -> tuple[ExportReceipt, ...]:
    """Export everything still inside the window. Nothing after it.

    Same attribution rule as :meth:`notices`, for a stronger reason: a
    receipt carries the state itself, so handing one to a guessed tenant
    is a cross-tenant *data* release rather than a mis-addressed warning.
    """
    if self._export is None or now >= self._horizon.cutoff:
        return ()
    return tuple(
        ExportReceipt(
            tenant_id=tenant_id,
            thread_id=ref.thread_id,
            checkpoint_id=ref.checkpoint_id,
            receipt=self._export(ref),
            at=now,
        )
        for ref, tenant_id in self._addressable(now=now)
    )

notices

notices(*, now: datetime) -> tuple[ExpiryNotice, ...]

Notices for the current moment; empty outside the notice window.

Notifying earlier would train tenants to ignore the notice, and notifying after the cutoff would be an obituary, not a warning. Refs listed by :meth:unattributed are skipped: a notice addressed to a guessed tenant is a cross-tenant disclosure, not a warning.

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

def notices(self, *, now: datetime) -> tuple[ExpiryNotice, ...]:
    """Notices for the current moment; empty outside the notice window.

    Notifying earlier would train tenants to ignore the notice, and
    notifying after the cutoff would be an obituary, not a warning.
    Refs listed by :meth:`unattributed` are skipped: a notice addressed to
    a guessed tenant is a cross-tenant disclosure, not a warning.
    """
    if not self._horizon.in_notice_window(now=now):
        return ()
    return tuple(
        ExpiryNotice(
            tenant_id=tenant_id,
            thread_id=ref.thread_id,
            checkpoint_id=ref.checkpoint_id,
            deadline=self._horizon.cutoff,
            support_route=self._horizon.support_route,
        )
        for ref, tenant_id in self._addressable(now=now)
    )

unattributed

unattributed(*, now: datetime) -> tuple[CheckpointRef, ...]

Expiring refs whose owning tenant cannot be named.

A thread key the legacy path wrote under a separator-bearing tenant parses to a prefix of that tenant, so addressing a notice from it would hand one tenant another tenant's thread and checkpoint ids. Those refs are held back here rather than mis-addressed — and rather than dropped, because an expiry nobody can be told about is exactly the silent expiry the horizon exists to prevent. The operator route is to record the owning tenant on the ref (tenant_id), after which it notifies normally.

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

def unattributed(self, *, now: datetime) -> tuple[CheckpointRef, ...]:
    """Expiring refs whose owning tenant cannot be named.

    A thread key the legacy path wrote under a separator-bearing tenant
    parses to a *prefix* of that tenant, so addressing a notice from it
    would hand one tenant another tenant's thread and checkpoint ids.
    Those refs are held back here rather than mis-addressed — and rather
    than dropped, because an expiry nobody can be told about is exactly
    the silent expiry the horizon exists to prevent. The operator route is
    to record the owning tenant on the ref (``tenant_id``), after which it
    notifies normally.
    """
    return tuple(
        ref for ref in self.expiring(now=now) if not ref.attribution_is_certain
    )

TranscriptQuery dataclass

TranscriptQuery(thread_id: str, speaker: Speaker = 'all', limit: int | None = None, index: int | None = None, since: datetime | None = None, until: datetime | None = None)

One transcript read. Validated at construction, not at the store.

TranscriptRow dataclass

TranscriptRow(index: int, role: TranscriptRole, content: str, message_id: str | None = None, timestamp: datetime | None = None)

One verbatim transcript row on the public surface.

index is the ordinal within the speaker-filtered view that produced it, and timestamp is checkpoint granularity — None when the source cannot resolve one. Both were load-bearing on the legacy surface.

TranscriptService

TranscriptService(*, source: TranscriptSourcePort)

Reads verbatim transcripts through a source port.

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

def __init__(self, *, source: TranscriptSourcePort) -> None:
    self._source = source

read async

read(query: TranscriptQuery) -> tuple[TranscriptRow, ...]

Filter, then select an ordinal, then cap — the legacy order.

limit is applied last and keeps the first rows of the resulting view (rows[:limit]), exactly as the legacy read did. Rows keep the ordinal of the speaker-filtered view they came from, so a capped or ordinal-selected read still correlates with an uncapped one.

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

async def read(self, query: TranscriptQuery) -> tuple[TranscriptRow, ...]:
    """Filter, then select an ordinal, then cap — the legacy order.

    ``limit`` is applied last and keeps the *first* rows of the resulting
    view (``rows[:limit]``), exactly as the legacy read did. Rows keep the
    ordinal of the speaker-filtered view they came from, so a capped or
    ordinal-selected read still correlates with an uncapped one.
    """
    messages = await self._source.messages(query.thread_id)
    stamps = await self._resolve_stamps(query)
    rows = self._project(messages, query, stamps)
    if query.index is not None:
        rows = _select_ordinal(rows, query.index)
    if query.limit is not None:
        rows = rows[: query.limit]
    return rows

TranscriptSourcePort

Bases: Protocol

Where verbatim turns are read from.

messages async

messages(thread_id: str) -> Any

Return the thread's messages, oldest first.

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

async def messages(self, thread_id: str) -> Any:
    """Return the thread's messages, oldest first."""

timestamps async

timestamps(thread_id: str) -> dict[str, datetime] | None

Map message id -> introducing checkpoint time, or None.

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

async def timestamps(self, thread_id: str) -> dict[str, datetime] | None:
    """Map message id -> introducing checkpoint time, or ``None``."""

TranscriptUnavailableError

Bases: ConversationServiceError

The transcript cannot be served — no durable source, or no timestamps.

Distinct from "the transcript is empty", which is a legitimate answer.

UnsafeBoundaryError

Bases: ConversationServiceError

The state is mid-frame and has no contract-tested safe boundary.

There is no translation path by design. Arbitrary mid-frame state carries node-local invariants nobody re-validated, so the only supported crossing is replay from a boundary that was tested as a boundary.

UpgradeGuidance dataclass

UpgradeGuidance(verdict: UpgradeVerdict, thread_id: str, checkpoint_id: str, reason: str, support_route: str, package_version: str, boundary: SafeBoundaryMarker | None = None, next_step: str = '')

What an adopter should do with one local artifact, and why.

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,
    )

installed_package_version

installed_package_version(distribution: str = DISTRIBUTION_NAME) -> str

The installed distribution version, or a conservative fallback.

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

def installed_package_version(distribution: str = DISTRIBUTION_NAME) -> str:
    """The installed distribution version, or a conservative fallback."""
    try:
        from importlib.metadata import version

        return version(distribution)
    except Exception:  # noqa: BLE001 - any metadata failure means "unknown"
        return _UNKNOWN_VERSION

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

parse_version

parse_version(raw: str) -> tuple[int, int, int]

Parse major.minor.patch, tolerating a pre-release suffix.

Refuses anything it cannot read rather than sorting it low: a version that silently compares as 0.0.0 would place every unparseable build inside every support window, which is the wrong direction to fail.

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

def parse_version(raw: str) -> tuple[int, int, int]:
    """Parse ``major.minor.patch``, tolerating a pre-release suffix.

    Refuses anything it cannot read rather than sorting it low: a version that
    silently compares as ``0.0.0`` would place every unparseable build inside
    every support window, which is the wrong direction to fail.
    """
    parts = raw.strip().split(".")
    if len(parts) < 2:
        raise ValueError(f"unparseable package version: {raw!r}")
    numbers: list[int] = []
    for part in parts[:3]:
        match = _VERSION_PART.match(part)
        if match is None:
            raise ValueError(f"unparseable package version: {raw!r}")
        numbers.append(int(match.group(1)))
    while len(numbers) < 3:
        numbers.append(0)
    return numbers[0], numbers[1], numbers[2]

resolve_history

resolve_history(strategy: HistoryStrategy | None, *, default: HistoryStrategy | None = None) -> HistoryDirective

Pick the governing directive: explicit strategy, then default, then the framework default.

A non-strategy argument is refused rather than duck-typed. Silently ignoring a misspelled strategy would leave the conversation unbounded while the caller believed it was windowed.

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

def resolve_history(
    strategy: HistoryStrategy | None, *, default: HistoryStrategy | None = None
) -> HistoryDirective:
    """Pick the governing directive: explicit strategy, then default, then the
    framework default.

    A non-strategy argument is refused rather than duck-typed. Silently
    ignoring a misspelled strategy would leave the conversation unbounded
    while the caller believed it was windowed.
    """
    for candidate in (strategy, default):
        if candidate is None:
            continue
        if not isinstance(candidate, HistoryStrategy):
            raise ConversationServiceError(
                f"{type(candidate).__name__} is not a history strategy"
            )
        return candidate.directive()
    return HistoryDirective.default()

tenant_segment_is_provable

tenant_segment_is_provable(thread_id: str) -> bool

Whether this key's tenant segment is provably the whole tenant id.

The derivation puts exactly two separators in a key. A key carrying more could have come from either an exempt session id (t:_:https://x) or a separator-bearing tenant the legacy path never validated (acme:eu:_:s1), and nothing in the key itself distinguishes the two. Positional parsing stays deterministic for both — but attribution does not, so this predicate gates naming a tenant, never reading the state.

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

def tenant_segment_is_provable(thread_id: str) -> bool:
    """Whether this key's tenant segment is provably the *whole* tenant id.

    The derivation puts exactly two separators in a key. A key carrying more
    could have come from either an exempt session id (``t:_:https://x``) or a
    separator-bearing tenant the legacy path never validated
    (``acme:eu:_:s1``), and nothing in the key itself distinguishes the two.
    Positional parsing stays deterministic for both — but *attribution* does
    not, so this predicate gates naming a tenant, never reading the state.
    """
    return thread_id.count(_THREAD_SEPARATOR) == _DERIVED_SEPARATORS