From e9e3c46cb76f564554357507102caf44cc79857d Mon Sep 17 00:00:00 2001 From: wwcchh0123 <1170142389@qq.com> Date: Tue, 25 Aug 2026 14:07:58 +0800 Subject: [PATCH 1/3] fix(ops): snapshot proxy on upstream errors --- .../upstream-error-proxy-attribution.md | 108 ++++++++++ .../service/antigravity_gateway_claude.go | 14 ++ .../service/antigravity_gateway_compat.go | 4 + .../service/antigravity_gateway_gemini.go | 14 ++ .../service/antigravity_gateway_retry.go | 6 + .../service/antigravity_gateway_streaming.go | 2 + .../service/gateway_anthropic_passthrough.go | 6 + backend/internal/service/gateway_bedrock.go | 6 + .../internal/service/gateway_count_tokens.go | 4 + backend/internal/service/gateway_forward.go | 18 ++ .../gateway_forward_as_chat_completions.go | 2 + .../service/gateway_forward_as_responses.go | 2 + .../service/gateway_upstream_response.go | 4 + .../gemini_chat_completions_compat_service.go | 8 + .../service/gemini_messages_compat_service.go | 30 +++ .../service/grok_credential_failure.go | 4 + .../service/grok_credential_failure_test.go | 28 +++ backend/internal/service/grok_media.go | 6 + backend/internal/service/openai_embeddings.go | 6 +- .../service/openai_first_output_timeout.go | 4 +- .../service/openai_gateway_cc_pipeline.go | 4 +- .../openai_gateway_chat_completions.go | 2 +- ...teway_chat_completions_anthropic_native.go | 2 +- .../openai_gateway_chat_completions_raw.go | 2 + .../service/openai_gateway_count_tokens.go | 4 +- .../service/openai_gateway_forward.go | 2 + .../internal/service/openai_gateway_grok.go | 4 + .../openai_gateway_grok_chat_bridge.go | 2 + .../service/openai_gateway_messages.go | 4 +- ...penai_gateway_messages_anthropic_native.go | 2 +- .../service/openai_gateway_passthrough.go | 6 + ...enai_gateway_responses_anthropic_native.go | 2 +- .../service/openai_gateway_service.go | 10 + .../service/openai_gateway_upstream_errors.go | 10 + .../service/openai_gateway_usage_integrity.go | 2 + backend/internal/service/openai_images.go | 4 + .../service/openai_images_responses.go | 14 +- .../internal/service/openai_silent_refusal.go | 4 + .../openai_upstream_transport_error.go | 2 + ...ai_upstream_transport_error_handle_test.go | 93 ++++++++- .../service/openai_ws_fallback_test.go | 24 +++ .../service/ops_queue_sanitize_test.go | 15 +- .../internal/service/ops_repo_mock_test.go | 4 + backend/internal/service/ops_service.go | 26 ++- .../internal/service/ops_upstream_context.go | 78 +++++++ .../service/ops_upstream_context_test.go | 196 ++++++++++++++++++ 46 files changed, 764 insertions(+), 30 deletions(-) create mode 100644 backend/dev-docs/upstream-error-proxy-attribution.md diff --git a/backend/dev-docs/upstream-error-proxy-attribution.md b/backend/dev-docs/upstream-error-proxy-attribution.md new file mode 100644 index 0000000000..d7b4ada37a --- /dev/null +++ b/backend/dev-docs/upstream-error-proxy-attribution.md @@ -0,0 +1,108 @@ +# Upstream error proxy attribution + +## Scope + +Gateway inference errors are stored in `ops_error_logs.upstream_errors`. Each +array item represents one upstream attempt, including same-account retries and +cross-account failover attempts. Proxy attribution is stored on that item; it +must not be reconstructed later from `accounts.proxy_id`. + +This repository has one gateway implementation under `backend/internal/service`. +There is no separate `gateway/backend` source copy to synchronize. + +## Event contract + +Every newly persisted event contains these fields. `proxy_id` is emitted even +when its value is JSON `null`: + +| Field | Meaning | +| --- | --- | +| `proxy_id` | Managed proxy ID used by the attempt; `null` for direct or unknown routes. | +| `proxy_name` | Snapshotted proxy name, `direct/no_proxy`, or `unknown`. | + +Only proxy ID and name are stored. Proxy URL, protocol endpoint, host, port, +username, password, and authorization data are excluded. + +The two sentinel names have strict meanings: + +- `direct/no_proxy`: the transport was explicitly given no proxy. +- `unknown`: the route cannot be proven from event-time evidence. This includes + legacy events, pre-transport credential failures, and OpenAI WebSocket use of + the default HTTP client. + +## Why the account snapshot is event-time evidence + +Account selection loads `ProxyID` and `Proxy` into one in-memory snapshot. HTTP, +TLS-fingerprint, Bedrock, Gemini, Antigravity, OpenAI, and OpenAI WebSocket +forwarding use that snapshot for managed proxy routing. +Inference transports use a managed proxy only when both the binding ID and its +hydrated proxy object are present; the attribution accessors use the same rule. +OpenAI WebSocket additionally retains the existing default-client environment +proxy behavior when that snapshot does not yield a usable proxy. The transport +does not switch to a database backup proxy during a request. + +Expired proxy fallback happens before scheduling by atomically rewriting the +account binding: + +- backup proxy: `proxy_id` becomes the backup ID and + `proxy_fallback_origin_id` stores the expired proxy ID; +- direct fallback: `proxy_id` becomes null and + `proxy_fallback_origin_id` stores the expired proxy ID. + +Custom Anthropic relays receive the same snapshotted proxy as their encoded +relay proxy parameter. OpenAI WebSocket requests keep the existing transport +behavior: configured account proxies use the proxy client, while an empty +account proxy leaves the HTTP client unset so `coder/websocket` uses +`http.DefaultClient`, including its `HTTP_PROXY`, `HTTPS_PROXY`, and `NO_PROXY` +handling. + +For WebSocket errors without a usable account proxy, the event stores +`proxy_id=null` and `proxy_name=unknown`. It does not infer whether the default +client selected an environment proxy or a direct connection after the fact. + +Credential acquisition failures that occur before the inference transport is +opened are also recorded as unknown. The account's proxy binding is not treated +as proof that the credential operation or an inference request used that route. + +Each failure site constructs a complete `OpsUpstreamErrorEvent`. Its literal +sets `proxy_id` and `proxy_name` from the same route input used by the attempt. +Credential-free accessors copy only the managed proxy ID and name. A fallback +proxy needs no special event flag: its event-time ID and name are the historical +route evidence. + +`appendOpsUpstreamError` accepts only the completed event. It does not receive +an `Account`, query current account state, or infer proxy attribution while +appending. Once appended, the event contains only scalar snapshot values, so +later account mutation or failover cannot change an earlier event. + +Queue sanitization preserves every non-nil attempt event. To bound asynchronous +queue memory, only the last 16 attempts retain the larger `detail` and +`upstream_response_body` values; older attempts keep their timestamps, account, +proxy, status, kind, message, and other scalar metadata. + +## Legacy events and analytics + +Legacy JSON is not rewritten in the database. Detail reads and the shared parser +materialize missing attribution as: + +```json +{ + "proxy_id": null, + "proxy_name": "unknown" +} +``` + +Proxy and region aggregation must use the event fields first. A safe grouping +rule is: + +1. A non-null event `proxy_id` groups by that ID and uses the event + `proxy_name` as the historical label. A join to `proxies` may enrich current + metadata but must not replace the event label. +2. `proxy_id=null` and `proxy_name=direct/no_proxy` groups as direct. +3. Missing attribution, or `proxy_id=null` and `proxy_name=unknown`, groups as + unknown. Do not join through the account's current `proxy_id` and present it + as historical attribution. + +If an operator deliberately produces a current-account cohort for legacy data, +the report must be labeled as a current snapshot; it is not historical proxy +attribution. diff --git a/backend/internal/service/antigravity_gateway_claude.go b/backend/internal/service/antigravity_gateway_claude.go index 3511f60288..9e3dc0f7dd 100644 --- a/backend/internal/service/antigravity_gateway_claude.go +++ b/backend/internal/service/antigravity_gateway_claude.go @@ -146,6 +146,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, logBody, maxBytes := s.getLogConfig() upstreamDetail := s.getUpstreamErrorDetail(respBody) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -203,6 +205,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, }) if retryErr != nil { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -242,6 +246,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, retryUpstreamDetail = truncateString(string(retryBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -278,6 +284,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, errMsg := strings.TrimSpace(extractAntigravityErrorMessage(respBody)) if isThinkingBudgetConstraintError(errMsg) && s.settingService.IsBudgetRectifierEnabled(ctx) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -359,6 +367,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, logger.LegacyPrintf("service.antigravity_gateway", "%s status=400 prompt_too_long=true upstream_message=%q request_id=%s body=%s", prefix, upstreamMsg, resp.Header.Get("x-request-id"), truncateForLog(respBody, maxBytes)) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -385,6 +395,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, upstreamDetail := s.getUpstreamErrorDetail(respBody) log.Printf("%s status=400 google_config_error failover=true upstream_message=%q account=%d", prefix, upstreamMsg, account.ID) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -403,6 +415,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) upstreamDetail := s.getUpstreamErrorDetail(respBody) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/antigravity_gateway_compat.go b/backend/internal/service/antigravity_gateway_compat.go index 1d6d39e1aa..5c6524489e 100644 --- a/backend/internal/service/antigravity_gateway_compat.go +++ b/backend/internal/service/antigravity_gateway_compat.go @@ -424,6 +424,8 @@ func (s *AntigravityGatewayService) handleAntigravityCompatHTTPError( if s.shouldFailoverUpstreamError(resp.StatusCode) { message := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractAntigravityErrorMessage(body))) event := OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -493,6 +495,8 @@ func (s *AntigravityGatewayService) writeMappedAntigravityCompatError( message := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractAntigravityErrorMessage(body))) setOpsUpstreamError(c, upstreamStatus, message, s.getUpstreamErrorDetail(body)) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/antigravity_gateway_gemini.go b/backend/internal/service/antigravity_gateway_gemini.go index 77a7be3df7..2093911100 100644 --- a/backend/internal/service/antigravity_gateway_gemini.go +++ b/backend/internal/service/antigravity_gateway_gemini.go @@ -227,6 +227,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractAntigravityErrorMessage(signatureCheckBody))) upstreamDetail := s.getUpstreamErrorDetail(signatureCheckBody) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -272,6 +274,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co retryOpsBody = retryUnwrapped } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -292,6 +296,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co } else { if switchErr, ok := IsAntigravityAccountSwitchError(retryErr); ok { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -305,6 +311,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co } } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -346,6 +354,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co if resp.StatusCode == http.StatusBadRequest && isGoogleProjectConfigError(strings.ToLower(upstreamMsg)) { log.Printf("%s status=400 google_config_error failover=true upstream_message=%q account=%d", prefix, upstreamMsg, account.ID) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -360,6 +370,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co if s.shouldFailoverUpstreamError(resp.StatusCode) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -375,6 +387,8 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co contentType = "application/json" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/antigravity_gateway_retry.go b/backend/internal/service/antigravity_gateway_retry.go index 5e16ba8692..78d46a4b23 100644 --- a/backend/internal/service/antigravity_gateway_retry.go +++ b/backend/internal/service/antigravity_gateway_retry.go @@ -548,6 +548,8 @@ urlFallbackLoop: if err != nil { safeErr := sanitizeUpstreamErrorMessage(err.Error()) appendOpsUpstreamError(p.c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(p.account), + ProxyName: opsUpstreamProxyName(p.account), Platform: p.account.Platform, AccountID: p.account.ID, AccountName: p.account.Name, @@ -625,6 +627,8 @@ urlFallbackLoop: upstreamMsg := strings.TrimSpace(extractAntigravityErrorMessage(respBody)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) appendOpsUpstreamError(p.c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(p.account), + ProxyName: opsUpstreamProxyName(p.account), Platform: p.account.Platform, AccountID: p.account.ID, AccountName: p.account.Name, @@ -660,6 +664,8 @@ urlFallbackLoop: upstreamMsg := strings.TrimSpace(extractAntigravityErrorMessage(respBody)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) appendOpsUpstreamError(p.c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(p.account), + ProxyName: opsUpstreamProxyName(p.account), Platform: p.account.Platform, AccountID: p.account.ID, AccountName: p.account.Name, diff --git a/backend/internal/service/antigravity_gateway_streaming.go b/backend/internal/service/antigravity_gateway_streaming.go index 2b68fc115a..002543f9ec 100644 --- a/backend/internal/service/antigravity_gateway_streaming.go +++ b/backend/internal/service/antigravity_gateway_streaming.go @@ -695,6 +695,8 @@ func (s *AntigravityGatewayService) writeMappedClaudeError(c *gin.Context, accou upstreamDetail := s.getUpstreamErrorDetail(body) setOpsUpstreamError(c, upstreamStatus, upstreamMsg, upstreamDetail) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gateway_anthropic_passthrough.go b/backend/internal/service/gateway_anthropic_passthrough.go index bf899211e8..10455f0581 100644 --- a/backend/internal/service/gateway_anthropic_passthrough.go +++ b/backend/internal/service/gateway_anthropic_passthrough.go @@ -140,6 +140,8 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput( respBody, _ := s.readUpstreamErrorBody(resp) _ = resp.Body.Close() appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -184,6 +186,8 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput( s.handleRetryExhaustedSideEffects(ctx, resp, account) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -218,6 +222,8 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput( s.handleFailoverSideEffects(ctx, resp, account, input.RequestModel) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gateway_bedrock.go b/backend/internal/service/gateway_bedrock.go index 77baf3f29b..5c46f25971 100644 --- a/backend/internal/service/gateway_bedrock.go +++ b/backend/internal/service/gateway_bedrock.go @@ -225,6 +225,8 @@ func (s *GatewayService) executeBedrockUpstream( respBody, _ := s.readUpstreamErrorBody(resp) _ = resp.Body.Close() appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -276,6 +278,8 @@ func (s *GatewayService) handleBedrockUpstreamErrors( s.handleRetryExhaustedSideEffects(ctx, resp, account) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -300,6 +304,8 @@ func (s *GatewayService) handleBedrockUpstreamErrors( s.handleFailoverSideEffects(ctx, resp, account) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gateway_count_tokens.go b/backend/internal/service/gateway_count_tokens.go index 003cbb7c1e..b3c7286c3f 100644 --- a/backend/internal/service/gateway_count_tokens.go +++ b/backend/internal/service/gateway_count_tokens.go @@ -270,6 +270,8 @@ func (s *GatewayService) forwardCountTokensAnthropicAPIKeyPassthrough(ctx contex if err != nil { setOpsUpstreamError(c, 0, sanitizeUpstreamErrorMessage(err.Error()), "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -324,6 +326,8 @@ func (s *GatewayService) forwardCountTokensAnthropicAPIKeyPassthrough(ctx contex } setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gateway_forward.go b/backend/internal/service/gateway_forward.go index 2571079403..7743b2d1c7 100644 --- a/backend/internal/service/gateway_forward.go +++ b/backend/internal/service/gateway_forward.go @@ -405,6 +405,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A if s.shouldRectifySignatureError(ctx, account, respBody, reqModel) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -466,6 +468,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A _ = 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, @@ -506,6 +510,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A _ = retryResp2.Body.Close() } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -545,6 +551,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A errMsg := extractUpstreamErrorMessage(respBody) if isThinkingBudgetConstraintError(errMsg) && s.settingService.IsBudgetRectifierEnabled(ctx) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -615,6 +623,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A respBody, _ := s.readUpstreamErrorBody(resp) _ = resp.Body.Close() appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -669,6 +679,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A s.handleRetryExhaustedSideEffects(ctx, resp, account) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -704,6 +716,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A s.handleFailoverSideEffects(ctx, resp, account, reqModel) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, UpstreamStatusCode: resp.StatusCode, @@ -746,6 +760,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -822,6 +838,8 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gateway_forward_as_chat_completions.go b/backend/internal/service/gateway_forward_as_chat_completions.go index b3d6c565ba..6a84426642 100644 --- a/backend/internal/service/gateway_forward_as_chat_completions.go +++ b/backend/internal/service/gateway_forward_as_chat_completions.go @@ -148,6 +148,8 @@ func (s *GatewayService) ForwardAsChatCompletions( if s.shouldFailoverUpstreamError(resp.StatusCode) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gateway_forward_as_responses.go b/backend/internal/service/gateway_forward_as_responses.go index 7c6cde8328..a72eb1a278 100644 --- a/backend/internal/service/gateway_forward_as_responses.go +++ b/backend/internal/service/gateway_forward_as_responses.go @@ -161,6 +161,8 @@ func (s *GatewayService) ForwardAsResponses( if s.shouldFailoverUpstreamError(resp.StatusCode) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gateway_upstream_response.go b/backend/internal/service/gateway_upstream_response.go index 5663333a91..cd9b1bf8b8 100644 --- a/backend/internal/service/gateway_upstream_response.go +++ b/backend/internal/service/gateway_upstream_response.go @@ -398,6 +398,8 @@ func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Res } setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, UpstreamStatusCode: resp.StatusCode, @@ -578,6 +580,8 @@ func (s *GatewayService) handleRetryExhaustedError(ctx context.Context, resp *ht } setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, UpstreamStatusCode: resp.StatusCode, diff --git a/backend/internal/service/gemini_chat_completions_compat_service.go b/backend/internal/service/gemini_chat_completions_compat_service.go index a56a1b3ebc..bb464bbe7c 100644 --- a/backend/internal/service/gemini_chat_completions_compat_service.go +++ b/backend/internal/service/gemini_chat_completions_compat_service.go @@ -127,6 +127,8 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions( if err != nil { safeErr := sanitizeUpstreamErrorMessage(err.Error()) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -172,6 +174,8 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions( upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -225,6 +229,8 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions( if s.shouldFailoverGeminiUpstreamError(resp.StatusCode) { upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(evBody))) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -816,6 +822,8 @@ func (s *GeminiMessagesCompatService) writeGeminiChatCompletionsMappedError( setOpsUpstreamError(c, upstreamStatus, upstreamMsg, "") if account != nil { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/gemini_messages_compat_service.go b/backend/internal/service/gemini_messages_compat_service.go index 5f9854167d..0f245b95fe 100644 --- a/backend/internal/service/gemini_messages_compat_service.go +++ b/backend/internal/service/gemini_messages_compat_service.go @@ -785,6 +785,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex if err != nil { safeErr := sanitizeUpstreamErrorMessage(err.Error()) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -823,6 +825,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -909,6 +913,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -976,6 +982,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1010,6 +1018,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex } log.Printf("[Gemini] status=400 google_config_error failover=true upstream_message=%q account=%d", upstreamMsg, account.ID) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1038,6 +1048,8 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1324,6 +1336,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. if err != nil { safeErr := sanitizeUpstreamErrorMessage(err.Error()) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1392,6 +1406,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1492,6 +1508,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. upstreamDetail = truncateString(string(evBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1523,6 +1541,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. } log.Printf("[Gemini] status=400 google_config_error failover=true upstream_message=%q account=%d", upstreamMsg, account.ID) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1548,6 +1568,8 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. upstreamDetail = truncateString(string(evBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1678,6 +1700,8 @@ func (s *GeminiMessagesCompatService) skippedErrorPolicyFailoverError(c *gin.Con upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(respBody))) upstreamDetail := s.upstreamErrorDetail(respBody) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1719,6 +1743,8 @@ func (s *GeminiMessagesCompatService) writeGeminiCustomCodeSkippedError(c *gin.C upstreamDetail := s.upstreamErrorDetail(body) setOpsUpstreamError(c, upstreamStatus, upstreamMsg, upstreamDetail) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1747,6 +1773,8 @@ func (s *GeminiMessagesCompatService) writeGeminiNativeUpstreamError(c *gin.Cont } setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1811,6 +1839,8 @@ func (s *GeminiMessagesCompatService) writeGeminiMappedError(c *gin.Context, acc } setOpsUpstreamError(c, upstreamStatus, upstreamMsg, upstreamDetail) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/grok_credential_failure.go b/backend/internal/service/grok_credential_failure.go index adad0c5932..cc3b86b49d 100644 --- a/backend/internal/service/grok_credential_failure.go +++ b/backend/internal/service/grok_credential_failure.go @@ -647,6 +647,10 @@ func (s *OpenAIGatewayService) newGrokCredentialFailover(c *gin.Context, account class.message = "Grok OAuth credentials are unavailable" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + // Credential acquisition happens before the inference transport opens, + // so the account binding is not evidence of an inference proxy route. + ProxyID: nil, + ProxyName: opsProxyNameUnknown, Platform: PlatformGrok, AccountID: account.ID, Stage: string(GatewayFailureStageAccountAuth), diff --git a/backend/internal/service/grok_credential_failure_test.go b/backend/internal/service/grok_credential_failure_test.go index 6665b8401d..cdc5f4f300 100644 --- a/backend/internal/service/grok_credential_failure_test.go +++ b/backend/internal/service/grok_credential_failure_test.go @@ -294,6 +294,34 @@ func TestGetRequestCredentialMapsPermanentGrokOAuthFailureAndRedactsSecrets(t *t require.NotContains(t, events[0].Message, "leaked-refresh") } +func TestNewGrokCredentialFailoverDoesNotAttributeInferenceProxy(t *testing.T) { + gin.SetMode(gin.TestMode) + proxyID := int64(43) + account := &Account{ + ID: 701, + Platform: PlatformGrok, + ProxyID: &proxyID, + Proxy: &Proxy{ID: proxyID, Name: "bound-proxy"}, + } + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + + err := (&OpenAIGatewayService{}).newGrokCredentialFailover(c, account, grokCredentialFailureClass{ + scope: GatewayFailureScopeAccount, + reason: GrokCredentialReasonAccountChanged, + action: NextAccountRetry, + message: "credential unavailable", + }) + require.Error(t, err) + + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 1) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameUnknown, events[0].ProxyName) +} + func TestGetRequestCredentialPermanentMappingsPersistAndInvalidate(t *testing.T) { gin.SetMode(gin.TestMode) tests := []struct { diff --git a/backend/internal/service/grok_media.go b/backend/internal/service/grok_media.go index 85dd1b14ec..9bc7bdc5ec 100644 --- a/backend/internal/service/grok_media.go +++ b/backend/internal/service/grok_media.go @@ -1239,6 +1239,8 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse( if isGrokContentPolicyRejection(resp.StatusCode, body) { clientMsg := grokContentPolicyClientMessage(body) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1269,6 +1271,8 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse( if !account.ShouldHandleErrorCode(resp.StatusCode) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1288,6 +1292,8 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_embeddings.go b/backend/internal/service/openai_embeddings.go index 87b1904923..1e2f9c8b60 100644 --- a/backend/internal/service/openai_embeddings.go +++ b/backend/internal/service/openai_embeddings.go @@ -89,7 +89,7 @@ func (s *OpenAIGatewayService) ForwardEmbeddings( account.ApplyHeaderOverrides(upstreamReq.Header) proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account) @@ -97,6 +97,8 @@ func (s *OpenAIGatewayService) ForwardEmbeddings( safeErr := sanitizeUpstreamErrorMessage(err.Error()) setOpsUpstreamError(c, 0, safeErr, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -126,6 +128,8 @@ func (s *OpenAIGatewayService) ForwardEmbeddings( upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_first_output_timeout.go b/backend/internal/service/openai_first_output_timeout.go index fbd985d067..45a4e05d07 100644 --- a/backend/internal/service/openai_first_output_timeout.go +++ b/backend/internal/service/openai_first_output_timeout.go @@ -262,7 +262,9 @@ func (s *OpenAIGatewayService) newOpenAIFirstOutputTimeoutError( ) requestID := strings.TrimSpace(responseHeaders.Get("x-request-id")) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, UpstreamStatusCode: http.StatusGatewayTimeout, UpstreamRequestID: requestID, Kind: "first_output_timeout", Message: "OpenAI upstream produced no semantic output before the deadline", Detail: fmt.Sprintf("phase=%s elapsed_ms=%d timeout_ms=%d", phase, elapsed.Milliseconds(), timeout.Milliseconds()), diff --git a/backend/internal/service/openai_gateway_cc_pipeline.go b/backend/internal/service/openai_gateway_cc_pipeline.go index a22288b70a..325081d9da 100644 --- a/backend/internal/service/openai_gateway_cc_pipeline.go +++ b/backend/internal/service/openai_gateway_cc_pipeline.go @@ -112,6 +112,8 @@ func (s *OpenAIGatewayService) failoverOpenAIUpstreamHTTPError( upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -223,7 +225,7 @@ func (s *OpenAIGatewayService) sendCCUpstreamRequest( account.ApplyHeaderOverrides(upstreamReq.Header) proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index c9a7e34b6a..fd0edd7cfd 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -348,7 +348,7 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions( // 7. Send request proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account) diff --git a/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go b/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go index 6e24fec392..d835556030 100644 --- a/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go +++ b/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go @@ -106,7 +106,7 @@ func (s *OpenAIGatewayService) forwardChatCompletionsViaNativeAnthropic( } proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } diff --git a/backend/internal/service/openai_gateway_chat_completions_raw.go b/backend/internal/service/openai_gateway_chat_completions_raw.go index a98c11e5e8..dc9dcc5a62 100644 --- a/backend/internal/service/openai_gateway_chat_completions_raw.go +++ b/backend/internal/service/openai_gateway_chat_completions_raw.go @@ -193,6 +193,8 @@ func (s *OpenAIGatewayService) forwardAsRawChatCompletions( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_gateway_count_tokens.go b/backend/internal/service/openai_gateway_count_tokens.go index 1480a2edd3..82270b3167 100644 --- a/backend/internal/service/openai_gateway_count_tokens.go +++ b/backend/internal/service/openai_gateway_count_tokens.go @@ -80,7 +80,7 @@ func (s *OpenAIGatewayService) ForwardResponsesInputTokens( } proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account) @@ -320,7 +320,7 @@ func (s *OpenAIGatewayService) ForwardCountTokensAsAnthropic( } proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } resp, err := s.doOpenAIUpstream(upstreamReq, proxyURL, account) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 18b5aff28e..f26b8a5806 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -1108,6 +1108,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco upstreamDetail = truncateString(string(respBody), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_gateway_grok.go b/backend/internal/service/openai_gateway_grok.go index efcbaa8d44..95c649ad97 100644 --- a/backend/internal/service/openai_gateway_grok.go +++ b/backend/internal/service/openai_gateway_grok.go @@ -170,6 +170,8 @@ func (s *OpenAIGatewayService) forwardGrokResponses( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1412,6 +1414,8 @@ func (s *OpenAIGatewayService) describeGrokComposerImage( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_gateway_grok_chat_bridge.go b/backend/internal/service/openai_gateway_grok_chat_bridge.go index 0050d10bfa..27e49773e9 100644 --- a/backend/internal/service/openai_gateway_grok_chat_bridge.go +++ b/backend/internal/service/openai_gateway_grok_chat_bridge.go @@ -623,6 +623,8 @@ func (s *OpenAIGatewayService) forwardGrokChatCompletionsViaResponses( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index b12ab68044..b63d4455e9 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -374,7 +374,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( // 7. Send request proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } // Grok may reject encrypted reasoning replayed under a different OAuth @@ -689,6 +689,8 @@ func (s *OpenAIGatewayService) recordOpenAIMessagesStreamUpstreamError(c *gin.Co message = sanitizeUpstreamErrorMessage(message) setOpsUpstreamError(c, http.StatusBadGateway, message, "") event := OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: PlatformOpenAI, UpstreamStatusCode: http.StatusBadGateway, UpstreamRequestID: strings.TrimSpace(upstreamRequestID), diff --git a/backend/internal/service/openai_gateway_messages_anthropic_native.go b/backend/internal/service/openai_gateway_messages_anthropic_native.go index 4fb5c2ec5a..f7addc90b0 100644 --- a/backend/internal/service/openai_gateway_messages_anthropic_native.go +++ b/backend/internal/service/openai_gateway_messages_anthropic_native.go @@ -86,7 +86,7 @@ func (s *OpenAIGatewayService) forwardAnthropicViaNativeAnthropicEndpoint( } proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index ecca154781..d2b49a2f9a 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -894,6 +894,8 @@ func (s *OpenAIGatewayService) handleFailoverErrorResponsePassthrough( canonicalModel := canonicalOpenAIAccountSchedulingModel(account, reqModel) shouldDisable := s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, canonicalModel) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -960,6 +962,8 @@ func (s *OpenAIGatewayService) handleErrorResponsePassthrough( _ = s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, canonicalModel) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1676,6 +1680,8 @@ func (s *OpenAIGatewayService) recordOpenAIStreamUpstreamError( if c != nil { setOpsUpstreamError(c, statusCode, message, detail) event := OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: PlatformOpenAI, UpstreamStatusCode: statusCode, UpstreamRequestID: strings.TrimSpace(upstreamRequestID), diff --git a/backend/internal/service/openai_gateway_responses_anthropic_native.go b/backend/internal/service/openai_gateway_responses_anthropic_native.go index 423cba3459..6a405d65b2 100644 --- a/backend/internal/service/openai_gateway_responses_anthropic_native.go +++ b/backend/internal/service/openai_gateway_responses_anthropic_native.go @@ -111,7 +111,7 @@ func (s *OpenAIGatewayService) forwardResponsesViaNativeAnthropic( } proxyURL := "" - if account.Proxy != nil { + if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index aaa2e30af7..8f4c03d3dd 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -892,7 +892,17 @@ func (s *OpenAIGatewayService) writeOpenAIWSFallbackErrorResponse(c *gin.Context setOpsUpstreamError(c, statusCode, upstreamMessage, "") if account != nil { + proxyID := opsUpstreamProxyID(account) + proxyName := opsUpstreamProxyName(account) + // The existing WS client uses the configured proxy only when both fields + // are present; otherwise http.DefaultClient may select an environment proxy. + if account.ProxyID == nil || account.Proxy == nil { + proxyID = nil + proxyName = opsProxyNameUnknown + } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: proxyID, + ProxyName: proxyName, Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_gateway_upstream_errors.go b/backend/internal/service/openai_gateway_upstream_errors.go index 70c000e321..db2cd5d9c0 100644 --- a/backend/internal/service/openai_gateway_upstream_errors.go +++ b/backend/internal/service/openai_gateway_upstream_errors.go @@ -549,6 +549,8 @@ func (s *OpenAIGatewayService) handleErrorResponse( if isOpenAIRequestBodyTooLargeError(resp.StatusCode, upstreamMsg, body) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -596,6 +598,8 @@ func (s *OpenAIGatewayService) handleErrorResponse( // Check custom error codes if !account.ShouldHandleErrorCode(resp.StatusCode) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -633,6 +637,8 @@ func (s *OpenAIGatewayService) handleErrorResponse( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -798,6 +804,8 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( // return a generic error without exposing upstream details. if !account.ShouldHandleErrorCode(resp.StatusCode) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -828,6 +836,8 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_gateway_usage_integrity.go b/backend/internal/service/openai_gateway_usage_integrity.go index e0344d4f44..08dc475a35 100644 --- a/backend/internal/service/openai_gateway_usage_integrity.go +++ b/backend/internal/service/openai_gateway_usage_integrity.go @@ -53,6 +53,8 @@ func newGrokMissingUsageFailoverError(c *gin.Context, account *Account, upstream setOpsUpstreamError(c, http.StatusBadGateway, grokMissingUsageMessage, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: PlatformGrok, AccountID: accountID, AccountName: accountName, diff --git a/backend/internal/service/openai_images.go b/backend/internal/service/openai_images.go index c661aade36..bfbc11ac27 100644 --- a/backend/internal/service/openai_images.go +++ b/backend/internal/service/openai_images.go @@ -631,6 +631,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey( safeErr := sanitizeUpstreamErrorMessage(err.Error()) setOpsUpstreamError(c, 0, safeErr, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -650,6 +652,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey( upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_images_responses.go b/backend/internal/service/openai_images_responses.go index 434c352da2..8dd850098f 100644 --- a/backend/internal/service/openai_images_responses.go +++ b/backend/internal/service/openai_images_responses.go @@ -957,6 +957,8 @@ func (s *OpenAIGatewayService) handleOpenAIImagesErrorResponse( // handleCompatErrorResponse). if !account.ShouldHandleErrorCode(resp.StatusCode) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -997,6 +999,8 @@ func (s *OpenAIGatewayService) handleOpenAIImagesErrorResponse( kind = "failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1826,6 +1830,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth( safeErr := sanitizeUpstreamErrorMessage(err.Error()) setOpsUpstreamError(c, 0, safeErr, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1852,6 +1858,8 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth( upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -2024,7 +2032,9 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthResponseError( kind = "retry_exhausted_failover" } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, UpstreamStatusCode: statusCode, UpstreamRequestID: requestID, UpstreamURL: upstreamURL, Kind: kind, Message: message, }) @@ -2066,6 +2076,8 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthResponseError( } } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_silent_refusal.go b/backend/internal/service/openai_silent_refusal.go index 017c4c1f6d..1f04f31d37 100644 --- a/backend/internal/service/openai_silent_refusal.go +++ b/backend/internal/service/openai_silent_refusal.go @@ -247,6 +247,8 @@ func newOpenAISilentRefusalFailoverError(c *gin.Context, account *Account, upstr setOpsUpstreamError(c, http.StatusBadGateway, openAISilentRefusalUpstreamMessage, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: platform, AccountID: accountID, AccountName: accountName, @@ -284,6 +286,8 @@ func newOpenAIResponsesEmptyCompletedFailoverError(c *gin.Context, account *Acco setOpsUpstreamError(c, http.StatusBadGateway, openAIResponsesEmptyCompletedMessage, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: platform, AccountID: accountID, AccountName: accountName, diff --git a/backend/internal/service/openai_upstream_transport_error.go b/backend/internal/service/openai_upstream_transport_error.go index ed5c5f9c23..6072704f5b 100644 --- a/backend/internal/service/openai_upstream_transport_error.go +++ b/backend/internal/service/openai_upstream_transport_error.go @@ -109,6 +109,8 @@ func (s *OpenAIGatewayService) handleOpenAIUpstreamTransportError(ctx context.Co safeErr := sanitizeUpstreamErrorMessage(err.Error()) setOpsUpstreamError(c, 0, safeErr, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_upstream_transport_error_handle_test.go b/backend/internal/service/openai_upstream_transport_error_handle_test.go index 38e392893e..8aaed2eae4 100644 --- a/backend/internal/service/openai_upstream_transport_error_handle_test.go +++ b/backend/internal/service/openai_upstream_transport_error_handle_test.go @@ -40,17 +40,23 @@ func newOpenAITransportErrTestContext() (*gin.Context, *httptest.ResponseRecorde } type failingOpenAIHTTPUpstream struct { - err error - calls int + err error + calls int + proxyURL string + proxyURLs []string } -func (u *failingOpenAIHTTPUpstream) Do(_ *http.Request, _ string, _ int64, _ int) (*http.Response, error) { +func (u *failingOpenAIHTTPUpstream) Do(_ *http.Request, proxyURL string, _ int64, _ int) (*http.Response, error) { u.calls++ + u.proxyURL = proxyURL + u.proxyURLs = append(u.proxyURLs, proxyURL) return nil, u.err } -func (u *failingOpenAIHTTPUpstream) DoWithTLS(_ *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { +func (u *failingOpenAIHTTPUpstream) DoWithTLS(_ *http.Request, proxyURL string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { u.calls++ + u.proxyURL = proxyURL + u.proxyURLs = append(u.proxyURLs, proxyURL) return nil, u.err } @@ -60,7 +66,14 @@ func (u *failingOpenAIHTTPUpstream) DoWithTLS(_ *http.Request, _ string, _ int64 func TestHandleOpenAIUpstreamTransportError_PersistentEvictsAndFailsOver(t *testing.T) { repo := &openaiTransportAccountRepoStub{} svc := &OpenAIGatewayService{accountRepo: repo} - account := &Account{ID: 4627, Name: "proxy-expired", Platform: PlatformOpenAI} + proxyID := int64(10060) + account := &Account{ + ID: 4627, + Name: "proxy-expired", + Platform: PlatformOpenAI, + ProxyID: &proxyID, + Proxy: &Proxy{ID: proxyID, Name: "wldsg82-ipv6-10060"}, + } c, rec := newOpenAITransportErrTestContext() before := time.Now() @@ -85,6 +98,15 @@ func TestHandleOpenAIUpstreamTransportError_PersistentEvictsAndFailsOver(t *test // Must NOT write a response body — the handler owns the (failover) response. require.Equal(t, 0, rec.Body.Len()) + + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 1) + require.NotNil(t, events[0].ProxyID) + require.Equal(t, proxyID, *events[0].ProxyID) + require.Equal(t, "wldsg82-ipv6-10060", events[0].ProxyName) } // A transient blip should fail over but must NOT evict the account. @@ -191,12 +213,22 @@ func TestForwardAsRawChatCompletions_TransportErrorFailsOver(t *testing.T) { }, }, } + proxyID := int64(8001) + proxy := &Proxy{ + ID: proxyID, + Name: "oxylabs-uk-8001", + Protocol: "http", + Host: "proxy.example", + Port: 8080, + } account := &Account{ ID: 81, Name: "oc-20053", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "sk-test", "base_url": "https://opencode.ai/zen/v1"}, + ProxyID: &proxyID, + Proxy: proxy, } c, rec := newOpenAITransportErrTestContext() body := []byte(`{"model":"deepseek-v4-flash-free","messages":[{"role":"user","content":"hello"}]}`) @@ -204,11 +236,62 @@ func TestForwardAsRawChatCompletions_TransportErrorFailsOver(t *testing.T) { _, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.Equal(t, 1, upstream.calls) + require.Equal(t, proxy.URL(), upstream.proxyURL) var fo *UpstreamFailoverError require.True(t, errors.As(err, &fo), "transport error must trigger account failover") require.Equal(t, http.StatusBadGateway, fo.StatusCode) require.Empty(t, repo.tempUnschedCalls, "plain EOF is transient: fail over but do not evict") require.Equal(t, 0, rec.Body.Len(), "service must not write a hard 502 before handler can fail over") + + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 1) + require.NotNil(t, events[0].ProxyID) + require.Equal(t, proxyID, *events[0].ProxyID) + require.Equal(t, proxy.Name, events[0].ProxyName) +} + +func TestForwardAsRawChatCompletions_RecordsProxyPerAccountAttempt(t *testing.T) { + upstream := &failingOpenAIHTTPUpstream{err: errors.New("EOF")} + svc := &OpenAIGatewayService{ + accountRepo: &openaiTransportAccountRepoStub{}, + httpUpstream: upstream, + cfg: &config.Config{Security: config.SecurityConfig{ + URLAllowlist: config.URLAllowlistConfig{Enabled: false}, + }}, + } + proxyA := &Proxy{ID: 10060, Name: "proxy-a", Protocol: "http", Host: "proxy-a.example", Port: 8080} + proxyB := &Proxy{ID: 8001, Name: "proxy-b", Protocol: "http", Host: "proxy-b.example", Port: 8080} + fallbackOriginID := int64(10150) + accounts := []*Account{ + {ID: 81, Name: "account-a", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "a", "base_url": "https://example.com/v1"}, ProxyID: &proxyA.ID, Proxy: proxyA}, + {ID: 82, Name: "account-b", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "b", "base_url": "https://example.com/v1"}, ProxyID: &proxyB.ID, Proxy: proxyB, ProxyFallbackOriginID: &fallbackOriginID}, + {ID: 83, Name: "account-direct", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "c", "base_url": "https://example.com/v1"}}, + } + c, _ := newOpenAITransportErrTestContext() + body := []byte(`{"model":"test","messages":[{"role":"user","content":"hello"}]}`) + + for _, account := range accounts { + _, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + } + + require.Equal(t, []string{proxyA.URL(), proxyB.URL(), ""}, upstream.proxyURLs) + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 3) + for i, proxy := range []*Proxy{proxyA, proxyB} { + require.NotNil(t, events[i].ProxyID) + require.Equal(t, proxy.ID, *events[i].ProxyID) + require.Equal(t, proxy.Name, events[i].ProxyName) + } + require.Nil(t, events[2].ProxyID) + require.Equal(t, opsProxyNameDirect, events[2].ProxyName) } func TestHandleOpenAIUpstreamTransportError_RecordsOllamaActivityOnly(t *testing.T) { diff --git a/backend/internal/service/openai_ws_fallback_test.go b/backend/internal/service/openai_ws_fallback_test.go index ce06f6a21f..0d2a47645e 100644 --- a/backend/internal/service/openai_ws_fallback_test.go +++ b/backend/internal/service/openai_ws_fallback_test.go @@ -4,11 +4,13 @@ import ( "context" "errors" "net/http" + "net/http/httptest" "testing" "time" "github.com/Wei-Shaw/sub2api/internal/config" coderws "github.com/coder/websocket" + "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) @@ -160,6 +162,28 @@ func TestResolveOpenAIWSFallbackErrorResponse(t *testing.T) { }) } +func TestWriteOpenAIWSFallbackErrorResponseMarksDefaultClientRouteUnknown(t *testing.T) { + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + svc := &OpenAIGatewayService{} + account := &Account{ID: 42, Name: "default-client", Platform: PlatformOpenAI} + + written := svc.writeOpenAIWSFallbackErrorResponse( + c, + account, + wrapOpenAIWSFallback("auth_failed", errors.New("unauthorized")), + ) + require.True(t, written) + + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := raw.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 1) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameUnknown, events[0].ProxyName) +} + func TestOpenAIWSFallbackCooling(t *testing.T) { svc := &OpenAIGatewayService{cfg: &config.Config{}} svc.cfg.Gateway.OpenAIWS.FallbackCooldownSeconds = 1 diff --git a/backend/internal/service/ops_queue_sanitize_test.go b/backend/internal/service/ops_queue_sanitize_test.go index f2ae621c07..341c54e6b2 100644 --- a/backend/internal/service/ops_queue_sanitize_test.go +++ b/backend/internal/service/ops_queue_sanitize_test.go @@ -8,7 +8,10 @@ import ( func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) { entry := &OpsInsertErrorLogInput{} for i := 0; i < 20; i++ { + proxyID := int64(i + 1) entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{ + ProxyID: &proxyID, + ProxyName: "proxy-" + string(rune('a'+i)), Platform: strings.Repeat("p", 100), AccountName: strings.Repeat("a", 300), UpstreamStatusCode: 500, @@ -32,10 +35,13 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) { if err != nil { t.Fatal(err) } - if len(events) != 16 { - t.Fatalf("event count = %d, want 16", len(events)) + if len(events) != 20 { + t.Fatalf("event count = %d, want 20", len(events)) } - for _, event := range events { + for i, event := range events { + if event.ProxyID == nil || *event.ProxyID != int64(i+1) { + t.Fatalf("event %d proxy id = %v, want %d", i, event.ProxyID, i+1) + } if len(event.Platform) > 32 || len(event.AccountName) > 128 || len(event.UpstreamURL) > 2048 || len(event.Message) > 2048 { t.Fatalf("event fields were not bounded: %+v", event) } @@ -45,5 +51,8 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) { if strings.Contains(event.UpstreamResponseBody, "Bearer secret") || strings.Contains(event.Detail, `"secret"`) { t.Fatal("credential material was not redacted") } + if i < 4 && (event.UpstreamResponseBody != "" || event.Detail != "") { + t.Fatal("older event retained a large body outside the queue body window") + } } } diff --git a/backend/internal/service/ops_repo_mock_test.go b/backend/internal/service/ops_repo_mock_test.go index 4138ea77a2..77b9d17727 100644 --- a/backend/internal/service/ops_repo_mock_test.go +++ b/backend/internal/service/ops_repo_mock_test.go @@ -9,6 +9,7 @@ import ( type opsRepoMock struct { InsertErrorLogFn func(ctx context.Context, input *OpsInsertErrorLogInput) (int64, error) BatchInsertErrorLogsFn func(ctx context.Context, inputs []*OpsInsertErrorLogInput) (int64, error) + GetErrorLogByIDFn func(ctx context.Context, id int64) (*OpsErrorLogDetail, error) BatchInsertSystemLogsFn func(ctx context.Context, inputs []*OpsInsertSystemLogInput) (int64, error) ListSystemLogsFn func(ctx context.Context, filter *OpsSystemLogFilter) (*OpsSystemLogList, error) DeleteSystemLogsFn func(ctx context.Context, filter *OpsSystemLogCleanupFilter) (int64, error) @@ -34,6 +35,9 @@ func (m *opsRepoMock) ListErrorLogs(ctx context.Context, filter *OpsErrorLogFilt } func (m *opsRepoMock) GetErrorLogByID(ctx context.Context, id int64) (*OpsErrorLogDetail, error) { + if m.GetErrorLogByIDFn != nil { + return m.GetErrorLogByIDFn(ctx, id) + } return &OpsErrorLogDetail{}, nil } diff --git a/backend/internal/service/ops_service.go b/backend/internal/service/ops_service.go index c651b9184d..0571592368 100644 --- a/backend/internal/service/ops_service.go +++ b/backend/internal/service/ops_service.go @@ -544,24 +544,28 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { return nil } - const maxEvents = 16 events := entry.UpstreamErrors - if len(events) > maxEvents { - events = events[len(events)-maxEvents:] + const maxEventsWithBodies = 16 + firstEventWithBody := len(events) - maxEventsWithBodies + if firstEventWithBody < 0 { + firstEventWithBody = 0 } sanitized := make([]*OpsUpstreamErrorEvent, 0, len(events)) - for _, ev := range events { + for i, ev := range events { if ev == nil { continue } out := *ev + normalizeOpsUpstreamProxyAttribution(&out) out.Platform = truncateString(strings.TrimSpace(out.Platform), 32) out.AccountName = truncateString(strings.TrimSpace(out.AccountName), 128) + out.ProxyName = truncateString(strings.TrimSpace(out.ProxyName), 128) out.UpstreamRequestID = truncateString(strings.TrimSpace(out.UpstreamRequestID), 128) out.UpstreamURL = truncateString(strings.TrimSpace(out.UpstreamURL), 2048) - if body := strings.TrimSpace(out.UpstreamResponseBody); body != "" { + keepBody := i >= firstEventWithBody + if body := strings.TrimSpace(out.UpstreamResponseBody); keepBody && body != "" { out.UpstreamResponseBody, _ = sanitizeErrorBodyForStorage(body, OpsErrorLogQueueBodyMaxBytes) } else { out.UpstreamResponseBody = "" @@ -586,7 +590,7 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { out.Message = msg detail := strings.TrimSpace(out.Detail) - if detail != "" { + if keepBody && detail != "" { // Keep upstream detail small while the event waits in the queue. sanitizedDetail, _ := sanitizeErrorBodyForStorage(detail, OpsErrorLogQueueBodyMaxBytes) out.Detail = sanitizedDetail @@ -594,11 +598,6 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { out.Detail = "" } - // Drop fully-empty events (can happen if only status code was known). - if out.UpstreamStatusCode == 0 && out.Message == "" && out.Detail == "" { - continue - } - evCopy := out sanitized = append(sanitized, &evCopy) } @@ -682,6 +681,11 @@ func (s *OpsService) GetErrorLogByID(ctx context.Context, id int64) (*OpsErrorLo } return nil, infraerrors.InternalServer("OPS_ERROR_LOAD_FAILED", "Failed to load ops error log").WithCause(err) } + if detail != nil && strings.TrimSpace(detail.UpstreamErrors) != "" { + if normalized, normalizeErr := normalizeOpsUpstreamErrorsJSON(detail.UpstreamErrors); normalizeErr == nil { + detail.UpstreamErrors = normalized + } + } return detail, nil } diff --git a/backend/internal/service/ops_upstream_context.go b/backend/internal/service/ops_upstream_context.go index 509620fa1b..f6c17d4236 100644 --- a/backend/internal/service/ops_upstream_context.go +++ b/backend/internal/service/ops_upstream_context.go @@ -365,6 +365,13 @@ type OpsUpstreamErrorEvent struct { AccountID int64 `json:"account_id,omitempty"` AccountName string `json:"account_name,omitempty"` + // Proxy attribution is an immutable, credential-free snapshot of the route + // used by this attempt. ProxyID is null for direct and unknown routes; + // ProxyName distinguishes direct/no_proxy from unknown. + // Never add proxy URLs or credentials to this event. + ProxyID *int64 `json:"proxy_id"` + ProxyName string `json:"proxy_name"` + // Outcome UpstreamStatusCode int `json:"upstream_status_code,omitempty"` UpstreamRequestID string `json:"upstream_request_id,omitempty"` @@ -394,6 +401,11 @@ type OpsUpstreamErrorEvent struct { SkipMonitoring bool `json:"-"` } +const ( + opsProxyNameDirect = "direct/no_proxy" + opsProxyNameUnknown = "unknown" +) + func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) { if c == nil { return @@ -402,6 +414,7 @@ func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) { ev.AtUnixMs = time.Now().UnixMilli() } ev.Platform = strings.TrimSpace(ev.Platform) + normalizeOpsUpstreamProxyAttribution(&ev) ev.UpstreamRequestID = strings.TrimSpace(ev.UpstreamRequestID) ev.UpstreamResponseBody = strings.TrimSpace(ev.UpstreamResponseBody) ev.Kind = strings.TrimSpace(ev.Kind) @@ -429,6 +442,52 @@ func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) { checkSkipMonitoringForUpstreamEvent(c, &evCopy) } +// The event is assembled at the failure site. Forwarding code builds managed +// proxy routes from Proxy, so that event-time object is authoritative. +func opsUpstreamProxyID(account *Account) *int64 { + if account == nil || account.ProxyID == nil || account.Proxy == nil || account.Proxy.ID <= 0 { + return nil + } + proxyID := account.Proxy.ID + return &proxyID +} + +func opsUpstreamProxyName(account *Account) string { + if account == nil { + return opsProxyNameUnknown + } + if account.ProxyID == nil || account.Proxy == nil { + return opsProxyNameDirect + } + if name := strings.TrimSpace(account.Proxy.Name); name != "" { + return name + } + return "proxy" +} + +func setUnknownOpsUpstreamProxy(ev *OpsUpstreamErrorEvent) { + if ev == nil { + return + } + ev.ProxyID = nil + ev.ProxyName = opsProxyNameUnknown +} + +// normalizeOpsUpstreamProxyAttribution makes legacy events explicit without +// pretending that the account's current proxy is historical evidence. +func normalizeOpsUpstreamProxyAttribution(ev *OpsUpstreamErrorEvent) { + if ev == nil { + return + } + if ev.ProxyID != nil && *ev.ProxyID <= 0 { + ev.ProxyID = nil + } + ev.ProxyName = strings.TrimSpace(ev.ProxyName) + if ev.ProxyName == "" { + setUnknownOpsUpstreamProxy(ev) + } +} + // checkSkipMonitoringForUpstreamEvent snapshots whether this attempt matches a // skip_monitoring passthrough rule. The final failure decides whether the // request error is hidden; an intermediate recovered attempt cannot suppress a @@ -479,9 +538,28 @@ func ParseOpsUpstreamErrors(raw string) ([]*OpsUpstreamErrorEvent, error) { if err := json.Unmarshal([]byte(raw), &out); err != nil { return nil, err } + for _, ev := range out { + normalizeOpsUpstreamProxyAttribution(ev) + } return out, nil } +func normalizeOpsUpstreamErrorsJSON(raw string) (string, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return raw, nil + } + events, err := ParseOpsUpstreamErrors(raw) + if err != nil { + return "", err + } + normalized, err := json.Marshal(events) + if err != nil { + return "", err + } + return string(normalized), nil +} + // safeUpstreamURL returns scheme + host + path from a URL, stripping query/fragment // to avoid leaking sensitive query parameters (e.g. OAuth tokens). func safeUpstreamURL(rawURL string) string { diff --git a/backend/internal/service/ops_upstream_context_test.go b/backend/internal/service/ops_upstream_context_test.go index 711223f464..0ecb86e2a5 100644 --- a/backend/internal/service/ops_upstream_context_test.go +++ b/backend/internal/service/ops_upstream_context_test.go @@ -1,8 +1,11 @@ package service import ( + "context" + "net/http/httptest" "testing" + "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) @@ -26,3 +29,196 @@ func TestSafeUpstreamURL(t *testing.T) { }) } } + +func TestOpsUpstreamErrorEventKeepsExplicitProxySnapshotPerAttempt(t *testing.T) { + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + + oldProxyID := int64(10060) + oldProxy := &Proxy{ + ID: oldProxyID, + Name: "wldsg82-ipv6-10060", + Protocol: "socks5", + Host: "old-proxy.example", + Username: "proxy-user", + Password: "proxy-secret", + } + oldAccount := &Account{ + ID: 11, + Name: "old-account", + Platform: PlatformOpenAI, + ProxyID: &oldProxyID, + Proxy: oldProxy, + } + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + AccountID: oldAccount.ID, + ProxyID: opsUpstreamProxyID(oldAccount), + ProxyName: opsUpstreamProxyName(oldAccount), + Kind: "retry", + }) + + newProxyID := int64(8001) + newAccount := &Account{ + ID: 12, + Name: "new-account", + Platform: PlatformOpenAI, + ProxyID: &newProxyID, + Proxy: &Proxy{ID: newProxyID, Name: "oxylabs-uk-8001"}, + } + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + AccountID: newAccount.ID, + ProxyID: opsUpstreamProxyID(newAccount), + ProxyName: opsUpstreamProxyName(newAccount), + Kind: "failover", + }) + + // Mutating the selected account after both attempts must not rewrite history. + oldProxy.Name = "mutated-current-proxy" + oldAccount.ProxyID = &newProxyID + oldAccount.Proxy = newAccount.Proxy + + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := raw.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 2) + + require.Equal(t, int64(10060), *events[0].ProxyID) + require.Equal(t, "wldsg82-ipv6-10060", events[0].ProxyName) + + require.Equal(t, int64(8001), *events[1].ProxyID) + require.Equal(t, "oxylabs-uk-8001", events[1].ProxyName) + + encoded := marshalOpsUpstreamErrors(events) + require.NotNil(t, encoded) + require.NotContains(t, *encoded, "old-proxy.example") + require.NotContains(t, *encoded, "proxy-user") + require.NotContains(t, *encoded, "proxy-secret") +} + +func TestOpsUpstreamProxyFieldAccessors(t *testing.T) { + gin.SetMode(gin.TestMode) + + boundID := int64(7) + backupID := int64(8) + tests := []struct { + name string + account *Account + wantID *int64 + wantName string + }{ + { + name: "actual proxy object is authoritative", + account: &Account{ProxyID: &boundID, Proxy: &Proxy{ID: backupID, Name: "backup-uk"}}, + wantID: &backupID, + wantName: "backup-uk", + }, + { + name: "proxy object without binding id is not a configured route", + account: &Account{Proxy: &Proxy{ID: backupID, Name: "hydrated-proxy"}}, + wantName: opsProxyNameDirect, + }, + { + name: "account has no proxy", + account: &Account{}, + wantName: opsProxyNameDirect, + }, + { + name: "configured proxy was not hydrated and transport is direct", + account: &Account{ProxyID: &backupID}, + wantName: opsProxyNameDirect, + }, + { + name: "missing attempt account", + account: nil, + wantName: opsProxyNameUnknown, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(tt.account), + ProxyName: opsUpstreamProxyName(tt.account), + Kind: "request_error", + }) + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := raw.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 1) + event := events[0] + if tt.wantID == nil { + require.Nil(t, event.ProxyID) + } else { + require.NotNil(t, event.ProxyID) + require.Equal(t, *tt.wantID, *event.ProxyID) + } + require.Equal(t, tt.wantName, event.ProxyName) + }) + } +} + +func TestOpsUpstreamErrorEventRetryKeepsExplicitAttemptProxy(t *testing.T) { + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + proxyID := int64(10060) + account := &Account{ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "retry-proxy"}} + + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Kind: "retry", + }) + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Kind: "retry_exhausted", + }) + + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := raw.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 2) + for _, event := range events { + require.NotNil(t, event.ProxyID) + require.Equal(t, proxyID, *event.ProxyID) + require.Equal(t, "retry-proxy", event.ProxyName) + } +} + +func TestParseOpsUpstreamErrorsMarksLegacyProxyAttributionUnknown(t *testing.T) { + events, err := ParseOpsUpstreamErrors(`[{"account_id":42,"account_name":"legacy","kind":"http_error"}]`) + require.NoError(t, err) + require.Len(t, events, 1) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameUnknown, events[0].ProxyName) +} + +func TestParseOpsUpstreamErrorsPreservesExplicitDirectAttribution(t *testing.T) { + events, err := ParseOpsUpstreamErrors(`[{"proxy_id":null,"proxy_name":"direct/no_proxy","kind":"request_error"}]`) + require.NoError(t, err) + require.Len(t, events, 1) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameDirect, events[0].ProxyName) +} + +func TestOpsServiceGetErrorLogByIDNormalizesLegacyProxyAttribution(t *testing.T) { + repo := &opsRepoMock{ + GetErrorLogByIDFn: func(context.Context, int64) (*OpsErrorLogDetail, error) { + return &OpsErrorLogDetail{UpstreamErrors: `[{"account_id":42,"kind":"http_error"}]`}, nil + }, + } + svc := NewOpsService(repo, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) + + detail, err := svc.GetErrorLogByID(context.Background(), 1) + require.NoError(t, err) + require.NotNil(t, detail) + require.Contains(t, detail.UpstreamErrors, `"proxy_name":"unknown"`) + require.Contains(t, detail.UpstreamErrors, `"proxy_id":null`) + require.NotContains(t, detail.UpstreamErrors, `"proxy_mode"`) + require.NotContains(t, detail.UpstreamErrors, `"proxy_source"`) + require.NotContains(t, detail.UpstreamErrors, `"proxy_fallback"`) +} From 4c1f920d58d6f8b0e2421a42b76f2661fafe4a61 Mon Sep 17 00:00:00 2001 From: Wei-Shaw Date: Thu, 3 Sep 2026 15:11:36 +0800 Subject: [PATCH 2/3] fix(ops): close proxy attribution gaps and bound queued upstream events Follow-up to the event-time proxy snapshot: every upstream attempt now carries attribution, the direct/unknown contract is enforced everywhere, and the queued upstream_errors array has hard bounds again. Attribution coverage - compact fallback retry helper and the streaming compact failover event now stamp proxy_id/proxy_name; the non-streaming SSE compact retry previously dropped its first attempt on `continue` and now records it - raw chat-completions stream truncation events (new on main) attributed - Anthropic/Bedrock transport errors get attribution inside the shared handleUpstreamTransportError helper, so all five callers inherit it - OpenAI first-output timeout takes attribution from the caller: HTTP callers pass the HTTP rule, the WS passthrough relay passes the WS rule (no managed proxy => unknown, never direct, because http.DefaultClient may honor environment proxies) Contract hardening - single opsUpstreamProxyAttribution returns (id, name) from one decision; a hydrated Proxy without a durable ID is unknown, not id=null+name - both normalizers (struct at append time, JSON on detail reads) enforce proxy_id==null <=> sentinel name; blank managed name => "proxy" - detail-read normalization edits only the attribution keys via sjson, so keys from older struct versions and stored key order are preserved Queue bounds - keep at most 256 attempts and ~512 KiB serialized per entry, newest first; the oldest retained event records dropped_earlier_attempts - attempts outside the 16-event body window trim url/message to 512 bytes - restore the empty-event drop rule, judged before the body window clears detail so older real attempts survive Tests cover each site above (managed proxy, direct, WS unknown), the normalizer invariants, key/order preservation, and the count/byte bounds. --- .../upstream-error-proxy-attribution.md | 42 ++++- .../gateway_upstream_transport_error.go | 4 +- .../gateway_upstream_transport_error_test.go | 55 +++++++ .../service/openai_compact_fallback.go | 2 + .../service/openai_compact_fallback_test.go | 126 ++++++++++++++- .../service/openai_first_output_timeout.go | 10 +- .../openai_first_output_timeout_test.go | 26 ++++ ...penai_gateway_chat_completions_raw_test.go | 44 ++++++ .../service/openai_gateway_forward.go | 6 +- .../openai_gateway_response_handling.go | 3 +- .../service/openai_gateway_service.go | 9 +- .../service/openai_raw_stream_truncation.go | 2 + .../service/openai_ws_fallback_test.go | 34 +++++ .../openai_ws_v2_passthrough_adapter.go | 5 + .../service/ops_queue_sanitize_test.go | 143 ++++++++++++++++++ backend/internal/service/ops_service.go | 81 ++++++++-- .../internal/service/ops_upstream_context.go | 127 +++++++++++++--- .../service/ops_upstream_context_test.go | 87 +++++++++++ 18 files changed, 749 insertions(+), 57 deletions(-) diff --git a/backend/dev-docs/upstream-error-proxy-attribution.md b/backend/dev-docs/upstream-error-proxy-attribution.md index d7b4ada37a..1c449962a9 100644 --- a/backend/dev-docs/upstream-error-proxy-attribution.md +++ b/backend/dev-docs/upstream-error-proxy-attribution.md @@ -30,6 +30,12 @@ The two sentinel names have strict meanings: legacy events, pre-transport credential failures, and OpenAI WebSocket use of the default HTTP client. +Invariant: `proxy_id` is `null` if and only if `proxy_name` is one of the two +sentinels. A managed proxy with a blank name (defensive only; `proxies.name` is +non-empty) is labeled `proxy`. Both the struct normalizer used at append time +and the JSON normalizer used on detail reads enforce this, so an event can never +carry `proxy_id=null` together with a real proxy name. + ## Why the account snapshot is event-time evidence Account selection loads `ProxyID` and `Proxy` into one in-memory snapshot. HTTP, @@ -59,6 +65,10 @@ handling. For WebSocket errors without a usable account proxy, the event stores `proxy_id=null` and `proxy_name=unknown`. It does not infer whether the default client selected an environment proxy or a direct connection after the fact. +This applies to every event produced while a request is on the WebSocket +transport, including WS handshake/fallback errors and the first-output timeout +raised by the WS passthrough relay; `opsUpstreamWSProxyAttribution` is the +accessor for those sites and never returns `direct/no_proxy`. Credential acquisition failures that occur before the inference transport is opened are also recorded as unknown. The account's proxy binding is not treated @@ -68,22 +78,40 @@ Each failure site constructs a complete `OpsUpstreamErrorEvent`. Its literal sets `proxy_id` and `proxy_name` from the same route input used by the attempt. Credential-free accessors copy only the managed proxy ID and name. A fallback proxy needs no special event flag: its event-time ID and name are the historical -route evidence. +route evidence. Shared helpers that assemble the event on behalf of a caller +(`handleUpstreamTransportError`, `handleOpenAIUpstreamTransportError`, +`appendOpenAICompactFallbackRetryOps`, `recordOpenAIRawStreamTruncation`, +`newOpenAIFirstOutputTimeoutError`) stamp attribution themselves, so a caller +cannot forget it. Every non-test `OpsUpstreamErrorEvent{` literal in +`backend/internal/service` must either set `ProxyName` or be passed to one of +those helpers; grep for literals missing `ProxyName` before merging changes +that add a failure site. `appendOpsUpstreamError` accepts only the completed event. It does not receive an `Account`, query current account state, or infer proxy attribution while appending. Once appended, the event contains only scalar snapshot values, so later account mutation or failover cannot change an earlier event. -Queue sanitization preserves every non-nil attempt event. To bound asynchronous -queue memory, only the last 16 attempts retain the larger `detail` and -`upstream_response_body` values; older attempts keep their timestamps, account, -proxy, status, kind, message, and other scalar metadata. +Queue sanitization keeps attempt events newest-first under hard bounds: + +- the newest 16 attempts retain `detail` and `upstream_response_body` (each + capped at the queue body limit) plus `upstream_url`/`message` up to 2048 + bytes; +- older attempts keep timestamps, account, proxy, status, kind, stage, scope, + reason and a `upstream_url`/`message` trimmed to 512 bytes; an attempt whose + only payload was a cleared `detail` is still retained; +- at most 256 attempts and roughly 512 KiB of serialized JSON are kept per + request. When either bound drops attempts, the oldest retained event carries + `dropped_earlier_attempts` with the number of discarded earlier attempts. The + newest attempt is always retained because it decides the request outcome. ## Legacy events and analytics -Legacy JSON is not rewritten in the database. Detail reads and the shared parser -materialize missing attribution as: +Legacy JSON is not rewritten in the database. Detail reads edit only the +`proxy_id`/`proxy_name` keys of events that lack valid attribution, so keys +written by older struct versions and the stored key order are preserved. The +shared parser applies the same rule to the decoded structs. Both materialize +missing attribution as: ```json { diff --git a/backend/internal/service/gateway_upstream_transport_error.go b/backend/internal/service/gateway_upstream_transport_error.go index 32873a3ed2..66763b1199 100644 --- a/backend/internal/service/gateway_upstream_transport_error.go +++ b/backend/internal/service/gateway_upstream_transport_error.go @@ -27,7 +27,8 @@ var gatewayTransportFailoverBody = []byte(`{"type":"error","error":{"type":"upst // proxy / DNS / TCP / TLS). It: // 1. records the failure in Ops error logs (status 0, kind=request_error) — // the caller passes path-specific fields (UpstreamURL, Passthrough) via -// event; identity and classification fields are filled here; +// event; identity, proxy attribution and classification fields are +// filled here from the same account snapshot that built the transport; // 2. for durable faults (expired/rejected proxy creds, dead proxy, // DNS/routing) temporarily unschedules the account and logs a stable warn // event that alert rules can key on; @@ -40,6 +41,7 @@ var gatewayTransportFailoverBody = []byte(`{"type":"error","error":{"type":"upst func (s *GatewayService) handleUpstreamTransportError(ctx context.Context, c *gin.Context, account *Account, err error, event OpsUpstreamErrorEvent) error { safeErr := sanitizeUpstreamErrorMessage(err.Error()) setOpsUpstreamError(c, 0, safeErr, "") + event.ProxyID, event.ProxyName = opsUpstreamProxyAttribution(account) event.Platform = account.Platform event.AccountID = account.ID event.AccountName = account.Name diff --git a/backend/internal/service/gateway_upstream_transport_error_test.go b/backend/internal/service/gateway_upstream_transport_error_test.go index c3768522f7..dc41b08b89 100644 --- a/backend/internal/service/gateway_upstream_transport_error_test.go +++ b/backend/internal/service/gateway_upstream_transport_error_test.go @@ -72,6 +72,61 @@ func TestHandleUpstreamTransportError_TransientFailsOverWithoutEviction(t *testi } } +// TestHandleUpstreamTransportError_AttributesEventTimeProxy pins that the +// shared helper stamps proxy attribution from the same account snapshot the +// transport used, so every Anthropic/Bedrock caller inherits it. +func TestHandleUpstreamTransportError_AttributesEventTimeProxy(t *testing.T) { + proxyID := int64(10060) + tests := []struct { + name string + account *Account + wantID *int64 + wantName string + }{ + { + name: "managed proxy", + account: &Account{ID: 149, Name: "acc", Platform: PlatformAnthropic, ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "wldsg82-ipv6-10060"}}, + wantID: &proxyID, + wantName: "wldsg82-ipv6-10060", + }, + { + name: "direct", + account: &Account{ID: 149, Name: "acc", Platform: PlatformAnthropic}, + wantName: opsProxyNameDirect, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := &GatewayService{accountRepo: &transportTempUnschedRepoStub{}} + c := newTransportErrorTestGin(t) + _ = s.handleUpstreamTransportError(context.Background(), c, tt.account, + errors.New("EOF"), OpsUpstreamErrorEvent{UpstreamURL: "https://api.anthropic.com/v1/messages", Passthrough: true}) + raw, ok := c.Get(OpsUpstreamErrorsKey) + if !ok { + t.Fatal("no upstream events recorded") + } + events := raw.([]*OpsUpstreamErrorEvent) + if len(events) != 1 { + t.Fatalf("events = %d, want 1", len(events)) + } + ev := events[0] + if tt.wantID == nil { + if ev.ProxyID != nil { + t.Fatalf("proxy_id = %d, want null", *ev.ProxyID) + } + } else if ev.ProxyID == nil || *ev.ProxyID != *tt.wantID { + t.Fatalf("proxy_id = %v, want %d", ev.ProxyID, *tt.wantID) + } + if ev.ProxyName != tt.wantName { + t.Fatalf("proxy_name = %q, want %q", ev.ProxyName, tt.wantName) + } + if !ev.Passthrough || ev.UpstreamURL == "" || ev.Kind != "request_error" { + t.Fatalf("caller-supplied fields lost: %+v", ev) + } + }) + } +} + // TestHandleUpstreamTransportError_PersistentEvictsAccount pins the contract // for durable faults (dead endpoint / DNS / proxy credentials): fail over AND // temporarily unschedule the account for the transport cooldown. diff --git a/backend/internal/service/openai_compact_fallback.go b/backend/internal/service/openai_compact_fallback.go index 389a4e8beb..70aef1d376 100644 --- a/backend/internal/service/openai_compact_fallback.go +++ b/backend/internal/service/openai_compact_fallback.go @@ -215,6 +215,8 @@ func (s *OpenAIGatewayService) appendOpenAICompactFallbackRetryOps( detail = truncateString(string(payload), maxBytes) } appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, diff --git a/backend/internal/service/openai_compact_fallback_test.go b/backend/internal/service/openai_compact_fallback_test.go index 2e4c7050eb..3db5bb6705 100644 --- a/backend/internal/service/openai_compact_fallback_test.go +++ b/backend/internal/service/openai_compact_fallback_test.go @@ -225,6 +225,120 @@ func TestOpenAIGatewayForwardRetriesExplicitNativeCompactHTTPFailureOnce(t *test require.Equal(t, "retry", events[0].Kind) require.Equal(t, "compact_model_fallback", events[0].Reason) require.Equal(t, http.StatusBadRequest, events[0].UpstreamStatusCode) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameDirect, events[0].ProxyName) +} + +func compactFallbackManagedProxyAccount() (*Account, *Proxy) { + proxy := &Proxy{ID: 10060, Name: "wldsg82-ipv6-10060", Protocol: "http", Host: "proxy.example", Port: 8080} + return &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, + ProxyID: &proxy.ID, Proxy: proxy, + }, proxy +} + +func requireCompactEventsAttributedTo(t *testing.T, events []*OpsUpstreamErrorEvent, proxy *Proxy) { + t.Helper() + for i, ev := range events { + require.NotNil(t, ev.ProxyID, "event %d proxy_id", i) + require.Equal(t, proxy.ID, *ev.ProxyID, "event %d proxy_id", i) + require.Equal(t, proxy.Name, ev.ProxyName, "event %d proxy_name", i) + } +} + +// Non-streaming SSE compact retry: the first attempt's failure must be recorded +// as its own attempt (it previously vanished on `continue`), carrying the +// managed proxy the transport actually used. +func TestOpenAIGatewayForwardNonStreamCompactRetryRecordsAttemptWithManagedProxy(t *testing.T) { + gin.SetMode(gin.TestMode) + body := []byte(`{"model":"gpt-5.5","stream":false,"instructions":"compact-test","input":[{"type":"message","role":"user","content":"hello"},{"type":"compaction_trigger"}]}`) + c := newOpenAICompactFallbackTestContext(t, "/v1/responses") + c.Request.Body = io.NopCloser(bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + MarkOpenAINativeCompactionV2(c) + + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + { + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "X-Request-Id": []string{"rid_compact_1"}}, + Body: io.NopCloser(strings.NewReader("event: response.failed\n" + + `data: {"type":"response.failed","response":{"status":"failed","error":{"code":"context_length_exceeded","message":"context window exceeded"}}}` + "\n\n")), + }, + { + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"resp_compact","status":"completed","model":"gpt-5.4","output":[],"usage":{"input_tokens":1,"output_tokens":1}}`)), + }, + }} + svc := &OpenAIGatewayService{ + cfg: &config.Config{Gateway: config.GatewayConfig{OpenAICompactModel: "gpt-5.4"}}, + httpUpstream: upstream, + } + account, proxy := compactFallbackManagedProxyAccount() + + result, err := svc.Forward(context.Background(), c, account, body) + + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.bodies, 2) + require.Equal(t, proxy.URL(), upstream.lastProxyURL) + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 1) + require.Equal(t, "retry", events[0].Kind) + require.Equal(t, "compact_model_fallback", events[0].Reason) + require.Equal(t, "rid_compact_1", events[0].UpstreamRequestID) + requireCompactEventsAttributedTo(t, events, proxy) +} + +// Streaming compact fallback whose second attempt fails with a failover-class +// error: both the retry event and the failover event must carry the proxy. +func TestOpenAIGatewayForwardCompactFailoverEventCarriesManagedProxy(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) + + contextFailed := "event: response.failed\n" + + `data: {"type":"response.failed","response":{"status":"failed","error":{"code":"context_length_exceeded","message":"context window exceeded"}}}` + "\n\n" + // The fallback model's failure is an account-state error, which + // shouldFailoverOpenAIUpstreamResponse classifies as failover-worthy. + workspaceFailed := "event: response.failed\n" + + `data: {"type":"response.failed","response":{"status":"failed","error":{"code":"deactivated_workspace","message":"workspace deactivated"}}}` + "\n\n" + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(contextFailed))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(workspaceFailed))}, + }} + svc := &OpenAIGatewayService{ + cfg: &config.Config{Gateway: config.GatewayConfig{OpenAICompactModel: "gpt-5.4"}}, + httpUpstream: upstream, + } + account, proxy := compactFallbackManagedProxyAccount() + + result, err := svc.Forward(context.Background(), c, account, body) + + require.Error(t, err) + require.Nil(t, result) + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Len(t, upstream.bodies, 2) + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 2) + require.Equal(t, "retry", events[0].Kind) + require.Equal(t, "compact_model_fallback", events[0].Reason) + require.Equal(t, "failover", events[1].Kind) + requireCompactEventsAttributedTo(t, events, proxy) } func TestOpenAIGatewayForwardRetriesExplicitNativeCompactSSEFailureBeforeOutput(t *testing.T) { @@ -355,6 +469,10 @@ func TestOpenAIGatewayForwardDoesNotRecurseWhenCompactFallbackAlsoFails(t *testi require.Equal(t, "retry", events[0].Kind) require.Equal(t, "compact_model_fallback", events[0].Reason) require.Equal(t, "http_error", events[1].Kind) + for _, ev := range events { + require.Nil(t, ev.ProxyID) + require.Equal(t, opsProxyNameDirect, ev.ProxyName) + } } func TestOpenAIPassthroughCompactFallbackSecondStreamFailureUsesStandardErrorPath(t *testing.T) { @@ -376,11 +494,7 @@ func TestOpenAIPassthroughCompactFallbackSecondStreamFailureUsesStandardErrorPat 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, - } + account, proxy := compactFallbackManagedProxyAccount() result, err := svc.forwardOpenAIPassthrough( context.Background(), c, account, body, body, "gpt-5.5", false, nil, true, time.Now(), @@ -402,4 +516,6 @@ func TestOpenAIPassthroughCompactFallbackSecondStreamFailureUsesStandardErrorPat require.Equal(t, "compact_model_fallback", events[0].Reason) require.Equal(t, "http_error", events[1].Kind) require.True(t, events[1].Passthrough) + require.Equal(t, proxy.URL(), upstream.lastProxyURL) + requireCompactEventsAttributedTo(t, events, proxy) } diff --git a/backend/internal/service/openai_first_output_timeout.go b/backend/internal/service/openai_first_output_timeout.go index 45a4e05d07..3b47fdd7d4 100644 --- a/backend/internal/service/openai_first_output_timeout.go +++ b/backend/internal/service/openai_first_output_timeout.go @@ -243,10 +243,16 @@ func (s *OpenAIGatewayService) openAIFirstOutputTimeout(reasoningEffort string) return time.Duration(seconds) * time.Second } +// newOpenAIFirstOutputTimeoutError records the timeout as an upstream attempt +// and returns the failover error. proxyID/proxyName are supplied by the caller +// because the same deadline is enforced over HTTP and WebSocket transports, +// whose direct-route semantics differ (see opsUpstreamWSProxyAttribution). func (s *OpenAIGatewayService) newOpenAIFirstOutputTimeoutError( ctx context.Context, c *gin.Context, account *Account, + proxyID *int64, + proxyName string, startTime time.Time, originalModel string, reasoningEffort string, @@ -262,8 +268,8 @@ func (s *OpenAIGatewayService) newOpenAIFirstOutputTimeoutError( ) requestID := strings.TrimSpace(responseHeaders.Get("x-request-id")) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - ProxyID: opsUpstreamProxyID(account), - ProxyName: opsUpstreamProxyName(account), + ProxyID: proxyID, + ProxyName: proxyName, Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, UpstreamStatusCode: http.StatusGatewayTimeout, UpstreamRequestID: requestID, Kind: "first_output_timeout", Message: "OpenAI upstream produced no semantic output before the deadline", diff --git a/backend/internal/service/openai_first_output_timeout_test.go b/backend/internal/service/openai_first_output_timeout_test.go index 4ac568c541..d3970413e0 100644 --- a/backend/internal/service/openai_first_output_timeout_test.go +++ b/backend/internal/service/openai_first_output_timeout_test.go @@ -152,6 +152,32 @@ func TestOpenAINativeFirstOutputTimeoutIgnoresPreambleAndCleansReader(t *testing } } +func TestNewOpenAIFirstOutputTimeoutErrorRecordsCallerProxyAttribution(t *testing.T) { + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + svc := &OpenAIGatewayService{} + proxyID := int64(10060) + account := &Account{ID: 1, Name: "acc", Platform: PlatformOpenAI, ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "ws-proxy"}} + + // WebSocket callers pass the WS attribution; HTTP callers pass the HTTP one. + wsID, wsName := opsUpstreamWSProxyAttribution(&Account{ID: 2, Platform: PlatformOpenAI}) + _ = svc.newOpenAIFirstOutputTimeoutError(context.Background(), c, &Account{ID: 2, Name: "ws", Platform: PlatformOpenAI}, + wsID, wsName, time.Now(), "gpt-5.5", "", time.Second, "websocket_first_semantic_output", nil) + _ = svc.newOpenAIFirstOutputTimeoutError(context.Background(), c, account, + opsUpstreamProxyID(account), opsUpstreamProxyName(account), time.Now(), "gpt-5.5", "", time.Second, "semantic_output", nil) + + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events := raw.([]*OpsUpstreamErrorEvent) + require.Len(t, events, 2) + require.Equal(t, "first_output_timeout", events[0].Kind) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameUnknown, events[0].ProxyName) + require.NotNil(t, events[1].ProxyID) + require.Equal(t, proxyID, *events[1].ProxyID) + require.Equal(t, "ws-proxy", events[1].ProxyName) +} + func TestOpenAIFirstOutputTimeoutForReasoningEffort(t *testing.T) { svc := &OpenAIGatewayService{cfg: &config.Config{Gateway: config.GatewayConfig{ OpenAIFirstOutputTimeoutSeconds: 120, diff --git a/backend/internal/service/openai_gateway_chat_completions_raw_test.go b/backend/internal/service/openai_gateway_chat_completions_raw_test.go index 2f0b8dcc08..e0152816d7 100644 --- a/backend/internal/service/openai_gateway_chat_completions_raw_test.go +++ b/backend/internal/service/openai_gateway_chat_completions_raw_test.go @@ -727,6 +727,50 @@ func TestForwardAsRawChatCompletions_TruncatedStreamAfterOutputFailsRequest(t *t // 已写出的内容保持原样透传，客户端拿到的仍是它已经收到的那部分。 require.Contains(t, rec.Body.String(), `"content":"half an ans"`) require.NotContains(t, rec.Body.String(), "data: [DONE]") + + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events := rawEvents.([]*OpsUpstreamErrorEvent) + require.Len(t, events, 1) + require.Equal(t, "http_error", events[0].Kind) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameDirect, events[0].ProxyName) +} + +// 截断 failover 事件必须带上本次传输真实使用的托管代理快照。 +func TestForwardAsRawChatCompletions_TruncationFailoverAttributesManagedProxy(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_empty_proxy"}}, + Body: io.NopCloser(strings.NewReader("")), + }} + svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream} + proxy := &Proxy{ID: 8001, Name: "oxylabs-uk-8001", Protocol: "http", Host: "proxy.example", Port: 8080} + account := rawChatCompletionsTestAccount() + account.ProxyID = &proxy.ID + account.Proxy = proxy + + _, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Equal(t, proxy.URL(), upstream.lastProxyURL) + + rawEvents, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events := rawEvents.([]*OpsUpstreamErrorEvent) + require.Len(t, events, 1) + require.Equal(t, "failover", events[0].Kind) + require.NotNil(t, events[0].ProxyID) + require.Equal(t, proxy.ID, *events[0].ProxyID) + require.Equal(t, proxy.Name, events[0].ProxyName) } // 上游 200 但一个 SSE 字节都没发：响应头尚未提交，应换号重试而不是回 200 空流。 diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index f26b8a5806..119d446ccf 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -1015,7 +1015,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } headerGuard.close() return nil, s.newOpenAIFirstOutputTimeoutError( - ctx, c, account, startTime, originalModel, reasoningEffortValue, + ctx, c, account, opsUpstreamProxyID(account), opsUpstreamProxyName(account), + startTime, originalModel, reasoningEffortValue, firstOutputTimeout, "response_headers", nil, ) } @@ -1175,6 +1176,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco compactResp, compactBody := openAICompactFallbackErrorResponse(resp, signal) if s.shouldFailoverOpenAIUpstreamResponse(compactResp.StatusCode, signal.message, compactBody) { appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, @@ -1206,6 +1209,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if retryBody, fallbackModel, retry := s.prepareOpenAICompactFallbackRetry( c, account, requestedModel, body, http.StatusBadRequest, signal.message, signal.payload, compactModelFallbackRetried, ); retry { + s.appendOpenAICompactFallbackRetryOps(c, account, resp, signal.payload, signal.message, false) body = retryBody requestView = newOpenAIRequestView(body) upstreamModel = fallbackModel diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 104f319dc5..4ee9dca2b1 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -925,7 +925,8 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. markEventProcessed(ev) } return resultWithUsage(), s.newOpenAIFirstOutputTimeoutError( - ctx, c, account, startTime, originalModel, reasoningEffort, + ctx, c, account, opsUpstreamProxyID(account), opsUpstreamProxyName(account), + startTime, originalModel, reasoningEffort, firstOutputTimeout, "semantic_output", resp.Header, ) diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index 8f4c03d3dd..25436fdf4c 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -892,14 +892,7 @@ func (s *OpenAIGatewayService) writeOpenAIWSFallbackErrorResponse(c *gin.Context setOpsUpstreamError(c, statusCode, upstreamMessage, "") if account != nil { - proxyID := opsUpstreamProxyID(account) - proxyName := opsUpstreamProxyName(account) - // The existing WS client uses the configured proxy only when both fields - // are present; otherwise http.DefaultClient may select an environment proxy. - if account.ProxyID == nil || account.Proxy == nil { - proxyID = nil - proxyName = opsProxyNameUnknown - } + proxyID, proxyName := opsUpstreamWSProxyAttribution(account) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ ProxyID: proxyID, ProxyName: proxyName, diff --git a/backend/internal/service/openai_raw_stream_truncation.go b/backend/internal/service/openai_raw_stream_truncation.go index 678d5125b7..56a1739fb4 100644 --- a/backend/internal/service/openai_raw_stream_truncation.go +++ b/backend/internal/service/openai_raw_stream_truncation.go @@ -123,6 +123,8 @@ func recordOpenAIRawStreamTruncation( setOpsUpstreamError(c, http.StatusBadGateway, message, "") appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), Platform: platform, AccountID: accountID, AccountName: accountName, diff --git a/backend/internal/service/openai_ws_fallback_test.go b/backend/internal/service/openai_ws_fallback_test.go index 0d2a47645e..622e609e0f 100644 --- a/backend/internal/service/openai_ws_fallback_test.go +++ b/backend/internal/service/openai_ws_fallback_test.go @@ -184,6 +184,40 @@ func TestWriteOpenAIWSFallbackErrorResponseMarksDefaultClientRouteUnknown(t *tes require.Equal(t, opsProxyNameUnknown, events[0].ProxyName) } +func TestWriteOpenAIWSFallbackErrorResponseKeepsManagedProxy(t *testing.T) { + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + svc := &OpenAIGatewayService{} + proxyID := int64(10060) + account := &Account{ID: 42, Name: "proxied", Platform: PlatformOpenAI, ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "ws-proxy"}} + + require.True(t, svc.writeOpenAIWSFallbackErrorResponse(c, account, wrapOpenAIWSFallback("auth_failed", errors.New("unauthorized")))) + + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events := raw.([]*OpsUpstreamErrorEvent) + require.Len(t, events, 1) + require.NotNil(t, events[0].ProxyID) + require.Equal(t, proxyID, *events[0].ProxyID) + require.Equal(t, "ws-proxy", events[0].ProxyName) +} + +func TestOpsUpstreamWSProxyAttributionNeverReportsDirect(t *testing.T) { + proxyID := int64(7) + id, name := opsUpstreamWSProxyAttribution(nil) + require.Nil(t, id) + require.Equal(t, opsProxyNameUnknown, name) + + id, name = opsUpstreamWSProxyAttribution(&Account{}) + require.Nil(t, id) + require.Equal(t, opsProxyNameUnknown, name, "no managed proxy => http.DefaultClient => unknown, never direct") + + id, name = opsUpstreamWSProxyAttribution(&Account{ProxyID: &proxyID, Proxy: &Proxy{ID: proxyID, Name: "ws-proxy"}}) + require.NotNil(t, id) + require.Equal(t, proxyID, *id) + require.Equal(t, "ws-proxy", name) +} + func TestOpenAIWSFallbackCooling(t *testing.T) { svc := &OpenAIGatewayService{cfg: &config.Config{}} svc.cfg.Gateway.OpenAIWS.FallbackCooldownSeconds = 1 diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index 715a747918..8d29839b3b 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -1396,10 +1396,15 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( var firstOutputTimeoutErr *openAIWSPassthroughFirstOutputTimeoutError if errors.As(relayErr, &firstOutputTimeoutErr) { deadline := firstOutputTimeoutErr.deadline + // The relay ran over the WebSocket transport, so a missing managed + // proxy is an unknown route (http.DefaultClient), not a direct one. + wsProxyID, wsProxyName := opsUpstreamWSProxyAttribution(account) failoverErr := s.newOpenAIFirstOutputTimeoutError( ctx, c, account, + wsProxyID, + wsProxyName, deadline.startedAt, deadline.requestModel, deadline.reasoningEffort, diff --git a/backend/internal/service/ops_queue_sanitize_test.go b/backend/internal/service/ops_queue_sanitize_test.go index 341c54e6b2..9a1a1cd646 100644 --- a/backend/internal/service/ops_queue_sanitize_test.go +++ b/backend/internal/service/ops_queue_sanitize_test.go @@ -45,6 +45,12 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) { if len(event.Platform) > 32 || len(event.AccountName) > 128 || len(event.UpstreamURL) > 2048 || len(event.Message) > 2048 { t.Fatalf("event fields were not bounded: %+v", event) } + if i < 4 && (len(event.UpstreamURL) > opsUpstreamErrorsOlderURLMaxLen || len(event.Message) > opsUpstreamErrorsOlderMessageMaxLen) { + t.Fatalf("older event %d kept a full-size url/message: url=%d message=%d", i, len(event.UpstreamURL), len(event.Message)) + } + if event.DroppedEarlierAttempts != 0 { + t.Fatalf("event %d reported dropped attempts without any drop: %d", i, event.DroppedEarlierAttempts) + } if len(event.UpstreamResponseBody) > OpsErrorLogQueueBodyMaxBytes || len(event.Detail) > OpsErrorLogQueueBodyMaxBytes { t.Fatal("event body/detail exceeded queue limit") } @@ -56,3 +62,140 @@ func TestSanitizeOpsUpstreamErrorsForQueueBoundsAndRedacts(t *testing.T) { } } } + +func TestSanitizeOpsUpstreamErrorsForQueueHardBoundsEventCount(t *testing.T) { + entry := &OpsInsertErrorLogInput{} + total := opsUpstreamErrorsMaxEvents + 40 + for i := 0; i < total; i++ { + entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{ + AtUnixMs: int64(i + 1), + ProxyName: opsProxyNameDirect, + UpstreamStatusCode: 500, + Message: "attempt", + }) + } + if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil { + t.Fatal(err) + } + events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON) + if err != nil { + t.Fatal(err) + } + if len(events) != opsUpstreamErrorsMaxEvents { + t.Fatalf("event count = %d, want %d", len(events), opsUpstreamErrorsMaxEvents) + } + // Newest attempts win; the oldest retained one carries the drop count. + if events[len(events)-1].AtUnixMs != int64(total) { + t.Fatalf("newest event lost: at=%d want %d", events[len(events)-1].AtUnixMs, total) + } + if events[0].AtUnixMs != int64(total-opsUpstreamErrorsMaxEvents+1) { + t.Fatalf("oldest retained at=%d", events[0].AtUnixMs) + } + if events[0].DroppedEarlierAttempts != 40 { + t.Fatalf("dropped_earlier_attempts = %d, want 40", events[0].DroppedEarlierAttempts) + } + if events[1].DroppedEarlierAttempts != 0 { + t.Fatal("drop count must only be stamped on the oldest retained event") + } +} + +func TestSanitizeOpsUpstreamErrorsForQueueHardBoundsSerializedBytes(t *testing.T) { + entry := &OpsInsertErrorLogInput{} + // Every bounded field at its cap: the 16 newest attempts weigh ~20KB each + // (8KB body + 8KB detail + 2KB url + 2KB message), older ones ~2KB. At the + // count cap that is well past the byte budget, so bytes must win here. + const total = opsUpstreamErrorsMaxEvents + for i := 0; i < total; i++ { + entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{ + AtUnixMs: int64(i + 1), + Platform: strings.Repeat("p", 40), + AccountName: strings.Repeat("a", 200), + ProxyName: strings.Repeat("n", 200), + UpstreamRequestID: strings.Repeat("r", 200), + UpstreamStatusCode: 502, + UpstreamURL: "https://example.com/" + strings.Repeat("u", 3000), + // Plain-text payloads are truncated, not JSON-trimmed, so they keep + // their full queue-limit weight. + UpstreamResponseBody: strings.Repeat("b", 10_000), + Kind: strings.Repeat("k", 80), + Stage: strings.Repeat("s", 80), + Scope: strings.Repeat("c", 80), + Reason: strings.Repeat("e", 200), + Message: strings.Repeat("m", 3000), + Detail: strings.Repeat("d", 10_000), + }) + } + if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil { + t.Fatal(err) + } + // The newest event is always kept even when it alone exceeds the remaining + // budget, so allow one max-size event of slack. + if len(*entry.UpstreamErrorsJSON) > opsUpstreamErrorsQueueMaxBytes+24*1024 { + t.Fatalf("serialized upstream_errors = %d bytes, budget %d", len(*entry.UpstreamErrorsJSON), opsUpstreamErrorsQueueMaxBytes) + } + events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON) + if err != nil { + t.Fatal(err) + } + if len(events) == 0 || len(events) >= total { + t.Fatalf("expected the byte budget to drop some events, kept %d of %d", len(events), total) + } + if events[len(events)-1].AtUnixMs != int64(total) { + t.Fatal("newest attempt must always be retained") + } + if events[0].DroppedEarlierAttempts != total-len(events) { + t.Fatalf("dropped_earlier_attempts = %d, want %d", events[0].DroppedEarlierAttempts, total-len(events)) + } +} + +func TestSanitizeOpsUpstreamErrorsForQueueDropsEmptyButKeepsOlderTrimmedEvents(t *testing.T) { + entry := &OpsInsertErrorLogInput{} + // 20 events: the first 4 fall outside the body window. Event 0 is a real + // attempt whose only payload is a detail (status 0, no message); it must + // survive the window even though its detail is cleared. Event 1 is truly + // empty and must be dropped. + for i := 0; i < 20; i++ { + ev := &OpsUpstreamErrorEvent{AtUnixMs: int64(i + 1), ProxyName: opsProxyNameDirect} + switch i { + case 0: + ev.Detail = "proxy handshake failed" + case 1: + // fully empty + default: + ev.UpstreamStatusCode = 500 + } + entry.UpstreamErrors = append(entry.UpstreamErrors, ev) + } + if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil { + t.Fatal(err) + } + events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON) + if err != nil { + t.Fatal(err) + } + if len(events) != 19 { + t.Fatalf("event count = %d, want 19", len(events)) + } + if events[0].AtUnixMs != 1 || events[0].Detail != "" { + t.Fatalf("detail-only older attempt must be kept with detail cleared: %+v", events[0]) + } + if events[1].AtUnixMs != 3 { + t.Fatalf("fully-empty attempt must be dropped, got at=%d", events[1].AtUnixMs) + } +} + +func TestSanitizeOpsUpstreamErrorsForQueueIgnoresCallerSuppliedDropCount(t *testing.T) { + entry := &OpsInsertErrorLogInput{UpstreamErrors: []*OpsUpstreamErrorEvent{ + {UpstreamStatusCode: 500, ProxyName: opsProxyNameDirect, DroppedEarlierAttempts: 99}, + }} + if err := SanitizeOpsUpstreamErrorsForQueue(entry); err != nil { + t.Fatal(err) + } + events, err := ParseOpsUpstreamErrors(*entry.UpstreamErrorsJSON) + if err != nil { + t.Fatal(err) + } + if len(events) != 1 || events[0].DroppedEarlierAttempts != 0 { + t.Fatalf("caller-supplied drop count must be reset: %+v", events[0]) + } +} diff --git a/backend/internal/service/ops_service.go b/backend/internal/service/ops_service.go index 0571592368..0e110658b1 100644 --- a/backend/internal/service/ops_service.go +++ b/backend/internal/service/ops_service.go @@ -539,32 +539,53 @@ func (s *OpsService) prepareErrorLogInput(ctx context.Context, entry *OpsInsertE return entry, true, nil } +const ( + // opsUpstreamErrorsBodyWindow is how many of the newest attempts keep their + // larger detail / upstream_response_body payloads while queued. + opsUpstreamErrorsBodyWindow = 16 + // Attempts older than the body window keep scalar metadata only, with URL + // and message trimmed harder than the newest attempts (2048). + opsUpstreamErrorsOlderURLMaxLen = 512 + opsUpstreamErrorsOlderMessageMaxLen = 512 + // Hard bounds for the upstream_errors array of one queued entry. Newest + // attempts win; the oldest retained attempt records the dropped count. + opsUpstreamErrorsMaxEvents = 256 + opsUpstreamErrorsQueueMaxBytes = 512 * 1024 +) + func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { if entry == nil || len(entry.UpstreamErrors) == 0 { return nil } - events := entry.UpstreamErrors - const maxEventsWithBodies = 16 - firstEventWithBody := len(events) - maxEventsWithBodies + events := make([]*OpsUpstreamErrorEvent, 0, len(entry.UpstreamErrors)) + for _, ev := range entry.UpstreamErrors { + if ev != nil { + events = append(events, ev) + } + } + firstEventWithBody := len(events) - opsUpstreamErrorsBodyWindow if firstEventWithBody < 0 { firstEventWithBody = 0 } sanitized := make([]*OpsUpstreamErrorEvent, 0, len(events)) for i, ev := range events { - if ev == nil { - continue - } out := *ev normalizeOpsUpstreamProxyAttribution(&out) + // Only boundOpsUpstreamErrors may stamp this; never trust caller input. + out.DroppedEarlierAttempts = 0 + keepBody := i >= firstEventWithBody + urlMaxLen, messageMaxLen := 2048, 2048 + if !keepBody { + urlMaxLen, messageMaxLen = opsUpstreamErrorsOlderURLMaxLen, opsUpstreamErrorsOlderMessageMaxLen + } out.Platform = truncateString(strings.TrimSpace(out.Platform), 32) out.AccountName = truncateString(strings.TrimSpace(out.AccountName), 128) out.ProxyName = truncateString(strings.TrimSpace(out.ProxyName), 128) out.UpstreamRequestID = truncateString(strings.TrimSpace(out.UpstreamRequestID), 128) - out.UpstreamURL = truncateString(strings.TrimSpace(out.UpstreamURL), 2048) - keepBody := i >= firstEventWithBody + out.UpstreamURL = truncateString(strings.TrimSpace(out.UpstreamURL), urlMaxLen) if body := strings.TrimSpace(out.UpstreamResponseBody); keepBody && body != "" { out.UpstreamResponseBody, _ = sanitizeErrorBodyForStorage(body, OpsErrorLogQueueBodyMaxBytes) } else { @@ -586,10 +607,16 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { } msg := sanitizeUpstreamErrorMessage(strings.TrimSpace(out.Message)) - msg = truncateString(msg, 2048) + msg = truncateString(msg, messageMaxLen) out.Message = msg detail := strings.TrimSpace(out.Detail) + // Drop fully-empty events (can happen if only status code was known). + // Judged on the original detail so an older attempt whose detail is + // cleared by the body window below is still retained. + if out.UpstreamStatusCode == 0 && out.Message == "" && detail == "" { + continue + } if keepBody && detail != "" { // Keep upstream detail small while the event waits in the queue. sanitizedDetail, _ := sanitizeErrorBodyForStorage(detail, OpsErrorLogQueueBodyMaxBytes) @@ -602,11 +629,47 @@ func sanitizeOpsUpstreamErrors(entry *OpsInsertErrorLogInput) error { sanitized = append(sanitized, &evCopy) } + sanitized, _ = boundOpsUpstreamErrors(sanitized) entry.UpstreamErrorsJSON = marshalOpsUpstreamErrors(sanitized) entry.UpstreamErrors = nil return nil } +// boundOpsUpstreamErrors enforces the per-entry event count and serialized +// byte budget. It walks from the newest attempt backwards so the final, +// outcome-deciding attempts are always retained (the newest event is kept even +// if it alone exceeds the budget), and stamps the oldest retained event with +// the number of dropped earlier attempts. Events must already be sanitized +// copies owned by the caller. +func boundOpsUpstreamErrors(events []*OpsUpstreamErrorEvent) ([]*OpsUpstreamErrorEvent, int) { + if len(events) == 0 { + return events, 0 + } + keepFrom := len(events) + budget := opsUpstreamErrorsQueueMaxBytes + for i := len(events) - 1; i >= 0; i-- { + if len(events)-i > opsUpstreamErrorsMaxEvents { + break + } + raw, err := json.Marshal(events[i]) + size := 0 + if err == nil { + size = len(raw) + 1 // trailing comma / bracket share + } + if i != len(events)-1 && size > budget { + break + } + budget -= size + keepFrom = i + } + kept := events[keepFrom:] + dropped := keepFrom + if dropped > 0 { + kept[0].DroppedEarlierAttempts = dropped + } + return kept, dropped +} + func (s *OpsService) GetErrorLogs(ctx context.Context, filter *OpsErrorLogFilter) (*OpsErrorLogList, error) { if err := s.RequireMonitoringEnabled(ctx); err != nil { return nil, err diff --git a/backend/internal/service/ops_upstream_context.go b/backend/internal/service/ops_upstream_context.go index f6c17d4236..3e9e0b636a 100644 --- a/backend/internal/service/ops_upstream_context.go +++ b/backend/internal/service/ops_upstream_context.go @@ -2,11 +2,15 @@ package service import ( "encoding/json" + "errors" + "strconv" "strings" "time" "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey" "github.com/gin-gonic/gin" + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" ) // Gin context keys used by Ops error logger for capturing upstream error details. @@ -372,6 +376,10 @@ type OpsUpstreamErrorEvent struct { ProxyID *int64 `json:"proxy_id"` ProxyName string `json:"proxy_name"` + // DroppedEarlierAttempts is set on the oldest retained event when queue + // bounds forced earlier attempts of the same request to be discarded. + DroppedEarlierAttempts int `json:"dropped_earlier_attempts,omitempty"` + // Outcome UpstreamStatusCode int `json:"upstream_status_code,omitempty"` UpstreamRequestID string `json:"upstream_request_id,omitempty"` @@ -404,6 +412,9 @@ type OpsUpstreamErrorEvent struct { const ( opsProxyNameDirect = "direct/no_proxy" opsProxyNameUnknown = "unknown" + // opsProxyNameUnnamed labels a managed proxy whose name is blank. The + // proxies.name column is NOT NULL/non-empty, so this is a defensive value. + opsProxyNameUnnamed = "proxy" ) func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) { @@ -442,27 +453,57 @@ func appendOpsUpstreamError(c *gin.Context, ev OpsUpstreamErrorEvent) { checkSkipMonitoringForUpstreamEvent(c, &evCopy) } -// The event is assembled at the failure site. Forwarding code builds managed -// proxy routes from Proxy, so that event-time object is authoritative. -func opsUpstreamProxyID(account *Account) *int64 { - if account == nil || account.ProxyID == nil || account.Proxy == nil || account.Proxy.ID <= 0 { - return nil +// opsUpstreamProxyAttribution derives both attribution fields from one +// decision so they can never disagree. The event is assembled at the failure +// site; forwarding code builds managed proxy routes from Proxy only when the +// binding ID is also set, so the same rule decides the label here: +// +// - nil account -> (nil, unknown) +// - no binding or no hydrated Proxy -> (nil, direct/no_proxy) +// - hydrated Proxy without durable ID -> (nil, unknown): the transport did +// use that proxy, but nothing durable identifies it +// - otherwise -> (Proxy.ID, Proxy.Name or "proxy") +// +// Invariant: proxy_id == null implies proxy_name is one of the two sentinels. +func opsUpstreamProxyAttribution(account *Account) (*int64, string) { + if account == nil { + return nil, opsProxyNameUnknown + } + if account.ProxyID == nil || account.Proxy == nil { + return nil, opsProxyNameDirect + } + if account.Proxy.ID <= 0 { + return nil, opsProxyNameUnknown } proxyID := account.Proxy.ID - return &proxyID + name := strings.TrimSpace(account.Proxy.Name) + if name == "" { + name = opsProxyNameUnnamed + } + return &proxyID, name +} + +func opsUpstreamProxyID(account *Account) *int64 { + proxyID, _ := opsUpstreamProxyAttribution(account) + return proxyID } func opsUpstreamProxyName(account *Account) string { - if account == nil { - return opsProxyNameUnknown + _, name := opsUpstreamProxyAttribution(account) + return name +} + +// opsUpstreamWSProxyAttribution is the OpenAI WebSocket variant. The WS dialer +// sets an explicit proxy client only when the account has a usable managed +// proxy; otherwise coder/websocket falls back to http.DefaultClient, which +// honors HTTP_PROXY/HTTPS_PROXY/NO_PROXY. A missing managed proxy is therefore +// unknown, not direct. +func opsUpstreamWSProxyAttribution(account *Account) (*int64, string) { + proxyID, name := opsUpstreamProxyAttribution(account) + if proxyID == nil { + return nil, opsProxyNameUnknown } - if account.ProxyID == nil || account.Proxy == nil { - return opsProxyNameDirect - } - if name := strings.TrimSpace(account.Proxy.Name); name != "" { - return name - } - return "proxy" + return proxyID, name } func setUnknownOpsUpstreamProxy(ev *OpsUpstreamErrorEvent) { @@ -480,10 +521,19 @@ func normalizeOpsUpstreamProxyAttribution(ev *OpsUpstreamErrorEvent) { return } if ev.ProxyID != nil && *ev.ProxyID <= 0 { + // A non-positive ID never identifies a managed proxy. ev.ProxyID = nil } ev.ProxyName = strings.TrimSpace(ev.ProxyName) - if ev.ProxyName == "" { + if ev.ProxyID != nil { + if ev.ProxyName == "" { + ev.ProxyName = opsProxyNameUnnamed + } + return + } + // Invariant: proxy_id == null implies a sentinel name. Any other name + // without a durable ID is not historical evidence of a managed route. + if ev.ProxyName != opsProxyNameDirect { setUnknownOpsUpstreamProxy(ev) } } @@ -544,20 +594,51 @@ func ParseOpsUpstreamErrors(raw string) ([]*OpsUpstreamErrorEvent, error) { return out, nil } +// normalizeOpsUpstreamErrorsJSON materializes missing proxy attribution on +// stored JSON for detail reads. It edits only the attribution keys of events +// that lack them, so keys written by older struct versions and the original +// key order survive; nothing is re-marshaled through the current struct. func normalizeOpsUpstreamErrorsJSON(raw string) (string, error) { raw = strings.TrimSpace(raw) if raw == "" { return raw, nil } - events, err := ParseOpsUpstreamErrors(raw) - if err != nil { - return "", err + if !gjson.Valid(raw) { + return "", errors.New("upstream_errors is not valid JSON") } - normalized, err := json.Marshal(events) - if err != nil { - return "", err + parsed := gjson.Parse(raw) + if !parsed.IsArray() { + return "", errors.New("upstream_errors is not a JSON array") } - return string(normalized), nil + out := raw + for i, ev := range parsed.Array() { + if !ev.IsObject() { + continue + } + prefix := strconv.Itoa(i) + "." + proxyID := ev.Get("proxy_id") + proxyName := strings.TrimSpace(ev.Get("proxy_name").String()) + hasValidID := proxyID.Exists() && proxyID.Type == gjson.Number && proxyID.Int() > 0 + var err error + switch { + case hasValidID: + if proxyName == "" { + out, err = sjson.Set(out, prefix+"proxy_name", opsProxyNameUnnamed) + } + case proxyName == opsProxyNameDirect: + if !proxyID.Exists() || proxyID.Type != gjson.Null { + out, err = sjson.Set(out, prefix+"proxy_id", nil) + } + default: + if out, err = sjson.Set(out, prefix+"proxy_id", nil); err == nil { + out, err = sjson.Set(out, prefix+"proxy_name", opsProxyNameUnknown) + } + } + if err != nil { + return "", err + } + } + return out, nil } // safeUpstreamURL returns scheme + host + path from a URL, stripping query/fragment diff --git a/backend/internal/service/ops_upstream_context_test.go b/backend/internal/service/ops_upstream_context_test.go index 0ecb86e2a5..6e537cc461 100644 --- a/backend/internal/service/ops_upstream_context_test.go +++ b/backend/internal/service/ops_upstream_context_test.go @@ -3,6 +3,7 @@ package service import ( "context" "net/http/httptest" + "strings" "testing" "github.com/gin-gonic/gin" @@ -118,6 +119,17 @@ func TestOpsUpstreamProxyFieldAccessors(t *testing.T) { account: &Account{Proxy: &Proxy{ID: backupID, Name: "hydrated-proxy"}}, wantName: opsProxyNameDirect, }, + { + name: "hydrated proxy without durable id is unknown, never id=null plus a real name", + account: &Account{ProxyID: &boundID, Proxy: &Proxy{Name: "transient-proxy"}}, + wantName: opsProxyNameUnknown, + }, + { + name: "blank proxy name gets the unnamed placeholder", + account: &Account{ProxyID: &boundID, Proxy: &Proxy{ID: boundID, Name: " "}}, + wantID: &boundID, + wantName: opsProxyNameUnnamed, + }, { name: "account has no proxy", account: &Account{}, @@ -189,6 +201,81 @@ func TestOpsUpstreamErrorEventRetryKeepsExplicitAttemptProxy(t *testing.T) { } } +func TestNormalizeOpsUpstreamProxyAttributionEnforcesSentinelInvariant(t *testing.T) { + positive := int64(7) + zero := int64(0) + tests := []struct { + name string + in OpsUpstreamErrorEvent + wantID *int64 + wantName string + }{ + {name: "id with name kept", in: OpsUpstreamErrorEvent{ProxyID: &positive, ProxyName: "eu-1"}, wantID: &positive, wantName: "eu-1"}, + {name: "id without name gets placeholder", in: OpsUpstreamErrorEvent{ProxyID: &positive}, wantID: &positive, wantName: opsProxyNameUnnamed}, + {name: "null id with real name collapses to unknown", in: OpsUpstreamErrorEvent{ProxyName: "eu-1"}, wantName: opsProxyNameUnknown}, + {name: "zero id with real name collapses to unknown", in: OpsUpstreamErrorEvent{ProxyID: &zero, ProxyName: "eu-1"}, wantName: opsProxyNameUnknown}, + {name: "explicit direct preserved", in: OpsUpstreamErrorEvent{ProxyName: opsProxyNameDirect}, wantName: opsProxyNameDirect}, + {name: "empty is unknown", in: OpsUpstreamErrorEvent{}, wantName: opsProxyNameUnknown}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ev := tt.in + normalizeOpsUpstreamProxyAttribution(&ev) + if tt.wantID == nil { + require.Nil(t, ev.ProxyID) + } else { + require.NotNil(t, ev.ProxyID) + require.Equal(t, *tt.wantID, *ev.ProxyID) + } + require.Equal(t, tt.wantName, ev.ProxyName) + }) + } +} + +func TestNormalizeOpsUpstreamErrorsJSONPreservesUnknownKeysAndOrder(t *testing.T) { + // A legacy row written by an older struct version: it carries a key the + // current struct no longer has, and mixes explicit/legacy attribution. + raw := `[` + + `{"at_unix_ms":1,"account_id":42,"upstream_request_body":"legacy-only-key","kind":"http_error"},` + + `{"at_unix_ms":2,"proxy_id":null,"proxy_name":"direct/no_proxy","kind":"request_error"},` + + `{"at_unix_ms":3,"proxy_id":9,"proxy_name":"eu-9","kind":"failover"},` + + `{"at_unix_ms":4,"proxy_id":0,"proxy_name":"stale-name","kind":"retry"},` + + `{"at_unix_ms":5,"proxy_id":11,"proxy_name":"","kind":"retry"}` + + `]` + + out, err := normalizeOpsUpstreamErrorsJSON(raw) + require.NoError(t, err) + + // Legacy-only key and original key order survive. + require.Contains(t, out, `"upstream_request_body":"legacy-only-key"`) + require.True(t, strings.Index(out, `"at_unix_ms":1`) < strings.Index(out, `"account_id":42`)) + + events, err := ParseOpsUpstreamErrors(out) + require.NoError(t, err) + require.Len(t, events, 5) + require.Nil(t, events[0].ProxyID) + require.Equal(t, opsProxyNameUnknown, events[0].ProxyName) + require.Nil(t, events[1].ProxyID) + require.Equal(t, opsProxyNameDirect, events[1].ProxyName) + require.NotNil(t, events[2].ProxyID) + require.Equal(t, int64(9), *events[2].ProxyID) + require.Equal(t, "eu-9", events[2].ProxyName) + require.Nil(t, events[3].ProxyID) + require.Equal(t, opsProxyNameUnknown, events[3].ProxyName) + require.NotNil(t, events[4].ProxyID) + require.Equal(t, opsProxyNameUnnamed, events[4].ProxyName) + + // Already-normalized input is returned byte-for-byte. + again, err := normalizeOpsUpstreamErrorsJSON(out) + require.NoError(t, err) + require.Equal(t, out, again) + + _, err = normalizeOpsUpstreamErrorsJSON(`{"not":"an array"}`) + require.Error(t, err) + _, err = normalizeOpsUpstreamErrorsJSON(`[not json`) + require.Error(t, err) +} + func TestParseOpsUpstreamErrorsMarksLegacyProxyAttributionUnknown(t *testing.T) { events, err := ParseOpsUpstreamErrors(`[{"account_id":42,"account_name":"legacy","kind":"http_error"}]`) require.NoError(t, err) From abc07bb0799c7c8cc15795898d4acec247c7a59c Mon Sep 17 00:00:00 2001 From: Wei-Shaw Date: Thu, 3 Sep 2026 15:23:22 +0800 Subject: [PATCH 3/3] test(ops): satisfy errcheck and gofmt in proxy attribution tests --- .../service/openai_first_output_timeout_test.go | 3 ++- .../internal/service/openai_ws_fallback_test.go | 3 ++- .../internal/service/ops_queue_sanitize_test.go | 14 +++++++------- 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/backend/internal/service/openai_first_output_timeout_test.go b/backend/internal/service/openai_first_output_timeout_test.go index d3970413e0..fc4b5076ca 100644 --- a/backend/internal/service/openai_first_output_timeout_test.go +++ b/backend/internal/service/openai_first_output_timeout_test.go @@ -168,7 +168,8 @@ func TestNewOpenAIFirstOutputTimeoutErrorRecordsCallerProxyAttribution(t *testin raw, ok := c.Get(OpsUpstreamErrorsKey) require.True(t, ok) - events := raw.([]*OpsUpstreamErrorEvent) + events, ok := raw.([]*OpsUpstreamErrorEvent) + require.True(t, ok) require.Len(t, events, 2) require.Equal(t, "first_output_timeout", events[0].Kind) require.Nil(t, events[0].ProxyID) diff --git a/backend/internal/service/openai_ws_fallback_test.go b/backend/internal/service/openai_ws_fallback_test.go index 622e609e0f..b9f5c39b64 100644 --- a/backend/internal/service/openai_ws_fallback_test.go +++ b/backend/internal/service/openai_ws_fallback_test.go @@ -195,7 +195,8 @@ func TestWriteOpenAIWSFallbackErrorResponseKeepsManagedProxy(t *testing.T) { raw, ok := c.Get(OpsUpstreamErrorsKey) require.True(t, ok) - events := raw.([]*OpsUpstreamErrorEvent) + events, ok := raw.([]*OpsUpstreamErrorEvent) + require.True(t, ok) require.Len(t, events, 1) require.NotNil(t, events[0].ProxyID) require.Equal(t, proxyID, *events[0].ProxyID) diff --git a/backend/internal/service/ops_queue_sanitize_test.go b/backend/internal/service/ops_queue_sanitize_test.go index 9a1a1cd646..8ace57d9bb 100644 --- a/backend/internal/service/ops_queue_sanitize_test.go +++ b/backend/internal/service/ops_queue_sanitize_test.go @@ -107,13 +107,13 @@ func TestSanitizeOpsUpstreamErrorsForQueueHardBoundsSerializedBytes(t *testing.T const total = opsUpstreamErrorsMaxEvents for i := 0; i < total; i++ { entry.UpstreamErrors = append(entry.UpstreamErrors, &OpsUpstreamErrorEvent{ - AtUnixMs: int64(i + 1), - Platform: strings.Repeat("p", 40), - AccountName: strings.Repeat("a", 200), - ProxyName: strings.Repeat("n", 200), - UpstreamRequestID: strings.Repeat("r", 200), - UpstreamStatusCode: 502, - UpstreamURL: "https://example.com/" + strings.Repeat("u", 3000), + AtUnixMs: int64(i + 1), + Platform: strings.Repeat("p", 40), + AccountName: strings.Repeat("a", 200), + ProxyName: strings.Repeat("n", 200), + UpstreamRequestID: strings.Repeat("r", 200), + UpstreamStatusCode: 502, + UpstreamURL: "https://example.com/" + strings.Repeat("u", 3000), // Plain-text payloads are truncated, not JSON-trimmed, so they keep // their full queue-limit weight. UpstreamResponseBody: strings.Repeat("b", 10_000),