symfonic.capabilities.extensions.bridge

bridge

The plugin compatibility bridge — a legacy plugin as an extension (T4.2.2).

Every plugin an adopter has already written keeps working. The bridge reads one of them and answers typed contributions, so nothing downstream needs to know which API the plugin was written against.

What it maps, and what each mapping costs:

inject_contributions / inject_system_prompt → :class:~.declarations.PromptFragment per contribution, via :class:~.legacy_prompts.LegacyPromptReader. That module owns the position and tier mapping and the argument for each.

validate_state_transition → one :class:~.declarations.PolicyContribution. False becomes DENY, True becomes ALLOW, and a raise becomes ABSTAIN — a value naming the failure rather than a swallowed exception. What an abstention costs is the enforcement point's decision, not this bridge's: :func:~symfonic.agent.cutover.guardrails.enforce_guardrails denies on one (TA8.21, closing SEC-FCP-4). Keeping the distinction here is what lets a trace say "the only guard with an opinion was down" instead of "denied".

get_domain_tools → refused. A non-empty return raises :class:~.errors.PrivilegeWideningError at construction, keeping the v8.6.6 refusal but moving it off the engine and giving it a name that says what it is. Domain tools are constructor input, because the catalogue is closed when the graph compiles.

install / teardown / aclose → :class:~.declarations.LifecycleContribution. Optional; most legacy plugins define none, and the bridge contributing none for them is the honest answer rather than a synthesised no-op. A plugin holding an HTTP client previously had no place to close it, so the client closed when the process did.

The bridge never touches the plugin. It reads attributes, calls the hooks the plugin already exposes, and holds the results — no attribute is assigned, no method is wrapped in place, and the plugin object leaves a bridge exactly as it arrived.

LegacyPluginBridge

LegacyPluginBridge(plugin: Any, *, limits: TrustLimits | None = None)

One legacy plugin, read as an extension provider.

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

def __init__(self, plugin: Any, *, limits: TrustLimits | None = None) -> None:
    self._plugin = plugin
    self._limits = limits or TrustLimits()
    self._limits.validate()
    self._name = self._read_name(plugin)
    self._reader = LegacyPromptReader(plugin, self._name, limits=self._limits)
    self._prompts: tuple[PromptFragment, ...] = ()
    self._refuse_tools()

contribute

contribute() -> ExtensionContribution

The bundle, using whatever the last :meth:harvest produced.

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

def contribute(self) -> ExtensionContribution:
    """The bundle, using whatever the last :meth:`harvest` produced."""
    return ExtensionContribution.build(
        self._name,
        prompts=self._prompts,
        policies=self.policies(),
        lifecycle=self.lifecycle(),
    )

contribute_for async

contribute_for(state: Mapping[str, Any]) -> ExtensionContribution

Harvest this turn's prompts, then answer the whole bundle.

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

async def contribute_for(self, state: Mapping[str, Any]) -> ExtensionContribution:
    """Harvest this turn's prompts, then answer the whole bundle."""
    await self.harvest(state)
    return self.contribute()

harvest async

harvest(state: Mapping[str, Any]) -> tuple[PromptFragment, ...]

Call the plugin's prompt hook once and keep what it answered.

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

async def harvest(self, state: Mapping[str, Any]) -> tuple[PromptFragment, ...]:
    """Call the plugin's prompt hook once and keep what it answered."""
    self._prompts = await self._reader.read(state)
    return self._prompts

lifecycle

lifecycle() -> tuple[LifecycleContribution, ...]

Bind whichever optional lifecycle hooks the plugin actually defines.

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

def lifecycle(self) -> tuple[LifecycleContribution, ...]:
    """Bind whichever optional lifecycle hooks the plugin actually defines."""
    hooks: list[LifecycleContribution] = []
    for attribute, phase in _LIFECYCLE_HOOKS:
        hook = getattr(self._plugin, attribute, None)
        if not callable(hook):
            continue
        hooks.append(
            LifecycleContribution(
                hook_id=f"{self._name}.{attribute}",
                extension=self._name,
                phase=phase,
                run=hook,
            )
        )
    return tuple(hooks)

policies

policies() -> tuple[PolicyContribution, ...]

One policy when the plugin has a guard, none when it does not.

A plugin without validate_state_transition contributes no policy rather than an always-allow one. An always-allow stage in a trace is indistinguishable from a stage that examined the action and approved.

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

