Skip to content

symfonic.platform

platform

The platform layer: derive who is calling, decide whether it may proceed.

PLAT-ADR's one-sentence rule — the platform derives who the caller is, decides whether the request may proceed, and hands the request to the public facade; it implements nothing an invocation does — is what every module here is measured against.

T4.1.3 lands three of PLAT-ADR's five services, all transport-neutral (TRN-3):

  • :class:~symfonic.platform.scope.HeaderScopeResolver — the single derivation point (SCOPE-1), constructed and injected rather than registered in a module global, so two agents in one process no longer share one verifier;
  • :class:~symfonic.platform.budget.BudgetService — admission through the narrow BudgetCheck port with a typed refusal, retiring the str(exc).startswith("Budget exceeded:") branch and the raw-header read;
  • :class:~symfonic.platform.privacy.PrivacyService and :class:~symfonic.platform.erasure.ErasureSaga — export assembly and a durable, resumable, fail-closed erasure saga over the runtime-service SubjectDataStore registry.

Not here, by design: transport. No module in this package imports fastapi, constructs a response, or names a status code. Routers are T4.1.2's, and they will be mappers over these services (TRN-1) rather than the 3,759 lines of handler that reach into engine privates 54 times today.

AdmissionGate 

AdmissionGate(*, resolver: ScopeResolver, guard: SubjectAdmission | None = None, budget: BudgetService | None = None)

The three platform-side gates, composed in the one declared order.

Each collaborator is injected. There is no service locator and no lookup by name: a gate receives the two or three objects it uses, which is what keeps the order auditable — you can read it in the constructor, not chase it through a registry.

Source code in src/symfonic/platform/admission.py 
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    *,
    resolver: ScopeResolver,
    guard: SubjectAdmission | None = None,
    budget: BudgetService | None = None,
) -> None:
    self._resolver = resolver
    self._guard = guard
    self._budget = budget

admit async 

admit(credentials: RequestCredentials) -> AuthenticatedPrincipal

Run the gates in order and return the principal, or raise the first no.

Source code in src/symfonic/platform/admission.py 
57
58
59
60
61
62
63
64
async def admit(self, credentials: RequestCredentials) -> AuthenticatedPrincipal:
    """Run the gates in order and return the principal, or raise the first no."""
    principal = await self._resolver.resolve(credentials)
    if self._guard is not None:
        await self._guard.require_admission(principal.scope)
    if self._budget is not None:
        await self._budget.enforce(principal)
    return principal

AgentComposer

Bases: Protocol

Turns a scope and the process's shared resources into one Agent.

The adopter's composition root, expressed as a callable so the host never learns which capabilities a deployment folds. It receives the scope because capabilities are scope-bound at construction -- a memory capability is built for one tenant -- and that is precisely why an agent cannot be shared across scopes.

Synchronous on purpose. Composition assembles objects the host already opened; a composer that needed to await would be opening a resource of its own, which is the ownership this contract places elsewhere.

AgentHost 

AgentHost(*, composer: Any, resources: Any, open_resources: Any = None, continuation_factory: Any = None)

Owns the process's shared resources and the registry over them.

Built once by a composition root and closed once. open_resources is a separate callable rather than a method on the resources object because opening is not part of :class:~symfonic.platform.host.SharedResources: an adopter may hand over something already open, and requiring an aopen would make that a lie.

Source code in src/symfonic/platform/agent_host.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def __init__(
    self,
    *,
    composer: Any,
    resources: Any,
    open_resources: Any = None,
    continuation_factory: Any = None,
) -> None:
    self._resources = resources
    self._open = open_resources
    self._started = False
    self._closed = False
    self.registry = ScopedAgentRegistry(composer=composer, resources=resources)
    self._continuation = (
        None if continuation_factory is None else continuation_factory(self.agent_for)
    )

aclose async 

aclose() -> None

Close the registry, then the resources, exactly once.

In that order: an agent holding a closed pool is a worse failure than a pool held a moment longer, because the first surfaces mid-turn and the second surfaces nowhere.

Resources are closed only if they were opened. A host whose construction failed before start is still closed by the handler, and closing what was never opened is how a shutdown path raises during a shutdown that was already going badly.

Source code in src/symfonic/platform/agent_host.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
async def aclose(self) -> None:
    """Close the registry, then the resources, exactly once.

    In that order: an agent holding a closed pool is a worse failure than
    a pool held a moment longer, because the first surfaces mid-turn and
    the second surfaces nowhere.

    Resources are closed only if they were opened. A host whose
    construction failed before ``start`` is still closed by the handler,
    and closing what was never opened is how a shutdown path raises during
    a shutdown that was already going badly.
    """
    if self._closed:
        return
    self._closed = True
    await self.registry.aclose()
    if self._started and self._resources is not None:
        await self._resources.aclose()

resume async 

resume(token: str, answer: Any) -> Any

Continue a durable human pause through this host's scoped agents.

A host with no configured continuation fails early instead of accepting a token it cannot safely bind to a scope. Compose human_continuation(...) with a verified claim-to-scope resolver.

Source code in src/symfonic/platform/agent_host.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
async def resume(self, token: str, answer: Any) -> Any:
    """Continue a durable human pause through this host's scoped agents.

    A host with no configured continuation fails early instead of accepting
    a token it cannot safely bind to a scope.  Compose
    ``human_continuation(...)`` with a verified claim-to-scope resolver.
    """
    if self._closed:
        raise HostClosed("the host is closed; no continuation can be served")
    if self._continuation is None:
        from symfonic.platform.continuation import ContinuationUnavailable

        raise ContinuationUnavailable(
            "this AgentHost has no continuation service; compose "
            "human_continuation(capability=..., scope_for_claims=...)"
        )
    return await self._continuation.resume(token, answer)

start async 

start() -> None

Open shared resources once, however many times this is called.

Source code in src/symfonic/platform/agent_host.py 
135
136
137
138
139
140
141
142
143
async def start(self) -> None:
    """Open shared resources once, however many times this is called."""
    if self._started or self._closed:
        return
    # Set before awaiting, so a second caller arriving mid-open does not
    # start a second one. Two opens is worse than one wait: it is two pools.
    self._started = True
    if self._open is not None:
        await self._open()

AuditRecord dataclass 

AuditRecord(action: str, outcome: str, principal_id: str, scope_key: str, resource_type: str = 'tenant', resource_id: str = '', metadata: Mapping[str, Any] = dict(), at: float = time.time())

One append-only audit fact (AUD-1..3).

The credential screen runs in __post_init__ rather than in the sink, so a record that would leak cannot be constructed — a sink-side filter only protects the sinks that remember to run it.

AuditSeam 

AuditSeam(sink: AuditSink | None = None, *, on_degraded: Callable[[AuditRecord], None] | None = None)

The one path platform services write audit facts through.

Handlers do not construct sinks and do not reach for a module-level emitter; they are handed a seam. That is what makes "every administrative operation is audited before it mutates" (ADM-6) checkable rather than conventional.

Source code in src/symfonic/platform/audit.py 
81
82
83
84
85
86
87
88
89
def __init__(
    self,
    sink: AuditSink | None = None,
    *,
    on_degraded: Callable[[AuditRecord], None] | None = None,
) -> None:
    self._sink = sink if sink is not None else NullAuditSink()
    self._on_degraded = on_degraded
    self._degraded = False

degraded property 

degraded: bool

True once a record failed to land. Sticky: it describes the host.

record async 

record(record: AuditRecord, *, destructive: bool = False) -> None

Emit. On failure, degrade loudly — and for destruction, fail closed.

Source code in src/symfonic/platform/audit.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def record(self, record: AuditRecord, *, destructive: bool = False) -> None:
    """Emit. On failure, degrade loudly — and for destruction, fail closed."""
    try:
        await self._sink.emit(record)
    except Exception as exc:  # noqa: BLE001 - the failure is the subject
        self._degraded = True
        logger.error(
            "audit sink failed for action=%s scope=%s: %s",
            record.action,
            record.scope_key,
            exc,
        )
        if self._on_degraded is not None:
            self._on_degraded(record)
        if destructive:
            raise AuditSinkError(
                f"the audit seam could not record {record.action!r} and the "
                "operation is destructive; proceeding would destroy data with "
                "no trail (ADM-6, AUD-4)"
            ) from exc

AuditSink

Bases: Protocol

Registry row 3 — narrow, append-only, and never constructed in a handler.

AuditSinkError

Bases: SymfonicError

The seam could not record, and the operation may not proceed.

AuthPosture dataclass 

AuthPosture(environment: str, production: bool, verifier_registered: bool, insecure_override: bool, requests_served_untrusted: int = 0)

Continuously observable answer to "is this host trusting headers?".

Every field is something an operator would want on a dashboard, and requests_served_untrusted is the one that turns a silent misconfiguration into a rising number.

AuthenticatedPrincipal dataclass 

AuthenticatedPrincipal(principal_id: str, scope: SubjectScope, is_admin: bool = False, derivation: Mapping[str, Any] = (lambda: MappingProxyType({}))())

SCOPE-2 — whole, or not produced at all.

