Fix stale transport queue UI sync

Author: IM.codes

test/e2e/qwen-transport-flow.test.ts

@@ -716,6 +716,33 @@ describe('qwen transport flow e2e', () => {
     );
     expect(queuedStateAfterDrain).toBeUndefined();
 
+    const explicitEmptyRunningState = drainedEvents.find((e) =>
+      e.session === SESSION
+      && e.type === 'session.state'
+      && e.payload.state === 'running'
+      && Array.isArray(e.payload.pendingMessageEntries)
+      && e.payload.pendingMessageEntries.length === 0
+      && Array.isArray(e.payload.pendingMessages)
+      && e.payload.pendingMessages.length === 0
+      && typeof e.payload.pendingMessageVersion === 'number'
+    );
+    expect(explicitEmptyRunningState).toBeDefined();
+
+    const postDrainSessionListStart = serverLink.send.mock.calls.length;
+    handleWebCommand({ type: 'get_sessions' }, serverLink);
+    await flushAsync();
+
+    const postDrainSessionList = serverLink.send.mock.calls
+      .slice(postDrainSessionListStart)
+      .find(([msg]: [Record<string, unknown>]) => msg.type === 'session_list')?.[0] as
+        | { sessions?: Array<Record<string, unknown>> }
+        | undefined;
+    const postDrainSession = postDrainSessionList?.sessions?.find((session) => session.name === SESSION);
+    expect(postDrainSession).toEqual(expect.objectContaining({
+      transportPendingMessages: [],
+      transportPendingMessageEntries: [],
+      transportPendingMessageVersion: expect.any(Number),
+    }));
 
     const idleStateEvents = mocks.emitted.filter((e) =>
       e.session === SESSION

test/web/transport-queue.test.ts

@@ -110,4 +110,17 @@ describe('transport queue reconciliation', () => {
     expect(result.changed).toBe(false);
     expect(result.entries.map((entry) => entry.clientMessageId)).toEqual(['queued-a']);
   });
+
+  it('allows a real delivered id to clear a single matching legacy queued entry by text', () => {
+    const result = removeTransportPendingEntryForUserMessage(
+      [{ clientMessageId: 'session-a:legacy:0:same text', text: 'same text' }],
+      ['same text'],
+      { clientMessageId: 'real-command-id', text: 'same text' },
+      'session-a',
+    );
+
+    expect(result.changed).toBe(true);
+    expect(result.entries).toEqual([]);
+    expect(result.messages).toEqual([]);
+  });
 });

web/src/components/SessionControls.tsx

@@ -884,7 +884,8 @@ export function SessionControls({ ws, activeSession, inputRef, onAfterAction, on
           // Keep legacy messages visible when older state lacks structured
           // entries, but do not synthesize tails for partial structured entries.
           hasEntriesField: Array.isArray(activeSession?.transportPendingMessageEntries)
-            && activeSession.transportPendingMessageEntries.length > 0,
+            && (activeSession.transportPendingMessageEntries.length > 0
+              || typeof activeSession.transportPendingMessageVersion === 'number'),
           hasMessagesField: Array.isArray(activeSession?.transportPendingMessages),
         },
       )
@@ -1508,14 +1509,32 @@ export function SessionControls({ ws, activeSession, inputRef, onAfterAction, on
           : typeof event.payload.clientMessageId === 'string'
             ? event.payload.clientMessageId
             : '';
-        removeLocalQueuedEntry(commandId, typeof event.payload.text === 'string' ? event.payload.text : undefined);
+        const deliveredText = typeof event.payload.text === 'string' ? event.payload.text : undefined;
+        removeLocalQueuedEntry(commandId, deliveredText);
         // Record the delivered id so a stale daemon pending snapshot can't keep
         // showing it as queued (the incoming snapshot is not under our control,
-        // unlike the optimistic set cleared above).
-        if (commandId) {
+        // unlike the optimistic set cleared above).  Legacy snapshots synthesize
+        // ids from text, so also settle the single matching legacy id when the
+        // authoritative timeline echo carries a newer real command/client id.
+        const idsToSettle = commandId ? [commandId] : [];
+        const normalizedDeliveredText = typeof deliveredText === 'string' ? normalizeQueuedText(deliveredText) : '';
+        if (normalizedDeliveredText) {
+          const legacyTextMatches = incomingQueuedTransportEntries
+            .filter((entry) => isLegacyTransportPendingMessageId(entry.clientMessageId, activeSession.name)
+              && normalizeQueuedText(entry.text) === normalizedDeliveredText)
+            .map((entry) => entry.clientMessageId);
+          if (legacyTextMatches.length === 1) idsToSettle.push(legacyTextMatches[0]);
+        }
+        if (idsToSettle.length > 0) {
           setSettledQueuedIds((prev) => {
-            if (prev.has(commandId)) return prev;
-            const ids = [...prev, commandId];
+            let changed = false;
+            const ids = [...prev];
+            for (const id of idsToSettle) {
+              if (!id || prev.has(id)) continue;
+              ids.push(id);
+              changed = true;
+            }
+            if (!changed) return prev;
             return new Set(ids.length > 500 ? ids.slice(-500) : ids);
           });
         }
@@ -1539,7 +1558,7 @@ export function SessionControls({ ws, activeSession, inputRef, onAfterAction, on
         });
       }
     });
