From 26b64fa5236574b25b407f1eb5166b115a850514 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Wed, 22 Apr 2026 15:02:02 +0200 Subject: [PATCH] fix(coderd/x/chatd/chatdebug): record SSE attempts on EOF (#24565) `chat_turn` debug steps persist with `attempts: []` even when the streaming call to Anthropic completes successfully. Fantasy's Anthropic SSE adapter iterates the response to EOF via `for stream.Next()` and abandons the body without calling `Close()`, so `RecordingTransport`'s Close-only recording path never fires and the attempt is lost. Non-streaming runs (`quickgen`, `title_generation`) go through `model.Generate(...)` and are unaffected. Record on `io.EOF` for `text/event-stream` bodies specifically. Non-SSE responses stay on the Close-only path so JSON integrity, content-length validation, and inner-`Close()` error semantics are preserved. `record()` is already `sync.Once`-guarded, so a later `Close()` is a no-op for recording. --- coderd/x/chatd/chatdebug/recorder.go | 18 ++ coderd/x/chatd/chatdebug/transport.go | 156 ++++++++++++----- coderd/x/chatd/chatdebug/transport_test.go | 188 +++++++++++++++++++++ 3 files changed, 321 insertions(+), 41 deletions(-) diff --git a/coderd/x/chatd/chatdebug/recorder.go b/coderd/x/chatd/chatdebug/recorder.go index 04f5c8a11e..df015e31ec 100644 --- a/coderd/x/chatd/chatdebug/recorder.go +++ b/coderd/x/chatd/chatdebug/recorder.go @@ -56,6 +56,24 @@ func (s *attemptSink) record(a Attempt) { s.attempts = append(s.attempts, a) } +// replaceByNumber overwrites a previously recorded attempt whose Number +// matches. If no match is found, the attempt is appended. This supports +// the provisional-then-upgrade flow used for SSE bodies where Read() +// records a completed attempt on EOF and Close() later needs to replace +// it with a failed attempt when inner.Close() surfaces an error. +func (s *attemptSink) replaceByNumber(number int, a Attempt) { + s.mu.Lock() + defer s.mu.Unlock() + + for i := range s.attempts { + if s.attempts[i].Number == number { + s.attempts[i] = a + return + } + } + s.attempts = append(s.attempts, a) +} + func (s *attemptSink) snapshot() []Attempt { s.mu.Lock() defer s.mu.Unlock() diff --git a/coderd/x/chatd/chatdebug/transport.go b/coderd/x/chatd/chatdebug/transport.go index 20a05f344e..102de968af 100644 --- a/coderd/x/chatd/chatdebug/transport.go +++ b/coderd/x/chatd/chatdebug/transport.go @@ -191,6 +191,11 @@ type recordingBody struct { truncated bool sawEOF bool bytesRead int64 + // recordedProvisional is true when recordProvisional() has fired + // for an SSE body's Read-path EOF but Close() has not yet run. A + // subsequent inner.Close() error in Close() upgrades the + // provisional entry in the sink so the close error is not lost. + recordedProvisional bool recordOnce sync.Once closeOnce sync.Once @@ -225,12 +230,24 @@ func (r *recordingBody) Read(p []byte) (int, error) { r.accumulateReadLocked(p, n, err) r.mu.Unlock() - // Only record non-EOF errors immediately. io.EOF is deferred - // to Close() which runs more sophisticated validation (JSON - // completeness checks, content-length verification, etc.). - // Recording EOF here would preempt Close() via recordOnce. + // Record non-EOF errors immediately. EOF is handled + // below for SSE or deferred to Close() for validation. if err != nil && !errors.Is(err, io.EOF) { r.record(err) + return n, err + } + + // For server-sent-events bodies, record eagerly on EOF. Streaming + // consumers like fantasy's Anthropic SSE adapter iterate the + // response to EOF and abandon it without calling Close(), so the + // Close-only recording path would never fire and the attempt would + // be lost. The recording is provisional so Close() can still + // upgrade it to failed if inner.Close() surfaces a transport error. + // Non-SSE bodies stay on the Close-only path so that JSON + // integrity, content-length validation, and inner-Close errors + // keep their existing semantics. + if errors.Is(err, io.EOF) && isSSEContentType(r.contentType) { + r.recordProvisional(io.EOF) } return n, err } @@ -259,7 +276,27 @@ func (r *recordingBody) Close() error { closeErr = r.inner.Close() }) if closeErr != nil { - r.record(closeErr) + // Hold r.mu across the flag check AND the publish/replace so a + // concurrent recordProvisional cannot slip its recordOnce + // publish between our read of recordedProvisional and our call + // into the sink. Without this serialization, Close() could + // observe recordedProvisional=false, then lose the race and + // see r.record(closeErr) become a no-op once recordOnce has + // already fired from the SSE EOF path. + r.mu.Lock() + if r.recordedProvisional { + // The SSE EOF path already appended a completed attempt. + // inner.Close() surfaced a transport error, so upgrade + // that entry to failed instead of losing the close error. + upgraded := r.buildAttemptLocked(closeErr) + r.sink.replaceByNumber(upgraded.Number, upgraded) + r.recordedProvisional = false + } else { + r.recordOnce.Do(func() { + r.sink.record(r.buildAttemptLocked(closeErr)) + }) + } + r.mu.Unlock() return closeErr } @@ -334,6 +371,12 @@ func isNDJSONContentType(contentType string) bool { return parseMediaType(contentType) == "application/x-ndjson" } +// isSSEContentType reports whether contentType is a +// server-sent-events stream. +func isSSEContentType(contentType string) bool { + return parseMediaType(contentType) == "text/event-stream" +} + // maxDrainBytes caps how many trailing bytes drainToEOF will consume. // This prevents Close() from blocking indefinitely on a misbehaving // or extremely large chunked body. @@ -384,44 +427,75 @@ func isCompleteUnknownLengthJSONBody(contentType string, body []byte) bool { return errors.Is(decoder.Decode(&extra), io.EOF) } +// buildAttemptLocked materializes the final Attempt from the current +// buffered response data plus err. Callers use this from both the +// record-once append path and the provisional-upgrade replace path so +// both sites apply the same redaction and status rules. The caller +// must hold r.mu for the duration of the call. +func (r *recordingBody) buildAttemptLocked(err error) Attempt { + finishedAt := time.Now() + + truncated := r.truncated + responseBody := append([]byte(nil), r.buf.Bytes()...) + base := r.base + startedAt := r.startedAt + + contentType := r.contentType + switch { + case truncated: + base.ResponseBody = []byte("[TRUNCATED]") + case isNDJSONContentType(contentType): + base.ResponseBody = RedactNDJSONSecrets(responseBody) + case contentType == "" || isJSONLikeContentType(contentType): + // Redact JSON secrets when the content type is JSON-like + // or absent (unknown). For unknown types, RedactJSONSecrets + // fails closed by replacing non-JSON payloads with a + // diagnostic message. + base.ResponseBody = RedactJSONSecrets(responseBody) + default: + // Non-JSON content types (SSE, text/plain, HTML, etc.) + // are preserved as-is to avoid losing debug content. + base.ResponseBody = responseBody + } + base.StartedAt = startedAt.UTC().Format(time.RFC3339Nano) + base.FinishedAt = finishedAt.UTC().Format(time.RFC3339Nano) + // Recompute duration to include body read time. + base.DurationMs = finishedAt.Sub(startedAt).Milliseconds() + if err != nil && !errors.Is(err, io.EOF) { + base.Error = sanitizeErrorString(err.Error()) + base.Status = attemptStatusFailed + } else { + base.Status = attemptStatusCompleted + } + return base +} + +// record acquires r.mu before entering recordOnce.Do so it shares a +// single lock-acquisition order with recordProvisional. Without this, +// a concurrent Read (in recordProvisional, holding r.mu) and Close (in +// record, about to take r.mu inside the Do callback) would deadlock: +// the Do winner would block on r.mu while the loser would block on +// recordOnce. Callers must not hold r.mu. func (r *recordingBody) record(err error) { + r.mu.Lock() + defer r.mu.Unlock() r.recordOnce.Do(func() { - finishedAt := time.Now() + r.sink.record(r.buildAttemptLocked(err)) + }) +} - r.mu.Lock() - truncated := r.truncated - responseBody := append([]byte(nil), r.buf.Bytes()...) - base := r.base - startedAt := r.startedAt - r.mu.Unlock() - - contentType := r.contentType - switch { - case truncated: - base.ResponseBody = []byte("[TRUNCATED]") - case isNDJSONContentType(contentType): - base.ResponseBody = RedactNDJSONSecrets(responseBody) - case contentType == "" || isJSONLikeContentType(contentType): - // Redact JSON secrets when the content type is JSON-like - // or absent (unknown). For unknown types, RedactJSONSecrets - // fails closed by replacing non-JSON payloads with a - // diagnostic message. - base.ResponseBody = RedactJSONSecrets(responseBody) - default: - // Non-JSON content types (SSE, text/plain, HTML, etc.) - // are preserved as-is to avoid losing debug content. - base.ResponseBody = responseBody - } - base.StartedAt = startedAt.UTC().Format(time.RFC3339Nano) - base.FinishedAt = finishedAt.UTC().Format(time.RFC3339Nano) - // Recompute duration to include body read time. - base.DurationMs = finishedAt.Sub(startedAt).Milliseconds() - if err != nil && !errors.Is(err, io.EOF) { - base.Error = sanitizeErrorString(err.Error()) - base.Status = attemptStatusFailed - } else { - base.Status = attemptStatusCompleted - } - r.sink.record(base) +// recordProvisional records err via recordOnce and marks the entry as +// eligible for a later upgrade from Close(). Safe to call multiple +// times; only the first call appends. The publish and the provisional +// flag are committed atomically under r.mu so a concurrent Close() +// that takes r.mu to inspect the flag cannot observe a half-finished +// state where the attempt is in the sink but recordedProvisional is +// still false. +func (r *recordingBody) recordProvisional(err error) { + r.mu.Lock() + defer r.mu.Unlock() + r.recordOnce.Do(func() { + r.sink.record(r.buildAttemptLocked(err)) + r.recordedProvisional = true }) } diff --git a/coderd/x/chatd/chatdebug/transport_test.go b/coderd/x/chatd/chatdebug/transport_test.go index 06f2a29997..0b5b9448e8 100644 --- a/coderd/x/chatd/chatdebug/transport_test.go +++ b/coderd/x/chatd/chatdebug/transport_test.go @@ -8,10 +8,14 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" + "time" "github.com/stretchr/testify/require" "golang.org/x/xerrors" + + "github.com/coder/coder/v2/testutil" ) func newTestSinkContext(t *testing.T) (context.Context, *attemptSink) { @@ -825,6 +829,190 @@ func TestRecordingTransport_SSEReadToEOFMarksCompleted(t *testing.T) { require.Equal(t, ssePayload, string(attempts[0].ResponseBody)) } +// TestRecordingTransport_SSEReadToEOFWithoutCloseStillRecords verifies +// that SSE consumers that reach EOF and abandon the response without +// calling Close() (the pattern fantasy's Anthropic SSE adapter follows) +// still populate the attempt sink. Close()-only recording would leave +// the chat_turn step's attempts field permanently empty. +func TestRecordingTransport_SSEReadToEOFWithoutCloseStillRecords(t *testing.T) { + t.Parallel() + + ctx, sink := newTestSinkContext(t) + ssePayload := "data: {\"token\":\"secret\"}\n\ndata: [DONE]\n\n" + client := &http.Client{ + Transport: &RecordingTransport{ + Base: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ //nolint:exhaustruct // Test SSE content type. + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ssePayload)), + ContentLength: -1, + }, nil + }), + }, + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://example.invalid", nil) + require.NoError(t, err) + + resp, err := client.Do(req) //nolint:bodyclose // Intentionally skip Close() to verify EOF-only recording. + require.NoError(t, err) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, ssePayload, string(body)) + // Deliberately do NOT call resp.Body.Close(). The attempt must be + // recorded on EOF alone. + + attempts := sink.snapshot() + require.Len(t, attempts, 1) + require.Equal(t, attemptStatusCompleted, attempts[0].Status) + require.Empty(t, attempts[0].Error) + require.Equal(t, ssePayload, string(attempts[0].ResponseBody)) +} + +// TestRecordingTransport_SSEEmptyBodyRecordsOnEOF verifies that an SSE +// response with zero bytes (immediate EOF on the first Read) still +// records a completed attempt. This covers the n == 0 && err == io.EOF +// branch in accumulateReadLocked where the buffer path is skipped but +// sawEOF must still fire the Read-path recording. +func TestRecordingTransport_SSEEmptyBodyRecordsOnEOF(t *testing.T) { + t.Parallel() + + ctx, sink := newTestSinkContext(t) + client := &http.Client{ + Transport: &RecordingTransport{ + Base: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ //nolint:exhaustruct // Test SSE content type. + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader("")), + ContentLength: -1, + }, nil + }), + }, + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://example.invalid", nil) + require.NoError(t, err) + + resp, err := client.Do(req) //nolint:bodyclose // Intentionally skip Close() to verify EOF-only recording. + require.NoError(t, err) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Empty(t, body) + + attempts := sink.snapshot() + require.Len(t, attempts, 1) + require.Equal(t, attemptStatusCompleted, attempts[0].Status) + require.Empty(t, attempts[0].Error) + require.Empty(t, attempts[0].ResponseBody) +} + +// TestRecordingTransport_SSEReadToEOFWithCloseErrorUpgrades verifies +// that when an SSE consumer reads to EOF (which eagerly records the +// attempt as completed) and then Close() fails because inner.Close() +// returns an error, the recorded attempt is upgraded to failed with +// the close error rather than silently remaining completed. +func TestRecordingTransport_SSEReadToEOFWithCloseErrorUpgrades(t *testing.T) { + t.Parallel() + + ctx, sink := newTestSinkContext(t) + ssePayload := "data: {\"token\":\"secret\"}\n\ndata: [DONE]\n\n" + closeErr := xerrors.New("boom: connection reset") + client := &http.Client{ + Transport: &RecordingTransport{ + Base: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ //nolint:exhaustruct // Test SSE content type. + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: &failingCloseReader{ + inner: strings.NewReader(ssePayload), + closeErr: closeErr, + }, + ContentLength: -1, + }, nil + }), + }, + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://example.invalid", nil) + require.NoError(t, err) + + resp, err := client.Do(req) + require.NoError(t, err) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, ssePayload, string(body)) + + // Close must surface the inner close error to the caller... + gotCloseErr := resp.Body.Close() + require.ErrorIs(t, gotCloseErr, closeErr) + + // ...and the recorded attempt must reflect that failure instead of + // the provisional completed entry written on EOF. + attempts := sink.snapshot() + require.Len(t, attempts, 1) + require.Equal(t, attemptStatusFailed, attempts[0].Status) + require.Contains(t, attempts[0].Error, "boom: connection reset") + require.Equal(t, ssePayload, string(attempts[0].ResponseBody)) +} + +// TestRecordingBody_SSEConcurrentReadCloseNoDeadlock exercises the +// lock-ordering contract between record() and recordProvisional() +// under concurrent Read/Close on an SSE body. An earlier revision +// where record() entered recordOnce.Do before acquiring r.mu (while +// recordProvisional() acquired r.mu first) deadlocked when one +// goroutine won the Once but then blocked on r.mu while the other +// held r.mu and blocked on the Once. +func TestRecordingBody_SSEConcurrentReadCloseNoDeadlock(t *testing.T) { + t.Parallel() + + const iterations = 200 + ssePayload := []byte("data: ping\n\n") + + for i := range iterations { + sink := &attemptSink{} + body := &recordingBody{ + inner: io.NopCloser(strings.NewReader(string(ssePayload))), + contentLength: -1, + contentType: "text/event-stream", + sink: sink, + startedAt: time.Now(), + base: Attempt{Number: sink.nextAttemptNumber()}, + } + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + buf := make([]byte, 64) + for { + if _, err := body.Read(buf); err != nil { + return + } + } + }() + go func() { + defer wg.Done() + _ = body.Close() + }() + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(testutil.WaitShort): + t.Fatalf("deadlock detected on iteration %d", i) + } + } +} + func TestRecordingTransport_SSEClosedEarlyMarksFailed(t *testing.T) { t.Parallel()