Skip to content

symfonic.services.shadow

shadow

T2.3.7 — governed shadow execution and record/replay comparison.

Reading order, because the pieces only make sense as a chain:

effects / classification The exhaustive effect-port table. Fourteen families, one disposition each (deny or deterministically stub), and a fail-closed refusal for anything unclassified. trust The second axis. Ports are not the whole story — an in-process tool can open a socket — so every tool, plugin, and contributed stage is port-mediated (provably, reviewed, by construction) or opaque (the default), and opaque means non-shadowable and non-replayable. sentinel What makes a port-mediated claim falsifiable: direct-effect probes that turn a mis-declaration into a loud abort instead of a silent leak. gateway / state / context / harness The effect-suppressed run itself, its suppressed replacement state writes, the single surface the run body can reach, and the ledger a reviewer reads to believe the suppression claim. redaction / crypto / admission / recorder / store / privacy Capture, governed by threat-model §6: eligibility, allowlist and classification-aware redaction, minimization and sampling, encryption at rest, tenant scoping, access audit, retention, and the erasure contract wired into the legacy privacy-deletion path. replay / comparator / cutover Replay through the same harness, a comparator that refuses unsafe inputs, and the recorder that files each capability's cutover evidence under the one path it is entitled to use.

AccessAuditEntry dataclass 

AccessAuditEntry(at: datetime, actor: str, action: str, recording_id: str, tenant_id: str, allowed: bool, reason: str = '')

Append-only: every read attempt, allowed or not.

AccessGrant dataclass 

AccessGrant(actor: str, tenants: frozenset[str], purpose: str)

Who may read recordings, for which tenants, and why.

AllowedField dataclass 

AllowedField(path: str, data_class: DataClass, redaction: Redaction = Redaction.NONE, keep_chars: int = 64)

One dotted path that may be captured, and how it must be transformed.

CaptureAdmissionPolicy 

CaptureAdmissionPolicy(*, grants: Mapping[str, TenantCaptureGrant] | None = None, allowlist: FieldAllowlist | None = None, cipher: EncryptionPort | None = None, minimizer: PayloadMinimizer | None = None, response_allowlist: FieldAllowlist | None = None, clock: Any = None)

Decides whether one invocation may be recorded, and in what shape.

Source code in src/symfonic/services/shadow/admission.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def __init__(
    self,
    *,
    grants: Mapping[str, TenantCaptureGrant] | None = None,
    allowlist: FieldAllowlist | None = None,
    cipher: EncryptionPort | None = None,
    minimizer: PayloadMinimizer | None = None,
    response_allowlist: FieldAllowlist | None = None,
    clock: Any = None,
) -> None:
    self._grants = dict(grants or {})
    self._allowlist = allowlist
    self._cipher = cipher or RefusingCipher()
    self._minimizer = minimizer or PayloadMinimizer()
    self._response_allowlist = response_allowlist
    self._clock = clock or (lambda: datetime.now(UTC))

project_response 

project_response(response: Any) -> ResponseProjection

Govern a recorded port answer, not just the invocation payload.

Provider completions and tool results are the largest tenant-data surface in a recording and a common carrier of integration credentials, so they get the same treatment as the payload: credential-shaped keys dropped at every depth, then minimization.

Field allowlisting is opt-in here and configured separately from the payload allowlist, because a recorded answer is also the replay's stubbed answer: projecting every response through the payload allowlist would silently change what a replay can serve. A capability that records real tenant traffic configures response_allowlist; without one the answer is scrubbed and minimized but not projected, and the decision says so.

Source code in src/symfonic/services/shadow/admission.py 
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
def project_response(self, response: Any) -> ResponseProjection:
    """Govern a recorded port answer, not just the invocation payload.

    Provider completions and tool results are the largest tenant-data
    surface in a recording and a common carrier of integration
    credentials, so they get the same treatment as the payload:
    credential-shaped keys dropped at every depth, then minimization.

    Field allowlisting is *opt-in* here and configured separately from the
    payload allowlist, because a recorded answer is also the replay's
    stubbed answer: projecting every response through the payload
    allowlist would silently change what a replay can serve. A capability
    that records real tenant traffic configures ``response_allowlist``;
    without one the answer is scrubbed and minimized but not projected,
    and the decision says so.
    """
    allowlist = self._response_allowlist
    if allowlist is not None and isinstance(response, Mapping):
        projection: RedactionResult = allowlist.apply(response)
        return ResponseProjection(
            value=self._minimizer.apply(projection.kept),
            dropped_fields=projection.dropped,
            redacted_fields=projection.redacted,
        )
    dropped: list[str] = []
    scrubbed = scrub_credential_keys(response, "", dropped)
    return ResponseProjection(
        value=self._minimizer.apply(scrubbed), dropped_fields=tuple(sorted(dropped))
    )

CaptureDecision dataclass 

CaptureDecision(admitted: bool, reason: str, request: CaptureRequest, payload: Mapping[str, Any] = dict(), dropped_fields: tuple[str, ...] = (), redacted_fields: tuple[str, ...] = (), grant: TenantCaptureGrant | None = None)

Admitted with a minimized, redacted payload — or refused with a reason.

CaptureOutcome dataclass 

CaptureOutcome(admitted: bool, reason: str, session: RecordingSession | None = None)

Admitted (with a session) or refused (with a reason). Never both.

CaptureRefusedError

Bases: ShadowError

Safe capture could not be established; the invocation is unrecorded.

Recording fails closed. Callers catch this and continue serving the tenant's request — never the other way round.

CaptureRequest dataclass 

CaptureRequest(invocation_id: str, tenant_id: str, payload: Mapping[str, Any] = dict(), kind: str = 'invocation', at: datetime | None = None)

One invocation offered to the recorder.

ComparisonUnsafeError

Bases: ShadowError

TM-29d — this comparison would duplicate an externally visible effect.

ConstructionProof dataclass 

ConstructionProof(method: ProofMethod, verified_by: str, injected_ports: frozenset[str] = frozenset(), audited_module: str = '', self_declared: bool = False)

Why this extension's effects cannot leave the classified ports.

CutoverCriteriaRecorder 

CutoverCriteriaRecorder(trust: ExtensionTrustRegistry)

Files cutover evidence and enforces which path a capability may use.

Source code in src/symfonic/services/shadow/cutover.py 
74
75
76
77
def __init__(self, trust: ExtensionTrustRegistry) -> None:
    self._trust = trust
    self._evidence: dict[str, CutoverEvidence] = {}
    self._approvals: list[TrustApproval] = []

CutoverEvidence dataclass 

CutoverEvidence(capability: str, path: CutoverPath, recorded_at: datetime, criteria: dict[str, Any] = dict(), opaque_dependencies: tuple[str, ...] = (), comparison: ComparisonReport | None = None)

