# Synthetic large system prompt — committed seed for fixture 1.
# Composed deterministically from repo content so live capture is
# fully reproducible. Regenerate via tests/_replay/seeds/build_seed.py.
#
# Target: ~25k tokens (≈ 100k characters at 4 chars/token).
# Content: providers.py full + engine.py (lines 1-1200 and 2500-3700).

## Section A: src/symfonic/core/providers.py

"""ModelProvider protocol + concrete implementations.

Per ADR-PFX-003 and TDD SS2.4:
- ModelProvider: protocol for LLM instantiation
- AnthropicProvider: default, installed via symfonic-core[anthropic]
- OpenAIProvider: installed via symfonic-core[openai]
- DeepSeekProvider: installed via symfonic-core[deepseek]
- GoogleProvider: installed via symfonic-core[google]
- OllamaProvider: installed via symfonic-core[ollama]
"""

from __future__ import annotations

import logging
import os
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable

# BaseChatModel is used only as a return-type annotation and we have
# `from __future__ import annotations`, so all hints are stringified.
# Importing it at runtime pulls langchain_core.language_models which
# transitively eager-loads PIL.Image (via chat_template_utils). Keeping
# this under TYPE_CHECKING preserves the lazy-import contract of
# symfonic.agent.attachments.
if TYPE_CHECKING:
    from langchain_core.language_models import BaseChatModel

from .config import ModelConfig

logger = logging.getLogger("symfonic.core.providers")


def _api_key_fingerprint(api_key: str | None) -> str:
    """Return last 8 chars of an API key, or '<none>' if unset."""
    if not api_key:
        return "<none>"
    if len(api_key) < 12:
        return "***"
    return f"…{api_key[-8:]}"


@runtime_checkable
class ModelProvider(Protocol):
    """Protocol for LLM model construction."""

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel: ...

    def supports_thinking(self) -> bool: ...

    def supports_streaming(self) -> bool: ...

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        """v7.8.5: provider-level introspection for the v7.8.3
        ``procedural_force_first_action_tool`` lever.

        Returns ``True`` when the provider can honour
        ``llm.bind_tools(tools, tool_choice=<name>)`` under the given
        ``ModelConfig``.  Returns ``False`` when the provider would
        reject forced tool_choice (Anthropic's API returns ``400`` when
        ``thinking`` is enabled alongside ``tool_choice={type:tool,...}``;
        see Anthropic's published extended-thinking constraint).

        Default implementation returns ``True`` -- providers that have
        a no-force constraint override.  The engine's
        ``_maybe_resolve_forced_tool_choice`` consults this BEFORE
        stamping ``forced_tool_choice`` into the LangGraph state, so
        adopters who configure ``thinking`` see a clean WARN log and
        a forced no-op rather than a runtime 400 from the API
        boundary.

        Implementations MUST be pure functions of the passed
        ``ModelConfig`` -- no I/O, no provider state mutation -- so
        the engine can call this on every turn without cost.

        v7.8.3 architectural-line preservation: the framework
        constrains the choice set when it can; abstains cleanly when
        it cannot.  Auto-disabling ``config.thinking`` would cross the
        line (reasoning is model-owned).
        """
        return True

    # v7.19.2 NOTE: ``refusal_reason(config) -> str | None`` is an
    # OPTIONAL diagnostic companion the engine consults via
    # :func:`getattr` (see ``engine.py::_maybe_resolve_forced_tool_choice``).
    # It is NOT in the Protocol body deliberately so existing third-
    # party providers continue to satisfy the ``runtime_checkable``
    # Protocol without modification.  Providers that override
    # ``supports_forced_tool_choice`` are ENCOURAGED to also implement
    # ``refusal_reason`` so the per-turn refusal WARN names the
    # structural cause; providers that omit it fall through to the
    # engine's generic v7.8.5-contract fallback text.


class AnthropicProvider:
    """Anthropic-specific model construction. Requires symfonic-core[anthropic].

    Reads ``ANTHROPIC_API_KEY`` from the environment on first
    ``get_chat_model()`` call; logs an INFO line with the last 8 chars
    of the key (never the full value) so operators can tell whether
    the running process picked up the expected key.
    """

    #: v7.13.4: self-declare the wire dialect so
    #: :func:`_detect_provider_family` doesn't need to introspect via
    #: ``isinstance()`` (kept as a fallback for forks that subclass).
    #: Adopters who implement :class:`ModelProvider` Protocol directly
    #: (e.g. for per-request key injection) declare the same ClassVar
    #: on their own class with the appropriate value.
    _symfonic_provider_family: ClassVar[str] = "anthropic"

    _logged_init: bool = False

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_anthropic import ChatAnthropic
        except ImportError as e:
            raise ImportError(
                "langchain-anthropic is required for AnthropicProvider. "
                "Install it via: pip install symfonic-core[anthropic]"
            ) from e

        # Log once per provider instance: which key is in use.
        if not self._logged_init:
            logger.info(
                "AnthropicProvider initialized: api_key=%s model=%s",
                _api_key_fingerprint(os.environ.get("ANTHROPIC_API_KEY")),
                config.model_name,
            )
            self._logged_init = True

        kwargs: dict[str, Any] = {
            "model": config.model_name,
            "temperature": config.temperature,
            "max_tokens": config.max_tokens,
            "max_retries": config.max_retries,
        }
        if config.extra_headers:
            kwargs["extra_headers"] = config.extra_headers
        if config.thinking:
            # v7.13.2: pass ``thinking`` as a first-class kwarg.
            # ``langchain-anthropic>=0.3.20`` exposes ``thinking`` as a
            # native ChatAnthropic field; threading it via
            # ``model_kwargs`` triggers ``UserWarning: Parameters
            # {'thinking'} should be specified explicitly``.  Jarvio
            # flagged this informationally on 2026-05-31; pin bump in
            # pyproject.toml guarantees the field is available.
            kwargs["thinking"] = config.thinking
        return ChatAnthropic(**kwargs)

    def supports_thinking(self) -> bool:
        return True

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        """v7.8.5: Anthropic API rejects ``tool_choice={type:tool,
        name:X}`` and ``{type:any}`` when ``thinking`` is enabled.

        Per the published extended-thinking constraint at
        ``platform.claude.com/docs/en/docs/build-with-claude/
        extended-thinking``: when thinking is enabled, only
        ``{type:auto}`` and ``{type:none}`` are accepted; forced
        tool_choice returns ``400``.

        Returns ``False`` when ``config.thinking`` is set, so the
        engine's force lever cleanly abstains instead of crashing the
        API call at runtime.  Returns ``True`` otherwise (the
        unconstrained Anthropic path).
        """
        return not bool(config.thinking)

    def refusal_reason(self, config: ModelConfig) -> str | None:
        """v7.19.2: human-readable reason for the extended-thinking
        refusal, surfaced in the engine's per-turn refusal WARN."""
        if config.thinking:
            return (
                "Anthropic API rejects tool_choice={type:tool,name:X} "
                "when extended thinking is enabled (returns HTTP 400 "
                "per the published constraint)."
            )
        return None

    @staticmethod
    def extract_cache_ttl(input_token_details: Any) -> str | None:
        """v7.20.0 T-7.20.0.6: mine the Anthropic 1h-cache TTL from a
        LangChain ``usage_metadata.input_token_details`` dict.

        Anthropic's API surfaces the prompt-cache write rate via the
        ``cache_creation: {ephemeral_5m_input_tokens, ephemeral_1h_input_tokens}``
        sub-object on the response usage block.  langchain-anthropic
        (>= 0.3.20) flattens that into ``input_token_details`` with the
        same key names — see ``langchain_anthropic.chat_models:2600-2610``.

        When ``ephemeral_1h_input_tokens > 0`` (the 1h cache was used on
        this turn), this returns ``"1h"`` so the cost layer picks the
        ``cache_write_1h`` rate from the pricing row (T-7.20.0.5).
        Otherwise — pure cache_read, only 5m cache, or no cache_creation
        at all — returns ``None`` so the cost layer falls through to
        ``cache_write`` (the 5-min rate, which IS the documented default
        in the registry).

        Provider boundary placement (architect verdict 2026-06-05 §5.c):
        wire-shape translation lives on the provider that owns the wire
        contract.  The framework-level usage extractors in
        ``callbacks/emit.py`` and ``nodes/react.py`` consult this helper
        so they stay provider-agnostic — they pass through the
        ``input_token_details`` dict without coupling themselves to
        Anthropic-specific key names.

        Args:
            input_token_details: The ``input_token_details`` sub-dict of
                a LangChain ``usage_metadata`` payload.  May be ``None``
                (no cache info), an empty dict, or any non-dict value
                (defensive — never raises).

        Returns:
            ``"1h"`` when the 1h cache was used this turn.  ``None``
            otherwise (5m cache only, no cache, or non-dict input).
        """
        if not isinstance(input_token_details, dict):
            return None
        # ``or 0`` guards against the field being present but ``None``
        # (older langchain-anthropic surfaced a None when 1h cache was
        # absent rather than omitting the key).
        ephemeral_1h = input_token_details.get("ephemeral_1h_input_tokens") or 0
        if ephemeral_1h > 0:
            return "1h"
        return None


class OpenAIProvider:
    """OpenAI model construction. Requires langchain-openai.

    Install via: pip install symfonic-core[openai]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "openai"

    def __init__(self, base_url: str | None = None, api_key: str | None = None):
        self._base_url = base_url
        self._api_key = api_key

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_openai import ChatOpenAI
        except ImportError as e:
            raise ImportError(
                "langchain-openai is required for OpenAIProvider. "
                "Install it via: pip install symfonic-core[openai]"
            ) from e

        kwargs: dict[str, Any] = {
            "model": config.model_name,
            "temperature": config.temperature,
            "max_tokens": config.max_tokens,
            "max_retries": config.max_retries,
            # Emit a usage chunk at the end of every SSE stream so
            # input_tokens / output_tokens are non-zero on streamed calls.
            # LangChain defaults stream_usage=False.
            "stream_usage": True,
        }
        if self._base_url:
            kwargs["base_url"] = self._base_url
        if self._api_key:
            kwargs["api_key"] = self._api_key
        return ChatOpenAI(**kwargs)

    def supports_thinking(self) -> bool:
        return False  # Unless o1/o3 models

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: OpenAI honours ``tool_choice`` unconditionally.
        return True


class DeepSeekProvider:
    """DeepSeek via OpenAI-compatible API. Requires langchain-openai.

    Install via: pip install symfonic-core[deepseek]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "openai"

    def __init__(self, api_key: str | None = None):
        self._api_key = api_key

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_openai import ChatOpenAI
        except ImportError as e:
            raise ImportError(
                "langchain-openai is required for DeepSeekProvider. "
                "Install it via: pip install symfonic-core[deepseek]"
            ) from e

        kwargs: dict[str, Any] = {
            "model": config.model_name or "deepseek-chat",
            "temperature": config.temperature,
            "max_tokens": config.max_tokens,
            "base_url": "https://api.deepseek.com/v1",
            # Emit a usage chunk at the end of every SSE stream so
            # input_tokens / output_tokens are non-zero on streamed calls.
            # LangChain defaults stream_usage=False.
            "stream_usage": True,
        }
        if self._api_key:
            kwargs["api_key"] = self._api_key
        return ChatOpenAI(**kwargs)

    def supports_thinking(self) -> bool:
        return True  # DeepSeek-R1 has reasoning

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: DeepSeek goes through the OpenAI-compatible API
        # which honours ``tool_choice`` unconditionally.
        return True


class GoogleProvider:
    """Google Gemini models. Requires langchain-google-genai.

    Install via: pip install symfonic-core[google]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "google"

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_google_genai import ChatGoogleGenerativeAI
        except ImportError as e:
            raise ImportError(
                "langchain-google-genai is required for GoogleProvider. "
                "Install it via: pip install symfonic-core[google]"
            ) from e

        return ChatGoogleGenerativeAI(
            # v7.19.8 (2026-06-08 registry refresh): default flipped from
            # ``gemini-2.0-flash`` (shut down 2026-06-01 at the provider)
            # to ``gemini-2.5-flash``.  Adopters who relied on the old
            # default were already broken at the wire by the provider
            # retirement; new builds get a live default.
            model=config.model_name or "gemini-2.5-flash",
            temperature=config.temperature,
            max_tokens=config.max_tokens,
        )

    def supports_thinking(self) -> bool:
        return True  # Gemini 2.5 has thinking

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: Google Gemini honours ``tool_choice`` (function
        # calling mode) unconditionally.  No known thinking
        # constraint analogous to Anthropic's.
        return True


class OllamaProvider:
    """Local Ollama models. Requires langchain-ollama.

    Install via: pip install symfonic-core[ollama]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "openai"

    def __init__(self, base_url: str = "http://localhost:11434"):
        self._base_url = base_url

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_ollama import ChatOllama
        except ImportError as e:
            raise ImportError(
                "langchain-ollama is required for OllamaProvider. "
                "Install it via: pip install symfonic-core[ollama]"
            ) from e

        return ChatOllama(
            model=config.model_name or "llama3.2",
            base_url=self._base_url,
            temperature=config.temperature,
        )

    def supports_thinking(self) -> bool:
        return False

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: Ollama tool-call support depends on the underlying
        # model.  Default ``True`` -- adopters running tool-bearing
        # local models (e.g. llama3.1) honour ``tool_choice``.
        return True


