fix(agent): attribute background-review usage and add cost controls

Persist fork token usage under session_model_usage task=background_review,
emit a per-fork completion log line, and expose enabled/max_iterations/
prompt_file so operators can see and bound the automatic review cost.

Address review feedback: load auxiliary.background_review once per spawn,
classify completion logs by summarize action prefixes, treat explicit
api_call_count=None as the documented default of 1, and WARNING on the
fail-open enabled-gate path.

agent/background_review.py

@@ -22,6 +22,7 @@ import copy
 import json
 import logging
 import os
+from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 from agent.thread_scoped_output import thread_scoped_silence
@@ -43,8 +44,141 @@ logger = logging.getLogger(__name__)
 # digest. That's the whole policy.
 # ---------------------------------------------------------------------------
 
+# Historical hardcoded value — kept as the default so existing behavior is
+# unchanged when the config key is unset.
+_DEFAULT_REVIEW_MAX_ITERATIONS = 16
 
-def _resolve_review_runtime(agent: Any) -> Dict[str, Any]:
+
+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 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 for prompt / max_iterations 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_max_iterations(
+    task_cfg: Optional[Dict[str, Any]] = None,
+) -> int:
+    """Resolve the review fork's tool-calling iteration budget.
+
+    Default 16 (unchanged historical behavior). Configurable via
+    ``auxiliary.background_review.max_iterations`` so operators can bound the
+    worst-case cost of a review that never concludes "nothing to save".
+    Clamped to [1, 64]; any config-read or parse failure falls back to the
+    default.
+    """
+    try:
+        val = _background_review_task_config(task_cfg).get("max_iterations")
+        if val is None:
+            return _DEFAULT_REVIEW_MAX_ITERATIONS
+        return max(1, min(int(val), 64))
+    except Exception:
+        return _DEFAULT_REVIEW_MAX_ITERATIONS
+
+
+def _load_review_prompt_file(
+    task_cfg: Optional[Dict[str, Any]] = None,
+) -> Optional[str]:
+    """Load ``auxiliary.background_review.prompt_file`` if configured.
+
+    Returns the file contents (stripped) or ``None`` when unset / unreadable.
+    Relative paths resolve against ``HERMES_HOME`` (same pattern as TTS
+    ``persona_prompt_file``).
+    """
+    raw = _background_review_task_config(task_cfg).get("prompt_file")
+    if not isinstance(raw, str) or not raw.strip():
+        return None
+    expanded = os.path.expandvars(raw.strip())
+    path = Path(expanded).expanduser()
+    if not path.is_absolute():
+        try:
+            from hermes_constants import get_hermes_home
+
+            path = get_hermes_home() / path
+        except Exception:
+            path = Path.cwd() / path
+    try:
+        text = path.read_text(encoding="utf-8").strip()
+    except (OSError, UnicodeDecodeError) as exc:
+        logger.warning(
+            "background_review.prompt_file unavailable at %s: %s", path, exc
+        )
+        return None
+    return text or None
+
+
+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
@@ -70,13 +204,7 @@ def _resolve_review_runtime(agent: Any) -> Dict[str, Any]:
         "args": list(getattr(agent, "acp_args", []) or []),
         "routed": False,
     }
