"""Mixture-of-Agents runtime helpers for /moa turns.

The slash command marks one user turn as MoA-enabled; the normal agent loop still
owns tool calling and turn termination, while this module gathers reference-model
context before each model iteration.
"""

from __future__ import annotations

import contextlib
import functools
import hashlib
import json
import logging
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, wait as _futures_wait
from dataclasses import KW_ONLY, dataclass, replace
from types import SimpleNamespace
from typing import Any

from agent.auxiliary_client import call_llm
from agent.message_content import flatten_message_text
from agent.moa_alternation import destination_key, is_role_alternation_rejection, merge_same_role_messages
from agent.transports import get_transport
from agent.usage_pricing import CanonicalUsage

logger = logging.getLogger(__name__)

# Privacy filter (moa.privacy_filter: '' | display | full): PII classes agent.redact
# leaves alone. The phone pattern requires explicit delimiters so line numbers,
# dates, times, SHAs, IPs and versions never match.
# Advisor (reference) outputs can echo PII from the conversation — emails, phone numbers, credentials pasted
# by the user — into surfaces the user may not expect: the labelled reference blocks rendered in the UI,
# saved MoA trace files, and (in `full` mode) the guidance block injected into the aggregator prompt (issue
# #59959). Secret/credential shapes (API-key prefixes, JWTs, private keys, DB connection strings, E.164
# phone numbers) are handled by the repo's central redactor, ``agent.redact .redact_sensitive_text`` — the
# MoA filter never re-implements those. The two patterns below cover the PII classes the central redactor
# deliberately leaves alone for log/tool output (emails and formatted phone numbers). Pattern safety:
# advisory text is frequently code-review-shaped — line numbers, timestamps, git SHAs, IDs, IP addresses. A
# bare 10-digit match would mangle all of those, so the phone pattern requires clearly delimited formatting:
# a parenthesized area code and/or explicit `-`/`.` separators between groups ((555) 123-4567, 555-123-4567,
# 555.123.4567, +1 555-123-4567). Undelimited digit runs (5551234567), dates (2026-07-12), times (12:34:56),
# hex IDs, and dotted quads never match. International numbers in E.164 form (+14155551234) are already
# masked by the central redactor.
_MOA_EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
_MOA_PHONE_RE = re.compile(
    r"(?<![\w.+-])"                    # no leading word char / dot / + / - (kills IPs, IDs, versions)
    r"(?:\+?1[ .-])?"                  # optional NA country code
    r"(?:\(\d{3}\)[ .-]?|\d{3}[.-])"   # delimited area code: (555) or 555- / 555.
    r"\d{3}[.-]\d{4}"                  # exchange-subscriber with explicit separator
    r"(?![\w-])"                       # no trailing word char / hyphen
)


def _redact_reference_text(text: Any) -> Any:
    """Redact secrets (central redactor) then MoA PII patterns.

    force=True: the privacy filter is its own opt-in, independent of the global
    log-redaction toggle. code_file=True: keeps the ENV/JSON assignment heuristics
    (which mangle source snippets) off advisory prose/code.
    """
    if not isinstance(text, str) or not text:
        return text
    from agent.redact import redact_sensitive_text
    text = redact_sensitive_text(text, force=True, code_file=True)
    text = _MOA_EMAIL_RE.sub("[redacted email]", text)
    return _MOA_PHONE_RE.sub("[redacted phone]", text)


def _moa_privacy_mode(moa_raw: Any) -> str:
    """Normalized privacy-filter mode from a raw ``moa`` config."""
    from hermes_cli.moa_config import coerce_privacy_filter
    raw = moa_raw if isinstance(moa_raw, dict) else {}
    return coerce_privacy_filter(raw.get("privacy_filter"))


def _redact_reference_outputs(reference_outputs: list[tuple[str, str, Any]]) -> list[tuple[str, str, Any]]:
    """Redact advisor text in reference-output tuples; accounting slot untouched."""
    return [(label, _redact_reference_text(text), acct) for label, text, acct in reference_outputs]


def _redact_message_content(content: Any) -> Any:
    """Redact a message's content: a string, or the text parts of a content-part list."""
    if isinstance(content, str):
        return _redact_reference_text(content)
    if isinstance(content, list):
        return [
            {**p, "text": _redact_reference_text(p.get("text"))}
            if isinstance(p, dict) and isinstance(p.get("text"), str)
            else p
            for p in content
        ]
    return content


def _redact_trace_messages(messages: Any) -> Any:
    """Redact message copies for trace persistence (string or content-part lists)."""
    if not isinstance(messages, list):
        return messages
    return [
        {**m, "content": _redact_message_content(m.get("content"))} if isinstance(m, dict) else m
        for m in messages
    ]


def _redact_trace_accounting(acct: Any) -> Any:
    """Copy a ``_RefAccounting`` with its trace text (messages/output) redacted."""
    if not isinstance(acct, _RefAccounting):
        return acct
    return replace(acct, messages=_redact_trace_messages(acct.messages), output=_redact_reference_text(acct.output))


# Cold-start caches: preset and per-(provider, model) runtime are immutable for a turn.
# A MoA preset switch used to re-resolve the full config + preset + every slot's provider runtime on EACH
# create() call (once per tool-loop iteration), serially before the parallel fan-out could start — adding
# 5-30s of "frozen" latency on complex presets (#66793).
_preset_cache_lock = threading.Lock()
_preset_cache: dict[tuple, Any] = {}


def _resolve_preset_cached(preset_name: str) -> tuple[dict[str, Any], Any]:
    """``(preset, raw moa config)``; the resolved preset is cached per config file signature
    (skips resolve_moa_preset's full validation of the moa block on every create())."""
    from hermes_cli.config import get_config_path, load_config
    from hermes_cli.moa_config import resolve_moa_preset
    from utils import file_signature
    try:
        cfg_stamp = file_signature(get_config_path().stat())
    except OSError:
        cfg_stamp = None
    moa_raw = load_config().get("moa") or {}
    key = (cfg_stamp, preset_name)
    with _preset_cache_lock:
        preset = _preset_cache.get(key) if cfg_stamp is not None else None
    if preset is None:
        preset = resolve_moa_preset(moa_raw, preset_name)
        if cfg_stamp is not None:
            with _preset_cache_lock:
                _preset_cache.clear()  # one live config stamp at a time
                _preset_cache[key] = preset
    return preset, moa_raw


_runtime_cache_lock = threading.Lock()
_runtime_cache: dict[tuple[str, str, str], tuple[float, dict[str, Any]]] = {}

# Short TTL so rotated keys / base_url edits are picked up within 5 minutes.
_RUNTIME_CACHE_TTL_SECONDS = 300.0

# Cap on concurrent reference calls (guards pathologically large presets).
_MAX_REFERENCE_WORKERS = 8


@dataclass(slots=True)
class _RefAccounting:
    """Per-reference usage, cost and full trace (third slot of a reference-output tuple).

    Cost is priced at the advisor's OWN rate and summed in dollars (advisors may run
    on a different model than the aggregator). Trace fields are only populated when
    tracing is on.
    """

    usage: Any
    cost_usd: Any = None
    cost_status: str | None = None
    cost_source: str | None = None
    _: KW_ONLY
    messages: Any = None
    output: str | None = None
    model: str | None = None
    provider: str | None = None
    temperature: Any = None


# Per-tool-result char budget for the advisory view: tool CALLS are kept in full,
# tool RESULTS are head+tail previewed. The aggregator always gets the full transcript.
_REFERENCE_TOOL_RESULT_BUDGET = 4000

# Reference system prompt: without this framing a reference assumes it is the acting
# agent and refuses ("I can't access repositories") or tries to call tools.
_REFERENCE_SYSTEM_PROMPT = (
    "You are a reference advisor in a Mixture of Agents (MoA) process. You are "
    "NOT the acting agent and you do NOT execute anything: you cannot call "
    "tools, run commands, browse, or access files, repositories, or URLs, and "
    "you should not try to or apologize for being unable to. A separate "
    "aggregator/orchestrator model holds those capabilities and will take the "
    "actual actions.\n\n"
    "CRITICAL: You must NEVER claim or imply that you have executed a command, "
    "downloaded a file, accessed a URL, or performed any action. You can only "
    "analyze and advise based on the conversation context. Examples of what to "
    "avoid:\n"
    "- Bad: \"I ran curl and got 404.\"\n"
    "- Bad: \"I downloaded the file successfully.\"\n"
    "- Bad: \"I checked the repository and found...\"\n"
    "- Good: \"Based on the error pattern, a curl request to that URL would likely return 404.\"\n"
    "- Good: \"The conversation suggests downloading this file may help.\"\n"
    "- Good: \"From the context, checking the repository would reveal...\"\n\n"
    "The conversation below is the current state of a task handled by that "
    "acting agent. Your job is to give your most intelligent analysis of that "
    "state: understand the goal, reason about the problem, and advise on what "
    "to do next. Surface the best approach, concrete next steps and tool-use "
    "strategy, likely pitfalls and risks, and anything the acting agent may "
    "have missed or gotten wrong. Assume any referenced files, URLs, or "
    "systems exist and reason about them from the context given rather than "
    "asking for access.\n\n"
    "Respond with your advice directly — no preamble, no disclaimers about "
    "tools or access. Advise in prose: never emit a tool call or a JSON "
    "tool-call object, because the aggregator replays what looks like one. "
    "Your response is private guidance handed to the "
    "aggregator, not an answer shown to the user. NEVER claim to have executed "
    "anything."
)


def _slot_label(slot: dict[str, Any]) -> str:
    label = f"{(slot.get('provider') or '').strip()}:{(slot.get('model') or '').strip()}"
    effort = str(slot.get("reasoning_effort") or "").strip()
    return f"{label}[reasoning={effort}]" if effort else label


def _slot_reasoning_config(slot: dict[str, Any]) -> dict[str, Any] | None:
    """Translate optional per-MoA-slot reasoning_effort into runtime config."""
    try:
        from hermes_constants import parse_reasoning_effort
        return parse_reasoning_effort(slot.get("reasoning_effort"))
    except Exception:  # pragma: no cover - bad config must not break MoA
        return None


def _aggregator_reasoning_config(aggregator: dict[str, Any]) -> dict[str, Any] | None:
    """Aggregator reasoning config: slot > per-model > global (shared chokepoint).

    References deliberately do NOT fall back: inheriting a global ``xhigh`` into
    every advisor would multiply cost.
    """
    cfg = _slot_reasoning_config(aggregator)
    if cfg is not None:
        return cfg
    try:
        from hermes_cli.config import load_config
        from hermes_constants import resolve_reasoning_config
        return resolve_reasoning_config(load_config() or {}, str(aggregator.get("model") or ""))
    except Exception:  # pragma: no cover - bad config must not break MoA
        return None


def _slot_runtime(slot: dict[str, Any]) -> dict[str, Any]:
    """Slot → ``call_llm`` kwargs with the provider's real api_mode/base_url/api_key.

    Cached per (profile home, provider, model) with a short TTL. Falls back to bare provider/model
    on error — never cached, or a transient error would pin bare kwargs for a TTL.
    """
    provider = str(slot.get("provider") or "").strip()
    model = str(slot.get("model") or "").strip()
    # hermes_home_key() in the key: the resolved api_key/base_url are per-profile, and under a
    # multiplex gateway two profiles can share (provider, model) with different accounts.
    from hermes_constants import hermes_home_key
    cache_key = (hermes_home_key(), provider, model)
    now = time.monotonic()
    with _runtime_cache_lock:
        entry = _runtime_cache.get(cache_key)
    if entry is not None and now - entry[0] < _RUNTIME_CACHE_TTL_SECONDS:
        return entry[1]
    out: dict[str, Any] = {"provider": provider, "model": model}
    try:
        from hermes_cli.runtime_provider import resolve_runtime_provider
        rt = resolve_runtime_provider(requested=provider, target_model=model)
        out.update({k: rt[k] for k in ("base_url", "api_key", "api_mode") if rt.get(k)})
        overrides = rt.get("request_overrides")
        extra_body = overrides.get("extra_body") if isinstance(overrides, dict) else None
        if isinstance(extra_body, dict) and extra_body:
            out["extra_body"] = dict(extra_body)
    except Exception as exc:
        logger.warning("MoA slot %s: provider '%s' could not be resolved (%s); calling with bare provider/model",
                       _slot_label(slot), provider, exc)
        return out
    with _runtime_cache_lock:
        _runtime_cache[cache_key] = (now, out)
    return out


def _merge_slot_extra_body(slot_extra_body: Any, caller_extra_body: Any) -> Any:
    """Merge slot defaults with a caller override (caller wins) for ``call_llm``."""
    if not (isinstance(slot_extra_body, dict) and slot_extra_body):
        return caller_extra_body
    if isinstance(caller_extra_body, dict):
        return {**slot_extra_body, **caller_extra_body}
    return caller_extra_body or dict(slot_extra_body)


def _agent_cache_opts(agent: Any) -> tuple[Any, Any]:
    """The live agent's ``(_cache_disabled, _cache_ttl)``; ``(None, None)`` without an agent."""
    return (None, None) if agent is None else (getattr(agent, "_cache_disabled", None), getattr(agent, "_cache_ttl", None))


def _with_cache_disabled(runtime: dict[str, Any], cache_disabled: Any) -> dict[str, Any]:
    """Pin the live agent's cache disable onto a runtime snapshot (None is a no-op)."""
    return runtime if cache_disabled is None else {**runtime, "_cache_disabled": cache_disabled}


def _maybe_apply_moa_cache_control(
    messages: list[dict[str, Any]], runtime: dict[str, Any], *, cache_disabled: bool | None = None,
    cache_ttl: str | None = None,
) -> list[dict[str, Any]]:
    """Apply cache_control to an advisor/aggregator request when its route honors it.

    Same policy/marker helpers as the main loop; MoA has no static prefix so the
    legacy system-and-3 fallback is used. ``cache_disabled`` is stamped onto the
    stub so ``cache_ttl: off`` is honored; ``cache_ttl`` is clamped per destination.
    Returns the messages unchanged on any error.

    ``cache_disabled`` (or the live config when omitted) is stamped onto the policy stub so
    ``prompt_caching.cache_ttl: off`` is not bypassed by the blank-agent pattern (#76085).
    """
    try:
        from agent.agent_runtime_helpers import anthropic_prompt_cache_policy, blank_cache_policy_stub
        from agent.prompt_caching import (
            apply_anthropic_cache_control, effective_cache_ttl, envelope_tool_part_cache_markers_supported,
        )

        # Explicit kwarg > runtime snapshot (threaded from the live agent) > config.
        if cache_disabled is None and "_cache_disabled" in runtime:
            cache_disabled = runtime.get("_cache_disabled")
        provider, model, base_url = runtime.get("provider") or "", runtime.get("model") or "", runtime.get("base_url") or ""
        # blank_cache_policy_stub is the only sanctioned stub (carries _cache_disabled).
        should_cache, native_layout = anthropic_prompt_cache_policy(
            blank_cache_policy_stub(cache_disabled), provider=provider, base_url=base_url,
            api_mode=runtime.get("api_mode") or "", model=model,
        )
        if not should_cache:
            return messages
        # None → "5m"; cache-disabled routes already returned above. Envelope routes
        # reject part-level markers in tool_result.content[].
        return apply_anthropic_cache_control(
            messages, cache_ttl=effective_cache_ttl(cache_ttl, provider=provider, model=model),
            native_anthropic=native_layout,
            tool_part_markers=envelope_tool_part_cache_markers_supported(provider, base_url),
        )
    except Exception as exc:  # pragma: no cover - decoration must never break a call
        logger.debug("MoA cache_control decoration skipped: %s", exc)
        return messages


def _price_reference_response(
    response: Any, slot: dict[str, Any], runtime: dict[str, Any]
) -> tuple[Any, Any, str | None, str | None]:
    """Normalize a reference's usage with the slot's OWN provider/api_mode and price it
    at its own rate (hence fan-out cost is summed in dollars). Never raises."""
    from agent.usage_pricing import estimate_usage_cost, normalize_usage
    usage = CanonicalUsage()
    raw_usage = getattr(response, "usage", None)
    if raw_usage:
        with contextlib.suppress(Exception):  # pragma: no cover - defensive
            usage = normalize_usage(raw_usage, provider=runtime.get("provider"), api_mode=runtime.get("api_mode"))
    try:
        cost = estimate_usage_cost(
            slot.get("model") or "", usage, provider=runtime.get("provider"),
            base_url=runtime.get("base_url"), api_key=runtime.get("api_key"),
        )
        return usage, cost.amount_usd, cost.status, cost.source
    except Exception:  # pragma: no cover - defensive
        return usage, None, None, None