class MultiProviderRouter:
    """v7.9.1: prefix-based ``ModelProvider`` dispatcher.

    Closes the v7.9.0 adoption gap: ``FrameworkConfig.role_models``
    lets adopters route different roles to different
    ``ModelConfig`` instances, but ``SymfonicAgent`` accepts a
    SINGLE ``model_provider`` -- so a ``ModelConfig(model_name=
    "gpt-4.1-nano")`` route handed to ``AnthropicProvider``
    builds ``ChatAnthropic(model="gpt-4.1-nano")`` and crashes at
    the API boundary with ``404 model_not_found``.

    This router satisfies the ``ModelProvider`` protocol and
    dispatches each call to a backing provider based on the
    ``ModelConfig.model_name`` prefix.  Adopters compose it once
    at agent construction; ``SymfonicAgent`` never sees the
    multi-provider topology.

    Example::

        from symfonic.core.providers import (
            AnthropicProvider, MultiProviderRouter, OpenAIProvider,
        )

        provider = MultiProviderRouter(
            default=AnthropicProvider(),
            routes={
                "gpt-": OpenAIProvider(),
                "o1-": OpenAIProvider(),
                "o3-": OpenAIProvider(),
            },
        )
        agent = SymfonicAgent(
            model_provider=provider,
            config=FrameworkConfig(
                agent=AgentConfig(
                    model=ModelConfig(model_name="claude-opus-4-6"),
                ),
                role_models={
                    "action": ModelConfig(model_name="gpt-4.1-nano"),
                },
            ),
        )

    Routes are matched in insertion order via ``startswith`` (the
    first matching prefix wins).  Unmatched ``model_name`` strings
    fall back to ``default``.  Empty / ``None`` ``model_name``
    falls back to ``default`` (the construction-time provider's
    default model).

    Architectural-line preserved (v7.8.3): the router is wiring,
    not policy.  It exposes the ``ModelProvider`` surface and
    delegates per call; it does not modify ``ModelConfig``
    instances, does not log model selection (adopter telemetry is
    the right place), and does not maintain per-call state.
    """

    def __init__(
        self,
        *,
        default: ModelProvider,
        routes: dict[str, ModelProvider] | None = None,
    ) -> None:
        """Construct the router.

        Args:
            default: The provider used for any ``ModelConfig`` whose
                ``model_name`` does not match a registered prefix
                (or has an empty / ``None`` ``model_name``).
            routes: Mapping from prefix string to ``ModelProvider``.
                Matched in insertion order via ``startswith``.  An
                empty / ``None`` mapping makes the router a
                pass-through to ``default``.
        """
        self._default: ModelProvider = default
        self._routes: dict[str, ModelProvider] = dict(routes or {})
        # v7.11.0 (closed-enumeration audit, Part C): warn adopters
        # when an earlier prefix would shadow a longer one that can
        # never fire.  Example: ``{"gpt-": OpenAIProvider(),
        # "gpt-4.1-": SpecificProvider()}`` -- ``"gpt-"`` matches
        # every gpt-4.1-* model first, so SpecificProvider() is dead
        # code.  Insertion-order semantics are documented but the
        # footgun is silent -- this WARN surfaces it once at
        # construction so adopters can re-order.  Architectural
        # principle: closed-set enumerations are framework
        # boundaries; cross-boundary extension (custom route maps)
        # needs an adopter seam AND honest feedback when the seam is
        # misconfigured.
        self._warn_shadowed_routes()

    def _warn_shadowed_routes(self) -> None:
        """Detect and WARN when a route's prefix is a strict prefix
        of a later-registered route's prefix.  The later route can
        never fire because the earlier one matches first."""
        keys = [k for k in self._routes if k]
        shadowed: list[tuple[str, str]] = []
        for i, earlier in enumerate(keys):
            for later in keys[i + 1:]:
                if later.startswith(earlier) and later != earlier:
                    shadowed.append((earlier, later))
        if shadowed:
            import logging
            _log = logging.getLogger(__name__)
            for earlier, later in shadowed:
                _log.warning(
                    "MultiProviderRouter: route %r is shadowed by "
                    "the earlier-registered route %r and will never "
                    "fire.  Re-register routes longest-prefix-first "
                    "to resolve.",
                    later, earlier,
                )

    def _pick(self, config: ModelConfig) -> ModelProvider:
        """Resolve the backing provider for the given ``config``.

        Pure function of ``config.model_name``; no side effects.
        """
        name = (config.model_name or "")
        for prefix, provider in self._routes.items():
            if prefix and name.startswith(prefix):
                return provider
        return self._default

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        return self._pick(config).get_chat_model(config)

    def supports_thinking(self) -> bool:
        """Delegate to ``default`` -- this method is called without a
        ``config`` per the v7.x protocol, so the safest answer is
        the default provider's capability."""
        return self._default.supports_thinking()

    def supports_streaming(self) -> bool:
        """Delegate to ``default`` for the same reason as
        ``supports_thinking``."""
        return self._default.supports_streaming()

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        """Delegate to the route-matched provider so per-role forcing
        eligibility tracks the right model.  v7.8.5 refuse-to-force
        contract: the engine queries this BEFORE binding tools, so
        the answer must reflect the model that will actually receive
        the call -- not the default."""
        return self._pick(config).supports_forced_tool_choice(config)

    def refusal_reason(self, config: ModelConfig) -> str | None:
        """v7.19.2: delegate refusal-reason diagnostics to the route-
        matched provider so the per-turn WARN names the right cause."""
        picked = self._pick(config)
        getter = getattr(picked, "refusal_reason", None)
        if callable(getter):
            try:
                return getter(config)
            except Exception:
                return None
        return None


## Section B: src/symfonic/agent/engine.py (lines 1-1200)

"""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.
"""

# ruff: noqa: E402  -- engine.py uses intentional late imports throughout
# (hygiene, prompts, jit_manifest) to defer optional-deps until usage.

from __future__ import annotations

import asyncio
import json as _json
import logging
import re as _re
import time
import uuid
from collections.abc import AsyncGenerator, Mapping, Sequence
from contextvars import ContextVar
from typing import TYPE_CHECKING, Any, Literal

_active_scope: ContextVar[FrameworkTenantScope | None] = ContextVar(
    "symfonic_active_scope", default=None,
)

def _with_active_scope(method):
    import functools
    import inspect
    if inspect.isasyncgenfunction(method):
        @functools.wraps(method)
        async def generator_wrapper(self, *args, **kwargs):
            scope = kwargs.get("scope")
            token = _active_scope.set(scope)
            try:
                async for item in method(self, *args, **kwargs):
                    yield item
            finally:
                _active_scope.reset(token)
        return generator_wrapper
    else:
        @functools.wraps(method)
        async def async_wrapper(self, *args, **kwargs):
            scope = kwargs.get("scope")
            token = _active_scope.set(scope)
            try:
                return await method(self, *args, **kwargs)
            finally:
                _active_scope.reset(token)
        return async_wrapper

if TYPE_CHECKING:
    from langchain_core.language_models import BaseChatModel

    from symfonic.core.config import ModelConfig
    from symfonic.core.contracts.elicitation import AskUserResponse
    from symfonic.core.contracts.types import AskUserQuestionEvent

from langchain_core.messages import BaseMessage, ToolMessage

from symfonic.agent.checkpointer import (
    CheckpointerFactory,
    MemoryCheckpointerFactory,
    PostgresCheckpointerFactory,
    SqliteCheckpointerFactory,
)
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.factory import HMSFactory, MemoryOrchestrator
from symfonic.agent.prompts import HMSSystemPromptSection, MemoryExtractionSection
from symfonic.agent.sessions import SessionManager
from symfonic.agent.types import (
    AgentResponse,
    Attachment,
    FrameworkTenantScope,
    HydratedMemory,
    StreamChunk,
    SymfonicAgentError,
)
from symfonic.core.callbacks.protocol import CallbackHandler
from symfonic.core.contracts.types import (
    ActivationEvent,
    ConsolidationScheduledEvent,
    ExtensionEvent,
    MemoryExtractedEvent,
    ResponseCompleteEvent,
    StreamEvent,
    TextDeltaEvent,
    ThinkingDeltaEvent,
)
from symfonic.core.deps import BaseAgentDeps
from symfonic.core.graph import AgentGraph
from symfonic.core.plugins.contribution import (
    render_position_group,
)
from symfonic.core.plugins.section import PluginPromptSection
from symfonic.core.providers import ModelProvider
from symfonic.core.runtime import AgentRuntime
from symfonic.core.streaming.citation_filter import CitationFilter
from symfonic.core.streaming.extraction_filter import ExtractionFilter
from symfonic.core.utils.compression import compress_memory
from symfonic.memory.protocols import EmbeddingProvider, GraphBackend, VectorBackend
from symfonic.memory.types import MemoryLayer

logger = logging.getLogger("symfonic.agent")

# Content part accepted alongside the semantic query string.  Callers may pass
# either an ``Attachment`` wrapper or a raw LangChain-style content-block dict.
ContentPart = Attachment | dict[str, Any]

# _MAX_HISTORY_MESSAGES was removed — use self._config.agent.max_conversation_messages

# Pattern 1: <MEMORY_EXTRACT>...</MEMORY_EXTRACT> or <GRAPH_OPERATIONS>...</GRAPH_OPERATIONS>
_EXTRACT_RE = _re.compile(
    r"\s*<(?:MEMORY_EXTRACT|GRAPH_OPERATIONS)>[\s\S]*?</(?:MEMORY_EXTRACT|GRAPH_OPERATIONS)>\s*",
    _re.DOTALL,
)

# Pattern 2: GRAPH_OPERATIONS without proper closing tag (LLM sometimes omits it)
_GRAPH_OPS_LOOSE_RE = _re.compile(
    r"\s*(?:<GRAPH_OPERATIONS>|GRAPH_OPERATIONS)\s*(?:```(?:json)?\s*)?\{[\s\S]*$",
    _re.DOTALL,
)

# Pattern 3: trailing ```json { ... } ``` code fence
_CODE_FENCE_RE = _re.compile(
    r"\s*```(?:json)?\s*\{[\s\S]*?\}\s*```\s*$",
    _re.DOTALL,
)

# Keys that signal extraction JSON (used in last-resort scan)
_EXTRACTION_KEYS = ("ops", "operations", "type")


# v6.1.9 memory-hygiene credential scrubber (F2 + F3).
#
# Defence-in-depth: the same pattern is applied on both sides of the
# HMS memory boundary so a slip on one side is still caught on the
# other.
#
#   - F3 (WRITE side): ``_consolidate`` drops credential-shaped keys
#     before persisting extracted ``properties`` to the graph.  The
#     extraction prompt (F4) tells the LLM not to emit them; this
#     regex is the enforcement tier.
#   - F2 (READ side): ``_hydrate`` drops credential-shaped keys before
#     formatting them into the system-prompt context string, so
#     pre-existing secrets written under older versions cannot be
#     echoed back into the wire payload.
#
# Key-name match is intentional — value scanning (regex over
# free-form content) is a separate design track that carries a
# false-positive tax.
#
# v6.2 T01: the pattern list is exposed through
# ``FrameworkConfig.credential_patterns``; see
# ``symfonic.agent.hygiene`` for the canonical implementation. The
# two legacy symbols below are kept as back-compat re-exports so
# external callers and the v6.1.9 test module keep working.
from symfonic.agent.hygiene import (  # noqa: E402
    DEFAULT_CREDENTIAL_KEY_PATTERN as _CREDENTIAL_KEY_PATTERN,
)
from symfonic.agent.hygiene import (  # noqa: E402
    scrub_credential_keys as _hygiene_scrub_credential_keys,
)


def _scrub_credential_keys(properties: Any) -> Any:
    """Drop dict keys whose names match the built-in credential pattern.

    Back-compat wrapper around
    :func:`symfonic.agent.hygiene.scrub_credential_keys`; preserves the
    v6.1.9 private-function signature. Callers that want a tunable
    pattern should use the agent-bound ``SymfonicAgent._scrub_props``
    helper or ``hygiene.scrub_credential_keys`` directly.
    """
    return _hygiene_scrub_credential_keys(
        properties, _CREDENTIAL_KEY_PATTERN,
    )