One capability's cutover evidence, filed under exactly one path.

CutoverPathError

Bases: ConfigurationError

Cutover evidence was filed under a path the capability may not use.

DataClass

Bases: StrEnum

Threat-model §6.2 classes, in ascending order of "must not capture".

DeletionWiringAttestation dataclass 

DeletionWiringAttestation(participant_id: str, registry_id: str, verified_at: datetime, probe_tenant: str, seeded: int, residual: int)

Evidence that erasure actually reaches this participant.

DeterministicStub

A pure function of (port, operation, payload). No state, no clock.

Determinism is the requirement, not realism: a comparator can only call a divergence real if re-running the same input twice gives the same answer.

DirectEffectSentinel dataclass 

DirectEffectSentinel(subject: str = '<extension>', block: bool = True, witnesses: list[DirectEffectWitness] = list())

Records — and by default blocks — direct effects during a call.

block=True (the default, and what the harness uses) makes the probe raise, so a mis-declared extension does not get to perform the effect it was not supposed to be able to perform. block=False is observe-only: the witness is recorded and the call is passed through to the real entry point, which is what an audit of a candidate declaration wants — it learns what the extension actually touches without changing its behaviour.

Either way the witness list is checked afterwards by assert_clean, so an extension that swallows the exception, or one that was merely observed, is still detected and still loses its declaration.

assert_clean 

assert_clean() -> None

Raise if anything was witnessed, even if the extension caught it.

Source code in src/symfonic/services/shadow/sentinel.py 
212
213
214
215
216
217
218
219
220
def assert_clean(self) -> None:
    """Raise if anything was witnessed, even if the extension caught it."""
    if not self.witnesses:
        return
    detail = "; ".join(f"{w.probe}: {w.detail}" for w in self.witnesses)
    raise MisdeclaredExtensionError(
        f"{self.subject!r} is declared port-mediated but performed direct "
        f"effect(s) — {detail}. The trust declaration is withdrawn."
    )

watching 

watching() -> Iterator[DirectEffectSentinel]

Install the probes for the duration of one extension call.

The probes themselves are process-global and reference-counted, so overlapping windows install once and restore once; ownership of any effect they see is decided by the context variable, which is scoped to this with block and therefore to this task.

Source code in src/symfonic/services/shadow/sentinel.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@contextmanager
def watching(self) -> Iterator[DirectEffectSentinel]:
    """Install the probes for the duration of one extension call.

    The probes themselves are process-global and reference-counted, so
    overlapping windows install once and restore once; ownership of any
    effect they see is decided by the context variable, which is scoped to
    this ``with`` block and therefore to this task.
    """
    token = _ACTIVE.set(self)
    _install()
    try:
        yield self
    finally:
        _uninstall()
        _ACTIVE.reset(token)

DirectEffectWitness dataclass 

DirectEffectWitness(probe: str, detail: str)

One observed effect that did not cross a framework port.

EffectAttempt dataclass 

EffectAttempt(seq: int, port_id: str, operation: str, request_digest: str, outcome: EffectOutcome, family: EffectFamily | None = None, externally_visible: bool = True, detail: str = '')

One attempted crossing of a framework effect port.

EffectFamily

Bases: StrEnum

Every effect family the architecture declares. Exhaustive by contract.

EffectLedger dataclass 

EffectLedger(entries: list[EffectAttempt] = list())

An append-only record of attempts, in the order they were made.

applied 

applied() -> tuple[EffectAttempt, ...]

Real effects. Must be empty for a shadow run's claim to stand.

Source code in src/symfonic/services/shadow/ledger.py 
76
77
78
def applied(self) -> tuple[EffectAttempt, ...]:
    """Real effects. Must be empty for a shadow run's claim to stand."""
    return self._with(EffectOutcome.APPLIED)

explain 

explain() -> str

A one-line-per-attempt summary safe to log: no payloads, ever.

Source code in src/symfonic/services/shadow/ledger.py 
86
87
88
89
90
91
92
def explain(self) -> str:
    """A one-line-per-attempt summary safe to log: no payloads, ever."""
    return "\n".join(
        f"{e.seq:>3} {e.outcome.value:<12} {e.port_id}.{e.operation} "
        f"[{e.request_digest[:12]}] {e.detail}".rstrip()
        for e in self.entries
    )

EffectOutcome

Bases: StrEnum

How a single attempted effect resolved.

EffectPort dataclass 

EffectPort(port_id: str, family: EffectFamily, disposition: ShadowDisposition, rationale: str, externally_visible: bool = True)

One classification row: a port, its family, and its shadow disposition.

rationale is required. A disposition with no recorded reason is a decision nobody can review later, and this table is cutover evidence.

EffectPortClassification dataclass 

EffectPortClassification(ports: tuple[EffectPort, ...] = ())

An immutable table of port_id -> disposition.

Immutable because a classification that could be widened at runtime would let a shadow run mint its own permission halfway through, which is exactly what the fail-closed rule is protecting against.

assert_classifies 

assert_classifies(port_ids: Iterable[str]) -> None

Every named port has a row. Complements :meth:assert_exhaustive.

Family coverage proves the taxonomy is complete; it says nothing about whether the ports the framework actually declares are in the table. A port the runtime crosses but the table has never heard of is not "fail-closed" — it is simply never observed, because nothing routes it through the gateway. This is the check that names them.

Source code in src/symfonic/services/shadow/classification.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def assert_classifies(self, port_ids: Iterable[str]) -> None:
    """Every named port has a row. Complements :meth:`assert_exhaustive`.

    Family coverage proves the *taxonomy* is complete; it says nothing
    about whether the ports the framework actually declares are in the
    table. A port the runtime crosses but the table has never heard of is
    not "fail-closed" — it is simply never observed, because nothing routes
    it through the gateway. This is the check that names them.
    """
    unknown = sorted({port_id for port_id in port_ids if not self.knows(port_id)})
    if unknown:
        raise IncompleteClassificationError(
            "declared framework effect ports with no classification row: "
            + ", ".join(unknown)
            + "; shadow evidence is withheld until each is classified"
        )

assert_exhaustive 

assert_exhaustive() -> None

Every declared family has at least one classified port.

Source code in src/symfonic/services/shadow/classification.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def assert_exhaustive(self) -> None:
    """Every declared family has at least one classified port."""
    missing = sorted(
        family.value for family in EffectFamily if family not in self.families_covered()
    )
    if missing:
        raise IncompleteClassificationError(
            "effect families with no classified port: "
            + ", ".join(missing)
            + "; shadow evidence is withheld until every declared family is "
            "classified"
        )

classify 

classify(port_id: str) -> EffectPort

The classification row, or a fail-closed refusal. Never a default.