-  }, [activeSession, ws]);
+  }, [activeSession, incomingQueuedTransportEntries, ws]);
 
   // Reset P2P mode on session change
   useEffect(() => { setP2pMode('solo'); setP2pOpen(false); }, [activeSession?.name]);

web/src/transport-queue.ts

@@ -178,15 +178,29 @@ export function removeTransportPendingEntryForUserMessage(
   const normalizedText = typeof payload.text === 'string'
     ? normalizeTransportPendingText(payload.text)
     : '';
-  const matchIndex = candidateIdSet.size > 0
+  const idMatchIndex = candidateIdSet.size > 0
     ? entries.findIndex((entry) => candidateIdSet.has(entry.clientMessageId))
-    : (() => {
-        if (!normalizedText) return -1;
-        const matches = entries
-          .map((entry, index) => ({ entry, index }))
-          .filter(({ entry }) => normalizeTransportPendingText(entry.text) === normalizedText);
-        return matches.length === 1 ? matches[0].index : -1;
-      })();
+    : -1;
+  const textFallbackIndex = (() => {
+    if (!normalizedText) return -1;
+    const matches = entries
+      .map((entry, index) => ({ entry, index }))
+      .filter(({ entry }) => normalizeTransportPendingText(entry.text) === normalizedText);
+    if (matches.length !== 1) return -1;
+    const match = matches[0];
+    // Stable ids normally win.  However, legacy snapshots synthesize ids from
+    // text (`session:legacy:*`) and can survive across client/daemon upgrades; a
+    // later authoritative user.message carries the real command/client id, so an
+    // id-only match leaves the old yellow queue card stuck beside the delivered
+    // user bubble.  Allow text fallback only for those synthetic legacy ids (or
+    // when no id was provided at all), preserving the duplicate-text safety for
+    // real structured queue entries.
+    if (candidateIdSet.size === 0 || isLegacyTransportPendingMessageId(match.entry.clientMessageId, scopeKey)) {
+      return match.index;
+    }
+    return -1;
+  })();
+  const matchIndex = idMatchIndex >= 0 ? idMatchIndex : textFallbackIndex;
   if (matchIndex < 0) return { messages, entries, changed: false };
   const nextEntries = entries.filter((_, index) => index !== matchIndex);
   return {

web/test/components/SessionControls.test.tsx

@@ -2824,6 +2824,63 @@ afterEach(() => {
     expect(screen.getByText('queued second')).toBeDefined();
   });
 
+  it('does not resurrect stale pendingMessages when a versioned empty entries snapshot is present', () => {
+    const ws = makeWs();
+    render(
+      <SessionControls
+        ws={ws as any}
+        activeSession={makeSession({
+          runtimeType: 'transport',
+          transportPendingMessages: ['stale queued'],
+          transportPendingMessageEntries: [],
+          transportPendingMessageVersion: 12,
+        })}
+        quickData={makeQuickData() as any}
+      />,
+    );
+
+    expect(document.querySelector('.controls-queued-hint')).toBeFalsy();
+    expect(screen.queryByText('stale queued')).toBeNull();
+  });
+
+  it('clears a legacy queued card by unique text when the delivered timeline message has the real id', () => {
+    const ws = makeWs();
+    render(
+      <SessionControls
+        ws={ws as any}
+        activeSession={makeSession({
+          name: 'qwen-session',
+          runtimeType: 'transport',
+          state: 'running',
+          transportPendingMessages: ['legacy stale send'],
+          transportPendingMessageEntries: [],
+        })}
+        quickData={makeQuickData() as any}
+      />,
+    );
+
+    expect(screen.getByText('legacy stale send')).toBeDefined();
+
+    act(() => {
+      ws.emit({
+        type: 'timeline.event',
+        event: {
+          eventId: 'transport-user:real-command-id',
+          sessionId: 'qwen-session',
+          type: 'user.message',
+          ts: Date.now(),
+          seq: 1,
+          epoch: 1,
+          source: 'daemon',
+          confidence: 'high',
+          payload: { text: 'legacy stale send', commandId: 'real-command-id' },
+        },
+      });
+    });
+
+    expect(screen.queryByText('legacy stale send')).toBeNull();
+  });
+
   it('treats partial queued transport entries as authoritative', () => {
     const ws = makeWs();
     render(