diff --git a/agent/background_review.py b/agent/background_review.py index 3a07658c64..f346767634 100644 --- a/agent/background_review.py +++ b/agent/background_review.py @@ -980,6 +980,10 @@ def build_cache_parity_fork( _warn_ignored_reasoning_effort(agent, task_cfg) review_agent = AIAgent(**_fork_init_kwargs(agent, _rt, _routed, max_iterations, task_cfg)) review_agent._memory_write_origin = review_agent._memory_write_context = write_origin + # Fork-turn log tag: the fork shares the parent's session_id (and model on the + # same-model path), so its turn-start/turn-exit log lines are otherwise + # indistinguishable from live turns (#118693). + review_agent._turn_origin = write_origin review_agent._memory_store = agent._memory_store review_agent._memory_enabled = agent._memory_enabled review_agent._user_profile_enabled = agent._user_profile_enabled diff --git a/agent/turn_context.py b/agent/turn_context.py index 2fc545ffd3..43f8ae03b4 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -1036,12 +1036,20 @@ def build_turn_context( _preview_text = summarize_user_message_for_log(user_message) _msg_preview = _preview_text[:80] + ("..." if len(_preview_text) > 80 else "") - logger.info( - "conversation turn: session=%s model=%s provider=%s platform=%s history=%d msg=%r", + _turn_fmt = ( + "conversation turn: session=%s model=%s provider=%s platform=%s history=%d msg=%r" + ) + _turn_args = [ agent.session_id or "none", agent.model, agent.provider or "unknown", agent.platform or "unknown", len(conversation_history or []), _msg_preview.replace("\n", " "), - ) + ] + # Fork turns (background review, side questions) reuse the parent's session_id and + # model; tag their turn-start line so logs can tell them apart (#118693). + if (_turn_origin := getattr(agent, "_turn_origin", None)): + _turn_fmt += " origin=%s" + _turn_args.append(_turn_origin) + logger.info(_turn_fmt, *_turn_args) # Copy so the caller's list is never mutated. messages = list(conversation_history) if conversation_history else [] diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index 3db49dd22f..c8460bb4d0 100644 --- a/agent/turn_finalizer.py +++ b/agent/turn_finalizer.py @@ -322,9 +322,14 @@ def _log_turn_exit(agent, messages, final_response, api_call_count, _turn_exit_r 1 for m in messages if isinstance(m, dict) and m.get("role") == "assistant" and m.get("tool_calls") ) + # Fork turns (background review, side questions) carry ``_turn_origin``; tagging the + # exit line keeps a fork's ``interrupted_during_api_call`` from reading as a killed + # foreground stream — the fork shares the parent's session_id and often its model (#118693). + _turn_origin = getattr(agent, "_turn_origin", None) _diag_msg = ( "Turn ended: reason=%s model=%s api_calls=%d/%d budget=%d/%d " "tool_turns=%d last_msg_role=%s response_len=%d session=%s" + + (" origin=%s" if _turn_origin else "") ) _diag_args = ( _turn_exit_reason, agent.model, api_call_count, agent.max_iterations, @@ -332,6 +337,7 @@ def _log_turn_exit(agent, messages, final_response, api_call_count, _turn_exit_r agent.iteration_budget.max_total if agent.iteration_budget else 0, _turn_tool_count, _last_msg_role, len(final_response) if final_response else 0, agent.session_id or "none", + *((_turn_origin,) if _turn_origin else ()), ) if _last_msg_role == "tool" and not interrupted: logger.warning( diff --git a/tests/agent/test_background_review.py b/tests/agent/test_background_review.py index eb05213dec..9f76606527 100644 --- a/tests/agent/test_background_review.py +++ b/tests/agent/test_background_review.py @@ -849,3 +849,150 @@ def test_skill_patch_off_silent_verbose_shows_diff(): ) assert len(verbose) == 1 assert "demo" in verbose[0] and "→" in verbose[0] + + +def _interrupt_scoped_agent() -> AIAgent: + """Bare AIAgent carrying the full interrupt-control surface so the REAL + ``InterruptControlMixin`` path runs (no stubbed ``interrupt()``).""" + agent = _bare_agent() + agent._interrupt_requested = False + agent._interrupt_message = None + agent._tool_interrupt_reason = None + agent._hard_interrupt_requested = threading.Event() + agent._execution_thread_id = None + agent._interrupt_thread_signal_pending = False + agent._pending_redirect = None + agent._pending_steer = None + agent._pending_redirect_lock = threading.Lock() + agent._pending_steer_lock = threading.Lock() + agent._tool_worker_threads = set() + agent._tool_worker_threads_lock = threading.Lock() + agent.quiet_mode = True + return agent + + +def test_supersede_hard_interrupt_targets_review_fork_only(): + """Regression for #118693: a live turn superseding the background review must + flag only the fork. The foreground agent's instance flags and its execution + thread's ``tools.interrupt`` bit must stay untouched, or a user-facing stream + mid-flight is exactly what gets killed.""" + import time + + from agent.background_review import ( + cancel_background_review_for_live_turn, + finish_background_review_run, + prepare_background_review_run, + ) + from tools.interrupt import ( + _interrupt_reasons, + _interrupted_threads, + is_thread_interrupted, + set_interrupt, + ) + + parent = _interrupt_scoped_agent() + fork = _interrupt_scoped_agent() + stop = threading.Event() + + def _hold(): + stop.wait(5.0) + + foreground_thread = threading.Thread(target=_hold, daemon=True) # live turn + review_thread = threading.Thread(target=_hold, daemon=True) # bg-review + foreground_thread.start() + review_thread.start() + parent._execution_thread_id = foreground_thread.ident + fork._execution_thread_id = review_thread.ident + parent._background_review_agent = None + + run = prepare_background_review_run(parent) + assert run is not None + assert run.begin_request(fork) + + # Production-shaped ack: the fork's turn unwinds only after its interrupt landed + # (the per-thread bit is the last signal interrupt() publishes). + def _ack_when_fork_interrupted(): + deadline = time.monotonic() + 1.5 + while time.monotonic() < deadline: + if is_thread_interrupted(fork._execution_thread_id): + break + time.sleep(0.01) + finish_background_review_run(parent, run) + + ack = threading.Thread(target=_ack_when_fork_interrupted, daemon=True) + ack.start() + + try: + cancel_background_review_for_live_turn(parent) + ack.join(timeout=3.0) + + # The fork owns the supersede. + assert fork._interrupt_requested is True + assert fork._tool_interrupt_reason == "background review superseded" + assert fork._hard_interrupt_requested.is_set() + assert is_thread_interrupted(fork._execution_thread_id) + assert ( + _interrupt_reasons.get(fork._execution_thread_id) + == "background review superseded" + ) + + # The foreground turn is untouched: no instance flags, no thread bit. + assert parent._interrupt_requested is False + assert parent._tool_interrupt_reason is None + assert not parent._hard_interrupt_requested.is_set() + assert parent._interrupt_thread_signal_pending is False + assert not is_thread_interrupted(foreground_thread.ident) + assert foreground_thread.ident not in _interrupted_threads + assert foreground_thread.ident not in _interrupt_reasons + finally: + set_interrupt(False, fork._execution_thread_id) + set_interrupt(False, foreground_thread.ident) + stop.set() + foreground_thread.join(timeout=2.0) + review_thread.join(timeout=2.0) + + +def test_turn_exit_log_carries_fork_origin_tag(): + """Regression for #118693: a fork's turn-exit line must carry its origin so a + background-review stream killed by supersede is never mistaken for a killed + foreground stream (the fork shares session_id and often the model).""" + import logging + + from agent.turn_finalizer import _log_turn_exit + + records = [] + + class _Capture(logging.Handler): + def emit(self, record): + records.append(record.getMessage()) + + logger = logging.getLogger("test_turn_exit_origin") + logger.addHandler(_Capture()) + logger.setLevel(logging.INFO) + try: + live = _bare_agent() + fork = _bare_agent() + fork._turn_origin = "background_review" + for agent in (live, fork): + agent.max_iterations = 10 + agent.iteration_budget = None + messages = [ + {"role": "user", "content": "q"}, + {"role": "assistant", "content": "a"}, + ] + _log_turn_exit(live, list(messages), "a", 1, "text_response", False, logger) + _log_turn_exit( + fork, + list(messages), + "a", + 1, + "interrupted_during_api_call(background_review_superseded)", + True, + logger, + ) + finally: + logger.handlers.clear() + + assert len(records) == 2 + assert "origin=" not in records[0], records[0] + assert "origin=background_review" in records[1], records[1]