fix(web): repair response stream sequence gaps

cmd/serve_response_runs.go

@@ -211,6 +211,8 @@ func (r *responseRun) appendEventLocked(event string, payload map[string]any, te
 
 	data, err := json.Marshal(payload)
 	if err != nil {
+		r.lastSequenceNumber--
+		delete(payload, "sequence_number")
 		return err
 	}
 
@@ -290,6 +292,7 @@ func (r *responseRun) appendTextDeltaEvent(outputIndex int, delta string) error
 
 	data, err := encodeTextDeltaPayload(outputIndex, delta, r.lastSequenceNumber)
 	if err != nil {
+		r.lastSequenceNumber--
 		return err
 	}
 

cmd/serve_response_runs_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"math"
 	"net/http"
 	"strings"
 	"sync"
@@ -32,6 +33,42 @@ func TestEncodeTextDeltaPayloadMatchesJSONMarshalForInvalidUTF8(t *testing.T) {
 	}
 }
 
+func TestResponseRunAppendEventMarshalErrorDoesNotBurnSequence(t *testing.T) {
+	run := newResponseRun("resp_marshal_gap", "sess_test", "", "mock", time.Now().Unix(), func() {})
+
+	if err := run.appendEvent("response.created", map[string]any{"ok": true}); err != nil {
+		t.Fatalf("append initial event: %v", err)
+	}
+	if err := run.appendEvent("response.bad", map[string]any{"bad": math.Inf(1)}); err == nil {
+		t.Fatal("appendEvent with non-marshalable payload succeeded, want error")
+	}
+	if err := run.appendEvent("response.phase", map[string]any{"text": "after"}); err != nil {
+		t.Fatalf("append event after marshal error: %v", err)
+	}
+
+	run.mu.Lock()
+	defer run.mu.Unlock()
+	if run.lastSequenceNumber != 2 {
+		t.Fatalf("lastSequenceNumber = %d, want 2", run.lastSequenceNumber)
+	}
+	if len(run.events) != 2 {
+		t.Fatalf("events = %d, want 2", len(run.events))
+	}
+	for i, ev := range run.events {
+		want := int64(i + 1)
+		if ev.Sequence != want {
+			t.Fatalf("event[%d].Sequence = %d, want %d", i, ev.Sequence, want)
+		}
+		var payload map[string]any
+		if err := json.Unmarshal(ev.Data, &payload); err != nil {
+			t.Fatalf("event[%d] payload unmarshal: %v", i, err)
+		}
+		if payload["sequence_number"] != float64(want) {
+			t.Fatalf("event[%d] payload sequence_number = %#v, want %d", i, payload["sequence_number"], want)
+		}
+	}
+}
+
 func TestAppendResponseRunEventEmitsPhase(t *testing.T) {
 	run := newResponseRun("resp_phase", "sess_test", "", "mock", time.Now().Unix(), func() {})
 	server := &serveServer{}

internal/serveui/static/app-stream.js

@@ -1125,6 +1125,11 @@ const consumeResponseStream = async (stream, session, streamState, options = {})
   const sessionId = String(session?.id || '').trim();
   const expectedResponseId = String(options.responseId || '').trim();
 
+  const eventSequenceNumber = (payload) => {
+    const seq = Number(payload?.sequence_number);
+    return Number.isFinite(seq) && seq > 0 ? seq : 0;
+  };
+
   const streamIsCurrent = () => {
     if (generation !== state.streamGeneration) return false;
     const currentSessionId = state.currentStreamSessionId;
@@ -1163,7 +1168,27 @@ const consumeResponseStream = async (stream, session, streamState, options = {})
       stale = true;
       return false;
     }
-    const result = applyResponseStreamEvent(session, streamState, event, payload);
+    const seq = eventSequenceNumber(payload);
+    const currentSeq = Number(session.lastSequenceNumber || 0);
+    if (seq > currentSeq + 1) {
+      terminalError = {
+        message: `response event stream gap: expected sequence ${currentSeq + 1}, received ${seq}`,
+        recoverableStreamGap: true,
+      };
+      return false;
+    }
+
+    let result;
+    try {
+      result = applyResponseStreamEvent(session, streamState, event, payload);
+    } catch (err) {
+      session.lastSequenceNumber = currentSeq;
+      terminalError = {
+        message: err?.message || 'response event projection failed',
+        recoverableStreamApplyFailure: true,
+      };
+      return false;
+    }
     if (result?.terminal) {
       sawTerminal = true;
     }
@@ -1443,6 +1468,25 @@ const resumeActiveResponseInner = async (session, responseId, options) => {
         setStreaming(Boolean(state.currentStreamResponseId));
         return false;
       }
+      if (result.error?.recoverableStreamGap || result.error?.recoverableStreamApplyFailure) {
+        const sequenceBeforeRecovery = Number(session.lastSequenceNumber || 0);
+        try {
+          const snapshot = await recoverResponseStateFromSnapshot(session, responseId);
+          streamState = createResponseStreamState(session);
+          if (snapshot?.status !== 'in_progress') {
+            setStreaming(Boolean(state.currentStreamResponseId));
+            return true;
+          }
+          if (Number(session.lastSequenceNumber || 0) > sequenceBeforeRecovery) {
+            continue;
+          }
+        } catch (snapshotErr) {
+          if (snapshotErr?.status === 401) {
+            handleAuthFailure();
+            return false;
+          }
+        }
+      }
       if (session.activeResponseId !== responseId) {
         setStreaming(Boolean(state.currentStreamResponseId));
         return true;

internal/serveui/static/app_stream_test.js

@@ -2249,6 +2249,81 @@ async function testResumeActiveResponseRecoversFromSnapshotBeforeReplaying() {
   pass(name);
 }
 
+async function testResumeActiveResponseRepairsSequenceGapWithSnapshot() {
+  const name = 'resumeActiveResponse repairs replay sequence gaps with snapshot';
+  const responseId = 'resp_gap_repair';
+  const harness = createHarness({
+    responseId,
+    snapshotPayload: {
+      id: responseId,
+      status: 'completed',
+      last_sequence_number: 9,
+      recovery: {
+        sequence_number: 9,
+        messages: [
+          { id: 'assistant-recovered', role: 'assistant', content: 'complete recovered answer', created: 1001 },
+        ],
+      },
+    },
+    eventsBody: [
+      'id: 7\n',
+      'event: response.output_text.delta\n',
+      'data: {"delta":"tail only","sequence_number":7}\n\n',
+      'id: 8\n',
+      'event: response.output_text.delta\n',
+      'data: {"delta":" should not apply","sequence_number":8}\n\n',
+      'id: 9\n',
+      'event: response.completed\n',
+      `data: {"response":{"id":"${responseId}","model":"test-model","status":"completed"},"sequence_number":9}\n\n`,
+      'data: [DONE]\n\n',
+    ].join(''),
+  });
+
+  const { app, state, fetchCalls, cleanup } = harness;
+  const session = {
+    id: 'session_gap_repair',
+    title: 'Gap repair',
+    messages: [
+      { id: 'msg_user_local', role: 'user', content: 'go', created: 1000 },
+    ],
+    activeResponseId: responseId,
+    lastSequenceNumber: 4,
+    number: 1,
+  };
+  state.sessions.push(session);
+  state.activeSessionId = session.id;
+
+  await app.resumeActiveResponse(session, { responseId });
+
+  const eventsCall = fetchCalls.find((call) => call.url.startsWith(`/ui/v1/responses/${responseId}/events?after=`));
+  if (!eventsCall || !eventsCall.url.endsWith('after=4')) {
+    fail(name, 'expected replay request from local sequence 4', eventsCall ? eventsCall.url : JSON.stringify(fetchCalls));
+    await cleanup();
+    return;
+  }
+  const snapshotCall = fetchCalls.find((call) => call.url === `/ui/v1/responses/${responseId}`);
+  if (!snapshotCall) {
+    fail(name, 'expected snapshot fetch after detecting sequence gap', JSON.stringify(fetchCalls));
+    await cleanup();
+    return;
+  }
+  const assistantMessages = session.messages.filter((message) => message.role === 'assistant');
+  if (assistantMessages.length !== 1 || assistantMessages[0].content !== 'complete recovered answer') {
+    fail(name, 'expected gapped replay tail to be discarded in favor of snapshot', JSON.stringify(session.messages));
+    await cleanup();
+    return;
+  }
+  if (session.lastSequenceNumber !== 0) {
+    // completed snapshot clears active tracking, which resets the replay cursor.
+    fail(name, `lastSequenceNumber = ${session.lastSequenceNumber}, want 0 after completed snapshot`);
+    await cleanup();
+    return;
+  }
+
+  await cleanup();
+  pass(name);
+}
+
 async function testRecoverySnapshotClearsSyntheticPendingInterjectionByText() {
   const name = 'recovery snapshot clears pending interjection tracked under synthetic id';
   const responseId = 'resp_snapshot_interjection_cleanup';
@@ -4739,6 +4814,7 @@ function testCompletedResponseClearsUnappliedQueuedEffort() {
   await testDrainInterruptQueueAfterResumeCompletes();
   await testDrainInterruptQueueIgnoresOtherSessionEntries();
   await testResumeActiveResponseRecoversFromSnapshotBeforeReplaying();
+  await testResumeActiveResponseRepairsSequenceGapWithSnapshot();
   await testRecoverySnapshotClearsSyntheticPendingInterjectionByText();
   await testRecoverySnapshotDoesNotDuplicateOptimisticInterjection();
   await testFunctionCallArgumentDeltasFillToolPrompt();