Skip to content

symfonic.services.switching

switching

Distributed route switching and invocation pinning (T2.3.6).

One binding contract, two backends (DMC). Operated platforms bind the authorized, audited control plane over a per-key-linearizable switch store; library mode binds a hermetic in-process static-generation backend. Both answer the same BindingSource port, and both feed the same per-invocation binding stage — so the kernel never learns which one it is running under.

ActiveInvocationRegistry 

ActiveInvocationRegistry(*, clock: Callable[[], float] = time.time)

Append-only from the data plane; indexed by bundle and by generation.

Source code in src/symfonic/services/switching/registry.py 
81
82
83
84
85
86
def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
    self._rows: dict[str, InvocationRecord] = {}
    self._by_bundle: dict[str, list[str]] = {}
    self._by_generation: dict[str, list[str]] = {}
    self._clock = clock
    self._lock = asyncio.Lock()

admission_lock property 

admission_lock: Lock

Held across bind-and-register so the pair is one atomic admission.

expire_lease 

expire_lease(invocation_id: str, *, reason: str = 'lease expired') -> None

EFX-L-5 — a crashed worker's record is closed by lease expiry.

Source code in src/symfonic/services/switching/registry.py 
111
112
113
def expire_lease(self, invocation_id: str, *, reason: str = "lease expired") -> None:
    """EFX-L-5 — a crashed worker's record is closed by lease expiry."""
    self._close(invocation_id, reason)

invalidate 

invalidate(invocation_id: str, *, actor_role: str) -> None

CUT-AIR-4 — only the control plane, and only via SCP-REV-4.

Source code in src/symfonic/services/switching/registry.py 
123
124
125
126
127
128
129
130
131
132
def invalidate(self, invocation_id: str, *, actor_role: str) -> None:
    """CUT-AIR-4 — only the control plane, and only via SCP-REV-4."""
    if actor_role != CONTROL_PLANE_ROLE:
        raise PermissionError(
            f"role {actor_role!r} may not invalidate registry rows; registry "
            "writes are append-only from the data plane and only the control "
            "plane marks rows invalidated."
        )
    row = self._rows[invocation_id]
    self._rows[invocation_id] = replace(row, invalidated_by=CONTROL_PLANE_ROLE)

quiescent_below 

quiescent_below(bundle_id: str, epoch: int) -> bool

CUT-AIR-3 — no open record on this bundle below epoch.

Source code in src/symfonic/services/switching/registry.py 
153
154
155
156
157
def quiescent_below(self, bundle_id: str, epoch: int) -> bool:
    """CUT-AIR-3 — no open record on this bundle below ``epoch``."""
    return not any(
        row.admitted_epoch < epoch for row in self.open_records(bundle_id)
    )

Actor dataclass 

Actor(identity: str, role: Role, bundles: frozenset[str] = frozenset(), kind: str = 'human')

SCP-AUTH-1 — an authenticated human or named workflow identity.

kind='service' exists so it can be refused: a shared service account is the exact anti-pattern the clause names, and a vocabulary that could not express it would turn a stated rule into an unenforceable one.

AdmissionController 

AdmissionController(*, source: Any, registry: ActiveInvocationRegistry | None = None, worker_id: str = 'worker', barrier: QuiescenceBarrier | None = None, clock: Callable[[], float] = time.time)

Binds a generation and registers the invocation under one lock.

Source code in src/symfonic/services/switching/admission.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    *,
    source: Any,
    registry: ActiveInvocationRegistry | None = None,
    worker_id: str = "worker",
    barrier: QuiescenceBarrier | None = None,
    clock: Callable[[], float] = time.time,
) -> None:
    self._source = source
    self._registry = registry
    self._worker_id = worker_id
    self._barrier = barrier
    self._clock = clock
    self._lock = registry.admission_lock if registry is not None else asyncio.Lock()

acknowledge_freeze async 

acknowledge_freeze(bundle_id: str) -> str | None

CUT-BR-6 — this worker states that it has observed the freeze.

Source code in src/symfonic/services/switching/admission.py 
123
124
125
126
127
128
129
async def acknowledge_freeze(self, bundle_id: str) -> str | None:
    """CUT-BR-6 — this worker states that it has observed the freeze."""
    binding = await self._source.resolve(bundle_id)
    epoch_id = binding.freeze_state.freeze_epoch_id
    if epoch_id is not None and self._barrier is not None:
        self._barrier.acknowledge(self._worker_id, epoch_id)
    return epoch_id

admit async 

admit(*, invocation_id: str, bundle_id: str, tenant_scope_hash: str, inherited_pin: InvocationPin | None = None) -> BoundGeneration

Resolve one snapshot and register it, atomically against the freeze.

Source code in src/symfonic/services/switching/admission.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
async def admit(
    self,
    *,
    invocation_id: str,
    bundle_id: str,
    tenant_scope_hash: str,
    inherited_pin: InvocationPin | None = None,
) -> BoundGeneration:
    """Resolve one snapshot and register it, atomically against the freeze."""
    if inherited_pin is not None:
        return await self._admit_pinned(
            inherited_pin, invocation_id=invocation_id, scope=tenant_scope_hash
        )
    async with self._lock:
        binding = await self._source.resolve(bundle_id)
        self._require_admissible(binding, binding)
        return self._register(
            binding.pin(), binding, invocation_id, tenant_scope_hash
        )

resume async 

resume(pin: InvocationPin | None, *, invocation_id: str, tenant_scope_hash: str = '') -> BoundGeneration

Resume under the artifact's pin, not under this worker's binding.

Source code in src/symfonic/services/switching/admission.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
async def resume(
    self,
    pin: InvocationPin | None,
    *,
    invocation_id: str,
    tenant_scope_hash: str = "",
) -> BoundGeneration:
    """Resume under the artifact's pin, not under this worker's binding."""
    if pin is None:
        raise PinlessArtifactError(
            "this artifact carries no generation pin, and no pin-less artifact "
            "policy is configured; resuming it would silently place it on "
            "whatever generation this worker happens to run (CUT-PIN-1)."
        )
    return await self._admit_pinned(
        pin, invocation_id=invocation_id, scope=tenant_scope_hash
    )

AuditRecord dataclass 

AuditRecord(seq: int, prev_hash: str, actor: str, role: str, bundle_id: str, prior_epoch: int, proposed_vector_hash: str, outcome: str, reason: str, timestamp: float)