-    try:
-        from hermes_cli.config import load_config_readonly
-        cfg = load_config_readonly()
-    except Exception:
-        return parent
-    aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {}
-    task = aux.get("background_review", {}) if isinstance(aux.get("background_review"), dict) else {}
+    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)
@@ -651,10 +779,140 @@ def build_memory_write_metadata(
     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 _run_review_in_thread(
     agent: Any,
     messages_snapshot: List[Dict],
     prompt: str,
+    task_cfg: Optional[Dict[str, Any]] = None,
 ) -> None:
     """Worker function executed in the background-review daemon thread.
 
@@ -684,6 +942,7 @@ def _run_review_in_thread(
 
     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.
@@ -733,7 +992,7 @@ def _run_review_in_thread(
             # 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)
+            _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,
@@ -811,7 +1070,7 @@ def _run_review_in_thread(
                         _fork_kwargs[_pref_attr] = _pref_val
             review_agent = AIAgent(
                 model=_rt.get("model") or agent.model,
-                max_iterations=16,
+                max_iterations=_resolve_review_max_iterations(task_cfg),
                 quiet_mode=True,
                 platform=agent.platform,
                 provider=_rt.get("provider") or agent.provider,
@@ -983,6 +1242,14 @@ def _run_review_in_thread(
                 )
             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)
                 # Unregister as soon as run_conversation() itself has
                 # returned — that's the only phase making outbound API
                 # calls, i.e. the only phase that can race the parent's
@@ -1039,6 +1306,10 @@ def _run_review_in_thread(
             )
             actions = []
 
+        _log_review_completion(
+            review_usage, _classify_review_result(actions)
+        )
+
         if actions:
             summary = " · ".join(dict.fromkeys(actions))
             agent._safe_print(
@@ -1055,6 +1326,8 @@ def _run_review_in_thread(
 
     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
@@ -1096,6 +1369,7 @@ def spawn_background_review_thread(
     review_memory: bool = False,
     review_skills: bool = False,
     focus: Optional[str] = None,
+    task_cfg: Optional[Dict[str, Any]] = None,
 ):
     """Build the review thread target and prompt for a background review.
 
@@ -1108,11 +1382,24 @@ def spawn_background_review_thread(
     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 (prompt file, max_iterations,
+    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:
+    # ``auxiliary.background_review.prompt_file`` replaces the built-in text
+    # when set — operators can make the reviewer conservative or redefine
+    # "worth saving" without patching core (issue #87250).
+    prompt_override = _load_review_prompt_file(task_cfg)
+    if prompt_override:
+        prompt = prompt_override
+    elif 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)
@@ -1129,7 +1416,7 @@ def spawn_background_review_thread(
         )
 
     def _target() -> None:
-        _run_review_in_thread(agent, messages_snapshot, prompt)
+        _run_review_in_thread(agent, messages_snapshot, prompt, task_cfg)
 
     return _target, prompt
 
@@ -1138,6 +1425,8 @@ __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",

cli-config.yaml.example

@@ -779,6 +779,16 @@ prompt_caching:
 #     provider: "auto"
 #     model: ""
 #     # max_concurrency: 2    # Optional: cap simultaneous compression calls
+#
+#   # Post-turn memory/skill self-improvement review fork. Runs after a turn
+#   # when the nudge intervals fire; writes skills/memories in a daemon thread.
+#   # Usage is recorded under session_model_usage task='background_review'.
+#   background_review:
+#     enabled: true            # false = skip automatic forks (/refine still works)
+#     provider: "auto"         # or pin a cheaper model (see memory.md)
+#     model: ""
+#     max_iterations: 16       # tool-calling budget per fork (clamped 1–64)
+#     prompt_file: ""          # optional custom review prompt (HERMES_HOME-relative)
 
 # =============================================================================
 # Persistent Memory

hermes_cli/config_defaults.py

@@ -1140,6 +1140,9 @@ DEFAULT_CONFIG = {
         # replay; different model = digest. Quality holds (memory capture
         # identical, skill near-identical in benchmarks).
         "background_review": {
+            # Master switch for automatic post-turn memory/skill review forks.
+            # false = skip automatic spawns (manual /refine still works).
+            "enabled": True,
             "provider": "auto",
             "model": "",
             "base_url": "",
@@ -1147,6 +1150,16 @@ DEFAULT_CONFIG = {
             "timeout": 120,
             "extra_body": {},
             "reasoning_effort": "",  # per-task thinking level: none|minimal|low|medium|high|xhigh|max|ultra (empty = provider default)
+            # Tool-calling iteration budget for the review fork. Default 16
+            # matches historical behavior. The review has no hard "nothing to
+            # do" early-out — a session with nothing worth saving can still
+            # burn the full budget. Lower this to bound worst-case cost;
+            # clamped to [1, 64].
+            "max_iterations": 16,
+            # Optional path to a custom review prompt (replaces the built-in
+            # memory/skill/combined prompts). Relative paths resolve under
+            # HERMES_HOME. Empty = use built-in prompts.
+            "prompt_file": "",
         },
         "moa_reference": {
             "provider": "auto",

hermes_state.py

@@ -7689,6 +7689,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         cache_write_tokens: int = 0,
         reasoning_tokens: int = 0,
         estimated_cost_usd: Optional[float] = None,
+        api_call_count: int = 1,
     ) -> None:
         """Record an auxiliary LLM call's usage against *session_id* (issue #23270).
 
@@ -7702,6 +7703,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         so folding aux tokens into the summary row would either be clobbered
         or double-counted. Insights/analytics read the union of both.
 
+        ``api_call_count`` defaults to 1 (one aux LLM call). Background-review
+        forks record an aggregate of N fork API calls in one write with
+        ``task='background_review'`` (issue #87250).
+
         Best-effort by contract: callers must never fail an aux call because
         accounting failed.
         """
@@ -7729,7 +7734,9 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 actual_cost_usd=None,
                 cost_status=None,
                 cost_source=None,
-                api_call_count=1,
+                api_call_count=(
+                    1 if api_call_count is None else int(api_call_count)
+                ),
                 task=task,
             )
         self._execute_write(_do)

run_agent.py

@@ -1829,6 +1829,17 @@ class AIAgent:
         # is a deliberate user request and still runs.
         if focus is None and getattr(self, "_delegate_depth", 0) > 0:
             return
+        # Explicit off-switch for automatic post-turn forks
+        # (``auxiliary.background_review.enabled: false``). Manual ``/refine``
+        # still works — same contract as zeroing the nudge intervals (#87250).
+        # Load the task block once here and pass it into the spawn path so
+        # prompt_file / max_iterations / aux routing do not re-read config.
+        task_cfg = None
+        if focus is None:
+            from agent.background_review import load_background_review_settings
+            enabled, task_cfg = load_background_review_settings()
+            if not enabled:
+                return
         from agent.background_review import spawn_background_review_thread
         from tools.thread_context import propagate_context_to_thread
         target, _prompt = spawn_background_review_thread(
@@ -1837,6 +1848,7 @@ class AIAgent:
             review_memory=review_memory,
             review_skills=review_skills,
             focus=focus,
+            task_cfg=task_cfg,
         )
         # Carry the active profile into the review thread so MEMORY.md / skill
         # review writes land in the right profile (#54937).

tests/agent/test_background_review_usage.py

@@ -0,0 +1,206 @@
+"""Background-review usage attribution (issue #87250).
+
+Background-review forks run with ``_session_db = None`` (persistence
+isolation), so their provider-billed API calls were never recorded in
+``session_model_usage``. ``_record_review_usage_to_parent`` closes that gap
+by snapshotting the fork's in-memory counters and recording them against the
+parent session via the aux-accounting chokepoint.
+"""
+
+from __future__ import annotations
+
+import logging
+from unittest.mock import patch
+
+import pytest
+
+from agent import background_review
+from hermes_state import SessionDB
+
+
+@pytest.fixture
+def db(tmp_path):
+    return SessionDB(tmp_path / "state.db")
+
+
+def _usage_rows(db, session_id):
+    with db._lock:
+        rows = db._conn.execute(
+            "SELECT * FROM session_model_usage WHERE session_id = ? ORDER BY task",
+            (session_id,),
+        ).fetchall()
+    return [dict(r) for r in rows]
+
+
+class _FakeParent:
+    def __init__(self, session_db, session_id="sess-parent"):
+        self._session_db = session_db
+        self.session_id = session_id
+
+
+def _usage(**overrides):
+    base = {
+        "model": "test-model",
+        "provider": "test-provider",
+        "base_url": "https://example.invalid/v1",
+        "input_tokens": 12000,
+        "output_tokens": 2400,
+        "cache_read_tokens": 190000,
+        "cache_write_tokens": 0,
+        "reasoning_tokens": 0,
+        "api_calls": 5,
+        "estimated_cost_usd": 0.05,
+    }
+    base.update(overrides)
+    return base
+
+
+def test_records_fork_usage_against_parent_session(db):
+    db.create_session("sess-parent", source="cli")
+
+    background_review._record_review_usage_to_parent(_FakeParent(db), _usage())
+
+    rows = _usage_rows(db, "sess-parent")
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["task"] == "background_review"
+    assert r["model"] == "test-model"
+    assert r["billing_provider"] == "test-provider"
+    assert r["input_tokens"] == 12000
+    assert r["output_tokens"] == 2400
+    assert r["cache_read_tokens"] == 190000
+    assert r["api_call_count"] == 5
+    assert r.get("estimated_cost_usd") == 0.05
+
+
+def test_accumulates_repeated_forks_same_model(db):
+    db.create_session("sess-parent", source="cli")
+    parent = _FakeParent(db)
+
+    background_review._record_review_usage_to_parent(parent, _usage(api_calls=5))
+    background_review._record_review_usage_to_parent(parent, _usage(api_calls=7))
+
+    rows = _usage_rows(db, "sess-parent")
+    assert len(rows) == 1
+    assert rows[0]["input_tokens"] == 24000
+    assert rows[0]["api_call_count"] == 12
+
+
+def test_noop_when_fork_made_no_calls(db):
+    db.create_session("sess-parent", source="cli")
+
+    background_review._record_review_usage_to_parent(
+        _FakeParent(db),
+        _usage(
+            input_tokens=0,
+            output_tokens=0,
+            cache_read_tokens=0,
+            cache_write_tokens=0,
+            reasoning_tokens=0,
+            api_calls=0,
+        ),
+    )
+
+    assert _usage_rows(db, "sess-parent") == []
+
+
+def test_noop_when_parent_has_no_session_db():
+    background_review._record_review_usage_to_parent(_FakeParent(None), _usage())
+
+
+def test_noop_when_parent_has_no_session_id(db):
+    db.create_session("sess-parent", source="cli")
+
+    background_review._record_review_usage_to_parent(
+        _FakeParent(db, session_id=""), _usage()
+    )
+
+    assert _usage_rows(db, "sess-parent") == []
+
+
+def test_survives_accounting_failure():
+    class _BoomDB:
+        def record_auxiliary_usage(self, *args, **kwargs):
+            raise RuntimeError("simulated accounting failure")
+
+    background_review._record_review_usage_to_parent(_FakeParent(_BoomDB()), _usage())
+
+
+def test_classify_review_result():
+    assert background_review._classify_review_result([]) == "none"
+    assert background_review._classify_review_result(["Memory updated"]) == "memory"
+    assert background_review._classify_review_result(["Skill 'x' patched"]) == "skill"
+    assert (
+        background_review._classify_review_result(
+            ["Memory updated", "Skill 'x' created"]
+        )
+        == "skill+memory"
+    )
+    # Prefix-based — free-text "skill"/"memory" elsewhere must not misclassify.
+    assert (
+        background_review._classify_review_result(
+            ["Skipped: no skill worth saving"]
+        )
+        == "none"
+    )
+    assert (
+        background_review._classify_review_result(
+            ["📝 Skill 'deploy' patched: \"a\" → \"b\""]
+        )
+        == "skill"
+    )
+    assert (
+        background_review._classify_review_result(["User profile ➕ prefers terse"])
+        == "memory"
+    )
+
+
+def test_enabled_config_failure_logs_warning(caplog):
+    with patch(
+        "hermes_cli.config.load_config_readonly",
+        side_effect=RuntimeError("boom"),
+    ), caplog.at_level(logging.WARNING, logger="agent.background_review"):
+        assert background_review.is_background_review_enabled() is True
+    assert any(
+        "fail-open" in r.message.lower() or "leaving automatic" in r.message.lower()
+        for r in caplog.records
+    )
+
+
+def test_spawn_reuses_provided_task_cfg_without_rereading(tmp_path):
+    """One config load per spawn — prompt + max_iterations share task_cfg."""
+    prompt_path = tmp_path / "review.md"
+    prompt_path.write_text("Custom review prompt only.\n", encoding="utf-8")
+    task = {
+        "enabled": True,
+        "max_iterations": 3,
+        "prompt_file": str(prompt_path),
+    }
+    agent = type("A", (), {})()
+    with patch(
+        "hermes_cli.config.load_config_readonly",
+        side_effect=AssertionError("config must not be re-read when task_cfg is passed"),
+    ):
+        _target, prompt = background_review.spawn_background_review_thread(
+            agent,
+            messages_snapshot=[{"role": "user", "content": "hi"}],
+            review_skills=True,
+            task_cfg=task,
+        )
+        assert "Custom review prompt only" in prompt
+        assert background_review._resolve_review_max_iterations(task) == 3
+        assert callable(_target)
+
+
+def test_log_review_completion_emits_thread_tag(caplog):
+    with caplog.at_level(logging.INFO, logger="agent.background_review"):
+        background_review._log_review_completion(
+            _usage(api_calls=8, input_tokens=53000, output_tokens=400),
+            "skill",
+        )
+    assert any(
+        "thread=bg-review" in r.message
+        and "calls=8" in r.message
+        and "result=skill" in r.message
+        for r in caplog.records
+    )

tests/hermes_state/test_aux_usage_accounting.py

@@ -67,7 +67,20 @@ class TestRecordAuxiliaryUsage:
         assert rows[0]["input_tokens"] == 3000
         assert rows[0]["api_call_count"] == 3
 
-
+    def test_explicit_none_api_call_count_uses_default_one(self, db):
+        """Explicit None must match the documented default of 1, not become 0."""
+        db.create_session("s1", source="cli")
+        db.record_auxiliary_usage(
+            "s1",
+            "vision",
+            model="gemini-3-flash",
+            input_tokens=10,
+            output_tokens=1,
+            api_call_count=None,
+        )
+        rows = _usage_rows(db, "s1")
+        assert len(rows) == 1
+        assert rows[0]["api_call_count"] == 1
 
     def test_main_loop_and_aux_rows_coexist(self, db):
         db.create_session("s1", source="cli")

tests/run_agent/test_background_review.py

@@ -198,6 +198,88 @@ def test_background_review_runs_at_top_level(monkeypatch):
     assert len(forks) == 1, "top-level review must still spawn the fork"
 
 
+def test_background_review_honors_configured_max_iterations(monkeypatch):
+    """``auxiliary.background_review.max_iterations`` must reach the fork's
+    ``AIAgent(max_iterations=...)`` (#87250)."""
+    from unittest.mock import patch
+
+    forks = []
+
+    class FakeReviewAgent:
+        def __init__(self, **kwargs):
+            forks.append(kwargs)
+
+        def run_conversation(self, **kwargs):
+            pass
+
+        def shutdown_memory_provider(self):
+            pass
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
+    monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
+
+    agent = _bare_agent()
+    agent._delegate_depth = 0
+
+    cfg = {"auxiliary": {"background_review": {"max_iterations": 4}}}
+    with patch("hermes_cli.config.load_config_readonly", return_value=cfg):
+        AIAgent._spawn_background_review(
+            agent,
+            messages_snapshot=[{"role": "user", "content": "hello"}],
+            review_memory=True,
+        )
+
+    assert len(forks) == 1
+    assert forks[0]["max_iterations"] == 4
+
+
+def test_background_review_disabled_skips_automatic_spawn(monkeypatch):
+    """``auxiliary.background_review.enabled: false`` must skip automatic
+    post-turn forks while leaving ``/refine`` (focus set) working (#87250)."""
+    from unittest.mock import patch
+
+    forks = []
+
+    class FakeReviewAgent:
+        def __init__(self, **kwargs):
+            forks.append(kwargs)
+
+        def run_conversation(self, **kwargs):
+            pass
+
+        def shutdown_memory_provider(self):
+            pass
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
+    monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
+
+    agent = _bare_agent()
+    agent._delegate_depth = 0
+    cfg = {"auxiliary": {"background_review": {"enabled": False}}}
+
+    with patch("hermes_cli.config.load_config_readonly", return_value=cfg):
+        AIAgent._spawn_background_review(
+            agent,
+            messages_snapshot=[{"role": "user", "content": "hello"}],
+            review_memory=True,
+        )
+        assert forks == [], "automatic review must not spawn when disabled"
+
+        AIAgent._spawn_background_review(
+            agent,
+            messages_snapshot=[{"role": "user", "content": "hello"}],
+            review_memory=True,
+            focus="save the deploy workflow",
+        )
+        assert len(forks) == 1, "/refine must still run when enabled=false"
+
+
 def test_background_review_explicit_focus_runs_even_in_subagent(monkeypatch):
     """An explicit ``/refine`` (``focus`` set) is a deliberate user request and
     is honored regardless of depth — only the automatic post-turn review is

tests/run_agent/test_background_review_cost_controls.py

@@ -158,3 +158,65 @@ def test_digest_records_tool_names_in_arc():
     digest = out[0]["content"]
     assert "USER: do the thing" in digest
     assert "tools: skill_view, patch" in digest
+
+
+# ---------------------------------------------------------------------------
+# Cost / configurability controls (issue #87250)
+# ---------------------------------------------------------------------------
+
+def test_max_iterations_unset_falls_back_to_default():
+    with patch("hermes_cli.config.load_config_readonly", return_value={}):
+        assert br._resolve_review_max_iterations() == 16
+
+
+def test_max_iterations_honors_config_override():
+    cfg = {"auxiliary": {"background_review": {"max_iterations": 4}}}
+    with patch("hermes_cli.config.load_config_readonly", return_value=cfg):
+        assert br._resolve_review_max_iterations() == 4
+
+
+def test_max_iterations_clamped_to_upper_bound():
+    cfg = {"auxiliary": {"background_review": {"max_iterations": 999}}}
+    with patch("hermes_cli.config.load_config_readonly", return_value=cfg):
+        assert br._resolve_review_max_iterations() == 64
+
+
+def test_max_iterations_clamped_to_lower_bound():
+    cfg = {"auxiliary": {"background_review": {"max_iterations": 0}}}
+    with patch("hermes_cli.config.load_config_readonly", return_value=cfg):
+        assert br._resolve_review_max_iterations() == 1
+
+
+def test_max_iterations_falls_back_to_default_on_config_error():
+    with patch(
+        "hermes_cli.config.load_config_readonly",
+        side_effect=RuntimeError("boom"),
+    ):
+        assert br._resolve_review_max_iterations() == 16
+
+
+def test_enabled_defaults_true():
+    with patch("hermes_cli.config.load_config_readonly", return_value={}):
+        assert br.is_background_review_enabled() is True
+
+
+def test_enabled_false_disables_automatic_review():
+    cfg = {"auxiliary": {"background_review": {"enabled": False}}}
+    with patch("hermes_cli.config.load_config_readonly", return_value=cfg):
+        assert br.is_background_review_enabled() is False
+
+
+def test_prompt_file_overrides_builtin(tmp_path):
+    prompt_path = tmp_path / "review.md"
+    prompt_path.write_text("Be conservative. Prefer Nothing to save.\n", encoding="utf-8")
+    cfg = {"auxiliary": {"background_review": {"prompt_file": str(prompt_path)}}}
+    agent = _FakeAgent()
+    with patch("hermes_cli.config.load_config_readonly", return_value=cfg):
+        _target, prompt = br.spawn_background_review_thread(
+            agent,
+            messages_snapshot=[{"role": "user", "content": "hi"}],
+            review_skills=True,
+        )
+    assert "Be conservative" in prompt
+    assert "ACTIVE" not in prompt  # built-in skill prompt not used
+    assert callable(_target)

website/docs/user-guide/features/memory.md

@@ -317,6 +317,29 @@ identical and skill capture near-identical to the main-model review.
 Leave it at `auto` (or set it to your main model) and nothing changes — the
 review keeps running on the main model with the full warm-cache replay.
 
+### Cost controls (`enabled`, `max_iterations`, `prompt_file`)
+
+The review fork can burn a meaningful share of total tokens on busy hosts.
+Operators can bound or disable it without zeroing nudge intervals:
+
+```yaml
+auxiliary:
+  background_review:
+    enabled: true              # false = skip automatic post-turn forks
+    max_iterations: 16         # tool-calling budget per fork (clamped 1–64)
+    prompt_file: ""            # optional custom prompt (HERMES_HOME-relative)
+```
+
+| Key | Behaviour |
+|-----|-----------|
+| `enabled: false` | Automatic post-turn forks do not spawn. Manual `/refine` still works. |
+| `max_iterations` | Caps how many tool-calling iterations the fork may run (default `16`). |
+| `prompt_file` | When set to a readable Markdown/text file, replaces the built-in review prompt so you can make the reviewer conservative or redefine "worth saving." |
+
+Fork usage is persisted in `session_model_usage` with `task='background_review'`
+and a completion line is written to `agent.log`
+(`Background review complete: thread=bg-review calls=… in=… out=… result=…`).
+
 ## Controlling skill writes (`skills.write_approval`)
 
 Skills use the same on/off gate, but the review UX differs because a