There is no partially-derived principal and no mutation after derivation. Code that needs a narrower scope derives a child by :meth:narrowed, which refuses to widen, rather than by editing the one it was given.

narrowed 

narrowed(child: SubjectScope) -> AuthenticatedPrincipal

SCOPE-9 / LAY-ADR §3.3: children narrow, compilation rejects widening.

Source code in src/symfonic/platform/values.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def narrowed(self, child: SubjectScope) -> AuthenticatedPrincipal:
    """SCOPE-9 / LAY-ADR §3.3: children narrow, compilation rejects widening."""
    if not child.narrows(self.scope):
        raise AuthorizationError(
            f"{child.scope_key!r} does not narrow {self.scope.scope_key!r}; a "
            "child scope is a subset of its parent's, never a sibling and "
            "never a widening"
        )
    return AuthenticatedPrincipal(
        principal_id=self.principal_id,
        scope=child,
        is_admin=self.is_admin,
        derivation=self.derivation,
    )

require_self 

require_self(tenant_id: str) -> None

SCOPE-8 / ADM-4: a self-scope route cannot be redirected.

Refused, not honoured, and refused before the service is called — a parameter, body field, or header naming another tenant is an attempt, not a preference.

Source code in src/symfonic/platform/values.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def require_self(self, tenant_id: str) -> None:
    """SCOPE-8 / ADM-4: a self-scope route cannot be redirected.

    Refused, not honoured, and refused *before* the service is called — a
    parameter, body field, or header naming another tenant is an attempt,
    not a preference.
    """
    if tenant_id != self.scope.tenant_id:
        raise AuthorizationError(
            f"this operation acts on the resolved tenant "
            f"{self.scope.tenant_id!r}; a request naming {tenant_id!r} is "
            "refused rather than redirected (SEC-AUTHZ-2)"
        )

AuthenticationError

Bases: PlatformError

No derivable credentials. 401.

Never a fallback to a default, shared, or root tenant (SCOPE-4, SEC-FCP-1).

AuthorizationError

Bases: PlatformError

Authenticated, but not for this scope. 403.

Also raised by :meth:AuthenticatedPrincipal.narrowed on an attempted widening and by require_self on a redirect attempt (SCOPE-8/SEC-AUTHZ-2), because "you may not act on that tenant" is one answer however it was asked.

BudgetExceededError 

BudgetExceededError(message: str, *, code: str = 'budget_exceeded', reason: str | None = None, retry_after: int = 3600, scope_key: str = '')

Bases: PlatformError

Admission refused by the budget port. 429 + Retry-After.

This class is BUD-2's whole point. The shipped path raises a generic error whose message starts "Budget exceeded:" and three routers branch on that prefix; here the class carries the decision, code distinguishes exhaustion from an unavailable ledger, and retry_after is a number rather than a hard-coded header string in three places.

Source code in src/symfonic/platform/errors.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(
    self,
    message: str,
    *,
    code: str = "budget_exceeded",
    reason: str | None = None,
    retry_after: int = 3600,
    scope_key: str = "",
) -> None:
    super().__init__(message)
    self.code = code
    self.reason = reason
    self.retry_after = retry_after
    self.scope_key = scope_key

BudgetService 

BudgetService(*, budget_check: BudgetCheck | None, cost_read_model: CostReadModel | None = None, audit: AuditSeam | None = None)

Admission from the narrow port, plus the cost read models.

Source code in src/symfonic/platform/budget.py 
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    *,
    budget_check: BudgetCheck | None,
    cost_read_model: CostReadModel | None = None,
    audit: AuditSeam | None = None,
) -> None:
    self._check = budget_check
    self._read_model = cost_read_model
    self._audit = audit if audit is not None else AuditSeam()

admit async 

admit(principal: AuthenticatedPrincipal) -> BudgetDecision

The decision, without raising. enforce is the gate.

Split because BUD-1 has two consumers: the platform admits a request, and the invocation re-checks before its own effects. Both read the same :class:BudgetDecision; only the first turns it into a refusal status.

Source code in src/symfonic/platform/budget.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
async def admit(self, principal: AuthenticatedPrincipal) -> BudgetDecision:
    """The decision, without raising. ``enforce`` is the gate.

    Split because BUD-1 has two consumers: the platform admits a request,
    and the invocation re-checks before its own effects. Both read the same
    :class:`BudgetDecision`; only the first turns it into a refusal status.
    """
    if self._check is None:
        return BudgetDecision(allowed=True, code="budget_unenforced")
    try:
        return await self._check.check(
            principal.scope, is_admin=principal.is_admin
        )
    except Exception as exc:  # noqa: BLE001 - fail closed, then say why
        logger.warning(
            "budget port failed for scope=%s: %s", principal.scope.scope_key, exc
        )
        return BudgetDecision(
            allowed=False,
            reason=f"budget port unavailable: {type(exc).__name__}",
            code="budget_unavailable",
        )

describe 

describe() -> dict[str, Any]

BUD-7: what this host does about budgets, stated rather than implied.

Source code in src/symfonic/platform/budget.py 
50
51
52
53
54
55
def describe(self) -> dict[str, Any]:
    """BUD-7: what this host does about budgets, stated rather than implied."""
    return {
        "enforcing": self._check is not None,
        "cost_read_model": self._read_model is not None,
    }

enforce async 

enforce(principal: AuthenticatedPrincipal) -> BudgetDecision

SCOPE-14 step 5: fail-closed, evented, typed.

A broken ledger denies rather than admits (SEC-FCP-1). The alternative — "the accountant is down, so everything is free" — is the failure mode a budget exists to prevent.

Source code in src/symfonic/platform/budget.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
async def enforce(self, principal: AuthenticatedPrincipal) -> BudgetDecision:
    """SCOPE-14 step 5: fail-closed, evented, typed.

    A broken ledger denies rather than admits (SEC-FCP-1). The alternative —
    "the accountant is down, so everything is free" — is the failure mode a
    budget exists to prevent.
    """
    decision = await self.admit(principal)
    if decision.allowed:
        return decision
    await self._audit.record(
        AuditRecord(
            action="budget_denied",
            outcome="denied",
            principal_id=principal.principal_id,
            scope_key=principal.scope.scope_key,
            resource_type="budget",
            metadata={"code": decision.code, "reason": decision.reason or ""},
        )
    )
    raise BudgetExceededError(
        f"budget refused for {principal.scope.scope_key}: {decision.reason}",
        code=decision.code if decision.code != "budget_ok" else "budget_exceeded",
        reason=decision.reason,
        retry_after=decision.retry_after_seconds,
        scope_key=principal.scope.scope_key,
    )

summarize async 

summarize(principal: AuthenticatedPrincipal, *, window: str) -> Any | None

BUD-6: a projection, scoped to the caller like every other route.

Returns None when no read model is bound — an absent projection, not an empty one. Reporting zeros for a ledger nobody is keeping is how a cost dashboard learns to lie.

Source code in src/symfonic/platform/budget.py 
108
109
110
111
112
113
114
115
116
117
118
119
async def summarize(
    self, principal: AuthenticatedPrincipal, *, window: str
) -> Any | None:
    """BUD-6: a projection, scoped to the caller like every other route.

    Returns ``None`` when no read model is bound — an absent projection, not
    an empty one. Reporting zeros for a ledger nobody is keeping is how a
    cost dashboard learns to lie.
    """
    if self._read_model is None:
        return None
    return await self._read_model.summarize(principal.scope, window=window)

ContinuationService

Bases: Protocol

Resumes what a pause stopped, over the host's own agents and stores.

Deliberately not a method on Agent. An agent is bound to one scope and does not know which of its turns are outstanding; resuming needs a token to be looked up, a scope to be recovered from it, and the agent for that scope to be found -- three things the host has and an agent does not.

resume async 

resume(token: str, answer: Any) -> Any

Continue the paused turn with answer and return its result.

Idempotency is the contract's hardest clause and it belongs here rather than in a transport: a person answering twice, or a client retrying, must not run the rest of the turn twice. An implementation either completes once and replays the result, or refuses the second attempt by name -- and it must say which, because "the answer was accepted" and "the answer was accepted again" look identical to a caller otherwise. The supported human adapter chooses at most once: it prepares the scoped agent before redemption, then rejects a second use after the durable claim; it does not replay effects if the resumed invocation itself later fails.

Raises:

Type Description
UnknownPauseToken

if the token names no resumable turn.
Source code in src/symfonic/platform/continuation.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def resume(self, token: str, answer: Any) -> Any:
    """Continue the paused turn with ``answer`` and return its result.

    Idempotency is the contract's hardest clause and it belongs here
    rather than in a transport: a person answering twice, or a client
    retrying, must not run the rest of the turn twice. An implementation
    either completes once and replays the result, or refuses the second
    attempt by name -- and it must say which, because "the answer was
    accepted" and "the answer was accepted again" look identical to a
    caller otherwise. The supported human adapter chooses **at most once**:
    it prepares the scoped agent before redemption, then rejects a second
    use after the durable claim; it does not replay effects if the resumed
    invocation itself later fails.

    Raises:
        UnknownPauseToken: if the token names no resumable turn.
    """
    ...

ticket async 

ticket(token: str) -> PauseTicket

