fix(gateway): a queue drain never becomes a live-turn correction

The desktop's queue promises "run this AFTER the current turn", but the
promise broke on a race the user can't see: a drain that fired when the
client observed idle while the server was still unwinding the turn landed
in _handle_busy_submit, which applied busy_input_mode — redirecting or
interrupting the live turn with text the user explicitly queued. That's
why force-sending the queue felt like a dice roll: the same gesture
steered, interrupted, or queued depending on a millisecond settle race.

prompt.submit now carries queued:true on every fromQueue drain (composer
auto-drain, send-now, background drain), and the gateway's busy path
honors it by forcing queue semantics — never steer, never redirect,
never interrupt. Lose the race and the text simply waits its turn, which
is what queueing meant all along.

Covered on both sides: a gateway test proving a queued drain cannot touch
redirect/steer/interrupt on the live agent, and the desktop drain tests
now assert the flag rides every prompt.submit shape (direct, background,
resume-retry).

apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx

@@ -1555,6 +1555,7 @@ describe('usePromptActions submit / queue drain semantics', () => {
     expect(requestGateway).toHaveBeenCalledWith(
       'prompt.submit',
       {
+        queued: true,
         session_id: RUNTIME_SESSION_ID,
         text: 'queued message'
       },
@@ -1590,6 +1591,7 @@ describe('usePromptActions submit / queue drain semantics', () => {
     expect(requestGateway).toHaveBeenCalledWith(
       'prompt.submit',
       {
+        queued: true,
         session_id: 'rt-session-a',
         text: 'queued for background session'
       },
@@ -1646,6 +1648,7 @@ describe('usePromptActions submit / queue drain semantics', () => {
     expect(requestGateway).toHaveBeenCalledWith(
       'prompt.submit',
       {
+        queued: true,
         session_id: 'rt-session-a-rebound',
         text: 'queued for background session'
       },
@@ -1696,6 +1699,7 @@ describe('usePromptActions submit / queue drain semantics', () => {
     expect(requestGateway).toHaveBeenCalledWith(
       'prompt.submit',
       {
+        queued: true,
         session_id: RUNTIME_SESSION_ID,
         text: 'please send me'
       },
@@ -2473,11 +2477,13 @@ describe('usePromptActions sleep/wake session recovery', () => {
     expect(ok).toBe(true)
     expect(calls.map(c => c.method)).toEqual(['prompt.submit', 'session.resume', 'prompt.submit'])
     expect(calls[0]?.params).toEqual({
+      queued: true,
       session_id: 'rt-background-stale',
       text: 'queued background message after wake'
     })
     expect(calls[1]?.params).toEqual({ session_id: STORED_SESSION_ID, source: 'desktop' })
     expect(calls[2]?.params).toEqual({
+      queued: true,
       session_id: RECOVERED_SESSION_ID,
       text: 'queued background message after wake'
     })

apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts

@@ -543,7 +543,13 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) {
         const submitParams = (targetId: string) => ({
           session_id: targetId,
           text,
-          ...(interrupted && { interrupted })
+          ...(interrupted && { interrupted }),
+          // A queue drain is a "run after" message, never a live-turn
+          // correction. The flag tells the gateway's busy path to hold it for
+          // the next turn untouched — without it, losing the settle race
+          // (client saw idle, server still unwinding) redirects or interrupts
+          // the live turn with text the user explicitly queued.
+          ...(options?.fromQueue && { queued: true })
         })
 
         // On sleep/wake the gateway's in-memory session may have been cleared

tests/test_tui_gateway_queue_on_busy.py

@@ -101,6 +101,35 @@ def test_busy_queue_mode_queues_without_interrupting(monkeypatch):
     assert session["queued_prompt"]["text"] == "later"
 
 
+def test_queued_flag_overrides_mode_never_touches_live_turn(monkeypatch):
+    # A client queue drain that loses the settle race (client saw idle, server
+    # still unwinding) must stay queue-semantics: run AFTER the live turn,
+    # never redirect or interrupt it. Without the override, busy_input_mode
+    # "interrupt" turned explicitly-queued text into a live-turn correction —
+    # the "force-sending the queue is a dice roll" bug.
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: (_ for _ in ()).throw(
+            AssertionError("queued drain must not redirect the live turn")
+        ),
+        steer=lambda text: (_ for _ in ()).throw(
+            AssertionError("queued drain must not steer the live turn")
+        ),
+        interrupt=lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("queued drain must not interrupt the live turn")
+        ),
+    )
+    session = _session(agent=agent, running=True)
+
+    resp = server._handle_busy_submit(
+        "r1", "sid", session, "next turn text", "ws-1", queued=True
+    )
+
+    assert resp["result"]["status"] == "queued"
+    assert session["queued_prompt"]["text"] == "next turn text"
+
+
 def test_busy_steer_mode_injects_when_accepted(monkeypatch):
     monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "steer")
     agent = types.SimpleNamespace(steer=lambda text: True, interrupt=lambda *a, **k: None)

tui_gateway/server.py

@@ -6661,7 +6661,7 @@ def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None:
 
 
 def _handle_busy_submit(
-    rid, sid: str, session: dict, text: Any, transport: Any
+    rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False
 ) -> dict | None:
     """Apply the ``display.busy_input_mode`` policy to a prompt that lands while
     a turn is in flight, instead of rejecting it with ``session busy``.
@@ -6674,8 +6674,15 @@ def _handle_busy_submit(
     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.
+
+    ``queued=True`` (client's queue drain, ``prompt.submit`` param) overrides
+    the mode entirely: the message was explicitly queued as "run after", so it
+    must NEVER become a live-turn correction or interrupt. Without this, a
+    drain that loses the settle race (client observed idle, server still
+    unwinding the turn) redirected the live turn with next-turn text — queue
+    semantics betrayed by a millisecond race the user can't see.
     """
-    mode = _load_busy_input_mode()
+    mode = "queue" if queued else _load_busy_input_mode()
     agent = session.get("agent")
     with session["history_lock"]:
         if not session.get("running"):
@@ -10853,7 +10860,10 @@ def _(rid, params: dict) -> dict:
                 busy_transport = t or session.get("transport")
             else:
                 break
-        busy_response = _handle_busy_submit(rid, sid, session, text, busy_transport)
+        busy_response = _handle_busy_submit(
+            rid, sid, session, text, busy_transport,
+            queued=bool(params.get("queued")),
+        )
         if busy_response is not None:
             return busy_response
         # The old turn finished between the two lock acquisitions. Retry the