symfonic.services.shadow.ports

ports

The declared port of the shadow / cutover service (LAY-ADR §2, port).

Same cell, same rule as :mod:symfonic.services.models.ports: facade-compiler → runtime-service is port, so the per-capability cutover switchboard in :mod:symfonic.agent.cutover may consult this service only through a declared narrow interface. Before TA2.1 it imported :mod:~symfonic.services.shadow.cutover, :mod:~symfonic.services.shadow.errors and :mod:~symfonic.services.shadow.trust directly — three internal modules of a twenty-two-module service package.

The names below are exactly what the switchboard needs to answer "may this capability leave shadow mode, and on which path" — the criteria table, the evidence recorder, the trust registry, and the one error the refusal raises. Nothing here is defined locally; every name is re-exported from its owning module.

PRIVACY_DELETION_PARTICIPANTS is here for a different importer. :mod:symfonic.agent.fastapi is ruled platform, whose runtime-service cell is yes — unrestricted — so tenant_privacy_router could reach :mod:symfonic.services.shadow.privacy directly and no rule would fire. It goes through this port anyway: the matrix leaving an edge unchecked is not the same as the edge being sound, and the rest of symfonic.agent was reworked onto declared ports. test_the_facade_reaches_runtime_services_only_through_a_port holds the whole symfonic.agent tree — fastapi included — to that rule.

CutoverCriteriaRecorder

CutoverCriteriaRecorder(trust: ExtensionTrustRegistry)

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

Source code in src/symfonic/services/shadow/cutover.py

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.

ExtensionTrustRegistry

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

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

Source code in src/symfonic/services/shadow/trust.py

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

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

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

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

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