feat(agent): add active-turn redirect core primitive

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

@@ -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).

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"

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.

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.

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",
 }
 
 

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

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: