aiturk-hermes-ide/agent/background_review.py

"""Background memory/skill review — fork the agent to evaluate the turn.

After every turn, ``AIAgent.run_conversation`` may call
:func:`spawn_background_review` to fire off a daemon thread that replays
the conversation snapshot in a forked :class:`AIAgent` and asks itself
"should any skill/memory be saved or updated?".  Writes go straight to
the memory + skill stores.  Main conversation and prompt cache are never
touched.

The fork inherits the parent's live runtime (provider, model, base_url,
credentials, cached system prompt) so it hits the same prefix cache and
uses the same auth.  It runs with a tool whitelist limited to memory and
skill management tools; everything else is denied at runtime.

See the ``hermes-agent-dev`` skill (``references/self-improvement-loop.md``)
for invariants and PR review criteria.
"""

from __future__ import annotations

import copy
import json
import logging
import os
from pathlib import Path
import threading
from typing import Any, Dict, List, Optional, Tuple

from agent.thread_scoped_output import thread_scoped_silence

logger = logging.getLogger(__name__)


_BACKGROUND_REVIEW_CANCEL_TIMEOUT_SECONDS = 2.0


class _BackgroundReviewRun:
    """Per-review cancellation and request-completion handshake."""

    def __init__(self) -> None:
        self.cancel_requested = threading.Event()
        self.request_done = threading.Event()
        self._lock = threading.Lock()
        self._review_agent = None
        self._request_finished = False
        self._cancel_dispatched = False

    def begin_request(self, review_agent: Any) -> bool:
        """Atomically admit the first provider-capable review phase."""
        with self._lock:
            if self.cancel_requested.is_set() or self._request_finished:
                return False
            self._review_agent = review_agent
            return True

    def cancel(self) -> Any:
        """Fence startup and return the running fork, if one was admitted."""
        with self._lock:
            self.cancel_requested.set()
            if self._review_agent is not None and not self._cancel_dispatched:
                self._cancel_dispatched = True
                return self._review_agent
            return None

    def mark_request_finished(self) -> bool:
        """Latch request completion once; the caller publishes the event."""
        with self._lock:
            if self._request_finished:
                return False
            self._request_finished = True
            self._review_agent = None
            return True


def prepare_background_review_run(agent: Any) -> Optional[_BackgroundReviewRun]:
    """Install a unique run token on the parent before ``Thread.start()``."""
    lock = getattr(agent, "_background_review_lock", None)
    if lock is None:
        try:
            lock = threading.Lock()
            agent._background_review_lock = lock
        except (AttributeError, TypeError):
            return None

    run = _BackgroundReviewRun()
    try:
        with lock:
            current = getattr(agent, "_background_review_run", None)
            if current is not None and not current.request_done.is_set():
                return None
            agent._background_review_run = run
    except (AttributeError, TypeError):
        return None
    return run


def finish_background_review_run(
    agent: Any,
    run: Optional[_BackgroundReviewRun],
) -> None:
    """Publish one run's request exit without clearing a successor (ABA-safe)."""
    if run is None or not run.mark_request_finished():
        return

    lock = getattr(agent, "_background_review_lock", None)
    if lock is not None:
        with lock:
            if getattr(agent, "_background_review_run", None) is run:
                agent._background_review_run = None
    elif getattr(agent, "_background_review_run", None) is run:
        agent._background_review_run = None
    run.request_done.set()


def _interrupt_background_review(review_agent: Any) -> None:
    """Request abort off-thread so a broken abort hook cannot stall foreground.

    The bounded wait on ``request_done`` in
    :func:`cancel_background_review_for_live_turn` is only effective if
    ``interrupt()`` returns quickly.  Off-loading to a daemon thread ensures
    a slow or wedged abort path cannot block the foreground turn (#84423).
    """

    def _interrupt() -> None:
        try:
            from agent.interrupt_compat import request_hard_interrupt

            request_hard_interrupt(
                review_agent,
                "superseded by a new live turn",
                tool_reason="background review superseded",
            )
        except Exception:
            logger.debug(
                "Failed to cancel in-flight background review for a new turn",
                exc_info=True,
            )

    try:
        threading.Thread(
            target=_interrupt,
            daemon=True,
            name="bg-review-cancel",
        ).start()
    except Exception:
        logger.debug(
            "Failed to start background-review cancellation thread",
            exc_info=True,
        )


def cancel_background_review_for_live_turn(agent: Any) -> None:
    """Cancel the current review and await its request-phase acknowledgement.

    Foreground priority is preserved: if the review does not acknowledge within
    the bounded deadline, a warning is logged and the live turn proceeds
    anyway. The review is non-critical self-improvement work and must never
    block a user-facing turn (#84423).
    """
    lock = getattr(agent, "_background_review_lock", None)
    if lock is not None:
        with lock:
            run = getattr(agent, "_background_review_run", None)
            legacy_agent = getattr(agent, "_background_review_agent", None)
    else:
        run = getattr(agent, "_background_review_run", None)
        legacy_agent = getattr(agent, "_background_review_agent", None)

    if run is None:
        if legacy_agent is None:
            return
        _interrupt_background_review(legacy_agent)
        return

    review_agent = run.cancel()
    if review_agent is not None:
        _interrupt_background_review(review_agent)

    acknowledged = run.request_done.wait(
        timeout=_BACKGROUND_REVIEW_CANCEL_TIMEOUT_SECONDS
    )
    if not acknowledged:
        logger.warning(
            "Background review did not acknowledge cancellation within %.1fs; "
            "proceeding with foreground live turn",
            _BACKGROUND_REVIEW_CANCEL_TIMEOUT_SECONDS,
        )


# ---------------------------------------------------------------------------
# Background-review aux-model selector + routed digest.
#
# The review fork runs on the MAIN model by default ("auto"), replaying the
# full conversation — already warm in the prompt cache, so cheap cache reads.
# Optimal and unchanged. A user can route the review to a different, cheaper
# model via auxiliary.background_review.{provider,model}. A different model
# cannot reuse the parent's cache (different key), so the fork is cold
# regardless — replaying the full transcript would just cold-write it. So when
# (and only when) routed to a different model, we replay a compact DIGEST to
# minimise cold-written tokens. Same model -> full replay; different model ->
# digest. That's the whole policy.
# ---------------------------------------------------------------------------

# Historical hardcoded iteration budget for the review fork.
_REVIEW_MAX_ITERATIONS = 16

# Default aggregate INPUT-token budget for one review fork (#93057). The
# fork's first request replays the full snapshot — a warm prompt-cache read
# that is cheap and intended (cache parity), which is why both compression
# gates are deferred until the first provider response arrives
# (_review_fork_first_request_pending in agent/turn_context.py). After that,
# detached in-memory compaction bounds each request to roughly the
# compression threshold, but nothing capped the SUM across the review's tool
# loop: one production review made 8 requests replaying 1,487,951 input
# tokens total (four of them at 350k-384k). This budget caps the aggregate;
# the review tool loop stops before the provider call that would cross it
# (see ``_review_input_budget_exhausted`` in agent/conversation_loop.py).
# 2x the historical 300k foreground trigger keeps legitimate reviews
# comfortable while capping the pathological case. Override with
# ``auxiliary.background_review.max_input_tokens``; 0 or a negative value
# disables the cap (unbounded = pre-fix behavior).
_REVIEW_MAX_INPUT_TOKENS_DEFAULT = 600_000


def _background_review_task_config(
    task_cfg: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """Return ``auxiliary.background_review`` (or ``{}`` on any failure).

    Pass ``task_cfg`` when the caller already loaded the block once so spawn /
    resolve / prompt paths do not re-read config on every turn.
    """
    if task_cfg is not None:
        return task_cfg if isinstance(task_cfg, dict) else {}
    try:
        from hermes_cli.config import load_config_readonly

        cfg = load_config_readonly()
    except Exception:
        return {}
    aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {}
    task = aux.get("background_review", {})
    return task if isinstance(task, dict) else {}


def _review_input_token_budget(
    task_cfg: Optional[Dict[str, Any]] = None,
) -> Optional[int]:
    """Aggregate input-token budget for one review fork (None = unlimited).

    Reads ``auxiliary.background_review.max_input_tokens``; falls back to
    :data:`_REVIEW_MAX_INPUT_TOKENS_DEFAULT`. ``0`` or a negative value
    disables the cap explicitly.
    """
    task = _background_review_task_config(task_cfg)
    raw = task.get("max_input_tokens", _REVIEW_MAX_INPUT_TOKENS_DEFAULT)
    try:
        budget = int(raw)
    except (TypeError, ValueError):
        budget = _REVIEW_MAX_INPUT_TOKENS_DEFAULT
    if budget <= 0:
        return None
    return budget


def load_background_review_settings() -> tuple[bool, Dict[str, Any]]:
    """Single config read for the automatic-review gate + task block.

    Returns ``(enabled, task_cfg)``. Fail-open on config errors (``enabled=True``)
    so a broken config file does not silently disable reviews — but log at
    WARNING so the cost-incurring path is visible.
    """
    try:
        from hermes_cli.config import load_config_readonly
        from utils import is_truthy_value

        cfg = load_config_readonly()
        aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {}
        task = aux.get("background_review", {})
        task = task if isinstance(task, dict) else {}
        return is_truthy_value(task.get("enabled"), default=True), task
    except Exception:
        logger.warning(
            "Failed to read background_review.enabled; leaving automatic "
            "review enabled (fail-open)",
            exc_info=True,
        )
        return True, {}


def is_background_review_enabled(
    task_cfg: Optional[Dict[str, Any]] = None,
) -> bool:
    """Return whether automatic post-turn background review may spawn.

    Controlled by ``auxiliary.background_review.enabled`` (default ``true``).
    Explicit ``/refine`` (``focus`` set) bypasses this gate — same contract as
    zeroing the nudge intervals, which stops automatic forks but leaves manual
    refine working (issue #87250).

    Prefer :func:`load_background_review_settings` at the spawn call site so
    the task block is not re-read on the same turn.
    """
    if task_cfg is not None:
        try:
            from utils import is_truthy_value

            return is_truthy_value(task_cfg.get("enabled"), default=True)
        except Exception:
            logger.warning(
                "Failed to interpret background_review.enabled; leaving "
                "automatic review enabled (fail-open)",
                exc_info=True,
            )
            return True
    enabled, _ = load_background_review_settings()
    return enabled



def _resolve_review_runtime(
    agent: Any,
    task_cfg: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """Resolve provider/model/credentials for the review fork.

    Default (auto / unset / same as parent): inherit the parent's live runtime
    (with codex_app_server -> codex_responses downgrade). ``routed`` is False —
    the fork uses the main model and the warm cache, exactly as before. When
    ``auxiliary.background_review.{provider,model}`` names a concrete model
    different from the parent's, resolve that runtime and set ``routed=True``.
    """
    parent_runtime = agent._current_main_runtime()
    parent_api_mode = parent_runtime.get("api_mode") or None
    if parent_api_mode == "codex_app_server":
        parent_api_mode = "codex_responses"
    parent = {
        "provider": agent.provider,
        "model": agent.model,
        "api_key": parent_runtime.get("api_key") or None,
        "base_url": parent_runtime.get("base_url") or None,
        "api_mode": parent_api_mode,
        "credential_pool": getattr(agent, "_credential_pool", None),
        "request_overrides": dict(getattr(agent, "request_overrides", {}) or {}),
        "max_tokens": getattr(agent, "max_tokens", None),
        "command": getattr(agent, "acp_command", None),
        "args": list(getattr(agent, "acp_args", []) or []),
        "routed": False,
    }
    task = _background_review_task_config(task_cfg)
    task_provider = (str(task.get("provider", "")).strip() or None)
    task_model = (str(task.get("model", "")).strip() or None)
    task_base_url = (str(task.get("base_url", "")).strip() or None)
    task_api_key = (str(task.get("api_key", "")).strip() or None)
    if not (task_provider and task_provider != "auto" and task_model):
        return parent
    if task_provider == (agent.provider or "") and task_model == (agent.model or ""):
        return parent  # same model/provider as parent -> not routed
    try:
        from hermes_cli.runtime_provider import resolve_runtime_provider
        rp = resolve_runtime_provider(
            requested=task_provider,
            target_model=task_model,
            explicit_api_key=task_api_key,
            explicit_base_url=task_base_url,
        )
        return {
            "provider": rp.get("provider") or task_provider,
            "model": rp.get("model") or task_model,
            "api_key": rp.get("api_key"),
            "base_url": rp.get("base_url"),
            "api_mode": rp.get("api_mode"),
            "credential_pool": rp.get("credential_pool"),
            "request_overrides": dict(rp.get("request_overrides") or {}),
            "max_tokens": rp.get("max_output_tokens"),
            "command": rp.get("command"),
            "args": list(rp.get("args") or []),
            "routed": True,
        }
    except Exception as e:
        logger.debug("background-review aux routing failed (%s); using main model", e)
        return parent


def _parent_can_emit_tool_calls(agent: Any) -> bool:
    """Whether a fork inheriting ``agent``'s runtime could act at all.

    The review fork's entire job is to emit ``memory`` / ``skill_manage`` tool
    calls. A provider that IS an autonomous agent reaches Hermes through a client
    shim, and a shim that cannot carry Hermes tool calls back turns the fork into
    a guaranteed no-op — one that still pays for a full agent spawn (a whole CLI
    process, sometimes a JVM) on every review cadence. The in-tree ACP client CAN
    carry them (it uses the text bridge in ``agent/acp_openai_bridge.py``); this
    exists so a shim that can't declares ``SUPPORTS_HERMES_TOOL_CALLS = False``
    and is skipped instead of burning a spawn. Anything that doesn't say
    otherwise is assumed capable, so ordinary providers are unaffected.
    """
    client = getattr(agent, "client", None)
    for candidate in (client, type(client) if client is not None else None):
        if candidate is None:
            continue
        supported = getattr(candidate, "SUPPORTS_HERMES_TOOL_CALLS", None)
        if supported is not None:
            return bool(supported)
    return True


def _msg_text(m: Dict) -> str:
    c = m.get("content")
    if isinstance(c, str):
        return c.strip()
    if isinstance(c, list):
        return " ".join(b.get("text", "") for b in c if isinstance(b, dict)).strip()
    return ""


def _digest_history(messages_snapshot: List[Dict], tail: int = 24) -> List[Dict]:
    """Compact replay for the routed (different-model) path only.

    Keeps the recent ``tail`` messages verbatim, collapses older turns into one
    synthetic user-role digest, preserving role alternation. Used ONLY when
    routed to a different model (cache cold regardless, so fewer cold-written
    tokens is a pure win). Never on the main-model path (full replay stays warm).
    """
    msgs = list(messages_snapshot or [])
    if len(msgs) <= tail:
        return msgs
    keep = msgs[-tail:]
    while keep and isinstance(keep[0], dict) and keep[0].get("role") == "tool":
        tail += 1
        if len(msgs) <= tail:
            return msgs
        keep = msgs[-tail:]
    old = msgs[:-len(keep)]
    lines: List[str] = []
    for m in old:
        if not isinstance(m, dict):
            continue
        role = m.get("role")
        text = _msg_text(m).replace("\n", " ")
        if role == "user" and text:
            lines.append(f"USER: {text[:300]}")
        elif role == "assistant":
            tcs = m.get("tool_calls") or []
            if tcs:
                names = [(tc.get("function") or {}).get("name", "?") for tc in tcs if isinstance(tc, dict)]
                lines.append(f"ASSISTANT[tools: {', '.join(names)}]")
            if text:
                lines.append(f"ASSISTANT: {text[:200]}")
    digest = {
        "role": "user",
        "content": (
            "[Earlier conversation digest — older turns summarised to bound the "
            "review's cold-write cost on the routed aux model. Recent turns "
            "follow verbatim below.]\n" + "\n".join(lines)
        ),
    }
    return [digest] + keep


# Review-prompt strings — used by ``spawn_background_review_thread`` to build
# the user-message that the forked review agent receives.  AIAgent exposes
# them as class attributes (``_MEMORY_REVIEW_PROMPT`` etc.) for back-compat;
# the actual text lives here so future edits are one-place.
_MEMORY_REVIEW_PROMPT = (
    "Review the conversation above and consider saving to memory if appropriate.\n\n"
    "Focus on:\n"
    "1. Has the user revealed things about themselves — their persona, desires, "
    "preferences, or personal details worth remembering?\n"
    "2. Has the user expressed expectations about how you should behave, their work "
    "style, or ways they want you to operate?\n\n"
    "If something stands out, save it using the memory tool. "
    "If nothing is worth saving, just say 'Nothing to save.' and stop."
)

_SKILL_REVIEW_PROMPT = (
    "Review the conversation above and update the skill library. Be "
    "ACTIVE — most sessions produce at least one skill update, even if "
    "small. A pass that does nothing is a missed learning opportunity, "
    "not a neutral outcome.\n\n"
    "Target shape of the library: CLASS-LEVEL skills, each with a rich "
    "SKILL.md and a `references/` directory for session-specific detail. "
    "Not a long flat list of narrow one-session-one-skill entries. This "
    "shapes HOW you update, not WHETHER you update.\n\n"
    "Signals to look for (any one of these warrants action):\n"
    "  • User corrected your style, tone, format, legibility, or "
    "verbosity. Frustration signals like 'stop doing X', 'this is too "
    "verbose', 'don't format like this', 'why are you explaining', "
    "'just give me the answer', 'you always do Y and I hate it', or an "
    "explicit 'remember this' are FIRST-CLASS skill signals, not just "
    "memory signals. Update the relevant skill(s) to embed the "
    "preference so the next session starts already knowing.\n"
    "  • User corrected your workflow, approach, or sequence of steps. "
    "Encode the correction as a pitfall or explicit step in the skill "
    "that governs that class of task.\n"
    "  • Non-trivial technique, fix, workaround, debugging path, or "
    "tool-usage pattern emerged that a future session would benefit "
    "from. Capture it.\n"
    "  • A skill that got loaded or consulted this session turned out "
    "to be wrong, missing a step, or outdated. Patch it NOW.\n\n"
    "Preference order — prefer the earliest action that fits, but do "
    "pick one when a signal above fired:\n"
    "  1. UPDATE A CURRENTLY-LOADED SKILL. Look back through the "
    "conversation for skills the user loaded via /skill-name or you "
    "read via skill_view. If any of them covers the territory of the "
    "new learning, PATCH that one first (re-load it with skill_view "
    "during this review — see Read-before-write below). It is the "
    "skill that was in play, so it's the right one to extend — but "
    "only if it is "
    "curator-managed. Bundled, hub, pinned, and user-owned skills are "
    "off-limits to you no matter how relevant (see Protected skills "
    "below); for those, fall through to the next option.\n"
    "  2. UPDATE AN EXISTING UMBRELLA (via skills_list + skill_view). "
    "If no loaded skill fits but an existing class-level skill does, "
    "patch it. Add a subsection, a pitfall, or broaden a trigger.\n"
    "  3. ADD A SUPPORT FILE under an existing umbrella. Skills can be "
    "packaged with three kinds of support files — use the right "
    "directory per kind:\n"
    "     • `references/<topic>.md` — session-specific detail (error "
    "transcripts, reproduction recipes, provider quirks) AND "
    "condensed knowledge banks: quoted research, API docs, external "
    "authoritative excerpts, or domain notes you found while working "
    "on the problem. Write it concise and for the value of the task, "
    "not as a full mirror of upstream docs.\n"
    "     • `templates/<name>.<ext>` — starter files meant to be "
    "copied and modified (boilerplate configs, scaffolding, a "
    "known-good example the agent can `reproduce with modifications`).\n"
    "     • `scripts/<name>.<ext>` — statically re-runnable actions "
    "the skill can invoke directly (verification scripts, fixture "
    "generators, deterministic probes, anything the agent should run "
    "rather than hand-type each time).\n"
    "     Add support files via skill_manage action=write_file with "
    "file_path starting 'references/', 'templates/', or 'scripts/'. "
    "The umbrella's SKILL.md should gain a one-line pointer to any "
    "new support file so future agents know it exists.\n"
    "  4. CREATE A NEW CLASS-LEVEL UMBRELLA SKILL when no existing "
    "skill covers the class. The name MUST be at the class level. "
    "The name MUST NOT be a specific PR number, error string, feature "
    "codename, library-alone name, or 'fix-X / debug-Y / audit-Z-today' "
    "session artifact. If the proposed name only makes sense for "
    "today's task, it's wrong — fall back to (1), (2), or (3).\n\n"
    "Read-before-write (ENFORCED — skill_manage refuses otherwise): "
    "before you patch or edit an existing skill's SKILL.md, call "
    "skill_view(name) for that skill during this review. Before you "
    "overwrite or remove an EXISTING supporting file, call "
    "skill_view(name, file_path=...) for that exact file. Content "
    "quoted earlier in the conversation transcript does NOT count — "
    "the guard requires a fresh load within this review, and your "
    "write must be based on what skill_view just returned. Creating "
    "a brand-new skill or adding a NEW supporting file needs no "
    "prior read. If a write is refused with a read-before-write "
    "error, call skill_view for the named target once and retry the "
    "write once; do not loop.\n\n"
    "User-preference embedding (important): when the user expressed a "
    "style/format/workflow preference, the update belongs in the "
    "SKILL.md body, not just in memory. Memory captures 'who the user "
    "is and what the current situation and state of your operations "
    "are'; skills capture 'how to do this class of task for this "
    "user'. When they complain about how you handled a task, the "
    "skill that governs that task needs to carry the lesson.\n\n"
    "If you notice two existing skills that overlap, note it in your "
    "reply — the background curator handles consolidation at scale.\n\n"
    "Protected skills (DO NOT edit these):\n"
    "  • Bundled skills (shipped with Hermes, e.g. 'hermes-agent').\n"
    "  • Hub-installed skills (installed via 'hermes skills install').\n"
    "  • Skills in skills.external_dirs (externally owned).\n"
    "  • PINNED skills (marked via 'hermes curator pin'). You are an "
    "autonomous no-user-present actor, so pin blocks your writes too — "
    "content updates included. Only the user, in a foreground session, "
    "can change a pinned skill.\n"
    "  • USER-OWNED skills — anything not curator-managed. A skill the "
    "user hand-wrote, installed by URL, or asked a foreground agent to "
    "create is theirs, not yours; your writes to it WILL be refused. "
    "This includes skills that were loaded or consulted this session: "
    "being in play does not make one yours to edit. If such a skill is "
    "wrong or outdated, say so in your reply and recommend "
    "'hermes curator adopt <name>' — do not try to patch it.\n"
    "If the only skills that need updating are protected, say\n"
    "'Nothing to save.' and stop.\n\n"
    "Do NOT capture (these become persistent self-imposed constraints "
    "that bite you later when the environment changes):\n"
    "  • Environment-dependent failures: missing binaries, fresh-install "
    "errors, post-migration path mismatches, 'command not found', "
    "unconfigured credentials, uninstalled packages. The user can fix "
    "these — they are not durable rules.\n"
    "  • Negative claims about tools or features ('browser tools do not "
    "work', 'X tool is broken', 'cannot use Y from execute_code'). These "
    "harden into refusals the agent cites against itself for months "
    "after the actual problem was fixed.\n"
    "  • Session-specific transient errors that resolved before the "
    "conversation ended. If retrying worked, the lesson is the retry "
    "pattern, not the original failure.\n"
    "  • One-off task narratives. A user asking 'summarize today's "
    "market' or 'analyze this PR' is not a class of work that warrants "
    "a skill.\n\n"
    "  • Unresolved failures: if the session ended WITHOUT actually "
    "finding a working method — you tried several things, none worked, "
    "and told the user to check manually — do NOT write those attempts "
    "up as a 'reliable workflow' or 'recommended approach'. That presents "
    "an untested sequence of failures as validated guidance a future "
    "session will trust and repeat. Either say 'Nothing to save', or, "
    "only if you are independently confident of a real working alternative "
    "(not something you are merely guessing might work), capture ONLY that "
    "alternative — never the dead ends, and never dressed up as best practice.\n\n"
    "If a tool failed because of setup state, capture the FIX (install "
    "command, config step, env var to set) under an existing setup or "
    "troubleshooting skill — never 'this tool does not work' as a "
    "standalone constraint.\n\n"
    "'Nothing to save.' is a real option but should NOT be the "
    "default. If the session ran smoothly with no corrections and "
    "produced no new technique, just say 'Nothing to save.' and stop. "
    "Otherwise, act."
)

_COMBINED_REVIEW_PROMPT = (
    "Review the conversation above and update two things:\n\n"
    "**Memory**: who the user is. Did the user reveal persona, "
    "desires, preferences, personal details, or expectations about "
    "how you should behave? Save facts about the user and durable "
    "preferences with the memory tool.\n\n"
    "**Skills**: how to do this class of task. Be ACTIVE — most "
    "sessions produce at least one skill update. A pass that does "
    "nothing is a missed learning opportunity, not a neutral outcome.\n\n"
    "Target shape of the skill library: CLASS-LEVEL skills with a rich "
    "SKILL.md and a `references/` directory for session-specific detail. "
    "Not a long flat list of narrow one-session-one-skill entries.\n\n"
    "Signals that warrant a skill update (any one is enough):\n"
    "  • User corrected your style, tone, format, legibility, "
    "verbosity, or approach. Frustration is a FIRST-CLASS skill "
    "signal, not just a memory signal. 'stop doing X', 'don't format "
    "like this', 'I hate when you Y' — embed the lesson in the skill "
    "that governs that task so the next session starts fixed.\n"
    "  • Non-trivial technique, fix, workaround, or debugging path "
    "emerged.\n"
    "  • A skill that was loaded or consulted turned out wrong, "
    "missing, or outdated — patch it now.\n\n"
    "Preference order for skills — pick the earliest that fits:\n"
    "  1. UPDATE A CURRENTLY-LOADED SKILL. Check what skills were "
    "loaded via /skill-name or skill_view in the conversation. If one "
    "of them covers the learning, PATCH it first (re-load it with "
    "skill_view during this review — see Read-before-write below). "
    "It was in play; "
    "it's the right place — provided it is curator-managed. Protected "
    "and user-owned skills are off-limits however relevant; fall "
    "through when one of those is the best fit.\n"
    "  2. UPDATE AN EXISTING UMBRELLA (skills_list + skill_view to "
    "find the right one). Patch it.\n"
    "  3. ADD A SUPPORT FILE under an existing umbrella via "
    "skill_manage action=write_file. Three kinds: "
    "`references/<topic>.md` for session-specific detail OR condensed "
    "knowledge banks (quoted research, API docs excerpts, domain "
    "notes) written concise and task-focused; `templates/<name>.<ext>` "
    "for starter files meant to be copied and modified; "
    "`scripts/<name>.<ext>` for statically re-runnable actions "
    "(verification, fixture generators, probes). Add a one-line "
    "pointer in SKILL.md so future agents find them.\n"
    "  4. CREATE A NEW CLASS-LEVEL UMBRELLA when nothing exists. "
    "Name at the class level — NOT a PR number, error string, "
    "codename, library-alone name, or 'fix-X / debug-Y' session "
    "artifact. If the name only fits today's task, fall back to (1), "
    "(2), or (3).\n\n"
    "Read-before-write (ENFORCED — skill_manage refuses otherwise): "
    "before patching or editing an existing skill's SKILL.md, call "
    "skill_view(name) during this review; before overwriting or "
    "removing an EXISTING supporting file, call skill_view(name, "
    "file_path=...) for that exact file. Content quoted earlier in "
    "the transcript does NOT count — base the write on what "
    "skill_view just returned. New skills and NEW supporting files "
    "need no prior read. On a read-before-write refusal: view the "
    "named target once, retry the write once, do not loop.\n\n"
    "User-preference embedding: when the user complains about how "
    "you handled a task, update the skill that governs that task — "
    "memory alone isn't enough. Memory says 'who the user is and "
    "what the current situation and state of your operations are'; "
    "skills say 'how to do this class of task for this user'. Both "
    "should carry user-preference lessons when relevant.\n\n"
    "If you notice overlapping existing skills, mention it — the "
    "background curator handles consolidation.\n\n"
    "Protected skills (DO NOT edit these):\n"
    "  • Bundled skills (shipped with Hermes, e.g. 'hermes-agent').\n"
    "  • Hub-installed skills (installed via 'hermes skills install').\n"
    "  • Skills in skills.external_dirs (externally owned).\n"
    "  • PINNED skills (marked via 'hermes curator pin'). Pin blocks "
    "autonomous writes entirely — content updates included — because no "
    "user is present to consent. Only a foreground session can change one.\n"
    "  • USER-OWNED skills — anything not curator-managed (hand-written, "
    "URL-installed, or created by a foreground agent at the user's "
    "request). Your writes to these WILL be refused, including to skills "
    "loaded or consulted this session. If one is wrong, say so in your "
    "reply and recommend 'hermes curator adopt <name>' instead.\n"
    "If the only skills that need updating are protected, say\n"
    "'Nothing to save.' and stop.\n\n"
    "Do NOT capture as skills (these become persistent self-imposed "
    "constraints that bite you later when the environment changes):\n"
    "  • Environment-dependent failures: missing binaries, fresh-install "
    "errors, post-migration path mismatches, 'command not found', "
    "unconfigured credentials, uninstalled packages. The user can fix "
    "these — they are not durable rules.\n"
    "  • Negative claims about tools or features ('browser tools do not "
    "work', 'X tool is broken', 'cannot use Y from execute_code'). These "
    "harden into refusals the agent cites against itself for months "
    "after the actual problem was fixed.\n"
    "  • Session-specific transient errors that resolved before the "
    "conversation ended. If retrying worked, the lesson is the retry "
    "pattern, not the original failure.\n"
    "  • One-off task narratives. A user asking 'summarize today's "
    "market' or 'analyze this PR' is not a class of work that warrants "
    "a skill.\n\n"
    "  • Unresolved failures: if the session ended WITHOUT actually "
    "finding a working method — you tried several things, none worked, "
    "and told the user to check manually — do NOT write those attempts "
    "up as a 'reliable workflow' or 'recommended approach'. That presents "
    "an untested sequence of failures as validated guidance a future "
    "session will trust and repeat. Either say 'Nothing to save', or, "
    "only if you are independently confident of a real working alternative "
    "(not something you are merely guessing might work), capture ONLY that "
    "alternative — never the dead ends, and never dressed up as best practice.\n\n"
    "If a tool failed because of setup state, capture the FIX (install "
    "command, config step, env var to set) under an existing setup or "
    "troubleshooting skill — never 'this tool does not work' as a "
    "standalone constraint.\n\n"
    "Act on whichever of the two dimensions has real signal. If "
    "genuinely nothing stands out on either, say 'Nothing to save.' "
    "and stop — but don't reach for that conclusion as a default."
)



def summarize_background_review_actions(
    review_messages: List[Dict],
    prior_snapshot: List[Dict],
    notification_mode: str = "on",
) -> List[str]:
    """Build the human-facing action summary for a background review pass.

    Walks the review agent's session messages and collects successful memory
    and skill-management actions to surface to the user. Tool messages already
    present in ``prior_snapshot`` are skipped so stale inherited results are
    not re-surfaced as fresh background work (issue #14944).

    ``notification_mode`` controls display detail:
    - ``off``: return no actions.
    - ``on``: generic "Memory updated"/tool messages.
    - ``verbose``: include compact content previews from tool-call arguments.
    """
    mode = str(notification_mode or "on").lower()
    if mode == "off":
        return []
    verbose = mode == "verbose"

    existing_tool_call_ids = set()
    existing_tool_contents = set()
    for prior in prior_snapshot or []:
        if not isinstance(prior, dict) or prior.get("role") != "tool":
            continue
        tcid = prior.get("tool_call_id")
        if tcid:
            existing_tool_call_ids.add(tcid)
        else:
            content = prior.get("content")
            if isinstance(content, str):
                existing_tool_contents.add(content)

    # Map review-agent tool results back to the calls that produced them.  The
    # result JSON only says "Entry added"; the call arguments contain action,
    # target, and content previews.  Restricting to notify_tools also prevents
    # helper tools from surfacing as memory work just because they succeeded.
    notify_tools = {"memory", "skill_manage"}
    all_tool_call_ids: set = set()
    call_details: dict = {}
    for msg in review_messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls", []) or []:
            if not isinstance(tc, dict):
                continue
            fn = tc.get("function", {}) or {}
            fn_name = fn.get("name", "")
            tcid = tc.get("id")
            if tcid:
                all_tool_call_ids.add(tcid)
            if fn_name not in notify_tools:
                continue
            try:
                args = json.loads(fn.get("arguments", "{}"))
            except (json.JSONDecodeError, TypeError):
                args = {}
            if tcid:
                call_details[tcid] = {
                    "tool": fn_name,
                    "action": args.get("action", "?"),
                    "target": args.get("target", "memory"),
                    "content": args.get("content", ""),
                    "old_text": args.get("old_text", ""),
                    "operations": args.get("operations") or [],
                    "name": args.get("name", ""),
                    "old_string": args.get("old_string", ""),
                    "new_string": args.get("new_string", ""),
                }

    actions: List[str] = []
    for msg in review_messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "tool":
            continue
        tcid = msg.get("tool_call_id")
        if tcid and tcid in existing_tool_call_ids:
            continue
        if not tcid:
            content_str = msg.get("content")
            if isinstance(content_str, str) and content_str in existing_tool_contents:
                continue
        if tcid and all_tool_call_ids and tcid not in call_details:
            continue
        try:
            data = json.loads(msg.get("content", "{}"))
        except (json.JSONDecodeError, TypeError):
            continue
        # ``data`` may not be a dict — some memory/skill tool responses in
        # older codepaths or wrapper MCP servers return a top-level JSON
        # list (e.g. ``[{"success": true, ...}]``) or a scalar.  The original
        # isinstance check below silently skips non-dict payloads, which
        # is correct, but ``data.get("_change")`` further down can still
        # hand back a list and break ``change.get("description", "")``.
        # Defensively normalize everything through a dict-typed alias so
        # the rest of the function can stay terse without per-call
        # ``isinstance`` guards (#59437).
        if not isinstance(data, dict) or not data.get("success"):
            continue
        message = data.get("message", "")
        detail = call_details.get(tcid) or {}
        if not isinstance(detail, dict):
            detail = {}
        target = data.get("target", "") or detail.get("target", "")
        is_skill = detail.get("tool") == "skill_manage"

        message_lower = message.lower()
        if not verbose:
            if "created" in message_lower:
                actions.append(message)
                continue
            if "updated" in message_lower:
                actions.append(message)
                continue
            if is_skill and "patched" in message_lower:
                actions.append(message)
                continue

        if is_skill:
            label = "Skill"
        elif target:
            label = "Memory" if target == "memory" else "User profile" if target == "user" else target
        else:
            continue

        if verbose:
            action = detail.get("action", "")
            content = detail.get("content", "")
            old_text = detail.get("old_text", "")
            skill_name = detail.get("name", "")
            # ``operations`` may be anything callable put into the JSON
            # arguments.  Anything non-iterable that isn't a list[str]
            # of dicts becomes unusable here, so coerce defensively.
            ops_raw = detail.get("operations")
            operations: list = (
                ops_raw if isinstance(ops_raw, list) else []
            )
            max_preview = 120
            if is_skill:
                # ``_change`` is a free-form dict the skill tool leaves in
                # the response.  Older / wrapper MCP backends return it
                # as a list, an int, or a JSON-shaped scalar — normalize
                # to a dict so the .get() calls downstream don't
                # AttributeError (#59437).
                change_raw = data.get("_change")
                change: dict = (
                    change_raw if isinstance(change_raw, dict) else {}
                )
                old_string = (
                    change.get("old", "") or detail.get("old_string", "")
                )
                new_string = (
                    change.get("new", "") or detail.get("new_string", "")
                )
                description = change.get("description", "")
                if action == "patch" and (old_string or new_string):
                    old_preview = old_string[:80].replace("\n", " ") + (
                        "…" if len(old_string) > 80 else ""
                    )
                    new_preview = new_string[:80].replace("\n", " ") + (
                        "…" if len(new_string) > 80 else ""
                    )
                    actions.append(
                        f"📝 Skill '{skill_name}' patched: "
                        f"\"{old_preview}\" → \"{new_preview}\""
                    )
                elif action == "create" and description:
                    actions.append(f"📝 Skill '{skill_name}' created: {description}")
                elif action == "edit" and description:
                    actions.append(f"📝 Skill '{skill_name}' rewritten: {description}")
                else:
                    actions.append(f"📝 {message}" if message else f"Skill {action}")
            elif operations:
                for op in operations:
                    # Each element must be a dict-of-fields; some
                    # legacy codepaths serialize the entry as a bare
                    # string and the message dict doesn't exist.  Skip
                    # non-dict items defensively — they have no
                    # actionable fields anyway (#59437).
                    if not isinstance(op, dict):
                        continue
                    op_act = op.get("action", "")
                    op_content = (op.get("content") or "")
                    op_old = (op.get("old_text") or "")
                    if op_act == "add" and op_content:
                        preview = op_content[:max_preview] + ("…" if len(op_content) > max_preview else "")
                        actions.append(f"{label} ➕ {preview}")
                    elif op_act == "replace" and op_content:
                        preview = op_content[:max_preview] + ("…" if len(op_content) > max_preview else "")
                        actions.append(f"{label} ✏️ {preview}")
                    elif op_act == "remove" and op_old:
                        preview = op_old[:60] + ("…" if len(op_old) > 60 else "")
                        actions.append(f"{label} ➖ {preview}")
            elif action == "add" and content:
                preview = content[:max_preview] + ("…" if len(content) > max_preview else "")
                actions.append(f"{label} ➕ {preview}")
            elif action == "replace" and content:
                preview = content[:max_preview] + ("…" if len(content) > max_preview else "")
                actions.append(f"{label} ✏️ {preview}")
            elif action == "remove" and old_text:
                preview = old_text[:60] + ("…" if len(old_text) > 60 else "")
                actions.append(f"{label} ➖ {preview}")
            else:
                actions.append(f"{label} updated")
        elif (
            "added" in message_lower
            or "replaced" in message_lower
            or "removed" in message_lower
            or "applied" in message_lower
            or (target and "add" in message.lower())
            or "Entry added" in message
        ):
            actions.append(f"{label} updated")
    return actions


def build_memory_write_metadata(
    agent: Any,
    *,
    write_origin: Optional[str] = None,
    execution_context: Optional[str] = None,
    task_id: Optional[str] = None,
    tool_call_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Build provenance metadata for external memory-provider mirrors."""
    metadata: Dict[str, Any] = {
        "write_origin": write_origin or getattr(agent, "_memory_write_origin", "assistant_tool"),
        "execution_context": (
            execution_context
            or getattr(agent, "_memory_write_context", "foreground")
        ),
        "session_id": agent.session_id or "",
        "parent_session_id": agent._parent_session_id or "",
        "platform": agent.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"),
        "tool_name": "memory",
    }
    if task_id:
        metadata["task_id"] = task_id
    if tool_call_id:
        metadata["tool_call_id"] = tool_call_id
    return {k: v for k, v in metadata.items() if v not in {None, ""}}


def _snapshot_review_usage(review_agent: Any) -> Dict[str, Any]:
    """Snapshot in-memory usage counters from a review fork (pre-close)."""
    return {
        "model": getattr(review_agent, "model", None),
        "provider": getattr(review_agent, "provider", None),
        "base_url": getattr(review_agent, "base_url", None),
        "input_tokens": int(getattr(review_agent, "session_input_tokens", 0) or 0),
        "output_tokens": int(getattr(review_agent, "session_output_tokens", 0) or 0),
        "cache_read_tokens": int(
            getattr(review_agent, "session_cache_read_tokens", 0) or 0
        ),
        "cache_write_tokens": int(
            getattr(review_agent, "session_cache_write_tokens", 0) or 0
        ),
        "reasoning_tokens": int(
            getattr(review_agent, "session_reasoning_tokens", 0) or 0
        ),
        "api_calls": int(getattr(review_agent, "session_api_calls", 0) or 0),
        "estimated_cost_usd": getattr(review_agent, "session_estimated_cost_usd", None),
    }


def _record_review_usage_to_parent(
    parent_agent: Any,
    usage: Dict[str, Any],
) -> None:
    """Record a background-review fork's usage against the parent session.

    Background-review forks run with ``_session_db = None`` for persistence
    isolation (see the PERSISTENCE ISOLATION comment in
    :func:`_run_review_in_thread`): the fork must never write its harness turn
    into the user's real session. A side effect of that isolation is that the
    fork's API calls — which the provider bills — were never recorded in
    ``session_model_usage``, because the accounting path in
    ``conversation_loop`` is gated on the DB handle. This hides the
    background-review volume from billing analytics (issue #87250).

    The fork still accumulates the same in-memory counters the main loop does
    (``session_input_tokens`` etc.) and shares the parent's ``session_id``, so
    its usage can be attributed to the parent session through the
    aux-accounting chokepoint, which writes only ``session_model_usage`` —
    never the transcript or the ``sessions`` summary row.

    Best-effort by contract: accounting must never fail the review.
    """
    try:
        session_db = getattr(parent_agent, "_session_db", None)
        session_id = getattr(parent_agent, "session_id", None)
        if session_db is None or not session_id:
            return
        input_tokens = int(usage.get("input_tokens") or 0)
        output_tokens = int(usage.get("output_tokens") or 0)
        cache_read = int(usage.get("cache_read_tokens") or 0)
        cache_write = int(usage.get("cache_write_tokens") or 0)
        reasoning = int(usage.get("reasoning_tokens") or 0)
        api_calls = int(usage.get("api_calls") or 0)
        if not (
            input_tokens
            or output_tokens
            or cache_read
            or cache_write
            or reasoning
            or api_calls
        ):
            return  # fork made no successful API calls (e.g. failed at spawn)
        session_db.record_auxiliary_usage(
            session_id,
            task="background_review",
            model=usage.get("model"),
            billing_provider=usage.get("provider"),
            billing_base_url=usage.get("base_url"),
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cache_read_tokens=cache_read,
            cache_write_tokens=cache_write,
            reasoning_tokens=reasoning,
            estimated_cost_usd=usage.get("estimated_cost_usd"),
            api_call_count=api_calls,
        )
    except Exception as e:
        logger.debug(
            "Background review usage recording failed (non-fatal): %s", e
        )


def _classify_review_result(actions: List[str]) -> str:
    """Map a review action summary to ``none`` / ``skill`` / ``memory`` / both.

    Matching is prefix-based on the formats
    :func:`summarize_background_review_actions` emits
    (``Skill …``, ``📝 Skill …``, ``Memory …``, ``User profile …``), not
    free-text substring search — so a line like
    ``Skipped: no skill worth saving`` stays ``none``.
    """
    if not actions:
        return "none"
    has_skill = False
    has_memory = False
    for action in actions:
        text = str(action).lstrip()
        if text.startswith("📝"):
            text = text[1:].lstrip()
        lower = text.lower()
        if lower.startswith("skill"):
            has_skill = True
        elif lower.startswith("memory") or lower.startswith("user profile"):
            has_memory = True
    if has_skill and has_memory:
        return "skill+memory"
    if has_skill:
        return "skill"
    if has_memory:
        return "memory"
    return "none"


def _log_review_completion(usage: Dict[str, Any], result: str) -> None:
    """Emit a per-fork completion line so cost is visible where it is incurred."""
    logger.info(
        "Background review complete: thread=bg-review calls=%d in=%d out=%d "
        "cache_read=%d result=%s",
        int(usage.get("api_calls") or 0),
        int(usage.get("input_tokens") or 0),
        int(usage.get("output_tokens") or 0),
        int(usage.get("cache_read_tokens") or 0),
        result,
    )


def build_cache_parity_fork(
    agent: Any,
    task_cfg: Optional[Dict[str, Any]] = None,
    *,
    max_iterations: int,
    write_origin: str = "background_review",
) -> Tuple[Any, Dict[str, Any], bool]:
    """Construct a detached AIAgent fork with warm prompt-cache parity.

    This is the fork recipe the self-improvement background review uses,
    extracted so other conversation-snapshot consumers (``/btw`` side
    questions) get the identical cache-parity guarantees: same runtime and
    credentials as the parent, byte-identical system prompt / tools[] /
    reasoning config on the same-model path, shared session_id for prefix
    warmth, and full persistence detachment (no state.db writes, no session
    rotation, no external memory providers, in-place-only compaction).

    Returns ``(fork_agent, runtime_dict, routed)`` where ``routed`` is True
    when auxiliary config redirected the fork to a different model (cache
    cold; callers should replay a digest instead of the full snapshot).

    The caller keeps ownership of: registering the fork on the parent's
    ``_active_children`` / ``_background_review_agent`` slots, thread tool
    whitelisting, running the conversation, usage attribution, and teardown
    (``shutdown_memory_provider()`` + ``close()``).
    """
    # Local import to avoid a hard circular dep at module load.
    from run_agent import AIAgent

    # Inherit the parent agent's live runtime (provider, model,
    # base_url, api_key, api_mode) so the fork uses the exact
    # same credentials the main turn is using.  Without this,
    # AIAgent.__init__ re-runs auto-resolution from env vars,
    # which fails for OAuth-only providers, session-scoped
    # creds, or credential-pool setups where the resolver can't
    # reconstruct auth from scratch -- producing the spurious
    # "No LLM provider configured" warning at end of turn.
    # _resolve_review_runtime() returns the parent's live runtime by
    # default (routed=False; main model, warm cache), or — when the user
    # set auxiliary.background_review.{provider,model} to a different
    # model — that model's runtime (routed=True). The codex_app_server
    # -> codex_responses downgrade is applied inside the resolver.
    _rt = _resolve_review_runtime(agent, task_cfg)
    _routed = bool(_rt.get("routed"))
    # skip_memory=True keeps the review fork from
    # touching external memory plugins (honcho, mem0,
    # supermemory, etc.).  Without it, the fork's
    # __init__ rebuilds its own _memory_manager from
    # config, scoped to the parent's session_id, and
    # run_conversation() then leaks the harness prompt
    # into the user's real memory namespace via three
    # ingestion sites: on_turn_start (cadence + turn
    # message), prefetch_all (recall query), and
    # sync_all (harness prompt + review output recorded
    # as a (user, assistant) turn pair).  Built-in
    # MEMORY.md / USER.md state is re-bound from the
    # parent below so memory(action="add") writes from
    # the review still land on disk; the review just
    # has zero side effects on external providers.
    # Match parent's toolset config so ``tools[]`` is byte-identical
    # in the request body — Anthropic's cache key includes it.
    # (The runtime whitelist below still restricts dispatch.)
    _fork_kwargs: Dict[str, Any] = {}
    if isinstance(_rt.get("max_tokens"), int):
        _fork_kwargs["max_tokens"] = _rt["max_tokens"]
    if isinstance(_rt.get("command"), str) and _rt["command"]:
        _fork_kwargs["acp_command"] = _rt["command"]
        _fork_kwargs["acp_args"] = _rt.get("args") or []
    # Match parent's reasoning config so the fork's ``thinking`` /
    # ``output_config`` are byte-identical in the request body —
    # Anthropic's cache key is namespaced by ``thinking`` presence.
    # Same-model path only: when routed to a different aux model the
    # cache is cold regardless (parity buys nothing) and the parent's
    # effort vocabulary may not be valid for the routed model/provider
    # (e.g. OpenRouter ``extra_body.reasoning.effort`` is forwarded
    # unclamped; codex_responses passes ``max``/``ultra`` through
    # unmapped except on gpt-5.6/xAI). Let the routed fork use
    # provider defaults — matching the ``not _routed`` gate on
    # _cached_system_prompt below.
    if not _routed:
        _fork_kwargs["reasoning_config"] = getattr(agent, "reasoning_config", None)
        # Gateway session context is appended to the parent's cached
        # system prompt at API-call time through this field.  Preserve
        # it on same-model forks so the complete effective system
        # prompt remains byte-identical and can reuse the warm prefix.
        _fork_kwargs["ephemeral_system_prompt"] = getattr(
            agent, "ephemeral_system_prompt", None
        )
        # Prefill messages are inserted immediately after the system
        # message at API-call time (chat_completion_helpers.py /
        # conversation_loop.py), so a parent with prefill configured
        # (gateway prefill_messages_file) would otherwise diverge
        # from the warm prefix at message index 1 — same bug class
        # as the ephemeral prompt above, one position later.
        # Deep copy: the unicode-error recovery path mutates
        # prefill entries IN PLACE (_sanitize_messages_surrogates
        # via conversation_loop), so sharing dicts would let a
        # fork-side sanitize rewrite the parent's prefill bytes.
        _parent_prefill = copy.deepcopy(
            getattr(agent, "prefill_messages", None) or []
        )
        if _parent_prefill:
            _fork_kwargs["prefill_messages"] = _parent_prefill
        # OpenRouter provider-routing pins: prompt caches live per
        # UPSTREAM provider, so a fork without the parent's pins can
        # be routed to a different upstream and miss the warm cache
        # even with byte-identical prompt/tools bytes.
        for _pref_attr in (
            "providers_allowed",
            "providers_ignored",
            "providers_order",
            "provider_sort",
            "provider_require_parameters",
            "provider_data_collection",
        ):
            _pref_val = getattr(agent, _pref_attr, None)
            if _pref_val:
                _fork_kwargs[_pref_attr] = _pref_val
    review_agent = AIAgent(
        model=_rt.get("model") or agent.model,
        max_iterations=max_iterations,
        quiet_mode=True,
        platform=agent.platform,
        provider=_rt.get("provider") or agent.provider,
        api_mode=_rt.get("api_mode"),
        base_url=_rt.get("base_url") or None,
        api_key=_rt.get("api_key") or None,
        credential_pool=_rt.get("credential_pool"),
        request_overrides=_rt.get("request_overrides") or {},
        parent_session_id=agent.session_id,
        enabled_toolsets=getattr(agent, "enabled_toolsets", None),
        disabled_toolsets=getattr(agent, "disabled_toolsets", None),
        skip_memory=True,
        **_fork_kwargs,
    )
    review_agent._memory_write_origin = write_origin
    review_agent._memory_write_context = write_origin
    # The review fork pins the parent's cached system prompt and keeps
    # ``tools[]`` byte-identical to the parent so its outbound request
    # hits the same provider cache prefix (see the toolset-parity note
    # above). The between-turns MCP refresh in build_turn_context would
    # add late-connecting MCP tools to this fork and break that parity,
    # so opt the review fork out of it.
    review_agent._skip_mcp_refresh = True
    review_agent._memory_store = agent._memory_store
    review_agent._memory_enabled = agent._memory_enabled
    review_agent._user_profile_enabled = agent._user_profile_enabled
    review_agent._memory_nudge_interval = 0
    review_agent._skill_nudge_interval = 0
    # PERSISTENCE ISOLATION (the curator-takeover root cause): the fork
    # shares the parent's session_id (set below, for prompt-cache
    # warmth), so without this it would write its harness turn ("Review
    # the conversation above and update the skill library…") + its own
    # response straight into the user's REAL session in state.db. On the
    # user's next live turn the agent re-reads that injected user message
    # as a standing instruction and "becomes" the curator, refusing the
    # actual task. _persist_disabled hard-stops every DB write/lazy-open
    # path (_flush_messages_to_session_db, _ensure_db_session,
    # _get_session_db_for_recall); the review writes only to the skill
    # and memory stores via its tools, which is all it needs.
    review_agent._persist_disabled = True
    review_agent._session_db = None
    review_agent._session_json_enabled = False
    # Suppress all status/warning emits from the fork so the
    # user only sees the final successful-action summary.
    # Without this, mid-review "Iteration budget exhausted",
    # rate-limit retries, compression warnings, and other
    # lifecycle messages bubble up through _emit_status ->
    # _vprint and leak past the stdout redirect (they go via
    # _print_fn/status_callback, which bypass sys.stdout).
    review_agent.suppress_status_output = True
    # Inherit the parent's cached system prompt verbatim so
    # the review fork's outbound HTTP request hits the same
    # Anthropic/OpenRouter prefix cache the parent warmed.
    # Without this, the fork rebuilds the system prompt from
    # scratch (fresh _hermes_now() timestamp, fresh
    # session_id, narrower toolset → different skills_prompt)
    # and the byte-exact prefix-cache key misses. See
    # issue #25322 and PR #17276 for the full analysis +
    # measured impact (~26% end-to-end cost reduction on
    # Sonnet 4.5).
    # Share the parent's warm cached system prompt ONLY when the review
    # runs on the SAME model (not routed). When routed to a different
    # model the parent's cached prompt is for the wrong model/cache key
    # and would miss anyway, so let the routed fork build its own.
    if not _routed:
        review_agent._cached_system_prompt = agent._cached_system_prompt
        # Defensive: pin session_start + session_id to the
        # parent's so any code path that re-renders parts of
        # the system prompt (compression, plugin hooks) still
        # produces byte-identical output. The cached-prompt
        # assignment above already short-circuits the normal
        # rebuild path, but these pins guarantee parity even
        # if a future code path bypasses the cache.
        review_agent.session_start = agent.session_start
    review_agent.session_id = agent.session_id
    # The fork shares the parent's live session_id (pinned above for
    # prefix-cache parity). It is single-lifecycle and calls close()
    # right after this run_conversation(); without opting out, close()
    # would finalize the parent's still-active session row mid
    # conversation (the review fires every ~10 turns). Leave session
    # finalization to the real owner (CLI close / gateway reset / cron).
    review_agent._end_session_on_close = False
    # DETACHED IN-MEMORY COMPACTION (issue #93057). The fork shares
    # the parent's session_id (pinned above for prefix-cache parity),
    # so the historical guard here was ``compression_enabled = False``:
    # if the fork ran the ordinary compression path it could rotate /
    # archive the parent's live session — the sibling-session race
    # behind #38727. But disabling compaction was a proxy for
    # detachment, and it removed the ONLY bound on the review's
    # private snapshot: as the review performs tool calls, every
    # follow-up provider request replayed the snapshot plus the
    # growing review tool loop (350k-384k input tokens per request in
    # production, 1.49M total across one 8-request review).
    #
    # The fix is detachment, not disablement:
    #   • Persistence is already off above (_persist_disabled /
    #     _session_db=None), so the commit site in compress_context
    #     (``if agent._session_db:``) skips every durable write and
    #     compaction can only ever rewrite the fork's private
    #     in-memory transcript.
    #   • The compressor's OWN session binding still needs severing:
    #     AIAgent.__init__ bound it to the parent's SessionDB and
    #     session_id before this function nulled the agent-level
    #     binding, so durable cooldown/streak/ineffective-count
    #     writes would otherwise land on the parent's row. Rebinding
    #     with session_db=None / session_id="" makes every
    #     compressor persist guard a no-op.
    #   • Force in-place mode (never rotation) even if the parent's
    #     config selected rotation, and re-enable compression ONLY
    #     after the rebind succeeds (fail-closed — see below). While
    #     enabled, both compression gates stay deferred until the
    #     fork's first provider response so request #1 replays the
    #     full snapshot as a warm cache read.
    _review_compressor = getattr(review_agent, "context_compressor", None)
    _bind_review_compressor = getattr(
        _review_compressor, "bind_session_state", None
    )
    _review_compression_detached = False
    if callable(_bind_review_compressor):
        try:
            # Plugin/third-party context engines may not accept these
            # kwargs; they own their own persistence policy, so a
            # failed rebind leaves the pre-existing flags in place
            # and must never abort the review (same tolerance as the
            # init-time binding in agent_init.py).
            _bind_review_compressor(session_db=None, session_id="")
            _review_compression_detached = True
        except Exception:
            # FAIL-CLOSED (adversarial review, #93057): if the rebind
            # could not sever the engine's session binding, the
            # compressor may still point at the parent's
            # SessionDB/session_id. Enabling compression in that
            # state would let durable cooldown/streak/ineffective-
            # count writes land on the parent's row and re-open the
            # #38727 sibling race. Keep the historical
            # compression_enabled=False behavior instead and warn;
            # the review still runs, bounded by the iteration cap
            # and the aggregate input budget below.
            logger.warning(
                "background-review compressor detachment failed; "
                "keeping compression DISABLED on this review fork "
                "(fail-closed, issue #93057 / #38727)",
                exc_info=True,
            )
    # Force in-place mode (never rotation) even if the parent's
    # config selected rotation. Re-enable compression ONLY after the
    # compressor's session binding was successfully severed; an
    # engine without a bind hook keeps the historical disabled
    # behavior as well.
    review_agent.compression_in_place = True
    review_agent.compression_enabled = _review_compression_detached
    if _review_compression_detached:
        # Warm-cache parity: the fork's FIRST provider request
        # replays the parent's full snapshot as a warm prompt-cache
        # read, so compaction must not rewrite the snapshot before
        # that first request goes out. Defer both compression gates
        # until the first provider response arrives (see
        # _review_fork_first_request_pending in agent/turn_context.py
        # and the pre-API gate in agent/conversation_loop.py); from
        # the second request on, the fork's transcript is its own and
        # compaction bounds it.
        review_agent._review_defer_compaction_before_first_response = True
    # Aggregate input budget: compaction bounds any single request;
    # this bounds the WHOLE review. Iterations are already capped by
    # _REVIEW_MAX_ITERATIONS. Checked in agent/conversation_loop.py
    # via _review_input_budget_exhausted (issue #93057).
    review_agent._review_input_token_budget = _review_input_token_budget(
        task_cfg
    )
    return review_agent, _rt, _routed


def _run_review_in_thread(
    agent: Any,
    messages_snapshot: List[Dict],
    prompt: str,
    task_cfg: Optional[Dict[str, Any]] = None,
    review_run: Optional[_BackgroundReviewRun] = None,
) -> None:
    """Worker function executed in the background-review daemon thread.

    Spawns a forked ``AIAgent`` inheriting the parent's runtime, runs the
    review prompt, and surfaces a compact action summary back to the user
    via ``agent._safe_print`` and ``agent.background_review_callback``.

    ``review_run`` is the per-review cancellation token from
    :func:`prepare_background_review_run`.  If a live turn bumps the
    cancel generation before this review reaches its first provider call,
    the review aborts without entering ``run_conversation()`` (#84423).
    """
    if review_run is not None and review_run.cancel_requested.is_set():
        finish_background_review_run(agent, review_run)
        return

    # Local import to avoid a hard circular dep at module load.
    from run_agent import AIAgent
    from tools.terminal_tool import set_approval_callback as _set_approval_callback

    # Install a non-interactive approval callback on this worker
    # thread so any dangerous-command guard the review agent trips
    # resolves to "deny" instead of falling back to input() -- which
    # deadlocks against the parent's prompt_toolkit TUI (#15216).
    # Same pattern as _subagent_auto_deny in tools/delegate_tool.py.
    def _bg_review_auto_deny(command, description, **kwargs):
        logger.warning(
            "Background review auto-denied dangerous command: %s (%s)",
            command, description,
        )
        return "deny"
    try:
        _set_approval_callback(_bg_review_auto_deny)
    except Exception:
        pass

    # An agent-as-provider whose client can't carry Hermes tool calls back would
    # produce a fork that spawns a whole agent and then cannot write anything.
    # Don't spawn it — point at the override that does work. Checked BEFORE the
    # thread-scoped silence below so the warning is not swallowed, and
    # cheap-check-first so the normal path never resolves the runtime twice.
    # Fixes the class, not one provider: any future agent-as-provider client
    # inherits the guard.
    if not _parent_can_emit_tool_calls(agent) and not bool(
        _resolve_review_runtime(agent, task_cfg).get("routed")
    ):
        logger.warning(
            "Background review skipped: provider %r cannot emit Hermes tool calls, "
            "so the review fork could not write memories or skills. Set "
            "auxiliary.background_review.{provider,model} to route the review to "
            "a normal model.",
            getattr(agent, "provider", "?"),
        )
        try:
            _set_approval_callback(None)
        except Exception:
            pass
        return

    review_agent = None
    review_messages: List[Dict] = []
    review_usage: Dict[str, Any] = {}

    def _unregister_review_agent(agent_ref) -> None:
        """Idempotent: clears the review fork from both tracking slots.
        Called from the run_conversation finally and the outer safety-net finally.
        """
        if agent_ref is None:
            return
        if hasattr(agent, "_background_review_agent"):
            _br_lock = getattr(agent, "_background_review_lock", None)
            if _br_lock is not None:
                with _br_lock:
                    if agent._background_review_agent is agent_ref:
                        agent._background_review_agent = None
            elif agent._background_review_agent is agent_ref:
                agent._background_review_agent = None
        if hasattr(agent, "_active_children"):
            try:
                _ac_lock = getattr(agent, "_active_children_lock", None)
                if _ac_lock is not None:
                    with _ac_lock:
                        agent._active_children.remove(agent_ref)
                else:
                    agent._active_children.remove(agent_ref)
            except (ValueError, AttributeError):
                pass

    def _finish_request_phase(agent_ref) -> None:
        _unregister_review_agent(agent_ref)
        finish_background_review_run(agent, review_run)

    try:
        # Silence stdout/stderr for THIS worker thread only.  A process-global
        # ``contextlib.redirect_stdout(devnull)`` here would also blank
        # ``sys.stdout``/``sys.stderr`` for every other thread — including a
        # gateway event-loop thread driving a Telegram long-poll — for the full
        # duration of the review (tens of seconds), swallowing their console
        # output (#55769 / #55925).  ``thread_scoped_silence`` routes only this
        # thread's writes to devnull and leaves all other threads on the real
        # streams.
        with thread_scoped_silence():
            review_agent, _rt, _routed = build_cache_parity_fork(
                agent, task_cfg, max_iterations=_REVIEW_MAX_ITERATIONS
            )

            # Register this fork on the PARENT's _active_children (the same
            # list interrupt() fans out to for subagent delegation) and
            # _background_review_agent (a direct pointer the next live turn
            # uses to interrupt an admitted request). The per-review run token
            # separately fences startup and acknowledges request-phase exit.
            # The legacy pointer/list remain best-effort for direct test stubs;
            # a prepared run token is the live-turn cancellation authority.
            if hasattr(agent, "_background_review_agent"):
                _br_lock = getattr(agent, "_background_review_lock", None)
                if _br_lock is not None:
                    with _br_lock:
                        agent._background_review_agent = review_agent
                else:
                    agent._background_review_agent = review_agent
            if hasattr(agent, "_active_children"):
                _ac_lock = getattr(agent, "_active_children_lock", None)
                if _ac_lock is not None:
                    with _ac_lock:
                        agent._active_children.append(review_agent)
                else:
                    agent._active_children.append(review_agent)

            from model_tools import get_tool_definitions
            from hermes_cli.plugins import (
                set_thread_tool_whitelist,
                clear_thread_tool_whitelist,
            )

            # Gate the built-in memory tool on the profile's memory_enabled flag.
            # Hardcoding ["memory", "skills"] granted the review LLM the MEMORY.md
            # read/write tool even when a profile set memory_enabled: false,
            # contaminating a memory-disabled profile (#54937 layer 2).
            review_toolsets = ["skills"]
            if review_agent._memory_enabled or review_agent._user_profile_enabled:
                review_toolsets.insert(0, "memory")
            review_whitelist = {
                t["function"]["name"]
                for t in get_tool_definitions(
                    enabled_toolsets=review_toolsets,
                    quiet_mode=True,
                )
            }
            # Read-only file tools are whitelisted too (#61521, #39996): the
            # model naturally reaches for read_file/search_files to inspect a
            # skill before patching it. Denying them caused a per-review
            # denial storm (~142 denials + ~204 read-before-write refusals
            # over 2 days on one deployment) that starved the self-improvement
            # loop — the model never loaded SKILL.md the way the
            # read-before-write guard requires, so almost no patch landed.
            # This is a DISPATCH-side change only: the advertised ``tools[]``
            # stays byte-identical to the parent's, so prompt-cache parity is
            # untouched. read_file registers the read with the
            # read-before-write guard (tools/file_tools.py), so a
            # read_file → skill_manage(patch) sequence now succeeds. Write
            # tools (write_file/patch/terminal) stay denied — autonomous
            # maintenance must go through skill_manage's validation, and the
            # deny message below names that substitute so one denial
            # redirects the model instead of a storm.
            review_whitelist |= {"read_file", "search_files"}
            # Profile-configured opt-in tools (#44672, salvage #82146 by
            # @BrinShadewater): ``auxiliary.background_review.extra_tools``
            # admits named parent tools to the review whitelist — e.g. a
            # human-gated proposal tool or a memory-provider write surface.
            # Default-empty; a listed tool must already exist in the parent's
            # inherited schema (the whitelist can only admit, never advertise),
            # and everything unlisted stays denied. Read from task_cfg (the
            # auxiliary.background_review block already loaded for this spawn)
            # so no extra config I/O happens per review.
            configured_extra_tools: set = set()
            try:
                _extra_raw = _background_review_task_config(task_cfg).get(
                    "extra_tools", []
                )
                if isinstance(_extra_raw, list):
                    configured_extra_tools = {
                        name.strip()
                        for name in _extra_raw
                        if isinstance(name, str) and name.strip()
                    }
                    review_whitelist |= configured_extra_tools
            except Exception:
                logger.debug(
                    "background_review extra_tools parse failed", exc_info=True
                )
            _extra_deny_note = (
                " Configured extra tools also allowed: "
                + ", ".join(sorted(configured_extra_tools)) + "."
                if configured_extra_tools
                else ""
            )
            set_thread_tool_whitelist(
                review_whitelist,
                deny_msg_fmt=(
                    "Background review denied non-whitelisted tool: "
                    "{tool_name}. Allowed here: skill_view/skills_list/"
                    "read_file/search_files to read, "
                    "skill_manage(action='patch'|...) to change skills, and "
                    "memory for notes." + _extra_deny_note
                    + " Do not retry {tool_name}."
                ),
            )
            try:
                from tools.skill_manager_tool import _reset_background_review_read_marks

                _reset_background_review_read_marks()
            except Exception:
                pass

            try:
                request_admitted = (
                    review_run is None or review_run.begin_request(review_agent)
                )
                if request_admitted:
                    # Routed to a different model -> replay a digest (cache is cold
                    # on that model anyway, so minimise cold-written tokens). Same
                    # model -> replay the full snapshot (warm cache reads).
                    _review_history = (
                        _digest_history(messages_snapshot) if _routed
                        else messages_snapshot
                    )
                    review_agent.run_conversation(
                        user_message=(
                            prompt
                            + "\n\nYou can only call memory and skill "
                            "management tools. Other tools will be denied "
                            "at runtime — do not attempt them."
                            + (
                                " Exception — these configured tools are "
                                "also allowed: "
                                + ", ".join(sorted(configured_extra_tools))
                                + "."
                                if configured_extra_tools
                                else ""
                            )
                        ),
                        conversation_history=_review_history,
                    )
            finally:
                clear_thread_tool_whitelist()
                # Attribute the review fork's usage to the PARENT session.
                # Snapshot BEFORE unregister/close so counters survive teardown.
                # Placed in this finally so a fork that consumed tokens and THEN
                # raised is still attributed (issue #87250). Best-effort: the
                # recorder never raises into the review thread.
                if review_agent is not None:
                    review_usage.update(_snapshot_review_usage(review_agent))
                    _record_review_usage_to_parent(agent, review_usage)
                # Publish completion as soon as the provider-capable phase has
                # returned or startup cancellation has fenced it out.
                _finish_request_phase(review_agent)

            # Snapshot review actions before teardown. close() is allowed to
            # clean per-session state, but the user-visible self-improvement
            # summary still needs the completed review agent's tool results.
            review_messages = list(getattr(review_agent, "_session_messages", []))

            # The fork shares the foreground session ID for prompt-cache
            # parity.  Do not call close() or shutdown_memory_provider():
            # both are session-bound lifecycle operations, and close() also
            # kills registered terminal processes and cleans environments for
            # that ID.  Releasing only this fork's clients leaves the live
            # session and its child processes untouched.
            try:
                review_agent.release_clients()
            except Exception:
                pass
            review_agent = None

        # Scan the review agent's messages for successful tool actions
        # and surface a compact summary to the user. Tool messages
        # already present in messages_snapshot must be skipped, since
        # the review agent inherits that history and would otherwise
        # re-surface stale "created"/"updated" messages from the prior
        # conversation as if they just happened (issue #14944).
        #
        # Wrapped in try/except: a buggy/legacy tool response shape
        # (e.g. ``_change`` returned as a list instead of a dict, #59437)
        # must NOT take down the whole review with an AttributeError,
        # since the caller's outer except logs only "Background
        # memory/skill review failed" and discards every successful
        # action the fork DID complete before the crash. Coerce an
        # exception into an empty actions list so the partial valid
        # actions from earlier in the messages are returned instead.
        try:
            actions = summarize_background_review_actions(
                review_messages,
                messages_snapshot,
                notification_mode=getattr(agent, "memory_notifications", "on"),
            )
        except Exception as e:
            logger.warning(
                "summarize_background_review_actions returned partial results "
                "after exception (treating as empty); suppressing AttributeError "
                "that previously aborted the entire review (#59437): %s",
                e,
            )
            actions = []

        _log_review_completion(
            review_usage, _classify_review_result(actions)
        )

        if actions:
            summary = " · ".join(dict.fromkeys(actions))
            agent._safe_print(
                f"  💾 Self-improvement review: {summary}"
            )
            _bg_cb = agent.background_review_callback
            if _bg_cb:
                try:
                    _bg_cb(
                        f"💾 Self-improvement review: {summary}"
                    )
                except Exception:
                    pass

    except Exception as e:
        logger.warning("Background memory/skill review failed: %s", e)
        if review_usage:
            _log_review_completion(review_usage, "error")
        agent._emit_auxiliary_failure("background review", e)
    finally:
        # Safety-net cleanup for the exception path. Normal completion already
        # released its clients inside the thread-scoped silence above. Re-enter
        # the thread-scoped silence here so exception-path cleanup output stays
        # quiet without blanking other threads' streams.
        # Also a safety-net completion: covers exceptions raised during setup
        # before the request-phase finally. Both tracking cleanup and the
        # per-run completion publication are identity-scoped and idempotent.
        _finish_request_phase(review_agent)
        if review_agent is not None:
            try:
                with thread_scoped_silence():
                    try:
                        review_agent.release_clients()
                    except Exception:
                        pass
            except Exception:
                pass
        # Clear the approval callback on this bg-review thread so a
        # recycled thread-id doesn't inherit a stale reference.
        try:
            _set_approval_callback(None)
        except Exception:
            pass


def spawn_background_review_thread(
    agent: Any,
    messages_snapshot: List[Dict],
    review_memory: bool = False,
    review_skills: bool = False,
    focus: Optional[str] = None,
    task_cfg: Optional[Dict[str, Any]] = None,
    review_run: Optional[_BackgroundReviewRun] = None,
):
    """Build the review thread target and prompt for a background review.

    Returns a ``(target, prompt)`` tuple.  The caller (``AIAgent._spawn_background_review``)
    owns the actual ``threading.Thread`` construction so test-level patches
    of ``run_agent.threading.Thread`` keep working.

    ``focus`` is optional user steering (the ``/refine [instructions]``
    path): appended to the chosen review prompt so the fork prioritizes what
    the user asked for while keeping the same guardrails. Automatic
    post-turn reviews pass ``None`` — their prompts are byte-identical to
    before this parameter existed.

    ``task_cfg`` is the already-loaded ``auxiliary.background_review`` block
    from :func:`load_background_review_settings`. When omitted, config is
    read once here and shared with the worker (aux routing) so a single
    turn does not re-parse the config file.
    """
    if task_cfg is None:
        task_cfg = _background_review_task_config()
    # Pick the right prompt based on which triggers fired.  Allow per-agent
    # override (the prompts moved to module-level constants but old code paths
    # that set agent._MEMORY_REVIEW_PROMPT etc. directly keep working).
    if review_memory and review_skills:
        prompt = getattr(agent, "_COMBINED_REVIEW_PROMPT", _COMBINED_REVIEW_PROMPT)
    elif review_memory:
        prompt = getattr(agent, "_MEMORY_REVIEW_PROMPT", _MEMORY_REVIEW_PROMPT)
    else:
        prompt = getattr(agent, "_SKILL_REVIEW_PROMPT", _SKILL_REVIEW_PROMPT)

    focus = (focus or "").strip()
    if focus:
        prompt = (
            f"{prompt}\n\n"
            f"The user explicitly requested this review with the following "
            f"focus — prioritize it over the general instructions above:\n"
            f"{focus}"
        )

    def _target() -> None:
        _run_review_in_thread(
            agent,
            messages_snapshot,
            prompt,
            task_cfg=task_cfg,
            review_run=review_run,
        )

    return _target, prompt


__all__ = [
    "_MEMORY_REVIEW_PROMPT",
    "_SKILL_REVIEW_PROMPT",
    "_COMBINED_REVIEW_PROMPT",
    "is_background_review_enabled",
    "load_background_review_settings",
    "spawn_background_review_thread",
    "summarize_background_review_actions",
    "build_memory_write_metadata",
]