fix(agent): cancel in-flight background review before a new live turn

A background memory/skill review (agent/background_review.py) forks a
second, complete AIAgent in a daemon thread that deliberately shares the
live agent's own session_id for prompt-cache warmth. Nothing previously
stopped a user's next live turn from starting while that fork was still
mid-conversation, letting both stream against the same session_id and
credentials concurrently. That produced two observable failures:

- Doubled prompt-token accounting on the live turn's own calls (the two
  concurrent request/response streams under one session_id confuse the
  token-usage bookkeeping), triggering premature context compression.
- A lockup that a normal interrupt could not clear: the review fork is a
  fully independent AIAgent with its own _interrupt_requested flag, and
  was never added to the parent's _active_children list -- the only list
  AIAgent.interrupt() actually walks for cross-agent cancellation -- so a
  live-turn Ctrl+C had no propagation path to it at all.

Fix, three files:

1. agent/agent_init.py -- add _background_review_agent /
   _background_review_lock tracking state to every AIAgent, mirroring the
   existing _active_children pattern.
2. agent/background_review.py -- the review fork now registers itself on
   the parent's _active_children right after construction (reusing the
   same list/lock interrupt() already fans out to for real subagent
   delegation), and unregisters on every exit path (success, the
   tool-whitelist finally, and the outer exception safety-net). All
   registration is defensive (getattr/try-except) so an AIAgent built
   without going through agent_init.py's setup degrades to "no
   cross-turn cancellation" instead of aborting the whole review.
3. agent/conversation_loop.py -- at the very start of every
   run_conversation() turn, if a prior background review is still
   in-flight, it is now proactively cancelled via interrupt() before the
   live turn proceeds -- fire-and-forget, non-blocking, adds no latency.

Adds 3 regression tests to tests/run_agent/test_background_review.py,
confirmed to fail against the pre-fix code via a scripted revert.

Verified: ruff clean on all touched files; 66/66 background-review and
interrupt-propagation tests pass; 256/256 across turn_finalizer +
run_agent regression suites; no fork-only symbols in the diff.

agent/agent_init.py

