symfonic.agent.engine

engine

SymfonicAgent -- the compatibility facade. Closed to new work.

Do not build here. The kernel is the front door: new capability work goes through public composition and the generated scaffold, not through this module. A change that adds behaviour to this file is almost certainly in the wrong place, and this file is 14k lines because that was not always said out loud.

What this module still is, stated precisely, because the difference matters and the repository's own invariants forbid collapsing it: this is not dormant and not retired. It still serves turns -- a turn that leaves the migrated envelope falls back to a body here, counted and attributed -- and the legacy implementations remain in the source on purpose. "Closed to new work" is a routing rule for contributors. It is not a claim that the code is unreachable, and it does not authorize deleting anything: physical removal is a separate, explicitly authorized operation behind the fourteen-condition retirement gate.

Fixes that keep a live route correct are still fixes. What does not belong here is new surface.

Provides run() and stream() entry points that transparently hydrate memory context before LLM execution and consolidate new memories afterward.

SymfonicAgent

SymfonicAgent(model_provider: ModelProvider, *, graph_preset: str = 'react_loop', config: FrameworkConfig | None = None, graph_backend: GraphBackend | None = None, vector_backend: VectorBackend | None = None, embedding_provider: EmbeddingProvider | None = None, tools: list[Any] | None = None, sub_agents: list[Any] | None = None, conversation_manager: Any | None = None, orchestrator: MemoryOrchestrator | None = None, sleep_consolidator: Any | None = None, metrics_collector: Any | None = None, legacy_pin: Sequence[str] | None = None)

High-level agent combining AgentRuntime + MemoryOrchestrator.

Minimal setup::

agent = SymfonicAgent(model_provider=my_provider)
response = await agent.run("Hello", scope=scope)

Construct the unified agent.

.. versionchanged:: 7.0 The deprecated enable_hms_prompt keyword argument has been removed. Use FrameworkConfig(enable_hms_prompt=True) instead — available since v6.0. Passing the removed kwarg now raises :class:TypeError at construction time.

.. versionadded:: TA8.54 legacy_pin — the documented legacy pin, as capability names (["invocation.stream"]). It holds exactly the capabilities it names on their legacy bodies for the life of this agent and moves no other; an unrecognised name -- or a bare string where a sequence of names belongs -- is refused by :func:~symfonic.agent.cutover.legacy_pin.build_legacy_pin rather than dropped, and a pin on a release at or past 11.0 is refused by LegacyPinRetiredError carrying the migration procedure. None means "this build sets no pin", which is not the same value as [] — an empty pin is refused, because a pin that moves nothing is the silence the mechanism exists to remove. See evidence/RET-PREP/legacy-pin-wiring.md.

Source code in src/symfonic/agent/engine.py

