mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
feat(desktop): move queue/steer to the agent-side TurnQueue so queued messages fire without the tab open
Queued messages in the desktop app lived in localStorage and were drained
by a React useEffect on the busy->false edge — if the session tab wasn't
mounted, the effect never ran and queued prompts sat dormant forever.
Move the queue into the agent process where steer already lives:
- agent/turn_queue.py: TurnQueue, a thread-safe FIFO on AIAgent (wired in
agent_init next to _pending_steer). Entries carry a `source` field
("queue" vs "busy_submit") so drain events tell clients whether the
text was already echoed optimistically.
- tui_gateway: _enqueue_prompt/_drain_queued_prompt delegate to
agent.turn_queue; new session.queue.add/list/remove/clear/promote/update
RPCs; queue.updated + queue.drained events; queue in session.info.
An idle-session enqueue drains immediately — the gateway owns every
drain path. session.interrupt clears the queue (keep_queue opts out
for the promote+interrupt "send now" gesture). A leftover pending_steer
returned by run_conversation is re-queued as the next turn instead of
being silently dropped.
- steer honesty: new agent._on_steer_event observer fires steer.applied
at the two real injection sites (pre-API drain + tool-batch drain) and
steer.dropped when an interrupt discards the pending steer. The desktop
shows steers as pending in the queue panel and only appends the steer:
transcript row when the model actually saw the text (both the primary
composer and session tiles previously painted it at RPC-accept time).
- desktop: composer-queue.ts rewritten as a gateway-backed mirror
(optimistic updates settled by queue.updated); the auto-drain effect is
deleted; "send now" promotes on the gateway; attachments resolve to
@file: refs at enqueue time so queued text is self-contained when the
gateway drains it later; one-time localStorage migration. Dead code
removed: fromQueue submit option, shouldAutoDrain, queueStuck i18n.
Tests: tests/tui_gateway/test_turn_queue.py (TurnQueue unit + RPC
integration + drain semantics), composer-queue.test.ts rewritten for the
gateway-backed store.
This commit is contained in:
ethernet
parent
c81afd6e50
commit
bd70904c26
33 changed files with 1431 additions and 498 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
188
agent/turn_queue.py
Normal file
188
agent/turn_queue.py
Normal file
|
|
@ -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]
|
||||
|
|
@ -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<string>
|
||||
focusInput: () => void
|
||||
loadIntoComposer: (text: string, attachments: ComposerAttachment[]) => void
|
||||
onCancel: ChatBarProps['onCancel']
|
||||
onSubmit: ChatBarProps['onSubmit']
|
||||
onQueue: ChatBarProps['onQueue']
|
||||
queueEditRef: RefObject<QueueEditState | null>
|
||||
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<QueueEditState | null>(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<string, number>())
|
||||
|
||||
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<boolean> => {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<string>
|
||||
drainNextQueued: () => Promise<boolean>
|
||||
drainNextQueued: () => boolean
|
||||
editorRef: RefObject<HTMLDivElement | null>
|
||||
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<boolean>
|
||||
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))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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. */}
|
||||
<ComposerStatusStack
|
||||
queue={
|
||||
activeQueueSessionKey && queuedPrompts.length > 0 ? (
|
||||
activeQueueSessionKey && (queuedPrompts.length > 0 || pendingSteers.length > 0) ? (
|
||||
<QueuePanel
|
||||
busy={busy}
|
||||
editingId={queueEdit?.entryId ?? null}
|
||||
|
|
@ -895,6 +897,7 @@ export function ChatBar({
|
|||
}}
|
||||
onEdit={beginQueuedEdit}
|
||||
onSendNow={id => void sendQueuedNow(id)}
|
||||
pendingSteers={pendingSteers}
|
||||
/>
|
||||
) : null
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 (
|
||||
<StatusSection
|
||||
icon={<Codicon className="text-muted-foreground/70" name="layers" size="0.8rem" />}
|
||||
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 => (
|
||||
<StatusRow key={steer.id}>
|
||||
<div className="min-w-0 flex-1">
|
||||
<p className="truncate text-[0.73rem] leading-4 text-foreground/92">{steer.text}</p>
|
||||
<div className="mt-0.5 flex items-center gap-1.5 text-[0.64rem] text-muted-foreground/75">
|
||||
<span>{c.steerPending}</span>
|
||||
</div>
|
||||
</div>
|
||||
</StatusRow>
|
||||
))}
|
||||
{entries.map(entry => {
|
||||
const isEditing = editingId === entry.id
|
||||
const attachmentsCount = entry.attachments.length
|
||||
|
|
|
|||
|
|
@ -52,10 +52,10 @@ export interface ChatBarProps {
|
|||
onPickImages?: () => void
|
||||
onRemoveAttachment?: (id: string) => void
|
||||
onSteer?: (text: string) => Promise<boolean> | boolean
|
||||
onSubmit: (
|
||||
value: string,
|
||||
options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }
|
||||
) => Promise<boolean> | 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> | boolean
|
||||
onSubmit: (value: string, options?: { attachments?: ComposerAttachment[] }) => Promise<boolean> | boolean
|
||||
onTranscribeAudio?: (audio: Blob) => Promise<string>
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -74,10 +74,8 @@ interface ChatViewProps extends Omit<React.ComponentProps<'div'>, 'onSubmit'> {
|
|||
onPickImages: () => void
|
||||
onRemoveAttachment: (id: string) => void
|
||||
onSteer: (text: string) => Promise<boolean> | boolean
|
||||
onSubmit: (
|
||||
text: string,
|
||||
options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }
|
||||
) => Promise<boolean> | boolean
|
||||
onQueue?: (text: string, options?: { attachments?: ComposerAttachment[] }) => Promise<boolean> | boolean
|
||||
onSubmit: (text: string, options?: { attachments?: ComposerAttachment[] }) => Promise<boolean> | boolean
|
||||
onThreadMessagesChange: (messages: readonly ThreadMessage[]) => void
|
||||
onEdit: (message: AppendMessage) => Promise<void>
|
||||
onReload: (parentId: string | null) => Promise<void>
|
||||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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<boolean> => {
|
||||
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
|
||||
]
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ export type ChatActions = Pick<
|
|||
| 'onRestoreToMessage'
|
||||
| 'onRetryResume'
|
||||
| 'onSteer'
|
||||
| 'onQueue'
|
||||
| 'onSubmit'
|
||||
| 'onThreadMessagesChange'
|
||||
| 'onToggleSelectedPin'
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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 }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ interface HarnessHandle {
|
|||
cancelRun: () => Promise<void>
|
||||
restoreToMessage: (messageId: string, target?: { text?: string; userOrdinal?: number | null }) => Promise<void>
|
||||
steerPrompt: (text: string) => Promise<boolean>
|
||||
submitText: (text: string, options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }) => Promise<boolean>
|
||||
submitText: (text: string, options?: { attachments?: ComposerAttachment[] }) => Promise<boolean>
|
||||
}
|
||||
|
||||
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(
|
||||
<Harness
|
||||
busyRef={busyRef}
|
||||
onReady={h => (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',
|
||||
|
|
|
|||
|
|
@ -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<boolean> => {
|
||||
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<boolean> => {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -59,10 +59,7 @@ interface SlashCommandDeps {
|
|||
requestGateway: GatewayRequest
|
||||
resumeStoredSession: (storedSessionId: string) => Promise<void> | void
|
||||
startFreshSessionDraft: () => void
|
||||
submitPromptText: (
|
||||
rawText: string,
|
||||
options?: { attachments?: ComposerAttachment[]; fromQueue?: boolean }
|
||||
) => Promise<boolean>
|
||||
submitPromptText: (rawText: string, options?: { attachments?: ComposerAttachment[] }) => Promise<boolean>
|
||||
}
|
||||
|
||||
/** The /slash command dispatcher, extracted from usePromptActions. */
|
||||
|
|
|
|||
|
|
@ -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 => ({
|
||||
|
|
|
|||
|
|
@ -225,5 +225,4 @@ export function visibleUserIndexAtOrdinal(messages: readonly ChatMessage[], targ
|
|||
|
||||
export interface SubmitTextOptions {
|
||||
attachments?: ComposerAttachment[]
|
||||
fromQueue?: boolean
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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}`,
|
||||
|
|
|
|||
|
|
@ -1672,9 +1672,6 @@ export const ja = defineLocale({
|
|||
queueSendNext: '次に送信',
|
||||
queueSend: '送信',
|
||||
queueDelete: '削除',
|
||||
queueStuckTitle: 'キュー内のメッセージを送信できません',
|
||||
queueStuckBody:
|
||||
'キューに入れたターンの送信が繰り返し失敗しました。まだキューに残っています。もう一度送信してください。',
|
||||
previewUnavailable: 'プレビューは利用できません',
|
||||
previewLabel: label => `${label} のプレビュー`,
|
||||
couldNotPreview: label => `${label} をプレビューできませんでした`,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -1622,8 +1622,6 @@ export const zhHant = defineLocale({
|
|||
queueSendNext: '下一個',
|
||||
queueSend: '傳送',
|
||||
queueDelete: '刪除',
|
||||
queueStuckTitle: '佇列訊息未送出',
|
||||
queueStuckBody: '佇列中的對話多次傳送失敗。它仍在佇列中，請重試傳送。',
|
||||
previewUnavailable: '預覽不可用',
|
||||
previewLabel: label => `預覽 ${label}`,
|
||||
couldNotPreview: label => `無法預覽 ${label}`,
|
||||
|
|
|
|||
|
|
@ -1931,8 +1931,7 @@ export const zh: Translations = {
|
|||
queueSendNext: '下一个',
|
||||
queueSend: '发送',
|
||||
queueDelete: '删除',
|
||||
queueStuckTitle: '排队消息未发送',
|
||||
queueStuckBody: '排队的对话多次发送失败。它仍在队列中，请重试发送。',
|
||||
steerPending: '正在引导 — 将在下一个工具步骤生效',
|
||||
previewUnavailable: '预览不可用',
|
||||
previewLabel: label => `预览 ${label}`,
|
||||
couldNotPreview: label => `无法预览 ${label}`,
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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<string, { text: string }[]>
|
||||
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'])
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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 = <T = unknown>(method: string, params?: Record<string, unknown>) => Promise<T>
|
||||
|
||||
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<string, QueuedPromptEntry[]>
|
||||
type SteerState = Record<string, PendingSteerEntry[]>
|
||||
|
||||
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<QueueState>({})
|
||||
export const $pendingSteersBySession = atom<SteerState>({})
|
||||
|
||||
// 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 <T = unknown>(method: string, params?: Record<string, unknown>): Promise<T | null> => {
|
||||
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<T>(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<QueueState>(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<null | QueuedPromptEntry> => {
|
||||
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<string, { text?: string }[]>
|
||||
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, [])
|
||||
}
|
||||
}
|
||||
|
|
|
|||
20
run_agent.py
20
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:
|
||||
"""
|
||||
|
|
|
|||
300
tests/tui_gateway/test_turn_queue.py
Normal file
300
tests/tui_gateway/test_turn_queue.py
Normal file
|
|
@ -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"
|
||||
|
|
@ -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))
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue