symfonic.agent.engine

engine

SymfonicAgent -- unified orchestrator wiring core runtime with HMS.

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,
)

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.

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,
) -> 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.
    """
    self._config = config or FrameworkConfig.with_defaults()

    # 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

    # 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).
    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

    # 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 (Jarvio 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,
        )

    # -- 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 (Jarvio 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 (Jarvio'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.
    self._auto_populate_tool_manifest(local_tools)

    # 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,
        )

    # v7.9.0 (Jarvio 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 Jarvio
    # 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.9.3 (Jarvio 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.
    self._agent_config_resolved = agent_config

    # v7.7.4 (Jarvio 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,
        )
    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 (Jarvio 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 (Jarvio 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 (Jarvio 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``. 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,
        ):
            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) -- Jarvio 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

    # -- 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 cortex/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 (Jarvio 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.exporter 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,
    )

config property

config: FrameworkConfig

Access the framework configuration.

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

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,
    )

load_plugin

load_plugin(plugin: Any) -> None

Load a domain plugin into the agent.

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

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().

Parameters:

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

Raises:

Type	Description
SymfonicAgentError	If the plugin provides tools (which cannot be registered after the graph has been compiled). Pass domain tools via SymfonicAgent(tools=[...]).

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

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

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

    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().

    Args:
        plugin: Any object satisfying the BaseDomainPlugin protocol.

    Raises:
        SymfonicAgentError: If the plugin provides tools (which cannot
            be registered after the graph has been compiled). Pass
            domain tools via ``SymfonicAgent(tools=[...])``.
    """
    domain_tools = plugin.get_domain_tools()
    if domain_tools:
        raise SymfonicAgentError(
            f"Plugin '{getattr(plugin, 'name', 'unknown')}' returned "
            f"{len(domain_tools)} tool(s) from get_domain_tools(). "
            "Domain tools must be registered at construction time via "
            "SymfonicAgent(tools=[...]) because the graph is compiled "
            "eagerly and the ToolRegistry is frozen after that point."
        )

    self._plugins.append(plugin)
    self._plugin_section.add_plugin(plugin)
    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.

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.
    """
    from symfonic.core.config import ModelConfig as _MC

    roles = self._config.role_models or {}
    override = roles.get(role)
    if isinstance(override, _MC):
        return override
    return self._config.agent.model

resume async

resume(
    pause_token: str,
    response: AskUserResponse,
    *,
    scope: FrameworkTenantScope,
    callbacks: Sequence[CallbackHandler] | None = None,
) -> AsyncGenerator[StreamEvent, None]

Resume a paused agent execution from an ask_user elicitation.

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	Optional per-invocation callback handlers.	None

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,
) -> AsyncGenerator[StreamEvent, None]:
    """Resume a paused agent execution from an ``ask_user`` elicitation.

    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: Optional per-invocation callback handlers.

    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.agent.middleware.pause_token import PauseToken
    from symfonic.core.contracts.elicitation import PauseTokenClaims, validate_against_request
    from symfonic.core.contracts.types import AskUserAnsweredEvent

    # v7.1.1: open the psycopg checkpointer pool + run idempotent DDL
    # before any checkpointer I/O. This entry point is special: a fresh
    # worker can receive /resume/{token} as its very first request
    # (pause/resume by definition spans worker lifecycles), so unlike
    # run/stream/stream_typed we cannot rely on a prior call having
    # opened the pool. Skipping this hook here would surface as
    # PoolClosed/PoolNotOpen on the load_original_request below.
    await self._ensure_checkpointer_ready()

    # Step ordering is critical for token hygiene:
    # (1) HMAC + expiry + scope validation (stateless)
    # (2) Load original request from checkpoint metadata
    # (3) Validate AskUserResponse against the original request
    # (4) Atomic consume (single-use enforcement)
    # (5) Build ToolMessage and resume graph
    #
    # Consume MUST come AFTER validate so that a malformed payload does
    # not burn the single-use token — the client can correct and retry.

    # 1. Decode and verify token (stateless first pass — HMAC + expiry)
    try:
        claims: PauseTokenClaims = PauseToken.decode(pause_token, self._config)
    except SymfonicAgentError:
        # Follow-up #9: best-effort expire-side OTel close. Decode
        # failed (expired or tampered HMAC); peek the unverified
        # claims to recover the span key, then re-raise.
        self._maybe_expire_interrupt_span(
            pause_token, default_key_attr="tool_call_id",
        )
        raise

    # 2. Scope binding check
    expected_scope_hash = PauseToken.hash_scope(scope)
    if claims.scope_hash != expected_scope_hash:
        raise SymfonicAgentError("Pause token scope mismatch", code="forbidden")

    # 3. Load original request from checkpoint and cross-validate response.
    # This must precede consume() so an invalid payload cannot burn the token.
    original_request = await PauseToken.load_original_request(
        self._checkpointer, claims
    )
    validate_against_request(response, original_request)

    # 4. Consumption check (single-use enforcement) — only after validation passes.
    # v7.1.3 Item 10.2: route through the per-agent manager so two agents
    # in the same process do not share consumption state. The legacy
    # ``PauseToken.consume`` classmethod still works via the process-wide
    # default manager but is now reserved for direct (non-engine) callers.
    if not await self._pause_token_manager.consume(claims.jti, scope):
        raise SymfonicAgentError(
            "Pause token already used or expired", code="gone"
        )

    # Follow-up #9: close the ``symfonic.interrupt.ask_user`` span
    # opened by the matching ``_mint_pause_token`` call. Keyed by
    # tool_call_id (the ask_user contract has no interrupt_id).
    if self._otel_handles is not None:
        try:
            resp_json = response.model_dump_json()
            self._otel_handles.tracer.recorder.resolve_interrupt_emit(
                interrupt_id=claims.tool_call_id,
                response_payload_size_bytes=len(resp_json.encode("utf-8")),
            )
        except Exception:
            logger.debug(
                "OTel resolve_interrupt_emit raised", exc_info=True,
            )

    # 5. Build synthetic ToolMessage
    # v7.1.0: response carries multiple answers. We dump the whole response
    # as the ToolMessage content.
    tool_msg = ToolMessage(
        content=response.model_dump_json(),
        tool_call_id=claims.tool_call_id,
    )

    # 6. Resume stream_events
    run_id = claims.run_id
    session_id = claims.session_id

    # Emit the success event before resuming
    all_labels = []
    for ans in response.answers:
        all_labels.extend(ans.selections)

    has_other = any(ans.other is not None for ans in response.answers)
    time_to_answer = time.time() - PauseToken.jti_to_timestamp(claims.jti)

    yield AskUserAnsweredEvent(
        run_id=run_id,
        session_id=session_id,
        tool_call_id=claims.tool_call_id,
        selected_labels=all_labels,
        has_other=has_other,
        time_to_answer_seconds=time_to_answer,
    )

    # Resume the graph directly via the runtime, bypassing the
    # hydration/budget/session setup in stream_typed().
    # _thread_id and _checkpoint_id are LangGraph configurable keys —
    # they are extracted by AgentRuntime._build_runnable_config() from
    # state_overrides and placed in config["configurable"], NOT injected
    # into the graph state directly.
    #
    # v8.7.1 C3: resume via ``Command(resume=...)`` — NOT a plain input.
    # The elicitation node called ``interrupt()`` and is parked at the
    # checkpoint; only a Command makes ``interrupt()`` RETURN the payload
    # (the node then builds the ToolMessage from it). Previously we
    # re-invoked with a plain "(resume)" query + a messages override,
    # which restarted the node and re-fired the interrupt forever. The
    # payload is the response JSON string; the elicitation node uses it
    # verbatim as the ToolMessage content (see core/nodes/elicitation.py).
    from langgraph.types import Command

    resume_state_overrides: dict[str, Any] = {
        "run_id": run_id,
        "_thread_id": claims.thread_id,
        "_resume_command": Command(resume=tool_msg.content),
    }

    async for event in self._runtime.stream_events(
        "(resume)",
        callbacks=self._with_metrics_callbacks(callbacks),
        **resume_state_overrides,
    ):
        yield event