def __init__(
    self,
    model_provider: ModelProvider,
    *,
    graph_preset: str = "react_loop",
    config: FrameworkConfig | None = None,
    graph_backend: GraphBackend | None = None,
    vector_backend: VectorBackend | None = None,
    embedding_provider: EmbeddingProvider | None = None,
    tools: list[Any] | None = None,
    sub_agents: list[Any] | None = None,
    conversation_manager: Any | None = None,
    orchestrator: MemoryOrchestrator | None = None,
    sleep_consolidator: Any | None = None,
    metrics_collector: Any | None = None,
    legacy_pin: Sequence[str] | None = None,
) -> None:
    """Construct the unified agent.

    .. versionchanged:: 7.0
        The deprecated ``enable_hms_prompt`` keyword argument has
        been removed. Use ``FrameworkConfig(enable_hms_prompt=True)``
        instead — available since v6.0.  Passing the removed kwarg
        now raises :class:`TypeError` at construction time.

    .. versionadded:: TA8.54
        ``legacy_pin`` — the documented legacy pin, as capability names
        (``["invocation.stream"]``). It holds exactly the capabilities it
        names on their legacy bodies for the life of this agent and moves
        no other; an unrecognised name -- or a bare string where a sequence
        of names belongs -- is refused by
        :func:`~symfonic.agent.cutover.legacy_pin.build_legacy_pin` rather
        than dropped, and a pin on a release at or past
        ``11.0`` is refused by ``LegacyPinRetiredError`` carrying the
        migration procedure. ``None`` means "this build sets no pin",
        which is not the same value as ``[]`` — an empty pin is refused,
        because a pin that moves nothing is the silence the mechanism
        exists to remove. See ``evidence/RET-PREP/legacy-pin-wiring.md``.
    """
    self._config = config or FrameworkConfig.with_defaults()
    # TA8.54 (review): kept exactly as passed, bare string included.
    # ``tuple("invocation.run")`` would explode a plausible typo for the
    # documented list form into single characters, and the refusal would
    # then name ``'i'`` -- a character the operator never typed. Both the
    # shape and the names are refused by ``build_legacy_pin`` below, which
    # is the one place that knows what a pin may be.
    self._legacy_pin_request = legacy_pin

    # v9.2.1 (review P1a): per-agent background-task set. Consolidation is
    # scheduled fire-and-forget; ``flush_background_tasks`` (and the 9.1
    # ``async with`` auto-flush) await THIS set only, so exiting agent A
    # never blocks on agent B's pending work. The process-global
    # ``_background_tasks`` is retained solely for the atexit shutdown
    # warning (a genuinely process-wide concern).
    self._background_tasks: set[asyncio.Task[Any]] = set()

    # v8.12.0: conversation-manager strategy objects. A named strategy
    # (sliding-window / summarizing / null) resolves into an AgentConfig
    # override (compaction + max_conversation_messages) and takes precedence
    # over whatever the passed config set. See symfonic.agent.conversation.
    self._conversation_manager = conversation_manager
    if conversation_manager is not None:
        _new_agent_cfg = conversation_manager.apply(self._config.agent)
        self._config = self._config.model_copy(
            update={"agent": _new_agent_cfg}
        )

    self._model_provider = model_provider

    # v6.2 T01: compile the tunable credential-pattern list ONCE at
    # construction time so the hot hydrate / consolidate paths never
    # re-compile. Invalid regex fragments surface as re.error HERE,
    # not later at scrub time, which matches the fail-fast contract
    # documented on ``hygiene.compile_credential_pattern``.
    from symfonic.agent.hygiene import compile_credential_pattern

    self._credential_pattern = compile_credential_pattern(
        self._config.credential_patterns,
    )

    # v7.26.1 Item B telemetry: per-agent counter of
    # ``scorer_hotpath_fallback`` events.  Increments whenever the
    # ``FrameworkConfig.scorer_on_hot_path=True`` engine path takes
    # the per-turn fallback to per-layer dispatch -- because
    # ``RetrievalEngine.retrieve`` raised (transient vector outage,
    # bad query embedding, etc.) OR because the orchestrator was
    # constructed without a wired engine (direct dataclass test
    # path).  Adopters reading the counter from an
    # observability hook can detect silent regression patterns
    # without grepping logs.  Reset on construction; not persisted.
    self._scorer_hotpath_fallback_count: int = 0

    # v7.8.0 (adopter billing-pipeline ask): install adopter-supplied
    # pricing overrides BEFORE any cost-computation site reads.  The
    # registry is module-level so this is a process-wide install --
    # documented as a one-agent-per-process contract.
    #
    # v8.7.1 (M7): install via ``install_pricing_overrides`` which SKIPS
    # an empty dict instead of clearing. Previously a default-config
    # agent B (empty overrides) wiped a prior agent A's process-wide
    # overrides; the empty-clear semantic still lives on
    # ``register_pricing_overrides({})`` for explicit callers. The
    # helper also WARNS once on a divergent non-empty replace and emits
    # the v7.20.0 audit INFO line (no rate values).
    from symfonic.core.observability.pricing import (
        install_pricing_overrides,
    )

    _resolved_overrides = dict(self._config.model_pricing_overrides or {})
    install_pricing_overrides(_resolved_overrides)

    # Sync FrameworkConfig.enabled_layers -> OrchestratorConfig.enabled_layers
    # so there is a single source of truth (FrameworkConfig).
    _synced_orchestrator_cfg = self._sync_enabled_layers(self._config)

    # -- Memory orchestrator -----------------------------------------------
    if orchestrator is not None:
        self._orchestrator = orchestrator
    elif embedding_provider is not None:
        if graph_backend is not None and vector_backend is not None:
            self._orchestrator = HMSFactory.build(
                graph_backend=graph_backend,
                vector_backend=vector_backend,
                embedding_provider=embedding_provider,
                config=_synced_orchestrator_cfg,
                stm_summary_mode=self._config.stm_summary_mode,
            )
        else:
            self._orchestrator = HMSFactory.build_in_memory(
                embedding_provider=embedding_provider,
                config=_synced_orchestrator_cfg,
                stm_summary_mode=self._config.stm_summary_mode,
            )
    else:
        # No memory -- create an empty orchestrator
        self._orchestrator = MemoryOrchestrator(
            config=_synced_orchestrator_cfg,
        )

    # v8.9.0 sub-agent delegation: declared children (see below for wiring).
    # v9.1.0 (issue #27): a declared child may be a pre-built ``SubAgent``
    # OR a declarative ``SubAgentSpec`` -- resolve specs into concrete
    # ``SubAgent`` instances now (needs the provider + config set above).
    #
    # v8.20 review fix (B3): this block moved BELOW ``self._orchestrator``
    # construction (was directly after the provider/config setup, well
    # before the orchestrator existed). ``_build_sub_agent_from_spec``
    # passes ``self._orchestrator`` to the child so it resolves the same
    # graph store as the parent; reading that attribute before this line
    # used to raise ``AttributeError`` the moment a spec child needed it,
    # which is every spec child now that the seam exists.
    from symfonic.agent.subagents.types import SubAgentSpec as _SubAgentSpec

    # v9.2.x (issue #27 leak fix): children WE build from a spec are
    # OWNED by this parent -- their checkpointer pools must be released
    # on parent shutdown. Pre-built ``SubAgent`` instances passed in by
    # the caller are NOT owned (the caller controls their lifecycle), so
    # we track only the ones we construct.
    self._sub_agents = []
    self._owned_child_agents: list[SymfonicAgent] = []
    for sa in (sub_agents or []):
        if isinstance(sa, _SubAgentSpec):
            built = self._build_sub_agent_from_spec(sa)
            self._sub_agents.append(built)
            self._owned_child_agents.append(built.agent)
        else:
            self._sub_agents.append(sa)
    self._sub_agent_registry: Any | None = None
    # TA8.12: the folded delegation capability, built lazily on the first
    # turn that needs it and kept. ``_UNBUILT`` rather than ``None``
    # because ``None`` is a real answer here -- "this agent folds no
    # delegation" -- and a sentinel is what keeps that answer from being
    # recomputed on every dispatch.
    self._delegation_capability_cache: Any = _UNBUILT
    # TA8.21: the folded extensions capability, on the same sentinel and
    # for the same reason. It closes over ``self._plugins`` through a
    # callable rather than over a snapshot of it, so a plugin loaded after
    # the first turn still contributes prompts and guardrails -- which is
    # what legacy did -- while the *tool* set it sealed at construction
    # stays sealed, which is what the compile seam requires.
    self._extensions_capability_cache: Any = _UNBUILT
    # The plugin count at which a failed extensions build was last reported.
    # Only a *successful* capability is cached (see
    # ``_extensions_capability``), so a deterministic failure is retried on
    # every guarded action once ``validate_action`` began asking -- and
    # without this latch each retry re-emitted a full traceback at WARNING
    # for the life of the agent. Keyed on the roster size rather than a bare
    # bool because ``load_plugin`` can change the answer, and a build that
    # starts failing for a *newly* loaded plugin is a new fact. The latch
    # pattern is the one ``core/plugins/section.py`` already uses for
    # ``_kernel_warned``.
    self._extensions_capability_warned: int | None = None
    # The same latch, for the report ``enforce_guardrails`` emits when it
    # falls back to the plugin list. A dict rather than an ``int`` because
    # the writer is a module function this object hands it to: an ``int``
    # would be copied and the latch would never close. See
    # ``cutover/guardrails.py``'s ``UNFOLDED_LATCH_KEY``.
    self._unfolded_guard_warned: dict[str, int] = {}
    # TA8.21. ``True`` once the kernel delegate has been built or the
    # extensions surface has sealed -- the compile seam. A plugin offered
    # after it cannot be honoured: the bundle the delegate compiled from is
    # fixed, so the plugin would be registered and never harvested. It is
    # refused at ``load_plugin`` instead, which is where an adopter can act
    # on it.
    self._plugin_admission_frozen: bool = False
    # The plugins whose admission was *refused*. Not a registry: nothing
    # here contributes a prompt fragment, a tool, a lifecycle hook or a
    # name to the composition, and ``_plugins`` never saw them. What
    # ``validate_action`` still does is ask their guardrails, deny-only,
    # because refusing a plugin may not delete a veto the adopter
    # installed. See ``cutover/guardrails.py``'s ``_quarantine_outcome``.
    self._refused_plugins: list[Any] = []

    # -- Episodic telemetry sink (v6.1 T01) --------------------------------
    # Default ``None`` keeps the hot write path cost-free.  When a sink
    # string is configured we resolve it once and attach to the live
    # EpisodicLayer -- the sink is fire-and-forget from EpisodicLayer's
    # perspective and will not block the write.
    self._attach_episodic_telemetry_sink()

    # -- Core graph + runtime ----------------------------------------------
    # v7.9.4 (adopter state-injection unblock): forward the
    # adopter-supplied ``FrameworkConfig.state_class`` to
    # ``AgentGraph`` so its ``StateGraph`` carries the adopter's
    # extended TypedDict schema.  Default ``None`` -> framework
    # ``BaseAgentState``.  The hard plumbing already exists at
    # ``core/graph.py:48-59`` (state_class param) and ``:169``
    # (``StateGraph(self._state_class)`` consumer).
    _state_class = getattr(self._config, "state_class", None)
    if _state_class is not None:
        from symfonic.core.state import BaseAgentState as _BAS

        # TypedDict cannot be validated via ``issubclass`` --
        # Python raises ``TypeError: TypedDict does not support
        # instance and class checks``.  We use structural
        # subtyping instead: the adopter class MUST carry every
        # annotation ``BaseAgentState`` declares.  This is the
        # protocol-style check that actually works for TypedDict.
        adopter_annos = getattr(_state_class, "__annotations__", None)
        if not isinstance(adopter_annos, dict):
            raise TypeError(
                f"FrameworkConfig.state_class must be a "
                f"TypedDict subclass of BaseAgentState; got "
                f"{getattr(_state_class, '__name__', _state_class)!r} "
                f"(no ``__annotations__`` attribute).",
            )
        missing = set(_BAS.__annotations__) - set(adopter_annos)
        if missing:
            raise TypeError(
                f"FrameworkConfig.state_class must subclass "
                f"symfonic.core.state.BaseAgentState; got "
                f"{_state_class.__name__!r} which is missing the "
                f"framework fields {sorted(missing)}.  v7.9.4 "
                f"contract: adopter TypedDicts inherit the "
                f"framework reserved-fields surface.",
            )
        # v7.11.0 field-collision check.  An adopter that
        # silently redeclares a framework field (e.g. annotating
        # ``messages: list[str]`` instead of inheriting the
        # ``Annotated[Sequence[BaseMessage], add_messages]``
        # reducer-bearing form) breaks LangGraph's state merge
        # without any error at construction time -- exactly the
        # "introspect the wrong source of truth" bug class
        # (v7.7.6 / v7.7.7 / v7.9.3 / v7.9.5) escalated to the
        # state schema layer.
        #
        # Policy: raise on type divergence; escape hatch via
        # ``__symfonic_shadow_reserved__`` class attribute on the
        # adopter state class, opting in field-by-field.  An
        # adopter who *knows* they want to tighten a framework
        # field's type (e.g. ``dict[str, Any]`` -> ``dict[str,
        # MyShape]``) must list that field by name; bare
        # redeclaration raises.
        #
        # Comparison shape: ``repr()`` over the type annotation.
        # Not perfect (``typing.get_type_hints`` would resolve
        # ForwardRefs and Annotated metadata) but it catches the
        # actual bugs we've seen (the adopter's hypothetical
        # ``messages: list[str]`` shadowing add_messages) without
        # a typing.get_type_hints round-trip that would surface
        # ForwardRef resolution errors from adopters with
        # circular imports.  v7.12+ can tighten if needed.
        shadow_optin = frozenset(
            getattr(
                _state_class,
                "__symfonic_shadow_reserved__",
                (),
            ) or (),
        )
        collisions: list[str] = []
        for field, framework_type in _BAS.__annotations__.items():
            if field not in adopter_annos:
                continue  # handled by ``missing`` check above
            if field in shadow_optin:
                continue  # adopter knowingly shadows; allow
            adopter_type = adopter_annos[field]
            # Identical re-declaration is fine (no-op subclass).
            if repr(adopter_type) != repr(framework_type):
                collisions.append(field)
        if collisions:
            raise TypeError(
                f"FrameworkConfig.state_class "
                f"{_state_class.__name__!r} silently redeclares "
                f"framework field(s) {sorted(collisions)} with a "
                f"different type than BaseAgentState.  This "
                f"breaks LangGraph's state merge (the framework's "
                f"reducer-bearing annotations are dropped).  To "
                f"opt in to redeclaration, list the field(s) on "
                f"the class attribute "
                f"``__symfonic_shadow_reserved__`` -- e.g. "
                f"``__symfonic_shadow_reserved__ = "
                f"frozenset({{{collisions[0]!r}}})``.  v7.11.0 "
                f"field-collision check; see "
                f"docs/concepts/state-scope-mismatch.md.",
            )
        self._graph = AgentGraph(
            topology=graph_preset, state_class=_state_class,
        )
    else:
        self._graph = AgentGraph(topology=graph_preset)
    # Roadmap Item 6: propagate the experimental ``@symfonic_tool``
    # gate into the registry BEFORE any tool is registered.  When
    # the flag is off, decorator-produced tools fail at register()
    # time so the experimental surface cannot ship by accident.
    self._graph.registry.set_experimental_tool_decorator(
        self._config.experimental_tool_decorator,
    )
    local_tools = list(tools) if tools is not None else []
    if local_tools:
        self._graph.add_tools(local_tools)

    # v7.1.0: register built-in ask_user tool if enabled.
    if self._config.ask_user_enabled:
        from langchain_core.tools import StructuredTool

        from symfonic.core.tools.ask_user import ASK_USER_DESCRIPTION_EN, ask_user

        # Use locale-specific description if configured.
        locale = self._config.agent.model.extra_headers.get("X-Language", "en")
        description = self._config.ask_user_tool_descriptions.get(
            locale, ASK_USER_DESCRIPTION_EN
        )

        # Wrap the function in a StructuredTool to apply the description.
        bound_ask_user = StructuredTool.from_function(
            func=None, # Not needed as it's intercepted
            coroutine=ask_user,
            name="ask_user",
            description=description,
        )
        self._graph.add_tool(bound_ask_user)

    # Auto-register hydrate_context in JIT mode (F1 [CRITICAL])
    from symfonic.agent.context import resolve_strategy_name
    self._strategy_name = resolve_strategy_name(self._config)
    if self._strategy_name == "jit":
        from symfonic.tools.hydrate_context import create_hydrate_context_tool

        def _resolve_scope():
            val = _active_scope.get()
            if val is None:
                raise RuntimeError(
                    "No active tenant scope is configured for the current run context.",
                )
            return val

        bound_hydrate_context = create_hydrate_context_tool(
            self._orchestrator,
            _resolve_scope,
        )
        self._graph.add_tool(bound_hydrate_context)

    # -- Auto-populate DomainTemplate.tool_manifest (Change 3) ------------
    # If tools were registered but the domain template ships no manifest,
    # derive one from the tool names so the HMS system prompt can list
    # them instead of rendering "(no tools registered)".  Respect an
    # explicit manifest if provided.  Both frozen models are rebuilt via
    # model_copy because they are immutable.
    #
    # Derive from the REGISTRY, not from ``local_tools``.  The caller's
    # list omits every framework-injected tool -- ``ask_user`` (added
    # just above) and the JIT ``hydrate_context`` are both registered on
    # the graph without ever entering ``local_tools``.  Deriving from the
    # caller's list therefore left the manifest empty for an agent whose
    # only tool was ``ask_user``, and the L1 prompt renders the manifest
    # under "Use ONLY these exact names when making tool calls:" -- so
    # the model was bound a tool the prompt simultaneously declined to
    # authorise, and no model called it.
    self._auto_populate_tool_manifest(self._graph.registry)

    # Propagate ask_user_enabled to core AgentConfig
    agent_config = self._config.agent
    if self._config.ask_user_enabled:
        import dataclasses
        agent_config = dataclasses.replace(agent_config, ask_user_enabled=True)
    # Roadmap Item 9: propagate the experimental_interrupt flag so
    # the preset can wire the generic ``InterruptNode`` into the
    # topology. Independent of ask_user_enabled so a graph can
    # ship with only the generic primitive enabled.
    if self._config.experimental_interrupt:
        import dataclasses
        agent_config = dataclasses.replace(
            agent_config, experimental_interrupt=True,
        )

    # PR #79 pre-merge review MEDIUM 2: propagate the ask_user pause
    # TTL so the preset wiring at ``core/presets.py`` can bind it
    # into the elicitation/interrupt/precondition_gate routers.
    # Without this, ``core.config.AgentConfig`` (a deliberately
    # narrow dataclass -- see its docstring) never carries the
    # field at all, so ``getattr(config, "ask_user_pause_ttl_seconds",
    # None)`` at wire time silently resolves to ``None`` and the
    # TTL guard added for MEDIUM 2 would be dead code in production.
    if self._config.ask_user_enabled or self._config.experimental_interrupt:
        import dataclasses
        agent_config = dataclasses.replace(
            agent_config,
            ask_user_pause_ttl_seconds=self._config.ask_user_pause_ttl_seconds,
        )

    # v7.9.0 (adopter model-tier compliance ask): per-role model
    # override.  When ``role_models["action"]`` is set, route the
    # react node's brain LLM through that ``ModelConfig`` instead
    # of the default ``config.agent.model``.  Closes the
    # claude-opus-4-6 vs gpt-4.1-nano compliance gap adopter
    # documented (legacy 6/6 vs symfonic 1/6 on the WLM probe).
    # Architectural principle (v7.8.3 preserved): the framework
    # orchestrates; the adopter assigns models per role.  Unset
    # roles fall back transparently.
    action_model_override = (self._config.role_models or {}).get(
        "action",
    )
    if action_model_override is not None:
        import dataclasses
        agent_config = dataclasses.replace(
            agent_config, model=action_model_override,
        )

    # v7.24.2: also propagate the WHOLE ``role_models`` map onto
    # ``AgentConfig.role_models`` so the ``ReactLoopPreset`` wiring
    # at ``core/presets.py`` can thread the ``"summary"`` override
    # into ``create_context_window_node(...)``.  The action role
    # was already consumed above via the ``model=`` swap; this
    # adds the rest of the taxonomy to the preset's input
    # surface.  Critic / router / consolidation continue to read
    # ``FrameworkConfig.role_models`` directly because those call
    # sites already have a FrameworkConfig reference.
    _role_models_snapshot = dict(self._config.role_models or {})
    if _role_models_snapshot:
        import dataclasses
        agent_config = dataclasses.replace(
            agent_config, role_models=_role_models_snapshot,
        )

    # v7.7.4 (adopter PRE-FLIGHT enforcement ask, authored tier):
    # thread the ``procedural_enforce_preconditions`` flag from
    # FrameworkConfig into AgentConfig so the preset wiring at
    # ``core/presets.py`` knows to add the precondition_gate node
    # as the third sibling of elicitation / interrupt.  Also
    # register the active-skills getter on the graph so the gate
    # can query the live ProceduralLayer at runtime without
    # holding a closure over the agent instance.
    #
    # v7.7.5 extends the getter registration: the same closure
    # also feeds the L1 PRE-FLIGHT contribution synthesiser when
    # ``procedural_render_preflight_in_l1=True``, so either knob
    # is enough to wire the getter (and both can be on together).
    _needs_skills_getter = (
        getattr(self._config, "procedural_enforce_preconditions", False)
        or getattr(
            self._config, "procedural_render_preflight_in_l1", False,
        )
    )
    if getattr(self._config, "procedural_enforce_preconditions", False):
        import dataclasses
        agent_config = dataclasses.replace(
            agent_config,
            procedural_enforce_preconditions=True,
        )

    # v7.9.3 (adopter engine.py:1385 dual-construction bug): cache
    # the resolved ``AgentConfig`` as instance state so BOTH
    # ``AgentRuntime.__init__`` call sites read from the same
    # source of truth.  Pre-v7.9.3 the first construction at
    # line 1235 read from the overridden ``agent_config`` local;
    # the SECOND construction at line 1382 silently re-read
    # ``self._config.agent`` (the raw FrameworkConfig field) and
    # overwrote ``self._runtime`` with a config that had
    # discarded the v7.9.0 role_models override entirely.  Net
    # effect: ``role_models["action"]`` was a no-op for every
    # adopter who did not hand-patch ``self._runtime`` after
    # ``__init__``.  Caching here AND consuming
    # ``self._agent_config_resolved`` at every construction site
    # closes the dual-construction bug class for future
    # role_models wirings (summary/critic/router) as well.
    #
    # PR #79 review fix (item 9): this assignment used to sit
    # ABOVE the ``procedural_enforce_preconditions`` replace that
    # now immediately precedes it, so it cached the AgentConfig
    # from BEFORE that flag was folded in -- the replace below
    # rebound the local ``agent_config`` to a config nobody read.
    # ``core/presets.py`` gates the ``precondition_gate`` node on
    # ``config.procedural_enforce_preconditions`` where ``config``
    # is exactly ``self._agent_config_resolved`` (threaded through
    # ``AgentRuntime`` -> ``AgentGraph.compile``), so the node was
    # never wired into the compiled graph for ANY agent, however
    # it was constructed: ``FrameworkConfig(procedural_enforce_
    # preconditions=True)`` produced a real, running agent whose
    # graph nodes were ``{"compaction", "react", ...}`` with no
    # ``"precondition_gate"`` -- the v7.7.4 feature has been dead
    # since it shipped. Moving this assignment below the replace
    # is the one-line fix; see the mutation-checked regression
    # test at ``tests/agent/test_procedural_enforce_preconditions_
    # graph_wiring.py`` for the reproduction and the safety
    # exercise this fix required before being declared complete.
    self._agent_config_resolved = agent_config

    if _needs_skills_getter:
        # T-7.21.5 (Slice B site #1 -- LOAD-BEARING).  The getter
        # closure is now built by the module-level factory so that
        # BOTH consumers (runtime precondition_gate AND L1
        # PRE-FLIGHT synthesiser) route through
        # ``TenantScope.from_state_dict``.  Closes the split-brain
        # class (Risk R-V7.21-A): pre-7.21 the bare construction at
        # this site dropped ``namespace`` and any new TenantScope
        # field, causing the cache-prefix (L1 PRE-FLIGHT) and the
        # runtime gate to potentially see different scopes.  The
        # closure-over-self pattern survives because the factory
        # captures the live ProceduralLayer reference; binding the
        # layer at agent-build-time means a layer swap (rare) would
        # also need a getter re-registration -- pre-existing
        # contract.
        proc_layer = self._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
        self._graph.set_active_skills_getter(
            _active_skills_getter_factory(proc_layer),
        )

    # v7.23 T-7.23.5: per-engine in-memory cache_state tracking.
    # ``_cache_state`` maps ``model_name -> last cache_creation
    # event timestamp (UTC)``.  Updated by ``_on_llm_response``
    # whenever a response carries ``cache_creation_tokens > 0``.
    # Exposed to the adopter resolver via
    # ``_AgentModelResolver.cache_state_view()`` and threaded into
    # ``DispatchContext.cache_state`` at the react dispatch site.
    # In-memory per-engine-instance -- does NOT cross worker
    # boundaries.  Cross-worker coordination is out-of-scope-for-
    # framework (v8.0 backlog §7); the concept doc spells out two
    # adopter patterns (session affinity, external Redis store).
    from datetime import datetime as _datetime
    self._cache_state: dict[str, _datetime] = {}

    self._deps = BaseAgentDeps(ModelProvider=model_provider)
    # v7.10: register the live force-resolver so the react node
    # can call it on every iteration instead of reading a stamp
    # that ran once at assemble-time.  Fixes the "force never
    # releases" bug (adopter post-v7.9.5) AND closes the latent
    # streaming-stamp gap (stream/stream_typed never invoked the
    # assemble-time resolver) in a single change -- because the
    # react node is shared across all three entry points, force
    # activation and release are now symmetric by construction.
    from ..core.force_resolver import ForcedToolChoiceResolver
    self._deps.register(
        ForcedToolChoiceResolver,
        _AgentForcedToolChoiceResolver(self),
    )
    # v7.23: register the per-iteration model resolver adapter.
    # Bridges adopter-supplied ``FrameworkConfig.role_model_resolver``
    # to a uniform handle the react node looks up via
    # ``deps.get(_AgentModelResolver)``.  ``resolver=None`` (the
    # default) means the adapter falls through to the snapshot
    # ``AgentConfig.model`` on every iteration -- pre-v7.23 wire
    # shape preserved.  See ``docs/concepts/per-dispatch-model-
    # resolver.md`` for the closure-factory adopter pattern.
    self._deps.register(
        _AgentModelResolver,
        _AgentModelResolver(
            resolver=getattr(
                self._config, "role_model_resolver", None,
            ),
            snapshot_default=self._agent_config_resolved.model,
            cache_state_provider=lambda: self._cache_state,
            # v8.1.0 C3: thread declarative policies to the PRE-model
            # seam.  Empty default -> byte-identical to v8.0.1.
            tool_call_policies=getattr(
                self._config, "tool_call_policies", (),
            ),
        ),
    )
    # v7.19.2 Item 4: startup-time WARN when the force lever is
    # silently advisory under the current configuration.  Fires
    # ONCE at construction (NOT per-turn) when ALL THREE hold:
    # (a) procedural_force_first_action_tool=True,
    # (b) the action-role ModelConfig has ``thinking`` set,
    # (c) the provider refuses forced tool_choice for that config.
    # Catches the misconfiguration in the boot log BEFORE any
    # production turn lands.  Complements (does not replace) the
    # per-turn refusal WARN at _maybe_resolve_forced_tool_choice
    # and the post-call non-compliance WARN at run()'s post-LLM
    # block.  Errors here MUST NOT stall agent construction --
    # this is a diagnostic, not a gate.
    self._warn_silent_advisory_force_lever_at_startup()
    # v7.11.0 (adopter Finding 5 nano-recursion fix): register the
    # role-aware tool-palette resolver next to the force resolver.
    # Default adapter reads ``FrameworkConfig.role_tools``;
    # adopters override by registering their own implementation.
    from ..core.tool_palette_resolver import ToolPaletteResolver
    self._deps.register(
        ToolPaletteResolver,
        _AgentToolPaletteResolver(self),
    )
    # v7.12.0 (adopter message-history cost-killer fix): register
    # the tool-result ledger.  The default in-memory adapter
    # serves intra-conversation recall; adopters who need cross-
    # process recall override with a durable-store implementation.
    # The react node calls ``ledger.record(...)`` on every
    # iteration that sees a verbose ToolMessage (idempotent on
    # tool_call_id) and the framework-provided
    # ``recall_tool_result`` tool calls ``ledger.fetch(...)``.
    from ..core.tool_result_ledger import (
        InMemoryToolResultLedger,
        ToolResultLedger,
    )
    self._deps.register(
        ToolResultLedger,
        InMemoryToolResultLedger(),
    )
    # v7.13.0 Path E: register the messages-region cache strategy
    # so the react node can place a ``cache_control`` breakpoint on
    # the most recent CLOSED AIMessage / HumanMessage when the
    # prefix exceeds the model-family threshold.  Default impl
    # (LastStableTurnBoundaryStrategy) auto-disables for non-
    # Anthropic providers and below the cacheable prefix threshold.
    # The ``"off"`` policy skips registration entirely so the
    # react node's lookup degrades cleanly to "no breakpoint."
    if self._config.messages_cache_policy != "off":
        from ..core.prompt.messages_cache import (
            LastStableTurnBoundaryStrategy,
            MessagesRegionCacheStrategy,
        )
        _cache_strategy: MessagesRegionCacheStrategy
        if self._config.messages_cache_policy == "rolling":
            # v8.7.0 opt-in: rolling ladder of messages-region
            # breakpoints (holds prior frontier as cache-reads, adds
            # one new marker per iteration) within Anthropic's
            # 4-marker budget.  Degrades to the single sticky marker
            # when fewer than 2 messages slots remain.
            from ..core.prompt.messages_cache_rolling import (
                RollingLadderStrategy,
            )
            _cache_strategy = RollingLadderStrategy()
        else:
            if self._config.messages_cache_policy == "window_aware":
                # v7.13.1 not yet shipped; one-time INFO log + fall
                # back to the default strategy.
                logger.info(
                    "messages_cache_policy='window_aware' is queued "
                    "for v7.13.1; falling back to v7.13.0 default "
                    "stateless placement for now",
                )
            _cache_strategy = LastStableTurnBoundaryStrategy()
        self._deps.register(
            MessagesRegionCacheStrategy,
            _cache_strategy,
        )
    # v7.12.0 Auto-register recall_tool_result whenever tool-result
    # compaction is enabled.  The compaction rewriter at
    # ``react.py`` swaps verbose ToolMessage.content for a compact
    # stub that names a ``mem_id``; the model recalls the full
    # payload through this tool when needed.  Without registering
    # the tool, the stub names a tool the model has no way to
    # invoke -- effectively a documentation-only stub.  MUST run
    # after the ledger is registered (above) and after the agent
    # registry exists (line ~1373 ``self._graph = AgentGraph(...)``).
    if self._config.tool_result_compaction_enabled:
        from symfonic.tools.recall_tool_result import (
            create_recall_tool_result_tool,
        )
        _ledger = self._deps.require(ToolResultLedger)
        bound_recall = create_recall_tool_result_tool(_ledger)
        self._graph.add_tool(bound_recall)
    # v8.9.0: sub-agent delegation. When the adopter declares ``sub_agents``,
    # build a concrete ``AgentStore`` backed by the child agents, register
    # the ``run_agent`` / ``list_agents`` tools BEFORE compile (the tool
    # registry freezes with the graph), and expose the store as a deps
    # capability. Delegation depth is read from the ``_active_agent_depth``
    # contextvar (set per-run by ``_with_active_scope``) so nested
    # delegation honours ``max_agent_depth``. The run's correlation id is
    # read the same way, from ``_active_run_id`` (issue #64): the tools are
    # bound once, here, and run inside whatever turn calls them, so every
    # per-run value they need arrives as a callable rather than a capture.
    # MUST run before the runtime is constructed (compile freezes the
    # registry).
    if self._sub_agents:
        from symfonic.agent.subagents.registry import SubAgentRegistry
        from symfonic.agent.subagents.tool import create_delegation_tools
        from symfonic.core.protocols import AgentStore

        self._sub_agent_registry = SubAgentRegistry(self._sub_agents)
        self._deps.register(AgentStore, self._sub_agent_registry)
        # Fail fast rather than silently clobber: the tool registry is
        # last-writer-wins, so an adopter tool literally named ``run_agent``
        # / ``list_agents`` would be overwritten (or overwrite ours)
        # invisibly. Surface the clash at construction time.
        _existing_names = {t.name for t in self._graph.registry.all_tools()}
        for _delegation_tool in create_delegation_tools(
            self._sub_agent_registry,
            resolve_scope=lambda: _active_scope.get(),
            resolve_depth=lambda: _active_agent_depth.get(),
            max_depth=self._agent_config_resolved.max_agent_depth,
            record_delegation=_record_delegation,
            resolve_run_id=lambda: _active_run_id.get(),
        ):
            if _delegation_tool.name in _existing_names:
                raise ConfigurationError(
                    f"Tool name {_delegation_tool.name!r} is reserved for "
                    "sub-agent delegation but is already registered by a "
                    "domain/user tool. Rename the conflicting tool or drop "
                    "sub_agents."
                )
            self._graph.add_tool(_delegation_tool)
    # v7.9.3: read from the cached resolved config so the
    # v7.9.0 role_models override survives.
    self._runtime = AgentRuntime(
        graph=self._graph,
        deps=self._deps,
        config=self._agent_config_resolved,
    )

    # -- Session manager ---------------------------------------------------
    self._session_manager = SessionManager()

    # -- HMS prompt sections (opt-in) -------------------------------------
    # Single source of truth: FrameworkConfig.enable_hms_prompt.  The
    # deprecated kwarg has already been folded into self._config above.
    self._enable_hms_prompt = self._config.enable_hms_prompt
    if self._enable_hms_prompt:
        self._hms_section = HMSSystemPromptSection()
        # v9.1.0 (issue #25.4b): honour a custom extraction directive
        # template when the adopter sets ``extraction_template_path`` --
        # the non-Claude escape hatch for HMS memory extraction. None
        # keeps the bundled template (byte-identical default path).
        #
        # v9.2.0 (issue #25.4c): when NO explicit override is set, pick a
        # provider-family-tuned bundled template. OpenAI-family / Google
        # models get the more explicit ``extraction_openai`` directive
        # (the default template is Anthropic-delimiter oriented and such
        # models silently drop memory when they ignore it). "unknown"
        # providers (incl. MockModelProvider and truly-custom dialects)
        # stay on the default template -- wire-neutral, byte-identical.
        _extract_tpl = self._config.extraction_template_path
        if _extract_tpl:
            self._extraction_section = MemoryExtractionSection(
                template_path=Path(_extract_tpl),
            )
        else:
            # v9.2.1 (review P1b): pass the resolved ModelConfig so a
            # MultiProviderRouter is classified from the LEAF that will
            # actually serve this agent (its configured model), not the
            # router's default provider -- otherwise an Anthropic-default
            # router serving gpt-* through OpenAI would wrongly keep the
            # Anthropic template and defeat the family-aware fix.
            _family = _detect_provider_family(
                self._model_provider, self._agent_config_resolved.model,
            )
            _tpl_name = (
                "extraction_openai"
                if _family in ("openai", "google")
                else "extraction"
            )
            self._extraction_section = MemoryExtractionSection(
                template_name=_tpl_name,
            )
    else:
        self._hms_section = None
        self._extraction_section = None

    # -- Dual-write mirror (set externally, e.g. by demo server) ----------
    self._mongo_mirror: Any = None

    # -- Quick-nap consolidation -------------------------------------------
    self._sleep_consolidator: Any = sleep_consolidator
    # v8.17 activation seam: when nap behaviour is requested via config
    # but no consolidator was injected, build one from the orchestrator.
    # Pre-8.17, ``quick_nap_interval`` (and ``enable_semantic_merge``)
    # were silently dead config on the AgentBuilder path: the nap guard
    # required an injected consolidator the builder had no seam for
    # (adopter-verified: naps never fired, Phase 13 was unreachable).
    if self._sleep_consolidator is None and (
        self._config.quick_nap_interval > 0
        or self._config.enable_semantic_merge
    ):
        self._sleep_consolidator = self._build_default_consolidator()
    self._turn_count: int = 0

    # -- Advisory budget warnings (emit-once flags) -----------------------
    # The extraction prompt template is static + large; once it exceeds
    # the advisory budget the warning would fire every turn.  Latch it
    # so the warning surfaces exactly once per agent lifetime.
    self._extraction_budget_warned: bool = False
    # v9.3.0 (issue #36): latch the lazy-tooling skill-resolution warning
    # so an unresolved procedural skill surfaces a Python warning exactly
    # once per agent instead of every turn.
    self._lazy_skill_resolution_warned: bool = False
    # v7.7: same once-per-instance latch for the HMS system-prompt
    # section budget warning.  Pre-7.7 the engine logged a fresh
    # warning every turn at the 3000-token cap (with a stale
    # ``"max 1500"`` literal in the message) -- adopter observed it
    # firing 50+ times per session.  v7.7 logs at most once.
    self._hms_budget_warned: bool = False
    # v7.0.2: latch ``domain.description`` truncation warning so it
    # fires at most once per agent lifetime (the raw description is
    # static across turns; re-warning every turn would drown logs).
    self._domain_description_truncate_warned: bool = False
    # v7.x stratigraphic prompting: per-call stash for the L0 kernel
    # version + L1 body hash produced by ``_build_hms_system_prompt`` /
    # ``_build_jit_system_prompt`` when ``prompt_layer_mode='stratigraphic'``.
    # The three run/stream call sites read this immediately after the
    # builder returns and pour the values into ``state_overrides`` via
    # ``_apply_strat_metadata_to_overrides``.  Reset to ``None`` in
    # legacy mode so stale values from a prior call cannot leak.
    self._strat_brain_version: str | None = None
    self._strat_system_hash: str | None = None

    # -- Domain plugins ---------------------------------------------------
    self._plugins: list[Any] = []
    self._plugin_section = PluginPromptSection()
    # v7.4 Item 14: agent-wide latches for plugin-contribution
    # observability logs.  ``_jit_volatile_warned`` fires once per
    # agent instance the first time a plugin emits
    # ``position="volatile"`` on the JIT path (which has no
    # separate L2 region, so volatile contributions provide no
    # cache benefit there).  Mirrors ``_extraction_budget_warned``.
    # The short-circuit and kernel sets live on the plugin section
    # itself (see ``PluginPromptSection._short_circuit_warned`` /
    # ``_kernel_warned``); the engine doesn't double-track them.
    self._jit_volatile_warned: bool = False
    # v7.4 Item 16: agent-wide latches for ``manifest_cache_position``
    # observability logs.  ``_manifest_jit_warned`` fires once per
    # agent instance the first time ``manifest_cache_position=
    # "volatile"`` is observed on the JIT path (which has no
    # separate L2 region, so the knob is a silent no-op there --
    # mirrors the Item 14 JIT decision).
    # ``_manifest_legacy_warned`` fires once per agent instance the
    # first time ``manifest_cache_position="volatile"`` is observed
    # on the stratified path while ``prompt_layer_mode='legacy'``
    # (no L1/L2 breakpoint, so the knob has nowhere to split to --
    # mirrors the Item 15 stratigraphic-only gating).
    self._manifest_jit_warned: bool = False
    self._manifest_legacy_warned: bool = False

    # -- v8.20 prompt blocks (stage 3) ------------------------------------
    # ``None`` whenever ``FrameworkConfig.prompt_blocks`` is empty --
    # the default -- so the block lane costs nothing to construct and
    # every prompt path takes the identical pre-8.20 branch.  Built
    # HERE rather than lazily on first prompt build so a declared
    # block whose source cannot be wired (a memory-lane block with no
    # graph-backed layer) fails at agent construction, where the
    # traceback names the misconfiguration, instead of on turn one
    # inside the prompt builder's broad ``except`` -- which would
    # degrade to "no system prompt at all".
    self._block_injector: PromptBlockInjector | None = (
        self._build_block_injector()
    )
    self._block_scope_warned: bool = False

    # -- Metrics collector (optional) -------------------------------------
    self._metrics_collector: Any = metrics_collector

    # -- Metacognitive middleware (opt-in, lazy import) -------------------
    self._metacog: Any = None
    if self._config.metacognition_enabled:
        from symfonic.agent.middleware.metacognitive import MetacognitiveMiddleware

        # v8.3.1 observability fix: pass the SAME always-on
        # ``ObservabilityHook`` react resolves (``react.py:837``:
        # ``deps.require(ObservabilityHook)``) so the critic LLM call is
        # visible on the OTel hook path.  Defaults to the
        # auto-registered ``NoOpObservabilityHook`` (deps.py:100-110), so
        # the unconfigured path stays byte-identical.
        from symfonic.core.observability.protocol import (
            ObservabilityHook as _ObservabilityHook,
        )

        self._metacog = MetacognitiveMiddleware(
            model_provider=self._model_provider,
            config=self._config,
            domain=self._config.domain,
            procedural_layer=self._orchestrator.get_layer(MemoryLayer.PROCEDURAL)
            if hasattr(self._orchestrator, "get_layer")
            else None,
            observability_hook=self._deps.require(_ObservabilityHook),
        )

    # -- Fabrication mode effective resolution (v7.0 R4) ------------------
    # ``fabrication_check_mode="revise"`` requires metacognition_enabled=True
    # to produce a genuine revised draft.  When metacog is off, v7.0 pre-R4
    # tagged the response with a synthetic trailer -- observable but not a
    # real revise.  R4 downgrades the effective mode to ``refuse`` in that
    # invalid combo and emits a UserWarning so callers notice.  Config is
    # frozen; we store the effective mode on the instance instead of
    # mutating the config.
    self._effective_fabrication_mode: str = self._config.fabrication_check_mode
    if (
        self._config.fabrication_check_mode == "revise"
        and not self._config.metacognition_enabled
    ):
        import warnings as _warnings

        _warnings.warn(
            "fabrication_check_mode='revise' requires "
            "metacognition_enabled=True; auto-downgrading to 'refuse'",
            UserWarning,
            stacklevel=2,
        )
        self._effective_fabrication_mode = "refuse"

    # -- Auto-seed tracking (per-tenant, avoids re-check every turn) ------
    # Bounded to avoid unbounded growth in multi-tenant deployments with
    # many distinct tenant IDs. Oldest entries are evicted first.
    from collections import OrderedDict as _OrderedDict
    self._seeded_tenants: _OrderedDict[str, None] = _OrderedDict()
    self._seeded_tenants_max = 1_000

    # -- v7.0 T06/T07: intent classifier (stateless, reused across turns) --
    # Constructed unconditionally -- the ``intent_filter_mode`` flag gates
    # whether ``classify`` is ever called. Keeping the instance hot avoids
    # per-turn allocation when the feature is in observe/enforce.
    from symfonic.agent.triage.intent import IntentClassifier as _IC
    self._intent_classifier: Any = _IC()

    # -- v7.1.0 Elicitation (ask_user) checkpointer ------------------------
    # v7.1.1: the engine now picks a CheckpointerFactory (Memory / Sqlite /
    # Postgres) and asks it for an unstarted saver. ``astart()`` runs once
    # on the first run/stream call via ``_ensure_checkpointer_ready``.
    self._checkpointer_factory: CheckpointerFactory | None = None
    self._checkpointer_ready: bool = False
    # v7.1.3 Item 10.2: per-agent pause-token manager. Replaces the old
    # class-level ``PauseToken._store`` ClassVar singleton which two agents
    # in the same process would clobber. The legacy classmethod facade on
    # ``PauseToken`` still exists for backward compatibility with tests.
    from symfonic.agent.middleware.pause_token import PauseTokenManager
    self._pause_token_manager: PauseTokenManager = PauseTokenManager()
    self._checkpointer = self._make_checkpointer()
    # v7.27.0 restart-resume (Q7): emit-once guard so the working-deque
    # rehydrate from the durable checkpointer runs at most once per
    # ``thread_id`` per process lifetime (idempotent, same shape as the
    # working-layer ``_multi_scope_warned`` guard).
    self._resume_rehydrated_threads: set[str] = set()

    # v7.9.3 (adopter dual-construction bug): read from the
    # cached resolved config so the v7.9.0 role_models override
    # survives the second construction.  Pre-v7.9.3 this read
    # ``self._config.agent`` and overwrote the first runtime
    # (line 1235) with a config that had discarded the override.
    self._runtime = AgentRuntime(
        graph=self._graph,
        deps=self._deps,
        config=self._agent_config_resolved,
        checkpointer=self._checkpointer,
    )

    self._validate_lazy_tooling_prerequisites()

    # -- FrameworkObservabilityHook registry (Roadmap Item 10) -----------
    # Engine code dispatches into ``self._framework_hooks`` for the
    # four memory-pipeline lifecycle events declared in
    # ``symfonic.agent.observability.FrameworkObservabilityHook``.
    # The list is mutated in place by the OTEL exporter (see
    # ``OTelExporterHandles.register_into``). Library callers can
    # append additional hooks via ``register_framework_hook``.
    self._framework_hooks: list[Any] = []

    # -- OpenTelemetry exporter (opt-in, Roadmap Item 10) ----------------
    # ``build_if_enabled`` returns ``None`` when ``config.otel_enabled``
    # is False, and the call DOES NOT import any module under
    # ``symfonic.observability.otel.tracer``/``callback_bridge``/
    # ``framework_bridge`` -- preserving the zero-cost invariant
    # verified by ``tests/observability/otel/test_lazy_import.py``.
    from symfonic.observability.otel.ports import build_if_enabled

    self._otel_handles: Any = build_if_enabled(self._config)
    if self._otel_handles is not None:
        # The metrics-collector pre-pending pattern handles
        # CallbackHandler registration via ``_with_metrics_callbacks``
        # so we only need to wire the framework-hook side here.
        self._framework_hooks.append(self._otel_handles.framework_bridge)

    # -- Generic interrupt registry (Roadmap Item 9, v7.2-bound) ---------
    # Holds :class:`InterruptRegistration` records keyed by name.
    # ``ask_user`` is auto-registered as a built-in so it appears in
    # the registry for introspection even though the engine routes
    # it through its dedicated codepath (preserving the byte-identical
    # event stream).  Caller-registered interrupts require
    # ``FrameworkConfig.experimental_interrupt=True``.
    from symfonic.core.contracts.elicitation import (
        AskUserRequest,
        AskUserResponse,
        validate_against_request,
    )
    from symfonic.core.contracts.interrupt import InterruptRegistration
    self._registered_interrupts: dict[str, InterruptRegistration] = {
        "ask_user": InterruptRegistration(
            name="ask_user",
            payload_schema=AskUserRequest,
            response_schema=AskUserResponse,
            validate_response=validate_against_request,
            cross_scope_allowed=False,
            built_in=True,
        ),
    }

    # -- Roadmap Item 8 / PR-6 (7.2): ContextManager strategy ------------
    # One context-assembly strategy per agent lifetime. The factory
    # dispatches on ``context_strategy`` (preferred) and falls back
    # to the legacy ``jit_context`` flag when unset. Strategies are
    # delegation adapters -- the engine's existing ``_hydrate`` /
    # ``_build_hms_system_prompt`` / ``_build_jit_system_prompt``
    # methods are injected and own the rendering logic, so the
    # byte-identity invariant is preserved by construction.
    # ``read_strat_overrides`` snapshots ``_strat_brain_version`` /
    # ``_strat_system_hash`` after the prompt builder runs -- the
    # strategy returns them via ``AssembledContext.state_overrides``
    # instead of the legacy ``_apply_strat_metadata_to_overrides``
    # pour.  ``_hydrate`` continues to fire the
    # ``on_hydration_complete`` framework hook (Phase 4 OTel
    # invariant) because the strategy delegates to it directly.
    # Wrap each engine helper in a thunk that re-looks up the attribute
    # on ``self`` so test-time ``monkeypatch.setattr(agent,
    # "_hydrate_for_strategy", ...)`` (and the equivalent for the HMS /
    # JIT prompt builders) still flows through the strategy. Without
    # the thunk the strategy would hold the bound method captured at
    # construction time, freezing the dispatch surface.
    async def _thunk_hydrate(*args: Any, **kwargs: Any) -> Any:
        return await self._hydrate_for_strategy(*args, **kwargs)

    async def _thunk_resolve_tools(*args: Any, **kwargs: Any) -> Any:
        return await self._lazy_resolve_tools(*args, **kwargs)

    async def _thunk_build_hms(*args: Any, **kwargs: Any) -> Any:
        return await self._build_hms_prompt_for_strategy(
            *args, **kwargs,
        )

    async def _thunk_build_jit(*args: Any, **kwargs: Any) -> Any:
        return await self._build_jit_prompt_for_strategy(
            *args, **kwargs,
        )

    from symfonic.agent.context import make_context_manager
    self._context_manager = make_context_manager(
        self._config,
        hydrate=_thunk_hydrate,
        resolve_tools=_thunk_resolve_tools,
        build_hms_prompt=_thunk_build_hms,
        build_jit_prompt=_thunk_build_jit,
        read_strat_overrides=self._snapshot_strat_overrides,
        enable_hms_prompt=self._enable_hms_prompt,
    )

    # -- T3.5.1: the per-capability cutover switchboard -------------------
    # Construction only reads the compiled-in criteria ledger: no
    # environment variable, no file, no network. The engine's own run and
    # stream bodies stay exactly where they are and stay reachable --
    # ``route_for`` decides which implementation serves each turn.
    #
    # Since TA8.5 there is no lever that puts a capability back on the
    # legacy body: ``rollback`` and ``rollback_process_wide`` refuse on the
    # 11.0 line, and a board whose criteria are unfiled refuses to
    # construct rather than degrading to legacy -- so this line raises if
    # the compiled-in ledger ever loses a citation. The bodies are still
    # entered out of the migrated envelope and on release lines that
    # predate the text-delta chunk contract. T4.4.6 deletes them, not this
    # task.
    # TA8.54: the build's legacy pin, resolved through the release profile
    # and the static binding source -- constructed on this line whether or
    # not a pin was passed, because a construction reachable only when an
    # adopter opts in is a construction nothing in a default deployment
    # takes. It is the *only* remaining supported way to put a capability
    # on its legacy body, and it is set here, at construction, rather than
    # pulled at runtime like the levers TA8.5 retired.
    from symfonic.agent.cutover import CutoverSwitchboard
    from symfonic.agent.cutover.legacy_pin import build_legacy_pin
    self._legacy_pin = build_legacy_pin(self._legacy_pin_request)
    self._cutover = CutoverSwitchboard(pin=self._legacy_pin)
    self._kernel_delegate: Any = None