def _extract_json_from_block(raw: str) -> list[dict[str, Any]]:
    """Parse JSON ops from raw extraction block content."""
    # Strip XML-like delimiters
    for tag in ("MEMORY_EXTRACT", "GRAPH_OPERATIONS"):
        raw = raw.replace(f"<{tag}>", "").replace(f"</{tag}>", "")
    # Strip code fences
    raw = _re.sub(r"```(?:json)?", "", raw).strip()
    if not raw:
        return []
    return _parse_ops_from_json(raw)


def _parse_ops_from_json(raw: str) -> list[dict[str, Any]]:
    """Parse a JSON string into a list of memory ops.

    Accepts dicts with "ops" / "operations" keys, or a single op dict
    with a "type" field.  Returns [] on malformed input.
    """
    try:
        parsed = _json.loads(raw)
    except (ValueError, TypeError):
        return []
    if isinstance(parsed, dict):
        if "ops" in parsed:
            return parsed["ops"]
        if "operations" in parsed:
            return parsed["operations"]
        # Single op object (has "type" field)
        if "type" in parsed:
            return [parsed]
    return []


def _last_resort_json_scan(text: str) -> tuple[str, list[dict[str, Any]]]:
    """Find the last JSON object in *text* that looks like extraction ops.

    Scans backwards for ``{`` characters and tries ``json.loads`` from
    each candidate position to end-of-text.  Returns the first match
    that contains an extraction-relevant key, or ``(text, [])``.
    """
    search_start = 0
    candidates: list[int] = []
    while True:
        idx = text.find("{", search_start)
        if idx == -1:
            break
        candidates.append(idx)
        search_start = idx + 1

    # Walk backwards — we want the LAST valid JSON object
    for pos in reversed(candidates):
        fragment = text[pos:].rstrip()
        try:
            parsed = _json.loads(fragment)
        except (ValueError, TypeError):
            continue
        if not isinstance(parsed, dict):
            continue
        if any(k in parsed for k in _EXTRACTION_KEYS):
            ops = _parse_ops_from_json(fragment)
            if ops:
                clean = text[:pos].rstrip()
                return clean, ops
    return text.rstrip(), []


_CITATION_TAG_RE = _re.compile(
    r"\s*\[(?:semantic|episodic|procedural|working|prospective)\]",
    _re.IGNORECASE,
)


def _strip_citation_tags(text: str) -> str:
    """Remove HMS layer citation tags from user-visible text."""
    if not text:
        return text
    return _CITATION_TAG_RE.sub("", text)


# Regex for the "SPREADING ACTIVATION LOG" directive section of
# hms_system.txt. Historically numbered "## 9" up to v6.1.9; v6.1.10
# inserted a new "## 4. TOOL DISCIPLINE" section which shifted this
# one to "## 10". We anchor on the heading title rather than the
# numeric prefix so future re-numberings stay regex-stable.
# Used by activation_log_mode="never" to suppress the instruction to
# emit the footer.
_ACTIVATION_SECTION_RE = _re.compile(
    r"## \d+\. SPREADING ACTIVATION LOG.*?(?=\n## \d+\.|\Z)",
    _re.DOTALL,
)

# Regex matching the footer the LLM emits on the output side. Matches
# both the Spanish template ("Nodos Activados") and a tolerant English
# fallback ("Activated Nodes"), anchored on the preceding "---" rule
# the template prescribes.
_ACTIVATION_FOOTER_RE = _re.compile(
    r"\n*-{3,}\s*\n+"
    r"(?:Nodos Activados|Activated Nodes)[^\n]*"
    r"(?:\n[^\n]*)*"
    r"\s*$",
    _re.IGNORECASE,
)


def _strip_activation_footer(text: str) -> str:
    """Remove the spreading-activation footer from a final response."""
    if not text:
        return text
    return _ACTIVATION_FOOTER_RE.sub("", text).rstrip()


def _response_needs_activation_footer(
    text: str, messages: list[Any] | None,
) -> bool:
    """Complex-mode heuristic: decide whether to retain the footer.

    Keeps the footer when the response contains tool calls or when the
    cleaned body (footer already stripped) exceeds 400 characters.
    Simple greetings and short acknowledgements therefore lose the
    footer, saving output tokens on trivial turns.
    """
    if not text:
        return False
    # Strip footer for length measurement so the heuristic measures the
    # substantive reply rather than the footer itself.
    body = _strip_activation_footer(text)
    if len(body) >= 400:
        return True
    if messages:
        for msg in messages:
            # LangChain messages expose .tool_calls; serialised dicts
            # expose "tool_calls" key.
            tcalls = (
                getattr(msg, "tool_calls", None)
                if not isinstance(msg, dict)
                else msg.get("tool_calls")
            )
            if tcalls:
                return True
    return False


def _is_episodic_entry(entry: Any) -> bool:
    """Return True if ``entry.layer`` identifies the episodic layer.

    Works for both the MemoryLayer enum (layer.value == "episodic")
    and plain string values (MemoryNode layer attribute).
    """
    layer = getattr(entry, "layer", None)
    if layer is None:
        return False
    layer_val = getattr(layer, "value", None) or str(layer)
    return layer_val.lower() == "episodic"


def _cap_episodic_entries(
    entries: list[Any], raw_cap: int,
) -> list[Any]:
    """Keep the most recent ``raw_cap`` episodic entries; preserve others.

    v6.1.8 working-memory knob. Non-episodic entries are left in place;
    episodic entries are ordered by ``created_at`` descending and
    truncated to ``raw_cap``.  Original order within the returned list
    is preserved (i.e. episodic hits appear in the same relative
    positions they occupied in the input list).

    Args:
        entries: Mixed-layer retrieval hits (MemoryEntry or MemoryNode).
        raw_cap: Maximum number of episodic entries to retain.

    Returns:
        New list with episodic entries capped; non-episodic entries
        unchanged.
    """
    if raw_cap < 0:
        return list(entries)

    episodic_indices = [
        (idx, e) for idx, e in enumerate(entries) if _is_episodic_entry(e)
    ]
    if not episodic_indices or len(episodic_indices) <= raw_cap:
        return list(entries)

    # Rank episodic entries by created_at desc (fall back to idx order).
    def _ts(e: Any) -> float:
        ts = getattr(e, "created_at", None)
        if ts is None:
            return 0.0
        try:
            return ts.timestamp()
        except Exception:
            return 0.0

    ranked = sorted(episodic_indices, key=lambda p: _ts(p[1]), reverse=True)
    keep_indices = {idx for idx, _ in ranked[:raw_cap]}

    # Preserve original order while dropping capped-out episodic entries.
    kept: list[Any] = []
    for idx, entry in enumerate(entries):
        if _is_episodic_entry(entry) and idx not in keep_indices:
            continue
        kept.append(entry)
    return kept


# Maximum characters kept from an assistant response when recording a
# turn summary in episodic memory (Finding 2 fix — was 200, raised to 800
# so structured offers / bullet menus at the tail are not truncated away).
_EPISODIC_RESPONSE_SUMMARY_MAX = 800

# Machine-label prefixes that must never reach user-facing output. Legacy
# nodes (seeded before the prose-content fix) stored the persona/profile
# label baked into ``content``; _hydrate strips a leading occurrence so the
# model cannot parrot the token back.
_MEMORY_LABEL_PREFIXES: tuple[str, ...] = ("AGENT_IDENTITY:", "SOUL:")


def _strip_memory_label_prefix(content: str) -> str:
    """Deprecated v6.x helper retained for backwards-compat imports.

    v7.7 moved the canonical implementation to
    :func:`symfonic.memory.render.strip_memory_label_prefix` (the
    underscore was dropped to mark it as a public helper of the new
    render module).  Existing engine callers continue to work via
    this delegation; new code should import from
    ``symfonic.memory.render`` directly.
    """
    from symfonic.memory.render import strip_memory_label_prefix

    return strip_memory_label_prefix(content)


# v8.0: ``_is_skill_aware_entry`` predicate removed (promised v7.8;
# the procedural skill-aware decision moved inline into
# :meth:`ProceduralLayer.render_for_prompt` in v7.7 as the engine's
# render loop became polymorphic dispatch).  The v7.6.4
# ``_SkillRenderInput`` adapter (already moved to
# :mod:`symfonic.memory.layers.procedural.router` in v7.7) was the
# last consumer; with that gone, the predicate became a pure-test-
# only symbol.  The v6.4-era parity guard at
# ``tests/agent/test_hydrate_legacy_parity.py`` is removed alongside.


def _entry_relevance_score(entry: Any) -> float | None:
    """Return an entry's retrieval relevance score, if it carries one.

    Vector-backed layers populate ``MemoryEntry.score`` with a SIMILARITY
    value (higher == better): in-memory cosine similarity, Postgres
    ``1 - (embedding <=> q)``, Chroma ``1.0 - distance``. Keyword-only
    layers leave it ``None`` -- such entries are NOT gated (no signal to
    judge relevance on).
    """
    score = getattr(entry, "score", None)
    if score is None:
        return None
    try:
        return float(score)
    except (TypeError, ValueError):
        return None


def _filter_manifest_by_keywords(
    manifest: list[str],
    trigger_keywords: dict[str, list[str]],
    query: str,
) -> list[str]:
    """Filter tool manifest entries by per-tool keyword gating.

    v6.1.8 JIT-context knob. Preserves pre-v6.1.8 behaviour when
    ``trigger_keywords`` is empty (default) -- every manifest entry is
    returned unchanged.

    For each ``"name: description"`` entry in ``manifest``:

    * If ``name`` has no entry in ``trigger_keywords`` OR the value is
      an empty list, the entry is included (back-compat, opt-in gating).
    * Otherwise the entry is included only when at least one of its
      keywords appears as a case-insensitive substring of ``query``.

    Args:
        manifest: List of ``"tool_name: short description"`` strings.
        trigger_keywords: Optional keyword lists keyed by tool name.
        query: Current user query (empty string disables matching).

    Returns:
        Filtered manifest preserving original order.
    """
    if not manifest:
        return []
    if not trigger_keywords:
        return list(manifest)
    q_lower = (query or "").lower()
    filtered: list[str] = []
    for entry in manifest:
        name = entry.split(":", 1)[0].strip()
        keywords = trigger_keywords.get(name) or []
        if not keywords:
            # No gating configured for this tool -- always include.
            filtered.append(entry)
            continue
        if not q_lower:
            # Tool is gated but the query is empty -- skip.
            continue
        if any(kw and kw.lower() in q_lower for kw in keywords):
            filtered.append(entry)
    return filtered


# v6.2 T04 -- JIT compressed tool-manifest helper.
# Implementation lives in symfonic.agent.prompts.jit_manifest so
# engine.py does not keep growing; this thin wrapper preserves the
# existing private symbol ``_compress_jit_manifest`` for callers that
# import it from engine (tests).
import contextlib

from symfonic.agent.prompts.jit_manifest import (  # noqa: E402
    compress_jit_manifest as _compress_jit_manifest_impl,
)


def _compress_jit_manifest(
    manifest: list[str],
    trigger_keywords: dict[str, list[str]],
    query: str,
    token_budget: int,
) -> str:
    """Thin wrapper binding the agent's filter function into the helper."""
    return _compress_jit_manifest_impl(
        manifest=manifest,
        trigger_keywords=trigger_keywords,
        query=query,
        token_budget=token_budget,
        filter_fn=_filter_manifest_by_keywords,
    )


# Keys in LangGraph state dicts whose values are user-facing text and must
# be scrubbed of citation tags on the non-typed ``stream()`` path.  These
# mirror the fields the UI renders (final_response) or that downstream
# transformers treat as response text (content, text).
_CITATION_SCRUB_KEYS = frozenset({"final_response", "content", "text"})


def _scrub_citation_from_stream_event(
    event: Any, citation_filter: Any,
) -> Any:
    """Scrub citation tags from text fields in a LangGraph stream event.

    The non-typed ``stream()`` yields raw LangGraph state dicts. Any
    user-facing text inside them (``final_response`` / ``content`` /
    ``text``) may contain citation markers emitted by the LLM.  Prior to
    v6.0.2 only the terminal ``done`` chunk stripped them; per-chunk
    ``data:`` events leaked partial tags when chunk boundaries split mid
    tag (e.g. ``[sem`` + ``antic]``).

    ``citation_filter`` is a long-lived :class:`CitationFilter` shared
    across the stream so partial tag state carries between chunks.  It
    strips complete tags in place and buffers any trailing partial tag
    for the next call.

    Return value: the (possibly new) event structure with scrubbed text
    fields.  Scalars and unknown types are returned unchanged.  Returns
    ``None`` if every text-bearing field was fully buffered as a partial
    tag (caller should skip yielding).  Otherwise the event is yielded
    verbatim even if some strings were replaced by empty text -- preserves
    the event shape for downstream consumers.
    """
    if isinstance(event, dict):
        any_emitted = False
        any_text_field = False
        new_dict: dict[str, Any] = {}
        for key, val in event.items():
            if key in _CITATION_SCRUB_KEYS and isinstance(val, str):
                any_text_field = True
                clean = citation_filter.feed(val)
                new_dict[key] = clean
                if clean:
                    any_emitted = True
            else:
                new_dict[key] = _scrub_citation_from_stream_event(
                    val, citation_filter,
                )
        # If the event contained ONLY a single text field and it was
        # entirely buffered, skip yielding -- we'll emit it on the next
        # chunk once the filter decides.  Multi-field events (the common
        # case) keep their shape so state-machine consumers don't break.
        if any_text_field and not any_emitted and len(event) == 1:
            return None
        return new_dict
    if isinstance(event, list):
        return [_scrub_citation_from_stream_event(v, citation_filter) for v in event]
    return event


