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

@@ -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()

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
 	})
 }

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()