symfonic.services.shadow.admission

admission

The capture-admission policy (threat-model §6.3).

Recording a tenant's traffic for engineering replay is a distinct processing purpose, lawful here only under §6's controls. This module is where those controls become one decision: eligibility and authorization per tenant, field allowlisting with classification-aware redaction, payload minimization and sampling, and a configured cipher.

decide never raises. Every failure mode ends in admitted=False with a reason, because the required behaviour is the invocation proceeds unrecorded — a capture policy that could take down a tenant's request would be a worse bug than the one it prevents. require is the raising variant for callers that are recording deliberately (a test, a fixture build).

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

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

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.

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.

ResponseProjection dataclass

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

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

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".