Keep queued transport messages visible until idle

Author: IM.codes

web/src/app.tsx

@@ -60,7 +60,7 @@ import { shouldSubscribeTerminalRaw, type TerminalSubscribeViewMode } from './te
 import { onWatchCommand } from './watch-bridge.js';
 import { watchProjectionStore } from './watch-projection.js';
 import { isIdleSessionStateTimelineEvent, isRunningTimelineEvent } from './timeline-running.js';
-import { extractTransportPendingMessages } from './transport-queue.js';
+import { extractTransportPendingMessages, mergeTransportPendingMessagesForRunningState } from './transport-queue.js';
 import { ingestTimelineEventForCache } from './hooks/useTimeline.js';
 import { getMobileKeyboardState } from './mobile-keyboard.js';
 import { pickReadableSessionDisplay } from '@shared/session-display.js';
@@ -1332,15 +1332,16 @@ export function App() {
                 : s,
             ));
           } else if (liveState === 'running') {
-            const pendingMessages = hasPendingMessagesField
-              ? extractTransportPendingMessages(event.payload.pendingMessages)
-              : null;
             setSessions((prev) => prev.map((s) =>
               s.name === event.sessionId
                 ? {
                     ...s,
                     state: 'running' as SessionInfo['state'],
-                    transportPendingMessages: pendingMessages ?? (s.transportPendingMessages ?? []),
+                    transportPendingMessages: mergeTransportPendingMessagesForRunningState(
+                      s.transportPendingMessages,
+                      event.payload.pendingMessages,
+                      hasPendingMessagesField,
+                    ),
                   }
                 : s,
             ));

web/src/hooks/useSubSessions.ts

@@ -11,7 +11,7 @@ import {
 } from '../api.js';
 import type { WsClient } from '../ws-client.js';
 import { isRunningTimelineEvent } from '../timeline-running.js';
-import { extractTransportPendingMessages } from '../transport-queue.js';
+import { extractTransportPendingMessages, mergeTransportPendingMessagesForRunningState } from '../transport-queue.js';
 
 export interface SubSession extends SubSessionData {
   sessionName: string;
@@ -252,12 +252,19 @@ export function useSubSessions(
         return;
       }
       if (state === 'running' && hasPendingMessagesField) {
-        const pendingMessages = extractTransportPendingMessages(msg.event.payload.pendingMessages);
         setSubSessions((prev) => {
           const idx = prev.findIndex((s) => s.sessionName === sessionName);
           if (idx === -1) return prev;
           const next = [...prev];
-          next[idx] = { ...next[idx], state: 'running', transportPendingMessages: pendingMessages };
+          next[idx] = {
+            ...next[idx],
+            state: 'running',
+            transportPendingMessages: mergeTransportPendingMessagesForRunningState(
+              next[idx].transportPendingMessages,
+              msg.event.payload.pendingMessages,
+              true,
+            ),
+          };
           return next;
         });
         return;

web/src/transport-queue.ts

@@ -4,3 +4,15 @@ export function extractTransportPendingMessages(value: unknown): string[] {
     .map((entry) => (typeof entry === 'string' ? entry.trim() : ''))
     .filter((entry) => entry.length > 0);
 }
+
+export function mergeTransportPendingMessagesForRunningState(
+  existing: string[] | null | undefined,
+  pendingFromEvent: unknown,
+  hasPendingMessagesField: boolean,
+): string[] {
+  const existingMessages = Array.isArray(existing) ? existing.filter((entry) => typeof entry === 'string' && entry.length > 0) : [];
+  if (!hasPendingMessagesField) return existingMessages;
+  const nextMessages = extractTransportPendingMessages(pendingFromEvent);
+  if (nextMessages.length > 0) return nextMessages;
+  return existingMessages;
+}

web/test/transport-queue.test.ts

@@ -0,0 +1,30 @@
+import { describe, expect, it } from 'vitest';
+
+import {
+  extractTransportPendingMessages,
+  mergeTransportPendingMessagesForRunningState,
+} from '../src/transport-queue.js';
+
+describe('extractTransportPendingMessages', () => {
+  it('keeps only non-empty string entries', () => {
+    expect(extractTransportPendingMessages([' one ', '', 1, null, 'two'])).toEqual(['one', 'two']);
+  });
+});
+
+describe('mergeTransportPendingMessagesForRunningState', () => {
+  it('preserves the existing queue when running reports no pendingMessages field', () => {
+    expect(mergeTransportPendingMessagesForRunningState(['queued one'], undefined, false)).toEqual(['queued one']);
+  });
+
+  it('preserves the existing queue when running reports an empty pendingMessages array', () => {
+    expect(mergeTransportPendingMessagesForRunningState(['queued one', 'queued two'], [], true)).toEqual(['queued one', 'queued two']);
+  });
+
+  it('replaces the queue when running reports a non-empty pendingMessages array', () => {
+    expect(mergeTransportPendingMessagesForRunningState(['queued one'], ['queued two'], true)).toEqual(['queued two']);
+  });
+
+  it('returns an empty queue when nothing is queued yet', () => {
+    expect(mergeTransportPendingMessagesForRunningState([], [], true)).toEqual([]);
+  });
+});

web/test/use-sub-sessions-metadata.test.tsx

@@ -217,7 +217,7 @@ describe('sub-session metadata via subsession.sync', () => {
     expect(captured[0].quotaUsageLabel).toBe('today 20/1000');
   });
 
-  it('preserves queued transport messages until running explicitly reports an empty queue', async () => {
+  it('preserves queued transport messages while the drained send is still running and clears on idle', async () => {
     const { ws, send } = createMockWs();
     render(<Harness ws={ws} connected={true} />);
     await waitFor(() => expect(ws.onMessage).toHaveBeenCalled());
@@ -261,6 +261,17 @@ describe('sub-session metadata via subsession.sync', () => {
       },
     }));
 
+    expect(captured[0].transportPendingMessages).toEqual(['queued one', 'queued two']);
+
+    act(() => send({
+      type: 'timeline.event',
+      event: {
+        type: 'session.state',
+        sessionId: 'deck_sub_q4',
+        payload: { state: 'idle' },
+      },
+    }));
+
     expect(captured[0].transportPendingMessages).toEqual([]);
   });
 });