def _clean_messages(raw_messages: list[Any]) -> list[dict[str, Any]]:
    """Serialize and sanitize messages for the API response.

    Strips MEMORY_EXTRACT blocks and citation tags from message content
    so internal LLM instructions never reach API consumers.
    """
    cleaned: list[dict[str, Any]] = []
    for msg in raw_messages:
        dumped = msg.model_dump() if hasattr(msg, "model_dump") else msg
        if isinstance(dumped, dict) and isinstance(dumped.get("content"), str):
            text, _ = _parse_and_strip_extraction(dumped["content"])
            dumped = {**dumped, "content": _strip_citation_tags(text)}
        cleaned.append(dumped)
    return cleaned


#: v7.13.5 framework invariants on per-row metadata.  Adopter-supplied
#: ``extra_metadata`` values for these keys are dropped with a WARNING
#: log so the framework's downstream consumers (Phase 12 promotion,
#: speaker-discriminator, ``_collect_tool_calls``, audit pipelines) can
#: continue to assume the keys carry framework semantics.
#:
#: Reserved scope is deliberately narrow: ``ttl_hours``, ``tool_calls``,
#: ``importance``, ``created_at``, ``description``, ``category`` etc.
#: are NOT reserved -- adopters may legitimately want to override them
#: per-row (e.g. ``ttl_hours`` on per-conversation memory rows).
_RESERVED_AUTO_TURN_METADATA_KEYS: frozenset[str] = frozenset(
    {"speaker", "turn_id", "source"},
)


def _safe_merge_metadata(
    base: dict[str, Any],
    extra: dict[str, Any] | None,
    reserved: frozenset[str] = _RESERVED_AUTO_TURN_METADATA_KEYS,
    *,
    _collision_seen: set[str] | None = None,
) -> dict[str, Any]:
    """v7.13.5: merge adopter-supplied ``extra`` provenance metadata
    into ``base`` (the framework-owned per-row dict) while protecting
    reserved framework keys.

    On a reserved-key collision, emit ``logging.WARNING`` once per
    consolidation invocation (via the optional ``_collision_seen``
    set the caller threads in to dedupe across multiple write sites)
    and drop the adopter value.  The framework value wins.

    On a non-reserved key collision (rare; adopter intentionally
    overrides a base key the framework added), the adopter value
    wins -- this lets adopters customize, for example, the
    ``importance`` derived in :meth:`_consolidate_impl`.

    Returns a new dict.  Never mutates ``base`` or ``extra``.
    """
    merged: dict[str, Any] = dict(base)
    if not extra:
        return merged
    seen = _collision_seen if _collision_seen is not None else set()
    for key, value in extra.items():
        if key in reserved:
            if key not in seen:
                seen.add(key)
                logger.warning(
                    "v7.13.5 extra_metadata: adopter-supplied key %r "
                    "collides with a framework-reserved key; dropping "
                    "adopter value (framework invariant).  Reserved keys "
                    "are %s.",
                    key, sorted(reserved),
                )
            continue
        merged[key] = value
    return merged