config property

config: FrameworkConfig

Access the framework configuration.

cutover property

cutover: Any

The per-capability cutover switchboard for this agent (T3.5.1).

Public because the route, its evidence and its fallback counts have to be readable from an operator's REPL at 03:00 without importing a private module: agent.cutover.describe() is the whole procedure.

It used to say agent.cutover.rollback("invocation.run", reason=...) instead, and that call now refuses: TA8.5 retired the lever on the 11.0 line. Refuses rather than disappears, and says so — an operator who types the old procedure is told which line took it and why, not handed an AttributeError.

metrics_collector property

metrics_collector: Any

Return the ConversationMetricsCollector, or None if not configured.

orchestrator property

orchestrator: MemoryOrchestrator

Access the underlying memory orchestrator.

scheduler property

scheduler: Any

Access the optional in-process scheduler.

Returns the scheduler instance if one was configured, or None.

aclose async

aclose() -> None

Release external resources held by the agent.

Closes the checkpointer factory's connection pool (psycopg pool for the Postgres path, aiosqlite connection for the SQLite path) and resets the _checkpointer_ready flag so a fresh astart runs the next time the agent is used.

Idempotent -- calling aclose twice is safe; the second call is a no-op because _checkpointer_factory is reset to None after the first run.

v7.1.3 (Item 10.3): closes the previously-unclosed psycopg_pool.AsyncConnectionPool opened by PostgresCheckpointerFactory.astart. Engine re-creation in the same process (test fixtures, hot-reload, multi-tenant routers) would otherwise leak pools indefinitely. Pair with FastAPI lifespan or atexit for clean shutdown.