def _run_reference(
    slot: dict[str, Any], ref_messages: list[dict[str, Any]], *, temperature: float | None = None,
    max_tokens: int | None = None, reference_timeout: float | None = None, context_length_cache: Any = None,
    cache_disabled: bool | None = None, cache_ttl: str | None = None,
) -> tuple[str, str, Any]:
    """Call one reference model; return ``(label, text, accounting)``. Never raises:
    a failed reference becomes a labelled ``[failed: …]`` note. Runs in a thread pool."""
    label = _slot_label(slot)
    runtime = _slot_runtime(slot)
    trace_fields = {"model": slot.get("model"), "provider": runtime.get("provider") or slot.get("provider"), "temperature": temperature}
    # The advisory view already stripped the agent's system prompt; this is the only one.
    messages = [{"role": "system", "content": _REFERENCE_SYSTEM_PROMPT}, *ref_messages]
    try:
        # Trim to THIS model's window (advisors may be smaller than the aggregator); the
        # advisory view is append-only across iterations, so cache_control lets
        # iteration N+1 replay N's cached prefix.
        # Reference models may have a smaller window than the aggregator (e.g. kimi-k2.7-code @ 262K
        # advising a glm-5.2 @ 1M conversation); without this trim the provider returns a hard HTTP 400
        # which the except below silently converts to a [failed: …] note (issue #60345). Estimated AFTER the
        # advisory system prompt is prepended so its tokens count against the budget too.
        trimmed = _trim_messages_for_reference(
            messages, slot, runtime, reserve_output_tokens=max_tokens, context_length_cache=context_length_cache,
        )
        trimmed = _maybe_apply_moa_cache_control(trimmed, _with_cache_disabled(runtime, cache_disabled), cache_ttl=cache_ttl)

        # Copilot gates premium models on request attribution; MoA fan-out serves the
        # user's current turn, so mirror the main agent's x-initiator header.
        from agent.auxiliary_client import _normalize_aux_provider
        is_copilot = _normalize_aux_provider(str(runtime.get("provider") or "")) in ("copilot", "copilot-acp")
        response = call_llm(
            task="moa_reference", messages=trimmed, temperature=temperature,
            max_tokens=max_tokens,
            timeout=reference_timeout, reasoning_config=_slot_reasoning_config(slot),
            extra_headers={"x-initiator": "user"} if is_copilot else None, **runtime,
        )
        output_text = _extract_text(response) or "(empty response)"
        acct = _RefAccounting(*_price_reference_response(response, slot, runtime), messages=trimmed, output=output_text, **trace_fields)
        return label, output_text, acct
    except Exception as exc:
        logger.warning("MoA reference model %s failed: %s", label, exc)
        note = f"[failed: {exc}]"
        return label, note, _RefAccounting(CanonicalUsage(), messages=messages, output=note, **trace_fields)


# Output headroom reserved in the reference window when reference_max_tokens is unset.
_REFERENCE_DEFAULT_OUTPUT_RESERVE = 8192

# Estimator slack: estimate_messages_tokens_rough is a rough chars/4 heuristic.
_REFERENCE_TRIM_SAFETY_FRACTION = 0.10


def _reference_context_length(slot: dict[str, Any], runtime: dict[str, Any], cache: Any) -> int | None:
    """Context window for a slot, memoized in ``cache`` per (provider, model) when given.

    Failures are cached too so a flaky metadata source is not re-probed per reference.
    """
    from agent.model_metadata import get_model_context_length

    model = str(slot.get("model") or "")
    provider = str(runtime.get("provider") or slot.get("provider") or "")
    key = (provider, model)
    has_cache = isinstance(cache, dict)
    if has_cache and key in cache:
        return cache[key]
    try:
        context_length = get_model_context_length(
            model=model, base_url=str(runtime.get("base_url") or ""), api_key=str(runtime.get("api_key") or ""), provider=provider,
        )
    except Exception:
        logger.debug("MoA reference context-length resolution failed for %s", _slot_label(slot))
        context_length = None
    if has_cache:
        cache[key] = context_length
    return context_length


def _trim_messages_for_reference(
    messages: list[dict[str, Any]], slot: dict[str, str], runtime: dict[str, Any], *,
    reserve_output_tokens: int | None = None, context_length_cache: Any = None,
) -> list[dict[str, Any]]:
    """Trim an advisory request to fit a reference model's context window.

    Budget = window − ``reserve_output_tokens`` (or a default) − a safety fraction.
    Drops the OLDEST frames after the system prompt, always keeping a user-first
    body and the trailing user turn plus one preceding turn (even if still over
    budget). ``context_length_cache`` memoizes the window per (provider, model);
    unresolvable windows leave messages unchanged.

    Reference models may have a smaller context window than the aggregator or the main conversation. Without
    this trim, a reference whose window is exceeded gets a hard HTTP 400 from the provider, which
    ``_run_reference``'s try/except silently converts to a ``[failed: …]`` note — the MoA turn silently
    degrades to fewer references (issue #60345).
    """
    if not messages or not slot.get("model"):
        return messages
    from agent.model_metadata import estimate_messages_tokens_rough

    context_length = _reference_context_length(slot, runtime, context_length_cache)
    if not isinstance(context_length, int) or context_length <= 0:
        return messages
    reserve = reserve_output_tokens if isinstance(reserve_output_tokens, int) and reserve_output_tokens > 0 else _REFERENCE_DEFAULT_OUTPUT_RESERVE
    budget = int(context_length * (1.0 - _REFERENCE_TRIM_SAFETY_FRACTION)) - reserve
    estimated = estimate_messages_tokens_rough(messages)
    if budget <= 0 or estimated <= budget:
        return messages

    has_system = messages[0].get("role") == "system"
    head = [messages[0]] if has_system else []
    body = list(messages[1:] if has_system else messages)

    # Keep the trailing user turn plus at least one preceding turn.
    while len(body) > 2 and estimate_messages_tokens_rough(head + body) > budget:
        body.pop(0)
        # Preserve the user-first invariant after each pop.
        while len(body) > 2 and body[0].get("role") == "assistant":
            body.pop(0)
    # Two frames left with an assistant first: still enforce user-first.
    while len(body) > 1 and body[0].get("role") == "assistant":
        body.pop(0)

    trimmed = head + body
    dropped = len(messages) - len(trimmed)
    if dropped:
        logger.info(
            "MoA reference %s: estimated %d tokens exceeds budget %d "
            "(window %d, output reserve %d); dropped %d oldest message(s).",
            _slot_label(slot), estimated, budget, context_length, reserve, dropped,
        )
    return trimmed


_REFERENCE_POLL_INTERVAL_S = 5.0

# Sentinel for a reference aborted by user interrupt; the facade must never cache it.
_INTERRUPTED_REFERENCE_NOTE = "[skipped: interrupted by user]"


def _placeholder_output(slot: dict[str, Any], note: str) -> tuple[str, str, Any]:
    """A reference-output tuple for a slot that was not (fully) run: zero accounting."""
    return _slot_label(slot), note, _RefAccounting(CanonicalUsage())


def _settle_interrupted(
    futures: dict[Any, int], results: list, reference_models: list[dict[str, Any]], late_accounting_sink: Any,
) -> None:
    """Fill every unfinished slot after a user interrupt: cancel never-dispatched
    futures (nothing billed), keep real output of ones that just finished, and hand
    running ones (cannot be killed, WILL bill) to ``late_accounting_sink``."""
    for future, idx in futures.items():
        if results[idx] is not None:
            continue
        # #38922: a slow confirmation does NOT necessarily mean the send failed — but we must distinguish
        # two cases via future.cancel()'s return value: cancel() == False -> the coroutine was already
        # running on the gateway loop when the timeout fired; the request is in flight on the wire and
        # cannot be un-sent. Re-sending via standalone would be a guaranteed DUPLICATE, so treat it as
        # delivered (assume-delivered). cancel() == True -> the scheduled callback never started executing
        # (loop wedged/backlogged for the full 60s), so nothing was sent. We MUST fall through to the
        # standalone path or the message is silently dropped (worse than a duplicate).
        cancelled = future.cancel()
        if not cancelled and future.done():
            results[idx] = future.result()
            continue
        results[idx] = _placeholder_output(reference_models[idx], _INTERRUPTED_REFERENCE_NOTE)
        if not cancelled and late_accounting_sink is not None:
            future.add_done_callback(functools.partial(_record_late, late_accounting_sink, results[idx][0]))


def _record_late(sink: Any, label: str, future: Any) -> None:
    """Done-callback for an abandoned reference future: forward its real accounting."""
    try:
        sink(label, future.result()[2])
    except Exception:  # pragma: no cover - defensive
        logger.debug("MoA: late accounting sink failed for %s", label)


