From 7e9af4c100fe3f8d97a89925828516f8f8e7d8e6 Mon Sep 17 00:00:00 2001 From: IanShaw027 Date: Fri, 21 Aug 2026 08:55:04 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=86=20compact=20fallback=20=E7=9A=84?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E9=87=8D=E8=AF=95=E6=94=B9=E4=B8=BA=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF=EF=BC=8C=E9=81=BF=E5=85=8D=E9=80=92=E5=BD=92=E9=87=8D?= =?UTF-8?q?=E7=BD=AE=E5=8D=95=E6=AC=A1=E9=87=8D=E8=AF=95=E6=A0=87=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/openai_compact_fallback.go | 37 ++++ .../service/openai_compact_fallback_test.go | 38 ++++ .../service/openai_gateway_passthrough.go | 205 +++++++++--------- 3 files changed, 177 insertions(+), 103 deletions(-) diff --git a/backend/internal/service/openai_compact_fallback.go b/backend/internal/service/openai_compact_fallback.go index 79595adbb6..517a7e2e7b 100644 --- a/backend/internal/service/openai_compact_fallback.go +++ b/backend/internal/service/openai_compact_fallback.go @@ -5,6 +5,7 @@ import ( "net/http" "strings" + "github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/gin-gonic/gin" "github.com/tidwall/gjson" ) @@ -143,3 +144,39 @@ func (s *OpenAIGatewayService) prepareOpenAICompactFallbackRetry( } return retryBody, fallbackModel, true } + +func (s *OpenAIGatewayService) applyOpenAIPassthroughCompactFallbackFromSignal( + c *gin.Context, + account *Account, + requestedModel string, + body []byte, + err error, + alreadyRetried bool, + resp *http.Response, +) ([]byte, string, bool) { + signal, ok := asOpenAICompactFallbackSignal(err) + if !ok { + return body, "", false + } + retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry( + c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, alreadyRetried, + ) + if !retry { + return body, "", false + } + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + fromModel := strings.TrimSpace(gjson.GetBytes(body, "model").String()) + accountName := "" + if account != nil { + accountName = account.Name + } + SetOpsUpstreamModel(c, fallbackModel) + logger.LegacyPrintf( + "service.openai_gateway", + "[OpenAI passthrough] Retrying explicit compact request once with fallback model (account: %s, from: %s, to: %s, upstream_code: %s)", + accountName, fromModel, fallbackModel, extractUpstreamErrorCode(signal.payload), + ) + return retryBody, fallbackModel, true +} diff --git a/backend/internal/service/openai_compact_fallback_test.go b/backend/internal/service/openai_compact_fallback_test.go index 377dea2212..944f7e1cfd 100644 --- a/backend/internal/service/openai_compact_fallback_test.go +++ b/backend/internal/service/openai_compact_fallback_test.go @@ -272,3 +272,41 @@ func TestOpenAIGatewayForwardRetriesStreamingCompactFailureBeforeOutput(t *testi require.NotContains(t, recorder.Body.String(), "context_length_exceeded") require.Contains(t, recorder.Body.String(), "response.completed") } + +func TestOpenAIGatewayForwardDoesNotRecurseWhenCompactFallbackAlsoFails(t *testing.T) { + gin.SetMode(gin.TestMode) + body := []byte(`{"model":"gpt-5.5","stream":true,"instructions":"compact-test","input":[{"type":"message","role":"user","content":"hello"},{"type":"compaction_trigger"}]}`) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + c.Request.Body = io.NopCloser(bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + MarkOpenAINativeCompactionV2(c) + + failed := "event: response.failed\n" + + `data: {"type":"response.failed","response":{"status":"failed","error":{"code":"model_not_found","message":"model not found"}}}` + "\n\n" + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(failed))}, + }} + svc := &OpenAIGatewayService{ + cfg: &config.Config{Gateway: config.GatewayConfig{OpenAICompactModel: "gpt-5.4"}}, + httpUpstream: upstream, + } + account := &Account{ + ID: 1, Name: "openai-oauth", Platform: PlatformOpenAI, Type: AccountTypeOAuth, Concurrency: 1, + Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-account"}, + Status: StatusActive, Schedulable: true, + } + + result, err := svc.Forward(context.Background(), c, account, body) + + require.Error(t, err) + require.Nil(t, result) + require.Len(t, upstream.bodies, 2) + require.Equal(t, "gpt-5.5", gjson.GetBytes(upstream.bodies[0], "model").String()) + require.Equal(t, "gpt-5.4", gjson.GetBytes(upstream.bodies[1], "model").String()) +} diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 40931138d4..06653094ce 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -326,6 +326,11 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( compactModelFallbackRetried := false rejectedFieldRetryState := openAIResponsesRejectedFieldRetryStateForRequest(c, body) var resp *http.Response + var usage *OpenAIUsage + var firstTokenMs *int + responseID := "" + imageCount := 0 + var imageOutputSizes []string for { actualModel := strings.TrimSpace(gjson.GetBytes(body, "model").String()) if actualModel == "" { @@ -347,116 +352,110 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( // a failover so the handler switches to a healthy account. return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, true) } - if resp.StatusCode < 400 { - break - } - - // Peek only to identify an invalid task. Restore the body so the existing - // passthrough error handling sees the same response after recovery fails. - probeBody := s.readUpstreamErrorBody(resp) - _ = resp.Body.Close() - resp.Body = io.NopCloser(bytes.NewReader(probeBody)) - if retryBody, reason, changed, retryErr := normalizeOpenAIResponsesRejectedFieldRetryBody(resp.StatusCode, body, probeBody); retryErr != nil { - return nil, fmt.Errorf("normalize passthrough rejected Responses field retry body: %w", retryErr) - } else if changed && rejectedFieldRetryState.Allow(retryBody) { - body = retryBody - logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying passthrough request after %s (account: %s)", reason, account.Name) - continue - } - if !agentTaskRecoveryTried && s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidHTTPResponse(resp.StatusCode, probeBody) { - agentTaskRecoveryTried = true - expectedTaskID := account.GetCredential("task_id") - if recoveryErr := s.recoverAgentIdentityTask(ctx, account, expectedTaskID); recoveryErr != nil { - return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr) + if resp.StatusCode >= 400 { + // Peek only to identify an invalid task. Restore the body so the existing + // passthrough error handling sees the same response after recovery fails. + probeBody := s.readUpstreamErrorBody(resp) + _ = resp.Body.Close() + resp.Body = io.NopCloser(bytes.NewReader(probeBody)) + if retryBody, reason, changed, retryErr := normalizeOpenAIResponsesRejectedFieldRetryBody(resp.StatusCode, body, probeBody); retryErr != nil { + return nil, fmt.Errorf("normalize passthrough rejected Responses field retry body: %w", retryErr) + } else if changed && rejectedFieldRetryState.Allow(retryBody) { + body = retryBody + logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying passthrough request after %s (account: %s)", reason, account.Name) + continue } - continue - } - upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(probeBody))) - if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry( - c, account, requestedModel, body, resp.StatusCode, upstreamMsg, probeBody, compactModelFallbackRetried, - ); retry { - fromModel := strings.TrimSpace(gjson.GetBytes(body, "model").String()) - body = retryBody - upstreamPassthroughModel = fallbackModel - compactModelFallbackRetried = true - SetOpsUpstreamModel(c, fallbackModel) - logger.LegacyPrintf( - "service.openai_gateway", - "[OpenAI passthrough] Retrying explicit compact request once with fallback model (account: %s, from: %s, to: %s, upstream_code: %s)", - account.Name, fromModel, fallbackModel, extractUpstreamErrorCode(probeBody), - ) - continue + if !agentTaskRecoveryTried && s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidHTTPResponse(resp.StatusCode, probeBody) { + agentTaskRecoveryTried = true + expectedTaskID := account.GetCredential("task_id") + if recoveryErr := s.recoverAgentIdentityTask(ctx, account, expectedTaskID); recoveryErr != nil { + return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr) + } + continue + } + upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(probeBody))) + if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry( + c, account, requestedModel, body, resp.StatusCode, upstreamMsg, probeBody, compactModelFallbackRetried, + ); retry { + fromModel := strings.TrimSpace(gjson.GetBytes(body, "model").String()) + body = retryBody + upstreamPassthroughModel = fallbackModel + compactModelFallbackRetried = true + SetOpsUpstreamModel(c, fallbackModel) + logger.LegacyPrintf( + "service.openai_gateway", + "[OpenAI passthrough] Retrying explicit compact request once with fallback model (account: %s, from: %s, to: %s, upstream_code: %s)", + account.Name, fromModel, fallbackModel, extractUpstreamErrorCode(probeBody), + ) + continue + } + + // 透传模式默认保持原样代理；容量错误以及 API-key 上游的瞬时 + // 5xx 应先触发多账号 failover，且此时尚未写入下游响应。 + // probeBody 已在上方任务探测时读取过一次，直接复用避免重复读取。 + if shouldFailoverOpenAIPassthroughResponse(account, resp.StatusCode, probeBody) { + return nil, s.handleFailoverErrorResponsePassthrough(ctx, resp, c, account, body, probeBody) + } + return nil, s.handleErrorResponsePassthrough(ctx, resp, c, account, body, probeBody) } - // 透传模式默认保持原样代理；容量错误以及 API-key 上游的瞬时 - // 5xx 应先触发多账号 failover，且此时尚未写入下游响应。 - // probeBody 已在上方任务探测时读取过一次，直接复用避免重复读取。 - if shouldFailoverOpenAIPassthroughResponse(account, resp.StatusCode, probeBody) { - return nil, s.handleFailoverErrorResponsePassthrough(ctx, resp, c, account, body, probeBody) + if mapping, ok := openAIResponsesClientToolMapping(c); ok && isEventStreamResponse(resp.Header) { + maxLineSize := defaultMaxLineSize + if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 { + maxLineSize = s.cfg.Gateway.MaxLineSize + } + resp.Body = newGrokResponsesClientToolStreamBody(resp.Body, mapping, maxLineSize) } - return nil, s.handleErrorResponsePassthrough(ctx, resp, c, account, body, probeBody) + + // x-codex-turn-state 溯源：下游回传由 writeOpenAIPassthroughResponseHeaders + // 在各 handler 的写头点强制放行，铸造账号在此统一记录，供出站守卫剥离 + // failover 换号后的跨账号回带（openai_codex_turn_state.go）。 + if extractOpenAICodexTurnState(resp.Header) != "" { + s.noteOpenAICodexTurnStateProvenance(c, account) + } + + if reqStream { + result, handleErr := s.handleStreamingResponsePassthrough(ctx, resp, c, account, startTime, reqModel, upstreamPassthroughModel) + if handleErr != nil { + if retryBody, fallbackModel, retry := s.applyOpenAIPassthroughCompactFallbackFromSignal( + c, account, requestedModel, body, handleErr, compactModelFallbackRetried, resp, + ); retry { + body = retryBody + upstreamPassthroughModel = fallbackModel + compactModelFallbackRetried = true + continue + } + _ = resp.Body.Close() + return nil, handleErr + } + usage = result.usage + firstTokenMs = result.firstTokenMs + responseID = strings.TrimSpace(result.responseID) + imageCount = result.imageCount + imageOutputSizes = result.imageOutputSizes + } else { + result, handleErr := s.handleNonStreamingResponsePassthrough(ctx, resp, c, account, reqModel, upstreamPassthroughModel) + if handleErr != nil { + if retryBody, fallbackModel, retry := s.applyOpenAIPassthroughCompactFallbackFromSignal( + c, account, requestedModel, body, handleErr, compactModelFallbackRetried, resp, + ); retry { + body = retryBody + upstreamPassthroughModel = fallbackModel + compactModelFallbackRetried = true + continue + } + _ = resp.Body.Close() + return nil, handleErr + } + usage = result.usage + responseID = strings.TrimSpace(result.responseID) + imageCount = result.imageCount + imageOutputSizes = result.imageOutputSizes + } + break } defer func() { _ = resp.Body.Close() }() - if mapping, ok := openAIResponsesClientToolMapping(c); ok && isEventStreamResponse(resp.Header) { - maxLineSize := defaultMaxLineSize - if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 { - maxLineSize = s.cfg.Gateway.MaxLineSize - } - resp.Body = newGrokResponsesClientToolStreamBody(resp.Body, mapping, maxLineSize) - } - serviceTier := extractOpenAIServiceTierFromBody(body) - - // x-codex-turn-state 溯源：下游回传由 writeOpenAIPassthroughResponseHeaders - // 在各 handler 的写头点强制放行，铸造账号在此统一记录，供出站守卫剥离 - // failover 换号后的跨账号回带（openai_codex_turn_state.go）。 - if extractOpenAICodexTurnState(resp.Header) != "" { - s.noteOpenAICodexTurnStateProvenance(c, account) - } - - var usage *OpenAIUsage - var firstTokenMs *int - responseID := "" - imageCount := 0 - var imageOutputSizes []string - if reqStream { - result, err := s.handleStreamingResponsePassthrough(ctx, resp, c, account, startTime, reqModel, upstreamPassthroughModel) - if err != nil { - if signal, ok := asOpenAICompactFallbackSignal(err); ok { - if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry( - c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, compactModelFallbackRetried, - ); retry { - body = retryBody - SetOpsUpstreamModel(c, fallbackModel) - return s.forwardOpenAIPassthrough(ctx, c, account, body, canonicalImageIntentBody, requestedModel, true, reasoningEffort, reqStream, startTime) - } - } - return nil, err - } - usage = result.usage - firstTokenMs = result.firstTokenMs - responseID = strings.TrimSpace(result.responseID) - imageCount = result.imageCount - imageOutputSizes = result.imageOutputSizes - } else { - result, err := s.handleNonStreamingResponsePassthrough(ctx, resp, c, account, reqModel, upstreamPassthroughModel) - if err != nil { - if signal, ok := asOpenAICompactFallbackSignal(err); ok { - if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry( - c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, compactModelFallbackRetried, - ); retry { - body = retryBody - SetOpsUpstreamModel(c, fallbackModel) - return s.forwardOpenAIPassthrough(ctx, c, account, body, canonicalImageIntentBody, requestedModel, true, reasoningEffort, reqStream, startTime) - } - } - return nil, err - } - usage = result.usage - responseID = strings.TrimSpace(result.responseID) - imageCount = result.imageCount - imageOutputSizes = result.imageOutputSizes - } s.bindHTTPResponseAccount(ctx, c, account, responseID) // 排除 spark 影子:其 codex_* 仅由 QueryUsage(/wham/usage bengalfox)更新(外审第7轮 P1)。