fix(agent): clear_interrupt drops an unconsumed steer only on hard cancel

The _pending_steer wipe in clear_interrupt() ran unconditionally behind a
comment claiming "a hard interrupt supersedes any pending /steer" — but
nothing gated it on an actual hard stop. Every soft caller wiped the buffer
too: the redirect rebuild (turn_api_call.handle_api_interrupt and the
response-crossing check), API-error recovery, and turn-boundary hygiene.
Production signature: a text-only follow-up lands while the agent is busy,
redirect() degrades to steer() and returns True, the gateway acknowledges
delivery, and the accepted message then evaporates during a later soft
clear — no payload, no leftover, no log.

Gate the wipe on the new hard_cancel flag and keep the steer buffered on
soft clears: the existing drains already deliver it (pre-API inject into
the newest tool result, post-batch append, or the finalizer's leftover
handoff to the next turn). abort_turn_on_interrupt is the one real hard
stop in the clear family and now passes hard_cancel=True, preserving the
documented drop-on-stop semantics.

agent/interrupt_control.py

@@ -212,9 +212,15 @@ class InterruptControlMixin:
         override interrupt(message=None) without hard_cancel."""
         InterruptControlMixin.interrupt(self, message, hard_cancel=True, tool_reason=tool_reason)
 
-    def clear_interrupt(self, *, preserve_redirect: bool = False) -> bool:
+    def clear_interrupt(self, *, preserve_redirect: bool = False, hard_cancel: bool = False) -> bool:
         """Clear the interrupt request and per-thread tool signal. ``preserve_redirect`` is only for the
-        conversation loop rebuilding the same logical turn after cancelling a model request."""
+        conversation loop rebuilding the same logical turn after cancelling a model request.
+        ``hard_cancel`` also drops an unconsumed pending steer: the aborted turn's next tool iteration
+        will no longer happen, and re-injecting the note into the post-stop turn would surprise the
+        user. Every other caller (redirect rebuild, error recovery, turn-boundary hygiene) continues
+        this session, so the already-accepted steer must survive: it stays buffered for the existing
+        drains — the pre-API inject, the post-batch append, or the finalizer's leftover handoff —
+        instead of silently vanishing after the surface was told it was delivered."""
         with _ic_lock(self, "_pending_redirect_lock"):
             if preserve_redirect and not _ic_slot(self, "_pending_redirect_lock", "_pending_redirect"):
                 return False
@@ -227,9 +233,12 @@ class InterruptControlMixin:
         if self._execution_thread_id is not None:
             _set_interrupt(False, self._execution_thread_id)
         _ic_signal_tool_workers(self, False)
-        # A hard interrupt supersedes any pending /steer — its target iteration will no longer happen.
-        with _ic_lock(self, "_pending_steer_lock"):
-            self._pending_steer = None
+        if hard_cancel:
+            # Hard stop only (see docstring). The comment that used to run unconditionally here
+            # claimed a hard interrupt supersedes the steer — but nothing gated this wipe on
+            # hard_cancel, so a soft clear dropped a live user message with no trace.
+            with _ic_lock(self, "_pending_steer_lock"):
+                self._pending_steer = None
         return True
 
     def steer(self, text: str) -> bool:

agent/turn_recovery.py

