diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index 0f910254d1a..f4151f5d660 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -146,7 +146,12 @@ export async function handleResponsesWithPolicyFallback( } : {}), onRequestBodyParsed: body => { options.onRequestBodyParsed?.(body); - if (body && typeof body === "object" && !Array.isArray(body)) rawBody = body as Record; + if (rawBody === null && body && typeof body === "object" && !Array.isArray(body)) { + // Recovery and other core preparation may mutate the parsed body in place. Keep an + // immutable snapshot of the original wire body so a retry cannot serialize those + // mutations while losing object-identity metadata attached by the first attempt. + rawBody = structuredClone(body as Record); + } }, onStoredPool401ReplayDispatched: () => { storedPool401ReplayDispatched = true; diff --git a/tests/routing/routing-policy-fallback.test.ts b/tests/routing/routing-policy-fallback.test.ts index 347b9bdaaac..c089df8aa4e 100644 --- a/tests/routing/routing-policy-fallback.test.ts +++ b/tests/routing/routing-policy-fallback.test.ts @@ -110,6 +110,66 @@ describe("policy candidate fallback", () => { expect(cloneCalls).toBe(0); }); + test("retries from an immutable snapshot of the initially parsed body", async () => { + const trace = policyTrace(); + const logCtx = { routeDecision: trace } as RequestLogContext; + const seenInputs: unknown[] = []; + let calls = 0; + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {}, { + runCore: async (req, _config, context, options) => { + calls += 1; + const body = await req.json() as { input: unknown; model: string }; + options.onRequestBodyParsed?.(body); + seenInputs.push(body.input); + context.routeDecision = trace; + if (calls === 1) { + body.input = "recovered plaintext"; + return Response.json({ error: { type: "rate_limit_error" } }, { status: 429 }); + } + return Response.json({ status: "completed" }); + }, + }); + + expect(response.status).toBe(200); + expect(seenInputs).toEqual(["hello", "hello"]); + }); + + test("the retry snapshot survives mutation inside the input array", async () => { + // The top-level field swap above also passes under a shallow `{...body}` copy. The + // real leaks mutate deeper: the sanitizer splices input entries in place and the + // assignment injector rewrites inside the same array. Pin a nested mutation so a + // shallow-copy regression cannot stay green. + const trace = policyTrace(); + const logCtx = { routeDecision: trace } as RequestLogContext; + const seenInputs: unknown[] = []; + let calls = 0; + const req = new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "policy/daily", input: [{ role: "user", content: "hello" }], stream: false }), + }); + const response = await handleResponsesWithPolicyFallback(req, {} as OcxConfig, logCtx, {}, { + runCore: async (req, _config, context, options) => { + calls += 1; + const body = await req.json() as { input: { role: string; content: string }[]; model: string }; + options.onRequestBodyParsed?.(body); + seenInputs.push(JSON.parse(JSON.stringify(body.input))); + context.routeDecision = trace; + if (calls === 1) { + body.input.splice(0, 1, { role: "assistant", content: "recovered plaintext" }); + return Response.json({ error: { type: "rate_limit_error" } }, { status: 429 }); + } + return Response.json({ status: "completed" }); + }, + }); + + expect(response.status).toBe(200); + expect(seenInputs).toEqual([ + [{ role: "user", content: "hello" }], + [{ role: "user", content: "hello" }], + ]); + }); + test("a local input-admission refusal hops instead of ending the chain (#1524)", async () => { // #1524: a candidate whose context window cannot fit the request used to TERMINATE the // fallback chain. It is a local preflight verdict about ONE candidate, not about the