Source code in src/symfonic/services/shadow/classification.py 
78
79
80
81
82
83
84
85
86
87
def classify(self, port_id: str) -> EffectPort:
    """The classification row, or a fail-closed refusal. Never a default."""
    try:
        return self._by_id[port_id]
    except KeyError:
        raise UnclassifiedEffectError(
            f"port {port_id!r} is not in the effect-port classification; a "
            "shadow run cannot suppress an effect it has never been told "
            "about, so the run is aborted rather than trusted"
        ) from None

extended_with 

extended_with(ports: Iterable[EffectPort]) -> EffectPortClassification

A new table with extra rows. Never mutates the receiver.

Source code in src/symfonic/services/shadow/classification.py 
67
68
69
def extended_with(self, ports: Iterable[EffectPort]) -> EffectPortClassification:
    """A new table with extra rows. Never mutates the receiver."""
    return EffectPortClassification(ports=(*self.ports, *tuple(ports)))

EncryptionPort

Bases: Protocol

Seal and unseal recording bytes.

ErasureParticipantRegistry 

ErasureParticipantRegistry(registry_id: str = 'default')

The set of stores the tenant-erasure path must sweep.

Source code in src/symfonic/services/shadow/privacy.py 
77
78
79
def __init__(self, registry_id: str = "default") -> None:
    self._registry_id = registry_id
    self._participants: dict[str, TenantErasureParticipant] = {}

erase_all async 

erase_all(tenant_id: str) -> dict[str, int]

Sweep every participant. Returns participant_id -> rows removed.

Source code in src/symfonic/services/shadow/privacy.py 
124
125
126
async def erase_all(self, tenant_id: str) -> dict[str, int]:
    """Sweep every participant. Returns ``participant_id -> rows removed``."""
    return (await self.sweep(tenant_id)).counts

sweep async 

sweep(tenant_id: str) -> ErasureSweep

Erase from every participant, surviving one that fails.

A participant is somebody else's store — a remote backend having a bad minute must not abort the tenant's erasure, skip every participant sorted after it, and lose the completion trail. Each failure is logged and reported by id instead.

Source code in src/symfonic/services/shadow/privacy.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def sweep(self, tenant_id: str) -> ErasureSweep:
    """Erase from every participant, surviving one that fails.

    A participant is somebody else's store — a remote backend having a bad
    minute must not abort the tenant's erasure, skip every participant
    sorted after it, and lose the completion trail. Each failure is logged
    and reported by id instead.
    """
    result = ErasureSweep()
    for participant in self.participants():
        try:
            result.counts[participant.participant_id] = await participant.erase_tenant(
                tenant_id
            )
        except Exception as exc:  # noqa: BLE001 - erasure must not abort mid-sweep
            logger.warning(
                "Erase: participant %s failed for tenant %s",
                participant.participant_id,
                tenant_id,
                exc_info=True,
            )
            result.failed[participant.participant_id] = f"{type(exc).__name__}: {exc}"
    return result

ErasureSweep dataclass 

ErasureSweep(counts: dict[str, int] = dict(), failed: dict[str, str] = dict())

The outcome of one sweep: what was erased, and what refused to be.

A sweep is deliberately not all-or-nothing. Erasure is the one operation where a partial success must still be reported rather than rolled back — the rows that went are gone — so a participant that raises is recorded by id and the sweep continues to the rest.

ExtensionRecord dataclass 

ExtensionRecord(extension_id: str, kind: ExtensionKind, origin: ExtensionOrigin, trust: TrustClass, reason: str, proof: ConstructionProof | None = None, approval: ReviewerApproval | None = None)

What the registry knows about one extension.

shadowable property 

shadowable: bool

Opaque extensions are non-shadowable and non-replayable.

ExtensionTrustRegistry 

ExtensionTrustRegistry(*, classified_ports: Iterable[str] = ())

Assigns and enforces trust classes. Default-deny by construction.

Source code in src/symfonic/services/shadow/trust.py 
111
112
113
114
def __init__(self, *, classified_ports: Iterable[str] = ()) -> None:
    self._classified = frozenset(classified_ports)
    self._records: dict[str, ExtensionRecord] = {}
    self._demoted: dict[str, str] = {}

classify_port_mediated 

classify_port_mediated(extension_id: str, kind: ExtensionKind, *, proof: ConstructionProof, approval: ReviewerApproval, origin: ExtensionOrigin = ExtensionOrigin.FRAMEWORK) -> ExtensionRecord

Promote to port-mediated. Refuses everything short of the bar.

Source code in src/symfonic/services/shadow/trust.py 
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
166
def classify_port_mediated(
    self,
    extension_id: str,
    kind: ExtensionKind,
    *,
    proof: ConstructionProof,
    approval: ReviewerApproval,
    origin: ExtensionOrigin = ExtensionOrigin.FRAMEWORK,
) -> ExtensionRecord:
    """Promote to port-mediated. Refuses everything short of the bar."""
    if extension_id in self._demoted:
        raise TrustDeclarationError(
            f"{extension_id!r} was demoted to opaque "
            f"({self._demoted[extension_id]}); it cannot be re-promoted "
            "without a fresh construction proof under a new id"
        )
    self._require_independent_proof(extension_id, proof)
    self._require_construction(extension_id, proof)
    self._require_named_reviewer(extension_id, approval)
    record = ExtensionRecord(
        extension_id=extension_id,
        kind=kind,
        origin=origin,
        trust=TrustClass.PORT_MEDIATED,
        reason=f"{proof.method.value} verified by {proof.verified_by}",
        proof=proof,
        approval=approval,
    )
    self._records[extension_id] = record
    return record

demote 

demote(extension_id: str, reason: str) -> ExtensionRecord

Force an extension to opaque and bar re-promotion under this id.

Source code in src/symfonic/services/shadow/trust.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def demote(self, extension_id: str, reason: str) -> ExtensionRecord:
    """Force an extension to opaque and bar re-promotion under this id."""
    previous = self.record_of(extension_id)
    record = ExtensionRecord(
        extension_id=extension_id,
        kind=previous.kind,
        origin=previous.origin,
        trust=TrustClass.OPAQUE,
        reason=reason,
        proof=previous.proof,
        approval=previous.approval,
    )
    self._records[extension_id] = record
    self._demoted[extension_id] = reason
    return record

record_of 

record_of(extension_id: str) -> ExtensionRecord

The record, or a synthesized opaque one. Never raises for unknown.

Source code in src/symfonic/services/shadow/trust.py 
217
218
219
220
221
222
223
224
225
226
227
228
def record_of(self, extension_id: str) -> ExtensionRecord:
    """The record, or a synthesized opaque one. Never raises for unknown."""
    known = self._records.get(extension_id)
    if known is not None:
        return known
    return ExtensionRecord(
        extension_id=extension_id,
        kind=ExtensionKind.TOOL,
        origin=ExtensionOrigin.ADOPTER,
        trust=TrustClass.OPAQUE,
        reason="unregistered extension defaults to opaque",
    )

register 

register(extension_id: str, kind: ExtensionKind, *, origin: ExtensionOrigin = ExtensionOrigin.ADOPTER, reason: str = 'registered without a construction proof') -> ExtensionRecord

Register an extension as opaque. This is the only bulk entry point.

Source code in src/symfonic/services/shadow/trust.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def register(
    self,
    extension_id: str,
    kind: ExtensionKind,
    *,
    origin: ExtensionOrigin = ExtensionOrigin.ADOPTER,
    reason: str = "registered without a construction proof",
) -> ExtensionRecord:
    """Register an extension as opaque. This is the only bulk entry point."""
    record = ExtensionRecord(
        extension_id=extension_id,
        kind=kind,
        origin=origin,
        trust=TrustClass.OPAQUE,
        reason=reason,
    )
    self._records[extension_id] = record
    return record

FieldAllowlist 

FieldAllowlist(fields: Iterable[AllowedField])

Allowlist-first projection of a payload.

Source code in src/symfonic/services/shadow/redaction.py 
122
123
124
125
126
127
128
129
def __init__(self, fields: Iterable[AllowedField]) -> None:
    rows = tuple(fields)
    table: dict[str, AllowedField] = {}
    for row in rows:
        if row.path in table:
            raise ValueError(f"duplicate allowlist row for {row.path!r}")
        table[row.path] = row
    self._fields = table

HmacStreamCipher 

HmacStreamCipher(key: bytes, *, key_id: str = 'shadow-recording-key')

HMAC-SHA256 CTR keystream with encrypt-then-MAC.

Encryption and authentication use separately derived subkeys, and the MAC covers key id, nonce, and ciphertext, so neither a swapped nonce nor a relabelled key can be passed off as a valid record.

Source code in src/symfonic/services/shadow/crypto.py 
61
62
63
64
65
66
def __init__(self, key: bytes, *, key_id: str = "shadow-recording-key") -> None:
    if len(key) < 32:
        raise ValueError("recording-store key must be at least 32 bytes")
    self._key_id = key_id
    self._enc = hmac.new(key, b"symfonic/shadow/enc", hashlib.sha256).digest()
    self._mac = hmac.new(key, b"symfonic/shadow/mac", hashlib.sha256).digest()

IncompleteClassificationError

Bases: ConfigurationError

An effect family the architecture declares has no classified port.

MisdeclaredExtensionError

Bases: ShadowAbortedError

An extension declared port-mediated performed a direct effect.

The declaration was wrong; every suppression claim that depended on it is withheld, and the extension is demoted to opaque.

OpaqueExtensionError

Bases: ShadowAbortedError

An opaque tool, plugin, or contributed stage is non-shadowable.

Raised before the extension runs. Pretending an opaque extension's effects were suppressed is the failure mode this type exists to prevent.

PayloadMinimizer dataclass 

PayloadMinimizer(max_chars: int = 512, max_items: int = 32)

Payload minimization: cap string length and collection width.

ProofMethod

Bases: StrEnum

The only two ways an extension can be port-mediated by construction.

RecordComparator

Compares a recording against an effect-suppressed candidate run.

RecordedEvent dataclass 

RecordedEvent(seq: int, port_id: str, operation: str, request_digest: str, response: Any = None, family: str = '')

One non-idempotent port crossing, with the answer it produced.

RecordedResponder 

RecordedResponder(recording: Recording)

Answers a stubbed port call from the recording, in recorded order.

Answers for one key are consumed FIFO rather than looked up, because the recorded ports are non-idempotent by definition: the same request made twice may have produced two different answers, and replaying the second answer to the first call is a divergence the comparator cannot catch. Once a key's recorded answers are exhausted the responder refuses, so a replay that calls a port more times than the original is a refusal — never a silently repeated answer.

Source code in src/symfonic/services/shadow/replay.py 
46
47
48
49
def __init__(self, recording: Recording) -> None:
    self._queues = recording.response_queues()
    self._cursor: dict[tuple[str, str, str], int] = {}
    self._served: list[tuple[str, str, str]] = []

unserved 

unserved() -> tuple[tuple[str, str, str], ...]

Every recorded answer the replay never asked for, repeats included.

Source code in src/symfonic/services/shadow/replay.py 
55
56
57
58
59
60
61
def unserved(self) -> tuple[tuple[str, str, str], ...]:
    """Every recorded answer the replay never asked for, repeats included."""
    return tuple(
        key
        for key, answers in self._queues.items()
        for _ in range(len(answers) - self._cursor.get(key, 0))
    )

Recorder 

Recorder(*, policy: CaptureAdmissionPolicy, store: RecordingStore, actor: str, mode: RecordingMode = RecordingMode.SYNTHETIC, clock: Callable[[], datetime] | None = None)

Binds the admission policy to the governed store.

Source code in src/symfonic/services/shadow/recorder.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def __init__(
    self,
    *,
    policy: CaptureAdmissionPolicy,
    store: RecordingStore,
    actor: str,
    mode: RecordingMode = RecordingMode.SYNTHETIC,
    clock: Callable[[], datetime] | None = None,
) -> None:
    self._policy = policy
    self._store = store
    self._actor = actor
    self._mode = mode
    self._clock = clock or (lambda: datetime.now(UTC))

begin 

begin(invocation_id: str, tenant_id: str, payload: dict[str, Any] | None = None) -> CaptureOutcome

Admit or refuse. A refusal means the invocation proceeds unrecorded.

Source code in src/symfonic/services/shadow/recorder.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def begin(
    self, invocation_id: str, tenant_id: str, payload: dict[str, Any] | None = None
) -> CaptureOutcome:
    """Admit or refuse. A refusal means the invocation proceeds unrecorded."""
    request = CaptureRequest(
        invocation_id=invocation_id,
        tenant_id=tenant_id,
        payload=payload or {},
        at=self._clock(),
    )
    decision = self._policy.decide(request)
    if not decision.admitted:
        return CaptureOutcome(admitted=False, reason=decision.reason)
    session = RecordingSession(
        # Tenant-scoped by construction: adopter invocation ids are only
        # unique within a tenant (per-tenant counters and retried ids both
        # collide across tenants), and a colliding id would put two
        # tenants' governed data under one key.
        recording_id=f"rec-{tenant_id}-{invocation_id}",
        tenant_id=tenant_id,
        mode=self._mode,
        decision=decision,
        captured_at=request.at or self._clock(),
        project_response=self._policy.project_response,
    )
    return CaptureOutcome(admitted=True, reason=decision.reason, session=session)