@@ -1113,7 +1113,9 @@ def abort_turn_on_interrupt(
     _vlines(agent, f"⚡ {abort_message}")
     close_interrupted_tool_sequence(messages, interrupt_text)
     agent._persist_session(messages, conversation_history)
-    agent.clear_interrupt()
+    # The turn was stopped, not rebuilt: a pending steer was aimed at this turn's next
+    # tool iteration, which will no longer happen — drop it (hard-cancel semantics).
+    agent.clear_interrupt(hard_cancel=True)
     return {
         "final_response": interrupt_text, "messages": messages, "api_calls": api_call_count,
         "completed": False, "interrupted": True,

tests/agent/test_steer.py

@@ -13,6 +13,28 @@ import pytest
 
 from agent.prompt_builder import STEER_MARKER_OPEN, format_steer_marker
 from run_agent import AIAgent
+from tools.registry import registry
+
+# Registry handler for the end-to-end steer-survival test below (module level,
+# like the built-in tool files — dispatch looks the tool up here at run time).
+_STEER_SURVIVAL_TOOL = "steer_survival_probe"
+
+
+def _steer_survival_tool(args, **_kwargs):
+    return "probe ok"
+
+
+registry.register(
+    name=_STEER_SURVIVAL_TOOL,
+    toolset="utility",
+    schema={
+        "name": _STEER_SURVIVAL_TOOL,
+        "description": "probe tool for the steer-survival regression test",
+        "parameters": {"type": "object", "properties": {}, "required": []},
+    },
+    handler=_steer_survival_tool,
+    override=True,
+)
 
 
 def _bare_agent() -> AIAgent:
@@ -645,10 +667,10 @@ class TestSteerThreadSafety:
 
 
 class TestSteerClearedOnInterrupt:
-    def test_clear_interrupt_drops_pending_steer(self):
+    def test_hard_cancel_drops_pending_steer(self):
         """A hard interrupt supersedes any pending steer — the agent's
         next tool iteration won't happen, so delivering the steer later
-        would be surprising."""
+        would be surprising. Only the explicit hard-cancel clear drops it."""
         agent = _bare_agent()
         # Minimal surface needed by clear_interrupt()
         agent._interrupt_requested = True
@@ -662,10 +684,135 @@ class TestSteerClearedOnInterrupt:
         agent._pending_redirect = "also drop this"
         assert agent._pending_steer == "will be dropped"
 
-        agent.clear_interrupt()
+        agent.clear_interrupt(hard_cancel=True)
         assert agent._pending_steer is None
         assert agent._pending_redirect is None
 
+    def test_soft_clear_preserves_pending_steer(self):
+        """A soft clear (redirect rebuild, error recovery, turn-boundary
+        hygiene) keeps the session alive, so an already-accepted steer must
+        survive it — the existing drains deliver it on the continued run.
+        Dropping it here silently lost a user message the surface had
+        already acknowledged as delivered."""
+        agent = _bare_agent()
+        agent._interrupt_requested = True
+        agent._interrupt_message = None
+        agent._interrupt_thread_signal_pending = False
+        agent._execution_thread_id = None
+        agent._tool_worker_threads = None
+        agent._tool_worker_threads_lock = None
+
+        agent.steer("must survive the rebuild")
+        agent._pending_redirect = "correction"
+
+        agent.clear_interrupt(preserve_redirect=True)
+        assert agent._pending_steer == "must survive the rebuild"
+        # preserve_redirect semantics unchanged: the correction survives too.
+        assert agent._pending_redirect == "correction"
+
+        # A plain soft clear (no flags) preserves the steer as well.
+        agent.clear_interrupt()
+        assert agent._pending_steer == "must survive the rebuild"
+        assert agent._pending_redirect is None
+
+
+class TestSteerSurvivesRedirectRebuild:
+    """A steer accepted while a redirect lands must still reach a later API
+    payload. Regression for the busy-redirect message loss: the rebuild's
+    ``clear_interrupt(preserve_redirect=True)`` used to wipe ``_pending_steer``
+    unconditionally, so a user message the surface had already acknowledged
+    as delivered evaporated with no trace (no payload, no leftover, no log)."""
+
+    STEER_TEXT = "STEER_TEXT_ONE"
+    REDIRECT_TEXT = "REDIRECT_TEXT_TWO"
+
+    def _loop_agent(self):
+        from unittest.mock import MagicMock, patch
+
+        from run_agent import AIAgent
+
+        tool_schema = {
+            "type": "function",
+            "function": {
+                "name": _STEER_SURVIVAL_TOOL,
+                "description": "probe tool for the steer-survival regression test",
+                "parameters": {"type": "object", "properties": {}, "required": []},
+            },
+        }
+        with (
+            patch("model_tools.get_tool_definitions", return_value=[tool_schema]),
+            patch("model_tools.check_toolset_requirements", return_value={}),
+            patch("agent.process_bootstrap.OpenAI"),
+        ):
+            agent = AIAgent(
+                api_key="test-key-1234567890",
+                base_url="https://openrouter.ai/api/v1",
+                quiet_mode=True,
+                skip_context_files=True,
+                skip_memory=True,
+            )
+        agent.client = MagicMock()
+        agent._disable_streaming = True
+        agent._cached_system_prompt = "You are helpful."
+        agent._use_prompt_caching = False
+        agent.tool_delay = 0
+        agent.compression_enabled = False
+        agent.save_trajectories = False
+        return agent
+
+    def test_steer_accepted_before_redirect_lands_in_rebuilt_payload(self):
+        from types import SimpleNamespace
+        from unittest.mock import patch
+
+        from tests.run_agent.test_run_agent import _mock_response
+
+        agent = self._loop_agent()
+        payloads = []
+
+        def model_call(api_kwargs):
+            payloads.append([dict(m) for m in api_kwargs["messages"]])
+            if len(payloads) == 1:
+                tool_call = SimpleNamespace(
+                    id="call_1", type="function",
+                    function=SimpleNamespace(name=_STEER_SURVIVAL_TOOL, arguments="{}"),
+                )
+                return _mock_response(
+                    content=None, finish_reason="tool_calls", tool_calls=[tool_call]
+                )
+            if len(payloads) == 2:
+                # Both mid-turn user messages race the in-flight request:
+                # the steer is accepted first, then the redirect kills the
+                # request and arms the rebuild.
+                assert agent.steer(self.STEER_TEXT) is True
+                assert agent.redirect(self.REDIRECT_TEXT) is True
+                raise InterruptedError("redirect cancelled the in-flight request")
+            return _mock_response(content="rebuilt reply", finish_reason="stop")
+
+        agent._interruptible_api_call = model_call
+
+        with (
+            patch.object(agent, "_flush_messages_to_session_db"),
+            patch.object(agent, "_persist_session"),
+            patch.object(agent, "_save_trajectory"),
+            patch.object(agent, "_cleanup_task_resources"),
+        ):
+            result = agent.run_conversation("start something")
+
+        blob = "\n".join(
+            str(m.get("content"))
+            for call in payloads
+            for m in call
+            if isinstance(m, dict)
+        )
+        # The rebuild reached the wire: the redirect correction is a real user message.
+        assert self.REDIRECT_TEXT in blob
+        # The steer accepted just before the redirect must ride the same rebuild
+        # (injected into the newest tool result by the pre-API drain).
+        assert self.STEER_TEXT in blob
+        # Fully consumed: nothing left for the finalizer's leftover handoff.
+        assert result.get("pending_steer") is None
+        assert result["completed"] is True
+
 
 class TestPreApiCallSteerDrain:
     """Test that steers arriving during an API call are drained before the