v9.2.x (issue #27 leak fix): also releases the resources of any child agents this parent BUILT from a SubAgentSpec -- otherwise each declarative sub-agent's checkpointer pool leaked on parent shutdown. Child aclose is pure resource-release + idempotent; each is guarded so one failure can't strand the others or our own pool. Caller-supplied pre-built SubAgent children are left untouched (the caller owns their lifecycle).

Source code in src/symfonic/agent/engine.py

async def aclose(self) -> None:
    """Release external resources held by the agent.

    Closes the checkpointer factory's connection pool (psycopg pool
    for the Postgres path, aiosqlite connection for the SQLite path)
    and resets the ``_checkpointer_ready`` flag so a fresh ``astart``
    runs the next time the agent is used.

    Idempotent -- calling ``aclose`` twice is safe; the second call
    is a no-op because ``_checkpointer_factory`` is reset to ``None``
    after the first run.

    v7.1.3 (Item 10.3): closes the previously-unclosed
    ``psycopg_pool.AsyncConnectionPool`` opened by
    ``PostgresCheckpointerFactory.astart``. Engine re-creation in the
    same process (test fixtures, hot-reload, multi-tenant routers)
    would otherwise leak pools indefinitely. Pair with FastAPI
    ``lifespan`` or ``atexit`` for clean shutdown.

    v9.2.x (issue #27 leak fix): also releases the resources of any
    child agents this parent BUILT from a ``SubAgentSpec`` -- otherwise
    each declarative sub-agent's checkpointer pool leaked on parent
    shutdown. Child ``aclose`` is pure resource-release + idempotent;
    each is guarded so one failure can't strand the others or our own
    pool. Caller-supplied pre-built ``SubAgent`` children are left
    untouched (the caller owns their lifecycle).
    """
    # Close owned children FIRST, before releasing our own factory.
    # The list is cleared so a second ``aclose`` is a no-op.
    owned_children, self._owned_child_agents = self._owned_child_agents, []
    for child in owned_children:
        try:
            await child.aclose()
        except Exception:
            logger.warning(
                "sub-agent aclose failed during parent shutdown",
                exc_info=True,
            )

    factory = self._checkpointer_factory
    if factory is None:
        # Already closed (or ask_user disabled). Idempotent no-op.
        return
    try:
        await factory.close()
    finally:
        # Whether or not close() raised, drop the references so a
        # second aclose() invocation is a no-op and a subsequent
        # ``_ensure_checkpointer_ready`` cannot reach a half-closed
        # factory.
        self._checkpointer_factory = None
        self._checkpointer_ready = False
        self._checkpointer = None

describe_memory_blocks async

describe_memory_blocks(scope: FrameworkTenantScope) -> list[MemoryBlockStatus]

Report per-block memory status for a scope.

Everything this used to require of a caller — construct a DiscoveryService over the private orchestrator, read domain.required_labels, then cross-reference enabled_layers — was configuration policy being re-derived in a route. The enabled/ required decisions belong to the agent that owns the configuration.

Source code in src/symfonic/agent/engine.py

async def describe_memory_blocks(
    self, scope: FrameworkTenantScope
) -> list[MemoryBlockStatus]:
    """Report per-block memory status for a scope.

    Everything this used to require of a caller — construct a
    ``DiscoveryService`` over the private orchestrator, read
    ``domain.required_labels``, then cross-reference ``enabled_layers`` —
    was configuration policy being re-derived in a route. The enabled/
    required decisions belong to the agent that owns the configuration.
    """
    from symfonic.memory.retrieval.discovery import DiscoveryService

    discovery = DiscoveryService(self._orchestrator)
    blocks = await discovery.scan_status(
        scope.to_memory_scope(),
        required_labels=self._config.domain.required_labels or None,
    )
    return [
        MemoryBlockStatus(
            label=block.label,
            exists=block.exists,
            entry_count=block.entry_count,
            last_updated=block.last_updated,
            importance=block.importance,
            layer=block.layer.value,
            enabled=block.layer.value in self._config.enabled_layers,
            is_required=block.is_required,
        )
        for block in blocks
    ]

describe_resume_dispatch

describe_resume_dispatch(pause_token: str) -> ResumeDispatch

Answer "which registration does this token belong to?" (T4.1.2).

A host needs three facts before it can validate a resume body: the interrupt name the token claims, whether the historical ask_user contract applies, and which schema validates the payload. Before this method a host got them by decoding the token with the agent's private config and reading the agent's private registry — two engine internals and a policy decision sitting in a route (TRN-1/TRN-2).

The preview decode is deliberately not authoritative and deliberately swallows every failure: :meth:resume and :meth:resume_interrupt re-decode and raise the typed error, so a malformed or expired token still fails in the layer that owns token validation and still produces the status the shipped contract returns. Making the preview authoritative would move that decision into transport, where none of the pause-token suites can observe it.

Two token schemes are previewed, not one (TA8.45). A pause minted on the kernel route is an EnvelopeSigner envelope rendered by the deployment's encode_token; PauseToken.decode cannot read one, so before TA8.45 every such token previewed as the default ask_user and POST /resume/{token} sent a registered interrupt's redemption to :meth:resume — the wrong door, with the wrong schema validating the body. That is a defect on the HTTP path only, which is why it is found by driving the HTTP path rather than by inferring it from the Python one. The capability transport is asked first because it is the scheme that can answer with a registration; the HMAC preview is the fallback, unchanged, for every token minted before this build.

Source code in src/symfonic/agent/engine.py

def describe_resume_dispatch(self, pause_token: str) -> ResumeDispatch:
    """Answer "which registration does this token belong to?" (T4.1.2).

    A host needs three facts before it can validate a resume body: the
    interrupt name the token claims, whether the historical ``ask_user``
    contract applies, and which schema validates the payload. Before this
    method a host got them by decoding the token with the agent's private
    config and reading the agent's private registry — two engine internals
    and a policy decision sitting in a route (TRN-1/TRN-2).

    The preview decode is deliberately *not* authoritative and deliberately
    swallows every failure: :meth:`resume` and :meth:`resume_interrupt`
    re-decode and raise the typed error, so a malformed or expired token
    still fails in the layer that owns token validation and still produces
    the status the shipped contract returns. Making the preview
    authoritative would move that decision into transport, where none of
    the pause-token suites can observe it.

    **Two token schemes are previewed, not one (TA8.45).** A pause minted
    on the kernel route is an ``EnvelopeSigner`` envelope rendered by the
    deployment's ``encode_token``; ``PauseToken.decode`` cannot read one, so
    before TA8.45 every such token previewed as the default ``ask_user``
    and ``POST /resume/{token}`` sent a registered interrupt's redemption
    to :meth:`resume` — the wrong door, with the wrong schema validating
    the body. That is a defect on the HTTP path *only*, which is why it is
    found by driving the HTTP path rather than by inferring it from the
    Python one. The capability transport is asked first because it is the
    scheme that can answer with a registration; the HMAC preview is the
    fallback, unchanged, for every token minted before this build.
    """
    from symfonic.agent.middleware.pause_token import PauseToken

    capability = self._human_capability()
    previewed = self._preview_capability_dispatch(capability, pause_token)
    if previewed is not None:
        return previewed

    name = "ask_user"
    try:
        claims = PauseToken.decode(pause_token, self._config)
        name = getattr(claims, "name", "ask_user") or "ask_user"
    except Exception:  # noqa: BLE001 - preview only; the real path re-decodes
        pass

    registration = self._registered_interrupts.get(name)
    is_ask_user = registration is None or (
        registration.built_in and name == "ask_user"
    )
    return ResumeDispatch(
        name=name,
        is_ask_user=is_ask_user,
        known=registration is not None,
        response_schema=(
            None if is_ask_user or registration is None
            else registration.response_schema
        ),
    )

flush_background_tasks async

flush_background_tasks() -> None

Await all pending background tasks (e.g. consolidation).

Call this before the event loop shuts down to ensure background work completes. Also shuts down the scheduler if one is configured. Safe to call multiple times.

v9.2.1 (review P1a): awaits only THIS agent's tasks, not the process-global set, so exiting one agent's context manager never blocks on another agent's pending consolidation.

Source code in src/symfonic/agent/engine.py

async def flush_background_tasks(self) -> None:
    """Await all pending background tasks (e.g. consolidation).

    Call this before the event loop shuts down to ensure background
    work completes.  Also shuts down the scheduler if one is
    configured.  Safe to call multiple times.

    v9.2.1 (review P1a): awaits only THIS agent's tasks, not the
    process-global set, so exiting one agent's context manager never
    blocks on another agent's pending consolidation.
    """
    tasks = list(self._background_tasks)
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)

    if self._config.scheduler is not None:
        await self._config.scheduler.shutdown()

get_chat_model

get_chat_model(role: str = 'action') -> BaseChatModel

v7.16.0 public helper: resolve a role to its bound chat model.

Thin wrapper around resolve_model_config(role) + self._model_provider.get_chat_model(config). Stateless and safe to call from any thread or process — the method reads only self._config and dispatches to the provider's get_chat_model; it does not touch graph state, the checkpointer, the orchestrator, or any per-turn mutable. This is the load-bearing property for background-worker adopters (Celery, RQ, APScheduler) who instantiate one SymfonicAgent per process and call this from worker tasks.

Closes the v7.16-era adopter workaround agent._model_provider.get_chat_model(agent._config.agent.model) which bypassed the role resolver entirely and therefore missed any role_models[role] overrides the adopter had set. The new public method threads through resolve_model_config(role) so role-routing flows into background workers the same way it flows into the engine's run() entrypoint.

Default role="action" matches the v7.9.0 wired role -- the tool-using brain LLM archetype most background workers want. Other roles defined on FrameworkConfig.role_models resolve through the same resolver fallback (see resolve_model_config for the semantics: known role with override -> returns override; unknown / unset role -> returns self._config.agent.model).

Note: the engine's internal call sites (react node, summary, metacognition, reflection, STM summary) continue using the private self._model_provider reference -- they're intra-class and already participate in role routing via the engine's own construction wiring. This public method is for out-of-graph consumers.

Source code in src/symfonic/agent/engine.py

def get_chat_model(self, role: str = "action") -> BaseChatModel:
    """v7.16.0 public helper: resolve a role to its bound chat model.

    Thin wrapper around ``resolve_model_config(role)`` +
    ``self._model_provider.get_chat_model(config)``.  **Stateless
    and safe to call from any thread or process** — the method
    reads only ``self._config`` and dispatches to the provider's
    ``get_chat_model``; it does not touch graph state, the
    checkpointer, the orchestrator, or any per-turn mutable.  This
    is the load-bearing property for background-worker adopters
    (Celery, RQ, APScheduler) who instantiate one ``SymfonicAgent``
    per process and call this from worker tasks.

    Closes the v7.16-era adopter workaround
    ``agent._model_provider.get_chat_model(agent._config.agent.model)``
    which bypassed the role resolver entirely and therefore missed
    any ``role_models[role]`` overrides the adopter had set.  The
    new public method threads through ``resolve_model_config(role)``
    so role-routing flows into background workers the same way it
    flows into the engine's ``run()`` entrypoint.

    Default ``role="action"`` matches the v7.9.0 wired role -- the
    tool-using brain LLM archetype most background workers want.
    Other roles defined on ``FrameworkConfig.role_models`` resolve
    through the same resolver fallback (see
    ``resolve_model_config`` for the semantics: known role with
    override -> returns override; unknown / unset role ->
    returns ``self._config.agent.model``).

    Note: the engine's internal call sites (react node, summary,
    metacognition, reflection, STM summary) continue using the
    private ``self._model_provider`` reference -- they're
    intra-class and already participate in role routing via the
    engine's own construction wiring.  This public method is for
    out-of-graph consumers.
    """
    return self._model_provider.get_chat_model(
        self.resolve_model_config(role),
    )

get_transcript async

get_transcript(scope: Any, session_id: str, *, index: int | None = None, time_range: tuple[datetime, datetime] | None = None, speaker: str = 'all', limit: int | None = None) -> list[TranscriptMessage]

Read the verbatim conversation transcript for a thread (v7.27.0).

Surface 1 of the three-surface model (verbatim replay) -- reads state['messages'] from the LangGraph checkpointer, NOT the lossy episodic HMS surface. Requires a wired checkpointer (transcript_persistence_enabled=True or ask_user_enabled=True) AND, for durability across restart, a durable saver (Postgres or dev_sqlite_checkpoint_path); see docs/concepts/conversation-persistence.md.

Parameters:

Name	Type	Description	Default
scope	Any	Tenant scope (tenant_id + optional sub_tenant_id) used with session_id to derive the thread_id.	required
session_id	str	Session identifier completing the thread_id.	required
index	int | None	0-based ordinal over the speaker-FILTERED view. index=0, speaker="user" is the first human turn (T14). Negative indices count from the end (Python-slice). Out-of-range returns [] (never raises). Mutually exclusive with time_range.	None
time_range	tuple[datetime, datetime] | None	(start, end) inclusive window at CHECKPOINT granularity (per-superstep ts, NOT per-message -- LangChain messages carry no native timestamp). Requires a saver with alist; otherwise raises :class:TranscriptUnsupportedError.	None
speaker	str	"user" / "assistant" / "all". Filters the view BEFORE indexing; "all" includes tool/system rows.	'all'
limit	int | None	Cap on rows returned (applied last, after index/time filtering). None = no cap.	None

Returns:

Type	Description
list[TranscriptMessage]	A list of :class:TranscriptMessage. Empty when no checkpointer
list[TranscriptMessage]	is wired, the thread is unknown, or the query matched nothing.

Raises:

Type	Description
TranscriptUnsupportedError	time_range requested on a saver without alist.

Source code in src/symfonic/agent/engine.py

async def get_transcript(
    self,
    scope: Any,
    session_id: str,
    *,
    index: int | None = None,
    time_range: tuple[datetime, datetime] | None = None,
    speaker: str = "all",
    limit: int | None = None,
) -> list[TranscriptMessage]:
    """Read the verbatim conversation transcript for a thread (v7.27.0).

    Surface 1 of the three-surface model (verbatim replay) -- reads
    ``state['messages']`` from the LangGraph checkpointer, NOT the lossy
    episodic HMS surface.  Requires a wired checkpointer
    (``transcript_persistence_enabled=True`` or ``ask_user_enabled=True``)
    AND, for durability across restart, a durable saver (Postgres or
    ``dev_sqlite_checkpoint_path``); see
    docs/concepts/conversation-persistence.md.

    Args:
        scope: Tenant scope (``tenant_id`` + optional ``sub_tenant_id``)
            used with ``session_id`` to derive the ``thread_id``.
        session_id: Session identifier completing the ``thread_id``.
        index: 0-based ordinal over the ``speaker``-FILTERED view.
            ``index=0, speaker="user"`` is the first human turn (T14).
            Negative indices count from the end (Python-slice).
            Out-of-range returns ``[]`` (never raises).  Mutually
            exclusive with ``time_range``.
        time_range: ``(start, end)`` inclusive window at
            CHECKPOINT granularity (per-superstep ts, NOT per-message --
            LangChain messages carry no native timestamp).  Requires a
            saver with ``alist``; otherwise raises
            :class:`TranscriptUnsupportedError`.
        speaker: ``"user"`` / ``"assistant"`` / ``"all"``.  Filters the
            view BEFORE indexing; ``"all"`` includes tool/system rows.
        limit: Cap on rows returned (applied last, after index/time
            filtering).  ``None`` = no cap.

    Returns:
        A list of :class:`TranscriptMessage`.  Empty when no checkpointer
        is wired, the thread is unknown, or the query matched nothing.

    Raises:
        TranscriptUnsupportedError: ``time_range`` requested on a saver
            without ``alist``.
    """
    from symfonic.agent.transcript import (
        build_transcript_rows,
        filter_by_time_range,
        slice_by_index,
    )

    if index is not None and time_range is not None:
        raise SymfonicAgentError(
            "get_transcript: `index` and `time_range` are mutually "
            "exclusive -- pass at most one.",
            code="bad_request",
        )

    if self._checkpointer is None:
        # No checkpointer wired (default config): no persisted
        # transcript.  Return empty rather than raise -- absence of a
        # transcript is a valid state, not an error.
        return []

    await self._ensure_checkpointer_ready()

    thread_id = self._thread_id_for(scope, session_id)
    config = {"configurable": {"thread_id": thread_id}}

    tuple_ = await self._checkpointer.aget_tuple(config)
    if tuple_ is None:
        return []  # unknown / never-persisted thread
    checkpoint = getattr(tuple_, "checkpoint", None) or {}
    messages = checkpoint.get("channel_values", {}).get("messages", [])
    if not messages:
        return []

    # Timestamp map: required (and fatal-if-absent) for time_range;
    # best-effort for ordinal/full reads (rows get timestamp=None when
    # the saver lacks alist).
    ts_map: dict[str, datetime] | None
    if time_range is not None:
        ts_map = await self._checkpoint_ts_by_message_id(config)
        if ts_map is None:
            raise TranscriptUnsupportedError(
                "get_transcript(time_range=...) requires a checkpointer "
                f"that supports `alist`; {type(self._checkpointer).__name__} "
                "does not.  Use `index` / full-transcript reads, or "
                "configure a Postgres / SQLite saver.",
                code="unsupported",
            )
    else:
        ts_map = await self._checkpoint_ts_by_message_id(config)

    speaker_filter = speaker if speaker in ("user", "assistant", "all") else "all"
    rows = build_transcript_rows(
        messages, speaker=speaker_filter, ts_by_message_id=ts_map
    )

    if index is not None:
        rows = slice_by_index(rows, index)
    elif time_range is not None:
        rows = filter_by_time_range(rows, time_range)

    if limit is not None and limit >= 0:
        rows = rows[:limit]
    return rows

interrupt

interrupt(name: str, payload: Any, *, tool_call_id: str | None = None) -> dict[str, Any]

Build the state-update marker that triggers a registered interrupt.

Roadmap Item 9 surface. Returned from a tool/node body so the graph's :class:InterruptNode picks it up on the next step and calls LangGraph's interrupt() after the AIMessage is safely checkpointed (mirrors the v7.1.0 ask_user topology fix).

Parameters:

Name	Type	Description	Default
name	str	A previously-registered interrupt name.	required
payload	Any	A Pydantic model matching the registered payload_schema.	required
tool_call_id	str | None	Optional tool-call id when the interrupt is triggered from a tool; the resume path injects a ToolMessage carrying the response in that case.	None

Returns:

Type	Description
dict[str, Any]	The state-update dict to be returned by the calling
dict[str, Any]	tool/node, e.g.:: async def my_tool(...) -> dict[str, Any]: return agent.interrupt( "approval_required", ApprovalPayload(amount=100, reason="..."), )

Raises:

Type	Description
SymfonicAgentError	bad_request when the flag is off, the name is not registered, or the payload does not match the registered schema.

Source code in src/symfonic/agent/engine.py

