From cbf5b05c704e3c6c3fb53b3dca2cbd6f0d1ff61e Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Mon, 13 Jul 2026 02:23:33 -0400 Subject: [PATCH] feat(agent): add active-turn redirect core primitive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A follow-up sent while the model is still generating previously ended the turn: Hermes kept only the visible partial text (reasoning was display-only), cleared the loop, and replayed the message as a fresh next turn. If the correction referred to something that only appeared in the thinking stream, the model no longer had that context. Add `AIAgent.redirect(text)`: a corrective interrupt distinct from a hard stop. It cancels only the in-flight model request (not tool workers or child agents), stashes the correction under a lock shared with `interrupt()` so a concurrent `/stop` always wins, and lets the loop rebuild the same logical iteration. `_apply_active_turn_redirect()` checkpoints the reasoning that was actually shown to the user plus any visible partial text as an ordinary assistant message, then appends the correction as a real user turn — never replaying incomplete signed/encrypted provider reasoning, and keeping strict role alternation and prompt-cache stability intact. During tool execution it degrades to `steer()` so a running tool finishes at a safe boundary. `_fire_reasoning_delta` now only records reasoning that a display callback actually consumed, so `show_reasoning: false` never leaks hidden provider thinking into the persisted transcript. --- agent/agent_init.py | 15 +++ agent/conversation_loop.py | 144 ++++++++++++++++++--- agent/turn_retry_state.py | 4 + run_agent.py | 179 ++++++++++++++++++++++++++- tests/agent/test_turn_retry_state.py | 1 + tests/run_agent/test_run_agent.py | 131 ++++++++++++++++++++ tests/run_agent/test_steer.py | 175 ++++++++++++++++++++++++++ 7 files changed, 627 insertions(+), 22 deletions(-) diff --git a/agent/agent_init.py b/agent/agent_init.py index 2094dd2b4f7f..a6d3b0481411 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -720,6 +720,8 @@ def init_agent( agent._execution_thread_id: int | None = None # Set at run_conversation() start agent._interrupt_thread_signal_pending = False agent._client_lock = threading.RLock() + agent._model_request_active = threading.Event() + agent._supports_active_turn_redirect = True # /steer mechanism — inject a user note into the next tool result # without interrupting the agent. Unlike interrupt(), steer() does @@ -731,6 +733,13 @@ def init_agent( agent._pending_steer: Optional[str] = None agent._pending_steer_lock = threading.Lock() + # Active-turn redirect mechanism. A regular follow-up sent while the model + # is generating is different from a hard /stop: preserve the valid turn + # prefix, cancel only the in-flight model request, and rebuild its tail with + # the correction. The loop drains this slot at a role-safe boundary. + agent._pending_redirect: Optional[str] = None + agent._pending_redirect_lock = threading.Lock() + # Concurrent-tool worker thread tracking. `_execute_tool_calls_concurrent` # runs each tool on its own ThreadPoolExecutor worker — those worker # threads have tids distinct from `_execution_thread_id`, so @@ -897,6 +906,12 @@ def init_agent( agent._stream_writer_tls = threading.local() agent._stream_writer_dropped = 0 + # Displayed reasoning text streamed during the current model response, + # captured only when a surface consumed it via a reasoning callback. Used + # by active-turn redirect to checkpoint what the user actually saw without + # ever persisting hidden provider reasoning. + agent._current_streamed_reasoning_text = "" + # Optional current-turn user-message override used when the API-facing # user message intentionally differs from the persisted transcript # (e.g. CLI voice mode adds a temporary prefix for the live call only). diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index df81c20be211..fb9d11b66fce 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -103,6 +103,53 @@ _API_CALL_MODULES = frozenset({ }) +def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text: str) -> None: + """Append a provider-safe checkpoint and correction to the live turn. + + Incomplete provider reasoning blocks are not valid replay items (Anthropic + signs them; Responses reasoning items require their following output). + Preserve only what Hermes actually displayed, demoted to ordinary text, + then add the correction as a real user message. This keeps role alternation + valid and leaves every previously cached message byte-for-byte unchanged. + """ + reasoning = str( + getattr(agent, "_current_streamed_reasoning_text", "") or "" + ).strip() + visible = agent._strip_think_blocks( + getattr(agent, "_current_streamed_assistant_text", "") or "" + ).strip() + + checkpoint_parts = ["[This response was interrupted by a user correction.]"] + if reasoning: + checkpoint_parts.extend( + ["Reasoning shown before the interruption:", reasoning] + ) + if visible: + checkpoint_parts.extend( + ["Visible response before the interruption:", visible] + ) + checkpoint = "\n\n".join(checkpoint_parts) + + # The normal live tail is user or tool, so an assistant checkpoint followed + # by the correction preserves strict alternation. If a transport already + # committed an assistant item, attribute the checkpoint inside the user + # correction instead of creating assistant→assistant. + if messages and messages[-1].get("role") == "assistant": + correction = ( + "[Context from the interrupted assistant response]\n" + f"{checkpoint}\n\n" + f"{text}" + ) + messages.append({"role": "user", "content": correction}) + else: + messages.append({"role": "assistant", "content": checkpoint}) + messages.append({"role": "user", "content": text}) + + agent._current_streamed_assistant_text = "" + agent._current_streamed_reasoning_text = "" + agent._stream_needs_break = True + + def _image_error_max_dimension(error: Exception) -> Optional[int]: """Extract a provider-reported image dimension ceiling, if present.""" parts = [] @@ -733,6 +780,16 @@ def run_conversation( ) while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call: + _redirect_text = agent._drain_pending_redirect() + if _redirect_text: + _apply_active_turn_redirect(agent, messages, _redirect_text) + if isinstance(original_user_message, str): + original_user_message = ( + f"{original_user_message}\n\n" + f"User correction during the turn: {_redirect_text}" + ) + agent._persist_session(messages, conversation_history) + # Reset per-turn checkpoint dedup so each iteration can take one snapshot agent._checkpoint_mgr.new_turn() @@ -1532,22 +1589,59 @@ def run_conversation( from hermes_cli.middleware import run_llm_execution_middleware - response = run_llm_execution_middleware( - api_kwargs, - _perform_api_call, - original_request=_original_api_kwargs, - task_id=effective_task_id, - turn_id=turn_id, - api_request_id=api_request_id, - session_id=agent.session_id or "", - platform=agent.platform or "", - model=agent.model, - provider=agent.provider, - base_url=agent.base_url, - api_mode=agent.api_mode, - api_call_count=api_call_count, - middleware_trace=list(_llm_middleware_trace), - ) + _model_request_active = getattr(agent, "_model_request_active", None) + _redirect_lock = getattr(agent, "_pending_redirect_lock", None) + if _redirect_lock is not None: + with _redirect_lock: + if _model_request_active is not None: + _model_request_active.set() + elif _model_request_active is not None: + _model_request_active.set() + _redirect_crossed_response = False + try: + response = run_llm_execution_middleware( + api_kwargs, + _perform_api_call, + original_request=_original_api_kwargs, + task_id=effective_task_id, + turn_id=turn_id, + api_request_id=api_request_id, + session_id=agent.session_id or "", + platform=agent.platform or "", + model=agent.model, + provider=agent.provider, + base_url=agent.base_url, + api_mode=agent.api_mode, + api_call_count=api_call_count, + middleware_trace=list(_llm_middleware_trace), + ) + finally: + if _redirect_lock is not None: + with _redirect_lock: + if _model_request_active is not None: + _model_request_active.clear() + _redirect_crossed_response = bool( + agent._pending_redirect + ) + else: + if _model_request_active is not None: + _model_request_active.clear() + _redirect_crossed_response = agent._has_pending_redirect() + if _redirect_crossed_response: + # The response and redirect can cross on different threads: + # redirect() observed the request as active just before this + # call returned. Discard that now-stale response and rebuild + # from the correction rather than silently losing it. + if thinking_spinner: + thinking_spinner.stop("") + thinking_spinner = None + if agent.thinking_callback: + agent.thinking_callback("") + if agent.clear_interrupt(preserve_redirect=True): + _retry.restart_with_redirected_messages = True + else: + interrupted = True + break api_duration = time.time() - api_start_time @@ -2509,6 +2603,15 @@ def run_conversation( thinking_spinner = None if agent.thinking_callback: agent.thinking_callback("") + if agent._has_pending_redirect(): + # redirect() deliberately used the interrupt machinery to + # cancel only this provider request. Keep its correction + # queued, clear the cancellation bit, and let the outer + # loop rebuild a clean request tail. Never materialize + # incomplete signed/encrypted reasoning items. + if agent.clear_interrupt(preserve_redirect=True): + _retry.restart_with_redirected_messages = True + break api_elapsed = time.time() - api_start_time agent._vprint(f"{agent.log_prefix}⚡ Interrupted during API call.", force=True) interrupted = True @@ -4457,6 +4560,15 @@ def run_conversation( f"{int(sleep_end - time.time())}s remaining" ) + if _retry.restart_with_redirected_messages: + # The cancelled request produced no valid assistant item. Reuse the + # same logical iteration after the outer loop appends the displayed + # partial context and correction to ``messages``. + api_call_count -= 1 + agent.iteration_budget.refund() + _retry.restart_with_redirected_messages = False + continue + # If the API call was interrupted, skip response processing if interrupted: _turn_exit_reason = "interrupted_during_api_call" diff --git a/agent/turn_retry_state.py b/agent/turn_retry_state.py index 3d231fef9ff4..59e343bfeda3 100644 --- a/agent/turn_retry_state.py +++ b/agent/turn_retry_state.py @@ -73,6 +73,10 @@ class TurnRetryState: # was rolled back off ``messages`` and the loop should re-issue the API # call against the newly-activated provider (#32421). restart_with_rebuilt_messages: bool = False + # A user correction cancelled the in-flight provider request. The outer + # loop must append a role-safe checkpoint + user message, rebuild the API + # payload, and retry the same logical iteration. + restart_with_redirected_messages: bool = False def __iter__(self): # Convenience for debugging / tests: iterate (name, value) pairs. diff --git a/run_agent.py b/run_agent.py index 7a37b9f3554b..4274c230e746 100644 --- a/run_agent.py +++ b/run_agent.py @@ -2809,8 +2809,33 @@ class AIAgent: if session_has_running_agent: running_agent.interrupt(new_message.text) """ - self._interrupt_requested = True - self._interrupt_message = message + # A hard stop and redirect share one lock so /stop cannot race with an + # accepted correction and accidentally turn itself into a retry. + _redirect_lock = getattr(self, "_pending_redirect_lock", None) + if _redirect_lock is not None: + with _redirect_lock: + self._interrupt_requested = True + self._interrupt_message = message + self._pending_redirect = None + else: + self._interrupt_requested = True + self._interrupt_message = message + self._pending_redirect = None + + # Codex app-server owns its model/tool loop and watches a private + # interrupt event rather than Hermes' per-thread flag. + if getattr(self, "api_mode", None) == "codex_app_server": + _codex_session = getattr(self, "_codex_session", None) + _request_interrupt = getattr(_codex_session, "request_interrupt", None) + if callable(_request_interrupt): + try: + _request_interrupt() + except Exception: + logger.debug( + "Failed to interrupt Codex app-server turn", + exc_info=True, + ) + # A cron turn performs its API request on the conversation thread to # avoid the nested interrupt-worker deadlock. Unlike the normal worker # path, its client is registered here so this cross-thread interrupt can @@ -2863,10 +2888,29 @@ class AIAgent: if not self.quiet_mode: print("\n⚡ Interrupt requested" + (f": '{message[:40]}...'" if message and len(message) > 40 else f": '{message}'" if message else "")) - def clear_interrupt(self) -> None: - """Clear any pending interrupt request and the per-thread tool interrupt signal.""" - self._interrupt_requested = False - self._interrupt_message = None + def clear_interrupt(self, *, preserve_redirect: bool = False) -> bool: + """Clear the interrupt request and per-thread tool signal. + + ``preserve_redirect`` is used only by the conversation loop after it + intentionally cancels a model request to rebuild that same logical + turn. Public hard-stop paths keep the default and clear everything. + """ + _redirect_lock = getattr(self, "_pending_redirect_lock", None) + if _redirect_lock is not None: + with _redirect_lock: + if preserve_redirect and not self._pending_redirect: + return False + self._interrupt_requested = False + self._interrupt_message = None + if not preserve_redirect: + self._pending_redirect = None + else: + if preserve_redirect and not getattr(self, "_pending_redirect", None): + return False + self._interrupt_requested = False + self._interrupt_message = None + if not preserve_redirect: + self._pending_redirect = None self._interrupt_thread_signal_pending = False if self._execution_thread_id is not None: _set_interrupt(False, self._execution_thread_id) @@ -2895,6 +2939,7 @@ class AIAgent: if _steer_lock is not None: with _steer_lock: self._pending_steer = None + return True def steer(self, text: str) -> bool: """ @@ -2932,6 +2977,118 @@ class AIAgent: self._pending_steer = cleaned return True + def redirect(self, text: str) -> bool: + """Redirect the active turn without converting it into a new task. + + During a normal Hermes model request this cancels only that request; + the conversation loop retains completed messages/tool results, records + the displayed partial reasoning as plain assistant context, appends the + correction as a real user message, and retries. During tool execution + it degrades to ``steer()`` so the tool can finish at a safe boundary. + Codex app-server has a native ``turn/steer`` operation and uses it + directly instead of cancelling. + + Returns ``False`` when there is no live turn or the text is empty, so + surfaces can fall back to their existing next-turn queue. + """ + if not text or not text.strip(): + return False + cleaned = text.strip() + + # Codex owns its internal reasoning/tool loop, so use its first-class + # active-turn steering protocol rather than interrupting the subprocess. + if getattr(self, "api_mode", None) == "codex_app_server": + _codex_session = getattr(self, "_codex_session", None) + _native_steer = getattr(_codex_session, "request_steer", None) + if callable(_native_steer): + _redirect_lock = getattr(self, "_pending_redirect_lock", None) + if _redirect_lock is not None: + with _redirect_lock: + if self._interrupt_requested: + return False + elif self._interrupt_requested: + return False + try: + return bool(_native_steer(cleaned)) + except Exception: + logger.debug("Codex app-server turn/steer failed", exc_info=True) + return False + + # Never kill a tool merely to deliver conversational guidance. The + # existing steer drain puts it on the final tool result before the next + # model decision, including delegate_task children. + if getattr(self, "_executing_tools", False): + return self.steer(cleaned) + + _model_active = getattr(self, "_model_request_active", None) + _redirect_lock = getattr(self, "_pending_redirect_lock", None) + if _redirect_lock is None: + if _model_active is None or not _model_active.is_set(): + return False + existing = getattr(self, "_pending_redirect", None) + if self._interrupt_requested and not existing: + return False + self._pending_redirect = ( + f"{existing}\n\n[Additional user correction]\n{cleaned}" + if existing + else cleaned + ) + self._interrupt_requested = True + self._interrupt_message = None + else: + with _redirect_lock: + if _model_active is None or not _model_active.is_set(): + # The response completed before we acquired the state lock. + # Reject so the surface queues a new turn. + return False + if self._interrupt_requested and not self._pending_redirect: + return False + if self._pending_redirect: + self._pending_redirect = ( + f"{self._pending_redirect}\n\n" + f"[Additional user correction]\n{cleaned}" + ) + else: + self._pending_redirect = cleaned + self._interrupt_requested = True + self._interrupt_message = None + + # Interrupt only the model request. Do not fan out to tool workers or + # child agents as interrupt() does. + _execution_thread_id = getattr(self, "_execution_thread_id", None) + if _execution_thread_id is not None: + _set_interrupt(True, _execution_thread_id) + self._interrupt_thread_signal_pending = False + else: + self._interrupt_thread_signal_pending = True + _abort_active_request = getattr(self, "_active_request_abort", None) + if callable(_abort_active_request): + try: + _abort_active_request("redirect_abort") + except Exception: + logger.debug("Failed to abort request for redirect", exc_info=True) + return True + + def _has_pending_redirect(self) -> bool: + """Return whether an active-turn redirect is waiting to be applied.""" + _redirect_lock = getattr(self, "_pending_redirect_lock", None) + if _redirect_lock is None: + return bool(getattr(self, "_pending_redirect", None)) + with _redirect_lock: + return bool(self._pending_redirect) + + def _drain_pending_redirect(self) -> Optional[str]: + """Return and clear pending active-turn correction text.""" + _redirect_lock = getattr(self, "_pending_redirect_lock", None) + if _redirect_lock is None: + text = getattr(self, "_pending_redirect", None) + self._pending_redirect = None + return text + with _redirect_lock: + text = self._pending_redirect + self._pending_redirect = None + return text + def _drain_pending_steer(self) -> Optional[str]: """Return the pending steer text (if any) and clear the slot. @@ -4918,6 +5075,7 @@ class AIAgent: pass self._record_streamed_assistant_text(tail) self._current_streamed_assistant_text = "" + self._current_streamed_reasoning_text = "" def _record_streamed_assistant_text(self, text: str) -> None: """Accumulate visible assistant text emitted through stream callbacks.""" @@ -5258,6 +5416,15 @@ class AIAgent: cb(text) except Exception: pass + else: + # Only checkpoint reasoning that a surface actually displayed. + # show_reasoning=false leaves the callback unset, so hidden + # provider thinking never becomes visible transcript content. + if isinstance(text, str) and text: + self._current_streamed_reasoning_text = ( + getattr(self, "_current_streamed_reasoning_text", "") + + text + ) def _fire_tool_gen_started(self, tool_name: str) -> None: """Notify display layer that the model is generating tool call arguments. diff --git a/tests/agent/test_turn_retry_state.py b/tests/agent/test_turn_retry_state.py index 687a63f41899..89309f5cbef1 100644 --- a/tests/agent/test_turn_retry_state.py +++ b/tests/agent/test_turn_retry_state.py @@ -32,6 +32,7 @@ EXPECTED_FIELDS = { "restart_with_compressed_messages", "restart_with_length_continuation", "restart_with_rebuilt_messages", + "restart_with_redirected_messages", } diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 073bdd892c42..26d82cbcfdf6 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -12,6 +12,7 @@ import json import logging import re import threading +import time import uuid from logging.handlers import RotatingFileHandler from pathlib import Path @@ -4850,6 +4851,136 @@ class TestRunConversation: "content": "Sure, here's how to do it: first", } + def test_redirect_during_thinking_retries_same_turn_with_context(self, agent): + """A corrective follow-up keeps displayed reasoning and does not end the turn.""" + self._setup_agent(agent) + agent.reasoning_callback = lambda _text: None + final = _mock_response(content="Using Postgres instead.", finish_reason="stop") + requests = [] + persisted = [] + + def _fake_api_call(api_kwargs): + requests.append(api_kwargs) + if len(requests) == 1: + agent._fire_reasoning_delta("I should implement this with SQLite.") + assert agent.redirect("No, use Postgres instead.") is True + raise InterruptedError("redirect cancelled the first request") + return final + + with ( + patch.object(agent, "_interruptible_api_call", side_effect=_fake_api_call), + patch.object( + agent, + "_persist_session", + side_effect=lambda messages, *_a, **_k: persisted.append( + [dict(message) for message in messages] + ), + ), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("Choose a database and implement it.") + + assert result["completed"] is True + assert result["interrupted"] is False + assert result["final_response"] == "Using Postgres instead." + assert len(requests) == 2 + + replay = requests[1]["messages"] + assert [m["role"] for m in replay[-3:]] == [ + "user", + "assistant", + "user", + ] + checkpoint = replay[-2]["content"] + assert "interrupted by a user correction" in checkpoint + assert "I should implement this with SQLite." in checkpoint + assert replay[-1]["content"] == "No, use Postgres instead." + assert agent._pending_redirect is None + assert any( + snapshot[-1].get("content") == "No, use Postgres instead." + and snapshot[-2].get("role") == "assistant" + for snapshot in persisted + if len(snapshot) >= 2 + ) + + def test_redirect_wins_race_with_response_completion(self, agent): + """If the provider returns as redirect lands, discard the stale answer.""" + self._setup_agent(agent) + stale = _mock_response(content="Using SQLite.", finish_reason="stop") + corrected = _mock_response(content="Using Postgres.", finish_reason="stop") + calls = 0 + + def _fake_api_call(_api_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + assert agent.redirect("Use Postgres instead.") is True + return stale + return corrected + + with ( + patch.object(agent, "_interruptible_api_call", side_effect=_fake_api_call), + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("Choose a database.") + + assert calls == 2 + assert result["final_response"] == "Using Postgres." + assert all( + message.get("content") != "Using SQLite." + for message in result["messages"] + ) + + def test_redirect_from_input_thread_cancels_live_model_request(self, agent): + """Exercise the real cross-thread path used by CLI and gateways.""" + self._setup_agent(agent) + agent.reasoning_callback = lambda _text: None + entered = threading.Event() + results = {} + calls = 0 + final = _mock_response(content="Corrected answer.", finish_reason="stop") + + def _fake_api_call(_api_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + agent._fire_reasoning_delta("Following the original approach.") + entered.set() + deadline = time.time() + 2 + while not agent._interrupt_requested and time.time() < deadline: + time.sleep(0.01) + raise InterruptedError("request cancelled by redirect") + return final + + with ( + patch.object(agent, "_interruptible_api_call", side_effect=_fake_api_call), + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + worker = threading.Thread( + target=lambda: results.update( + result=agent.run_conversation("Take the original approach.") + ) + ) + worker.start() + assert entered.wait(timeout=2) + assert agent.redirect("Use the corrected approach.") is True + worker.join(timeout=5) + + assert worker.is_alive() is False + assert calls == 2 + assert results["result"]["completed"] is True + assert results["result"]["final_response"] == "Corrected answer." + checkpoint = results["result"]["messages"][-3] + assert "Following the original approach." in checkpoint["content"] + assert results["result"]["messages"][-2]["content"] == ( + "Use the corrected approach." + ) + def test_interrupt_before_any_stream_keeps_sentinel(self, agent): """An interrupt with no streamed text falls back to the metadata sentinel.""" from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX diff --git a/tests/run_agent/test_steer.py b/tests/run_agent/test_steer.py index 99feb56343e7..ebf84c617d4b 100644 --- a/tests/run_agent/test_steer.py +++ b/tests/run_agent/test_steer.py @@ -23,6 +23,24 @@ def _bare_agent() -> AIAgent: agent = object.__new__(AIAgent) agent._pending_steer = None agent._pending_steer_lock = threading.Lock() + agent._pending_redirect = None + agent._pending_redirect_lock = threading.Lock() + agent._model_request_active = threading.Event() + agent._executing_tools = False + agent._execution_thread_id = None + agent._interrupt_thread_signal_pending = False + agent._interrupt_requested = False + agent._interrupt_message = None + agent._active_children = [] + agent._active_children_lock = threading.Lock() + agent._tool_worker_threads = None + agent._tool_worker_threads_lock = None + agent._current_streamed_reasoning_text = "" + agent._current_streamed_assistant_text = "" + agent._stream_needs_break = False + agent._strip_think_blocks = lambda content: content + agent.quiet_mode = True + agent.api_mode = "chat_completions" return agent @@ -72,6 +90,161 @@ class TestSteerDrain: assert agent._drain_pending_steer() is None +class TestActiveTurnRedirect: + def test_rejects_when_no_turn_is_active(self): + agent = _bare_agent() + assert agent.redirect("change course") is False + assert agent._pending_redirect is None + + def test_cancels_only_an_active_model_request(self): + agent = _bare_agent() + agent._model_request_active.set() + + assert agent.redirect("use Postgres") is True + assert agent._pending_redirect == "use Postgres" + assert agent._interrupt_requested is True + assert agent._interrupt_message is None + + def test_multiple_redirects_preserve_message_boundaries(self): + agent = _bare_agent() + agent._model_request_active.set() + + assert agent.redirect("first correction") is True + assert agent.redirect("second correction") is True + assert agent._pending_redirect == ( + "first correction\n\n" + "[Additional user correction]\n" + "second correction" + ) + + def test_hard_interrupt_wins_over_new_redirect(self): + agent = _bare_agent() + agent._model_request_active.set() + agent._interrupt_requested = True + + assert agent.redirect("too late") is False + assert agent._pending_redirect is None + + def test_hidden_reasoning_is_not_checkpointed(self): + agent = _bare_agent() + agent.reasoning_callback = None + agent._current_streamed_reasoning_text = "" + + agent._fire_reasoning_delta("private provider thinking") + + assert agent._current_streamed_reasoning_text == "" + + def test_response_completion_before_redirect_lock_rejects_correction(self): + agent = _bare_agent() + agent._model_request_active.set() + started = threading.Event() + outcome = {} + + def redirect(): + started.set() + outcome["accepted"] = agent.redirect("late correction") + + with agent._pending_redirect_lock: + worker = threading.Thread(target=redirect) + worker.start() + assert started.wait(timeout=1) + # Mirrors conversation_loop clearing the request-active marker + # under this same lock before redirect can commit its slot. + agent._model_request_active.clear() + worker.join(timeout=1) + + assert outcome["accepted"] is False + assert agent._pending_redirect is None + + def test_hard_stop_wins_concurrent_redirect(self): + agent = _bare_agent() + agent._model_request_active.set() + start = threading.Barrier(3) + outcome = {} + + def redirect(): + start.wait() + outcome["redirect"] = agent.redirect("change course") + + def hard_stop(): + start.wait() + agent.interrupt("stop requested") + + redirect_thread = threading.Thread(target=redirect) + stop_thread = threading.Thread(target=hard_stop) + redirect_thread.start() + stop_thread.start() + start.wait() + redirect_thread.join(timeout=1) + stop_thread.join(timeout=1) + + assert redirect_thread.is_alive() is False + assert stop_thread.is_alive() is False + assert agent._interrupt_requested is True + assert agent._interrupt_message == "stop requested" + assert agent._pending_redirect is None + + def test_codex_app_server_hard_stop_reaches_native_session(self): + agent = _bare_agent() + calls = [] + agent.api_mode = "codex_app_server" + agent._codex_session = type( + "_CodexSession", + (), + {"request_interrupt": lambda self: calls.append("interrupt")}, + )() + + agent.interrupt() + + assert calls == ["interrupt"] + + def test_codex_app_server_redirect_rejects_after_hard_stop(self): + agent = _bare_agent() + calls = [] + agent.api_mode = "codex_app_server" + agent._interrupt_requested = True + agent._codex_session = type( + "_CodexSession", + (), + {"request_steer": lambda self, text: calls.append(text) or True}, + )() + + assert agent.redirect("too late") is False + assert calls == [] + + def test_redirect_during_tool_execution_uses_safe_steer_boundary(self): + agent = _bare_agent() + agent._executing_tools = True + + assert agent.redirect("also check migrations") is True + assert agent._pending_redirect is None + assert agent._pending_steer == "also check migrations" + assert agent._interrupt_requested is False + + +class TestActiveTurnRedirectCheckpoint: + def test_assistant_tail_puts_correction_last(self): + from agent.conversation_loop import _apply_active_turn_redirect + + agent = _bare_agent() + agent._current_streamed_reasoning_text = "Shown reasoning." + agent._current_streamed_assistant_text = "Visible draft." + messages = [ + {"role": "user", "content": "start"}, + {"role": "assistant", "content": "committed assistant item"}, + ] + + _apply_active_turn_redirect(agent, messages, "Use Postgres instead.") + + assert [m["role"] for m in messages] == ["user", "assistant", "user"] + assert messages[-1]["role"] == "user" + assert messages[-1]["content"].endswith("Use Postgres instead.") + assert sum(1 for m in messages if m["role"] == "assistant") == 1 + assert "Shown reasoning." in messages[-1]["content"] + assert "Visible draft." in messages[-1]["content"] + assert "Context from the interrupted assistant response" in messages[-1]["content"] + + class TestSteerInjection: def test_appends_to_last_tool_result(self): agent = _bare_agent() @@ -196,10 +369,12 @@ class TestSteerClearedOnInterrupt: agent._tool_worker_threads_lock = None agent.steer("will be dropped") + agent._pending_redirect = "also drop this" assert agent._pending_steer == "will be dropped" agent.clear_interrupt() assert agent._pending_steer is None + assert agent._pending_redirect is None class TestPreApiCallSteerDrain: