From 42ca9151e67805410497f4ffcb60192ee100efad Mon Sep 17 00:00:00 2001 From: qer Date: Fri, 17 Jul 2026 17:11:23 +0800 Subject: [PATCH 1/6] fix(kimi-web): stop the prompt queue from ghost-sending stale attachments Sending while a turn was running queues prompts locally. A failed flush left entries stuck, and every later session open silently re-submitted them with their old file attachments. Gate the drain on locally witnessed turns, re-drive stuck entries FIFO from real events with a failure budget, restore merged entries on steer failure, persist the queue per session so a refresh loses nothing, and converge queues across tabs via storage events. --- .../web-prompt-queue-stale-attachments.md | 5 + .../composables/client/useWorkspaceState.ts | 193 +++++++++++++-- .../src/composables/useKimiWebClient.ts | 16 +- apps/kimi-web/src/lib/storage.ts | 2 + apps/kimi-web/test/workspace-state.test.ts | 227 +++++++++++++++++- 5 files changed, 416 insertions(+), 27 deletions(-) create mode 100644 .changeset/web-prompt-queue-stale-attachments.md diff --git a/.changeset/web-prompt-queue-stale-attachments.md b/.changeset/web-prompt-queue-stale-attachments.md new file mode 100644 index 0000000000..71ed465a3b --- /dev/null +++ b/.changeset/web-prompt-queue-stale-attachments.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Fix queued messages silently re-sending previously uploaded files when a session is reopened, and keep unsent queued messages across page refreshes. diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index 90df990aa7..85b2c8cc6d 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -7,7 +7,7 @@ // the view-model computeds stay in the facade; cross-dependencies are injected // here as params. -import { reactive, type ComputedRef, type Ref } from 'vue'; +import { reactive, watch, type ComputedRef, type Ref } from 'vue'; import { getKimiWebApi } from '../../api'; import { i18n } from '../../i18n'; import { useConfirmDialog } from '../useConfirmDialog'; @@ -28,7 +28,9 @@ import type { } from '../../api/types'; import { loadWorkspaceNameOverrides, + safeGetJson, safeRemove, + safeSetJson, saveWorkspaceNameOverrides, STORAGE_KEYS, } from '../../lib/storage'; @@ -118,6 +120,67 @@ const pendingLocalTurnStarts = new Map>(); const afterLocalTurnsSettled = new Map void>(); let nextLocalTurnToken = 0; +/** + * Consecutive flushQueueHead failures per session. A rejected queued prompt + * goes back at the head and waits for the next turn end (or idle send) to + * retry; beyond MAX_QUEUE_FLUSH_FAILURES it is dropped so a permanently + * rejected entry (e.g. one whose uploaded files no longer exist server-side) + * cannot block every later prompt behind it. Module-level singleton — the + * queue itself is per-session on rawState, so a page reload resets both. + */ +const queueFlushFailures = new Map(); +const MAX_QUEUE_FLUSH_FAILURES = 3; + +type QueuedPromptEntry = { text: string; attachments?: PromptAttachment[] }; + +/** + * The local prompt queue is persisted per session (STORAGE_KEYS.promptQueue) + * so messages the user already sent — but that only made it as far as the + * local queue because a turn was still running — survive a page refresh + * instead of vanishing from memory. A restored queue is display-only until a + * real flush driver runs (finishPromptLocal's drain gate requires a locally + * witnessed turn, and sendPrompt flushes FIFO), so it can never ghost-send + * just because a session was re-opened. + */ +function loadQueuedPrompts(): Record { + const parsed = safeGetJson(STORAGE_KEYS.promptQueue); + if (!parsed || typeof parsed !== 'object') return {}; + const out: Record = {}; + for (const [sid, entries] of Object.entries(parsed as Record)) { + if (!Array.isArray(entries)) continue; + const valid = entries.filter(isQueuedPromptLike); + if (valid.length > 0) out[sid] = valid; + } + return out; +} + +function isQueuedPromptLike(value: unknown): value is QueuedPromptEntry { + if (!value || typeof value !== 'object') return false; + const entry = value as { text?: unknown; attachments?: unknown }; + if (typeof entry.text !== 'string') return false; + if (entry.attachments === undefined) return true; + if (!Array.isArray(entry.attachments)) return false; + return entry.attachments.every(isPromptAttachmentLike); +} + +function isPromptAttachmentLike(value: unknown): value is PromptAttachment { + if (!value || typeof value !== 'object') return false; + const att = value as { fileId?: unknown; kind?: unknown }; + return ( + typeof att.fileId === 'string' && + (att.kind === 'image' || att.kind === 'video' || att.kind === 'file') + ); +} + +function saveQueuedPrompts(queue: Record): void { + // Drop empty arrays so a fully drained session leaves no key behind. + const compact: Record = {}; + for (const [sid, entries] of Object.entries(queue)) { + if (entries.length > 0) compact[sid] = entries; + } + safeSetJson(STORAGE_KEYS.promptQueue, compact); +} + export interface LocalTurnStartState { generation: number; pending: boolean; @@ -161,6 +224,7 @@ export function forgetLocalTurnState(sid: string): void { promptGenerationBySession.delete(sid); pendingLocalTurnStarts.delete(sid); afterLocalTurnsSettled.delete(sid); + queueFlushFailures.delete(sid); } /** Whether a snapshot request can still be applied without overwriting a @@ -295,6 +359,36 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } = deps; let exportInFlight = false; + // Hydrate the local prompt queue from storage (see loadQueuedPrompts) so + // queued prompts survive a page refresh. Only fills an empty in-memory + // queue — a live one is never clobbered. + const storedQueue = loadQueuedPrompts(); + if (Object.keys(rawState.queuedBySession).length === 0 && Object.keys(storedQueue).length > 0) { + rawState.queuedBySession = storedQueue; + } + // Every queue mutation is a whole-record replace or a key delete, so one + // deep watch keeps storage in sync — including the facade's session-forget + // delete, which bypasses this module entirely. + watch( + () => rawState.queuedBySession, + (queue) => { + saveQueuedPrompts(queue); + }, + { deep: true }, + ); + + /** + * Multi-tab convergence: adopt the queue another tab just persisted. Unlike + * startup hydration this ALWAYS replaces the in-memory queue — the other + * tab's write is the newer truth, and adopting it is what stops two tabs + * from flushing the same queued prompts twice. (Our own not-yet-persisted + * mutation is at most a microtask old and is re-written by the watch right + * after, becoming the next newer truth.) + */ + function syncQueuedPromptsFromStorage(): void { + rawState.queuedBySession = loadQueuedPrompts(); + } + async function loadOlderMessages(sessionId: string): Promise { if (rawState.messagesLoadingMoreBySession[sessionId]) return; const current = rawState.messagesBySession[sessionId]; @@ -1562,6 +1656,19 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta return; } + // The queue should be empty by the time the session is idle, so a + // non-empty queue means earlier prompts never made it out (e.g. a drain + // that raced a still-busy daemon right after an abort). Preserve FIFO: + // 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) { + enqueue(text, attachments); + flushQueueHead(sid); + return; + } + await submitPromptInternal(sid, text, attachments); } @@ -1594,9 +1701,19 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } 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] }; + }; + // Idle and nothing in flight — there is no turn to steer into; normal send. if (activity.value === 'idle' && !rawState.inFlightBySession[sid]) { - await submitPromptInternal(sid, merged, mergedAttachments); + const ok = await submitPromptInternal(sid, merged, mergedAttachments); + if (!ok) restoreQueue(); return; } @@ -1672,8 +1789,10 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } } catch (err) { // Submit failed: drop the optimistic echo so the transcript doesn't show - // a delivered-looking message the daemon never received. + // a delivered-looking message the daemon never received, and put the + // merged queue entries back for a later steer/drain. updateSessionMessages(sid, (msgs) => msgs.filter((m) => m.id !== tempId)); + restoreQueue(); pushOperationFailure('steer', err, { sessionId: sid }); } finally { settleLocalTurn(sid, localTurnToken); @@ -1706,6 +1825,38 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta }; } + /** + * Submit the head of the session's local prompt queue. On failure the + * entry goes back at the head with NO immediate retry — the daemon that + * just rejected it is the one we'd race against (e.g. right after an + * abort); the next turn end or the next idle sendPrompt drives the next + * attempt. After MAX_QUEUE_FLUSH_FAILURES consecutive failures the entry + * is dropped instead, so one permanently rejected prompt cannot wedge the + * queue. Every failure is already surfaced via pushOperationFailure. + */ + function flushQueueHead(sid: string): void { + const [next, ...rest] = rawState.queuedBySession[sid] ?? []; + if (next === undefined) return; + rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: rest }; + void submitPromptInternal(sid, next.text, next.attachments).then((ok) => { + if (ok) { + queueFlushFailures.delete(sid); + return; + } + const failures = (queueFlushFailures.get(sid) ?? 0) + 1; + if (failures >= MAX_QUEUE_FLUSH_FAILURES) { + queueFlushFailures.delete(sid); + return; + } + queueFlushFailures.set(sid, failures); + const current = rawState.queuedBySession[sid] ?? []; + rawState.queuedBySession = { + ...rawState.queuedBySession, + [sid]: [next, ...current], + }; + }); + } + /** * Shared prompt-finish cleanup, used by BOTH the main-turn-ended path * (facade `onMainTurnEnd`) and the authoritative-snapshot path @@ -1718,8 +1869,15 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta * duplicate idle event) therefore cannot drain more than one message per * real turn end. Callers layer their own side effects (notify, sound, * unread) on top; the snapshot path deliberately adds none. + * + * The drain is GATED on this client having actually witnessed a live + * prompt/turn: a finish with no local in-flight prompt, no locally + * tracked active turn, and no `turnWasActive` hint must NOT submit + * queued prompts. That is what fired stale queued messages (and their + * old file attachments) spontaneously when a session was merely + * re-opened after an earlier drain had failed. */ - function finishPromptLocal(sid: string): boolean { + function finishPromptLocal(sid: string, opts?: { turnWasActive?: boolean }): boolean { const wasInFlight = rawState.inFlightBySession[sid] === true; rawState.inFlightBySession = { ...rawState.inFlightBySession, [sid]: false }; // Drop any cached prompt_id so a later skill activation (which has no @@ -1733,23 +1891,10 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta resetFastMoon(); } - const queue = rawState.queuedBySession[sid] ?? []; - if (queue.length > 0) { - const [next, ...rest] = queue; - rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: rest }; - // Flush the first queued message; on failure put it back at the head so - // a transient error doesn't silently drop the prompt. - if (next !== undefined) { - void submitPromptInternal(sid, next.text, next.attachments).then((ok) => { - if (!ok) { - const current = rawState.queuedBySession[sid] ?? []; - rawState.queuedBySession = { - ...rawState.queuedBySession, - [sid]: [next, ...current], - }; - } - }); - } + const mayDrain = + wasInFlight || opts?.turnWasActive === true || (rawState.turnActiveBySession[sid] ?? false); + if (mayDrain) { + flushQueueHead(sid); } return wasInFlight; @@ -1763,7 +1908,10 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta * the next prompt queues behind a turn that already ended. * * Unlike the WS path this adds NO completion side effects (no notification, - * sound, or unread): opening a historical session must not cry wolf. + * sound, or unread): opening a historical session must not cry wolf. The + * queue drain inside finishPromptLocal is additionally gated on a locally + * witnessed prompt/turn, so merely re-opening a session can never + * spontaneously submit queued prompts (with their stale attachments). */ function handleSessionSnapshot( sid: string, @@ -2648,6 +2796,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta enqueue, unqueue, reorderQueue, + syncQueuedPromptsFromStorage, abortCurrentPrompt, respondApproval, respondQuestion, diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 7ffc1b0d93..5de057a85c 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -505,6 +505,12 @@ if (typeof window !== 'undefined') { rawState.unreadBySession = loadUnread(); clearActiveUnread(); } + // Another tab enqueued/flushed/discarded queued prompts — adopt its queue + // so both tabs don't flush the same entries twice. workspaceState is + // referenced lazily inside the event callback, after module init. + if (event.key === STORAGE_KEYS.promptQueue) { + workspaceState.syncQueuedPromptsFromStorage(); + } }); } @@ -944,9 +950,15 @@ function processEvent(appEvent: AppEvent, meta: KimiEventMeta): void { meta.seq > prevSeq ) { const reason = appEvent.reason; + // wasMainTurnActive was captured BEFORE the reducer consumed this event + // (the reducer clears turnActiveBySession on turn end), so it is the only + // remaining signal that this client witnessed a live turn — pass it down + // so finishPromptLocal may drain queued prompts behind a turn the user + // actually watched (including one started by another client). onMainTurnEnd( appEvent.sessionId, reason === 'cancelled' || reason === 'failed' || reason === 'blocked' ? 'aborted' : 'idle', + wasMainTurnActive, ); } @@ -2638,7 +2650,7 @@ function clearWorkingFlags(sid: string): void { } } -function onMainTurnEnd(sid: string, status: 'idle' | 'aborted'): void { +function onMainTurnEnd(sid: string, status: 'idle' | 'aborted', turnWasActive: boolean): void { // Capture before finishPromptLocal drops it — it keys the completion // notification's dedup tag so each finished turn alerts once. const finishedPromptId = rawState.promptIdBySession[sid]; @@ -2646,7 +2658,7 @@ function onMainTurnEnd(sid: string, status: 'idle' | 'aborted'): void { // queued message. The notification/sound/unread side effects below stay // WS-event-only — the snapshot path (handleSessionSnapshot) must not cry // wolf when opening a historical session. - workspaceState.finishPromptLocal(sid); + workspaceState.finishPromptLocal(sid, { turnWasActive }); // For the session on screen, refresh git status (edits the agent just made) // and runtime status (model/context usage may have changed this turn). diff --git a/apps/kimi-web/src/lib/storage.ts b/apps/kimi-web/src/lib/storage.ts index 5cd60760c6..a462ee3cf1 100644 --- a/apps/kimi-web/src/lib/storage.ts +++ b/apps/kimi-web/src/lib/storage.ts @@ -34,6 +34,8 @@ export const STORAGE_KEYS = { notifyOnApproval: 'kimi-web.notify-on-approval', soundOnComplete: 'kimi-web.sound-on-complete', inputHistory: 'kimi-web.input-history', + // useWorkspaceState — per-session local prompt queue + promptQueue: 'kimi-web.prompt-queue', // cross-file locale: 'kimi-locale', clientId: 'kimi-web.client-id', diff --git a/apps/kimi-web/test/workspace-state.test.ts b/apps/kimi-web/test/workspace-state.test.ts index 7c9f8c8966..b3c087ca6a 100644 --- a/apps/kimi-web/test/workspace-state.test.ts +++ b/apps/kimi-web/test/workspace-state.test.ts @@ -3,14 +3,14 @@ // Wiring: the composable is real; daemon requests and unrelated facade collaborators are stubbed. // Run: pnpm --filter @moonshot-ai/kimi-web exec vitest run test/workspace-state.test.ts -import { computed, ref, type Ref } from 'vue'; +import { computed, nextTick, reactive, ref, type Ref } from 'vue'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { AppApprovalRequest, AppQuestionRequest, AppSession, AppTask } from '../src/api/types'; import { DaemonApiError } from '../src/api/errors'; import { createInitialState } from '../src/api/daemon/eventReducer'; import { mergeWorkspaces } from '../src/lib/mergeWorkspaces'; -import { loadWorkspaceNameOverrides, saveWorkspaceNameOverrides } from '../src/lib/storage'; -import { useWorkspaceState, type UseWorkspaceStateDeps } from '../src/composables/client/useWorkspaceState'; +import { loadWorkspaceNameOverrides, saveWorkspaceNameOverrides, STORAGE_KEYS } from '../src/lib/storage'; +import { useWorkspaceState, forgetLocalTurnState, type UseWorkspaceStateDeps } from '../src/composables/client/useWorkspaceState'; import type { ExtendedState } from '../src/composables/useKimiWebClient'; import { clearTrace, traceKeyEvent } from '../src/debug/trace'; @@ -1539,6 +1539,8 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { beforeEach(() => { apiMock.submitPrompt.mockReset(); apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_new' }); + // Module-level flush failure budget must not leak between tests. + forgetLocalTurnState('sess_1'); }); it('clears a finished prompt from a terminal snapshot so the next send is immediate', async () => { @@ -1595,6 +1597,108 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { ]); }); + // Regression: re-opening a session after a failed drain must NOT fire the + // stuck queued prompts (with their stale attachments) out of nowhere. + it('does not drain the queue on a bare session-open snapshot with no locally witnessed prompt', () => { + const state = createState(); + state.queuedBySession = { + sess_1: [{ text: 'stuck queued', attachments: [{ fileId: 'f_old', kind: 'image' }] }], + }; + const ws = useWorkspaceState(state, promptDeps()); + + ws.handleSessionSnapshot('sess_1', { inFlightTurn: null, busy: false }); + + expect(apiMock.submitPrompt).not.toHaveBeenCalled(); + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'stuck queued', attachments: [{ fileId: 'f_old', kind: 'image' }] }], + ); + }); + + it('drains one queued prompt when the finished turn was locally witnessed', () => { + const state = createState(); + state.queuedBySession = { + sess_1: [ + { text: 'first queued', attachments: undefined }, + { text: 'second queued', attachments: undefined }, + ], + }; + const ws = useWorkspaceState(state, promptDeps()); + + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'second queued', attachments: undefined }, + ]); + }); + + 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 }] }; + const ws = useWorkspaceState(state, promptDeps({ activity: computed(() => 'idle') })); + + await ws.sendPrompt('next'); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(apiMock.submitPrompt).toHaveBeenCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'stuck queued' }] }), + ); + expect(state.queuedBySession.sess_1).toEqual([{ text: 'next', attachments: undefined }]); + }); + + 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 }] }; + apiMock.submitPrompt.mockRejectedValue(new Error('turn.agent_busy')); + const ws = useWorkspaceState(state, promptDeps()); + const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); + + // Failures 1-2 (e.g. racing a still-busy daemon after an abort): the + // entry goes back at the head and waits for the next flush driver. + for (let i = 0; i < 2; i += 1) { + state.inFlightBySession = { sess_1: true }; + ws.handleSessionSnapshot('sess_1', { inFlightTurn: null, busy: false }); + await settle(); + expect(state.queuedBySession.sess_1).toEqual([{ text: 'first queued', attachments: undefined }]); + } + + // Failure 3: a permanently rejected head is dropped rather than blocking + // every later prompt behind it forever. + state.inFlightBySession = { sess_1: true }; + ws.handleSessionSnapshot('sess_1', { inFlightTurn: null, busy: false }); + await settle(); + expect(state.queuedBySession.sess_1).toEqual([]); + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(3); + }); + + it('restores the merged queue entries when a steer submit fails', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [{ text: 'queued', attachments: [{ fileId: 'f_q', kind: 'image' }] }], + }; + apiMock.submitPrompt.mockRejectedValue(new Error('boom')); + const ws = useWorkspaceState(state, promptDeps()); + + await ws.steerPrompt('live text', [{ fileId: 'f_live', kind: 'image' }]); + + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'queued', attachments: [{ fileId: 'f_q', kind: 'image' }] }], + ); + }); + + it('restores the queue when an idle steer falls back to a normal send that fails', async () => { + const state = createState(); + state.queuedBySession = { sess_1: [{ text: 'queued', attachments: undefined }] }; + apiMock.submitPrompt.mockRejectedValue(new Error('boom')); + const ws = useWorkspaceState(state, promptDeps({ activity: computed(() => 'idle') })); + + await ws.steerPrompt('live text'); + + expect(state.queuedBySession.sess_1).toEqual([{ text: 'queued', attachments: undefined }]); + }); + it('clears local prompt state when busy disproves a stale snapshot turn', () => { const state = createState(); state.inFlightBySession = { sess_1: true }; @@ -1686,6 +1790,123 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { }); }); +describe('useWorkspaceState — prompt queue persistence', () => { + function deps(overrides: Partial = {}): UseWorkspaceStateDeps { + return { + ...createDeps(), + modelProvider: { models: ref([]) } as unknown as UseWorkspaceStateDeps['modelProvider'], + ...overrides, + }; + } + + beforeEach(() => { + installStorage(createMemoryStorage()); + apiMock.submitPrompt.mockReset(); + apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_new' }); + // Module-level flush failure budget must not leak between tests. + forgetLocalTurnState('sess_1'); + }); + + it('restores a persisted queue on startup so a refresh loses nothing', () => { + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: [{ text: 'old queued', attachments: [{ fileId: 'f_old', kind: 'image' }] }], + }), + ); + const state = createState(); + useWorkspaceState(state, deps()); + + expect(state.queuedBySession.sess_1).toEqual([ + { text: 'old queued', attachments: [{ fileId: 'f_old', kind: 'image' }] }], + ); + }); + + it('keeps only well-formed persisted entries and ignores the rest', () => { + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: [{ text: 5 }, 'nope', { text: 'ok', attachments: [{ fileId: 'f', kind: 'file' }] }], + sess_2: 'not-an-array', + sess_3: [{ text: 'bad kind', attachments: [{ fileId: 'f', kind: 'exe' }] }], + }), + ); + const state = createState(); + useWorkspaceState(state, deps()); + + expect(state.queuedBySession).toEqual({ + sess_1: [{ text: 'ok', attachments: [{ fileId: 'f', kind: 'file' }] }], + }); + }); + + it('does not clobber an in-memory queue with the persisted one', () => { + localStorage.setItem(STORAGE_KEYS.promptQueue, JSON.stringify({ sess_1: [{ text: 'stored' }] })); + const state = createState(); + state.queuedBySession = { sess_1: [{ text: 'in memory', attachments: undefined }] }; + useWorkspaceState(state, deps()); + + expect(state.queuedBySession.sess_1).toEqual([{ text: 'in memory', attachments: undefined }]); + }); + + it('persists enqueue, flush, and session-forget mutations', async () => { + // reactive() so the persistence watch actually tracks the state, the same + // way the facade wraps rawState in production. + const state = reactive(createState()); + const ws = useWorkspaceState(state, deps()); + + // A send while the session is running lands in the queue AND in storage. + await ws.sendPrompt('queued while running', [{ fileId: 'f_q', kind: 'image' }]); + await nextTick(); + expect(JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!)).toEqual({ + sess_1: [{ text: 'queued while running', attachments: [{ fileId: 'f_q', kind: 'image' }] }], + }); + + // A witnessed turn end flushes the head — the persisted queue drains too. + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + await nextTick(); + expect(JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!)).toEqual({}); + + // Deleting a session key (the facade's session-forget path) syncs via the + // same deep watch. + state.queuedBySession = { + sess_1: [{ text: 'a', attachments: undefined }], + sess_2: [{ text: 'b', attachments: undefined }], + }; + await nextTick(); + delete state.queuedBySession.sess_1; + await nextTick(); + expect(JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!)).toEqual({ + sess_2: [{ text: 'b' }], + }); + }); + + it('adopts the queue another tab persisted, so two tabs never flush the same entries twice', () => { + const state = createState(); + state.queuedBySession = { sess_1: [{ text: 'stale local', attachments: undefined }] }; + const ws = useWorkspaceState(state, deps()); + + // Simulate another tab flushing part of the queue and persisting the rest. + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ sess_1: [{ text: 'newer from tab B' }] }), + ); + ws.syncQueuedPromptsFromStorage(); + + expect(state.queuedBySession).toEqual({ sess_1: [{ text: 'newer from tab B' }] }); + }); + + it('clears the in-memory queue when another tab drained it', () => { + const state = createState(); + state.queuedBySession = { sess_1: [{ text: 'stale local', attachments: undefined }] }; + const ws = useWorkspaceState(state, deps()); + + localStorage.setItem(STORAGE_KEYS.promptQueue, JSON.stringify({})); + ws.syncQueuedPromptsFromStorage(); + + expect(state.queuedBySession).toEqual({}); + }); +}); + // Regression: a search-triggered full session-list reload must not clobber the // live usage (context ring) with the list endpoint's all-zero placeholder. describe('useWorkspaceState — loadAllSessions usage preservation', () => { From 932e228f7b696b2acc4cafa100ee6658dfffc991 Mon Sep 17 00:00:00 2001 From: qer Date: Fri, 17 Jul 2026 17:36:09 +0800 Subject: [PATCH 2/6] fix(kimi-web): merge cross-tab queue updates by entry id (review P1/P2) Whole-record adoption could silently discard a prompt another tab enqueued concurrently. Queue entries now carry a stable id and enqueue timestamp; adoption union-merges snapshots by id, a shared TTL'd removal set stops flushed/discarded entries from resurrecting (and being flushed twice), an in-flight marker covers the submit window, and manual reorders re-stamp timestamps so they survive merges. The flush failure budget is also tracked per entry instead of per session, so removing or reordering the head no longer hands its strikes to the next entry. --- .../composables/client/useWorkspaceState.ts | 296 ++++++++++++++++-- .../src/composables/useKimiWebClient.ts | 12 +- apps/kimi-web/test/workspace-state.test.ts | 234 ++++++++++++-- 3 files changed, 483 insertions(+), 59 deletions(-) diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index 85b2c8cc6d..e73d29abe8 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -29,8 +29,9 @@ import type { import { loadWorkspaceNameOverrides, safeGetJson, + safeGetString, safeRemove, - safeSetJson, + safeSetString, saveWorkspaceNameOverrides, STORAGE_KEYS, } from '../../lib/storage'; @@ -128,10 +129,109 @@ let nextLocalTurnToken = 0; * cannot block every later prompt behind it. Module-level singleton — the * queue itself is per-session on rawState, so a page reload resets both. */ -const queueFlushFailures = new Map(); +/** + * Consecutive flushQueueHead failures per queue ENTRY (not per session) — + * keyed by entry id, falling back to its text for entries without one. + * Keying by entry keeps a removed or reordered head from handing its + * failure budget down to the next entry. + */ +const queueFlushFailures = new Map(); const MAX_QUEUE_FLUSH_FAILURES = 3; -type QueuedPromptEntry = { text: string; attachments?: PromptAttachment[] }; +/** + * Entries this tab popped and is submitting right now: already out of the + * local queue but not yet acknowledged. Cross-tab merges treat these as + * removed so an adopt can't bring a submitting entry back early — a failed + * submit re-queues it under the SAME id and the merge dedupes. + */ +const queueInFlightSubmits = new Map>(); + +/** How long a removal stays remembered in the shared snapshot (per id). */ +const QUEUE_GONE_TTL_MS = 30 * 60 * 1000; + +type QueuedPromptEntry = { + text: string; + attachments?: PromptAttachment[]; + id?: string; + enqueuedAt?: number; +}; + +/** + * Shared (localStorage) per-session queue snapshot. `gone` is the removal + * set — ids this or another tab finished/discarded, with timestamps — and is + * what stops a union merge from RESURRECTING an entry a concurrent tab + * already flushed or discarded (without it, the tab still holding the entry + * would merge it back in and eventually flush it twice). + */ +interface SessionQueueSnapshot { + entries: QueuedPromptEntry[]; + gone?: Record; +} + +let queueEntryCounter = 0; +function nextQueueEntryId(): string { + queueEntryCounter += 1; + return `${Date.now().toString(36)}-${queueEntryCounter}-${Math.random().toString(36).slice(2, 8)}`; +} + +function pruneGoneMap(gone: Record | undefined): Record { + if (gone === undefined) return {}; + const cutoff = Date.now() - QUEUE_GONE_TTL_MS; + const out: Record = {}; + for (const [id, at] of Object.entries(gone)) { + if (at >= cutoff) out[id] = at; + } + return out; +} + +/** Stable, deterministic order for a session's queue. Manual reorder + * re-stamps enqueuedAt, so a user's drag order survives sorting. */ +function sortQueuedEntries(entries: QueuedPromptEntry[]): QueuedPromptEntry[] { + return entries.toSorted( + (a, b) => (a.enqueuedAt ?? Number.POSITIVE_INFINITY) - (b.enqueuedAt ?? Number.POSITIVE_INFINITY), + ); +} + +/** + * Union-merge two per-session entry maps by entry id. Merging instead of + * last-writer-wins replacement is what keeps two tabs enqueueing different + * prompts close together from silently discarding each other's entry: the + * tab that lost the storage write still holds its entry in memory, the + * adopt merges it back, and the resulting watch write converges both tabs. + * Entries without an id (legacy persisted data) dedupe by content. Ids in + * `goneBySid` or submitting in this tab are filtered out. + */ +function mergeQueuedSnapshots( + a: Record, + b: Record, + goneBySid: Record>, +): Record { + const out: Record = {}; + const sids = new Set([...Object.keys(a), ...Object.keys(b)]); + for (const sid of sids) { + const gone = goneBySid[sid] ?? {}; + const submitting = queueInFlightSubmits.get(sid); + const seenIds = new Set(); + const seenContent = new Set(); + const merged: QueuedPromptEntry[] = []; + for (const entry of [...(a[sid] ?? []), ...(b[sid] ?? [])]) { + if (entry.id !== undefined) { + if (seenIds.has(entry.id) || gone[entry.id] !== undefined || submitting?.has(entry.id)) { + continue; + } + seenIds.add(entry.id); + } else { + const contentKey = `${entry.text}\n${JSON.stringify(entry.attachments ?? null)}`; + if (seenContent.has(contentKey)) continue; + seenContent.add(contentKey); + } + merged.push(entry); + } + const sorted = sortQueuedEntries(merged); + if (sorted.length > 0) out[sid] = sorted; + } + return out; +} /** * The local prompt queue is persisted per session (STORAGE_KEYS.promptQueue) @@ -142,27 +242,57 @@ type QueuedPromptEntry = { text: string; attachments?: PromptAttachment[] }; * witnessed turn, and sendPrompt flushes FIFO), so it can never ghost-send * just because a session was re-opened. */ -function loadQueuedPrompts(): Record { +function loadQueueSnapshots(): Record { const parsed = safeGetJson(STORAGE_KEYS.promptQueue); if (!parsed || typeof parsed !== 'object') return {}; - const out: Record = {}; - for (const [sid, entries] of Object.entries(parsed as Record)) { - if (!Array.isArray(entries)) continue; - const valid = entries.filter(isQueuedPromptLike); - if (valid.length > 0) out[sid] = valid; + const out: Record = {}; + for (const [sid, value] of Object.entries(parsed as Record)) { + // Legacy shape: a bare entry array with no removal set. + const raw = Array.isArray(value) + ? { entries: value as unknown[] } + : (value as { entries?: unknown; gone?: unknown } | null); + if (!raw || !Array.isArray(raw.entries)) continue; + const entries = raw.entries.filter(isQueuedPromptLike); + const gone: Record = {}; + if (raw.gone && typeof raw.gone === 'object') { + for (const [id, at] of Object.entries(raw.gone as Record)) { + if (typeof at === 'number') gone[id] = at; + } + } + const pruned = pruneGoneMap(gone); + if (entries.length > 0 || Object.keys(pruned).length > 0) { + out[sid] = { entries, gone: pruned }; + } } return out; } function isQueuedPromptLike(value: unknown): value is QueuedPromptEntry { if (!value || typeof value !== 'object') return false; - const entry = value as { text?: unknown; attachments?: unknown }; + const entry = value as { text?: unknown; attachments?: unknown; id?: unknown; enqueuedAt?: unknown }; if (typeof entry.text !== 'string') return false; + if (entry.id !== undefined && typeof entry.id !== 'string') return false; + if (entry.enqueuedAt !== undefined && typeof entry.enqueuedAt !== 'number') return false; if (entry.attachments === undefined) return true; if (!Array.isArray(entry.attachments)) return false; return entry.attachments.every(isPromptAttachmentLike); } +/** Entries view of the shared snapshots (removal-set ids filtered out). */ +function queuedEntriesFromSnapshots( + snapshots: Record, +): Record { + const out: Record = {}; + for (const [sid, snap] of Object.entries(snapshots)) { + const gone = snap.gone ?? {}; + const entries = sortQueuedEntries( + snap.entries.filter((entry) => entry.id === undefined || gone[entry.id] === undefined), + ); + if (entries.length > 0) out[sid] = entries; + } + return out; +} + function isPromptAttachmentLike(value: unknown): value is PromptAttachment { if (!value || typeof value !== 'object') return false; const att = value as { fileId?: unknown; kind?: unknown }; @@ -172,13 +302,44 @@ function isPromptAttachmentLike(value: unknown): value is PromptAttachment { ); } +function writeQueueSnapshots(snapshots: Record): void { + const out: Record = {}; + for (const [sid, snap] of Object.entries(snapshots)) { + const entries = sortQueuedEntries(snap.entries); + const gone = pruneGoneMap(snap.gone); + if (entries.length === 0 && Object.keys(gone).length === 0) continue; + out[sid] = { entries, ...(Object.keys(gone).length > 0 ? { gone } : {}) }; + } + const serialized = JSON.stringify(out); + // Skip the write when nothing changed — otherwise two tabs' writes would + // ping-pong storage events forever. + if (safeGetString(STORAGE_KEYS.promptQueue) === serialized) return; + safeSetString(STORAGE_KEYS.promptQueue, serialized); +} + +/** Persist the current in-memory queue, preserving stored removal sets. */ function saveQueuedPrompts(queue: Record): void { - // Drop empty arrays so a fully drained session leaves no key behind. - const compact: Record = {}; + const stored = loadQueueSnapshots(); + for (const [sid, snap] of Object.entries(stored)) { + snap.entries = queue[sid] ?? []; + } for (const [sid, entries] of Object.entries(queue)) { - if (entries.length > 0) compact[sid] = entries; + stored[sid] ??= { entries }; } - safeSetJson(STORAGE_KEYS.promptQueue, compact); + writeQueueSnapshots(stored); +} + +/** Read-modify-write the shared removal set for `sid`. */ +function tombstoneQueuedIds(sid: string, ids: Array): void { + const realIds = ids.filter((id): id is string => id !== undefined); + if (realIds.length === 0) return; + const stored = loadQueueSnapshots(); + const snap = stored[sid] ?? { entries: [] }; + const gone = { ...snap.gone }; + const now = Date.now(); + for (const id of realIds) gone[id] = now; + stored[sid] = { ...snap, gone }; + writeQueueSnapshots(stored); } export interface LocalTurnStartState { @@ -359,10 +520,12 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } = deps; let exportInFlight = false; - // Hydrate the local prompt queue from storage (see loadQueuedPrompts) so - // queued prompts survive a page refresh. Only fills an empty in-memory - // queue — a live one is never clobbered. - const storedQueue = loadQueuedPrompts(); + // Hydrate the local prompt queue from storage so queued prompts survive a + // page refresh. Only fills an empty in-memory queue — a live one is never + // clobbered. Entries already removed elsewhere (in the shared `gone` set) + // are filtered out. + const storedSnapshots = loadQueueSnapshots(); + const storedQueue = queuedEntriesFromSnapshots(storedSnapshots); if (Object.keys(rawState.queuedBySession).length === 0 && Object.keys(storedQueue).length > 0) { rawState.queuedBySession = storedQueue; } @@ -378,15 +541,40 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta ); /** - * Multi-tab convergence: adopt the queue another tab just persisted. Unlike - * startup hydration this ALWAYS replaces the in-memory queue — the other - * tab's write is the newer truth, and adopting it is what stops two tabs - * from flushing the same queued prompts twice. (Our own not-yet-persisted - * mutation is at most a microtask old and is re-written by the watch right - * after, becoming the next newer truth.) + * Multi-tab convergence: UNION-merge the queue another tab just persisted + * with our own by entry id (never last-writer-wins replacement — that + * would silently discard whichever tab lost a concurrent write). Entries + * in the shared `gone` set or currently submitting here are left out, so a + * flushed/discarded entry can't resurrect and be flushed twice. The watch + * above then persists the merged result, converging both tabs. */ function syncQueuedPromptsFromStorage(): void { - rawState.queuedBySession = loadQueuedPrompts(); + const snapshots = loadQueueSnapshots(); + const goneBySid: Record> = {}; + for (const [sid, snap] of Object.entries(snapshots)) { + goneBySid[sid] = snap.gone ?? {}; + } + rawState.queuedBySession = mergeQueuedSnapshots( + rawState.queuedBySession, + queuedEntriesFromSnapshots(snapshots), + goneBySid, + ); + } + + /** + * Drop a session's queue for good (the facade's session-forget path, e.g. + * after archive): every current entry lands in the shared removal set so + * another tab's merge can't resurrect it into a session that no longer + * exists. + */ + function discardQueuedPrompts(sid: string): void { + const current = rawState.queuedBySession[sid] ?? []; + tombstoneQueuedIds(sid, current.map((entry) => entry.id)); + if (rawState.queuedBySession[sid] !== undefined) { + const next = { ...rawState.queuedBySession }; + delete next[sid]; + rawState.queuedBySession = next; + } } async function loadOlderMessages(sessionId: string): Promise { @@ -1714,6 +1902,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta if (activity.value === 'idle' && !rawState.inFlightBySession[sid]) { const ok = await submitPromptInternal(sid, merged, mergedAttachments); if (!ok) restoreQueue(); + else tombstoneQueuedIds(sid, queue.map((q) => q.id)); return; } @@ -1761,6 +1950,11 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta swarmMode: rawState.swarmModeBySession[sid] ?? false, }); + // The merged queue entries are consumed by this prompt no matter how + // the steer below turns out — tombstone them so no other tab + // re-flushes them. + tombstoneQueuedIds(sid, queue.map((q) => q.id)); + // 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 @@ -1819,9 +2013,17 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const sid = rawState.activeSessionId; if (!sid) return; const current = rawState.queuedBySession[sid] ?? []; + // The id is the entry's stable identity for cross-tab merge/dedupe; the + // timestamp orders merged snapshots (and can be re-stamped by reorder). + const entry: QueuedPromptEntry = { + text, + attachments, + id: nextQueueEntryId(), + enqueuedAt: Date.now(), + }; rawState.queuedBySession = { ...rawState.queuedBySession, - [sid]: [...current, { text, attachments }], + [sid]: [...current, entry], }; } @@ -1832,23 +2034,46 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta * abort); the next turn end or the next idle sendPrompt drives the next * attempt. After MAX_QUEUE_FLUSH_FAILURES consecutive failures the entry * is dropped instead, so one permanently rejected prompt cannot wedge the - * queue. Every failure is already surfaced via pushOperationFailure. + * 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. + * + * Cross-tab bookkeeping: the popped id is marked in-flight for the whole + * submit window so a concurrent adopt can't resurrect it early; success + * (or a final drop) moves it into the shared removal set, and a failed + * submit re-queues it under the SAME id, which the merge dedupes. */ function flushQueueHead(sid: string): void { const [next, ...rest] = rawState.queuedBySession[sid] ?? []; if (next === undefined) return; rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: rest }; + if (next.id !== undefined) { + const submitting = queueInFlightSubmits.get(sid) ?? new Set(); + submitting.add(next.id); + queueInFlightSubmits.set(sid, submitting); + } void submitPromptInternal(sid, next.text, next.attachments).then((ok) => { + if (next.id !== undefined) { + const submitting = queueInFlightSubmits.get(sid); + submitting?.delete(next.id); + if (submitting?.size === 0) queueInFlightSubmits.delete(sid); + } if (ok) { queueFlushFailures.delete(sid); + tombstoneQueuedIds(sid, [next.id]); return; } - const failures = (queueFlushFailures.get(sid) ?? 0) + 1; - if (failures >= MAX_QUEUE_FLUSH_FAILURES) { + // 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); + tombstoneQueuedIds(sid, [next.id]); return; } - queueFlushFailures.set(sid, failures); + queueFlushFailures.set(sid, { key, count }); const current = rawState.queuedBySession[sid] ?? []; rawState.queuedBySession = { ...rawState.queuedBySession, @@ -2581,6 +2806,8 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta if (!sid) return; const current = rawState.queuedBySession[sid] ?? []; if (index < 0 || index >= current.length) return; + // Tombstone first so a concurrent cross-tab merge can't resurrect it. + tombstoneQueuedIds(sid, [current[index]?.id]); const next = [...current]; next.splice(index, 1); rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: next }; @@ -2589,6 +2816,8 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta /** * Move a queued message within the active session's queue (drag-to-reorder). * Defensive: no-op if indices are equal, out of range, or no active session. + * Re-stamps enqueuedAt in display order — the merge sorts by it, so the + * user's order survives cross-tab adoption. */ function reorderQueue(from: number, to: number): void { const sid = rawState.activeSessionId; @@ -2600,7 +2829,11 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const [moved] = next.splice(from, 1); if (moved === undefined) return; next.splice(to, 0, moved); - rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: next }; + const base = Date.now(); + rawState.queuedBySession = { + ...rawState.queuedBySession, + [sid]: next.map((entry, index) => ({ ...entry, enqueuedAt: base + index })), + }; } /** @@ -2797,6 +3030,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta unqueue, reorderQueue, syncQueuedPromptsFromStorage, + discardQueuedPrompts, abortCurrentPrompt, respondApproval, respondQuestion, diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 5de057a85c..0caf9c916d 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -296,10 +296,15 @@ export type PromptAttachment = { }; /** A prompt waiting for the session to go idle. Keeps the uploaded - fileIds so attachments survive queueing (not just the text). */ + fileIds so attachments survive queueing (not just the text). The id is + the entry's stable identity for cross-tab merge/dedupe; enqueuedAt orders + merged snapshots and is re-stamped on manual reorder. Both are assigned + at enqueue time and persisted. */ interface QueuedPrompt { text: string; attachments?: PromptAttachment[]; + id?: string; + enqueuedAt?: number; } export interface ExtendedState extends KimiClientState { @@ -625,9 +630,10 @@ function forgetSession(sessionId: string): void { // In-flight / queued prompt state: drop these too so a queued follow-up // can't be submitted to a session that was just archived when its turn later // ends (onMainTurnEnd drains queuedBySession[sid] without re-checking - // that the session still exists). + // that the session still exists). discardQueuedPrompts also tombstones the + // entries in the shared snapshot so another tab can't merge them back. forgetLocalTurnState(sessionId); - delete rawState.queuedBySession[sessionId]; + workspaceState.discardQueuedPrompts(sessionId); delete rawState.promptIdBySession[sessionId]; delete rawState.inFlightBySession[sessionId]; delete rawState.turnActiveBySession[sessionId]; diff --git a/apps/kimi-web/test/workspace-state.test.ts b/apps/kimi-web/test/workspace-state.test.ts index b3c087ca6a..33c305dfe6 100644 --- a/apps/kimi-web/test/workspace-state.test.ts +++ b/apps/kimi-web/test/workspace-state.test.ts @@ -959,7 +959,7 @@ describe('useWorkspaceState — createGoal from an empty composer', () => { // 'running'), sendPrompt queues rather than posting immediately. expect(apiMock.submitPrompt).not.toHaveBeenCalled(); expect(state.queuedBySession['sess_1']).toEqual([ - { text: 'improve test coverage', attachments: undefined }, + expect.objectContaining({ text: 'improve test coverage', attachments: undefined }), ]); }); @@ -1574,7 +1574,9 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { expect(state.inFlightBySession.sess_1).toBe(true); expect(apiMock.submitPrompt).not.toHaveBeenCalled(); - expect(state.queuedBySession.sess_1).toEqual([{ text: 'next', attachments: undefined }]); + expect(state.queuedBySession.sess_1).toEqual([ + expect.objectContaining({ text: 'next', attachments: undefined }), + ]); }); it('drains one queued prompt when only background work remains', async () => { @@ -1644,7 +1646,9 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { 'sess_1', expect.objectContaining({ content: [{ type: 'text', text: 'stuck queued' }] }), ); - expect(state.queuedBySession.sess_1).toEqual([{ text: 'next', attachments: undefined }]); + expect(state.queuedBySession.sess_1).toEqual([ + expect.objectContaining({ text: 'next', attachments: undefined }), + ]); }); it('re-queues a failed flush at the head and drops it after repeated failures', async () => { @@ -1853,57 +1857,237 @@ describe('useWorkspaceState — prompt queue persistence', () => { // way the facade wraps rawState in production. const state = reactive(createState()); const ws = useWorkspaceState(state, deps()); + const readStored = () => + JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue) ?? '{}') as Record< + string, + { entries: Array>; gone?: Record } + >; + const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); - // A send while the session is running lands in the queue AND in storage. + // A send while the session is running lands in the queue AND in storage, + // stamped with a stable id for cross-tab merge/dedupe. await ws.sendPrompt('queued while running', [{ fileId: 'f_q', kind: 'image' }]); await nextTick(); - expect(JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!)).toEqual({ - sess_1: [{ text: 'queued while running', attachments: [{ fileId: 'f_q', kind: 'image' }] }], - }); - - // A witnessed turn end flushes the head — the persisted queue drains too. + let stored = readStored(); + expect( + stored.sess_1?.entries.map((e) => ({ text: e.text, attachments: e.attachments })), + ).toEqual([{ text: 'queued while running', attachments: [{ fileId: 'f_q', kind: 'image' }] }]); + const firstId = stored.sess_1?.entries[0]?.id as string; + expect(firstId).toBeTruthy(); + + // A witnessed turn end flushes the head — the entry leaves the stored + // queue and its id lands in the shared removal set. ws.finishPromptLocal('sess_1', { turnWasActive: true }); await nextTick(); - expect(JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!)).toEqual({}); + await settle(); + stored = readStored(); + expect(stored.sess_1?.entries ?? []).toEqual([]); + expect(Object.keys(stored.sess_1?.gone ?? {})).toEqual([firstId]); - // Deleting a session key (the facade's session-forget path) syncs via the - // same deep watch. + // The facade's session-forget path discards the queue too. state.queuedBySession = { sess_1: [{ text: 'a', attachments: undefined }], sess_2: [{ text: 'b', attachments: undefined }], }; await nextTick(); - delete state.queuedBySession.sess_1; + ws.discardQueuedPrompts('sess_1'); await nextTick(); - expect(JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!)).toEqual({ - sess_2: [{ text: 'b' }], - }); + stored = readStored(); + expect(stored.sess_2?.entries.map((e) => e.text)).toEqual(['b']); + expect(stored.sess_1?.entries ?? []).toEqual([]); }); - it('adopts the queue another tab persisted, so two tabs never flush the same entries twice', () => { + it('union-merges a queue another tab persisted instead of discarding concurrent entries', async () => { + const state = reactive(createState()); + state.queuedBySession = { + sess_1: [{ text: 'from tab A', attachments: undefined, id: 'id-a', enqueuedAt: 1000 }], + }; + const ws = useWorkspaceState(state, deps()); + + // Tab B persisted its own entry close in time — tab A's write lost the + // race in storage, but A's entry must not be silently discarded. + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: { + entries: [{ text: 'from tab B', attachments: undefined, id: 'id-b', enqueuedAt: 2000 }], + }, + }), + ); + ws.syncQueuedPromptsFromStorage(); + + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['from tab A', 'from tab B']); + await nextTick(); + // The merged queue converges back into storage for tab B to adopt. + const stored = JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!) as Record< + string, + { entries: Array<{ text: string }> } + >; + expect(stored.sess_1?.entries.map((e) => e.text)).toEqual(['from tab A', 'from tab B']); + }); + + it('drops entries the shared removal set says another tab already flushed', () => { const state = createState(); - state.queuedBySession = { sess_1: [{ text: 'stale local', attachments: undefined }] }; + state.queuedBySession = { + sess_1: [{ text: 'stale local', attachments: undefined, id: 'id-x', enqueuedAt: 1000 }], + }; const ws = useWorkspaceState(state, deps()); - // Simulate another tab flushing part of the queue and persisting the rest. localStorage.setItem( STORAGE_KEYS.promptQueue, - JSON.stringify({ sess_1: [{ text: 'newer from tab B' }] }), + JSON.stringify({ sess_1: { entries: [], gone: { 'id-x': Date.now() } } }), ); ws.syncQueuedPromptsFromStorage(); - expect(state.queuedBySession).toEqual({ sess_1: [{ text: 'newer from tab B' }] }); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); }); - it('clears the in-memory queue when another tab drained it', () => { + it('hydrates the v2 snapshot shape and filters entries already removed elsewhere', () => { + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: { + entries: [ + { text: 'kept', id: 'id-keep', enqueuedAt: 1 }, + { text: 'gone', id: 'id-gone', enqueuedAt: 2 }, + ], + gone: { 'id-gone': Date.now() }, + }, + }), + ); const state = createState(); - state.queuedBySession = { sess_1: [{ text: 'stale local', attachments: undefined }] }; + useWorkspaceState(state, deps()); + + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['kept']); + }); + + it('does not resurrect an entry this tab already flushed when adopting a stale snapshot', async () => { + const state = reactive(createState()); + state.queuedBySession = { + sess_1: [{ text: 'mine', attachments: undefined, id: 'id-mine', enqueuedAt: 1000 }], + }; const ws = useWorkspaceState(state, deps()); - localStorage.setItem(STORAGE_KEYS.promptQueue, JSON.stringify({})); + // This tab flushes its entry successfully... + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + + // ...but a snapshot from another tab (written with the removal set + // preserved) still holds the entry. It must not resurrect and flush twice. + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: { + entries: [{ text: 'mine', attachments: undefined, id: 'id-mine', enqueuedAt: 1000 }], + gone: { 'id-mine': Date.now() }, + }, + }), + ); + ws.syncQueuedPromptsFromStorage(); + + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + }); + + it('keeps a submitting entry out of adopted snapshots until its submit resolves', async () => { + let rejectSubmit!: (err: Error) => void; + apiMock.submitPrompt.mockImplementation( + () => + new Promise<{ promptId: string }>((_resolve, reject) => { + rejectSubmit = reject; + }), + ); + const state = createState(); + state.queuedBySession = { + sess_1: [{ text: 'submitting', attachments: undefined, id: 'id-busy', enqueuedAt: 1000 }], + }; + const ws = useWorkspaceState(state, deps()); + + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + + // A stale snapshot still containing the entry must not bring it back + // while its submit is in flight. + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: { + entries: [{ text: 'submitting', attachments: undefined, id: 'id-busy', enqueuedAt: 1000 }], + }, + }), + ); + ws.syncQueuedPromptsFromStorage(); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + + // The submit fails: the entry is re-queued ONCE under the same id, and a + // follow-up adopt still holds exactly one copy (dedupe by id). + rejectSubmit(new Error('turn.agent_busy')); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['submitting']); + ws.syncQueuedPromptsFromStorage(); + expect(state.queuedBySession.sess_1?.length).toBe(1); + }); + + it('resets the flush failure budget when the queue head changes', async () => { + apiMock.submitPrompt.mockRejectedValue(new Error('turn.agent_busy')); + const state = createState(); + state.queuedBySession = { + sess_1: [ + { text: 'first', attachments: undefined, id: 'id-first', enqueuedAt: 1 }, + { text: 'second', attachments: undefined, id: 'id-second', enqueuedAt: 2 }, + ], + }; + const ws = useWorkspaceState(state, deps()); + const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); + const flushOnce = async () => { + state.inFlightBySession = { sess_1: true }; + ws.handleSessionSnapshot('sess_1', { inFlightTurn: null, busy: false }); + await settle(); + }; + + // 'first' fails once, then the user discards it. + await flushOnce(); + ws.unqueue(0); + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['second']); + + // 'second' gets its OWN budget: two failures leave it queued... + await flushOnce(); + await flushOnce(); + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['second']); + // ...and only the third consecutive failure drops it. + await flushOnce(); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + }); + + it('keeps a manual reorder stable across cross-tab adoption', () => { + const state = createState(); + state.queuedBySession = { + sess_1: [ + { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1 }, + { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 2 }, + ], + }; + const ws = useWorkspaceState(state, deps()); + + ws.reorderQueue(1, 0); + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['b', 'a']); + + // Adopting an older snapshot must not un-reorder: reorder re-stamps + // enqueuedAt, and the merge sorts by it (dedupe by id). + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: { + entries: [ + { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1 }, + { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 2 }, + ], + }, + }), + ); ws.syncQueuedPromptsFromStorage(); - expect(state.queuedBySession).toEqual({}); + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['b', 'a']); }); }); From 56889eb5a38f56bc250e078dc426af7f87cb939f Mon Sep 17 00:00:00 2001 From: qer Date: Fri, 17 Jul 2026 18:06:51 +0800 Subject: [PATCH 3/6] fix(kimi-web): single-flusher queue entries, merge order convergence, forget guard (review round 2) Turn-end events reach every open tab and the server accepts concurrent submissions as distinct prompts, so two tabs holding the same adopted entry could both submit it. Entries now record their owner tab and only the owner flushes (ownerless legacy entries flush anywhere); an idle send adopts entries left behind by closed tabs so a stranded queue can still drain. Cross-tab merges now keep the newer enqueuedAt copy per id so a manual reorder converges instead of ping-ponging writes, and the flush failure callback no longer resurrects a queue whose session was forgotten while the submit was pending. --- .../composables/client/useWorkspaceState.ts | 81 ++++++++++++-- .../src/composables/useKimiWebClient.ts | 6 +- apps/kimi-web/test/workspace-state.test.ts | 100 ++++++++++++++++++ 3 files changed, 175 insertions(+), 12 deletions(-) diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index e73d29abe8..2cd244074c 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -146,6 +146,16 @@ const MAX_QUEUE_FLUSH_FAILURES = 3; */ const queueInFlightSubmits = new Map>(); +/** + * Per-page-load tab identity. Queue entries record their owner tab; only the + * owner tab flushes an entry (turn-end events reach EVERY tab, and the + * server happily accepts concurrent submissions as distinct prompts — so + * flush ownership is what stops two tabs from double-submitting a shared + * entry). Stored in memory, not persisted: after a refresh the old owner is + * gone and entries become adoptable by the next tab the user sends from. + */ +const TAB_ID = `tab_${globalThis.crypto?.randomUUID?.() ?? Math.random().toString(36).slice(2)}`; + /** How long a removal stays remembered in the shared snapshot (per id). */ const QUEUE_GONE_TTL_MS = 30 * 60 * 1000; @@ -154,6 +164,7 @@ type QueuedPromptEntry = { attachments?: PromptAttachment[]; id?: string; enqueuedAt?: number; + ownerTabId?: string; }; /** @@ -211,23 +222,32 @@ function mergeQueuedSnapshots( for (const sid of sids) { const gone = goneBySid[sid] ?? {}; const submitting = queueInFlightSubmits.get(sid); - const seenIds = new Set(); + const byId = new Map(); const seenContent = new Set(); - const merged: QueuedPromptEntry[] = []; + const idless: QueuedPromptEntry[] = []; for (const entry of [...(a[sid] ?? []), ...(b[sid] ?? [])]) { if (entry.id !== undefined) { - if (seenIds.has(entry.id) || gone[entry.id] !== undefined || submitting?.has(entry.id)) { - continue; + if (gone[entry.id] !== undefined || submitting?.has(entry.id)) continue; + // Prefer the copy with the NEWER enqueuedAt. A manual reorder + // re-stamps timestamps, so max-timestamp-per-id is the deterministic + // winner every tab converges to — keeping the local copy + // unconditionally would ping-pong the order between tabs forever. + const prev = byId.get(entry.id); + if ( + prev === undefined || + (entry.enqueuedAt ?? Number.NEGATIVE_INFINITY) > + (prev.enqueuedAt ?? Number.NEGATIVE_INFINITY) + ) { + byId.set(entry.id, entry); } - seenIds.add(entry.id); } else { const contentKey = `${entry.text}\n${JSON.stringify(entry.attachments ?? null)}`; if (seenContent.has(contentKey)) continue; seenContent.add(contentKey); + idless.push(entry); } - merged.push(entry); } - const sorted = sortQueuedEntries(merged); + const sorted = sortQueuedEntries([...byId.values(), ...idless]); if (sorted.length > 0) out[sid] = sorted; } return out; @@ -269,10 +289,17 @@ function loadQueueSnapshots(): Record { function isQueuedPromptLike(value: unknown): value is QueuedPromptEntry { if (!value || typeof value !== 'object') return false; - const entry = value as { text?: unknown; attachments?: unknown; id?: unknown; enqueuedAt?: unknown }; + const entry = value as { + text?: unknown; + attachments?: unknown; + id?: unknown; + enqueuedAt?: unknown; + ownerTabId?: unknown; + }; if (typeof entry.text !== 'string') return false; if (entry.id !== undefined && typeof entry.id !== 'string') return false; if (entry.enqueuedAt !== undefined && typeof entry.enqueuedAt !== 'number') return false; + if (entry.ownerTabId !== undefined && typeof entry.ownerTabId !== 'string') return false; if (entry.attachments === undefined) return true; if (!Array.isArray(entry.attachments)) return false; return entry.attachments.every(isPromptAttachmentLike); @@ -569,7 +596,12 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta */ function discardQueuedPrompts(sid: string): void { const current = rawState.queuedBySession[sid] ?? []; - tombstoneQueuedIds(sid, current.map((entry) => entry.id)); + // Tombstone in-flight submits of this session too: their .then callback + // runs later and (on failure) must not resurrect the entry — the shared + // removal set also tells every other tab to drop it. + const inFlight = queueInFlightSubmits.get(sid); + queueInFlightSubmits.delete(sid); + tombstoneQueuedIds(sid, [...current.map((entry) => entry.id), ...(inFlight ?? [])]); if (rawState.queuedBySession[sid] !== undefined) { const next = { ...rawState.queuedBySession }; delete next[sid]; @@ -1853,6 +1885,19 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // stuck entries without a flush driver. if ((rawState.queuedBySession[sid]?.length ?? 0) > 0) { enqueue(text, attachments); + // Take over entries enqueued by other (possibly long-closed) tabs so + // this user-initiated send can actually flush them — only the owner + // tab flushes, so a foreign head would otherwise strand the queue. + // Adoption happens ONLY here, never on passive sync, which would let + // two open tabs steal entries back and forth. + rawState.queuedBySession = { + ...rawState.queuedBySession, + [sid]: (rawState.queuedBySession[sid] ?? []).map((entry) => + entry.ownerTabId !== undefined && entry.ownerTabId !== TAB_ID + ? { ...entry, ownerTabId: TAB_ID } + : entry, + ), + }; flushQueueHead(sid); return; } @@ -2014,12 +2059,14 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta if (!sid) return; const current = rawState.queuedBySession[sid] ?? []; // The id is the entry's stable identity for cross-tab merge/dedupe; the - // timestamp orders merged snapshots (and can be re-stamped by reorder). + // timestamp orders merged snapshots (and can be re-stamped by reorder); + // the owner tab is the only one allowed to flush it. const entry: QueuedPromptEntry = { text, attachments, id: nextQueueEntryId(), enqueuedAt: Date.now(), + ownerTabId: TAB_ID, }; rawState.queuedBySession = { ...rawState.queuedBySession, @@ -2046,6 +2093,12 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta function flushQueueHead(sid: string): void { const [next, ...rest] = rawState.queuedBySession[sid] ?? []; if (next === undefined) return; + // Only the tab that enqueued an entry flushes it. Turn-end events reach + // every tab, and the server accepts concurrent submissions as distinct + // prompts — without this ownership check, two tabs holding the same + // adopted entry would both submit it. Ownerless entries (legacy + // persisted data, or tests) flush anywhere. + if (next.ownerTabId !== undefined && next.ownerTabId !== TAB_ID) return; rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: rest }; if (next.id !== undefined) { const submitting = queueInFlightSubmits.get(sid) ?? new Set(); @@ -2063,6 +2116,14 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta tombstoneQueuedIds(sid, [next.id]); return; } + // The session was forgotten (e.g. archived) while the submit was + // pending — its queue was already discarded, so don't resurrect the + // entry; just make sure no other tab picks it up either. + if (!rawState.sessions.some((s) => s.id === sid)) { + queueFlushFailures.delete(sid); + tombstoneQueuedIds(sid, [next.id]); + 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; diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 0caf9c916d..d712fe180e 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -298,13 +298,15 @@ export type PromptAttachment = { /** A prompt waiting for the session to go idle. Keeps the uploaded fileIds so attachments survive queueing (not just the text). The id is the entry's stable identity for cross-tab merge/dedupe; enqueuedAt orders - merged snapshots and is re-stamped on manual reorder. Both are assigned - at enqueue time and persisted. */ + merged snapshots and is re-stamped on manual reorder; ownerTabId marks + the only tab allowed to flush it (turn-end events reach every tab). + All three are assigned at enqueue time and persisted. */ interface QueuedPrompt { text: string; attachments?: PromptAttachment[]; id?: string; enqueuedAt?: number; + ownerTabId?: string; } export interface ExtendedState extends KimiClientState { diff --git a/apps/kimi-web/test/workspace-state.test.ts b/apps/kimi-web/test/workspace-state.test.ts index 33c305dfe6..dc8706b224 100644 --- a/apps/kimi-web/test/workspace-state.test.ts +++ b/apps/kimi-web/test/workspace-state.test.ts @@ -2089,6 +2089,106 @@ describe('useWorkspaceState — prompt queue persistence', () => { expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['b', 'a']); }); + + it('does not flush a queue entry owned by another tab on a witnessed turn end', () => { + const state = createState(); + state.queuedBySession = { + sess_1: [ + { text: 'foreign', attachments: undefined, id: 'id-foreign', enqueuedAt: 1, ownerTabId: 'tab_other' }, + ], + }; + const ws = useWorkspaceState(state, deps()); + + // Both tabs witness the same turn end; only the owner tab may submit — + // the server would otherwise accept both as distinct prompts. + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + + expect(apiMock.submitPrompt).not.toHaveBeenCalled(); + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['foreign']); + }); + + it('adopts foreign-owned entries on an idle send so a stranded queue can flush', async () => { + const state = createState(); + state.queuedBySession = { + sess_1: [ + { text: 'foreign head', attachments: undefined, id: 'id-f', enqueuedAt: 1, ownerTabId: 'tab_other' }, + ], + }; + const ws = useWorkspaceState(state, deps({ activity: computed(() => 'idle') })); + + await ws.sendPrompt('mine'); + + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + expect(apiMock.submitPrompt).toHaveBeenCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'foreign head' }] }), + ); + // Ownership transferred to this tab — nothing stays owned by the dead tab. + expect(state.queuedBySession.sess_1?.every((e) => e.ownerTabId !== 'tab_other')).toBe(true); + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['mine']); + }); + + it('prefers the newer enqueuedAt per id so a reorder converges instead of ping-ponging', async () => { + const state = reactive(createState()); + state.queuedBySession = { + sess_1: [ + { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1 }, + { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 2 }, + ], + }; + const ws = useWorkspaceState(state, deps()); + + // Another tab reordered (b first — reorder re-stamps display order as + // ascending timestamps, so b carries the SMALLER newer stamp) and + // persisted that; this tab must adopt the newer order even though its + // own copy disagrees. + localStorage.setItem( + STORAGE_KEYS.promptQueue, + JSON.stringify({ + sess_1: { + entries: [ + { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 1001 }, + { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1002 }, + ], + }, + }), + ); + ws.syncQueuedPromptsFromStorage(); + + expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['b', 'a']); + // ...and convergence is stable: a second adopt writes nothing new. + await nextTick(); + const once = localStorage.getItem(STORAGE_KEYS.promptQueue); + ws.syncQueuedPromptsFromStorage(); + await nextTick(); + expect(localStorage.getItem(STORAGE_KEYS.promptQueue)).toBe(once); + }); + + it('does not resurrect the queue when a submit fails after the session was forgotten', async () => { + let rejectSubmit!: (err: Error) => void; + apiMock.submitPrompt.mockImplementation( + () => + new Promise<{ promptId: string }>((_resolve, reject) => { + rejectSubmit = reject; + }), + ); + const state = createState(); + state.queuedBySession = { + sess_1: [{ text: 'doomed', attachments: undefined, id: 'id-doomed', enqueuedAt: 1 }], + }; + const ws = useWorkspaceState(state, deps()); + + ws.finishPromptLocal('sess_1', { turnWasActive: true }); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + + // Session archived (facade forget path) while the submit is pending. + state.sessions = []; + ws.discardQueuedPrompts('sess_1'); + rejectSubmit(new Error('network down')); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(state.queuedBySession.sess_1).toBeUndefined(); + }); }); // Regression: a search-triggered full session-list reload must not clobber the From 9781b50ac31c77a3ef7275b0cac7082bb3658af9 Mon Sep 17 00:00:00 2001 From: qer Date: Fri, 17 Jul 2026 18:26:29 +0800 Subject: [PATCH 4/6] refactor(kimi-web): drop the cross-tab queue persistence, keep the minimal fix Cross-tab queue sync over localStorage is a distributed-systems problem (claim/lease, conflict merge, ownership) that keeps generating review findings far beyond this PR's scope. Remove the persistence/hydration/ adoption machinery wholesale; keep the bug fix proper: gated queue drain, event-driven FIFO retry with a per-entry failure budget, steer queue restore, and the forgotten-session flush guard. Durable queued prompts will be designed together with the server-side prompt queue. --- .../composables/client/useWorkspaceState.ts | 376 +----------------- .../src/composables/useKimiWebClient.ts | 20 +- apps/kimi-web/src/lib/storage.ts | 2 - apps/kimi-web/test/workspace-state.test.ts | 358 +---------------- 4 files changed, 22 insertions(+), 734 deletions(-) diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index 2cd244074c..e8326e6d03 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -7,7 +7,7 @@ // the view-model computeds stay in the facade; cross-dependencies are injected // here as params. -import { reactive, watch, type ComputedRef, type Ref } from 'vue'; +import { reactive, type ComputedRef, type Ref } from 'vue'; import { getKimiWebApi } from '../../api'; import { i18n } from '../../i18n'; import { useConfirmDialog } from '../useConfirmDialog'; @@ -28,10 +28,7 @@ import type { } from '../../api/types'; import { loadWorkspaceNameOverrides, - safeGetJson, - safeGetString, safeRemove, - safeSetString, saveWorkspaceNameOverrides, STORAGE_KEYS, } from '../../lib/storage'; @@ -121,252 +118,20 @@ const pendingLocalTurnStarts = new Map>(); const afterLocalTurnsSettled = new Map void>(); let nextLocalTurnToken = 0; -/** - * Consecutive flushQueueHead failures per session. A rejected queued prompt - * goes back at the head and waits for the next turn end (or idle send) to - * retry; beyond MAX_QUEUE_FLUSH_FAILURES it is dropped so a permanently - * rejected entry (e.g. one whose uploaded files no longer exist server-side) - * cannot block every later prompt behind it. Module-level singleton — the - * queue itself is per-session on rawState, so a page reload resets both. - */ /** * Consecutive flushQueueHead failures per queue ENTRY (not per session) — * keyed by entry id, falling back to its text for entries without one. * Keying by entry keeps a removed or reordered head from handing its - * failure budget down to the next entry. + * failure budget down to the next entry. Module-level singleton — the queue + * itself is per-session on rawState, so a page reload resets both. */ const queueFlushFailures = new Map(); const MAX_QUEUE_FLUSH_FAILURES = 3; -/** - * Entries this tab popped and is submitting right now: already out of the - * local queue but not yet acknowledged. Cross-tab merges treat these as - * removed so an adopt can't bring a submitting entry back early — a failed - * submit re-queues it under the SAME id and the merge dedupes. - */ -const queueInFlightSubmits = new Map>(); - -/** - * Per-page-load tab identity. Queue entries record their owner tab; only the - * owner tab flushes an entry (turn-end events reach EVERY tab, and the - * server happily accepts concurrent submissions as distinct prompts — so - * flush ownership is what stops two tabs from double-submitting a shared - * entry). Stored in memory, not persisted: after a refresh the old owner is - * gone and entries become adoptable by the next tab the user sends from. - */ -const TAB_ID = `tab_${globalThis.crypto?.randomUUID?.() ?? Math.random().toString(36).slice(2)}`; - -/** How long a removal stays remembered in the shared snapshot (per id). */ -const QUEUE_GONE_TTL_MS = 30 * 60 * 1000; - -type QueuedPromptEntry = { - text: string; - attachments?: PromptAttachment[]; - id?: string; - enqueuedAt?: number; - ownerTabId?: string; -}; - -/** - * Shared (localStorage) per-session queue snapshot. `gone` is the removal - * set — ids this or another tab finished/discarded, with timestamps — and is - * what stops a union merge from RESURRECTING an entry a concurrent tab - * already flushed or discarded (without it, the tab still holding the entry - * would merge it back in and eventually flush it twice). - */ -interface SessionQueueSnapshot { - entries: QueuedPromptEntry[]; - gone?: Record; -} - let queueEntryCounter = 0; function nextQueueEntryId(): string { queueEntryCounter += 1; - return `${Date.now().toString(36)}-${queueEntryCounter}-${Math.random().toString(36).slice(2, 8)}`; -} - -function pruneGoneMap(gone: Record | undefined): Record { - if (gone === undefined) return {}; - const cutoff = Date.now() - QUEUE_GONE_TTL_MS; - const out: Record = {}; - for (const [id, at] of Object.entries(gone)) { - if (at >= cutoff) out[id] = at; - } - return out; -} - -/** Stable, deterministic order for a session's queue. Manual reorder - * re-stamps enqueuedAt, so a user's drag order survives sorting. */ -function sortQueuedEntries(entries: QueuedPromptEntry[]): QueuedPromptEntry[] { - return entries.toSorted( - (a, b) => (a.enqueuedAt ?? Number.POSITIVE_INFINITY) - (b.enqueuedAt ?? Number.POSITIVE_INFINITY), - ); -} - -/** - * Union-merge two per-session entry maps by entry id. Merging instead of - * last-writer-wins replacement is what keeps two tabs enqueueing different - * prompts close together from silently discarding each other's entry: the - * tab that lost the storage write still holds its entry in memory, the - * adopt merges it back, and the resulting watch write converges both tabs. - * Entries without an id (legacy persisted data) dedupe by content. Ids in - * `goneBySid` or submitting in this tab are filtered out. - */ -function mergeQueuedSnapshots( - a: Record, - b: Record, - goneBySid: Record>, -): Record { - const out: Record = {}; - const sids = new Set([...Object.keys(a), ...Object.keys(b)]); - for (const sid of sids) { - const gone = goneBySid[sid] ?? {}; - const submitting = queueInFlightSubmits.get(sid); - const byId = new Map(); - const seenContent = new Set(); - const idless: QueuedPromptEntry[] = []; - for (const entry of [...(a[sid] ?? []), ...(b[sid] ?? [])]) { - if (entry.id !== undefined) { - if (gone[entry.id] !== undefined || submitting?.has(entry.id)) continue; - // Prefer the copy with the NEWER enqueuedAt. A manual reorder - // re-stamps timestamps, so max-timestamp-per-id is the deterministic - // winner every tab converges to — keeping the local copy - // unconditionally would ping-pong the order between tabs forever. - const prev = byId.get(entry.id); - if ( - prev === undefined || - (entry.enqueuedAt ?? Number.NEGATIVE_INFINITY) > - (prev.enqueuedAt ?? Number.NEGATIVE_INFINITY) - ) { - byId.set(entry.id, entry); - } - } else { - const contentKey = `${entry.text}\n${JSON.stringify(entry.attachments ?? null)}`; - if (seenContent.has(contentKey)) continue; - seenContent.add(contentKey); - idless.push(entry); - } - } - const sorted = sortQueuedEntries([...byId.values(), ...idless]); - if (sorted.length > 0) out[sid] = sorted; - } - return out; -} - -/** - * The local prompt queue is persisted per session (STORAGE_KEYS.promptQueue) - * so messages the user already sent — but that only made it as far as the - * local queue because a turn was still running — survive a page refresh - * instead of vanishing from memory. A restored queue is display-only until a - * real flush driver runs (finishPromptLocal's drain gate requires a locally - * witnessed turn, and sendPrompt flushes FIFO), so it can never ghost-send - * just because a session was re-opened. - */ -function loadQueueSnapshots(): Record { - const parsed = safeGetJson(STORAGE_KEYS.promptQueue); - if (!parsed || typeof parsed !== 'object') return {}; - const out: Record = {}; - for (const [sid, value] of Object.entries(parsed as Record)) { - // Legacy shape: a bare entry array with no removal set. - const raw = Array.isArray(value) - ? { entries: value as unknown[] } - : (value as { entries?: unknown; gone?: unknown } | null); - if (!raw || !Array.isArray(raw.entries)) continue; - const entries = raw.entries.filter(isQueuedPromptLike); - const gone: Record = {}; - if (raw.gone && typeof raw.gone === 'object') { - for (const [id, at] of Object.entries(raw.gone as Record)) { - if (typeof at === 'number') gone[id] = at; - } - } - const pruned = pruneGoneMap(gone); - if (entries.length > 0 || Object.keys(pruned).length > 0) { - out[sid] = { entries, gone: pruned }; - } - } - return out; -} - -function isQueuedPromptLike(value: unknown): value is QueuedPromptEntry { - if (!value || typeof value !== 'object') return false; - const entry = value as { - text?: unknown; - attachments?: unknown; - id?: unknown; - enqueuedAt?: unknown; - ownerTabId?: unknown; - }; - if (typeof entry.text !== 'string') return false; - if (entry.id !== undefined && typeof entry.id !== 'string') return false; - if (entry.enqueuedAt !== undefined && typeof entry.enqueuedAt !== 'number') return false; - if (entry.ownerTabId !== undefined && typeof entry.ownerTabId !== 'string') return false; - if (entry.attachments === undefined) return true; - if (!Array.isArray(entry.attachments)) return false; - return entry.attachments.every(isPromptAttachmentLike); -} - -/** Entries view of the shared snapshots (removal-set ids filtered out). */ -function queuedEntriesFromSnapshots( - snapshots: Record, -): Record { - const out: Record = {}; - for (const [sid, snap] of Object.entries(snapshots)) { - const gone = snap.gone ?? {}; - const entries = sortQueuedEntries( - snap.entries.filter((entry) => entry.id === undefined || gone[entry.id] === undefined), - ); - if (entries.length > 0) out[sid] = entries; - } - return out; -} - -function isPromptAttachmentLike(value: unknown): value is PromptAttachment { - if (!value || typeof value !== 'object') return false; - const att = value as { fileId?: unknown; kind?: unknown }; - return ( - typeof att.fileId === 'string' && - (att.kind === 'image' || att.kind === 'video' || att.kind === 'file') - ); -} - -function writeQueueSnapshots(snapshots: Record): void { - const out: Record = {}; - for (const [sid, snap] of Object.entries(snapshots)) { - const entries = sortQueuedEntries(snap.entries); - const gone = pruneGoneMap(snap.gone); - if (entries.length === 0 && Object.keys(gone).length === 0) continue; - out[sid] = { entries, ...(Object.keys(gone).length > 0 ? { gone } : {}) }; - } - const serialized = JSON.stringify(out); - // Skip the write when nothing changed — otherwise two tabs' writes would - // ping-pong storage events forever. - if (safeGetString(STORAGE_KEYS.promptQueue) === serialized) return; - safeSetString(STORAGE_KEYS.promptQueue, serialized); -} - -/** Persist the current in-memory queue, preserving stored removal sets. */ -function saveQueuedPrompts(queue: Record): void { - const stored = loadQueueSnapshots(); - for (const [sid, snap] of Object.entries(stored)) { - snap.entries = queue[sid] ?? []; - } - for (const [sid, entries] of Object.entries(queue)) { - stored[sid] ??= { entries }; - } - writeQueueSnapshots(stored); -} - -/** Read-modify-write the shared removal set for `sid`. */ -function tombstoneQueuedIds(sid: string, ids: Array): void { - const realIds = ids.filter((id): id is string => id !== undefined); - if (realIds.length === 0) return; - const stored = loadQueueSnapshots(); - const snap = stored[sid] ?? { entries: [] }; - const gone = { ...snap.gone }; - const now = Date.now(); - for (const id of realIds) gone[id] = now; - stored[sid] = { ...snap, gone }; - writeQueueSnapshots(stored); + return `${Date.now().toString(36)}-${queueEntryCounter}`; } export interface LocalTurnStartState { @@ -547,68 +312,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } = deps; let exportInFlight = false; - // Hydrate the local prompt queue from storage so queued prompts survive a - // page refresh. Only fills an empty in-memory queue — a live one is never - // clobbered. Entries already removed elsewhere (in the shared `gone` set) - // are filtered out. - const storedSnapshots = loadQueueSnapshots(); - const storedQueue = queuedEntriesFromSnapshots(storedSnapshots); - if (Object.keys(rawState.queuedBySession).length === 0 && Object.keys(storedQueue).length > 0) { - rawState.queuedBySession = storedQueue; - } - // Every queue mutation is a whole-record replace or a key delete, so one - // deep watch keeps storage in sync — including the facade's session-forget - // delete, which bypasses this module entirely. - watch( - () => rawState.queuedBySession, - (queue) => { - saveQueuedPrompts(queue); - }, - { deep: true }, - ); - - /** - * Multi-tab convergence: UNION-merge the queue another tab just persisted - * with our own by entry id (never last-writer-wins replacement — that - * would silently discard whichever tab lost a concurrent write). Entries - * in the shared `gone` set or currently submitting here are left out, so a - * flushed/discarded entry can't resurrect and be flushed twice. The watch - * above then persists the merged result, converging both tabs. - */ - function syncQueuedPromptsFromStorage(): void { - const snapshots = loadQueueSnapshots(); - const goneBySid: Record> = {}; - for (const [sid, snap] of Object.entries(snapshots)) { - goneBySid[sid] = snap.gone ?? {}; - } - rawState.queuedBySession = mergeQueuedSnapshots( - rawState.queuedBySession, - queuedEntriesFromSnapshots(snapshots), - goneBySid, - ); - } - - /** - * Drop a session's queue for good (the facade's session-forget path, e.g. - * after archive): every current entry lands in the shared removal set so - * another tab's merge can't resurrect it into a session that no longer - * exists. - */ - function discardQueuedPrompts(sid: string): void { - const current = rawState.queuedBySession[sid] ?? []; - // Tombstone in-flight submits of this session too: their .then callback - // runs later and (on failure) must not resurrect the entry — the shared - // removal set also tells every other tab to drop it. - const inFlight = queueInFlightSubmits.get(sid); - queueInFlightSubmits.delete(sid); - tombstoneQueuedIds(sid, [...current.map((entry) => entry.id), ...(inFlight ?? [])]); - if (rawState.queuedBySession[sid] !== undefined) { - const next = { ...rawState.queuedBySession }; - delete next[sid]; - rawState.queuedBySession = next; - } - } - async function loadOlderMessages(sessionId: string): Promise { if (rawState.messagesLoadingMoreBySession[sessionId]) return; const current = rawState.messagesBySession[sessionId]; @@ -1885,19 +1588,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // stuck entries without a flush driver. if ((rawState.queuedBySession[sid]?.length ?? 0) > 0) { enqueue(text, attachments); - // Take over entries enqueued by other (possibly long-closed) tabs so - // this user-initiated send can actually flush them — only the owner - // tab flushes, so a foreign head would otherwise strand the queue. - // Adoption happens ONLY here, never on passive sync, which would let - // two open tabs steal entries back and forth. - rawState.queuedBySession = { - ...rawState.queuedBySession, - [sid]: (rawState.queuedBySession[sid] ?? []).map((entry) => - entry.ownerTabId !== undefined && entry.ownerTabId !== TAB_ID - ? { ...entry, ownerTabId: TAB_ID } - : entry, - ), - }; flushQueueHead(sid); return; } @@ -1947,7 +1637,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta if (activity.value === 'idle' && !rawState.inFlightBySession[sid]) { const ok = await submitPromptInternal(sid, merged, mergedAttachments); if (!ok) restoreQueue(); - else tombstoneQueuedIds(sid, queue.map((q) => q.id)); return; } @@ -1995,11 +1684,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta swarmMode: rawState.swarmModeBySession[sid] ?? false, }); - // The merged queue entries are consumed by this prompt no matter how - // the steer below turns out — tombstone them so no other tab - // re-flushes them. - tombstoneQueuedIds(sid, queue.map((q) => q.id)); - // 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 @@ -2058,16 +1742,9 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const sid = rawState.activeSessionId; if (!sid) return; const current = rawState.queuedBySession[sid] ?? []; - // The id is the entry's stable identity for cross-tab merge/dedupe; the - // timestamp orders merged snapshots (and can be re-stamped by reorder); - // the owner tab is the only one allowed to flush it. - const entry: QueuedPromptEntry = { - text, - attachments, - id: nextQueueEntryId(), - enqueuedAt: Date.now(), - ownerTabId: TAB_ID, - }; + // The id keys the per-entry flush failure budget (removing/reordering + // the head then resets the next entry's budget). + const entry = { text, attachments, id: nextQueueEntryId() }; rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: [...current, entry], @@ -2084,44 +1761,20 @@ 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. - * - * Cross-tab bookkeeping: the popped id is marked in-flight for the whole - * submit window so a concurrent adopt can't resurrect it early; success - * (or a final drop) moves it into the shared removal set, and a failed - * submit re-queues it under the SAME id, which the merge dedupes. */ function flushQueueHead(sid: string): void { const [next, ...rest] = rawState.queuedBySession[sid] ?? []; if (next === undefined) return; - // Only the tab that enqueued an entry flushes it. Turn-end events reach - // every tab, and the server accepts concurrent submissions as distinct - // prompts — without this ownership check, two tabs holding the same - // adopted entry would both submit it. Ownerless entries (legacy - // persisted data, or tests) flush anywhere. - if (next.ownerTabId !== undefined && next.ownerTabId !== TAB_ID) return; rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: rest }; - if (next.id !== undefined) { - const submitting = queueInFlightSubmits.get(sid) ?? new Set(); - submitting.add(next.id); - queueInFlightSubmits.set(sid, submitting); - } void submitPromptInternal(sid, next.text, next.attachments).then((ok) => { - if (next.id !== undefined) { - const submitting = queueInFlightSubmits.get(sid); - submitting?.delete(next.id); - if (submitting?.size === 0) queueInFlightSubmits.delete(sid); - } if (ok) { queueFlushFailures.delete(sid); - tombstoneQueuedIds(sid, [next.id]); return; } // The session was forgotten (e.g. archived) while the submit was - // pending — its queue was already discarded, so don't resurrect the - // entry; just make sure no other tab picks it up either. + // pending — its queue was already discarded, so don't resurrect it. if (!rawState.sessions.some((s) => s.id === sid)) { queueFlushFailures.delete(sid); - tombstoneQueuedIds(sid, [next.id]); return; } // Per-entry budget: a different head (removed/reordered since) starts @@ -2131,7 +1784,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const count = previous !== undefined && previous.key === key ? previous.count + 1 : 1; if (count >= MAX_QUEUE_FLUSH_FAILURES) { queueFlushFailures.delete(sid); - tombstoneQueuedIds(sid, [next.id]); return; } queueFlushFailures.set(sid, { key, count }); @@ -2867,8 +2519,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta if (!sid) return; const current = rawState.queuedBySession[sid] ?? []; if (index < 0 || index >= current.length) return; - // Tombstone first so a concurrent cross-tab merge can't resurrect it. - tombstoneQueuedIds(sid, [current[index]?.id]); const next = [...current]; next.splice(index, 1); rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: next }; @@ -2877,8 +2527,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta /** * Move a queued message within the active session's queue (drag-to-reorder). * Defensive: no-op if indices are equal, out of range, or no active session. - * Re-stamps enqueuedAt in display order — the merge sorts by it, so the - * user's order survives cross-tab adoption. */ function reorderQueue(from: number, to: number): void { const sid = rawState.activeSessionId; @@ -2890,11 +2538,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const [moved] = next.splice(from, 1); if (moved === undefined) return; next.splice(to, 0, moved); - const base = Date.now(); - rawState.queuedBySession = { - ...rawState.queuedBySession, - [sid]: next.map((entry, index) => ({ ...entry, enqueuedAt: base + index })), - }; + rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: next }; } /** @@ -3090,8 +2734,6 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta enqueue, unqueue, reorderQueue, - syncQueuedPromptsFromStorage, - discardQueuedPrompts, abortCurrentPrompt, respondApproval, respondQuestion, diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index d712fe180e..97445231be 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -296,17 +296,12 @@ export type PromptAttachment = { }; /** A prompt waiting for the session to go idle. Keeps the uploaded - fileIds so attachments survive queueing (not just the text). The id is - the entry's stable identity for cross-tab merge/dedupe; enqueuedAt orders - merged snapshots and is re-stamped on manual reorder; ownerTabId marks - the only tab allowed to flush it (turn-end events reach every tab). - All three are assigned at enqueue time and persisted. */ + fileIds so attachments survive queueing (not just the text). The id keys + the per-entry flush failure budget locally (assigned at enqueue). */ interface QueuedPrompt { text: string; attachments?: PromptAttachment[]; id?: string; - enqueuedAt?: number; - ownerTabId?: string; } export interface ExtendedState extends KimiClientState { @@ -512,12 +507,6 @@ if (typeof window !== 'undefined') { rawState.unreadBySession = loadUnread(); clearActiveUnread(); } - // Another tab enqueued/flushed/discarded queued prompts — adopt its queue - // so both tabs don't flush the same entries twice. workspaceState is - // referenced lazily inside the event callback, after module init. - if (event.key === STORAGE_KEYS.promptQueue) { - workspaceState.syncQueuedPromptsFromStorage(); - } }); } @@ -632,10 +621,9 @@ function forgetSession(sessionId: string): void { // In-flight / queued prompt state: drop these too so a queued follow-up // can't be submitted to a session that was just archived when its turn later // ends (onMainTurnEnd drains queuedBySession[sid] without re-checking - // that the session still exists). discardQueuedPrompts also tombstones the - // entries in the shared snapshot so another tab can't merge them back. + // that the session still exists). forgetLocalTurnState(sessionId); - workspaceState.discardQueuedPrompts(sessionId); + delete rawState.queuedBySession[sessionId]; delete rawState.promptIdBySession[sessionId]; delete rawState.inFlightBySession[sessionId]; delete rawState.turnActiveBySession[sessionId]; diff --git a/apps/kimi-web/src/lib/storage.ts b/apps/kimi-web/src/lib/storage.ts index a462ee3cf1..5cd60760c6 100644 --- a/apps/kimi-web/src/lib/storage.ts +++ b/apps/kimi-web/src/lib/storage.ts @@ -34,8 +34,6 @@ export const STORAGE_KEYS = { notifyOnApproval: 'kimi-web.notify-on-approval', soundOnComplete: 'kimi-web.sound-on-complete', inputHistory: 'kimi-web.input-history', - // useWorkspaceState — per-session local prompt queue - promptQueue: 'kimi-web.prompt-queue', // cross-file locale: 'kimi-locale', clientId: 'kimi-web.client-id', diff --git a/apps/kimi-web/test/workspace-state.test.ts b/apps/kimi-web/test/workspace-state.test.ts index dc8706b224..700dbebb3e 100644 --- a/apps/kimi-web/test/workspace-state.test.ts +++ b/apps/kimi-web/test/workspace-state.test.ts @@ -3,13 +3,13 @@ // Wiring: the composable is real; daemon requests and unrelated facade collaborators are stubbed. // Run: pnpm --filter @moonshot-ai/kimi-web exec vitest run test/workspace-state.test.ts -import { computed, nextTick, reactive, ref, type Ref } from 'vue'; +import { computed, ref, type Ref } from 'vue'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { AppApprovalRequest, AppQuestionRequest, AppSession, AppTask } from '../src/api/types'; import { DaemonApiError } from '../src/api/errors'; import { createInitialState } from '../src/api/daemon/eventReducer'; import { mergeWorkspaces } from '../src/lib/mergeWorkspaces'; -import { loadWorkspaceNameOverrides, saveWorkspaceNameOverrides, STORAGE_KEYS } from '../src/lib/storage'; +import { loadWorkspaceNameOverrides, saveWorkspaceNameOverrides } from '../src/lib/storage'; import { useWorkspaceState, forgetLocalTurnState, type UseWorkspaceStateDeps } from '../src/composables/client/useWorkspaceState'; import type { ExtendedState } from '../src/composables/useKimiWebClient'; import { clearTrace, traceKeyEvent } from '../src/debug/trace'; @@ -1792,252 +1792,17 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { }), ); }); -}); - -describe('useWorkspaceState — prompt queue persistence', () => { - function deps(overrides: Partial = {}): UseWorkspaceStateDeps { - return { - ...createDeps(), - modelProvider: { models: ref([]) } as unknown as UseWorkspaceStateDeps['modelProvider'], - ...overrides, - }; - } - - beforeEach(() => { - installStorage(createMemoryStorage()); - apiMock.submitPrompt.mockReset(); - apiMock.submitPrompt.mockResolvedValue({ promptId: 'prompt_new' }); - // Module-level flush failure budget must not leak between tests. - forgetLocalTurnState('sess_1'); - }); - - it('restores a persisted queue on startup so a refresh loses nothing', () => { - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: [{ text: 'old queued', attachments: [{ fileId: 'f_old', kind: 'image' }] }], - }), - ); - const state = createState(); - useWorkspaceState(state, deps()); - - expect(state.queuedBySession.sess_1).toEqual([ - { text: 'old queued', attachments: [{ fileId: 'f_old', kind: 'image' }] }], - ); - }); - - it('keeps only well-formed persisted entries and ignores the rest', () => { - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: [{ text: 5 }, 'nope', { text: 'ok', attachments: [{ fileId: 'f', kind: 'file' }] }], - sess_2: 'not-an-array', - sess_3: [{ text: 'bad kind', attachments: [{ fileId: 'f', kind: 'exe' }] }], - }), - ); - const state = createState(); - useWorkspaceState(state, deps()); - - expect(state.queuedBySession).toEqual({ - sess_1: [{ text: 'ok', attachments: [{ fileId: 'f', kind: 'file' }] }], - }); - }); - - it('does not clobber an in-memory queue with the persisted one', () => { - localStorage.setItem(STORAGE_KEYS.promptQueue, JSON.stringify({ sess_1: [{ text: 'stored' }] })); - const state = createState(); - state.queuedBySession = { sess_1: [{ text: 'in memory', attachments: undefined }] }; - useWorkspaceState(state, deps()); - - expect(state.queuedBySession.sess_1).toEqual([{ text: 'in memory', attachments: undefined }]); - }); - - it('persists enqueue, flush, and session-forget mutations', async () => { - // reactive() so the persistence watch actually tracks the state, the same - // way the facade wraps rawState in production. - const state = reactive(createState()); - const ws = useWorkspaceState(state, deps()); - const readStored = () => - JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue) ?? '{}') as Record< - string, - { entries: Array>; gone?: Record } - >; - const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); - - // A send while the session is running lands in the queue AND in storage, - // stamped with a stable id for cross-tab merge/dedupe. - await ws.sendPrompt('queued while running', [{ fileId: 'f_q', kind: 'image' }]); - await nextTick(); - let stored = readStored(); - expect( - stored.sess_1?.entries.map((e) => ({ text: e.text, attachments: e.attachments })), - ).toEqual([{ text: 'queued while running', attachments: [{ fileId: 'f_q', kind: 'image' }] }]); - const firstId = stored.sess_1?.entries[0]?.id as string; - expect(firstId).toBeTruthy(); - - // A witnessed turn end flushes the head — the entry leaves the stored - // queue and its id lands in the shared removal set. - ws.finishPromptLocal('sess_1', { turnWasActive: true }); - await nextTick(); - await settle(); - stored = readStored(); - expect(stored.sess_1?.entries ?? []).toEqual([]); - expect(Object.keys(stored.sess_1?.gone ?? {})).toEqual([firstId]); - - // The facade's session-forget path discards the queue too. - state.queuedBySession = { - sess_1: [{ text: 'a', attachments: undefined }], - sess_2: [{ text: 'b', attachments: undefined }], - }; - await nextTick(); - ws.discardQueuedPrompts('sess_1'); - await nextTick(); - stored = readStored(); - expect(stored.sess_2?.entries.map((e) => e.text)).toEqual(['b']); - expect(stored.sess_1?.entries ?? []).toEqual([]); - }); - - it('union-merges a queue another tab persisted instead of discarding concurrent entries', async () => { - const state = reactive(createState()); - state.queuedBySession = { - sess_1: [{ text: 'from tab A', attachments: undefined, id: 'id-a', enqueuedAt: 1000 }], - }; - const ws = useWorkspaceState(state, deps()); - - // Tab B persisted its own entry close in time — tab A's write lost the - // race in storage, but A's entry must not be silently discarded. - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: { - entries: [{ text: 'from tab B', attachments: undefined, id: 'id-b', enqueuedAt: 2000 }], - }, - }), - ); - ws.syncQueuedPromptsFromStorage(); - - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['from tab A', 'from tab B']); - await nextTick(); - // The merged queue converges back into storage for tab B to adopt. - const stored = JSON.parse(localStorage.getItem(STORAGE_KEYS.promptQueue)!) as Record< - string, - { entries: Array<{ text: string }> } - >; - expect(stored.sess_1?.entries.map((e) => e.text)).toEqual(['from tab A', 'from tab B']); - }); - - it('drops entries the shared removal set says another tab already flushed', () => { - const state = createState(); - state.queuedBySession = { - sess_1: [{ text: 'stale local', attachments: undefined, id: 'id-x', enqueuedAt: 1000 }], - }; - const ws = useWorkspaceState(state, deps()); - - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ sess_1: { entries: [], gone: { 'id-x': Date.now() } } }), - ); - ws.syncQueuedPromptsFromStorage(); - - expect(state.queuedBySession.sess_1 ?? []).toEqual([]); - }); - - it('hydrates the v2 snapshot shape and filters entries already removed elsewhere', () => { - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: { - entries: [ - { text: 'kept', id: 'id-keep', enqueuedAt: 1 }, - { text: 'gone', id: 'id-gone', enqueuedAt: 2 }, - ], - gone: { 'id-gone': Date.now() }, - }, - }), - ); - const state = createState(); - useWorkspaceState(state, deps()); - - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['kept']); - }); - - it('does not resurrect an entry this tab already flushed when adopting a stale snapshot', async () => { - const state = reactive(createState()); - state.queuedBySession = { - sess_1: [{ text: 'mine', attachments: undefined, id: 'id-mine', enqueuedAt: 1000 }], - }; - const ws = useWorkspaceState(state, deps()); - - // This tab flushes its entry successfully... - ws.finishPromptLocal('sess_1', { turnWasActive: true }); - await new Promise((resolve) => setTimeout(resolve, 0)); - expect(state.queuedBySession.sess_1 ?? []).toEqual([]); - - // ...but a snapshot from another tab (written with the removal set - // preserved) still holds the entry. It must not resurrect and flush twice. - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: { - entries: [{ text: 'mine', attachments: undefined, id: 'id-mine', enqueuedAt: 1000 }], - gone: { 'id-mine': Date.now() }, - }, - }), - ); - ws.syncQueuedPromptsFromStorage(); - - expect(state.queuedBySession.sess_1 ?? []).toEqual([]); - }); - - it('keeps a submitting entry out of adopted snapshots until its submit resolves', async () => { - let rejectSubmit!: (err: Error) => void; - apiMock.submitPrompt.mockImplementation( - () => - new Promise<{ promptId: string }>((_resolve, reject) => { - rejectSubmit = reject; - }), - ); - const state = createState(); - state.queuedBySession = { - sess_1: [{ text: 'submitting', attachments: undefined, id: 'id-busy', enqueuedAt: 1000 }], - }; - const ws = useWorkspaceState(state, deps()); - - ws.finishPromptLocal('sess_1', { turnWasActive: true }); - expect(state.queuedBySession.sess_1 ?? []).toEqual([]); - - // A stale snapshot still containing the entry must not bring it back - // while its submit is in flight. - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: { - entries: [{ text: 'submitting', attachments: undefined, id: 'id-busy', enqueuedAt: 1000 }], - }, - }), - ); - ws.syncQueuedPromptsFromStorage(); - expect(state.queuedBySession.sess_1 ?? []).toEqual([]); - - // The submit fails: the entry is re-queued ONCE under the same id, and a - // follow-up adopt still holds exactly one copy (dedupe by id). - rejectSubmit(new Error('turn.agent_busy')); - await new Promise((resolve) => setTimeout(resolve, 0)); - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['submitting']); - ws.syncQueuedPromptsFromStorage(); - expect(state.queuedBySession.sess_1?.length).toBe(1); - }); it('resets the flush failure budget when the queue head changes', async () => { apiMock.submitPrompt.mockRejectedValue(new Error('turn.agent_busy')); const state = createState(); state.queuedBySession = { sess_1: [ - { text: 'first', attachments: undefined, id: 'id-first', enqueuedAt: 1 }, - { text: 'second', attachments: undefined, id: 'id-second', enqueuedAt: 2 }, + { text: 'first', attachments: undefined, id: 'id-first' }, + { text: 'second', attachments: undefined, id: 'id-second' }, ], }; - const ws = useWorkspaceState(state, deps()); + const ws = useWorkspaceState(state, promptDeps()); const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); const flushOnce = async () => { state.inFlightBySession = { sess_1: true }; @@ -2059,111 +1824,6 @@ describe('useWorkspaceState — prompt queue persistence', () => { expect(state.queuedBySession.sess_1 ?? []).toEqual([]); }); - it('keeps a manual reorder stable across cross-tab adoption', () => { - const state = createState(); - state.queuedBySession = { - sess_1: [ - { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1 }, - { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 2 }, - ], - }; - const ws = useWorkspaceState(state, deps()); - - ws.reorderQueue(1, 0); - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['b', 'a']); - - // Adopting an older snapshot must not un-reorder: reorder re-stamps - // enqueuedAt, and the merge sorts by it (dedupe by id). - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: { - entries: [ - { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1 }, - { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 2 }, - ], - }, - }), - ); - ws.syncQueuedPromptsFromStorage(); - - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['b', 'a']); - }); - - it('does not flush a queue entry owned by another tab on a witnessed turn end', () => { - const state = createState(); - state.queuedBySession = { - sess_1: [ - { text: 'foreign', attachments: undefined, id: 'id-foreign', enqueuedAt: 1, ownerTabId: 'tab_other' }, - ], - }; - const ws = useWorkspaceState(state, deps()); - - // Both tabs witness the same turn end; only the owner tab may submit — - // the server would otherwise accept both as distinct prompts. - ws.finishPromptLocal('sess_1', { turnWasActive: true }); - - expect(apiMock.submitPrompt).not.toHaveBeenCalled(); - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['foreign']); - }); - - it('adopts foreign-owned entries on an idle send so a stranded queue can flush', async () => { - const state = createState(); - state.queuedBySession = { - sess_1: [ - { text: 'foreign head', attachments: undefined, id: 'id-f', enqueuedAt: 1, ownerTabId: 'tab_other' }, - ], - }; - const ws = useWorkspaceState(state, deps({ activity: computed(() => 'idle') })); - - await ws.sendPrompt('mine'); - - expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); - expect(apiMock.submitPrompt).toHaveBeenCalledWith( - 'sess_1', - expect.objectContaining({ content: [{ type: 'text', text: 'foreign head' }] }), - ); - // Ownership transferred to this tab — nothing stays owned by the dead tab. - expect(state.queuedBySession.sess_1?.every((e) => e.ownerTabId !== 'tab_other')).toBe(true); - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['mine']); - }); - - it('prefers the newer enqueuedAt per id so a reorder converges instead of ping-ponging', async () => { - const state = reactive(createState()); - state.queuedBySession = { - sess_1: [ - { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1 }, - { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 2 }, - ], - }; - const ws = useWorkspaceState(state, deps()); - - // Another tab reordered (b first — reorder re-stamps display order as - // ascending timestamps, so b carries the SMALLER newer stamp) and - // persisted that; this tab must adopt the newer order even though its - // own copy disagrees. - localStorage.setItem( - STORAGE_KEYS.promptQueue, - JSON.stringify({ - sess_1: { - entries: [ - { text: 'b', attachments: undefined, id: 'id-b', enqueuedAt: 1001 }, - { text: 'a', attachments: undefined, id: 'id-a', enqueuedAt: 1002 }, - ], - }, - }), - ); - ws.syncQueuedPromptsFromStorage(); - - expect(state.queuedBySession.sess_1?.map((e) => e.text)).toEqual(['b', 'a']); - // ...and convergence is stable: a second adopt writes nothing new. - await nextTick(); - const once = localStorage.getItem(STORAGE_KEYS.promptQueue); - ws.syncQueuedPromptsFromStorage(); - await nextTick(); - expect(localStorage.getItem(STORAGE_KEYS.promptQueue)).toBe(once); - }); - it('does not resurrect the queue when a submit fails after the session was forgotten', async () => { let rejectSubmit!: (err: Error) => void; apiMock.submitPrompt.mockImplementation( @@ -2174,16 +1834,16 @@ describe('useWorkspaceState — prompt queue persistence', () => { ); const state = createState(); state.queuedBySession = { - sess_1: [{ text: 'doomed', attachments: undefined, id: 'id-doomed', enqueuedAt: 1 }], + sess_1: [{ text: 'doomed', attachments: undefined, id: 'id-doomed' }], }; - const ws = useWorkspaceState(state, deps()); + const ws = useWorkspaceState(state, promptDeps()); ws.finishPromptLocal('sess_1', { turnWasActive: true }); expect(state.queuedBySession.sess_1 ?? []).toEqual([]); - // Session archived (facade forget path) while the submit is pending. + // Facade forget path (e.g. archive) while the submit is pending. state.sessions = []; - ws.discardQueuedPrompts('sess_1'); + delete state.queuedBySession.sess_1; rejectSubmit(new Error('network down')); await new Promise((resolve) => setTimeout(resolve, 0)); From a576efbea561f98958a4c43408a0c2303b014f61 Mon Sep 17 00:00:00 2001 From: qer Date: Fri, 17 Jul 2026 18:27:05 +0800 Subject: [PATCH 5/6] chore: align the changeset wording with the reduced scope --- .changeset/web-prompt-queue-stale-attachments.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/web-prompt-queue-stale-attachments.md b/.changeset/web-prompt-queue-stale-attachments.md index 71ed465a3b..340604fb20 100644 --- a/.changeset/web-prompt-queue-stale-attachments.md +++ b/.changeset/web-prompt-queue-stale-attachments.md @@ -2,4 +2,4 @@ "@moonshot-ai/kimi-code": patch --- -web: Fix queued messages silently re-sending previously uploaded files when a session is reopened, and keep unsent queued messages across page refreshes. +web: Fix queued messages silently re-sending previously uploaded files when a session is reopened. From 9c56b1d42f41429c53d94991c85631cf9cf1510c Mon Sep 17 00:00:00 2001 From: qer Date: Fri, 17 Jul 2026 18:49:04 +0800 Subject: [PATCH 6/6] fix(kimi-web): never duplicate on ambiguous submit failures; advance after drop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings: (1) restoring merged queue entries after ANY steer failure could re-submit prompts the daemon had already accepted when the failure was a lost response — submits now report ok/rejected/uncertain and restores + flush re-queues happen only on definitive daemon rejections, while ambiguous failures drop the entry (the failure toast still tells the user); (2) dropping an exhausted queue head no longer strands the entries behind it — the new head is submitted immediately, carrying its own failure budget. --- .../composables/client/useWorkspaceState.ts | 65 ++++++++++---- apps/kimi-web/test/workspace-state.test.ts | 87 +++++++++++++++++-- 2 files changed, 129 insertions(+), 23 deletions(-) diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index e8326e6d03..a921951342 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -1429,8 +1429,16 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } /** Internal: submit a prompt to a specific session, bypassing the queue check. - Returns true when the daemon accepted the prompt. */ - async function submitPromptInternal(sid: string, text: string, attachments?: PromptAttachment[]): Promise { + Returns 'ok' when the daemon accepted the prompt, 'rejected' on a + definitive refusal (structured API error — the server holds nothing, so + re-queueing is safe), and 'uncertain' when the failure was ambiguous + (dropped response, network error): the merged prompt may already be + queued server-side, and callers must NOT re-queue or they'd duplicate it. */ + async function submitPromptInternal( + 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 @@ -1457,7 +1465,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } if (content.length === 0) { rawState.inFlightBySession = { ...rawState.inFlightBySession, [sid]: false }; - return false; + return 'rejected'; } // OPTIMISTICALLY add the user message to local state BEFORE awaiting the @@ -1498,7 +1506,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta updateSessionMessages(sid, (msgs) => msgs.some((m) => m.id === tempId) ? msgs.filter((m) => m.id !== tempId) : msgs, ); - return false; + return 'rejected'; } } @@ -1547,18 +1555,20 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // session.meta.updated (projected to sessionMetaUpdated). PATCHing a title // locally would mark the session isCustomTitle=true and SUPPRESS the // daemon's auto-title, so we let the daemon own it. - return true; + return 'ok'; } catch (err) { // 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- - // looking message the daemon never received. + // looking message the daemon never received. A structured API error is a + // definitive refusal; anything else (network, truncated response) is + // ambiguous — the prompt may already sit in the server's queue. rawState.inFlightBySession = { ...rawState.inFlightBySession, [sid]: false }; updateSessionMessages(sid, (msgs) => msgs.some((m) => m.id === tempId) ? msgs.filter((m) => m.id !== tempId) : msgs, ); pushOperationFailure('sendPrompt', err, { sessionId: sid }); - return false; + return isDaemonApiError(err) ? 'rejected' : 'uncertain'; } finally { // The daemon answered the submit (accepted or rejected) — the pending // window in which a snapshot can't reflect this turn is over. @@ -1635,8 +1645,10 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta // Idle and nothing in flight — there is no turn to steer into; normal send. if (activity.value === 'idle' && !rawState.inFlightBySession[sid]) { - const ok = await submitPromptInternal(sid, merged, mergedAttachments); - if (!ok) restoreQueue(); + const outcome = await submitPromptInternal(sid, merged, mergedAttachments); + // Same never-duplicate rule as the running-path catch below: restore + // the merged entries only on a definitive rejection. + if (outcome === 'rejected') restoreQueue(); return; } @@ -1712,10 +1724,15 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } } catch (err) { // Submit failed: drop the optimistic echo so the transcript doesn't show - // a delivered-looking message the daemon never received, and put the - // merged queue entries back for a later steer/drain. + // a delivered-looking message the daemon never received. updateSessionMessages(sid, (msgs) => msgs.filter((m) => m.id !== tempId)); - restoreQueue(); + // Restore the merged queue entries ONLY on a definitive daemon rejection + // (a structured API error means nothing was accepted). On an ambiguous + // failure — dropped response, network error — the merged prompt may + // 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(); pushOperationFailure('steer', err, { sessionId: sid }); } finally { settleLocalTurn(sid, localTurnToken); @@ -1766,13 +1783,21 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const [next, ...rest] = rawState.queuedBySession[sid] ?? []; if (next === undefined) return; rawState.queuedBySession = { ...rawState.queuedBySession, [sid]: rest }; - void submitPromptInternal(sid, next.text, next.attachments).then((ok) => { - if (ok) { + void submitPromptInternal(sid, next.text, next.attachments).then((outcome) => { + if (outcome === 'ok') { queueFlushFailures.delete(sid); return; } - // The session was forgotten (e.g. archived) while the submit was - // pending — its queue was already discarded, so don't resurrect it. + // 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; @@ -1784,6 +1809,14 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta 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; } queueFlushFailures.set(sid, { key, count }); diff --git a/apps/kimi-web/test/workspace-state.test.ts b/apps/kimi-web/test/workspace-state.test.ts index 700dbebb3e..9b761a29e0 100644 --- a/apps/kimi-web/test/workspace-state.test.ts +++ b/apps/kimi-web/test/workspace-state.test.ts @@ -1654,7 +1654,9 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { 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 }] }; - apiMock.submitPrompt.mockRejectedValue(new Error('turn.agent_busy')); + apiMock.submitPrompt.mockRejectedValue( + new DaemonApiError({ code: 50000, msg: 'turn.agent_busy', requestId: 'r' }), + ); const ws = useWorkspaceState(state, promptDeps()); const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); @@ -1676,13 +1678,15 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { expect(apiMock.submitPrompt).toHaveBeenCalledTimes(3); }); - it('restores the merged queue entries when a steer submit fails', async () => { + it('restores the merged queue entries when a steer submit is definitively rejected', async () => { const state = createState(); state.inFlightBySession = { sess_1: true }; state.queuedBySession = { sess_1: [{ text: 'queued', attachments: [{ fileId: 'f_q', kind: 'image' }] }], }; - apiMock.submitPrompt.mockRejectedValue(new Error('boom')); + apiMock.submitPrompt.mockRejectedValue( + new DaemonApiError({ code: 50000, msg: 'boom', requestId: 'r' }), + ); const ws = useWorkspaceState(state, promptDeps()); await ws.steerPrompt('live text', [{ fileId: 'f_live', kind: 'image' }]); @@ -1692,10 +1696,28 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { ); }); + it('does NOT restore merged queue entries when a steer failure is network-ambiguous', async () => { + const state = createState(); + state.inFlightBySession = { sess_1: true }; + state.queuedBySession = { + sess_1: [{ text: 'queued', attachments: [{ fileId: 'f_q', kind: 'image' }] }], + }; + // Response lost mid-flight: the merged prompt may already be queued + // server-side, so restoring would duplicate it on a later drain. + apiMock.submitPrompt.mockRejectedValue(new TypeError('fetch failed')); + const ws = useWorkspaceState(state, promptDeps()); + + await ws.steerPrompt('live text', [{ fileId: 'f_live', kind: 'image' }]); + + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + }); + it('restores the queue when an idle steer falls back to a normal send that fails', async () => { const state = createState(); state.queuedBySession = { sess_1: [{ text: 'queued', attachments: undefined }] }; - apiMock.submitPrompt.mockRejectedValue(new Error('boom')); + apiMock.submitPrompt.mockRejectedValue( + new DaemonApiError({ code: 50000, msg: 'boom', requestId: 'r' }), + ); const ws = useWorkspaceState(state, promptDeps({ activity: computed(() => 'idle') })); await ws.steerPrompt('live text'); @@ -1793,8 +1815,58 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { ); }); + it('advances to the next queued entry after dropping an exhausted head', async () => { + const state = createState(); + state.queuedBySession = { + sess_1: [ + { text: 'poisoned head', attachments: undefined, id: 'id-bad' }, + { text: 'good next', attachments: undefined, id: 'id-good' }, + ], + }; + apiMock.submitPrompt + .mockRejectedValueOnce(new DaemonApiError({ code: 50000, msg: 'gone', requestId: 'r' })) + .mockRejectedValueOnce(new DaemonApiError({ code: 50000, msg: 'gone', requestId: 'r' })) + .mockRejectedValueOnce(new DaemonApiError({ code: 50000, msg: 'gone', requestId: 'r' })) + .mockResolvedValueOnce({ promptId: 'prompt_good' }); + const ws = useWorkspaceState(state, promptDeps()); + const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); + + for (let i = 0; i < 3; i += 1) { + state.inFlightBySession = { sess_1: true }; + ws.handleSessionSnapshot('sess_1', { inFlightTurn: null, busy: false }); + await settle(); + } + + // The exhausted head is gone AND the next entry was submitted right + // away — entries behind a dropped head must not wait for another send. + expect(apiMock.submitPrompt).toHaveBeenCalledTimes(4); + expect(apiMock.submitPrompt).toHaveBeenLastCalledWith( + 'sess_1', + expect.objectContaining({ content: [{ type: 'text', text: 'good next' }] }), + ); + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + }); + + it('drops (never duplicates) a flush whose failure was network-ambiguous', async () => { + const state = createState(); + state.queuedBySession = { sess_1: [{ text: 'maybe sent', attachments: undefined, id: 'id-x' }] }; + apiMock.submitPrompt.mockRejectedValue(new TypeError('fetch failed')); + const ws = useWorkspaceState(state, promptDeps()); + + state.inFlightBySession = { sess_1: true }; + ws.handleSessionSnapshot('sess_1', { inFlightTurn: null, busy: false }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + // The response was lost mid-flight — the daemon may already hold the + // prompt. Re-queueing could submit it twice, so the entry is dropped + // instead (the failure was surfaced via pushOperationFailure). + expect(state.queuedBySession.sess_1 ?? []).toEqual([]); + }); + it('resets the flush failure budget when the queue head changes', async () => { - apiMock.submitPrompt.mockRejectedValue(new Error('turn.agent_busy')); + apiMock.submitPrompt.mockRejectedValue( + new DaemonApiError({ code: 50000, msg: 'turn.agent_busy', requestId: 'r' }), + ); const state = createState(); state.queuedBySession = { sess_1: [ @@ -1841,10 +1913,11 @@ describe('useWorkspaceState — snapshot prompt recovery', () => { ws.finishPromptLocal('sess_1', { turnWasActive: true }); expect(state.queuedBySession.sess_1 ?? []).toEqual([]); - // Facade forget path (e.g. archive) while the submit is pending. + // Facade forget path (e.g. archive) while the submit is pending. The + // daemon definitively rejects afterwards — even then, no resurrection. state.sessions = []; delete state.queuedBySession.sess_1; - rejectSubmit(new Error('network down')); + rejectSubmit(new DaemonApiError({ code: 50000, msg: 'network down', requestId: 'r' })); await new Promise((resolve) => setTimeout(resolve, 0)); expect(state.queuedBySession.sess_1).toBeUndefined();