From 34d0de80e64a47cb1022d99b964d3755a3c2bd3d Mon Sep 17 00:00:00 2001
From: Brooklyn Nicholson <brooklyn.bb.nicholson@gmail.com>
Date: Mon, 13 Jul 2026 02:23:39 -0400
Subject: [PATCH] feat(surfaces): route busy-input corrections through
 active-turn redirect

The default `busy_input_mode: interrupt` now redirects the live turn instead
of hard-stopping it and re-queuing a fresh turn, wired consistently across
every first-party surface via the shared core primitive.

- CLI, gateway (busy + PRIORITY paths), TUI (`_handle_busy_submit`), desktop
  (`session.redirect` RPC), and ACP call `redirect()` when the agent advertises
  `_supports_active_turn_redirect`, and fall back to the proven interrupt +
  next-turn queue for older runtimes.
- Redirect is gated to plain text with no attachments: captioned or
  attachment-bearing events (including adapters that classify unknown media as
  `TEXT`) stay queued so media is never dropped.
- ACP `cancel()` records the interrupted prompt, sets its cancel event, and
  hard-stops the agent while holding `runtime_lock`, closing the
  cancel-then-correct ordering gap; connection I/O happens after the lock is
  released.
- Desktop appends the correction as a real user transcript message so the live
  view matches the durable history after reload.
- `/busy` help, onboarding hints, and the new `session.redirect` RPC describe
  the redirect behavior; `/stop` remains the hard stop.
---
 acp_adapter/server.py                         | 98 +++++++++++++++----
 agent/onboarding.py                           | 13 +++
 .../composer/hooks/use-composer-submit.ts     |  6 +-
 .../hooks/use-prompt-actions/index.test.tsx   | 42 +++++---
 .../session/hooks/use-prompt-actions/index.ts | 31 +++---
 apps/desktop/src/app/types.ts                 |  5 +
 cli.py                                        | 44 ++++++---
 gateway/run.py                                | 65 +++++++++++-
 hermes_cli/cli_commands_mixin.py              |  6 +-
 tests/acp_adapter/test_acp_commands.py        | 62 ++++++++++++
 tests/gateway/test_busy_session_ack.py        | 48 +++++++++
 tests/test_tui_gateway_queue_on_busy.py       | 88 ++++++++++++++++-
 tests/test_tui_gateway_server.py              | 29 ++++++
 tui_gateway/server.py                         | 98 +++++++++++++++++--
 ui-tui/src/app/useSubmission.ts               | 19 ++--
 15 files changed, 563 insertions(+), 91 deletions(-)

diff --git a/acp_adapter/server.py b/acp_adapter/server.py
index 266d587b0743..f664c47070e4 100644
--- a/acp_adapter/server.py
+++ b/acp_adapter/server.py
@@ -1218,12 +1218,19 @@ class HermesACPAgent(acp.Agent):
             with state.runtime_lock:
                 if state.is_running and state.current_prompt_text:
                     state.interrupted_prompt_text = state.current_prompt_text