One mutation attempt — accepted or denied, both recorded identically.

BindingSource

Bases: Protocol

Resolve a route bundle's binding. Async-first, read-only, fail-closed.

describe 

describe() -> str

CUT-PIN-3 — the effective vector and its source, never credentials.

Source code in src/symfonic/services/switching/ports.py 
60
61
62
def describe(self) -> str:
    """CUT-PIN-3 — the effective vector and its source, never credentials."""
    ...

resolve async 

resolve(bundle_id: str) -> BundleBinding

The current binding snapshot, or a denial. Never a default.

Source code in src/symfonic/services/switching/ports.py 
47
48
49
async def resolve(self, bundle_id: str) -> BundleBinding:
    """The current binding snapshot, or a denial. Never a default."""
    ...

resolve_epoch async 

resolve_epoch(bundle_id: str, epoch: int) -> BundleBinding

CUT-SS-7 — the binding an earlier admission ran under.

Resume paths need it: a pin names an epoch, and placing the resumed work back on that epoch's vector is the only way a switch that landed while the work was paused does not silently follow it.

Source code in src/symfonic/services/switching/ports.py 
51
52
53
54
55
56
57
58
async def resolve_epoch(self, bundle_id: str, epoch: int) -> BundleBinding:
    """CUT-SS-7 — the binding an earlier admission ran under.

    Resume paths need it: a pin names an epoch, and placing the resumed
    work back on that epoch's vector is the only way a switch that landed
    while the work was paused does not silently follow it.
    """
    ...

BindingStage dataclass 

BindingStage(bundle_id: str, tenant_scope_hash: str = '', inherited_pin: InvocationPin | None = None)

A bind-phase stage: what to bind, never what was bound.

Deliberately holds no vector, epoch, or binding source. Two invocations of one compiled plan must be able to observe two different generations, and a stage that cached a resolved vector could not express that.

child 

child(pin: InvocationPin) -> BindingStage

Derive the stage a sub-agent runs under (EFX-L-2: never widening).

A child inherits the parent's pin rather than resolving its own. A child that re-resolved could land on a newer generation mid-way through its parent's invocation, which is the exact mixed-generation run that CUT-SS-4 forbids.

Source code in src/symfonic/services/switching/binding_stage.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def child(self, pin: InvocationPin) -> BindingStage:
    """Derive the stage a sub-agent runs under (EFX-L-2: never widening).

    A child inherits the parent's pin rather than resolving its own. A
    child that re-resolved could land on a newer generation mid-way through
    its parent's invocation, which is the exact mixed-generation run that
    CUT-SS-4 forbids.
    """
    if pin.bundle_id != self.bundle_id:
        raise ContractViolationError(
            f"a child binding stage inherits its parent's pin; the supplied pin "
            f"names bundle {pin.bundle_id!r} but the stage binds "
            f"{self.bundle_id!r}."
        )
    return replace(self, inherited_pin=pin)

execute async 

execute(admission: AdmissionController, *, invocation_id: str) -> BoundGeneration

Capture exactly one generation vector for one invocation.

Source code in src/symfonic/services/switching/binding_stage.py 
36
37
38
39
40
41
42
43
44
45
async def execute(
    self, admission: AdmissionController, *, invocation_id: str
) -> BoundGeneration:
    """Capture exactly one generation vector for one invocation."""
    return await admission.admit(
        invocation_id=invocation_id,
        bundle_id=self.bundle_id,
        tenant_scope_hash=self.tenant_scope_hash,
        inherited_pin=self.inherited_pin,
    )

BindingUnavailableError

Bases: SwitchingError

CUT-BR-4/CUT-BR-5 — no admissible binding. Never a default fallback.

BoundGeneration dataclass 

BoundGeneration(pin: InvocationPin, binding: BundleBinding, record: InvocationRecord | None = None)

What one invocation captured: a pin, its binding, and its registry row.

record is None in library mode: CUT-AIR-5 makes the registry operated-platform-only, and a library adopter must not find themselves maintaining a drain-proof table they have no control plane to drain for.

BreakGlassCredential dataclass 

BreakGlassCredential(credential_id: str, bundle_id: str, identity: str = 'bg:on-call')

SCP-BG-1/3 — pre-provisioned, sealed, single-use, bundle-scoped.

BundleBinding dataclass 

BundleBinding(bundle_id: str, epoch: int, generation_vector: GenerationVector, freeze_state: FreezeState = NO_FREEZE, source: str = 'release-static', fetched_at: float = 0.0, stale: bool = False)

DMC-1 — what a BindingSource answers with.

source is one of release-static, control-plane or local-override (CUT-PIN-3), which is what makes "where did this binding come from?" answerable in a diagnostic without guessing.

BundleRecord dataclass 

BundleRecord(bundle_id: str, epoch: int, generation_vector: GenerationVector, freeze_state: FreezeState = NO_FREEZE, audit_head: str = '0' * 64, rollback_vector: GenerationVector | None = None, committed_at: float = 0.0)

CUT-SS-1/CUT-SS-3 — the entire binding state of one bundle.

One record, one atomic write. There is deliberately no way to express a half-switched bundle: the vector, the freeze and the audit head move together or not at all.

CompatibilityConstraint dataclass 

CompatibilityConstraint(subject: str, subject_generation: str, requires: str, minimum: str)

subject_generation requires requires at minimum or newer.

ConstraintSet 

ConstraintSet(constraints: Sequence[CompatibilityConstraint] = ())

The declared constraints plus the well-formedness rules every vector obeys.

Source code in src/symfonic/services/switching/constraints.py 
67
68
def __init__(self, constraints: Sequence[CompatibilityConstraint] = ()) -> None:
    self._constraints = tuple(constraints)

validate 

validate(vector: GenerationVector, *, context: str = 'vector') -> None

Raise on the first illegal combination (fail-closed, SEC-FCP-1).

Source code in src/symfonic/services/switching/constraints.py 
 94
 95
 96
 97
 98
 99
100
101
102
def validate(self, vector: GenerationVector, *, context: str = "vector") -> None:
    """Raise on the first illegal combination (fail-closed, SEC-FCP-1)."""
    reasons = self.violations(vector)
    if reasons:
        joined = "; ".join(reasons)
        raise ConstraintViolationError(
            f"{context} {vector.describe()!r} violates the compatibility "
            f"constraint set: {joined}."
        )

violations 