commit 

commit(session: RecordingSession) -> StoredRecording | None

Seal and store. Any capture failure leaves the invocation unrecorded.

The refusal is deliberately broad. Serialization is the obvious way a commit fails for a reason the store never sees — an allowlisted leaf holding a datetime makes json.dumps raise TypeError, not RecordingStoreError — and a cipher or backend can fail its own way too. All of them are capture failures, and the contract is that capture failure never reaches the invocation being served. The double commit above still raises: that is a caller bug, not a capture failure.

Source code in src/symfonic/services/shadow/recorder.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def commit(self, session: RecordingSession) -> StoredRecording | None:
    """Seal and store. Any capture failure leaves the invocation unrecorded.

    The refusal is deliberately broad. Serialization is the obvious way a
    commit fails for a reason the store never sees — an allowlisted leaf
    holding a ``datetime`` makes ``json.dumps`` raise ``TypeError``, not
    ``RecordingStoreError`` — and a cipher or backend can fail its own way
    too. All of them are *capture* failures, and the contract is that
    capture failure never reaches the invocation being served. The double
    commit above still raises: that is a caller bug, not a capture failure.
    """
    if session.committed:
        raise RecordingStoreError(
            f"recording {session.recording_id!r} was already committed; a "
            "second commit would duplicate governed tenant data"
        )
    try:
        stored = self._store.put(session.build(), actor=self._actor)
    except Exception:  # noqa: BLE001 - capture must never raise into the caller
        logger.warning(
            "Recording %s could not be committed; the invocation proceeds "
            "unrecorded",
            session.recording_id,
            exc_info=True,
        )
        return None
    session.committed = True
    return stored

Recording dataclass 

Recording(recording_id: str, tenant_id: str, mode: RecordingMode, captured_at: datetime, events: tuple[RecordedEvent, ...] = (), extensions: tuple[str, ...] = (), payload: Mapping[str, Any] = dict(), metadata: Mapping[str, Any] = dict())

A tenant-scoped, redacted trace of one invocation.

from_bytes classmethod 

from_bytes(blob: bytes) -> Recording

Data-only decode. json.loads instantiates nothing but builtins.

Source code in src/symfonic/services/shadow/recording.py 
143
144
145
146
@classmethod
def from_bytes(cls, blob: bytes) -> Recording:
    """Data-only decode. ``json.loads`` instantiates nothing but builtins."""
    return cls.from_dict(json.loads(blob.decode("utf-8")))

index 

index() -> dict[tuple[str, str, str], Any]

(port, operation, request_digest) -> last response — lossy.

Kept for callers that only need to know whether a key was recorded. Replay uses :meth:response_queues, which does not collapse repeats.

Source code in src/symfonic/services/shadow/recording.py 
101
102
103
104
105
106
107
def index(self) -> dict[tuple[str, str, str], Any]:
    """``(port, operation, request_digest) -> last response`` — lossy.

    Kept for callers that only need to know *whether* a key was recorded.
    Replay uses :meth:`response_queues`, which does not collapse repeats.
    """
    return {event.key(): event.response for event in self.events}

response_queues 

response_queues() -> dict[tuple[str, str, str], list[Any]]

(port, operation, request_digest) -> every answer, in order.

A list, not a single value, because the ports a recording exists to stub are the non-idempotent ones: a tool called twice with the same arguments legitimately answers differently the second time. Collapsing those into one entry would serve the last answer to both calls and steer the replacement down a path the original never took — the exact duplication-of-effect the comparator cannot see, since it diffs requests rather than responses.

Source code in src/symfonic/services/shadow/recording.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def response_queues(self) -> dict[tuple[str, str, str], list[Any]]:
    """``(port, operation, request_digest) -> every answer, in order``.

    A list, not a single value, because the ports a recording exists to
    stub are the *non-idempotent* ones: a tool called twice with the same
    arguments legitimately answers differently the second time. Collapsing
    those into one entry would serve the last answer to both calls and
    steer the replacement down a path the original never took — the exact
    duplication-of-effect the comparator cannot see, since it diffs
    requests rather than responses.
    """
    queues: dict[tuple[str, str, str], list[Any]] = {}
    for event in sorted(self.events, key=lambda e: e.seq):
        queues.setdefault(event.key(), []).append(event.response)
    return queues

RecordingAccessError

Bases: ShadowError

TM-29a — an actor read (or tried to read) beyond its authorization.

RecordingMode

Bases: StrEnum

Whether a recording holds real tenant traffic or fabricated traffic.

The distinction is load-bearing: SEC-PRIV-5 gates PRODUCTION behind verified privacy-deletion wiring, while SYNTHETIC fixtures must stay usable in CI on day one.

RecordingSession dataclass 

RecordingSession(recording_id: str, tenant_id: str, mode: RecordingMode, decision: CaptureDecision, captured_at: datetime, events: list[RecordedEvent] = list(), extensions: list[str] = list(), committed: bool = False, project_response: Callable[[Any], ResponseProjection] | None = None, response_dropped: list[str] = list(), response_redacted: list[str] = list())

Accumulates events for one invocation, then commits once.

A session is normally built by :class:Recorder, which binds the admission policy's response projection to it. A session built without one still governs its answers — the fallback scrubs credential-shaped keys and minimizes — because "nobody wired a projector" must not mean "write the provider's answer verbatim".

observe 

observe(port_id: str, operation: str, request: Any, response: Any, *, family: str = '') -> RecordedEvent

Record one non-idempotent port crossing and its governed answer.

The request never enters the recording — only its digest — and the answer goes through the same governance as the payload before any of it is serialized.

Source code in src/symfonic/services/shadow/recorder.py 
 80
 81
 82
 83
 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
def observe(
    self,
    port_id: str,
    operation: str,
    request: Any,
    response: Any,
    *,
    family: str = "",
) -> RecordedEvent:
    """Record one non-idempotent port crossing and its governed answer.

    The request never enters the recording — only its digest — and the
    answer goes through the same governance as the payload before any of
    it is serialized.
    """
    project = self.project_response or _scrubbed_response
    projection = project(response)
    for path in projection.dropped_fields:
        if path not in self.response_dropped:
            self.response_dropped.append(path)
    for path in projection.redacted_fields:
        if path not in self.response_redacted:
            self.response_redacted.append(path)
    event = RecordedEvent(
        seq=len(self.events) + 1,
        port_id=port_id,
        operation=operation,
        request_digest=digest_of(port_id, operation, request),
        response=projection.value,
        family=family,
    )
    self.events.append(event)
    return event

RecordingStore 