resume_interrupt async

resume_interrupt(
    pause_token: str,
    response: Any,
    *,
    scope: FrameworkTenantScope,
    callbacks: Sequence[CallbackHandler] | None = None,
) -> 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).

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,
) -> 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).
    """
    from symfonic.agent.middleware.interrupt_minter import (
        load_original_payload,
    )
    from symfonic.agent.middleware.pause_token import PauseToken
    from symfonic.core.contracts.elicitation import PauseTokenClaims
    from symfonic.core.contracts.interrupt import InterruptResolvedEvent

    await self._ensure_checkpointer_ready()

    # 1. Decode + verify (stateless: HMAC + expiry)
    try:
        claims: PauseTokenClaims = PauseToken.decode(pause_token, self._config)
    except SymfonicAgentError:
        # Follow-up #9: close the matching OTel span ERROR/EXPIRED
        # before re-raising. The generic path keys spans by
        # ``interrupt_id``.
        self._maybe_expire_interrupt_span(
            pause_token, default_key_attr="interrupt_id",
        )
        raise

    registration = self._registered_interrupts.get(claims.name)
    if registration is None or registration.built_in:
        raise SymfonicAgentError(
            f"Token references unknown interrupt name {claims.name!r}",
            code="bad_request",
        )

    # 2. Scope binding (or cross-scope opt-in path).
    expected_scope_hash = PauseToken.hash_scope(scope)
    if claims.scope_hash != expected_scope_hash:
        if not registration.cross_scope_allowed:
            raise SymfonicAgentError(
                "Pause token scope mismatch", code="forbidden",
            )
        # Cross-scope redemption: log for audit and continue.
        logger.warning(
            "Cross-scope interrupt redemption permitted "
            "(name=%s interrupt_id=%s expected=%s got=%s)",
            claims.name, claims.interrupt_id,
            claims.scope_hash, expected_scope_hash,
        )

    # 3. Recover the original payload + cross-validate response.
    original_payload = await load_original_payload(
        checkpointer=self._checkpointer, claims=claims,
    )
    if original_payload is None:
        from symfonic.agent.middleware.pause_token import (
            _is_non_persistent_checkpointer,
        )
        if _is_non_persistent_checkpointer(self._checkpointer):
            raise SymfonicAgentError(
                "Checkpoint lost on restart; non-persistent "
                "checkpointers (MemorySaver) are not supported "
                "with experimental_interrupt=True. Configure "
                "AsyncPostgresSaver or AsyncSqliteSaver.",
                code="failed_dependency",
            )
        raise SymfonicAgentError(
            "Original interrupt payload not found or hash "
            "mismatch -- possible tampering",
            code="bad_request",
        )

    if registration.validate_response is not None:
        try:
            payload_obj = registration.payload_schema.model_validate(
                original_payload
            )
            registration.validate_response(response, payload_obj)
        except Exception as exc:
            raise SymfonicAgentError(
                f"Resume response failed validation: {exc}",
                code="bad_request",
            ) from exc

    # 4. Single-use consume AFTER validate (don't burn the token).
    if not await self._pause_token_manager.consume(
        claims.jti, scope, name=claims.name,
    ):
        raise SymfonicAgentError(
            "Pause token already used or expired", code="gone",
        )

    # 5. Emit InterruptResolvedEvent before resuming the graph.
    response_dict = (
        response.model_dump() if hasattr(response, "model_dump")
        else dict(response)
    )

    # Follow-up #9: close ``symfonic.interrupt.<name>`` span here.
    # The mint site keyed it by claims.interrupt_id.
    if self._otel_handles is not None:
        try:
            import json as _json
            resp_size = len(_json.dumps(response_dict, default=str).encode("utf-8"))
            self._otel_handles.tracer.recorder.resolve_interrupt_emit(
                interrupt_id=claims.interrupt_id,
                response_payload_size_bytes=resp_size,
            )
        except Exception:
            logger.debug(
                "OTel resolve_interrupt_emit raised", exc_info=True,
            )
    time_to_resolve = time.time() - PauseToken.jti_to_timestamp(claims.jti)
    yield InterruptResolvedEvent(
        name=claims.name,
        interrupt_id=claims.interrupt_id,
        response=response_dict,
        run_id=claims.run_id,
        session_id=claims.session_id,
        time_to_resolve_seconds=time_to_resolve,
    )

    # 6. Resume the graph. The InterruptNode receives this dict
    # as the return value of LangGraph's ``interrupt()`` and
    # routes the response back into the messages channel.
    resume_state_overrides: dict[str, Any] = {
        "run_id": claims.run_id,
        "_thread_id": claims.thread_id,
        "_interrupt_resume_value": response_dict,
    }
    async for event in self._runtime.stream_events(
        "(resume_interrupt)",
        callbacks=self._with_metrics_callbacks(callbacks),
        **resume_state_overrides,
    ):
        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,
    **state_overrides: Any,
) -> AgentResponse

Execute the agent graph with automatic HMS hydration and consolidation.

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 HumanMessage 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 (LangChain BaseMessage instances) to prepend before the current query. Capped at the most recent max_conversation_messages entries (from AgentConfig) to avoid token explosion.	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).	None
response_model	type[Any] | None	Optional Pydantic BaseModel subclass. When supplied, the agent runs its normal tool loop and then a terminal structured-output pass validates the final answer against this schema; the validated instance is returned on AgentResponse.structured. Raises StructuredOutputError if the provider cannot do structured output or validation fails. See symfonic.agent.structured.	None
**state_overrides	Any	Arbitrary state values to merge into the graph input (e.g. _thread_id, _checkpoint_id for resume).	{}

Returns:

Type	Description
AgentResponse	AgentResponse with the final response and metadata.

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,
    **state_overrides: Any,
) -> AgentResponse:
    """Execute the agent graph with automatic HMS hydration and consolidation.

    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 ``HumanMessage`` 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 (LangChain
            BaseMessage instances) to prepend before the current query.
            Capped at the most recent ``max_conversation_messages`` entries
            (from AgentConfig) to avoid token explosion.
        is_admin: True to bypass budget checks (v7.0.3).
        run_id: Optional run identifier for correlation (v7.1.0).
        response_model: Optional Pydantic ``BaseModel`` subclass. When
            supplied, the agent runs its normal tool loop and then a
            terminal structured-output pass validates the final answer
            against this schema; the validated instance is returned on
            ``AgentResponse.structured``. Raises ``StructuredOutputError``
            if the provider cannot do structured output or validation
            fails. See ``symfonic.agent.structured``.
        **state_overrides: Arbitrary state values to merge into the graph
            input (e.g. ``_thread_id``, ``_checkpoint_id`` for resume).

    Returns:
        AgentResponse with the final response and metadata.
    """
    # v7.1.1: fail-loud guard for the ask_user contract. When
    # ask_user_enabled is True the caller MUST supply both `scope` and
    # `session_id` on the first turn so the engine can derive a
    # deterministic _thread_id (tenant:sub:session) for LangGraph's
    # checkpointer. Without it a pause token cannot be bound to a stable
    # thread, and a subsequent /resume cannot locate the checkpoint.
    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 the first turn of a resumed
    # session.  One-shot per thread_id; no-op when no durable checkpoint
    # exists or the deque is already populated.
    await self._maybe_rehydrate_working(scope, session_id)

    run_id = run_id or uuid.uuid4().hex[:12]
    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)

    # Defence-in-depth: reject over-budget tenants BEFORE we spend
    # tokens on hydration / LLM calls.  Raises SymfonicAgentError.
    await self._enforce_budget(scope, is_admin=is_admin)

    # Auto-seed AGENT_IDENTITY on first interaction per tenant.
    if scope is not None and scope.tenant_id not in self._seeded_tenants:
        await self._ensure_agent_identity(scope)
        self._mark_tenant_seeded(scope.tenant_id)

    # Track session
    if scope is not None:
        session_id = self._session_manager.ensure_session(
            scope.tenant_id, session_id,
        )
        self._session_manager.touch(session_id)
        # Fix #5: associate run_id with the session so metrics aggregate
        # per-session instead of creating an ephemeral per-run bucket.
        set_conv = getattr(self._metrics_collector, "set_conversation", None)
        if set_conv is not None:
            set_conv(run_id, session_id or run_id)

    # Roadmap Item 10 / PR-5b: open the OTEL root span for the run.
    # When otel_enabled=False ``_otel_run_span`` returns a nullcontext
    # so the cold path is byte-for-byte unchanged.
    _otel_cm = self._otel_run_span(
        run_id=run_id,
        scope=scope,
        session_id=session_id,
        query=query,
        entry_point="run",
    )
    _otel_cm.__enter__()
    _otel_exited = False
    try:

        invocation_state: dict[str, Any] = {"run_id": run_id, **state_overrides}
        if scope is not None and session_id:
            invocation_state.setdefault("_thread_id",
                f"{scope.tenant_id}:{scope.sub_tenant_id or '_'}:{session_id}"
            )

        # v7.1: pre-LLM concerns (intent filter + dynamic tool routing)
        # consolidated into one helper so all three entry points
        # (run / stream / stream_typed) share identical behaviour.
        _intent_verdict = await self._prepare_invocation_state(
            query=query,
            callbacks=callbacks,
            run_id=run_id,
            state_overrides=invocation_state,
            scope=scope,
        )

        # Roadmap Item 8 / PR-6 (7.2): one call collapses hydration +
        # prompt build + stratigraphic-metadata pour.
        # v7.7.8 (Jarvio 9th-recurrence): pass ``scope`` unconditionally
        # and gate hydration on the explicit ``auto_hydrate`` flag.
        # Pre-7.7.8 the engine nulled scope when ``auto_hydrate=False``
        # to signal "skip hydration" -- but that ALSO stripped
        # ``tenant_id`` from the prompt template (rendered as
        # ``Tenant: unknown``) and from the plugin-render state dict.
        # Result: the L1 PRE-FLIGHT block silently degraded on every
        # adopter running ``auto_hydrate=False`` (Jarvio's prod
        # config), defeating the v7.7.6/v7.7.7 fixes upstream.
        assembled = await self._context_manager.assemble(
            query=query,
            scope=scope,
            intent_verdict=_intent_verdict,
            auto_hydrate=self._config.auto_hydrate,
        )
        hydrated = assembled.memory_context or HydratedMemory()
        if assembled.memory_context is not None and hydrated.context:
            invocation_state["memory_context"] = hydrated.model_dump()

        # Inject scope (independent of hydration -- ``tenant_id``
        # must reach the LLM even when ``auto_hydrate=False``).
        # T-7.21.6 (Slice B site #3): splat the FULL state-dict so
        # ``namespace`` survives and any new TenantScope field added
        # in v8.0 propagates without an engine edit.  Pre-7.21 the
        # bare-field assignment dropped namespace silently.
        if scope is not None:
            invocation_state.update(scope.to_state_dict())
        # v7.12.0: thread tool-result compaction knobs into state
        # so the react node helper can read them with a single
        # ``state.get("_tool_compaction")`` lookup.  Engine reads
        # from ``self._config`` once per invocation; node reads
        # from state every iteration.  Pattern mirrors how
        # ``state["resolved_tools"]`` and ``state["forced_tool_choice"]``
        # are threaded.
        invocation_state["_tool_compaction"] = {
            "enabled": self._config.tool_result_compaction_enabled,
            "size_chars": self._config.tool_result_compaction_size_chars,
            "keep_last_n": self._config.tool_result_compaction_keep_last_n,
            # v8.6.0 size-tiered large-tier offload (default-OFF).
            "offload_enabled": getattr(
                self._config, "tool_result_offload_enabled", False,
            ),
            "large_offload_threshold_chars": getattr(
                self._config,
                "tool_result_large_offload_threshold_chars",
                60_000,
            ),
            "expected_conversation_depth": getattr(
                self._config, "tool_result_offload_horizon_default", 5,
            ),
            "expected_recall_probability": getattr(
                self._config, "tool_result_offload_recall_probability", 0.5,
            ),
        }
        # v7.24.0 §4: thread ``tools_cache_ttl`` so ``_bind_tools``
        # in the react node post-processes the bound runnable.  None
        # default preserves v7.23 wire shape byte-for-byte.
        invocation_state["_tools_cache_ttl"] = getattr(
            self._config, "tools_cache_ttl", None,
        )
        # v8.6.4 Ask 2(a): thread the opt-in name-only tool-loop cut
        # threshold so ``tools_condition`` can apply an args-ignoring
        # cut.  None (default) -> the edge skips the name-only branch,
        # byte-identical to v8.6.3.
        invocation_state["_tool_loop_name_only_threshold"] = getattr(
            self._config, "tool_loop_name_only_threshold", None,
        )
        # v8.1.0 C4: thread ``tool_call_policies`` so the react node's
        # POST-model redirect seam can evaluate them.  Empty tuple
        # (default) -> the seam short-circuits, byte-identical to v8.0.1.
        invocation_state["_tool_call_policies"] = getattr(
            self._config, "tool_call_policies", (),
        )
        # v7.10: force-resolution is now owned by the react node
        # (lifecycle-correct: runs per iteration so force AND
        # release are symmetric by construction).  The previous
        # v7.8.3-v7.9.6 assemble-time stamp lived here; relocating
        # it to the node also fixes the latent ``stream()`` /
        # ``stream_typed()`` gap where force was never activated
        # on streaming entry points.  The
        # ``ForcedToolChoiceResolver`` adapter is registered on
        # ``self._deps`` at engine __init__ and looked up by the
        # react node via ``deps.get(ForcedToolChoiceResolver)``.
        # ``state["forced_tool_choice"]`` remains as an OPTIONAL
        # adopter override -- the node honours it if set, then
        # applies the v7.9.6 release check as defense-in-depth.

        # System-prompt + stratigraphic state-override propagation.
        system_prompt_tokens = 0
        if assembled.system_prompt:
            invocation_state["custom_system_prompt"] = assembled.system_prompt
            invocation_state.update(assembled.state_overrides)
            system_prompt_tokens = HMSSystemPromptSection.estimate_tokens(
                assembled.system_prompt,
            )

        # Prepend conversation history so the LLM sees prior turns, and/or
        # inject attachments alongside the query as a multimodal content list.
        # runtime._build_input_state creates [HumanMessage(query)] as the
        # messages list; overriding "messages" here gives the react node full
        # context (history + current turn with optional attachments).
        if history or attachments:
            from langchain_core.messages import HumanMessage
            self._maybe_log_history_trim(history)
            trimmed = (
                _pair_aware_history_slice(
                    history, self._config.agent.max_conversation_messages,
                )
                if history
                else []
            )
            content = _build_human_content(
                query, attachments, _detect_provider_family(self._model_provider),
            )
            invocation_state["messages"] = [*trimmed, HumanMessage(content=content)]

        # Execute
        # v7.18.0 (Option B emission-ordering): suppress the
        # runtime success-path ``on_agent_end`` dispatch -- the
        # engine fires it itself AFTER the full transformation
        # chain so the event carries the rewritten content.  The
        # runtime error-path ``on_agent_end`` still fires for
        # BaseException terminators (engine can't intercept those).
        invocation_state["_suppress_agent_end"] = True
        # v8.6.4 Ask 2(b): opt into runtime recursion-exhaustion
        # salvage ONLY when synthesize_on_empty_final is on.  This
        # switches runtime.run to values-streaming so the gathered
        # ToolMessages survive a recursion_limit cut; the engine then
        # synthesizes a completion from them below (same logic as the
        # normal-return empty-final salvage) and emits the
        # recursion_exhausted ReactLoopEndEvent.  When the gate is OFF
        # the flag is absent -> runtime keeps the ainvoke path and
        # re-raises GraphRecursionError (byte-identical default).
        if self._config.synthesize_on_empty_final:
            invocation_state["_recursion_salvage"] = True
        # v7.19.2 Item 2 prep: reset the per-run audit slot so a
        # stale record from a prior turn cannot contaminate the
        # non-compliance WARN below.  Single-flight assumption
        # documented on _record_force_audit.
        self._last_force_audit = None
        result = await self._runtime.run(
            query, callbacks=self._with_metrics_callbacks(callbacks), **invocation_state
        )

        # v7.19.2 Item 2: post-call non-compliance WARN.  Fires
        # when force was configured AND a procedure matched with
        # action_tool=X AND the AIMessage's tool_calls do NOT
        # include X.  The load-bearing ``forced_upstream`` field
        # disambiguates three failure modes (see plan-of-record
        # at .agent/team/plans/2026-06-05-v7.19.2-...md).
        self._maybe_warn_force_non_compliance(result)

        duration_ms = (time.monotonic() - start) * 1000

        # Strip hidden extraction block and parse ops before consolidation
        raw_response = result.get("final_response") or ""

        # v7.17.0 React-loop terminator synthesis fallback.  The
        # salvage path at ``core/nodes/react.py`` already recovered
        # textual content from terminal AIMessages with tool_calls.
        # If salvage produced nothing AND ``synthesize_on_empty_final``
        # is opted-in, issue one follow-up no-tool LLM call to
        # summarise the work.  Scope is the sync ``run()`` path
        # only (the streaming paths are documented not to support
        # this knob because injecting a synchronous LLM call
        # mid-stream is incompatible with the streaming contract).
        # v8.6.4 Ask 2(b): emit a recursion_exhausted signal when the
        # runtime salvaged a recursion_limit cut, so genuine runaways
        # still alarm/observe even though the engine recovers below.
        _recursion_exhausted = bool(result.get("_recursion_exhausted"))
        if _recursion_exhausted:
            logger.warning(
                "recursion_exhausted: LangGraph hit recursion_limit; "
                "salvaging gathered ToolMessages (run_id=%s)", run_id,
            )
            _rec_mgr = self._resolve_callback_manager(callbacks)
            if _rec_mgr is not None and _rec_mgr.has_hook(
                "on_react_loop_end"
            ):
                from langchain_core.messages import (
                    AIMessage as _AIMessage,
                )

                from symfonic.core.contracts.callbacks import (
                    ReactLoopEndEvent as _ReactLoopEndEvent,
                )

                _rec_tool_calls = _collect_tool_calls(result)
                with contextlib.suppress(Exception):
                    await _rec_mgr.on_react_loop_end(
                        _ReactLoopEndEvent(
                            run_id=run_id,
                            iteration_index=sum(
                                1
                                for _m in result.get("messages", [])
                                if isinstance(_m, _AIMessage)
                            ),
                            content_type="NoneType",
                            has_tool_calls=bool(_rec_tool_calls),
                            tool_call_count=len(_rec_tool_calls),
                            termination_reason="recursion_exhausted",
                        )
                    )

        if not raw_response and self._config.synthesize_on_empty_final:
            logger.info(
                "synthesize_on_empty_final: react bailed with empty "
                "final_response after salvage; issuing one no-tool "
                "synthesis call (run_id=%s)",
                run_id,
            )
            raw_response = await self._synthesize_empty_final(
                messages=list(result.get("messages", [])),
                run_id=run_id,
                callback_manager=self._resolve_callback_manager(callbacks),
            )

        final_response, extracted_ops = _parse_and_strip_extraction(raw_response)
        # Strip HMS citation tags (e.g. "[semantic]", "[episodic]") that
        # the prompt instructs the LLM to add but should never reach users.
        final_response = _strip_citation_tags(final_response)
        # v6.1.8 activation_log_mode: remove the footer per mode:
        #   never   -> always strip
        #   complex -> strip unless response has tool calls or length>=400
        #   always  -> leave it alone (default, pre-v6.1.8 behaviour)
        _alm = self._config.activation_log_mode
        if _alm == "never" and final_response or (
            _alm == "complex"
            and final_response
            and not _response_needs_activation_footer(
                final_response, result.get("messages", []),
            )
        ):
            final_response = _strip_activation_footer(final_response)
        run_thinking = _extract_thinking_text(result)

        # v7.0 T11: fabrication check (off / observe / revise / refuse)
        # v7.0 R4: pass the effective mode (auto-downgrade applied at
        # agent-construction when revise was requested without metacog).
        from symfonic.agent.middleware.fabrication_wiring import (
            apply_fabrication_check,
        )

        _fab_config = (
            self._config
            if self._effective_fabrication_mode
            == self._config.fabrication_check_mode
            else self._config.model_copy(
                update={
                    "fabrication_check_mode": self._effective_fabrication_mode,
                },
            )
        )
        # v8.2.0 T1: compute tool calls ONCE and reuse for BOTH the
        # fabrication check AND the metacognition selective gate.  Was
        # previously collected inline here and discarded.
        _tool_calls_this_turn = _collect_tool_calls(result)
        _fab_outcome = await apply_fabrication_check(
            config=_fab_config,
            response_text=final_response,
            tool_calls_this_turn=_tool_calls_this_turn,
            hydrated_context=hydrated.context,
            user_message=query,
            identifier_keys=list(
                getattr(self._config.domain, "identifier_keys", []) or []
            ),
            callbacks=callbacks,
            # v7.0.6: thread the SAME verdict ``run_intent_filter``
            # already produced for the hydration / tool-routing path
            # so the classifier executes ONCE per turn. ``None`` when
            # both intent_filter_mode and tool_routing_mode are
            # ``"off"`` -- preserves v7.0.5 behaviour for adopters
            # who have not opted into intent classification.
            intent_verdict=_intent_verdict,
            # v7.0.7: thread the engine's run_id into the typed
            # FabricationDetectedEvent so SRE dashboards can correlate
            # findings with the rest of the run telemetry.
            run_id=run_id,
        )
        if _fab_outcome.replacement_draft is not None:
            final_response = _fab_outcome.replacement_draft

        # Metacognitive reflection (post-draft, pre-consolidation)
        _reflection_event = None
        if self._metacog is not None and final_response:
            # v7.4.3 Jarvio Ask 5: thread the resolved callback manager
            # so the critic LLM call fires on_llm_end with
            # node_name="metacognition_critic".
            _reflection_event = await self._metacog.evaluate(
                run_id=run_id,
                draft=final_response,
                activation_log=hydrated.activation_log,
                scope=scope,
                callback_manager=self._resolve_callback_manager(callbacks),
                # v8.2.0 selective gate: thread the deterministic
                # pre-critic signals (computed once above) so the smart
                # gate can skip provably-trivial acks without an LLM call.
                tool_calls_this_turn=_tool_calls_this_turn,
                fabrication_findings=_fab_outcome.findings,
                intent_verdict=_intent_verdict,
            )
            if (
                _reflection_event.verdict == "revise"
                and _reflection_event.revised_draft
            ):
                final_response = _reflection_event.revised_draft
            elif (
                _reflection_event.verdict == "refuse"
                and _reflection_event.refusal_message
            ):
                final_response = _reflection_event.refusal_message
        elif (
            _fab_outcome.forced_metacog_verdict == "revise"
            and final_response
        ):
            # v7.0 T11 ``revise`` mode without metacog enabled: keep the
            # replacement path minimal so the behaviour is observable
            # (tests pin this). We tag the response with the findings.
            final_response = (
                f"{final_response}\n\n"
                f"[fabrication findings: "
                f"{len(_fab_outcome.findings)} unresolved identifier(s)]"
            )

        # v7.18.0 (Option B): response-rewriter hook.  Fires AFTER
        # the full transformation chain (salvage -> extraction
        # stripping -> fabrication -> metacog -> revise/refuse) and
        # BEFORE ``on_agent_end``.  The rewriter receives the
        # adopter-visible final_response and may return a modified
        # string (e.g. Markdown -> Slack mrkdwn translation).
        #
        # Composability rule (locked v7.18.0):
        #   salvage -> fabrication -> metacog -> extraction-stripping
        #   -> on_response_render -> on_agent_end.
        #
        # The rewriter runs at most ONCE per turn.  Adopters that
        # don't subscribe see byte-identical behaviour to v7.17.x.
        _callback_mgr = self._resolve_callback_manager(callbacks)
        original_final_response = final_response
        from symfonic.core.contracts.callbacks import (
            AgentEndEvent,
            ResponseRenderEvent,
        )

        # v7.18.1: structural diagnostic for the on_response_render
        # dispatch gate.  Adopters debugging "is the gate reached?"
        # incidents (Jarvio 2026-06-04 followup) enable via
        # ``logging.getLogger("symfonic.agent.engine").setLevel(DEBUG)``.
        # Zero-cost when DEBUG is not enabled.  Surfaces:
        #   - run_id (correlate with the rest of the run telemetry)
        #   - final_response_present (distinguishes "empty response,
        #     short-circuit" from "non-empty, evaluated has_hook")
        #   - mgr_module + mgr_qualname (a forked CallbackManager
        #     shows up immediately as a non-symfonic module)
        #   - has_hook + handler_count (confirms the registration)
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(
                "on_response_render gate: run_id=%s "
                "final_response_present=%s mgr_module=%s "
                "mgr_qualname=%s has_hook=%s handler_count=%d",
                run_id,
                final_response is not None and bool(final_response),
                type(_callback_mgr).__module__,
                type(_callback_mgr).__qualname__,
                _callback_mgr.has_hook("on_response_render"),
                len(getattr(_callback_mgr, "_handlers", [])),
            )

        if (
            final_response
            and _callback_mgr.has_hook("on_response_render")
        ):
            _render_event = ResponseRenderEvent(
                run_id=run_id,
                # iteration_index is unavailable at the engine
                # post-chain layer (lives inside the react loop's
                # local state).  -1 signals "engine-emitted, not
                # react-loop-emitted".  Adopters that need the
                # exact iteration count subscribe to
                # on_react_loop_end (v7.17.0) instead.
                iteration_index=-1,
                model=(
                    getattr(self._config.agent.model, "model_name", "")
                    or str(self._config.agent.model)
                ),
                node_name="react",
                original_content=original_final_response,
            )
            # The state dict handed to the rewriter is the pre-
            # runtime invocation state (which carries adopter
            # overrides like ``page_context`` / ``tenant_id``)
            # overlaid with the post-runtime messages + final
            # response.  ``runtime.run()`` returns a shaped
            # response dict (final_response / messages /
            # node_execution_log) that does NOT preserve adopter
            # state overrides -- handlers conditioning on
            # ``state['page_context']['platform']`` would see
            # nothing without this overlay.  v7.18.0 contract:
            # adopter state IS visible to the rewriter.
            _rewriter_state: dict[str, Any] = dict(invocation_state)
            _rewriter_state.update({
                "messages": result.get("messages", []),
                "final_response": final_response,
            })
            try:
                rewritten = await _callback_mgr.on_response_render(
                    final_response, _rewriter_state, _render_event,
                )
            except Exception:
                logger.exception(
                    "on_response_render: dispatch raised; "
                    "delivering unrewritten content",
                )
                rewritten = final_response
            if isinstance(rewritten, str):
                final_response = rewritten

        # v7.18.0 (Option B): engine-level ``on_agent_end`` dispatch.
        # The runtime success-path was suppressed via
        # ``_suppress_agent_end=True`` -- engine fires the event
        # here so it carries the post-transformation-chain,
        # post-rewriter content the adopter receives.  Runtime
        # still owns the error-path event for BaseException
        # terminators.
        _runtime_duration_ms = result.get(
            "_runtime_duration_ms", duration_ms,
        )
        _runtime_node_count = result.get(
            "_runtime_node_count", 0,
        )
        await _callback_mgr.on_agent_end(
            AgentEndEvent(
                run_id=run_id,
                final_response=final_response or None,
                duration_ms=_runtime_duration_ms,
                node_count=_runtime_node_count,
            )
        )

        # Post-response consolidation (non-blocking)
        if self._config.auto_consolidate and scope is not None:
            self._schedule_background(
                self._consolidate(
                    query, result, scope,
                    extracted_ops=extracted_ops,
                    thinking_text=run_thinking or None,
                    callback_manager=self._resolve_callback_manager(callbacks),
                    extra_metadata=extra_metadata,
                )
            )

        # Quick-nap: lightweight consolidation every N turns
        self._turn_count += 1
        self._maybe_schedule_quick_nap(scope)

        raw_messages = result.get("messages", [])
        serialized_messages = _clean_messages(raw_messages)

        import dataclasses as _dc

        _reflection_dict: dict[str, Any] | None = None
        if _reflection_event is not None:
            if _dc.is_dataclass(_reflection_event) and not isinstance(
                _reflection_event, type
            ):
                _reflection_dict = _dc.asdict(_reflection_event)
            else:
                # Fallback for mocks or non-dataclass objects in tests
                _reflection_dict = {
                    "verdict": getattr(_reflection_event, "verdict", "approve"),
                    "revised_draft": getattr(_reflection_event, "revised_draft", None),
                    "refusal_message": getattr(
                        _reflection_event, "refusal_message", None
                    ),
                }

        _intent_payload = (
            _intent_verdict.to_event_payload()
            if _intent_verdict is not None
            else None
        )
        _fab_findings_payload = (
            [f.to_event_payload() for f in _fab_outcome.findings]
            if _fab_outcome.findings
            else None
        )

        # Structured-output terminal pass. The loop is already complete;
        # we extract a schema-validated instance from the final
        # conversation without perturbing tool behaviour. Fail-loud: the
        # caller opted in by passing response_model.
        _structured: Any | None = None
        if response_model is not None:
            from .structured import extract_structured_output

            _structured = await extract_structured_output(
                self._model_provider.get_chat_model(
                    self.resolve_model_config("action")
                ),
                raw_messages,
                response_model,
            )

        _response = AgentResponse(
            final_response=final_response or None,
            messages=serialized_messages,
            memory_entries_used=hydrated.entries_count,
            system_prompt_tokens=system_prompt_tokens,
            run_id=run_id,
            # session_id is resolved only when scope is provided; echo it
            # back so stateless callers can look up per-session metrics.
            session_id=session_id or "",
            duration_ms=duration_ms,
            extracted_ops=extracted_ops,
            graph_edges=hydrated.graph_edges,
            activation_log=hydrated.activation_log,
            reflection_event=_reflection_dict,
            intent_verdict=_intent_payload,
            fabrication_findings=_fab_findings_payload,
            structured=_structured,
        )
        # Close the run-span on the SUCCESS path *before* returning.
        # This previously lived in an ``else:`` clause -- but a ``return``
        # inside the ``try`` body skips ``else``, so on success the span
        # was never closed when OTEL was enabled: the run-span leaked and
        # its context token was never detached (stream()/stream_typed()
        # already close before yielding -- run() was the lone regression).
        # The sentinel mirrors the streaming idiom and prevents a double
        # ``__exit__`` if this close is what raises.
        if not _otel_exited:
            _otel_cm.__exit__(None, None, None)
            _otel_exited = True
        return _response
    except BaseException:
        import sys as _sys
        # Neither ``nullcontext`` nor OTel's ``start_as_current_span``
        # suppresses exceptions, so the return value of __exit__ is
        # always falsy here. We re-raise unconditionally.
        if not _otel_exited:
            _otel_cm.__exit__(*_sys.exc_info())
            _otel_exited = True
        raise

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,
    **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	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).	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

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,
    **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: 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).
    Yields:
        StreamChunk events with appropriate event_type progression:
        thinking -> acting/text_delta/tool_call/tool_result -> consolidating -> done
    """
    token = _active_scope.set(scope)
    try:
        if not self._config.streaming_enabled:
            raise SymfonicAgentError("Streaming is disabled in configuration")

        # 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.
        await self._enforce_budget(scope, is_admin=is_admin)

        # Auto-seed AGENT_IDENTITY on first interaction per tenant.
        if scope is not None and scope.tenant_id not in self._seeded_tenants:
            await self._ensure_agent_identity(scope)
            self._mark_tenant_seeded(scope.tenant_id)

        run_id = run_id or uuid.uuid4().hex[:12]
        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
        if scope is not None:
            session_id = self._session_manager.ensure_session(
                scope.tenant_id, session_id,
            )
            self._session_manager.touch(session_id)
            # Fix #5: aggregate metrics by session_id rather than run_id.
            set_conv = getattr(self._metrics_collector, "set_conversation", None)
            if set_conv is not None:
                set_conv(run_id, session_id or 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
        try:
            async for chunk in 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,
            ):
                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:
        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.
    """
    async for event in self.stream_typed(
        query,
        attachments=attachments,
        scope=scope,
        callbacks=callbacks,
        session_id=session_id,
        history=history,
    ):
        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,
    **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).	None
**state_overrides	Any	Arbitrary state values to merge into the graph input (e.g. _thread_id, _checkpoint_id for resume).	{}

Yields:

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

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

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,
    **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).
        **state_overrides: Arbitrary state values to merge into the graph
            input (e.g. ``_thread_id``, ``_checkpoint_id`` for resume).

    Yields:
        Typed StreamEvent objects from the EventTranspiler.
    """
    token = _active_scope.set(scope)
    try:
        if not self._config.streaming_enabled:
            raise SymfonicAgentError("Streaming is disabled in configuration")

        # 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",
            )

        # 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
        # (Jarvio's WebSocket endpoint) previously skipped rehydration, so a
        # resumed session streamed with an empty working deque.
        await self._maybe_rehydrate_working(scope, session_id)

        # Defence-in-depth: same breaker as run() / stream().
        await self._enforce_budget(scope, is_admin=is_admin)

        # Auto-seed AGENT_IDENTITY on first interaction per tenant.
        if scope is not None and scope.tenant_id not in self._seeded_tenants:
            await self._ensure_agent_identity(scope)
            self._mark_tenant_seeded(scope.tenant_id)

        run_id = run_id or uuid.uuid4().hex[:12]
        # ``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
        if scope is not None:
            session_id = self._session_manager.ensure_session(
                scope.tenant_id, session_id,
            )
            self._session_manager.touch(session_id)
            # Fix #5: aggregate metrics by session_id rather than run_id.
            set_conv = getattr(self._metrics_collector, "set_conversation", None)
            if set_conv is not None:
                set_conv(run_id, session_id or 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.
        _otel_cm = 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
        try:
            async for typed_event in self._stream_typed_impl(
                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,
            ):
                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:
        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.

Iterates every registered plugin and calls its validate_state_transition hook. Returns False immediately if any plugin rejects the action. Plugin errors are non-fatal and default to allowing the action.

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.

    Iterates every registered plugin and calls its
    ``validate_state_transition`` hook. Returns False immediately if
    any plugin rejects the action. Plugin errors are non-fatal and
    default to allowing the action.

    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.
    """
    for plugin in self._plugins:
        try:
            if hasattr(plugin, "validate_state_transition"):
                allowed = await plugin.validate_state_transition(
                    action, context
                )
                if not allowed:
                    logger.warning(
                        "Plugin '%s' blocked action: %s",
                        getattr(plugin, "name", "unknown"),
                        action,
                    )
                    return False
        except Exception:
            logger.debug(
                "Plugin '%s' validate_state_transition raised an exception",
                getattr(plugin, "name", "unknown"),
                exc_info=True,
            )
    return True