fix(server): classify client-disconnect during streaming + live-preview e2e scenarios

internal/server/handlers_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"maps"
 	"mime/multipart"
@@ -15,6 +16,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"syscall"
 	"testing"
 	"time"
 
@@ -2651,6 +2653,180 @@ func TestHandleStreamingResponse_RecordsStreamingError(t *testing.T) {
 	}
 }
 
+func TestHandleStreamingResponse_ClientDisconnectBeforeUpstream(t *testing.T) {
+	e := echo.New()
+	handler := NewHandler(&mockProvider{}, nil, nil, nil)
+
+	req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+	ctx, cancel := context.WithCancel(req.Context())
+	cancel() // simulate client gone before streamFn returns
+	req = req.WithContext(ctx)
+	rec := httptest.NewRecorder()
+	c := e.NewContext(req, rec)
+	entry := &auditlog.LogEntry{
+		ID:        "entry-cancel",
+		Timestamp: time.Now(),
+		Method:    http.MethodPost,
+		Path:      "/v1/chat/completions",
+		Data:      &auditlog.LogData{},
+	}
+	c.Set(string(auditlog.LogEntryKey), entry)
+
+	err := handler.translatedInference().handleStreamingResponse(c, nil, "gpt-4o-mini", "openai", "primary-openai", func() (io.ReadCloser, error) {
+		return nil, context.Canceled
+	})
+	if err != nil {
+		t.Fatalf("handler returned error: %v", err)
+	}
+
+	if !entry.Stream {
+		t.Fatalf("expected entry.Stream=true, got false")
+	}
+	if entry.ErrorType != "client_disconnected" {
+		t.Fatalf("expected error_type client_disconnected, got %q", entry.ErrorType)
+	}
+}
+
+// At pre-flush dispatch time the only socket in play is the upstream
+// provider connection, so EPIPE / ECONNRESET on the error from streamFn
+// belong to the provider and must surface as upstream failures rather than
+// be swallowed as client disconnects.
+func TestHandleStreamingResponse_UpstreamResetIsNotClassifiedAsClientDisconnect(t *testing.T) {
+	logger := &capturingAuditLogger{
+		config: auditlog.Config{Enabled: true},
+	}
+
+	e := echo.New()
+	handler := NewHandler(&mockProvider{}, logger, nil, nil)
+
+	for _, tt := range []struct {
+		name string
+		err  error
+	}{
+		{name: "bare syscall.ECONNRESET", err: syscall.ECONNRESET},
+		{name: "wrapped syscall.EPIPE", err: fmt.Errorf("dial upstream: %w", syscall.EPIPE)},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+			rec := httptest.NewRecorder()
+			c := e.NewContext(req, rec)
+			entry := &auditlog.LogEntry{
+				ID:        "entry-upstream-reset",
+				Timestamp: time.Now(),
+				Method:    http.MethodPost,
+				Path:      "/v1/chat/completions",
+				Data:      &auditlog.LogData{},
+			}
+			c.Set(string(auditlog.LogEntryKey), entry)
+
+			err := handler.translatedInference().handleStreamingResponse(c, nil, "gpt-4o-mini", "openai", "primary-openai", func() (io.ReadCloser, error) {
+				return nil, tt.err
+			})
+
+			// handleStreamingResponse always swallows the error by writing a
+			// JSON response via handleError; the gateway response must be the
+			// upstream failure, not an empty 200.
+			if err != nil {
+				t.Fatalf("handler returned error: %v", err)
+			}
+			if rec.Code == http.StatusOK {
+				t.Fatalf("upstream reset surfaced as 200 OK; want non-2xx, got body=%q", rec.Body.String())
+			}
+			if entry.ErrorType == "client_disconnected" {
+				t.Fatalf("upstream reset misclassified as client_disconnected (err=%v)", tt.err)
+			}
+			if !entry.Stream {
+				t.Fatalf("expected entry.Stream=true regardless of classification, got false")
+			}
+		})
+	}
+}
+
+func TestRecordStreamingError_ClassifiesClientDisconnect(t *testing.T) {
+	canceledCtx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	tests := []struct {
+		name      string
+		ctx       context.Context
+		err       error
+		wantType  string
+	}{
+		{
+			name:     "explicit context.Canceled",
+			ctx:      context.Background(),
+			err:      context.Canceled,
+			wantType: "client_disconnected",
+		},
+		{
+			name:     "wrapped context.Canceled",
+			ctx:      context.Background(),
+			err:      fmt.Errorf("upstream send failed: %w", context.Canceled),
+			wantType: "client_disconnected",
+		},
+		{
+			name:     "syscall.EPIPE",
+			ctx:      context.Background(),
+			err:      syscall.EPIPE,
+			wantType: "client_disconnected",
+		},
+		{
+			name:     "wrapped syscall.EPIPE",
+			ctx:      context.Background(),
+			err:      fmt.Errorf("write to client: %w", syscall.EPIPE),
+			wantType: "client_disconnected",
+		},
+		{
+			name:     "syscall.ECONNRESET",
+			ctx:      context.Background(),
+			err:      syscall.ECONNRESET,
+			wantType: "client_disconnected",
+		},
+		{
+			name:     "canceled ctx racing real upstream error stays stream_error",
+			ctx:      canceledCtx,
+			err:      errors.New("upstream malformed"),
+			wantType: "stream_error",
+		},
+		{
+			name:     "clean ctx and generic error",
+			ctx:      context.Background(),
+			err:      errors.New("upstream malformed"),
+			wantType: "stream_error",
+		},
+		{
+			// Exercises the err==nil branch of isClientDisconnect and the
+			// matching nil-guard fallback in recordStreamingError. Must not
+			// panic and must record the context error as the message.
+			name:     "canceled ctx with nil err",
+			ctx:      canceledCtx,
+			err:      nil,
+			wantType: "client_disconnected",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			entry := &auditlog.LogEntry{Data: &auditlog.LogData{}}
+			recordStreamingError(entry, "gpt-4o-mini", "openai", "/v1/chat/completions", "req-"+tt.name, tt.ctx, tt.err)
+			if entry.ErrorType != tt.wantType {
+				t.Fatalf("error_type = %q, want %q", entry.ErrorType, tt.wantType)
+			}
+
+			wantMessage := ""
+			switch {
+			case tt.err != nil:
+				wantMessage = tt.err.Error()
+			case tt.ctx != nil && tt.ctx.Err() != nil:
+				wantMessage = tt.ctx.Err().Error()
+			}
+			if entry.Data.ErrorMessage != wantMessage {
+				t.Fatalf("error_message = %q, want %q", entry.Data.ErrorMessage, wantMessage)
+			}
+		})
+	}
+}
+
 func TestChatCompletionStreaming_FlushesBeforeNextChunkArrives(t *testing.T) {
 	secondChunkStarted := make(chan struct{})
 	releaseSecondChunk := make(chan struct{})

internal/server/passthrough_support.go

@@ -305,7 +305,7 @@ func (s *passthroughService) proxyPassthroughResponse(c *echo.Context, providerT
 
 		c.Response().WriteHeader(resp.StatusCode)
 		if err := flushStream(c.Response(), wrappedStream); err != nil {
-			recordStreamingError(streamEntry, model, providerType, c.Request().URL.Path, requestID, err)
+			recordStreamingError(streamEntry, model, providerType, c.Request().URL.Path, requestID, c.Request().Context(), err)
 			return err
 		}
 		return nil

internal/server/translated_inference_service.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"syscall"
 
 	"github.com/labstack/echo/v5"
 
@@ -94,7 +95,7 @@ func (s *translatedInferenceService) dispatchChatCompletion(c *echo.Context, req
 		}
 		result, err := s.inference().StreamChatCompletion(ctx, workflow, req)
 		if err != nil {
-			return handleError(c, err)
+			return handleStreamingDispatchError(c, err)
 		}
 		if result.Meta.UsedFallback {
 			markRequestFallbackUsed(c)
@@ -237,7 +238,7 @@ func (s *translatedInferenceService) dispatchResponses(c *echo.Context, req *cor
 	if req.Stream {
 		result, err := s.inference().StreamResponses(ctx, workflow, req)
 		if err != nil {
-			return handleError(c, err)
+			return handleStreamingDispatchError(c, err)
 		}
 		if result.Meta.UsedFallback {
 			markRequestFallbackUsed(c)
@@ -485,7 +486,7 @@ func (s *translatedInferenceService) handleStreamingReadCloser(
 
 	c.Response().WriteHeader(http.StatusOK)
 	if err := flushStream(c.Response(), wrappedStream); err != nil {
-		recordStreamingError(streamEntry, model, provider, c.Request().URL.Path, requestID, err)
+		recordStreamingError(streamEntry, model, provider, c.Request().URL.Path, requestID, c.Request().Context(), err)
 	}
 	return nil
 }
@@ -498,29 +499,89 @@ func (s *translatedInferenceService) handleStreamingResponse(
 ) error {
 	stream, err := streamFn()
 	if err != nil {
-		return handleError(c, err)
+		return handleStreamingDispatchError(c, err)
 	}
 	return s.handleStreamingReadCloser(c, workflow, model, provider, providerName, "", stream)
 }
 
-func recordStreamingError(streamEntry *auditlog.LogEntry, model, provider, path, requestID string, err error) {
+// handleStreamingDispatchError records audit context for a streaming request
+// that failed before any chunks could be flushed. It marks the entry as
+// streaming and distinguishes client cancellations from upstream failures so
+// the audit log reflects the actual cause.
+func handleStreamingDispatchError(c *echo.Context, err error) error {
+	auditlog.EnrichEntryWithStream(c, true)
+	if isClientDisconnectDuringDispatch(c.Request().Context(), err) {
+		auditlog.EnrichEntryWithError(c, "client_disconnected", err.Error(), "")
+		return nil
+	}
+	return handleError(c, err)
+}
+
+func recordStreamingError(streamEntry *auditlog.LogEntry, model, provider, path, requestID string, ctx context.Context, err error) {
+	errorType := "stream_error"
+	if isClientDisconnect(ctx, err) {
+		errorType = "client_disconnected"
+	}
+
+	// The nil-err branch in isClientDisconnect is reachable for callers that
+	// only have a canceled context to report. Fall back to the context error
+	// in that case so we never dereference a nil error.
+	logErr := err
+	errorMessage := ""
+	switch {
+	case err != nil:
+		errorMessage = err.Error()
+	case ctx != nil && ctx.Err() != nil:
+		logErr = ctx.Err()
+		errorMessage = logErr.Error()
+	}
+
 	if streamEntry != nil {
-		streamEntry.ErrorType = "stream_error"
+		streamEntry.ErrorType = errorType
 		if streamEntry.Data == nil {
 			streamEntry.Data = &auditlog.LogData{}
 		}
-		streamEntry.Data.ErrorMessage = err.Error()
+		streamEntry.Data.ErrorMessage = errorMessage
 	}
 
 	slog.Warn("stream terminated abnormally",
-		"error", err,
+		"error", logErr,
+		"error_type", errorType,
 		"model", model,
 		"provider", provider,
 		"path", path,
 		"request_id", requestID,
 	)
 }
 
+// isClientDisconnect classifies write-phase streaming errors (errors returned
+// after the gateway has begun writing the SSE response back to the client). At
+// this phase EPIPE / ECONNRESET on the response writer can only come from the
+// downstream client connection, so they are treated as client disconnects. The
+// nil-err / canceled-context branch supports callers that only have a context
+// signal to report.
+func isClientDisconnect(ctx context.Context, err error) bool {
+	if errors.Is(err, context.Canceled) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) {
+		return true
+	}
+	return err == nil && ctx != nil && ctx.Err() == context.Canceled
+}
+
+// isClientDisconnectDuringDispatch classifies a streaming dispatch error - one
+// that happened before any response bytes were flushed to the client. At this
+// phase the only socket in play is the upstream provider connection, so
+// EPIPE / ECONNRESET on err belong to the provider and must NOT be swallowed
+// as client disconnects. Only a cancellation of the request context proves
+// the client is gone. The ctx-only branch still requires err == nil so a
+// concrete upstream failure racing with a cancellation surfaces as a real
+// upstream error.
+func isClientDisconnectDuringDispatch(ctx context.Context, err error) bool {
+	if errors.Is(err, context.Canceled) {
+		return true
+	}
+	return err == nil && ctx != nil && ctx.Err() == context.Canceled
+}
+
 func providerNameFromWorkflow(workflow *core.Workflow) string {
 	return gateway.ProviderNameFromWorkflow(workflow)
 }