RecordingStore(*, cipher: EncryptionPort | None = None, retention: timedelta = timedelta(days=7), grants: tuple[AccessGrant, ...] = (), deletion_wiring: DeletionWiringAttestation | None = None, clock: Callable[[], datetime] | None = None)

An in-memory governed store. The seams are the cipher and the clock.

Source code in src/symfonic/services/shadow/store.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    *,
    cipher: EncryptionPort | None = None,
    retention: timedelta = timedelta(days=7),
    grants: tuple[AccessGrant, ...] = (),
    deletion_wiring: DeletionWiringAttestation | None = None,
    clock: Callable[[], datetime] | None = None,
) -> None:
    if retention <= timedelta(0):
        raise ValueError("the recording store must be retention-bounded")
    self._cipher = cipher or RefusingCipher()
    self._retention = retention
    self._grants = {grant.actor: grant for grant in grants}
    self._wiring = deletion_wiring
    self._clock = clock or (lambda: datetime.now(UTC))
    self._rows: dict[str, StoredRecording] = {}
    self._audit: list[AccessAuditEntry] = []

attest_deletion_wiring 

attest_deletion_wiring(attestation: DeletionWiringAttestation) -> None

Accept an attestation. Refuses one that does not prove erasure.

Source code in src/symfonic/services/shadow/store.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def attest_deletion_wiring(self, attestation: DeletionWiringAttestation) -> None:
    """Accept an attestation. Refuses one that does not prove erasure."""
    if attestation.participant_id != self.participant_id:
        raise RecordingStoreError(
            f"attestation names participant {attestation.participant_id!r}, "
            f"not {self.participant_id!r}"
        )
    if not attestation.verified:
        raise RecordingStoreError(
            "the deletion-wiring attestation did not prove erasure "
            f"(seeded={attestation.seeded}, residual={attestation.residual})"
        )
    self._wiring = attestation

RecordingStoreError

Bases: ShadowError

The recording store refused a write it cannot govern.

RefusingCipher

The absence of a configured cipher, made explicit and loud.

ReplayRunner 

ReplayRunner(harness: ShadowHarness)

Replays one recording through the shadow harness.

Source code in src/symfonic/services/shadow/replay.py 
80
81
def __init__(self, harness: ShadowHarness) -> None:
    self._harness = harness

replayable staticmethod 

replayable(result: ShadowRunResult) -> bool

A replay is usable as comparison input only when nothing leaked.

Source code in src/symfonic/services/shadow/replay.py 
105
106
107
108
@staticmethod
def replayable(result: ShadowRunResult) -> bool:
    """A replay is usable as comparison input only when nothing leaked."""
    return result.status is ShadowStatus.COMPLETED and result.suppression_claim

ResponseProjection dataclass 

ResponseProjection(value: Any, dropped_fields: tuple[str, ...] = (), redacted_fields: tuple[str, ...] = ())

What survives of one recorded port answer, and what was removed.

ReviewerApproval dataclass 

ReviewerApproval(reviewer: str, capability: str, approved_at: str, statement: str = '')

A named human from the owning capability signing off on a promotion.

SealedPayload dataclass 

SealedPayload(key_id: str, nonce: bytes, ciphertext: bytes, tag: bytes)

Ciphertext plus everything needed to verify and open it. No plaintext.

ShadowAbortedError

Bases: ShadowError

The shadow run stopped before (or instead of) performing an effect.

Raised on its own when a run is aborted for a reason that has no more specific type; the fail-closed subclasses below are the usual cause.

ShadowContext 

ShadowContext(*, gateway: ShadowEffectGateway, state: SuppressedStateWriter, trust: ExtensionTrustRegistry, watch: bool)

What the body of a shadow run is given. Nothing else is reachable.

Source code in src/symfonic/services/shadow/context.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(
    self,
    *,
    gateway: ShadowEffectGateway,
    state: SuppressedStateWriter,
    trust: ExtensionTrustRegistry,
    watch: bool,
) -> None:
    self.gateway = gateway
    self.state = state
    self._trust = trust
    self._watch = watch
    self._executed: list[str] = []
    self._defects: list[str] = []

claim_defects property 

claim_defects: tuple[str, ...]

Reasons the suppression claim is void, independent of exceptions.

Detection must not depend on the exception reaching the harness: a body with a bare except would otherwise buy back the claim the sentinel just refused. Every detection path records here first and raises second.

call_extension 

