diff --git a/agent/agent_init.py b/agent/agent_init.py index 0c700c279b9..6fdbc82d880 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -597,6 +597,21 @@ def init_agent( agent._pending_steer: Optional[str] = None agent._pending_steer_lock = threading.Lock() + # Steer lifecycle observer — set by the gateway/CLI to learn when a + # pending steer is actually injected into the conversation ("applied") + # or discarded by a hard interrupt ("dropped"). Lets clients render the + # steer in the transcript at the moment it really lands instead of + # optimistically painting it the instant the RPC returns. + # Signature: callback(kind: str, text: str) — kind is "applied"|"dropped". + agent._on_steer_event = None + + # Turn queue — prompts to run as the next turn(s) after the current one + # finishes. Replaces the ad-hoc session["queued_prompt"] dict slot in the + # TUI gateway and the localStorage-backed queue in the desktop renderer. + # Lives on the agent so it drains even when no client window is open. + from agent.turn_queue import TurnQueue + agent.turn_queue = TurnQueue() + # Concurrent-tool worker thread tracking. `_execute_tool_calls_concurrent` # runs each tool on its own ThreadPoolExecutor worker — those worker # threads have tids distinct from `_execution_thread_id`, so diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index c6ed459e93d..04d0d7bc8f8 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -3216,6 +3216,10 @@ def apply_pending_steer_to_tool_results(agent, messages: list, num_tool_msgs: in messages[target_idx]["content"] = f"{existing_content}{marker}" else: messages[target_idx]["content"] = existing_content + marker + try: + agent._emit_steer_event("applied", steer_text) + except Exception: + pass _ra().logger.info( "Delivered /steer to agent after tool batch (%d chars): %s", len(steer_text), diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 420d12670e6..4e684df3adc 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -752,6 +752,11 @@ def run_conversation( _si, ) break + if _injected: + try: + agent._emit_steer_event("applied", _pre_api_steer) + except Exception: + pass if not _injected: # No tool message to inject into — put it back so # the post-tool-execution drain picks it up later. diff --git a/agent/turn_queue.py b/agent/turn_queue.py new file mode 100644 index 00000000000..8e2f8bf4ac7 --- /dev/null +++ b/agent/turn_queue.py @@ -0,0 +1,188 @@ +"""TurnQueue — a thread-safe FIFO of prompts to run as the next turn(s). + +This is the agent-side queue that unifies message queueing across CLI, TUI +gateway, desktop, and messaging platforms. It lives on the AIAgent instance, +alongside the existing ``_pending_steer`` mechanism. + +Unlike steer (which injects into the *current* turn's tool results), queued +prompts become the *next* user turn after the current one finishes. The +drain point is the agent-loop / gateway-runner tail — wherever the current +turn ends and ``running`` flips to False. + +Thread-safety: all mutations go through ``self._lock``. Safe to call from +gateway threads, CLI process_loop, and the agent execution thread. +""" + +from __future__ import annotations + +import threading +import time +import uuid +from dataclasses import dataclass, field +from typing import Any, Optional + + +@dataclass +class QueuedTurn: + """A single queued prompt entry.""" + + 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 + # as an optimistic user message). Lets drain events tell clients whether + # the text still needs painting. + source: str = "queue" + + def to_dict(self) -> dict: + """Serialise for RPC / event emission (no transport internals).""" + return { + "id": self.id, + "text": self.text, + "mode": self.mode, + "queued_at": self.queued_at, + "attachments": self.attachments, + "source": self.source, + } + + +class TurnQueue: + """Thread-safe FIFO of prompts waiting to become the next turn. + + Replaces the ad-hoc ``session["queued_prompt"]`` dict slot in the TUI + gateway and the ``localStorage``-backed queue in the desktop renderer. + The queue lives in the agent process so it drains even when no client + window is open. + """ + + def __init__(self) -> None: + self._entries: list[QueuedTurn] = [] + self._lock = threading.Lock() + + # ── enqueue ────────────────────────────────────────────── + + 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. + + Consecutive text entries are *not* merged here — the gateway's + ``_handle_busy_submit`` still does merge if it wants to (mirroring + ``repair_message_sequence``). Callers that want merge semantics + should check ``peek()`` first. + + Returns the created :class:`QueuedTurn`. + """ + 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 [], + ) + with self._lock: + self._entries.insert(0, entry) + return entry + + # ── drain ──────────────────────────────────────────────── + + def drain(self) -> Optional[QueuedTurn]: + """Pop and return the head entry, or None if the queue is empty.""" + with self._lock: + if not self._entries: + return None + return self._entries.pop(0) + + # ── peek / inspect ─────────────────────────────────────── + + def peek(self) -> list[QueuedTurn]: + """Return a snapshot copy of pending entries (for UI sync).""" + with self._lock: + return list(self._entries) + + def is_empty(self) -> bool: + with self._lock: + return len(self._entries) == 0 + + def __len__(self) -> int: + with self._lock: + return len(self._entries) + + def __bool__(self) -> bool: + return len(self) > 0 + + # ── mutate ─────────────────────────────────────────────── + + def remove(self, entry_id: str) -> bool: + """Remove a specific entry by id. Returns True if found.""" + with self._lock: + before = len(self._entries) + self._entries = [e for e in self._entries if e.id != entry_id] + return len(self._entries) < before + + def promote(self, entry_id: str) -> bool: + """Move an entry to the front of the queue. Returns True if found.""" + with self._lock: + idx = None + for i, e in enumerate(self._entries): + if e.id == entry_id: + idx = i + break + if idx is None or idx == 0: + return False + entry = self._entries.pop(idx) + self._entries.insert(0, entry) + return True + + def update_text(self, entry_id: str, text: str) -> bool: + """Update the text of a queued entry. Returns True if found.""" + with self._lock: + for e in self._entries: + if e.id == entry_id: + e.text = text + return True + return False + + def clear(self) -> int: + """Clear all entries. Returns the number removed.""" + with self._lock: + n = len(self._entries) + self._entries.clear() + return n + + # ── serialisation ──────────────────────────────────────── + + def to_list(self) -> list[dict]: + """Return a list of plain dicts for RPC / event emission.""" + with self._lock: + return [e.to_dict() for e in self._entries] 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 761d6830a1b..4a06ff50422 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 @@ -1,22 +1,17 @@ import { type RefObject, useCallback, useEffect, useRef, useState } from 'react' -import { useI18n } from '@/i18n' import { triggerHaptic } from '@/lib/haptics' import { useSessionSlice } from '@/lib/use-session-slice' import { type ComposerAttachment } from '@/store/composer' -import { resetBrowseState } from '@/store/composer-input-history' import { + $pendingSteersBySession, $queuedPromptsBySession, - enqueueQueuedPrompt, - MAX_AUTO_DRAIN_ATTEMPTS, + migrateLegacyQueue, migrateQueuedPrompts, promoteQueuedPrompt, type QueuedPromptEntry, - removeQueuedPrompt, - shouldAutoDrain, updateQueuedPrompt } from '@/store/composer-queue' -import { notify } from '@/store/notifications' import { cloneAttachments, type QueueEditState } from '../composer-utils' import { useComposerScope } from '../scope' @@ -30,21 +25,18 @@ interface UseComposerQueueArgs { draftRef: RefObject focusInput: () => void loadIntoComposer: (text: string, attachments: ComposerAttachment[]) => void - onCancel: ChatBarProps['onCancel'] - onSubmit: ChatBarProps['onSubmit'] + onQueue: ChatBarProps['onQueue'] queueEditRef: RefObject queueSessionKey: ChatBarProps['queueSessionKey'] sessionId: string | null | undefined } /** - * The composer's queue engine — everything about queued turns: the per-session - * queue store binding, in-place queued-prompt editing (begin/step/exit), the - * shared drain lock + send-then-remove sequence, manual send-now, and the - * edge-independent auto-drain with bounded retries. It consumes the draft API - * (draftRef/clearDraft/loadIntoComposer/focusInput) and writes the - * coordinator-owned `queueEditRef` so the draft engine can read the edit state - * without a back-reference. Behaviour-identical to the inline original. + * The composer's queue view — the gateway owns the queue (agent.turn_queue) + * and drains it at the end of every turn, so there is NO auto-drain here and + * no drain lock. This hook covers what's left client-side: the per-session + * mirror binding, in-place queued-prompt editing (begin/step/exit), enqueueing + * the current draft, and "send now" (promote + interrupt on the gateway). */ export function useComposerQueue({ activeQueueSessionKey, @@ -54,13 +46,11 @@ export function useComposerQueue({ draftRef, focusInput, loadIntoComposer, - onCancel, - onSubmit, + onQueue, queueEditRef, queueSessionKey, - sessionId + sessionId: _sessionId }: UseComposerQueueArgs) { - const { t } = useI18n() const scope = useComposerScope() // Per-session slice (edge): re-renders only when THIS session's queue changes, @@ -68,6 +58,9 @@ export function useComposerQueue({ // write; the keyed array does not). const queuedPrompts = useSessionSlice($queuedPromptsBySession, activeQueueSessionKey) + // Steers accepted by the gateway but not yet injected into the live turn. + const pendingSteers = useSessionSlice($pendingSteersBySession, activeQueueSessionKey) + const [queueEdit, setQueueEdit] = useState(null) queueEditRef.current = queueEdit @@ -82,8 +75,6 @@ export function useComposerQueue({ const editingQueuedPrompt = queueEdit ? (queuedPrompts.find(entry => entry.id === queueEdit.entryId) ?? null) : null const prevQueueKeyRef = useRef(activeQueueSessionKey) - const drainingQueueRef = useRef(false) - const drainFailuresRef = useRef(new Map()) const beginQueuedEdit = (entry: QueuedPromptEntry) => { if (!activeQueueSessionKey || queueEdit) { @@ -117,7 +108,6 @@ export function useComposerQueue({ } const saved = updateQueuedPrompt(queueEdit.sessionKey, queueEdit.entryId, { - attachments: cloneAttachments(attachments), text: draftRef.current }) @@ -144,13 +134,12 @@ export function useComposerQueue({ if (action === 'save') { const text = draftRef.current - const next = cloneAttachments(attachments) - if (!text.trim() && next.length === 0) { + if (!text.trim() && attachments.length === 0) { return false } - const saved = updateQueuedPrompt(queueEdit.sessionKey, queueEdit.entryId, { attachments: next, text }) + const saved = updateQueuedPrompt(queueEdit.sessionKey, queueEdit.entryId, { text }) triggerHaptic(saved ? 'success' : 'selection') } else { triggerHaptic('cancel') @@ -163,14 +152,21 @@ export function useComposerQueue({ return true } - const queueCurrentDraft = useCallback(() => { + // Queue the current draft on the gateway. onQueue (use-prompt-actions' + // queuePromptText) resolves attachments into refs and fires + // session.queue.add; the gateway drains it as the next turn, so nothing + // more happens client-side. Clears the draft only after the gateway + // accepts — a rejected enqueue keeps the words in the composer. + const queueCurrentDraft = useCallback(async () => { const text = draftRef.current - if (!activeQueueSessionKey || (!text.trim() && attachments.length === 0)) { + if (!activeQueueSessionKey || !onQueue || (!text.trim() && attachments.length === 0)) { return false } - if (!enqueueQueuedPrompt(activeQueueSessionKey, { text, attachments })) { + const accepted = await Promise.resolve(onQueue(text, { attachments: cloneAttachments(attachments) })) + + if (!accepted) { return false } @@ -179,124 +175,41 @@ export function useComposerQueue({ triggerHaptic('selection') return true - }, [activeQueueSessionKey, attachments, clearDraft, draftRef, scope.attachments]) - - // All queue drain paths share one lock + send-then-remove sequence. - // `pickEntry` lets each caller choose head, by-id, or skip-edited. - const runDrain = useCallback( - async (pickEntry: (entries: QueuedPromptEntry[]) => QueuedPromptEntry | undefined): Promise => { - if (drainingQueueRef.current || !activeQueueSessionKey) { - return false - } - - const entry = pickEntry(queuedPrompts) - - if (!entry) { - return false - } - - drainingQueueRef.current = true - - try { - const accepted = await Promise.resolve( - onSubmit(entry.text, { attachments: entry.attachments, fromQueue: true }) - ) - - if (accepted === false) { - return false - } - - drainFailuresRef.current.delete(entry.id) - removeQueuedPrompt(activeQueueSessionKey, entry.id) - resetBrowseState(sessionId) - - return true - } finally { - drainingQueueRef.current = false - } - }, - [activeQueueSessionKey, onSubmit, queuedPrompts, sessionId] - ) - - const pickDrainHead = useCallback( - (entries: QueuedPromptEntry[]) => { - const skip = queueEditRef.current?.entryId - - return skip ? entries.find(e => e.id !== skip) : entries[0] - }, - [queueEditRef] // reads the edit id off a ref so the lock-holder always sees the latest - ) - - const drainNextQueued = useCallback(() => runDrain(pickDrainHead), [pickDrainHead, runDrain]) + }, [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. const sendQueuedNow = useCallback( (id: string) => { if (!activeQueueSessionKey || id === queueEdit?.entryId) { return false } - if (busy) { - // Promote to the head, then interrupt. The gateway always emits a - // settle (message.complete + session.info running:false) when the - // turn unwinds, and the busy→false auto-drain below sends this entry. - promoteQueuedPrompt(activeQueueSessionKey, id) - triggerHaptic('selection') - void Promise.resolve(onCancel()) + triggerHaptic('selection') - return true - } - - // A manual send clears the auto-drain backoff so a stuck entry the user - // taps gets a fresh attempt (and re-enables auto-retry on success). - drainFailuresRef.current.delete(id) - - return runDrain(entries => entries.find(e => e.id === id)) + return promoteQueuedPrompt(activeQueueSessionKey, id, { interrupt: busy }) }, - [activeQueueSessionKey, busy, onCancel, queueEdit, runDrain] + [activeQueueSessionKey, busy, queueEdit] ) - // Edge-independent auto-drain: send the head whenever the session is idle and - // the queue is non-empty, bounding retries so a thrown/rejected onSubmit (e.g. - // a stale-session 404) can't strand the entry permanently nor spin-loop. The - // drain lock serializes sends; a remount/reconnect resets the failure counts. - const autoDrainNext = useCallback(() => { - if (busy || drainingQueueRef.current || !activeQueueSessionKey) { - return - } + // Manual "fire the next queued turn" gesture (Cmd/Ctrl+Shift+K, empty + // Enter). The gateway normally drains on its own the moment the session + // idles; this nudges a head entry that got stuck (e.g. its idle-drain + // attempt failed while the backend was restarting). + const drainNextQueued = useCallback(() => { + const head = queuedPrompts.find(e => e.id !== queueEditRef.current?.entryId) - const entry = pickDrainHead(queuedPrompts) - - if (!entry || (drainFailuresRef.current.get(entry.id) ?? 0) >= MAX_AUTO_DRAIN_ATTEMPTS) { - return - } - - const onFail = () => { - const fails = (drainFailuresRef.current.get(entry.id) ?? 0) + 1 - drainFailuresRef.current.set(entry.id, fails) - - if (fails >= MAX_AUTO_DRAIN_ATTEMPTS) { - notify({ - id: 'composer-queue-stuck', - kind: 'error', - title: t.composer.queueStuckTitle, - message: t.composer.queueStuckBody - }) - } - } - - void runDrain(() => entry) - .then(sent => { - if (!sent) { - onFail() - } - }) - .catch(onFail) - }, [activeQueueSessionKey, busy, pickDrainHead, queuedPrompts, runDrain, t]) + return head ? sendQueuedNow(head.id) : false + }, [queueEditRef, queuedPrompts, sendQueuedNow]) // Re-key on a runtime session-id change. A stable stored id (queueSessionKey) // never churns, so a change there is a real session switch and must NOT // migrate; only the runtime-derived key (queueSessionKey falsy → key is // sessionId) churns on a backend bounce/resume of the same conversation. + // Local-mirror-only: the gateway re-syncs authoritative state via + // session.info/queue.updated after the resume. useEffect(() => { const prev = prevQueueKeyRef.current prevQueueKeyRef.current = activeQueueSessionKey @@ -308,14 +221,12 @@ export function useComposerQueue({ migrateQueuedPrompts(prev, activeQueueSessionKey) }, [activeQueueSessionKey, queueSessionKey]) - // Queued turns flow whenever the session is idle — on the busy→false settle - // edge, on mount/reconnect, and after a re-key — so a swallowed edge can't - // strand them. To cancel queued turns, the user deletes them from the panel. + // One-time legacy migration: entries stranded in the localStorage queue + // (from before the queue moved to the gateway) are pushed to + // session.queue.add and the storage key is cleared. useEffect(() => { - if (shouldAutoDrain({ isBusy: busy, queueLength: queuedPrompts.length })) { - autoDrainNext() - } - }, [autoDrainNext, busy, queuedPrompts.length]) + migrateLegacyQueue(activeQueueSessionKey) + }, [activeQueueSessionKey]) // Queue-edit cleanup: on session swap the scope effect already stashed the // edit snapshot; only restore into the composer when still on the same scope. @@ -343,6 +254,7 @@ export function useComposerQueue({ drainNextQueued, editingQueuedPrompt, exitQueuedEdit, + pendingSteers, queueCurrentDraft, queueEdit, queuedPrompts, 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 adf44e34a8d..c891adabd8c 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 @@ -4,7 +4,7 @@ import { SLASH_COMMAND_RE } from '@/lib/chat-runtime' import { triggerHaptic } from '@/lib/haptics' import { clearSessionDraft, type ComposerAttachment } from '@/store/composer' import { resetBrowseState } from '@/store/composer-input-history' -import { enqueueQueuedPrompt, type QueuedPromptEntry } from '@/store/composer-queue' +import { type QueuedPromptEntry } from '@/store/composer-queue' import { cloneAttachments, type QueueEditState } from '../composer-utils' import { onComposerSubmitRequest } from '../focus' @@ -21,16 +21,17 @@ interface UseComposerSubmitArgs { clearDraft: () => void disabled: boolean draftRef: RefObject - drainNextQueued: () => Promise + drainNextQueued: () => boolean editorRef: RefObject exitQueuedEdit: (action: 'cancel' | 'save') => boolean focusInput: () => void inputDisabled: boolean loadIntoComposer: (text: string, attachments: ComposerAttachment[]) => void onCancel: ChatBarProps['onCancel'] + onQueue: ChatBarProps['onQueue'] onSteer: ChatBarProps['onSteer'] onSubmit: ChatBarProps['onSubmit'] - queueCurrentDraft: () => boolean + queueCurrentDraft: () => Promise queueEdit: QueueEditState | null queuedPrompts: QueuedPromptEntry[] sessionId: string | null | undefined @@ -63,6 +64,7 @@ export function useComposerSubmit({ inputDisabled, loadIntoComposer, onCancel, + onQueue, onSteer, onSubmit, queueCurrentDraft, @@ -152,7 +154,7 @@ export function useComposerSubmit({ clearDraft() dispatchSubmit(text) } else if (payloadPresent) { - queueCurrentDraft() + void queueCurrentDraft() } else { // Stop button (the only way to reach here while busy with an empty // composer — empty Enter is short-circuited in the keydown handler). @@ -175,7 +177,8 @@ export function useComposerSubmit({ // 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. + // are re-queued on the gateway so nothing is lost — same safety net as a + // plain queue. const steerDraft = () => { if (!onSteer || !canSteer) { return @@ -187,8 +190,8 @@ export function useComposerSubmit({ clearDraft() void Promise.resolve(onSteer(text)).then(accepted => { - if (!accepted && activeQueueSessionKey) { - enqueueQueuedPrompt(activeQueueSessionKey, { text, attachments: [] }) + if (!accepted && onQueue) { + void Promise.resolve(onQueue(text)) } }) } diff --git a/apps/desktop/src/app/chat/composer/index.tsx b/apps/desktop/src/app/chat/composer/index.tsx index 41e1c813309..23f2966fbb1 100644 --- a/apps/desktop/src/app/chat/composer/index.tsx +++ b/apps/desktop/src/app/chat/composer/index.tsx @@ -81,6 +81,7 @@ export function ChatBar({ onPickImages, onRemoveAttachment, onSteer, + onQueue, onSubmit: onSubmitProp, onTranscribeAudio }: ChatBarProps) { @@ -180,13 +181,14 @@ export function ChatBar({ onAddUrl }) - // The queue engine — queued turns, in-place editing, the shared drain lock, - // and bounded auto-drain. Consumes the draft API and writes `queueEditRef`. + // The queue view — queued turns + in-place editing. The gateway owns the + // queue and drains it (agent.turn_queue); there is no client auto-drain. const { beginQueuedEdit, drainNextQueued, editingQueuedPrompt, exitQueuedEdit, + pendingSteers, queueCurrentDraft, queueEdit, queuedPrompts, @@ -200,8 +202,7 @@ export function ChatBar({ draftRef, focusInput, loadIntoComposer, - onCancel, - onSubmit, + onQueue, queueEditRef, queueSessionKey, sessionId @@ -236,6 +237,7 @@ export function ChatBar({ inputDisabled, loadIntoComposer, onCancel, + onQueue, onSteer, onSubmit, queueCurrentDraft, @@ -883,7 +885,7 @@ export function ChatBar({ accounts for it. Collapses to nothing when every status is empty. */} 0 ? ( + activeQueueSessionKey && (queuedPrompts.length > 0 || pendingSteers.length > 0) ? ( void sendQueuedNow(id)} + pendingSteers={pendingSteers} /> ) : null } diff --git a/apps/desktop/src/app/chat/composer/queue-panel.tsx b/apps/desktop/src/app/chat/composer/queue-panel.tsx index 8f38fb89303..7235eae233b 100644 --- a/apps/desktop/src/app/chat/composer/queue-panel.tsx +++ b/apps/desktop/src/app/chat/composer/queue-panel.tsx @@ -6,12 +6,13 @@ import { Tip } from '@/components/ui/tooltip' import { type Translations, useI18n } from '@/i18n' import { ArrowUp, iconSize, Pencil, Trash2 } from '@/lib/icons' import { cn } from '@/lib/utils' -import type { QueuedPromptEntry } from '@/store/composer-queue' +import type { PendingSteerEntry, QueuedPromptEntry } from '@/store/composer-queue' interface QueuePanelProps { busy: boolean editingId: null | string entries: QueuedPromptEntry[] + pendingSteers?: PendingSteerEntry[] onDelete: (id: string) => void onEdit: (entry: QueuedPromptEntry) => void onSendNow: (id: string) => void @@ -20,19 +21,39 @@ interface QueuePanelProps { const entryPreview = (entry: QueuedPromptEntry, c: Translations['composer']) => entry.text.trim() || (entry.attachments.length > 0 ? c.attachmentOnly : c.emptyTurn) -export function QueuePanel({ busy, editingId, entries, onDelete, onEdit, onSendNow }: QueuePanelProps) { +export function QueuePanel({ + busy, + editingId, + entries, + onDelete, + onEdit, + onSendNow, + pendingSteers = [] +}: QueuePanelProps) { const { t } = useI18n() const c = t.composer - if (entries.length === 0) { + if (entries.length === 0 && pendingSteers.length === 0) { return null } return ( } - label={c.queued(entries.length)} + label={c.queued(entries.length + pendingSteers.length)} > + {/* Steers accepted by the gateway but not yet injected into the live + turn — they land in the transcript on the steer.applied event. */} + {pendingSteers.map(steer => ( + +
+

{steer.text}

+
+ {c.steerPending} +
+
+ + ))} {entries.map(entry => { const isEditing = editingId === entry.id const attachmentsCount = entry.attachments.length diff --git a/apps/desktop/src/app/chat/composer/types.ts b/apps/desktop/src/app/chat/composer/types.ts index 59c7c17274c..b5e7f53036c 100644 --- a/apps/desktop/src/app/chat/composer/types.ts +++ b/apps/desktop/src/app/chat/composer/types.ts @@ -52,10 +52,10 @@ export interface ChatBarProps { onPickImages?: () => void onRemoveAttachment?: (id: string) => void onSteer?: (text: string) => Promise | boolean - onSubmit: ( - value: string, - options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean } - ) => Promise | boolean + /** Queue text on the gateway's agent-side turn queue (session.queue.add). + * The gateway drains it as the next turn — even with the tab closed. */ + onQueue?: (text: string, options?: { attachments?: ComposerAttachment[] }) => Promise | boolean + onSubmit: (value: string, options?: { attachments?: ComposerAttachment[] }) => Promise | boolean onTranscribeAudio?: (audio: Blob) => Promise } diff --git a/apps/desktop/src/app/chat/index.tsx b/apps/desktop/src/app/chat/index.tsx index 4b417404bed..fe7a91159c6 100644 --- a/apps/desktop/src/app/chat/index.tsx +++ b/apps/desktop/src/app/chat/index.tsx @@ -74,10 +74,8 @@ interface ChatViewProps extends Omit, 'onSubmit'> { onPickImages: () => void onRemoveAttachment: (id: string) => void onSteer: (text: string) => Promise | boolean - onSubmit: ( - text: string, - options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean } - ) => Promise | boolean + onQueue?: (text: string, options?: { attachments?: ComposerAttachment[] }) => Promise | boolean + onSubmit: (text: string, options?: { attachments?: ComposerAttachment[] }) => Promise | boolean onThreadMessagesChange: (messages: readonly ThreadMessage[]) => void onEdit: (message: AppendMessage) => Promise onReload: (parentId: string | null) => Promise @@ -223,6 +221,7 @@ export function ChatView({ onPickImages, onRemoveAttachment, onSteer, + onQueue, onSubmit, onThreadMessagesChange, onEdit, @@ -515,6 +514,7 @@ export function ChatView({ onPickFiles={onPickFiles} onPickFolders={onPickFolders} onPickImages={onPickImages} + onQueue={onQueue} onRemoveAttachment={onRemoveAttachment} onSteer={onSteer} onSubmit={onSubmit} diff --git a/apps/desktop/src/app/chat/session-tile-actions.ts b/apps/desktop/src/app/chat/session-tile-actions.ts index f75ff6bf4da..c8bfff3cec2 100644 --- a/apps/desktop/src/app/chat/session-tile-actions.ts +++ b/apps/desktop/src/app/chat/session-tile-actions.ts @@ -15,11 +15,11 @@ import { useGatewayRequest } from '@/app/gateway/hooks/use-gateway-request' import type { ClientSessionState } from '@/app/types' import { PROMPT_SUBMIT_REQUEST_TIMEOUT_MS } from '@/hermes' import { useI18n } from '@/i18n' -import { textPart } from '@/lib/chat-messages' import { SLASH_COMMAND_RE } from '@/lib/chat-runtime' import { triggerHaptic } from '@/lib/haptics' import { clearClarifyRequest } from '@/store/clarify' import type { ComposerAttachment } from '@/store/composer' +import { addPendingSteer, enqueueQueuedPrompt } from '@/store/composer-queue' import { resetSessionBackground } from '@/store/composer-status' import { notifyError } from '@/store/notifications' import { clearPreviewArtifacts } from '@/store/preview-status' @@ -170,19 +170,6 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses [scope.attachments.$attachments, submitPromptText] ) - const appendSystemNote = useCallback( - (text: string) => { - update(state => ({ - ...state, - messages: [ - ...state.messages, - { id: `system-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, role: 'system', parts: [textPart(text)] } - ] - })) - }, - [update] - ) - const cancelRun = useCallback(async () => { const sessionId = runtimeIdRef.current @@ -226,7 +213,10 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses if (result?.status === 'queued') { triggerHaptic('submit') - appendSystemNote(`steer:${text}`) + // Pending until the gateway's steer.applied event lands — the + // transcript row is appended there (use-message-stream), when the + // model actually saw the nudge. + addPendingSteer(runtimeIdRef.current, text) return true } @@ -236,9 +226,21 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses return false }, - [appendSystemNote, requestGateway] + [requestGateway] ) + // Queue on the gateway's agent-side turn queue (the tile equivalent of the + // primary queuePromptText, minus attachment sync — tiles queue text-only). + const queuePromptText = useCallback(async (rawText: string): Promise => { + const text = rawText.trim() + + if (!text) { + return false + } + + return (await enqueueQueuedPrompt(runtimeIdRef.current, { text })) !== null + }, []) + // Rewind primitive (interrupt-first for live turns, busy-retry) — shared with // the primary chat so the two can't diverge. const submitRewind = useCallback( @@ -349,11 +351,22 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses dismissError, editMessage, handleThreadMessagesChange, + queuePromptText, reloadFromMessage, restoreToMessage, steerPrompt, submitText }), - [cancelRun, dismissError, editMessage, handleThreadMessagesChange, reloadFromMessage, restoreToMessage, steerPrompt, submitText] + [ + cancelRun, + dismissError, + editMessage, + handleThreadMessagesChange, + queuePromptText, + reloadFromMessage, + restoreToMessage, + steerPrompt, + submitText + ] ) } diff --git a/apps/desktop/src/app/chat/session-tile.tsx b/apps/desktop/src/app/chat/session-tile.tsx index 04ddaf1cc63..1dd253e6b5b 100644 --- a/apps/desktop/src/app/chat/session-tile.tsx +++ b/apps/desktop/src/app/chat/session-tile.tsx @@ -149,6 +149,7 @@ function TileChat({ onPickFiles={() => void composer.pickContextPaths('file')} onPickFolders={() => void composer.pickContextPaths('folder')} onPickImages={() => void composer.pickImages()} + onQueue={actions.queuePromptText} onReload={actions.reloadFromMessage} onRemoveAttachment={id => void composer.removeAttachment(id)} onRestoreToMessage={actions.restoreToMessage} diff --git a/apps/desktop/src/app/contrib/surfaces.tsx b/apps/desktop/src/app/contrib/surfaces.tsx index 1d3d475fc16..ccbbe226549 100644 --- a/apps/desktop/src/app/contrib/surfaces.tsx +++ b/apps/desktop/src/app/contrib/surfaces.tsx @@ -152,6 +152,7 @@ export const ChatRoutesSurface = memo(function ChatRoutesSurface({ onPickFiles={actions.onPickFiles} onPickFolders={actions.onPickFolders} onPickImages={actions.onPickImages} + onQueue={actions.onQueue} onReload={actions.onReload} onRemoveAttachment={actions.onRemoveAttachment} onRestoreToMessage={actions.onRestoreToMessage} diff --git a/apps/desktop/src/app/contrib/types.ts b/apps/desktop/src/app/contrib/types.ts index 1e2c60ac776..ded384971c8 100644 --- a/apps/desktop/src/app/contrib/types.ts +++ b/apps/desktop/src/app/contrib/types.ts @@ -46,6 +46,7 @@ export type ChatActions = Pick< | 'onRestoreToMessage' | 'onRetryResume' | 'onSteer' + | 'onQueue' | 'onSubmit' | 'onThreadMessagesChange' | 'onToggleSelectedPin' diff --git a/apps/desktop/src/app/contrib/wiring.tsx b/apps/desktop/src/app/contrib/wiring.tsx index ca7a043a916..90b15b31988 100644 --- a/apps/desktop/src/app/contrib/wiring.tsx +++ b/apps/desktop/src/app/contrib/wiring.tsx @@ -492,6 +492,7 @@ export function ContribWiring({ children }: { children: ReactNode }) { editMessage, executeSlashCommand, handleThreadMessagesChange, + queuePromptText, reloadFromMessage, restoreToMessage, steerPrompt, @@ -726,6 +727,7 @@ export function ContribWiring({ children }: { children: ReactNode }) { }, onRetryResume: sessionId => void resumeSession(sessionId, true), onSteer: steerPrompt, + onQueue: queuePromptText, onSubmit: submitText, onThreadMessagesChange: handleThreadMessagesChange, onToggleSelectedPin: toggleSelectedPin, diff --git a/apps/desktop/src/app/gateway/hooks/use-gateway-request.ts b/apps/desktop/src/app/gateway/hooks/use-gateway-request.ts index d6c9ab0e029..bf422599834 100644 --- a/apps/desktop/src/app/gateway/hooks/use-gateway-request.ts +++ b/apps/desktop/src/app/gateway/hooks/use-gateway-request.ts @@ -3,6 +3,7 @@ import { useStore } from '@nanostores/react' import { useCallback, useEffect, useRef } from 'react' import type { HermesGateway } from '@/hermes' +import { setGatewayRequester } from '@/store/composer-queue' import { $gateway, ensureActiveGatewayOpen, isActivePrimary } from '@/store/gateway' import { $activeGatewayProfile } from '@/store/profile' import { $gatewayState, setConnection } from '@/store/session' @@ -134,5 +135,13 @@ export function useGatewayRequest() { [ensureGatewayOpen] ) + // Give the gateway-backed composer-queue store a way to fire RPCs without + // being a React component (module-level seam, set once per mount). + useEffect(() => { + setGatewayRequester(requestGateway) + + return () => setGatewayRequester(null) + }, [requestGateway]) + return { connectionRef, gatewayRef, requestGateway } } 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 58cabd78e9a..29a9e3ca110 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,6 +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 { refreshBackgroundProcesses } from '@/store/composer-status' import { $gateway } from '@/store/gateway' import { dispatchNativeNotification } from '@/store/native-notifications' @@ -136,11 +137,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { // model output or tool event proves summarization has finished and the // turn has resumed, so retire the phase label without waiting for the // whole turn to complete. - if ( - sessionId && - COMPACTION_RESUME_EVENT_TYPES.has(event.type) && - compactedTurnRef.current.has(sessionId) - ) { + if (sessionId && COMPACTION_RESUME_EVENT_TYPES.has(event.type) && compactedTurnRef.current.has(sessionId)) { setSessionCompacting(sessionId, false) } @@ -275,6 +272,10 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { reportInstallMethodWarning(payload?.install_warning) } + if (sessionId && Array.isArray(payload?.queue)) { + setSessionQueue(sessionId, payload.queue) + } + void refreshHermesConfig() if (modelChanged || providerChanged) { @@ -534,7 +535,9 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { setApprovalRequest({ // false only when a tirith warning forbids it; backend omits the field otherwise. allowPermanent: payload?.allow_permanent !== false, - choices: Array.isArray(payload?.choices) ? payload.choices.filter(choice => typeof choice === 'string') : undefined, + choices: Array.isArray(payload?.choices) + ? payload.choices.filter(choice => typeof choice === 'string') + : undefined, command, description, sessionId: sessionId ?? null, @@ -658,6 +661,62 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { ] })) } + } else if (event.type === 'queue.updated') { + // Agent-side turn queue changed (enqueue/drain/remove/clear/promote). + // The gateway owns the queue; this mirror only feeds the panel UI. + if (sessionId && Array.isArray(payload?.entries)) { + setSessionQueue(sessionId, payload.entries) + } + } else if (event.type === 'queue.drained') { + // A queued entry just became the next user turn. Entries queued from + // the panel (source "queue") were never in the transcript — paint them + // now. Busy-submit entries ("busy_submit") were already echoed + // optimistically at submit time; painting again would duplicate them. + if (sessionId && payload?.source !== 'busy_submit') { + const text = coerceGatewayText(payload?.text).trim() + + if (text) { + updateSessionState(sessionId, state => ({ + ...state, + messages: [ + ...state.messages, + { + id: `queued-user-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, + role: 'user', + parts: [textPart(text)] + } + ] + })) + } + } + } else if (event.type === 'steer.applied') { + // The steer text actually reached the model (injected into a tool + // result). NOW it belongs in the transcript — painting it at RPC-accept + // time would show a nudge the model hadn't seen yet (and might never + // see, e.g. when an interrupt drops it). + if (sessionId) { + const text = coerceGatewayText(payload?.text).trim() + + if (text) { + settlePendingSteer(sessionId, text) + updateSessionState(sessionId, state => ({ + ...state, + messages: [ + ...state.messages, + { + id: `steer-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, + role: 'system', + parts: [textPart(`steer:${text}`)] + } + ] + })) + } + } + } else if (event.type === 'steer.dropped') { + // An interrupt discarded the pending steer before the model saw it. + if (sessionId) { + settlePendingSteer(sessionId, coerceGatewayText(payload?.text)) + } } else if (event.type === 'error') { const errorMessage = payload?.message || 'Hermes reported an error' const looksLikeProviderSetup = isProviderSetupErrorMessage(errorMessage) 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 e35589010bd..24f073c65d2 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 @@ -59,7 +59,7 @@ interface HarnessHandle { cancelRun: () => Promise restoreToMessage: (messageId: string, target?: { text?: string; userOrdinal?: number | null }) => Promise steerPrompt: (text: string) => Promise - submitText: (text: string, options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }) => Promise + submitText: (text: string, options?: { attachments?: ComposerAttachment[] }) => Promise } function Harness({ @@ -505,40 +505,11 @@ describe('usePromptActions submit / queue drain semantics', () => { ) }) - it('a fromQueue drain sends even when busyRef is still true on the settle edge', async () => { - // busyRef lags $busy by one effect tick on the busy→false settle edge, so a - // drained queue send would otherwise hit the busy guard and silently no-op. - const busyRef = { current: true } - const requestGateway = vi.fn(async () => ({}) as never) - - let handle: HarnessHandle | null = null - await actRender( - (handle = h)} - refreshSessions={async () => undefined} - requestGateway={requestGateway} - /> - ) - - const accepted = await handle!.submitText('queued message', { fromQueue: true }) - - expect(accepted).toBe(true) - expect(requestGateway).toHaveBeenCalledWith( - 'prompt.submit', - { - session_id: RUNTIME_SESSION_ID, - text: 'queued message' - }, - 1_800_000 - ) - }) - - it('a rejected fromQueue drain returns false (entry stays queued) and a later retry sends it', async () => { - // A stale-session 404 must not strand the queued entry: submitPrompt returns - // false on failure so the composer keeps it, and the edge-independent - // auto-drain re-attempts once the session is idle again. storedSessionId is - // null so the session.resume recovery path is skipped and the error surfaces. + it('a failed submit returns false and a later retry sends it', async () => { + // A stale-session 404 must not strand the user's words: submitPrompt + // returns false on failure so the composer keeps the draft for a retry. + // storedSessionId is null so the session.resume recovery path is skipped + // and the error surfaces. let attempt = 0 const requestGateway = vi.fn(async (method: string) => { @@ -563,10 +534,10 @@ describe('usePromptActions submit / queue drain semantics', () => { /> ) - const first = await handle!.submitText('please send me', { fromQueue: true }) + const first = await handle!.submitText('please send me') expect(first).toBe(false) - const second = await handle!.submitText('please send me', { fromQueue: true }) + const second = await handle!.submitText('please send me') expect(second).toBe(true) expect(requestGateway).toHaveBeenCalledWith( 'prompt.submit', 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 2e163fcc80d..031be43e0ff 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 @@ -18,6 +18,7 @@ import { setComposerAttachmentUploadState, updateComposerAttachment } from '@/store/composer' +import { addPendingSteer, enqueueQueuedPrompt } from '@/store/composer-queue' import { resetSessionBackground } from '@/store/composer-status' import { clearNotifications, notify, notifyError } from '@/store/notifications' import { clearPreviewArtifacts } from '@/store/preview-status' @@ -594,8 +595,13 @@ export function usePromptActions({ // 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. + // (desktop parity with `/steer`). The RPC accepting the steer does NOT mean + // the model saw it yet — that happens at the next tool-batch boundary, and + // the gateway announces it with a `steer.applied` event. We track the steer + // as *pending* here; the transcript row is appended by the steer.applied + // handler in use-message-stream, so the chat log reflects when the nudge + // actually landed. Returns false on reject so the caller can fall back to + // queueing the words for the next turn. const steerPrompt = useCallback( async (rawText: string): Promise => { const text = sanitizeComposerInput(rawText).trim() @@ -610,10 +616,7 @@ export function usePromptActions({ if (result?.status === 'queued') { 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}`) + addPendingSteer(sessionId, text) return true } @@ -623,7 +626,7 @@ export function usePromptActions({ return false }, - [activeSessionId, activeSessionIdRef, appendSessionTextMessage, requestGateway] + [activeSessionId, activeSessionIdRef, requestGateway] ) const reloadFromMessage = useCallback( @@ -789,6 +792,55 @@ export function usePromptActions({ [activeSessionIdRef, updateSessionState] ) + // Queue a prompt on the gateway's agent-side turn queue. Attachments are + // synced (uploaded / rewritten to @file: refs) NOW, at enqueue time — the + // drain happens later in the gateway process where no attachment bytes + // exist, so the queued text must already be self-contained. + const queuePromptText = useCallback( + async (rawText: string, options?: { attachments?: ComposerAttachment[] }): Promise => { + const visibleText = sanitizeComposerInput(rawText).trim() + const sessionId = activeSessionId || activeSessionIdRef.current + const attachments = (options?.attachments ?? []).filter((a): a is ComposerAttachment => Boolean(a)) + + if ((!visibleText && attachments.length === 0) || !sessionId) { + return false + } + + let text = visibleText + + if (attachments.length > 0) { + try { + const synced = await syncAttachmentsForSubmit(sessionId, attachments, { + updateComposerAttachments: false + }) + + const contextRefs = synced + .filter((a): a is ComposerAttachment => Boolean(a)) + .map(a => a.refText) + .filter(Boolean) + .join('\n') + + text = + [contextRefs, visibleText].filter(Boolean).join('\n\n') || + (synced.some(a => a?.kind === 'image') ? 'What do you see in this image?' : '') + } catch (err) { + notifyError(err, copy.promptFailed) + + return false + } + } + + if (!text) { + return false + } + + const entry = await enqueueQueuedPrompt(sessionId, { text }) + + return entry !== null + }, + [activeSessionId, activeSessionIdRef, copy.promptFailed, syncAttachmentsForSubmit] + ) + return { cancelRun, editMessage, @@ -797,6 +849,7 @@ export function usePromptActions({ executeSlashCommand, handleThreadMessagesChange, handoffSession, + queuePromptText, reloadFromMessage, restoreToMessage, steerPrompt, diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts index def08fe6eb4..af8263f8663 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts @@ -59,10 +59,7 @@ interface SlashCommandDeps { requestGateway: GatewayRequest resumeStoredSession: (storedSessionId: string) => Promise | void startFreshSessionDraft: () => void - submitPromptText: ( - rawText: string, - options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean } - ) => Promise + submitPromptText: (rawText: string, options?: { attachments?: ComposerAttachment[] }) => Promise } /** The /slash command dispatcher, extracted from usePromptActions. */ 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 162d9e41011..166fa7c4dda 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 @@ -24,7 +24,6 @@ import { inlineErrorMessage, isGatewayTimeoutError, isProviderSetupError, - isSessionBusyError, isSessionNotFoundError, type SubmitTextOptions, withSessionBusyRetry @@ -127,13 +126,9 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { ) } - // Queue drains fire on the busy→false settle edge, where busyRef (synced - // from $busy by a separate effect) may still read true — honoring it would - // bounce the drained send. The drain lock serializes them; the user path - // keeps the guard so a stray Enter mid-turn can't double-submit. const hasSendable = Boolean(visibleText || terminalContextBlocks || attachments.length || hasImage) - if (!hasSendable || (!options?.fromQueue && busyRef.current)) { + if (!hasSendable || busyRef.current) { return false } @@ -147,8 +142,7 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { let startingRouteToken = getRouteToken() const sessionContextDrifted = (): boolean => - selectedStoredSessionIdRef.current !== startingStoredSessionId || - getRouteToken() !== startingRouteToken + selectedStoredSessionIdRef.current !== startingStoredSessionId || getRouteToken() !== startingRouteToken // One submit in flight per session — drop any concurrent re-fire so a // stalled turn can't stack the same prompt into multiple real turns. @@ -366,10 +360,7 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { requestGateway('prompt.submit', { session_id: sessionId, text }, PROMPT_SUBMIT_REQUEST_TIMEOUT_MS) ) } catch (firstErr) { - if ( - (isSessionNotFoundError(firstErr) || isGatewayTimeoutError(firstErr)) && - startingStoredSessionId - ) { + if ((isSessionNotFoundError(firstErr) || isGatewayTimeoutError(firstErr)) && startingStoredSessionId) { // Re-register the session in the gateway and get a fresh live ID. // Timeouts recover the same way as "session not found": a starved // backend loop (#55578 symptom d) rejects the submit even though @@ -415,13 +406,6 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { } catch (err) { releaseBusy() - // A queued drain that raced a not-yet-settled turn gets a transient - // "session busy" (4009). Don't surface an error bubble/toast — the entry - // stays queued and the composer's bounded auto-drain retries when idle. - if (options?.fromQueue && isSessionBusyError(err)) { - return false - } - const message = inlineErrorMessage(err, copy.promptFailed) updateSessionState(sessionId, state => ({ diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts index de501a52bbc..30ed63d2d03 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts @@ -225,5 +225,4 @@ export function visibleUserIndexAtOrdinal(messages: readonly ChatMessage[], targ export interface SubmitTextOptions { attachments?: ComposerAttachment[] - fromQueue?: boolean } diff --git a/apps/desktop/src/i18n/en.ts b/apps/desktop/src/i18n/en.ts index 00a799e3102..a2d19de1158 100644 --- a/apps/desktop/src/i18n/en.ts +++ b/apps/desktop/src/i18n/en.ts @@ -1755,8 +1755,7 @@ export const en: Translations = { queueSendNext: 'Next', queueSend: 'Send', queueDelete: 'Delete', - queueStuckTitle: 'Queued message not sent', - queueStuckBody: 'A queued turn kept failing to send. It is still in the queue — try sending it again.', + steerPending: 'Steering — lands at the next tool step', previewUnavailable: 'Preview unavailable', previewLabel: label => `Preview ${label}`, couldNotPreview: label => `Could not preview ${label}`, diff --git a/apps/desktop/src/i18n/ja.ts b/apps/desktop/src/i18n/ja.ts index 1cb2f86e165..e2bea2e777a 100644 --- a/apps/desktop/src/i18n/ja.ts +++ b/apps/desktop/src/i18n/ja.ts @@ -1672,9 +1672,6 @@ export const ja = defineLocale({ queueSendNext: '次に送信', queueSend: '送信', queueDelete: '削除', - queueStuckTitle: 'キュー内のメッセージを送信できません', - queueStuckBody: - 'キューに入れたターンの送信が繰り返し失敗しました。まだキューに残っています。もう一度送信してください。', previewUnavailable: 'プレビューは利用できません', previewLabel: label => `${label} のプレビュー`, couldNotPreview: label => `${label} をプレビューできませんでした`, diff --git a/apps/desktop/src/i18n/types.ts b/apps/desktop/src/i18n/types.ts index 35a30da7579..824a1119a79 100644 --- a/apps/desktop/src/i18n/types.ts +++ b/apps/desktop/src/i18n/types.ts @@ -1446,8 +1446,7 @@ export interface Translations { queueSendNext: string queueSend: string queueDelete: string - queueStuckTitle: string - queueStuckBody: string + steerPending: string previewUnavailable: string previewLabel: (label: string) => string couldNotPreview: (label: string) => string diff --git a/apps/desktop/src/i18n/zh-hant.ts b/apps/desktop/src/i18n/zh-hant.ts index 31a72122e84..ce81b0456e3 100644 --- a/apps/desktop/src/i18n/zh-hant.ts +++ b/apps/desktop/src/i18n/zh-hant.ts @@ -1622,8 +1622,6 @@ export const zhHant = defineLocale({ queueSendNext: '下一個', queueSend: '傳送', queueDelete: '刪除', - queueStuckTitle: '佇列訊息未送出', - queueStuckBody: '佇列中的對話多次傳送失敗。它仍在佇列中，請重試傳送。', previewUnavailable: '預覽不可用', previewLabel: label => `預覽 ${label}`, couldNotPreview: label => `無法預覽 ${label}`, diff --git a/apps/desktop/src/i18n/zh.ts b/apps/desktop/src/i18n/zh.ts index f97cd27a503..9afc67021b3 100644 --- a/apps/desktop/src/i18n/zh.ts +++ b/apps/desktop/src/i18n/zh.ts @@ -1931,8 +1931,7 @@ export const zh: Translations = { queueSendNext: '下一个', queueSend: '发送', queueDelete: '删除', - queueStuckTitle: '排队消息未发送', - queueStuckBody: '排队的对话多次发送失败。它仍在队列中，请重试发送。', + steerPending: '正在引导 — 将在下一个工具步骤生效', previewUnavailable: '预览不可用', previewLabel: label => `预览 ${label}`, couldNotPreview: label => `无法预览 ${label}`, diff --git a/apps/desktop/src/lib/chat-messages.ts b/apps/desktop/src/lib/chat-messages.ts index c622b90d85e..4a8d2c10e09 100644 --- a/apps/desktop/src/lib/chat-messages.ts +++ b/apps/desktop/src/lib/chat-messages.ts @@ -83,6 +83,11 @@ export type GatewayEventPayload = { label?: string index?: number aggregator?: string + // queue.updated (agent-side turn queue mirror) + session.info.queue + entries?: unknown[] + queue?: unknown[] + // queue.drained — where the drained entry came from ("queue" | "busy_submit") + source?: string } export function textPart(text: string): ChatMessagePart { diff --git a/apps/desktop/src/store/composer-queue.test.ts b/apps/desktop/src/store/composer-queue.test.ts index 8012e2870f0..3aa73e07652 100644 --- a/apps/desktop/src/store/composer-queue.test.ts +++ b/apps/desktop/src/store/composer-queue.test.ts @@ -1,170 +1,167 @@ -import { beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import type { ComposerAttachment } from './composer' import { + $pendingSteersBySession, $queuedPromptsBySession, + addPendingSteer, clearQueuedPrompts, - dequeueQueuedPrompt, enqueueQueuedPrompt, + getPendingSteers, getQueuedPrompts, migrateQueuedPrompts, promoteQueuedPrompt, removeQueuedPrompt, - shouldAutoDrain, - updateQueuedPrompt, + setGatewayRequester, + setSessionQueue, + settlePendingSteer, updateQueuedPromptText } from './composer-queue' -const SESSION_KEY = 'session-abc' -const QUEUE_STORAGE_KEY = 'hermes.desktop.composerQueue.v1' +const SESSION_KEY = 'session-1' -function attachment(id: string, kind: ComposerAttachment['kind'] = 'file'): ComposerAttachment { - return { - id, - kind, - label: id, - refText: `@file:${id}` - } -} +describe('composer-queue (gateway-backed)', () => { + const requester = vi.fn() -describe('composer queue store', () => { beforeEach(() => { - window.localStorage.removeItem(QUEUE_STORAGE_KEY) $queuedPromptsBySession.set({}) + $pendingSteersBySession.set({}) + requester.mockReset() + requester.mockResolvedValue({ entry_id: 'gw-1', status: 'queued' }) + setGatewayRequester(requester as never) }) - it('queues prompts in FIFO order', () => { - enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'first' }) - enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'second' }) - - expect(dequeueQueuedPrompt(SESSION_KEY)?.text).toBe('first') - expect(dequeueQueuedPrompt(SESSION_KEY)?.text).toBe('second') - expect(dequeueQueuedPrompt(SESSION_KEY)).toBeNull() + afterEach(() => { + setGatewayRequester(null) }) - it('clones attachments when queueing', () => { - const source = [attachment('a-1')] - const queued = enqueueQueuedPrompt(SESSION_KEY, { attachments: source, text: 'check clones' }) + it('enqueue fires session.queue.add and mirrors the entry locally', async () => { + const entry = await enqueueQueuedPrompt(SESSION_KEY, { text: 'queued draft' }) - expect(queued).not.toBeNull() - expect(getQueuedPrompts(SESSION_KEY)[0]?.attachments[0]).toEqual(source[0]) - expect(getQueuedPrompts(SESSION_KEY)[0]?.attachments[0]).not.toBe(source[0]) + expect(requester).toHaveBeenCalledWith('session.queue.add', { session_id: SESSION_KEY, text: 'queued draft' }) + expect(entry?.id).toBe('gw-1') + expect(getQueuedPrompts(SESSION_KEY).map(e => e.text)).toEqual(['queued draft']) }) - it('updates and removes queued entries by id', () => { - const first = enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'draft one' }) - const second = enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'draft two' }) + it('enqueue returns null (and keeps the mirror empty) when the gateway rejects', async () => { + requester.mockRejectedValue(new Error('gateway down')) - expect(first).not.toBeNull() - expect(second).not.toBeNull() + const entry = await enqueueQueuedPrompt(SESSION_KEY, { text: 'lost?' }) - expect(updateQueuedPromptText(SESSION_KEY, first!.id, 'draft one edited')).toBe(true) - expect(getQueuedPrompts(SESSION_KEY).map(entry => entry.text)).toEqual(['draft one edited', 'draft two']) - - expect(removeQueuedPrompt(SESSION_KEY, first!.id)).toBe(true) - expect(getQueuedPrompts(SESSION_KEY).map(entry => entry.text)).toEqual(['draft two']) + expect(entry).toBeNull() + expect(getQueuedPrompts(SESSION_KEY)).toEqual([]) }) - it('promotes a queued entry to the front', () => { - const first = enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'first' }) - const second = enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'second' }) - const third = enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'third' }) + it('enqueue returns null with no requester wired', async () => { + setGatewayRequester(null) - expect(first).not.toBeNull() - expect(second).not.toBeNull() - expect(third).not.toBeNull() - - expect(promoteQueuedPrompt(SESSION_KEY, third!.id)).toBe(true) - expect(getQueuedPrompts(SESSION_KEY).map(entry => entry.text)).toEqual(['third', 'first', 'second']) - expect(promoteQueuedPrompt(SESSION_KEY, third!.id)).toBe(false) + expect(await enqueueQueuedPrompt(SESSION_KEY, { text: 'no gateway' })).toBeNull() }) - it('updates queued text and attachment snapshot', () => { - const first = enqueueQueuedPrompt(SESSION_KEY, { attachments: [attachment('f-1')], text: 'draft one' }) - const editedAttachments = [attachment('f-2'), attachment('f-3', 'image')] + it('setSessionQueue replaces the mirror with the gateway entry list', () => { + setSessionQueue(SESSION_KEY, [ + { id: 'a', text: 'first', queued_at: 1700000000.5 }, + { id: 'b', text: 'second' } + ]) - expect(first).not.toBeNull() - expect( - updateQueuedPrompt(SESSION_KEY, first!.id, { - attachments: editedAttachments, - text: 'edited text' - }) - ).toBe(true) + const entries = getQueuedPrompts(SESSION_KEY) + expect(entries.map(e => e.text)).toEqual(['first', 'second']) + expect(entries[0]?.queuedAt).toBe(1700000000500) - const queue = getQueuedPrompts(SESSION_KEY) - expect(queue[0]?.text).toBe('edited text') - expect(queue[0]?.attachments).toEqual(editedAttachments) - expect(queue[0]?.attachments[0]).not.toBe(editedAttachments[0]) + setSessionQueue(SESSION_KEY, []) + expect(getQueuedPrompts(SESSION_KEY)).toEqual([]) }) - it('clears queue state for a session', () => { - enqueueQueuedPrompt(SESSION_KEY, { attachments: [attachment('img-1', 'image')], text: 'queued' }) + it('removeQueuedPrompt updates the mirror optimistically and fires the RPC', () => { + setSessionQueue(SESSION_KEY, [ + { id: 'a', text: 'first' }, + { id: 'b', text: 'second' } + ]) + + expect(removeQueuedPrompt(SESSION_KEY, 'a')).toBe(true) + expect(getQueuedPrompts(SESSION_KEY).map(e => e.id)).toEqual(['b']) + expect(requester).toHaveBeenCalledWith('session.queue.remove', { session_id: SESSION_KEY, entry_id: 'a' }) + + expect(removeQueuedPrompt(SESSION_KEY, 'missing')).toBe(false) + }) + + it('promoteQueuedPrompt moves the entry to the head and forwards interrupt', () => { + setSessionQueue(SESSION_KEY, [ + { id: 'a', text: 'first' }, + { id: 'b', text: 'second' }, + { id: 'c', text: 'third' } + ]) + + expect(promoteQueuedPrompt(SESSION_KEY, 'c', { interrupt: true })).toBe(true) + expect(getQueuedPrompts(SESSION_KEY).map(e => e.id)).toEqual(['c', 'a', 'b']) + expect(requester).toHaveBeenCalledWith('session.queue.promote', { + session_id: SESSION_KEY, + entry_id: 'c', + interrupt: true + }) + + // Head entry: still accepted (fires the RPC so an idle gateway can drain). + expect(promoteQueuedPrompt(SESSION_KEY, 'c')).toBe(true) + expect(promoteQueuedPrompt(SESSION_KEY, 'missing')).toBe(false) + }) + + it('updateQueuedPromptText edits in place and fires the RPC', () => { + setSessionQueue(SESSION_KEY, [{ id: 'a', text: 'before' }]) + + expect(updateQueuedPromptText(SESSION_KEY, 'a', 'after')).toBe(true) + expect(getQueuedPrompts(SESSION_KEY)[0]?.text).toBe('after') + expect(requester).toHaveBeenCalledWith('session.queue.update', { + session_id: SESSION_KEY, + entry_id: 'a', + text: 'after' + }) + + expect(updateQueuedPromptText(SESSION_KEY, 'a', 'after')).toBe(false) + }) + + it('clearQueuedPrompts wipes the mirror and fires the RPC', () => { + setSessionQueue(SESSION_KEY, [{ id: 'a', text: 'first' }]) + addPendingSteer(SESSION_KEY, 'nudge') clearQueuedPrompts(SESSION_KEY) expect(getQueuedPrompts(SESSION_KEY)).toEqual([]) - expect($queuedPromptsBySession.get()[SESSION_KEY]).toBeUndefined() - expect(window.localStorage.getItem(QUEUE_STORAGE_KEY)).toBeNull() + expect(getPendingSteers(SESSION_KEY)).toEqual([]) + expect(requester).toHaveBeenCalledWith('session.queue.clear', { session_id: SESSION_KEY }) }) - it('persists queue entries into local storage', () => { - enqueueQueuedPrompt(SESSION_KEY, { attachments: [], text: 'persist me' }) - - const raw = window.localStorage.getItem(QUEUE_STORAGE_KEY) - expect(raw).toBeTruthy() - - const parsed = JSON.parse(String(raw)) as Record - expect(parsed[SESSION_KEY]?.[0]?.text).toBe('persist me') - }) -}) - -describe('migrateQueuedPrompts', () => { - beforeEach(() => { - window.localStorage.removeItem(QUEUE_STORAGE_KEY) - $queuedPromptsBySession.set({}) - }) - - it('moves entries from a dead runtime key onto the live one', () => { - enqueueQueuedPrompt('rt-old', { attachments: [], text: 'stranded' }) + it('migrateQueuedPrompts re-keys the local mirror on a runtime id change', () => { + setSessionQueue('rt-old', [{ id: 'a', text: 'stranded' }]) + setSessionQueue('rt-new', [{ id: 'b', text: 'already here' }]) expect(migrateQueuedPrompts('rt-old', 'rt-new')).toBe(true) expect(getQueuedPrompts('rt-old')).toEqual([]) - expect(getQueuedPrompts('rt-new').map(e => e.text)).toEqual(['stranded']) - // The dead key is dropped from the store entirely. - expect($queuedPromptsBySession.get()['rt-old']).toBeUndefined() + expect(getQueuedPrompts('rt-new').map(e => e.text)).toEqual(['already here', 'stranded']) + + expect(migrateQueuedPrompts('rt-new', 'rt-new')).toBe(false) + expect(migrateQueuedPrompts('rt-empty', 'rt-new')).toBe(false) }) - it('appends after existing target entries (FIFO preserved)', () => { - enqueueQueuedPrompt('rt-new', { attachments: [], text: 'already here' }) - enqueueQueuedPrompt('rt-old', { attachments: [], text: 'migrated' }) + describe('pending steers', () => { + it('tracks a steer until steer.applied settles it', () => { + addPendingSteer(SESSION_KEY, 'go left') + addPendingSteer(SESSION_KEY, 'then right') - migrateQueuedPrompts('rt-old', 'rt-new') + expect(getPendingSteers(SESSION_KEY).map(s => s.text)).toEqual(['go left', 'then right']) - expect(getQueuedPrompts('rt-new').map(e => e.text)).toEqual(['already here', 'migrated']) - }) + // The agent concatenates queued steers with newlines before injecting — + // one applied event can cover both. + settlePendingSteer(SESSION_KEY, 'go left\nthen right') - it('is a no-op when source is empty or keys match', () => { - expect(migrateQueuedPrompts('rt-old', 'rt-new')).toBe(false) - expect(migrateQueuedPrompts('rt-x', 'rt-x')).toBe(false) - }) -}) - -describe('shouldAutoDrain', () => { - it('drains whenever idle with a non-empty queue', () => { - expect(shouldAutoDrain({ isBusy: false, queueLength: 1 })).toBe(true) - }) - - it('drains on mount/reconnect with no observed busy edge', () => { - // The whole point of dropping the edge: a remount resets the busy ref, so an - // edge-gated drain would strand the entry. Idle + non-empty must still fire. - expect(shouldAutoDrain({ isBusy: false, queueLength: 2 })).toBe(true) - }) - - it('does not drain mid-turn', () => { - expect(shouldAutoDrain({ isBusy: true, queueLength: 1 })).toBe(false) - }) - - it('does not drain an empty queue', () => { - expect(shouldAutoDrain({ isBusy: false, queueLength: 0 })).toBe(false) + expect(getPendingSteers(SESSION_KEY)).toEqual([]) + }) + + it('settles only matching entries', () => { + addPendingSteer(SESSION_KEY, 'go left') + addPendingSteer(SESSION_KEY, 'stay') + + settlePendingSteer(SESSION_KEY, 'go left') + + expect(getPendingSteers(SESSION_KEY).map(s => s.text)).toEqual(['stay']) + }) }) }) diff --git a/apps/desktop/src/store/composer-queue.ts b/apps/desktop/src/store/composer-queue.ts index 922e990fdce..4a94b94095b 100644 --- a/apps/desktop/src/store/composer-queue.ts +++ b/apps/desktop/src/store/composer-queue.ts @@ -1,7 +1,24 @@ +/** + * Gateway-backed turn queue store for the desktop renderer. + * + * The queue lives on the agent (`agent.turn_queue`) in the gateway process — + * this store is a mirror, populated from `queue.updated` events and the + * `queue` field of `session.info`. The renderer never owns the queue: the + * gateway drains it at the end of every turn (and immediately on enqueue when + * the session is idle), so queued messages fire even when the session tab is + * closed or the window is hidden. There is NO client-side auto-drain. + * + * Mutations are optimistic: the local mirror updates immediately for snappy + * UI, then the RPC fires and the authoritative `queue.updated` event settles + * the final state. + */ + import { atom } from 'nanostores' import type { ComposerAttachment } from './composer' +export type GatewayRequester = (method: string, params?: Record) => Promise + export interface QueuedPromptEntry { id: string text: string @@ -9,57 +26,43 @@ export interface QueuedPromptEntry { queuedAt: number } +/** A steer accepted by the gateway but not yet injected into the live turn. */ +export interface PendingSteerEntry { + id: string + text: string + steeredAt: number +} + type QueueState = Record +type SteerState = Record -const STORAGE_KEY = 'hermes.desktop.composerQueue.v1' +// Legacy localStorage key from the client-owned queue era. Read once by +// migrateLegacyQueue() to push stranded entries to the gateway, then cleared. +const LEGACY_STORAGE_KEY = 'hermes.desktop.composerQueue.v1' -const load = (): QueueState => { - if (typeof window === 'undefined') { - return {} +export const $queuedPromptsBySession = atom({}) +export const $pendingSteersBySession = atom({}) + +// Module-level gateway request function — set once by the gateway boot hook so +// this store can fire RPCs without being a React component. +let _callGateway: GatewayRequester | null = null + +export const setGatewayRequester = (fn: GatewayRequester | null) => { + _callGateway = fn +} + +const callGateway = async (method: string, params?: Record): Promise => { + if (!_callGateway) { + return null } try { - const raw = window.localStorage.getItem(STORAGE_KEY) - const parsed = raw ? JSON.parse(raw) : null - - return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? (parsed as QueueState) : {} + return await _callGateway(method, params) } catch { - return {} + return null } } -const save = (state: QueueState) => { - if (typeof window === 'undefined') { - return - } - - try { - if (Object.keys(state).length === 0) { - window.localStorage.removeItem(STORAGE_KEY) - } else { - window.localStorage.setItem(STORAGE_KEY, JSON.stringify(state)) - } - } catch { - // best-effort: storage may be unavailable, queue still works in-memory - } -} - -export const $queuedPromptsBySession = atom(load()) - -const writeSession = (sid: string, queue: QueuedPromptEntry[]) => { - const current = $queuedPromptsBySession.get() - const next = { ...current } - - if (queue.length === 0) { - delete next[sid] - } else { - next[sid] = queue - } - - $queuedPromptsBySession.set(next) - save(next) -} - const sidOf = (key: string | null | undefined): null | string => { const trimmed = key?.trim() @@ -68,9 +71,45 @@ const sidOf = (key: string | null | undefined): null | string => { const queueFor = (sid: string) => $queuedPromptsBySession.get()[sid] ?? [] -const nextId = () => `queued-${Date.now()}-${Math.random().toString(36).slice(2, 8)}` +const writeSession = (sid: string, queue: QueuedPromptEntry[]) => { + const next = { ...$queuedPromptsBySession.get() } -const cloneAttachments = (attachments: ComposerAttachment[]) => attachments.map(a => ({ ...a })) + if (queue.length === 0) { + delete next[sid] + } else { + next[sid] = queue + } + + $queuedPromptsBySession.set(next) +} + +interface GatewayQueueEntry { + id?: string + text?: string + queued_at?: number +} + +/** Replace a session's mirror with the gateway's authoritative entry list + * (called from `queue.updated` events and `session.info.queue`). */ +export const setSessionQueue = (key: string | null | undefined, entries: unknown[]) => { + const sid = sidOf(key) + + if (!sid) { + return + } + + writeSession( + sid, + entries + .filter((e): e is GatewayQueueEntry => Boolean(e) && typeof e === 'object') + .map(e => ({ + id: String(e.id ?? ''), + text: String(e.text ?? ''), + attachments: [], + queuedAt: typeof e.queued_at === 'number' ? Math.round(e.queued_at * 1000) : 0 + })) + ) +} export const getQueuedPrompts = (key: string | null | undefined): QueuedPromptEntry[] => { const sid = sidOf(key) @@ -78,46 +117,48 @@ export const getQueuedPrompts = (key: string | null | undefined): QueuedPromptEn return sid ? queueFor(sid) : [] } -export const enqueueQueuedPrompt = ( +// ── RPC-backed mutations ───────────────────────────────────────────── +// Each mutation updates the local mirror optimistically, then fires the RPC; +// the gateway's queue.updated event is the authoritative settle. + +/** + * Queue a prompt on the gateway. The gateway drains it as the next turn (or + * immediately when the session is idle). Returns the created entry on accept, + * null when the gateway is unreachable — the caller keeps the draft so no + * words are lost. + */ +export const enqueueQueuedPrompt = async ( key: string | null | undefined, - payload: { text: string; attachments: ComposerAttachment[] } -): null | QueuedPromptEntry => { + payload: { text: string } +): Promise => { const sid = sidOf(key) if (!sid) { return null } + const result = await callGateway<{ entry_id?: string; status?: string }>('session.queue.add', { + session_id: sid, + text: payload.text + }) + + if (result?.status !== 'queued') { + return null + } + const entry: QueuedPromptEntry = { - id: nextId(), + id: result.entry_id ?? `local-${Date.now()}`, text: payload.text, - attachments: cloneAttachments(payload.attachments), + attachments: [], queuedAt: Date.now() } - writeSession(sid, [...queueFor(sid), entry]) + // Optimistic append — queue.updated settles the authoritative list. + writeSession(sid, [...queueFor(sid).filter(e => e.id !== entry.id), entry]) return entry } -export const dequeueQueuedPrompt = (key: string | null | undefined): null | QueuedPromptEntry => { - const sid = sidOf(key) - - if (!sid) { - return null - } - - const [head, ...rest] = queueFor(sid) - - if (!head) { - return null - } - - writeSession(sid, rest) - - return head -} - export const removeQueuedPrompt = (key: string | null | undefined, id: string): boolean => { const sid = sidOf(key) @@ -133,11 +174,21 @@ export const removeQueuedPrompt = (key: string | null | undefined, id: string): } writeSession(sid, next) + void callGateway('session.queue.remove', { session_id: sid, entry_id: id }) return true } -export const promoteQueuedPrompt = (key: string | null | undefined, id: string): boolean => { +/** + * Move an entry to the head so it fires next. With `interrupt`, also winds + * down the live turn (queue preserved) so the entry fires as soon as the turn + * settles — the "send now" gesture. + */ +export const promoteQueuedPrompt = ( + key: string | null | undefined, + id: string, + options?: { interrupt?: boolean } +): boolean => { const sid = sidOf(key) if (!sid) { @@ -147,12 +198,20 @@ export const promoteQueuedPrompt = (key: string | null | undefined, id: string): const queue = queueFor(sid) const index = queue.findIndex(e => e.id === id) - if (index <= 0) { + if (index < 0) { return false } - const entry = queue[index]! - writeSession(sid, [entry, ...queue.slice(0, index), ...queue.slice(index + 1)]) + if (index > 0) { + const entry = queue[index]! + writeSession(sid, [entry, ...queue.slice(0, index), ...queue.slice(index + 1)]) + } + + void callGateway('session.queue.promote', { + session_id: sid, + entry_id: id, + interrupt: Boolean(options?.interrupt) + }) return true } @@ -172,19 +231,13 @@ export const updateQueuedPrompt = ( let changed = false const next = queue.map(entry => { - if (entry.id !== id) { - return entry - } - - const attachments = update.attachments ? cloneAttachments(update.attachments) : entry.attachments - - if (entry.text === update.text && !update.attachments) { + if (entry.id !== id || entry.text === update.text) { return entry } changed = true - return { ...entry, text: update.text, attachments } + return { ...entry, text: update.text } }) if (!changed) { @@ -192,6 +245,7 @@ export const updateQueuedPrompt = ( } writeSession(sid, next) + void callGateway('session.queue.update', { session_id: sid, entry_id: id, text: update.text }) return true } @@ -199,23 +253,25 @@ export const updateQueuedPrompt = ( export const updateQueuedPromptText = (key: string | null | undefined, id: string, text: string): boolean => updateQueuedPrompt(key, id, { text }) +/** Clear the local mirror and the gateway queue (session close/delete). */ export const clearQueuedPrompts = (key: string | null | undefined) => { const sid = sidOf(key) - if (!sid || !(sid in $queuedPromptsBySession.get())) { + if (!sid) { return } - writeSession(sid, []) + if (sid in $queuedPromptsBySession.get()) { + writeSession(sid, []) + } + + clearPendingSteers(sid) + void callGateway('session.queue.clear', { session_id: sid }) } -/** - * Move pending entries from a dead session key onto a live one, preserving FIFO - * (existing target entries first, migrated entries appended). A backend bounce / - * resume can mint a fresh runtime session id for the *same* conversation; the - * entries enqueued under the old id would otherwise be stranded under a key - * nothing reads anymore. No-op unless both keys resolve and differ. - */ +/** Local-only mirror re-key when a backend bounce mints a fresh runtime id + * for the same conversation. The gateway re-syncs via session.info/queue + * events; this just keeps the panel from flashing empty in between. */ export const migrateQueuedPrompts = (fromKey: string | null | undefined, toKey: string | null | undefined): boolean => { const from = sidOf(fromKey) const to = sidOf(toKey) @@ -235,29 +291,123 @@ export const migrateQueuedPrompts = (fromKey: string | null | undefined, toKey: next[to] = [...queueFor(to), ...pending] $queuedPromptsBySession.set(next) - save(next) return true } -/** Inputs to {@link shouldAutoDrain}. */ -export interface AutoDrainInput { - isBusy: boolean - queueLength: number +/** One-time migration: push any entries stranded in the legacy localStorage + * queue (client-owned era) to the gateway, then clear the storage key. */ +export const migrateLegacyQueue = (key: string | null | undefined) => { + const sid = sidOf(key) + + if (!sid || typeof window === 'undefined') { + return + } + + try { + const raw = window.localStorage.getItem(LEGACY_STORAGE_KEY) + + if (!raw) { + return + } + + const parsed: unknown = JSON.parse(raw) + + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + window.localStorage.removeItem(LEGACY_STORAGE_KEY) + + return + } + + const state = parsed as Record + const entries = state[sid] + + if (entries?.length) { + for (const entry of entries) { + if (entry?.text?.trim()) { + void enqueueQueuedPrompt(sid, { text: entry.text }) + } + } + } + + delete state[sid] + + if (Object.keys(state).length === 0) { + window.localStorage.removeItem(LEGACY_STORAGE_KEY) + } else { + window.localStorage.setItem(LEGACY_STORAGE_KEY, JSON.stringify(state)) + } + } catch { + // Best-effort — a broken legacy blob shouldn't take down the composer. + } } -/** - * Decide whether the composer should auto-drain the next queued prompt. - * - * Edge-independent on purpose: the queue must advance whenever the session is - * idle and has pending entries, NOT only on an observed busy true → false edge. - * A backend bounce / websocket reconnect remounts the composer and resets the - * busy ref to the current value, swallowing the settle edge — an edge-gated - * drain would then strand the entry forever. The caller's drain lock - * (`drainingQueueRef`) serializes sends so being edge-free can't double-submit. - */ -export const shouldAutoDrain = ({ isBusy, queueLength }: AutoDrainInput): boolean => !isBusy && queueLength > 0 +// ── Pending steers ─────────────────────────────────────────────────── +// A steer RPC is accepted instantly, but the text only reaches the model at +// the next tool-batch boundary. These helpers track that in-between state so +// the transcript can show the steer when it actually lands (steer.applied) +// instead of pretending it was instant. -/** Auto-drain attempts for one entry before we stop retrying and toast. The - * entry stays queued for a manual send; a remount/reconnect resets the count. */ -export const MAX_AUTO_DRAIN_ATTEMPTS = 4 +const steersFor = (sid: string) => $pendingSteersBySession.get()[sid] ?? [] + +const writeSteers = (sid: string, steers: PendingSteerEntry[]) => { + const next = { ...$pendingSteersBySession.get() } + + if (steers.length === 0) { + delete next[sid] + } else { + next[sid] = steers + } + + $pendingSteersBySession.set(next) +} + +export const getPendingSteers = (key: string | null | undefined): PendingSteerEntry[] => { + const sid = sidOf(key) + + return sid ? steersFor(sid) : [] +} + +export const addPendingSteer = (key: string | null | undefined, text: string) => { + const sid = sidOf(key) + + if (!sid) { + return + } + + writeSteers(sid, [ + ...steersFor(sid), + { id: `steer-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, text, steeredAt: Date.now() } + ]) +} + +/** 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. */ +export const settlePendingSteer = (key: string | null | undefined, text: string) => { + const sid = sidOf(key) + + if (!sid) { + return + } + + const settled = new Set( + text + .split('\n') + .map(line => line.trim()) + .filter(Boolean) + ) + + writeSteers( + sid, + steersFor(sid).filter(entry => !settled.has(entry.text.trim())) + ) +} + +export const clearPendingSteers = (key: string | null | undefined) => { + const sid = sidOf(key) + + if (sid && sid in $pendingSteersBySession.get()) { + writeSteers(sid, []) + } +} diff --git a/run_agent.py b/run_agent.py index bcacec3909b..04422da29b0 100644 --- a/run_agent.py +++ b/run_agent.py @@ -2829,8 +2829,28 @@ class AIAgent: # late injection on the post-interrupt turn. _steer_lock = getattr(self, "_pending_steer_lock", None) if _steer_lock is not None: + _dropped = None with _steer_lock: + _dropped = self._pending_steer self._pending_steer = None + if _dropped: + self._emit_steer_event("dropped", _dropped) + + def _emit_steer_event(self, kind: str, text: str) -> None: + """Notify the steer lifecycle observer (if any). + + ``kind`` is ``"applied"`` when the steer text was really injected + into the conversation (the model will see it), or ``"dropped"`` when + a hard interrupt discarded it. Never raises — the observer is a + client-side courtesy, not part of the turn's control flow. + """ + callback = getattr(self, "_on_steer_event", None) + if callback is None: + return + try: + callback(kind, text) + except Exception: + pass def steer(self, text: str) -> bool: """ diff --git a/tests/tui_gateway/test_turn_queue.py b/tests/tui_gateway/test_turn_queue.py new file mode 100644 index 00000000000..99e3550aeaa --- /dev/null +++ b/tests/tui_gateway/test_turn_queue.py @@ -0,0 +1,300 @@ +"""Tests for the agent-side TurnQueue and the session.queue.* gateway RPCs. + +The queue lives on the agent (agent.turn_queue) so it drains even when no +client window is open — see agent/turn_queue.py. The gateway RPCs are thin +manipulators over it; _drain_queued_prompt is the single drain point. +""" + +import io +import sys +import threading +import time +from unittest.mock import MagicMock, patch + +import pytest + +from agent.turn_queue import TurnQueue + +_original_stdout = sys.stdout + + +@pytest.fixture(autouse=True) +def _restore_stdout(): + yield + sys.stdout = _original_stdout + + +# ── TurnQueue unit behaviour ───────────────────────────────────────── + + +def test_enqueue_drain_fifo(): + q = TurnQueue() + a = q.enqueue("first") + b = q.enqueue("second") + + assert len(q) == 2 + assert bool(q) + assert q.drain().id == a.id + assert q.drain().id == b.id + assert q.drain() is None + assert not q + + +def test_remove_promote_update_clear(): + q = TurnQueue() + a = q.enqueue("a") + b = q.enqueue("b") + c = q.enqueue("c") + + assert q.promote(c.id) is True + assert [e.id for e in q.peek()] == [c.id, a.id, b.id] + # Already at head → False (callers treat membership as "will go next"). + assert q.promote(c.id) is False + assert q.promote("missing") is False + + assert q.update_text(a.id, "a2") is True + assert q.update_text("missing", "x") is False + + assert q.remove(b.id) is True + assert q.remove(b.id) is False + + assert q.clear() == 2 + assert q.drain() is None + + +def test_to_dict_shape_excludes_transport(): + q = TurnQueue() + q.enqueue("hello", transport=object(), source="busy_submit") + (d,) = q.to_list() + + assert d["text"] == "hello" + assert d["source"] == "busy_submit" + assert "transport" not in d + assert isinstance(d["queued_at"], float) + + +def test_thread_safety_under_concurrent_enqueue_drain(): + q = TurnQueue() + drained = [] + + def producer(): + for i in range(200): + q.enqueue(f"msg-{i}") + + def consumer(): + deadline = time.time() + 5 + while len(drained) < 200 and time.time() < deadline: + entry = q.drain() + if entry is not None: + drained.append(entry.text) + + t1 = threading.Thread(target=producer) + t2 = threading.Thread(target=consumer) + t1.start(); t2.start() + t1.join(); t2.join() + + assert len(drained) == 200 + assert len(set(drained)) == 200 + + +# ── Gateway RPC integration ────────────────────────────────────────── + + +@pytest.fixture() +def server(): + with patch.dict("sys.modules", { + "hermes_constants": MagicMock(get_hermes_home=MagicMock(return_value="/tmp/hermes_test")), + "hermes_cli.env_loader": MagicMock(), + "hermes_cli.banner": MagicMock(), + "hermes_state": MagicMock(), + }): + import importlib + mod = importlib.import_module("tui_gateway.server") + yield mod + mod._sessions.clear() + mod._pending.clear() + mod._answers.clear() + + +def _make_session(server, sid, running=True): + agent = MagicMock(spec=["turn_queue", "steer", "interrupt"]) + agent.turn_queue = TurnQueue() + session = { + "session_key": sid, + "agent": agent, + "running": running, + "history_lock": threading.Lock(), + "last_active": 0, + } + server._sessions[sid] = session + return session + + +def test_queue_add_lists_and_emits(server): + sid = "s1" + session = _make_session(server, sid, running=True) + events = [] + with patch.object(server, "_emit", side_effect=lambda ev, s, p=None: events.append((ev, p))): + resp = server.handle_request({ + "id": "r1", + "method": "session.queue.add", + "params": {"session_id": sid, "text": "queued msg"}, + }) + + assert "error" not in resp + assert resp["result"]["status"] == "queued" + entry_id = resp["result"]["entry_id"] + assert entry_id + + # Mirrors into the agent queue; busy session → NOT drained. + assert [e.text for e in session["agent"].turn_queue.peek()] == ["queued msg"] + assert ("queue.updated", {"entries": session["agent"].turn_queue.to_list()}) in events + + listed = server.handle_request({ + "id": "r2", + "method": "session.queue.list", + "params": {"session_id": sid}, + }) + assert [e["id"] for e in listed["result"]["entries"]] == [entry_id] + + +def test_queue_add_empty_text_rejected(server): + _make_session(server, "s1") + resp = server.handle_request({ + "id": "r1", + "method": "session.queue.add", + "params": {"session_id": "s1", "text": " "}, + }) + assert resp["error"]["code"] == 4002 + + +def test_queue_add_idle_session_drains_immediately(server): + sid = "s1" + session = _make_session(server, sid, running=False) + 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.add", + "params": {"session_id": sid, "text": "run now"}, + }) + assert resp["result"]["status"] == "queued" + assert drained.wait(timeout=5), "idle enqueue should drain immediately" + + assert submitted["text"] == "run now" + assert session["agent"].turn_queue.drain() is None + + +def test_queue_remove_promote_update_clear_rpcs(server): + sid = "s1" + session = _make_session(server, sid, running=True) + q = session["agent"].turn_queue + a = q.enqueue("a") + b = q.enqueue("b") + + with patch.object(server, "_emit"): + promoted = server.handle_request({ + "id": "r1", "method": "session.queue.promote", + "params": {"session_id": sid, "entry_id": b.id}, + }) + assert promoted["result"]["promoted"] is True + assert [e.id for e in q.peek()] == [b.id, a.id] + + updated = server.handle_request({ + "id": "r2", "method": "session.queue.update", + "params": {"session_id": sid, "entry_id": a.id, "text": "a-edited"}, + }) + assert updated["result"]["updated"] is True + + removed = server.handle_request({ + "id": "r3", "method": "session.queue.remove", + "params": {"session_id": sid, "entry_id": b.id}, + }) + assert removed["result"]["removed"] is True + + cleared = server.handle_request({ + "id": "r4", "method": "session.queue.clear", + "params": {"session_id": sid}, + }) + assert cleared["result"]["cleared"] == 1 + assert q.drain() is None + + +def test_queue_promote_interrupt_forwards_to_agent(server): + sid = "s1" + session = _make_session(server, sid, running=True) + q = session["agent"].turn_queue + q.enqueue("a") + b = q.enqueue("b") + + with patch.object(server, "_emit"): + resp = server.handle_request({ + "id": "r1", "method": "session.queue.promote", + "params": {"session_id": sid, "entry_id": b.id, "interrupt": True}, + }) + + assert resp["result"]["promoted"] is True + session["agent"].interrupt.assert_called_once() + # The queue survives the interrupt — that's the whole point of send-now. + assert len(q) == 2 + + +def test_queue_rpcs_unknown_session(server): + for method in ( + "session.queue.list", + "session.queue.clear", + ): + resp = server.handle_request({ + "id": "r", "method": method, "params": {"session_id": "nope"}, + }) + assert resp["error"]["code"] == 4001 + + +def test_drain_queued_prompt_pops_agent_queue_and_emits_drained(server): + sid = "s1" + session = _make_session(server, sid, running=False) + session["agent"].turn_queue.enqueue("next turn", source="queue") + events = [] + with patch.object(server, "_run_prompt_submit"), \ + 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.updated" in kinds + drained = dict(events)["queue.drained"] + assert drained == {"text": "next turn", "source": "queue"} + + +def test_drain_queued_prompt_skips_when_running(server): + sid = "s1" + session = _make_session(server, sid, running=True) + session["agent"].turn_queue.enqueue("waiting") + + with patch.object(server, "_run_prompt_submit") as submit: + assert server._drain_queued_prompt("rid", sid, session) is False + submit.assert_not_called() + + assert len(session["agent"].turn_queue) == 1 + + +def test_busy_submit_enqueues_on_agent_queue(server): + """_handle_busy_submit routes mid-turn prompts into agent.turn_queue with + source=busy_submit so drain events tell clients the text was already echoed.""" + sid = "s1" + session = _make_session(server, sid, running=True) + + with patch.object(server, "_emit"), \ + patch.object(server, "_load_busy_input_mode", return_value="queue"): + resp = server._handle_busy_submit("rid", sid, session, "mid-turn msg", None) + + assert resp["result"]["status"] == "queued" + (entry,) = session["agent"].turn_queue.peek() + assert entry.text == "mid-turn msg" + assert entry.source == "busy_submit" diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 2f6e833934f..35c7ad14d04 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -3355,6 +3355,13 @@ def _current_profile_name() -> str: DESKTOP_BACKEND_CONTRACT = 3 +def _queue_info(agent) -> list: + """Return the agent's turn queue as a list of dicts (for session.info).""" + if agent is not None and hasattr(agent, "turn_queue"): + return agent.turn_queue.to_list() + return [] + + def _session_info(agent, session: dict | None = None) -> dict: if session is None: for candidate in _sessions.values(): @@ -3411,6 +3418,7 @@ def _session_info(agent, session: dict | None = None) -> dict: "branch": _git_branch_for_cwd(cwd), "personality": str(personality or ""), "running": bool((session or {}).get("running")), + "queue": _queue_info(agent), "title": _session_live_title(session or {}, session_key) if session_key else "", "desktop_contract": DESKTOP_BACKEND_CONTRACT, "version": "", @@ -5101,15 +5109,35 @@ def _clear_inflight_turn(session: dict) -> None: session["inflight_turn"] = None +def _emit_queue_update(sid: str, session: dict) -> None: + """Emit a ``queue.updated`` event so clients can mirror queue state. + + Called after every mutation (enqueue, drain, remove, clear, promote). + The desktop renderer subscribes to this instead of managing its own + localStorage-backed queue. + """ + agent = session.get("agent") + if agent is None or not hasattr(agent, "turn_queue"): + return + _emit("queue.updated", sid, {"entries": agent.turn_queue.to_list()}) + + def _enqueue_prompt(session: dict, text: Any, transport: Any) -> None: """Stash a message to run as the very next turn once the live one ends. - Used when a prompt arrives mid-turn (see ``_handle_busy_submit``). A single - slot is kept; a second arrival is merged (lossless, mirroring the - consecutive-user merge in ``repair_message_sequence``) so nothing the user - typed is dropped. ``transport`` is pinned so the drained turn streams back to - the client that sent it even if the session transport is rebound meanwhile. + Used when a prompt arrives mid-turn (see ``_handle_busy_submit``). The + message is appended to the agent's :class:`TurnQueue` — the queue lives + on the agent so it drains even when no client window is open. + ``transport`` is pinned so the drained turn streams back to the client + that sent it even if the session transport is rebound meanwhile. """ + agent = session.get("agent") + if agent is not None and hasattr(agent, "turn_queue"): + agent.turn_queue.enqueue(text=str(text), transport=transport, source="busy_submit") + return + # Fallback: the legacy single-slot session dict, for sessions without a + # built agent yet (shouldn't happen in normal flow — prompt.submit builds + # the agent before any busy-submit can occur — but keeps the contract safe). existing = session.get("queued_prompt") if ( existing @@ -5152,6 +5180,7 @@ def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any) pass _enqueue_prompt(session, text, transport) session["last_active"] = time.time() + _emit_queue_update(sid, session) return _ok(rid, {"status": "queued"}) @@ -5162,14 +5191,45 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool: lower-priority follow-ups this cycle — the user's message wins). Mirrors the claim-under-lock pattern used by the goal-continuation re-fire. """ + agent = session.get("agent") with session["history_lock"]: - queued = session.get("queued_prompt") - if not queued or session.get("running"): + if session.get("running"): return False - session["queued_prompt"] = None + # Drain from the agent's TurnQueue (preferred), falling back to the + # legacy session-dict slot (populated only when no agent existed at + # enqueue time). + queued = None + if agent is not None and hasattr(agent, "turn_queue"): + entry = agent.turn_queue.drain() + if entry is not None: + queued = { + "text": entry.text, + "transport": entry.transport, + "source": entry.source, + } + if queued is None: + legacy = session.get("queued_prompt") + if not legacy: + return False + session["queued_prompt"] = None + queued = { + "text": legacy.get("text"), + "transport": legacy.get("transport"), + "source": "busy_submit", + } session["running"] = True 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: @@ -8174,7 +8234,17 @@ def _(rid, params: dict) -> dict: session["agent"].interrupt() with session["history_lock"]: 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() session["queued_prompt"] = None + _emit_queue_update(str(params.get("session_id", "")), session or {}) if not run_thread_alive: with session["history_lock"]: if session.get("running"): @@ -8449,6 +8519,144 @@ def _(rid, params: dict) -> dict: return _ok(rid, {"status": "queued" if accepted else "rejected", "text": text}) +# ── Methods: session.queue ──────────────────────────────────────────── +# +# The turn queue lives on the agent (agent.turn_queue) — see +# agent/turn_queue.py. These RPCs let clients manipulate it without owning +# it; the gateway drains automatically at the end of every turn +# (_drain_queued_prompt), so queued messages fire even when no client +# window is open. Every mutation emits queue.updated so clients mirror +# state instead of persisting their own copy. + + +@method("session.queue.add") +def _(rid, params: dict) -> dict: + """Add a prompt to the session's turn queue. + + The prompt runs as the next turn once the current one finishes. When the + session is already idle, the entry is drained immediately — the gateway + owns all drain paths, so clients never need a "send it myself" fallback. + """ + text = str(params.get("text") or "") + if not text.strip(): + return _err(rid, 4002, "text is required") + session, err = _sess_nowait(params, rid) + if err or session is None: + return err or _err(rid, 4001, "session not found") + sid = str(params.get("session_id", "")) + agent = session.get("agent") + if agent is None or not hasattr(agent, "turn_queue"): + return _err(rid, 4010, "session has no turn queue") + entry = agent.turn_queue.enqueue(text=text) + session["last_active"] = time.time() + _emit_queue_update(sid, session) + # Idle session → drain right away (in a thread; the drain runs a whole + # turn and this RPC must return promptly). + if not session.get("running"): + threading.Thread( + target=_drain_queued_prompt, + args=(rid, sid, session), + daemon=True, + ).start() + return _ok(rid, {"status": "queued", "entry_id": entry.id}) + + +@method("session.queue.list") +def _(rid, params: dict) -> dict: + """List pending queued prompts for the session.""" + session, err = _sess_nowait(params, rid) + if err or session is None: + return err or _err(rid, 4001, "session not found") + agent = session.get("agent") + return _ok(rid, {"entries": _queue_info(agent)}) + + +@method("session.queue.remove") +def _(rid, params: dict) -> dict: + """Remove a specific queued entry by id.""" + entry_id = str(params.get("entry_id") or "") + if not entry_id: + return _err(rid, 4002, "entry_id is required") + session, err = _sess_nowait(params, rid) + if err or session is None: + return err or _err(rid, 4001, "session not found") + sid = str(params.get("session_id", "")) + agent = session.get("agent") + removed = False + if agent is not None and hasattr(agent, "turn_queue"): + removed = agent.turn_queue.remove(entry_id) + _emit_queue_update(sid, session) + return _ok(rid, {"removed": removed}) + + +@method("session.queue.clear") +def _(rid, params: dict) -> dict: + """Clear all queued entries for the session.""" + session, err = _sess_nowait(params, rid) + if err or session is None: + return err or _err(rid, 4001, "session not found") + sid = str(params.get("session_id", "")) + agent = session.get("agent") + count = 0 + if agent is not None and hasattr(agent, "turn_queue"): + count = agent.turn_queue.clear() + session["queued_prompt"] = None + _emit_queue_update(sid, session) + return _ok(rid, {"cleared": count}) + + +@method("session.queue.promote") +def _(rid, params: dict) -> dict: + """Move a queued entry to the front of the queue (send-next priority). + + 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. + """ + entry_id = str(params.get("entry_id") or "") + if not entry_id: + return _err(rid, 4002, "entry_id is required") + session, err = _sess_nowait(params, rid) + if err or session is None: + return err or _err(rid, 4001, "session not found") + sid = str(params.get("session_id", "")) + agent = session.get("agent") + promoted = False + if agent is not None and hasattr(agent, "turn_queue"): + # promote() returns False for an entry already at the head — that + # still counts as "it will go next", so verify membership instead. + promoted = agent.turn_queue.promote(entry_id) or any( + e.id == entry_id for e in agent.turn_queue.peek() + ) + _emit_queue_update(sid, session) + if promoted and params.get("interrupt") and session.get("running"): + if hasattr(agent, "interrupt"): + try: + agent.interrupt() + except Exception: + pass + return _ok(rid, {"promoted": promoted}) + + +@method("session.queue.update") +def _(rid, params: dict) -> dict: + """Update the text of a queued entry.""" + entry_id = str(params.get("entry_id") or "") + text = str(params.get("text") or "") + if not entry_id or not text.strip(): + return _err(rid, 4002, "entry_id and text are required") + session, err = _sess_nowait(params, rid) + if err or session is None: + return err or _err(rid, 4001, "session not found") + sid = str(params.get("session_id", "")) + agent = session.get("agent") + updated = False + if agent is not None and hasattr(agent, "turn_queue"): + updated = agent.turn_queue.update_text(entry_id, text) + _emit_queue_update(sid, session) + return _ok(rid, {"updated": updated}) + + @method("terminal.resize") def _(rid, params: dict) -> dict: session, err = _sess_nowait(params, rid) @@ -8964,6 +9172,17 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None: agent.clear_interrupt() except Exception: pass + # Steer lifecycle → client events. A steer is accepted instantly by the + # RPC but only *lands* when the agent loop injects it into a tool result + # (or drops it on interrupt). Clients render the steer as pending until + # 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} + ) + except Exception: + pass _emit("message.start", sid) def run(): @@ -9150,6 +9369,15 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None: last_reasoning = None status_note = None if isinstance(result, dict): + # A /steer that landed after the final assistant message has + # no tool batch left to inject into — run_conversation hands + # it back as pending_steer. Queue it as the next user turn + # (the post-turn drain below fires it) instead of silently + # dropping it. + _leftover_steer = result.get("pending_steer") + if _leftover_steer and hasattr(agent, "turn_queue"): + agent.turn_queue.enqueue(text=str(_leftover_steer)) + _emit_queue_update(sid, session) if isinstance(result.get("messages"), list): with session["history_lock"]: current_version = int(session.get("history_version", 0))