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