diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index 8853bffbe6..bf52f4adc0 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -769,8 +769,8 @@ func (h *GatewayHandler) Messages(c *gin.Context) { return } } - // Bedrock CC 兼容:渠道模型映射后,清理 Anthropic API 专有字段、注入 Bedrock 必需字段 - if err := attemptParsedReq.ReplaceBody(h.gatewayService.ApplyBedrockCCCompat(c.Request.Context(), attemptParsedReq.Body.Bytes(), attemptParsedReq.Model, account, apiKey.GroupID)); err != nil { + // Bedrock CC 兼容:清理 body 专有字段 + 过滤 anthropic-beta header,适用于所有转发路径 + if err := attemptParsedReq.ReplaceBody(h.gatewayService.ApplyBedrockCCCompat(c, attemptParsedReq.Body.Bytes(), attemptParsedReq.Model, account, apiKey.GroupID)); err != nil { h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "Failed to parse request body") return } @@ -1605,6 +1605,9 @@ func (h *GatewayHandler) ensureForwardErrorResponse(c *gin.Context, streamStarte if c == nil || c.Writer == nil { return false } + if service.IsResponseCommitted(c) { + return false + } if c.Writer.Written() { streamStarted = true } diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 6730529743..80e00b7fc4 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -1847,11 +1847,9 @@ func (h *OpenAIGatewayHandler) ensureForwardErrorResponse(c *gin.Context, stream if c == nil || c.Writer == nil { return false } - // 旧实现在 Writer.Written 时直接 return false,导致 ping 已 flush 之后的 - // 上游错误(http2 timeout、连接中断等)完全无法把错误传给客户端—— - // HTTP 200 已锁死,TCP 直接 EOF,Codex CLI 报 "stream closed before response.completed"。 - // 这里改成:Writer 已写过时强制走 streamStarted 分支,让 - // handleStreamingAwareError 通过 SSE 发协议合规的 response.failed。 + if service.IsResponseCommitted(c) { + return false + } if c.Writer.Written() { streamStarted = true } diff --git a/backend/internal/service/antigravity_gateway_service.go b/backend/internal/service/antigravity_gateway_service.go index f79d20a2ca..7bdda2e507 100644 --- a/backend/internal/service/antigravity_gateway_service.go +++ b/backend/internal/service/antigravity_gateway_service.go @@ -2447,6 +2447,7 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co Detail: upstreamDetail, }) logger.LegacyPrintf("service.antigravity_gateway", "[antigravity-Forward] upstream error status=%d body=%s", resp.StatusCode, truncateForLog(unwrappedForOps, 500)) + MarkResponseCommitted(c) c.Data(resp.StatusCode, contentType, unwrappedForOps) return nil, fmt.Errorf("antigravity upstream error: %d", resp.StatusCode) } @@ -3644,6 +3645,7 @@ func mergeTextPartsToResponse(response map[string]any, textParts []string) map[s } func (s *AntigravityGatewayService) writeClaudeError(c *gin.Context, status int, errType, message string) error { + MarkResponseCommitted(c) c.JSON(status, gin.H{ "type": "error", "error": gin.H{"type": errType, "message": message}, @@ -3657,6 +3659,7 @@ func (s *AntigravityGatewayService) WriteMappedClaudeError(c *gin.Context, accou } func (s *AntigravityGatewayService) writeMappedClaudeError(c *gin.Context, account *Account, upstreamStatus int, upstreamRequestID string, body []byte) error { + MarkResponseCommitted(c) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) logBody, maxBytes := s.getLogConfig() @@ -3734,6 +3737,7 @@ func (s *AntigravityGatewayService) writeMappedClaudeError(c *gin.Context, accou } func (s *AntigravityGatewayService) writeGoogleError(c *gin.Context, status int, message string) error { + MarkResponseCommitted(c) statusStr := "UNKNOWN" switch status { case 400: diff --git a/backend/internal/service/bedrock_request.go b/backend/internal/service/bedrock_request.go index 69830a6768..595f687ac1 100644 --- a/backend/internal/service/bedrock_request.go +++ b/backend/internal/service/bedrock_request.go @@ -215,8 +215,14 @@ func PrepareBedrockRequestBodyWithTokens(body []byte, modelID string, betaTokens return nil, fmt.Errorf("inject anthropic_beta: %w", err) } logger.LegacyPrintf("service.gateway", "[Bedrock] Injected beta tokens: %v (model=%s ccCompat=%v)", betaTokens, modelID, ccCompat) + } else { + body, _ = sjson.DeleteBytes(body, "anthropic_beta") } + // 移除 Bedrock 不支持的 Anthropic 直连 API 专有顶层字段 + body, _ = sjson.DeleteBytes(body, "provider") + body, _ = sjson.DeleteBytes(body, "metadata") + // 移除 model 字段(Bedrock 通过 URL 指定模型) body, err = sjson.DeleteBytes(body, "model") if err != nil { @@ -468,11 +474,12 @@ func parseAnthropicBetaHeader(header string) []string { // 参考: AWS Bedrock 官方文档 + litellm anthropic_beta_headers_config.json // 更新策略: 当 AWS Bedrock 新增支持的 beta token 时需同步更新此白名单 var bedrockSupportedBetaTokens = map[string]bool{ - "computer-use-2025-01-24": true, - "computer-use-2025-11-24": true, - "context-1m-2025-08-07": true, - // "context-management-2025-06-27": false, // 无官方文档支持 - "compact-2026-01-12": true, // 官方支持,仅 InvokeModel API(Opus 4.6+) + "computer-use-2025-01-24": true, + "computer-use-2025-11-24": true, + "context-1m-2025-08-07": true, + "context-management-2025-06-27": true, // compaction + clear_thinking,AWS 文档已支持 + "compact-2026-01-12": true, // 官方支持,仅 InvokeModel API(Opus 4.6+) + "fine-grained-tool-streaming-2025-05-14": true, // AWS Tool Use 文档已支持 // "interleaved-thinking-2025-05-14": false, // 无官方文档支持 "tool-search-tool-2025-10-19": true, "tool-examples-2025-10-29": true, diff --git a/backend/internal/service/bedrock_request_test.go b/backend/internal/service/bedrock_request_test.go index f209ef0749..d580f8c1ba 100644 --- a/backend/internal/service/bedrock_request_test.go +++ b/backend/internal/service/bedrock_request_test.go @@ -412,7 +412,7 @@ func TestPrepareBedrockRequestBodyWithTokens_ContextManagementRequiresSupportedB assert.Equal(t, int64(100), gjson.GetBytes(result, "max_tokens").Int()) }) - t.Run("filters explicit unsupported context-management beta and strips field", func(t *testing.T) { + t.Run("keeps supported context-management beta and retains field", func(t *testing.T) { input := `{ "messages":[{"role":"user","content":"hi"}], "max_tokens":100, @@ -427,8 +427,8 @@ func TestPrepareBedrockRequestBodyWithTokens_ContextManagementRequiresSupportedB ) require.NoError(t, err) - assert.False(t, gjson.GetBytes(result, "context_management").Exists()) - assert.Equal(t, []string{"context-1m-2025-08-07"}, bedrockAnthropicBetaNames(result)) + assert.True(t, gjson.GetBytes(result, "context_management").Exists()) + assert.Equal(t, []string{bedrockContextManagementBetaToken, "context-1m-2025-08-07"}, bedrockAnthropicBetaNames(result)) }) } diff --git a/backend/internal/service/error_passthrough_runtime_test.go b/backend/internal/service/error_passthrough_runtime_test.go index 7032d15b95..73a4bfab19 100644 --- a/backend/internal/service/error_passthrough_runtime_test.go +++ b/backend/internal/service/error_passthrough_runtime_test.go @@ -251,6 +251,90 @@ func TestApplyErrorPassthroughRule_NoSkipMonitoringDoesNotSetContextKey(t *testi assert.False(t, exists, "OpsSkipPassthroughKey should NOT be set when skip_monitoring=false") } +// ---- ResponseCommittedKey: service 层写完错误响应后标记,handler 层检查跳过兜底写入 ---- + +func TestHandleErrorResponse_SetsResponseCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + svc := &GatewayService{} + resp := &http.Response{ + StatusCode: http.StatusBadRequest, + Body: io.NopCloser(bytes.NewReader([]byte(`{"error":{"message":"temperature: range: 0..1"}}`))), + Header: http.Header{}, + } + account := &Account{ID: 100, Platform: PlatformAnthropic, Type: AccountTypeAPIKey} + + _, err := svc.handleErrorResponse(context.Background(), resp, c, account) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "non-failover error path must mark response committed") + var payload map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &payload)) +} + +func TestHandleErrorResponse_PassthroughRuleSetsCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + ruleSvc := &ErrorPassthroughService{} + ruleSvc.setLocalCache([]*model.ErrorPassthroughRule{ + newNonFailoverPassthroughRule(http.StatusBadRequest, "temperature", http.StatusBadRequest, "参数错误"), + }) + BindErrorPassthroughService(c, ruleSvc) + + svc := &GatewayService{} + resp := &http.Response{ + StatusCode: http.StatusBadRequest, + Body: io.NopCloser(bytes.NewReader([]byte(`{"error":{"message":"temperature: range: 0..1"}}`))), + Header: http.Header{}, + } + account := &Account{ID: 200, Platform: PlatformAnthropic, Type: AccountTypeAPIKey} + + _, err := svc.handleErrorResponse(context.Background(), resp, c, account) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "passthrough rule path must mark response committed") + assert.Equal(t, http.StatusBadRequest, rec.Code) + var payload map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &payload)) + errField, ok := payload["error"].(map[string]any) + require.True(t, ok, "payload[\"error\"] should be map[string]any") + assert.Equal(t, "参数错误", errField["message"]) +} + +func TestOpenAIHandleErrorResponse_SetsResponseCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + svc := &OpenAIGatewayService{} + resp := &http.Response{ + StatusCode: http.StatusTooManyRequests, + Body: io.NopCloser(bytes.NewReader([]byte(`{"error":{"message":"rate limit exceeded"}}`))), + Header: http.Header{}, + } + account := &Account{ID: 101, Platform: PlatformOpenAI, Type: AccountTypeAPIKey} + + _, err := svc.handleErrorResponse(context.Background(), resp, c, account, nil) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "OpenAI non-failover path must mark response committed") +} + +func TestGeminiWriteGeminiMappedError_SetsResponseCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + svc := &GeminiMessagesCompatService{} + body := []byte(`{"error":{"message":"invalid field"}}`) + account := &Account{ID: 102, Platform: PlatformGemini, Type: AccountTypeAPIKey} + + err := svc.writeGeminiMappedError(c, account, http.StatusBadRequest, "req-99", body) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "Gemini path must mark response committed") +} + func newNonFailoverPassthroughRule(statusCode int, keyword string, respCode int, customMessage string) *model.ErrorPassthroughRule { return &model.ErrorPassthroughRule{ ID: 1, diff --git a/backend/internal/service/gateway_forward_as_chat_completions.go b/backend/internal/service/gateway_forward_as_chat_completions.go index 16c10a3298..67324d63ba 100644 --- a/backend/internal/service/gateway_forward_as_chat_completions.go +++ b/backend/internal/service/gateway_forward_as_chat_completions.go @@ -498,6 +498,7 @@ func (s *GatewayService) handleCCStreamingFromAnthropic( // writeGatewayCCError writes an error in OpenAI Chat Completions format for // the Anthropic-upstream CC forwarding path. func writeGatewayCCError(c *gin.Context, statusCode int, errType, message string) { + MarkResponseCommitted(c) c.JSON(statusCode, gin.H{ "error": gin.H{ "type": errType, diff --git a/backend/internal/service/gateway_forward_as_responses.go b/backend/internal/service/gateway_forward_as_responses.go index f883ae870f..83adce6cad 100644 --- a/backend/internal/service/gateway_forward_as_responses.go +++ b/backend/internal/service/gateway_forward_as_responses.go @@ -520,6 +520,7 @@ func appendRawJSON(existing json.RawMessage, fragment string) json.RawMessage { // writeResponsesError writes an error response in OpenAI Responses API format. func writeResponsesError(c *gin.Context, statusCode int, code, message string) { + MarkResponseCommitted(c) c.JSON(statusCode, gin.H{ "error": gin.H{ "code": code, diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 531c93e4e7..eb2d547fde 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -5860,15 +5860,24 @@ func writeAnthropicPassthroughResponseHeaders(dst http.Header, src http.Header, } // ApplyBedrockCCCompat 应用 Bedrock CC 兼容转换(渠道级模型映射后调用) -// 清理 Anthropic API 专有字段、注入 Bedrock 必需字段、修复 thinking/tool_use ID -func (s *GatewayService) ApplyBedrockCCCompat(ctx context.Context, body []byte, model string, account *Account, groupID *int64) []byte { - if !s.isBedrockCCCompatEnabled(ctx, account, groupID) { +// 清理 body 中 Anthropic API 专有字段、修复 thinking/tool_use ID、过滤 beta token, +// 同时过滤 HTTP header 中的 anthropic-beta(防止 Passthrough 路径透传不支持的 token)。 +func (s *GatewayService) ApplyBedrockCCCompat(c *gin.Context, body []byte, model string, account *Account, groupID *int64) []byte { + if !s.isBedrockCCCompatEnabled(c.Request.Context(), account, groupID) { return body } body = sanitizeBedrockCCFields(body) body = sanitizeBedrockThinking(body, model) body = sanitizeBedrockToolUseIDs(body) body = sanitizeBedrockCCBetaTokens(body, model) + // 过滤 HTTP header 中的 anthropic-beta,只保留 Bedrock 支持的 token + if betaHeader := c.GetHeader("anthropic-beta"); betaHeader != "" { + if filtered := ResolveBedrockBetaTokens(betaHeader, body, model); len(filtered) > 0 { + c.Request.Header.Set("anthropic-beta", strings.Join(filtered, ", ")) + } else { + c.Request.Header.Del("anthropic-beta") + } + } return body } @@ -7353,6 +7362,8 @@ func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Res return nil, &UpstreamFailoverError{StatusCode: resp.StatusCode, ResponseBody: body} } + MarkResponseCommitted(c) + // 记录上游错误响应体摘要便于排障(可选:由配置控制;不回显到客户端) if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { logger.LegacyPrintf("service.gateway", @@ -7476,6 +7487,7 @@ func (s *GatewayService) handleFailoverSideEffects(ctx context.Context, resp *ht // OAuth 403:标记账号异常 // API Key 未配置错误码:仅返回错误,不标记账号 func (s *GatewayService) handleRetryExhaustedError(ctx context.Context, resp *http.Response, c *gin.Context, account *Account) (*ForwardResult, error) { + MarkResponseCommitted(c) // Capture upstream error body before side-effects consume the stream. respBody, _ := s.readUpstreamErrorBody(resp) _ = resp.Body.Close() diff --git a/backend/internal/service/gemini_messages_compat_service.go b/backend/internal/service/gemini_messages_compat_service.go index 86073d9ca5..375fb05326 100644 --- a/backend/internal/service/gemini_messages_compat_service.go +++ b/backend/internal/service/gemini_messages_compat_service.go @@ -1449,6 +1449,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. if contentType == "" { contentType = "application/json" } + MarkResponseCommitted(c) c.Data(http.StatusInternalServerError, contentType, respBody) return nil, fmt.Errorf("gemini upstream error: %d (skipped by error policy)", resp.StatusCode) case ErrorPolicyMatched, ErrorPolicyTempUnscheduled: @@ -1561,6 +1562,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. if contentType == "" { contentType = "application/json" } + MarkResponseCommitted(c) c.Data(resp.StatusCode, contentType, respBody) if upstreamMsg == "" { return nil, fmt.Errorf("gemini upstream error: %d", resp.StatusCode) @@ -1700,6 +1702,7 @@ func sanitizeUpstreamErrorMessage(msg string) string { } func (s *GeminiMessagesCompatService) writeGeminiMappedError(c *gin.Context, account *Account, upstreamStatus int, upstreamRequestID string, body []byte) error { + MarkResponseCommitted(c) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) upstreamDetail := "" @@ -2243,6 +2246,7 @@ func randomHex(nBytes int) string { } func (s *GeminiMessagesCompatService) writeClaudeError(c *gin.Context, status int, errType, message string) error { + MarkResponseCommitted(c) c.JSON(status, gin.H{ "type": "error", "error": gin.H{"type": errType, "message": message}, @@ -2251,6 +2255,7 @@ func (s *GeminiMessagesCompatService) writeClaudeError(c *gin.Context, status in } func (s *GeminiMessagesCompatService) writeGoogleError(c *gin.Context, status int, message string) error { + MarkResponseCommitted(c) c.JSON(status, gin.H{ "error": gin.H{ "code": status, diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index e6d438baea..8b75097349 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -853,6 +853,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( // writeChatCompletionsError writes an error response in OpenAI Chat Completions format. func writeChatCompletionsError(c *gin.Context, statusCode int, errType, message string) { + MarkResponseCommitted(c) c.JSON(statusCode, gin.H{ "error": gin.H{ "type": errType, diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index be4dc539ff..6374ee91bd 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -3552,6 +3552,7 @@ func (s *OpenAIGatewayService) handleErrorResponsePassthrough( account *Account, requestBody []byte, ) error { + MarkResponseCommitted(c) body := s.readUpstreamErrorBody(resp) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) @@ -4289,6 +4290,7 @@ func (s *OpenAIGatewayService) handleErrorResponse( "upstream_error", "Upstream request failed", ); matched { + MarkResponseCommitted(c) c.JSON(status, gin.H{ "error": gin.H{ "type": errType, @@ -4316,6 +4318,7 @@ func (s *OpenAIGatewayService) handleErrorResponse( Message: upstreamMsg, Detail: upstreamDetail, }) + MarkResponseCommitted(c) c.JSON(http.StatusInternalServerError, gin.H{ "error": gin.H{ "type": "upstream_error", @@ -4359,6 +4362,8 @@ func (s *OpenAIGatewayService) handleErrorResponse( } } + MarkResponseCommitted(c) + // Return appropriate error response var errType, errMsg string var statusCode int @@ -4438,6 +4443,7 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( c, account.Platform, resp.StatusCode, body, http.StatusBadGateway, "api_error", "Upstream request failed", ); matched { + MarkResponseCommitted(c) writeError(c, status, errType, errMsg) if upstreamMsg == "" { upstreamMsg = errMsg @@ -4461,6 +4467,7 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( Message: upstreamMsg, Detail: upstreamDetail, }) + MarkResponseCommitted(c) writeError(c, http.StatusInternalServerError, "api_error", "Upstream gateway error") if upstreamMsg == "" { return nil, fmt.Errorf("upstream error: %d (not in custom error codes)", resp.StatusCode) @@ -4498,6 +4505,8 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( } } + MarkResponseCommitted(c) + // Map status code to error type and write response errType := "api_error" switch { diff --git a/backend/internal/service/ops_upstream_context.go b/backend/internal/service/ops_upstream_context.go index 2405f306fe..4a1e36aa6f 100644 --- a/backend/internal/service/ops_upstream_context.go +++ b/backend/internal/service/ops_upstream_context.go @@ -34,6 +34,10 @@ const ( // Client-side configuration denials should remain visible in ops_error_logs, // but should be excluded from SLA/error-rate calculations. + // ResponseCommittedKey 由 handleErrorResponse 系列函数在写完 HTTP 错误响应后设置。 + // ensureForwardErrorResponse 检查此 key,为 true 时跳过兜底写入,避免在已完成的 JSON 后追加 SSE。 + ResponseCommittedKey = "response_committed" + OpsClientBusinessLimitedKey = "ops_client_business_limited" OpsClientBusinessLimitedReasonKey = "ops_client_business_limited_reason" OpsClientBusinessLimitedReasonIPRestriction = "api_key_ip_restriction" @@ -43,6 +47,17 @@ const ( OpsClientBusinessLimitedReasonLocalPolicyDenied = "local_policy_denied" ) +func MarkResponseCommitted(c *gin.Context) { c.Set(ResponseCommittedKey, true) } + +func IsResponseCommitted(c *gin.Context) bool { + v, ok := c.Get(ResponseCommittedKey) + if !ok { + return false + } + b, _ := v.(bool) + return b +} + func SetOpsLatencyMs(c *gin.Context, key string, value int64) { if c == nil || strings.TrimSpace(key) == "" || value < 0 { return diff --git a/frontend/src/views/admin/ChannelsView.vue b/frontend/src/views/admin/ChannelsView.vue index 38f470a48b..57ea1f3879 100644 --- a/frontend/src/views/admin/ChannelsView.vue +++ b/frontend/src/views/admin/ChannelsView.vue @@ -1205,8 +1205,7 @@ function apiToForm(channel: Channel): PlatformSection[] { const webSearchEnabled = wsEmulation?.[platform] === true const codexImageGenerationBridge = fc?.codex_image_generation_bridge as Record | undefined const codexImageGenerationBridgeEnabled = codexImageGenerationBridge?.[platform] === true - const bedrockCCCompat = fc?.bedrock_cc_compat as Record | undefined - const bedrockCCCompatEnabled = bedrockCCCompat?.[platform] === true + const bedrockCCCompatEnabled = fc?.bedrock_cc_compat === true sections.push({ platform,