symfonic.capabilities.extensions.composition

composition

Composing extensions into one frozen value (T4.2.2).

:func:compose is a pure function over contributions. It holds no module state, touches no engine, and returns a :class:ComposedExtensions — so "what do my extensions add up to?" is a question with an answer you can print, diff between two deployments, and assert equal across two runs.

The admission rules — reserved names, first-declaration-wins, total ordering — live in :mod:~symfonic.capabilities.extensions.admission. One rule stays here because it is a property of the composed value rather than of any single contribution:

Policies may only refuse. :meth:ComposedExtensions.decide combines verdicts deny-wins and treats ALLOW and ABSTAIN alike as "no objection". An extension cannot vote to permit something the host would refuse, so adding one can only narrow what the agent does (AS-INT-3).

ComposedExtensions dataclass

ComposedExtensions(extensions: tuple[str, ...] = (), tools: tuple[ToolContribution, ...] = (), prompts: tuple[PromptFragment, ...] = (), policies: tuple[PolicyContribution, ...] = (), lifecycle: tuple[LifecycleContribution, ...] = (), diagnostics: tuple[ExtensionDiagnostic, ...] = ())

What a set of extensions adds up to, and what was refused on the way.

refusals property

refusals: tuple[ExtensionDiagnostic, ...]

Only the error-severity diagnostics: what an operator must see.

decide async

decide(request: PolicyRequest) -> PolicyVerdict

Combine every applicable policy's verdict, deny-wins.

Answers only the combined verdict. Use :meth:decide_detailed when the caller needs to know which guards failed to answer as well.

Source code in src/symfonic/capabilities/extensions/composition.py

async def decide(self, request: PolicyRequest) -> PolicyVerdict:
    """Combine every applicable policy's verdict, deny-wins.

    Answers only the combined verdict. Use :meth:`decide_detailed` when the
    caller needs to know *which* guards failed to answer as well.
    """
    return (await self.decide_detailed(request)).verdict

decide_detailed async

decide_detailed(request: PolicyRequest) -> PolicyOutcome

Ask every applicable policy and report the verdict with the abstentions.

The first DENY short-circuits: once one policy has refused, asking the rest costs latency on a turn that is already refused, and no later verdict could change the answer. The abstentions reported are therefore those gathered before the refusal, which is the whole set whenever nothing denied.

A policy that raises abstains. That is the legacy engine's fail-open behaviour preserved on purpose — flipping it to fail-closed would turn a broken guard into a total outage — but the abstention is now a value the caller can see rather than a DEBUG log nobody reads.

Source code in src/symfonic/capabilities/extensions/composition.py

async def decide_detailed(self, request: PolicyRequest) -> PolicyOutcome:
    """Ask every applicable policy and report the verdict with the abstentions.

    The first DENY short-circuits: once one policy has refused, asking the
    rest costs latency on a turn that is already refused, and no later
    verdict could change the answer. The abstentions reported are therefore
    those gathered *before* the refusal, which is the whole set whenever
    nothing denied.

    A policy that raises abstains. That is the legacy engine's fail-open
    behaviour preserved on purpose — flipping it to fail-closed would turn
    a broken guard into a total outage — but the abstention is now a value
    the caller can see rather than a DEBUG log nobody reads.
    """
    abstentions: list[PolicyVerdict] = []
    for policy in self.policies:
        if not policy.applies_to(request.action):
            continue
        verdict = await self._ask(policy, request)
        if verdict.denied:
            return PolicyOutcome(verdict=verdict, abstentions=tuple(abstentions))
        if verdict.abstained:
            abstentions.append(verdict)
    return PolicyOutcome(
        verdict=PolicyVerdict.allow(reason="no contributed policy objected"),
        abstentions=tuple(abstentions),
    )

install async

install() -> tuple[ExtensionDiagnostic, ...]

Run every install hook. A hook that raises costs a diagnostic.

Source code in src/symfonic/capabilities/extensions/composition.py

async def install(self) -> tuple[ExtensionDiagnostic, ...]:
    """Run every install hook. A hook that raises costs a diagnostic."""
    return await self._run_phase(LifecyclePhase.INSTALL)

