From 8b1c1dfa1d68eb4a1e212c181563d9b34d40bd5d Mon Sep 17 00:00:00 2001 From: ethernet Date: Wed, 15 Jul 2026 17:54:21 -0400 Subject: [PATCH] fix(desktop): review fixes for the agent-side TurnQueue Review findings on the TurnQueue PR, fixed in one pass: 1. "Send now" on an idle session was a silent no-op: session.queue.promote reordered the queue but never drained it, so the promoted entry (and the drainNextQueued rescue gesture built on it) just sat there. Promote now fires _drain_queued_prompt in a thread when the session is idle, same as an idle session.queue.add. 2. A drained entry whose dispatch raised was lost: _drain_queued_prompt popped the entry and emitted queue.drained (painting a user turn in the client transcript) before _run_prompt_submit. On exception the entry was gone and the transcript lied. The drain now requeues the entry at the head (same id, so client mirrors stay consistent) via the new TurnQueue.requeue_front(), and queue.drained is only emitted after a successful dispatch. 3. Multi-line steers never settled: settlePendingSteer split the applied text into a line-set, so an entry that itself contained newlines (Cmd+Enter on a multi-line draft) matched nothing and pinned a "Steering..." row forever. Now matches by whole-entry containment, plus a message.complete backstop sweep (a steer can't outlive its turn: applied, dropped, or re-queued as the next turn). 4. Speculative surface removed per the contribution rubric: QueuedTurn.mode (written, never read), QueuedTurn.attachments (clients resolve attachments to @file: refs at enqueue time), enqueue_front() (replaced by the requeue_front() that finding 2 actually needs), and the keep_queue param on session.interrupt (documented for a promote+interrupt flow that actually interrupts via agent.interrupt() directly, so it was dead). Also: unused sessionId arg dropped from useComposerQueue, the steer-event lambda no longer shadows the enclosing text parameter, and a rejected session.queue.add now surfaces an i18n'd error toast instead of silently no-oping (draft is kept either way). Tests: idle-promote drains immediately, failed dispatch requeues at head without emitting queue.drained, interrupt clears the queue, multi-line steer settles. 16 gateway tests pass; desktop tsc/eslint/vitest clean. --- agent/turn_queue.py | 31 +++------ .../chat/composer/hooks/use-composer-queue.ts | 13 ++-- apps/desktop/src/app/chat/composer/index.tsx | 3 +- .../hooks/use-message-stream/gateway-event.ts | 8 ++- apps/desktop/src/i18n/en.ts | 2 + apps/desktop/src/i18n/types.ts | 2 + apps/desktop/src/i18n/zh.ts | 2 + apps/desktop/src/store/composer-queue.test.ts | 8 +++ apps/desktop/src/store/composer-queue.ts | 30 ++++++--- tests/tui_gateway/test_turn_queue.py | 67 +++++++++++++++++++ tui_gateway/server.py | 66 ++++++++++++------ 11 files changed, 172 insertions(+), 60 deletions(-) diff --git a/agent/turn_queue.py b/agent/turn_queue.py index 8e2f8bf4ac7..0aaa53ac6f6 100644 --- a/agent/turn_queue.py +++ b/agent/turn_queue.py @@ -28,10 +28,8 @@ class QueuedTurn: id: str text: str - mode: str = "queue" # "queue" | "interrupt" | "steer" transport: Any = None # pinned transport for the drained turn queued_at: float = field(default_factory=time.time) - attachments: list = field(default_factory=list) # Where the entry came from: "queue" (explicit session.queue.add — the # client shows it in a queue panel, not the transcript) or "busy_submit" # (a prompt.submit that landed mid-turn — the client already echoed it @@ -44,9 +42,7 @@ class QueuedTurn: return { "id": self.id, "text": self.text, - "mode": self.mode, "queued_at": self.queued_at, - "attachments": self.attachments, "source": self.source, } @@ -69,9 +65,7 @@ class TurnQueue: def enqueue( self, text: str, - mode: str = "queue", transport: Any = None, - attachments: list | None = None, source: str = "queue", ) -> QueuedTurn: """Add a prompt to the back of the queue. @@ -86,30 +80,21 @@ class TurnQueue: entry = QueuedTurn( id=str(uuid.uuid4()), text=text, - mode=mode, transport=transport, - attachments=list(attachments) if attachments else [], source=source, ) with self._lock: self._entries.append(entry) return entry - def enqueue_front( - self, - text: str, - mode: str = "queue", - transport: Any = None, - attachments: list | None = None, - ) -> QueuedTurn: - """Add a prompt to the *front* of the queue (priority send).""" - entry = QueuedTurn( - id=str(uuid.uuid4()), - text=text, - mode=mode, - transport=transport, - attachments=list(attachments) if attachments else [], - ) + def requeue_front(self, entry: QueuedTurn) -> QueuedTurn: + """Put a drained entry back at the *front* of the queue. + + Used by the gateway when the turn dispatch for a drained entry raises + — the pop already happened, so the entry (same id, so client mirrors + stay consistent) is restored to the head for the next drain attempt + instead of being silently lost. + """ with self._lock: self._entries.insert(0, entry) return entry diff --git a/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts b/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts index ddda3de145d..286369a1cef 100644 --- a/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts +++ b/apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts @@ -27,7 +27,6 @@ interface UseComposerQueueArgs { onQueue: ChatBarProps['onQueue'] queueEditRef: RefObject queueSessionKey: ChatBarProps['queueSessionKey'] - sessionId: string | null | undefined } /** @@ -47,8 +46,7 @@ export function useComposerQueue({ loadIntoComposer, onQueue, queueEditRef, - queueSessionKey, - sessionId: _sessionId + queueSessionKey }: UseComposerQueueArgs) { const scope = useComposerScope() @@ -176,10 +174,11 @@ export function useComposerQueue({ return true }, [activeQueueSessionKey, attachments, clearDraft, draftRef, onQueue, scope.attachments]) - // "Send now": promote the entry to the queue head and interrupt the live - // turn on the gateway (queue preserved). The gateway drains the promoted - // entry the moment the turn unwinds — no client-side send at all. When - // idle the gateway drains promoted entries on the same RPC. + // "Send now": promote the entry to the queue head on the gateway. While a + // turn is live, also interrupt it (queue preserved) — the gateway drains + // the promoted entry the moment the turn unwinds. When idle, the gateway + // drains the promoted entry immediately on the same RPC. No client-side + // send in either case. const sendQueuedNow = useCallback( (id: string) => { if (!activeQueueSessionKey || id === queueEdit?.entryId) { diff --git a/apps/desktop/src/app/chat/composer/index.tsx b/apps/desktop/src/app/chat/composer/index.tsx index 23f2966fbb1..68fc00ed87e 100644 --- a/apps/desktop/src/app/chat/composer/index.tsx +++ b/apps/desktop/src/app/chat/composer/index.tsx @@ -204,8 +204,7 @@ export function ChatBar({ loadIntoComposer, onQueue, queueEditRef, - queueSessionKey, - sessionId + queueSessionKey }) const statusStackVisible = queuedPrompts.length > 0 || statusPresent diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts index 29a9e3ca110..dd04cdaa213 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts @@ -15,7 +15,7 @@ import { isProviderSetupErrorMessage } from '@/lib/provider-setup-errors' import { reconcileApprovalModeForProfile } from '@/store/approval-mode' import { clearClarifyRequest, setClarifyRequest } from '@/store/clarify' import { setSessionCompacting } from '@/store/compaction' -import { setSessionQueue, settlePendingSteer } from '@/store/composer-queue' +import { clearPendingSteers, setSessionQueue, settlePendingSteer } from '@/store/composer-queue' import { refreshBackgroundProcesses } from '@/store/composer-status' import { $gateway } from '@/store/gateway' import { dispatchNativeNotification } from '@/store/native-notifications' @@ -380,6 +380,12 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { // last item stuck pending/in_progress. Finished lists keep their linger. clearActiveSessionTodos(sessionId) setSessionCompacting(sessionId, false) + // A steer cannot outlive its turn: it was either injected + // (steer.applied settled it), dropped by an interrupt + // (steer.dropped), or handed back as pending_steer and re-queued as + // the next turn by the gateway. Sweep any stragglers so a missed + // event can't pin a "Steering…" row forever. + clearPendingSteers(sessionId) flushQueuedDeltas(sessionId) diff --git a/apps/desktop/src/i18n/en.ts b/apps/desktop/src/i18n/en.ts index a2d19de1158..f41bc63b6a3 100644 --- a/apps/desktop/src/i18n/en.ts +++ b/apps/desktop/src/i18n/en.ts @@ -1755,6 +1755,8 @@ export const en: Translations = { queueSendNext: 'Next', queueSend: 'Send', queueDelete: 'Delete', + queueRejectedTitle: 'Message not queued', + queueRejectedBody: 'The gateway did not accept the queued message. Your draft is unchanged — try again.', steerPending: 'Steering — lands at the next tool step', previewUnavailable: 'Preview unavailable', previewLabel: label => `Preview ${label}`, diff --git a/apps/desktop/src/i18n/types.ts b/apps/desktop/src/i18n/types.ts index 824a1119a79..b17afc57cb1 100644 --- a/apps/desktop/src/i18n/types.ts +++ b/apps/desktop/src/i18n/types.ts @@ -1446,6 +1446,8 @@ export interface Translations { queueSendNext: string queueSend: string queueDelete: string + queueRejectedTitle: string + queueRejectedBody: string steerPending: string previewUnavailable: string previewLabel: (label: string) => string diff --git a/apps/desktop/src/i18n/zh.ts b/apps/desktop/src/i18n/zh.ts index 9afc67021b3..33bfe32ef83 100644 --- a/apps/desktop/src/i18n/zh.ts +++ b/apps/desktop/src/i18n/zh.ts @@ -1931,6 +1931,8 @@ export const zh: Translations = { queueSendNext: '下一个', queueSend: '发送', queueDelete: '删除', + queueRejectedTitle: '消息未加入队列', + queueRejectedBody: '网关未接受该排队消息。草稿未变 — 请重试。', steerPending: '正在引导 — 将在下一个工具步骤生效', previewUnavailable: '预览不可用', previewLabel: label => `预览 ${label}`, diff --git a/apps/desktop/src/store/composer-queue.test.ts b/apps/desktop/src/store/composer-queue.test.ts index 3aa73e07652..e5e979642f8 100644 --- a/apps/desktop/src/store/composer-queue.test.ts +++ b/apps/desktop/src/store/composer-queue.test.ts @@ -163,5 +163,13 @@ describe('composer-queue (gateway-backed)', () => { expect(getPendingSteers(SESSION_KEY).map(s => s.text)).toEqual(['stay']) }) + + it('settles a multi-line steer entry (Cmd+Enter on a multi-line draft)', () => { + addPendingSteer(SESSION_KEY, 'line one\nline two') + + settlePendingSteer(SESSION_KEY, 'line one\nline two') + + expect(getPendingSteers(SESSION_KEY)).toEqual([]) + }) }) }) diff --git a/apps/desktop/src/store/composer-queue.ts b/apps/desktop/src/store/composer-queue.ts index a8c71ff876f..1132b129f8d 100644 --- a/apps/desktop/src/store/composer-queue.ts +++ b/apps/desktop/src/store/composer-queue.ts @@ -15,7 +15,10 @@ import { atom } from 'nanostores' +import { translateNow } from '@/i18n' + import type { ComposerAttachment } from './composer' +import { notify } from './notifications' export type GatewayRequester = (method: string, params?: Record) => Promise @@ -139,6 +142,16 @@ export const enqueueQueuedPrompt = async ( }) if (result?.status !== 'queued') { + // Surface the reject (gateway unreachable, or a session whose agent + // isn't built yet → 4010) instead of silently no-oping; the caller + // keeps the draft either way, so the words survive. + notify({ + id: 'composer-queue-rejected', + kind: 'error', + title: translateNow('composer.queueRejectedTitle'), + message: translateNow('composer.queueRejectedBody') + }) + return null } @@ -332,7 +345,9 @@ export const addPendingSteer = (key: string | null | undefined, text: string) => /** Settle pending steers covered by an applied/dropped gateway event. The * agent concatenates queued-up steers with newlines before injecting, so one - * event can cover several pending entries — match by containment. */ + * event can cover several pending entries — and each entry can itself be + * multi-line (Cmd+Enter on a multi-line draft), so line-set matching would + * miss it. Match by whole-entry containment in the applied text instead. */ export const settlePendingSteer = (key: string | null | undefined, text: string) => { const sid = sidOf(key) @@ -340,16 +355,15 @@ export const settlePendingSteer = (key: string | null | undefined, text: string) return } - const settled = new Set( - text - .split('\n') - .map(line => line.trim()) - .filter(Boolean) - ) + const applied = text.trim() writeSteers( sid, - steersFor(sid).filter(entry => !settled.has(entry.text.trim())) + steersFor(sid).filter(entry => { + const entryText = entry.text.trim() + + return entryText.length === 0 || !applied.includes(entryText) + }) ) } diff --git a/tests/tui_gateway/test_turn_queue.py b/tests/tui_gateway/test_turn_queue.py index 99e3550aeaa..23a317b42b8 100644 --- a/tests/tui_gateway/test_turn_queue.py +++ b/tests/tui_gateway/test_turn_queue.py @@ -246,6 +246,73 @@ def test_queue_promote_interrupt_forwards_to_agent(server): assert len(q) == 2 +def test_queue_promote_idle_session_drains_immediately(server): + """Send-now on an idle session must fire the entry, not just reorder it — + there's no turn about to unwind and trigger the end-of-turn drain.""" + sid = "s1" + session = _make_session(server, sid, running=False) + q = session["agent"].turn_queue + q.enqueue("a") + b = q.enqueue("b") + drained = threading.Event() + submitted = {} + + def fake_submit(rid, s, sess, text): + submitted["text"] = text + drained.set() + + with patch.object(server, "_run_prompt_submit", side_effect=fake_submit), \ + patch.object(server, "_emit"): + resp = server.handle_request({ + "id": "r1", "method": "session.queue.promote", + "params": {"session_id": sid, "entry_id": b.id}, + }) + assert resp["result"]["promoted"] is True + assert drained.wait(timeout=5), "idle promote should drain immediately" + + assert submitted["text"] == "b" + session["agent"].interrupt.assert_not_called() + + +def test_drain_dispatch_failure_requeues_entry_at_head(server): + """A drained entry whose dispatch raises must go back to the queue head + (same id) instead of being lost — and queue.drained must NOT be emitted, + or the client would paint a user turn that never ran.""" + sid = "s1" + session = _make_session(server, sid, running=False) + entry = session["agent"].turn_queue.enqueue("precious words", source="queue") + events = [] + + with patch.object(server, "_run_prompt_submit", side_effect=RuntimeError("boom")), \ + patch.object(server, "_emit", side_effect=lambda ev, s, p=None: events.append((ev, p))): + assert server._drain_queued_prompt("rid", sid, session) is True + + kinds = [ev for ev, _ in events] + assert "queue.drained" not in kinds + (requeued,) = session["agent"].turn_queue.peek() + assert requeued.id == entry.id + assert requeued.text == "precious words" + assert session["running"] is False + + +def test_interrupt_clears_queue(server): + """Stop discards pending next-turn prompts — they'd otherwise fire the + instant the turn unwinds, the opposite of what Stop means.""" + sid = "s1" + session = _make_session(server, sid, running=True) + session["agent"].turn_queue.enqueue("stale follow-up") + session["queued_prompt"] = None + session["_run_thread"] = None + + with patch.object(server, "_emit"): + server.handle_request({ + "id": "r1", "method": "session.interrupt", + "params": {"session_id": sid}, + }) + + assert len(session["agent"].turn_queue) == 0 + + def test_queue_rpcs_unknown_session(server): for method in ( "session.queue.list", diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 35c7ad14d04..93d8882b328 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -5198,6 +5198,7 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool: # Drain from the agent's TurnQueue (preferred), falling back to the # legacy session-dict slot (populated only when no agent existed at # enqueue time). + entry = None queued = None if agent is not None and hasattr(agent, "turn_queue"): entry = agent.turn_queue.drain() @@ -5221,15 +5222,6 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool: if queued.get("transport") is not None: session["transport"] = queued["transport"] _emit_queue_update(sid, session) - # Tell clients the drained text is becoming the next user turn so they - # can move it from their queue panel into the transcript. `source` lets - # them skip painting text they already echoed optimistically - # (busy_submit) vs panel-queued entries that were never in the transcript. - _emit( - "queue.drained", - sid, - {"text": queued.get("text") or "", "source": queued.get("source") or "queue"}, - ) try: _run_prompt_submit(rid, sid, session, queued["text"]) except Exception as exc: @@ -5240,6 +5232,31 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool: ) with session["history_lock"]: session["running"] = False + # The pop already happened but the turn never dispatched — put the + # entry back at the head (same id, so client mirrors stay + # consistent) instead of silently losing the user's words. The + # next drain trigger (end-of-turn, queue.add, promote) retries it. + if entry is not None: + agent.turn_queue.requeue_front(entry) + else: + session["queued_prompt"] = { + "text": queued.get("text"), + "transport": queued.get("transport"), + } + _emit_queue_update(sid, session) + return True + # Tell clients the drained text is becoming the next user turn so they + # can move it from their queue panel into the transcript. Emitted only + # after a successful dispatch — a failed dispatch requeues the entry + # above, and painting a user row for a turn that never ran would lie. + # `source` lets clients skip painting text they already echoed + # optimistically (busy_submit) vs panel-queued entries that were never + # in the transcript. + _emit( + "queue.drained", + sid, + {"text": queued.get("text") or "", "source": queued.get("source") or "queue"}, + ) return True @@ -8236,13 +8253,11 @@ def _(rid, params: dict) -> dict: session["_turn_cancel_requested"] = True # Stop discards pending next-turn prompts too (they'd otherwise fire # the instant the turn unwinds — the opposite of what Stop means). - # `keep_queue` opts out for the promote-then-interrupt "send queued - # entry now" flow, where the queued entry must survive to be drained - # as the next turn. - if not params.get("keep_queue"): - agent = (session or {}).get("agent") - if agent is not None and hasattr(agent, "turn_queue"): - agent.turn_queue.clear() + # The "send queued entry now" flow is unaffected: session.queue.promote + # interrupts via agent.interrupt() directly, never through this RPC. + agent = (session or {}).get("agent") + if agent is not None and hasattr(agent, "turn_queue"): + agent.turn_queue.clear() session["queued_prompt"] = None _emit_queue_update(str(params.get("session_id", "")), session or {}) if not run_thread_alive: @@ -8611,7 +8626,9 @@ def _(rid, params: dict) -> dict: With ``interrupt=True``, also interrupts the live turn (keeping the queue intact) so the promoted entry fires as soon as the turn unwinds — - the "send now" gesture. + the "send now" gesture. On an idle session there is no turn to wait for: + the promoted entry is drained immediately (same as an idle + session.queue.add), so "send now" never silently no-ops. """ entry_id = str(params.get("entry_id") or "") if not entry_id: @@ -8635,6 +8652,17 @@ def _(rid, params: dict) -> dict: agent.interrupt() except Exception: pass + # Idle session → nothing will unwind and trigger the end-of-turn drain, + # so fire it now (in a thread; the drain runs a whole turn and this RPC + # must return promptly). Also the retry path for a head entry stranded + # by a failed drain attempt (which requeues it and leaves the session + # idle). + if promoted and not session.get("running"): + threading.Thread( + target=_drain_queued_prompt, + args=(rid, sid, session), + daemon=True, + ).start() return _ok(rid, {"promoted": promoted}) @@ -9178,8 +9206,8 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None: # steer.applied arrives, so the transcript reflects what the model # actually saw, when it saw it. try: - agent._on_steer_event = lambda kind, text: _emit( - f"steer.{kind}", sid, {"text": text} + agent._on_steer_event = lambda kind, steer_text: _emit( + f"steer.{kind}", sid, {"text": steer_text} ) except Exception: pass