violations(vector: GenerationVector) -> tuple[str, ...]

Every reason this vector is illegal, in declaration order.

Source code in src/symfonic/services/switching/constraints.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def violations(self, vector: GenerationVector) -> tuple[str, ...]:
    """Every reason this vector is illegal, in declaration order."""
    found: list[str] = []
    for name, generation in vector.entries:
        try:
            subject, _ = parse_generation(generation)
        except ConstraintViolationError as exc:
            found.append(str(exc))
            continue
        if subject != name:
            found.append(
                f"vector entry {name!r} binds generation {generation!r}, whose "
                f"subject is {subject!r}; an entry may not rename its own generation"
            )
    for constraint in self._constraints:
        reason = constraint.violated_by(vector)
        if reason is not None:
            found.append(reason)
    return tuple(found)

ConstraintViolationError

Bases: ConfigurationError

CUT-RB-4/CUT-RB-5 — the proposed vector is not a legal combination.

A ConfigurationError on purpose: an illegal vector is a configuration fault whether it arrives from a control-plane proposal or a local override, and library-mode construction must reject it with the taxonomy adopters already catch.

ControlPlaneBindingSource 

ControlPlaneBindingSource(store: InMemorySwitchStore, constraints: ConstraintSet, *, t_stale: float = DEFAULT_T_STALE, t_outage: float = DEFAULT_T_OUTAGE, cutover_window: float = DEFAULT_T_CUTOVER_WINDOW, clock: Callable[[], float] = time.monotonic)

Cache-first BindingSource implementing CUT-BR-1..6.

Source code in src/symfonic/services/switching/control_plane_source.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    store: InMemorySwitchStore,
    constraints: ConstraintSet,
    *,
    t_stale: float = DEFAULT_T_STALE,
    t_outage: float = DEFAULT_T_OUTAGE,
    cutover_window: float = DEFAULT_T_CUTOVER_WINDOW,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    self._store = store
    self._constraints = constraints
    self._t_stale = t_stale
    self._t_outage = t_outage
    self._cutover_window = cutover_window
    self._clock = clock
    self._cache: dict[str, _CacheEntry] = {}

refresh async 

refresh(bundle_id: str) -> BundleRecord

The watcher/poller write point. Never called from the hot path.

Source code in src/symfonic/services/switching/control_plane_source.py 
80
81
82
83
84
async def refresh(self, bundle_id: str) -> BundleRecord:
    """The watcher/poller write point. Never called from the hot path."""
    record = await self._store.read(bundle_id)
    self._cache[bundle_id] = _CacheEntry(record=record, fetched_at=self._clock())
    return record

resolve_epoch async 

resolve_epoch(bundle_id: str, epoch: int) -> BundleBinding

CUT-SS-7 — reconstruct the binding an earlier admission ran under.

Source code in src/symfonic/services/switching/control_plane_source.py 
101
102
103
104
105
106
107
108
109
async def resolve_epoch(self, bundle_id: str, epoch: int) -> BundleBinding:
    """CUT-SS-7 — reconstruct the binding an earlier admission ran under."""
    for record in await self._store.history(bundle_id):
        if record.epoch == epoch:
            return self._binding(record, stale=False)
    raise BindingUnavailableError(
        f"the switch store retains no epoch {epoch} for bundle {bundle_id!r}; a "
        "pinned resume denies rather than falling forward onto the current epoch."
    )

EnvelopeSigner 

EnvelopeSigner(keyset: Keyset, *, producer_package_version: str, clock: Callable[[], float] = time.time)

Mints and verifies pin envelopes against a keyset (ENV-2, ENV-6, KEY-5).

Source code in src/symfonic/services/switching/envelope.py 
123
124
125
126
127
128
129
130
131
132
def __init__(
    self,
    keyset: Keyset,
    *,
    producer_package_version: str,
    clock: Callable[[], float] = time.time,
) -> None:
    self._keyset = keyset
    self._producer_version = producer_package_version
    self._clock = clock

mint 

mint(*, pin: InvocationPin, payload: bytes, schema_id: str, envelope_version: int = 2) -> PinEnvelope

ENV-6 — refuse to mint when the active key is unavailable.

Source code in src/symfonic/services/switching/envelope.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def mint(
    self,
    *,
    pin: InvocationPin,
    payload: bytes,
    schema_id: str,
    envelope_version: int = 2,
) -> PinEnvelope:
    """ENV-6 — refuse to mint when the active key is unavailable."""
    key_id = self._keyset.active_key_id()
    unsigned = PinEnvelope(
        envelope_version=envelope_version,
        generation_vector_hash=pin.vector_hash,
        schema_id=schema_id,
        producer_package_version=self._producer_version,
        created_at=self._clock(),
        key_id=key_id,
        pin=pin,
        payload=payload,
    )
    signature = self._keyset.sign(key_id, unsigned.signing_input())
    return PinEnvelope(
        envelope_version=unsigned.envelope_version,
        generation_vector_hash=unsigned.generation_vector_hash,
        schema_id=unsigned.schema_id,
        producer_package_version=unsigned.producer_package_version,
        created_at=unsigned.created_at,
        key_id=unsigned.key_id,
        pin=unsigned.pin,
        payload=unsigned.payload,
        signature=signature,
    )

verify 

verify(envelope: PinEnvelope | None, *, pinless_policy: PinlessArtifactPolicy | None = None, schema_id: str = '') -> InvocationPin

Verify first, then dispatch on version. Never the other way round.

Source code in src/symfonic/services/switching/envelope.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def verify(
    self,
    envelope: PinEnvelope | None,
    *,
    pinless_policy: PinlessArtifactPolicy | None = None,
    schema_id: str = "",
) -> InvocationPin:
    """Verify first, then dispatch on version. Never the other way round."""
    if envelope is None:
        policy = pinless_policy or PinlessArtifactPolicy()
        policy.resolve(schema_id=schema_id)
        raise PinlessArtifactError(
            "a pin-less artifact resolves to an attributed vector, not to an "
            "invocation pin; callers must handle it through the policy."
        )
    unsigned = PinEnvelope(
        envelope_version=envelope.envelope_version,
        generation_vector_hash=envelope.generation_vector_hash,
        schema_id=envelope.schema_id,
        producer_package_version=envelope.producer_package_version,
        created_at=envelope.created_at,
        key_id=envelope.key_id,
        pin=envelope.pin,
        payload=envelope.payload,
    )
    if not self._keyset.verifies(
        envelope.key_id, unsigned.signing_input(), envelope.signature
    ):
        raise EnvelopeVerificationError(
            f"envelope signed by key {envelope.key_id!r} does not verify; it is "
            "treated as tampered or corrupt and never partially honored."
        )
    if envelope.envelope_version not in SUPPORTED_ENVELOPE_VERSIONS:
        raise EnvelopeVersionError(
            f"envelope layout version {envelope.envelope_version} is outside "
            f"this build's supported window {sorted(SUPPORTED_ENVELOPE_VERSIONS)}; "
            "readers reject rather than guess at a layout."
        )
    return envelope.pin