The paused turn token names.

Raises:

Type Description
UnknownPauseToken

if it names none, or has expired.
Source code in src/symfonic/platform/continuation.py 
 98
 99
100
101
102
103
104
async def ticket(self, token: str) -> PauseTicket:
    """The paused turn ``token`` names.

    Raises:
        UnknownPauseToken: if it names none, or has expired.
    """
    ...

ContinuationUnavailable

Bases: RuntimeError

Raised when a host was not composed with durable continuation wiring.

DecisionLog dataclass 

DecisionLog(decisions: list[GovernanceDecision] = list())

The in-process sink, for a deployment that wants one and nothing more.

Not a default: governance() records nothing unless handed a sink, so a deployment that never asked for a decision log does not accumulate one for the life of the process.

by_rule 

by_rule(rule_id: str) -> tuple[GovernanceDecision, ...]

Every decision one rule took part in.

The question a dashboard asks, and the reason rule_ids is a tuple rather than a string: a decision two rules made is found by either.

Source code in src/symfonic/platform/governance_decisions.py 
 95
 96
 97
 98
 99
100
101
def by_rule(self, rule_id: str) -> tuple[GovernanceDecision, ...]:
    """Every decision one rule took part in.

    The question a dashboard asks, and the reason ``rule_ids`` is a tuple
    rather than a string: a decision two rules made is found by either.
    """
    return tuple(d for d in self.decisions if rule_id in d.rule_ids)

DecisionSink

Where decisions go. A protocol in all but name.

Anything with record(decision) is one, so a deployment writes to its own store without inheriting from this package.

DestructiveConfirmationError

Bases: PlatformError

A destructive operation arrived without intent bound to the resolved scope.

ADM-5: the confirmation is compared against the derived tenant id, never against a value taken from the same request that supplied the confirmation — otherwise the check confirms only that the caller can copy a string.

ErasureOutcome dataclass 

ErasureOutcome(scope_key: str, complete: bool, counts: dict[str, int] = dict(), unconfirmed: tuple[str, ...] = (), errors: dict[str, str] = dict(), generation: int = 0)

What one pass of the saga achieved. Partial success is a real answer.

ErasureSaga 

ErasureSaga(*, registry: SubjectDataStoreRegistry, fence: ErasureFence, saga_store: ErasureSagaStore, audit: AuditSeam | None = None, max_attempts: int = 3)

Drives every registered participant to a confirmed absence.

Source code in src/symfonic/platform/erasure.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    *,
    registry: SubjectDataStoreRegistry,
    fence: ErasureFence,
    saga_store: ErasureSagaStore,
    audit: AuditSeam | None = None,
    max_attempts: int = 3,
) -> None:
    if max_attempts < 1:
        raise ValueError("max_attempts must be at least 1")
    self._registry = registry
    self._fence = fence
    self._saga_store = saga_store
    self._audit = audit if audit is not None else AuditSeam()
    self._max_attempts = max_attempts

run async 

run(scope: SubjectScope) -> ErasureOutcome

One bounded pass. Idempotent, and safe to call again after a crash.

Source code in src/symfonic/platform/erasure.py 
 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
113
114
115
116
117
118
119
120
121
122
123
124
125
async def run(self, scope: SubjectScope) -> ErasureOutcome:
    """One bounded pass. Idempotent, and safe to call again after a crash."""
    generation = await self._fence.read_generation(scope.scope_key)
    if not generation.tombstoned:
        raise ErasureFenceDenied(
            f"no subject tombstone is published for {scope.scope_key!r}; "
            "running the saga first would race every live writer, which is "
            "the whole reason the tombstone goes first (PRIV-4)",
            reason="generation_mismatch",
            scope_key=scope.scope_key,
            observed_generation=generation.generation,
        )

    state = await self._saga_store.start(
        scope.scope_key, participants=self._registry.participant_ids()
    )
    for participant_id in state.unconfirmed():
        state = await self._sweep_one(scope, participant_id)

    if state.complete and await self._saga_store.claim_generation_advance(
        scope.scope_key
    ):
        # Claim first, advance second. "Read the flag, then advance" puts an
        # await between the check and the act, and two runs for one scope —
        # an operator retry racing a scheduled `resume` — would then both
        # advance the generation. Claiming loses at most one advance if the
        # fence then fails; the tombstone, which is what actually bars
        # writers, stands either way.
        generation = await self._fence.complete_erasure(scope.scope_key)
        state = await self._saga_store.read(scope.scope_key) or state
    else:
        generation = await self._fence.read_generation(scope.scope_key)

    return ErasureOutcome(
        scope_key=scope.scope_key,
        complete=state.complete,
        counts=state.counts(),
        unconfirmed=state.unconfirmed(),
        errors=state.errors(),
        generation=generation.generation,
    )

ExtensionCapability 

ExtensionCapability(composed: Any)

The composed extensions, in the shape Agent accepts.

Source code in src/symfonic/platform/extensions.py 
119
120
def __init__(self, composed: Any) -> None:
    self._composed = composed

composed property 

composed: Any

The merged extension set, for a caller that wants the diagnostics.

sources property 

sources: tuple[Any, ...]

The extensions' prompt fragments, for PromptingCapability.

capabilities=[
    exts,
    PromptingCapability(sources=[*persona, *exts.sources]),
]

Every fragment is declared untrusted. A plugin's text is not the deployment's own, and rendering it verbatim at an authored tier is how an extension writes instructions nobody in the deployment approved.

aclose async 

aclose() -> None

Run the teardown hooks the composed extensions declared.

Called by whatever owns this capability -- the host, for a generated app. teardown is a kernel-owned phase, so a capability cannot register a stage there; the hook reaches its moment through the close path instead. Every hook runs even if an earlier one raises, because a server left open by a failed close is worse than a traceback.

Source code in src/symfonic/platform/extensions.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def aclose(self) -> None:
    """Run the ``teardown`` hooks the composed extensions declared.

    Called by whatever owns this capability -- the host, for a generated
    app. ``teardown`` is a kernel-owned phase, so a capability cannot
    register a stage there; the hook reaches its moment through the close
    path instead. Every hook runs even if an earlier one raises, because a
    server left open by a failed close is worse than a traceback.
    """
    failures: list[BaseException] = []
    for hook in self._composed.lifecycle:
        if str(hook.phase) != "teardown":
            continue
        try:
            outcome = hook.run()
            if hasattr(outcome, "__await__"):
                await outcome
        except Exception as failed:  # noqa: BLE001, PERF203
            failures.append(failed)
    if failures:
        raise ExtensionContractError(
            f"{len(failures)} extension teardown hook(s) failed; the first "
            f"was {failures[0]!r}"
        )

contribute 

contribute(request: Any) -> Any

Offer the extensions' tools to the turn.

Tools only. CapabilityContribution carries tools, stages, handlers and grants -- there is no field for a prompt contribution, because the prompt is compiled by the prompting capability from sources. So the fragments come out of :attr:sources instead and go where every other source goes, which is the same answer the knowledge door reached.

request is read for its grants and found to need none: an extension contributes what it declared and performs no effect of its own.

Source code in src/symfonic/platform/extensions.py 
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
162
def contribute(self, request: Any) -> Any:
    """Offer the extensions' tools to the turn.

    Tools only. ``CapabilityContribution`` carries tools, stages, handlers
    and grants -- there is no field for a prompt contribution, because the
    prompt is compiled by the prompting capability from *sources*. So the
    fragments come out of :attr:`sources` instead and go where every other
    source goes, which is the same answer the knowledge door reached.

    ``request`` is read for its grants and found to need none: an extension
    contributes what it declared and performs no effect of its own.
    """
    from symfonic.kernel.contracts.contributions import CapabilityContribution

    # #147: policies too, not tools only. A contributed policy used to
    # compose cleanly and be consulted before nothing, because this method
    # offered no stage for it to be asked in. Contributed only when there
    # is a policy: a deployment with none pays for no rung.
    stages: tuple[Any, ...] = ()
    handlers: dict[str, Any] = {}
    if getattr(self._composed, "policies", ()):  # noqa: PLC0415
        from symfonic.platform.extension_policy_stage import (
            POLICY_STAGE_ID,
            policy_handler,
            policy_stage,
        )

        stages = (policy_stage(),)
        handlers = {POLICY_STAGE_ID: policy_handler(self._composed)}

    return CapabilityContribution(
        capability="extensions",
        tools=tuple(_executable(tool) for tool in self._composed.tools),
        stages=stages,
        handlers=handlers,
    )

ExtensionContractError

Bases: ExtensionError

A contribution's own declaration is malformed.

Raised at declaration time — when the contribution value is validated — not at use time. A tool with no name is not a tool that fails when called; it is a tool that should never have been admitted to the catalogue.

ExtensionContribution dataclass 

ExtensionContribution(extension: str, tools: tuple[ToolContribution, ...] = (), prompts: tuple[PromptFragment, ...] = (), policies: tuple[PolicyContribution, ...] = (), lifecycle: tuple[LifecycleContribution, ...] = ())

Everything one extension offers, in one value.

An extension is asked, it answers, and the answer is validated, ordered, and refused as a unit. Nothing here holds a reference to an engine, a registry, or a graph, which is why an extension cannot install itself.