def _pair_aware_history_slice(
    history: list[Any], cap: int,
) -> list[Any]:
    """v7.12.0: cap a conversation-history list to the most recent
    ``cap`` entries WITHOUT cutting between an AIMessage's
    ``tool_use`` block and its matching ``ToolMessage``.

    Background.  Anthropic's API rejects (HTTP 400) any message list
    where an ``AIMessage`` carries a ``tool_use`` block without a
    following ``tool_result`` block with the matching
    ``tool_use_id``.  The pre-v7.12.0 slice ``history[-cap:]`` was
    blunt -- it would happily cut between an AIMessage and its
    ToolMessage, leaving the leading message dangling and producing
    a 400 on the next call.  With v7.12.0 tool-result compaction
    shrinking individual ToolMessages, more messages fit under the
    cap and this latent bug surfaced more often.

    Algorithm.  Walk from the end of ``history`` backwards.  Keep
    messages until the count >= cap.  After hitting cap, continue
    walking back through any AIMessage whose ``tool_calls`` are not
    all paired by ToolMessages already in the retained slice.  This
    guarantees every retained AIMessage's tool_use blocks have a
    matching ToolMessage in the slice.

    Defensive: returns ``history[-cap:]`` unchanged when ``cap >=
    len(history)`` (no slicing needed) and when ``cap <= 0``
    (returns empty list, matching the pre-existing semantics).
    """
    if cap <= 0:
        return []
    # Step 1: start with the naive tail-slice as the seed.  When
    # ``cap >= len(history)`` this is the full list, but we still
    # need to run Step 5 to drop orphaned AIMessages whose matching
    # ToolMessages were upstream-truncated -- the input might already
    # be malformed.
    retained = list(history if cap >= len(history) else history[-cap:])
    retained_ids = {id(m) for m in retained}
    # Step 2: collect the set of tool_call_ids that the retained
    # slice carries as ToolMessages (the "satisfied" set).
    satisfied: set[str] = set()
    for m in retained:
        tcid = getattr(m, "tool_call_id", None)
        if tcid:
            satisfied.add(str(tcid))
    # Step 3: collect the set of tool_call_ids the retained slice
    # REQUIRES via AIMessage.tool_calls but does NOT have ToolMessages
    # for.  These are the dangling refs we need to either (a) extend
    # backwards to pick up their ToolMessages, or (b) drop the
    # AIMessage that requires them.
    def _required_ids(messages_list: list[Any]) -> set[str]:
        needed: set[str] = set()
        for msg in messages_list:
            tool_calls = getattr(msg, "tool_calls", None)
            if not isinstance(tool_calls, list):
                continue
            for tc in tool_calls:
                tid = (
                    tc.get("id") if isinstance(tc, dict)
                    else getattr(tc, "id", None)
                )
                if tid:
                    needed.add(str(tid))
        return needed

    # Step 4: walk further back to pull in matching ToolMessages for
    # any dangling required ids.  Bounded by the size of `history`
    # so it cannot loop forever.
    cut_index = len(history) - cap  # first index NOT yet in retained
    while True:
        required = _required_ids(retained)
        missing = required - satisfied
        if not missing or cut_index <= 0:
            break
        cut_index -= 1
        candidate = history[cut_index]
        # Only pull this message in if it satisfies one of the missing
        # ids; this prevents reverse-walking through irrelevant
        # messages just to satisfy a near-orphan.
        cand_tcid = str(getattr(candidate, "tool_call_id", "") or "")
        if cand_tcid in missing:
            if id(candidate) not in retained_ids:
                retained.insert(0, candidate)
                retained_ids.add(id(candidate))
                satisfied.add(cand_tcid)
        else:
            # Stop the walk-back the moment we hit a message that
            # doesn't satisfy a dangling id -- otherwise we'd grow
            # the slice past the user's intended cap chasing edges.
            break

    # Step 5: drop any AIMessage at the FRONT of the retained slice
    # whose tool_calls still aren't all satisfied (defensive cleanup
    # for the case where the matching ToolMessages weren't in
    # ``history`` at all -- e.g. truncated upstream).
    while retained:
        head = retained[0]
        head_tool_calls = getattr(head, "tool_calls", None)
        if not isinstance(head_tool_calls, list) or not head_tool_calls:
            break
        head_ids = {
            str(tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", ""))
            for tc in head_tool_calls
        }
        head_ids.discard("")
        if head_ids.issubset(satisfied):
            break
        # Head AIMessage requires ToolMessages we don't have; drop it.
        retained.pop(0)
    return retained


def _collect_tool_calls(result: dict[str, Any]) -> list[dict[str, Any]]:
    """Pull tool-call metadata out of a completed LangGraph run.

    Used by the v7.0 T11 fabrication check to ground identifier claims
    against what actually executed this turn.  Returns a list of
    ``{name, args, result}`` dicts; never raises.  Empty list when the
    run did not exercise any tools.
    """
    calls: list[dict[str, Any]] = []
    # Pull from messages: AI messages may carry ``tool_calls``, Tool
    # messages carry the return ``content``.
    msgs = result.get("messages") if isinstance(result, dict) else None
    if not isinstance(msgs, list):
        return calls
    # Map tool_call_id -> tool result content for pairing.
    results_by_id: dict[str, Any] = {}
    for msg in msgs:
        dumped = msg.model_dump() if hasattr(msg, "model_dump") else msg
        if not isinstance(dumped, dict):
            continue
        msg_type = dumped.get("type") or dumped.get("role")
        if msg_type == "tool":
            tid = dumped.get("tool_call_id") or dumped.get("id") or ""
            results_by_id[str(tid)] = dumped.get("content")
    for msg in msgs:
        dumped = msg.model_dump() if hasattr(msg, "model_dump") else msg
        if not isinstance(dumped, dict):
            continue
        tool_calls = dumped.get("tool_calls")
        if not isinstance(tool_calls, list):
            continue
        for tc in tool_calls:
            if not isinstance(tc, dict):
                continue
            tid = str(tc.get("id") or "")
            calls.append(
                {
                    "name": tc.get("name"),
                    "args": tc.get("args"),
                    "result": results_by_id.get(tid),
                }
            )
    return calls


def _flatten_content_blocks(content: Any) -> str:
    """Flatten an Anthropic content-block list to a plain text string.

    Extended thinking returns ``AIMessage.content`` as a list of blocks
    (``thinking``, ``text``, ``redacted_thinking``, ``tool_use``, ...). The
    React node forwards that list verbatim into ``state["final_response"]``
    (see ``core/nodes/react.py:327``), so downstream consumers that assume
    ``str`` (e.g. ``re.Pattern.search``) crash.  Mirrors the block
    discrimination in ``core/streaming/transpiler.py:_handle_content_block_list``
    -- only ``type == "text"`` blocks contribute; ``thinking`` and other
    block types are dropped (the transpiler surfaces them via separate
    ``ThinkingDeltaEvent`` / ``ToolCall*`` events).

    The helper is idempotent on ``str`` input and defensive on unexpected
    shapes -- returns ``""`` for ``None``, dict, int, etc.  This matches
    the ``or ""`` semantics at the original call sites.
    """
    if isinstance(content, str):
        return content
    if not isinstance(content, list):
        return ""
    parts: list[str] = []
    for block in content:
        if isinstance(block, dict):
            btype = block.get("type")
            text = block.get("text", "")
        else:
            btype = getattr(block, "type", None)
            text = getattr(block, "text", "")
        if btype == "text" and text:
            parts.append(str(text))
    return "\n".join(parts)


def _find_final_response(data: dict[str, Any]) -> str:
    """Recursively find 'final_response' in a nested LangGraph state dict.

    v7.4.1: flattens list-shaped ``final_response`` (Anthropic extended
    thinking) via ``_flatten_content_blocks`` so streaming consumers no
    longer silently receive ``""`` when thinking is enabled.
    """
    if not isinstance(data, dict):
        return ""
    final = data.get("final_response")
    if isinstance(final, list):
        flattened = _flatten_content_blocks(final)
        if flattened:
            return flattened
        # Empty flatten (e.g. thinking-only blocks) -- fall through to the
        # nested-dict scan in case a sibling node carries a usable value.
    elif isinstance(final, str):
        return final
    for val in data.values():
        if isinstance(val, dict):
            found = _find_final_response(val)
            if found:
                return found
    return ""


def _extract_thinking_text(data: dict[str, Any]) -> str:
    """Extract accumulated thinking text from a LangGraph state dict.

    Anthropic extended thinking returns AIMessage content as a list of
    blocks, where thinking blocks have type="thinking" and carry the
    reasoning text in a "thinking" key.  This helper walks the messages
    list in *data* (or nested dicts keyed by LangGraph node name) and
    collects all such text.

    Returns up to 500 characters to avoid bloat in working memory.
    """
    parts: list[str] = []
    _collect_thinking_parts(data, parts)
    combined = " ".join(parts).strip()
    return combined[:500] if combined else ""


def _collect_thinking_parts(data: dict[str, Any], parts: list[str]) -> None:
    """Recursively collect thinking text from *data* into *parts*."""
    if not isinstance(data, dict):
        return

    messages = data.get("messages")
    if isinstance(messages, list):
        for msg in messages:
            content = getattr(msg, "content", None) if hasattr(msg, "content") else None
            if not isinstance(content, list):
                continue
            for block in content:
                if isinstance(block, dict):
                    btype = block.get("type")
                    text = block.get("thinking", "")
                else:
                    btype = getattr(block, "type", None)
                    text = getattr(block, "thinking", "")
                if btype == "thinking" and text:
                    parts.append(str(text))

    # Always traverse nested dicts so LangGraph node-keyed results are handled.
    for val in data.values():
        if isinstance(val, dict):
            _collect_thinking_parts(val, parts)


def _parse_and_strip_extraction(text: Any) -> tuple[str, list[dict[str, Any]]]:
    """Strip extraction blocks and parse the JSON ops.

    Tries these strategies in order:
      1. ``<MEMORY_EXTRACT>...</MEMORY_EXTRACT>``
      2. ``<GRAPH_OPERATIONS>...</GRAPH_OPERATIONS>`` (with or without closing tag)
      3. Trailing ````` ```json {...} ``` ````` code fence
      4. Last-resort: find the last bare JSON object containing extraction keys

    v7.4.1: accepts ``Any`` and normalises list-shaped input (Anthropic
    extended-thinking ``AIMessage.content``) at the boundary via
    ``_flatten_content_blocks``. The ``or ""`` guards at the three vulnerable
    call sites (engine.py:1986, 5802, 5832) treat a non-empty list as truthy
    and forwarded it verbatim, raising ``TypeError`` inside ``re.search``.

    Returns:
        Tuple of (clean_text, ops_list).  ops_list is empty when no block
        is present or when the JSON is malformed (graceful degradation).
    """
    if not isinstance(text, str):
        text = _flatten_content_blocks(text)
    # Strategy 1: XML-delimited blocks
    match = _EXTRACT_RE.search(text)
    if match:
        clean = _EXTRACT_RE.sub("", text).rstrip()
        ops = _extract_json_from_block(match.group(0))
        return clean, ops

    # Strategy 2: Loose GRAPH_OPERATIONS without closing tag
    loose_match = _GRAPH_OPS_LOOSE_RE.search(text)
    if loose_match:
        clean = _GRAPH_OPS_LOOSE_RE.sub("", text).rstrip()
        ops = _extract_json_from_block(loose_match.group(0))
        return clean, ops

    # Strategy 3: Trailing code-fenced JSON
    fence_match = _CODE_FENCE_RE.search(text)
    if fence_match:
        clean = _CODE_FENCE_RE.sub("", text).rstrip()
        ops = _extract_json_from_block(fence_match.group(0))
        return clean, ops

    # Strategy 4: Last-resort — find trailing bare JSON with extraction keys
    return _last_resort_json_scan(text)


ProviderFamily = Literal["anthropic", "openai", "google", "unknown"]


def _detect_provider_family(
    model_provider: object,
    resolved_config: Any = None,
) -> ProviderFamily:
    """Classify a ``ModelProvider`` into the content-block dialect it expects.

    v7.23 T-7.23.7 co-fix: when ``resolved_config`` is supplied AND the
    provider is a :class:`MultiProviderRouter`, descend into the route
    map via ``_pick(resolved_config)`` so the family matches the LEAF
    provider that will actually serve the request.  Without this, per-
    turn dispatch (T-7.23.6) surfaces the pre-v7.23 latent misroute --
    the family detector returns the wrapper's ``_default`` leaf even
    when the route map picks a different family.

    Pre-v7.23 callers (``resolved_config=None``) get byte-identical
    behaviour: the wrapper's ``_default`` is introspected via the
    existing recursion path.

    Detection precedence (v7.13.4):

    1. **Duck-typed declaration:** ``_symfonic_provider_family: ClassVar[str]``
       on the provider class.  This is the canonical channel for
       adopter forks that implement :class:`ModelProvider` Protocol
       directly (e.g. for per-request key injection that doesn't fit
       :class:`AnthropicProvider`'s env-read construction model).
       Set ``_symfonic_provider_family: ClassVar[str] = "anthropic"``
       (or ``"openai"`` / ``"google"``) on your provider class.
    2. **``isinstance()`` check:** adopter forks that subclass one of
       :class:`AnthropicProvider` / :class:`OpenAIProvider` /
       :class:`DeepSeekProvider` / :class:`OllamaProvider` /
       :class:`GoogleProvider` classify correctly without an explicit
       declaration.
    3. **Routing-wrapper descent (v7.13.3):** ``_default`` or
       ``default`` attribute pointing at a wrapped provider triggers
       recursion (bounded depth 4 + identity-cycle guard).  Covers
       :class:`MultiProviderRouter` and any future routing wrapper
       that exposes the same attribute name.

    ``OpenAIProvider``, ``DeepSeekProvider`` and ``OllamaProvider`` all
    speak the OpenAI-style ``image_url`` shape, so they collapse to
    the ``openai`` family.  ``AnthropicProvider`` owns the Anthropic
    ``image/source`` shape.  ``GoogleProvider`` needs its own Gemini-
    native encoding (not yet implemented here -- raised at
    ``_to_block`` time).

    Returns ``"unknown"`` only when none of the above match.
    ``"unknown"`` providers are treated as wire-neutral:
    :class:`LastStableTurnBoundaryStrategy` refuses to place a
    ``cache_control`` annotation (preserving the v7.12.1 Qwen wire-
    neutrality contract).  Adopters using truly-custom wire dialects
    (custom Llama / vLLM / Bedrock) keep the byte-identical wire
    payload they currently rely on.

    History:

    * v7.13.2 and earlier: string-match on ``type(provider).__name__ ==
      "AnthropicProvider"``.  Misclassified routing wrappers (Jarvio
      $4.19/turn regression) AND adopter forks that didn't share the
      string name.
    * v7.13.3: added bounded recursion through ``_default`` (fixed
      router case).  Leaf check still string-matched.
    * v7.13.4 (this revision): replaced string-match with
      ``isinstance()`` + duck-typed ``_symfonic_provider_family``
      ClassVar (fixes adopter-fork case).  Architect ADR
      ``a2b049d7dfabeee97``.
    """
    # Local imports avoid circular-import risk: providers.py imports
    # from core.config which transitively touches engine.py.
    from symfonic.core.providers import (
        AnthropicProvider,
        DeepSeekProvider,
        GoogleProvider,
        MultiProviderRouter,
        OllamaProvider,
        OpenAIProvider,
    )

    # v7.23 T-7.23.7 co-fix: when a router wraps providers and the
    # caller passed the resolved ``ModelConfig`` for this turn, descend
    # into the route map so the family matches the LEAF that will serve
    # the request.  Without ``resolved_config`` (pre-v7.23 callers) we
    # fall through to the existing ``_default`` introspection -- byte-
    # identical legacy behaviour preserved.
    if resolved_config is not None and isinstance(
        model_provider, MultiProviderRouter,
    ):
        try:
            leaf = model_provider._pick(resolved_config)
        except Exception:
            leaf = None
        if leaf is not None and leaf is not model_provider:
            return _detect_provider_family(leaf, None)

    seen: set[int] = set()
    cur: Any = model_provider
    for _ in range(4):
        if id(cur) in seen:
            break
        seen.add(id(cur))

        # v7.13.4 Part 1: explicit duck-typed declaration wins.
        # Adopters who implement the Protocol directly (no subclass)
        # can opt their provider into a family with one ClassVar:
        # ``_symfonic_provider_family: ClassVar[str] = "anthropic"``.
        declared = getattr(type(cur), "_symfonic_provider_family", None)
        if declared in ("anthropic", "openai", "google"):
            return declared  # type: ignore[return-value]

        # v7.13.4 Part 2: ``isinstance()`` replaces the pre-v7.13.4
        # string-match.  Adopter forks that subclass canonical
        # providers (the common case: per-request behavior overrides
        # while inheriting the construction shape) classify correctly
        # without an explicit ClassVar.  Resilient against future
        # renames of the canonical class names.
        if isinstance(cur, AnthropicProvider):
            return "anthropic"
        if isinstance(cur, (OpenAIProvider, DeepSeekProvider, OllamaProvider)):
            return "openai"
        if isinstance(cur, GoogleProvider):
            return "google"

        # v7.13.3 routing-wrapper descent (preserved).  ``_default`` /
        # ``default`` duck-type covers :class:`MultiProviderRouter`
        # without coupling the engine to that concrete class.
        inner = getattr(cur, "_default", None) or getattr(cur, "default", None)
        if inner is None or inner is cur:
            break
        cur = inner
    return "unknown"


def _to_block(
    att: ContentPart,
    family: ProviderFamily = "anthropic",
) -> dict[str, Any]:
    """Normalise an ``Attachment`` or raw content-block dict into a LangChain
    content-block ``dict`` in the wire shape expected by the target provider
    *family*.

    Raw dicts pass through unchanged -- callers that hand-craft blocks take
    full responsibility for the per-provider shape.  ``Attachment`` instances
    are serialised per *family*:

    * ``anthropic`` / ``unknown`` -- Anthropic ``image`` / ``document`` shape.
    * ``openai`` -- OpenAI-style ``image_url`` with ``data:<mime>;base64,...``
      URL for base64 payloads.  Document attachments are unsupported on this
      family (OpenAI vision is image-only); a ``NotImplementedError`` is
      raised with a clear redirect.
    * ``google`` -- Gemini requires its own content-part encoding that is
      intentionally out of scope for v1; ``NotImplementedError`` is raised
      so callers see a clear error instead of silently malformed requests.

    Raises:
        TypeError: when *att* is neither a dict nor an ``Attachment``, or
            when an ``Attachment.kind`` is unsupported.
        NotImplementedError: when *family* cannot encode the requested
            attachment kind (e.g. OpenAI + document, Google + any).
    """
    if isinstance(att, dict):
        # Hand-crafted dicts are forwarded verbatim -- we cannot infer the
        # caller's intended family, so responsibility is theirs.
        return att
    if not isinstance(att, Attachment):
        raise TypeError(
            f"attachments[i] must be Attachment or dict, got {type(att).__name__}"
        )

    if family == "google":
        raise NotImplementedError(
            "Google/Gemini attachment encoding is not implemented yet. "
            "Either pass a hand-crafted content-block dict (see "
            "google-genai docs) or use the Anthropic / OpenAI family."
        )

    if family == "openai":
        return _to_openai_block(att)

    # Default: Anthropic (also the "unknown" fallback -- preserves the v1
    # wire shape that existing tests and providers assert on).
    return _to_anthropic_block(att)


def _to_anthropic_block(att: Attachment) -> dict[str, Any]:
    """Serialise an ``Attachment`` into Anthropic's content-block shape."""
    if att.kind == "image":
        if att.source_type == "url":
            return {
                "type": "image",
                "source": {"type": "url", "url": att.data},
            }
        return {
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": att.media_type,
                "data": att.data,
            },
        }
    if att.kind == "document":
        if att.source_type == "url":
            block: dict[str, Any] = {
                "type": "document",
                "source": {"type": "url", "url": att.data},
            }
        else:
            block = {
                "type": "document",
                "source": {
                    "type": "base64",
                    "media_type": att.media_type,
                    "data": att.data,
                },
            }
        if att.filename:
            block["filename"] = att.filename
        return block

    raise TypeError(f"Unsupported attachment kind: {att.kind!r}")