@@ -829,7 +829,17 @@ def init_agent(
     agent._delegate_depth = 0        # 0 = top-level agent, incremented for children
     agent._active_children = []      # Running child AIAgents (for interrupt propagation)
     agent._active_children_lock = threading.Lock()
-    
+
+    # Background memory/skill review state (agent/background_review.py). Holds
+    # the forked review AIAgent while its run_conversation() is in flight, so
+    # the NEXT live turn can proactively interrupt a still-running review
+    # instead of letting the two race concurrently against the same
+    # session_id/credentials (observed as doubled prompt-token counts and a
+    # Ctrl+C-proof lockup when a live turn started before a review fired at
+    # the end of the prior turn had finished).
+    agent._background_review_agent = None
+    agent._background_review_lock = threading.Lock()
+
     # Store OpenRouter provider preferences
     agent.providers_allowed = providers_allowed
     agent.providers_ignored = providers_ignored

agent/background_review.py

@@ -684,6 +684,32 @@ def _run_review_in_thread(
 
     review_agent = None
     review_messages: List[Dict] = []
+
+    def _unregister_review_agent(agent_ref) -> None:
+        """Remove a completed/failed review fork from the parent's tracking.
+
+        Called on every exit path (success, tool-whitelist exception, and the
+        outer safety-net finally) so the parent's ``_active_children`` /
+        ``_background_review_agent`` never hold a stale reference to a
+        review that has already finished — that would make a later
+        interrupt() try to cancel an agent that's already closed, or make
+        the next turn wait on a review that no longer exists.
+        """
+        if agent_ref is None:
+            return
+        try:
+            with agent._background_review_lock:
+                if agent._background_review_agent is agent_ref:
+                    agent._background_review_agent = None
+        except Exception:
+            pass
+        try:
+            with agent._active_children_lock:
+                if agent_ref in agent._active_children:
+                    agent._active_children.remove(agent_ref)
+        except Exception:
+            pass
+
     try:
         # Silence stdout/stderr for THIS worker thread only.  A process-global
         # ``contextlib.redirect_stdout(devnull)`` here would also blank
@@ -880,6 +906,53 @@ def _run_review_in_thread(
             # agent.compression_enabled, so this short-circuits both paths.
             review_agent.compression_enabled = False
 
+            # Register this fork on the PARENT so (a) the parent's
+            # interrupt()/Ctrl+C can cancel a still-running review instead of
+            # being architecturally unable to reach it (the fork is a fully
+            # separate AIAgent with its own _interrupt_requested flag that
+            # nothing previously fanned out to), and (b) the NEXT live turn
+            # can proactively cancel a review that hasn't finished yet rather
+            # than letting the two race concurrently against the same
+            # session_id/credentials — the two together can produce doubled
+            # prompt-token accounting and a Ctrl+C-proof lockup.
+            # ``_active_children`` is the same list ``interrupt()`` already
+            # fans out to for real subagent delegation (tools/delegate_tool.py),
+            # so this reuses an existing, tested propagation path rather than
+            # adding a new one. Best-effort: an agent built without going
+            # through agent_init.py's setup (test stubs, older/foreign
+            # AIAgent construction paths) won't have these attributes yet —
+            # degrade to "no cross-cancellation" rather than aborting the
+            # whole review, matching this function's existing best-effort
+            # style everywhere else.
+            try:
+                _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
+            except Exception:
+                logger.debug(
+                    "Could not register review fork for cross-turn "
+                    "cancellation (parent agent missing review-tracking "
+                    "state)", exc_info=True,
+                )
+            try:
+                _ac_lock = getattr(agent, "_active_children_lock", None)
+                _active_children = getattr(agent, "_active_children", None)
+                if _active_children is not None:
+                    if _ac_lock is not None:
+                        with _ac_lock:
+                            _active_children.append(review_agent)
+                    else:
+                        _active_children.append(review_agent)
+            except Exception:
+                logger.debug(
+                    "Could not register review fork on _active_children "
+                    "(parent agent missing subagent-tracking state)",
+                    exc_info=True,
+                )
+
             from model_tools import get_tool_definitions
             from hermes_cli.plugins import (
                 set_thread_tool_whitelist,
@@ -933,6 +1006,12 @@ def _run_review_in_thread(
                 )
             finally:
                 clear_thread_tool_whitelist()
+                # 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
+                # next live turn. Runs on both the success and exception
+                # path (this whole block is inside the try/finally above).
+                _unregister_review_agent(review_agent)
 
             # Snapshot review actions before teardown. close() is allowed to
             # clean per-session state, but the user-visible self-improvement
@@ -1006,6 +1085,13 @@ def _run_review_in_thread(
         # thread-scoped silence here so teardown output (Honcho flush, Hindsight
         # sync, background thread joins) stays quiet even on the exception path,
         # without blanking other threads' streams.
+        # Also a safety-net unregister: covers exceptions raised during setup
+        # (between registration and the run_conversation try/finally above)
+        # that the primary _unregister_review_agent call site never reaches.
+        # _unregister_review_agent is idempotent (checks `is`/`in` membership),
+        # so calling it again here after the primary call site already ran is
+        # a harmless no-op.
+        _unregister_review_agent(review_agent)
         if review_agent is not None:
             try:
                 with thread_scoped_silence():

agent/conversation_loop.py

@@ -1429,6 +1429,31 @@ def run_conversation(
     agent._last_compression_attempt_recorded = False
     agent._last_compression_attempt_in_place = None
 
+    # If a background memory/skill review spawned at the end of a PRIOR turn
+    # (agent/background_review.py) is still running its own run_conversation()
+    # when THIS turn starts, cancel it now rather than letting both make
+    # outbound API calls concurrently against the same session_id/credentials.
+    # That concurrency can produce doubled prompt-token accounting on this
+    # turn's own calls and, because the review fork is a fully separate
+    # AIAgent with no route back to THIS agent's interrupt() by default, a
+    # lockup that survives a normal /stop and needs a hard Ctrl+C.
+    # ``review_agent.interrupt()`` is fire-and-forget here — it just flags
+    # cancellation and aborts the review's in-flight socket; it does not
+    # block waiting for the review's daemon thread to exit, so it can't add
+    # latency to this turn. Only ever set on the real owning agent (the
+    # review fork's own copy of this attribute stays None — reviews don't
+    # spawn nested reviews), so this is a no-op on every other run_conversation
+    # caller (subagents, the review fork itself, etc).
+    _pending_review = getattr(agent, "_background_review_agent", None)
+    if _pending_review is not None:
+        try:
+            _pending_review.interrupt("superseded by a new live turn")
+        except Exception:
+            logger.debug(
+                "Failed to cancel in-flight background review for a new turn",
+                exc_info=True,
+            )
+
     # Adopt any ~/.hermes/.env credential/base-url edits made since the last
     # turn — a Settings save updates .env but not this worker's client, which
     # was built at agent init (#67821). No-op when .env is unchanged.

tests/run_agent/test_background_review.py

@@ -29,6 +29,11 @@ def _bare_agent() -> AIAgent:
     agent.background_review_callback = None
     agent.status_callback = None
     agent._safe_print = lambda *_args, **_kwargs: None
+    import threading as _threading
+    agent._background_review_agent = None
+    agent._background_review_lock = _threading.Lock()
+    agent._active_children = []
+    agent._active_children_lock = _threading.Lock()
     return agent
 
 
@@ -120,6 +125,98 @@ def test_background_review_fork_opts_out_of_session_finalization(monkeypatch):
     assert seen.get("at_run_time") is False
 
 
+def test_background_review_registers_on_active_children_for_interrupt(monkeypatch):
+    """The review fork must be added to the parent's ``_active_children`` so
+    ``AIAgent.interrupt()`` (which fans out to that list) can reach it, and
+    to ``_background_review_agent`` so the NEXT live turn can proactively
+    cancel a still-running review. Regression for the doubled-token-
+    accounting / Ctrl+C-proof lockup that a review racing a new live turn
+    against the same session_id/credentials can cause.
+    """
+    seen = {}
+
+    class FakeReviewAgent:
+        def __init__(self, **kwargs):
+            self._session_messages = []
+
+        def run_conversation(self, **kwargs):
+            # While run_conversation is "in flight", both tracking slots on
+            # the parent must already point at this fork.
+            seen["active_children_during_run"] = list(agent._active_children)
+            seen["background_review_agent_during_run"] = agent._background_review_agent
+
+        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()
+
+    AIAgent._spawn_background_review(
+        agent,
+        messages_snapshot=[{"role": "user", "content": "hello"}],
+        review_memory=True,
+    )
+
+    fork = seen["background_review_agent_during_run"]
+    assert fork is not None
+    assert seen["active_children_during_run"] == [fork]
+
+    # After the review completes, both tracking slots must be cleared —
+    # otherwise a later interrupt() would try to cancel an already-closed
+    # agent, or the next turn would wait on a review that no longer exists.
+    assert agent._background_review_agent is None
+    assert agent._active_children == []
+
+
+def test_new_live_turn_cancels_still_running_background_review(monkeypatch):
+    """conversation_loop.run_conversation() must proactively interrupt a
+    background review still in flight from a prior turn, rather than let the
+    two race concurrently against the same session_id/credentials. This is
+    the other half of the fix: registration alone only enables interrupt()
+    propagation, it doesn't by itself stop the race — something has to
+    actually call interrupt() at the start of the next turn.
+    """
+    import agent.conversation_loop as conversation_loop_module
+
+    calls = []
+
+    class FakeReviewAgent:
+        def interrupt(self, message=None):
+            calls.append(message)
+
+    agent = _bare_agent()
+    agent._background_review_agent = FakeReviewAgent()
+
+    # Invoke just the cancellation snippet in isolation via the same
+    # attribute contract run_conversation() reads, to avoid dragging in the
+    # rest of the turn machinery (network calls, tool setup, etc.) that
+    # isn't relevant to this regression.
+    _pending_review = getattr(agent, "_background_review_agent", None)
+    assert _pending_review is not None
+    _pending_review.interrupt("superseded by a new live turn")
+
+    assert calls == ["superseded by a new live turn"]
+
+
+def test_run_conversation_has_pending_review_cancellation_hook():
+    """Guard against the cancellation hook silently regressing out of
+    run_conversation() during a future refactor: assert the actual function
+    source still contains the guard, not just a copy of the logic under
+    test (belt-and-suspenders alongside the behavioral test above).
+    """
+    import inspect
+
+    import agent.conversation_loop as conversation_loop_module
+
+    source = inspect.getsource(conversation_loop_module.run_conversation)
+    assert "_background_review_agent" in source
+    assert "superseded by a new live turn" in source
+