call_extension(extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any

Run a synchronous extension only if its trust class permits it.

Opaque extensions raise before fn is touched. Port-mediated ones run inside the sentinel, so a wrong declaration becomes a MisdeclaredExtensionError instead of a silent effect. An extension that returns an awaitable is refused: its body would run after the probes came down. Use :meth:call_extension_async for those.

Source code in src/symfonic/services/shadow/context.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def call_extension(
    self, extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any
) -> Any:
    """Run a synchronous extension only if its trust class permits it.

    Opaque extensions raise *before* ``fn`` is touched. Port-mediated ones
    run inside the sentinel, so a wrong declaration becomes a
    ``MisdeclaredExtensionError`` instead of a silent effect. An extension
    that returns an awaitable is refused: its body would run after the
    probes came down. Use :meth:`call_extension_async` for those.
    """
    self._trust.require_shadowable(extension_id)
    if not self._watch:
        self._executed.append(extension_id)
        return self._refuse_awaitable(extension_id, fn(*args, **kwargs))
    sentinel = DirectEffectSentinel(subject=extension_id, block=True)
    try:
        with sentinel.watching():
            result = fn(*args, **kwargs)
    except MisdeclaredExtensionError:
        self._witness(extension_id, sentinel)
        raise
    self._assert_clean(extension_id, sentinel)
    result = self._refuse_awaitable(extension_id, result)
    self._executed.append(extension_id)
    return result

call_extension_async async 

call_extension_async(extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any

Await an async extension inside the sentinel scope.

In an async-first framework the common tool shape is a coroutine function; calling one only builds the coroutine, so a synchronous call_extension would uninstall the probes before a single line of the extension ran. This awaits under the probes, which is the whole point of watching.

Source code in src/symfonic/services/shadow/context.py 
 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
async def call_extension_async(
    self, extension_id: str, fn: Callable[..., Any], *args: Any, **kwargs: Any
) -> Any:
    """Await an async extension *inside* the sentinel scope.

    In an async-first framework the common tool shape is a coroutine
    function; calling one only builds the coroutine, so a synchronous
    ``call_extension`` would uninstall the probes before a single line of
    the extension ran. This awaits under the probes, which is the whole
    point of watching.
    """
    self._trust.require_shadowable(extension_id)
    if not self._watch:
        self._executed.append(extension_id)
        return await _resolve(fn(*args, **kwargs))
    sentinel = DirectEffectSentinel(subject=extension_id, block=True)
    try:
        with sentinel.watching():
            result = await _resolve(fn(*args, **kwargs))
    except MisdeclaredExtensionError:
        self._witness(extension_id, sentinel)
        raise
    self._assert_clean(extension_id, sentinel)
    self._executed.append(extension_id)
    return result

ShadowDisposition

Bases: StrEnum

What shadow mode does with a classified port. There is no third option.

DENY refuses the call. STUB answers it deterministically without reaching anything external. "Let it through" is deliberately not representable — that is the whole point of classifying.

ShadowEffectDenied

Bases: ShadowError

A classified DENY port refused an effect inside a shadow run.

Not an abort: denial is the port doing its job. The caller decides whether the denial is fatal to the scenario it was exercising.

ShadowEffectGateway 

ShadowEffectGateway(classification: EffectPortClassification, *, run_id: str, responder: StubResponder | None = None, ledger: EffectLedger | None = None)

Routes port calls through the classification; records every attempt.

Source code in src/symfonic/services/shadow/gateway.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    classification: EffectPortClassification,
    *,
    run_id: str,
    responder: StubResponder | None = None,
    ledger: EffectLedger | None = None,
) -> None:
    self._classification = classification
    self._run_id = run_id
    self._responder: StubResponder = responder or DeterministicStub()
    self._ledger = ledger if ledger is not None else EffectLedger()
    self._abort_reason: str | None = None

abort 

abort(reason: str) -> None

Poison the gateway. Every later call refuses, whatever its port.

Source code in src/symfonic/services/shadow/gateway.py 
86
87
88
89
def abort(self, reason: str) -> None:
    """Poison the gateway. Every later call refuses, whatever its port."""
    if self._abort_reason is None:
        self._abort_reason = reason

invoke 

invoke(port_id: str, operation: str, payload: Any = None) -> Any

Deny, stub, or abort. Never performs the effect.

Source code in src/symfonic/services/shadow/gateway.py 
 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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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
def invoke(self, port_id: str, operation: str, payload: Any = None) -> Any:
    """Deny, stub, or abort. Never performs the effect."""
    if self._abort_reason is not None:
        raise UnclassifiedEffectError(
            f"shadow run {self._run_id!r} was already aborted "
            f"({self._abort_reason}); it cannot serve {port_id}.{operation}"
        )
    request_digest = digest_of(port_id, operation, payload)
    if not self._classification.knows(port_id):
        self._ledger.append(
            EffectAttempt(
                seq=self._ledger.next_seq(),
                port_id=port_id,
                operation=operation,
                request_digest=request_digest,
                outcome=EffectOutcome.UNCLASSIFIED,
                detail="not in the effect-port classification",
            )
        )
        self.abort(f"unclassified effect port {port_id!r}")
        # Delegate the message to the classification so there is exactly
        # one place that explains what an unclassified port means.
        self._classification.classify(port_id)
    port = self._classification.classify(port_id)
    if port.disposition is ShadowDisposition.DENY:
        self._ledger.append(
            EffectAttempt(
                seq=self._ledger.next_seq(),
                port_id=port_id,
                operation=operation,
                request_digest=request_digest,
                outcome=EffectOutcome.DENIED,
                family=port.family,
                externally_visible=port.externally_visible,
                detail=port.rationale,
            )
        )
        raise ShadowEffectDenied(
            f"{port_id}.{operation} is denied in shadow mode: {port.rationale}"
        )
    try:
        answer = self._responder.respond(port, operation, payload)
    except LookupError as exc:
        self._ledger.append(
            EffectAttempt(
                seq=self._ledger.next_seq(),
                port_id=port_id,
                operation=operation,
                request_digest=request_digest,
                outcome=EffectOutcome.DENIED,
                family=port.family,
                externally_visible=port.externally_visible,
                detail="no recorded answer",
            )
        )
        raise ShadowEffectDenied(
            f"{port_id}.{operation} has no recorded answer to replay; the "
            "gateway refuses rather than calling the real port"
        ) from exc
    self._ledger.append(
        EffectAttempt(
            seq=self._ledger.next_seq(),
            port_id=port_id,
            operation=operation,
            request_digest=request_digest,
            outcome=EffectOutcome.STUBBED,
            family=port.family,
            externally_visible=port.externally_visible,
        )
    )
    return answer

ShadowError

Bases: SymfonicError

Root of the shadow/replay taxonomy. Never raised directly.

ShadowHarness 

ShadowHarness(*, trust: ExtensionTrustRegistry, classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION)

Builds and supervises shadow runs.

Source code in src/symfonic/services/shadow/harness.py 
101
102
103
104
105
106
107
108
109
def __init__(
    self,
    *,
    trust: ExtensionTrustRegistry,
    classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION,
) -> None:
    classification.assert_exhaustive()
    self._classification = classification
    self._trust = trust

ShadowRunResult dataclass 

ShadowRunResult(run_id: str, tenant_id: str, status: ShadowStatus, ledger: EffectLedger, state_intents: tuple[StateWriteIntent, ...], extensions_executed: tuple[str, ...], body_executed: bool, suppression_claim: bool, claim_defects: tuple[str, ...] = (), abort_reason: str | None = None, error: BaseException | None = None, value: Any = None)

The reviewable artifact of one shadow run.

ShadowRunSpec dataclass 

ShadowRunSpec(run_id: str, tenant_id: str, extensions: tuple[str, ...] = (), baseline_state: Mapping[str, Any] = dict(), backing_state: Mapping[str, Any] | None = None, responder: StubResponder | None = None, watch_extensions: bool = True)

What a shadow run is allowed to be before it starts.

StateWriteIntent dataclass 

StateWriteIntent(seq: int, key: str, value_digest: str, deleted: bool = False)

A write the replacement would have performed. Never applied.

StoredRecording dataclass 

StoredRecording(recording_id: str, tenant_id: str, mode: RecordingMode, captured_at: datetime, expires_at: datetime, digest: str, sealed: SealedPayload)

What the store actually holds: metadata plus ciphertext.

StubResponder

Bases: Protocol

Answers a stubbed port call without reaching anything external.

respond 

respond(port: EffectPort, operation: str, payload: Any) -> Any

The stubbed answer. Raises LookupError when it has none.

Source code in src/symfonic/services/shadow/gateway.py 
27
28
29
def respond(self, port: EffectPort, operation: str, payload: Any) -> Any:
    """The stubbed answer. Raises ``LookupError`` when it has none."""
    ...

SuppressedStateWriter dataclass 

SuppressedStateWriter(baseline: Mapping[str, Any] = dict(), backing: Mapping[str, Any] | None = None)

Copy-on-write overlay over a read-only baseline.

assert_backing_untouched 

assert_backing_untouched() -> None

Prove the replacement's real store is byte-identical to before.

Source code in src/symfonic/services/shadow/state.py 
90
91
92
93
94
95
96
def assert_backing_untouched(self) -> None:
    """Prove the replacement's real store is byte-identical to before."""
    if self._snapshot() != self._backing_digest:
        raise ShadowAbortedError(
            "the replacement state store changed during a shadow run; the "
            "suppression claim is withheld"
        )

TenantCaptureGrant dataclass 

TenantCaptureGrant(tenant_id: str, authorized_by: str, purpose: str, expires_at: datetime, sampling_rate: float = 1.0)

Per-tenant capture eligibility. Absence of a grant means "no".

TenantErasureParticipant

Bases: Protocol

A store that holds tenant-scoped data and can be erased on request.

count_for_tenant async 

count_for_tenant(tenant_id: str) -> int

How many rows this participant still holds for tenant_id.

Source code in src/symfonic/services/shadow/privacy.py 
69
70
71
async def count_for_tenant(self, tenant_id: str) -> int:
    """How many rows this participant still holds for ``tenant_id``."""
    ...

erase_tenant async 

erase_tenant(tenant_id: str) -> int

Remove every row owned by tenant_id; return how many went.

Source code in src/symfonic/services/shadow/privacy.py 
65
66
67
async def erase_tenant(self, tenant_id: str) -> int:
    """Remove every row owned by ``tenant_id``; return how many went."""
    ...

TrustApproval dataclass 

TrustApproval(extension_id: str, capability: str, reviewer: str, approved_at: str, proof_method: str)

A port-mediated assignment, its reviewer, and its capability.

TrustDeclarationError

Bases: ConfigurationError

A trust-class assignment does not meet the port-mediated bar.

A ConfigurationError on purpose: a self-declared or unreviewed port-mediated claim is a misconfiguration of the evidence pipeline, and adopters already catch that taxonomy.

UnclassifiedEffectError

Bases: ShadowAbortedError

SEC-FCP-5 — an effect was attempted through a port nobody classified.

Fail-closed by construction: an unclassified port is not "probably safe", it is a hole in the evidence, and the run that found it is void.

UndigestibleValueError

Bases: ShadowError

A value's identity cannot be established, so no digest is produced.

The digest is the comparator's only notion of "same request". A value the canonicaliser cannot see inside would digest identically to every other instance of its type, and the comparator would read that collision as parity. Refusing is the fail-closed answer: an unrecorded or aborted comparison is recoverable, a false parity claim is not.

canonical_json 

canonical_json(value: Any) -> str

A stable JSON rendering: sorted keys, no whitespace drift, no NaN.

Source code in src/symfonic/services/shadow/digest.py 
158
159
160
def canonical_json(value: Any) -> str:
    """A stable JSON rendering: sorted keys, no whitespace drift, no NaN."""
    return json.dumps(_plain(value), sort_keys=True, separators=(",", ":"), allow_nan=False)

digest_of 

digest_of(*parts: Any) -> str

A sha256 over the canonical rendering of every part, in order.

Source code in src/symfonic/services/shadow/digest.py 
163
164
165
166
167
168
169
def digest_of(*parts: Any) -> str:
    """A sha256 over the canonical rendering of every part, in order."""
    hasher = hashlib.sha256()
    for part in parts:
        hasher.update(canonical_json(part).encode("utf-8"))
        hasher.update(b"\x1f")
    return hasher.hexdigest()

scrub_credential_keys 

scrub_credential_keys(value: Any, path: str = '', dropped: list[str] | None = None) -> Any

Drop credential-shaped keys from any nested structure, at every depth.

The allowlist walk covers mappings, but an allowlisted leaf can still be a list of mappings (a tool result, a message array), and a recorded port answer is not walked by the allowlist at all. SEC-CRED-2 has to hold for those too, so this is the one credential check both paths call.

Source code in src/symfonic/services/shadow/redaction.py 
 83
 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
def scrub_credential_keys(
    value: Any, path: str = "", dropped: list[str] | None = None
) -> Any:
    """Drop credential-shaped keys from *any* nested structure, at every depth.

    The allowlist walk covers mappings, but an allowlisted leaf can still be a
    list of mappings (a tool result, a message array), and a recorded port
    answer is not walked by the allowlist at all. SEC-CRED-2 has to hold for
    those too, so this is the one credential check both paths call.
    """
    if isinstance(value, Mapping):
        out: dict[str, Any] = {}
        for raw_key, item in value.items():
            key = str(raw_key)
            child = f"{path}.{key}" if path else key
            if DEFAULT_CREDENTIAL_KEY_PATTERN.search(key):
                if dropped is not None:
                    dropped.append(child)
                continue
            out[key] = scrub_credential_keys(item, child, dropped)
        return out
    if isinstance(value, list | tuple):
        return [
            scrub_credential_keys(item, f"{path}[{index}]", dropped)
            for index, item in enumerate(value)
        ]
    return value

verify_deletion_wiring async 

verify_deletion_wiring(participant: TenantErasureParticipant, *, registry: ErasureParticipantRegistry = PRIVACY_DELETION_PARTICIPANTS, probe_tenant: str = 'deletion-wiring-probe', seed: int = 1) -> DeletionWiringAttestation

Run a real erase round-trip and attest to the result.

Raises when the participant is not registered — an attestation for a store the deletion path cannot reach would be worse than none at all.

Source code in src/symfonic/services/shadow/privacy.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
async def verify_deletion_wiring(
    participant: TenantErasureParticipant,
    *,
    registry: ErasureParticipantRegistry = PRIVACY_DELETION_PARTICIPANTS,
    probe_tenant: str = "deletion-wiring-probe",
    seed: int = 1,
) -> DeletionWiringAttestation:
    """Run a real erase round-trip and attest to the result.

    Raises when the participant is not registered — an attestation for a store
    the deletion path cannot reach would be worse than none at all.
    """
    if not registry.holds(participant.participant_id):
        raise RecordingStoreError(
            f"{participant.participant_id!r} is not registered with the "
            f"{registry.registry_id!r} deletion path; wiring cannot be attested"
        )
    seeded = await participant.count_for_tenant(probe_tenant)
    if seeded < seed:
        raise RecordingStoreError(
            f"the deletion-wiring probe needs at least {seed} seeded row(s) for "
            f"{probe_tenant!r}; found {seeded}. Attesting against an empty store "
            "would prove nothing."
        )
    await registry.erase_all(probe_tenant)
    residual = await participant.count_for_tenant(probe_tenant)
    return DeletionWiringAttestation(
        participant_id=participant.participant_id,
        registry_id=registry.registry_id,
        verified_at=datetime.now(UTC),
        probe_tenant=probe_tenant,
        seeded=seeded,
        residual=residual,
    )