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 a0f0c20b80e6..c9cd19c64063 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 @@ -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' }) diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts index e7cc0d439917..d8988a0c926d 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts +++ b/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 diff --git a/tests/test_tui_gateway_queue_on_busy.py b/tests/test_tui_gateway_queue_on_busy.py index 1fffb7db3f68..44f2363e5c9a 100644 --- a/tests/test_tui_gateway_queue_on_busy.py +++ b/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) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 98151358765f..14228035c434 100644 --- a/tui_gateway/server.py +++ b/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