EnvelopeVerificationError

Bases: SwitchingError

ENV-2/ENV-5 — the envelope's signature did not verify.

EnvelopeVersionError

Bases: SwitchingError

ENV-3 — a known-good envelope names a layout or schema we cannot read.

EvidenceInvalidated dataclass 

EvidenceInvalidated(freeze_epoch_id: str, bundle_id: str, reason: str, evidence_ids: tuple[str, ...], kinds: tuple[str, ...], invalidated_at: float = 0.0)

The event T4.4.7 consumes when a freeze revocation lands.

FreezeState dataclass 

FreezeState(freeze_epoch_id: str | None = None, frozen_legacy_vector: GenerationVector | None = None, retiring: frozenset[str] = frozenset(), created_by: str | None = None, approved_by: str | None = None, created_at: float = 0.0)

SCP-FRZ-1 — the retirement-freeze epoch object, or its absence.

retiring is the set of generation ids the freeze disables for new admissions. It is computed once, when the freeze is committed, so a later vector change cannot silently widen or narrow what the freeze barred.

bars 

bars(vector: GenerationVector) -> frozenset[str]

The retiring generations this vector would newly admit work onto.

Source code in src/symfonic/services/switching/values.py 
100
101
102
103
104
def bars(self, vector: GenerationVector) -> frozenset[str]:
    """The retiring generations this vector would newly admit work onto."""
    if not self.active:
        return frozenset()
    return self.retiring & vector.generations()

FreezeViolationError

Bases: SwitchingError

SCP-FRZ-1/CUT-RB-7 — a retirement freeze bars this generation.

GenerationVector dataclass 

GenerationVector(entries: tuple[tuple[str, str], ...] = ())

CUT-RB-3 — {capability_or_port → generation_id}, canonically ordered.

Stored as a sorted tuple rather than a mapping so two vectors built from differently-ordered dicts are the same value, compare equal, and hash to the same vector_hash. Ordering is the whole reason the hash is stable enough to travel inside a checkpoint envelope.

vector_hash property 

vector_hash: str

A stable digest of the whole vector (ENV-1 generation_vector_hash).

generations 

generations() -> frozenset[str]

The generation ids this vector selects, without their port names.

Source code in src/symfonic/services/switching/values.py 
66
67
68
def generations(self) -> frozenset[str]:
    """The generation ids this vector selects, without their port names."""
    return frozenset(generation for _, generation in self.entries)

HashChainAuditLog 

HashChainAuditLog(*, clock: Callable[[], float] = time.time)

An in-process append-only hash chain (the reference implementation).

Source code in src/symfonic/services/switching/audit.py 
71
72
73
def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
    self._records: list[AuditRecord] = []
    self._clock = clock

head property 

head: str

The chain head the bundle record binds itself to (SCP-AUD-1).

assert_no_credentials staticmethod 

assert_no_credentials(text: str) -> None

SCP-AUD-3 — refuse credential-shaped free text before it is chained.

Source code in src/symfonic/services/switching/audit.py 
75
76
77
78
79
80
81
82
@staticmethod
def assert_no_credentials(text: str) -> None:
    """SCP-AUD-3 — refuse credential-shaped free text before it is chained."""
    if _CREDENTIAL_SHAPES.search(text):
        raise ValueError(
            "audit reasons carry identities, hashes and outcomes — never "
            "credential material; refusing to chain the supplied text."
        )

commit 

commit(record: AuditRecord) -> AuditRecord

Chain a prepared record, refusing one that no longer fits the chain.

Source code in src/symfonic/services/switching/audit.py 
118
119
120
121
122
123
124
125
126
def commit(self, record: AuditRecord) -> AuditRecord:
    """Chain a prepared record, refusing one that no longer fits the chain."""
    if record.seq != len(self._records) or record.prev_hash != self.head:
        raise ValueError(
            "the prepared audit record no longer chains onto the log head; "
            "prepare and commit must not be interleaved with another append."
        )
    self._records.append(record)
    return record

prepare 

prepare(*, actor: str, role: str, bundle_id: str, prior_epoch: int, proposed_vector_hash: str, outcome: str, reason: str = '') -> AuditRecord

Build the next record without chaining it.

Preparing and committing are separate because the bundle record must store the audit head that attests it (SCP-AUD-1), and the record it attests is only durable once the CAS lands. Preparing lets the caller compute that head, commit the bundle, and only then chain the entry — so a lost CAS race never leaves an "accepted" record for a mutation that never happened.

Source code in src/symfonic/services/switching/audit.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def prepare(
    self,
    *,
    actor: str,
    role: str,
    bundle_id: str,
    prior_epoch: int,
    proposed_vector_hash: str,
    outcome: str,
    reason: str = "",
) -> AuditRecord:
    """Build the next record without chaining it.

    Preparing and committing are separate because the bundle record must
    store the audit head that attests it (SCP-AUD-1), and the record it
    attests is only durable once the CAS lands. Preparing lets the caller
    compute that head, commit the bundle, and only then chain the entry —
    so a lost CAS race never leaves an "accepted" record for a mutation
    that never happened.
    """
    self.assert_no_credentials(reason)
    return AuditRecord(
        seq=len(self._records),
        prev_hash=self.head,
        actor=actor,
        role=role,
        bundle_id=bundle_id,
        prior_epoch=prior_epoch,
        proposed_vector_hash=proposed_vector_hash,
        outcome=outcome,
        reason=reason,
        timestamp=self._clock(),
    )

tamper_for_test 

tamper_for_test(index: int, *, reason: str) -> None

Rewrite one record in place — the only way to exercise SCP-AUD-2.

It lives on the log rather than in a test helper so that "the chain detects this" is asserted against the same object production uses, and so any future storage backend inherits the same characterisation.

