diff --git a/cmd/serve_response_runs.go b/cmd/serve_response_runs.go index ccc6a18f..359a8c63 100644 --- a/cmd/serve_response_runs.go +++ b/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 } diff --git a/cmd/serve_response_runs_test.go b/cmd/serve_response_runs_test.go index d9a1eaf5..dc4d7a27 100644 --- a/cmd/serve_response_runs_test.go +++ b/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{} diff --git a/internal/serveui/static/app-stream.js b/internal/serveui/static/app-stream.js index c3242ee3..b6cc0957 100644 --- a/internal/serveui/static/app-stream.js +++ b/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; diff --git a/internal/serveui/static/app_stream_test.js b/internal/serveui/static/app_stream_test.js index 43b4c977..bb202e5b 100644 --- a/internal/serveui/static/app_stream_test.js +++ b/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();