Skip to content

symfonic.services.shadow.recording

recording

The recording value objects, and their data-only serialization.

AS-ING-4 applies to the replay read-back path: recorded bytes are untrusted input on read (TM-29f). So a recording serializes to and from plain JSON — no pickle, no object hooks, no type resolution — and every field is rebuilt by explicit construction rather than by handing a decoder a class to instantiate.

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.

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

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.