Source code in src/symfonic/services/switching/audit.py 
151
152
153
154
155
156
157
158
def tamper_for_test(self, index: int, *, reason: str) -> None:
    """Rewrite one record in place — the only way to exercise SCP-AUD-2.

    It lives on the log rather than in a test helper so that "the chain
    detects this" is asserted against the same object production uses, and
    so any future storage backend inherits the same characterisation.
    """
    self._records[index] = replace(self._records[index], reason=reason)

verify 

verify() -> bool

SCP-AUD-2 — recompute the chain from genesis.

Source code in src/symfonic/services/switching/audit.py 
142
143
144
145
146
147
148
149
def verify(self) -> bool:
    """SCP-AUD-2 — recompute the chain from genesis."""
    expected = GENESIS_HASH
    for index, record in enumerate(self._records):
        if record.seq != index or record.prev_hash != expected:
            return False
        expected = record.digest()
    return True

InMemorySwitchStore 

InMemorySwitchStore()

Per-key linearizable CAS over an in-process dict, with epoch history.

Every mutation takes the same lock and re-reads the current epoch inside it, so a lost race raises SwitchConflictError instead of overwriting. set_unreachable models the control-plane outage the availability rules in CUT-BR exist to survive.

Source code in src/symfonic/services/switching/store.py 
53
54
55
56
57
def __init__(self) -> None:
    self._records: dict[str, BundleRecord] = {}
    self._history: dict[str, list[BundleRecord]] = {}
    self._lock = asyncio.Lock()
    self._unreachable = False

compare_and_swap async 

compare_and_swap(record: BundleRecord, *, expected_epoch: int) -> BundleRecord

CUT-SS-2 — commit record iff the stored epoch is still expected.

Source code in src/symfonic/services/switching/store.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def compare_and_swap(
    self, record: BundleRecord, *, expected_epoch: int
) -> BundleRecord:
    """CUT-SS-2 — commit ``record`` iff the stored epoch is still expected."""
    async with self._lock:
        self._guard()
        current = self._records.get(record.bundle_id)
        if current is None:
            raise BindingUnavailableError(
                f"cannot compare-and-swap unknown bundle {record.bundle_id!r}."
            )
        if current.epoch != expected_epoch:
            raise SwitchConflictError(
                f"compare-and-swap on bundle {record.bundle_id!r} expected epoch "
                f"{expected_epoch} but the store is at {current.epoch}; the "
                "mutation was not applied."
            )
        stored = replace(record, epoch=current.epoch + 1)
        self._records[stored.bundle_id] = stored
        self._history.setdefault(stored.bundle_id, []).append(stored)
        return stored

history async 

history(bundle_id: str) -> tuple[BundleRecord, ...]

CUT-SS-7 — the full epoch history, for audit reconstruction.

Source code in src/symfonic/services/switching/store.py 
118
119
120
121
async def history(self, bundle_id: str) -> tuple[BundleRecord, ...]:
    """CUT-SS-7 — the full epoch history, for audit reconstruction."""
    self._guard()
    return tuple(self._history.get(bundle_id, ()))

InvocationPin dataclass 

InvocationPin(bundle_id: str, epoch: int, vector_hash: str, source: str, stale_binding: bool = False, freeze_epoch_id: str | None = None)

The travelling half of a binding: what a checkpoint or token carries.

A pin names an epoch and a vector hash, never the vector's contents. A resuming worker resolves the contents from its own binding source and refuses if they disagree, so a pin cannot smuggle an unvalidated vector across a process boundary.

token 

token() -> str

The compact form recorded on a RequestContext (RCX-5).

Source code in src/symfonic/services/switching/values.py 
156
157
158
def token(self) -> str:
    """The compact form recorded on a ``RequestContext`` (RCX-5)."""
    return f"{self.bundle_id}@{self.epoch}:{self.vector_hash}"

InvocationRecord dataclass 

InvocationRecord(invocation_id: str, tenant_scope_hash: str, bundle_id: str, admitted_epoch: int, generation_vector_hash: str, admitted_at: float, stale_binding: bool = False, completed_at: float | None = None, closed_reason: str | None = None, invalidated_by: str | None = None)

One admitted invocation. Carries a scope hash, never tenant content.

Keyset 

Keyset()

key_id → (material, state), with exactly one active key.

Key material never leaves this object: there is no accessor that returns it, __repr__ names only ids and states, and signing happens here rather than in the caller. A getter would be convenient exactly once and then live forever in a log line.

Source code in src/symfonic/services/switching/keys.py 
38
39
40
def __init__(self) -> None:
    self._keys: dict[str, tuple[bytes, KeyState]] = {}
    self._available = True

retire 

retire(key_id: str) -> None

KEY-4 — after the overlap window; its envelopes now verify-fail.

Source code in src/symfonic/services/switching/keys.py 
60
61
62
63
def retire(self, key_id: str) -> None:
    """KEY-4 — after the overlap window; its envelopes now verify-fail."""
    material, _ = self._require(key_id)
    self._keys[key_id] = (material, KeyState.RETIRED)

rotate 

rotate(key_id: str, *, secret: bytes) -> None

KEY-3 — the new key becomes active; the previous one verifies only.

Source code in src/symfonic/services/switching/keys.py 
56
57
58
def rotate(self, key_id: str, *, secret: bytes) -> None:
    """KEY-3 — the new key becomes active; the previous one verifies only."""
    self.provision(key_id, secret=secret, state=KeyState.ACTIVE)

set_available 

set_available(available: bool) -> None

KEY-5 — model secret-manager/keyset unavailability.

Source code in src/symfonic/services/switching/keys.py 
42
43
44
def set_available(self, available: bool) -> None:
    """KEY-5 — model secret-manager/keyset unavailability."""
    self._available = available

KeysetUnavailableError

Bases: SwitchingError

KEY-5 — no active key to mint with, or no keyset to verify against.

Availability, never authorization (EMAP-6). A verifier with no keyset must deny, but the holder's token is provably still redeemable — reporting the outage as 401 tells them to stop retrying something that will work again in a minute.

LegacyPinRetiredError

Bases: ConfigurationError

LIB-OV-5 — this release removed the pinned legacy generation.

PinEnvelope dataclass 

PinEnvelope(envelope_version: int, generation_vector_hash: str, schema_id: str, producer_package_version: str, created_at: float, key_id: str, pin: InvocationPin, payload: bytes, signature: str = '')