build classmethod 

build(extension: str, *, tools: Sequence[ToolContribution] = (), prompts: Sequence[PromptFragment] = (), policies: Sequence[PolicyContribution] = (), lifecycle: Sequence[LifecycleContribution] = ()) -> ExtensionContribution

Build a bundle from any sequences, freezing each into a tuple.

Source code in src/symfonic/capabilities/extensions/contracts.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def build(
    cls,
    extension: str,
    *,
    tools: Sequence[ToolContribution] = (),
    prompts: Sequence[PromptFragment] = (),
    policies: Sequence[PolicyContribution] = (),
    lifecycle: Sequence[LifecycleContribution] = (),
) -> ExtensionContribution:
    """Build a bundle from any sequences, freezing each into a tuple."""
    return cls(
        extension=extension,
        tools=tuple(tools),
        prompts=tuple(prompts),
        policies=tuple(policies),
        lifecycle=tuple(lifecycle),
    )

validate 

validate() -> None

Validate the bundle, every member, and the attribution of each.

Attribution is checked here rather than on each member because it is a property of the pair: a well-formed tool contributed under someone else's name is exactly the confused-deputy shape AS-INT-5 is about, and the member alone cannot see that it was misfiled.

Source code in src/symfonic/capabilities/extensions/contracts.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def validate(self) -> None:
    """Validate the bundle, every member, and the attribution of each.

    Attribution is checked here rather than on each member because it is a
    property of the *pair*: a well-formed tool contributed under someone
    else's name is exactly the confused-deputy shape AS-INT-5 is about, and
    the member alone cannot see that it was misfiled.
    """
    if not self.extension:
        raise ExtensionContractError("an extension contribution must name itself.")
    require_id(self.extension, what="extension", owner=self.extension)
    for group in (self.tools, self.prompts, self.policies, self.lifecycle):
        for member in group:
            member.validate()
            if member.extension != self.extension:
                raise PrivilegeWideningError(
                    f"{self.extension}: contributed a member attributed to "
                    f"{member.extension!r}. An extension may not contribute on "
                    "another extension's behalf."
                )

ExtensionProvider

Bases: Protocol

What the composer asks. Two members, both deliberate.

name is a declared member so an isinstance check requires it: an anonymous provider produces contributions nobody can revoke.

contribute is synchronous. Anything an extension needs to discover over the network happens before composition — the MCP adapter's discover() is its own async step — so composition itself is a pure function over values, replayable and comparable without an event loop.

GovernanceDecision dataclass 

GovernanceDecision(state: str, phase: str, stage: str, rule_ids: tuple[str, ...], call_id: str, tool: str, reason: str, at: datetime = (lambda: datetime.now(UTC))())

One governance outcome for one call.

state is the vocabulary, and it is three words on purpose:

applied A rule rewrote the call and the tool ran with the rewrite. refused A rule objected and the call did not run. discarded A rule asked for a change this phase has no way to make. It is neither of the above and must not be reported as either: an operator reading applied for a change that never happened is worse off than one reading nothing.

GovernanceRefused 

GovernanceRefused(*, stage: str, phase: str, disposition: Disposition, reason: str, rule_ids: tuple[str, ...] = ())

Bases: ContractViolationError

A governance stage refused, and the turn must stop.

Raised rather than returned as rejected(...), and the distinction is the kernel's rather than a style choice. REJECTED means this contribution was not applied -- a well-formed answer the dispatcher declined, after which the turn continues; tests/kernel pins that ("a rung that ended the turn on a rejection would make refusal and failure the same event"). What governance means is do not proceed, and the kernel's word for that is an exception, which every phase already turns into a stopped turn through require_no_crashed_stage.

The first version of this door used rejected for both. It read as a refusal in the trace and the tool call ran anyway -- ten rounds refused, ten calls executed -- because nothing consumes REJECTED and nothing was supposed to.

Source code in src/symfonic/platform/governance_refusal.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    *,
    stage: str,
    phase: str,
    disposition: Disposition,
    reason: str,
    rule_ids: tuple[str, ...] = (),
) -> None:
    self.stage = stage
    self.phase = phase
    self.disposition = disposition
    self.reason = reason
    #: Which rules refused, individually. A stage is a place and a rule is
    #: a decision: three tenant rules run at ``policy_steering``, and a
    #: caller told only the stage knows which file to open rather than
    #: which rule to change.
    self.rule_ids = rule_ids
    super().__init__(
        f"governance {disposition.value} at {stage} during {phase}: {reason}"
    )

GovernanceStages 

GovernanceStages(capability: Any, decisions: Any = None)

A composable wrapper around a built governance pipeline.

The object a composition root passes to Agent(capabilities=[...]). It holds the capability rather than subclassing it: governance answers "what does this turn get, and why", and this answers "which rungs does that run on" -- two questions, and merging them is what would put kernel imports back inside the contained package.

Source code in src/symfonic/platform/governance.py 
30
31
32
def __init__(self, capability: Any, decisions: Any = None) -> None:
    self._capability = capability
    self._decisions = decisions

capability property 

capability: Any

The governed pipeline, for a caller that wants the trace directly.

contribute 

contribute(request: Any) -> Any

Declare the rungs this pipeline runs on, and the handlers for them.

Classification and reflection may call a model and declare that effect when composed. Refusal and deterministic checks need no additional effect grant.

Source code in src/symfonic/platform/governance.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def contribute(self, request: Any) -> Any:
    """Declare the rungs this pipeline runs on, and the handlers for them.

    Classification and reflection may call a model and declare that
    effect when composed. Refusal and deterministic checks need no
    additional effect grant.
    """
    from symfonic.kernel.contracts.contributions import CapabilityContribution

    descriptors, handlers = kernel_stages(
        self._capability.pipeline, self._decisions
    )
    return CapabilityContribution(
        capability=GOVERNANCE_CAPABILITY,
        stages=descriptors,
        handlers=handlers,
        effect_grants=frozenset(effect for stage in descriptors for effect in stage.effects),
    )

GrantEffects 

GrantEffects(*families: str)

An adopter's explicit authorization, written at the composition root.

Usage, and the whole point of the shape::

Agent(provider, capabilities=[
    GrantEffects("memory-read"),
    MemoryCapability(bridge),
])

It is not a capability. It contributes no stage, no handler and no tool; the facade recognises it by exact type, takes its families, and removes it from the sequence before folding. That ordering is the security property: a contribution can only ever narrow a set that was established without its participation, and it cannot manufacture another top-level entry.

Why a separate type rather than an Agent parameter: FAC-4 freezes the facade at four constructor arguments, and extension happens through capabilities=. Why not "installing the capability implies the grant": that conflates installation with authorization and hides the authority from the place a reviewer looks for it.

The honest limit, stated rather than papered over: nothing stops third-party code from instantiating this type. The boundary is that the composition root chose the top-level sequence. If untrusted code controls that sequence, nominal framework provenance cannot recover authority — that needs policy outside this process. And a granted family is granted for the whole invocation, not to one capability: once memory-read is granted, any capability in the same list may use it. Per-capability authority would need grants keyed by capability rather than today's frozenset[str].

Source code in src/symfonic/kernel/contracts/effects.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def __init__(self, *families: str) -> None:
    declared = frozenset(families)
    if not declared:
        raise ConfigurationError(
            "GrantEffects() declares no family. An empty authorization is a "
            "no-op that reads at the composition root as though something was "
            "granted; name the families or remove the entry."
        )
    require_known_families(declared, subject="a GrantEffects declaration")
    ungrantable = sorted(declared - ADOPTER_GRANTABLE)
    if ungrantable:
        raise ConfigurationError(
            f"GrantEffects declares {ungrantable}, which an adopter does not "
            f"grant. Grantable at a composition root: {sorted(ADOPTER_GRANTABLE)}. "
            "model_call and tool_call are derived from the invocation itself — a "
            "turn calls the model, and carries tools or does not — so declaring "
            "them here would be a second source of truth for a fact the plan "
            "already knows."
        )
    # Private plus a read-only property: every other invariant in this
    # module is established at construction and the object then treated as a
    # value, and a writable attribute broke that. ``partition_grants``
    # re-reads this without re-validating, so assigning to it granted a
    # family ADOPTER_GRANTABLE deliberately withholds -- and since __eq__
    # and __hash__ are defined over it, a GrantEffects in a set could have
    # its hash change underneath it.
    self._families: frozenset[str] = declared

HeaderScopeResolver 

HeaderScopeResolver(*, verifier: TenantAuthVerifier | None = None, environment: str | None = None, allow_insecure_prod: bool = False, environ: dict[str, str] | None = None)

Registry row 16, default implementation: headers plus an injected verifier.

X-Tenant-ID on its own is untrusted input — the framework cannot assume an auth model — so the verifier is what turns a claim into a principal. With no verifier the header is accepted on trust, which is fine for a unit test and never for production; :meth:posture is how that state stops being invisible (SCOPE-5).