def interrupt(
    self,
    name: str,
    payload: Any,
    *,
    tool_call_id: str | None = None,
) -> dict[str, Any]:
    """Build the state-update marker that triggers a registered interrupt.

    Roadmap Item 9 surface.  Returned from a tool/node body so the
    graph's :class:`InterruptNode` picks it up on the next step and
    calls LangGraph's ``interrupt()`` after the AIMessage is safely
    checkpointed (mirrors the v7.1.0 ``ask_user`` topology fix).

    Args:
        name: A previously-registered interrupt name.
        payload: A Pydantic model matching the registered
            ``payload_schema``.
        tool_call_id: Optional tool-call id when the interrupt is
            triggered from a tool; the resume path injects a
            ``ToolMessage`` carrying the response in that case.

    Returns:
        The state-update dict to be returned by the calling
        tool/node, e.g.::

            async def my_tool(...) -> dict[str, Any]:
                return agent.interrupt(
                    "approval_required",
                    ApprovalPayload(amount=100, reason="..."),
                )

    Raises:
        SymfonicAgentError: ``bad_request`` when the flag is off, the
            name is not registered, or the payload does not match
            the registered schema.
    """
    from symfonic.core.nodes.interrupt import build_interrupt_marker

    if not self._config.experimental_interrupt:
        raise SymfonicAgentError(
            "experimental_interrupt must be True to use "
            "agent.interrupt(). This flag will be removed in 7.3.",
            code="bad_request",
        )
    registration = self._registered_interrupts.get(name)
    if registration is None or registration.built_in:
        raise SymfonicAgentError(
            f"Unknown interrupt name {name!r} -- call "
            f"agent.register_interrupt({name!r}, ...) first.",
            code="bad_request",
        )
    try:
        validated = registration.payload_schema.model_validate(
            payload.model_dump() if hasattr(payload, "model_dump") else payload
        )
    except Exception as exc:
        raise SymfonicAgentError(
            f"Interrupt {name!r}: payload does not match "
            f"registered schema {registration.payload_schema.__name__}: "
            f"{exc}",
            code="bad_request",
        ) from exc

    interrupt_id = f"i-{uuid.uuid4().hex[:12]}"
    return build_interrupt_marker(
        name=name,
        interrupt_id=interrupt_id,
        payload=validated.model_dump(),
        tool_call_id=tool_call_id,
    )

list_sessions

list_sessions(tenant_id: str) -> list[dict[str, Any]]

List the tenant's active chat sessions.

The conversation surface a host needs, so that listing chats is a call rather than a reach into _session_manager (TRN-2). Scope filtering stays where it already was — in the session registry — because a host that could pass any tenant id and get rows back would be the isolation bug this method exists to make impossible to write by accident.

Source code in src/symfonic/agent/engine.py

def list_sessions(self, tenant_id: str) -> list[dict[str, Any]]:
    """List the tenant's active chat sessions.

    The conversation surface a host needs, so that listing chats is a call
    rather than a reach into ``_session_manager`` (TRN-2). Scope filtering
    stays where it already was — in the session registry — because a host
    that could pass any tenant id and get rows back would be the isolation
    bug this method exists to make impossible to write by accident.
    """
    return self._session_manager.list_sessions(tenant_id)

load_plugin

load_plugin(plugin: Any) -> None

Load a domain plugin into the agent -- entirely, or not at all.

Registers the plugin's system prompt contribution and validation hook. Multiple plugins may be loaded; their contributions are chained in registration order.

Admission is atomic (TA8.21). Every reason the kernel route could not honour the plugin's whole surface is established before any registry is touched -- see :meth:_admission_refusal -- and the write that follows rolls back if its second half fails. There is no outcome in which some of this plugin is loaded. The failure modes this replaced were all one shape: load_plugin accepted the plugin and a later stage quietly served less than it declared -- a duplicate name's policies dropped by compose, a plugin loaded after the seam never harvested, an unbridgeable plugin's fragment discarded.

Domain tools must be provided at construction time via the tools= parameter. If a plugin returns tools from get_domain_tools(), this method raises SymfonicAgentError because the AgentGraph's ToolRegistry is frozen after compile().

A refused plugin keeps its veto. It is not registered -- it contributes no prompt fragment, no tool, no lifecycle hook and no name to any composition, and it is absent from every count this agent reports -- but it is remembered in a quarantine that :meth:validate_action asks, deny-only. Refusing a plugin may not delete a guardrail the adopter installed: an adopter who wrote a veto and received an exception has not thereby consented to running without it. See cutover/guardrails.py's _quarantine_outcome.

Parameters:

Name	Type	Description	Default
plugin	Any	Any object satisfying the BaseDomainPlugin protocol.	required

Raises:

Type	Description
SymfonicAgentError	The plugin provides tools (which cannot be registered after the graph has been compiled -- pass domain tools via SymfonicAgent(tools=[...])); or it was offered after the kernel plan compiled; or its surface cannot be carried whole, which includes a name another loaded plugin has already claimed.

Source code in src/symfonic/agent/engine.py

def load_plugin(self, plugin: Any) -> None:
    """Load a domain plugin into the agent -- entirely, or not at all.

    Registers the plugin's system prompt contribution and validation
    hook. Multiple plugins may be loaded; their contributions are
    chained in registration order.

    **Admission is atomic (TA8.21).** Every reason the kernel route could
    not honour the plugin's whole surface is established *before* any
    registry is touched -- see :meth:`_admission_refusal` -- and the write
    that follows rolls back if its second half fails. There is no outcome
    in which some of this plugin is loaded. The failure modes this replaced
    were all one shape: ``load_plugin`` accepted the plugin and a later
    stage quietly served less than it declared -- a duplicate name's
    policies dropped by ``compose``, a plugin loaded after the seam never
    harvested, an unbridgeable plugin's fragment discarded.

    Domain tools must be provided at construction time via the
    ``tools=`` parameter.  If a plugin returns tools from
    ``get_domain_tools()``, this method raises ``SymfonicAgentError``
    because the AgentGraph's ToolRegistry is frozen after compile().

    **A refused plugin keeps its veto.** It is not registered -- it
    contributes no prompt fragment, no tool, no lifecycle hook and no name
    to any composition, and it is absent from every count this agent
    reports -- but it is remembered in a quarantine that
    :meth:`validate_action` asks, deny-only. Refusing a plugin may not
    *delete* a guardrail the adopter installed: an adopter who wrote a veto
    and received an exception has not thereby consented to running without
    it. See ``cutover/guardrails.py``'s ``_quarantine_outcome``.

    Args:
        plugin: Any object satisfying the BaseDomainPlugin protocol.

    Raises:
        SymfonicAgentError: The plugin provides tools (which cannot be
            registered after the graph has been compiled -- pass domain
            tools via ``SymfonicAgent(tools=[...])``); or it was offered
            after the kernel plan compiled; or its surface cannot be
            carried whole, which includes a name another loaded plugin has
            already claimed.
    """
    refusal = self._admission_refusal(plugin)
    if refusal is not None:
        self._quarantine(plugin, refusal)
        raise SymfonicAgentError(refusal)

    self._plugins.append(plugin)
    try:
        self._plugin_section.add_plugin(plugin)
    except Exception as exc:  # noqa: BLE001 - roll back, then refuse
        # The second half of a two-registry write. Leaving the first half
        # in place would be exactly the partial registration the check
        # above exists to prevent, reached from the other direction.
        self._plugins.pop()
        detail = (
            f"Plugin '{getattr(plugin, 'name', 'unknown')}' could not be "
            f"registered with the prompt section ({type(exc).__name__}: "
            f"{exc}). Nothing was left loaded."
        )
        self._quarantine(plugin, detail)
        raise SymfonicAgentError(detail) from exc
    logger.info(
        "Plugin '%s' loaded (no tools)",
        getattr(plugin, "name", "unknown"),
    )

on_user_correction async

on_user_correction(*, turn_id: str, original_response: str, correction_text: str, activation_log: dict[str, Any], scope: FrameworkTenantScope, callbacks: Sequence[CallbackHandler] | None = None) -> None

Dispatch a user correction event to all handlers that support it.

Non-breaking: uses hasattr check at the call site so handlers written before v5.5 are silently skipped.

Source code in src/symfonic/agent/engine.py

async def on_user_correction(
    self,
    *,
    turn_id: str,
    original_response: str,
    correction_text: str,
    activation_log: dict[str, Any],
    scope: FrameworkTenantScope,
    callbacks: Sequence[CallbackHandler] | None = None,
) -> None:
    """Dispatch a user correction event to all handlers that support it.

    Non-breaking: uses hasattr check at the call site so handlers
    written before v5.5 are silently skipped.
    """
    from symfonic.core.contracts.callbacks import UserCorrectionEvent

    event = UserCorrectionEvent(
        turn_id=turn_id,
        original_response=original_response,
        correction_text=correction_text,
        activation_log=activation_log,
        tenant_id=scope.tenant_id,
        # T-7.21.8 (Slice B site #7): stamp the full scope so
        # NegativeReinforcementHandler decays edges in the same
        # partition the original turn lived in.
        scope=scope,
    )
    for h in callbacks or []:
        if hasattr(h, "on_user_correction"):
            try:
                await h.on_user_correction(event)
            except Exception:
                logger.warning(
                    "on_user_correction handler failed", exc_info=True,
                )

register_interrupt

register_interrupt(name: str, payload_schema: type[Any], response_schema: type[Any], *, validate_response: Any = None, cross_scope_allowed: bool = False) -> None

Register a named interrupt point in the graph (Roadmap Item 9).

Parameters:

Name	Type	Description	Default
name	str	Stable identifier (e.g. "approval_required"). Must not collide with a built-in ("ask_user").	required
payload_schema	type[Any]	Pydantic BaseModel subclass for the payload the agent emits when the interrupt fires.	required
response_schema	type[Any]	Pydantic BaseModel subclass for the user-supplied resume body. /resume/{token} validates the request body against this schema.	required
validate_response	Any	Optional cross-validator (response, payload) -> None; raises on mismatch.	None
cross_scope_allowed	bool	If True, the resume endpoint accepts a token issued under one scope to be redeemed under another (logs the event for audit). Default False.	False

Raises:

Type	Description
SymfonicAgentError	bad_request when the flag is off or the name collides.

Source code in src/symfonic/agent/engine.py

def register_interrupt(
    self,
    name: str,
    payload_schema: type[Any],
    response_schema: type[Any],
    *,
    validate_response: Any = None,
    cross_scope_allowed: bool = False,
) -> None:
    """Register a named interrupt point in the graph (Roadmap Item 9).

    Args:
        name: Stable identifier (e.g. ``"approval_required"``). Must
            not collide with a built-in (``"ask_user"``).
        payload_schema: Pydantic ``BaseModel`` subclass for the
            payload the agent emits when the interrupt fires.
        response_schema: Pydantic ``BaseModel`` subclass for the
            user-supplied resume body. ``/resume/{token}`` validates
            the request body against this schema.
        validate_response: Optional cross-validator
            ``(response, payload) -> None``; raises on mismatch.
        cross_scope_allowed: If True, the resume endpoint accepts a
            token issued under one scope to be redeemed under another
            (logs the event for audit). Default False.

    Raises:
        SymfonicAgentError: ``bad_request`` when the flag is off or the
            name collides.
    """
    from symfonic.core.contracts.interrupt import InterruptRegistration

    if not self._config.experimental_interrupt:
        raise SymfonicAgentError(
            "experimental_interrupt must be True to use "
            "register_interrupt(). This flag will be removed in 7.3.",
            code="bad_request",
        )
    if name in self._registered_interrupts:
        existing = self._registered_interrupts[name]
        if existing.built_in:
            raise SymfonicAgentError(
                f"Cannot override built-in interrupt {name!r}",
                code="bad_request",
            )
        raise SymfonicAgentError(
            f"Interrupt {name!r} is already registered",
            code="bad_request",
        )
    self._registered_interrupts[name] = InterruptRegistration(
        name=name,
        payload_schema=payload_schema,
        response_schema=response_schema,
        validate_response=validate_response,
        cross_scope_allowed=cross_scope_allowed,
        built_in=False,
    )

resolve_model_config

resolve_model_config(role: str) -> ModelConfig

v7.9.0 public helper: resolve the ModelConfig to use for a given role.

Returns self._config.role_models[role] when set, falls back to self._config.agent.model otherwise. Pure function of the config; safe to call on every turn.