def policies(self) -> tuple[PolicyContribution, ...]:
    """One policy when the plugin has a guard, none when it does not.

    A plugin without ``validate_state_transition`` contributes *no* policy
    rather than an always-allow one. An always-allow stage in a trace is
    indistinguishable from a stage that examined the action and approved.
    """
    return legacy_guard_policies(self._plugin, self._name)

tools

tools() -> tuple[ToolContribution, ...]

Always empty. Present so the bridge reads like every other adapter.

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

def tools(self) -> tuple[ToolContribution, ...]:
    """Always empty. Present so the bridge reads like every other adapter."""
    return ()

legacy_guard_policies

legacy_guard_policies(plugin: Any, name: str) -> tuple[PolicyContribution, ...]

One legacy plugin's validate_state_transition, as a policy — or none.

Module-level rather than a :class:LegacyPluginBridge method because the guard outlives the bridge: a plugin the bridge refuses still has a veto, and dropping it would fail open in a way the pre-refactor inline loop over self._plugins never did. :func:~symfonic.agent.cutover.guardrails.enforce_guardrails is that caller -- it asks the engine's quarantine of refused plugins, and the plugin list itself when no capability folded -- and sharing this function is what keeps those from being separate implementations of deny-wins that agree by inspection.

A plugin without the hook contributes no policy rather than an always-allow one: an always-allow stage in a trace is indistinguishable from a stage that examined the action and approved it.

The hook is handed a copy, and that is a narrowing. The inline loop this replaced passed SymfonicAgent.validate_action's own context object straight through, so a guard could annotate it and the caller would read the annotation back. request.context is frozen by :class:~symfonic.capabilities.extensions.values.PolicyRequest (AS-INT-3 -- a policy answers a question, it does not edit the question) and dict() of it is a fresh mutable mapping the legacy signature can accept, so that side-channel is closed on both routes. Named in validate_action's docstring as the one observable that did change.

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

def legacy_guard_policies(
    plugin: Any, name: str
) -> tuple[PolicyContribution, ...]:
    """One legacy plugin's ``validate_state_transition``, as a policy — or none.

    Module-level rather than a :class:`LegacyPluginBridge` method because the
    guard outlives the bridge: a plugin the bridge *refuses* still has a veto,
    and dropping it would fail open in a way the pre-refactor inline loop over
    ``self._plugins`` never did.
    :func:`~symfonic.agent.cutover.guardrails.enforce_guardrails` is that
    caller -- it asks the engine's quarantine of refused plugins, and the
    plugin list itself when no capability folded -- and sharing this function
    is what keeps those from being separate implementations of deny-wins that
    agree by inspection.

    A plugin without the hook contributes *no* policy rather than an
    always-allow one: an always-allow stage in a trace is indistinguishable from
    a stage that examined the action and approved it.

    **The hook is handed a copy, and that is a narrowing.** The inline loop this
    replaced passed ``SymfonicAgent.validate_action``'s own ``context`` object
    straight through, so a guard could annotate it and the caller would read the
    annotation back. ``request.context`` is frozen by
    :class:`~symfonic.capabilities.extensions.values.PolicyRequest` (AS-INT-3 --
    a policy answers a question, it does not edit the question) and ``dict()``
    of it is a fresh mutable mapping the legacy signature can accept, so that
    side-channel is closed on both routes. Named in ``validate_action``'s
    docstring as the one observable that did change.
    """
    if not callable(getattr(plugin, "validate_state_transition", None)):
        return ()
    policy = f"{name}.validate_state_transition"

    async def decide(request: PolicyRequest) -> PolicyVerdict:
        validator = getattr(plugin, "validate_state_transition", None)
        if not callable(validator):  # pragma: no cover - gated above
            return PolicyVerdict.abstain(policy=policy, reason="hook is absent")
        try:
            allowed = await validator(request.action, dict(request.context))
        except Exception as exc:  # noqa: BLE001 - abstain, and say why
            logger.warning(
                "Plugin %r validate_state_transition raised; the guarded "
                "action is denied",
                name,
                exc_info=True,
            )
            return PolicyVerdict.abstain(
                policy=policy, reason=f"guard raised {type(exc).__name__}: {exc}"
            )
        if allowed:
            return PolicyVerdict.allow(policy=policy)
        return PolicyVerdict.deny(
            reason=f"plugin {name} blocked action {request.action!r}",
            policy=policy,
        )

    return (
        PolicyContribution(policy_id=policy, extension=name, decide=decide),
    )