Source code in src/symfonic/platform/scope.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    *,
    verifier: TenantAuthVerifier | None = None,
    environment: str | None = None,
    allow_insecure_prod: bool = False,
    environ: dict[str, str] | None = None,
) -> None:
    self._verifier = verifier
    self._environment = (
        environment if environment is not None else detect_environment(environ)
    ).strip().lower()
    self._allow_insecure_prod = allow_insecure_prod
    self._untrusted_requests = 0

posture 

posture() -> AuthPosture

A fresh reading every call — this is a gauge, not a one-shot log.

Source code in src/symfonic/platform/scope.py 
70
71
72
73
74
75
76
77
78
def posture(self) -> AuthPosture:
    """A fresh reading every call — this is a gauge, not a one-shot log."""
    return AuthPosture(
        environment=self._environment,
        production=self._environment in PRODUCTION_VALUES,
        verifier_registered=self._verifier is not None,
        insecure_override=self._allow_insecure_prod,
        requests_served_untrusted=self._untrusted_requests,
    )

resolve async 

resolve(credentials: RequestCredentials) -> AuthenticatedPrincipal

SCOPE-1: exactly once per request, and this is the once.

Source code in src/symfonic/platform/scope.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def resolve(self, credentials: RequestCredentials) -> AuthenticatedPrincipal:
    """SCOPE-1: exactly once per request, and this is the once."""
    tenant_id = (credentials.header(TENANT_HEADER) or "").strip()
    if not tenant_id:
        raise AuthenticationError(
            f"Missing required {TENANT_HEADER} header",
        )

    facts = await self._verify(credentials, tenant_id)
    scope = self._build_scope(credentials, tenant_id)
    return AuthenticatedPrincipal(
        principal_id=str(facts.get("principal_id") or tenant_id),
        scope=scope,
        is_admin=bool(facts.get("is_admin", False)),
        derivation={
            "source": "header",
            "route": credentials.route,
            "verified": self._verifier is not None,
            "environment": self._environment,
        },
    )

HostClosed

Bases: RuntimeError

Raised when an agent is requested from a host that has been closed.

A distinct type rather than a bare RuntimeError because the caller's correct response differs: a closed host during shutdown is expected and should end the request, while a closed host during normal operation is a lifecycle bug in the composition root. A message alone cannot be branched on without matching strings, which is the pattern this codebase has been removing.

HostContinuationFactory 

HostContinuationFactory(capability: Any, scope_for_claims: Callable[[PauseClaims], Any])

Bind host-owned agent lookup after a deployment has composed its ports.

Source code in src/symfonic/platform/human_continuation.py 
133
134
135
def __init__(self, capability: Any, scope_for_claims: Callable[[PauseClaims], Any]) -> None:
    self._capability = capability
    self._scope_for_claims = scope_for_claims

HumanContinuationService 

HumanContinuationService(*, capability: Any, scope_for_claims: Callable[[PauseClaims], Any], agent_for: Callable[[Any], Awaitable[Any]])

Redeem durable human pauses and route the resulting turn through a host.

scope_for_claims receives only claims that passed signature and expiry verification. It must resolve the deployment's durable scope record and may return None when the scope is no longer known; that is rejected before a capability or agent is asked to continue the turn.

Source code in src/symfonic/platform/human_continuation.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    *,
    capability: Any,
    scope_for_claims: Callable[[PauseClaims], Any],
    agent_for: Callable[[Any], Awaitable[Any]],
) -> None:
    if not callable(getattr(capability, "resume", None)):
        raise TypeError(
            "human continuation needs a HumanInteractionCapability with resume(); "
            f"got {type(capability).__name__}"
        )
    if not callable(getattr(capability, "decode_token", None)):
        raise TypeError(
            "human continuation needs decode_token so an opaque pause can be "
            "authenticated on a fresh host"
        )
    self._capability = capability
    self._scope_for_claims = scope_for_claims
    self._agent_for = agent_for

MountedRoute

Bases: tuple

One served endpoint: (path, methods), comparable and printable.

NullAuditSink

Drops records. The default only because a host that wants none says so.

ObservabilityServices dataclass 

ObservabilityServices(collector: Any, metrics: Any = None, budget: Any = None, event_sinks: tuple[Any, ...] = ())

What a process needs to account for what its agents did.

Satisfies :class:~symfonic.platform.host.SharedResources structurally, so a composition root hands this to the host it already builds rather than installing it anywhere.

There is deliberately no setter. Ownership runs one way: the host holds these, and nothing installs them into a place other code discovers.

budget_is_shared property 

budget_is_shared: bool

Whether two replicas would enforce one limit or two halves of it.

Exposed because in-memory is a legitimate choice and a silent one is not: a deployment may accept per-replica budgets, but it may not be unable to tell that is what it has.

aclose async 

aclose() -> None

Release what these own. Idempotent, because shutdown paths repeat.

Source code in src/symfonic/platform/observability.py 
73
74
75
76
77
78
async def aclose(self) -> None:
    """Release what these own. Idempotent, because shutdown paths repeat."""
    for held in (self.metrics, self.budget, *self.event_sinks):
        closer = getattr(held, "aclose", None)
        if callable(closer):
            await closer()

PauseTicket

Bases: Protocol

What a paused turn hands back, and what resuming it needs.

The token is the whole of the caller's half. Everything else on this protocol is recovered from it rather than supplied: the scope the pause was issued under, the session it belongs to, and what the turn was waiting for. A caller that supplied any of them could contradict the token, and the contradiction would have to be resolved by trusting one of them.

question property 

question: Any

What the turn stopped to ask. Shaped by whoever registered the pause point; the platform does not interpret it.

scope property 

scope: Any

The scope the pause was issued under. Recovered, never supplied.

token property 

token: str

The opaque handle a transport gives back to whoever must answer.

PlatformAgentHost

Bases: Protocol

Owns the process's shared resources and the registry over them.

The composition root a generated project's main builds once and closes once. Everything a request needs comes from here; nothing a request does happens here.

aclose async 

aclose() -> None

Close the registry and then the shared resources, exactly once.

Idempotent, because shutdown paths run twice more often than they run cleanly: a lifespan that closes and an atexit that closes are both correct and both fire.

Source code in src/symfonic/platform/host.py 
158
159
160
161
162
163
164
165
async def aclose(self) -> None:
    """Close the registry and then the shared resources, exactly once.

    Idempotent, because shutdown paths run twice more often than they run
    cleanly: a lifespan that closes and an atexit that closes are both
    correct and both fire.
    """
    ...

agent_for async 

agent_for(scope: Any) -> Any

The agent bound to scope.

Raises:

Type Description
HostClosed

if the host has been closed.
Source code in src/symfonic/platform/host.py 
146
147
148
149
150
151
152
async def agent_for(self, scope: Any) -> Any:
    """The agent bound to ``scope``.

    Raises:
        HostClosed: if the host has been closed.
    """
    ...

resume async 

resume(token: str, answer: Any) -> Any

Resume an explicitly configured durable human continuation.

Source code in src/symfonic/platform/host.py 
154
155
156
async def resume(self, token: str, answer: Any) -> Any:
    """Resume an explicitly configured durable human continuation."""
    ...

start async 

start() -> None

Open shared resources. Idempotent: starting twice opens once.

Source code in src/symfonic/platform/host.py 
142
143
144
async def start(self) -> None:
    """Open shared resources. Idempotent: starting twice opens once."""
    ...

PlatformError

Bases: SymfonicError

Root of the platform taxonomy.

PrivacyService 

PrivacyService(*, registry: SubjectDataStoreRegistry, fence: ErasureFence, saga_store: ErasureSagaStore, audit: AuditSeam | None = None, max_attempts: int = 3, bind_fence_audit: bool = True)

Export (Article 20) and erasure (Article 17), behind one boundary.

Source code in src/symfonic/platform/privacy.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(
    self,
    *,
    registry: SubjectDataStoreRegistry,
    fence: ErasureFence,
    saga_store: ErasureSagaStore,
    audit: AuditSeam | None = None,
    max_attempts: int = 3,
    bind_fence_audit: bool = True,
) -> None:
    self._registry = registry
    self._fence = fence
    self._audit = audit if audit is not None else AuditSeam()
    # EFX-ER-4: the fence refuses writers racing an erasure; this is what
    # makes those refusals visible outside the process. The hook is optional
    # rather than part of row 14 because an adapter whose backend already
    # audits its own conditional-update failures should not be made to
    # accept a second path for the same fact.
    #
    # The binding is *claimed*, not assumed. HOST-3's two-agents-in-one-
    # process case can hand the same fence to two services, and the second
    # constructor silently rebinding would send the first host's denials to
    # the second host's sink. The fence refuses the overwrite; a host that
    # deliberately shares a fence passes ``bind_fence_audit=False`` and
    # leaves the binding to whoever made it.
    if bind_fence_audit:
        attach = getattr(fence, "set_denial_recorder", None)
        if callable(attach):
            attach(fence_denial_recorder(self._audit))
    self._guard = SubjectGuard(fence=fence, saga_store=saga_store)
    self._saga = ErasureSaga(
        registry=registry,
        fence=fence,
        saga_store=saga_store,
        audit=self._audit,
        max_attempts=max_attempts,
    )

admit async 

admit(principal: AuthenticatedPrincipal) -> None

