fix(web): recover stale response streams

frontend/src/domain/response.test.ts

@@ -52,6 +52,23 @@ describe('response projection', () => {
     expect(new Set(RESPONSE_EVENT_TYPES).size).toBe(RESPONSE_EVENT_TYPES.length);
   });
 
+  it('adopts the server epoch from the initial response.created event', () => {
+    const projection = reduceResponse(
+      initialProjection(run),
+      event('response.created', 1, { run_epoch: 1788084563000000 }),
+    );
+
+    expect(projection.run).toMatchObject({
+      responseId: 'r1',
+      epoch: 1788084563000000,
+      lastSequence: 1,
+      status: 'streaming',
+    });
+    expect(() =>
+      reduceResponse(projection, event('response.output_text.delta', 2, { run_epoch: 1 })),
+    ).toThrow(ResponseProtocolError);
+  });
+
   it('does not project the compaction resume handoff below its transcript boundary', () => {
     let projection = initialProjection(run);
     projection = reduceResponse(

frontend/src/domain/response.ts

@@ -254,7 +254,11 @@ function validate(
   if (responseId !== projection.run.responseId)
     throw new ResponseProtocolError(`Response owner mismatch: ${responseId}`, 'owner');
   const epoch = number(event.run_epoch);
-  if (!epoch || epoch !== projection.run.epoch)
+  // Before sequence 1 there is no projected server state to contradict; the
+  // response.created event establishes the authoritative epoch for this run.
+  const adoptingCreatedEpoch =
+    event.type === 'response.created' && projection.run.lastSequence === 0;
+  if (!epoch || (!adoptingCreatedEpoch && epoch !== projection.run.epoch))
     throw new ResponseProtocolError(`Response epoch mismatch: ${epoch}`, 'epoch');
   const sequence = number(event.sequence_number);
   if (!sequence)

frontend/src/stores/app-store.test.ts

@@ -2033,6 +2033,71 @@ describe('AppStore compatibility behavior', () => {
     }
   });
 
+  it('keeps the admitted POST stream when response.created supplies the server epoch', async () => {
+    const store = new AppStore(config);
+    let postSignal: AbortSignal | undefined;
+    try {
+      store.sessions.value = [session()];
+      store.activeSessionId.value = 's1';
+      store.draftActive.value = false;
+      store.prompt.value = 'Use the original response stream';
+      store.endpoints.response = vi.fn(async () => {
+        throw new Error('snapshot recovery should not run');
+      });
+      store.endpoints.createResponse = vi.fn(async (_body, _sessionId, _requestId, signal) => {
+        postSignal = signal;
+        const encoder = new TextEncoder();
+        const frames = [
+          ['response.created', { response: { id: 'r1', status: 'in_progress' } }],
+          ['response.output_text.delta', { delta: 'Done.' }],
+          ['response.completed', { response: { id: 'r1', status: 'completed' }, final_rev: 2 }],
+        ] as const;
+        const body = new ReadableStream<Uint8Array>({
+          start(controller) {
+            frames.forEach(([type, payload], index) =>
+              controller.enqueue(
+                encoder.encode(
+                  `event: ${type}\ndata: ${JSON.stringify({
+                    ...payload,
+                    response_id: 'r1',
+                    run_epoch: 1788084563000000,
+                    sequence_number: index + 1,
+                  })}\n\n`,
+                ),
+              ),
+            );
+            controller.enqueue(encoder.encode('data: [DONE]\n\n'));
+            controller.close();
+          },
+        });
+        return new Response(body, {
+          headers: { 'x-response-id': 'r1', 'x-session-id': 's1' },
+        });
+      });
+      store.endpoints.selectedSession = vi.fn(async () => ({
+        selected_session: { id: 's1', transcript_rev: 2 },
+        selected_transcript: { bodies: { rev: 2, messages: [] } },
+      }));
+      store.endpoints.sessionState = vi.fn(async () => ({}));
+
+      await store.send();
+
+      expect(postSignal?.aborted).toBe(true);
+      expect(store.endpoints.response).not.toHaveBeenCalled();
+      expect(store.runs.value.s1.run).toMatchObject({
+        responseId: 'r1',
+        epoch: 1788084563000000,
+        status: 'completed',
+        lastSequence: 3,
+      });
+      expect(store.runs.value.s1.messages).toEqual([
+        expect.objectContaining({ role: 'assistant', content: 'Done.' }),
+      ]);
+    } finally {
+      store.dispose();
+    }
+  });
+
   it('ignores late transport failures after a response is already complete', async () => {
     const store = new AppStore(config);
     store.sessions.value = [session()];
@@ -2618,6 +2683,87 @@ describe('AppStore compatibility behavior', () => {
     }
   });
 
+  it('retries when a replacement response stream never returns headers', async () => {
+    vi.useFakeTimers();
+    const store = new AppStore(config);
+    let connectSignal: AbortSignal | undefined;
+    try {
+      store.sessions.value = [{ ...session(), activeRun: true, activeResponseId: 'r1' }];
+      store.activeSessionId.value = 's1';
+      store.runs.value = {
+        s1: initialProjection({
+          responseId: 'r1',
+          sessionId: 's1',
+          epoch: 1,
+          status: 'connecting',
+          lastSequence: 3,
+          startedRev: 0,
+          reconnects: 0,
+        }),
+      };
+      store.endpoints.responseEvents = vi
+        .fn()
+        .mockImplementationOnce((_responseId, _after, signal) => {
+          connectSignal = signal;
+          // Model WebKit leaving fetch pending even after its signal is aborted.
+          return new Promise<Response>(() => undefined);
+        })
+        .mockImplementationOnce(async () => {
+          const body = new ReadableStream<Uint8Array>({
+            start(controller) {
+              controller.enqueue(
+                new TextEncoder().encode(
+                  `event: response.completed\ndata: ${JSON.stringify({
+                    response_id: 'r1',
+                    run_epoch: 1,
+                    sequence_number: 5,
+                    response: { id: 'r1', status: 'completed' },
+                    final_rev: 1,
+                  })}\n\ndata: [DONE]\n\n`,
+                ),
+              );
+              controller.close();
+            },
+          });
+          return new Response(body);
+        });
+      store.endpoints.response = vi.fn(async () => ({
+        id: 'r1',
+        status: 'in_progress',
+        run_epoch: 1,
+        last_sequence_number: 4,
+        recovery: { sequence_number: 4 },
+      }));
+      store.endpoints.selectedSession = vi.fn(async () => ({
+        selected_session: { id: 's1' },
+        selected_transcript: { bodies: { rev: 1, messages: [] } },
+      }));
+      store.endpoints.sessionState = vi.fn(async () => ({}));
+
+      void store.streamResponse('r1', 's1', 3);
+      await Promise.resolve();
+      expect(connectSignal?.aborted).toBe(false);
+
+      await vi.advanceTimersByTimeAsync(15_000);
+
+      expect(connectSignal?.aborted).toBe(true);
+      expect(store.runs.value.s1.run).toMatchObject({ status: 'connecting', reconnects: 1 });
+
+      await vi.advanceTimersByTimeAsync(1_500);
+      await vi.waitFor(() => expect(store.endpoints.response).toHaveBeenCalledOnce());
+      await vi.waitFor(() => expect(store.endpoints.responseEvents).toHaveBeenCalledTimes(2));
+      expect(store.endpoints.responseEvents).toHaveBeenLastCalledWith(
+        'r1',
+        4,
+        expect.any(AbortSignal),
+      );
+      expect(store.runEngine.currentSupervisor('s1')).toBeUndefined();
+    } finally {
+      store.dispose();
+      vi.useRealTimers();
+    }
+  });
+
   it('recovers an active run when no response transport is owned', async () => {
     const store = new AppStore(config);
     try {
@@ -2691,6 +2837,64 @@ describe('AppStore compatibility behavior', () => {
     expect(internals.resumeResponse).toHaveBeenCalledWith('s1', 'finished-while-suspended');
   });
 
+  it('retires an expired response after authoritative idle transcript reconciliation', async () => {
+    const store = new AppStore(config);
+    try {
+      store.sessions.value = [
+        {
+          ...session(),
+          activeRun: true,
+          activeResponseId: 'expired-response',
+          transcriptRev: 1,
+          messageBodiesRev: 1,
+        },
+      ];
+      store.activeSessionId.value = 's1';
+      store.runs.value = {
+        s1: initialProjection({
+          responseId: 'expired-response',
+          sessionId: 's1',
+          epoch: 99,
+          status: 'streaming',
+          lastSequence: 25,
+          startedRev: 1,
+          reconnects: 0,
+        }),
+      };
+      store.endpoints.sessionStatus = vi.fn(async () => ({
+        sessions: [{ id: 's1', transcript_rev: 2 }],
+        __etag: 'idle-etag',
+      }));
+      store.endpoints.response = vi.fn(async () => {
+        throw new APIError('response not found', 404);
+      });
+      store.endpoints.selectedSession = vi.fn(async () => ({
+        selected_session: { id: 's1', transcript_rev: 2 },
+        selected_transcript: { bodies: { rev: 2, messages: [] } },
+      }));
+      store.endpoints.sessionState = vi.fn(async () => ({}));
+      const internals = store as unknown as {
+        refreshStatus(authoritative?: boolean): Promise<void>;
+      };
+
+      await internals.refreshStatus(true);
+
+      await vi.waitFor(() => expect(store.runs.value.s1).toBeUndefined());
+      expect(store.endpoints.sessionStatus).toHaveBeenCalledWith('s1', false, ['all'], '');
+      expect(store.sessions.value[0]).toMatchObject({
+        activeRun: false,
+        activeResponseId: null,
+        messageBodiesRev: 2,
+        lastResponseId: 'expired-response',
+      });
+      expect(store.runEngine.currentSupervisor('s1')).toBeUndefined();
+      expect(store.runActive.value).toBe(false);
+      expect(store.streaming.value).toBe(false);
+    } finally {
+      store.dispose();
+    }
+  });
+
   it('does not probe a provisional response before the server admits it', async () => {
     const store = new AppStore(config);
     store.sessions.value = [{ ...session(), activeRun: true }];

frontend/src/stores/app-store.ts

@@ -480,6 +480,8 @@ export class AppStore {
       reconcile: (reason, authoritative) => this.reconcile(reason, { authoritative }),
       refreshSidebar: (authoritative) => this.refreshSidebar(authoritative),
       resumeResponse: (sessionId, responseId) => this.resumeResponse(sessionId, responseId),
+      reconcileServerIdleResponse: (sessionId, responseId, transcriptRev) =>
+        this.runEngine.reconcileServerIdleResponse(sessionId, responseId, transcriptRev),
       refreshSessionMessages: (sessionId, targetRev) =>
         this.refreshSessionMessages(sessionId, targetRev),
       syncSessionMessagesForAttach: (sessionId, targetRev) =>

frontend/src/stores/interaction-store.test.ts

@@ -1,6 +1,9 @@
+import { signal } from '@preact/signals';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import type { AskUserPrompt } from '../domain/types';
 import { AppStore } from './app-store';
+import { InteractionStore } from './interaction-store';
+import type { Modal } from './store-types';
 import { testConfig } from './store-test-fixtures';
 
 beforeEach(() => localStorage.clear());
@@ -36,6 +39,31 @@ describe('InteractionStore', () => {
     }
   });
 
+  it('publishes only authoritative interaction transitions, not recovery refreshes', () => {
+    const app = new AppStore(testConfig);
+    const publish = vi.fn();
+    const interactions = new InteractionStore(app.services, signal<Modal>(''), publish);
+    const prompt: AskUserPrompt = {
+      sessionId: 's1',
+      callId: 'ask-1',
+      questions: [{ header: 'Choice', question: 'Continue?', options: [] }],
+    };
+    try {
+      interactions.upsert('ask-user', 's1', '', 'ask-1', prompt);
+      interactions.upsert('ask-user', 's1', 'r1', 'ask-1', { ...prompt });
+      expect(publish).toHaveBeenCalledTimes(1);
+      expect(interactions.order.value).toHaveLength(1);
+
+      interactions.resolve('ask-user', 's1', 'r2', 'ask-1', 'answered', 10);
+      interactions.resolve('ask-user', 's1', 'r1', 'ask-1', 'answered', 20);
+      expect(publish).toHaveBeenCalledTimes(2);
+      expect(interactions.order.value).toHaveLength(1);
+      expect(interactions.shouldOpen('ask-user', 's1', 'ask-1')).toBe(false);
+    } finally {
+      app.dispose();
+    }
+  });
+
   it('deduplicates concurrent submissions inside the interaction owner', async () => {
     const store = new AppStore(testConfig);
     try {

frontend/src/stores/interaction-store.ts

@@ -32,8 +32,10 @@ export class InteractionStore {
     requestId: string,
     prompt: ApprovalPrompt | AskUserPrompt,
   ): string {
-    const key = `${sessionId}:${responseId}:${requestId}`;
-    const existing = this.interactions.peek()[key];
+    const discoveredKey = `${sessionId}:${responseId}:${requestId}`;
+    const existing =
+      this.find(kind, sessionId, requestId, responseId) || this.find(kind, sessionId, requestId);
+    const key = existing?.key || discoveredKey;
     const record: InteractionRecord = existing || {
       key,
       sessionId,
@@ -47,10 +49,12 @@ export class InteractionStore {
     };
     this.interactions.value = {
       ...this.interactions.peek(),
-      [key]: { ...record, prompt },
+      [key]: { ...record, responseId: record.responseId || responseId, prompt },
     };
-    if (!existing) this.order.value = [...this.order.peek(), key];
-    this.publish('interaction-changed', sessionId, responseId);
+    if (!existing) {
+      this.order.value = [...this.order.peek(), key];
+      this.publish('interaction-changed', sessionId, responseId);
+    }
     return key;
   }
 
@@ -62,7 +66,8 @@ export class InteractionStore {
     outcome: string,
     resolvedAt = Date.now(),
   ): void {
-    const existing = this.find(kind, sessionId, requestId, responseId);
+    const existing =
+      this.find(kind, sessionId, requestId, responseId) || this.find(kind, sessionId, requestId);
     const key = existing?.key || `${sessionId}:${responseId}:${requestId}`;
     const normalized = outcome.replaceAll('_', '-');
     const state: InteractionRecord['state'] =
@@ -80,6 +85,7 @@ export class InteractionStore {
       (kind === 'approval'
         ? ({ sessionId, id: requestId, title: 'Access request' } satisfies ApprovalPrompt)
         : ({ sessionId, callId: requestId, questions: [] } satisfies AskUserPrompt));
+    const changed = !existing || existing.state !== state || existing.outcome !== outcome;
     this.interactions.value = {
       ...this.interactions.peek(),
       [key]: {
@@ -101,6 +107,7 @@ export class InteractionStore {
     this.services.bumpDiagnostic('interactionReconciliations');
     if (kind === 'approval' && this.approval.peek()?.id === requestId) this.approval.value = null;
     if (kind === 'ask-user' && this.askUser.peek()?.callId === requestId) this.askUser.value = null;
+    if (changed) this.publish('interaction-changed', sessionId, responseId);
   }
 
   find(