ENV-1 — the authenticated wrapper around an opaque payload.

describe 

describe() -> str

A diagnostic line: ids and hashes only (SEC-CRED-2, KEY-6).

Source code in src/symfonic/services/switching/envelope.py 
75
76
77
78
79
80
81
def describe(self) -> str:
    """A diagnostic line: ids and hashes only (SEC-CRED-2, KEY-6)."""
    return (
        f"envelope v{self.envelope_version} schema={self.schema_id} "
        f"vector={self.generation_vector_hash} key={self.key_id} "
        f"pin={self.pin.token()}"
    )

signing_input 

signing_input() -> bytes

Every envelope field plus the payload — ENV-2 covers both.

Source code in src/symfonic/services/switching/envelope.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def signing_input(self) -> bytes:
    """Every envelope field plus the payload — ENV-2 covers both."""
    header = json.dumps(
        {
            "envelope_version": self.envelope_version,
            "generation_vector_hash": self.generation_vector_hash,
            "schema_id": self.schema_id,
            "producer_package_version": self.producer_package_version,
            "created_at": self.created_at,
            "key_id": self.key_id,
            "pin": [
                self.pin.bundle_id,
                self.pin.epoch,
                self.pin.vector_hash,
                self.pin.source,
                self.pin.stale_binding,
                self.pin.freeze_epoch_id,
            ],
        },
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
    return header + b"\x00" + self.payload

PinlessArtifactError

Bases: SwitchingError

CUT-PIN-1 — an un-pinned artifact with no configured policy to place it.

PinlessArtifactPolicy dataclass 

PinlessArtifactPolicy(accept: bool = False, attributed_vector_hash: str | None = None, reason: str = '')

CUT-PIN-1/2 — what happens to an artifact that carries no pin.

The default is refusal. Accepting one requires naming, at construction, the vector it should be attributed to and why — because the alternative is a baked-in default binding, which is the thing CUT-PIN-1 exists to forbid.

QuiescenceBarrier 

QuiescenceBarrier(workers: Sequence[str] = ())

CUT-BR-6 — the worker-acknowledged half of the drain proof.

A freeze is only propagated when every worker says so. Waiting out T_stale + T_outage and hoping is the alternative this replaces: it cannot distinguish "every worker saw the freeze" from "every worker is wedged", and those two states need opposite responses.

Source code in src/symfonic/services/switching/registry.py 
53
54
55
def __init__(self, workers: Sequence[str] = ()) -> None:
    self._workers = frozenset(workers)
    self._acks: dict[str, set[str]] = {}

ReleaseProfile dataclass 

ReleaseProfile(package_version: str, static_vectors: Mapping[str, GenerationVector], constraints: ConstraintSet, override_table: Mapping[str, Sequence[GenerationVector]] = dict())

Everything the hermetic library backend is allowed to know (DMC-3).

A profile is compiled data: the static vector the release was validated against, the constraint set it shipped, and the documented override table. There is no field for "where to look this up", because looking anything up is precisely what library mode may not do.

RetirementEvidence dataclass 

RetirementEvidence(evidence_id: str, freeze_epoch_id: str, bundle_id: str, kind: str, detail: str, recorded_at: float = 0.0, invalidated_by: str | None = None)

One gathered result, permanently attributed to one freeze epoch.

RetirementEvidenceStore 

RetirementEvidenceStore(*, clock: Callable[[], float] = time.time)

Append-only evidence with epoch-scoped invalidation and a subscriber seam.

Source code in src/symfonic/services/switching/evidence.py 
58
59
60
61
62
def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
    self._rows: list[RetirementEvidence] = []
    self._subscribers: list[Callable[[EvidenceInvalidated], None]] = []
    self._seq = itertools.count(1)
    self._clock = clock

covers 

covers(freeze_epoch_id: str, kinds: Sequence[str] = ()) -> bool

Whether every required evidence kind is present and still valid.

Source code in src/symfonic/services/switching/evidence.py 
92
93
94
95
96
97
98
def covers(
    self, freeze_epoch_id: str, kinds: Sequence[str] = ()
) -> bool:
    """Whether every required evidence kind is present and still valid."""
    required = frozenset(kinds) if kinds else REQUIRED_KINDS
    present = {row.kind for row in self.valid_for(freeze_epoch_id)}
    return required <= present

invalidate_epoch 

invalidate_epoch(freeze_epoch_id: str, *, reason: str) -> EvidenceInvalidated

Atomically mark every row of this epoch invalid and emit the event.

Source code in src/symfonic/services/switching/evidence.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def invalidate_epoch(self, freeze_epoch_id: str, *, reason: str) -> EvidenceInvalidated:
    """Atomically mark every row of this epoch invalid and emit the event."""
    affected: list[RetirementEvidence] = []
    for index, row in enumerate(self._rows):
        if row.freeze_epoch_id == freeze_epoch_id and row.valid:
            marked = replace(row, invalidated_by=freeze_epoch_id)
            self._rows[index] = marked
            affected.append(marked)
    event = EvidenceInvalidated(
        freeze_epoch_id=freeze_epoch_id,
        bundle_id=affected[0].bundle_id if affected else "",
        reason=reason,
        evidence_ids=tuple(row.evidence_id for row in affected),
        kinds=tuple(sorted({row.kind for row in affected})),
        invalidated_at=self._clock(),
    )
    for subscriber in self._subscribers:
        subscriber(event)
    return event

valid_for 

valid_for(freeze_epoch_id: str) -> tuple[RetirementEvidence, ...]

SCP-REV-4: reads filter on non-invalidated AND matching epoch.

Source code in src/symfonic/services/switching/evidence.py 
81
82
83
84
85
86
87
def valid_for(self, freeze_epoch_id: str) -> tuple[RetirementEvidence, ...]:
    """SCP-REV-4: reads filter on non-invalidated AND matching epoch."""
    return tuple(
        row
        for row in self._rows
        if row.freeze_epoch_id == freeze_epoch_id and row.valid
    )

StaticBindingSource 

StaticBindingSource(profile: ReleaseProfile, *, overrides: Mapping[str, GenerationVector] | None = None)

Resolve bindings from the release's static vector plus local overrides.

Source code in src/symfonic/services/switching/static_source.py 
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self,
    profile: ReleaseProfile,
    *,
    overrides: Mapping[str, GenerationVector] | None = None,
) -> None:
    self._profile = profile
    self._vectors: dict[str, GenerationVector] = dict(profile.static_vectors)
    self._sources: dict[str, str] = dict.fromkeys(self._vectors, "release-static")
    self._freeze = self._package_freeze(profile)
    for bundle_id, override in (overrides or {}).items():
        self._apply_override(bundle_id, override)

