fix(cloud-agent-next): hide superseded delivery failures (#3451)

Author: eshurakov

services/cloud-agent-next/src/session/session-message-queue.test.ts

@@ -1088,6 +1088,92 @@ describe('SessionMessageQueue', () => {
     ]);
   });
 
+  it('omits a reconnect delivery failure after a later turn completed', async () => {
+    const harness = createQueueHarness();
+    await putSessionMessageState(harness.storage, {
+      ...createQueuedSessionMessageState(
+        {
+          turn: {
+            type: 'prompt',
+            messageId: FIRST_MESSAGE_ID,
+            prompt: 'failed before acceptance',
+          },
+          agent: { mode: 'code', model: 'default-model' },
+        },
+        undefined,
+        10
+      ),
+      status: 'failed',
+      terminalAt: 30,
+      completionSource: 'delivery_failure',
+      failureReason: 'exhausted',
+      error: 'delivery exhausted',
+      attempts: 1,
+    });
+    await putSessionMessageState(harness.storage, {
+      ...createQueuedSessionMessageState(
+        {
+          turn: {
+            type: 'prompt',
+            messageId: SECOND_MESSAGE_ID,
+            prompt: 'delivered later',
+          },
+          agent: { mode: 'code', model: 'default-model' },
+        },
+        undefined,
+        40
+      ),
+      status: 'completed',
+      acceptedAt: 50,
+      terminalAt: 60,
+      completionSource: 'assistant_message_event',
+    });
+
+    await expect(harness.queue.snapshotForStreamConnect()).resolves.toEqual([]);
+  });
+
+  it('omits a reconnect delivery failure while a later turn is accepted', async () => {
+    const harness = createQueueHarness();
+    await putSessionMessageState(harness.storage, {
+      ...createQueuedSessionMessageState(
+        {
+          turn: {
+            type: 'prompt',
+            messageId: FIRST_MESSAGE_ID,
+            prompt: 'failed before acceptance',
+          },
+          agent: { mode: 'code', model: 'default-model' },
+        },
+        undefined,
+        10
+      ),
+      status: 'failed',
+      terminalAt: 30,
+      completionSource: 'delivery_failure',
+      failureReason: 'exhausted',
+      error: 'delivery exhausted',
+      attempts: 1,
+    });
+    await putSessionMessageState(harness.storage, {
+      ...createQueuedSessionMessageState(
+        {
+          turn: {
+            type: 'prompt',
+            messageId: SECOND_MESSAGE_ID,
+            prompt: 'currently running',
+          },
+          agent: { mode: 'code', model: 'default-model' },
+        },
+        undefined,
+        40
+      ),
+      status: 'accepted',
+      acceptedAt: 50,
+    });
+
+    await expect(harness.queue.snapshotForStreamConnect()).resolves.toEqual([]);
+  });
+
   it('terminalizes pending queued work before deleting it during interrupt handoff', async () => {
     const harness = createQueueHarness();
     const first = createPendingSessionMessage({

services/cloud-agent-next/src/session/session-message-queue.ts

@@ -38,7 +38,7 @@ import {
 import {
   createQueuedSessionMessageState,
   getSessionMessageState,
-  listNeverAcceptedTerminalQueuedMessages,
+  listReconnectVisibleTerminalQueuedMessages,
   putSessionMessageState,
   type SessionMessageStorage,
   type TerminalizeParams,
@@ -848,7 +848,7 @@ export function createSessionMessageQueue(
   async function snapshotForStreamConnect(): Promise<QueuedMessageSnapshot[]> {
     const pending = await listPendingSessionMessages(storage);
     const pendingMessageIds = new Set(pending.map(message => message.messageId));
-    const terminalQueued = await listNeverAcceptedTerminalQueuedMessages(storage);
+    const terminalQueued = await listReconnectVisibleTerminalQueuedMessages(storage);
 
     return [
       ...pending.map(message => ({

services/cloud-agent-next/src/session/session-message-state.ts

@@ -419,7 +419,7 @@ type NeverAcceptedTerminalQueuedMessageState = SessionMessageState & {
   status: 'failed' | 'interrupted';
 };
 
-export async function listNeverAcceptedTerminalQueuedMessages(
+export async function listReconnectVisibleTerminalQueuedMessages(
   storage: SessionMessageStorage
 ): Promise<NeverAcceptedTerminalQueuedMessageState[]> {
   const entries = await listSessionMessageStates(storage);
@@ -428,7 +428,14 @@ export async function listNeverAcceptedTerminalQueuedMessages(
       state.acceptedAt === undefined &&
       state.queuedAt !== undefined &&
       (state.status === 'failed' || state.status === 'interrupted') &&
-      (state.completionSource === 'delivery_failure' || state.completionSource === 'interrupt')
+      (state.completionSource === 'delivery_failure' || state.completionSource === 'interrupt') &&
+      !entries.some(
+        laterState =>
+          (laterState.acceptedAt !== undefined ||
+            laterState.status === 'accepted' ||
+            laterState.status === 'completed') &&
+          (laterState.queuedAt ?? laterState.createdAt) > (state.queuedAt ?? state.createdAt)
+      )
   );
 }