From 2f00c21f251c9d7ea4c553b73dd06983db124361 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 8 Oct 2026 17:34:30 +0800 Subject: [PATCH 1/2] fix: classify Claude stream errors and recover thinking signatures safely --- .github/workflows/downstream-verify.yml | 4 + DOWNSTREAM.md | 20 + .../gateway_claude_stream_error_test.go | 162 ++++++ backend/internal/handler/gateway_handler.go | 7 + backend/internal/handler/ops_error_logger.go | 21 +- .../service/gateway_claude_prelude_test.go | 206 ++++++++ .../gateway_claude_signature_recovery.go | 236 +++++++++ .../gateway_claude_signature_recovery_test.go | 465 ++++++++++++++++++ .../service/gateway_claude_stream_error.go | 122 +++++ .../gateway_claude_stream_error_test.go | 89 ++++ backend/internal/service/gateway_forward.go | 392 +++++++++------ .../gateway_forward_partial_usage_test.go | 8 +- backend/internal/service/gateway_service.go | 3 + .../service/gateway_upstream_response.go | 158 ++++-- .../ops_claude_signature_evidence_test.go | 48 ++ backend/internal/service/ops_service.go | 4 + .../internal/service/ops_upstream_context.go | 6 + frontend/src/api/admin/ops.ts | 1 + frontend/src/i18n/locales/en/admin/ops.ts | 2 + frontend/src/i18n/locales/zh/admin/ops.ts | 2 + .../ops/components/OpsErrorDetailModal.vue | 42 +- .../__tests__/OpsErrorDetailModal.spec.ts | 82 ++- 22 files changed, 1884 insertions(+), 196 deletions(-) create mode 100644 backend/internal/handler/gateway_claude_stream_error_test.go create mode 100644 backend/internal/service/gateway_claude_prelude_test.go create mode 100644 backend/internal/service/gateway_claude_signature_recovery.go create mode 100644 backend/internal/service/gateway_claude_signature_recovery_test.go create mode 100644 backend/internal/service/gateway_claude_stream_error.go create mode 100644 backend/internal/service/gateway_claude_stream_error_test.go create mode 100644 backend/internal/service/ops_claude_signature_evidence_test.go diff --git a/.github/workflows/downstream-verify.yml b/.github/workflows/downstream-verify.yml index e3255bae73a5..16eaa8c4ccd9 100644 --- a/.github/workflows/downstream-verify.yml +++ b/.github/workflows/downstream-verify.yml @@ -41,6 +41,9 @@ jobs: - name: Service tests working-directory: backend run: go test -p=1 -tags unit ./internal/service -run 'Test.*(Refusal|CyberPolicy|Continuation|InvalidEncryptedContent|PreviousResponseNotFound|CapacityShed|RequestScopedTransient|ServerOverloaded|AlphaSearch|SelectedWorkspace)|Test(UpdateService|NormalizeAccountTaxonomyName|AccountTaxonomyRules|AccountTestReasoningValidated|FilterConsoleAccounts|AccountConsolePlan|AccountFacetMatcher)' -count=1 + - name: Claude stream failure and signature recovery contracts + working-directory: backend + run: go test -p=1 -tags unit ./internal/service ./internal/handler -run '^TestClaude(ClassifyStreamError|ThinkingSignatureError|Prelude_|Signature|StreamError|StreamFailureStatus|OpsErrorLogger)|^TestGatewayService_Forward_(PreOutputSSE|PostOutputSSE)' -count=1 - name: Native gateway borrowing contracts working-directory: backend run: | @@ -140,6 +143,7 @@ jobs: src/components/payment/__tests__/StripePaymentInline.contrast.spec.ts \ src/views/user/__tests__/StripePaymentView.spec.ts \ src/views/admin/__tests__/ReasoningRecoveryView.spec.ts \ + src/views/admin/ops/components/__tests__/OpsErrorDetailModal.spec.ts \ src/api/admin/__tests__/codexGatewayBorrow.spec.ts \ src/components/admin/codex/__tests__/BorrowPelicanPreview.spec.ts \ src/views/admin/__tests__/CodexGatewayBorrowView.spec.ts \ diff --git a/DOWNSTREAM.md b/DOWNSTREAM.md index 44bab9bf1dd5..2eba511d356a 100644 --- a/DOWNSTREAM.md +++ b/DOWNSTREAM.md @@ -96,6 +96,26 @@ is the downloaded `model_catalog_json` file; a response over 1 MiB switches the Codex configuration to that file mode. OpenCode continues reading the manifest regardless of the Codex remote/file selection. +## Claude streaming failures and signature recovery + +Ordinary Claude Messages forwarding classifies SSE error events by their error +type, with a narrow thinking-signature fallback for third-party errors missing +the type. A stream error is not an HTTP permission rejection. In Ops attempt +JSON, `stream_error.upstream_status_code` retains the error's semantic status; +the optional `upstream_http_status_code` records the actual response status. +Existing records and HTTP error semantics remain compatible without a migration. + +When the existing signature rectifier permits it, ordinary Claude forwarding +can repair rejected thinking history on the same account before any message +event is committed. It buffers only `message_start`, up to 64 KiB, while sending +ping immediately. Committing the next message event or exceeding the bound +ends recovery eligibility. The first thinking repair has one attempt independent +of the initial request's ten-second backoff budget; the existing HTTP-only tool +fallback retains its original conditions. Automatic passthrough and other +provider paths do not gain this repair. Fingerprint-only recovery diagnostics +stay attached to the upstream attempt and do not replace its failure or expose +new thinking text or signature values. + ## OpenAI forwarding and model identity Explicit `reasoning.effort=none` is preserved on OpenAI-wire accounts, including diff --git a/backend/internal/handler/gateway_claude_stream_error_test.go b/backend/internal/handler/gateway_claude_stream_error_test.go new file mode 100644 index 000000000000..4d37d53d36a0 --- /dev/null +++ b/backend/internal/handler/gateway_claude_stream_error_test.go @@ -0,0 +1,162 @@ +package handler + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/model" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func TestClaudeStreamErrorTerminal(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, test := range []struct { + errorType string + message string + status int + }{ + {"invalid_request_error", "Invalid thinking signature", 400}, + {"overloaded_error", "Overloaded", 503}, + {"service_error", "Service temporarily unavailable", 502}, + } { + for _, started := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/stream=%v", test.errorType, started), func(t *testing.T) { + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + if started { + _, err := c.Writer.WriteString(partialMessageStartSSE) + require.NoError(t, err) + } + body := []byte(fmt.Sprintf(`{"type":"error","error":{"type":%q,"message":%q}}`, test.errorType, test.message)) + failoverErr := service.ClassifyClaudeStreamError(body).FailoverError(body) + (&GatewayHandler{}).handleFailoverExhausted(c, failoverErr, service.PlatformAnthropic, started) + if started { + require.Equal(t, 200, recorder.Code) + require.Equal(t, 1, strings.Count(recorder.Body.String(), "event: message_start")) + require.Equal(t, 1, strings.Count(recorder.Body.String(), `"type":"error"`)) + parsed := parseOpsErrorResponse(recorder.Body.Bytes()) + require.Equal(t, test.errorType, parsed.ErrorType) + require.Equal(t, test.message, parsed.Message) + } else { + require.Equal(t, test.status, recorder.Code) + require.Equal(t, test.errorType, gjson.Get(recorder.Body.String(), "error.type").String()) + require.Equal(t, test.message, gjson.Get(recorder.Body.String(), "error.message").String()) + } + require.NotContains(t, recorder.Body.String(), "Upstream access forbidden") + }) + } + } +} + +type claudeStreamPassthroughRepo struct { + service.ErrorPassthroughRepository + rules []*model.ErrorPassthroughRule +} + +func (repo *claudeStreamPassthroughRepo) List(context.Context) ([]*model.ErrorPassthroughRule, error) { + return repo.rules, nil +} + +func TestClaudeStreamErrorTerminal_PassthroughRulePrecedence(t *testing.T) { + gin.SetMode(gin.TestMode) + responseCode, message := 418, "configured overload response" + repo := &claudeStreamPassthroughRepo{rules: []*model.ErrorPassthroughRule{{ + Enabled: true, ErrorCodes: []int{529}, Platforms: []string{service.PlatformAnthropic}, MatchMode: model.MatchModeAny, + ResponseCode: &responseCode, CustomMessage: &message, SkipMonitoring: true, + }}} + handler := &GatewayHandler{errorPassthroughService: service.NewErrorPassthroughService(repo, nil)} + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + body := []byte(`{"error":{"type":"overloaded_error","message":"Overloaded"}}`) + handler.handleFailoverExhausted(c, service.ClassifyClaudeStreamError(body).FailoverError(body), service.PlatformAnthropic, false) + require.Equal(t, responseCode, recorder.Code) + require.Equal(t, message, gjson.Get(recorder.Body.String(), "error.message").String()) + require.Equal(t, true, c.GetBool(service.OpsSkipPassthroughKey)) +} + +func TestClaudeStreamErrorTerminal_RealHTTPAuthMappingUnchanged(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, status := range []int{401, 403} { + t.Run(fmt.Sprint(status), func(t *testing.T) { + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + (&GatewayHandler{}).handleFailoverExhausted(c, &service.UpstreamFailoverError{StatusCode: status}, service.PlatformAnthropic, false) + require.Equal(t, 502, recorder.Code) + require.Equal(t, "upstream_error", gjson.Get(recorder.Body.String(), "error.type").String()) + require.Contains(t, gjson.Get(recorder.Body.String(), "error.message").String(), "please contact administrator") + }) + } +} + +func TestClaudeStreamFailureStatus_EventGate(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, test := range []struct { + name string + events []*service.OpsUpstreamErrorEvent + want int + }{ + {"new overload", []*service.OpsUpstreamErrorEvent{{Platform: service.PlatformAnthropic, Kind: "stream_error", UpstreamStatusCode: 529, UpstreamHTTPStatusCode: 200}}, 529}, + {"new service failure", []*service.OpsUpstreamErrorEvent{{Platform: service.PlatformAnthropic, Kind: "stream_error", UpstreamStatusCode: 500, UpstreamHTTPStatusCode: 200}}, 500}, + {"other platform", []*service.OpsUpstreamErrorEvent{{Platform: service.PlatformOpenAI, Kind: "stream_error", UpstreamStatusCode: 500, UpstreamHTTPStatusCode: 200}}, 503}, + {"legacy stream", []*service.OpsUpstreamErrorEvent{{Kind: "stream_error", UpstreamStatusCode: 403}}, 503}, + {"ordinary HTTP", []*service.OpsUpstreamErrorEvent{{Kind: "http_error", UpstreamStatusCode: 502, UpstreamHTTPStatusCode: 502}}, 503}, + {"invalid semantic status", []*service.OpsUpstreamErrorEvent{{Kind: "stream_error", UpstreamStatusCode: 200, UpstreamHTTPStatusCode: 200}}, 503}, + {"later HTTP attempt wins", []*service.OpsUpstreamErrorEvent{{Kind: "stream_error", UpstreamStatusCode: 529, UpstreamHTTPStatusCode: 200}, {Kind: "http_error", UpstreamStatusCode: 502}}, 503}, + } { + t.Run(test.name, func(t *testing.T) { + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Set(service.OpsUpstreamErrorsKey, test.events) + require.Equal(t, test.want, inferStreamFailureStatus(c, parsedOpsError{ErrorType: "overloaded_error"})) + }) + } +} + +func TestClaudeOpsErrorLogger_StreamErrorSemanticStatus(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, test := range []struct { + errorType string + status int + }{ + {"invalid_request_error", 400}, + {"rate_limit_error", 429}, + {"overloaded_error", 529}, + {"service_error", 500}, + } { + t.Run(test.errorType, func(t *testing.T) { + setupOpsErrorLogTestQueue(t, 2) + ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) + router := gin.New() + router.Use(OpsErrorLoggerMiddleware(ops)) + router.POST("/v1/messages", func(c *gin.Context) { + body := []byte(fmt.Sprintf(`{"type":"error","error":{"type":%q,"message":"provider failure"}}`, test.errorType)) + c.Set(service.OpsUpstreamErrorsKey, []*service.OpsUpstreamErrorEvent{{Platform: service.PlatformAnthropic, Kind: "stream_error", UpstreamStatusCode: test.status, UpstreamHTTPStatusCode: 200, Message: "provider failure"}}) + _, err := c.Writer.WriteString(partialMessageStartSSE) + require.NoError(t, err) + (&GatewayHandler{}).handleFailoverExhausted(c, service.ClassifyClaudeStreamError(body).FailoverError(body), service.PlatformAnthropic, true) + }) + recorder := httptest.NewRecorder() + router.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/v1/messages", nil)) + require.Equal(t, 200, recorder.Code) + require.Equal(t, int64(1), OpsErrorLogQueueLength()) + job := <-opsErrorLogQueue + require.Equal(t, test.status, job.entry.StatusCode) + require.Equal(t, test.status, *job.entry.UpstreamStatusCode) + require.Equal(t, "provider failure", *job.entry.UpstreamErrorMessage) + require.NotNil(t, job.entry.UpstreamErrorsJSON) + events, err := service.ParseOpsUpstreamErrors(*job.entry.UpstreamErrorsJSON) + require.NoError(t, err) + require.Equal(t, 200, events[0].UpstreamHTTPStatusCode) + require.Equal(t, test.status, events[0].UpstreamStatusCode) + }) + } +} diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index 68eccaecbd7c..258c115d1a26 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -1957,6 +1957,13 @@ func (h *GatewayHandler) handleFailoverExhausted(c *gin.Context, failoverErr *se upstreamMsg := service.ExtractUpstreamErrorMessage(responseBody) service.SetOpsUpstreamError(c, statusCode, upstreamMsg, "") + // Claude SSE classifications carry their own client terminal. Keep rule + // precedence above, and leave ordinary HTTP errors on the existing mapping. + if platform == service.PlatformAnthropic && failoverErr.ClientStatusCode > 0 && failoverErr.ClientErrorType != "" { + h.handleStreamingAwareErrorWithCode(c, failoverErr.ClientStatusCode, failoverErr.ClientErrorType, failoverErr.ClientErrorCode, failoverErr.ClientMessage, streamStarted) + return + } + // 使用默认的错误映射 status, errType, errMsg := h.mapUpstreamError(statusCode) h.handleStreamingAwareError(c, status, errType, errMsg, streamStarted) diff --git a/backend/internal/handler/ops_error_logger.go b/backend/internal/handler/ops_error_logger.go index 7d146eebad78..7a922acc62a5 100644 --- a/backend/internal/handler/ops_error_logger.go +++ b/backend/internal/handler/ops_error_logger.go @@ -2140,7 +2140,26 @@ func inferResponsesFailedOpsErrorType(code string) string { } } -func inferStreamFailureStatus(_ *gin.Context, parsed parsedOpsError) int { +func inferStreamFailureStatus(c *gin.Context, parsed parsedOpsError) int { + // New Claude SSE attempts preserve the existing semantic status convention + // and independently capture the real carrying HTTP response. Only those + // events may override generic terminal inference (529 would otherwise be + // reduced to its client-facing 503). Ordinary HTTP attempts stay independent + // of the terminal semantics, including earlier attempts on a recovered path. + if c != nil { + if value, ok := c.Get(service.OpsUpstreamErrorsKey); ok { + if events, ok := value.([]*service.OpsUpstreamErrorEvent); ok { + for i := len(events) - 1; i >= 0; i-- { + if event := events[i]; event != nil { + if event.Platform == service.PlatformAnthropic && event.Kind == "stream_error" && event.UpstreamHTTPStatusCode > 0 && event.UpstreamStatusCode >= 400 && event.UpstreamStatusCode <= 599 { + return event.UpstreamStatusCode + } + break + } + } + } + } + } if parsed.StatusCode >= 400 && parsed.StatusCode <= 599 { return parsed.StatusCode } diff --git a/backend/internal/service/gateway_claude_prelude_test.go b/backend/internal/service/gateway_claude_prelude_test.go new file mode 100644 index 000000000000..ad3be5dc0235 --- /dev/null +++ b/backend/internal/service/gateway_claude_prelude_test.go @@ -0,0 +1,206 @@ +package service + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +const claudePreludeSignatureError = `{"type":"error","error":{"type":"invalid_request_error","message":"Invalid signature in thinking block"}}` + +func claudePreludeFrame(event, data string) string { + return "event: " + event + "\ndata: " + data + "\n\n" +} + +func claudePreludeStart(id string, inputTokens int) string { + return claudePreludeFrame("message_start", fmt.Sprintf(`{"type":"message_start","message":{"id":%q,"model":"claude-sonnet-5-5","usage":{"input_tokens":%d}}}`, id, inputTokens)) +} + +func claudePreludeTestContext() (*httptest.ResponseRecorder, *gin.Context) { + gin.SetMode(gin.TestMode) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + return recorder, c +} + +type claudePreludeFlushObserver struct { + gin.ResponseWriter + onFlush func() +} + +func (w *claudePreludeFlushObserver) Flush() { + w.ResponseWriter.Flush() + w.onFlush() +} + +func runClaudePreludeTestStream(t *testing.T, svc *GatewayService, c *gin.Context, stream string, recoverPrelude bool) (*streamingResult, error) { + t.Helper() + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(stream))} + defer resp.Body.Close() + return svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Platform: PlatformAnthropic}, time.Now(), "claude-sonnet-5-5", "claude-sonnet-5-5", false, recoverPrelude) +} + +func TestClaudePrelude_SignatureRejectionDiscardsStartButKeepsAttemptUsage(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + stream := claudePreludeStart("failed-message", 123) + claudePreludeFrame("ping", `{"type":"ping"}`) + ": keepalive\n\n" + claudePreludeFrame("error", claudePreludeSignatureError) + + result, err := runClaudePreludeTestStream(t, svc, c, stream, true) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr)) + require.False(t, sseErr.MessageStarted, "ping and comments do not start this message") + require.NotNil(t, result) + require.Equal(t, 123, result.usage.InputTokens, "discarding a client prelude must not discard this attempt's observed metering") + require.NotContains(t, recorder.Body.String(), "failed-message") + require.Contains(t, recorder.Body.String(), "event: ping") + require.Contains(t, recorder.Body.String(), ": keepalive") + + // A successful repaired attempt contributes its own usage and one message ID. + beginUpstreamResponseModelObservation(c) + success := claudePreludeStart("repaired-message", 7) + claudePreludeFrame("message_delta", `{"type":"message_delta","usage":{"output_tokens":3}}`) + claudePreludeFrame("message_stop", `{"type":"message_stop"}`) + result, err = runClaudePreludeTestStream(t, svc, c, success, true) + require.NoError(t, err) + require.Equal(t, 7, result.usage.InputTokens) + require.Equal(t, 3, result.usage.OutputTokens) + require.Equal(t, 1, strings.Count(recorder.Body.String(), "event: message_start")) + require.Equal(t, 1, strings.Count(recorder.Body.String(), "event: message_stop")) + require.Contains(t, recorder.Body.String(), "repaired-message") +} + +func TestClaudePrelude_PingIsFlushedBeforeStartCommit(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + beforeCommit := "" + c.Writer = &claudePreludeFlushObserver{ResponseWriter: c.Writer, onFlush: func() { + if beforeCommit == "" { + beforeCommit = recorder.Body.String() + } + }} + stream := claudePreludeStart("message", 5) + claudePreludeFrame("ping", `{"type":"ping"}`) + claudePreludeFrame("message_stop", `{"type":"message_stop"}`) + _, err := runClaudePreludeTestStream(t, svc, c, stream, true) + require.NoError(t, err) + require.Contains(t, beforeCommit, "event: ping") + require.NotContains(t, beforeCommit, "event: message_start") +} + +func TestClaudePrelude_CommitsStartBeforeContentAndPreservesSignatureDelta(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + start := claudePreludeStart("message", 5) + contentStart := claudePreludeFrame("content_block_start", `{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}`) + thinkingDelta := claudePreludeFrame("content_block_delta", `{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"A thought"}}`) + signatureDelta := claudePreludeFrame("content_block_delta", `{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"opaque+/=fragment"}}`) + contentStop := claudePreludeFrame("content_block_stop", `{"type":"content_block_stop","index":0}`) + stop := claudePreludeFrame("message_stop", `{"type":"message_stop"}`) + + _, err := runClaudePreludeTestStream(t, svc, c, start+contentStart+thinkingDelta+signatureDelta+contentStop+stop, true) + require.NoError(t, err) + require.Equal(t, start+contentStart+thinkingDelta+signatureDelta+contentStop+stop, recorder.Body.String(), "buffering must preserve all event order and signature bytes") +} + +func TestClaudePrelude_ContentOutputPreventsSignatureRecovery(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + stream := claudePreludeStart("message", 5) + claudePreludeFrame("content_block_start", `{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`) + claudePreludeFrame("error", claudePreludeSignatureError) + result, err := runClaudePreludeTestStream(t, svc, c, stream, true) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr)) + require.True(t, sseErr.MessageStarted) + require.Equal(t, 5, result.usage.InputTokens) + require.Equal(t, 1, strings.Count(recorder.Body.String(), "event: message_start")) + require.Contains(t, recorder.Body.String(), "event: content_block_start") +} + +func TestClaudePrelude_OversizedStartCommitsAndPreventsRecovery(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + start := claudePreludeFrame("message_start", fmt.Sprintf(`{"type":"message_start","message":{"id":%q,"usage":{"input_tokens":5}}}`, strings.Repeat("x", claudeMessageStartPreludeLimit))) + _, err := runClaudePreludeTestStream(t, svc, c, start+claudePreludeFrame("error", claudePreludeSignatureError), true) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr)) + require.True(t, sseErr.MessageStarted) + require.Equal(t, start, recorder.Body.String()) +} + +func TestClaudePrelude_NonSignatureRejectionCommitsStart(t *testing.T) { + for _, errorJSON := range []string{ + `{"type":"error","error":{"type":"service_error","message":"Service temporarily unavailable"}}`, + `{"type":"error","error":{"type":"service_error","message":"Invalid signature in thinking block"}}`, + `{"type":"error","error":{"type":"permission_error","message":"Invalid signature in thinking block"}}`, + } { + t.Run(errorJSON, func(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + start := claudePreludeStart("message", 5) + result, err := runClaudePreludeTestStream(t, svc, c, start+claudePreludeFrame("error", errorJSON), true) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr)) + require.True(t, sseErr.MessageStarted) + require.Equal(t, 5, result.usage.InputTokens) + require.Equal(t, start, recorder.Body.String()) + }) + } +} + +func TestClaudePrelude_EOFCommitsStartAndEndsWithSingleError(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + start := claudePreludeStart("message", 5) + result, err := runClaudePreludeTestStream(t, svc, c, start, true) + require.ErrorContains(t, err, "missing terminal event") + require.Equal(t, 5, result.usage.InputTokens) + require.True(t, strings.HasPrefix(recorder.Body.String(), start)) + require.Equal(t, 1, strings.Count(recorder.Body.String(), "event: error")) + require.True(t, IsResponseCommitted(c), "the handler must not append a second terminal after the prelude EOF error") +} + +func TestClaudePrelude_DisabledRecoveryDoesNotHoldStart(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + recorder, c := claudePreludeTestContext() + start := claudePreludeStart("message", 5) + _, err := runClaudePreludeTestStream(t, svc, c, start+claudePreludeFrame("error", claudePreludeSignatureError), false) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr)) + require.True(t, sseErr.MessageStarted) + require.Equal(t, start, recorder.Body.String()) +} + +func TestClaudePrelude_RepairTimeoutSkipsHealthPenaltyAndNormalTimeoutKeepsIt(t *testing.T) { + for _, suppressPenalty := range []bool{false, true} { + t.Run(fmt.Sprintf("suppress_%t", suppressPenalty), func(t *testing.T) { + svc := newStreamingResponseTestGatewayService() + svc.cfg.Gateway.StreamDataIntervalTimeout = 1 + repo := &gatewayForwardErrorPolicyRepoStub{} + settings := NewSettingService(&gatewayTTLSettingRepo{data: map[string]string{ + SettingKeyStreamTimeoutSettings: `{"enabled":true,"action":"temp_unsched","threshold_count":1,"threshold_window_minutes":1,"temp_unsched_minutes":1}`, + }}, svc.cfg) + svc.rateLimitService = NewRateLimitService(repo, nil, svc.cfg, nil, nil) + svc.rateLimitService.SetSettingService(settings) + recorder, c := claudePreludeTestContext() + pr, pw := io.Pipe() + defer pr.Close() + defer pw.Close() + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: pr} + + _, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false, false, suppressPenalty) + require.ErrorContains(t, err, "stream data interval timeout") + if suppressPenalty { + require.Zero(t, repo.tempCalls, "a failed signature repair must not park the account") + } else { + require.Equal(t, 1, repo.tempCalls, "normal streaming timeout keeps its existing health action") + } + require.Equal(t, 1, strings.Count(recorder.Body.String(), "event: error")) + require.Contains(t, recorder.Body.String(), `"type":"stream_timeout"`) + }) + } +} diff --git a/backend/internal/service/gateway_claude_signature_recovery.go b/backend/internal/service/gateway_claude_signature_recovery.go new file mode 100644 index 000000000000..d535904899ae --- /dev/null +++ b/backend/internal/service/gateway_claude_signature_recovery.go @@ -0,0 +1,236 @@ +package service + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "net/http" + "regexp" + "time" + + "github.com/tidwall/gjson" +) + +const claudeSignatureFingerprintLimit = 16 + +// ClaudeSignatureValueFingerprint identifies a signature without retaining the +// signature itself or any message content. Length is the decoded UTF-8 byte size. +type ClaudeSignatureValueFingerprint struct { + Path string `json:"path"` + Length int `json:"length"` + SHA256 string `json:"sha256"` +} + +type ClaudeSignatureSnapshot struct { + TotalCount int `json:"total_count"` + OmittedCount int `json:"omitted_count,omitempty"` + Signatures []ClaudeSignatureValueFingerprint `json:"signatures,omitempty"` +} + +// ClaudeSignatureRecoveryDiagnostic is attached to the rejection that triggered +// recovery. It does not create a synthetic upstream attempt or change health. +type ClaudeSignatureRecoveryDiagnostic struct { + Source string `json:"source"` + Outcome string `json:"outcome"` + Reason string `json:"reason,omitempty"` + Attempts int `json:"attempts"` + ElapsedMs int64 `json:"elapsed_ms"` + MessageStarted bool `json:"message_started"` + Inbound ClaudeSignatureSnapshot `json:"inbound"` + FirstWire ClaudeSignatureSnapshot `json:"first_wire"` +} + +type claudeSignatureRecoveryState struct { + used bool + inbound []byte + firstWire []byte + firstWireSeen bool + now func() time.Time +} + +func newClaudeSignatureRecoveryState(inbound []byte) *claudeSignatureRecoveryState { + return &claudeSignatureRecoveryState{inbound: inbound, now: time.Now} +} + +func claudeSignatureSnapshot(body []byte) ClaudeSignatureSnapshot { + var snapshot ClaudeSignatureSnapshot + for mi, message := range gjson.GetBytes(body, "messages").Array() { + for bi, block := range message.Get("content").Array() { + if block.Get("type").String() != "thinking" { + continue + } + signature := block.Get("signature") + if signature.Type != gjson.String { + continue + } + snapshot.TotalCount++ + if len(snapshot.Signatures) >= claudeSignatureFingerprintLimit { + continue + } + value := signature.String() + hash := sha256.Sum256([]byte(value)) + snapshot.Signatures = append(snapshot.Signatures, ClaudeSignatureValueFingerprint{ + Path: fmt.Sprintf("/messages/%d/content/%d/signature", mi, bi), + Length: len(value), SHA256: hex.EncodeToString(hash[:]), + }) + } + } + snapshot.OmittedCount = snapshot.TotalCount - len(snapshot.Signatures) + return snapshot +} + +func (s *claudeSignatureRecoveryState) observeFirstWire(body []byte) { + if !s.firstWireSeen { + s.firstWireSeen = true + s.firstWire = body + } +} + +func (s *claudeSignatureRecoveryState) diagnostic(source string, messageStarted bool) *ClaudeSignatureRecoveryDiagnostic { + return &ClaudeSignatureRecoveryDiagnostic{ + Source: source, Outcome: "skipped", MessageStarted: messageStarted, + Inbound: claudeSignatureSnapshot(s.inbound), FirstWire: claudeSignatureSnapshot(s.firstWire), + } +} + +func (s *GatewayService) claudeSignatureRecoveryEnabled(ctx context.Context, account *Account, model string) bool { + if account == nil || account.Platform != PlatformAnthropic || account.IsBedrock() || account.IsAnthropicAPIKeyPassthroughEnabled() || s.settingService == nil || !ShouldRectifyThinkingSignatureError(model) { + return false + } + settings, err := s.settingService.GetRectifierSettings(ctx) + if err != nil { + // Match the existing account-type policy on a settings read failure. + return account.Type != AccountTypeAPIKey + } + if !settings.Enabled { + return false + } + if account.Type == AccountTypeAPIKey { + return settings.APIKeySignatureEnabled + } + return settings.ThinkingSignatureEnabled +} + +func (s *GatewayService) claudeSignatureErrorCandidate(ctx context.Context, account *Account, body []byte) bool { + if s.isThinkingBlockSignatureError(body) { + return true + } + return s.settingService != nil && s.isSignatureErrorPattern(ctx, account, body) +} + +// run provides one first-stage repair shared by HTTP and SSE rejection paths. +// Its decision budget begins at the rejection, not before the first upstream +// request. Normal transport/stream timeouts remain in force after sending. +func (s *claudeSignatureRecoveryState) run( + ctx context.Context, diagnostic *ClaudeSignatureRecoveryDiagnostic, baseBody []byte, model string, + build func([]byte) (*http.Request, []byte, error), send func(*http.Request) (*http.Response, error), +) (resp *http.Response, wireBody []byte, err error) { + started := s.now() + defer func() { diagnostic.ElapsedMs = s.now().Sub(started).Milliseconds() }() + if ctx.Err() != nil { + diagnostic.Reason = "context_canceled" + return nil, nil, nil + } + if diagnostic.MessageStarted { + diagnostic.Reason = "output_started" + return nil, nil, nil + } + if s.used { + diagnostic.Reason = "attempt_limit" + return nil, nil, nil + } + filteredBody := FilterThinkingBlocksForRetry(baseBody, model) + // The existing HTTP two-stage fallback can need a no-op first stage + // before a tool-specific rejection authorizes its second-stage filter. + // New SSE recovery must not replay an unchanged body. + if bytes.Equal(filteredBody, baseBody) && diagnostic.Source == "sse_error" { + diagnostic.Reason = "body_unchanged" + return nil, nil, nil + } + s.used = true + req, wireBody, err := build(filteredBody) + if err != nil { + diagnostic.Outcome, diagnostic.Reason = "build_failed", "build_failed" + return nil, nil, err + } + if ctx.Err() != nil { + diagnostic.Reason = "context_canceled" + return nil, nil, nil + } + if s.now().Sub(started) >= maxRetryElapsed { + diagnostic.Reason = "budget_exhausted" + return nil, nil, nil + } + diagnostic.Attempts++ + diagnostic.Outcome, diagnostic.Reason = "attempted", "" + resp, err = send(req) + if err != nil { + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + diagnostic.Outcome, diagnostic.Reason = "transport_failed", "transport_failed" + return nil, nil, err + } + if resp == nil || resp.Body == nil { + diagnostic.Outcome, diagnostic.Reason = "transport_failed", "empty_response" + return nil, nil, fmt.Errorf("signature recovery: empty upstream response") + } + if resp.StatusCode < 400 { + diagnostic.Outcome = "accepted" + } else { + diagnostic.Outcome = "upstream_rejected" + } + return resp, wireBody, nil +} + +var claudeSignaturePathPattern = regexp.MustCompile(`^/messages/[0-9]{1,10}/content/[0-9]{1,10}/signature$`) +var claudeSignatureHashPattern = regexp.MustCompile(`^[0-9a-f]{64}$`) + +// sanitizeClaudeSignatureRecoveryDiagnostic makes a bounded, detached copy for +// asynchronous persistence. No caller-supplied strings other than a signature +// JSON pointer and a digest can enter these diagnostics. +func sanitizeClaudeSignatureRecoveryDiagnostic(in *ClaudeSignatureRecoveryDiagnostic) *ClaudeSignatureRecoveryDiagnostic { + if in == nil { + return nil + } + out := *in + out.Source = claudeSignatureEnum(in.Source, "http_400", "sse_error") + out.Outcome = claudeSignatureEnum(in.Outcome, "skipped", "attempted", "accepted", "upstream_rejected", "build_failed", "transport_failed", "response_read_failed") + out.Reason = claudeSignatureEnum(in.Reason, "disabled", "passthrough", "output_started", "attempt_limit", "budget_exhausted", "body_unchanged", "context_canceled", "build_failed", "transport_failed", "empty_response", "response_read_failed", "stream_incomplete") + out.Attempts = min(max(in.Attempts, 0), 2) + if out.ElapsedMs < 0 { + out.ElapsedMs = 0 + } else if limit := int64(24 * time.Hour / time.Millisecond); out.ElapsedMs > limit { + out.ElapsedMs = limit + } + out.Inbound = sanitizeClaudeSignatureSnapshot(in.Inbound) + out.FirstWire = sanitizeClaudeSignatureSnapshot(in.FirstWire) + return &out +} + +func claudeSignatureEnum(value string, allowed ...string) string { + for _, candidate := range allowed { + if value == candidate { + return value + } + } + return "" +} + +func sanitizeClaudeSignatureSnapshot(in ClaudeSignatureSnapshot) ClaudeSignatureSnapshot { + out := ClaudeSignatureSnapshot{TotalCount: min(max(in.TotalCount, 0), 1_000_000)} + for _, signature := range in.Signatures { + if len(out.Signatures) == claudeSignatureFingerprintLimit { + break + } + if !claudeSignaturePathPattern.MatchString(signature.Path) || !claudeSignatureHashPattern.MatchString(signature.SHA256) { + continue + } + signature.Length = min(max(signature.Length, 0), 100*1024*1024) + out.Signatures = append(out.Signatures, signature) + } + out.OmittedCount = max(out.TotalCount-len(out.Signatures), 0) + return out +} diff --git a/backend/internal/service/gateway_claude_signature_recovery_test.go b/backend/internal/service/gateway_claude_signature_recovery_test.go new file mode 100644 index 000000000000..8c6db7f1d923 --- /dev/null +++ b/backend/internal/service/gateway_claude_signature_recovery_test.go @@ -0,0 +1,465 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +const claudeRecoveryTestBody = `{"model":"claude-sonnet-5-5","stream":true,"thinking":{"type":"between_tools"},"output_config":{"effort":"high"},"messages":[{"role":"assistant","content":[{"type":"thinking","thinking":"old reasoning","signature":"private-signature-value"},{"type":"text","text":"old answer"}]},{"role":"user","content":"continue"}]}` +const claudeRecoverySignatureError = `{"type":"error","error":{"message":"Invalid signature in thinking block"}}` + +type claudeRecoverySequenceUpstream struct { + responses []*http.Response + requests []*http.Request + bodies [][]byte + accountIDs []int64 + proxyURLs []string + beforeSend func(int) +} + +func (u *claudeRecoverySequenceUpstream) Do(req *http.Request, proxy string, id int64, _ int) (*http.Response, error) { + if u.beforeSend != nil { + u.beforeSend(len(u.requests)) + } + u.requests = append(u.requests, req) + u.accountIDs = append(u.accountIDs, id) + u.proxyURLs = append(u.proxyURLs, proxy) + body, err := io.ReadAll(req.Body) + if err != nil { + return nil, err + } + u.bodies = append(u.bodies, body) + if len(u.responses) == 0 { + return nil, errors.New("unexpected additional request") + } + response := u.responses[0] + u.responses = u.responses[1:] + return response, nil +} + +type claudeRecoveryCloseObservedBody struct { + io.ReadCloser + closed bool +} + +func (body *claudeRecoveryCloseObservedBody) Close() error { + body.closed = true + return body.ReadCloser.Close() +} + +func (u *claudeRecoverySequenceUpstream) DoWithTLS(req *http.Request, proxy string, id int64, concurrency int, _ *tlsfingerprint.Profile) (*http.Response, error) { + return u.Do(req, proxy, id, concurrency) +} + +func claudeRecoveryResponse(status int, body, id string) *http.Response { + return &http.Response{StatusCode: status, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "X-Request-Id": []string{id}}, Body: io.NopCloser(strings.NewReader(body))} +} + +func claudeRecoverySuccessStream() string { + return "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"final-message\",\"model\":\"claude-sonnet-5-5\",\"usage\":{\"input_tokens\":7}}}\n\n" + + "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n" + + "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"done\"}}\n\n" + + "event: message_delta\ndata: {\"type\":\"message_delta\",\"usage\":{\"output_tokens\":3}}\n\n" + + "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n" +} + +func claudeRecoveryService(upstream *claudeRecoverySequenceUpstream, enabled bool) *GatewayService { + cfg := &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}} + settings := fmt.Sprintf(`{"enabled":true,"thinking_signature_enabled":true,"apikey_signature_enabled":%t}`, enabled) + return &GatewayService{cfg: cfg, responseHeaderFilter: compileResponseHeaderFilter(cfg), httpUpstream: upstream, + rateLimitService: &RateLimitService{}, deferredService: &DeferredService{}, + settingService: NewSettingService(&gatewayTTLSettingRepo{data: map[string]string{SettingKeyRectifierSettings: settings}}, cfg)} +} + +func TestClaudeSignatureRecoveryForwardHTTPAndSSE(t *testing.T) { + for _, tc := range []struct { + name string + status int + prelude string + }{ + {"http_400", 400, ""}, + {"sse_before_message", 200, ""}, + {"sse_after_ping", 200, "event: ping\ndata: {\"type\":\"ping\"}\n\n"}, + {"sse_after_held_message_start", 200, "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"rejected-message\",\"model\":\"other-model\",\"usage\":{\"input_tokens\":900}}}\n\n"}, + } { + t.Run(tc.name, func(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + firstBody := claudeRecoverySignatureError + if tc.status == 200 { + firstBody = tc.prelude + "event: error\ndata: " + firstBody + "\n\n" + } + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{ + claudeRecoveryResponse(tc.status, firstBody, "initial-rid"), claudeRecoveryResponse(200, claudeRecoverySuccessStream(), "repaired-rid"), + }} + initialBody := &claudeRecoveryCloseObservedBody{ReadCloser: upstream.responses[0].Body} + upstream.responses[0].Body = initialBody + upstream.beforeSend = func(index int) { + if index == 1 { + require.True(t, initialBody.closed, "the rejected response must release its connection before repair sends") + } + } + account := newAnthropicAPIKeyAccountForTest() + account.Extra = nil + parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(claudeRecoveryTestBody)), PlatformAnthropic) + require.NoError(t, err) + acceptedCalls := 0 + parsed.OnUpstreamAccepted = func() { acceptedCalls++ } + result, err := claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.requests, 2) + require.Equal(t, []int64{account.ID, account.ID}, upstream.accountIDs) + require.Equal(t, upstream.requests[0].URL, upstream.requests[1].URL) + require.Equal(t, upstream.requests[0].Header.Get("X-Api-Key"), upstream.requests[1].Header.Get("X-Api-Key")) + require.Equal(t, 1, acceptedCalls) + require.Equal(t, 7, result.Usage.InputTokens) + require.Equal(t, 3, result.Usage.OutputTokens) + require.Equal(t, "repaired-rid", result.RequestID) + require.False(t, result.UpstreamResponseModelConflict) + require.NotContains(t, rec.Body.String(), "rejected-message") + require.Equal(t, 1, strings.Count(rec.Body.String(), "event: message_start")) + require.NotContains(t, string(upstream.bodies[1]), "private-signature-value") + require.Equal(t, "between_tools", gjson.GetBytes(upstream.bodies[1], "thinking.type").String()) + require.Equal(t, "high", gjson.GetBytes(upstream.bodies[1], "output_config.effort").String()) + require.True(t, bytes.Equal(parsed.Body.Bytes(), upstream.bodies[1])) + events, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + trigger := events.([]*OpsUpstreamErrorEvent)[0] + require.NotNil(t, trigger.SignatureRecovery) + require.Equal(t, "accepted", trigger.SignatureRecovery.Outcome) + require.Equal(t, 1, trigger.SignatureRecovery.Attempts) + require.Equal(t, 1, trigger.SignatureRecovery.Inbound.TotalCount) + diagnosticJSON, err := json.Marshal(trigger.SignatureRecovery) + require.NoError(t, err) + require.NotContains(t, string(diagnosticJSON), "private-signature-value") + require.NotContains(t, string(diagnosticJSON), "old reasoning") + }) + } +} + +func TestClaudeSignatureRecoveryDoesNotReplaySSEAgainOrAfterOutput(t *testing.T) { + for _, prefix := range []string{"", "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":4}}}\n\nevent: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"partial\"}}\n\n"} { + t.Run(fmt.Sprintf("output_%t", prefix != ""), func(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + errStream := prefix + "event: error\ndata: " + claudeRecoverySignatureError + "\n\n" + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{claudeRecoveryResponse(200, errStream, "one"), claudeRecoveryResponse(200, errStream, "two")}} + account := newAnthropicAPIKeyAccountForTest() + account.Extra = nil + parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(claudeRecoveryTestBody)), PlatformAnthropic) + require.NoError(t, err) + _, err = claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) + require.Error(t, err) + var failure *UpstreamFailoverError + require.ErrorAs(t, err, &failure) + require.Equal(t, 400, failure.StatusCode) + require.False(t, failure.ShouldRetryNextAccount()) + require.False(t, failure.ShouldReportAccountScheduleFailure()) + wantRequests := 2 + if prefix != "" { + wantRequests = 1 + } + require.Len(t, upstream.requests, wantRequests) + events, _ := c.Get(OpsUpstreamErrorsKey) + last := events.([]*OpsUpstreamErrorEvent)[wantRequests-1] + wantReason := "attempt_limit" + if prefix != "" { + wantReason = "output_started" + } + require.Equal(t, wantReason, last.SignatureRecovery.Reason) + }) + } +} + +func TestClaudeSignatureRecoveryDisabledAndPassthrough(t *testing.T) { + for _, passthrough := range []bool{false, true} { + t.Run(fmt.Sprintf("passthrough_%t", passthrough), func(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{claudeRecoveryResponse(400, claudeRecoverySignatureError, "original")}} + account := newAnthropicAPIKeyAccountForTest() + if !passthrough { + account.Extra = nil + } + parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(claudeRecoveryTestBody)), PlatformAnthropic) + require.NoError(t, err) + _, err = claudeRecoveryService(upstream, passthrough).Forward(context.Background(), c, account, parsed) + require.Error(t, err) + require.Len(t, upstream.requests, 1) + require.Contains(t, string(upstream.bodies[0]), "private-signature-value") + }) + } +} + +func TestClaudeSignatureRecoveryDecisionAndFailureEvidence(t *testing.T) { + for _, reason := range []string{"build_failed", "transport_failed", "empty_response", "budget_exhausted", "context_canceled", "body_unchanged", "output_started", "attempt_limit"} { + t.Run(reason, func(t *testing.T) { + state := newClaudeSignatureRecoveryState([]byte(claudeRecoveryTestBody)) + diag := state.diagnostic("http_400", false) + ctx := context.Background() + body := []byte(claudeRecoveryTestBody) + sent := 0 + clock := time.Unix(0, 0) + state.now = func() time.Time { return clock } + if reason == "context_canceled" { + var cancel context.CancelFunc + ctx, cancel = context.WithCancel(ctx) + cancel() + } + if reason == "body_unchanged" { + diag.Source = "sse_error" + body = []byte(`{"messages":[{"role":"user","content":"hi"}]}`) + } + if reason == "output_started" { + diag.MessageStarted = true + } + if reason == "attempt_limit" { + state.used = true + } + _, _, _ = state.run(ctx, diag, body, "claude-sonnet-5-5", func(body []byte) (*http.Request, []byte, error) { + if reason == "build_failed" { + return nil, nil, errors.New("build detail") + } + if reason == "budget_exhausted" { + clock = clock.Add(11 * time.Second) + } + req, err := http.NewRequest(http.MethodPost, "https://example.invalid/v1/messages", bytes.NewReader(body)) + return req, body, err + }, func(*http.Request) (*http.Response, error) { + sent++ + if reason == "transport_failed" { + return nil, errors.New("transport detail") + } + return nil, nil + }) + require.Equal(t, reason, diag.Reason) + require.Equal(t, sent, diag.Attempts) + if reason == "transport_failed" || reason == "empty_response" { + require.Equal(t, 1, sent) + } else { + require.Zero(t, sent) + } + }) + } +} + +func TestClaudeSignatureRecoveryFingerprintAndSanitizer(t *testing.T) { + blocks := make([]string, 20) + for i := range blocks { + blocks[i] = fmt.Sprintf(`{"type":"thinking","signature":"secret-%d"}`, i) + } + body := []byte(`{"messages":[{"role":"assistant","content":[` + strings.Join(blocks, ",") + `]}]}`) + state := newClaudeSignatureRecoveryState(body) + state.observeFirstWire(body) + state.observeFirstWire([]byte(`{}`)) + diag := state.diagnostic("sse_error", false) + require.Equal(t, 20, diag.Inbound.TotalCount) + require.Equal(t, 4, diag.Inbound.OmittedCount) + require.Len(t, diag.Inbound.Signatures, 16) + require.Equal(t, diag.Inbound, diag.FirstWire) + require.Equal(t, "/messages/0/content/0/signature", diag.Inbound.Signatures[0].Path) + diag.Attempts = 100 + diag.ElapsedMs = -1 + diag.Reason = "private text" + diag.FirstWire.Signatures[0].Path = "/messages/private-text" + safe := sanitizeClaudeSignatureRecoveryDiagnostic(diag) + require.Equal(t, 2, safe.Attempts) + require.Zero(t, safe.ElapsedMs) + require.Empty(t, safe.Reason) + require.Len(t, safe.FirstWire.Signatures, 15) + safe.Inbound.Signatures[0].Length = 0 + require.NotZero(t, diag.Inbound.Signatures[0].Length) + encoded, err := json.Marshal(safe) + require.NoError(t, err) + require.NotContains(t, string(encoded), "secret-") + require.NotContains(t, string(encoded), "private text") +} + +func TestClaudeSignatureRecoverySharedHTTPAndSSELimit(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + errStream := "event: error\ndata: " + claudeRecoverySignatureError + "\n\n" + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{ + claudeRecoveryResponse(400, claudeRecoverySignatureError, "http-rejection"), + claudeRecoveryResponse(200, errStream, "sse-rejection"), + }} + account := newAnthropicAPIKeyAccountForTest() + account.Extra = nil + parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(claudeRecoveryTestBody)), PlatformAnthropic) + require.NoError(t, err) + _, err = claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) + require.Error(t, err) + require.Len(t, upstream.requests, 2) + events, _ := c.Get(OpsUpstreamErrorsKey) + attempts := events.([]*OpsUpstreamErrorEvent) + require.Equal(t, "attempt_limit", attempts[len(attempts)-1].SignatureRecovery.Reason) +} + +func TestClaudeSignatureRecoveryFailureStopsWithoutHealthPenalty(t *testing.T) { + for _, streamFailure := range []bool{false, true} { + t.Run(fmt.Sprintf("stream_failure_%t", streamFailure), func(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + serviceError := `{"type":"error","error":{"type":"service_error","message":"temporarily unavailable"}}` + repairedStatus := 503 + repairedBody := serviceError + if streamFailure { + repairedStatus = 200 + repairedBody = "event: error\ndata: " + serviceError + "\n\n" + } + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{ + claudeRecoveryResponse(200, "event: error\ndata: "+claudeRecoverySignatureError+"\n\n", "initial"), + claudeRecoveryResponse(repairedStatus, repairedBody, "retry"), + }} + account := newAnthropicAPIKeyAccountForTest() + account.Extra = nil + parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(claudeRecoveryTestBody)), PlatformAnthropic) + require.NoError(t, err) + _, err = claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) + require.Error(t, err) + var failure *UpstreamFailoverError + require.ErrorAs(t, err, &failure) + require.False(t, failure.ShouldRetryNextAccount()) + require.False(t, failure.ShouldReportAccountScheduleFailure()) + require.Equal(t, serviceError, string(failure.ResponseBody)) + require.Len(t, upstream.requests, 2) + }) + } +} + +func TestClaudeSignatureRecoveryExplicitRequestPatternMatchesPrelude(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + // This is an existing HTTP rectifier match without a "thinking" word; it + // must agree with the prelude handler when the provider declares HTTP 400. + first := "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"discard-me\"}}\n\nevent: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"signature is invalid\"}}\n\n" + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{claudeRecoveryResponse(200, first, "first"), claudeRecoveryResponse(200, claudeRecoverySuccessStream(), "second")}} + account := newAnthropicAPIKeyAccountForTest() + account.Extra = nil + parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(claudeRecoveryTestBody)), PlatformAnthropic) + require.NoError(t, err) + result, err := claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.requests, 2) + require.NotContains(t, rec.Body.String(), "discard-me") + require.Equal(t, 1, strings.Count(rec.Body.String(), "event: message_start")) +} + +func TestClaudeSignatureRecoverySlowInitialResponseDoesNotSpendDecisionBudget(t *testing.T) { + clock := time.Unix(0, 0) + state := newClaudeSignatureRecoveryState([]byte(claudeRecoveryTestBody)) + state.now = func() time.Time { return clock } + state.observeFirstWire([]byte(claudeRecoveryTestBody)) + // Twenty seconds pass while waiting for the first upstream rejection. + clock = clock.Add(20 * time.Second) + diag := state.diagnostic("http_400", false) + sent := 0 + resp, _, err := state.run(context.Background(), diag, []byte(claudeRecoveryTestBody), "claude-sonnet-5-5", + func(body []byte) (*http.Request, []byte, error) { + req, err := http.NewRequest(http.MethodPost, "https://example.invalid/v1/messages", bytes.NewReader(body)) + return req, body, err + }, + func(*http.Request) (*http.Response, error) { + sent++ + clock = clock.Add(time.Second) + return claudeRecoveryResponse(200, claudeRecoverySuccessStream(), "fresh"), nil + }) + require.NoError(t, err) + require.NotNil(t, resp) + require.Equal(t, 1, sent) + require.Equal(t, "accepted", diag.Outcome) + require.Equal(t, int64(1000), diag.ElapsedMs) +} + +func TestClaudeSignatureRecoveryIncompleteRepairedStreamRecordsActualAttempt(t *testing.T) { + for _, initialHTTP := range []bool{false, true} { + t.Run(fmt.Sprintf("initial_http_%t", initialHTTP), func(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + firstStatus := 200 + firstBody := "event: error\ndata: " + claudeRecoverySignatureError + "\n\n" + if initialHTTP { + firstStatus = 400 + firstBody = claudeRecoverySignatureError + } + incomplete := strings.Split(claudeRecoverySuccessStream(), "event: message_stop")[0] + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{claudeRecoveryResponse(firstStatus, firstBody, "initial"), claudeRecoveryResponse(200, incomplete, "incomplete-retry")}} + account := newAnthropicAPIKeyAccountForTest() + account.Extra = nil + parsed, err := ParseGatewayRequest(NewRequestBodyRef([]byte(claudeRecoveryTestBody)), PlatformAnthropic) + require.NoError(t, err) + _, err = claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) + require.Error(t, err) + require.Contains(t, err.Error(), "missing terminal") + events, _ := c.Get(OpsUpstreamErrorsKey) + attempts := events.([]*OpsUpstreamErrorEvent) + require.Equal(t, "response_read_failed", attempts[0].SignatureRecovery.Outcome) + require.Equal(t, "stream_incomplete", attempts[0].SignatureRecovery.Reason) + last := attempts[len(attempts)-1] + require.Equal(t, "stream_failure", last.Kind) + require.Equal(t, 502, last.UpstreamStatusCode) + require.Equal(t, 200, last.UpstreamHTTPStatusCode) + require.Equal(t, "incomplete-retry", last.UpstreamRequestID) + require.Len(t, upstream.requests, 2) + }) + } +} + +func TestClaudeSignatureRecoveryHTTPToolOnlyKeepsExistingSecondStage(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + body := []byte(`{"model":"claude-opus-5-5","messages":[{"role":"assistant","content":[{"type":"tool_use","id":"tool1","name":"lookup","input":{}}]},{"role":"user","content":[{"type":"tool_result","tool_use_id":"tool1","content":"result"}]}]}`) + errBody := `{"type":"error","error":{"type":"invalid_request_error","message":"Invalid signature in tool_use block"}}` + upstream := &claudeRecoverySequenceUpstream{responses: []*http.Response{ + claudeRecoveryResponse(400, errBody, "original"), claudeRecoveryResponse(400, errBody, "first-repair"), + claudeRecoveryResponse(200, `{"type":"message","content":[{"type":"text","text":"done"}],"usage":{"input_tokens":7,"output_tokens":3}}`, "tool-repair"), + }} + account := newAnthropicAPIKeyAccountForTest() + account.Extra = nil + parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), PlatformAnthropic) + require.NoError(t, err) + result, err := claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.requests, 3) + require.Equal(t, gjson.GetBytes(upstream.bodies[0], "messages").Raw, gjson.GetBytes(upstream.bodies[1], "messages").Raw) + require.NotContains(t, string(upstream.bodies[2]), `"type":"tool_use"`) + require.NotContains(t, string(upstream.bodies[2]), `"type":"tool_result"`) + events, _ := c.Get(OpsUpstreamErrorsKey) + first := events.([]*OpsUpstreamErrorEvent)[0] + require.Equal(t, 2, first.SignatureRecovery.Attempts) + require.Equal(t, "accepted", first.SignatureRecovery.Outcome) +} diff --git a/backend/internal/service/gateway_claude_stream_error.go b/backend/internal/service/gateway_claude_stream_error.go new file mode 100644 index 000000000000..cf548f968de0 --- /dev/null +++ b/backend/internal/service/gateway_claude_stream_error.go @@ -0,0 +1,122 @@ +package service + +import ( + "net/http" + "strings" + + "github.com/tidwall/gjson" +) + +// ClaudeStreamErrorClassification describes an error carried inside an +// otherwise successful upstream SSE response. StatusCode is its semantic error +// status; it is not the HTTP status of the response carrying the event. +type ClaudeStreamErrorClassification struct { + StatusCode int + ClientStatusCode int + ClientErrorType string + ClientErrorCode string + ClientMessage string + RequestError bool +} + +// ClassifyClaudeStreamError classifies the original event without using whether +// client output has started. Output safety and signature recovery are separate +// decisions made by the caller. +func ClassifyClaudeStreamError(body []byte) ClaudeStreamErrorClassification { + errType := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.type").String())) + result := ClaudeStreamErrorClassification{ + StatusCode: http.StatusBadGateway, + ClientStatusCode: http.StatusBadGateway, + ClientErrorType: "upstream_error", + ClientMessage: sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(body))), + } + if code := gjson.GetBytes(body, "error.code"); code.Type == gjson.String { + result.ClientErrorCode = strings.TrimSpace(code.String()) + } + switch errType { + case "invalid_request_error": + result.StatusCode = http.StatusBadRequest + result.RequestError = true + case "authentication_error": + result.StatusCode = http.StatusUnauthorized + case "billing_error": + result.StatusCode = http.StatusPaymentRequired + case "permission_error": + result.StatusCode = http.StatusForbidden + case "not_found_error": + result.StatusCode = http.StatusNotFound + result.RequestError = true + case "conflict_error": + result.StatusCode = http.StatusConflict + result.RequestError = true + case "request_too_large", "request_too_large_error": + result.StatusCode = http.StatusRequestEntityTooLarge + result.RequestError = true + case "rate_limit_error": + result.StatusCode = http.StatusTooManyRequests + case "api_error", "service_error": + result.StatusCode = http.StatusInternalServerError + case "overloaded_error": + result.StatusCode = 529 + case "timeout_error": + result.StatusCode = http.StatusGatewayTimeout + default: + errType = "upstream_error" + if IsClaudeThinkingSignatureError(body) { + result.StatusCode = http.StatusBadRequest + result.RequestError = true + errType = "invalid_request_error" + } + } + result.ClientErrorType = errType + if result.RequestError || result.StatusCode == http.StatusTooManyRequests { + result.ClientStatusCode = result.StatusCode + } else if result.StatusCode == 529 { + result.ClientStatusCode = http.StatusServiceUnavailable + } + if result.ClientMessage == "" { + result.ClientMessage = "Upstream returned an error in the response stream" + } + return result +} + +// FailoverError retains the original event for rules and administrator +// diagnostics. Request validation failures must not be replayed on another +// account or counted as account health failures. +func (classification ClaudeStreamErrorClassification) FailoverError(body []byte) *UpstreamFailoverError { + err := &UpstreamFailoverError{ + StatusCode: classification.StatusCode, + ResponseBody: body, + ClientStatusCode: classification.ClientStatusCode, + ClientErrorType: classification.ClientErrorType, + ClientErrorCode: classification.ClientErrorCode, + ClientMessage: classification.ClientMessage, + } + if classification.RequestError { + err.NextAccountAction = NextAccountStop + err.SuppressAccountHealthPenalty = true + } + return err +} + +// IsClaudeThinkingSignatureError recognizes explicit rejection of a thinking +// signature. Merely mentioning a signature or thinking is not sufficient: such +// messages can describe unrelated provider failures or user content. +func IsClaudeThinkingSignatureError(body []byte) bool { + message := strings.ToLower(strings.TrimSpace(extractUpstreamErrorMessage(body))) + if !strings.Contains(message, "signature") || + (!strings.Contains(message, "thinking") && !strings.Contains(message, "thought_signature")) { + return false + } + for _, rejection := range []string{ + "invalid", "not valid", "missing", "required", "malformed", "mismatch", + "does not match", "doesn't match", "cannot verify", "cannot be verified", + "unable to verify", "verification failed", "failed to verify", "failed to validate", + "different conversation", "another conversation", "different session", "another session", + } { + if strings.Contains(message, rejection) { + return true + } + } + return false +} diff --git a/backend/internal/service/gateway_claude_stream_error_test.go b/backend/internal/service/gateway_claude_stream_error_test.go new file mode 100644 index 000000000000..5864814f161f --- /dev/null +++ b/backend/internal/service/gateway_claude_stream_error_test.go @@ -0,0 +1,89 @@ +package service + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestClaudeClassifyStreamError(t *testing.T) { + tests := []struct { + name string + errorType string + message string + status int + clientStatus int + clientType string + requestError bool + }{ + {"invalid request", "invalid_request_error", "Invalid request body", 400, 400, "invalid_request_error", true}, + {"authentication", "authentication_error", "Authentication failed", 401, 502, "authentication_error", false}, + {"billing", "billing_error", "Payment required", 402, 502, "billing_error", false}, + {"permission", "permission_error", "Permission denied", 403, 502, "permission_error", false}, + {"not found", "not_found_error", "Model not found", 404, 404, "not_found_error", true}, + {"conflict", "conflict_error", "Request conflict", 409, 409, "conflict_error", true}, + {"too large", "request_too_large", "Request too large", 413, 413, "request_too_large", true}, + {"too large alias", "request_too_large_error", "Request too large", 413, 413, "request_too_large_error", true}, + {"rate limited", "rate_limit_error", "Rate limit exceeded", 429, 429, "rate_limit_error", false}, + {"api error", "api_error", "Internal error", 500, 502, "api_error", false}, + {"third party service", "service_error", "Service temporarily unavailable", 500, 502, "service_error", false}, + {"timeout", "timeout_error", "Request timed out", 504, 502, "timeout_error", false}, + {"overloaded", "overloaded_error", "Overloaded", 529, 503, "overloaded_error", false}, + {"unknown", "provider_specific_error", "Provider failure", 502, 502, "upstream_error", false}, + {"missing type signature", "", "Invalid `signature` in `thinking` block", 400, 400, "invalid_request_error", true}, + {"nil type signature", "", "Invalid `signature` in `thinking` block", 400, 400, "invalid_request_error", true}, + {"unknown type signature", "provider_specific_error", "Invalid `signature` in `thinking` block", 400, 400, "invalid_request_error", true}, + {"explicit permission beats signature text", "permission_error", "Invalid `signature` in `thinking` block", 403, 502, "permission_error", false}, + {"explicit authentication beats signature text", "authentication_error", "Invalid `signature` in `thinking` block", 401, 502, "authentication_error", false}, + {"explicit service beats signature text", "service_error", "Invalid `signature` in `thinking` block", 500, 502, "service_error", false}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + body, err := json.Marshal(map[string]any{"type": "error", "error": map[string]string{"type": test.errorType, "message": test.message, "code": "provider_code"}}) + require.NoError(t, err) + classification := ClassifyClaudeStreamError(body) + require.Equal(t, test.status, classification.StatusCode) + require.Equal(t, test.clientStatus, classification.ClientStatusCode) + require.Equal(t, test.clientType, classification.ClientErrorType) + require.Equal(t, test.message, classification.ClientMessage) + require.Equal(t, "provider_code", classification.ClientErrorCode) + require.Equal(t, test.requestError, classification.RequestError) + failoverErr := classification.FailoverError(body) + require.Equal(t, !test.requestError, failoverErr.ShouldRetryNextAccount()) + require.Equal(t, test.requestError, failoverErr.SuppressAccountHealthPenalty) + require.False(t, failoverErr.RetryableOnSameAccount) + require.Equal(t, body, failoverErr.ResponseBody) + }) + } +} + +func TestClaudeThinkingSignatureError(t *testing.T) { + for _, test := range []struct { + message string + want bool + }{ + {"Invalid `signature` in `thinking` block", true}, + {"messages.2.content.0.thinking.signature: Field required", true}, + {"thinking signature is bound to a different conversation", true}, + {"thought_signature verification failed", true}, + {"Invalid API key signature", false}, + {"thinking signature service is temporarily unavailable", false}, + {"The response uses thinking and a signature", false}, + {"all messages must have non-empty content", false}, + } { + t.Run(test.message, func(t *testing.T) { + body, err := json.Marshal(map[string]any{"error": map[string]string{"message": test.message}}) + require.NoError(t, err) + require.Equal(t, test.want, IsClaudeThinkingSignatureError(body)) + }) + } +} + +func TestClaudeClassifyStreamError_MalformedBody(t *testing.T) { + classification := ClassifyClaudeStreamError([]byte("not JSON")) + require.Equal(t, 502, classification.StatusCode) + require.Equal(t, "upstream_error", classification.ClientErrorType) + require.NotEmpty(t, classification.ClientMessage) + require.False(t, classification.RequestError) +} diff --git a/backend/internal/service/gateway_forward.go b/backend/internal/service/gateway_forward.go index 0e88aef75413..ea525548ddad 100644 --- a/backend/internal/service/gateway_forward.go +++ b/backend/internal/service/gateway_forward.go @@ -14,7 +14,6 @@ import ( "github.com/Wei-Shaw/sub2api/internal/pkg/claude" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" - "github.com/tidwall/gjson" "github.com/gin-gonic/gin" ) @@ -93,6 +92,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A if parsed == nil { return nil, fmt.Errorf("parse request: empty request") } + signatureRecovery := newClaudeSignatureRecoveryState(parsed.Body.Bytes()) // API-key mappings and OAuth native IDs are resolved before mimicry. validationModel := parsed.Model if account != nil { @@ -387,6 +387,10 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A } // 重试循环 + // Keep the unsigned request view for signature recovery, even after an + // accepted attempt updates ParsedRequest to its actual signed wire body. + signatureRecoveryBase := body + var activeRecoveryDiagnostic *ClaudeSignatureRecoveryDiagnostic var resp *http.Response lastWireBody := body retryStart := time.Now() @@ -400,6 +404,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A } // 记录本次实际发送的 wire body；只有请求成功后才写回 ParsedRequest，避免 400 retry 基于已签名 CCH 再改写。 lastWireBody = wireBody + signatureRecovery.observeFirstWire(wireBody) // 发送请求 resp, err = s.httpUpstream.DoWithTLS(upstreamReq, proxyURL, account.ID, account.Concurrency, tlsProfile) @@ -418,7 +423,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A if readErr == nil { _ = resp.Body.Close() - if s.shouldRectifySignatureError(ctx, account, respBody, reqModel) { + if ShouldRectifyThinkingSignatureError(reqModel) && s.claudeSignatureErrorCandidate(ctx, account, respBody) { + recoveryDiagnostic := signatureRecovery.diagnostic("http_400", false) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ ProxyID: opsUpstreamProxyID(account), ProxyName: opsUpstreamProxyName(account), @@ -430,6 +436,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A UpstreamURL: safeUpstreamURL(upstreamReq.URL.String()), Kind: "signature_error", Message: extractUpstreamErrorMessage(respBody), + SignatureRecovery: recoveryDiagnostic, Detail: func() string { if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { return truncateString(string(respBody), s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes) @@ -437,6 +444,11 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A return "" }(), }) + if !s.claudeSignatureRecoveryEnabled(ctx, account, reqModel) { + recoveryDiagnostic.Reason = "disabled" + resp.Body = io.NopCloser(bytes.NewReader(respBody)) + break + } looksLikeToolSignatureError := func(msg string) bool { m := strings.ToLower(msg) @@ -448,114 +460,122 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A strings.Contains(m, "function_response") } - // 避免在重试预算已耗尽时再发起额外请求 - if time.Since(retryStart) >= maxRetryElapsed { - resp.Body = io.NopCloser(bytes.NewReader(respBody)) - break - } - logger.LegacyPrintf("service.gateway", "[warn] Account %d: thinking blocks have invalid signature, retrying with filtered blocks", account.ID) - // Conservative two-stage fallback: // 1) Disable thinking + thinking->text (preserve content) // 2) Only if upstream still errors AND error message points to tool/function signature issues: // also downgrade tool_use/tool_result blocks to text. - filteredBody := FilterThinkingBlocksForRetry(body, reqModel) - retryCtx, releaseRetryCtx := detachStreamUpstreamContext(ctx, reqStream) - retryReq, retryWireBody, buildErr := s.buildUpstreamRequest(retryCtx, c, account, filteredBody, token, tokenType, reqModel, reqStream, shouldMimicClaudeCode) - releaseRetryCtx() - if buildErr == nil { - retryResp, retryErr := s.httpUpstream.DoWithTLS(retryReq, proxyURL, account.ID, account.Concurrency, tlsProfile) - if retryErr == nil { - if retryResp.StatusCode < 400 { - // 重试请求被上游接受后同步 ParsedRequest，保证 usage/日志看到真实请求体。 - lastWireBody = retryWireBody - if err := replaceBody(retryWireBody); err != nil { - _ = retryResp.Body.Close() - return nil, err - } - logger.LegacyPrintf("service.gateway", "Account %d: thinking block retry succeeded (blocks downgraded)", account.ID) - resp = retryResp - break + var retryReq *http.Request + retryResp, retryWireBody, retryErr := signatureRecovery.run(ctx, recoveryDiagnostic, body, reqModel, + func(filteredBody []byte) (*http.Request, []byte, error) { + retryCtx, releaseRetryCtx := detachStreamUpstreamContext(ctx, reqStream) + defer releaseRetryCtx() + var wire []byte + var buildErr error + retryReq, wire, buildErr = s.buildUpstreamRequest(retryCtx, c, account, filteredBody, token, tokenType, reqModel, reqStream, shouldMimicClaudeCode) + return retryReq, wire, buildErr + }, + func(req *http.Request) (*http.Response, error) { + return s.httpUpstream.DoWithTLS(req, proxyURL, account.ID, account.Concurrency, tlsProfile) + }) + if retryResp != nil && retryErr == nil { + if retryResp.StatusCode < 400 { + activeRecoveryDiagnostic = recoveryDiagnostic + // 重试请求被上游接受后同步 ParsedRequest，保证 usage/日志看到真实请求体。 + lastWireBody = retryWireBody + if err := replaceBody(retryWireBody); err != nil { + _ = retryResp.Body.Close() + return nil, err } + logger.LegacyPrintf("service.gateway", "Account %d: thinking block retry succeeded (blocks downgraded)", account.ID) + resp = retryResp + break + } - retryRespBody, retryReadErr := s.readUpstreamErrorBody(retryResp) - _ = retryResp.Body.Close() - if retryReadErr == nil && retryResp.StatusCode == 400 && s.isSignatureErrorPattern(ctx, account, retryRespBody) { - appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - ProxyID: opsUpstreamProxyID(account), - ProxyName: opsUpstreamProxyName(account), - Platform: account.Platform, - AccountID: account.ID, - AccountName: account.Name, - UpstreamStatusCode: retryResp.StatusCode, - UpstreamRequestID: retryResp.Header.Get("x-request-id"), - UpstreamURL: safeUpstreamURL(retryReq.URL.String()), - Kind: "signature_retry_thinking", - Message: extractUpstreamErrorMessage(retryRespBody), - Detail: func() string { - if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { - return truncateString(string(retryRespBody), s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes) - } - return "" - }(), - }) - msg2 := extractUpstreamErrorMessage(retryRespBody) - if looksLikeToolSignatureError(msg2) && time.Since(retryStart) < maxRetryElapsed { - logger.LegacyPrintf("service.gateway", "Account %d: signature retry still failing and looks tool-related, retrying with tool blocks downgraded", account.ID) - filteredBody2 := FilterSignatureSensitiveBlocksForRetry(body, reqModel) - retryCtx2, releaseRetryCtx2 := detachStreamUpstreamContext(ctx, reqStream) - retryReq2, retryWireBody2, buildErr2 := s.buildUpstreamRequest(retryCtx2, c, account, filteredBody2, token, tokenType, reqModel, reqStream, shouldMimicClaudeCode) - releaseRetryCtx2() - if buildErr2 == nil { - retryResp2, retryErr2 := s.httpUpstream.DoWithTLS(retryReq2, proxyURL, account.ID, account.Concurrency, tlsProfile) - if retryErr2 == nil { - if retryResp2.StatusCode < 400 { - // 二阶段工具块降级成功时也必须更新当前 body。 - lastWireBody = retryWireBody2 - if err := replaceBody(retryWireBody2); err != nil { - _ = retryResp2.Body.Close() - return nil, err - } + retryRespBody, retryReadErr := s.readUpstreamErrorBody(retryResp) + _ = retryResp.Body.Close() + if retryReadErr != nil { + recoveryDiagnostic.Outcome, recoveryDiagnostic.Reason = "response_read_failed", "response_read_failed" + resp.Body = io.NopCloser(bytes.NewReader(respBody)) + break + } + if retryReadErr == nil && retryResp.StatusCode == 400 && s.isSignatureErrorPattern(ctx, account, retryRespBody) { + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, + AccountID: account.ID, + AccountName: account.Name, + UpstreamStatusCode: retryResp.StatusCode, + UpstreamRequestID: retryResp.Header.Get("x-request-id"), + UpstreamURL: safeUpstreamURL(retryReq.URL.String()), + Kind: "signature_retry_thinking", + Message: extractUpstreamErrorMessage(retryRespBody), + Detail: func() string { + if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { + return truncateString(string(retryRespBody), s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes) + } + return "" + }(), + }) + msg2 := extractUpstreamErrorMessage(retryRespBody) + if looksLikeToolSignatureError(msg2) && time.Since(retryStart) < maxRetryElapsed { + logger.LegacyPrintf("service.gateway", "Account %d: signature retry still failing and looks tool-related, retrying with tool blocks downgraded", account.ID) + filteredBody2 := FilterSignatureSensitiveBlocksForRetry(body, reqModel) + retryCtx2, releaseRetryCtx2 := detachStreamUpstreamContext(ctx, reqStream) + retryReq2, retryWireBody2, buildErr2 := s.buildUpstreamRequest(retryCtx2, c, account, filteredBody2, token, tokenType, reqModel, reqStream, shouldMimicClaudeCode) + releaseRetryCtx2() + if buildErr2 == nil { + recoveryDiagnostic.Attempts++ + retryResp2, retryErr2 := s.httpUpstream.DoWithTLS(retryReq2, proxyURL, account.ID, account.Concurrency, tlsProfile) + if retryErr2 == nil { + recoveryDiagnostic.Outcome = "upstream_rejected" + if retryResp2.StatusCode < 400 { + activeRecoveryDiagnostic = recoveryDiagnostic + recoveryDiagnostic.Outcome = "accepted" + // 二阶段工具块降级成功时也必须更新当前 body。 + lastWireBody = retryWireBody2 + if err := replaceBody(retryWireBody2); err != nil { + _ = retryResp2.Body.Close() + return nil, err } - resp = retryResp2 - break - } - if retryResp2 != nil && retryResp2.Body != nil { - _ = retryResp2.Body.Close() } - appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - ProxyID: opsUpstreamProxyID(account), - ProxyName: opsUpstreamProxyName(account), - Platform: account.Platform, - AccountID: account.ID, - AccountName: account.Name, - UpstreamStatusCode: 0, - UpstreamURL: safeUpstreamURL(retryReq2.URL.String()), - Kind: "signature_retry_tools_request_error", - Message: sanitizeUpstreamErrorMessage(retryErr2.Error()), - }) - logger.LegacyPrintf("service.gateway", "Account %d: tool-downgrade signature retry failed: %v", account.ID, retryErr2) - } else { - logger.LegacyPrintf("service.gateway", "Account %d: tool-downgrade signature retry build failed: %v", account.ID, buildErr2) + resp = retryResp2 + break } + if retryResp2 != nil && retryResp2.Body != nil { + _ = retryResp2.Body.Close() + } + recoveryDiagnostic.Outcome, recoveryDiagnostic.Reason = "transport_failed", "transport_failed" + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, + AccountID: account.ID, + AccountName: account.Name, + UpstreamStatusCode: 0, + UpstreamURL: safeUpstreamURL(retryReq2.URL.String()), + Kind: "signature_retry_tools_request_error", + Message: sanitizeUpstreamErrorMessage(retryErr2.Error()), + }) + logger.LegacyPrintf("service.gateway", "Account %d: tool-downgrade signature retry failed: %v", account.ID, retryErr2) + } else { + recoveryDiagnostic.Outcome, recoveryDiagnostic.Reason = "build_failed", "build_failed" + logger.LegacyPrintf("service.gateway", "Account %d: tool-downgrade signature retry build failed: %v", account.ID, buildErr2) } } - - // Fall back to the original retry response context. - resp = &http.Response{ - StatusCode: retryResp.StatusCode, - Header: retryResp.Header.Clone(), - Body: io.NopCloser(bytes.NewReader(retryRespBody)), - } - break } - if retryResp != nil && retryResp.Body != nil { - _ = retryResp.Body.Close() + + // Fall back to the original retry response context. + resp = &http.Response{ + StatusCode: retryResp.StatusCode, + Header: retryResp.Header.Clone(), + Body: io.NopCloser(bytes.NewReader(retryRespBody)), } - logger.LegacyPrintf("service.gateway", "Account %d: signature error retry failed: %v", account.ID, retryErr) - } else { - logger.LegacyPrintf("service.gateway", "Account %d: signature error retry build request failed: %v", account.ID, buildErr) + break + } + if retryErr != nil { + logger.LegacyPrintf("service.gateway", "Account %d: signature recovery failed (%s): %v", account.ID, recoveryDiagnostic.Reason, retryErr) } // Retry failed: restore original response body and continue handling. @@ -821,71 +841,135 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A var firstTokenMs *int var clientDisconnect bool if reqStream { - writerSizeBeforeStream := c.Writer.Size() - streamResult, err := s.handleStreamingResponse(ctx, resp, c, account, startTime, originalModel, reqModel, shouldMimicClaudeCode) - if err != nil { - var sseErr *sseStreamErrorEventError - if errors.As(err, &sseErr) { - // 上游 HTTP 200 + SSE 流体内出现 event:error 帧。 - body := []byte(sseErr.RawData) - semanticStatus := http.StatusForbidden - if c.Writer.Size() == writerSizeBeforeStream && gjson.GetBytes(body, "error.type").String() == "overloaded_error" { - semanticStatus = 529 - syntheticResp := &http.Response{ - StatusCode: semanticStatus, - Header: resp.Header.Clone(), - Body: io.NopCloser(bytes.NewReader(body)), + allowSignatureRecovery := s.claudeSignatureRecoveryEnabled(ctx, account, reqModel) + for { + streamResult, err := s.handleStreamingResponse(ctx, resp, c, account, startTime, originalModel, reqModel, shouldMimicClaudeCode, + allowSignatureRecovery && !signatureRecovery.used, activeRecoveryDiagnostic != nil) + if err != nil { + var sseErr *sseStreamErrorEventError + if errors.As(err, &sseErr) { + streamErrorBody := []byte(sseErr.RawData) + classification := ClassifyClaudeStreamError(streamErrorBody) + if activeRecoveryDiagnostic != nil { + activeRecoveryDiagnostic.Outcome = "upstream_rejected" + } + var recoveryDiagnostic *ClaudeSignatureRecoveryDiagnostic + if classification.StatusCode == http.StatusBadRequest && s.claudeSignatureErrorCandidate(ctx, account, streamErrorBody) { + recoveryDiagnostic = signatureRecovery.diagnostic("sse_error", sseErr.MessageStarted) + if !allowSignatureRecovery { + recoveryDiagnostic.Reason = "disabled" + } + } + upstreamDetail := "" + if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { + maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes + if maxBytes <= 0 { + maxBytes = 2048 + } + upstreamDetail = truncateString(sseErr.RawData, maxBytes) } - s.handleFailoverSideEffects(ctx, syntheticResp, account, reqModel) - } - - upstreamMsg := sanitizeUpstreamErrorMessage( - strings.TrimSpace(extractUpstreamErrorMessage(body)), - ) - upstreamDetail := "" - if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { - maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes - if maxBytes <= 0 { - maxBytes = 2048 + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, + AccountID: account.ID, + AccountName: account.Name, + UpstreamStatusCode: classification.StatusCode, + UpstreamHTTPStatusCode: resp.StatusCode, + UpstreamRequestID: resp.Header.Get("x-request-id"), + Kind: "stream_error", + Message: classification.ClientMessage, + Detail: upstreamDetail, + SignatureRecovery: recoveryDiagnostic, + }) + if recoveryDiagnostic != nil && allowSignatureRecovery { + // Release the rejected response before acquiring another + // connection, including pools with a single host slot. + _ = resp.Body.Close() + retryResp, retryWireBody, recoveryErr := signatureRecovery.run(ctx, recoveryDiagnostic, signatureRecoveryBase, reqModel, + func(filteredBody []byte) (*http.Request, []byte, error) { + retryCtx, releaseRetryCtx := detachStreamUpstreamContext(ctx, reqStream) + defer releaseRetryCtx() + return s.buildUpstreamRequest(retryCtx, c, account, filteredBody, token, tokenType, reqModel, reqStream, shouldMimicClaudeCode) + }, + func(req *http.Request) (*http.Response, error) { + return s.httpUpstream.DoWithTLS(req, proxyURL, account.ID, account.Concurrency, tlsProfile) + }) + if recoveryErr != nil { + logger.LegacyPrintf("service.gateway", "Account %d: SSE signature recovery failed (%s): %v", account.ID, recoveryDiagnostic.Reason, recoveryErr) + } + if retryResp != nil { + _ = resp.Body.Close() + resp = retryResp + if resp.StatusCode < 400 { + if err := replaceBody(retryWireBody); err != nil { + return nil, err + } + beginUpstreamResponseModelObservation(c) + activeRecoveryDiagnostic = recoveryDiagnostic + continue + } + // A repair can fail with a real HTTP response, including while + // earlier keepalive pings have committed HTTP 200. Let the + // streaming-aware handler write the terminal, never append JSON. + retryBody, readErr := s.readUpstreamErrorBody(resp) + if readErr != nil { + recoveryDiagnostic.Outcome, recoveryDiagnostic.Reason = "response_read_failed", "response_read_failed" + } + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, + UpstreamStatusCode: resp.StatusCode, UpstreamRequestID: resp.Header.Get("x-request-id"), + Kind: "http_error", Message: extractUpstreamErrorMessage(retryBody), + }) + failure := &UpstreamFailoverError{StatusCode: resp.StatusCode, ResponseBody: retryBody, NextAccountAction: NextAccountStop, SuppressAccountHealthPenalty: true} + if resp.StatusCode == http.StatusBadRequest || resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusRequestEntityTooLarge { + failure.ClientStatusCode = resp.StatusCode + failure.ClientErrorType = "invalid_request_error" + failure.ClientMessage = sanitizeUpstreamErrorMessage(extractUpstreamErrorMessage(retryBody)) + } + return nil, failure + } + } + if activeRecoveryDiagnostic == nil && !sseErr.MessageStarted && classification.StatusCode == 529 { + syntheticResp := &http.Response{StatusCode: classification.StatusCode, Header: resp.Header.Clone(), Body: io.NopCloser(bytes.NewReader(streamErrorBody))} + s.handleFailoverSideEffects(ctx, syntheticResp, account, reqModel) + } + failure := classification.FailoverError(streamErrorBody) + if activeRecoveryDiagnostic != nil { + failure.NextAccountAction = NextAccountStop + failure.SuppressAccountHealthPenalty = true } - upstreamDetail = truncateString(sseErr.RawData, maxBytes) + return nil, failure } - - appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - ProxyID: opsUpstreamProxyID(account), - ProxyName: opsUpstreamProxyName(account), - Platform: account.Platform, - AccountID: account.ID, - AccountName: account.Name, - UpstreamStatusCode: semanticStatus, - UpstreamRequestID: resp.Header.Get("x-request-id"), - Kind: "stream_error", - Message: upstreamMsg, - Detail: upstreamDetail, - }) - - logger.LegacyPrintf("service.gateway", - "[Forward] SSE error event in stream: Account=%d(%s) RequestID=%s Body=%s", - account.ID, account.Name, resp.Header.Get("x-request-id"), - truncateString(sseErr.RawData, 1000), - ) - - return nil, &UpstreamFailoverError{ - StatusCode: semanticStatus, - ResponseBody: body, + if activeRecoveryDiagnostic != nil { + activeRecoveryDiagnostic.Outcome, activeRecoveryDiagnostic.Reason = "response_read_failed", "stream_incomplete" + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, + UpstreamStatusCode: http.StatusBadGateway, UpstreamHTTPStatusCode: resp.StatusCode, + UpstreamRequestID: resp.Header.Get("x-request-id"), + Kind: "stream_failure", Message: sanitizeStreamError(err), + }) + var failure *UpstreamFailoverError + if errors.As(err, &failure) { + failure.NextAccountAction = NextAccountStop + failure.SuppressAccountHealthPenalty = true + } } + // 流中断（缺失 terminal 事件、读错误、数据间隔超时等）时保留已观测到的 + // usage 与错误一起返回，handler 在错误处理完成后照常提交 usage 记录。 + if partial := partialStreamUsageResult(c, resp, streamResult, originalModel, mappedModel, startTime, err); partial != nil { + return partial, err + } + return nil, err } - // 流中断（缺失 terminal 事件、读错误、数据间隔超时等）时保留已观测到的 - // usage 与错误一起返回，handler 在错误处理完成后照常提交 usage 记录。 - if partial := partialStreamUsageResult(c, resp, streamResult, originalModel, mappedModel, startTime, err); partial != nil { - return partial, err - } - return nil, err + usage = streamResult.usage + firstTokenMs = streamResult.firstTokenMs + clientDisconnect = streamResult.clientDisconnect + break } - usage = streamResult.usage - firstTokenMs = streamResult.firstTokenMs - clientDisconnect = streamResult.clientDisconnect } else { usage, err = s.handleNonStreamingResponse(ctx, resp, c, account, originalModel, reqModel) if err != nil { diff --git a/backend/internal/service/gateway_forward_partial_usage_test.go b/backend/internal/service/gateway_forward_partial_usage_test.go index 90f6757e0119..617940ce9d13 100644 --- a/backend/internal/service/gateway_forward_partial_usage_test.go +++ b/backend/internal/service/gateway_forward_partial_usage_test.go @@ -261,7 +261,7 @@ func TestGatewayService_Forward_PreOutputSSEOverloadedErrorUsesSemantic529(t *te require.Empty(t, rec.Body.String(), "pre-output overload must remain eligible for account failover") } -func TestGatewayService_Forward_PostOutputSSEOverloadedErrorKeepsExistingStatus(t *testing.T) { +func TestGatewayService_Forward_PostOutputSSEOverloadedErrorUsesSemantic529(t *testing.T) { gin.SetMode(gin.TestMode) rec := httptest.NewRecorder() @@ -274,6 +274,7 @@ func TestGatewayService_Forward_PostOutputSSEOverloadedErrorKeepsExistingStatus( const errorJSON = `{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}` fixture := "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":1}}}\n\n" + + "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n" + "event: error\ndata: " + errorJSON + "\n\n" upstream := &anthropicHTTPUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, @@ -296,10 +297,13 @@ func TestGatewayService_Forward_PostOutputSSEOverloadedErrorKeepsExistingStatus( var failoverErr *UpstreamFailoverError require.ErrorAs(t, err, &failoverErr) - require.Equal(t, http.StatusForbidden, failoverErr.StatusCode) + require.Equal(t, 529, failoverErr.StatusCode, "stream output must not change the upstream error classification") require.JSONEq(t, errorJSON, string(failoverErr.ResponseBody)) require.Zero(t, repo.tempCalls) + require.Zero(t, repo.overloadCalls, "a post-output overload must not add the pre-output failover cooldown") + require.Empty(t, repo.modelRateLimitCalls) require.Contains(t, rec.Body.String(), "message_start") + require.Contains(t, rec.Body.String(), "content_block_start") } func TestGatewayService_AnthropicAPIKeyPassthrough_ForwardStreamMissingTerminalPreservesPartialUsage(t *testing.T) { diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 70132eafad00..4b64b3477ab5 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -784,6 +784,9 @@ func (e *UpstreamFailoverError) ShouldReportAccountScheduleFailure() bool { // 提取 RawData 并构造 UpstreamFailoverError.ResponseBody。 type sseStreamErrorEventError struct { RawData string + // MessageStarted excludes keepalive pings/comments and HTTP headers. Once + // a message event reaches the client, another attempt cannot replace it. + MessageStarted bool } func (e *sseStreamErrorEventError) Error() string { return "have error in stream" } diff --git a/backend/internal/service/gateway_upstream_response.go b/backend/internal/service/gateway_upstream_response.go index 6aaaaf4b16d5..32317756cf2c 100644 --- a/backend/internal/service/gateway_upstream_response.go +++ b/backend/internal/service/gateway_upstream_response.go @@ -658,6 +658,11 @@ type streamingResult struct { clientDisconnect bool // 客户端是否在流式传输过程中断开 } +// Only message_start is held back, so a signature rejection before content can +// be repaired without exposing two message IDs to the client. Larger preludes +// are committed immediately rather than retaining an unbounded upstream frame. +const claudeMessageStartPreludeLimit = 64 * 1024 + // hasObservedTokens 报告流式过程中是否已观测到任何上游计量的 token。 func (u *ClaudeUsage) hasObservedTokens() bool { if u == nil { @@ -700,7 +705,7 @@ func partialStreamUsageResult(c *gin.Context, resp *http.Response, streamResult } } -func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http.Response, c *gin.Context, account *Account, startTime time.Time, originalModel, mappedModel string, mimicClaudeCode bool) (*streamingResult, error) { +func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http.Response, c *gin.Context, account *Account, startTime time.Time, originalModel, mappedModel string, mimicClaudeCode bool, allowPreludeRecovery ...bool) (*streamingResult, error) { observer := upstreamResponseModelObserverFromContext(c) if observer == nil { observer = beginUpstreamResponseModelObservation(c) @@ -845,6 +850,42 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http needModelReplace := originalModel != mappedModel clientDisconnected := false // 客户端断开标志，断开后继续读取上游以获取完整usage + messageStarted := false // 本次尝试已提交非 ping 事件；HTTP headers/ping 不妨碍签名恢复。 + preludeRecoveryEnabled := len(allowPreludeRecovery) > 0 && allowPreludeRecovery[0] + // The second optional flag is used only by a sent signature repair. A + // failed repair belongs to the rejected request, not the account's health. + suppressAccountHealthPenalty := len(allowPreludeRecovery) > 1 && allowPreludeRecovery[1] + bufferPrelude := preludeRecoveryEnabled + var pendingPrelude []string + pendingPreludeBytes := 0 + writeBlocks := func(blocks []string, eventType string) { + for _, block := range blocks { + if clientDisconnected { + continue + } + restored := reverseToolNamesIfPresent(c, []byte(block)) + written, werr := fmt.Fprint(w, string(restored)) + if written > 0 && eventType != "ping" { + messageStarted = true + } + if werr != nil { + clientDisconnected = true + logger.LegacyPrintf("service.gateway", "Client disconnected during streaming, continuing to drain upstream for billing") + continue + } + flusher.Flush() + lastDataAt = time.Now() + resetKeepaliveTimer() + } + } + commitPrelude := func() { + if len(pendingPrelude) == 0 { + return + } + writeBlocks(pendingPrelude, "message_start") + pendingPrelude = nil + pendingPreludeBytes = 0 + } sawTerminalEvent := false useNoopDeltaKeepalive := c != nil && c.Request != nil && shouldUseClaudeCodeNoopDeltaKeepalive(c.GetHeader("User-Agent")) noopDeltaKeepaliveBlockIndex := -1 @@ -852,9 +893,9 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http pendingEventLines := make([]string, 0, 4) - processSSEEvent := func(lines []string) ([]string, string, *sseUsagePatch, error) { + processSSEEvent := func(lines []string) ([]string, string, *sseUsagePatch, string, error) { if len(lines) == 0 { - return nil, "", nil, nil + return nil, "", nil, "", nil } eventName := "" @@ -871,11 +912,16 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http } if eventName == "error" { - return nil, dataLine, nil, &sseStreamErrorEventError{RawData: dataLine} + return nil, dataLine, nil, "error", &sseStreamErrorEventError{RawData: dataLine} } if dataLine == "" { - return []string{strings.Join(lines, "\n") + "\n\n"}, "", nil, nil + // SSE comments are keepalives too; they must not commit a message prelude. + kind := eventName + if kind == "" { + kind = "ping" + } + return []string{strings.Join(lines, "\n") + "\n\n"}, "", nil, kind, nil } if dataLine == "[DONE]" { @@ -885,7 +931,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http block = "event: " + eventName + "\n" } block += "data: " + dataLine + "\n\n" - return []string{block}, dataLine, nil, nil + return []string{block}, dataLine, nil, "message_stop", nil } var event map[string]any @@ -896,14 +942,29 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http block = "event: " + eventName + "\n" } block += "data: " + dataLine + "\n\n" - return []string{block}, dataLine, nil, nil + kind := eventName + if kind == "" { + kind = "unknown" + } + return []string{block}, dataLine, nil, kind, nil } eventType, _ := event["type"].(string) + if eventType == "error" { + return nil, dataLine, nil, "error", &sseStreamErrorEventError{RawData: dataLine} + } observer.ObserveAnthropic([]byte(dataLine)) if eventName == "" { eventName = eventType } + eventKind := eventName + if eventType != "" && eventKind != eventType { + // A contradictory event name/payload is not a safe ping or prelude. + eventKind = "unknown" + } + if eventKind == "" { + eventKind = "unknown" + } eventChanged := false if useNoopDeltaKeepalive { @@ -991,7 +1052,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http block = "event: " + eventName + "\n" } block += "data: " + dataLine + "\n\n" - return []string{block}, dataLine, usagePatch, nil + return []string{block}, dataLine, usagePatch, eventKind, nil } newData, err := json.Marshal(event) @@ -1002,7 +1063,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http block = "event: " + eventName + "\n" } block += "data: " + dataLine + "\n\n" - return []string{block}, dataLine, usagePatch, nil + return []string{block}, dataLine, usagePatch, eventKind, nil } block := "" @@ -1010,20 +1071,29 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http block = "event: " + eventName + "\n" } block += "data: " + string(newData) + "\n\n" - return []string{block}, string(newData), usagePatch, nil + return []string{block}, string(newData), usagePatch, eventKind, nil } for { select { case ev, ok := <-events: if !ok { + hadPendingPrelude := len(pendingPrelude) > 0 + commitPrelude() // 上游完成，返回结果 if !sawTerminalEvent { + if hadPendingPrelude && !clientDisconnected { + // Committing the held start makes the response streaming; finish it + // explicitly rather than relying on a handler JSON fallback. + sendErrorEvent("upstream_error", "Upstream stream ended before completion") + MarkResponseCommitted(c) + } return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: clientDisconnected}, fmt.Errorf("stream usage incomplete: missing terminal event") } return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: clientDisconnected}, nil } if ev.err != nil { + commitPrelude() if sawTerminalEvent { return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: clientDisconnected}, nil } @@ -1074,39 +1144,58 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http continue } - outputBlocks, data, usagePatch, err := processSSEEvent(pendingEventLines) + outputBlocks, data, usagePatch, eventType, err := processSSEEvent(pendingEventLines) pendingEventLines = pendingEventLines[:0] if err != nil { if clientDisconnected { return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: true}, nil } + var sseErr *sseStreamErrorEventError + if errors.As(err, &sseErr) { + // Custom rectifier patterns share the same matcher as HTTP recovery. + requestRejected := ClassifyClaudeStreamError([]byte(data)).StatusCode == http.StatusBadRequest + signatureRejected := requestRejected && s.isThinkingBlockSignatureError([]byte(data)) + if requestRejected && !signatureRejected && bufferPrelude && s.settingService != nil { + signatureRejected = s.isSignatureErrorPattern(ctx, account, []byte(data)) + } + if bufferPrelude && signatureRejected && !messageStarted { + pendingPrelude = nil + pendingPreludeBytes = 0 + } else { + commitPrelude() + } + sseErr.MessageStarted = messageStarted + } + if usage.hasObservedTokens() { + return &streamingResult{usage: usage, firstTokenMs: firstTokenMs}, err + } return nil, err } - for _, block := range outputBlocks { - if !clientDisconnected { - restored := reverseToolNamesIfPresent(c, []byte(block)) - if _, werr := fmt.Fprint(w, string(restored)); werr != nil { - clientDisconnected = true - logger.LegacyPrintf("service.gateway", "Client disconnected during streaming, continuing to drain upstream for billing") - // 不 break：客户端断开后仍需继续合并本事件及后续事件的 usage， - // 否则会漏计当前事件携带的 usage 导致少计费。后续写入由 - // clientDisconnected 守卫跳过。 - } else { - flusher.Flush() - lastDataAt = time.Now() - resetKeepaliveTimer() - } + if data != "" { + if firstTokenMs == nil && data != "[DONE]" && (!(preludeRecoveryEnabled || suppressAccountHealthPenalty) || eventType != "ping") { + ms := int(time.Since(startTime).Milliseconds()) + firstTokenMs = &ms } - if data != "" { - if firstTokenMs == nil && data != "[DONE]" { - ms := int(time.Since(startTime).Milliseconds()) - firstTokenMs = &ms - } - if usagePatch != nil { - mergeSSEUsagePatch(usage, usagePatch) - } + if usagePatch != nil { + // Metering belongs to this attempt even while its prelude is held. + mergeSSEUsagePatch(usage, usagePatch) + } + } + if bufferPrelude && !messageStarted && len(pendingPrelude) == 0 && eventType == "message_start" { + pendingPrelude = append(pendingPrelude, outputBlocks...) + for _, block := range outputBlocks { + pendingPreludeBytes += len(block) + } + if pendingPreludeBytes > claudeMessageStartPreludeLimit { + commitPrelude() + bufferPrelude = false + } + } else { + if eventType != "ping" { + commitPrelude() } + writeBlocks(outputBlocks, eventType) } continue } @@ -1118,12 +1207,13 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http if time.Since(lastRead) < streamInterval { continue } + commitPrelude() if clientDisconnected { return &streamingResult{usage: usage, firstTokenMs: firstTokenMs, clientDisconnect: true}, fmt.Errorf("stream usage incomplete after timeout") } logger.LegacyPrintf("service.gateway", "Stream data interval timeout: account=%d model=%s interval=%s", account.ID, originalModel, streamInterval) // 处理流超时，可能标记账户为临时不可调度或错误状态 - if s.rateLimitService != nil { + if s.rateLimitService != nil && !suppressAccountHealthPenalty { s.rateLimitService.HandleStreamTimeout(ctx, account, originalModel) } sendErrorEvent("stream_timeout", fmt.Sprintf("upstream stream idle for %s", streamInterval)) diff --git a/backend/internal/service/ops_claude_signature_evidence_test.go b/backend/internal/service/ops_claude_signature_evidence_test.go new file mode 100644 index 000000000000..9698c18e5ce4 --- /dev/null +++ b/backend/internal/service/ops_claude_signature_evidence_test.go @@ -0,0 +1,48 @@ +package service + +import ( + "strings" + "testing" +) + +// Queuing drops older raw bodies, but must retain the facts explaining a +// signature recovery and must not confuse semantic status with actual HTTP. +func TestClaudeSignatureEvidenceSurvivesOpsQueue(t *testing.T) { + diagnostic := &ClaudeSignatureRecoveryDiagnostic{ + Source: "sse_error", Outcome: "upstream_rejected", Attempts: 1, + Inbound: ClaudeSignatureSnapshot{TotalCount: 1, Signatures: []ClaudeSignatureValueFingerprint{{ + Path: "/messages/0/content/0/signature", Length: 13, SHA256: strings.Repeat("a", 64), + }}}, + } + entry := &OpsInsertErrorLogInput{UpstreamErrors: []*OpsUpstreamErrorEvent{{ + Kind: "stream_error", UpstreamStatusCode: 400, UpstreamHTTPStatusCode: 200, + Message: "Invalid signature in thinking block", Detail: "old raw error", + SignatureRecovery: diagnostic, + }}} + for i := 0; i < opsUpstreamErrorsBodyWindow; i++ { + entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{ + Kind: "http_error", UpstreamStatusCode: 500, Message: "later failure", + }) + } + if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil { + t.Fatal(err) + } + diagnostic.Inbound.Signatures[0].SHA256 = "mutated after queueing" + events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON) + if err != nil { + t.Fatal(err) + } + first := events[0] + if first.UpstreamStatusCode != 400 || first.UpstreamHTTPStatusCode != 200 { + t.Fatalf("semantic/HTTP status changed: %d/%d", first.UpstreamStatusCode, first.UpstreamHTTPStatusCode) + } + if first.Detail != "" { + t.Fatal("older body must still be dropped by the existing queue window") + } + if first.SignatureRecovery == nil || first.SignatureRecovery.Inbound.Signatures[0].SHA256 != strings.Repeat("a", 64) { + t.Fatal("bounded recovery fingerprints must survive independent of raw bodies") + } + if last := events[len(events)-1]; last.UpstreamStatusCode != 500 || last.SignatureRecovery != nil { + t.Fatal("recovery bookkeeping must not replace the terminal upstream failure") + } +} diff --git a/backend/internal/service/ops_service.go b/backend/internal/service/ops_service.go index 9beeca51f95d..891736f739f3 100644 --- a/backend/internal/service/ops_service.go +++ b/backend/internal/service/ops_service.go @@ -593,6 +593,7 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { // Continuation diagnostics are self-bounded and stay on every retained // attempt; boundOpsUpstreamErrors still enforces the entry byte budget. out.ContinuationDiagnostic = sanitizeOpenAIContinuationDiagnostic(ev.ContinuationDiagnostic) + out.SignatureRecovery = sanitizeClaudeSignatureRecoveryDiagnostic(ev.SignatureRecovery) urlMaxLen, messageMaxLen := 2048, 2048 if !keepBody { urlMaxLen, messageMaxLen = opsUpstreamErrorsOlderURLMaxLen, opsUpstreamErrorsOlderMessageMaxLen @@ -619,6 +620,9 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { if out.UpstreamStatusCode < 0 { out.UpstreamStatusCode = 0 } + if out.UpstreamHTTPStatusCode < 100 || out.UpstreamHTTPStatusCode > 599 { + out.UpstreamHTTPStatusCode = 0 + } if out.AtUnixMs < 0 { out.AtUnixMs = 0 } diff --git a/backend/internal/service/ops_upstream_context.go b/backend/internal/service/ops_upstream_context.go index c920b7f08dda..7f9db2fd27e2 100644 --- a/backend/internal/service/ops_upstream_context.go +++ b/backend/internal/service/ops_upstream_context.go @@ -424,6 +424,9 @@ type OpsUpstreamErrorEvent struct { // Outcome UpstreamStatusCode int `json:"upstream_status_code,omitempty"` UpstreamRequestID string `json:"upstream_request_id,omitempty"` + // For Claude stream_error events, UpstreamStatusCode is the error's semantic + // status. Keep the actual HTTP response separately: SSE can fail after 200. + UpstreamHTTPStatusCode int `json:"upstream_http_status_code,omitempty"` // UpstreamURL is the actual upstream URL that was called (host + path, query/fragment stripped). // Helps debug 404/routing errors by showing which endpoint was targeted. @@ -448,6 +451,9 @@ type OpsUpstreamErrorEvent struct { // this for passthrough-rule matching or to decide whether a request can be // retried. ContinuationDiagnostic *OpenAIContinuationDiagnostic `json:"continuation_diagnostic,omitempty"` + // Signature recovery evidence is bounded and contains fingerprints only; + // bookkeeping must not replace the corresponding upstream failure. + SignatureRecovery *ClaudeSignatureRecoveryDiagnostic `json:"signature_recovery,omitempty"` // SkipMonitoring is request-local rule state. It is intentionally excluded // from persisted attempt JSON. The logger consults it only when this event is diff --git a/frontend/src/api/admin/ops.ts b/frontend/src/api/admin/ops.ts index 8754f1958370..1476026df7e1 100644 --- a/frontend/src/api/admin/ops.ts +++ b/frontend/src/api/admin/ops.ts @@ -19,6 +19,7 @@ export type OpsUpstreamErrorEvent = { account_id?: number account_name?: string upstream_status_code?: number + upstream_http_status_code?: number upstream_request_id?: string kind?: string message?: string diff --git a/frontend/src/i18n/locales/en/admin/ops.ts b/frontend/src/i18n/locales/en/admin/ops.ts index 8576736b9c8b..71a2ef999fc7 100644 --- a/frontend/src/i18n/locales/en/admin/ops.ts +++ b/frontend/src/i18n/locales/en/admin/ops.ts @@ -351,6 +351,8 @@ export default { phase: 'Phase', status: 'Status', upstreamStatus: 'Upstream Status', + upstreamStreamStatus: 'Upstream Stream Error', + upstreamHttpStatus: 'Upstream HTTP', message: 'Message', rootCause: 'Root Cause', diagnosticPayloads: 'Diagnostic Payloads', diff --git a/frontend/src/i18n/locales/zh/admin/ops.ts b/frontend/src/i18n/locales/zh/admin/ops.ts index ace6fff1e4f6..51f40c6bb37a 100644 --- a/frontend/src/i18n/locales/zh/admin/ops.ts +++ b/frontend/src/i18n/locales/zh/admin/ops.ts @@ -351,6 +351,8 @@ export default { phase: '阶段', status: '状态码', upstreamStatus: '上游状态码', + upstreamStreamStatus: '上游流内错误', + upstreamHttpStatus: '上游 HTTP', message: '消息', rootCause: '根因', diagnosticPayloads: '诊断载荷', diff --git a/frontend/src/views/admin/ops/components/OpsErrorDetailModal.vue b/frontend/src/views/admin/ops/components/OpsErrorDetailModal.vue index ca3d74efed6c..b433a4b498b7 100644 --- a/frontend/src/views/admin/ops/components/OpsErrorDetailModal.vue +++ b/frontend/src/views/admin/ops/components/OpsErrorDetailModal.vue @@ -94,10 +94,21 @@
-
{{ t('admin.ops.errorDetail.upstreamStatus') }}
+
+ {{ terminalStreamError ? t('admin.ops.errorDetail.upstreamStreamStatus') : t('admin.ops.errorDetail.upstreamStatus') }} +
- {{ detail.upstream_status_code ?? '—' }} + {{ upstreamStatusCode ?? '—' }} + +
+
+ +
+
{{ t('admin.ops.errorDetail.upstreamHttpStatus') }}
+
+ + {{ terminalStreamError.upstream_http_status_code }}
@@ -233,7 +244,7 @@ import BaseDialog from '@/components/common/BaseDialog.vue' import Icon from '@/components/icons/Icon.vue' import CodexContinuationDiagnostics from '@/components/admin/codex/CodexContinuationDiagnostics.vue' import { useAppStore } from '@/stores' -import { opsAPI, type OpsErrorDetail } from '@/api/admin/ops' +import { opsAPI, type OpsErrorDetail, type OpsUpstreamErrorEvent } from '@/api/admin/ops' import { formatDateTime } from '@/utils/format' import { resolveUpstreamPayload } from '../utils/errorDetailResponse' @@ -262,6 +273,27 @@ const showUpstreamList = computed(() => props.errorType === 'request') const requestId = computed(() => detail.value?.request_id || detail.value?.client_request_id || '') +const terminalStreamError = computed(() => { + const raw = detail.value?.upstream_errors + if (!raw) return null + try { + const events: unknown = JSON.parse(raw) + if (!Array.isArray(events) || !events.length) return null + const event = events[events.length - 1] + if (!event || typeof event !== 'object' || event.kind !== 'stream_error') return null + if (!isHttpStatusCode(event.upstream_http_status_code) || !isHttpStatusCode(event.upstream_status_code) || event.upstream_status_code < 400) return null + return event as OpsUpstreamErrorEvent + } catch { + return null + } +}) + +const upstreamStatusCode = computed(() => terminalStreamError.value?.upstream_status_code ?? detail.value?.upstream_status_code) + +function isHttpStatusCode(value: unknown): value is number { + return typeof value === 'number' && Number.isInteger(value) && value >= 100 && value <= 599 +} + type DiagnosticPayloadKey = 'client' | 'upstream_message' | 'upstream_detail' | 'upstream_events' const rootCauseMessage = computed(() => { @@ -435,6 +467,8 @@ function statusBadgeClass(code: number): string { const statusClass = computed(() => statusBadgeClass(detail.value?.status_code ?? 0)) -const upstreamStatusClass = computed(() => statusBadgeClass(detail.value?.upstream_status_code ?? 0)) +const upstreamStatusClass = computed(() => statusBadgeClass(upstreamStatusCode.value ?? 0)) + +const upstreamHttpStatusClass = computed(() => statusBadgeClass(terminalStreamError.value?.upstream_http_status_code ?? 0)) diff --git a/frontend/src/views/admin/ops/components/__tests__/OpsErrorDetailModal.spec.ts b/frontend/src/views/admin/ops/components/__tests__/OpsErrorDetailModal.spec.ts index f8bfe2272023..8403d47e3f4e 100644 --- a/frontend/src/views/admin/ops/components/__tests__/OpsErrorDetailModal.spec.ts +++ b/frontend/src/views/admin/ops/components/__tests__/OpsErrorDetailModal.spec.ts @@ -1,4 +1,4 @@ -import { flushPromises, shallowMount } from '@vue/test-utils' +import { flushPromises, shallowMount, type VueWrapper } from '@vue/test-utils' import { beforeEach, describe, expect, it, vi } from 'vitest' import OpsErrorDetailModal from '../OpsErrorDetailModal.vue' @@ -31,6 +31,36 @@ vi.mock('vue-i18n', async (importOriginal) => { } }) +async function mountErrorDetail(upstreamErrors: string, upstreamStatusCode = 403) { + mocks.getRequestErrorDetail.mockResolvedValue({ + id: 1, + created_at: '2026-10-08T08:00:00Z', + phase: 'request', + type: 'upstream_error', + error_owner: 'provider', + status_code: upstreamStatusCode, + upstream_status_code: upstreamStatusCode, + upstream_errors: upstreamErrors, + upstream_error_message: 'Original upstream error', + platform: 'anthropic' + }) + const wrapper = shallowMount(OpsErrorDetailModal, { + props: { show: true, errorId: 1, errorType: 'request' }, + global: { + stubs: { + BaseDialog: { template: '
' }, + Icon: true + } + } + }) + await flushPromises() + return wrapper +} + +function summaryCard(wrapper: VueWrapper, label: string) { + return wrapper.findAll('[data-ui="ops-kv-grid"] > div').find(card => card.find('div').text() === label) +} + describe('OpsErrorDetailModal', () => { beforeEach(() => { mocks.getRequestErrorDetail.mockReset() @@ -81,4 +111,54 @@ describe('OpsErrorDetailModal', () => { expect(wrapper.text()).not.toContain('admin.ops.errorDetail.payloads.upstream_detail') expect(wrapper.findComponent({ name: 'CodexContinuationDiagnostics' }).props('errorId')).toBe(1) }) + + it.each([400, 529])('separates stream error %i from the actual upstream HTTP response', async (semanticStatus) => { + const events = [{ + kind: 'stream_error', + upstream_status_code: semanticStatus, + upstream_http_status_code: 200, + message: 'Original stream error', + detail: '{"signature_comparison":{"omitted":0}}' + }] + const wrapper = await mountErrorDetail(JSON.stringify(events)) + + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamStreamStatus')?.text()).toContain(String(semanticStatus)) + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamHttpStatus')?.text()).toContain('200') + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamStatus')).toBeUndefined() + expect(wrapper.text()).toContain('Original upstream error') + expect(wrapper.findAll('pre').some(pre => pre.text() === JSON.stringify(events, null, 2))).toBe(true) + }) + + it('keeps the final HTTP error visible after an earlier stream error', async () => { + const wrapper = await mountErrorDetail(JSON.stringify([ + { kind: 'stream_error', upstream_status_code: 400, upstream_http_status_code: 200 }, + { kind: 'http_error', upstream_status_code: 403 } + ])) + + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamStatus')?.text()).toContain('403') + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamStreamStatus')).toBeUndefined() + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamHttpStatus')).toBeUndefined() + }) + + it.each([ + ['historical stream error', '[{"kind":"stream_error","upstream_status_code":403}]'], + ['ordinary HTTP error', '[{"kind":"http_error","upstream_status_code":403,"upstream_http_status_code":403}]'], + ['invalid JSON', '{invalid'], + ['JSON object', '{"kind":"stream_error","upstream_status_code":400,"upstream_http_status_code":200}'], + ['null final event', '[null]'], + ['zero HTTP code', '[{"kind":"stream_error","upstream_status_code":400,"upstream_http_status_code":0}]'], + ['negative HTTP code', '[{"kind":"stream_error","upstream_status_code":400,"upstream_http_status_code":-1}]'], + ['string HTTP code', '[{"kind":"stream_error","upstream_status_code":400,"upstream_http_status_code":"200"}]'], + ['out-of-range HTTP code', '[{"kind":"stream_error","upstream_status_code":400,"upstream_http_status_code":999}]'], + ['success semantic code', '[{"kind":"stream_error","upstream_status_code":200,"upstream_http_status_code":200}]'], + ['out-of-range semantic code', '[{"kind":"stream_error","upstream_status_code":999,"upstream_http_status_code":200}]'], + ['missing semantic code', '[{"kind":"stream_error","upstream_http_status_code":200}]'] + ])('preserves the existing status display for %s', async (_name, rawEvents) => { + const wrapper = await mountErrorDetail(rawEvents) + + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamStatus')?.text()).toContain('403') + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamStreamStatus')).toBeUndefined() + expect(summaryCard(wrapper, 'admin.ops.errorDetail.upstreamHttpStatus')).toBeUndefined() + expect(wrapper.text()).toContain('Original upstream error') + }) }) From e35f3ca80dbde268e3b15238f3466684a36fb89b Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 8 Oct 2026 17:43:21 +0800 Subject: [PATCH 2/2] fix: satisfy Claude recovery error and assertion checks --- .../service/gateway_claude_prelude_test.go | 6 ++--- .../gateway_claude_signature_recovery_test.go | 25 +++++++++++-------- .../service/gateway_upstream_response.go | 2 +- 3 files changed, 18 insertions(+), 15 deletions(-) diff --git a/backend/internal/service/gateway_claude_prelude_test.go b/backend/internal/service/gateway_claude_prelude_test.go index ad3be5dc0235..f9592a82c21f 100644 --- a/backend/internal/service/gateway_claude_prelude_test.go +++ b/backend/internal/service/gateway_claude_prelude_test.go @@ -46,7 +46,7 @@ func (w *claudePreludeFlushObserver) Flush() { func runClaudePreludeTestStream(t *testing.T, svc *GatewayService, c *gin.Context, stream string, recoverPrelude bool) (*streamingResult, error) { t.Helper() resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(stream))} - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() return svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Platform: PlatformAnthropic}, time.Now(), "claude-sonnet-5-5", "claude-sonnet-5-5", false, recoverPrelude) } @@ -188,8 +188,8 @@ func TestClaudePrelude_RepairTimeoutSkipsHealthPenaltyAndNormalTimeoutKeepsIt(t svc.rateLimitService.SetSettingService(settings) recorder, c := claudePreludeTestContext() pr, pw := io.Pipe() - defer pr.Close() - defer pw.Close() + defer func() { _ = pr.Close() }() + defer func() { _ = pw.Close() }() resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: pr} _, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false, false, suppressPenalty) diff --git a/backend/internal/service/gateway_claude_signature_recovery_test.go b/backend/internal/service/gateway_claude_signature_recovery_test.go index 8c6db7f1d923..e9d77ad0e624 100644 --- a/backend/internal/service/gateway_claude_signature_recovery_test.go +++ b/backend/internal/service/gateway_claude_signature_recovery_test.go @@ -23,6 +23,15 @@ import ( const claudeRecoveryTestBody = `{"model":"claude-sonnet-5-5","stream":true,"thinking":{"type":"between_tools"},"output_config":{"effort":"high"},"messages":[{"role":"assistant","content":[{"type":"thinking","thinking":"old reasoning","signature":"private-signature-value"},{"type":"text","text":"old answer"}]},{"role":"user","content":"continue"}]}` const claudeRecoverySignatureError = `{"type":"error","error":{"message":"Invalid signature in thinking block"}}` +func claudeRecoveryOpsEvents(t *testing.T, c *gin.Context) []*OpsUpstreamErrorEvent { + t.Helper() + value, found := c.Get(OpsUpstreamErrorsKey) + require.True(t, found) + events, valid := value.([]*OpsUpstreamErrorEvent) + require.True(t, valid) + return events +} + type claudeRecoverySequenceUpstream struct { responses []*http.Response requests []*http.Request @@ -140,9 +149,7 @@ func TestClaudeSignatureRecoveryForwardHTTPAndSSE(t *testing.T) { require.Equal(t, "between_tools", gjson.GetBytes(upstream.bodies[1], "thinking.type").String()) require.Equal(t, "high", gjson.GetBytes(upstream.bodies[1], "output_config.effort").String()) require.True(t, bytes.Equal(parsed.Body.Bytes(), upstream.bodies[1])) - events, ok := c.Get(OpsUpstreamErrorsKey) - require.True(t, ok) - trigger := events.([]*OpsUpstreamErrorEvent)[0] + trigger := claudeRecoveryOpsEvents(t, c)[0] require.NotNil(t, trigger.SignatureRecovery) require.Equal(t, "accepted", trigger.SignatureRecovery.Outcome) require.Equal(t, 1, trigger.SignatureRecovery.Attempts) @@ -180,8 +187,7 @@ func TestClaudeSignatureRecoveryDoesNotReplaySSEAgainOrAfterOutput(t *testing.T) wantRequests = 1 } require.Len(t, upstream.requests, wantRequests) - events, _ := c.Get(OpsUpstreamErrorsKey) - last := events.([]*OpsUpstreamErrorEvent)[wantRequests-1] + last := claudeRecoveryOpsEvents(t, c)[wantRequests-1] wantReason := "attempt_limit" if prefix != "" { wantReason = "output_started" @@ -314,8 +320,7 @@ func TestClaudeSignatureRecoverySharedHTTPAndSSELimit(t *testing.T) { _, err = claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) require.Error(t, err) require.Len(t, upstream.requests, 2) - events, _ := c.Get(OpsUpstreamErrorsKey) - attempts := events.([]*OpsUpstreamErrorEvent) + attempts := claudeRecoveryOpsEvents(t, c) require.Equal(t, "attempt_limit", attempts[len(attempts)-1].SignatureRecovery.Reason) } @@ -422,8 +427,7 @@ func TestClaudeSignatureRecoveryIncompleteRepairedStreamRecordsActualAttempt(t * _, err = claudeRecoveryService(upstream, true).Forward(context.Background(), c, account, parsed) require.Error(t, err) require.Contains(t, err.Error(), "missing terminal") - events, _ := c.Get(OpsUpstreamErrorsKey) - attempts := events.([]*OpsUpstreamErrorEvent) + attempts := claudeRecoveryOpsEvents(t, c) require.Equal(t, "response_read_failed", attempts[0].SignatureRecovery.Outcome) require.Equal(t, "stream_incomplete", attempts[0].SignatureRecovery.Reason) last := attempts[len(attempts)-1] @@ -458,8 +462,7 @@ func TestClaudeSignatureRecoveryHTTPToolOnlyKeepsExistingSecondStage(t *testing. require.Equal(t, gjson.GetBytes(upstream.bodies[0], "messages").Raw, gjson.GetBytes(upstream.bodies[1], "messages").Raw) require.NotContains(t, string(upstream.bodies[2]), `"type":"tool_use"`) require.NotContains(t, string(upstream.bodies[2]), `"type":"tool_result"`) - events, _ := c.Get(OpsUpstreamErrorsKey) - first := events.([]*OpsUpstreamErrorEvent)[0] + first := claudeRecoveryOpsEvents(t, c)[0] require.Equal(t, 2, first.SignatureRecovery.Attempts) require.Equal(t, "accepted", first.SignatureRecovery.Outcome) } diff --git a/backend/internal/service/gateway_upstream_response.go b/backend/internal/service/gateway_upstream_response.go index 32317756cf2c..575341264c10 100644 --- a/backend/internal/service/gateway_upstream_response.go +++ b/backend/internal/service/gateway_upstream_response.go @@ -1173,7 +1173,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http } if data != "" { - if firstTokenMs == nil && data != "[DONE]" && (!(preludeRecoveryEnabled || suppressAccountHealthPenalty) || eventType != "ping") { + if firstTokenMs == nil && data != "[DONE]" && ((!preludeRecoveryEnabled && !suppressAccountHealthPenalty) || eventType != "ping") { ms := int(time.Since(startTime).Milliseconds()) firstTokenMs = &ms }