diff --git a/.changeset/web-steer-capslock-fix.md b/.changeset/web-steer-capslock-fix.md new file mode 100644 index 0000000000..46fa41b2d0 --- /dev/null +++ b/.changeset/web-steer-capslock-fix.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Fix Ctrl+S steering not triggering when CapsLock is on. diff --git a/.changeset/web-steer-queued-messages.md b/.changeset/web-steer-queued-messages.md new file mode 100644 index 0000000000..db2deac77a --- /dev/null +++ b/.changeset/web-steer-queued-messages.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Add "Steer now" buttons to the queued-message list to inject queued messages into the running turn — one message at a time or the whole queue from the queue header. Click the bolt button next to a queued message to steer it. diff --git a/.changeset/web-steer-telemetry.md b/.changeset/web-steer-telemetry.md new file mode 100644 index 0000000000..c8495a6630 --- /dev/null +++ b/.changeset/web-steer-telemetry.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Count steered messages sent from the web UI in usage telemetry, the same way steers from the TUI are counted. diff --git a/apps/kimi-web/src/App.vue b/apps/kimi-web/src/App.vue index 5708bcc9bf..299e6ae591 100644 --- a/apps/kimi-web/src/App.vue +++ b/apps/kimi-web/src/App.vue @@ -837,6 +837,7 @@ function openPr(url: string): void { @interrupt="client.abortCurrentPrompt()" @unqueue="handleUnqueue" @edit-queued="handleEditQueued" + @steer-queued="client.steerQueuedPrompt($event)" @reorder-queue="handleReorderQueue" @set-permission="client.setPermission($event)" @set-thinking="client.setThinking($event)" diff --git a/apps/kimi-web/src/components/chat/ChatPane.vue b/apps/kimi-web/src/components/chat/ChatPane.vue index 79ad6c5d19..f71b1ee5fe 100644 --- a/apps/kimi-web/src/components/chat/ChatPane.vue +++ b/apps/kimi-web/src/components/chat/ChatPane.vue @@ -15,6 +15,7 @@ import AttachmentChip from './AttachmentChip.vue'; import MoonSpinner from '../ui/MoonSpinner.vue'; import Spinner from '../ui/Spinner.vue'; import Icon from '../ui/Icon.vue'; +import Button from '../ui/Button.vue'; import Tooltip from '../ui/Tooltip.vue'; import { useConfirmDialog } from '../../composables/useConfirmDialog'; import { copyTextToClipboard } from '../../lib/clipboard'; @@ -213,6 +214,11 @@ const emit = defineEmits<{ unqueue: [index: number]; /** Load a queued message back into the composer for editing (and dequeue it). */ editQueued: [index: number]; + /** Steer the whole queue into the running turn now (like Ctrl+S, but the + * live composer draft stays behind). */ + steerQueue: []; + /** Steer one queued message (by index) into the running turn now. */ + steerQueued: [index: number]; /** Drag-to-reorder a queued message within the active session's queue. */ reorderQueue: [payload: { from: number; to: number }]; }>(); @@ -689,6 +695,16 @@ function isStreamingRenderBlock(turn: ChatTurn, block: { sourceIndex: number }): {{ t('composer.queueLabel') }} · {{ queued.length }} {{ t('composer.queueAutoDrain') }} +
{{ t('composer.queueNext') }} #{{ qi + 1 }} +
diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index 03a541fe0a..87b670d768 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -115,7 +115,7 @@ const startingFirstPromptWorkspaces = reactive(new Set()); */ const promptGenerationBySession = new Map(); const pendingLocalTurnStarts = new Map>(); -const afterLocalTurnsSettled = new Map void>(); +const afterLocalTurnsSettled = new Map void>>(); let nextLocalTurnToken = 0; /** @@ -134,6 +134,124 @@ function nextQueueEntryId(): string { return `${Date.now().toString(36)}-${queueEntryCounter}`; } +/** + * Per-session serialized operation queue for every queue-mutating prompt + * submission: per-entry steers, whole-queue steers, and automatic queue + * drains. Running them through one chain guarantees at most one submission + * per session is in flight at a time, so removals, restores, and drains + * always interleave in the order the operations were issued and the queue's + * FIFO order survives any interleaving of the three. + */ +const sessionOpQueue = new Map>(); + +interface SessionArchiveBarrier { + promise: Promise; + resolve: () => void; +} + +/** + * Archive starts before the daemon has confirmed whether the session was + * forgotten. Queue-mutating work must pause in that window: success invalidates + * its incarnation, while failure releases it against the still-live session. + */ +const sessionArchiveBarriers = new Map(); + +interface SessionOperation { + sid: string; + incarnation: number; +} + +/** + * A queue drain is reserved synchronously, before its serialized operation can + * start on the promise microtask queue. This closes the window where another + * send or turn-end callback could observe an idle session and schedule a second + * drain before submitPromptInternal marks the first prompt in flight. + * + * Store the owning operation, rather than a boolean, so a late completion from + * an archived incarnation cannot clear a reservation created after restore. + */ +const sessionQueueDrainReservations = new Map(); + +/** + * Monotonic session incarnation retained across teardown. A session id can be + * restored after archive, so presence alone cannot distinguish late work from + * the old instance. Never delete these markers: resetting to the default would + * let an old incarnation compare equal again after an archive/restore ABA. + */ +const sessionIncarnation = new Map(); +let nextSessionIncarnation = 0; + +function captureSessionOperation(sid: string): SessionOperation { + return { sid, incarnation: sessionIncarnation.get(sid) ?? 0 }; +} + +function hasCurrentIncarnation(operation: SessionOperation): boolean { + return operation.incarnation === (sessionIncarnation.get(operation.sid) ?? 0); +} + +function invalidateSessionOperations(sid: string): void { + nextSessionIncarnation += 1; + sessionIncarnation.set(sid, nextSessionIncarnation); +} + +function reserveSessionQueueDrain(operation: SessionOperation): boolean { + if (!hasCurrentIncarnation(operation) || sessionQueueDrainReservations.has(operation.sid)) { + return false; + } + sessionQueueDrainReservations.set(operation.sid, operation); + return true; +} + +function releaseSessionQueueDrain(operation: SessionOperation): void { + if (sessionQueueDrainReservations.get(operation.sid) === operation) { + sessionQueueDrainReservations.delete(operation.sid); + } +} + +function beginSessionArchive(sid: string): SessionArchiveBarrier | undefined { + if (sessionArchiveBarriers.has(sid)) return undefined; + let resolve!: () => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + const barrier = { promise, resolve }; + sessionArchiveBarriers.set(sid, barrier); + return barrier; +} + +function sessionArchiveBarrierPromise(sid: string): Promise | undefined { + return sessionArchiveBarriers.get(sid)?.promise; +} + +function endSessionArchive(sid: string, barrier: SessionArchiveBarrier): void { + if (sessionArchiveBarriers.get(sid) === barrier) { + sessionArchiveBarriers.delete(sid); + } + barrier.resolve(); +} + +/** Serialize `op` behind every previously queued operation for the session. + * Rejections are absorbed for SEQUENCING only — the chain keeps later ops + * running — while the caller still receives the op's original error. */ +function enqueueSessionOp( + operation: SessionOperation, + op: () => Promise, +): Promise { + const { sid } = operation; + const chain = sessionOpQueue.get(sid) ?? Promise.resolve(); + const run = chain.then(async () => { + await sessionArchiveBarrierPromise(sid); + if (!hasCurrentIncarnation(operation)) return; + await op(); + }); + const tail = run.catch(() => undefined); + sessionOpQueue.set(sid, tail); + void tail.finally(() => { + if (sessionOpQueue.get(sid) === tail) sessionOpQueue.delete(sid); + }); + return run; +} + export interface LocalTurnStartState { generation: number; pending: boolean; @@ -167,17 +285,33 @@ export function settleLocalTurn(sid: string, token: number): void { pending.delete(token); if (pending.size > 0) return; pendingLocalTurnStarts.delete(sid); - const callback = afterLocalTurnsSettled.get(sid); + const callbacks = afterLocalTurnsSettled.get(sid); afterLocalTurnsSettled.delete(sid); - callback?.(); + if (callbacks !== undefined) { + // Isolate each callback so one throwing registrant can't skip the rest + // (e.g. a deferred drain behind a snapshot retry). + for (const callback of callbacks) { + try { + callback(); + } catch { + // Registrants are fire-and-forget; a failure must not block the others. + } + } + } } /** Drop lifecycle state with the rest of a forgotten session. */ export function forgetLocalTurnState(sid: string): void { + invalidateSessionOperations(sid); promptGenerationBySession.delete(sid); pendingLocalTurnStarts.delete(sid); afterLocalTurnsSettled.delete(sid); queueFlushFailures.delete(sid); + sessionQueueDrainReservations.delete(sid); + // Safe to drop the chain itself: queued ops are no-ops under the bumped + // incarnation, and a fresh incarnation must never queue behind an op that + // may never settle (e.g. a wedged request from the archived instance). + sessionOpQueue.delete(sid); } /** Whether a snapshot request can still be applied without overwriting a @@ -186,13 +320,18 @@ export function isLocalTurnSnapshotCurrent(sid: string, atRequest: LocalTurnStar return !atRequest.pending && atRequest.generation === (promptGenerationBySession.get(sid) ?? 0); } -/** Coalesce a skipped snapshot into one retry after local turn-start requests settle. */ +/** Run `callback` once every in-flight local turn-start for the session has + * settled (immediately when none are pending). Multiple registrants — e.g. a + * deferred queue drain and a stale-snapshot retry landing in the same + * window — are all preserved and run in registration order. */ export function afterLocalTurnStartsSettle(sid: string, callback: () => void): void { if ((pendingLocalTurnStarts.get(sid)?.size ?? 0) === 0) { callback(); return; } - afterLocalTurnsSettled.set(sid, callback); + const callbacks = afterLocalTurnsSettled.get(sid) ?? new Set<() => void>(); + callbacks.add(callback); + afterLocalTurnsSettled.set(sid, callbacks); } type SyncSessionResult = 'ok' | 'not-found' | 'failed'; @@ -315,6 +454,13 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } = deps; let exportInFlight = false; + function isSessionOperationCurrent(operation: SessionOperation): boolean { + return ( + hasCurrentIncarnation(operation) && + rawState.sessions.some((session) => session.id === operation.sid) + ); + } + async function loadOlderMessages(sessionId: string): Promise { if (rawState.messagesLoadingMoreBySession[sessionId]) return; const current = rawState.messagesBySession[sessionId]; @@ -1456,14 +1602,26 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta sid: string, text: string, attachments?: PromptAttachment[], - ): Promise<'ok' | 'rejected' | 'uncertain'> { - // Mark this session as having a prompt in flight BEFORE any await, so a racing - // sendPrompt sees it and enqueues. Cleared when the main turn ends (or the - // prompt dies without one). beginLocalTurn also bumps the snapshot generation - // and marks the submit pending, so a racing terminal snapshot can't clear - // this prompt (see handleSessionSnapshot). + operation: SessionOperation = captureSessionOperation(sid), + ): Promise<'ok' | 'rejected' | 'uncertain' | 'cancelled'> { + // Mark this session as having a prompt in flight BEFORE any await — + // including the archive-barrier wait, so two sends racing an in-flight + // archive cannot both proceed as concurrent submissions when the archive + // fails. Cleared when the main turn ends (or the prompt dies without + // one). beginLocalTurn also bumps the snapshot generation and marks the + // submit pending, so a racing terminal snapshot can't clear this prompt + // (see handleSessionSnapshot). const localTurnToken = beginLocalTurn(sid); rawState.inFlightBySession = { ...rawState.inFlightBySession, [sid]: true }; + const archiveBarrier = sessionArchiveBarrierPromise(sid); + if (archiveBarrier !== undefined) await archiveBarrier; + if (!hasCurrentIncarnation(operation)) { + // An archive completed while we waited and forgot the session — release + // our own marks and cancel (the session's state is already discarded). + rawState.inFlightBySession = { ...rawState.inFlightBySession, [sid]: false }; + settleLocalTurn(sid, localTurnToken); + return 'cancelled'; + } const tempId = nextOptimisticMsgId(); try { const api = getKimiWebApi(); @@ -1518,7 +1676,9 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta if (goalMode && text) { try { await api.updateSession(sid, { goalObjective: text.trim() }); + if (!hasCurrentIncarnation(operation)) return 'cancelled'; } catch (err) { + if (!hasCurrentIncarnation(operation)) return 'cancelled'; pushOperationFailure('createGoal', err, { sessionId: sid }); rawState.inFlightBySession = { ...rawState.inFlightBySession, [sid]: false }; updateSessionMessages(sid, (msgs) => @@ -1528,6 +1688,11 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } } + const thinking = + (await modelProvider.resolveThinkingForPrompt(sid, model)) ?? rawState.thinking; + await sessionArchiveBarrierPromise(sid); + if (!hasCurrentIncarnation(operation)) return 'cancelled'; + const result = await api.submitPrompt(sid, { content, model, @@ -1537,11 +1702,12 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // tracks whatever session the user is looking at now: a queue drain for // a background session would otherwise submit the level of the session // the user switched to since enqueueing. - thinking: (await modelProvider.resolveThinkingForPrompt(sid, model)) ?? rawState.thinking, + thinking, permissionMode: rawState.permission, planMode, swarmMode, }); + if (!hasCurrentIncarnation(operation)) return 'cancelled'; // Goal mode is a one-shot flag: consumed by this send, then cleared. if (goalMode) { @@ -1579,6 +1745,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // daemon's auto-title, so we let the daemon own it. return 'ok'; } catch (err) { + if (!hasCurrentIncarnation(operation)) return 'cancelled'; // Submit failed — clear the in-flight flag so the next prompt isn't stuck // queued forever (turn.ended will never arrive), and roll back the // optimistic user message so the transcript doesn't show a delivered- @@ -1605,8 +1772,13 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // If the session is not idle OR a prompt is already in flight (submitted but // the WS turn.started hasn't arrived yet), enqueue instead of submitting // directly. The in-flight flag closes the window where two rapid prompts - // would both submit and race. - if (activity.value !== 'idle' || rawState.inFlightBySession[sid]) { + // would both submit and race. A reserved queue drain (scheduled but not yet + // submitting, e.g. parked behind a pending steer) counts as busy too. + if ( + activity.value !== 'idle' || + rawState.inFlightBySession[sid] || + sessionQueueDrainReservations.has(sid) + ) { enqueue(text, attachments); return; } @@ -1617,8 +1789,14 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // enqueue this prompt behind them and flush the head now — the flush // re-arms the in-flight flag, and each later turn end drains the next // entry. Submitting directly here would jump the queue AND leave the - // stuck entries without a flush driver. - if ((rawState.queuedBySession[sid]?.length ?? 0) > 0) { + // stuck entries without a flush driver. A pending local turn-start (e.g. + // a steer submit) takes the same path: the serialized drain submits + // behind it instead of racing a second concurrent POST, which could start + // its turn first and pull the older steered text into the wrong turn. + if ( + (rawState.queuedBySession[sid]?.length ?? 0) > 0 || + localTurnStartState(sid).pending + ) { enqueue(text, attachments); flushQueueHead(sid); return; @@ -1637,40 +1815,129 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta async function steerPrompt(text: string, attachments?: PromptAttachment[]): Promise { const sid = rawState.activeSessionId; if (!sid) return; - - // Merge queued texts (oldest first) + the live text, like the TUI does. - const queue = rawState.queuedBySession[sid] ?? []; - const parts: string[] = []; - const mergedAttachments: PromptAttachment[] = []; - for (const q of queue) { - const trimmed = q.text.trim(); - if (trimmed) parts.push(trimmed); - if (q.attachments?.length) mergedAttachments.push(...q.attachments); - } + const operation = captureSessionOperation(sid); const live = text.trim(); - if (live) parts.push(live); - if (attachments?.length) mergedAttachments.push(...attachments); - if (parts.length === 0 && mergedAttachments.length === 0) return; - if (queue.length > 0) { - rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: [] }; - } - const merged = parts.join('\n\n'); - - // Put back every entry that was merged into this steer when its submit - // fails, so the queued prompts aren't silently lost. Entries enqueued - // while the submit was in flight stay behind them. - const restoreQueue = (): void => { - if (queue.length === 0) return; + + // The merge runs inside the session's op queue (serialized with per-entry + // steers and drains), so the queue is read and cleared at execution time. + await enqueueSessionOp(operation, async () => { + if (!isSessionOperationCurrent(operation)) return; + // Merge queued texts (oldest first) + the live text, like the TUI does. + const queue = rawState.queuedBySession[sid] ?? []; + const parts: string[] = []; + const mergedAttachments: PromptAttachment[] = []; + for (const q of queue) { + const trimmed = q.text.trim(); + if (trimmed) parts.push(trimmed); + if (q.attachments?.length) mergedAttachments.push(...q.attachments); + } + if (live) parts.push(live); + if (attachments?.length) mergedAttachments.push(...attachments); + if (parts.length === 0 && mergedAttachments.length === 0) return; + if (queue.length > 0) { + rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: [] }; + } + const merged = parts.join('\n\n'); + + // Put back every entry that was merged into this steer when its submit + // fails, so the queued prompts aren't silently lost. Entries enqueued + // while the submit was in flight stay behind them. + const restoreQueue = (): void => { + if (queue.length === 0) return; + const current = rawState.queuedBySession[sid] ?? []; + rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: [...queue, ...current] }; + }; + + await steerContentInternal(sid, merged, mergedAttachments, restoreQueue, operation); + }).catch((err: unknown) => { + // steerContentInternal handles its own failures; surface an unexpected + // op-queue error instead of an unhandled rejection. + pushOperationFailure('steer', err, { sessionId: sid }); + }); + } + + /** + * steerQueuedPrompt() — steer a SINGLE queued entry into the running turn, + * leaving the rest of the queue untouched. Same submit+steer flow (and the + * same restore-on-rejection rule) as steerPrompt, scoped to one entry. + * Serialized with every other queue-mutating operation (see sessionOpQueue); + * the removal re-resolves the clicked entry by identity at execution time + * because reorders and restores may have moved it since the click. + */ + async function steerQueuedPrompt(index: number): Promise { + const sid = rawState.activeSessionId; + if (!sid) return; + const operation = captureSessionOperation(sid); + const entry = (rawState.queuedBySession[sid] ?? [])[index]; + if (!entry) return; + const text = entry.text.trim(); + const entryAttachments = entry.attachments ?? []; + if (!text && entryAttachments.length === 0) return; + + await enqueueSessionOp(operation, async () => { + if (!isSessionOperationCurrent(operation)) return; const current = rawState.queuedBySession[sid] ?? []; - rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: [...queue, ...current] }; - }; + const currentIndex = current.indexOf(entry); + // Edited, unqueued, or drained while an earlier op was in flight. + if (currentIndex === -1) return; + rawState.queuedBySession = { + ...rawState.queuedBySession, + [sid]: current.filter((_, i) => i !== currentIndex), + }; + + // Operations are serialized, so the removal slot is stable with respect + // to other steers. If the user reorders the visible queue while the + // request is pending, put the rejected entry back into that same slot in + // the latest order instead of anchoring it ahead of a moved successor. + const restoreQueue = (): void => { + const latest = rawState.queuedBySession[sid] ?? []; + if (latest.includes(entry)) return; + const next = [...latest]; + next.splice(Math.min(currentIndex, next.length), 0, entry); + rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: next }; + }; + await steerContentInternal(sid, text, entryAttachments, restoreQueue, operation); + }).catch((err: unknown) => { + // steerContentInternal handles its own failures; surface an unexpected + // op-queue error instead of an unhandled rejection. + pushOperationFailure('steer', err, { sessionId: sid }); + }); + } + + /** + * Shared steer tail: idle fallback, optimistic echo, submit (parks behind the + * active prompt) then POST /prompts:steer. `restoreQueue` undoes the caller's + * queue removal on a definitive daemon rejection. + */ + async function steerContentInternal( + sid: string, + merged: string, + mergedAttachments: PromptAttachment[], + restoreQueue: () => void, + operation: SessionOperation, + ): Promise { + if (!isSessionOperationCurrent(operation)) return; + // Same guard as the queue drain: if the session was forgotten (e.g. + // archived) while the submit was pending, its queue was already discarded + // — restoring would resurrect a stale message that could drain later. + const restoreIfSessionKnown = (): void => { + if (rawState.sessions.some((s) => s.id === sid)) restoreQueue(); + }; + // Liveness of THIS session's main conversation — not the active view's + // `activity`, which follows whichever session the user has selected (a + // serialized per-entry steer can fire after a session switch). + const turnRunning = + (rawState.turnActiveBySession[sid] ?? false) || + rawState.inFlightBySession[sid] === true || + (rawState.sessions.find((session) => session.id === sid)?.mainTurnActive ?? false); // Idle and nothing in flight — there is no turn to steer into; normal send. - if (activity.value === 'idle' && !rawState.inFlightBySession[sid]) { - const outcome = await submitPromptInternal(sid, merged, mergedAttachments); + if (!turnRunning) { + const outcome = await submitPromptInternal(sid, merged, mergedAttachments, operation); + if (!isSessionOperationCurrent(operation)) return; // Same never-duplicate rule as the running-path catch below: restore // the merged entries only on a definitive rejection. - if (outcome === 'rejected') restoreQueue(); + if (outcome === 'rejected') restoreIfSessionKnown(); return; } @@ -1708,21 +1975,28 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta (promptSession?.model && promptSession.model.length > 0 ? promptSession.model : rawState.defaultModel) ?? undefined; + const thinking = + (await modelProvider.resolveThinkingForPrompt(sid, model)) ?? rawState.thinking; + await sessionArchiveBarrierPromise(sid); + if (!isSessionOperationCurrent(operation)) return; + const result = await api.submitPrompt(sid, { content, model, // Resolved against this prompt's own session + model, same as a normal // send (see submitPromptInternal). - thinking: (await modelProvider.resolveThinkingForPrompt(sid, model)) ?? rawState.thinking, + thinking, permissionMode: rawState.permission, planMode: rawState.planModeBySession[sid] ?? false, swarmMode: rawState.swarmModeBySession[sid] ?? false, }); + await sessionArchiveBarrierPromise(sid); + if (!isSessionOperationCurrent(operation)) return; - // Stamp the real prompt_id onto the optimistic echo. Unlike a normal send, - // a steered prompt IS echoed back by the daemon as a messageCreated user - // event; matching that echo by prompt_id (instead of content) is what keeps - // an image steer from rendering two user bubbles. + // Stamp the real prompt_id onto the optimistic echo. kap-server emits + // no messageCreated event for a steered prompt, but the stamp lets any + // later snapshot or resync match the persisted message by id instead of + // falling back to looser content heuristics. updateSessionMessages(sid, (msgs) => { const idx = msgs.findIndex((m) => m.id === tempId); if (idx === -1) return msgs; @@ -1746,6 +2020,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // the parked prompt as its own turn. Nothing to roll back. } } catch (err) { + if (!isSessionOperationCurrent(operation)) return; // Submit failed: drop the optimistic echo so the transcript doesn't show // a delivered-looking message the daemon never received. updateSessionMessages(sid, (msgs) => msgs.filter((m) => m.id !== tempId)); @@ -1755,7 +2030,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // already be queued server-side; re-queueing the originals would // duplicate it (the exact ghost-send behavior this change exists to // prevent). The failure toast below tells the user what happened. - if (isDaemonApiError(err)) restoreQueue(); + if (isDaemonApiError(err)) restoreIfSessionKnown(); pushOperationFailure('steer', err, { sessionId: sid }); } finally { settleLocalTurn(sid, localTurnToken); @@ -1801,54 +2076,93 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta * queue. The budget is tracked PER ENTRY (by id), so removing or * reordering the head resets it for the next entry. Every failure is * already surfaced via pushOperationFailure. + * + * The submission runs inside the session's op queue (see sessionOpQueue), + * serialized with steers and other drains, so a drain can never submit + * concurrently with a steer that is still in flight. */ function flushQueueHead(sid: string): void { + const operation = captureSessionOperation(sid); + flushQueueHeadForOperation(operation); + } + + function flushQueueHeadForOperation(operation: SessionOperation): void { + if ((rawState.queuedBySession[operation.sid]?.length ?? 0) === 0) return; + if (!reserveSessionQueueDrain(operation)) return; + + const enqueueReservedDrain = () => { + // submitPromptInternal reports its own failures; this catch is the + // last-resort surface for an unexpected error inside the drain itself, + // now that op-queue rejections propagate to the caller. + void enqueueSessionOp(operation, () => flushQueueHeadNow(operation)) + .catch((err: unknown) => { + if (!isSessionOperationCurrent(operation)) return; + pushOperationFailure('queueFlush', err, { sessionId: operation.sid }); + }) + .finally(() => { + releaseSessionQueueDrain(operation); + }); + }; + + // Reserve before deferring behind a local turn start. Otherwise a same-tick + // send can schedule another drain while this one is waiting to enter the op + // queue. + if (localTurnStartState(operation.sid).pending) { + afterLocalTurnStartsSettle(operation.sid, enqueueReservedDrain); + } else { + enqueueReservedDrain(); + } + } + + async function flushQueueHeadNow(operation: SessionOperation): Promise { + if (!isSessionOperationCurrent(operation)) return; + const { sid } = operation; const [next, ...rest] = rawState.queuedBySession[sid] ?? []; if (next === undefined) return; rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: rest }; - void submitPromptInternal(sid, next.text, next.attachments).then((outcome) => { - if (outcome === 'ok') { - queueFlushFailures.delete(sid); - return; - } - // Ambiguous failure: the daemon may have accepted the prompt and lost - // the response — the entry is dropped (the failure was already toasted) - // rather than re-queued and possibly submitted twice. - if (outcome === 'uncertain') { - queueFlushFailures.delete(sid); - return; - } - // Definitively rejected below this point. If the session was forgotten - // (e.g. archived) while the submit was pending, its queue was already - // discarded — don't resurrect it. - if (!rawState.sessions.some((s) => s.id === sid)) { - queueFlushFailures.delete(sid); - return; - } - // Per-entry budget: a different head (removed/reordered since) starts - // fresh instead of inheriting the previous entry's failures. - const key = next.id ?? next.text; - const previous = queueFlushFailures.get(sid); - const count = previous !== undefined && previous.key === key ? previous.count + 1 : 1; - if (count >= MAX_QUEUE_FLUSH_FAILURES) { - queueFlushFailures.delete(sid); - // Advance the queue instead of stranding the entries behind the - // dropped head: the failed submit produced no turn, so nothing else - // will drive the next entry until the user sends again. The new head - // carries its own budget, so a poisoned successor just goes through - // the same retry cycle. - if ((rawState.queuedBySession[sid]?.length ?? 0) > 0) { - flushQueueHead(sid); - } - return; + const outcome = await submitPromptInternal(sid, next.text, next.attachments, operation); + if (!isSessionOperationCurrent(operation) || outcome === 'cancelled') return; + if (outcome === 'ok') { + queueFlushFailures.delete(sid); + return; + } + // Ambiguous failure: the daemon may have accepted the prompt and lost + // the response — the entry is dropped (the failure was already toasted) + // rather than re-queued and possibly submitted twice. + if (outcome === 'uncertain') { + queueFlushFailures.delete(sid); + return; + } + // Definitively rejected below this point. If the session was forgotten + // (e.g. archived) while the submit was pending, its queue was already + // discarded — don't resurrect it. + if (!rawState.sessions.some((s) => s.id === sid)) { + queueFlushFailures.delete(sid); + return; + } + // Per-entry budget: a different head (removed/reordered since) starts + // fresh instead of inheriting the previous entry's failures. + const key = next.id ?? next.text; + const previous = queueFlushFailures.get(sid); + const count = previous !== undefined && previous.key === key ? previous.count + 1 : 1; + if (count >= MAX_QUEUE_FLUSH_FAILURES) { + queueFlushFailures.delete(sid); + // Advance the queue instead of stranding the entries behind the + // dropped head: the failed submit produced no turn, so nothing else + // will drive the next entry until the user sends again. The new head + // carries its own budget, so a poisoned successor just goes through + // the same retry cycle. + if ((rawState.queuedBySession[sid]?.length ?? 0) > 0) { + await flushQueueHeadNow(operation); } - queueFlushFailures.set(sid, { key, count }); - const current = rawState.queuedBySession[sid] ?? []; - rawState.queuedBySession = { - ...rawState.queuedBySession, - [sid]: [next, ...current], - }; - }); + return; + } + queueFlushFailures.set(sid, { key, count }); + const current = rawState.queuedBySession[sid] ?? []; + rawState.queuedBySession = { + ...rawState.queuedBySession, + [sid]: [next, ...current], + }; } /** @@ -1888,7 +2202,8 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const mayDrain = wasInFlight || opts?.turnWasActive === true || (rawState.turnActiveBySession[sid] ?? false); if (mayDrain) { - flushQueueHead(sid); + const operation = captureSessionOperation(sid); + flushQueueHeadForOperation(operation); } return wasInFlight; @@ -2369,6 +2684,16 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta /** Archive a session — calls API, persists the archive flag, removes locally, picks another active session or none */ async function archiveSession(id: string): Promise { + // Another archive for this session is already in flight: wait for it, + // then run our own attempt — a second click is a retry intent, not a + // dedupe, when the first attempt failed. + const existing = sessionArchiveBarrierPromise(id); + if (existing !== undefined) await existing; + // The first attempt may already have archived (and forgotten) the + // session — retrying would re-POST against an archived session. + if (!rawState.sessions.some((s) => s.id === id)) return; + const barrier = beginSessionArchive(id); + if (barrier === undefined) return; // a concurrent caller beat us to it try { const api = getKimiWebApi(); await api.archiveSession(id); @@ -2391,6 +2716,8 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } } catch (err) { pushOperationFailure('archiveSession', err, { sessionId: id }); + } finally { + endSessionArchive(id, barrier); } } @@ -2786,6 +3113,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta handleSessionSnapshot, sendPrompt, steerPrompt, + steerQueuedPrompt, uploadImage, enqueue, unqueue, diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index f6e3f6a694..03df7c8abd 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -2873,6 +2873,7 @@ export function useKimiWebClient() { sendPrompt: workspaceState.sendPrompt, steerPrompt: workspaceState.steerPrompt, + steerQueuedPrompt: workspaceState.steerQueuedPrompt, // Side chat (BTW side-channel agent) sideChatVisible: sideChat.sideChatVisible, sideChatSessionId: sideChat.sideChatSessionId, diff --git a/apps/kimi-web/src/i18n/locales/en/composer.ts b/apps/kimi-web/src/i18n/locales/en/composer.ts index 5f566c6729..6d4b237fcb 100644 --- a/apps/kimi-web/src/i18n/locales/en/composer.ts +++ b/apps/kimi-web/src/i18n/locales/en/composer.ts @@ -6,6 +6,9 @@ export default { starting: 'Sending…', queueAutoDrain: 'sends automatically when the current turn ends', queueNext: 'Up next', + queueSteerNow: 'Steer now', + queueSteerTitle: 'Inject the queued message(s) into the running turn (Ctrl+S)', + queueSteerOneTitle: 'Inject this message into the running turn', queueDragTitle: 'Drag to reorder', editQueued: 'Edit (load back into the input)', queuedAttachments: 'attachment ×{n}', diff --git a/apps/kimi-web/src/i18n/locales/zh/composer.ts b/apps/kimi-web/src/i18n/locales/zh/composer.ts index a0ef9f952b..85156de2af 100644 --- a/apps/kimi-web/src/i18n/locales/zh/composer.ts +++ b/apps/kimi-web/src/i18n/locales/zh/composer.ts @@ -6,6 +6,9 @@ export default { starting: '正在发送…', queueAutoDrain: '当前回合结束后自动逐条发送', queueNext: '下一条', + queueSteerNow: '立即插入', + queueSteerTitle: '把排队的消息立即插入运行中的回合（Ctrl+S）', + queueSteerOneTitle: '把这条消息立即插入运行中的回合', queueDragTitle: '拖拽排序', editQueued: '编辑（载入到输入框）', queuedAttachments: '附件 ×{n}', diff --git a/apps/kimi-web/test/workspace-state.test.ts b/apps/kimi-web/test/workspace-state.test.ts index d8871d9f8b..a1248cb27c 100644 --- a/apps/kimi-web/test/workspace-state.test.ts +++ b/apps/kimi-web/test/workspace-state.test.ts @@ -18,11 +18,14 @@ const apiMock = vi.hoisted(() => ({ abortPrompt: vi.fn(), abortSession: vi.fn(), addWorkspace: vi.fn(), + archiveSession: vi.fn(), updateWorkspace: vi.fn(), createSession: vi.fn(), exportSession: vi.fn(), + restoreSession: vi.fn(), updateSession: vi.fn(), submitPrompt: vi.fn(), + steerPrompts: vi.fn(), respondQuestion: vi.fn(), respondApproval: vi.fn(), dismissQuestion: vi.fn(), @@ -1576,6 +1579,12 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { beforeEach(() => { apiMock.submitPrompt.mockReset(); apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_new' }); + apiMock.steerPrompts.mockReset(); + apiMock.steerPrompts.mockResolvedValue({ steered: true, promptIds: ['prompt_new'] }); + apiMock.archiveSession.mockReset(); + apiMock.archiveSession.mockResolvedValue(undefined); + apiMock.restoreSession.mockReset(); + apiMock.restoreSession.mockResolvedValue(createSession()); // Module-level flush failure budget must not leak between tests. forgetLocalTurnState('sess_1'); }); @@ -1671,6 +1680,31 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { ]); }); + it('drains only one queued prompt when a send follows a turn finish in the same tick', async () => { + const state = createState(); + state.queuedBySession = { + sess_1: [ + { text: 'first queued', attachments: undefined }, + { text: 'second queued', attachments: undefined }, + ], + }; + const ws = useWorkspaceState( + state, + promptDeps({ activity: computed(() => 'idle') }), + ); + + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + const send = ws.sendPrompt('new prompt'); + await send; + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(state.queuedBySession.sess_1?.map((entry) => entry.text)).toEqual([ + 'second queued', + 'new prompt', + ]); + }); + it('flushes the stuck queue head before the new prompt when sending while idle', async () => { const state = createState(); state.queuedBySession = { sess_1: [{ text: 'stuck queued', attachments: undefined }] }; @@ -1678,7 +1712,8 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { await ws.sendPrompt('next'); - expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + // The flush runs inside the session's op queue — one microtask behind. + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledOnce()); expect(apiMock.submitPrompt).toHaveBeenCalledWith( 'sess_1', expect.objectContaining({ content: [{ type: 'text', text: 'stuck queued' }] }), @@ -1688,6 +1723,28 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { ]); }); + it('drains only the stuck queue head when two idle sends happen in the same tick', async () => { + const state = createState(); + state.queuedBySession = { + sess_1: [{ text: 'stuck queued', attachments: undefined }], + }; + const ws = useWorkspaceState( + state, + promptDeps({ activity: computed(() => 'idle') }), + ); + + const firstSend = ws.sendPrompt('first new prompt'); + const secondSend = ws.sendPrompt('second new prompt'); + await Promise.all([firstSend, secondSend]); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(state.queuedBySession.sess_1?.map((entry) => entry.text)).toEqual([ + 'first new prompt', + 'second new prompt', + ]); + }); + it('re-queues a failed flush at the head and drops it after repeated failures', async () => { const state = createState(); state.queuedBySession = { sess_1: [{ text: 'first queued', attachments: undefined }] }; @@ -1733,6 +1790,75 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { ); }); + it('steers whole-queue text and attachments in visible order before live composer content', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'first', attachments: [{ fileId: 'img_q', kind: 'image' }] }, + { text: ' ', attachments: [{ fileId: 'video_q', kind: 'video' }] }, + { + text: 'third', + attachments: [{ + fileId: 'file_q', + kind: 'file', + name: 'notes.txt', + mediaType: 'text/plain', + size: 12, + }], + }, + ], + }; + apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_merged', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps()); + + await ws.steerPrompt('live', [{ fileId: 'img_live', kind: 'image' }]); + + expect(apiMock.submitPrompt).toHaveBeenCalledWith( + 'sess_1', + expect.objectContaining({ + content: [ + { type: 'text', text: 'first\n\nthird\n\nlive' }, + { type: 'image', source: { kind: 'file', fileId: 'img_q' } }, + { type: 'video', source: { kind: 'file', fileId: 'video_q' } }, + { + type: 'file', + fileId: 'file_q', + name: 'notes.txt', + mediaType: 'text/plain', + size: 12, + }, + { type: 'image', source: { kind: 'file', fileId: 'img_live' } }, + ], + }), + ); + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_merged']); + expect(state.queuedBySession.sess_1).toEqual([]); + }); + + it('normal-sends the whole queue when no main turn is running', async () => { + const state = createState(); + state.sessions = [{ ...createSession(), busy: false, mainTurnActive: false }]; + state.queuedBySession = { + sess_1: [ + { text: 'first', attachments: undefined }, + { text: 'second', attachments: undefined }, + ], + }; + const ws = useWorkspaceState(state, promptDeps({ activity: computed(() => 'idle') })); + + await ws.steerPrompt(''); + + expect(apiMock.submitPrompt).toHaveBeenCalledWith( + 'sess_1', + expect.objectContaining({ + content: [{ type: 'text', text: 'first\n\nsecond' }], + }), + ); + expect(apiMock.steerPrompts).not.toHaveBeenCalled(); + expect(state.queuedBySession.sess_1).toEqual([]); + }); + it('does NOT restore merged queue entries when a steer failure is network-ambiguous', async () => { const state = createState(); state.inFlightBySession = { sess_1: true }; @@ -1762,6 +1888,671 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { expect(state.queuedBySession.sess_1).toEqual([{ text: 'queued', attachments: undefined }]); }); + it('steers only the selected queued entry and leaves the rest of the queue', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'first', attachments: undefined }, + { text: 'second', attachments: [{ fileId: 'f_2', kind: 'image' }] }, + { text: 'third', attachments: undefined }, + ], + }; + apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_new', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps()); + + await ws.steerQueuedPrompt(1); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(apiMock.submitPrompt).toHaveBeenCalledWith( + 'sess_1', + expect.objectContaining({ + content: [ + { type: 'text', text: 'second' }, + { type: 'image', source: { kind: 'file', fileId: 'f_2' } }, + ], + }), + ); + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_new']); + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'first', attachments: undefined }, + { text: 'third', attachments: undefined }, + ]); + }); + + it('restores a single-steered entry at its original position on a definitive rejection', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'first', attachments: undefined }, + { text: 'second', attachments: undefined }, + { text: 'third', attachments: undefined }, + ], + }; + apiMock.submitPrompt.mockRejectedValue( + new DaemonApiError({ code: 50000, msg: 'boom', requestId: 'r' }), + ); + const ws = useWorkspaceState(state, promptDeps()); + + await ws.steerQueuedPrompt(1); + + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'first', attachments: undefined }, + { text: 'second', attachments: undefined }, + { text: 'third', attachments: undefined }, + ]); + }); + + it('keeps the latest user order when a pending row steer is rejected', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + { text: 'C', attachments: undefined }, + ], + }; + let rejectSubmit!: (err: Error) => void; + apiMock.submitPrompt.mockImplementation( + () => + new Promise<{ promptId: string }>((_resolve, reject) => { + rejectSubmit = reject; + }), + ); + const ws = useWorkspaceState(state, promptDeps()); + + const steer = ws.steerQueuedPrompt(1); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledOnce()); + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'A', attachments: undefined }, + { text: 'C', attachments: undefined }, + ]); + + // The user moves C to the front while B's request is pending. + ws.reorderQueue(1, 0); + rejectSubmit(new DaemonApiError({ code: 50000, msg: 'boom', requestId: 'r' })); + await steer; + + // C remains first; the rejected B returns to its prior numeric slot. + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'C', attachments: undefined }, + { text: 'B', attachments: undefined }, + { text: 'A', attachments: undefined }, + ]); + }); + + // A turn that ends while a single-steered entry's submit is still awaiting + // the daemon must not drain the remaining queue head concurrently — the + // drained prompt could overtake the parked steer server-side and the steered + // message would land in the wrong turn. The drain waits for the steer to + // settle, then submits the head behind it. + it('defers the queue drain until an in-flight steer submission settles', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'first', attachments: undefined }, + { text: 'second', attachments: undefined }, + ], + }; + let resolveSubmit!: (value: { promptId: string; status: string }) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => new Promise((resolve) => { resolveSubmit = resolve; }), + ) + .mockResolvedValue({ promptId: 'prompt_next', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps()); + + const steerPromise = ws.steerQueuedPrompt(1); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + + // The turn ends mid-submit: the drain is deferred, no second submission. + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1); + + resolveSubmit({ promptId: 'prompt_new', status: 'queued' }); + await steerPromise; + + // Once the steer settles, the deferred drain submits the remaining head. + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(2)); + expect(apiMock.submitPrompt).toHaveBeenLastCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'first' }] }), + ); + }); + + // The exact B/C scenario from review: row-steer B out of [A, B, C], the + // turn ends mid-submit, and the drain must wait for the steer to settle + // before submitting head A — leaving C queued and FIFO intact. + it('steers B and drains head A only after the steer settles, keeping C queued', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + { text: 'C', attachments: undefined }, + ], + }; + let resolveSteer!: (value: { promptId: string; status: string }) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => { + return new Promise((resolve) => { resolveSteer = resolve; }); + }, + ) + .mockResolvedValue({ promptId: 'prompt_next', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps()); + + const steer = ws.steerQueuedPrompt(1); // row-steer B + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + + // The turn ends while B's submit is in flight: the drain is deferred. + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1); + + resolveSteer({ promptId: 'prompt_b', status: 'queued' }); + await steer; + + // B was steered; only then does the drain submit head A. C stays queued. + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_b']); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(2)); + expect(apiMock.submitPrompt).toHaveBeenLastCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'A' }] }), + ); + expect(state.queuedBySession.sess_1).toEqual([{ text: 'C', attachments: undefined }]); + }); + + // The full race the op queue exists to prevent: TWO row steers queued + // before the turn ends, plus the deferred drain. Submissions must come out + // in click order — B, then C (normal-sent: the turn has ended by the time + // its op runs), then the drained head A — never concurrently. + it('serializes two row steers and the turn-end drain in click order', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + { text: 'C', attachments: undefined }, + ], + }; + let resolveSteer!: (value: { promptId: string; status: string }) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => { + return new Promise((resolve) => { resolveSteer = resolve; }); + }, + ) + .mockResolvedValue({ promptId: 'prompt_next', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps()); + + const steerB = ws.steerQueuedPrompt(1); // row-steer B → queue [A, C] + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + const steerC = ws.steerQueuedPrompt(1); // row-steer C (now at index 1) + + // The turn ends while B's submit is in flight: the drain is deferred + // behind both steers. The macrotask flush keeps this assertion honest — + // an undeferred drain would have submitted by then. + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1); + + resolveSteer({ promptId: 'prompt_b', status: 'queued' }); + await steerB; + await steerC; + + // B was steered; C's op ran after the turn ended so it normal-sent; the + // deferred drain submitted head A last. Exactly one op at a time. + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(3)); + expect(apiMock.steerPrompts).toHaveBeenCalledTimes(1); + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_b']); + const submittedTexts = apiMock.submitPrompt.mock.calls.map( + (call) => (call[1] as { content: { type: string; text?: string }[] }).content[0]?.text, + ); + expect(submittedTexts).toEqual(['B', 'C', 'A']); + expect(state.queuedBySession.sess_1).toEqual([]); + }); + + // The turn ends while a steer submit is still in flight and the user then + // sends a fresh message: it must enqueue and let the serialized drain + // submit it BEHIND the steer, never race a second concurrent POST that + // could start its turn first and pull the older steered text into it. + it('enqueues a send behind an in-flight steer submit instead of racing it', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + let resolveSteer!: (value: { promptId: string; status: string }) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => { + return new Promise((resolve) => { resolveSteer = resolve; }); + }, + ) + .mockResolvedValue({ promptId: 'prompt_next', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps({ activity: computed(() => 'idle') })); + + const steer = ws.steerPrompt('S'); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + const send = ws.sendPrompt('M'); + await send; + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1); + expect(state.queuedBySession.sess_1?.map((entry) => entry.text)).toEqual(['M']); + + resolveSteer({ promptId: 'prompt_s', status: 'queued' }); + await steer; + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(2); + expect(apiMock.submitPrompt).toHaveBeenLastCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'M' }] }), + ); + }); + + // The drain deferral is the only guard when the pending turn-start bypasses + // the op queue (a direct idle send): without it the drain would submit + // concurrently with that in-flight POST. + it('defers the drain behind a direct send submit that bypasses the op queue', async () => { + const state = createState(); + let resolveSend!: (value: { promptId: string }) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => { + return new Promise((resolve) => { resolveSend = resolve; }); + }, + ) + .mockResolvedValue({ promptId: 'prompt_next', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps({ activity: computed(() => 'idle') })); + + const send = ws.sendPrompt('X'); // direct submit, still in flight + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + + // A queued entry appears and the turn ends while X's POST is in flight. + state.queuedBySession = { sess_1: [{ text: 'Y', attachments: undefined }] }; + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1); + + resolveSend({ promptId: 'prompt_x' }); + await send; + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(2)); + expect(apiMock.submitPrompt).toHaveBeenLastCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'Y' }] }), + ); + }); + + // A deferred drain and a stale-snapshot retry can land in the same pending + // window — both callbacks must survive registration, not overwrite each + // other (single-slot storage used to drop the earlier one). + it('runs every afterLocalTurnStartsSettle callback, not just the last one', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [{ text: 'entry', attachments: undefined }], + }; + let resolveSubmit!: (value: { promptId: string; status: string }) => void; + apiMock.submitPrompt.mockImplementation( + () => new Promise((resolve) => { resolveSubmit = resolve; }), + ); + const ws = useWorkspaceState(state, promptDeps()); + + const steerPromise = ws.steerQueuedPrompt(0); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + + const calls: string[] = []; + ws.afterLocalTurnStartsSettle('sess_1', () => calls.push('first')); + ws.afterLocalTurnStartsSettle('sess_1', () => calls.push('second')); + + resolveSubmit({ promptId: 'prompt_new', status: 'queued' }); + await steerPromise; + + expect(calls).toEqual(['first', 'second']); + }); + + // Two rapid per-row steers whose submits are both definitively rejected + // must restore their entries without corrupting FIFO order: removals and + // restores interleave in click order because per-entry steers are + // serialized per session, and each entry returns to its removal slot in + // the latest visible order. + it('preserves queue order when two concurrent per-entry steers are both rejected', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + { text: 'C', attachments: undefined }, + ], + }; + apiMock.submitPrompt.mockRejectedValue( + new DaemonApiError({ code: 50000, msg: 'boom', requestId: 'r' }), + ); + const ws = useWorkspaceState(state, promptDeps()); + + await Promise.all([ws.steerQueuedPrompt(0), ws.steerQueuedPrompt(1)]); + + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + { text: 'C', attachments: undefined }, + ]); + }); + + // Steer liveness must be judged from the captured session's own state: a + // serialized per-entry steer can fire after the user switched to an idle + // session, and the active view's `activity` would wrongly report idle while + // the captured session's turn (observed from another client) still runs. + it('judges steer liveness from the captured session, not the active view', async () => { + const state = createState(); + // Turn observed from another client: turnActive, not locally in flight. + state.turnActiveBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [{ text: 'entry', attachments: undefined }], + }; + apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_new', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps({ activity: computed(() => 'idle') })); + + await ws.steerQueuedPrompt(0); + + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_new']); + expect(state.queuedBySession.sess_1).toEqual([]); + }); + + // Row steers, whole-queue steers, and drains all share one per-session op + // queue: a whole-queue steer clicked while a per-entry steer is in flight + // must not submit concurrently — it merges and submits only after the + // first op settles. + it('serializes a whole-queue steer behind an in-flight per-entry steer', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + ], + }; + let resolveFirst!: (value: { promptId: string; status: string }) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => { + return new Promise((resolve) => { resolveFirst = resolve; }); + }, + ) + .mockResolvedValueOnce({ promptId: 'prompt_merged', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps()); + + const row = ws.steerQueuedPrompt(0); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + const all = ws.steerPrompt(''); + // The whole-queue steer is queued behind the in-flight one: nothing else + // submits (and the queue is not cleared) until the first op settles. A + // macrotask flush makes this assertion meaningful — a concurrent op would + // have reached the API by then. + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1); + expect(state.queuedBySession.sess_1).toEqual([{ text: 'B', attachments: undefined }]); + + resolveFirst({ promptId: 'prompt_a', status: 'queued' }); + await row; + await all; + + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(2); + expect(apiMock.submitPrompt).toHaveBeenLastCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'B' }] }), + ); + expect(apiMock.steerPrompts).toHaveBeenNthCalledWith(1, 'sess_1', ['prompt_a']); + expect(apiMock.steerPrompts).toHaveBeenNthCalledWith(2, 'sess_1', ['prompt_merged']); + }); + + // Same rule as the queue drain (see 'does not resurrect the queue …' + // below): a definitive rejection after the session was forgotten must not + // recreate its discarded queue. + it('does not resurrect the queue when a steer is rejected after the session was forgotten', async () => { + let rejectSubmit!: (err: Error) => void; + apiMock.submitPrompt.mockImplementation( + () => { + return new Promise<{ promptId: string }>((_resolve, reject) => { + rejectSubmit = reject; + }); + }, + ); + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [{ text: 'doomed', attachments: undefined }], + }; + const ws = useWorkspaceState(state, promptDeps()); + + const steer = ws.steerQueuedPrompt(0); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalled()); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + + // Facade forget path (e.g. archive) while the steer submit is pending. + state.sessions = []; + delete state.queuedBySession.sess_1; + rejectSubmit(new DaemonApiError({ code: 50000, msg: 'gone', requestId: 'r' })); + await steer; + + expect(state.queuedBySession.sess_1).toBeUndefined(); + }); + + it('invalidates pending steers when archive and restore reuse the session id', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + ], + }; + let rejectSubmit!: (err: Error) => void; + apiMock.submitPrompt.mockImplementation( + () => + new Promise<{ promptId: string }>((_resolve, reject) => { + rejectSubmit = reject; + }), + ); + const deps = promptDeps({ + sideChat: { + clearSideChatForSession: vi.fn(), + } as unknown as UseWorkspaceStateDeps['sideChat'], + forgetSession: vi.fn((id: string) => { + forgetLocalTurnState(id); + state.sessions = state.sessions.filter((session) => session.id !== id); + delete state.queuedBySession[id]; + delete state.inFlightBySession[id]; + delete state.turnActiveBySession[id]; + }), + upsertSessionFront: vi.fn((session: AppSession) => { + state.sessions = [ + session, + ...state.sessions.filter((existing) => existing.id !== session.id), + ]; + }), + }); + const ws = useWorkspaceState(state, deps); + + const rowSteer = ws.steerQueuedPrompt(0); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledOnce()); + const queuedWholeSteer = ws.steerPrompt('live'); + + // Archive invalidates both the in-flight row steer and the whole-queue + // operation waiting behind it. Restoring the same id must not revive either. + state.activeSessionId = undefined; + await ws.archiveSession('sess_1'); + await ws.restoreSession('sess_1'); + rejectSubmit(new DaemonApiError({ code: 50000, msg: 'gone', requestId: 'r' })); + await Promise.all([rowSteer, queuedWholeSteer]); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(apiMock.steerPrompts).not.toHaveBeenCalled(); + expect(state.queuedBySession.sess_1).toBeUndefined(); + expect(deps.pushOperationFailure).not.toHaveBeenCalledWith( + 'steer', + expect.anything(), + expect.anything(), + ); + }); + + it('blocks chained steers as soon as archive starts, before the archive response', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + ], + }; + let resolveSubmit!: (value: { promptId: string; status: string }) => void; + let resolveArchive!: () => void; + apiMock.submitPrompt.mockImplementation( + () => new Promise((resolve) => { resolveSubmit = resolve; }), + ); + apiMock.archiveSession.mockImplementation( + () => new Promise((resolve) => { resolveArchive = resolve; }), + ); + const deps = promptDeps({ + sideChat: { + clearSideChatForSession: vi.fn(), + } as unknown as UseWorkspaceStateDeps['sideChat'], + forgetSession: vi.fn((id: string) => { + forgetLocalTurnState(id); + state.sessions = state.sessions.filter((session) => session.id !== id); + delete state.queuedBySession[id]; + delete state.inFlightBySession[id]; + delete state.turnActiveBySession[id]; + }), + }); + const ws = useWorkspaceState(state, deps); + + const rowSteer = ws.steerQueuedPrompt(0); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledOnce()); + const queuedWholeSteer = ws.steerPrompt('live'); + const archive = ws.archiveSession('sess_1'); + + resolveSubmit({ promptId: 'prompt_a', status: 'queued' }); + await Promise.resolve(); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(apiMock.steerPrompts).not.toHaveBeenCalled(); + + resolveArchive(); + await Promise.all([archive, rowSteer, queuedWholeSteer]); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(apiMock.steerPrompts).not.toHaveBeenCalled(); + expect(state.queuedBySession.sess_1).toBeUndefined(); + }); + + it('releases chained steers when archive fails', async () => { + const archiveError = new Error('archive failed'); + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + ], + }; + let resolveFirstSubmit!: (value: { promptId: string; status: string }) => void; + let rejectArchive!: (error: Error) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => new Promise((resolve) => { resolveFirstSubmit = resolve; }), + ) + .mockResolvedValueOnce({ promptId: 'prompt_b', status: 'queued' }); + apiMock.archiveSession.mockImplementation( + () => new Promise((_resolve, reject) => { rejectArchive = reject; }), + ); + const deps = promptDeps(); + const ws = useWorkspaceState(state, deps); + + const rowSteer = ws.steerQueuedPrompt(0); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledOnce()); + const queuedWholeSteer = ws.steerPrompt('live'); + const archive = ws.archiveSession('sess_1'); + + resolveFirstSubmit({ promptId: 'prompt_a', status: 'queued' }); + await Promise.resolve(); + expect(apiMock.steerPrompts).not.toHaveBeenCalled(); + + rejectArchive(archiveError); + await Promise.all([archive, rowSteer, queuedWholeSteer]); + + expect(apiMock.steerPrompts).toHaveBeenNthCalledWith(1, 'sess_1', ['prompt_a']); + expect(apiMock.steerPrompts).toHaveBeenNthCalledWith(2, 'sess_1', ['prompt_b']); + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(2); + expect(deps.pushOperationFailure).toHaveBeenCalledWith('archiveSession', archiveError, { + sessionId: 'sess_1', + }); + }); + + // The full session-switch scenario: a serialized per-entry steer fires + // after the user switched to an idle session — liveness still comes from + // the captured session, so the entry is steered, not normal-sent. + it('steers a serialized per-entry steer after a mid-flight session switch', async () => { + const state = createState(); + state.activeSessionId = 'sess_1'; + state.turnActiveBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [ + { text: 'A', attachments: undefined }, + { text: 'B', attachments: undefined }, + ], + }; + let resolveFirst!: (value: { promptId: string; status: string }) => void; + apiMock.submitPrompt + .mockImplementationOnce( + () => { + return new Promise((resolve) => { resolveFirst = resolve; }); + }, + ) + .mockResolvedValueOnce({ promptId: 'prompt_b', status: 'queued' }); + // Activity follows the active view, like the real computed. + const ws = useWorkspaceState( + state, + promptDeps({ + activity: computed(() => { + return state.activeSessionId === 'sess_1' ? 'running' : 'idle'; + }), + }), + ); + + const first = ws.steerQueuedPrompt(0); + await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalledTimes(1)); + const second = ws.steerQueuedPrompt(0); + state.activeSessionId = 'sess_2'; // the user switches before the first settles + + resolveFirst({ promptId: 'prompt_a', status: 'queued' }); + await first; + await second; + + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_b']); + }); + + it('steers a queued prompt when only session metadata marks the main turn active', async () => { + const state = createState(); + state.sessions = [{ ...createSession(), mainTurnActive: true }]; + state.queuedBySession = { + sess_1: [{ text: 'queued', attachments: undefined }], + }; + apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_queued', status: 'queued' }); + const ws = useWorkspaceState(state, promptDeps()); + + await ws.steerQueuedPrompt(0); + + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_queued']); + }); + // A background session's drained prompt must not inherit the thinking level // of whichever session is active when the drain happens — the level is // resolved from the prompt's OWN model, never the active-view global. @@ -2007,11 +2798,11 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { const ws = useWorkspaceState(state, promptDeps()); ws.finishPromptLocal('sess_1', { turnWasActive: true }); - expect(state.queuedBySession.sess_1 ?? []).toEqual([]); // Facade forget path (e.g. archive) while the submit is pending. The // daemon definitively rejects afterwards — even then, no resurrection. await vi.waitFor(() => expect(apiMock.submitPrompt).toHaveBeenCalled()); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); state.sessions = []; delete state.queuedBySession.sess_1; rejectSubmit(new DaemonApiError({ code: 50000, msg: 'network down', requestId: 'r' })); diff --git a/packages/agent-core-v2/docs/state-manifest.d.ts b/packages/agent-core-v2/docs/state-manifest.d.ts index 4f6bded88c..6de0b48061 100644 --- a/packages/agent-core-v2/docs/state-manifest.d.ts +++ b/packages/agent-core-v2/docs/state-manifest.d.ts @@ -1010,7 +1010,7 @@ export interface AgentStateSnapshot { 'llmRequester.lastConfigLogSignature': string | undefined; 'llmRequester.mediaDegradedTurns': Set; 'llmRequester.mediaStrippedTurns': Map; 'llmRequester.turnConfigs': Map { @@ -146,6 +149,7 @@ export class AgentPromptService implements IAgentPromptService { }, () => {}); const turn = (await this.loop.enqueue(request).assigned).turn; if (turn === undefined) throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); + this.telemetry.track2('input_steer', { parts: message.content.length }); for (const item of selected) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...selected]); this.eventBus.publish({ type: 'prompt.steered', activePromptId: this.active.id, promptIds: selected.map((x) => x.id), content: rerouted.content as ContentPart[], steeredAt: new Date().toISOString() }); diff --git a/packages/agent-core-v2/src/agent/rpc/rpcService.ts b/packages/agent-core-v2/src/agent/rpc/rpcService.ts index bb7ffef1d4..361a0f39da 100644 --- a/packages/agent-core-v2/src/agent/rpc/rpcService.ts +++ b/packages/agent-core-v2/src/agent/rpc/rpcService.ts @@ -108,7 +108,6 @@ export class AgentRPCService implements IAgentRPCService { } async steer(payload: SteerPayload): Promise { - this.telemetry.track2('input_steer', { parts: payload.input.length }); const queued = await this.promptService.enqueue({ message: { role: 'user', content: [...payload.input], diff --git a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts index 013b919830..f39474a3d9 100644 --- a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts +++ b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts @@ -22,6 +22,7 @@ import { AgentSystemReminderService } from '#/agent/systemReminder/systemReminde import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; import { IEventBus } from '#/app/event/eventBus'; import { EventBusService } from '#/app/event/eventBusService'; +import { ITelemetryService } from '#/app/telemetry/telemetry'; import { ErrorCodes, Error2 } from '#/errors'; import { createHooks } from '#/hooks'; import { IWireService } from '#/wire/wire'; @@ -29,6 +30,7 @@ import { IWireService } from '#/wire/wire'; import { stubContextMemory } from '../contextMemory/stubs'; import { stubLoopWithHooks, stubToolExecutor, stubWire } from '../loop/stubs'; import { registerStateServices } from '../../state/stubs'; +import { recordingTelemetry, type TelemetryRecord } from '../../app/telemetry/stubs'; function message(text: string): ContextMessage { return { role: 'user', content: [{ type: 'text', text }], toolCalls: [], origin: { kind: 'user' } }; @@ -46,6 +48,7 @@ function harness() { hooks: createHooks(['onWillCompact']), onDidFinishCompaction: Event.None, } as unknown as IAgentFullCompactionService; + const telemetryRecords: TelemetryRecord[] = []; const ix = createServices(disposables, { strict: true, additionalServices: (reg) => { registerStateServices(reg); @@ -54,12 +57,13 @@ function harness() { reg.defineInstance(IWireService, stubWire()); reg.defineInstance(IAgentToolExecutorService, stubToolExecutor()); reg.defineInstance(IAgentFullCompactionService, fullCompaction); + reg.defineInstance(ITelemetryService, recordingTelemetry(telemetryRecords)); reg.define(IEventBus, EventBusService); reg.define(IAgentSystemReminderService, AgentSystemReminderService); reg.define(IAgentPromptService, AgentPromptService); } }); - return { prompt: ix.get(IAgentPromptService), loop, context, fullCompaction }; + return { prompt: ix.get(IAgentPromptService), loop, context, fullCompaction, telemetryRecords }; } describe('AgentPromptService', () => { @@ -98,6 +102,18 @@ describe('AgentPromptService', () => { loop.drainNextBatch(context); }); + it('emits input_steer telemetry when pending prompts are steered', async () => { + const { prompt, telemetryRecords } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const one = await prompt.enqueue({ message: message('one') }); + const two = await prompt.enqueue({ message: message('two') }); + await prompt.steer([two.id, one.id]); + expect(telemetryRecords.filter((record) => record.event === 'input_steer')).toEqual([ + { event: 'input_steer', properties: { parts: 2 } }, + ]); + }); + it('aborts pending prompts and settles completion', async () => { const { prompt } = harness(); await prompt.enqueue({ message: message('active') });