SCOPE-14 step 2, exposed for the admission gate.

Source code in src/symfonic/platform/privacy.py 
84
85
86
async def admit(self, principal: AuthenticatedPrincipal) -> None:
    """SCOPE-14 step 2, exposed for the admission gate."""
    await self._guard.require_admission(principal.scope)

erase async 

erase(principal: AuthenticatedPrincipal, *, confirmation: str) -> dict[str, Any]

Article 17. Confirm, audit intent, tombstone, then sweep.

Source code in src/symfonic/platform/privacy.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
167
168
169
170
171
172
173
174
175
176
177
178
179
async def erase(
    self, principal: AuthenticatedPrincipal, *, confirmation: str
) -> dict[str, Any]:
    """Article 17. Confirm, audit intent, tombstone, then sweep."""
    expected = f"DELETE-{principal.scope.tenant_id}"
    if confirmation != expected:
        raise DestructiveConfirmationError(
            "Missing or incorrect confirmation. Pass "
            "?confirmation=DELETE-<tenant_id> matching the *resolved* tenant "
            "to proceed."
        )

    # ADM-6: intent first, and fail closed if it cannot be recorded.
    await self._audit.record(
        AuditRecord(
            action="erase_all",
            outcome="initiated",
            principal_id=principal.principal_id,
            scope_key=principal.scope.scope_key,
            metadata={"stage": "initiated", "confirmation": expected},
        ),
        destructive=True,
    )

    await self._fence.publish_tombstone(
        principal.scope.scope_key, reason=f"erasure requested by {principal.principal_id}"
    )
    outcome = await self._saga.run(principal.scope)

    await self._audit.record(
        AuditRecord(
            action="erase_all",
            outcome="completed" if outcome.complete else "partial",
            principal_id=principal.principal_id,
            scope_key=principal.scope.scope_key,
            metadata={
                "stage": "completed",
                "counts": dict(outcome.counts),
                "unconfirmed": list(outcome.unconfirmed),
            },
        )
    )
    return self._render(outcome)

export async 

export(principal: AuthenticatedPrincipal) -> dict[str, Any]

PRIV-8 — exactly what the tenant's own queries could see.

Suppressed for a tombstoned subject like any other read: an export is not a privileged back door around the erasure it follows.

Source code in src/symfonic/platform/privacy.py 
 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
async def export(self, principal: AuthenticatedPrincipal) -> dict[str, Any]:
    """PRIV-8 — exactly what the tenant's own queries could see.

    Suppressed for a tombstoned subject like any other read: an export is
    not a privileged back door around the erasure it follows.
    """
    await self._guard.require_readable(principal.scope)
    fragments: dict[str, list[Any]] = {}
    unavailable: list[str] = []
    for store in self._registry.participants():
        fragment = await store.export_subject(principal.scope)
        if not fragment.exportable:
            unavailable.append(fragment.participant_id)
            continue
        fragments[fragment.participant_id] = [dict(r) for r in fragment.records]

    payload: dict[str, Any] = {
        "scope_key": principal.scope.scope_key,
        "tenant_id": principal.scope.tenant_id,
        "exported_at": datetime.now(UTC).isoformat(),
        "schema_version": EXPORT_SCHEMA_VERSION,
        "fragments": fragments,
    }
    if unavailable:
        # Honest about the gap rather than quietly complete.
        payload["not_exportable"] = sorted(unavailable)

    await self._audit.record(
        AuditRecord(
            action="export_data",
            outcome="ok",
            principal_id=principal.principal_id,
            scope_key=principal.scope.scope_key,
            metadata={
                "record_count": sum(len(v) for v in fragments.values()),
                "participants": len(fragments),
            },
        )
    )
    return payload

read_guard async 

read_guard(principal: AuthenticatedPrincipal) -> None

PRIV-5: reads are suppressed while any participant is unconfirmed.

Source code in src/symfonic/platform/privacy.py 
88
89
90
async def read_guard(self, principal: AuthenticatedPrincipal) -> None:
    """PRIV-5: reads are suppressed while any participant is unconfirmed."""
    await self._guard.require_readable(principal.scope)

resume async 

resume(scope: SubjectScope) -> dict[str, Any]

Drive an interrupted erasure forward. Safe to call repeatedly.

This is what makes PRIV-3's "resumable" operational rather than theoretical: a scheduler, an operator, or a restart hook calls it, and the saga picks up exactly the participants that never confirmed.

Source code in src/symfonic/platform/privacy.py 
181
182
183
184
185
186
187
188
async def resume(self, scope: SubjectScope) -> dict[str, Any]:
    """Drive an interrupted erasure forward. Safe to call repeatedly.

    This is what makes PRIV-3's "resumable" operational rather than
    theoretical: a scheduler, an operator, or a restart hook calls it, and
    the saga picks up exactly the participants that never confirmed.
    """
    return self._render(await self._saga.run(scope))

PromptFragment dataclass 

PromptFragment(fragment_id: str, text: str, extension: str, layer: str = 'l1', tier: str = 'session', scope: str = 'deployment', order: int = 100, truncated: bool = False)

Text an extension contributes to the compiled prompt.

The fields mirror the prompt compiler's contribution contract by name — layer, tier, scope, order — because the composition root's job is then a lookup rather than a translation. What it does not mirror is the tier range: a contributed fragment is restricted to the learned tiers, so no extension can place text where the model reads operator instruction.

Reflection dataclass 

Reflection(revise: bool, reason: str = '', confidence: float | None = None)

The critic's answer about the draft.

RequestCredentials dataclass 

RequestCredentials(headers: Mapping[str, str] = dict(), peer: str | None = None, user_agent: str | None = None, route: str = '')

Everything a resolver may look at, and nothing a handler would add.

A frozen mapping plus the connection facts. Header lookup is case-insensitive because HTTP header names are, and a resolver that only matched X-Tenant-ID exactly would authenticate one proxy and refuse another for no reason a user could see.

ScopeDerivationError

Bases: PlatformError

The credentials named a scope that is not well formed. 400.

ScopeResolver

Bases: Protocol

Registry row 16 — credentials to an authenticated principal.

The single derivation point (SEC-AUTHZ-1). Returns a whole principal or raises; there is no third outcome, because a resolver that could return None would make "no principal" a value a handler might forget to check.

ScopedAgentRegistry 

ScopedAgentRegistry(*, composer: Any, resources: Any)

One agent per scope, composed on first use.

Concurrency is handled with a per-scope lock rather than one global lock, so a slow composition for one tenant does not serialise every other tenant's first request. The double check inside the lock is what makes the composer run exactly once: the first waiter composes, the rest find the entry already there.

Source code in src/symfonic/platform/agent_host.py 
53
54
55
56
57
58
59
60
61
62
def __init__(self, *, composer: Any, resources: Any) -> None:
    self._composer = composer
    self._resources = resources
    self._agents: dict[str, Any] = {}
    self._locks: dict[str, asyncio.Lock] = {}
    # Guards the lock map itself. Without it two coroutines racing on a
    # brand-new scope each build their own lock and then each compose --
    # the race this class exists to prevent, moved one level down.
    self._guard = asyncio.Lock()
    self._closed = False

aclose async 

aclose() -> None

Drop every agent. Idempotent, and does not touch shared resources.

Ownership runs one way: the registry borrows what the host opened, so closing it here would close a pool other registries may still hold -- and would do it from whichever one happened to close first.

Source code in src/symfonic/platform/agent_host.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def aclose(self) -> None:
    """Drop every agent. Idempotent, and does not touch shared resources.

    Ownership runs one way: the registry borrows what the host opened, so
    closing it here would close a pool other registries may still hold --
    and would do it from whichever one happened to close first.
    """
    self._closed = True
    self._agents.clear()
    self._locks.clear()

SharedResources

Bases: Protocol

What every agent in a process borrows and none of them owns.

Deliberately opaque: the host holds it, hands it to the composer, and closes it. Naming its members here would make the host know what a deployment's memory stack is, which is the coupling this package exists to avoid -- an adopter with a different backend implements this and changes nothing else.

aclose async 

aclose() -> None

Release everything this object owns. Called once, by the host.

Source code in src/symfonic/platform/host.py 
76
77
78
async def aclose(self) -> None:
    """Release everything this object owns. Called once, by the host."""
    ...

TelemetryCapability 

TelemetryCapability(collector: Any, tenant: str = '', event_sinks: tuple[Any, ...] = ())

Reports each model round to the collector the host built.

Source code in src/symfonic/platform/telemetry.py 
102
103
104
105
106
107
108
def __init__(
    self, collector: Any, tenant: str = "", event_sinks: tuple[Any, ...] = ()
) -> None:
    self._collector = collector
    self._tenant = tenant
    self._event_sinks = event_sinks
    self._sink = TelemetryEventSink(collector, event_sinks, tenant)

event_sink property 

event_sink: Any

The collector's opt-in kernel event observer.

tenant property 

tenant: str

Who these calls are charged to, or "" for nobody.

contribute 

contribute(request: Any) -> Any

Declare the post-model stage and the handler that reports the call.

request is read for its grants and found to need none. Counting a call is not an effect on the turn: nothing downstream reads what this stage writes, and a grant would be authority never exercised.

