The library-mode upgrade-across-retirement journey.
An adopter running symfonic as a library has no operator, no calendar horizon,
and no support queue. Three consequences, each enforced here rather than
documented:
- The migration path ships in the package. Safe-boundary replay lives in
symfonic.services.conversation.migration and imports nothing from the
legacy engine, so it keeps working after T4.4.6 retires that engine. An
upgrade path that lived in the thing being retired would vanish precisely
when it was needed.
- The horizon is keyed by package version. A local artifact written years
ago is still migratable while the adopter's installed package is inside the
supported range. Expiring adopter-local state on the operated platform's
calendar would retire state that nothing had actually stopped supporting.
- Crossing the boundary is guided, then explicit. Inside the horizon the
adopter is told which boundary to replay from and what to run. Beyond it
they get a documented error naming the migration and expiry route — never
silence, and never a best-effort translation.
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
)
|
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.
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
|