teardown async

teardown() -> tuple[ExtensionDiagnostic, ...]

Run every teardown hook, in reverse declaration order.

Reverse because unwinding is the inverse of building: an extension installed after another may depend on it, and closing in declaration order would tear the dependency out first.

Source code in src/symfonic/capabilities/extensions/composition.py

async def teardown(self) -> tuple[ExtensionDiagnostic, ...]:
    """Run every teardown hook, in reverse declaration order.

    Reverse because unwinding is the inverse of building: an extension
    installed after another may depend on it, and closing in declaration
    order would tear the dependency out first.
    """
    return await self._run_phase(LifecyclePhase.TEARDOWN, reverse=True)

PolicyOutcome dataclass

PolicyOutcome(verdict: PolicyVerdict, abstentions: tuple[PolicyVerdict, ...] = ())

The combined verdict and every policy that failed to answer.

Exists because "nobody objected" and "the only guard that had an opinion was down" are different facts about a turn, and :meth:ComposedExtensions.decide can only return one verdict. An abstention recorded here is the value that keeps a permanently broken guard distinguishable from an approving one.

compose

compose(contributions: Sequence[ExtensionContribution], *, reserved_tool_names: frozenset[str] = frozenset()) -> ComposedExtensions

Validate, de-duplicate, and order every contribution into one value.

Parameters:

Name	Type	Description	Default
contributions	Sequence[ExtensionContribution]	The bundles, in the order the deployment declared them. Order is the tie-break for every collision, so it is the deployment's statement of precedence.	required
reserved_tool_names	frozenset[str]	Names the host already owns. A contributed tool claiming one is refused.	frozenset()

Raises:

Type	Description
ExtensionContractError	A contribution is malformed.
PrivilegeWideningError	A contribution claims authority it lacks — an authored tier, a foreign attribution, the kernel layer.

Source code in src/symfonic/capabilities/extensions/composition.py

def compose(
    contributions: Sequence[ExtensionContribution],
    *,
    reserved_tool_names: frozenset[str] = frozenset(),
) -> ComposedExtensions:
    """Validate, de-duplicate, and order every contribution into one value.

    Args:
        contributions: The bundles, in the order the deployment declared them.
            Order is the tie-break for every collision, so it is the
            deployment's statement of precedence.
        reserved_tool_names: Names the host already owns. A contributed tool
            claiming one is refused.

    Raises:
        ExtensionContractError: A contribution is malformed.
        PrivilegeWideningError: A contribution claims authority it lacks —
            an authored tier, a foreign attribution, the kernel layer.
    """
    diagnostics: list[ExtensionDiagnostic] = []
    tools: list[ToolContribution] = []
    prompts: list[PromptFragment] = []
    policies: list[PolicyContribution] = []
    lifecycle: list[LifecycleContribution] = []
    seen: list[str] = []
    tool_names: set[str] = set()
    fragment_ids: set[str] = set()

    for contribution in contributions:
        contribution.validate()
        if contribution.extension in seen:
            diagnostics.append(
                ExtensionDiagnostic(
                    extension=contribution.extension,
                    kind=ContributionKind.TOOL,
                    detail="refused: a second contribution under this name; one "
                    "extension contributes once per composition.",
                    severity=Severity.ERROR,
                )
            )
            continue
        seen.append(contribution.extension)
        for tool in contribution.tools:
            admit_tool(tool, tools, tool_names, reserved_tool_names, diagnostics)
        for fragment in contribution.prompts:
            admit_fragment(fragment, prompts, fragment_ids, diagnostics)
        policies.extend(contribution.policies)
        lifecycle.extend(contribution.lifecycle)

    return ComposedExtensions(
        extensions=tuple(seen),
        tools=tuple(sorted(tools, key=lambda t: (t.extension, t.name))),
        prompts=tuple(sorted(prompts, key=fragment_key)),
        policies=tuple(sorted(policies, key=lambda p: (p.extension, p.policy_id))),
        lifecycle=tuple(lifecycle),
        diagnostics=tuple(diagnostics),
    )