fix: classify Claude stream failures and safely recover thinking signatures

.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 \

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

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

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)

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
 	}