fix(agent): defer background review during TUI turns

agent/agent_init.py

@@ -773,6 +773,14 @@ def init_agent(
     agent._client_lock = threading.RLock()
     agent._model_request_active = threading.Event()
     agent._supports_active_turn_redirect = True
+    agent._foreground_turn_generation = 0
+    agent._background_review_idle_delay_seconds = 60.0 if platform == "tui" else 0.0
+    try:
+        _bg_idle_raw = os.getenv("HERMES_BACKGROUND_REVIEW_IDLE_DELAY_SECONDS")
+        if _bg_idle_raw not in (None, ""):
+            agent._background_review_idle_delay_seconds = max(0.0, float(_bg_idle_raw))
+    except Exception:
+        pass
 
     # /steer mechanism — inject a user note into the next tool result
     # without interrupting the agent. Unlike interrupt(), steer() does

agent/background_review.py

@@ -21,6 +21,7 @@
 import json
 import logging
 import os
+import time
 from typing import Any, Dict, List, Optional
 
 from agent.thread_scoped_output import thread_scoped_silence
@@ -636,6 +637,8 @@ def _run_review_in_thread(
     agent: Any,
     messages_snapshot: List[Dict],
     prompt: str,
+    foreground_generation: Optional[int] = None,
+    idle_delay_seconds: float = 0.0,
 ) -> None:
     """Worker function executed in the background-review daemon thread.
 
@@ -647,6 +650,19 @@ def _run_review_in_thread(
     from run_agent import AIAgent
     from tools.terminal_tool import set_approval_callback as _set_approval_callback
 
+    if idle_delay_seconds > 0:
+        time.sleep(idle_delay_seconds)
+        current_generation = getattr(agent, "_foreground_turn_generation", None)
+        if (
+            foreground_generation is not None
+            and current_generation is not None
+            and current_generation != foreground_generation
+        ):
+            logger.info(
+                "Background review skipped: foreground turn resumed during idle delay"
+            )
+            return
+
     # 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
@@ -866,6 +882,16 @@ def _bg_review_auto_deny(command, description, **kwargs):
                     _digest_history(messages_snapshot) if _routed
                     else messages_snapshot
                 )
+                current_generation = getattr(agent, "_foreground_turn_generation", None)
+                if (
+                    foreground_generation is not None
+                    and current_generation is not None
+                    and current_generation != foreground_generation
+                ):
+                    logger.info(
+                        "Background review skipped: foreground turn resumed before model call"
+                    )
+                    return
                 review_agent.run_conversation(
                     user_message=(
                         prompt
@@ -976,6 +1002,8 @@ def spawn_background_review_thread(
     messages_snapshot: List[Dict],
     review_memory: bool = False,
     review_skills: bool = False,
+    foreground_generation: Optional[int] = None,
+    idle_delay_seconds: float = 0.0,
 ):
     """Build the review thread target and prompt for a background review.
 
@@ -994,7 +1022,13 @@ def spawn_background_review_thread(
         prompt = getattr(agent, "_SKILL_REVIEW_PROMPT", _SKILL_REVIEW_PROMPT)
 
     def _target() -> None:
-        _run_review_in_thread(agent, messages_snapshot, prompt)
+        _run_review_in_thread(
+            agent,
+            messages_snapshot,
+            prompt,
+            foreground_generation=foreground_generation,
+            idle_delay_seconds=idle_delay_seconds,
+        )
 
     return _target, prompt
 

agent/turn_context.py

@@ -500,6 +500,12 @@ def build_turn_context(
         agent.platform or "unknown", len(conversation_history or []),
         _msg_preview,
     )
+    try:
+        agent._foreground_turn_generation = (
+            int(getattr(agent, "_foreground_turn_generation", 0) or 0) + 1
+        )
+    except Exception:
+        agent._foreground_turn_generation = 1
 
     # Initialize conversation (copy to avoid mutating the caller's list).
     messages = list(conversation_history) if conversation_history else []

run_agent.py

@@ -1732,11 +1732,16 @@ def _spawn_background_review(
         """
         from agent.background_review import spawn_background_review_thread
         from tools.thread_context import propagate_context_to_thread
+        foreground_generation = int(getattr(self, "_foreground_turn_generation", 0) or 0)
         target, _prompt = spawn_background_review_thread(
             self,
             messages_snapshot,
             review_memory=review_memory,
             review_skills=review_skills,
+            foreground_generation=foreground_generation,
+            idle_delay_seconds=float(
+                getattr(self, "_background_review_idle_delay_seconds", 0.0) or 0.0
+            ),
         )
         # Carry the active profile into the review thread so MEMORY.md / skill
         # review writes land in the right profile (#54937).

tests/run_agent/test_background_review.py

@@ -120,6 +120,94 @@ def close(self):
     assert seen.get("at_run_time") is False
 
 
+def test_background_review_idle_delay_cancels_when_foreground_turn_resumes(monkeypatch):
+    """Interactive users can type immediately after a turn.
+
+    The delayed review fork must notice that a newer foreground turn started
+    during the idle window and exit before it spends model/GPU capacity.
+    """
+    import agent.background_review as bg_review
+
+    events = []
+
+    class FakeReviewAgent:
+        def __init__(self, **kwargs):
+            events.append(("init", kwargs))
+            self._session_messages = []
+
+        def run_conversation(self, **kwargs):
+            events.append(("run_conversation", kwargs))
+
+        def shutdown_memory_provider(self):
+            events.append(("shutdown_memory_provider", None))
+
+        def close(self):
+            events.append(("close", None))
+
+    agent = _bare_agent()
+    agent._foreground_turn_generation = 7
+    agent._background_review_idle_delay_seconds = 8.0
+
+    def fake_sleep(seconds):
+        events.append(("sleep", seconds))
+        agent._foreground_turn_generation += 1
+
+    monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
+    monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
+    monkeypatch.setattr(bg_review.time, "sleep", fake_sleep)
+
+    AIAgent._spawn_background_review(
+        agent,
+        messages_snapshot=[{"role": "user", "content": "hello"}],
+        review_memory=True,
+    )
+
+    assert events == [("sleep", 8.0)]
+
+
+def test_background_review_idle_delay_runs_when_foreground_stays_idle(monkeypatch):
+    """The idle gate is a defer/cancel guard, not a blanket disable."""
+    import agent.background_review as bg_review
+
+    events = []
+
+    class FakeReviewAgent:
+        def __init__(self, **kwargs):
+            events.append(("init", kwargs))
+            self._session_messages = []
+
+        def run_conversation(self, **kwargs):
+            events.append(("run_conversation", kwargs))
+
+        def shutdown_memory_provider(self):
+            events.append(("shutdown_memory_provider", None))
+
+        def close(self):
+            events.append(("close", None))
+
+    agent = _bare_agent()
+    agent._foreground_turn_generation = 3
+    agent._background_review_idle_delay_seconds = 8.0
+
+    monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
+    monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
+    monkeypatch.setattr(bg_review.time, "sleep", lambda seconds: events.append(("sleep", seconds)))
+
+    AIAgent._spawn_background_review(
+        agent,
+        messages_snapshot=[{"role": "user", "content": "hello"}],
+        review_memory=True,
+    )
+
+    assert [name for name, _payload in events] == [
+        "sleep",
+        "init",
+        "run_conversation",
+        "shutdown_memory_provider",
+        "close",
+    ]
+
+
 def test_background_review_summarizer_receives_captured_messages_after_close(monkeypatch):
     """The action summarizer must see review messages even after close cleanup.