From b6ce1af390ca0a7f9cffe8be716a3e533add0236 Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sun, 17 May 2026 15:12:13 +0200 Subject: [PATCH 1/5] fix(server): classify client-disconnect during streaming dispatch Streaming requests that failed before any chunks were flushed were audited as upstream provider errors (status 502, error_type provider_error, stream null) even when the cause was the client closing the connection mid-handshake. The audit row no longer reflected the request's streaming intent at all, hiding cancellations from monitoring. Mark the audit entry as a streaming request and tag the error type as client_disconnected when ctx.Err is set or the error wraps context.Canceled / EPIPE / ECONNRESET. Routed both the dispatch-time failures (StreamChatCompletion, StreamResponses, handleStreamingResponse, passthrough flushStream) through a single helper so the classification stays consistent. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/server/handlers_test.go | 50 +++++++++++++++++++ internal/server/passthrough_support.go | 2 +- .../server/translated_inference_service.go | 41 ++++++++++++--- 3 files changed, 86 insertions(+), 7 deletions(-) diff --git a/internal/server/handlers_test.go b/internal/server/handlers_test.go index 74b48680..36f34e94 100644 --- a/internal/server/handlers_test.go +++ b/internal/server/handlers_test.go @@ -2651,6 +2651,56 @@ 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) + } +} + +func TestRecordStreamingError_ClassifiesClientDisconnect(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + entry := &auditlog.LogEntry{Data: &auditlog.LogData{}} + recordStreamingError(entry, "gpt-4o-mini", "openai", "/v1/chat/completions", "req-1", ctx, errors.New("broken pipe")) + if entry.ErrorType != "client_disconnected" { + t.Fatalf("expected client_disconnected, got %q", entry.ErrorType) + } + + entry2 := &auditlog.LogEntry{Data: &auditlog.LogData{}} + recordStreamingError(entry2, "gpt-4o-mini", "openai", "/v1/chat/completions", "req-2", context.Background(), errors.New("upstream malformed")) + if entry2.ErrorType != "stream_error" { + t.Fatalf("expected stream_error, got %q", entry2.ErrorType) + } +} + func TestChatCompletionStreaming_FlushesBeforeNextChunkArrives(t *testing.T) { secondChunkStarted := make(chan struct{}) releaseSecondChunk := make(chan struct{}) diff --git a/internal/server/passthrough_support.go b/internal/server/passthrough_support.go index c91ad322..82fe6ac2 100644 --- a/internal/server/passthrough_support.go +++ b/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 diff --git a/internal/server/translated_inference_service.go b/internal/server/translated_inference_service.go index 0483f2f5..719319b8 100644 --- a/internal/server/translated_inference_service.go +++ b/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,14 +499,32 @@ 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 isClientDisconnect(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" + } + if streamEntry != nil { - streamEntry.ErrorType = "stream_error" + streamEntry.ErrorType = errorType if streamEntry.Data == nil { streamEntry.Data = &auditlog.LogData{} } @@ -514,6 +533,7 @@ func recordStreamingError(streamEntry *auditlog.LogEntry, model, provider, path, slog.Warn("stream terminated abnormally", "error", err, + "error_type", errorType, "model", model, "provider", provider, "path", path, @@ -521,6 +541,15 @@ func recordStreamingError(streamEntry *auditlog.LogEntry, model, provider, path, ) } +// isClientDisconnect reports whether the streaming error was caused by the +// client closing the connection rather than an upstream failure. +func isClientDisconnect(ctx context.Context, err error) bool { + if ctx != nil && ctx.Err() != nil { + return true + } + return errors.Is(err, context.Canceled) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) +} + func providerNameFromWorkflow(workflow *core.Workflow) string { return gateway.ProviderNameFromWorkflow(workflow) } From 554ea8f8d4ffd3fa358b59f80cb9fdb3e0506a10 Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sun, 17 May 2026 15:12:43 +0200 Subject: [PATCH 2/5] test(e2e): refresh S22 model and add live-preview scenarios MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the retired grok-3-mini target in S22 with xai/grok-4.3 so the xAI smoke probe again hits a live model. Add ยง17 Dashboard live preview covering /admin/live/logs: S91 idle subscriber receives reset + heartbeat events S92 chat completion produces matching audit and usage events S93 types=usage filter excludes audit events S94 invalid cursor is rejected with 400 invalid_request_error S95 streaming client disconnect is audited as client_disconnected S95 acts as the regression test for the audit fix shipped in the previous commit. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/e2e/release-e2e-scenarios.md | 121 ++++++++++++++++++++++++++++- 1 file changed, 119 insertions(+), 2 deletions(-) diff --git a/tests/e2e/release-e2e-scenarios.md b/tests/e2e/release-e2e-scenarios.md index e99e7abb..d523931f 100644 --- a/tests/e2e/release-e2e-scenarios.md +++ b/tests/e2e/release-e2e-scenarios.md @@ -1,6 +1,6 @@ # Release E2E Curl Matrix -This file contains 89 end-to-end curl scenarios for release validation. +This file contains 95 end-to-end curl scenarios for release validation. These scenarios are prepared for execution across these local gateways: - `http://localhost:18080` - SQLite-backed main test gateway @@ -445,7 +445,7 @@ Checks translated chat on xAI and reasoning-token accounting. ```bash curl -fsS "$BASE_URL/v1/chat/completions" \ -H 'Content-Type: application/json' \ - -d '{"model":"grok-3-mini","messages":[{"role":"user","content":"Reply with exactly QA_XAI_OK"}],"max_tokens":20}' \ + -d '{"model":"xai/grok-4.3","messages":[{"role":"user","content":"Reply with exactly QA_XAI_OK"}],"max_tokens":20}' \ | jq -e '{model,provider,usage,answer:.choices[0].message.content}' ``` @@ -1548,3 +1548,120 @@ curl -fsS -X POST "$BASE_URL/admin/usage/recalculate-pricing" \ -d '{"confirmation":"recalculate"}' \ | jq -e '.status == "ok" and (.matched | type == "number") and (.recalculated | type == "number")' ``` + +## 17. Dashboard live preview + +These scenarios exercise the `/admin/live/logs` SSE feed that powers the +dashboard's realtime audit/usage panel. The auth/cache gateway is used because +it serves the dashboard and requires master-key authentication for admin +routes. + +### S91 Live preview heartbeat from an idle subscriber + +Subscribes with a future cursor (no replay), waits past one heartbeat interval, and asserts the SSE stream emits `event: reset` followed by at least one `event: heartbeat`. + +```bash +LIVE_OUT="$QA_RUN_DIR/s91.live.sse" +curl -sS --no-buffer -N "$AUTH_BASE_URL/admin/live/logs?types=audit,usage&cursor=999999999" \ + -H "$ADMIN_AUTH_HEADER" \ + --max-time 20 > "$LIVE_OUT" || true +grep -cE '^event: reset' "$LIVE_OUT" | jq -R -e 'tonumber >= 1' >/dev/null +grep -cE '^event: heartbeat' "$LIVE_OUT" | jq -R -e 'tonumber >= 1' >/dev/null +``` + +### S92 Live preview emits audit + usage events for a fresh chat + +Opens an SSE subscription, triggers one chat completion with a unique request id, then asserts the captured stream contains an `audit.*` event whose JSON payload references that request id and at least one `usage.*` event. + +```bash +LIVE_OUT="$QA_RUN_DIR/s92.live.sse" +RID="qa-live-preview-$QA_SUFFIX" +curl -sS --no-buffer -N "$AUTH_BASE_URL/admin/live/logs?types=audit,usage" \ + -H "$ADMIN_AUTH_HEADER" \ + --max-time 18 > "$LIVE_OUT" & +LIVE_PID=$! +sleep 1 +curl -fsS "$AUTH_BASE_URL/v1/chat/completions" \ + -H "$ADMIN_AUTH_HEADER" \ + -H 'Content-Type: application/json' \ + -H "X-Request-ID: $RID" \ + -d '{"model":"openai/gpt-4.1-nano","messages":[{"role":"user","content":"reply OK"}],"max_tokens":12}' \ + > "$QA_RUN_DIR/s92.chat.json" +sleep 8 +kill "$LIVE_PID" 2>/dev/null || true +wait "$LIVE_PID" 2>/dev/null || true +jq -e '.choices[0].message.content | type == "string" and length > 0' "$QA_RUN_DIR/s92.chat.json" >/dev/null +grep -cE '^event: audit\.' "$LIVE_OUT" | jq -R -e 'tonumber >= 1' >/dev/null +grep -cE '^event: usage\.' "$LIVE_OUT" | jq -R -e 'tonumber >= 1' >/dev/null +grep '^data: {' "$LIVE_OUT" | sed 's/^data: //' \ + | jq -e --arg rid "$RID" 'select(.. | strings? | tostring | contains($rid)) | .seq | type == "number"' \ + | head -n1 >/dev/null +``` + +### S93 Live preview type filter excludes off-list categories + +Subscribes with `types=usage` only, fires another chat, and asserts the captured stream contains `usage.*` events with no `audit.*` events leaking through. + +```bash +LIVE_OUT="$QA_RUN_DIR/s93.live.sse" +RID="qa-live-filter-$QA_SUFFIX" +curl -sS --no-buffer -N "$AUTH_BASE_URL/admin/live/logs?types=usage" \ + -H "$ADMIN_AUTH_HEADER" \ + --max-time 18 > "$LIVE_OUT" & +LIVE_PID=$! +sleep 1 +curl -fsS "$AUTH_BASE_URL/v1/chat/completions" \ + -H "$ADMIN_AUTH_HEADER" \ + -H 'Content-Type: application/json' \ + -H "X-Request-ID: $RID" \ + -d '{"model":"openai/gpt-4.1-nano","messages":[{"role":"user","content":"reply OK"}],"max_tokens":12}' \ + > "$QA_RUN_DIR/s93.chat.json" +sleep 8 +kill "$LIVE_PID" 2>/dev/null || true +wait "$LIVE_PID" 2>/dev/null || true +grep -cE '^event: usage\.' "$LIVE_OUT" | jq -R -e 'tonumber >= 1' >/dev/null +if grep -qE '^event: audit\.' "$LIVE_OUT"; then + echo "error: audit.* event leaked through types=usage filter" >&2 + exit 1 +fi +``` + +### S94 Live preview rejects invalid cursor with 400 + +Verifies the endpoint validates the `cursor` query parameter rather than silently dropping it. + +```bash +HEADERS_FILE=$(mktemp "$QA_RUN_DIR/s94.headers.XXXXXX") +BODY_FILE=$(mktemp "$QA_RUN_DIR/s94.body.XXXXXX") +curl -sS -D "$HEADERS_FILE" -o "$BODY_FILE" "$AUTH_BASE_URL/admin/live/logs?cursor=not-a-number" \ + -H "$ADMIN_AUTH_HEADER" +sed -n '1,10p' "$HEADERS_FILE" +jq . "$BODY_FILE" +grep -Eiq '^HTTP/.* 400 ' "$HEADERS_FILE" +jq -e '.error.type == "invalid_request_error" and (.error.message | test("cursor"; "i"))' "$BODY_FILE" >/dev/null +``` + +### S95 Streaming client disconnect is audited as `client_disconnected` + +Starts a streaming chat completion, aborts it within 400 ms (before the upstream connection completes), then asserts the audit row reflects the request as a streaming request that was cancelled by the client rather than as an upstream provider failure. + +```bash +RID="qa-stream-cancel-$QA_SUFFIX" +timeout 0.4 curl -sS --no-buffer "$BASE_URL/v1/chat/completions" \ + -H 'Content-Type: application/json' \ + -H "X-Request-ID: $RID" \ + -d '{"model":"gpt-4.1-nano","stream":true,"messages":[{"role":"user","content":"Write a 200 word poem about caches."}],"max_tokens":256}' \ + > "$QA_RUN_DIR/s95.partial.sse" 2>/dev/null || true +sleep 6 +AUDIT_JSON_FILE="$QA_RUN_DIR/s95.audit.json" +curl -fsS "$BASE_URL/admin/audit/log?search=$RID&limit=5" > "$AUDIT_JSON_FILE" +jq --arg rid "$RID" '{entries:(.entries|map(select(.request_id==$rid))|map({request_id,status_code,stream,error_type,path}))}' "$AUDIT_JSON_FILE" +jq -e --arg rid "$RID" ' + any(.entries[]?; + .request_id == $rid + and .path == "/v1/chat/completions" + and .stream == true + and .error_type == "client_disconnected" + ) + ' "$AUDIT_JSON_FILE" >/dev/null +``` From 10b766aed3e4fbe6015674ab1d6bccbdeb8c2aaa Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sun, 17 May 2026 20:16:21 +0200 Subject: [PATCH 3/5] fix(server): only attribute streaming error to client when err is one isClientDisconnect returned true for any error once ctx.Err was set, so a real upstream stream failure that races with a client disconnect was audited as client_disconnected. That hides upstream incidents from the audit log. Require the error itself to be context.Canceled / syscall.EPIPE / syscall.ECONNRESET (or a chain that unwraps to one). The context-only path now only fires when no concrete error was returned by the call. Expand TestRecordStreamingError_ClassifiesClientDisconnect into a table-driven test covering syscall.EPIPE, syscall.ECONNRESET, wrapped context.Canceled, the race case that must stay stream_error, and the clean-context generic error case. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/server/handlers_test.go | 69 ++++++++++++++++--- .../server/translated_inference_service.go | 9 ++- 2 files changed, 66 insertions(+), 12 deletions(-) diff --git a/internal/server/handlers_test.go b/internal/server/handlers_test.go index 36f34e94..2bd5db39 100644 --- a/internal/server/handlers_test.go +++ b/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" @@ -2686,18 +2688,67 @@ func TestHandleStreamingResponse_ClientDisconnectBeforeUpstream(t *testing.T) { } func TestRecordStreamingError_ClassifiesClientDisconnect(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) + canceledCtx, cancel := context.WithCancel(context.Background()) cancel() - entry := &auditlog.LogEntry{Data: &auditlog.LogData{}} - recordStreamingError(entry, "gpt-4o-mini", "openai", "/v1/chat/completions", "req-1", ctx, errors.New("broken pipe")) - if entry.ErrorType != "client_disconnected" { - t.Fatalf("expected client_disconnected, got %q", entry.ErrorType) + + 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", + }, } - entry2 := &auditlog.LogEntry{Data: &auditlog.LogData{}} - recordStreamingError(entry2, "gpt-4o-mini", "openai", "/v1/chat/completions", "req-2", context.Background(), errors.New("upstream malformed")) - if entry2.ErrorType != "stream_error" { - t.Fatalf("expected stream_error, got %q", entry2.ErrorType) + 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) + } + }) } } diff --git a/internal/server/translated_inference_service.go b/internal/server/translated_inference_service.go index 719319b8..eb491743 100644 --- a/internal/server/translated_inference_service.go +++ b/internal/server/translated_inference_service.go @@ -542,12 +542,15 @@ func recordStreamingError(streamEntry *auditlog.LogEntry, model, provider, path, } // isClientDisconnect reports whether the streaming error was caused by the -// client closing the connection rather than an upstream failure. +// client closing the connection rather than an upstream failure. A real +// upstream error that races with a cancellation must still be classified as a +// provider/stream error, so the context-only path only fires when no concrete +// error was returned by the call path. func isClientDisconnect(ctx context.Context, err error) bool { - if ctx != nil && ctx.Err() != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) { return true } - return errors.Is(err, context.Canceled) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) + return err == nil && ctx != nil && ctx.Err() == context.Canceled } func providerNameFromWorkflow(workflow *core.Workflow) string { From 30aec8e420a63eab670965a0e58583caa25c728d Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sun, 17 May 2026 20:23:27 +0200 Subject: [PATCH 4/5] fix(server): guard recordStreamingError against nil err isClientDisconnect classifies (ctx canceled, err == nil) as a client disconnect, but recordStreamingError still called err.Error() unconditionally. The nil-err branch was unreachable from today's two callsites (both gate on err != nil) but it is documented behaviour that a future caller could rely on, and a latent nil deref is not the right place to leave a footgun. Compute the audit message defensively: prefer err.Error() when available, otherwise fall back to the context error. The slog "error" field follows the same source so the log line never carries a stale nil. Extend the classifier test with a (canceledCtx, nil) row and assert the recorded error_message on every row to lock in both the classification and the message-source fallback. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/server/handlers_test.go | 20 +++++++++++++++++++ .../server/translated_inference_service.go | 17 ++++++++++++++-- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/internal/server/handlers_test.go b/internal/server/handlers_test.go index 2bd5db39..b5d8bac9 100644 --- a/internal/server/handlers_test.go +++ b/internal/server/handlers_test.go @@ -2739,6 +2739,15 @@ func TestRecordStreamingError_ClassifiesClientDisconnect(t *testing.T) { 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 { @@ -2748,6 +2757,17 @@ func TestRecordStreamingError_ClassifiesClientDisconnect(t *testing.T) { 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) + } }) } } diff --git a/internal/server/translated_inference_service.go b/internal/server/translated_inference_service.go index eb491743..5a9d9a99 100644 --- a/internal/server/translated_inference_service.go +++ b/internal/server/translated_inference_service.go @@ -523,16 +523,29 @@ func recordStreamingError(streamEntry *auditlog.LogEntry, model, provider, path, 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 = 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, From 87ead6cea91dab053dd7441a017c81089d0f147b Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sun, 17 May 2026 20:36:42 +0200 Subject: [PATCH 5/5] fix(server): only treat ctx cancellation as client disconnect at dispatch handleStreamingDispatchError ran the same isClientDisconnect classifier as the write-phase recordStreamingError. That helper treats syscall.EPIPE / syscall.ECONNRESET as client disconnects, which is correct only after the gateway has begun writing the SSE response to the client. Before the first chunk is flushed the only socket in play is the upstream provider connection, so an EPIPE / ECONNRESET there belongs to the provider and must surface as an upstream failure - not be swallowed as client_disconnected and returned as an empty 200. Introduce isClientDisconnectDuringDispatch covering only request- context cancellation (errors.Is(context.Canceled) or the err==nil race fallback) and wire handleStreamingDispatchError to it. Keep isClientDisconnect unchanged for the write-phase callers in recordStreamingError where EPIPE / ECONNRESET genuinely signal the downstream client going away. Pin the new boundary with a test that feeds bare and wrapped EPIPE / ECONNRESET into handleStreamingResponse via streamFn and asserts the audit entry is not classified as client_disconnected and the response is not an empty 200. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/server/handlers_test.go | 55 +++++++++++++++++++ .../server/translated_inference_service.go | 28 ++++++++-- 2 files changed, 77 insertions(+), 6 deletions(-) diff --git a/internal/server/handlers_test.go b/internal/server/handlers_test.go index b5d8bac9..68ab2c71 100644 --- a/internal/server/handlers_test.go +++ b/internal/server/handlers_test.go @@ -2687,6 +2687,61 @@ func TestHandleStreamingResponse_ClientDisconnectBeforeUpstream(t *testing.T) { } } +// 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() diff --git a/internal/server/translated_inference_service.go b/internal/server/translated_inference_service.go index 5a9d9a99..e5f2a33d 100644 --- a/internal/server/translated_inference_service.go +++ b/internal/server/translated_inference_service.go @@ -510,7 +510,7 @@ func (s *translatedInferenceService) handleStreamingResponse( // the audit log reflects the actual cause. func handleStreamingDispatchError(c *echo.Context, err error) error { auditlog.EnrichEntryWithStream(c, true) - if isClientDisconnect(c.Request().Context(), err) { + if isClientDisconnectDuringDispatch(c.Request().Context(), err) { auditlog.EnrichEntryWithError(c, "client_disconnected", err.Error(), "") return nil } @@ -554,11 +554,12 @@ func recordStreamingError(streamEntry *auditlog.LogEntry, model, provider, path, ) } -// isClientDisconnect reports whether the streaming error was caused by the -// client closing the connection rather than an upstream failure. A real -// upstream error that races with a cancellation must still be classified as a -// provider/stream error, so the context-only path only fires when no concrete -// error was returned by the call path. +// 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 @@ -566,6 +567,21 @@ func isClientDisconnect(ctx context.Context, err error) bool { 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) }