diff --git a/src/lib/request-execution-budget.ts b/src/lib/request-execution-budget.ts index 429c0bc1e9d..31c5097aa12 100644 --- a/src/lib/request-execution-budget.ts +++ b/src/lib/request-execution-budget.ts @@ -187,6 +187,8 @@ export interface RequestExecutionBudget extends TransientSendBudget { * that cannot reach it has no operator override, which is the fail-closed answer. */ claimAmbiguousResend?(limit: number): boolean; + /** True once any scope has claimed a replacement for this logical request. */ + readonly ambiguousResendSpent?: boolean; } const RESERVE_FUNDED_CLASSES: ReadonlySet = new Set([ @@ -224,6 +226,7 @@ interface SharedSendLedger { * it in -- it has to ask whoever holds the request's grant. */ claimAmbiguousResend(limit: number): boolean; + readonly ambiguousResendSpent: boolean; readonly observer?: RequestSendObserver; } @@ -240,20 +243,23 @@ const sharedSendLedgers = new WeakMap( * already spent the one replacement a strict row granted buy another as soon as a more * permissive row asked, which is a second duplicate inference of one turn. */ -function createAmbiguousResendGrant(): (limit: number) => boolean { +function createAmbiguousResendGrant(): Pick { let claimed = 0; let ceiling: number | undefined; - return (limit: number): boolean => { - const presented = Number.isFinite(limit) ? Math.trunc(limit) : 0; - // A zero or nonsense ceiling refuses on its own and leaves the request's alone. It is a - // caller that cannot state a grant, not an operator narrowing this request: a leg with no - // policy is refused before it ever claims, so binding the request to a malformed number - // would only let such a caller cancel a grant an opted-in row really made. - if (presented <= 0) return false; - ceiling = ceiling === undefined ? presented : Math.min(ceiling, presented); - if (claimed >= ceiling) return false; - claimed += 1; - return true; + return { + get ambiguousResendSpent(): boolean { return claimed > 0; }, + claimAmbiguousResend(limit: number): boolean { + const presented = Number.isFinite(limit) ? Math.trunc(limit) : 0; + // A zero or nonsense ceiling refuses on its own and leaves the request's alone. It is a + // caller that cannot state a grant, not an operator narrowing this request: a leg with no + // policy is refused before it ever claims, so binding the request to a malformed number + // would only let such a caller cancel a grant an opted-in row really made. + if (presented <= 0) return false; + ceiling = ceiling === undefined ? presented : Math.min(ceiling, presented); + if (claimed >= ceiling) return false; + claimed += 1; + return true; + }, }; } @@ -302,6 +308,7 @@ function createRequestExecutionBudgetWithLedger( claimAmbiguousResend(limit: number): boolean { return counter.claimAmbiguousResend(limit); }, + get ambiguousResendSpent(): boolean { return counter.ambiguousResendSpent; }, reserveDispatch(intent: DispatchIntent): DispatchDecision { if (intent.replaySafe === false) return { allowed: false, reason: "not-replay-safe" }; if (counter.spent >= policy.maxTotalModelSends) return { allowed: false, reason: "total-exhausted" }; @@ -396,10 +403,12 @@ export function createRequestExecutionBudget( logicalRequestId?: string, observer?: RequestSendObserver, ): RequestExecutionBudget { + const grant = createAmbiguousResendGrant(); return createRequestExecutionBudgetWithLedger(policy, logicalRequestId, { spent: 0, pendingExternalSends: 0, - claimAmbiguousResend: createAmbiguousResendGrant(), + claimAmbiguousResend: grant.claimAmbiguousResend, + get ambiguousResendSpent(): boolean { return grant.ambiguousResendSpent; }, ...(observer ? { observer } : {}), }); } @@ -445,8 +454,11 @@ function ledgerFor(parent: RequestExecutionBudget): SharedSendLedger { // observer genuinely cannot cross this boundary because they are private to the factory, // but the grant can -- `claimAmbiguousResend` is public on the parent. A parent that does // not implement it grants nothing, which is the fail-closed answer for a send whose - // upstream state is unknown. - claimAmbiguousResend: (limit: number): boolean => parent.claimAmbiguousResend?.(limit) === true, + // upstream state is unknown. So does a parent that cannot report the grant as spent: a claim + // nobody can read back would let combo failover hop after the replacement went out. + claimAmbiguousResend: (limit: number): boolean => + typeof parent.ambiguousResendSpent === "boolean" && parent.claimAmbiguousResend?.(limit) === true, + get ambiguousResendSpent(): boolean { return parent.ambiguousResendSpent === true; }, }; } diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index 3419b8ffb8a..2b2b546388c 100644 --- a/src/lib/upstream-retry.ts +++ b/src/lib/upstream-retry.ts @@ -440,6 +440,25 @@ function invitesResendAfterReplacement(status: number): boolean { || status === 307 || status === 308 || status === 413 || status >= 500; } +/** + * The answer a request keeps once its one operator replacement has gone out. + * + * A status that invites another send settles as the refusal. Any other answer keeps its real + * status: no client retries it, and the caller needs the evidence (a 400 names the request + * defect). The marker still stops this process from using it as a recovery trigger, such as the + * opaque-blob rebuild of a 400 or a combo hop on a context overflow, because each of those checks + * it before sending again. + */ +export function settleOperatorReplacement(response: Response): Response { + if (response.ok) return response; + if (invitesResendAfterReplacement(response.status)) { + cancelResponseBodyBestEffort(response); + return replayRefusalResponse(); + } + markResponseNonReplayable(response); + return response; +} + export async function fetchWithAttemptDeadline( url: string, init: RequestInit, @@ -624,18 +643,7 @@ export async function fetchWithResetRetry( opts.onSendsConsumed?.(1); try { const response = await doFetch(attempt === 0 ? firstRecovery : "connection-reset"); - if (spentOperatorReplacement && !response.ok) { - if (invitesResendAfterReplacement(response.status)) { - cancelResponseBodyBestEffort(response); - return replayRefusalResponse(); - } - // Any other answer keeps its real status: no client retries it, and the caller needs the - // evidence (a 400 names the request defect). The marker still stops this process from - // using it as a recovery trigger, such as the opaque-blob rebuild of a 400 or a combo hop - // on a context overflow, because each of those checks it before sending again. - markResponseNonReplayable(response); - } - return response; + return spentOperatorReplacement ? settleOperatorReplacement(response) : response; } catch (err) { if (opts.abortSignal?.aborted) throw err; if (!isConnectionResetError(err)) { diff --git a/src/server/responses/core-combo.ts b/src/server/responses/core-combo.ts index 7712904aaf4..1178c4f86a8 100644 --- a/src/server/responses/core-combo.ts +++ b/src/server/responses/core-combo.ts @@ -73,6 +73,7 @@ import { import { preflightComboStreamResponse } from "./combo-stream-preflight"; import { streamingContextOverflowResponse, jsonContextOverflowResponse } from "./context-overflow"; import { mandatoryResponsesReasoningReplayUnavailable } from "./core-replay"; +import { settleOperatorReplacement } from "../../lib/upstream-retry"; /** * Sends one combo target may run on its own before the ladder moves on. A target is a whole @@ -678,9 +679,20 @@ export async function executeComboResponses( attemptRetained = true; lastFailure = failure.response; lastFailedChildLog = childLog; + // A replacement that answers 200 is unmarked, and its zero-output failure only exists once + // preflight has rebuilt the stream as a fresh Response. A spent grant never hops: a status the + // client would resend becomes the refusal, and anything else reaches the client as it is. + const spentReplacement = !failure.nonReplayable && comboSendScope?.ambiguousResendSpent === true; + if (spentReplacement) { + const settled = settleOperatorReplacement(failure.response); + if (settled !== failure.response) { + adoptFailedChildLog(childLog); + return settled; + } + } // A non-replayable failure (the answer to a spent ambiguous-reset replacement) may follow a // send that already ran the turn, so no later target may receive it, whatever its status says. - const failureDecision = failure.nonReplayable + const failureDecision = failure.nonReplayable || spentReplacement ? "stop" : comboFailureDecision(failure.response.status, failure.classificationText, { code: failure.upstreamCode, diff --git a/src/server/responses/core-opaque-recovery.ts b/src/server/responses/core-opaque-recovery.ts index 788e2f2c1ab..b436ebdc9e6 100644 --- a/src/server/responses/core-opaque-recovery.ts +++ b/src/server/responses/core-opaque-recovery.ts @@ -311,14 +311,15 @@ export function shouldAttemptOpaqueBlobRecovery(args: { * Peek the upstream error body for the reasoning-effort downgrade. Only 400/403 are considered * and the body must be complete and display-safe, the same contract the other rejection peeks * use. The match is deliberately narrow: the upstream has to name reasoning effort, so an - * unrelated 400 never triggers a replay. + * unrelated 400 never triggers a replay. A non-replayable answer, such as one to a spent operator + * replacement, is never read: the first send may already have run the turn. */ export async function reasoningEffortRejectionText( response: Response, alreadyAttempted: boolean, signal: AbortSignal, ): Promise { - if (alreadyAttempted) return undefined; + if (alreadyAttempted || isNonReplayableResponse(response)) return undefined; if (response.status !== 400 && response.status !== 403) return undefined; try { const body = await readBoundedResponseBody(response.clone(), { signal }); diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 733081b0bdc..42e0781f115 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -112,6 +112,7 @@ import { fetchWithTransientRetry, applyUpstreamRecoveryInit, isNonReplayableResponse, + settleOperatorReplacement, refetchAfterProtocolSafeReset, prepareSameTarget429Wait, sleepWithAbort, @@ -212,6 +213,7 @@ export async function preparePassthroughExchange( | "recoveryClassFor" | "sendBudgetExhausted" | "claimAmbiguousResend" + | "ambiguousResendSpent" | "reserveCredentialHop" | "pendingHopPermit" | "workflowRootId" @@ -1554,7 +1556,8 @@ export async function preparePassthroughExchange( if (options.abortSignal?.aborted) return transportFailureResponse(options.abortSignal.reason); upstreamResponse = preflight.response; if (preflight.kind === "failed") { - if (!configuredTransientSendBudgetExhausted()) { + // A zero-output failure does not undo an ambiguous replacement already sent. + if (!configuredTransientSendBudgetExhausted() && !sendBudgetState.ambiguousResendSpent) { const streamedOpaqueRecovery = await attemptOpaqueBlobRecovery({ response: upstreamResponse, outboundBody: request.body, @@ -1574,6 +1577,8 @@ export async function preparePassthroughExchange( logCtx.terminalHttpStatus = preflightLog.terminalHttpStatus; logCtx.terminalErrorCode = preflightLog.terminalErrorCode; logCtx.terminalIncompleteReason = preflightLog.terminalIncompleteReason; + // The projected failure must not invite another send after the replacement was spent. + if (sendBudgetState.ambiguousResendSpent) upstreamResponse = settleOperatorReplacement(upstreamResponse); } } // Console Go (opencode-zen / opencode-go) intermittently rejects a body it accepts seconds diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index 5d8bc2f9132..c78809913e2 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -1,5 +1,6 @@ import { comboFailureDecision } from "../../combos/failover"; import { readBoundedResponseBody } from "../../lib/bounded-body"; +import { isNonReplayableResponse } from "../../lib/upstream-retry"; import { finishRequestAttempt, type RequestLogContext } from "../request-log"; import { linkRequestSessionLane } from "../request-log-conversation"; import type { OcxConfig } from "../../types"; @@ -84,6 +85,8 @@ function errorCodeFromText(text: string): string | undefined { async function shouldHopPolicyCandidate(response: Response, signal?: AbortSignal): Promise { if (response.status < 400 || signal?.aborted) return false; + // A response that must not be sent again cannot open a policy-candidate retry either. + if (isNonReplayableResponse(response)) return false; try { const inspected = await readBoundedResponseBody(response.clone(), { signal }); const text = inspected.displaySafe ? inspected.text : ""; diff --git a/src/server/responses/request-send-budget.ts b/src/server/responses/request-send-budget.ts index 44ca16bc4d1..3dfe4f086ca 100644 --- a/src/server/responses/request-send-budget.ts +++ b/src/server/responses/request-send-budget.ts @@ -148,7 +148,8 @@ export function createResponsesSendBudget( * upstream state is unknown. */ const claimAmbiguousResend = (limit: number): boolean => - isRequestExecutionBudget(sendBudget) && sendBudget.claimAmbiguousResend?.(limit) === true; + isRequestExecutionBudget(sendBudget) && typeof sendBudget.ambiguousResendSpent === "boolean" + && sendBudget.claimAmbiguousResend?.(limit) === true; /** * A credential hop reserves the send its own replay will make, and that replay is a recovery * leg. The leg must SPEND the hop's reservation instead of taking a second one: the @@ -274,6 +275,9 @@ export function createResponsesSendBudget( noteAdapterRecoveryWithheld, sendBudgetExhausted, claimAmbiguousResend, + get ambiguousResendSpent(): boolean { + return isRequestExecutionBudget(sendBudget) && sendBudget.ambiguousResendSpent === true; + }, get pendingHopPermit(): SingleUseDispatchPermit | undefined { return pendingHopPermit; }, @@ -312,7 +316,9 @@ function adapterDispatchBudgetView( get targetTransitions(): number { return budget.targetTransitions; }, get lastTargetKey(): string | undefined { return budget.lastTargetKey; }, remainingBaseSends: (cap: number): number => budget.remainingBaseSends(cap), - claimAmbiguousResend: (limit: number): boolean => budget.claimAmbiguousResend?.(limit) === true, + claimAmbiguousResend: (limit: number): boolean => + typeof budget.ambiguousResendSpent === "boolean" && budget.claimAmbiguousResend?.(limit) === true, + get ambiguousResendSpent(): boolean { return budget.ambiguousResendSpent === true; }, reserveDispatch(intent: DispatchIntent): DispatchDecision { // A dispatch whose upstream state is unknown is refused on its own merits. A hop that // already paid does not make an unsafe replay safe, so that check stays with the budget. diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index acbe91a6044..dca417f03cc 100644 --- a/structure/transports/responses-failover.md +++ b/structure/transports/responses-failover.md @@ -270,6 +270,13 @@ response, so `consumeComboFailure` records `nonReplayable` and the combo stops r on, say, a context overflow. The cost is that a real 401, 402 or 429 on a replacement send is not recorded against its credential on that request. +A 2xx replacement carries no marker, and its stream can still fail before any output. Preflight +then rebuilds that failure as a fresh Response, so the request execution budget's +`ambiguousResendSpent` makes the combo stop: a status the client would resend becomes the refusal, +and anything else keeps its status and the non-replayable marker. The direct path skips the +streamed opaque-blob rebuild and settles the preflight's projected failure by the same rule. +Policy fallback does not hop on a marked answer. + **An upstream reset observed mid-stream or after a terminal keeps its existing behaviour.** The passthrough read path still settles a genuine upstream reset as a synthetic 502, and the Codex WebSocket transport still settles `upstream_closed_before_response` (socket closed diff --git a/tests/lib/ambiguous-resend-composition.test.ts b/tests/lib/ambiguous-resend-composition.test.ts index f9dd56e9bec..aed2e4377bd 100644 --- a/tests/lib/ambiguous-resend-composition.test.ts +++ b/tests/lib/ambiguous-resend-composition.test.ts @@ -1,10 +1,11 @@ -import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"; import { readFileSync } from "node:fs"; import { join } from "node:path"; import { CODEX_TEXT_GUARDED_BUDGET_POLICY, createRequestExecutionBudget, deriveRequestExecutionBudget, + isRequestExecutionBudget, type RequestExecutionBudget, type SendClass, } from "../../src/lib/request-execution-budget"; @@ -17,6 +18,7 @@ import { TRANSIENT_RETRY_MAX_ATTEMPTS, } from "../../src/lib/upstream-retry"; import { ambiguousResendAllowanceFor } from "../../src/server/responses/reset-replay"; +import { createResponsesSendBudget } from "../../src/server/responses/request-send-budget"; import { resetReplayPolicyFor } from "../../src/providers/key-failover"; import { repoPath } from "../helpers/repo-root"; import type { OcxProviderConfig } from "../../src/types"; @@ -131,6 +133,86 @@ function oneLogicalRequest() { } describe("one resend budget across composed recovery legs", () => { + test("a fresh budget has not spent an ambiguous resend", () => { + const budget = createRequestExecutionBudget(); + expect(budget.ambiguousResendSpent).toBe(false); + expect(budget.claimAmbiguousResend?.(0)).toBe(false); + expect(budget.ambiguousResendSpent).toBe(false); + }); + + test("a derived scope observes the parent's spent ambiguous resend", () => { + const parent = createRequestExecutionBudget(); + const child = deriveRequestExecutionBudget(parent, CODEX_TEXT_GUARDED_BUDGET_POLICY); + expect(parent.claimAmbiguousResend?.(GRANT)).toBe(true); + expect(child.ambiguousResendSpent).toBe(true); + }); + + test("a parent observes a derived scope's spent ambiguous resend", () => { + const parent = createRequestExecutionBudget(); + const child = deriveRequestExecutionBudget(parent, CODEX_TEXT_GUARDED_BUDGET_POLICY); + expect(child.claimAmbiguousResend?.(GRANT)).toBe(true); + expect(parent.ambiguousResendSpent).toBe(true); + }); + + for (const view of ["derived scope", "adapter dispatch", "Responses send"] as const) { + const buildView = (parent: RequestExecutionBudget) => { + if (view === "derived scope") { + return deriveRequestExecutionBudget(parent, CODEX_TEXT_GUARDED_BUDGET_POLICY); + } + const state = createResponsesSendBudget({ + options: { sendBudget: parent }, + req: new Request("http://localhost/v1/responses"), + logCtx: { model: "", provider: "" }, + }); + if (state instanceof Response) throw new Error("unexpected workflow refusal"); + if (view === "Responses send") return state; + if (!state.adapterDispatchBudget) throw new Error("missing adapter dispatch budget"); + return state.adapterDispatchBudget; + }; + const handBuiltParent = () => ({ + used: 0, + logicalRequestId: "stub", + policyVersion: "stub", + policy: CODEX_TEXT_GUARDED_BUDGET_POLICY, + reserveSpent: false, + alternateTargetSends: 0, + targetTransitions: 0, + lastTargetKey: undefined, + remainingBaseSends: () => 0, + reserveDispatch: () => ({ allowed: false, reason: "total-exhausted" }), + claimAmbiguousResend: mock((_limit: number) => true), + } satisfies RequestExecutionBudget); + + test(`${view} refuses a claim when the parent omits its spent state`, () => { + const parent = handBuiltParent(); + expect(isRequestExecutionBudget(parent)).toBe(true); + expect("ambiguousResendSpent" in parent).toBe(false); + const budget = buildView(parent); + + expect(budget.ambiguousResendSpent).toBe(false); + expect(budget.claimAmbiguousResend?.(GRANT)).toBe(false); + expect(parent.claimAmbiguousResend).not.toHaveBeenCalled(); + expect(budget.ambiguousResendSpent).toBe(false); + }); + + test(`${view} grants a claim when the parent exposes its spent state`, () => { + const parent = handBuiltParent(); + const observableParent = { + ...parent, + get ambiguousResendSpent(): boolean { return parent.claimAmbiguousResend.mock.calls.length > 0; }, + }; + expect(isRequestExecutionBudget(observableParent)).toBe(true); + const budget = buildView(observableParent); + + expect(budget.ambiguousResendSpent).toBe(false); + expect(budget.claimAmbiguousResend?.(GRANT)).toBe(true); + expect(parent.claimAmbiguousResend).toHaveBeenCalledTimes(1); + expect(parent.claimAmbiguousResend).toHaveBeenCalledWith(GRANT); + expect(budget.ambiguousResendSpent).toBe(true); + expect(observableParent.ambiguousResendSpent).toBe(true); + }); + } + test("the whole chain spends the grant once, whatever each leg was separately entitled to", async () => { silenceWarn(); const request = oneLogicalRequest(); @@ -233,12 +315,16 @@ describe("one resend budget across composed recovery legs", () => { get lastTargetKey(): string | undefined { return owner.lastTargetKey; }, remainingBaseSends: (cap: number): number => owner.remainingBaseSends(cap), claimAmbiguousResend: (limit: number): boolean => owner.claimAmbiguousResend?.(limit) === true, + get ambiguousResendSpent(): boolean | undefined { return owner.ambiguousResendSpent; }, reserveDispatch: intent => owner.reserveDispatch(intent), }; const first = deriveRequestExecutionBudget(view, CODEX_TEXT_GUARDED_BUDGET_POLICY); const second = deriveRequestExecutionBudget(view, CODEX_TEXT_GUARDED_BUDGET_POLICY); + expect(second.ambiguousResendSpent).toBe(false); expect(first.claimAmbiguousResend?.(GRANT)).toBe(true); + expect(second.ambiguousResendSpent).toBe(true); + expect(view.ambiguousResendSpent).toBe(true); expect(second.claimAmbiguousResend?.(GRANT)).toBe(false); expect(view.claimAmbiguousResend?.(GRANT)).toBe(false); expect(owner.claimAmbiguousResend?.(GRANT)).toBe(false); diff --git a/tests/routing/routing-policy-fallback.test.ts b/tests/routing/routing-policy-fallback.test.ts index 299b96b70ea..c09d720d0d7 100644 --- a/tests/routing/routing-policy-fallback.test.ts +++ b/tests/routing/routing-policy-fallback.test.ts @@ -2,7 +2,7 @@ import { describe, expect, test } from "bun:test"; import { formatErrorResponse } from "../../src/bridge"; import { RequestPacingQueueOverloadError } from "../../src/providers/request-pacing"; -import { fetchWithTransientRetry, isNonReplayableResponse } from "../../src/lib/upstream-retry"; +import { fetchWithTransientRetry, isNonReplayableResponse, markResponseNonReplayable } from "../../src/lib/upstream-retry"; import { shouldRetryCodexPoolAccountQuota } from "../../src/server/responses/core-codex-account"; import type { OcxConfig } from "../../src/types"; import { beginRequestAttempt, type RequestLogContext } from "../../src/server/request-log"; @@ -52,6 +52,27 @@ function seedAttempt(logCtx: RequestLogContext, provider: string, model: string) } describe("policy candidate fallback", () => { + test("a marked context overflow never tries another policy route", async () => { + const failure = Response.json({ error: { + type: "invalid_request_error", code: "context_length_exceeded", message: "Context window exceeded", + } }, { status: 400 }); + markResponseNonReplayable(failure); + let coreCalls = 0; + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, {} as RequestLogContext, {}, { + runCore: async (req, _config, context, options) => { + coreCalls += 1; + options.onRequestBodyParsed?.(await req.json()); + context.routeDecision = policyTrace(); + return coreCalls === 1 ? failure : Response.json({ status: "completed" }); + }, + }); + + expect(coreCalls).toBe(1); + expect(response).toBe(failure); + expect(response.status).toBe(400); + expect(isNonReplayableResponse(response)).toBe(true); + }); + test.each([false, true])("reset refusal stays terminal across policy and account recovery (replacement=%s)", async replacement => { let sends = 0; let coreCalls = 0; diff --git a/tests/server/replay-refusal-parity.test.ts b/tests/server/replay-refusal-parity.test.ts index d4b3fce38e5..201fa358640 100644 --- a/tests/server/replay-refusal-parity.test.ts +++ b/tests/server/replay-refusal-parity.test.ts @@ -3,6 +3,7 @@ import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { saveConfig } from "../../src/config"; +import { clearComboSelectionState, clearComboTargetCooldowns } from "../../src/combos"; import { startServer } from "../../src/server"; import { REPLAY_REFUSAL_NO_RETRY_HEADER, @@ -33,6 +34,8 @@ let previousHome: string | undefined; let isolatedCodexHome: IsolatedCodexHome | null = null; beforeEach(() => { + clearComboSelectionState(); + clearComboTargetCooldowns(); previousHome = process.env.OPENCODEX_HOME; isolatedCodexHome = installIsolatedCodexHome("ocx-replay-refusal-"); testDir = mkdtempSync(join(tmpdir(), "ocx-replay-refusal-")); @@ -41,6 +44,8 @@ beforeEach(() => { }); afterEach(() => { + clearComboSelectionState(); + clearComboTargetCooldowns(); globalThis.fetch = originalFetch; if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; @@ -231,6 +236,157 @@ function comboReplayConfig(): OcxConfig { } as unknown as OcxConfig; } +test("a combo refuses replay after a spent replacement's zero-output stream failure", async () => { + saveConfig(comboReplayConfig()); + let firstSends = 0; + let secondSends = 0; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (url.includes(COMBO_FIRST_HOST)) { + firstSends += 1; + if (firstSends === 1) preHeaderReset(); + const failure = { type: "response.failed", response: { + id: "resp_failed", object: "response", status: "failed", output: [], + error: { code: "server_is_overloaded", message: "Server is overloaded" }, + } }; + return new Response(`event: response.failed\ndata: ${JSON.stringify(failure)}\n\n`, { + headers: { "content-type": "text/event-stream" }, + }); + } + if (url.includes(COMBO_SECOND_HOST)) { + secondSends += 1; + return Response.json({ error: { code: "unexpected_second_target" } }, { status: 400 }); + } + return originalFetch(input as RequestInfo, init); + }) as typeof fetch; + const server = startServer(0); + try { + const { response, attempts, json } = await sendWithClientRetries(new URL("/v1/responses", server.url), { + model: "combo/pair", store: false, stream: true, ...RESPONSES_TURN, + }); + expect({ firstSends, secondSends, attempts }).toEqual({ firstSends: 2, secondSends: 0, attempts: 1 }); + expect(response.status).toBe(REPLAY_REFUSED_STATUS); + expect(json.error?.code).toBe(UPSTREAM_RESET_REPLAY_REFUSED_CODE); + expect(response.headers.get(REPLAY_REFUSAL_NO_RETRY_HEADER)).toBe(REPLAY_REFUSAL_NO_RETRY_VALUE); + } finally { + await server.stop(true); + } +}); + +test("a combo keeps a spent replacement's zero-output context overflow", async () => { + saveConfig(comboReplayConfig()); + let firstSends = 0; + let secondSends = 0; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (url.includes(COMBO_FIRST_HOST)) { + firstSends += 1; + if (firstSends === 1) preHeaderReset(); + const failure = { type: "response.failed", response: { + id: "resp_overflow", object: "response", status: "failed", output: [], + error: { type: "invalid_request_error", code: "context_length_exceeded", + message: "Input exceeds the model context window." }, + } }; + return new Response(`event: response.failed\ndata: ${JSON.stringify(failure)}\n\n`, { + headers: { "content-type": "text/event-stream" }, + }); + } + if (url.includes(COMBO_SECOND_HOST)) { + secondSends += 1; + return Response.json({ error: { code: "unexpected_second_target" } }, { status: 400 }); + } + return originalFetch(input as RequestInfo, init); + }) as typeof fetch; + const server = startServer(0); + try { + const { response, attempts, json } = await sendWithClientRetries(new URL("/v1/responses", server.url), { + model: "combo/pair", store: false, stream: true, ...RESPONSES_TURN, + }); + expect({ firstSends, secondSends, attempts }).toEqual({ firstSends: 2, secondSends: 0, attempts: 1 }); + expect(response.status).toBe(400); + expect(json.error?.code).toBe("context_length_exceeded"); + expect(JSON.stringify(json)).toContain("Input exceeds the model context window."); + } finally { + await server.stop(true); + } +}); + +test("the direct path refuses replay after a spent replacement's decrypt failure", async () => { + const config = parityConfig(); + config.providers.bridged.retryOnReset = {}; + saveConfig(config); + let upstreamSends = 0; + const sends = countingUpstream(() => { + upstreamSends += 1; + if (upstreamSends === 1) preHeaderReset(); + const failure = { type: "response.failed", response: { + id: "resp_decrypt_failed", status: "failed", + error: { type: "server_error", code: "upstream_server_error", + message: "Encrypted function output content could not be decrypted or decoded." }, + } }; + return new Response(`event: response.failed\ndata: ${JSON.stringify(failure)}\n\ndata: [DONE]\n\n`, { + headers: { "content-type": "text/event-stream" }, + }); + }); + // Canonical key-independent Fernet structure, as in the opaque-blob recovery fixtures. + const encryptedContent = `${Buffer.concat([ + Buffer.from([0x80]), Buffer.alloc(8), Buffer.alloc(16), Buffer.alloc(16), Buffer.alloc(32), + ]).toString("base64url")}==`; + const server = startServer(0); + try { + const response = await originalFetch(new URL("/v1/responses", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "bridged/model", store: false, stream: true, + input: [ + { type: "function_call", call_id: "call-encrypted-output", name: "browser_capture", arguments: "{}" }, + { type: "function_call_output", call_id: "call-encrypted-output", output: [ + { type: "encrypted_content", encrypted_content: encryptedContent }, + { type: "input_text", text: "visible tool output" }, + { type: "input_image", image_url: "data:image/png;base64,AAAA", detail: "high" }, + ] }, + { role: "user", content: [{ type: "input_text", text: "continue" }] }, + ], + }), + }); + expect(sends()).toBe(2); + expect(response.status).toBe(REPLAY_REFUSED_STATUS); + expect((await response.json()).error.code).toBe(UPSTREAM_RESET_REPLAY_REFUSED_CODE); + expect(response.headers.get(REPLAY_REFUSAL_NO_RETRY_HEADER)).toBe(REPLAY_REFUSAL_NO_RETRY_VALUE); + } finally { + await server.stop(true); + } +}); + +test("the direct path keeps a spent replacement's reasoning-effort rejection without a downgrade", async () => { + const config = parityConfig(); + config.providers.bridged.retryOnReset = {}; + config.providers.bridged.reasoningEfforts = ["low", "high"]; + saveConfig(config); + const rejection = { error: { + param: "reasoning.effort", type: "invalid_request_error", + message: "reasoning_effort high is not supported for model model.", + } }; + const sends = countingUpstream(() => { + if (sends() === 1) preHeaderReset(); + return Response.json(rejection, { status: 400 }); + }); + const server = startServer(0); + try { + const { response, attempts, json } = await sendWithClientRetries(new URL("/v1/responses", server.url), { + model: "bridged/model", store: false, stream: false, + reasoning: { effort: "high" }, ...RESPONSES_TURN, + }); + expect(sends()).toBe(2); + expect(attempts).toBe(1); + expect(response.status).toBe(400); + expect(json).toEqual(rejection); + } finally { + await server.stop(true); + } +}); + test.each([ { name: "a context overflow", status: 400, expectedStatus: 400 }, { name: "a 413", status: 413, expectedStatus: REPLAY_REFUSED_STATUS },