fix(kimi-web): stop the prompt queue from ghost-sending stale attachments

.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.

apps/kimi-web/src/composables/client/useWorkspaceState.ts

@@ -118,6 +118,22 @@ const pendingLocalTurnStarts = new Map<string, Set<number>>();
 const afterLocalTurnsSettled = new Map<string, () => void>();
 let nextLocalTurnToken = 0;
 
+/**
+ * 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. Module-level singleton — the queue
+ * itself is per-session on rawState, so a page reload resets both.
+ */
+const queueFlushFailures = new Map<string, { key: string; count: number }>();
+const MAX_QUEUE_FLUSH_FAILURES = 3;
+
+let queueEntryCounter = 0;
+function nextQueueEntryId(): string {
+  queueEntryCounter += 1;
+  return `${Date.now().toString(36)}-${queueEntryCounter}`;
+}
+
 export interface LocalTurnStartState {
   generation: number;
   pending: boolean;
@@ -161,6 +177,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
@@ -1412,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<boolean> {
+      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
@@ -1440,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
@@ -1481,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';
         }
       }
 
@@ -1530,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.
@@ -1562,6 +1589,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 +1634,21 @@ 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 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;
     }
 
@@ -1674,6 +1726,13 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta
       // Submit failed: drop the optimistic echo so the transcript doesn't show
       // a delivered-looking message the daemon never received.
       updateSessionMessages(sid, (msgs) => msgs.filter((m) => m.id !== tempId));
+      // 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);
@@ -1700,12 +1759,75 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta
     const sid = rawState.activeSessionId;
     if (!sid) return;
     const current = rawState.queuedBySession[sid] ?? [];
+    // 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, { text, attachments }],
+      [sid]: [...current, entry],
     };
   }
 
+  /**
+   * 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. 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.
+   */
+  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((outcome) => {
+      if (outcome === 'ok') {
+        queueFlushFailures.delete(sid);
+        return;
+      }
+      // Ambiguous failure: the daemon may have accepted the prompt and lost
+      // the response — the entry is dropped (the failure was already toasted)
+      // rather than re-queued and possibly submitted twice.
+      if (outcome === 'uncertain') {
+        queueFlushFailures.delete(sid);
+        return;
+      }
+      // Definitively rejected below this point. If the session was forgotten
+      // (e.g. archived) while the submit was pending, its queue was already
+      // discarded — don't resurrect it.
+      if (!rawState.sessions.some((s) => s.id === sid)) {
+        queueFlushFailures.delete(sid);
+        return;
+      }
+      // Per-entry budget: a different head (removed/reordered since) starts
+      // fresh instead of inheriting the previous entry's failures.
+      const key = next.id ?? next.text;
+      const previous = queueFlushFailures.get(sid);
+      const count = previous !== undefined && previous.key === key ? previous.count + 1 : 1;
+      if (count >= MAX_QUEUE_FLUSH_FAILURES) {
+        queueFlushFailures.delete(sid);
+        // Advance the queue instead of stranding the entries behind the
+        // dropped head: the failed submit produced no turn, so nothing else
+        // will drive the next entry until the user sends again. The new head
+        // carries its own budget, so a poisoned successor just goes through
+        // the same retry cycle.
+        if ((rawState.queuedBySession[sid]?.length ?? 0) > 0) {
+          flushQueueHead(sid);
+        }
+        return;
+      }
+      queueFlushFailures.set(sid, { key, count });
+      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 +1840,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 +1862,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 +1879,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,

apps/kimi-web/src/composables/useKimiWebClient.ts

@@ -296,10 +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). */
+    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;
 }
 
 export interface ExtendedState extends KimiClientState {
@@ -944,9 +946,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,15 +2646,15 @@ 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];
   // Shared finish cleanup: clears in-flight/prompt-id and drains one
   // 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).