Source code in src/symfonic/platform/telemetry.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def contribute(self, request: Any) -> Any:
    """Declare the post-model stage and the handler that reports the call.

    ``request`` is read for its grants and found to need none. Counting a
    call is not an effect on the turn: nothing downstream reads what this
    stage writes, and a grant would be authority never exercised.
    """
    from symfonic.kernel.contracts.contributions import CapabilityContribution

    return CapabilityContribution(
        capability=TELEMETRY_CAPABILITY,
        stages=(
            StageDescriptor(
                stage_id=TELEMETRY_STAGE,
                phase=Phase.POST_MODEL,
                capability=TELEMETRY_CAPABILITY,
                priority=_PRIORITY,
            ),
        ),
        handlers={TELEMETRY_STAGE: self._handle},
    )

TenantAgentRegistry

Bases: Protocol

The scope -> agent mapping, and the only place agents are created.

Concurrency is part of the contract, not an implementation detail: two coroutines asking for the same scope at once receive the same agent, and the composer runs once. A registry that raced would give one tenant two agents with two capability sets over one store, and the second would silently win.

aclose async 

aclose() -> None

Drop every agent. Idempotent. Does not close shared resources.

Source code in src/symfonic/platform/host.py 
128
129
130
async def aclose(self) -> None:
    """Drop every agent. Idempotent. Does not close shared resources."""
    ...

agent_for async 

agent_for(scope: Any) -> Any

The agent bound to scope, composing it on first use.

Raises:

Type Description
HostClosed

if the registry has been closed.
Source code in src/symfonic/platform/host.py 
112
113
114
115
116
117
118
async def agent_for(self, scope: Any) -> Any:
    """The agent bound to ``scope``, composing it on first use.

    Raises:
        HostClosed: if the registry has been closed.
    """
    ...

TenantAuthVerifier

Bases: Protocol

The deployment's own answer to "is this caller allowed in this tenant?".

Returns a mapping of principal facts (principal_id, is_admin) or None for a bare allow, and raises to deny. The admin bit comes from here and nowhere else (SCOPE-7): the shipped X-Admin-Override header convention is trusted only when a verifier has already authenticated the caller, which makes it a verifier-side deployment convention — the one place it can actually be reasoned about — rather than a transport rule.

ToolContribution dataclass 

ToolContribution(name: str, extension: str, invoke: Callable[[Mapping[str, Any]], Awaitable[str]], description: str = '', input_schema: Mapping[str, Any] = dict(), origin: str = '')

One callable an extension offers the agent.

invoke is an async callable taking the bound arguments and answering a string. It is captured at declaration time and never looked up again: the legacy MCP provider routed each call through a mutable name→server dict, so a later discovery could re-point an already-advertised tool at a different server. Holding the callable makes that unrepresentable.

UnknownPauseToken

Bases: LookupError

Raised when a token names no resumable turn.

A distinct type because the caller's correct response differs and cannot be derived from a message: an expired token is an ordinary outcome a transport turns into a 404 or a "this conversation moved on", while a malformed one is a bug in whoever minted it. Both are refusals, and only one is worth waking somebody for.

api_routes 

api_routes(app: Any, prefix: str = '/api/v1') -> list[MountedRoute]

The subset under prefix. The question most callers are asking.

Source code in src/symfonic/platform/routes.py 
62
63
64
def api_routes(app: Any, prefix: str = "/api/v1") -> list[MountedRoute]:
    """The subset under ``prefix``. The question most callers are asking."""
    return [route for route in mounted_routes(app) if route.path.startswith(prefix)]

attribution_is_bound 

attribution_is_bound(capability: TelemetryCapability) -> bool

Whether these calls can be charged to a tenant at all.

False means the agent was composed without a scope, so every call is recorded against the run and no per-tenant ceiling can ever fire. That is a legitimate choice for a single-tenant deployment and a silent billing hole for any other -- the same reason budget_is_shared exists next door.

Source code in src/symfonic/platform/telemetry.py 
210
211
212
213
214
215
216
217
218
def attribution_is_bound(capability: TelemetryCapability) -> bool:
    """Whether these calls can be charged to a tenant at all.

    ``False`` means the agent was composed without a scope, so every call is
    recorded against the run and no per-tenant ceiling can ever fire. That is a
    legitimate choice for a single-tenant deployment and a silent billing hole
    for any other -- the same reason ``budget_is_shared`` exists next door.
    """
    return bool(capability.tenant)

create_kernel_agent_router 

create_kernel_agent_router(host: Any, *, prefix: str = DEFAULT_PREFIX, memory: Any = None, graph: Any = None, scope_resolver: Any = None, memory_audit: Any = None) -> Any

A router that maps HTTP onto host.

Parameters:

Name Type Description Default
host Any

anything with async agent_for(scope). Structural, so a test double or a decorating host stands in without inheriting.

 required
prefix str

where the routes mount.

 DEFAULT_PREFIX
memory_audit Any

explicit acknowledged audit sink for point deletion. DELETE requires both this sink and a verified scope resolver; a missing or failing audit sink prevents storage mutation.

 None
scope_resolver Any

what turns a request's credentials into an authenticated principal, normally a :class:~symfonic.platform.HeaderScopeResolver. Without it the tenant header is taken on trust, which is a development posture: a caller who can set a header can then read any tenant. A deployment that authenticates must pass one.

 None
graph Any

an object with edges, neighborhood and export -- normally a :class:~symfonic.capabilities.memory.graph_admin.GraphAdminService. Optional, and its absence is why the routes below are mounted conditionally: a deployment with no graph view answers 404 rather than returning an empty list a client would read as "no relationships".

 None
memory Any

an object with records(scope, layer=, limit=) -- normally a :class:~symfonic.capabilities.memory.admin.MemoryAdminService over the host's store. Optional: a deployment that exposes no memory browser passes nothing and the route is not mounted at all, which is a 404 rather than an endpoint that answers with an empty list and lets a caller believe the tenant has no memories.

 None
Source code in src/symfonic/platform/transport.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 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
 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
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
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
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
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def create_kernel_agent_router(
    host: Any,
    *,
    prefix: str = DEFAULT_PREFIX,
    memory: Any = None,
    graph: Any = None,
    scope_resolver: Any = None,
    memory_audit: Any = None,
) -> Any:
    """A router that maps HTTP onto ``host``.

    Args:
        host: anything with ``async agent_for(scope)``. Structural, so a test
            double or a decorating host stands in without inheriting.
        prefix: where the routes mount.
        memory_audit: explicit acknowledged audit sink for point deletion.
            DELETE requires both this sink and a verified scope resolver;
            a missing or failing audit sink prevents storage mutation.
        scope_resolver: what turns a request's credentials into an
            authenticated principal, normally a
            :class:`~symfonic.platform.HeaderScopeResolver`. **Without it the
            tenant header is taken on trust**, which is a development posture:
            a caller who can set a header can then read any tenant. A
            deployment that authenticates must pass one.
        graph: an object with ``edges``, ``neighborhood`` and ``export`` --
            normally a
            :class:`~symfonic.capabilities.memory.graph_admin.GraphAdminService`.
            Optional, and its absence is why the routes below are mounted
            conditionally: a deployment with no graph view answers 404 rather
            than returning an empty list a client would read as "no
            relationships".
        memory: an object with ``records(scope, layer=, limit=)`` -- normally a
            :class:`~symfonic.capabilities.memory.admin.MemoryAdminService` over
            the host's store. Optional: a deployment that exposes no memory
            browser passes nothing and the route is not mounted at all, which
            is a 404 rather than an endpoint that answers with an empty list
            and lets a caller believe the tenant has no memories.
    """
    try:
        from fastapi import APIRouter, Body, Header, HTTPException
        from fastapi.responses import StreamingResponse
    except ImportError as missing:  # pragma: no cover - exercised by the gate
        # IA-4. A missing extra reaches the caller as an install command
        # rather than as a bare import error: the person who hits this is
        # standing up a deployment, and "no module named fastapi" does not
        # tell them which extra of which distribution provides it.
        raise ImportError(
            "the kernel-native router needs FastAPI, which ships with the "
            "'agent-api' extra: pip install 'symfonic-core[agent-api]'"
        ) from missing

    router = APIRouter(prefix=prefix)

    async def _authenticated_scope(
        tenant: str | None, authorization: str | None, *, principal_only: bool = False,
    ) -> Any:
        """The scope this request is *allowed* to name, not the one it claims.

        With a resolver, the tenant header is a claim the resolver checks
        against the caller's credentials. Without one, the header is taken at
        face value -- which is why the parameter exists and why a deployment
        that skips it has no tenant isolation at the transport.
        """
        if not tenant or not tenant.strip():
            raise HTTPException(
                status_code=401,
                detail=f"missing {TENANT_HEADER}: the caller is unidentified, "
                "so there is no tenant whose agent could serve this request",
            )
        if scope_resolver is None:
            if principal_only:
                raise HTTPException(401, "record deletion requires a verified identity")
            return scope_for_tenant(tenant.strip())

        principal = await verified_principal(scope_resolver, tenant.strip(), authorization)
        return principal if principal_only else principal.scope

    async def _agent_for(tenant: str | None, authorization: str | None) -> Any:
        scope = await _authenticated_scope(tenant, authorization)
        try:
            return await host.agent_for(scope)
        except HostClosed as closed:
            raise HTTPException(
                status_code=503,
                detail="the host is shutting down and is composing no further "
                "agents",
            ) from closed

    def _query(payload: dict[str, Any]) -> str:
        text = str(payload.get("query") or "").strip()
        if not text:
            raise HTTPException(
                status_code=422,
                detail="query must be a non-empty string",
            )
        return text

    def _history(payload: dict[str, Any]) -> tuple[Any, ...]:
        """Validate the browser's transcript at the transport boundary.

        ``session_id`` is correlation, not conversation state.  Treating it
        as though the facade would load a transcript made every request from
        the generated Chat page a first turn, even inside the same chat.
        Only user/assistant text crosses this public endpoint: a caller may
        not inject a system instruction or fabricate a tool result.
        """
        raw = payload.get("history", ())
        if raw in (None, ()):
            return ()
        if not isinstance(raw, list) or len(raw) > 100:
            raise HTTPException(
                status_code=422, detail="history must be a list of at most 100 messages"
            )
        from symfonic.agent.facade_types import Message

        parsed = []
        for item in raw:
            if not isinstance(item, dict):
                raise HTTPException(status_code=422, detail="invalid history message")
            role = item.get("role")
            content = item.get("content")
            if role not in ("user", "assistant") or not isinstance(content, str):
                raise HTTPException(
                    status_code=422,
                    detail="history messages require a user/assistant role and text content",
                )
            parsed.append(Message(role=role, content=content))
        return tuple(parsed)

    def _session_id(payload: dict[str, Any]) -> str:
        value = payload.get("session_id", "")
        if not isinstance(value, str):
            raise HTTPException(status_code=422, detail="session_id must be text")
        value = value.strip()
        if len(value) > 64:
            raise HTTPException(status_code=422, detail="session_id too long (max 64)")
        return value
    @router.post("/chat")
    async def chat(
        payload: dict[str, Any] = Body(...),  # noqa: B008 - FastAPI idiom
        x_tenant_id: str | None = Header(default=None),  # noqa: B008
        authorization: str | None = Header(default=None),  # noqa: B008
    ) -> dict[str, Any]:
        agent = await _agent_for(x_tenant_id, authorization)
        try:
            result = await agent.run(
                _query(payload), history=_history(payload), session_id=_session_id(payload)
            )
        except HTTPException:
            raise
        except asyncio.CancelledError:
            # Never mapped: a cancelled request has no response to send, and
            # swallowing it here would leave the turn's teardown to a task
            # nobody is waiting on.
            raise
        except Exception as failed:  # noqa: BLE001
            # A turn that fails upstream is a 502, not a 500: the caller's
            # request was well formed and the fault is behind us. Left
            # unmapped it escaped as an unhandled ASGI exception, which some
            # clients re-raise instead of reporting a status at all.
            raise HTTPException(
                status_code=502,
                detail=f"the turn failed: {failed}",
            ) from failed
        return {"response": text_of(result)}

    @router.post("/stream")
    async def stream(
        payload: dict[str, Any] = Body(...),  # noqa: B008 - FastAPI idiom
        x_tenant_id: str | None = Header(default=None),  # noqa: B008
        authorization: str | None = Header(default=None),  # noqa: B008
    ) -> Any:
        agent = await _agent_for(x_tenant_id, authorization)
        query = _query(payload)
        history = _history(payload)

        async def events():
            async for chunk in agent.stream(
                query, history=history, session_id=_session_id(payload)
            ):
                # Only chunks that CARRY text. A stream yields terminal events
                # too -- the last one holds the whole result and no text -- and
                # falling back to ``str(event)`` for those put a full repr on
                # the wire as though it were content. A client would render it.
                text = getattr(chunk, "text", None)
                if isinstance(text, str) and text:
                    yield f"data: {json.dumps({'chunk': text})}\n\n"
            yield "data: [DONE]\n\n"

        return StreamingResponse(events(), media_type="text/event-stream")

    @router.post("/stream/typed")
    async def stream_typed(
        payload: dict[str, Any] = Body(...),  # noqa: B008 - FastAPI idiom
        x_tenant_id: str | None = Header(default=None),  # noqa: B008
        authorization: str | None = Header(default=None),  # noqa: B008
    ) -> Any:
        """The same turn, with each event's kind on the wire.
        Derived from ``stream`` rather than from a typed API, because the
        facade has one stream and it already carries the kind. Every event is
        forwarded -- including the ones with no text, which is the difference
        from ``/stream``: a client that needs to tell a tool call from a token
        needs the events ``/stream`` deliberately drops.
        """
        agent = await _agent_for(x_tenant_id, authorization)
        query = _query(payload)
        history = _history(payload)

        async def events():
            async for event in agent.stream(
                query, history=history, session_id=_session_id(payload)
            ):
                body: dict[str, Any] = {"type": kind_of(event)}
                for field in ("text", "error", "index"):
                    value = getattr(event, field, None)
                    if value is not None:
                        body[field] = value
                yield f"data: {json.dumps(body)}\n\n"
            yield "data: [DONE]\n\n"

        return StreamingResponse(events(), media_type="text/event-stream")

    mount_data_routes(
        router, memory=memory, graph=graph, resolve=_authenticated_scope
    )
    mount_record_routes(router, memory=memory, resolve=_authenticated_scope,
                        principal=_authenticated_scope, sink=memory_audit)

    return router

fence_denial_recorder 

fence_denial_recorder(audit: AuditSeam) -> Callable[[FenceDenial], Awaitable[None]]

A recorder a row-14 fence can call on every denial.

Not destructive: the denial prevented a mutation, so failing the already-refused write closed a second time would add nothing. The seam still degrades loudly (AUD-4) if the sink is down.

Source code in src/symfonic/platform/fence_audit.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def fence_denial_recorder(audit: AuditSeam) -> Callable[[FenceDenial], Awaitable[None]]:
    """A recorder a row-14 fence can call on every denial.

    Not ``destructive``: the denial *prevented* a mutation, so failing the
    already-refused write closed a second time would add nothing. The seam still
    degrades loudly (AUD-4) if the sink is down.
    """

    async def record(denial: FenceDenial) -> None:
        await audit.record(
            AuditRecord(
                action=FENCE_DENIED_ACTION,
                outcome=str(denial.reason),
                principal_id="erasure-fence",
                scope_key=denial.scope_key,
                resource_type="subject",
                resource_id=denial.scope_key,
                metadata={
                    "reason": str(denial.reason),
                    "expected_generation": denial.expected_generation,
                    "observed_generation": denial.observed_generation,
                },
            )
        )

    return record

mounted_routes 

mounted_routes(app: Any) -> list[MountedRoute]

Every endpoint app serves, whichever FastAPI version built it.

Walks included routers rather than trusting app.routes to be flat. A route that carries no HTTP method -- a mount, a static files app -- is skipped: this answers "what can be called", and those cannot.

Source code in src/symfonic/platform/routes.py 
52
53
54
55
56
57
58
59
def mounted_routes(app: Any) -> list[MountedRoute]:
    """Every endpoint ``app`` serves, whichever FastAPI version built it.

    Walks included routers rather than trusting ``app.routes`` to be flat. A
    route that carries no HTTP method -- a mount, a static files app -- is
    skipped: this answers "what can be called", and those cannot.
    """
    return sorted(set(_walk(getattr(app, "routes", ()) or ())))

production_auth_gate 

production_auth_gate(resolver: object) -> None

HOST-5/HOST-6 — fail closed at startup, or log CRITICAL and continue.

Takes the resolver structurally (anything with posture()) so host assembly can gate an adopter's own resolver, not only the shipped one. A host that cannot build its auth path does not start in a degraded "allow everything" mode.

Source code in src/symfonic/platform/posture.py 
 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
def production_auth_gate(resolver: object) -> None:
    """HOST-5/HOST-6 — fail closed at startup, or log CRITICAL and continue.

    Takes the resolver structurally (anything with ``posture()``) so host
    assembly can gate an adopter's own resolver, not only the shipped one. A
    host that cannot build its auth path does not start in a degraded
    "allow everything" mode.
    """
    posture_of = getattr(resolver, "posture", None)
    if not callable(posture_of):
        raise RuntimeError(
            "the production auth gate needs a resolver that can report its "
            "posture(); a resolver that cannot describe whether it authenticates "
            "cannot be gated, and an ungateable auth path is not a safe default"
        )
    posture: AuthPosture = posture_of()
    if not posture.production:
        return
    if posture.insecure_override:
        logger.critical(
            "ALLOW_INSECURE_PROD detected — tenant auth disabled in production. "
            "Data WILL leak across tenants. Register a verifier immediately.",
        )
        return
    if not posture.verifier_registered:
        raise RuntimeError(
            "Production environment detected but no tenant auth verifier is "
            "registered. Construct the ScopeResolver with a verifier before "
            "mounting routers, or set ALLOW_INSECURE_PROD=true to bypass "
            "(NOT RECOMMENDED).",
        )