fix(agent): tag background-review fork turns in logs and pin supersede interrupt scoping

Fork turns (background review, side questions) reuse the parent's session_id
and, when unrouted, the model, so their `conversation turn:` and `Turn ended:`
lines are indistinguishable from live turns. In #118693 this made the fork's
own superseded exit read as a killed foreground stream.

- build_cache_parity_fork stamps _turn_origin on fork agents
- _log_turn_exit and the conversation-turn line append origin=<write_origin>
  for fork turns only; live turns are byte-identical
- new invariant test: cancel_background_review_for_live_turn flags only the
  fork (parent instance flags and thread bit untouched) via the real
  InterruptControlMixin path
- new regression test: fork turn-exit lines carry origin=background_review
  (red on base)

Related to #118693

Co-authored-by: CommandCodeBot <noreply@commandcode.ai>

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

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 []

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(

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]