diff --git a/src/lib/errors.ts b/src/lib/errors.ts index 7f8839eeb86..18b27d34827 100644 --- a/src/lib/errors.ts +++ b/src/lib/errors.ts @@ -256,6 +256,16 @@ export function isClientClosedMessage(text: string): boolean { ); } +/** + * Ambiguous-reset refusal wording owned by this proxy (src/lib/upstream-retry.ts): + * the upstream connection closed before any response arrived, so the request may + * already have been processed and automatic replay was stopped. Matched narrowly + * so a provider-sent message is never relabeled by it. + */ +export function isUpstreamResetReplayRefusedMessage(text: string): boolean { + return text.toLowerCase().includes("connection closed before a response was received"); +} + export function classifyError(status: number, type: string, message: string): OcxErrorPayload { const text = message.toLowerCase(); if (type === "previous_response_not_found") { diff --git a/src/providers/key-failover.ts b/src/providers/key-failover.ts index 3235cfb97bd..e4552df141a 100644 --- a/src/providers/key-failover.ts +++ b/src/providers/key-failover.ts @@ -298,7 +298,7 @@ const OPENCODE_GO_RATE_LIMIT_RETRY = { } as const satisfies Required; /** True when the provider row points at the OpenCode Go destination. */ -function isOpenCodeGoDestination( +export function isOpenCodeGoDestination( provider: Partial>, ): boolean { const raw = typeof provider.baseUrl === "string" ? provider.baseUrl : ""; diff --git a/src/server/request-log.ts b/src/server/request-log.ts index 176ec0c5a2c..f3d73ba8984 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -12,6 +12,7 @@ import { isCyberPolicyCode, isCyberPolicyMessage, isRateLimitOrQuotaFailureMessage, + isUpstreamResetReplayRefusedMessage, upstreamErrorMessageFromPayload, } from "../lib/errors"; import { CODEX_CONFIG_PATH, readRootTomlString } from "../codex/paths"; @@ -22,6 +23,7 @@ import { normalizeRouteDecisionTrace, type RouteDecisionTraceV1 } from "../routi import type { AdapterRequest } from "../adapters/base"; import type { RequestSpendSettlement } from "./responses/request-spend"; import type { AdapterTierMetadata } from "../providers/fastwire"; +import { UPSTREAM_RESET_REPLAY_REFUSED_CODE } from "../lib/upstream-retry"; import { redactSecretString, sanitizeLogMetadataString } from "../lib/redact"; import { appendUsageEntry, @@ -729,7 +731,15 @@ export function requestLogErrorCode( } return "permission_denied"; } - if (status === 429) return "rate_limit_exceeded"; + if (status === 429) { + // A refused ambiguous reset answers 429 by design (it must not invite a client + // retry that could duplicate inference); classify it by its message so the log + // distinguishes a proxy refusal from provider throttling. + if (upstreamError?.trim() && isUpstreamResetReplayRefusedMessage(upstreamError)) { + return UPSTREAM_RESET_REPLAY_REFUSED_CODE; + } + return "rate_limit_exceeded"; + } if (status === 503) return "server_is_overloaded"; if (status >= 500) return "upstream_server_error"; return `http_${status}`; diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 4e458f135f6..e4ca80ccb22 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -121,6 +121,7 @@ import { recordCodexUpstreamOutcome } from "../../codex/routing"; import { describeUpstreamConnectFailure } from "./upstream-error"; import type { OpaqueBlobRecoveryGuard } from "./core-opaque-recovery"; import { + isOpenCodeGoDestination, rateLimitRetryPolicyFor, rateLimitRetryDelayMs, transientRetryPolicyFor, @@ -841,7 +842,15 @@ export async function preparePassthroughExchange( // retry wrapper replaces — proves the host was reached (#914 review). .then(adoptObservedResponse); }, - { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends }, + { abortSignal: upstream.signal, label: safeHostLabel(request.url), + attempts: remainingTransientSendBudget(transientSendAttempts()), onSendsConsumed: noteTransientSends, + // The OpenCode Go destination stalls-then-drops inference sends (ambiguous + // pre-header resets surfacing as refused 429s); its subscription traffic is + // inference-only, so a bounded reset replay here absorbs the blip instead of + // failing the turn. Recovery legs keep the fail-closed refusal; only this + // initial send is replay-eligible. Attempts stay budget-bounded via attempts. + replaySafe: isOpenCodeGoDestination(route.provider), + }, ); } catch (err) { return transportFailureResponse(err); diff --git a/tests/responses/responses-passthrough-transient-policy.test.ts b/tests/responses/responses-passthrough-transient-policy.test.ts index c598ac4fa1c..77bc00487fe 100644 --- a/tests/responses/responses-passthrough-transient-policy.test.ts +++ b/tests/responses/responses-passthrough-transient-policy.test.ts @@ -175,3 +175,10 @@ describe("a configured ladder is bounded by the request budget", () => { expect(budget.reserveSpent).toBe(false); }); }); + + const goPacked = dense(readResponsesCoreModule("passthrough-dispatch.ts")); +describe("the Go destination replays ambiguous resets on the initial send", () => { + test("replaySafe is destination-scoped to exactly one leg", () => { + expect(occurrences(goPacked, "replaySafe:isOpenCodeGoDestination(route.provider)")).toBe(1); + }); +}); diff --git a/tests/usage/request-log.test.ts b/tests/usage/request-log.test.ts index f4842c27809..ae4c9b9493d 100644 --- a/tests/usage/request-log.test.ts +++ b/tests/usage/request-log.test.ts @@ -830,6 +830,10 @@ describe("request log metadata", () => { "Provider error 401: this model requires a subscription, upgrade for access", )).toBe("invalid_api_key"); expect(requestLogErrorCode(429)).toBe("rate_limit_exceeded"); + expect(requestLogErrorCode( + 429, + "The upstream connection closed before a response was received. The request may already have been processed; automatic replay was stopped.", + )).toBe("upstream_reset_replay_refused"); expect(requestLogErrorCode(499)).toBe("client_closed_request"); expect(requestLogErrorCode(502, "client closed request during web-search")).toBe("client_closed_request"); expect(requestLogErrorCode(400, "blocked", "cyber_policy")).toBe("cyber_policy");