diff --git a/agent/agent_init.py b/agent/agent_init.py index 428dc69c8b2df..797c18c6dfcdc 100644 --- a/agent/agent_init.py +++ b/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 diff --git a/agent/background_review.py b/agent/background_review.py index b5728f3fe8695..61de81aa3b448 100644 --- a/agent/background_review.py +++ b/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(): diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 40543644cc8a0..4d2f3eea4eca6 100644 --- a/agent/conversation_loop.py +++ b/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. diff --git a/tests/run_agent/test_background_review.py b/tests/run_agent/test_background_review.py index f92f8ac4b4d25..e32b486522f24 100644 --- a/tests/run_agent/test_background_review.py +++ b/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 +