describe 

describe() -> str

The startup log line: vector hash and where it came from, never keys.

Source code in src/symfonic/services/switching/static_source.py 
152
153
154
155
156
157
158
def describe(self) -> str:
    """The startup log line: vector hash and where it came from, never keys."""
    parts = [
        f"{bundle_id} vector={vector.vector_hash} source={self._sources[bundle_id]}"
        for bundle_id, vector in sorted(self._vectors.items())
    ]
    return f"binding[static,{self._profile.package_version}] " + "; ".join(parts)

resolve_epoch async 

resolve_epoch(bundle_id: str, epoch: int) -> BundleBinding

A library release has exactly one epoch: the one it compiled in.

Source code in src/symfonic/services/switching/static_source.py 
135
136
137
138
139
140
141
142
143
144
async def resolve_epoch(self, bundle_id: str, epoch: int) -> BundleBinding:
    """A library release has exactly one epoch: the one it compiled in."""
    binding = await self.resolve(bundle_id)
    if epoch != binding.epoch:
        raise BindingUnavailableError(
            f"this release binds bundle {bundle_id!r} at epoch {binding.epoch}; "
            f"there is no epoch {epoch} to resume onto. A pin from another "
            "deployment mode is not portable into library mode."
        )
    return binding

StoreUnavailableError

Bases: SwitchingError

The switch-state store could not be reached (control-plane dependency).

SwitchAuthorizationError

Bases: SwitchingError

SCP-AUTH — the actor may not perform this mutation, or authz is down.

SwitchAuthorizer 

SwitchAuthorizer(*, available: bool = True)

Evaluates SCP-AUTH. available=False models an identity-provider outage.

Source code in src/symfonic/services/switching/authz.py 
82
83
def __init__(self, *, available: bool = True) -> None:
    self._available = available

SwitchConflictError

Bases: SwitchingError

CUT-SS-2 — a compare-and-swap lost its race. Never a silent overwrite.

SwitchControlPlane 

SwitchControlPlane(*, store: InMemorySwitchStore, constraints: ConstraintSet, audit: HashChainAuditLog, authorizer: SwitchAuthorizer, validated_vectors: Sequence[GenerationVector] = (), evidence: RetirementEvidenceStore | None = None, clock: Callable[[], float] = time.time)

The authorized, audited switch-state service.

Source code in src/symfonic/services/switching/control_plane.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    *,
    store: InMemorySwitchStore,
    constraints: ConstraintSet,
    audit: HashChainAuditLog,
    authorizer: SwitchAuthorizer,
    validated_vectors: Sequence[GenerationVector] = (),
    evidence: RetirementEvidenceStore | None = None,
    clock: Callable[[], float] = time.time,
) -> None:
    self._store = store
    self._evidence = evidence
    self._clock = clock
    self._mutator = SwitchMutator(
        store=store,
        constraints=constraints,
        audit=audit,
        authorizer=authorizer,
        validated_vectors=validated_vectors,
        clock=clock,
    )
    self._break_glass: dict[str, BreakGlassCredential] = {}
    self._consumed: set[str] = set()

break_glass_revert async 

break_glass_revert(credential: BreakGlassCredential, bundle_id: str, *, expected_epoch: int | None = None, reason: str) -> BundleRecord

SCP-BG — revert-only, single-use, always audited, never a freeze verb.

Source code in src/symfonic/services/switching/control_plane.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def break_glass_revert(
    self,
    credential: BreakGlassCredential,
    bundle_id: str,
    *,
    expected_epoch: int | None = None,
    reason: str,
) -> BundleRecord:
    """SCP-BG — revert-only, single-use, always audited, never a freeze verb."""
    provisioned = self._break_glass.get(credential.credential_id)
    if provisioned is None or provisioned.bundle_id != bundle_id:
        raise SwitchAuthorizationError(
            f"break-glass credential {credential.credential_id!r} is not "
            f"provisioned for bundle {bundle_id!r}."
        )
    if credential.credential_id in self._consumed:
        raise SwitchAuthorizationError(
            f"break-glass credential {credential.credential_id!r} was already "
            "consumed; each credential is single-use and re-provisioning runs "
            "through the normal authenticated path."
        )
    record = await self._store.read(bundle_id)
    stored = await self._mutator.swap(
        Actor(identity=provisioned.identity, role=Role.BREAK_GLASS),
        Action.BREAK_GLASS_REVERT,
        bundle_id,
        vector=record.rollback_vector or record.generation_vector,
        expected_epoch=expected_epoch,
        reason=reason,
    )
    self._consumed.add(credential.credential_id)
    return stored

commit_vector async 

commit_vector(actor: Actor, bundle_id: str, vector: GenerationVector, *, expected_epoch: int | None = None, reason: str = '') -> BundleRecord

CUT-RB-2/CUT-SS-2 — one bundle, one validated vector, one CAS.

Source code in src/symfonic/services/switching/control_plane.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def commit_vector(
    self,
    actor: Actor,
    bundle_id: str,
    vector: GenerationVector,
    *,
    expected_epoch: int | None = None,
    reason: str = "",
) -> BundleRecord:
    """CUT-RB-2/CUT-SS-2 — one bundle, one validated vector, one CAS."""
    return await self._mutator.swap(
        actor,
        Action.COMMIT_VECTOR,
        bundle_id,
        vector=vector,
        expected_epoch=expected_epoch,
        reason=reason,
    )

freeze async 

freeze(actor: Actor, bundle_id: str, *, approver: Actor, expected_epoch: int | None = None, retiring: Sequence[str] | None = None, reason: str = 'freeze') -> BundleRecord

SCP-FRZ-1 — commit a reversible retirement-freeze epoch object.