def _run_references_parallel(
    reference_models: list[dict[str, Any]], ref_messages: list[dict[str, Any]], *,
    temperature: float | None = None, max_tokens: int | None = None, progress_callback: Any = None,
    reference_timeout: float | None = None, agent: Any = None, late_accounting_sink: Any = None,
) -> list[tuple[str, str, Any]]:
    """Fan out all reference models in parallel; ``(label, text, _RefAccounting)`` per
    slot in ``reference_models`` order.

    ``provider == "moa"`` slots are skipped with a note (recursion guard).
    ``progress_callback(refs_done, refs_total, label)`` fires per completion. With
    *agent*, the wait polls every ``_REFERENCE_POLL_INTERVAL_S`` so a user interrupt
    can abort it; in-flight calls cannot be killed and bill via ``late_accounting_sink``.
    """
    if not reference_models:
        return []

    total = len(reference_models)
    results: list[tuple[str, str, Any] | None] = [None] * total
    futures: dict[Any, int] = {}
    # Propagate the turn's contextvars (approval callbacks, Nous conversation tag).
    from tools.thread_context import propagate_context_to_thread
    completed = 0
    executor = ThreadPoolExecutor(max_workers=min(_MAX_REFERENCE_WORKERS, total))
    interrupted = False
    # Shared per-fan-out context-length cache (dict get/set is GIL-atomic).
    ctx_len_cache: dict[tuple[str, str], int | None] = {}
    cache_disabled, cache_ttl = _agent_cache_opts(agent)
    try:
        for idx, slot in enumerate(reference_models):
            if slot.get("provider") == "moa":
                results[idx] = _placeholder_output(slot, "[skipped: MoA presets cannot recursively reference MoA]")
                continue
            futures[executor.submit(
                propagate_context_to_thread(_run_reference), slot, ref_messages, temperature=temperature,
                max_tokens=max_tokens, reference_timeout=reference_timeout, context_length_cache=ctx_len_cache,
                cache_disabled=cache_disabled, cache_ttl=cache_ttl,
            )] = idx

        # Collect every reference (no early exit except a user interrupt).
        pending = set(futures)
        while pending:
            done, pending = _futures_wait(pending, timeout=_REFERENCE_POLL_INTERVAL_S)
            for future in done:
                idx = futures[future]
                results[idx] = future.result()
                completed += 1
                if progress_callback is not None:
                    try:
                        progress_callback(completed, total, _slot_label(reference_models[idx]))
                    except Exception as exc:  # pragma: no cover - display must never break
                        logger.debug("MoA progress_callback failed: %s", exc)
            if pending and agent is not None and getattr(agent, "_interrupt_requested", False):
                interrupted = True
                _settle_interrupted(futures, results, reference_models, late_accounting_sink)
                break
    finally:
        executor.shutdown(wait=not interrupted, cancel_futures=interrupted)

    return [r for r in results if r is not None]


def _truncate_tool_result(text: str, budget: int = _REFERENCE_TOOL_RESULT_BUDGET) -> str:
    """Head+tail preview of a tool result with an ``[... N chars omitted ...]`` marker."""
    if not text or len(text) <= budget:
        return text
    half = budget // 2
    omitted = len(text) - 2 * half
    return f"{text[:half]}\n[... {omitted} chars omitted ...]\n{text[-half:]}"


def _field(obj: Any, name: str) -> Any:
    """Read ``name`` from a dict or an attribute object (tool calls arrive as either)."""
    if obj is None:
        return None
    return obj.get(name) if isinstance(obj, dict) else getattr(obj, name, None)


def _render_tool_calls(tool_calls: Any) -> str:
    """Render an assistant turn's tool_calls as ``[called tool: name(args)]`` lines.

    Tolerates dict- and SimpleNamespace-shaped entries (and nested ``function``).
    """
    lines: list[str] = []
    for tc in tool_calls or []:
        fn = _field(tc, "function")
        name = _field(fn, "name") or _field(tc, "name") or "tool"
        fn_args = _field(fn, "arguments")
        if isinstance(fn_args, str) or fn_args is None:
            args_text = fn_args or ""
        else:
            try:
                args_text = json.dumps(fn_args, ensure_ascii=False)
            except Exception:
                args_text = str(fn_args)
        lines.append(f"[called tool: {name}({args_text})]" if args_text else f"[called tool: {name}]")
    return "\n".join(lines)


# Cached guidance (user_turn / off-cadence every_n fanout) is reused on later iterations of the
# same turn, where it predates the tool results the acting model now sees. Without this line the
# block reads as fresh instruction and an advisor's suggested tool call gets replayed after it
# already ran.
_STALE_GUIDANCE_NOTE = (
    "This guidance was produced earlier in this turn, before the tool results below it. "
    "Check the transcript before acting on it: a step it suggests may already have run, and "
    "repeating a completed tool call is never the next step.\n"
)


_ADVISORY_INSTRUCTION = (
    "[The conversation above is the current state of the task. Give your "
    "most intelligent judgement: what is going on, what should happen next, "
    "what risks or mistakes you see, and how the acting agent should "
    "proceed.]"
)


def _tool_activity_since_last_user(messages: list[dict[str, Any]]) -> bool:
    """Whether the acting model has already called tools since the last real user turn.
    Guidance cached from the start of the turn predates those results, so replaying a
    tool call it suggests can repeat work the transcript already shows as done."""
    for msg in reversed(messages):
        role = msg.get("role")
        if role == "user":
            return False
        if role == "tool" or (role == "assistant" and msg.get("tool_calls")):
            return True
    return False


