From a5497826f7552b37cd1f6af8469afa63f2909149 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sat, 29 Aug 2026 05:25:57 +0900 Subject: [PATCH] fix(responses): recover encrypted combo agent tasks Run the opt-in recovery pass before an all-third-party combo fails closed, then propagate the recovered-plaintext marker to every cloned child body so continuation state cannot persist the decrypted assignment. Preserve the canonical OpenAI target bypass and single-attempt failure path. Credit: @yxr1995-maker authored the original fix in #2744. --- src/server/responses/core.ts | 38 +++++++- tests/agent-task-recovery-combo.test.ts | 124 ++++++++++++++++++++++++ tests/agent-task-recovery.test.ts | 13 +-- 3 files changed, 167 insertions(+), 8 deletions(-) create mode 100644 tests/agent-task-recovery-combo.test.ts diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index c018751cfd7..03856a5cdcb 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1339,6 +1339,7 @@ export interface HandleResponsesOptions { sourceBody: unknown; previousResponseInputExpanded: boolean; providerContinuation: OcxProviderContinuationState | undefined; + recoveredPlaintext: boolean; }; /** Internal combo handoff: allow a later same-provider model after a reset-derived 429/402. */ deferCodexResetDerivedCooldown?: boolean; @@ -1894,6 +1895,7 @@ export async function handleComboResponses( providerContinuation: !scopeMismatch && body !== rawBody && requestedPreviousId ? previousResponseProviderState(requestedPreviousId) : undefined, + recoveredPlaintext: false, }; const adoptFailedChildLog = (childLog: RequestLogContext): void => { // Attempts remain the complete physical history; the logical row mirrors the most recent @@ -1923,11 +1925,40 @@ export async function handleComboResponses( return false; } }; + let comboPayloadReadable = false; const payloadEligible = (target: (typeof combo.targets)[number]): boolean => - !unreadableEncryptedAgentTask || canDecryptUnreadableAgentTask(target); + comboPayloadReadable || !unreadableEncryptedAgentTask || canDecryptUnreadableAgentTask(target); if (unreadableEncryptedAgentTask && !combo.targets.some(canDecryptUnreadableAgentTask)) { - return unreadableEncryptedAgentTaskResponse(); + const recovery = agentTaskRecoveryConfig(config); + let recovered = false; + if ( + (options.inboundWire ?? "responses") === "responses" + && isThreadSpawnRequest(req.headers) + && recovery + && !options.comboAttempt + ) { + try { + recovered = await recoverEncryptedAgentTask( + req, + (body as { input?: unknown } | undefined)?.input, + recovery, + config, + { parentThreadId: inboundClientThreadId, abortSignal: options.abortSignal }, + ); + } catch { + recovered = false; + } + } + // Recovery has the same in-place input mutation contract as the direct routed path. + if ( + !recovered + || hasUnreadableEncryptedAgentTask((body as { input?: unknown } | undefined)?.input) + ) { + return unreadableEncryptedAgentTaskResponse(); + } + comboPayloadReadable = true; + comboReplaySnapshot.recoveredPlaintext = true; } const initialNow = Date.now(); @@ -2335,6 +2366,9 @@ async function handleResponsesInner( let toolBridgeMaps: ReturnType; try { parsed = parseRequest(body); + if (options.comboReplaySnapshot?.recoveredPlaintext) { + markBodyNonPersistable(parsed._rawBody); + } toolBridgeMaps = buildToolBridgeMaps(parsed, translatorBudget); if (previousResponseInputExpanded) parsed._previousResponseInputExpanded = true; const providerContinuationCandidate = options.comboReplaySnapshot diff --git a/tests/agent-task-recovery-combo.test.ts b/tests/agent-task-recovery-combo.test.ts new file mode 100644 index 00000000000..ffb889959ae --- /dev/null +++ b/tests/agent-task-recovery-combo.test.ts @@ -0,0 +1,124 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + clearResponseStateForTests, + clearResponseStateMemoryForTests, + responseContinuationRetainedStoreSnapshot, + runPendingResponseStatePersistForTests, +} from "../src/responses/state"; +import { resetAgentTaskRecoveryState } from "../src/server/responses/agent-task-recovery"; +import { + codexHeaders, + encryptedInput, + FERNET_TASK, + originalFetch, + post, + providerResponse, + recoverySse, + routedConfig, +} from "./helpers/agent-task-recovery"; + +function providerCompletion(): Response { + return Response.json({ + id: "chatcmpl_combo_recovery", + object: "chat.completion", + choices: [{ + index: 0, + message: { role: "assistant", content: "done" }, + finish_reason: "stop", + }], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }); +} + +function comboConfig(targets: Array<{ provider: string; model: string }>) { + const config = routedConfig(); + config.combos = { + routed: { + strategy: "failover", + targets, + }, + }; + return config; +} + +describe("combo path encrypted agent task recovery", () => { + const priorHome = process.env["OPENCODEX_HOME"]; + let home: string; + + beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "ocx-agent-task-combo-")); + process.env["OPENCODEX_HOME"] = home; + clearResponseStateMemoryForTests(); + resetAgentTaskRecoveryState(); + }); + + afterEach(() => { + globalThis.fetch = originalFetch; + resetAgentTaskRecoveryState(); + clearResponseStateForTests(); + rmSync(home, { recursive: true, force: true }); + if (priorHome === undefined) delete process.env["OPENCODEX_HOME"]; + else process.env["OPENCODEX_HOME"] = priorHome; + }); + + test("recovers an all-third-party combo once without retaining plaintext continuation state", async () => { + const assignment = "RECOVERED-COMBO-PLAINTEXT-SENTINEL"; + const fetchedUrls: string[] = []; + globalThis.fetch = (async (input) => { + const url = String(input); + fetchedUrls.push(url); + if (url.includes("chatgpt.com")) { + return new Response(recoverySse(assignment), { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); + } + return providerCompletion(); + }) as typeof fetch; + + const response = await post( + comboConfig([{ provider: "xai", model: "grok-4.5" }]), + "combo/routed", + encryptedInput(), + codexHeaders(), + ); + await runPendingResponseStatePersistForTests(); + const responsePayload = await response.clone().json() as { id?: string }; + + expect(response.status).toBe(200); + expect(typeof responsePayload.id).toBe("string"); + expect(fetchedUrls).toHaveLength(2); + expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex/responses"); + expect(responseContinuationRetainedStoreSnapshot().count).toBe(0); + const snapshotPath = join(home, "responses-state.json"); + const snapshot = existsSync(snapshotPath) ? readFileSync(snapshotPath, "utf8") : ""; + expect(snapshot).not.toContain(assignment); + expect(snapshot).not.toContain(responsePayload.id!); + }); + + test("keeps the canonical target bypass in a mixed combo without running recovery", async () => { + const forwardedBodies: string[] = []; + globalThis.fetch = (async (_input, init) => { + forwardedBodies.push(typeof init?.body === "string" ? init.body : ""); + return providerResponse(); + }) as typeof fetch; + + const response = await post( + comboConfig([ + { provider: "xai", model: "grok-4.5" }, + { provider: "openai", model: "gpt-5.5" }, + ]), + "combo/routed", + encryptedInput(), + codexHeaders(), + ); + + expect(response.status).toBe(200); + expect(forwardedBodies).toHaveLength(1); + expect(forwardedBodies[0]).toContain(FERNET_TASK); + expect(forwardedBodies[0]).not.toContain("capture_assignment"); + }); +}); diff --git a/tests/agent-task-recovery.test.ts b/tests/agent-task-recovery.test.ts index 78647e1075a..fe6200b01af 100644 --- a/tests/agent-task-recovery.test.ts +++ b/tests/agent-task-recovery.test.ts @@ -551,7 +551,7 @@ describe("agent task recovery (opt-in, default off)", () => { expect(forwardedBody).not.toContain("capture_assignment"); }); - test("does not enable recovery inside combo attempts", async () => { + test("fails closed after a single failed combo recovery pass", async () => { const config = routedConfig(); config.combos = { routed: { @@ -559,10 +559,10 @@ describe("agent task recovery (opt-in, default off)", () => { targets: [{ provider: "xai", model: "grok-4.5" }], }, }; - let fetchCalls = 0; - globalThis.fetch = (async () => { - fetchCalls += 1; - throw new Error("combo must fail before dispatch"); + const fetchedUrls: string[] = []; + globalThis.fetch = (async (input) => { + fetchedUrls.push(String(input)); + throw new Error("every upstream call must fail"); }) as typeof fetch; const response = await post( @@ -573,7 +573,8 @@ describe("agent task recovery (opt-in, default off)", () => { ); expect(response.status).toBe(400); - expect(fetchCalls).toBe(0); + expect(fetchedUrls).toHaveLength(1); + expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex/responses"); expect(await response.json()).toMatchObject({ error: { code: "unreadable_encrypted_agent_task" }, });