v7.9.0 ships "action" as the only wired role (consumed by the react node's brain LLM at engine.py:1144-1158 via AgentConfig injection). Other roles defined on FrameworkConfig.role_models are reserved for v7.9.1+ wiring at the summary / critic / router / reflection / consolidation sites. Adopters can still set them today -- the helper resolves them correctly -- but the sites that consume them haven't been re-pointed yet. Calling this helper for an unwired role returns the override anyway, so adopter test code can pin the intended assignment ahead of the v7.9.1 wiring.

T3.1.1: the precedence itself now lives in :class:~symfonic.services.models.ModelResolutionService, which is the same chain agent.backend.model and the per-dispatch resolver read. The behaviour here is unchanged -- role_models[role] when it holds a ModelConfig, otherwise the snapshot default -- but it is no longer a third independent copy of that rule.

Source code in src/symfonic/agent/engine.py

def resolve_model_config(self, role: str) -> ModelConfig:
    """v7.9.0 public helper: resolve the ``ModelConfig`` to use
    for a given role.

    Returns ``self._config.role_models[role]`` when set, falls
    back to ``self._config.agent.model`` otherwise.  Pure function
    of the config; safe to call on every turn.

    v7.9.0 ships ``"action"`` as the only wired role (consumed by
    the react node's brain LLM at ``engine.py:1144-1158`` via
    ``AgentConfig`` injection).  Other roles defined on
    ``FrameworkConfig.role_models`` are reserved for v7.9.1+
    wiring at the summary / critic / router / reflection /
    consolidation sites.  Adopters can still set them today --
    the helper resolves them correctly -- but the sites that
    consume them haven't been re-pointed yet.  Calling this
    helper for an unwired role returns the override anyway, so
    adopter test code can pin the intended assignment ahead of
    the v7.9.1 wiring.

    T3.1.1: the precedence itself now lives in
    :class:`~symfonic.services.models.ModelResolutionService`, which is
    the same chain ``agent.backend.model`` and the per-dispatch resolver
    read.  The behaviour here is unchanged -- ``role_models[role]`` when it
    holds a ``ModelConfig``, otherwise the snapshot default -- but it is no
    longer a third independent copy of that rule.
    """
    from symfonic.services.models.ports import ModelResolutionService

    return ModelResolutionService(
        self._model_provider,
        default_config=self._config.agent.model,
        role_models=self._config.role_models or {},
    ).resolve_config(role)

resume async

resume(pause_token: str, response: AskUserResponse, *, scope: FrameworkTenantScope, callbacks: Sequence[CallbackHandler] | None = None, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, extra_metadata: dict[str, Any] | None = None, **state_overrides: Any) -> AsyncGenerator[StreamEvent, None]

Resume a paused agent execution from an ask_user elicitation.

Dispatched as of TA8.45. This entry point reads invocation.continuation on the board and, for a deployment whose transport can turn the consumer's token back into the envelope it was minted as, continues the paused turn on the kernel through TA8.35's redemption. Otherwise :meth:_legacy_continuation_impl serves it, and still serves every redemption if the switch is put back.

Parameters:

Name	Type	Description	Default
pause_token	str	Opaque token emitted in the original AskUserQuestionEvent.	required
response	AskUserResponse	The user's selections and optional free-text 'other' field.	required
scope	FrameworkTenantScope	Tenant scope — must match the scope that initiated the pause.	required
callbacks	Sequence[CallbackHandler] | None	Retired by TA8.45 on this door. It was a real, honoured parameter here before — a non-empty sequence was forwarded into the graph re-entry — and a non-empty sequence now raises RetiredArgumentError before anything is redeemed, which is the rule run, stream and stream_typed have carried since TA8.18. None and [] are unchanged. See :func:_refuse_retired_arguments and §2.1 of the continuation certificate.	None
session_id	str | None	The conversation this redemption belongs to, if the transport knows it. Stated axes are checked; see :meth:_continuation_kernel_impl for why an unstated one is filled from the token rather than left unchecked.	None
run_id	str | None	The turn this redemption belongs to, if the transport knows it.	None
call_id	str | None	The question of that turn, if the transport knows it.	None
extra_metadata	dict[str, Any] | None	Never a parameter of this method. Accepted into the signature only so an adopter who passes one meets the shared refusal that names the argument and the line rather than a bare TypeError; the wording it shares says "retired on the 11.0 line" because one rule owns one wording.	None
**state_overrides	Any	Never a parameter of this method either, and kept for the same reason.	{}

Yields:

Type	Description
AsyncGenerator[StreamEvent, None]	Typed StreamEvent objects as the run continues.

Raises:

Type	Description
SymfonicAgentError	If the token is invalid, expired, or already used, or if the response fails cross-validation against the original request stored in the checkpoint.

Source code in src/symfonic/agent/engine.py

@_with_active_scope
async def resume(
    self,
    pause_token: str,
    response: AskUserResponse,
    *,
    scope: FrameworkTenantScope,
    callbacks: Sequence[CallbackHandler] | None = None,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    extra_metadata: dict[str, Any] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Resume a paused agent execution from an ``ask_user`` elicitation.

    **Dispatched as of TA8.45.** This entry point reads
    ``invocation.continuation`` on the board and, for a deployment whose
    transport can turn the consumer's token back into the envelope it was
    minted as, continues the paused turn on the kernel through TA8.35's
    redemption. Otherwise :meth:`_legacy_continuation_impl` serves it, and
    still serves every redemption if the switch is put back.

    Args:
        pause_token: Opaque token emitted in the original AskUserQuestionEvent.
        response: The user's selections and optional free-text 'other' field.
        scope: Tenant scope — must match the scope that initiated the pause.
        callbacks: **Retired by TA8.45 on this door.** It was a real,
            honoured parameter here before — a non-empty sequence was
            forwarded into the graph re-entry — and a non-empty sequence
            now raises ``RetiredArgumentError`` before anything is
            redeemed, which is the rule ``run``, ``stream`` and
            ``stream_typed`` have carried since TA8.18. ``None`` and ``[]``
            are unchanged. See :func:`_refuse_retired_arguments` and §2.1
            of the continuation certificate.
        session_id: The conversation this redemption belongs to, if the
            transport knows it. Stated axes are checked; see
            :meth:`_continuation_kernel_impl` for why an *unstated* one is
            filled from the token rather than left unchecked.
        run_id: The turn this redemption belongs to, if the transport
            knows it.
        call_id: The question of that turn, if the transport knows it.
        extra_metadata: Never a parameter of this method. Accepted into
            the signature only so an adopter who passes one meets the
            shared refusal that names the argument and the line rather than
            a bare ``TypeError``; the wording it shares says "retired on
            the 11.0 line" because one rule owns one wording.
        **state_overrides: Never a parameter of this method either, and
            kept for the same reason.

    Yields:
        Typed StreamEvent objects as the run continues.

    Raises:
        SymfonicAgentError: If the token is invalid, expired, or already used,
            or if the response fails cross-validation against the original
            request stored in the checkpoint.
    """
    from symfonic.capabilities.human.registration import (  # noqa: PLC0415
        ASK_USER,
    )

    _refuse_retired_arguments(
        "resume",
        callbacks=callbacks,
        extra_metadata=extra_metadata,
        state_overrides=state_overrides,
    )
    # TA8.48. The same four guards, at the same pre-dispatch position they
    # occupy on run, stream and stream_typed, inside the migration window
    # that keeps a pause minted before they existed redeemable.
    with _continuation_guard_window(
        capability=self._human_capability(), pause_token=pause_token
    ):
        _refuse_retired_configuration("resume", self._config)
        _refuse_unowned_container("resume", self._config)
        _refuse_replaced_pause_setting("resume", self._config)
        _refuse_replaced_orchestrator_setting("resume", self._config)

    # v7.1.1, hoisted here by TA8.45 from the legacy body. A fresh worker
    # can receive /resume/{token} as its very first request -- pause/resume
    # spans worker lifecycles by definition -- so unlike run/stream we
    # cannot rely on a prior call having opened the psycopg pool. It is at
    # the door rather than in one body because *both* continuations touch a
    # checkpointer: the kernel route reads the paused turn's state through
    # the capability's checkpoint port.
    await self._ensure_checkpointer_ready()

    recognition = self._continuation_route(pause_token, scope, session_id)
    body = (
        self._continuation_kernel_impl(
            name=ASK_USER,
            envelope=recognition.envelope,
            response=response,
            scope=scope,
            session_id=session_id,
            run_id=run_id,
            call_id=call_id,
        )
        if recognition is not None
        else self._legacy_continuation_impl(
            name=ASK_USER,
            pause_token=pause_token,
            response=response,
            scope=scope,
            callbacks=callbacks,
        )
    )
    # No ``try`` that returns to the legacy body on failure, and the
    # omission is the point -- see :meth:`_continuation_route`.
    async with contextlib.aclosing(body):
        async for event in body:
            yield event

resume_interrupt async

resume_interrupt(pause_token: str, response: Any, *, scope: FrameworkTenantScope, callbacks: Sequence[CallbackHandler] | None = None, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, extra_metadata: dict[str, Any] | None = None, **state_overrides: Any) -> AsyncGenerator[StreamEvent, None]

Resume a generic interrupt (Roadmap Item 9).

Sibling of :meth:resume (which is ask_user-specific). The router decodes the token first, reads the name claim, and dispatches here when name != 'ask_user'.

response is an already-validated instance of the registered response_schema (the router validates the request body before calling this method).

Dispatched as of TA8.45, and on the same switch as :meth:resume. They are one atomic segment of a turn — the same token contract, the same rehydration, the same kernel entry — and giving them two switches would let a deployment flip ask_user redemption and leave a registered interrupt on the graph runtime, which is two continuation semantics for one paused run. The cost of sharing is that a driver exercising one would certify the other by assumption, so the reachability probe drives both in every configuration it reports.

Source code in src/symfonic/agent/engine.py

@_with_active_scope
async def resume_interrupt(
    self,
    pause_token: str,
    response: Any,
    *,
    scope: FrameworkTenantScope,
    callbacks: Sequence[CallbackHandler] | None = None,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    extra_metadata: dict[str, Any] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Resume a generic interrupt (Roadmap Item 9).

    Sibling of :meth:`resume` (which is ``ask_user``-specific).
    The router decodes the token first, reads the ``name`` claim,
    and dispatches here when ``name != 'ask_user'``.

    ``response`` is an already-validated instance of the registered
    ``response_schema`` (the router validates the request body
    before calling this method).

    **Dispatched as of TA8.45, and on the same switch as :meth:`resume`.**
    They are one atomic segment of a turn — the same token contract, the
    same rehydration, the same kernel entry — and giving them two switches
    would let a deployment flip ``ask_user`` redemption and leave a
    registered interrupt on the graph runtime, which is two continuation
    semantics for one paused run. The cost of sharing is that a driver
    exercising one would certify the other by assumption, so the
    reachability probe drives **both** in every configuration it reports.
    """
    from symfonic.capabilities.human.registration import (  # noqa: PLC0415
        ASK_USER,
    )

    _refuse_retired_arguments(
        "resume_interrupt",
        callbacks=callbacks,
        extra_metadata=extra_metadata,
        state_overrides=state_overrides,
    )
    # TA8.48, and the same block ``resume`` carries: one continuation
    # capability, two public contracts, and a guard that landed on one of
    # them would be exactly the "a driver exercising one certifies the
    # other by assumption" hole this pair is written against.
    with _continuation_guard_window(
        capability=self._human_capability(), pause_token=pause_token
    ):
        _refuse_retired_configuration("resume_interrupt", self._config)
        _refuse_unowned_container("resume_interrupt", self._config)
        _refuse_replaced_pause_setting("resume_interrupt", self._config)
        _refuse_replaced_orchestrator_setting("resume_interrupt", self._config)
    await self._ensure_checkpointer_ready()

    recognition = self._continuation_route(pause_token, scope, session_id)
    body = (
        self._continuation_kernel_impl(
            name=None,
            envelope=recognition.envelope,
            response=response,
            scope=scope,
            session_id=session_id,
            run_id=run_id,
            call_id=call_id,
            forbidden_name=ASK_USER,
        )
        if recognition is not None
        else self._legacy_continuation_impl(
            name=None,
            pause_token=pause_token,
            response=response,
            scope=scope,
            callbacks=callbacks,
        )
    )
    async with contextlib.aclosing(body):
        async for event in body:
            yield event

run async

run(query: str, *, attachments: Sequence[ContentPart] | None = None, scope: FrameworkTenantScope | None = None, callbacks: Sequence[CallbackHandler] | None = None, session_id: str | None = None, history: list[BaseMessage] | None = None, is_admin: bool = False, run_id: str | None = None, extra_metadata: dict[str, Any] | None = None, response_model: type[Any] | None = None, agent_depth: int | None = None, **state_overrides: Any) -> AgentResponse

Execute one turn, on whichever implementation the switch names.

The signature, the return type, and every keyword are unchanged: this is a delegate, not a new API. Two things decide where the turn goes.

1. cutover.route_for("invocation.run") — the capability switch. It is derived from filed cutover criteria (T3.5.2 files the parity and live-cutover criteria this switch is still waiting on), so it starts on legacy and an operator can put it back there at any time.
2. The migrated envelope — whether this invocation needs a capability that has not been flipped yet. A refusal names the feature and is counted, because a fallback nobody can see is a cutover that can be 100% fallback and still look complete.

Everything below the dispatch is unchanged and lives in :meth:_legacy_run_impl, which stays reachable until T4.4.6 retires it after the W5 pre-retirement gates pass.

callbacks, extra_metadata and any keyword landing in **state_overrides are retired on the 11.0 line (TA8.18) and refuse by name before dispatch — see :func:~symfonic.agent.engine._refuse_retired_arguments. The keywords stay in the signature so the failure is a named error citing the line rather than a bare TypeError.

agent_depth is a typed parameter as of TA8.18. It used to ride in **state_overrides, which is retired on the 11.0 line; it is the one key travelling through that map with a home on the migrated side, so it got a signature rather than an error. Delegation keeps calling run(..., agent_depth=N) with the same text it always used. It is refused by the migrated envelope at the same depths it was refused at while it rode in the map, so the promotion did not re-route delegated child turns onto the kernel as a side effect -- see the agent_depth paragraph in :func:~symfonic.agent.cutover.envelope.admit_invocation.

See :meth:_legacy_run_impl for the full argument documentation.

Source code in src/symfonic/agent/engine.py

@_with_active_scope
async def run(
    self,
    query: str,
    *,
    attachments: Sequence[ContentPart] | None = None,
    scope: FrameworkTenantScope | None = None,
    callbacks: Sequence[CallbackHandler] | None = None,
    session_id: str | None = None,
    history: list[BaseMessage] | None = None,
    is_admin: bool = False,
    run_id: str | None = None,
    extra_metadata: dict[str, Any] | None = None,
    response_model: type[Any] | None = None,
    agent_depth: int | None = None,
    **state_overrides: Any,
) -> AgentResponse:
    """Execute one turn, on whichever implementation the switch names.

    The signature, the return type, and every keyword are unchanged: this
    is a delegate, not a new API. Two things decide where the turn goes.

    1. ``cutover.route_for("invocation.run")`` — the capability switch. It
       is derived from filed cutover criteria (T3.5.2 files the parity and
       live-cutover criteria this switch is still waiting on), so it starts
       on ``legacy`` and an operator can put it back there at any time.
    2. The migrated envelope — whether *this* invocation needs a capability
       that has not been flipped yet. A refusal names the feature and is
       counted, because a fallback nobody can see is a cutover that can be
       100% fallback and still look complete.

    Everything below the dispatch is unchanged and lives in
    :meth:`_legacy_run_impl`, which stays reachable until T4.4.6 retires it
    after the W5 pre-retirement gates pass.

    ``callbacks``, ``extra_metadata`` and any keyword landing in
    ``**state_overrides`` are **retired on the 11.0 line** (TA8.18) and
    refuse by name before dispatch — see
    :func:`~symfonic.agent.engine._refuse_retired_arguments`. The keywords
    stay in the signature so the failure is a named error citing the line
    rather than a bare ``TypeError``.

    ``agent_depth`` is a typed parameter as of TA8.18. It used to ride in
    ``**state_overrides``, which is retired on the 11.0 line; it is the one
    key travelling through that map with a home on the migrated side, so it
    got a signature rather than an error. Delegation keeps calling
    ``run(..., agent_depth=N)`` with the same text it always used. It is
    refused by the migrated envelope at the same depths it was refused at
    while it rode in the map, so the promotion did not re-route delegated
    child turns onto the kernel as a side effect -- see the ``agent_depth``
    paragraph in :func:`~symfonic.agent.cutover.envelope.admit_invocation`.

    See :meth:`_legacy_run_impl` for the full argument documentation.
    """
    _refuse_retired_arguments(
        "run",
        callbacks=callbacks,
        extra_metadata=extra_metadata,
        state_overrides=state_overrides,
    )
    _refuse_retired_configuration("run", self._config)
    _refuse_unowned_container("run", self._config)
    _refuse_replaced_pause_setting("run", self._config)
    _refuse_replaced_orchestrator_setting("run", self._config)
    # TA8.43 (C1-K): refuse the self-asserted keyword, then rebind the local
    # to the authority an authenticated principal actually carries. Every
    # reader below -- the kernel branch's ``_apply_scope_effects`` and
    # ``_legacy_run_impl``'s -- then reads one derived value on both routes,
    # rather than the two routes agreeing by inspection.
    is_admin = _admin_authority("run", is_admin, agent_depth)
    _refuse_unaddressable_transcript(
        "run", self._config, scope, session_id
    )

    from symfonic.agent.cutover import INVOCATION_RUN, Route

    if self._cutover.route_for(INVOCATION_RUN) is Route.KERNEL:
        verdict = self._cutover_verdict(
            scope=scope,
            session_id=session_id,
            history=history,
            attachments=attachments,
            callbacks=callbacks,
            extra_metadata=extra_metadata,
            state_overrides=state_overrides,
            agent_depth=agent_depth,
        )
        if verdict.admitted:
            # TA8.11: the turn-scope effects that are neither recall nor
            # prompt tenancy, in the position ``_legacy_run_impl`` runs
            # them. Without this an admitted scoped turn skipped the
            # tenant budget breaker and the AGENT_IDENTITY seed, so the
            # scope was honoured for what it fetched and dropped for what
            # it *gates* -- the shape of drop this envelope exists to stop.
            await self._apply_scope_effects(
                scope, is_admin=is_admin, top_level_only=True
            )
            _run_id = _resolve_run_id(run_id)
            # TA8.19: the session the legacy body would have issued,
            # issued here, in the position ``_legacy_run_impl`` issues it
            # and *before* the delegate is called. Order matters: a
            # cross-tenant collision makes ``ensure_session`` hand back a
            # different id than the caller passed, and the delegate echoes
            # verbatim onto ``AgentResponse.session_id`` -- resolving
            # inside it would report the refused id as accepted.
            _session_id = self._resolve_session(scope, session_id, run_id=_run_id)
            return await self._cutover_delegate().run(
                query,
                run_id=_run_id,
                session_id=_session_id or "",
                response_model=response_model,
                # The turn's scope, translated once here. Without it the
                # bundle's construction-time default answered for every
                # tenant: the query went to the wrong scope, came back
                # empty, and the prompt lost its recall silently.
                scope=_capability_scope(scope),
                # TA8.10 (G1). Both are forwarded verbatim; the delegate
                # trims the history with this engine's own
                # ``_pair_aware_history_slice`` and hands the attachments
                # to this engine's own ``_build_human_content``. Admitting
                # them in the envelope without forwarding them here would
                # serve the turn on the kernel with the conversation and
                # the images dropped -- admitted, answered, and wrong.
                history=history,
                attachments=attachments,
                # TA8.20: telemetry identity, taken from the same expression
                # ``_legacy_run_impl`` hands to ``_otel_run_span`` so an
                # admitted run is attributed to the tenant the replaced path
                # attributed it to. Not ``_capability_scope(scope).tenant``:
                # that translation answers ``None`` for a scope it cannot
                # read, which is the right fallback for recall and the wrong
                # one for a billing attribution.
                tenant_id=scope.tenant_id if scope is not None else None,
                # TA8.12 (G5a). The delegate opens the delegation run scope
                # at this depth, which is what the ceiling check reads. The
                # envelope admits ``agent_depth`` on the strength of that
                # consumer, so *not* forwarding it here would turn the
                # admission into the silent drop the allowlist forbids: a
                # delegated grandchild would be offered ``run_agent`` at
                # depth 0 and the ceiling would never be reached.
                agent_depth=agent_depth,
            )
        self._cutover.record_fallback(
            INVOCATION_RUN, verdict.reason or "outside the migrated envelope"
        )

    return await self._legacy_run_impl(
        query,
        attachments=attachments,
        scope=scope,
        callbacks=callbacks,
        session_id=session_id,
        history=history,
        is_admin=is_admin,
        run_id=run_id,
        extra_metadata=extra_metadata,
        response_model=response_model,
        agent_depth=agent_depth,
        **state_overrides,
    )

scrub_properties

scrub_properties(properties: Any) -> Any

Apply the agent's configured credential-key scrubber (public since T4.1.2).

Wraps :func:symfonic.agent.hygiene.scrub_credential_keys with the pattern compiled from self._config.credential_patterns. When credential_patterns=[] the pattern is None and this returns the input unchanged (explicit opt-out).

It is public because a host has to be able to scrub the properties it accepts on a write path, and reaching for _scrub_props to do it (five call sites in the FastAPI adapter) made a data-visibility rule depend on a private attribute — which is exactly the kind of rule a second host forgets. Removing the credential scrub from a write path is a security regression, so the call has to be available, not hidden.

Source code in src/symfonic/agent/engine.py

def scrub_properties(self, properties: Any) -> Any:
    """Apply the agent's configured credential-key scrubber (public since
    T4.1.2).

    Wraps :func:`symfonic.agent.hygiene.scrub_credential_keys` with the
    pattern compiled from ``self._config.credential_patterns``. When
    ``credential_patterns=[]`` the pattern is ``None`` and this returns the
    input unchanged (explicit opt-out).

    It is public because a host has to be able to scrub the properties it
    accepts on a write path, and reaching for ``_scrub_props`` to do it
    (five call sites in the FastAPI adapter) made a data-visibility rule
    depend on a private attribute — which is exactly the kind of rule a
    second host forgets. Removing the credential scrub from a write path is
    a security regression, so the call has to be available, not hidden.
    """
    return _hygiene_scrub_credential_keys(
        properties, self._credential_pattern,
    )

stream async

stream(query: str, *, attachments: Sequence[ContentPart] | None = None, scope: FrameworkTenantScope | None = None, callbacks: Sequence[CallbackHandler] | None = None, session_id: str | None = None, history: list[BaseMessage] | None = None, is_admin: bool = False, run_id: str | None = None, extra_metadata: dict[str, Any] | None = None, response_model: type[Any] | None = None, agent_depth: int | None = None, **state_overrides: Any) -> AsyncGenerator[StreamChunk, None]

Stream agent execution as a sequence of StreamChunks.

Parameters:

Name	Type	Description	Default
query	str	User query text -- the canonical semantic anchor used for HMS hydration, embeddings, and tool routing.	required
attachments	Sequence[ContentPart] | None	Optional non-text content (Attachment wrappers or raw LangChain content-block dicts) forwarded to the LLM as additional content blocks. Attachments do NOT participate in memory hydration or tool routing in v1.	None
scope	FrameworkTenantScope | None	Tenant scope for multi-tenant isolation.	None
callbacks	Sequence[CallbackHandler] | None	Retired on the 11.0 line (TA8.18); supplying handlers raises RetiredArgumentError before dispatch.	None
session_id	str | None	Optional session identifier for session tracking.	None
history	list[BaseMessage] | None	Optional list of prior conversation messages to prepend.	None
is_admin	bool	True to bypass budget checks (v7.0.3).	False
run_id	str | None	Optional run identifier for correlation (v7.1.0). It is the identity the kernel route's observability suite attributes every lifecycle and telemetry observation to (TA8.41, C1-R); one is generated when it is omitted.	None
response_model	type[Any] | None	Not served here. Structured output is a blocking-turn contract on the 11.0 line -- it binds to the compiled plan and is delivered on AgentResponse.structured, and neither streaming projection has a field to carry it. Supplying one raises StructuredOutputUnsupportedError naming this entry point (TA8.41, C1-L); call run(query, response_model=...) instead. The keyword is in the signature so the refusal names response_model rather than the retired **state_overrides the argument used to land in.	None
agent_depth	int | None	Delegation depth to stamp on this run. A typed parameter as of TA8.18; it used to ride in **state_overrides, which is retired on the 11.0 line.	None

Yields: StreamChunk events with appropriate event_type progression: thinking -> acting/text_delta/tool_call/tool_result -> consolidating -> done

Source code in src/symfonic/agent/engine.py

@_owned_async_generator
async def stream(
    self,
    query: str,
    *,
    attachments: Sequence[ContentPart] | None = None,
    scope: FrameworkTenantScope | None = None,
    callbacks: Sequence[CallbackHandler] | None = None,
    session_id: str | None = None,
    history: list[BaseMessage] | None = None,
    is_admin: bool = False,
    run_id: str | None = None,
    extra_metadata: dict[str, Any] | None = None,
    response_model: type[Any] | None = None,
    agent_depth: int | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[StreamChunk, None]:
    """Stream agent execution as a sequence of StreamChunks.

    Args:
        query: User query text -- the canonical semantic anchor used for
            HMS hydration, embeddings, and tool routing.
        attachments: Optional non-text content (``Attachment`` wrappers or
            raw LangChain content-block dicts) forwarded to the LLM as
            additional content blocks.  Attachments do NOT participate in
            memory hydration or tool routing in v1.
        scope: Tenant scope for multi-tenant isolation.
        callbacks: Retired on the 11.0 line (TA8.18); supplying handlers
            raises ``RetiredArgumentError`` before dispatch.
        session_id: Optional session identifier for session tracking.
        history: Optional list of prior conversation messages to prepend.
        is_admin: True to bypass budget checks (v7.0.3).
        run_id: Optional run identifier for correlation (v7.1.0). It is
            the identity the kernel route's observability suite attributes
            every lifecycle and telemetry observation to (TA8.41, C1-R);
            one is generated when it is omitted.
        response_model: Not served here. Structured output is a
            blocking-turn contract on the 11.0 line -- it binds to the
            compiled plan and is delivered on
            ``AgentResponse.structured``, and neither streaming
            projection has a field to carry it. Supplying one raises
            ``StructuredOutputUnsupportedError`` naming this entry point
            (TA8.41, C1-L); call ``run(query, response_model=...)``
            instead. The keyword is in the signature so the refusal names
            ``response_model`` rather than the retired ``**state_overrides``
            the argument used to land in.
        agent_depth: Delegation depth to stamp on this run. A typed
            parameter as of TA8.18; it used to ride in
            ``**state_overrides``, which is retired on the 11.0 line.
    Yields:
        StreamChunk events with appropriate event_type progression:
        thinking -> acting/text_delta/tool_call/tool_result -> consolidating -> done
    """
    _refuse_retired_arguments(
        "stream",
        callbacks=callbacks,
        extra_metadata=extra_metadata,
        state_overrides=state_overrides,
    )
    _refuse_retired_configuration("stream", self._config)
    _refuse_unowned_container("stream", self._config)
    _refuse_replaced_pause_setting("stream", self._config)
    _refuse_replaced_orchestrator_setting("stream", self._config)
    # TA8.43 (C1-K), as in ``run`` and for the same reason.
    is_admin = _admin_authority("stream", is_admin, agent_depth)
    _refuse_unaddressable_transcript(
        "stream", self._config, scope, session_id
    )
    _refuse_streaming_response_model("stream", response_model)

    token = _active_scope.set(scope)
    run_id_token = _active_run_id.set(str(run_id or "") or None)
    depth_token, snapshot_token = _open_stream_depth(agent_depth)
    try:
        # TA8.41 (C1-L): the same guard, in the same position, saying
        # which field declined and on which surface. It was a bare
        # ``SymfonicAgentError`` whose message named no field, which is why
        # the admission inventory could only record the outcome as
        # ``unknown``. ``StreamingDisabledError`` subclasses that class, so
        # every existing ``except`` and every existing message match still
        # catch it -- a widening, never a rename.
        _refuse_disabled_streaming("stream", self._config)

        # T3.5.1: the streaming projection of the same cutover decision
        # ``run`` makes. Same semantics, separate switch: a streaming
        # regression must be rollback-able without dragging the blocking
        # path back to legacy with it. The legacy body (``_stream_impl``)
        # is untouched below and stays reachable until T4.4.6.
        from symfonic.agent.cutover import INVOCATION_STREAM, Route
        from symfonic.agent.stream_contract import (  # noqa: PLC0415
            text_delta_contract_is_published,
        )

        _stream_switched = (
            self._cutover.route_for(INVOCATION_STREAM) is Route.KERNEL
        )
        # TA8.2: the filed parity is scoped to the line that publishes the
        # text-delta chunk contract, and the switch alone cannot carry it
        # past that. ``route_for`` derives the route from criteria
        # completeness, which is deliberate -- a route is evidence, not a
        # knob -- so the schedule half of task #25 option C is enforced
        # *here*, at the dispatch, where the release line is a fact about
        # the running install rather than about the ledger.
        #
        # On 9.x/10.x the legacy body yields the graph runtime's node-update
        # mappings, which is those lines' *published* contract; taking the
        # delegate there would ship option C's breaking change on a line
        # that deliberately withheld it. So the turn falls back, and the
        # fallback is counted like every other one: a flipped switch that
        # still runs legacy is data, never a silent detour.
        if _stream_switched and not text_delta_contract_is_published():
            self._cutover.record_fallback(
                INVOCATION_STREAM,
                "release line predates the text-delta chunk contract",
            )
            _stream_switched = False
        if _stream_switched:
            _verdict = self._cutover_verdict(
                scope=scope,
                session_id=session_id,
                history=history,
                attachments=attachments,
                callbacks=callbacks,
                extra_metadata=extra_metadata,
                state_overrides=state_overrides,
                agent_depth=agent_depth,
            )
            if _verdict.admitted:
                # TA8.11: the same two scope effects the legacy body owes,
                # before the first chunk. The breaker in particular has to
                # raise instead of yield, which is why it is here rather
                # than inside the delegate's generator.
                #
                # ``top_level_only`` matches the legacy body below and
                # ``run`` on both routes (issue #63). A route-conditional
                # identity guard would be the worst shape this control can
                # take: which agent owns the tenant's persona would depend
                # on which body served the first turn.
                await self._apply_scope_effects(
                    scope, is_admin=is_admin, top_level_only=True
                )
                _kernel_run_id = _resolve_run_id(run_id)
                # TA8.19: the streaming half of the same resolution. The
                # legacy copy of this block lives in ``stream``'s own
                # fall-through below rather than in ``_stream_impl``, which
                # is why the kernel branch returning here used to mean the
                # streaming route registered no session at all.
                _kernel_session_id = self._resolve_session(
                    scope, session_id, run_id=_kernel_run_id
                )
                _kernel_stream = self._cutover_delegate().stream(
                    query,
                    run_id=_kernel_run_id,
                    session_id=_kernel_session_id or "",
                    # The streaming half of the same binding. Admission
                    # already accepted a caller scope here, so omitting it
                    # admitted a tenant's stream and then recalled from the
                    # bundle's default -- the one shape where a wrong scope
                    # can reach a different tenant's data.
                    scope=_capability_scope(scope),
                    # The streaming half of TA8.10's G1 seam. Same two
                    # arguments, same delegate translation: a stream that
                    # admitted a conversation and then asked the model
                    # without it would answer the wrong question and yield
                    # the answer one token at a time.
                    history=history,
                    attachments=attachments,
                    # TA8.20, the streaming half. Same expression the
                    # streaming legacy body hands to ``_otel_run_span``.
                    tenant_id=scope.tenant_id if scope is not None else None,
                    # TA8.12, the streaming half. The delegate holds the
                    # delegation scope open for the whole drain, because a
                    # hand-off happens mid-stream and a scope closed after
                    # the first chunk would read depth 0 for every one.
                    agent_depth=agent_depth,
                )
                async with contextlib.aclosing(_kernel_stream):
                    async for _chunk in _kernel_stream:
                        yield _chunk
                return
            self._cutover.record_fallback(
                INVOCATION_STREAM,
                _verdict.reason or "outside the migrated envelope",
            )

        # v7.1.1: fail-loud guard mirroring run() — see the equivalent
        # guard in `run()` for rationale.
        if self._config.ask_user_enabled and (scope is None or not session_id):
            raise SymfonicAgentError(
                "ask_user_enabled=True requires both `scope` and `session_id` "
                "to be supplied on the first turn so a deterministic thread_id "
                "can be derived. Pause-resume cycles cannot bind a token to an "
                "ephemeral LangGraph-generated thread.",
                code="bad_request",
            )

        # v7.1.1: open the psycopg checkpointer pool + run idempotent DDL on
        # the first real invocation. No-op for Memory/Sqlite factories.
        await self._ensure_checkpointer_ready()

        # v7.27.0 restart-resume (Q7): replay the persisted transcript
        # tail into an empty working deque on first turn of a resumed
        # session.  One-shot per thread_id; no-op otherwise.
        await self._maybe_rehydrate_working(scope, session_id)

        # Defence-in-depth: same breaker as run().  Raises before we
        # yield any chunks so the caller gets an immediate failure. Then
        # auto-seed AGENT_IDENTITY on first interaction per tenant.
        #
        # TA8.11: both live in ``_apply_scope_effects`` now, which the
        # kernel branch of :meth:`stream` calls before it delegates.
        #
        # ``top_level_only`` was False here through TA8.11 -- this body
        # never had ``run``'s depth check and acquiring one under a cutover
        # task would have been a behaviour change smuggled in. It is True
        # now because that behaviour change is issue #63's remaining half,
        # made deliberately: ``agent_depth`` is a supported keyword on
        # ``stream`` (``_open_stream_depth``), so a delegated child
        # consumed through this body seeded the PARENT tenant's persona
        # under the CHILD's domain name -- permanently, since the node is
        # importance 9.0 (decay-exempt) and AGENT_IDENTITY is excluded from
        # semantic merge. Top-level streaming turns are untouched: they run
        # at depth 0 and still seed.
        await self._apply_scope_effects(
            scope, is_admin=is_admin, top_level_only=True
        )

        run_id = _resolve_run_id(run_id)
        start = time.monotonic()

        # Register tenant with metrics collector so on_llm_end can fan
        # usage into the TokenBudgetTracker (budget ceilings + /billing).
        if self._metrics_collector is not None and scope is not None:
            set_tenant = getattr(self._metrics_collector, "set_tenant", None)
            if set_tenant is not None:
                set_tenant(run_id, scope.tenant_id)

        # Track session. TA8.19: one derivation site, shared with the
        # kernel branch above and with ``run``'s two routes.
        session_id = self._resolve_session(scope, session_id, run_id=run_id)

        # Roadmap Item 10 / PR-5b (follow-up #8): open the OTel root span for
        # the duration of the stream so child node/llm/tool spans nest
        # underneath. When ``otel_enabled=False`` ``_otel_run_span`` returns
        # a ``nullcontext`` so the cold path is byte-for-byte unchanged.
        # The body is factored into ``_stream_impl`` so the wrap stays thin
        # and async-generator teardown (consumer disconnect via
        # ``GeneratorExit``) cleanly closes the span exactly once.
        _otel_cm = self._otel_run_span(
            run_id=run_id,
            scope=scope,
            session_id=session_id,
            query=query,
            entry_point="stream",
        )
        _otel_cm.__enter__()
        _otel_exited = False
        iterator = self._stream_impl(
            query=query,
            attachments=attachments,
            scope=scope,
            callbacks=callbacks,
            session_id=session_id,
            history=history,
            run_id=run_id,
            start=start,
            state_overrides=state_overrides,
            extra_metadata=extra_metadata,
            agent_depth=agent_depth,
        )
        try:
            async with contextlib.aclosing(iterator):
                async for chunk in iterator:
                    yield chunk
        except GeneratorExit:
            # Consumer disconnect: close the span cleanly (no ERROR status)
            # because the cancellation is intentional, not a fault.
            if not _otel_exited:
                _otel_cm.__exit__(None, None, None)
                _otel_exited = True
            raise
        except BaseException:
            import sys as _sys
            if not _otel_exited:
                _otel_cm.__exit__(*_sys.exc_info())
                _otel_exited = True
            raise
        else:
            if not _otel_exited:
                _otel_cm.__exit__(None, None, None)
                _otel_exited = True
        finally:
            # Backstop: the sentinel ensures we never invoke ``__exit__``
            # twice on the underlying ``@contextmanager`` generator (which
            # would raise ``RuntimeError("generator didn't stop")``).
            if not _otel_exited:
                _otel_cm.__exit__(None, None, None)
    finally:
        _close_stream_depth(depth_token, snapshot_token)
        with contextlib.suppress(ValueError):
            _active_run_id.reset(run_id_token)
        with contextlib.suppress(ValueError):
            _active_scope.reset(token)

stream_text async

stream_text(query: str, *, attachments: Sequence[ContentPart] | None = None, scope: FrameworkTenantScope | None = None, callbacks: Sequence[CallbackHandler] | None = None, session_id: str | None = None, history: list[BaseMessage] | None = None) -> AsyncGenerator[str, None]

Stream only clean response text — the simplest streaming API.

Wraps stream_typed() and yields only the str payload from TextDeltaEvent objects. All other events (lifecycle, tool calls, activation, etc.) are silently consumed so consolidation still runs. Extraction blocks are automatically stripped by the underlying filter.

Parameters:

Name	Type	Description	Default
query	str	User query text -- the canonical semantic anchor used for HMS hydration, embeddings, and tool routing.	required
attachments	Sequence[ContentPart] | None	Optional non-text content (Attachment wrappers or raw LangChain content-block dicts) forwarded to the LLM as additional content blocks. Attachments do NOT participate in memory hydration or tool routing in v1.	None
scope	FrameworkTenantScope | None	Tenant scope for multi-tenant isolation.	None
callbacks	Sequence[CallbackHandler] | None	Optional per-invocation callback handlers.	None
session_id	str | None	Optional session identifier for session tracking.	None
history	list[BaseMessage] | None	Optional list of prior conversation messages to prepend.	None

Yields:

Type	Description
AsyncGenerator[str, None]	Clean text strings suitable for direct display in a chat UI.

Source code in src/symfonic/agent/engine.py

async def stream_text(
    self,
    query: str,
    *,
    attachments: Sequence[ContentPart] | None = None,
    scope: FrameworkTenantScope | None = None,
    callbacks: Sequence[CallbackHandler] | None = None,
    session_id: str | None = None,
    history: list[BaseMessage] | None = None,
) -> AsyncGenerator[str, None]:
    """Stream only clean response text — the simplest streaming API.

    Wraps ``stream_typed()`` and yields only the ``str`` payload from
    ``TextDeltaEvent`` objects.  All other events (lifecycle, tool calls,
    activation, etc.) are silently consumed so consolidation still runs.
    Extraction blocks are automatically stripped by the underlying filter.

    Args:
        query: User query text -- the canonical semantic anchor used for
            HMS hydration, embeddings, and tool routing.
        attachments: Optional non-text content (``Attachment`` wrappers or
            raw LangChain content-block dicts) forwarded to the LLM as
            additional content blocks.  Attachments do NOT participate in
            memory hydration or tool routing in v1.
        scope: Tenant scope for multi-tenant isolation.
        callbacks: Optional per-invocation callback handlers.
        session_id: Optional session identifier for session tracking.
        history: Optional list of prior conversation messages to prepend.

    Yields:
        Clean text strings suitable for direct display in a chat UI.
    """
    iterator = self.stream_typed(
        query,
        attachments=attachments,
        scope=scope,
        callbacks=callbacks,
        session_id=session_id,
        history=history,
    )
    async with contextlib.aclosing(iterator):
        async for event in iterator:
            if isinstance(event, TextDeltaEvent):
                yield event.text

stream_typed async

stream_typed(query: str, *, attachments: Sequence[ContentPart] | None = None, scope: FrameworkTenantScope | None = None, callbacks: Sequence[CallbackHandler] | None = None, session_id: str | None = None, history: list[BaseMessage] | None = None, is_admin: bool = False, run_id: str | None = None, extra_metadata: dict[str, Any] | None = None, response_model: type[Any] | None = None, agent_depth: int | None = None, **state_overrides: Any) -> AsyncGenerator[StreamEvent, None]

Stream agent execution as typed StreamEvent objects.

Identical hydration, scope injection, and HMS prompt composition as stream(), but yields typed StreamEvent instances (TextDeltaEvent, ToolCallStartEvent, ThinkingDeltaEvent, etc.) instead of StreamChunk wrappers. Designed for library consumers that want structured event access without parsing raw LangGraph dicts.

Parameters:

Name	Type	Description	Default
query	str	User query text -- the canonical semantic anchor used for HMS hydration, embeddings, and tool routing.	required
attachments	Sequence[ContentPart] | None	Optional non-text content (Attachment wrappers or raw LangChain content-block dicts) forwarded to the LLM as additional content blocks. Attachments do NOT participate in memory hydration or tool routing in v1.	None
scope	FrameworkTenantScope | None	Tenant scope for multi-tenant isolation.	None
callbacks	Sequence[CallbackHandler] | None	Optional per-invocation callback handlers.	None
session_id	str | None	Optional session identifier for session tracking.	None
history	list[BaseMessage] | None	Optional list of prior conversation messages to prepend.	None
is_admin	bool	True to bypass budget checks (v7.0.3).	False
run_id	str | None	Optional run identifier for correlation (v7.1.0). It is the identity the kernel route's observability suite attributes every lifecycle and telemetry observation to (TA8.41, C1-R); one is generated when it is omitted.	None
response_model	type[Any] | None	Not served here. Structured output is a blocking-turn contract on the 11.0 line -- it binds to the compiled plan and is delivered on AgentResponse.structured, and neither streaming projection has a field to carry it. Supplying one raises StructuredOutputUnsupportedError naming this entry point (TA8.41, C1-L); call run(query, response_model=...) instead. The keyword is in the signature so the refusal names response_model rather than the retired **state_overrides the argument used to land in.	None
agent_depth	int | None	Delegation depth to stamp on this run. A typed parameter as of TA8.18; it used to ride in **state_overrides, which is retired on the 11.0 line.	None
**state_overrides	Any	Retired on the 11.0 line (TA8.18). The keyword is kept in the signature so an adopter who passes one meets a refusal that names the argument, cites the line and says what to do instead, rather than a bare TypeError that names none of the three.	{}

Yields:

Type	Description
AsyncGenerator[StreamEvent, None]	Typed StreamEvent objects from the EventTranspiler.

Source code in src/symfonic/agent/engine.py

@_owned_async_generator
async def stream_typed(
    self,
    query: str,
    *,
    attachments: Sequence[ContentPart] | None = None,
    scope: FrameworkTenantScope | None = None,
    callbacks: Sequence[CallbackHandler] | None = None,
    session_id: str | None = None,
    history: list[BaseMessage] | None = None,
    is_admin: bool = False,
    run_id: str | None = None,
    extra_metadata: dict[str, Any] | None = None,
    response_model: type[Any] | None = None,
    agent_depth: int | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Stream agent execution as typed StreamEvent objects.

    Identical hydration, scope injection, and HMS prompt composition
    as ``stream()``, but yields typed ``StreamEvent`` instances
    (TextDeltaEvent, ToolCallStartEvent, ThinkingDeltaEvent, etc.)
    instead of ``StreamChunk`` wrappers.  Designed for library
    consumers that want structured event access without parsing raw
    LangGraph dicts.

    Args:
        query: User query text -- the canonical semantic anchor used for
            HMS hydration, embeddings, and tool routing.
        attachments: Optional non-text content (``Attachment`` wrappers or
            raw LangChain content-block dicts) forwarded to the LLM as
            additional content blocks.  Attachments do NOT participate in
            memory hydration or tool routing in v1.
        scope: Tenant scope for multi-tenant isolation.
        callbacks: Optional per-invocation callback handlers.
        session_id: Optional session identifier for session tracking.
        history: Optional list of prior conversation messages to prepend.
        is_admin: True to bypass budget checks (v7.0.3).
        run_id: Optional run identifier for correlation (v7.1.0). It is
            the identity the kernel route's observability suite attributes
            every lifecycle and telemetry observation to (TA8.41, C1-R);
            one is generated when it is omitted.
        response_model: Not served here. Structured output is a
            blocking-turn contract on the 11.0 line -- it binds to the
            compiled plan and is delivered on
            ``AgentResponse.structured``, and neither streaming
            projection has a field to carry it. Supplying one raises
            ``StructuredOutputUnsupportedError`` naming this entry point
            (TA8.41, C1-L); call ``run(query, response_model=...)``
            instead. The keyword is in the signature so the refusal names
            ``response_model`` rather than the retired ``**state_overrides``
            the argument used to land in.
        agent_depth: Delegation depth to stamp on this run. A typed
            parameter as of TA8.18; it used to ride in
            ``**state_overrides``, which is retired on the 11.0 line.
        **state_overrides: Retired on the 11.0 line (TA8.18). The keyword
            is kept in the signature so an adopter who passes one meets a
            refusal that names the argument, cites the line and says what
            to do instead, rather than a bare ``TypeError`` that names
            none of the three.

    Yields:
        Typed StreamEvent objects from the EventTranspiler.
    """
    _refuse_retired_arguments(
        "stream_typed",
        callbacks=callbacks,
        extra_metadata=extra_metadata,
        state_overrides=state_overrides,
    )
    _refuse_retired_configuration("stream_typed", self._config)
    _refuse_unowned_container("stream_typed", self._config)
    _refuse_replaced_pause_setting("stream_typed", self._config)
    _refuse_replaced_orchestrator_setting("stream_typed", self._config)
    # TA8.43 (C1-K), the third entry point. Two of three is the error this
    # programme spent its last third correcting.
    is_admin = _admin_authority("stream_typed", is_admin, agent_depth)
    _refuse_unaddressable_transcript(
        "stream_typed", self._config, scope, session_id
    )
    _refuse_streaming_response_model("stream_typed", response_model)

    token = _active_scope.set(scope)
    run_id_token = _active_run_id.set(str(run_id or "") or None)
    depth_token, snapshot_token = _open_stream_depth(agent_depth)
    try:
        # TA8.41 (C1-L): the same guard, in the same position, saying
        # which field declined and on which surface. It was a bare
        # ``SymfonicAgentError`` whose message named no field, which is why
        # the admission inventory could only record the outcome as
        # ``unknown``. ``StreamingDisabledError`` subclasses that class, so
        # every existing ``except`` and every existing message match still
        # catch it -- a widening, never a rename.
        _refuse_disabled_streaming("stream_typed", self._config)

        # v7.1.1: fail-loud guard mirroring run()/stream() — see the
        # equivalent guard in `run()` for rationale.
        if self._config.ask_user_enabled and (scope is None or not session_id):
            raise SymfonicAgentError(
                "ask_user_enabled=True requires both `scope` and `session_id` "
                "to be supplied on the first turn so a deterministic thread_id "
                "can be derived. Pause-resume cycles cannot bind a token to an "
                "ephemeral LangGraph-generated thread.",
                code="bad_request",
            )

        # TA8.30 (ST3): the typed projection's own cutover dispatch, and
        # the reason it is its own rather than a read of
        # ``invocation.stream``'s. ST1 measured the shared-body option and
        # recorded ``facade_over_stream.viable = false`` -- the typed
        # surface cannot be served by projecting ``stream``, so the two are
        # not one capability, and a shared switch would flip and roll back
        # two public APIs together whether or not their evidence agreed.
        #
        # There is no release-line clause here, deliberately.
        # ``stream``'s dispatch consults
        # ``text_delta_contract_is_published()`` because ``StreamChunk``
        # has a *published* payload shape on 9.x/10.x that option C
        # changes; the typed projection has no such published divergence to
        # schedule, and TA8.17 verifies the historic lines against their own
        # wheels. The compatibility window is kept by not backporting, not
        # by a branch in 11.0 that simulates 10.4.
        from symfonic.agent.cutover import (  # noqa: PLC0415
            INVOCATION_CONTINUATION,
            INVOCATION_STREAM_TYPED,
            Route,
        )
        from symfonic.agent.cutover.pause_admission import (  # noqa: PLC0415
            unredeemable_pause_reason,
        )

        _typed_switched = (
            self._cutover.route_for(INVOCATION_STREAM_TYPED) is Route.KERNEL
        )
        if _typed_switched:
            _typed_verdict = self._cutover_verdict(
                scope=scope,
                session_id=session_id,
                history=history,
                attachments=attachments,
                callbacks=callbacks,
                extra_metadata=extra_metadata,
                state_overrides=state_overrides,
                agent_depth=agent_depth,
            )
            # Two refusals, counted separately because they are different
            # facts. The first is about the *configuration* this turn was
            # called with; the second is about what this agent's
            # composition root wired, which no config field reports --
            # ``_human_capability`` is an attribute seam, deliberately not
            # a construction parameter, so the envelope cannot see it.
            #
            # A fallback is data, never a silent detour: a flipped switch
            # that still runs legacy is counted and attributed, exactly as
            # ``run`` and ``stream`` count theirs.
            _typed_refusal = (
                None
                if _typed_verdict.admitted
                else (_typed_verdict.reason or "outside the migrated envelope")
            )
            if _typed_refusal is None:
                # A pause this build can mint but no public method can
                # redeem would strand the consumer on the flipped route,
                # which is worse than not flipping it for them. The legacy
                # body's pause *is* redeemable through ``resume``, so this
                # deployment keeps the route where its cycle closes. See
                # ``cutover.pause_admission`` for why this is a dispatch
                # decision rather than a note in the changelog.
                #
                # Both halves of the cycle are asked, not one. A wired
                # ``decode_token`` closes it only while
                # ``invocation.continuation`` routes to the kernel; revert
                # that lever and ``resume`` hands the capability envelope
                # this route mints to ``_legacy_continuation_impl``, which
                # reads only the HMAC scheme -- ST3's condition exactly.
                _typed_refusal = unredeemable_pause_reason(
                    self._human_capability(),
                    continuation_served_by_kernel=(
                        self._cutover.route_for(INVOCATION_CONTINUATION)
                        is Route.KERNEL
                    ),
                )
            if _typed_refusal is not None:
                self._cutover.record_fallback(
                    INVOCATION_STREAM_TYPED, _typed_refusal
                )
                _typed_switched = False

        # v7.1.1: open the psycopg checkpointer pool + run idempotent DDL on
        # the first real invocation. No-op for Memory/Sqlite factories.
        await self._ensure_checkpointer_ready()

        # v7.27.0 restart-resume (Q7): replay the persisted transcript tail
        # into an empty working deque on the first turn of a resumed
        # session. Parity with run()/stream() -- the typed streaming path
        # (the adopter's WebSocket endpoint) previously skipped rehydration, so a
        # resumed session streamed with an empty working deque.
        #
        # TA8.30: this stays **outside** the route decision above. ST1's
        # ``working_rehydration`` row is a ``carry``, and ``stream``'s
        # kernel branch returns before its own copy -- so folding the typed
        # flip into ``stream``'s shape would have re-opened, on the flipped
        # route, exactly the omission this row exists to keep closed.
        await self._maybe_rehydrate_working(scope, session_id)

        # TA8.11: the same two effects, through the same derivation site as
        # ``run`` / ``stream``. It kept a third inline copy of the pair,
        # which is what the "one derivation site" claim on
        # ``_apply_scope_effects`` exists to prevent: a later change made at
        # the named site would have skipped the typed WebSocket path
        # silently. Shared by both routes as of TA8.30, in the position the
        # legacy body already ran it.
        #
        # ``top_level_only`` for the reason ``stream`` passes it (issue
        # #63), and this body was reproduced rather than assumed: the
        # handoff had the typed path from inspection only, and
        # ``test_child_stream_typed_leaves_no_identity_node`` failed on the
        # unfixed tree before this line changed.
        await self._apply_scope_effects(
            scope, is_admin=is_admin, top_level_only=True
        )

        run_id = _resolve_run_id(run_id)
        # ``start`` lives inside ``_stream_typed_impl`` because the typed
        # path measures ``duration_ms`` from the first runtime event onward
        # (preserving the v7.0 timing semantics) rather than from
        # ``stream_typed`` entry.

        # Register tenant with metrics collector so on_llm_end can fan
        # usage into the TokenBudgetTracker (budget ceilings + /billing).
        if self._metrics_collector is not None and scope is not None:
            set_tenant = getattr(self._metrics_collector, "set_tenant", None)
            if set_tenant is not None:
                set_tenant(run_id, scope.tenant_id)

        # Track session. TA8.19: one derivation site, shared with the
        # kernel branch above and with ``run``'s two routes.
        session_id = self._resolve_session(scope, session_id, run_id=run_id)

        # Roadmap Item 10 / PR-5b (follow-up #8): open the OTel root span for
        # the duration of stream_typed() so child node/llm/tool spans nest
        # underneath. Same teardown invariants as ``stream`` -- the wrap
        # closes the span exactly once across normal completion, exception,
        # and consumer disconnect (``GeneratorExit``) paths.
        #
        # TA8.40: **the legacy body's root span, and only the legacy
        # body's.** ``run`` and ``stream`` open theirs inside their own
        # fall-through, after the kernel branch has already returned; this
        # one was hoisted above the body selection by TA8.30 and so ran on
        # both. It was invisible while ``otel_enabled`` was refused --
        # ``_typed_switched`` was always ``False`` for an OTEL agent -- and
        # became reachable the moment TA8.40 admitted the field: on the
        # kernel route ``RunSpanTable.open`` opens a root span for the same
        # ``run_id`` with the same tenant, session, query and entry point,
        # so a typed turn exported *two* ``symfonic.run`` spans and every
        # backend counted the run twice. The kernel route's root span
        # belongs to the observability suite the delegate composes, which
        # is what makes the three entry points agree; the legacy route
        # keeps this one, unchanged.
        _otel_cm = (
            contextlib.nullcontext()
            if _typed_switched
            else self._otel_run_span(
                run_id=run_id,
                scope=scope,
                session_id=session_id,
                query=query,
                entry_point="stream_typed",
            )
        )
        _otel_cm.__enter__()
        _otel_exited = False
        # TA8.30: the verdict decides which body serves, and nothing else
        # does. There is deliberately **no** ``try`` around the drain that
        # returns to ``_stream_typed_impl`` on failure: a kernel error the
        # caller never sees would make every later "zero legacy reaches"
        # meaningless, because the probe would find the legacy body
        # unreached on a healthy run and reached on a broken one, and no
        # certificate can tell those two apart. If the typed kernel path
        # raises, the caller gets that exception.
        _typed_body = (
            self._stream_typed_kernel_impl
            if _typed_switched
            else self._stream_typed_impl
        )
        iterator = _typed_body(
            query=query,
            attachments=attachments,
            scope=scope,
            callbacks=callbacks,
            session_id=session_id,
            history=history,
            run_id=run_id,
            state_overrides=state_overrides,
            extra_metadata=extra_metadata,
            agent_depth=agent_depth,
        )
        try:
            async with contextlib.aclosing(iterator):
                async for typed_event in iterator:
                    yield typed_event
        except GeneratorExit:
            # Consumer disconnect: close cleanly (no ERROR status).
            if not _otel_exited:
                _otel_cm.__exit__(None, None, None)
                _otel_exited = True
            raise
        except BaseException:
            import sys as _sys
            if not _otel_exited:
                _otel_cm.__exit__(*_sys.exc_info())
                _otel_exited = True
            raise
        else:
            if not _otel_exited:
                _otel_cm.__exit__(None, None, None)
                _otel_exited = True
        finally:
            # Backstop: sentinel guarantees ``__exit__`` is called exactly
            # once on the ``@contextmanager`` cm, avoiding the
            # ``RuntimeError("generator didn't stop")`` double-close.
            if not _otel_exited:
                _otel_cm.__exit__(None, None, None)
    finally:
        _close_stream_depth(depth_token, snapshot_token)
        with contextlib.suppress(ValueError):
            _active_run_id.reset(run_id_token)
        with contextlib.suppress(ValueError):
            _active_scope.reset(token)

validate_action async

validate_action(action: str, context: dict[str, Any]) -> bool

Check all loaded plugins' guardrails for an action.

Asks every registered plugin's validate_state_transition hook and returns False as soon as one rejects. Plugin errors are non-fatal and default to allowing the action.

The one enforcement point, on both routes (TA8.21). The body used to be an inline loop over self._plugins, which meant a turn served by the kernel had no guardrail path at all -- the plugin's policy was a contribution nothing folded. It now delegates to :func:symfonic.agent.cutover.guardrails.enforce_guardrails over the capability the bundle folds, which is the shape TA8.19 gave session_id: one derivation site, called from the position the legacy body inlined it, rather than two implementations that agree by inspection until one of them changes.

The observable contract is unchanged and that is deliberate:

• deny-wins, and the first refusal short-circuits;
• a plugin with no validate_state_transition contributes no policy at all, rather than an always-allow one. An always-allow stage in a trace is indistinguishable from a stage that examined the action.

One deliberate change of verdict, and it is a security fix. A hook that raises abstains, and an abstention now denies (TA8.21). The inline loop this replaced swallowed the exception and allowed, which is SEC-FCP-4 / TM-17, filed HIGH: a guard whose backend was down was indistinguishable from a guard that approved, and one unhandled exception inside a policy was an agent with no guardrails. The abstention remains a value on the outcome -- "the only guard with an opinion was down" and "nobody objected" are still different readings of a turn -- they simply no longer produce the same answer.

Refused plugins are asked too, deny-only. load_plugin admits a plugin whole or not at all, and a plugin it refused is not registered anywhere. Its veto is not thereby deleted: the quarantine is passed alongside, and it can refuse an action but never approve one. Otherwise a refusal -- a duplicate name, an unbridgeable surface -- would be a way to remove a guardrail, and "no path turns a veto into an allow" is the property this method exists to hold.

One narrowing, named rather than left to be found. The inline loop handed each hook the caller's own context object; a guard now receives a fresh mutable dict taken from :class:~symfonic.capabilities.extensions.values.PolicyRequest's frozen view (capabilities/extensions/bridge.py's legacy_guard_policies). So a plugin that annotated the context inside its guard -- stamping a reason the caller read back afterwards -- no longer reaches the caller's mapping, on either route. That is deliberate and is AS-INT-3's rule: a policy answers a question, it does not edit the question, and a guard handed the live tool arguments could otherwise rewrite the very call it was only allowed to veto. The verdict contract is unchanged; the side-channel is gone.

The plugin list is passed alongside the capability as the floor, not as a second route. A capability that could not be built answers None (see :meth:_extensions_capability), and delegating that to an unconditional allow would mean an agent with plugins loaded enforcing no guardrail at all -- strictly more permissive than the loop this replaced, on both routes at once. enforce_guardrails asks the population directly in that case, reusing the same deny-wins combinator, so a failed build costs the fragments it was carrying and not the vetoes.

Parameters:

Name	Type	Description	Default
action	str	Name of the action/tool being attempted.	required
context	dict[str, Any]	Current execution context (tenant, params, etc.)	required

Returns:

Type	Description
bool	True if all plugins allow the action, False if any block it.

Source code in src/symfonic/agent/engine.py

async def validate_action(
    self, action: str, context: dict[str, Any]
) -> bool:
    """Check all loaded plugins' guardrails for an action.

    Asks every registered plugin's ``validate_state_transition`` hook and
    returns False as soon as one rejects. Plugin errors are non-fatal and
    default to allowing the action.

    **The one enforcement point, on both routes (TA8.21).** The body used
    to be an inline loop over ``self._plugins``, which meant a turn served
    by the kernel had no guardrail path at all -- the plugin's policy was a
    contribution nothing folded. It now delegates to
    :func:`symfonic.agent.cutover.guardrails.enforce_guardrails` over the
    capability the bundle folds, which is the shape TA8.19 gave
    ``session_id``: one derivation site, called from the position the legacy
    body inlined it, rather than two implementations that agree by
    inspection until one of them changes.

    The observable contract is unchanged and that is deliberate:

    * deny-wins, and the first refusal short-circuits;
    * a plugin with no ``validate_state_transition`` contributes no policy
      at all, rather than an always-allow one. An always-allow stage in a
      trace is indistinguishable from a stage that examined the action.

    **One deliberate change of verdict, and it is a security fix.** A hook
    that raises abstains, and an abstention now **denies** (TA8.21). The
    inline loop this replaced swallowed the exception and allowed, which is
    SEC-FCP-4 / TM-17, filed HIGH: a guard whose backend was down was
    indistinguishable from a guard that approved, and one unhandled
    exception inside a policy was an agent with no guardrails. The
    abstention remains a *value* on the outcome -- "the only guard with an
    opinion was down" and "nobody objected" are still different readings of
    a turn -- they simply no longer produce the same answer.

    **Refused plugins are asked too, deny-only.** ``load_plugin`` admits a
    plugin whole or not at all, and a plugin it refused is not registered
    anywhere. Its veto is not thereby deleted: the quarantine is passed
    alongside, and it can refuse an action but never approve one. Otherwise
    a refusal -- a duplicate name, an unbridgeable surface -- would be a way
    to *remove* a guardrail, and "no path turns a veto into an allow" is
    the property this method exists to hold.

    **One narrowing, named rather than left to be found.** The inline loop
    handed each hook the caller's own ``context`` object; a guard now
    receives a fresh mutable ``dict`` taken from
    :class:`~symfonic.capabilities.extensions.values.PolicyRequest`'s frozen
    view (``capabilities/extensions/bridge.py``'s ``legacy_guard_policies``).
    So a plugin that *annotated* the context inside its guard -- stamping a
    reason the caller read back afterwards -- no longer reaches the caller's
    mapping, on either route. That is deliberate and is AS-INT-3's rule:
    a policy answers a question, it does not edit the question, and a guard
    handed the live tool arguments could otherwise rewrite the very call it
    was only allowed to veto. The *verdict* contract is unchanged; the
    side-channel is gone.

    The plugin list is passed alongside the capability as the *floor*, not
    as a second route. A capability that could not be built answers ``None``
    (see :meth:`_extensions_capability`), and delegating that to an
    unconditional allow would mean an agent with plugins loaded enforcing no
    guardrail at all -- strictly more permissive than the loop this
    replaced, on both routes at once. ``enforce_guardrails`` asks the
    population directly in that case, reusing the same deny-wins combinator,
    so a failed build costs the fragments it was carrying and not the vetoes.

    Args:
        action: Name of the action/tool being attempted.
        context: Current execution context (tenant, params, etc.)

    Returns:
        True if all plugins allow the action, False if any block it.
    """
    from symfonic.agent.cutover.guardrails import enforce_guardrails

    return await enforce_guardrails(
        self._extensions_capability(),
        action,
        context,
        plugins=tuple(self._plugins),
        quarantined=tuple(self._refused_plugins),
        warned=self._unfolded_guard_warned,
    )