symfonic.agent.cutover.extensions

extensions

Extensions, folded into the bundle the migrated path compiles (TA8.21).

RET-PREP/envelope-guard-inventory.json recorded plugins as status: absent with three blockers. This module is the answer to all three, and each is named where it is closed:

1. "src/symfonic/capabilities/extensions/ has no capability.py and therefore no contribute(), so fold_contributions cannot fold it." :class:ExtensionsCapability is that contribute(). It lives here, at the composition root, rather than inside the capability: that package's own attestation (tests/capabilities/extensions/test_no_engine_mutation.py) forbids it from naming symfonic.kernel at all, and a CapabilityContribution cannot be built without it. The RCH-1 waiver named the gap precisely — "ExtensionBridge.contribute() answers an ExtensionContribution; fold_contributions needs a CapabilityContribution. Nothing adapts one to the other." Adapting one to the other is a root's job, and this is the root.
2. "nothing outside the package imports it." This module does, at module scope, which is what makes the package reached for RCH-1.
3. "the extension contract includes ToolContribution, so it inherits the same missing bundle tool path as sub_agents." TA8.12 built that path — bind_contributed_tools → RetrievalBundle.tools → merge_capability_tools — and this capability offers its tools into it.

The prompt half. Fragments arrive through the bundle's prompt seam, not beside it: a resolution stage harvests each plugin's hook once per turn and leaves compiler-ready mappings in the turn's resolved-input snapshot, which is where PromptingCapability._contributions_from reads. That is the seam memory already uses, used as intended, and it is why nothing here imports the prompt compiler. What the remap costs an adopter is decided in evidence/RET-PREP/decision-plugin-tier-layer.md and implemented in :mod:symfonic.agent.cutover.fragments.

The guardrail half. :func:~symfonic.agent.cutover.guardrails.enforce_guardrails is the enforcement point for contributed policies, and SymfonicAgent.validate_action is its one caller — the same shape TA8.19 gave session_id: one derivation site, in the position the legacy body inlined it, so both routes enforce the same guard rather than two that agree by inspection.

Nothing here registers a tool late. ExtensionSurface seals the tool set when this capability is constructed, which is before AgentPlanFactory is. See :mod:symfonic.capabilities.extensions.harvest.

