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