def _reference_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Build the advisory (reference-model) view of the conversation.

    Plain user/assistant TEXT turns only: system prompt dropped, tool_calls rendered
    inline, tool results folded into the preceding assistant turn as previews (no
    tool-role messages / tool_calls arrays, so strict providers do not 400). Always
    ends on a ``user`` turn (Anthropic treats a trailing assistant turn as prefill)
    by APPENDING a synthetic request. The aggregator always gets the full transcript.
    """
    rendered: list[dict[str, Any]] = []
    last_user_content: str | None = None
    for msg in messages:
        role = msg.get("role")
        content = msg.get("content")
        # Decorated (cache_control parts) and undecorated transcripts must yield a
        # byte-identical view so the advisory prefix stays cache-stable.
        text = flatten_message_text(content)
        if role == "user":
            if not text.strip() and isinstance(content, list) and content:
                # Image-only turn: strict providers reject empty user messages and
                # skipping would break alternation.
                text = "[user sent non-text content (e.g. an image attachment)]"
            if not text.strip():
                continue  # genuinely empty user turn: strict providers 400 on it
            last_user_content = text
            rendered.append({"role": "user", "content": text})
        elif role == "assistant":
            parts = [text.strip()] if text.strip() else []
            calls_text = _render_tool_calls(msg.get("tool_calls"))
            if calls_text:
                parts.append(calls_text)
            if parts:  # empty assistant turns carry nothing advisory
                rendered.append({"role": "assistant", "content": "\n".join(parts)})
        elif role == "tool":
            # Fold the tool result into the preceding assistant turn as text (a leading
            # tool result with no assistant turn opens one).
            block = f"[tool result: {_truncate_tool_result(text)}]"
            if rendered and rendered[-1].get("role") == "assistant":
                rendered[-1]["content"] = rendered[-1]["content"] + "\n" + block
            else:
                rendered.append({"role": "assistant", "content": block})
        # system and any other role are ignored.

    # Anthropic rejects trailing assistant prefill: end on a synthetic user request.
    if rendered and rendered[-1].get("role") == "assistant":
        rendered.append({"role": "user", "content": _ADVISORY_INSTRUCTION})
    if not rendered:
        # Nothing rendered: fall back to the latest user turn.
        if last_user_content is not None:
            return [{"role": "user", "content": last_user_content}]
        for msg in reversed(messages):
            fallback_text = flatten_message_text(msg.get("content")) if msg.get("role") == "user" else ""
            if fallback_text.strip():
                return [{"role": "user", "content": fallback_text}]
    return rendered


def _extract_text(response: Any) -> str:
    """Assistant text of a completed response: transport-normalized, else ``choices[0]``."""
    with contextlib.suppress(Exception):
        text = (get_transport("chat_completions").normalize_response(response).content or "").strip()
        if text:
            return text
    try:
        message = response.choices[0].message
        content = message.get("content") if isinstance(message, dict) else getattr(message, "content", message)
        if not isinstance(content, str):
            content = str(content) if content else ""
        return content.strip()
    except Exception:
        return ""


def _preset_temperature(preset: dict[str, Any], key: str) -> float | None:
    """Read an optional preset temperature; None (absent/empty/null) = provider default."""
    value = preset.get(key)
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        logger.warning("ignoring non-numeric %s=%r in MoA preset", key, value)
        return None


def _hash_messages(msgs: list[dict[str, Any]]) -> str:
    return hashlib.sha256("\u0000".join(f"{m.get('role')}:{m.get('content')}" for m in msgs).encode("utf-8", "replace")).hexdigest()


def _is_failed_reference(text: str) -> bool:
    """Whether a reference output is a ``[failed: …]`` / ``[skipped: …]`` sentinel."""
    return text.lstrip().lower().startswith(("[failed:", "[skipped:"))


def _join_reference_outputs(outputs: list[tuple[str, str, Any]], degraded: str = "") -> str:
    """Render numbered reference blocks for the aggregator, appending any degraded notice."""
    joined = "\n\n".join(f"Reference {idx} — {label}:\n{text}" for idx, (label, text, _acct) in enumerate(outputs, start=1))
    if degraded:
        joined = f"{joined}\n\n{degraded}" if joined else degraded
    return joined


def _sum_reference_accounting(outputs: list[tuple[str, str, Any]]) -> tuple[Any, Any]:
    """Sum fan-out usage AND cost in dollars (each advisor priced at its OWN rate)."""
    usage = CanonicalUsage()
    cost: Any = None
    for _lbl, _txt, acct in outputs:
        if isinstance(acct, _RefAccounting):
            if isinstance(acct.usage, CanonicalUsage):
                usage = usage + acct.usage
            if acct.cost_usd is not None:
                cost = (cost or 0) + acct.cost_usd
    return usage, cost


def _degraded_notice(failed_labels: list[str], policy: str) -> str:
    if not failed_labels or policy.strip().lower() == "silent":
        return ""
    return f"[Reference models unavailable: {', '.join(failed_labels)}]"


def _slot_labels(slots: list[dict[str, Any]]) -> str:
    return ", ".join(_slot_label(slot) for slot in slots)


def _guidance_inputs(
    reference_outputs: list[tuple[str, str, Any]], privacy_full: bool, policy: str,
) -> tuple[list[tuple[str, str, Any]], str, bool]:
    """``(advisor outputs for the aggregator, degraded notice, all_failed)``.

    'full' privacy mode redacts advisor text reaching the aggregator (applied to a
    per-call copy — caches hold raw text). Failed refs are filtered out first.
    """
    successful = [o for o in reference_outputs if not _is_failed_reference(o[1])]
    failed_labels = [label for label, text, _acct in reference_outputs if _is_failed_reference(text)]
    agg_refs = _redact_reference_outputs(successful) if privacy_full else successful
    return agg_refs, _degraded_notice(failed_labels, policy), bool(reference_outputs) and not successful


def aggregate_moa_context(
    *, user_prompt: str, api_messages: list[dict[str, Any]], reference_models: list[dict[str, Any]],
    aggregator: dict[str, Any], temperature: float | None = None, aggregator_temperature: float | None = None,
    reference_max_tokens: int | None = None, reference_timeout: float | None = None,
    degraded_reference_policy: str = "loud", agent: Any = None,
) -> str:
    """Run configured reference models and synthesize their advice (one-shot /moa).

    Failures become model-specific notes instead of aborting the loop.
    ``reference_max_tokens`` caps ONLY the fan-out (capping the aggregator truncated
    long syntheses). ``agent`` makes the fan-out interruptible.

    ``reference_max_tokens`` applies ONLY to the reference fan-out — the aggregator's own synthesis call is
    not given an advisor budget. Omission uses provider-specific defaults; native protocols may
    still require an internal wire limit.
    A hardcoded cap on the aggregator call previously truncated long aggregator syntheses (#53580) — passing
    ``reference_max_tokens`` to both calls here would silently reintroduce that regression.
    """
    reference_models = [slot for slot in reference_models if slot.get("enabled", True)]
    reference_outputs = _run_references_parallel(
        reference_models, _reference_messages(api_messages), temperature=temperature,
        max_tokens=reference_max_tokens, reference_timeout=reference_timeout, agent=agent,
    )
    privacy_full = False
    try:
        from hermes_cli.config import load_config as _load_config
        privacy_full = _moa_privacy_mode((_load_config() or {}).get("moa")) == "full"
    except Exception:  # pragma: no cover - privacy filter must never break a turn
        logger.debug("MoA privacy filter check failed", exc_info=True)
    agg_refs, degraded, all_failed = _guidance_inputs(reference_outputs, privacy_full, degraded_reference_policy)
    joined = _join_reference_outputs(agg_refs, degraded)
    if all_failed:
        # Skip the aggregator: synthesizing over nothing can block for the full provider timeout.
        logger.warning("MoA: all %d reference(s) failed — skipping aggregator synthesis", len(reference_outputs))
        return (
            "[Mixture of Agents context — all reference models failed. "
            "Proceeding without aggregated guidance.]\n"
            f"References: {_slot_labels(reference_models)}\n\n"
            f"{degraded or '[Reference models unavailable]'}"
        )

    synth_prompt = (
        "You are the aggregator in a Mixture of Agents process. Synthesize the "
        "reference responses into concise, actionable guidance for the main "
        "Hermes agent. Focus on next steps, tool-use strategy, risks, and any "
        "disagreements. Do not answer the user directly unless that is all that "
        "is needed; produce context the main agent should use in its normal loop.\n\n"
        f"Original user prompt:\n{user_prompt}\n\n"
        f"Reference responses:\n{joined}"
    )

    agg_label = _slot_label(aggregator)
    agg_runtime = _slot_runtime(aggregator)
    cache_disabled, cache_ttl = _agent_cache_opts(agent)
    try:
        # Same cache_control decoration as the advisor calls; this synthesis call is
        # a third independent MoA call path that otherwise re-bills its full input.
        agg_messages = _maybe_apply_moa_cache_control(
            [{"role": "user", "content": synth_prompt}], _with_cache_disabled(agg_runtime, cache_disabled), cache_ttl=cache_ttl,
        )
        synthesis = _extract_text(call_llm(
            task="moa_aggregator", messages=agg_messages, temperature=aggregator_temperature,
            reasoning_config=_aggregator_reasoning_config(aggregator), **agg_runtime,
        ))
    except Exception as exc:
        logger.warning("MoA aggregator model %s failed: %s", agg_label, exc)
        synthesis = ""

    return (
        "[Mixture of Agents context — use this as private guidance for the "
        "normal Hermes agent loop. You may call tools, continue reasoning, or "
        "finish normally.]\n"
        f"Aggregator: {agg_label}\n"
        f"References: {_slot_labels(reference_models)}\n\n"
        f"{(synthesis or joined).strip()}"
    )


def _completed_response_as_stream_chunk(response: Any) -> Any:
    """Adapt a completed response into one delta stream chunk (facade boundary only)."""
    choices = getattr(response, "choices", None)
    first_choice = choices[0] if isinstance(choices, (list, tuple)) and choices else None
    message = getattr(first_choice, "message", None)
    raw_tool_calls = getattr(message, "tool_calls", None)
    tool_call_deltas = None
    if isinstance(raw_tool_calls, (list, tuple)) and raw_tool_calls:
        tool_call_deltas = [
            SimpleNamespace(
                index=getattr(tc, "index", index), id=getattr(tc, "id", None), type=getattr(tc, "type", None) or "function",
                function=SimpleNamespace(
                    name=getattr(getattr(tc, "function", None), "name", None),
                    arguments=getattr(getattr(tc, "function", None), "arguments", None),
                ),
            )
            for index, tc in enumerate(raw_tool_calls)
        ]
    delta = SimpleNamespace(
        content=getattr(message, "content", None), tool_calls=tool_call_deltas,
        **{k: getattr(message, k, None) for k in ("reasoning_content", "reasoning", "reasoning_details")},
    )
    choice = SimpleNamespace(
        index=getattr(first_choice, "index", 0), delta=delta, finish_reason=getattr(first_choice, "finish_reason", None) or "stop",
    )
    return SimpleNamespace(
        id=getattr(response, "id", None), model=getattr(response, "model", None), choices=[choice], usage=getattr(response, "usage", None),
    )


def _attach_reference_guidance(agg_messages: list[dict[str, Any]], guidance: str) -> None:
    """Attach the per-turn reference block as its OWN trailing user message.

    The block varies per turn; appending keeps ``[system][task][tool-history]``
    cache-stable. It is never merged into a trailing user turn: iteration 1 of a
    tool loop ends on ``user(task)``, and a merged ``user(task + guidance)`` byte-differs
    from the ``user(task)`` every later iteration replays, so the provider prefix cache
    collapsed to the system prompt on iteration 2 of every turn (#112358). Converters
    that require strict alternation (Anthropic Messages, Converse, native Gemini) merge
    the two user turns as SEPARATE content blocks, so the task block stays byte-stable
    there too; on the OpenAI-compatible wire the request ends ``user(task), user(guidance)``,
    which a chat template that enforces strict user/assistant alternation rejects.
    """
    agg_messages.append({"role": "user", "content": guidance})


def peel_reference_guidance(messages: list[dict[str, Any]], guidance: Any) -> list[dict[str, Any]]:
    """Exact inverse of ``_attach_reference_guidance`` (plain string, or its cache-decorated
    single-text-part form), so a cache breakpoint never lands on the turn-varying guidance.
    Inputs are not mutated."""
    if not guidance or not messages:
        return messages
    guidance_text = str(guidance)
    last = messages[-1]
    if not isinstance(last, dict) or last.get("role") != "user":
        return messages
    content = last.get("content")
    if content == guidance_text:
        return list(messages[:-1])
    if isinstance(content, list) and len(content) == 1:
        part = content[0]
        if isinstance(part, dict) and part.get("type", "text") == "text" and (part.get("text") or "") == guidance_text:
            return list(messages[:-1])
    return messages


class MoAChatCompletions:
    """OpenAI-chat-compatible facade where the aggregator is the acting model.

    ``reference_callback(event, **kwargs)`` is an optional best-effort display hook
    (events: ``moa.reference``, ``moa.progress``, ``moa.phase``, ``moa.aggregating``;
    kwargs per ``_RELAY_EVENTS``). ``agent`` is the owning AIAgent; it lets the
    fan-out check ``_interrupt_requested``.
    """

    def __init__(self, preset_name: str, reference_callback: Any = None, agent: Any = None):
        self.preset_name = preset_name or "default"
        self.reference_callback = reference_callback
        self._agent = agent
        # Reference cache keyed on the advisory-view signature (HIT = no re-run, no re-emit).
        self._ref_cache_key: tuple | None = None
        self._ref_cache_outputs: list[tuple[str, str, Any]] = []
        # Fan-out spend awaiting consume_reference_usage (nothing deposited on a HIT so
        # spend counts once); the lock guards late-accounting callbacks on worker threads.
        self._pending_reference_usage: Any = CanonicalUsage()
        self._pending_reference_cost: Any = None
        self._accounting_lock = threading.Lock()
        # Real aggregator slot so cost accounting prices the acting turn at its model.
        self.last_aggregator_slot: Any = None
        # Full-turn trace parts from a cache-MISS create(), flushed by consume_and_save_trace.
        self._pending_trace: Any = None
        # Per-advisor metrics for observability hooks; NOT consumed (post_api_request
        # fires on a different branch than consume_and_save_trace).
        self._last_reference_metrics: Any = None
        # every_n cadence state, scoped to one USER TURN so iteration 1 is on-cadence.
        self._fanout_iteration_count = 0
        self._fanout_turn_sig: str | None = None
        self._fanout_last_state_sig: str | None = None
        self._privacy_mode: str = ""  # normalized moa.privacy_filter, refreshed per create()
        # Destinations (route, model) that 400'd on adjacent same-role messages this session:
        # their aggregator requests are pre-merged; every other destination keeps the split,
        # cache-stable shape (agent/moa_alternation.py).
        self._merge_same_role_destinations: set[tuple[str, str]] = set()

    def consume_reference_usage(self) -> tuple[Any, Any]:
        """Pop pending fan-out ``(CanonicalUsage, cost_usd_or_None)`` and reset both
        (so a streaming retry re-entering accounting cannot double-count)."""
        with self._accounting_lock:
            usage = self._pending_reference_usage or CanonicalUsage()
            cost = self._pending_reference_cost
            self._pending_reference_usage = CanonicalUsage()
            self._pending_reference_cost = None
        return usage, cost

    def last_reference_metrics(self) -> Any:
        """Per-advisor metrics from the most recent fan-out, or None (read-only)."""
        return self._last_reference_metrics

    def _record_late_reference_accounting(self, label: str, accounting: Any) -> None:
        """Done-callback for abandoned (still billing) futures: fold their real spend in."""
        if not isinstance(accounting, _RefAccounting):
            return
        self._fold_pending_accounting(*_sum_reference_accounting([(label, "", accounting)]))
        logger.debug("MoA: recorded late accounting for interrupted reference %s", label)

    def _fold_pending_accounting(self, usage: Any, cost: Any) -> None:
        """Add (never overwrite) fan-out spend so late interrupted-reference deposits survive."""
        with self._accounting_lock:
            self._pending_reference_usage = (self._pending_reference_usage or CanonicalUsage()) + usage
            if cost is not None:
                self._pending_reference_cost = (self._pending_reference_cost or 0) + cost

    def consume_and_save_trace(self, session_id: Any = None, aggregator_output_fallback: Any = None) -> None:
        """Flush the pending full-turn trace to disk (no-op when nothing is pending).

        ``aggregator_output_fallback`` is the caller's resolved acting text for the
        streaming path (not capturable at ``create()`` time). Never raises.
        """
        pending = self._pending_trace
        self._pending_trace = None
        if not pending or "aggregator_input_messages" not in pending:
            return
        try:
            from agent.moa_trace import save_moa_turn
            agg_slot = pending.get("aggregator_slot") or {}
            # Inline capture (non-streaming) beats the caller's streamed text.
            agg_output = pending.get("aggregator_output")
            if agg_output is None and aggregator_output_fallback:
                agg_output = aggregator_output_fallback
            save_moa_turn(
                session_id=session_id, preset_name=pending.get("preset", ""),
                reference_outputs=pending.get("reference_outputs", []),
                aggregator_label=pending.get("aggregator_label", ""), aggregator_model=agg_slot.get("model"),
                aggregator_provider=agg_slot.get("provider"), aggregator_temperature=pending.get("aggregator_temperature"),
                aggregator_input_messages=pending.get("aggregator_input_messages"),
                aggregator_output=agg_output, aggregator_streamed=bool(pending.get("aggregator_streamed")),
            )
        except Exception as exc:  # pragma: no cover - tracing must never break a turn
            logger.debug("MoA trace flush failed: %s", exc)

    def _emit(self, event: str, **kwargs: Any) -> None:
        if self.reference_callback is None:
            return
        try:
            self.reference_callback(event, **kwargs)
        except Exception as exc:  # pragma: no cover - display must never break the turn
            logger.debug("MoA reference_callback failed for %s: %s", event, exc)

    def prepare(self, messages: list[dict[str, Any]]) -> dict[str, Any]:
        """Run the advisor fan-out and return the exact aggregator request, which the
        loop measures before its compression gate and hands back to ``create()``."""
        return self.create(messages=messages, _moa_prepare_only=True)

    def rebase_prepared_request(self, prepared: dict[str, Any], messages: list[dict[str, Any]]) -> dict[str, Any]:
        """Re-attach already-generated guidance to a rebuilt (compressed) transcript."""
        guidance = prepared.get("guidance")
        agg_messages = [dict(message) for message in messages]
        if guidance:
            _attach_reference_guidance(agg_messages, str(guidance))
        return {**prepared, "messages": agg_messages}

    def _plan_aggregator_cache(
        self, agg_messages: list[dict[str, Any]], tools: Any, guidance: Any, agg_runtime: dict[str, Any],
    ) -> tuple[list[dict[str, Any]], Any]:
        """Cache-breakpoint the aggregator request for its destination.

        Guidance is peeled before planning and re-attached after so a breakpoint never
        lands on the turn-varying block. Any error → undecorated request (warning, not
        debug: this is the aggregator's ONLY decoration path).
        """
        try:
            from agent.agent_runtime_helpers import plan_cache_sections_for_destination
            planning_messages = peel_reference_guidance(agg_messages, str(guidance)) if guidance else agg_messages
            # Tri-state cache_disabled: facades built via __new__ have no _agent; forcing
            # False would suppress the planner's config fallback.
            # plan_cache_sections_for_destination never mutates its inputs and always returns request-local
            # copies, so the prepared state stays canonical. Tri-state: only pass a bool when a live agent
            # snapshot exists. See #76085.
            _agent = getattr(self, "_agent", None)
            cache_disabled, cache_ttl = _agent_cache_opts(_agent)
            # Agent TTL + stable system prefix so MoA does not regress 1h → 5m.
            agg_messages, tools = plan_cache_sections_for_destination(
                planning_messages, tools, provider=agg_runtime.get("provider") or "",
                base_url=agg_runtime.get("base_url") or "", api_mode=agg_runtime.get("api_mode") or "",
                model=agg_runtime.get("model") or "", cache_disabled=cache_disabled, cache_ttl=cache_ttl,
                static_system_prefix=getattr(_agent, "_cached_system_prompt_static", None),
            )
            if guidance:
                _attach_reference_guidance(agg_messages, str(guidance))
        except Exception as exc:  # pragma: no cover - cache planning must not block MoA
            logger.warning(
                "MoA aggregator cache plan failed — sending undecorated "
                "request (cache misses expected): %s", exc,
            )
        return agg_messages, tools

    def _call_prepared_aggregator(self, prepared: dict[str, Any], api_kwargs: dict[str, Any]) -> Any:
        """Send an already prepared MoA aggregator request exactly once."""
        aggregator = prepared["aggregator"]
        if aggregator.get("provider") == "moa":
            raise RuntimeError("MoA aggregator cannot be another MoA preset")
        agg_runtime = _slot_runtime(aggregator)
        agg_messages, tools = self._plan_aggregator_cache(
            prepared["messages"], api_kwargs.get("tools"), prepared.get("guidance"), agg_runtime
        )
        trace = self._pending_trace
        if trace is not None:
            trace["aggregator_label"] = _slot_label(aggregator)
        # stream=True returns the RAW token stream (consumer reassembles + retries);
        # the non-streaming path forwards no stream/stream_options/timeout. The
        # consumer's stream-read timeout must govern the aggregator stream.
        stream = bool(api_kwargs.get("stream"))
        stream_kwargs: dict[str, Any] = {}
        if stream:
            stream_kwargs = {"stream": True, "stream_options": api_kwargs.get("stream_options") or {"include_usage": True}}
            if api_kwargs.get("timeout") is not None:
                stream_kwargs["timeout"] = api_kwargs["timeout"]
        # Pop the runtime's extra_body so the explicit kwarg never collides with **agg_runtime.
        agg_extra_body = _merge_slot_extra_body(agg_runtime.pop("extra_body", None), api_kwargs.get("extra_body"))
        destination = destination_key(agg_runtime)
        # Facades built via __new__ (tests, swaps) have no __init__ state.
        remembered = getattr(self, "_merge_same_role_destinations", None)
        if remembered is None:
            remembered = self._merge_same_role_destinations = set()
        merged = destination in remembered
        if merged:
            agg_messages = merge_same_role_messages(agg_messages)
        send = functools.partial(
            call_llm, task="moa_aggregator", temperature=prepared["aggregator_temperature"],
            max_tokens=api_kwargs.get("max_tokens"), tools=tools, extra_body=agg_extra_body,
            reasoning_config=_aggregator_reasoning_config(aggregator),  # same policy as direct create()
            **stream_kwargs, **agg_runtime,
        )
        try:
            agg_response = send(messages=agg_messages)
        except Exception as exc:
            # Strict-alternation template rejected ``user(task), user(guidance)``: merge the pair for
            # THIS destination only and retry once; remember it so later iterations pre-merge.
            retry_messages = None if merged else merge_same_role_messages(agg_messages)
            if retry_messages is None or retry_messages is agg_messages or not is_role_alternation_rejection(exc, agg_runtime):
                raise
            remembered.add(destination)
            logger.warning(
                "MoA aggregator %s rejected adjacent same-role messages — merging them for this "
                "destination for the rest of the session and retrying once: %.200s", _slot_label(aggregator), exc,
            )
            agg_messages = retry_messages
            agg_response = send(messages=agg_messages)
        if trace is not None:
            # Trace the exact aggregator INPUT as sent (persisted copy redacted; live input raw).
            trace["aggregator_input_messages"] = (
                _redact_trace_messages([dict(m) for m in agg_messages]) if getattr(self, "_privacy_mode", "") else agg_messages
            )
            # Streaming output lands as the turn's assistant message; the trace marks it.
            trace["aggregator_streamed"] = stream
            output = None
            if not stream:
                with contextlib.suppress(Exception):  # pragma: no cover - defensive
                    output = _extract_text(agg_response)
            trace["aggregator_output"] = output
        if stream and hasattr(agg_response, "choices"):
            # Some adapters (openai-codex Responses) return a completed response even
            # when streaming was requested; hand the loop a one-chunk iterator.
            return iter((_completed_response_as_stream_chunk(agg_response),))
        return agg_response

    def _fanout_cache_key(
        self, preset: dict[str, Any], ref_messages: list[dict[str, Any]], reference_models: list[dict[str, Any]],
    ) -> tuple:
        """Turn-scoped reference cache key per the preset's fan-out cadence.

        "user_turn" (default) hashes only the prefix up to the LAST USER message, so
        later tool iterations are HITs. "per_iteration" re-runs whenever the advisory
        view changes. "every_n:<N>": iteration 1 of a turn, then every Nth; in-between
        iterations return the pinned last on-cadence key (HIT: no calls, no re-emit).
        """
        # "user_turn" (default — cheapest cadence, #67199): advisors run ONCE per user turn; subsequent tool
        # iterations reuse that turn's advice and the aggregator acts alone (the original MoA shape:
        # synthesize at the start, then let the acting model work). Implemented by hashing only the prefix
        # up to the LAST USER message so mid-turn growth doesn't change the signature — iteration 2+ becomes
        # a cache HIT. "per_iteration": advisors re-run whenever the advisory view changes — i.e. every tool
        # iteration, since the view grows with each tool result; advice tracks live task state at the cost
        # of multiplying advisor latency/spend by tool-loop depth. "every_n:<N>" (N >= 2): the middle ground
        # (issue #63393 — advisor fan-out multiplies latency/cost by the tool-iteration count). Advisors run
        # on iteration 1 of a user turn and then every Nth tool iteration; the iterations in between REUSE
        # the cached guidance from the last on-cadence run (same mechanism as user_turn's cache HIT — the
        # aggregator still gets advice every iteration, it's just not refreshed against the very latest tool
        # results). The iteration counter is scoped per user turn and resets on a new user message, so every
        # turn starts with fresh advice.
        fanout_mode = str(preset.get("fanout") or "user_turn").strip().lower()
        every_n = 0
        if fanout_mode.startswith("every_n:"):
            with contextlib.suppress(TypeError, ValueError):
                every_n = int(fanout_mode.split(":", 1)[1])
            if every_n < 2:
                fanout_mode = "per_iteration"  # every_n:1 IS per-iteration (mirrors _coerce_fanout)
        sig_messages = turn_prefix = ref_messages
        if fanout_mode == "user_turn" or every_n >= 2:
            # Last REAL user message: the synthetic _ADVISORY_INSTRUCTION marker must not
            # count or the prefix would grow (and re-sign) every iteration.
            last_user = next(
                (i for i in range(len(ref_messages) - 1, -1, -1)
                 if ref_messages[i].get("role") == "user" and ref_messages[i].get("content") != _ADVISORY_INSTRUCTION),
                None,
            )
            if last_user is not None:
                turn_prefix = ref_messages[: last_user + 1]
            if fanout_mode == "user_turn":
                sig_messages = turn_prefix

        # every_n bookkeeping: advance the counter only when the advisory STATE changed
        # (a streaming retry must not consume a cadence slot); reset on a new turn prefix.
        if every_n >= 2:
            turn_sig = _hash_messages(turn_prefix)
            if turn_sig != self._fanout_turn_sig:
                self._fanout_turn_sig = turn_sig
                self._fanout_iteration_count = 0
                self._fanout_last_state_sig = None
            state_sig = _hash_messages(ref_messages)
            if state_sig != self._fanout_last_state_sig:
                self._fanout_last_state_sig = state_sig
                self._fanout_iteration_count += 1
            # Iteration 1 is on-cadence; then every Nth iteration after it.
            if (self._fanout_iteration_count - 1) % every_n != 0 and self._ref_cache_outputs:
                return self._ref_cache_key

        return (self.preset_name, _hash_messages(sig_messages), tuple(_slot_label(s) for s in reference_models))

    def _run_fanout(
        self, preset: dict[str, Any], ref_messages: list[dict[str, Any]], reference_models: list[dict[str, Any]],
        aggregator: dict[str, Any], aggregator_temperature: Any, cache_key: tuple,
    ) -> list[tuple[str, str, Any]]:
        """Cache-MISS path of ``create``: run the advisors, account, trace and emit.

        A preset MAY cap ADVISOR output (dominant MoA latency); the acting aggregator
        is never capped. None timeout = inherit auxiliary.moa_reference.timeout.
        """
        raw_reference_timeout = preset.get("reference_timeout")
        reference_outputs = _run_references_parallel(
            reference_models, ref_messages, temperature=_preset_temperature(preset, "reference_temperature"),

            progress_callback=lambda done, total, label: self._emit("moa.progress", refs_done=done, refs_total=total, label=label),
            reference_timeout=float(raw_reference_timeout) if raw_reference_timeout else None,
            agent=self._agent, late_accounting_sink=self._record_late_reference_accounting,
        )
        # An interrupted fan-out is a partial snapshot: never cache it (a HIT would
        # replay placeholder notes every iteration).
        interrupted = any(text == _INTERRUPTED_REFERENCE_NOTE for _lbl, text, _acct in reference_outputs)
        self._ref_cache_key = None if interrupted else cache_key
        self._ref_cache_outputs = [] if interrupted else list(reference_outputs)
        # Fold advisor spend into accounting exactly once per turn.
        self._fold_pending_accounting(*_sum_reference_accounting(reference_outputs))
        # Stash the fan-out for trace persistence (aggregator parts filled in later).
        # Traces are persisted, so ANY active privacy mode redacts them.
        privacy_mode = self._privacy_mode
        trace_refs = [
            (label, _redact_reference_text(text), _redact_trace_accounting(acct)) for label, text, acct in reference_outputs
        ] if privacy_mode else list(reference_outputs)
        self._pending_trace = {
            "preset": self.preset_name, "reference_outputs": trace_refs,
            "aggregator_slot": aggregator, "aggregator_temperature": aggregator_temperature,
        }
        # Derived from the privacy-redacted trace_refs.
        try:
            from agent.moa_trace import slot_metrics
            self._last_reference_metrics = [slot_metrics(acct, label, output=text) for label, text, acct in trace_refs]
        except Exception as exc:  # pragma: no cover - never break a turn
            logger.debug("MoA reference metrics render failed: %s", exc)
            self._last_reference_metrics = None

        # Surface each answer BEFORE the aggregator acts; the cache keeps RAW text.
        ref_count = len(reference_outputs)
        for idx, (label, text, _accounting) in enumerate(reference_outputs, start=1):
            self._emit(
                "moa.reference", index=idx, count=ref_count, label=label,
                text=_redact_reference_text(text) if privacy_mode else text,
            )
        if ref_count:
            # Phase transition: fan-out complete, aggregator about to act.
            agg_label = _slot_label(aggregator)
            self._emit("moa.phase", phase="aggregator", refs_done=ref_count, refs_total=ref_count, aggregator=agg_label)
            self._emit("moa.aggregating", aggregator=agg_label, ref_count=ref_count)
        return reference_outputs

    def _build_guidance(
        self, reference_outputs: list[tuple[str, str, Any]], aggregator: dict[str, Any], degraded_reference_policy: str,
        stale: bool = False,
    ) -> str | None:
        """Render the reference block attached to the aggregator prompt (None = nothing)."""
        agg_refs, degraded, all_failed = _guidance_inputs(
            reference_outputs, self._privacy_mode == "full", degraded_reference_policy
        )
        header = (
            "[Mixture of Agents reference context]\n"
            f"Preset: {self.preset_name}\n"
            f"Aggregator/acting model: {_slot_label(aggregator)}\n"
        )
        if all_failed:
            # Every reference failed: the aggregator acts alone (loud policy → notice).
            logger.warning(
                "MoA: all %d reference(s) failed — acting aggregator-alone "
                "without reference guidance",
                len(reference_outputs),
            )
            if degraded:
                return (
                    f"{header}\n"
                    "All reference models failed this turn — no advisory "
                    "guidance is available. Act on your own judgment.\n\n"
                    f"{degraded}"
                )
            return None
        if agg_refs or degraded:
            return (
                f"{header}"
                f"References: {', '.join(label for label, _, _ in agg_refs)}\n\n"
                "Use the reference responses below as private context. You are the aggregator and acting model: "
                "answer the user directly or call tools as needed.\n"
                f"{_STALE_GUIDANCE_NOTE if stale else ''}\n"
                f"{_join_reference_outputs(agg_refs, degraded)}"
            )
        return None

    def create(self, **api_kwargs: Any) -> Any:
        prepared_request = api_kwargs.pop("_moa_prepared_request", None)
        if prepared_request is not None:
            if not isinstance(prepared_request, dict):
                raise TypeError("_moa_prepared_request must be a dict")
            return self._call_prepared_aggregator(prepared_request, api_kwargs)

        preset, moa_raw = _resolve_preset_cached(self.preset_name)
        # Remembered on self so _call_prepared_aggregator redacts the trace consistently.
        self._privacy_mode = _moa_privacy_mode(moa_raw)
        messages = list(api_kwargs.get("messages") or [])
        # A disabled preset = "use the aggregator directly".
        reference_models = [
            slot for slot in (preset.get("reference_models") or []) if slot.get("enabled", True)
        ] if preset.get("enabled", True) else []
        aggregator = preset.get("aggregator") or {}
        # The MoA path's virtual model/provider have no pricing entry; expose the real slot.
        self.last_aggregator_slot = dict(aggregator) if aggregator else None
        # None = provider default (see _preset_temperature); the acting agent's own
        # temperature applies to the aggregator (the acting model).
        aggregator_temperature = _preset_temperature(preset, "aggregator_temperature")
        if aggregator_temperature is None and api_kwargs.get("temperature") is not None:
            aggregator_temperature = api_kwargs.get("temperature")

        ref_messages = _reference_messages(messages)
        cache_key = self._fanout_cache_key(preset, ref_messages, reference_models)
        cache_hit = bool(cache_key == self._ref_cache_key and self._ref_cache_outputs)
        if cache_hit:
            # HIT: already ran and accounted. Do NOT zero pending totals (a late
            # interrupted reference may have deposited) and no trace (not a new turn).
            reference_outputs = list(self._ref_cache_outputs)
            self._pending_trace = None
        else:
            reference_outputs = self._run_fanout(preset, ref_messages, reference_models, aggregator, aggregator_temperature, cache_key)

        agg_messages = [dict(m) for m in messages]
        guidance = self._build_guidance(
            reference_outputs, aggregator, str(preset.get("degraded_reference_policy") or "loud"),
            stale=cache_hit and _tool_activity_since_last_user(messages),
        )
        if guidance:
            _attach_reference_guidance(agg_messages, guidance)

        prepared_request = {
            "messages": agg_messages, "guidance": guidance, "aggregator": aggregator,
            "aggregator_temperature": aggregator_temperature,
        }
        if api_kwargs.pop("_moa_prepare_only", False):
            return prepared_request
        return self._call_prepared_aggregator(prepared_request, api_kwargs)


class MoAClient:
    """OpenAI-client-shaped wrapper: ``client.chat.completions`` is a ``MoAChatCompletions``;
    the accounting/trace surface below is delegated to that facade."""

    def __init__(self, preset_name: str, reference_callback: Any = None, agent: Any = None):
        self.chat = type("_MoAChat", (), {})()
        self.chat.completions = MoAChatCompletions(preset_name, reference_callback=reference_callback, agent=agent)

    def consume_reference_usage(self) -> Any:
        return self.chat.completions.consume_reference_usage()

    @property
    def last_aggregator_slot(self) -> Any:
        return getattr(self.chat.completions, "last_aggregator_slot", None)

    def consume_and_save_trace(self, session_id: Any = None, aggregator_output_fallback: Any = None) -> None:
        return self.chat.completions.consume_and_save_trace(session_id, aggregator_output_fallback=aggregator_output_fallback)

    def last_reference_metrics(self) -> Any:
        return self.chat.completions.last_reference_metrics()


# Relay table: event -> (primary kwarg, secondary kwarg or None, {cb kwarg: emit kwarg}).
# The callback signature is ``cb(event, label, text, None, **moa_*)``.
_RELAY_EVENTS: dict[str, tuple[str, str | None, dict[str, str]]] = {
    "moa.reference": ("label", "text", {"moa_index": "index", "moa_count": "count"}),
    "moa.progress": ("label", None, {"moa_refs_done": "refs_done", "moa_refs_total": "refs_total"}),
    "moa.phase": ("aggregator", None, {"moa_phase": "phase", "moa_refs_done": "refs_done", "moa_refs_total": "refs_total"}),
    "moa.aggregating": ("aggregator", None, {"moa_ref_count": "ref_count"}),
}


def build_moa_facade(agent, preset_name: Any = None) -> MoAClient:
    """Single construction point for ``MoAClient``: a bare ``MoAClient(preset)`` would
    drop the ``reference_callback`` relay and silence display events for the session.
    The relay reads ``agent.tool_progress_callback`` at emit time."""
    def _moa_reference_relay(event: str, **kwargs: Any) -> None:
        cb = getattr(agent, "tool_progress_callback", None)
        spec = _RELAY_EVENTS.get(event)
        if cb is None or spec is None:
            return
        primary, secondary, extra_map = spec
        with contextlib.suppress(Exception):
            cb(
                event, str(kwargs.get(primary) or ""), str(kwargs.get(secondary) or "") if secondary else None, None,
                **{out: kwargs.get(src) for out, src in extra_map.items()},
            )

    resolved_preset = preset_name
    if resolved_preset is None and getattr(agent, "provider", None) == "moa":
        resolved_preset = getattr(agent, "model", None)
    resolved_preset = str(resolved_preset or "default")
    try:
        from hermes_cli.config import load_config
        from hermes_cli.moa_config import normalize_moa_config
        moa_cfg = normalize_moa_config(load_config().get("moa") or {})
        if resolved_preset not in (moa_cfg.get("presets") or {}):
            resolved_preset = moa_cfg.get("default_preset") or "default"
    except Exception:
        resolved_preset = "default"
    # ``agent`` lets the fan-out wait be aborted on a user interrupt.
    return MoAClient(resolved_preset, reference_callback=_moa_reference_relay, agent=agent)


def bind_moa_runtime(agent, preset_name: Any, api_key: Any = None) -> None:
    """Make ``agent`` act as the MoA preset: pin the virtual runtime fields and install the facade.

    Every site that puts an agent onto ``provider: moa`` (init, ``/model`` switch, fallback
    activation) must pin the same fields — the facade speaks only chat.completions, has no HTTP
    endpoint and no OpenAI client kwargs — or the next dispatch/rebuild reaches a real wire with a
    virtual identity (``moa://local`` 404, or the preset name sent as a model id).
    """
    agent.model = str(preset_name or "default")
    agent.provider = agent.requested_provider = "moa"
    agent.api_mode = "chat_completions"
    agent.api_key = api_key or "moa-virtual-provider"
    agent.base_url = "moa://local"
    agent._client_kwargs = {}
    agent.client = build_moa_facade(agent, agent.model)