Source code in src/symfonic/services/switching/control_plane.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def freeze(
    self,
    actor: Actor,
    bundle_id: str,
    *,
    approver: Actor,
    expected_epoch: int | None = None,
    retiring: Sequence[str] | None = None,
    reason: str = "freeze",
) -> BundleRecord:
    """SCP-FRZ-1 — commit a reversible retirement-freeze epoch object."""
    record = await self._store.read(bundle_id)
    legacy = record.rollback_vector or record.generation_vector
    state = FreezeState(
        freeze_epoch_id=f"frz-{uuid.uuid4().hex[:12]}",
        frozen_legacy_vector=legacy,
        retiring=(
            frozenset(retiring)
            if retiring is not None
            else self._default_retiring(record.generation_vector, legacy)
        ),
        created_by=actor.identity,
        approved_by=approver.identity,
        created_at=self._clock(),
    )
    return await self._mutator.swap(
        actor,
        Action.FREEZE,
        bundle_id,
        vector=record.generation_vector,
        expected_epoch=expected_epoch,
        reason=reason,
        approver=approver,
        freeze_state=state,
    )

retirement_ready 

retirement_ready(bundle_id: str, freeze_epoch_id: str, *, registry: object | None = None, barrier: object | None = None, epoch: int | None = None) -> bool

SCP-FRZ-3 — every leg of the freeze-and-drain gate, or False.

Deliberately returns a boolean rather than raising: this is a gate a runbook polls, and "not yet" is its normal answer, not an exception.

Source code in src/symfonic/services/switching/control_plane.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def retirement_ready(
    self,
    bundle_id: str,
    freeze_epoch_id: str,
    *,
    registry: object | None = None,
    barrier: object | None = None,
    epoch: int | None = None,
) -> bool:
    """SCP-FRZ-3 — every leg of the freeze-and-drain gate, or ``False``.

    Deliberately returns a boolean rather than raising: this is a gate a
    runbook polls, and "not yet" is its normal answer, not an exception.
    """
    if not self.audit.verify():
        return False
    if self._evidence is None or not self._evidence.covers(freeze_epoch_id):
        return False
    if barrier is not None and not barrier.acknowledged(freeze_epoch_id):  # type: ignore[attr-defined]
        return False
    if registry is not None and epoch is not None:
        return bool(registry.quiescent_below(bundle_id, epoch))  # type: ignore[attr-defined]
    return True

revoke_freeze async 

revoke_freeze(actor: Actor, bundle_id: str, *, expected_epoch: int | None = None, reason: str) -> BundleRecord

SCP-REV — release-owner-only, audited, evidence-invalidating.

Source code in src/symfonic/services/switching/control_plane.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
async def revoke_freeze(
    self,
    actor: Actor,
    bundle_id: str,
    *,
    expected_epoch: int | None = None,
    reason: str,
) -> BundleRecord:
    """SCP-REV — release-owner-only, audited, evidence-invalidating."""
    record = await self._store.read(bundle_id)
    state = record.freeze_state
    if not state.active or state.frozen_legacy_vector is None:
        raise FreezeViolationError(
            f"bundle {bundle_id!r} carries no active freeze epoch to revoke."
        )
    stored = await self._mutator.swap(
        actor,
        Action.REVOKE_FREEZE,
        bundle_id,
        vector=state.frozen_legacy_vector,
        expected_epoch=expected_epoch,
        reason=reason,
        freeze_state=NO_FREEZE,
        bypass_freeze=True,
    )
    if self._evidence is not None:
        self._evidence.invalidate_epoch(
            state.freeze_epoch_id or "", reason=f"freeze revoked: {reason}"
        )
    return stored

rollback async 

rollback(actor: Actor, bundle_id: str, *, expected_epoch: int | None = None, reason: str = 'rollback') -> BundleRecord

CUT-RB-6 — rollback is a CAS to the named vector, never an edit.

Source code in src/symfonic/services/switching/control_plane.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def rollback(
    self,
    actor: Actor,
    bundle_id: str,
    *,
    expected_epoch: int | None = None,
    reason: str = "rollback",
) -> BundleRecord:
    """CUT-RB-6 — rollback is a CAS to the *named* vector, never an edit."""
    record = await self._store.read(bundle_id)
    return await self._mutator.swap(
        actor,
        Action.ROLLBACK,
        bundle_id,
        vector=record.rollback_vector or record.generation_vector,
        expected_epoch=expected_epoch,
        reason=reason,
    )

SwitchingError

Bases: SymfonicError

Root of the cutover taxonomy. Never raised directly.

Every refusal carries a transport-neutral code (ERR-5). Nothing here knows about HTTP; the code exists so a caller — including one across a package boundary, such as the pause-token reader that only sees an opaque signer port — can tell "you may not" from "I cannot right now" without matching on the exception's type or its message text.

derive_library_key 

derive_library_key(secret: str, *, salt: bytes, info: bytes = b'symfonic-envelope') -> tuple[str, bytes]

LIB-EA-1 — HKDF-SHA256 from an adopter secret; key_id is its fingerprint.

LIB-EA-2 is the important half: there is no default secret. An adopter who configures nothing gets a construction error, not a package-wide shared key.

Source code in src/symfonic/services/switching/keys.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def derive_library_key(
    secret: str, *, salt: bytes, info: bytes = b"symfonic-envelope"
) -> tuple[str, bytes]:
    """LIB-EA-1 — HKDF-SHA256 from an adopter secret; ``key_id`` is its fingerprint.

    LIB-EA-2 is the important half: there is no default secret. An adopter who
    configures nothing gets a construction error, not a package-wide shared key.
    """
    if not secret:
        raise ConfigurationError(
            "library mode requires an adopter-provided signing secret (explicit "
            "configuration parameter or the documented environment variable); "
            "there is no default secret and unsigned envelopes are forbidden."
        )
    prk = hmac.new(salt, secret.encode("utf-8"), hashlib.sha256).digest()
    material = hmac.new(prk, info + b"\x01", hashlib.sha256).digest()
    key_id = hashlib.sha256(material).hexdigest()[:16]
    return key_id, material

parse_generation 

parse_generation(generation_id: str) -> tuple[str, int]

Split name@N (CUT-RB-3 vocabulary). Unparsable ids never guess.

Source code in src/symfonic/services/switching/constraints.py 
24
25
26
27
28
29
30
31
32
def parse_generation(generation_id: str) -> tuple[str, int]:
    """Split ``name@N`` (CUT-RB-3 vocabulary). Unparsable ids never guess."""
    match = _GENERATION.match(generation_id)
    if match is None:
        raise ConstraintViolationError(
            f"generation id {generation_id!r} is not of the form 'name@N'; "
            "a generation identifies a schema+behavior generation, not a package version."
        )
    return match["name"], int(match["version"])