symfonic.services.shadow.recorder

recorder

Recording capture — admission first, bytes second.

The recorder is deliberately shaped so that not recording is the easy path. begin returns a session or None; a caller that ignores the return value records nothing, which is the safe failure. Nothing here can raise into the invocation being served: a refusal is a CaptureOutcome 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.

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

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

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

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

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

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