def _to_openai_block(att: Attachment) -> dict[str, Any]:
    """Serialise an ``Attachment`` into OpenAI's content-block shape.

    OpenAI vision expects ``{"type": "image_url", "image_url": {"url": ...}}``
    where the URL is either an HTTPS link or an inline ``data:`` URI for
    base64 payloads.  Document attachments are not supported on this family.
    """
    if att.kind == "image":
        if att.source_type == "url":
            return {
                "type": "image_url",
                "image_url": {"url": att.data},
            }
        # base64 -> inline data URL
        return {
            "type": "image_url",
            "image_url": {
                "url": f"data:{att.media_type};base64,{att.data}",
            },
        }
    if att.kind == "document":
        raise NotImplementedError(
            "OpenAI-family providers do not accept document attachments. "
            "Convert the document to an image (one block per page) or "

## Section C: src/symfonic/agent/engine.py (lines 2500-3700)

                messages=messages,
                scope=scope,
                forced_tool_choice=forced_tc,
                resolved_tool_names=None,
                cache_state=adapter.cache_state_view(),
                run_id=run_id,
            )
            return adapter.resolve(ctx)
        except Exception:
            logger.exception(
                "v7.23 T-7.23.8: fresh resolver call failed; falling "
                "back to static resolve_model_config('action')",
            )
            return self.resolve_model_config("action")

    def _on_llm_response(
        self,
        model_name: str,
        usage: Mapping[str, Any] | Any,
    ) -> None:
        """v7.23 T-7.23.5: stamp ``_cache_state[model_name]`` when a
        response carries ``cache_creation_tokens > 0``.

        Called from the react node after every LLM response so the
        adopter's per-iteration resolver can read the most recent
        cache_creation timestamp via ``DispatchContext.cache_state``.

        Accepts both ``Mapping`` and ``TokenUsage``-shaped objects so
        the caller (react node) can pass whatever shape is convenient
        without coupling to a single type.  Looks for
        ``cache_creation_tokens`` first, then
        ``cache_creation_input_tokens`` (the wire-key spelling).
        """
        if not model_name:
            return
        try:
            if isinstance(usage, Mapping):
                tokens = (
                    usage.get("cache_creation_tokens")
                    or usage.get("cache_creation_input_tokens")
                    or 0
                )
            else:
                tokens = (
                    getattr(usage, "cache_creation_tokens", 0)
                    or getattr(usage, "cache_creation_input_tokens", 0)
                    or 0
                )
        except Exception:
            tokens = 0
        if tokens and int(tokens) > 0:
            from datetime import UTC
            from datetime import datetime as _dt
            self._cache_state[model_name] = _dt.now(UTC)

    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

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

    async def _synthesize_empty_final(
        self,
        *,
        messages: list[Any],
        run_id: str,
        callback_manager: Any | None,
    ) -> str:
        """v7.17.0 React-loop terminator synthesis fallback.

        Issue ONE additional LLM call with a fixed prompt, asking the
        model to summarise the work it just did.  The follow-up call is
        bound WITHOUT tools so the model cannot start a fresh tool-call
        loop -- bound cost is exactly one extra completion per silent-
        bail turn.

        Returns the synthesised text on success, or ``""`` if the call
        fails or returns no usable content.  Never raises -- exceptions
        are logged and the empty string is returned so the caller can
        fall back to its pre-v7.17.0 behaviour (empty ``final_response``).

        Emits ``LLMEndEvent`` with ``node_name="react_synthesize_fallback"``
        so adopters can attribute cost / latency separately from the
        main React loop's emissions.

        Args:
            messages: The conversation history from the LangGraph state
                (``result["messages"]``) -- includes the terminal
                AIMessage that produced the empty final.  Threaded into
                the synthesis prompt verbatim so the model can ground
                the summary in its prior tool-calls / reasoning.
            run_id: The agent run identifier so telemetry correlates
                with the surrounding turn.
            callback_manager: Resolved CallbackManager for LLMEndEvent
                emission.  ``None`` or zero-handler manager fires no
                event (preserves the zero-cost guarantee).

        Returns:
            The synthesised summary text, stripped.  ``""`` when the
            model returns nothing usable or the call raises.
        """
        from langchain_core.messages import HumanMessage

        from symfonic.core.callbacks.emit import (
            emit_llm_end_from_result,
            llm_timing,
            resolve_model_name,
        )

        # Use the action-role model.  The synthesis prompt is a
        # one-paragraph summary request -- no reasoning, no tools, no
        # follow-up turn.  Action-role is the right archetype for the
        # adopter's standard production model (matches what produced
        # the empty-final in the first place, so the summary is
        # consistent with the user-facing voice).
        try:
            chat_model = self.get_chat_model("action")
        except Exception:
            logger.exception(
                "synthesize_on_empty_final: failed to resolve action-role "
                "chat model; degrading to empty final",
            )
            return ""

        synthesis_prompt = (
            "Based on the work you just did in the conversation above, "
            "write a one-paragraph summary for the user. Describe what "
            "you accomplished, what you found, or what conclusion you "
            "reached. Do not call any tools. Do not ask the user a "
            "question. Respond with plain text only."
        )
        synthesis_messages = [
            *messages,
            HumanMessage(content=synthesis_prompt),
        ]

        requested_model = getattr(
            self._config.agent.model, "model_name", "",
        ) or str(self._config.agent.model)

        try:
            async with llm_timing() as timing:
                result = await chat_model.ainvoke(synthesis_messages)
        except Exception:
            logger.exception(
                "synthesize_on_empty_final: chat_model.ainvoke raised; "
                "degrading to empty final",
            )
            return ""

        # Emit LLMEndEvent for cost / latency attribution.  Best-effort:
        # event-emission failure must not mask the synthesised content
        # the caller is waiting for.
        try:
            await emit_llm_end_from_result(
                callback_manager,
                model=resolve_model_name(chat_model, requested_model),
                result=result,
                run_id=run_id,
                node_name="react_synthesize_fallback",
                timing=timing,
            )
        except Exception:
            logger.exception(
                "synthesize_on_empty_final: LLMEndEvent emission failed; "
                "continuing with synthesised content",
            )

        content = getattr(result, "content", "")
        if isinstance(content, list):
            content = _flatten_content_blocks(content)
        if not isinstance(content, str):
            content = ""
        return content.strip()

    async def _maybe_build_preflight_contribs(
        self,
        state: Any,
    ) -> list[Any]:
        """v7.7.5 PRE-FLIGHT L1 injection synthesiser.

        Returns ``[PluginContribution]`` to append to the plugin
        contribution list (which the engine then feeds to
        ``render_position_group``) when ALL of the following hold:

        * ``FrameworkConfig.procedural_render_preflight_in_l1`` is True.
        * The ``AgentGraph`` has an active-skills getter registered
          (the engine wires this in ``__init__`` when either
          ``procedural_enforce_preconditions`` OR
          ``procedural_render_preflight_in_l1`` is True).
        * The active skill snapshot for the current turn yields at
          least one bare-identifier ``(action_tool, precondition_tool)``
          pair.

        Returns ``[]`` (NOT ``None``) when the knob is off or any
        guard fails so the call site's concatenation stays simple.
        Errors degrade silently to ``[]`` -- a broken getter must not
        stall prompt assembly.
        """
        if not getattr(
            self._config, "procedural_render_preflight_in_l1", False,
        ):
            return []
        getter = getattr(self._graph, "_active_skills_getter", None)
        if getter is None:
            return []
        try:
            skills = await getter(state)
        except Exception:
            logger.debug(
                "preflight L1 synthesis: active-skill getter raised; "
                "degrading to no-op",
                exc_info=True,
            )
            return []
        if not skills:
            return []
        try:
            from symfonic.agent.preflight_l1 import (
                build_preflight_contribution,
            )

            # v7.18.0 (Option A): forward state so ``{"state": {...}}``
            # predicates filter skills BEFORE bullet emission.  The L1
            # cached prefix invalidates per platform class only --
            # skills with no state predicates render byte-identical to
            # v7.17.x output for the same snapshot, preserving
            # Anthropic prompt-cache hit rates.
            contrib = build_preflight_contribution(skills, state=state)
        except Exception:
            logger.debug(
                "preflight L1 synthesis: build_preflight_contribution "
                "raised; degrading to no-op",
                exc_info=True,
            )
            return []
        return [contrib] if contrib is not None else []

    async def _maybe_resolve_forced_tool_choice(
        self,
        state: Any,
        messages: list[Any] | None,
    ) -> str | None:
        """v7.8.3 Lever 1: proactively force the first action tool.

        Closes the structural gap that ``procedural_enforce_preconditions``
        (v7.7.4) cannot: the reactive gate at
        ``core/nodes/precondition_gate.py:337`` only fires AFTER the
        model emits a ``tool_use`` block, which is exactly the failure
        mode (7 consecutive zero-tool-call probes under
        claude-opus-4-6) the v7.5.1 -> v7.8.2 chain could not address.

        Returns the bare-identifier tool name when ALL of the
        following hold:

        * ``FrameworkConfig.procedural_force_first_action_tool`` is True.
        * The active-skills getter is registered.
        * The snapshot resolves EXACTLY ONE distinct bare-identifier
          precondition tool name (so the forced choice is
          unambiguous).
        * No prior ``ToolMessage`` in ``messages`` already called that
          tool (idempotency: a turn that already satisfied the
          precondition does not re-force the call).

        Returns ``None`` (no forcing) in every other case.  ``None``
        IS the safe default -- the caller must treat it as "let the
        model choose".  Errors degrade silently.

        Architectural principle (v7.8.3): the framework constrains
        the CHOICE SET deterministically.  The model still authors
        arguments and reasoning.  This helper never returns a tool
        argument payload, only a tool name.

        Allow-list applicability (v7.13.1 doc-debt fix -- pattern
        instance #1 of the model-family abstraction-over-allow-list
        Jarvio surfaced on 2026-05-31):

        This helper consults
        :meth:`AnthropicProvider.supports_forced_tool_choice` (defined
        at ``src/symfonic/core/providers.py:129-144``), which returns
        ``False`` whenever ``ModelConfig.thinking`` is set on an
        Anthropic model.  The introspector is action-role-aware
        (v7.9.5): it consults :meth:`resolve_model_config` with
        ``"action"``, NOT ``self._config.agent.model``, so asymmetric
        routing (default=opus+thinking, action-role=gpt-4.1-nano)
        correctly forces on the action turn even though the default
        would abstain.

        **Applicability matrix:**

        * Anthropic + ``thinking`` ON  -> abstain (provider rejects
          ``tool_choice={type:tool,name:X}`` per the extended-
          thinking constraint, verified at
          ``platform.claude.com/docs/en/docs/build-with-claude/extended-thinking``
          2026-05-29; returning 400 if attempted).
        * Anthropic + ``thinking`` OFF -> force when all other gates
          pass (knob on, single skill, no prior matching ToolMessage).
        * OpenAI                       -> force when all other gates
          pass (OpenAI's ``tool_choice`` accepts ``{type:"function",
          function:{name:...}}`` without a thinking-mode constraint).
        * Other providers              -> defer to the provider's
          :meth:`supports_forced_tool_choice` contract.  Default
          provider Protocol returns ``True`` (opt-out is provider-
          implemented).

        The behavior gates on the PROVIDER'S allow-list, not a
        framework-side string-prefix check.  v7.13.1 governance:
        ``tests/unit/governance/test_allowlist_docstrings.py``
        asserts this docstring names
        ``AnthropicProvider.supports_forced_tool_choice`` explicitly
        so the next reviewer can audit applicability without reading
        the implementation.
        """
        # v7.20.0 T-7.20.0.8: consult the single-source-of-truth
        # ``resolve_force_tool_choice()`` helper on FrameworkConfig.
        # Maps both the new ``procedural_force_tool_choice`` Literal
        # enum and the legacy ``procedural_force_first_action_tool``
        # bool through one method so precedence lives in one place.
        # See FrameworkConfig.resolve_force_tool_choice() for the
        # precedence rule + the v7.19.5 backward-compat mapping.
        resolver_mode = "off"
        resolver = getattr(self._config, "resolve_force_tool_choice", None)
        if callable(resolver):
            try:
                resolver_mode = resolver()
            except Exception:
                resolver_mode = "off"
        else:
            # Defensive: a custom FrameworkConfig subclass that doesn't
            # expose the resolver helper still works via the legacy bool.
            if getattr(
                self._config, "procedural_force_first_action_tool", False,
            ):
                resolver_mode = "soft"
        if resolver_mode == "off":
            return None
        # v7.8.5 (Jarvio thinking+forcing 400 crash fix): consult the
        # provider before resolving.  Anthropic's API rejects
        # ``tool_choice={type:tool,name:X}`` when ``thinking`` is
        # enabled with a ``400`` -- the v8.4 force lever would crash
        # the API call instead of degrading.  We abstain cleanly.
        # Architectural principle (v7.8.3): framework constrains the
        # choice set when it can; abstains cleanly when it cannot.
        # Auto-disabling ``config.thinking`` would cross the line
        # (reasoning is model-owned).
        #
        # v7.19.2 (Jarvio diagnostic gap): the previous one-shot WARN
        # gate (``_force_abstain_logged``) silently buried the refusal
        # after the first turn for the agent's lifetime.  Adopters
        # debugging "HARD lever did not fire" had no per-turn signal
        # for the v7.8.5 refusal.  v7.19.2 emits a per-turn WARNING
        # with structured kwargs naming the provider, action-role
        # model, procedure_id, and the provider's refusal_reason so
        # the diagnostic mirrors the post-call non-compliance WARN at
        # the engine's ``run()`` site.  Item 4 (startup-time WARN at
        # ``__init__``) complements this with a one-time boot-log
        # signal so misconfigurations surface before any turn runs.
        provider = self._model_provider
        if provider is not None:
            supports = getattr(
                provider, "supports_forced_tool_choice", None,
            )
            if callable(supports):
                try:
                    # v7.9.5 (Jarvio asymmetric-routing introspector
                    # bug): consult the model the ACTION role will
                    # actually receive, NOT the default
                    # ``self._config.agent.model``.  Pre-v7.9.5 the
                    # introspector checked the default (opus +
                    # thinking) on every adopter using
                    # ``role_models["action"]`` to route procedural-
                    # rule turns to a non-thinking model -- so
                    # AnthropicProvider.supports_forced_tool_choice
                    # returned False (refuse-to-force on thinking)
                    # and the resolver abstained, even though the
                    # ACTUAL chat model for the action turn was
                    # ``gpt-4.1-nano`` via ``OpenAIProvider`` which
                    # would have honoured ``tool_choice``.  Same
                    # architectural class as v7.7.6 / v7.7.7 / v7.9.3
                    # "introspect the wrong source of truth."  Use
                    # the v7.9.0 resolver at ``engine.py:1796-1819``
                    # which already returns the role-specific
                    # ``ModelConfig``.
                    #
                    # v7.23 T-7.23.8 (R-V7.23-A): per-iteration
                    # synchronisation.  Source ``action_model`` from
                    # the SAME ``_AgentModelResolver`` the react node
                    # consults at react.py:678.  Without this swap,
                    # adopters whose per-iteration resolver picks
                    # Opus+thinking for THIS turn (while static
                    # ``role_models["action"]`` says Sonnet+no-thinking)
                    # see the introspector return True (Sonnet accepts)
                    # while the actual LLM call targets Opus+thinking
                    # and Anthropic 400-crashes.  The v7.8.5 refuse-to-
                    # force-under-thinking contract repeats here.
                    #
                    # Double-call avoidance: the react node stashes the
                    # resolved config on ``state["_resolved_action_model"]``
                    # BEFORE this code runs (in the iteration's
                    # dispatch site).  We read from the stash first;
                    # only when absent (out-of-react direct callers,
                    # tests) do we re-invoke the adapter.  Single
                    # resolve call per iteration -- adopter's
                    # classifier cost doesn't double.
                    action_model = self._resolve_action_model_for_force(
                        state,
                    )
                    if not supports(action_model):
                        # v7.19.2: per-turn refusal WARN.  Reads the
                        # provider's ``refusal_reason`` companion when
                        # available, falls back to the v7.8.5 contract
                        # default text otherwise.  Resolution of the
                        # candidate procedure (procedure_id + action_
                        # tool) happens BELOW the refusal path, so we
                        # surface what we know here and let the post-
                        # call WARN cover the matched-procedure case.
                        candidate_proc_id, candidate_action_tool = (
                            self._peek_forced_candidate(state)
                        )
                        refusal_reason: str | None = None
                        reason_getter = getattr(
                            provider, "refusal_reason", None,
                        )
                        if callable(reason_getter):
                            try:
                                refusal_reason = reason_getter(
                                    action_model,
                                )
                            except Exception:
                                refusal_reason = None
                        if not refusal_reason:
                            refusal_reason = (
                                "provider refused under current "
                                "ModelConfig (v7.8.5 contract)"
                            )
                        logger.warning(
                            "procedural_force_first_action_tool=True "
                            "but provider refused to force "
                            "tool_choice "
                            "(provider=%s, model=%s, "
                            "thinking_enabled=%s, "
                            "procedure_id=%s, action_tool=%s, "
                            "reason=%s, tool_routing_mode=%r)",
                            type(provider).__name__,
                            getattr(action_model, "model_name", "<unknown>"),
                            bool(getattr(action_model, "thinking", None)),
                            candidate_proc_id or "<unresolved>",
                            candidate_action_tool or "<unresolved>",
                            refusal_reason,
                            getattr(
                                self._config, "tool_routing_mode", "off",
                            ),
                        )
                        # v7.19.2 Item 2 bookkeeping: record the
                        # refusal so the engine post-LLM block can
                        # disambiguate forced_upstream=False on the
                        # subsequent non-compliance WARN.
                        self._record_force_audit(
                            procedure_id=candidate_proc_id,
                            action_tool=candidate_action_tool,
                            forced_upstream=False,
                            refusal_reason=refusal_reason,
                        )
                        # v7.20.0 T-7.20.0.8: ``procedural_force_tool_choice="hard"``
                        # opts into strict-fail semantics — provider
                        # refusal raises a RuntimeError so the adopter's
                        # outer handler (compliance / audit boundary)
                        # can fail the request explicitly rather than
                        # accepting a silent abstain.  ``"soft"`` (and
                        # the legacy ``=True`` bool mapping to ``"soft"``)
                        # preserves v7.19.5 abstain-cleanly behaviour
                        # below.
                        if resolver_mode == "hard":
                            provider_name = type(provider).__name__
                            model_name = getattr(
                                action_model, "model_name", "<unknown>",
                            )
                            thinking_enabled = bool(
                                getattr(action_model, "thinking", None),
                            )
                            raise RuntimeError(
                                "procedural_force_tool_choice='hard' but "
                                "provider refused to force tool_choice: "
                                f"provider={provider_name}, "
                                f"model={model_name!r}, "
                                f"thinking_enabled={thinking_enabled}, "
                                f"procedure_id={candidate_proc_id or '<unresolved>'}, "
                                f"action_tool={candidate_action_tool or '<unresolved>'}, "
                                f"reason={refusal_reason}.  Set "
                                "procedural_force_tool_choice='soft' "
                                "to opt back into abstain-cleanly "
                                "semantics (the v7.19.5 default)."
                            )
                        return None
                except RuntimeError:
                    # v7.20.0: hard-mode RAISE must propagate.  The
                    # generic-Exception swallow below catches custom-
                    # provider misbehaviour; RuntimeError from our own
                    # strict-fail branch is intentional.
                    raise
                except Exception:
                    # Defensive: a misbehaving custom provider must
                    # not stall prompt assembly.  Treat the
                    # introspector failure as "abstain conservatively".
                    return None
        getter = getattr(self._graph, "_active_skills_getter", None)
        if getter is None:
            return None
        try:
            skills = await getter(state)
        except Exception:
            logger.debug(
                "forced tool_choice: active-skill getter raised; "
                "degrading to no-op",
                exc_info=True,
            )
            return None
        if not skills:
            return None
        try:
            from symfonic.core.nodes.precondition_gate import (
                _extract_bare_identifier,
                _skill_action_tool,
            )
        except Exception:
            return None

        candidate_pre_tools: set[str] = set()
        # v7.19.2: remember the first matching skill so the audit
        # record (Item 2 post-call WARN) can name procedure_id +
        # action_tool.  When candidate_pre_tools resolves to
        # exactly one element, this records the procedure that
        # contributed it.
        winning_procedure_id: str | None = None
        winning_action_tool: str | None = None
        for skill in skills:
            action_tool = _skill_action_tool(skill)
            if action_tool is None:
                continue
            md = getattr(skill, "metadata", None) or {}
            if not isinstance(md, dict):
                continue
            pre = md.get("precondition")
            contributed = False
            if isinstance(pre, str):
                bare = _extract_bare_identifier(pre)
                if bare:
                    candidate_pre_tools.add(bare)
                    contributed = True
            elif isinstance(pre, list):
                for entry in pre:
                    if isinstance(entry, str):
                        bare = _extract_bare_identifier(entry)
                        if bare:
                            candidate_pre_tools.add(bare)
                            contributed = True
            if contributed and winning_procedure_id is None:
                winning_procedure_id = (
                    str(getattr(skill, "node_id", None) or "")
                    or str(getattr(skill, "id", "") or "")
                )
                winning_action_tool = str(action_tool)
        # Unambiguous-choice contract: only force when exactly one
        # bare-identifier pre_tool resolves across the active-skill
        # set.  Multi-skill ambiguity (different pre_tools per skill)
        # abstains -- the model retains the reasoning role.
        if len(candidate_pre_tools) != 1:
            return None
        pre_tool = next(iter(candidate_pre_tools))

        # Idempotency: if a ``ToolMessage`` for this tool name is
        # already in history, do not re-force.  The precondition is
        # satisfied; further forcing would deny the model agency on
        # the downstream action.
        if messages:
            from langchain_core.messages import ToolMessage

            for msg in messages:
                if isinstance(msg, ToolMessage) and getattr(msg, "name", None) == pre_tool:
                    return None
        # v7.19.2 Item 2 bookkeeping: stamp the audit record before
        # returning the forced tool name so the engine post-LLM block
        # can compare against the first AIMessage's tool_calls and
        # emit the non-compliance WARN with forced_upstream=True.
        self._record_force_audit(
            procedure_id=winning_procedure_id,
            action_tool=winning_action_tool or pre_tool,
            forced_upstream=True,
            refusal_reason=None,
            expected_tool=pre_tool,
        )
        # v7.20.0 T-7.20.0.9: emit ProcedureSelectionEvent so adopter-
        # registered callback handlers observe the procedure-selection
        # decision (billing attribution, A/B-test buckets, audit trail).
        # Zero-cost when no handler subscribes: gated on has_hook so
        # the event object is built only when at least one handler
        # listens.  Fail-soft — a misbehaving handler must not stall
        # the resolver path.
        try:
            from symfonic.core.callbacks.manager import CallbackManager

            mgr = self._deps.get(CallbackManager)
            if mgr is not None and mgr.has_hook("on_procedure_selection"):
                from symfonic.core.contracts.callbacks import (
                    ProcedureSelectionEvent,
                )

                event = ProcedureSelectionEvent(
                    procedure_id=winning_procedure_id,
                    action_tool=winning_action_tool or pre_tool,
                    pre_tool=pre_tool,
                    force_mode=resolver_mode,  # "soft" or "hard" at this point
                    candidate_count=len(candidate_pre_tools),
                )
                await mgr.on_procedure_selection(event)
        except Exception:
            logger.debug(
                "ProcedureSelectionEvent emit failed; continuing",
                exc_info=True,
            )
        return pre_tool

    def _peek_forced_candidate(
        self, state: Any,
    ) -> tuple[str | None, str | None]:
        """v7.19.2 helper: best-effort lookup of the candidate
        procedure_id + action_tool BEFORE the resolver's unambiguous-
        choice gate runs.  Used by the per-turn refusal WARN at
        :meth:`_maybe_resolve_forced_tool_choice` so the log line
        surfaces what the resolver WOULD have forced if the provider
        had not refused.

        Returns ``(None, None)`` when the active-skills getter is
        absent, raises, or yields no skill with a structured
        precondition.  The first matching skill wins (matches the
        resolver's existing iteration order so the two log lines name
        the same procedure).

        Synchronous-readable: the active-skills getter is async, so
        this helper is a no-op when there is no running event loop.
        That keeps the refusal WARN cheap on the hot path; adopters
        who want richer diagnostics can read the post-call WARN at
        the engine's ``run()`` site, which has full message context.
        """
        getter = getattr(self._graph, "_active_skills_getter", None)
        if getter is None:
            return (None, None)
        # The getter is async; we can't await from a sync helper
        # without scheduling.  Trade richness for honesty: return
        # ``<unresolved>`` placeholders and let the post-call WARN
        # carry the structural answer.  Callers that need the names
        # for the refusal WARN itself accept the placeholder.
        return (None, None)

    def _record_force_audit(
        self,
        *,
        procedure_id: str | None,
        action_tool: str | None,
        forced_upstream: bool,
        refusal_reason: str | None,
        expected_tool: str | None = None,
    ) -> None:
        """v7.19.2 helper: stash the most recent force-resolution
        outcome on the agent so the engine post-LLM block can emit
        the Item 2 non-compliance WARN with the load-bearing
        ``forced_upstream`` field.

        Single-slot record (overwritten per call) -- matches the
        single-flight-per-agent assumption that v7.8.5 already made
        with ``_force_abstain_logged``.  Concurrent ``run()`` calls
        on the same agent would race here, but the existing engine
        state has the same limitation; documenting it as a diagnostic
        knob suitable for single-flight investigation.
        """
        self._last_force_audit = {
            "procedure_id": procedure_id,
            "action_tool": action_tool,
            "forced_upstream": forced_upstream,
            "refusal_reason": refusal_reason,
            "expected_tool": expected_tool or action_tool,
        }

    def _maybe_warn_force_non_compliance(self, result: Any) -> None:
        """v7.19.2 Item 2: post-call WARN when the HARD lever did not
        fire.

        Reads the audit slot stamped by
        :meth:`_maybe_resolve_forced_tool_choice` during the react
        loop and compares the recorded expected_tool against the
        first AIMessage's ``tool_calls``.  Emits a structured WARN
        with the load-bearing ``forced_upstream: bool`` field that
        disambiguates three failure modes:

        - forced_upstream=True, actual_tools=[]
            -> model defied the force; Anthropic compliance issue.
        - forced_upstream=False, actual_tools=[]
            -> force never sent (provider refused); see the per-turn
            refusal WARN at _maybe_resolve_forced_tool_choice for
            the structural reason.
        - forced_upstream=True, actual_tools=[other_tool]
            -> model picked the wrong tool despite forcing; provider
            edge case.

        Fail-soft -- a diagnostic must not crash the run path.
        """
        try:
            # v7.20.0 T-7.20.0.8: consult the resolved mode (covers
            # both the new enum and the legacy bool).
            resolver = getattr(self._config, "resolve_force_tool_choice", None)
            mode = "off"
            if callable(resolver):
                try:
                    mode = resolver()
                except Exception:
                    mode = "off"
            elif getattr(
                self._config, "procedural_force_first_action_tool", False,
            ):
                mode = "soft"
            if mode == "off":
                return
            audit = getattr(self, "_last_force_audit", None)
            if not isinstance(audit, dict):
                return
            expected = audit.get("expected_tool") or audit.get(
                "action_tool",
            )
            if not expected:
                return
            messages = result.get("messages") if isinstance(
                result, dict,
            ) else None
            if not messages:
                return
            # First AIMessage with tool_calls (or empty) is the
            # turn-1 response the force lever targets.  Skip ahead
            # of HumanMessage / system seed; pick the first AI
            # message whether it carries tool_calls or not.
            actual_tools: list[str] = []
            found_ai = False
            for msg in messages:
                # Lazy import to avoid a top-level langchain dep at
                # diagnostic-time.
                try:
                    from langchain_core.messages import AIMessage
                except Exception:
                    return
                if isinstance(msg, AIMessage):
                    found_ai = True
                    tc = getattr(msg, "tool_calls", None) or []
                    if isinstance(tc, list):
                        for entry in tc:
                            name = None
                            if isinstance(entry, dict):
                                name = entry.get("name")
                            else:
                                name = getattr(entry, "name", None)
                            if isinstance(name, str) and name:
                                actual_tools.append(name)
                    break
            if not found_ai:
                return
            if expected in actual_tools:
                return  # HARD lever fired as expected; no WARN
            logger.warning(
                "HARD lever did not fire "
                "(procedure_id=%s, expected_tool=%s, "
                "actual_tools=%s, forced_upstream=%s, "
                "tool_routing_mode=%r)",
                audit.get("procedure_id") or "<unresolved>",
                expected,
                actual_tools,
                bool(audit.get("forced_upstream", False)),
                getattr(self._config, "tool_routing_mode", "off"),
            )
        except Exception:
            logger.debug(
                "v7.19.2 Item 2 non-compliance WARN diagnostic "
                "raised; continuing",
                exc_info=True,
            )

    def _warn_silent_advisory_force_lever_at_startup(self) -> None:
        """v7.19.2 Item 4: once-per-agent boot-log WARN when the
        configured action-role model + provider combination guarantees
        the HARD force lever degrades to silently advisory for every
        turn the agent serves.

        Three-condition match:

        1. ``FrameworkConfig.procedural_force_first_action_tool`` is
           ``True`` (the adopter asked for forcing).
        2. The action-role ``ModelConfig`` (resolved via
           ``role_models['action']`` with fallback to
           ``agent.model``) has a truthy ``thinking`` setting.
        3. ``provider.supports_forced_tool_choice(action_model)``
           returns ``False`` (the v7.8.5 refuse-to-force contract).

        When all three hold, the per-turn refusal WARN at
        :meth:`_maybe_resolve_forced_tool_choice` will fire on EVERY
        action turn for the lifetime of this agent.  This boot-log
        WARN tells adopters once, up front, so the misconfiguration
        surfaces before the first production request lands.

        Errors fail-soft -- a diagnostic must not stall agent
        construction.
        """
        try:
            # v7.20.0 T-7.20.0.8: consult the resolved mode.  Fire only
            # when ``"soft"`` is in effect — ``"hard"`` adopters get a
            # RuntimeError at first-turn instead of a silent advisory,
            # so the startup warning is redundant.  ``"off"`` short-
            # circuits.
            resolver = getattr(self._config, "resolve_force_tool_choice", None)
            mode = "off"
            if callable(resolver):
                try:
                    mode = resolver()
                except Exception:
                    mode = "off"
            elif getattr(
                self._config, "procedural_force_first_action_tool", False,
            ):
                mode = "soft"
            if mode != "soft":
                return
            action_model = self.resolve_model_config("action")
            if not getattr(action_model, "thinking", None):
                return
            provider = self._model_provider
            if provider is None:
                return
            supports = getattr(
                provider, "supports_forced_tool_choice", None,
            )
            if not callable(supports):
                return
            if supports(action_model):
                return
            logger.warning(
                "procedural_force_tool_choice='soft' (or legacy "
                "procedural_force_first_action_tool=True) but the "
                "configured action model has thinking enabled and "
                "the provider refuses tool_choice forcing under "
                "thinking.  The HARD lever will be SILENTLY "
                "ADVISORY for the lifetime of this agent.  See "
                "docs/concepts/hard-lever-tools.md (v7.20) or "
                "disable thinking on action turns to enforce the "
                "lever.  (provider=%s, model=%s)",
                type(provider).__name__,
                getattr(action_model, "model_name", "<unknown>"),
            )
        except Exception:
            # Defensive: the startup WARN MUST NOT stall agent
            # construction.  A misbehaving custom provider's
            # introspector should not block the boot path.
            logger.debug(
                "v7.19.2 Item 4 startup-WARN diagnostic raised; "
                "continuing construction",
                exc_info=True,
            )

    def _effective_context_strategy(self) -> str:
        """Return ``"jit"`` or ``"stratified"`` based on the resolved
        strategy.

        v7.7 helper.  The strategy is configured via
        ``FrameworkConfig.context_strategy`` (preferred, explicit) or
        inferred from the legacy ``jit_context`` flag.  When neither is
        set, ``"stratified"`` is the v7.x default (matches the engine's
        actual hydration path).
        """
        explicit = getattr(self._config, "context_strategy", None)
        if explicit in ("jit", "stratified"):
            return str(explicit)
        if getattr(self._config, "jit_context", False):
            return "jit"
        return "stratified"

    def _resolve_hms_budget(self) -> int:
        """Resolve the HMS system-prompt token budget for this turn.

        v7.7.  When ``FrameworkConfig.hms_system_prompt_token_budget``
        is set explicitly, return it.  Otherwise pick a per-strategy
        default that matches the rendered surface:

        * ``jit`` -> 1500 (minimal JIT system prompt; matches the
          pre-7.7 stale log literal Jarvio observed and now makes it
          the actual cap so warnings are accurate)
        * ``stratified`` -> 5500 (full HMS render with extraction
          cached prefix; the pre-7.7 hardcoded 3000 cap triggered
          every turn at the observed 4900-5200 token range)
        """
        explicit = self._config.hms_system_prompt_token_budget
        if explicit is not None:
            return int(explicit)
        return 5500 if self._effective_context_strategy() == "stratified" else 1500

    def _scrub_props(self, properties: Any) -> Any:
        """Apply the agent's configured credential-key scrubber.

        v6.2 T01: wraps :func:`symfonic.agent.hygiene.scrub_credential_keys`
        with the pattern compiled from ``self._config.credential_patterns``.
        When ``credential_patterns=[]`` the pattern is ``None`` and this
        helper returns the input unchanged (explicit opt-out).
        """
        return _hygiene_scrub_credential_keys(
            properties, self._credential_pattern,
        )

    def _otel_run_span(
        self,
        *,
        run_id: str,
        scope: FrameworkTenantScope | None,
        session_id: str | None,
        query: str,
        entry_point: str,
    ) -> Any:
        """Return the OTEL run-span context manager or a no-op stand-in.

        When OTEL is disabled (the default), returns a ``nullcontext`` so
        the call sites can ``with self._otel_run_span(...):`` without a
        runtime branch on ``self._otel_handles is None``.
        """
        if self._otel_handles is None:
            import contextlib as _ctxlib

            return _ctxlib.nullcontext()
        return self._otel_handles.tracer.start_run_span(
            run_id=run_id,
            tenant_id=scope.tenant_id if scope is not None else None,
            session_id=session_id,
            query=query,
            entry_point=entry_point,
        )

    def _resolve_callback_manager(
        self,
        callbacks: Sequence[CallbackHandler] | None,
    ) -> Any:
        """Build a CallbackManager mirroring ``runtime._resolve_callback_manager``.

        v7.4.3 (Jarvio Ask 5): used to fire ``on_llm_end`` from engine call
        sites that bypass the react node -- specifically the metacognition
        critic at ``engine.py:2117`` and the skill-gate critic at
        ``engine.py:5630-5648``.  The result combines:

        1. The deps-level manager (capabilities registered at construction).
        2. The per-invocation handlers passed into ``run()`` / ``stream()``.
        3. The metrics collector + OTel bridge (so the new emissions also
           feed metrics dashboards and trace exporters).

        Returns an empty manager when no handlers are registered -- the
        zero-cost path in ``CallbackManager.is_noop`` short-circuits the
        emission helpers downstream.
        """
        from symfonic.core.callbacks.manager import CallbackManager

        deps_mgr = self._deps.get(CallbackManager) or CallbackManager()
        merged_handlers = self._with_metrics_callbacks(callbacks) or ()
        if merged_handlers:
            override = CallbackManager(handlers=list(merged_handlers))
            return deps_mgr.merge(override)
        return deps_mgr

    def _with_metrics_callbacks(
        self,
        callbacks: Sequence[CallbackHandler] | None,
    ) -> Sequence[CallbackHandler] | None:
        """Prepend metrics_collector + OTEL callback bridge when present.

        Zero overhead when neither is enabled. The OTEL bridge is appended
        AFTER the metrics collector so metrics fire first (preserving the
        v7.0 ordering contract) and spans receive the same events.
        """
        otel_bridge = (
            self._otel_handles.callback_bridge
            if self._otel_handles is not None
            else None
        )
        if self._metrics_collector is None and otel_bridge is None:
            return callbacks
        extras: list[Any] = []
        if self._metrics_collector is not None:
            extras.append(self._metrics_collector)
        if otel_bridge is not None:
            extras.append(otel_bridge)
        if callbacks:
            extras.extend(callbacks)
        return extras

    async def _enforce_budget(
        self,
        scope: FrameworkTenantScope | None,
        *,
        is_admin: bool = False,
    ) -> None:
        """Agent-level circuit breaker: reject LLM calls when tenant is broke.

        Defence-in-depth pair to the FastAPI ``check_budget_before_llm``
        dependency.  When a caller bypasses the HTTP layer (direct library
        use), this check still fires before we spend tokens.

        No-op when:
            * no :class:`TokenBudgetTracker` is registered; OR
            * ``scope`` is None (cannot key usage without a tenant); OR
            * ``is_admin`` is true.

        Raises:
            SymfonicAgentError: wraps ``reason`` from the breaker so the
                FastAPI router can translate it to a 429.  Direct library
                callers get the same typed exception.
        """
        from symfonic.core.observability.budget import get_budget_tracker

        tracker = get_budget_tracker()
        if tracker is None or scope is None:
            return
        result = await tracker.check_budget(scope.tenant_id, is_admin=is_admin)
        if not result.allowed:
            raise SymfonicAgentError(f"Budget exceeded: {result.reason}")

    @staticmethod
    def _sync_enabled_layers(cfg: FrameworkConfig) -> Any:
        """Return OrchestratorConfig with enabled_layers synced from FrameworkConfig.

        FrameworkConfig.enabled_layers (set[str]) is the single source of truth.
        This propagates it into OrchestratorConfig.enabled_layers (frozenset[MemoryLayer])
        so the two configs can never diverge.
        """

        framework_layers = frozenset(
            MemoryLayer(name) for name in cfg.enabled_layers
        )
        return cfg.orchestrator.model_copy(update={"enabled_layers": framework_layers})

    def _attach_episodic_telemetry_sink(self) -> None:
        """Attach the configured episodic telemetry sink, if any.

        Resolves ``FrameworkConfig.episodic_telemetry_sink`` (added in
        v6.1 for T01) via :func:`resolve_episodic_sink` and installs the
        result on the live ``EpisodicLayer``. Default is ``None`` which
        short-circuits inside the layer's write path at zero cost.

        Safe on every agent topology:
          * No orchestrator / no episodic layer -> no-op.
          * Non-default layer subclass missing ``_telemetry_sink`` slot
            -> no-op (we only attach if the attribute already exists).
        """
        sink_cfg = self._config.episodic_telemetry_sink
        if not sink_cfg:
            return
        from symfonic.memory.telemetry.episodic_sink import resolve_episodic_sink

        episodic = self._orchestrator.get_layer(MemoryLayer.EPISODIC)
        if episodic is None:
            return
        if not hasattr(episodic, "_telemetry_sink"):
            return
        sink = resolve_episodic_sink(sink_cfg)
        if sink is not None:
            episodic._telemetry_sink = sink

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

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