Three modules, one seam. This one is the kernel-facing adapter; the legacy end (bridging an adopter's plugin, and the state its hook is handed) is :mod:~symfonic.agent.cutover.plugin_providers, and the guardrail enforcement point is :mod:~symfonic.agent.cutover.guardrails. Both left here when this module passed its line budget, the same split rounds made from delegate and authorised from bundle. Both are re-exported below, because this is the address importers and evidence files already name.

ExtensionsCapability

ExtensionsCapability(surface: ExtensionSurface, *, state_for: Callable[[Any], Mapping[str, Any]] = plugin_state)

One agent's extensions, as something fold_contributions can fold.

Thin on purpose. It owns no rule: the admission rules are capabilities/extensions/admission.py's, the seam is :class:~symfonic.capabilities.extensions.harvest.ExtensionSurface's, and the remap is :mod:~symfonic.agent.cutover.fragments'. What is here is the two-way translation those three cannot perform without importing each other.

Source code in src/symfonic/agent/cutover/extensions.py

def __init__(
    self,
    surface: ExtensionSurface,
    *,
    state_for: Callable[[Any], Mapping[str, Any]] = plugin_state,
) -> None:
    self.surface = surface
    self._state_for = state_for
    # Bound **here**, at construction, over the tuple the surface sealed at
    # its own construction. Two seams, one moment: both happen before
    # ``AgentPlanFactory`` exists, so the tool set the plan compiles from is
    # decided before a turn can be admitted. Binding lazily in
    # ``contribute`` would have moved a schema decision onto the per-turn
    # fold; binding in the delegate would have moved a ``ConfigurationError``
    # into an already-admitted turn.
    self._tools = tuple(bind_extension_tool(tool) for tool in surface.tools)

contribute

contribute(request: CapabilityRequest) -> CapabilityContribution

Declare the resolution stage, its handler, and the sealed tools.

request is read for its grants and found to need none. A plugin's hook is arbitrary code and may reach a network, which is exactly why this is a RESOLUTION stage rather than a compilation one -- but the framework grants effect families, and there is no family that means "whatever a third party does". Declaring one it cannot bound would make the grant list a description rather than a bound, so it declares none and the containment stays where it already is: every hook failure is caught by LegacyPromptReader and becomes a diagnostic.

Source code in src/symfonic/agent/cutover/extensions.py

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Declare the resolution stage, its handler, and the sealed tools.

    ``request`` is read for its grants and found to need none. A plugin's
    hook is arbitrary code and may reach a network, which is exactly why
    this is a ``RESOLUTION`` stage rather than a compilation one -- but the
    framework grants effect *families*, and there is no family that means
    "whatever a third party does". Declaring one it cannot bound would make
    the grant list a description rather than a bound, so it declares none
    and the containment stays where it already is: every hook failure is
    caught by ``LegacyPromptReader`` and becomes a diagnostic.
    """

    async def handle(context: Any) -> StageResult[Any]:
        turn_request = getattr(context, "request", None)
        if turn_request is None:  # pragma: no cover - defensive
            return no_change("no turn request on the stage context")

        composed = await self.surface.harvest(self._state_for(turn_request))
        for refusal in composed.refusals:
            logger.warning(
                "extension %r: %s", refusal.extension, refusal.detail
            )
        if not composed.prompts:
            # An extension with nothing to say this turn and one that raised
            # are different facts; the diagnostics above carry the second.
            # Contributing an empty entry would cost the compiler a
            # delimiter and a budget slot for no content.
            return no_change(
                "no extension contributed a prompt fragment this turn"
            )
        return applied(
            ResolvedInput(
                capability=EXTENSIONS_CAPABILITY,
                value=tuple(
                    fragment_spec(fragment) for fragment in composed.prompts
                ),
                # Which extensions were asked, not a digest of what they
                # said: it is what gets a reader of a compiled prompt from a
                # rendered line back to the plugin that produced it.
                provenance=",".join(composed.extensions),
            )
        )

    return CapabilityContribution(
        capability=EXTENSIONS_CAPABILITY,
        stages=(
            StageDescriptor(
                stage_id=EXTENSIONS_STAGE,
                phase=Phase.PROMPT_ASSEMBLY,
                capability=EXTENSIONS_CAPABILITY,
                priority=_EXTENSIONS_PRIORITY,
                optional_before=(PROMPT_COMPILER_STAGE,),
                effects=frozenset(),
                kind=StageKind.RESOLUTION,
            ),
        ),
        handlers={EXTENSIONS_STAGE: handle},
        effect_grants=frozenset(),
        # Sealed at construction, already bound. For a bridged legacy plugin
        # this is always empty -- ``get_domain_tools`` must return ``[]``
        # because the tool catalogue closes at compile -- and the field
        # exists so a provider that legitimately declares one *before* the
        # seam reaches the plan through TA8.12's path instead of vanishing.
        tools=self._tools,
    )

PluginProviders

PluginProviders(plugins: Callable[[], Sequence[Any]], *, limits: TrustLimits | None = None)

The engine's live plugin list, read as extension providers.

One bridge per plugin object, kept across turns. Rebuilding them per turn would re-run LegacyPluginBridge.__init__, which calls the plugin's get_domain_tools() and re-validates its name on every dispatch -- work bridge.py states happens once, at construction and never per run.

A callable rather than a fixed sequence, because the roster is read again on every turn and on every guarded action. It does not reopen the tool question: :class:~symfonic.capabilities.extensions.harvest.ExtensionSurface seals that at the compile seam.

A plugin that cannot be bridged, or whose name is already claimed, is refused whole. Both used to be survivable here -- the first by dropping the plugin, the second by re-emitting it under a synthetic identity -- and both were the same defect: an adopter's plugin was accepted by load_plugin and then served by less than the whole of what it declared, with the shortfall visible only in a log line. Raising :class:~symfonic.capabilities.extensions.errors.ExtensionAdmissionError instead moves the decision to the one place an adopter can act on it. The engine calls this on a trial roster from load_plugin before it registers anything, so the raise becomes a refused load rather than an agent that cannot take a turn.

A refusal never costs a veto. The plugin the engine refused goes into its quarantine and validate_action still asks it, deny-only; and if a roster is somehow made unbridgeable after admission -- a plugin appended to _plugins directly, a name mutated in place -- this raise reaches _extensions_capability, which answers None, and :func:~symfonic.agent.cutover.guardrails.enforce_guardrails asks the whole population directly. There is no path from a refusal to an allow.

Source code in src/symfonic/agent/cutover/plugin_providers.py

def __init__(
    self,
    plugins: Callable[[], Sequence[Any]],
    *,
    limits: TrustLimits | None = None,
) -> None:
    self._plugins = plugins
    self._limits = limits
    self._bridges: list[tuple[Any, Any]] = []

enforce_guardrails async

enforce_guardrails(capability: Any, action: str, context: Mapping[str, Any], *, plugins: Sequence[Any] = (), quarantined: Sequence[Any] = (), warned: MutableMapping[str, int] | None = None) -> bool

Ask every contributed policy about action; False if one refuses.

Answers bool because that is what the public method has always answered and an adopter's if not await agent.validate_action(...) has to keep working. The richer :class:~symfonic.capabilities.extensions.composition.PolicyOutcome -- the verdict and every guard that failed to answer -- is available from ExtensionSurface.decide for a caller that wants it, which is what makes the abstention below a value rather than only a log line.

None with no plugins is an allow, and it is the branch every agent in the fleet takes: an agent that loaded no plugin folded no capability and has no contributed guard to consult.

None with plugins loaded is not the same fact, and is not treated as one -- see :func:_unfolded_outcome. The capability is the ordinary route and the plugin list is the floor beneath it, so a build that failed costs the fragments and the tools it was carrying and does not cost the vetoes.

warned is the caller's latch for the report that path emits, and is passed through untouched; it changes no verdict.

quarantined are the plugins whose admission this agent refused. They are asked first and deny-only -- see :func:_quarantine_outcome. Nothing about them is registered; what a refusal may not do is delete a veto the adopter installed.

Fail-closed. A guard that raises, or answers something that is not a verdict, abstains -- and an abstention denies. That closes SEC-FCP-4 / TM-17, filed HIGH against the inline loop this replaced, which swallowed the exception and allowed. The abstention is still a value rather than a swallowed exception, so "the only guard with an opinion was down" and "nobody objected" remain different readings of a turn; they simply no longer produce the same verdict.

Source code in src/symfonic/agent/cutover/guardrails.py

async def enforce_guardrails(
    capability: Any,
    action: str,
    context: Mapping[str, Any],
    *,
    plugins: Sequence[Any] = (),
    quarantined: Sequence[Any] = (),
    warned: MutableMapping[str, int] | None = None,
) -> bool:
    """Ask every contributed policy about ``action``; ``False`` if one refuses.

    Answers ``bool`` because that is what the public method has always answered
    and an adopter's ``if not await agent.validate_action(...)`` has to keep
    working. The richer
    :class:`~symfonic.capabilities.extensions.composition.PolicyOutcome` -- the
    verdict *and* every guard that failed to answer -- is available from
    ``ExtensionSurface.decide`` for a caller that wants it, which is what makes
    the abstention below a value rather than only a log line.

    ``None`` with no ``plugins`` is an allow, and it is the branch every agent
    in the fleet takes: an agent that loaded no plugin folded no capability and
    has no contributed guard to consult.

    ``None`` *with* plugins loaded is not the same fact, and is not treated as
    one -- see :func:`_unfolded_outcome`. The capability is the ordinary route
    and the plugin list is the floor beneath it, so a build that failed costs
    the fragments and the tools it was carrying and does not cost the vetoes.

    ``warned`` is the caller's latch for the report that path emits, and is
    passed through untouched; it changes no verdict.

    ``quarantined`` are the plugins whose *admission* this agent refused. They
    are asked first and deny-only -- see :func:`_quarantine_outcome`. Nothing
    about them is registered; what a refusal may not do is delete a veto the
    adopter installed.

    **Fail-closed.** A guard that raises, or answers something that is not a
    verdict, abstains -- and an abstention denies. That closes SEC-FCP-4 /
    TM-17, filed HIGH against the inline loop this replaced, which swallowed
    the exception and allowed. The abstention is still a *value* rather than a
    swallowed exception, so "the only guard with an opinion was down" and
    "nobody objected" remain different readings of a turn; they simply no
    longer produce the same verdict.
    """
    if quarantined:
        refused = await _quarantine_outcome(quarantined, action, context)
        if not _allows(refused, action):
            return False
    if capability is None:
        if not plugins:
            return True
        outcome = await _unfolded_outcome(plugins, action, context, warned)
    else:
        outcome = await capability.surface.decide(action, context)
    return _allows(outcome, action)

extensions_for_plugins

extensions_for_plugins(plugins: Callable[[], Sequence[Any]], *, limits: TrustLimits | None = None, reserved_tool_names: frozenset[str] = frozenset()) -> ExtensionsCapability | None

Wrap the engine's plugin list as the capability the bundle folds.

None when nothing composed, so the caller folds no extensions rather than folding a capability whose stage can only ever report NO_CHANGE. That is the same rule delegation_for_children applies to an empty roster, one layer out.

Source code in src/symfonic/agent/cutover/extensions.py

def extensions_for_plugins(
    plugins: Callable[[], Sequence[Any]],
    *,
    limits: TrustLimits | None = None,
    reserved_tool_names: frozenset[str] = frozenset(),
) -> ExtensionsCapability | None:
    """Wrap the engine's plugin list as the capability the bundle folds.

    ``None`` when nothing composed, so the caller folds no extensions rather
    than folding a capability whose stage can only ever report ``NO_CHANGE``.
    That is the same rule ``delegation_for_children`` applies to an empty
    roster, one layer out.
    """
    surface = ExtensionSurface(
        PluginProviders(plugins, limits=limits),
        reserved_tool_names=reserved_tool_names,
    )
    capability = ExtensionsCapability(surface)
    return capability if capability.active else None

plugin_state

plugin_state(request: Any) -> Mapping[str, Any]

The state a plugin's prompt hook is handed on the migrated path (D4).

Built from the turn, not from a rendered HMS template. The legacy HMS-shaped keys -- TOOL_MANIFEST, MEMORY_CONTEXT, SOUL_SCHEMA and the rest -- are deliberately absent, and evidence/RET-PREP/decision-plugin-tier-layer.md D4 is where that is decided rather than discovered: several of them are the output of the legacy prompt build this path replaces, so supplying them would mean running that build in order to feed a plugin whose output then goes somewhere else entirely.

Read-only, because it is handed to third-party code: a hook that edited the mapping would otherwise change what the next plugin was told about the same turn.

Source code in src/symfonic/agent/cutover/plugin_providers.py

def plugin_state(request: Any) -> Mapping[str, Any]:
    """The state a plugin's prompt hook is handed on the migrated path (D4).

    Built from the *turn*, not from a rendered HMS template. The legacy
    HMS-shaped keys -- ``TOOL_MANIFEST``, ``MEMORY_CONTEXT``, ``SOUL_SCHEMA``
    and the rest -- are deliberately absent, and
    ``evidence/RET-PREP/decision-plugin-tier-layer.md`` D4 is where that is
    decided rather than discovered: several of them are the *output* of the
    legacy prompt build this path replaces, so supplying them would mean running
    that build in order to feed a plugin whose output then goes somewhere else
    entirely.

    Read-only, because it is handed to third-party code: a hook that edited the
    mapping would otherwise change what the next plugin was told about the same
    turn.
    """
    scope = getattr(request, "scope", None)
    path = str(getattr(scope, "path", "") or "") if scope is not None else ""
    return MappingProxyType(
        {
            "QUERY": str(getattr(request, "prompt", "") or ""),
            # The tenant is the scope path's root segment; a deeper scope still
            # belongs to its own tenant. Same derivation as
            # ``capabilities/prompting/hms._tenant_of``, and named the same key
            # legacy used, so a plugin reading ``TENANT_ID`` keeps working.
            "TENANT_ID": path.split("/", 1)[0] if path else "",
            "SCOPE_PATH": path,
        }
    )