-            state.cancel_event.set()
-            try:
-                if getattr(state, "agent", None) and hasattr(state.agent, "interrupt"):
-                    state.agent.interrupt()
-            except Exception:
-                logger.debug("Failed to interrupt ACP session %s", session_id, exc_info=True)
+                # Publish cancellation and hard-stop the agent before another
+                # prompt can acquire this lock and mistake the turn for
+                # redirectable work.
+                state.cancel_event.set()
+                try:
+                    if getattr(state, "agent", None) and hasattr(state.agent, "interrupt"):
+                        state.agent.interrupt()
+                except Exception:
+                    logger.debug(
+                        "Failed to interrupt ACP session %s",
+                        session_id,
+                        exc_info=True,
+                    )
             logger.info("Cancelled session %s", session_id)
 
     async def fork_session(
@@ -1352,6 +1359,26 @@ class HermesACPAgent(acp.Agent):
             elif rewrite_idle:
                 user_text = steer_text
                 user_content = steer_text
+        elif (
+            text_only_prompt
+            and isinstance(user_content, str)
+            and not user_text.startswith("/")
+        ):
+            # Some ACP clients implement "stop and send" as two protocol calls:
+            # cancel the active prompt, then submit plain correction text. Keep
+            # the cancelled request attached so deictic follow-ups ("not that
+            # file") still have an explicit target.
+            interrupted_prompt = ""
+            with state.runtime_lock:
+                if not state.is_running and state.interrupted_prompt_text:
+                    interrupted_prompt = state.interrupted_prompt_text
+                    state.interrupted_prompt_text = ""
+            if interrupted_prompt:
+                user_text = (
+                    f"{interrupted_prompt}\n\n"
+                    f"User correction/guidance after interrupt: {user_text}"
+                )
+                user_content = user_text
 
         # Intercept slash commands — handle locally without calling the LLM.
         # Slash commands are text-only; if the client included images/resources,
@@ -1366,23 +1393,54 @@ class HermesACPAgent(acp.Agent):
                     await self._send_usage_update(state)
                 return PromptResponse(stop_reason="end_turn")
 
-        # If Zed sends another regular prompt while the same ACP session is
-        # still running, queue it instead of racing two AIAgent loops against
-        # the same state.history. /steer and /queue are handled above and can
-        # land immediately.
+        # If the client sends another regular text prompt while this ACP session
+        # is running, route it through the core active-turn redirect. Rich media
+        # and older runtimes retain the proven next-turn queue fallback.
+        redirected = False
+        queued_depth: int | None = None
         with state.runtime_lock:
             if state.is_running:
-                queued_text = user_text or "[Image attachment]"
-                state.queued_prompts.append(queued_text)
-                depth = len(state.queued_prompts)
-                if self._conn:
-                    update = acp.update_agent_message_text(
-                        f"Queued for the next turn. ({depth} queued)"
+                if (
+                    text_only_prompt
+                    and isinstance(user_content, str)
+                    and getattr(
+                        state.agent,
+                        "_supports_active_turn_redirect",
+                        False,
                     )
-                    await self._conn.session_update(session_id, update)
-                return PromptResponse(stop_reason="end_turn")
-            state.is_running = True
-            state.current_prompt_text = user_text or "[Image attachment]"
+                    is True
+                    and hasattr(state.agent, "redirect")
+                ):
+                    try:
+                        redirected = bool(state.agent.redirect(user_content))
+                    except Exception:
+                        logger.debug(
+                            "ACP active-turn redirect failed for %s",
+                            session_id,
+                            exc_info=True,
+                        )
+                if not redirected:
+                    queued_text = user_text or "[Image attachment]"
+                    state.queued_prompts.append(queued_text)
+                    queued_depth = len(state.queued_prompts)
+            else:
+                state.is_running = True
+                state.current_prompt_text = user_text or "[Image attachment]"
+
+        if redirected:
+            if self._conn:
+                update = acp.update_agent_message_text(
+                    "Redirected the active turn with your correction."
+                )
+                await self._conn.session_update(session_id, update)
+            return PromptResponse(stop_reason="end_turn")
+        if queued_depth is not None:
+            if self._conn:
+                update = acp.update_agent_message_text(
+                    f"Queued for the next turn. ({queued_depth} queued)"
+                )
+                await self._conn.session_update(session_id, update)
+            return PromptResponse(stop_reason="end_turn")
 
         logger.info("Prompt on session %s: %s", session_id, user_text[:100])
 
diff --git a/agent/onboarding.py b/agent/onboarding.py
index c29ea1529fb0..148fbcc9fdae 100644
--- a/agent/onboarding.py
+++ b/agent/onboarding.py
@@ -52,6 +52,13 @@ def busy_input_hint_gateway(mode: str) -> str:
             "Send `/busy interrupt` or `/busy queue` to change this, or "
             "`/busy status` to check. This notice won't appear again."
         )
+    if mode == "redirect":
+        return (
+            "💡 First-time tip — I redirected the current run using your message. "
+            "Completed work stays in context, and `/stop` still cancels the task. "
+            "Send `/busy queue` to wait for a separate turn, or `/busy status` "
+            "to check. This notice won't appear again."
+        )
     return (
         "💡 First-time tip — I just interrupted my current task to answer you. "
         "Send `/busy queue` to queue follow-ups for after the current task instead, "
@@ -74,6 +81,12 @@ def busy_input_hint_cli(mode: str) -> str:
             "after the next tool call. Use /busy interrupt or /busy queue to "
             "change this. This tip only shows once."
         )
+    if mode == "redirect":
+        return (
+            "(tip) Your correction redirected the current run without discarding "
+            "completed work. Use /stop to cancel or /busy queue to wait for a "
+            "separate turn. This tip only shows once."
+        )
     return (
         "(tip) Your message interrupted the current run. "
         "Use /busy queue to queue messages for the next turn instead, "
diff --git a/apps/desktop/src/app/chat/composer/hooks/use-composer-submit.ts b/apps/desktop/src/app/chat/composer/hooks/use-composer-submit.ts
index adf44e34a8d7..01ff0ecf8e4b 100644
--- a/apps/desktop/src/app/chat/composer/hooks/use-composer-submit.ts
+++ b/apps/desktop/src/app/chat/composer/hooks/use-composer-submit.ts
@@ -173,9 +173,9 @@ export function useComposerSubmit({
     focusInput()
   }
 
-  // Steer the live turn (nudge without interrupting). Clears the draft up front
-  // for snappy feedback; if the gateway rejects (no live tool window) the words
-  // are re-queued so nothing is lost — same safety net as a plain queue.
+  // Redirect the live turn with a correction. The gateway either restarts the
+  // active model request with its displayed context or waits for the current
+  // tool boundary. If the turn already ended, queue the words instead.
   const steerDraft = () => {
     if (!onSteer || !canSteer) {
       return
diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx
index a06ee1294c08..6fb71e1b6c35 100644
--- a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx
+++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx
@@ -61,6 +61,8 @@ interface HarnessHandle {
   activeSessionIdRef: MutableRefObject<string | null>
   cancelRun: () => Promise<void>
   restoreToMessage: (messageId: string, target?: { text?: string; userOrdinal?: number | null }) => Promise<void>
+  redirectPrompt: (text: string) => Promise<boolean>
+  /** @deprecated Use `redirectPrompt`. */
   steerPrompt: (text: string) => Promise<boolean>
   submitText: (text: string, options?: SubmitTextOptions) => Promise<boolean>
 }
@@ -160,6 +162,8 @@ function Harness({
         act(async () => actions.cancelRun(...args)) as Promise<void>,
       restoreToMessage: (...args: Parameters<typeof actions.restoreToMessage>) =>
         act(async () => actions.restoreToMessage(...args)) as Promise<void>,
+      redirectPrompt: (...args: Parameters<typeof actions.redirectPrompt>) =>
+        act(async () => actions.redirectPrompt(...args)) as Promise<boolean>,
       steerPrompt: (...args: Parameters<typeof actions.steerPrompt>) =>
         act(async () => actions.steerPrompt(...args)) as Promise<boolean>,
       submitText: (...args: Parameters<typeof actions.submitText>) =>
@@ -168,6 +172,7 @@ function Harness({
   }, [
     actions.cancelRun,
     actions.restoreToMessage,
+    actions.redirectPrompt,
     actions.steerPrompt,
     actions.submitText,
     activeSessionIdRef,
@@ -703,32 +708,41 @@ describe('usePromptActions submit / queue drain semantics', () => {
   })
 })
 
-describe('usePromptActions steerPrompt', () => {
+describe('usePromptActions redirectPrompt', () => {
   afterEach(() => {
     cleanup()
     vi.restoreAllMocks()
   })
 
-  it('injects the trimmed text via session.steer and reports acceptance on a queued status', async () => {
-    const requestGateway = vi.fn(async () => ({ status: 'queued' }) as never)
+  it('redirects the live turn with trimmed correction text', async () => {
+    const requestGateway = vi.fn(async () => ({ status: 'redirected' }) as never)
 
     let handle: HarnessHandle | null = null
+    const capturedStates: Record<string, unknown>[] = []
     await actRender(
-      <Harness onReady={h => (handle = h)} refreshSessions={async () => undefined} requestGateway={requestGateway} />
+      <Harness
+        onReady={h => (handle = h)}
+        onSeedState={state => capturedStates.push(state)}
+        refreshSessions={async () => undefined}
+        requestGateway={requestGateway}
+      />
     )
 
-    const accepted = await handle!.steerPrompt('  nudge the run  ')
+    const accepted = await handle!.redirectPrompt('  nudge the run  ')
 
     expect(accepted).toBe(true)
-    // Steer never starts a turn — it rides the live run via session.steer only.
-    expect(requestGateway).toHaveBeenCalledWith('session.steer', {
+    expect(requestGateway).toHaveBeenCalledWith('session.redirect', {
       session_id: RUNTIME_SESSION_ID,
       text: 'nudge the run'
     })
     expect(requestGateway).not.toHaveBeenCalledWith('prompt.submit', expect.anything())
+    expect((capturedStates.at(-1)?.messages as unknown[]).at(-1)).toMatchObject({
+      role: 'user',
+      parts: [{ type: 'text', text: 'nudge the run' }]
+    })
   })
 
-  it('reports rejection (so the caller queues) when the gateway has no live tool window', async () => {
+  it('reports rejection so the caller queues when the turn already ended', async () => {
     const requestGateway = vi.fn(async () => ({ status: 'rejected' }) as never)
 
     let handle: HarnessHandle | null = null
@@ -736,12 +750,12 @@ describe('usePromptActions steerPrompt', () => {
       <Harness onReady={h => (handle = h)} refreshSessions={async () => undefined} requestGateway={requestGateway} />
     )
 
-    expect(await handle!.steerPrompt('too late')).toBe(false)
+    expect(await handle!.redirectPrompt('too late')).toBe(false)
   })
 
-  it('reports rejection (never throws) when the steer RPC errors', async () => {
+  it('reports rejection without throwing when the redirect RPC errors', async () => {
     const requestGateway = vi.fn(async () => {
-      throw new Error('agent does not support steer')
+      throw new Error('agent does not support redirect')
     })
 
     let handle: HarnessHandle | null = null
@@ -749,18 +763,18 @@ describe('usePromptActions steerPrompt', () => {
       <Harness onReady={h => (handle = h)} refreshSessions={async () => undefined} requestGateway={requestGateway} />
     )
 
-    expect(await handle!.steerPrompt('boom')).toBe(false)
+    expect(await handle!.redirectPrompt('boom')).toBe(false)
   })
 
   it('skips the RPC entirely for empty text', async () => {
-    const requestGateway = vi.fn(async () => ({ status: 'queued' }) as never)
+    const requestGateway = vi.fn(async () => ({ status: 'redirected' }) as never)
 
     let handle: HarnessHandle | null = null
     await actRender(
       <Harness onReady={h => (handle = h)} refreshSessions={async () => undefined} requestGateway={requestGateway} />
     )
 
-    expect(await handle!.steerPrompt('   ')).toBe(false)
+    expect(await handle!.redirectPrompt('   ')).toBe(false)
     expect(requestGateway).not.toHaveBeenCalled()
   })
 })
diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.ts
index 1fde9070d4a3..852e822b1896 100644
--- a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.ts
+++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.ts
@@ -41,7 +41,7 @@ import type {
   HandoffRequestResponse,
   HandoffStateResponse,
   ImageAttachResponse,
-  SessionSteerResponse
+  SessionRedirectResponse
 } from '../../../types'
 
 import {
@@ -599,11 +599,11 @@ export function usePromptActions({
     }
   }, [activeSessionIdRef, busyRef, copy.stopFailed, requestGateway, selectedStoredSessionIdRef, updateSessionState])
 
-  // Steer = nudge the live turn without interrupting: the gateway appends the
-  // text to the next tool result so the model reads it on its next iteration
-  // (desktop parity with `/steer`). Returns false on reject (no live tool
-  // window) so the caller can fall back to queueing the words for the next turn.
-  const steerPrompt = useCallback(
+  // The desktop steering action is an immediate correction: the core cancels
+  // model generation and rebuilds the live turn with displayed reasoning and
+  // completed work intact. During a tool it waits for the safe result boundary.
+  // Returns false when the turn raced to completion so the composer can queue.
+  const redirectPrompt = useCallback(
     async (rawText: string): Promise<boolean> => {
       const text = sanitizeComposerInput(rawText).trim()
       const sessionId = activeSessionId || activeSessionIdRef.current
@@ -613,14 +613,17 @@ export function usePromptActions({
       }
 
       try {
-        const result = await requestGateway<SessionSteerResponse>('session.steer', { session_id: sessionId, text })
+        const result = await requestGateway<SessionRedirectResponse>('session.redirect', {
+          session_id: sessionId,
+          text
+        })
 
-        if (result?.status === 'queued') {
+        if (result?.status === 'redirected') {
           triggerHaptic('submit')
-          // Inline note (not a toast) so the nudge lives in the transcript next
-          // to the turn it steered. The `steer:` prefix is rendered as a codicon
-          // row by SystemMessage (see STEER_NOTE_RE), same style as slash output.
-          appendSessionTextMessage(sessionId, 'system', `steer:${text}`)
+          // Match the durable core transcript: the correction is a real user
+          // message after the interrupted assistant checkpoint, not a system
+          // note that changes role after reload.
+          appendSessionTextMessage(sessionId, 'user', text)
 
           return true
         }
@@ -806,7 +809,9 @@ export function usePromptActions({
     handoffSession,
     reloadFromMessage,
     restoreToMessage,
-    steerPrompt,
+    redirectPrompt,
+    /** @deprecated Use `redirectPrompt` — this is an active-turn redirect, not tool steer. */
+    steerPrompt: redirectPrompt,
     submitText,
     transcribeVoiceAudio
   }
diff --git a/apps/desktop/src/app/types.ts b/apps/desktop/src/app/types.ts
index 3f8da4144334..a9d09165c763 100644
--- a/apps/desktop/src/app/types.ts
+++ b/apps/desktop/src/app/types.ts
@@ -60,6 +60,11 @@ export interface SessionSteerResponse {
   text?: string
 }
 
+export interface SessionRedirectResponse {
+  status?: 'redirected' | 'rejected'
+  text?: string
+}
+
 export interface SessionTitleResponse {
   title?: string
   // True when the session row isn't persisted yet and the title was queued
diff --git a/cli.py b/cli.py
index cac922c850b0..b9641e53ee20 100644
--- a/cli.py
+++ b/cli.py
@@ -3816,7 +3816,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
             enabled=CLI_CONFIG["display"].get("persistent_output", True),
             max_lines=CLI_CONFIG["display"].get("persistent_output_max_lines", 200),
         )
-        # busy_input_mode: "interrupt" (Enter interrupts current run),
+        # busy_input_mode: "interrupt" (Enter redirects current run),
         # "queue" (Enter queues for next turn), or "steer" (Enter injects
         # mid-run via /steer, arriving after the next tool call).
         _bim = str(CLI_CONFIG["display"].get("busy_input_mode", "interrupt")).strip().lower()
@@ -13479,6 +13479,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
                 payload = (text, images) if images else text
                 if self._agent_running and not (text and _looks_like_slash_command(text)):
                     _effective_mode = self.busy_input_mode
+                    redirected = False
                     if _effective_mode == "steer":
                         # Route Enter through /steer — inject mid-run after the
                         # next tool call.  Images can't ride along (steer only
@@ -13506,15 +13507,35 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
                         preview = text if text else f"[{len(images)} image{'s' if len(images) != 1 else ''} attached]"
                         _cprint(f"  Queued for the next turn: {preview[:80]}{'...' if len(preview) > 80 else ''}")
                     elif _effective_mode == "interrupt":
-                        self._interrupt_queue.put(payload)
-                        # Debug: log to file when message enters interrupt queue
-                        try:
-                            _dbg = _hermes_home / "interrupt_debug.log"
-                            with open(_dbg, "a", encoding="utf-8") as _f:
-                                _f.write(f"{time.strftime('%H:%M:%S')} ENTER: queued interrupt msg={str(payload)[:60]!r}, "
-                                         f"agent_running={self._agent_running}\n")
-                        except Exception:
-                            pass
+                        if not images and text:
+                            try:
+                                if (
+                                    self.agent is not None
+                                    and getattr(
+                                        self.agent,
+                                        "_supports_active_turn_redirect",
+                                        False,
+                                    )
+                                    is True
+                                    and hasattr(self.agent, "redirect")
+                                ):
+                                    redirected = bool(self.agent.redirect(text))
+                            except Exception:
+                                redirected = False
+                        if redirected:
+                            preview = text[:80] + ("..." if len(text) > 80 else "")
+                            _cprint(f"  {_ACCENT}↪ Redirected current turn: '{preview}'{_RST}")
+                        else:
+                            # Compatibility path for older agents, multimodal
+                            # follow-ups, or a turn that finished in the race.
+                            self._interrupt_queue.put(payload)
+                            try:
+                                _dbg = _hermes_home / "interrupt_debug.log"
+                                with open(_dbg, "a", encoding="utf-8") as _f:
+                                    _f.write(f"{time.strftime('%H:%M:%S')} ENTER: queued interrupt msg={str(payload)[:60]!r}, "
+                                             f"agent_running={self._agent_running}\n")
+                            except Exception:
+                                pass
                     # First-touch onboarding: on the very first busy-while-running
                     # event for this install, print a one-line tip explaining the
                     # /busy knob.  Flag persists to config.yaml and never fires
@@ -13528,7 +13549,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
                             mark_seen,
                         )
                         if not is_seen(CLI_CONFIG, BUSY_INPUT_FLAG):
-                            _cprint(f"  {_DIM}{busy_input_hint_cli(self.busy_input_mode)}{_RST}")
+                            _hint_mode = "redirect" if redirected else _effective_mode
+                            _cprint(f"  {_DIM}{busy_input_hint_cli(_hint_mode)}{_RST}")
                             mark_seen(_hermes_home / "config.yaml", BUSY_INPUT_FLAG)
                             CLI_CONFIG.setdefault("onboarding", {}).setdefault("seen", {})[BUSY_INPUT_FLAG] = True
                     except Exception:
diff --git a/gateway/run.py b/gateway/run.py
index 42dd61e02776..7653ffb1e9b8 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -6030,10 +6030,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             )
             effective_mode = "queue"
         steered = False
+        redirected = False
         if effective_mode == "steer":
             steer_text = (event.text or "").strip()
             can_steer = (
                 steer_text
+                and event.message_type == MessageType.TEXT
+                and not event.media_urls
+                and not event.media_types
                 and running_agent is not None
                 and running_agent is not _AGENT_PENDING_SENTINEL
                 and hasattr(running_agent, "steer")
@@ -6047,6 +6051,21 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             if not steered:
                 # Fall back to queue (merge into pending messages, no interrupt)
                 effective_mode = "queue"
+        elif (
+            effective_mode == "interrupt"
+            and event.message_type == MessageType.TEXT
+            and not event.media_urls
+            and not event.media_types
+            and running_agent is not None
+            and running_agent is not _AGENT_PENDING_SENTINEL
+            and getattr(running_agent, "_supports_active_turn_redirect", False) is True
+            and hasattr(running_agent, "redirect")
+        ):
+            try:
+                redirected = bool(running_agent.redirect((event.text or "").strip()))
+            except Exception as exc:
+                logger.warning("Gateway redirect failed for session %s: %s", session_key, exc)
+                redirected = False
 
         # Store the message so it's processed as the next turn after the
         # current run finishes (or is interrupted).  Skip this for a
@@ -6063,16 +6082,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # turn (#43066 sub-bug 2). The FIFO path gives each text its own
         # turn in arrival order while still preserving photo-burst / album
         # merge semantics for media.
-        if not steered:
+        if not steered and not redirected:
             self._queue_or_replace_pending_event(session_key, event)
 
         is_queue_mode = effective_mode == "queue"
         is_steer_mode = effective_mode == "steer"
+        is_redirect_mode = effective_mode == "interrupt" and redirected
 
         # If not in queue/steer mode, interrupt the running agent immediately.
         # This aborts in-flight tool calls and causes the agent loop to exit
         # at the next check point.
-        if effective_mode == "interrupt" and running_agent and running_agent is not _AGENT_PENDING_SENTINEL:
+        if (
+            effective_mode == "interrupt"
+            and not redirected
+            and running_agent
+            and running_agent is not _AGENT_PENDING_SENTINEL
+        ):
             try:
                 _interrupt_text = event.text
                 _media_urls = getattr(event, "media_urls", None) or []
@@ -6170,6 +6195,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 f"⏩ Steered into current run{status_detail}. "
                 f"Your message arrives after the next tool call."
             )
+        elif is_redirect_mode:
+            message = (
+                f"↪ Redirected current run{status_detail}. "
+                f"I'll adjust using your correction."
+            )
         elif is_queue_mode and demoted_for_subagents:
             # #30170 — explain the demotion so the user knows their
             # follow-up didn't accidentally kill the subagent and
@@ -6211,6 +6241,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     _hint_mode = "steer"
                 elif is_queue_mode:
                     _hint_mode = "queue"
+                elif is_redirect_mode:
+                    _hint_mode = "redirect"
                 else:
                     _hint_mode = "interrupt"
                 message = (
@@ -10952,7 +10984,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 # is empty, the agent lacks steer(), or steer() rejects.
                 steer_text = (event.text or "").strip()
                 steered = False
-                if steer_text and hasattr(running_agent, "steer"):
+                if (
+                    event.message_type == MessageType.TEXT
+                    and not event.media_urls
+                    and not event.media_types
+                    and steer_text
+                    and hasattr(running_agent, "steer")
+                ):
                     try:
                         steered = bool(running_agent.steer(steer_text))
                     except Exception as exc:
@@ -10996,6 +11034,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 )
                 self._queue_or_replace_pending_event(_quick_key, event)
                 return None
+            # Text-only corrections redirect the live turn (preserving
+            # displayed context) when the runtime supports it; media/voice and
+            # older runtimes fall back to the proven interrupt path below.
+            if (
+                event.message_type == MessageType.TEXT
+                and not event.media_urls
+                and not event.media_types
+                and getattr(running_agent, "_supports_active_turn_redirect", False)
+                is True
+                and hasattr(running_agent, "redirect")
+            ):
+                try:
+                    if running_agent.redirect((event.text or "").strip()):
+                        logger.debug("PRIORITY redirect for session %s", _quick_key)
+                        return None
+                except Exception as exc:
+                    logger.warning(
+                        "PRIORITY redirect failed for session %s: %s",
+                        _quick_key,
+                        exc,
+                    )
             logger.debug("PRIORITY interrupt for session %s", _quick_key)
             _interrupt_text = event.text
             _media_urls = getattr(event, "media_urls", None) or []
diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py
index 4c98e70dbc66..f28c07602312 100644
--- a/hermes_cli/cli_commands_mixin.py
+++ b/hermes_cli/cli_commands_mixin.py
@@ -2620,7 +2620,7 @@ class CLICommandsMixin:
             /busy status        Show current busy input mode
             /busy queue         Queue input for the next turn instead of interrupting
             /busy steer         Inject Enter mid-run via /steer (after next tool call)
-            /busy interrupt     Interrupt the current run on Enter (default)
+            /busy interrupt     Redirect the current run on Enter (default)
         """
         from cli import _ACCENT, _DIM, _RST, _cprint, save_config_value
         parts = cmd.strip().split(maxsplit=1)
@@ -2631,7 +2631,7 @@ class CLICommandsMixin:
             elif self.busy_input_mode == "steer":
                 _behavior = "steers into current run (after next tool call)"
             else:
-                _behavior = "interrupts current run"
+                _behavior = "redirects current run immediately"
             _cprint(f"  {_DIM}Enter while busy: {_behavior}{_RST}")
             _cprint(f"  {_DIM}Usage: /busy [queue|steer|interrupt|status]{_RST}")
             return
@@ -2649,7 +2649,7 @@ class CLICommandsMixin:
             elif arg == "steer":
                 behavior = "Enter will steer your message into the current run (after the next tool call)."
             else:
-                behavior = "Enter will interrupt the current run while Hermes is busy."
+                behavior = "Enter will redirect the current run while Hermes is busy; /stop still cancels it."
             _cprint(f"  {_ACCENT}✓ Busy input mode set to '{arg}' (saved to config){_RST}")
             _cprint(f"  {_DIM}{behavior}{_RST}")
         else:
diff --git a/tests/acp_adapter/test_acp_commands.py b/tests/acp_adapter/test_acp_commands.py
index 4a95367a6ba5..4f8ca69ed85e 100644
--- a/tests/acp_adapter/test_acp_commands.py
+++ b/tests/acp_adapter/test_acp_commands.py
@@ -16,13 +16,19 @@ class FakeAgent:
         self.disabled_toolsets = []
         self.tools = []
         self.valid_tool_names = set()
+        self._supports_active_turn_redirect = True
         self.steers = []
+        self.redirects = []
         self.runs = []
 
     def steer(self, text):
         self.steers.append(text)
         return True
 
+    def redirect(self, text):
+        self.redirects.append(text)
+        return True
+
     def run_conversation(self, *, user_message, conversation_history, task_id, **kwargs):
         self.runs.append(user_message)
         messages = list(conversation_history or [])
@@ -147,6 +153,62 @@ async def test_acp_steer_after_zed_interrupt_replays_interrupted_prompt_with_gui
     assert state.interrupted_prompt_text == ""
 
 
+@pytest.mark.asyncio
+async def test_acp_plain_correction_redirects_running_turn():
+    acp_agent, state, fake, _conn = make_agent_and_state()
+    state.is_running = True
+
+    response = await acp_agent.prompt(
+        session_id=state.session_id,
+        prompt=[TextContentBlock(type="text", text="No, use Postgres instead")],
+    )
+
+    assert response.stop_reason == "end_turn"
+    assert fake.redirects == ["No, use Postgres instead"]
+    assert state.queued_prompts == []
+    assert fake.runs == []
+
+
+@pytest.mark.asyncio
+async def test_acp_plain_correction_after_cancel_replays_original_prompt():
+    acp_agent, state, fake, _conn = make_agent_and_state()
+    state.interrupted_prompt_text = "implement it with SQLite"
+
+    response = await acp_agent.prompt(
+        session_id=state.session_id,
+        prompt=[TextContentBlock(type="text", text="No, use Postgres instead")],
+    )
+
+    assert response.stop_reason == "end_turn"
+    assert fake.runs == [
+        "implement it with SQLite\n\n"
+        "User correction/guidance after interrupt: No, use Postgres instead"
+    ]
+    assert state.interrupted_prompt_text == ""
+
+
+@pytest.mark.asyncio
+async def test_acp_cancel_publishes_hard_stop_while_holding_runtime_lock():
+    acp_agent, state, fake, _conn = make_agent_and_state()
+    state.is_running = True
+    state.current_prompt_text = "original request"
+    observed = {}
+
+    def interrupt():
+        acquired = state.runtime_lock.acquire(blocking=False)
+        observed["lock_held"] = not acquired
+        if acquired:
+            state.runtime_lock.release()
+
+    fake.interrupt = interrupt
+
+    await acp_agent.cancel(state.session_id)
+
+    assert observed["lock_held"] is True
+    assert state.cancel_event.is_set()
+    assert state.interrupted_prompt_text == "original request"
+
+
 @pytest.mark.asyncio
 async def test_acp_steer_on_idle_session_runs_as_regular_prompt():
     # /steer on an idle session (no running turn, nothing to salvage) should
diff --git a/tests/gateway/test_busy_session_ack.py b/tests/gateway/test_busy_session_ack.py
index 66c4672f21c3..8b8598757992 100644
--- a/tests/gateway/test_busy_session_ack.py
+++ b/tests/gateway/test_busy_session_ack.py
@@ -211,6 +211,54 @@ class TestBusySessionAck:
         # Verify agent interrupt was called
         agent.interrupt.assert_called_once_with("Are you working?")
 
+    @pytest.mark.asyncio
+    async def test_interrupt_mode_redirects_capable_core_agent(self):
+        runner, _sentinel = _make_runner()
+        runner._busy_input_mode = "interrupt"
+        adapter = _make_adapter()
+        event = _make_event(text="No, use Postgres")
+        sk = build_session_key(event.source)
+
+        agent = MagicMock()
+        agent._supports_active_turn_redirect = True
+        agent.redirect.return_value = True
+        agent._active_children = []
+        agent.get_activity_summary.return_value = {}
+        runner._running_agents[sk] = agent
+        runner.adapters[event.source.platform] = adapter
+
+        assert await runner._handle_active_session_busy_message(event, sk) is True
+
+        agent.redirect.assert_called_once_with("No, use Postgres")
+        agent.interrupt.assert_not_called()
+        assert sk not in adapter._pending_messages
+        content = adapter._send_with_retry.call_args.kwargs.get("content", "")
+        assert "Redirected current run" in content
+
+    @pytest.mark.asyncio
+    async def test_text_event_with_attachment_is_queued_not_redirected(self):
+        runner, _sentinel = _make_runner()
+        runner._busy_input_mode = "interrupt"
+        adapter = _make_adapter()
+        event = _make_event(text="use this attachment")
+        # QQBot and other adapters may retain unknown attachment MIME types on
+        # a TEXT event, so message_type alone is not a safe redirect gate.
+        event.media_urls = ["https://example.invalid/attachment.bin"]
+        event.media_types = ["application/octet-stream"]
+        sk = build_session_key(event.source)
+
+        agent = MagicMock()
+        agent._supports_active_turn_redirect = True
+        agent._active_children = []
+        runner._running_agents[sk] = agent
+        runner.adapters[event.source.platform] = adapter
+
+        assert await runner._handle_active_session_busy_message(event, sk) is True
+
+        agent.redirect.assert_not_called()
+        assert adapter._pending_messages[sk] is event
+        assert adapter._pending_messages[sk].media_urls == event.media_urls
+
     @pytest.mark.asyncio
     async def test_queue_mode_suppresses_interrupt_and_updates_ack(self):
         """When busy_input_mode is 'queue', message is queued WITHOUT interrupt."""
diff --git a/tests/test_tui_gateway_queue_on_busy.py b/tests/test_tui_gateway_queue_on_busy.py
index 804a3b5505cf..0543d7e8b6b2 100644
--- a/tests/test_tui_gateway_queue_on_busy.py
+++ b/tests/test_tui_gateway_queue_on_busy.py
@@ -1,12 +1,12 @@
-"""A prompt that lands mid-turn is interrupted + queued, never dropped.
+"""A prompt that lands mid-turn is redirected or queued, never dropped.
 
 Before this, ``prompt.submit`` on a running session returned ``session busy``,
 forcing clients into a deadline-bounded busy-retry. When turn teardown outlived
 the deadline — e.g. a slow, non-interruptible tool (``web_search``) still
 running when the user hit stop — the resubmitted message was silently dropped
 ("it just doesn't listen"). The gateway now applies the ``busy_input_mode``
-policy: interrupt the live turn (default) and queue the message to run as the
-next turn, drained in ``run``'s tail.
+policy: redirect the live turn by default, with the legacy interrupt + queue
+path retained as a compatibility fallback.
 """
 
 import threading
@@ -49,7 +49,26 @@ def test_enqueue_merges_second_arrival_losslessly():
 
 # ── _handle_busy_submit (policy) ───────────────────────────────────────────
 
-def test_busy_interrupt_mode_interrupts_and_queues(monkeypatch):
+def test_busy_interrupt_mode_redirects_active_turn(monkeypatch):
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    seen = []
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: seen.append(text) or True,
+        interrupt=lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("redirect must not hard-interrupt")
+        ),
+    )
+    session = _session(agent=agent, running=True)
+
+    resp = server._handle_busy_submit("r1", "sid", session, "redirect", "ws-1")
+
+    assert resp["result"]["status"] == "redirected"
+    assert seen == ["redirect"]
+    assert session.get("queued_prompt") is None
+
+
+def test_busy_interrupt_mode_falls_back_for_legacy_agent(monkeypatch):
     monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
     calls = {"interrupt": 0}
     agent = types.SimpleNamespace(interrupt=lambda *a, **k: calls.__setitem__("interrupt", calls["interrupt"] + 1))
@@ -134,6 +153,67 @@ def test_busy_helper_retries_when_turn_finished(monkeypatch):
     assert session.get("queued_prompt") is None
 
 
+def test_busy_interrupt_mode_normalizes_rich_text_before_redirect(monkeypatch):
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    seen = []
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: seen.append(text) or True,
+        interrupt=lambda *a, **k: None,
+    )
+    session = _session(agent=agent, running=True)
+    rich = [{"type": "text", "text": "  redirect me  "}]
+
+    resp = server._handle_busy_submit(
+        "r1",
+        "sid",
+        session,
+        rich,
+        "ws-1",
+    )
+
+    assert resp["result"]["status"] == "redirected"
+    assert seen == ["redirect me"]
+    assert session.get("queued_prompt") is None
+
+
+def test_busy_queue_fallback_preserves_original_structured_text(monkeypatch):
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    rich = [{"type": "text", "text": "  keep me  "}]
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: False,
+        interrupt=lambda *a, **k: None,
+    )
+    session = _session(agent=agent, running=True)
+
+    resp = server._handle_busy_submit("r1", "sid", session, rich, "ws-1")
+
+    assert resp["result"]["status"] == "queued"
+    assert session["queued_prompt"]["text"] == rich
+
+
+def test_busy_interrupt_mode_queues_multimodal_payload_instead_of_redirect(monkeypatch):
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    seen = []
+    rich = [
+        {"type": "text", "text": "caption"},
+        {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}},
+    ]
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: seen.append(text) or True,
+        interrupt=lambda *a, **k: None,
+    )
+    session = _session(agent=agent, running=True)
+
+    resp = server._handle_busy_submit("r1", "sid", session, rich, "ws-1")
+
+    assert resp["result"]["status"] == "queued"
+    assert seen == []
+    assert session["queued_prompt"]["text"] == rich
+
+
 # ── _drain_queued_prompt ───────────────────────────────────────────────────
 
 def test_drain_fires_queued_prompt_and_claims_running(monkeypatch):
diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py
index 125d3eb584e1..48fa63c7b821 100644
--- a/tests/test_tui_gateway_server.py
+++ b/tests/test_tui_gateway_server.py
@@ -6835,6 +6835,35 @@ def test_session_steer_errors_when_agent_has_no_steer_method():
     assert resp["error"]["code"] == 4010
 
 
+def test_session_redirect_calls_capable_core_agent(monkeypatch):
+    calls = []
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: calls.append(text) or True,
+    )
+    session = _session(agent=agent)
+    server._sessions["sid"] = session
+    try:
+        before = session.get("last_active")
+        resp = server.handle_request(
+            {
+                "id": "1",
+                "method": "session.redirect",
+                "params": {"session_id": "sid", "text": "use Postgres"},
+            }
+        )
+    finally:
+        server._sessions.pop("sid", None)
+
+    assert resp["result"] == {
+        "status": "redirected",
+        "text": "use Postgres",
+    }
+    assert calls == ["use Postgres"]
+    assert session.get("last_active") is not None
+    assert before is None or session["last_active"] >= before
+
+
 def test_session_info_includes_mcp_servers(monkeypatch):
     fake_status = [
         {"name": "github", "transport": "http", "tools": 12, "connected": True},
diff --git a/tui_gateway/server.py b/tui_gateway/server.py
index 03d8492d932c..15ed6b4a7ad5 100644
--- a/tui_gateway/server.py
+++ b/tui_gateway/server.py
@@ -5553,6 +5553,38 @@ def _coerce_message_text(content: Any) -> str:
     return str(content)
 
 
+_TEXT_ONLY_BUSY_PART_KINDS = frozenset({"text", "input_text", "output_text"})
+
+
+def _is_text_only_busy_payload(content: Any) -> bool:
+    """True when a busy submit carries only plain text, not attachments/media."""
+    if content is None:
+        return False
+    if isinstance(content, (str, int, float)):
+        return True
+    if isinstance(content, list):
+        if not content:
+            return False
+        for part in content:
+            if isinstance(part, str):
+                continue
+            if not isinstance(part, dict):
+                return False
+            kind = part.get("type")
+            if kind in _TEXT_ONLY_BUSY_PART_KINDS:
+                continue
+            if kind is None and isinstance(part.get("text"), str):
+                continue
+            return False
+        return True
+    if isinstance(content, dict):
+        kind = content.get("type")
+        if kind in _TEXT_ONLY_BUSY_PART_KINDS:
+            return True
+        return kind is None and isinstance(content.get("text"), str)
+    return False
+
+
 def _history_to_messages(history: list[dict]) -> list[dict]:
     messages = []
     tool_call_args = {}
@@ -5761,15 +5793,13 @@ def _handle_busy_submit(
     a turn is in flight, instead of rejecting it with ``session busy``.
 
     The old rejection forced clients into a deadline-bounded busy-retry that
-    silently dropped the send when turn teardown outlived the deadline (e.g. a
-    slow, non-interruptible tool like ``web_search`` running when the user hits
-    stop). The message is instead queued to run as the next turn — and, for the
-    default ``interrupt`` policy, the live turn is interrupted so it winds down
-    promptly. Drained in ``run``'s tail (see ``_run_prompt_submit``).
+    silently dropped the send when turn teardown outlived the deadline. The
+    default policy now redirects a capable core agent in place; older agents
+    retain the proven interrupt-and-queue path drained from ``run``'s tail.
 
-    Modes: ``interrupt`` (default) → interrupt + queue; ``queue`` → queue
-    without interrupting; ``steer`` → inject into the live turn if accepted,
-    else queue.
+    Modes: ``interrupt`` (default) → redirect the live turn, falling back to
+    hard interrupt + queue for older agents; ``queue`` → queue without
+    interrupting; ``steer`` → inject after the current atomic action.
     """
     mode = _load_busy_input_mode()
     agent = session.get("agent")
@@ -5778,14 +5808,34 @@ def _handle_busy_submit(
             # The turn ended between prompt.submit's first busy check and this
             # helper. Let the caller retry and claim the now-idle session.
             return None
-    if mode == "steer" and agent is not None and hasattr(agent, "steer"):
+    text_only = _is_text_only_busy_payload(text)
+    plain_text = _coerce_message_text(text).strip() if text_only else ""
+    if mode == "steer" and text_only and plain_text and agent is not None and hasattr(agent, "steer"):
         try:
-            if agent.steer(text):
+            if agent.steer(plain_text):
                 with session["history_lock"]:
                     session["last_active"] = time.time()
                 return _ok(rid, {"status": "steered"})
         except Exception:
             pass  # fall through to queue
+    # Text-only corrections redirect the live turn in place when the runtime
+    # supports it; media/attachment payloads and older agents fall through to
+    # the proven interrupt + queue path below.
+    if (
+        mode == "interrupt"
+        and text_only
+        and plain_text
+        and agent is not None
+        and getattr(agent, "_supports_active_turn_redirect", False) is True
+        and hasattr(agent, "redirect")
+    ):
+        try:
+            if agent.redirect(plain_text):
+                with session["history_lock"]:
+                    session["last_active"] = time.time()
+                return _ok(rid, {"status": "redirected"})
+        except Exception:
+            pass  # preserve the proven interrupt + queue fallback below
     # Queue before asking the live turn to stop. In particular, never call a
     # provider or compute-host method while holding history_lock: an interrupt
     # can wait behind the very operation it is trying to cancel.
@@ -9585,6 +9635,34 @@ def _(rid, params: dict) -> dict:
     return _ok(rid, {"status": "queued" if accepted else "rejected", "text": text})
 
 
+@method("session.redirect")
+def _(rid, params: dict) -> dict:
+    """Redirect the active model turn while preserving valid work/context."""
+    text = (params.get("text") or "").strip()
+    if not text:
+        return _err(rid, 4002, "text is required")
+    session, err = _sess_nowait(params, rid)
+    if err:
+        return err
+    agent = session.get("agent")
+    if (
+        agent is None
+        or getattr(agent, "_supports_active_turn_redirect", False) is not True
+        or not hasattr(agent, "redirect")
+    ):
+        return _err(rid, 4010, "agent does not support active-turn redirect")
+    try:
+        accepted = agent.redirect(text)
+    except Exception as exc:
+        return _err(rid, 5000, f"redirect failed: {exc}")
+    if accepted:
+        session["last_active"] = time.time()
+    return _ok(
+        rid,
+        {"status": "redirected" if accepted else "rejected", "text": text},
+    )
+
+
 @method("terminal.resize")
 def _(rid, params: dict) -> dict:
     session, err = _sess_nowait(params, rid)
diff --git a/ui-tui/src/app/useSubmission.ts b/ui-tui/src/app/useSubmission.ts
index a5c484cc288c..a70b1fd7390d 100644
--- a/ui-tui/src/app/useSubmission.ts
+++ b/ui-tui/src/app/useSubmission.ts
@@ -158,9 +158,9 @@ export function useSubmission(opts: UseSubmissionOptions) {
   //   - 'steer'     : inject into the current turn via session.steer; falls
   //                   back to queue when steer is rejected (no agent / no
   //                   tool window).
-  //   - 'interrupt' (default): queue the text + interrupt with `keepBusy`; the
-  //                   busy→false settle edge drains it once (desktop parity).
-  //                   No optimistic send → no duplicate bubble / race note.
+  //   - 'interrupt' (default): submit immediately; the backend redirects the
+  //                   active model request (or safely steers after a tool),
+  //                   with legacy interrupt + queue as its compatibility path.
   //
   // `opts.fallbackToFront` re-inserts at the queue head (queue-edit picks keep
   // their position); the mainline submit path appends.
@@ -201,14 +201,13 @@ export function useSubmission(opts: UseSubmissionOptions) {
         return
       }
 
-      // 'interrupt': queue + interrupt(keepBusy); the settle edge drains it once.
-      enqueueText()
-
-      if (live.sid) {
-        turnController.interruptTurn({ appendMessage, gw, sid: live.sid, sys }, { keepBusy: true })
-      }
+      // The gateway owns the atomic redirect decision because it knows whether
+      // the agent is in model generation, tool execution, or an older runtime.
+      // Reuse the normal submit pipeline so the correction gets its user bubble
+      // and file-drop interpolation exactly once.
+      send(full)
     },
-    [appendMessage, composerActions, composerRefs, gw, sys]
+    [composerActions, composerRefs, gw, send, sys]
   )
 
   const dispatchSubmission = useCallback(