From ab471859546add9927a071565fdea2350746e9d8 Mon Sep 17 00:00:00 2001 From: Thierno Bah Date: Wed, 12 Aug 2026 15:45:00 +0200 Subject: [PATCH 1/4] fix(agents): recover encrypted v2 tasks Routed providers cannot consume backend-encrypted native Codex task payloads. Add a disabled-by-default recovery path that uses only the fixed authenticated ChatGPT Codex endpoint and fails closed for unsupported callers. Refs #92 --- .../content/docs/guides/sub-agent-surface.md | 11 + .../docs/ja/guides/sub-agent-surface.md | 2 + .../docs/ko/guides/sub-agent-surface.md | 2 + .../docs/reference/configuration/agents.md | 65 +++ .../docs/ru/guides/sub-agent-surface.md | 6 + .../docs/zh-cn/guides/sub-agent-surface.md | 2 + .../docs/zh-tw/guides/sub-agent-surface.md | 6 +- src/config.ts | 38 ++ src/server/index.ts | 5 + .../responses/agent-task-recovery-cache.ts | 137 +++++ src/server/responses/agent-task-recovery.ts | 466 +++++++++++++++++ src/server/responses/core.ts | 61 ++- src/server/responses/encrypted-payload.ts | 5 +- src/types.ts | 10 + tests/agent-task-recovery-cache.test.ts | 95 ++++ tests/agent-task-recovery-security.test.ts | 393 ++++++++++++++ tests/agent-task-recovery.test.ts | 494 ++++++++++++++++++ tests/config.test.ts | 55 ++ tests/helpers/agent-task-recovery.ts | 176 +++++++ 19 files changed, 2026 insertions(+), 3 deletions(-) create mode 100644 src/server/responses/agent-task-recovery-cache.ts create mode 100644 src/server/responses/agent-task-recovery.ts create mode 100644 tests/agent-task-recovery-cache.test.ts create mode 100644 tests/agent-task-recovery-security.test.ts create mode 100644 tests/agent-task-recovery.test.ts create mode 100644 tests/helpers/agent-task-recovery.ts diff --git a/docs-site/src/content/docs/guides/sub-agent-surface.md b/docs-site/src/content/docs/guides/sub-agent-surface.md index 4c48e8f396..526e0fd107 100644 --- a/docs-site/src/content/docs/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/guides/sub-agent-surface.md @@ -131,6 +131,17 @@ Recovery options are to select a native ChatGPT child, add a native ChatGPT targ v1 for heterogeneous-provider delegation, or resend the task as plaintext v2 `agent_message` content when you control the caller. +An experimental, disabled-by-default `agentTaskRecovery` option can recover this specific native- +to-routed shape through an additional authenticated request to ChatGPT before provider dispatch. +It consumes quota, adds latency, briefly retains recovered plaintext in a bounded in-memory cache, +and depends on undocumented ChatGPT backend behavior. Because a model returns the recovered text, +byte-for-byte fidelity is not guaranteed. It rejects generic/API-key proxy callers and preserves +`unreadable_encrypted_agent_task` on any failure. See +[Agent configuration: Encrypted v2 task recovery](/reference/configuration/agents/#encrypted-v2-task-recovery) +for the full trust boundary and configuration. +Combo routing remains unchanged and continues to consider only canonical native ChatGPT targets for +encrypted tasks. + ## Changing the mode ### GUI diff --git a/docs-site/src/content/docs/ja/guides/sub-agent-surface.md b/docs-site/src/content/docs/ja/guides/sub-agent-surface.md index daa1c2b839..ec0c290829 100644 --- a/docs-site/src/content/docs/ja/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/ja/guides/sub-agent-surface.md @@ -88,6 +88,8 @@ opencodex は、空のタスクまたは読み取り不可能なタスクを転 回復オプションは、ネイティブ ChatGPT 子の選択、コンボへのネイティブ ChatGPT ターゲットの追加、異種プロバイダーの委任に v1 を使用する、または呼び出し元を制御するときにタスクをプレーンテキスト v2 `agent_message` コンテンツとして再送信することです。 +実験的な `agentTaskRecovery` はデフォルトで無効です。明示的に有効にすると、固定された ChatGPT エンドポイントへの追加の認証済みリクエストでこの形式を回復できますが、クォータと待ち時間が増え、非公開のバックエンド動作に依存します。失敗時は従来の `unreadable_encrypted_agent_task` を維持します。詳細は[英語版の設定リファレンス](/reference/configuration/agents/#encrypted-v2-task-recovery)を参照してください。 + ## モードを変更する ### GUI diff --git a/docs-site/src/content/docs/ko/guides/sub-agent-surface.md b/docs-site/src/content/docs/ko/guides/sub-agent-surface.md index cd987ed761..9eac7b3b61 100644 --- a/docs-site/src/content/docs/ko/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/ko/guides/sub-agent-surface.md @@ -86,6 +86,8 @@ opencodex는 읽을 수 없거나 빈 작업을 그대로 넘기지 않고 안 복구 방법은 네이티브 ChatGPT 자식을 선택하거나, 콤보에 네이티브 ChatGPT 대상을 추가하거나, 이종 프로바이더 위임에는 v1을 사용하거나, 호출자를 제어할 수 있을 때 작업을 평문 v2 `agent_message` 콘텐츠로 다시 보내는 것입니다. +실험적인 `agentTaskRecovery`는 기본적으로 꺼져 있습니다. 명시적으로 켜면 고정된 ChatGPT 엔드포인트로 인증된 요청을 하나 더 보내 이 형식을 복구할 수 있지만, 할당량과 지연 시간이 늘고 비공개 백엔드 동작에 의존합니다. 실패하면 기존 `unreadable_encrypted_agent_task` 오류를 그대로 유지합니다. 자세한 내용은 [영문 설정 참고 문서](/reference/configuration/agents/#encrypted-v2-task-recovery)를 보세요. + ## 모드 변경 ### GUI diff --git a/docs-site/src/content/docs/reference/configuration/agents.md b/docs-site/src/content/docs/reference/configuration/agents.md index dddf8d4656..ba043ecacc 100644 --- a/docs-site/src/content/docs/reference/configuration/agents.md +++ b/docs-site/src/content/docs/reference/configuration/agents.md @@ -22,6 +22,7 @@ routes, and limits delegated work. | `subagentModelFallbackPollMs?` | `number` | `60000` | Availability-probe cache interval. Values below 1000 ms fall back to the default. | | `effortCap?` | `string` | — | Hard ceiling for qualifying v2 main turns and marked spawned-child turns. Accepts `low` through `ultra`. | | `subagentEffortCap?` | `string` | — | Additional ceiling for spawned-child turns only. When both caps apply, the lower wins. | +| `agentTaskRecovery?` | `object` | — | Experimental opt-in recovery for backend-encrypted v2 tasks sent to routed providers. Disabled unless `enabled: true`; see [Encrypted v2 task recovery](#encrypted-v2-task-recovery). | Manage the surface with the dashboard or `ocx v2 status|on|off|mode |threads |mode-hint `. @@ -118,6 +119,70 @@ fails instead of routing unreadable ciphertext elsewhere. } ``` +## Encrypted v2 task recovery + +`agentTaskRecovery` is an experimental compatibility path for a native ChatGPT parent spawning a +routed v2 child. It is disabled by default. When explicitly enabled and the final routed child task +contains an otherwise unreadable Fernet payload, opencodex sends the isolated `agent_message` to +the fixed authenticated ChatGPT Codex Responses endpoint. ChatGPT returns the plaintext assignment +through a forced function call; opencodex then converts only that task item to a standard user +message before routed-provider dispatch. + +This is not local decryption and does not fix the Codex wire protocol. It depends on undocumented +ChatGPT backend behavior and may stop working after a backend change. The recovered assignment is +model output, not a cryptographically verified plaintext, so byte-for-byte fidelity is not +guaranteed. Each scoped cache miss adds one authenticated ChatGPT request, consumes account quota, +and adds latency before the routed request. Concurrent requests for the same scoped task share one +recovery request. Startup prints a warning whenever the feature is enabled. + +Admission and retention are deliberately narrow: + +- only a native Codex caller with a matching ChatGPT bearer/account pair is eligible; +- callers using `x-opencodex-api-key`, `x-api-key`, generic API credentials, or a proxy admission + secret keep the existing `unreadable_encrypted_agent_task` failure; +- raw ChatGPT credentials are sent only to the hard-coded ChatGPT endpoint and are never placed in + the request body, logs, cache keys, or provider request; the in-memory cache scope uses only a + process-random keyed digest of the caller credential and account; +- recovered plaintext is never logged or persisted; the process-local cache is credential-, parent- + thread-, and ciphertext-scoped, expires after 15 minutes, and is bounded by both configured entry + count (200 by default, 512 maximum) and 8 MiB total; +- any malformed envelope, failed recovery, timeout, or validation failure preserves the existing + fail-closed error; client cancellation returns 499. Neither path forwards ciphertext to the + routed provider. + +### Threat model + +This path assumes the local native Codex caller already holds a valid ChatGPT credential and that +the fixed ChatGPT endpoint is trusted to authenticate it. It protects against generic proxy/API-key +callers using the feature as a plaintext oracle, redirecting credentials to another destination, +cross-account or cross-thread cache reuse, and sensitive-data logging or persistence. Admission +checks token issuer, audience, Codex client, expiry/not-before bounds, and exact account match before +every cache lookup; the endpoint remains the signature authority. + +It does not protect against another process running as the same OS user, a compromised ChatGPT +backend or recovery model, prompt injection inside the encrypted task, model transcription errors, +or memory inspection of the running proxy. Recovery output must therefore be treated as untrusted +model output rather than authenticated plaintext. + +```json +{ + "agentTaskRecovery": { + "enabled": true, + "model": "gpt-5.6-sol", + "timeoutMs": 45000, + "cacheEntries": 200 + } +} +``` + +Enable this only when the additional authenticated request, quota use, plaintext-in-process boundary, +and private-backend dependency are acceptable. Prefer a native ChatGPT child or v1 heterogeneous +delegation when they are not. + +This recovery path applies to direct routed children. At most 32 recovery requests can be active at +once; additional misses fail closed. Combo routing keeps its existing native-only filter for +encrypted tasks and does not invoke recovery. + ## Effort caps Caps apply only to the v2 collaboration feature: a main turn qualifies when its tools expose v2, diff --git a/docs-site/src/content/docs/ru/guides/sub-agent-surface.md b/docs-site/src/content/docs/ru/guides/sub-agent-surface.md index 416c6d9d02..a8be75722a 100644 --- a/docs-site/src/content/docs/ru/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/ru/guides/sub-agent-surface.md @@ -132,6 +132,12 @@ opencodex завершаетcя безопасно и не пересылает combo, использовать v1 для делегирования между разнородными провайдерами либо повторно отправить задачу как plaintext в содержимом v2 `agent_message`, если вы управляете вызывающей стороной. +Экспериментальная опция `agentTaskRecovery` по умолчанию выключена. После явного включения она +может восстановить этот формат дополнительным аутентифицированным запросом к фиксированному +endpoint ChatGPT, но расходует квоту, добавляет задержку и зависит от закрытого поведения backend. +При любом сбое сохраняется прежняя ошибка `unreadable_encrypted_agent_task`. Подробности приведены +в [английском справочнике конфигурации](/reference/configuration/agents/#encrypted-v2-task-recovery). + ## Как сменить режим ### GUI diff --git a/docs-site/src/content/docs/zh-cn/guides/sub-agent-surface.md b/docs-site/src/content/docs/zh-cn/guides/sub-agent-surface.md index 2ac9637416..89a5996b3f 100644 --- a/docs-site/src/content/docs/zh-cn/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/zh-cn/guides/sub-agent-surface.md @@ -86,6 +86,8 @@ opencodex 会安全失败，而不是转发空任务或不可读任务： 恢复选项是选择原生 ChatGPT 子级、在 combo 中添加原生 ChatGPT 目标、在异构 provider 委派中使用 v1，或者在你控制调用方时将任务作为明文 v2 `agent_message` 内容重新发送。 +实验性的 `agentTaskRecovery` 默认关闭。显式启用后，它可以通过向固定 ChatGPT 端点发送额外的认证请求来恢复这种格式，但会消耗配额、增加延迟，并依赖非公开的后端行为。任何失败都会保留原有的 `unreadable_encrypted_agent_task` 错误。详见[英文配置参考](/reference/configuration/agents/#encrypted-v2-task-recovery)。 + ## 更改模式 ### GUI diff --git a/docs-site/src/content/docs/zh-tw/guides/sub-agent-surface.md b/docs-site/src/content/docs/zh-tw/guides/sub-agent-surface.md index d67957018b..e4affe9649 100644 --- a/docs-site/src/content/docs/zh-tw/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/zh-tw/guides/sub-agent-surface.md @@ -6,7 +6,7 @@ description: 全域控制 Codex 在所有模型上生成和管理子代理的方 opencodex 允許你為目錄中的所有模型選擇多代理協作介面。儀表板和 Models 頁面中的 **Sub-agent** 開關會全域控制這一設定。 :::note -在 v2 介面（`multi_agent_v2`）上，子代理**預設**繼承父會話的模型：`fork_turns` 預設為 `all`，而全量歷史 fork 會拒絕覆蓋。自 v2.7.2 起，opencodex 注入的指引會教模型如何打破繼承 —— 將 `fork_turns` 設為 `"none"`（或如 `"3"` 的部分 fork）的 `spawn_agent` 呼叫可以傳入 `model` / `reasoning_effort` 引數；即使公開的工具 schema 中看不到這些引數，Codex 執行環境也會解析並應用。已知傳輸限制：當**原生**父代理 spawn 一個路由到**非原生** provider 的子代理時，Codex 用戶端可能只以後端加密的 `encrypted_content` 傳送 `NEW_TASK` 載荷（[#92](https://github.com/lidge-jun/opencodex/issues/92)）。opencodex 不會把這種無法讀取的任務轉發給外部 provider：直接路由會回傳 HTTP 400 和錯誤碼 `unreadable_encrypted_agent_task`；組合路由則會跳過無法解密的目標，並在存在可用目標時選擇規範的原生 ChatGPT 目標。恢復方法：異構 provider 委派改用 v1、選擇原生 ChatGPT 子代理，或將任務重新作為明文 v2 `agent_message` 內容傳送。 +在 v2 介面（`multi_agent_v2`）上，子代理**預設**繼承父會話的模型：`fork_turns` 預設為 `all`，而全量歷史 fork 會拒絕覆蓋。自 v2.7.2 起，opencodex 注入的指引會教模型如何打破繼承 —— 將 `fork_turns` 設為 `"none"`（或如 `"3"` 的部分 fork）的 `spawn_agent` 呼叫可以傳入 `model` / `reasoning_effort` 引數；即使公開的工具 schema 中看不到這些引數，Codex 執行環境也會解析並應用。已知傳輸限制：當**原生**父代理 spawn 一個路由到**非原生** provider 的子代理時，Codex 用戶端可能只以後端加密的 `encrypted_content` 傳送 `NEW_TASK` 載荷（[#92](https://github.com/lidge-jun/opencodex/issues/92)）。opencodex 不會把這種無法讀取的任務轉發給外部 provider：直接路由會回傳 HTTP 400 和錯誤碼 `unreadable_encrypted_agent_task`；組合路由則會跳過無法解密的目標，並在存在可用目標時選擇規範的原生 ChatGPT 目標。恢復方法：異構 provider 委派改用 v1、選擇原生 ChatGPT 子代理，或將任務重新作為明文 v2 `agent_message` 內容傳送。另有預設停用的實驗性 `agentTaskRecovery`；它會增加 ChatGPT 配額用量與延遲，且依賴非公開後端行為。 ::: ## What sub-agents are @@ -106,6 +106,10 @@ opencodex 會安全失敗，而不是轉發空或無法讀取的任務： 恢復方法：選擇原生 ChatGPT 子代理、在組合中加入原生 ChatGPT 目標、異構 provider 委派改用 v1， 或在你能控制呼叫方時將任務重新作為明文 v2 `agent_message` 內容傳送。 +實驗性的 `agentTaskRecovery` 預設停用。明確啟用後，它可透過固定 ChatGPT 端點的額外已驗證請求 +恢復此格式，但會消耗配額、增加延遲，並依賴非公開後端行為。任何失敗都保留原本的 +`unreadable_encrypted_agent_task` 錯誤。詳見[英文設定參考](/reference/configuration/agents/#encrypted-v2-task-recovery)。 + ## 更改模式 ### GUI diff --git a/src/config.ts b/src/config.ts index 7f99b2e470..b1ab384bdd 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1130,6 +1130,13 @@ const clientIntegrationsSchema = z.object({ "claude-desktop": z.boolean().optional().catch(undefined), }).passthrough(); +const agentTaskRecoverySchema = z.object({ + enabled: z.boolean().optional(), + model: z.string().trim().min(1).optional(), + timeoutMs: z.number().int().min(1_000).max(120_000).optional(), + cacheEntries: z.number().int().min(1).max(512).optional(), +}).strict(); + const configSchema = z.object({ port: z.number().int().min(0).max(65535).default(10100), managementUsageMaxReadBytes: z.number().int().positive().default(64 * 1024 * 1024), @@ -1168,6 +1175,8 @@ const configSchema = z.object({ providerContextCaps: z.record(z.string(), z.number().int().positive()).optional(), contextCapValue: z.number().int().positive().optional(), multiAgentGuidanceEnabled: z.boolean().optional(), + // Invalid optional recovery config must not discard unrelated provider/account state. + agentTaskRecovery: agentTaskRecoverySchema.optional().catch(undefined), // These selections pre-date schema validation and used to pass through as // unknown fields. Invalid hand edits must disable only the optional // delegation/native-default feature, not reject the whole config and hide @@ -1903,6 +1912,20 @@ function warnDegradedUpstreamHostCircuitThreshold(rawParsed: unknown): void { if (warning) console.warn(`⚠️ config.json ${warning}. Other settings were preserved.`); } +function malformedAgentTaskRecoveryWarning(rawParsed: unknown): string | null { + const raw = rawConfigRecord(rawParsed); + if (!raw || !Object.hasOwn(raw, "agentTaskRecovery")) return null; + const result = agentTaskRecoverySchema.safeParse(raw.agentTaskRecovery); + if (result.success) return null; + const field = result.error.issues[0]?.path.join("."); + return `agentTaskRecovery${field ? `.${field}` : ""} ignored: invalid experimental recovery configuration`; +} + +function warnDegradedAgentTaskRecovery(rawParsed: unknown): void { + const warning = malformedAgentTaskRecoveryWarning(rawParsed); + if (warning) console.warn(`⚠️ config.json ${warning}. Other settings were preserved.`); +} + type NativeSubagentPersistedField = "injectionModel" | "injectionEffort" | "syncCodexSubagentDefaults"; function rawConfigRecord(rawParsed: unknown): Record | null { @@ -2005,6 +2028,7 @@ export function loadConfig(): OcxConfig { warnDegradedNativeSubagentConfig(parsed, config); warnDegradedCodexAccountPicker(parsed); warnDegradedUpstreamHostCircuitThreshold(parsed); + warnDegradedAgentTaskRecovery(parsed); return withRefreshedCostOverlays(normalizeClaudeSubagentEffort(normalizeNativeSubagentSync(config, parsed), parsed)); } // Schema validation failed — merge defaults into the raw object instead of @@ -2027,6 +2051,7 @@ export function loadConfig(): OcxConfig { warnDegradedNativeSubagentConfig(parsed, config); warnDegradedCodexAccountPicker(parsed); warnDegradedUpstreamHostCircuitThreshold(parsed); + warnDegradedAgentTaskRecovery(parsed); return withRefreshedCostOverlays(normalizeClaudeSubagentEffort(normalizeNativeSubagentSync(config, parsed), parsed)); } // Merge couldn't fix it — truly broken config @@ -2086,6 +2111,8 @@ function validFileConfigDiagnostics(config: OcxConfig, rawParsed: unknown): Conf if (pickerWarning) warnings.push(pickerWarning); const hostCircuitWarning = malformedUpstreamHostCircuitThresholdWarning(rawParsed); if (hostCircuitWarning) warnings.push(hostCircuitWarning); + const recoveryWarning = malformedAgentTaskRecoveryWarning(rawParsed); + if (recoveryWarning) warnings.push(recoveryWarning); if (syncDisabledReason) { warnings.push(`syncCodexSubagentDefaults ignored: ${syncDisabledReason}`); } @@ -2167,6 +2194,16 @@ function upstreamHostCircuitThresholdError(value: unknown): string | null { return `schema_invalid: upstreamHostCircuitThreshold: must be an integer from 0 to ${UPSTREAM_HOST_CIRCUIT_MAX_THRESHOLD}`; } +function agentTaskRecoveryError(value: unknown): string | null { + const raw = rawConfigRecord(value); + if (!raw || !Object.hasOwn(raw, "agentTaskRecovery") || raw.agentTaskRecovery === undefined) return null; + const result = agentTaskRecoverySchema.safeParse(raw.agentTaskRecovery); + if (result.success) return null; + const issue = result.error.issues[0]; + const field = issue?.path.join("."); + return `schema_invalid: agentTaskRecovery${field ? `.${field}` : ""}: ${issue?.message ?? "invalid configuration"}`; +} + /** * Same reasoning as {@link blankHostnameError}, and more urgent: the read path degrades a * malformed selection-order map to undefined, which on a write would drop every entry the @@ -2262,6 +2299,7 @@ export function validateConfigCandidate(value: unknown): { ok: true; config: Ocx ?? claudeSubagentEffortError(value) ?? appOwnedMemoryBudgetError(value) ?? upstreamHostCircuitThresholdError(value) + ?? agentTaskRecoveryError(value) ?? googleAntigravityStaticCatalogVersionError(value) ?? codexAccountPrioritiesError(value) ?? codexAccountPickerEnabledError(value) diff --git a/src/server/index.ts b/src/server/index.ts index 96520bf156..905fbed2dd 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -481,6 +481,11 @@ export function consumeStartupCacheInvalidationWrite(): boolean { export function startServer(port?: number, deps: StartServerDeps = {}): Server { const localAttestationSecret = deps.localAttestationSecret ?? createLocalAttestationSecret(); const config = runAlibabaRegionStartupMigration(runOpenAiTierStartupMigration(loadConfig())); + if (config.agentTaskRecovery?.enabled === true) { + console.warn("⚠️ Experimental encrypted V2 task recovery is enabled."); + console.warn(" Each scoped cache miss sends an additional authenticated request to ChatGPT and may consume quota or add latency."); + console.warn(" Recovered model output is retained only in a bounded in-memory cache; exact fidelity is not guaranteed and the path depends on undocumented backend behavior."); + } setLiveStateStoreConfig(config); applyProxyEnv(config); assertServerAuthConfig(config); diff --git a/src/server/responses/agent-task-recovery-cache.ts b/src/server/responses/agent-task-recovery-cache.ts new file mode 100644 index 0000000000..709bbe787b --- /dev/null +++ b/src/server/responses/agent-task-recovery-cache.ts @@ -0,0 +1,137 @@ +const MAX_CACHE_BYTES = 8 * 1024 * 1024; +const MAX_CONCURRENT_RECOVERIES = 32; +const CACHE_TTL_MS = 15 * 60 * 1000; + +interface RecoveryCacheEntry { + assignment: string; + bytes: number; + expiresAt: number; + expiryTimer: ReturnType | null; +} + +interface RecoveryFlight { + controller: AbortController; + promise: Promise; + waiters: number; + settled: boolean; +} + +const RECOVERY_CACHE = new Map(); +const RECOVERY_FLIGHTS = new Map(); +let recoveryCacheBytes = 0; + +function deleteRecoveryCacheEntry(key: string, expected?: RecoveryCacheEntry): void { + const entry = RECOVERY_CACHE.get(key); + if (!entry || (expected && entry !== expected)) return; + RECOVERY_CACHE.delete(key); + if (entry.expiryTimer) clearTimeout(entry.expiryTimer); + recoveryCacheBytes = Math.max(0, recoveryCacheBytes - entry.bytes); +} + +function sweepRecoveryCache(now: number, maxEntries: number): void { + for (const [key, entry] of RECOVERY_CACHE) { + if (entry.expiresAt > now) continue; + deleteRecoveryCacheEntry(key, entry); + } + while (RECOVERY_CACHE.size > maxEntries || recoveryCacheBytes > MAX_CACHE_BYTES) { + const oldest = RECOVERY_CACHE.keys().next().value; + if (oldest === undefined) break; + deleteRecoveryCacheEntry(oldest); + } +} + +function insertRecoveryCacheEntry(key: string, assignment: string, maxEntries: number): void { + const replaced = RECOVERY_CACHE.get(key); + if (replaced) deleteRecoveryCacheEntry(key, replaced); + const insertedAt = Date.now(); + const entry: RecoveryCacheEntry = { + assignment, + bytes: Buffer.byteLength(assignment), + expiresAt: insertedAt + CACHE_TTL_MS, + expiryTimer: null, + }; + entry.expiryTimer = setTimeout( + () => deleteRecoveryCacheEntry(key, entry), + CACHE_TTL_MS, + ); + entry.expiryTimer.unref?.(); + RECOVERY_CACHE.set(key, entry); + recoveryCacheBytes += entry.bytes; + sweepRecoveryCache(insertedAt, maxEntries); +} + +function startRecoveryFlight( + key: string, + maxEntries: number, + request: (signal: AbortSignal) => Promise, +): RecoveryFlight | null { + const active = RECOVERY_FLIGHTS.get(key); + if (active) return active; + if (RECOVERY_FLIGHTS.size >= MAX_CONCURRENT_RECOVERIES) return null; + + const controller = new AbortController(); + const flight: RecoveryFlight = { + controller, + promise: Promise.resolve(null), + waiters: 0, + settled: false, + }; + flight.promise = request(controller.signal) + .then((assignment) => { + if (!assignment || controller.signal.aborted) return null; + insertRecoveryCacheEntry(key, assignment, maxEntries); + return assignment; + }) + .finally(() => { + flight.settled = true; + if (RECOVERY_FLIGHTS.get(key) === flight) RECOVERY_FLIGHTS.delete(key); + }); + RECOVERY_FLIGHTS.set(key, flight); + return flight; +} + +async function waitForRecoveryFlight( + flight: RecoveryFlight, + abortSignal?: AbortSignal, +): Promise { + if (abortSignal?.aborted) return null; + flight.waiters += 1; + let onAbort: (() => void) | undefined; + try { + if (!abortSignal) return await flight.promise; + const cancelled = new Promise((resolve) => { + onAbort = () => resolve(null); + abortSignal.addEventListener("abort", onAbort, { once: true }); + if (abortSignal.aborted) onAbort(); + }); + return await Promise.race([flight.promise, cancelled]); + } finally { + if (onAbort) abortSignal?.removeEventListener("abort", onAbort); + flight.waiters = Math.max(0, flight.waiters - 1); + if (flight.waiters === 0 && !flight.settled) { + flight.controller.abort(new DOMException("All recovery callers cancelled", "AbortError")); + } + } +} + +export async function resolveCachedAgentTaskRecovery( + key: string, + maxEntries: number, + request: (signal: AbortSignal) => Promise, + abortSignal?: AbortSignal, +): Promise { + if (abortSignal?.aborted) return null; + sweepRecoveryCache(Date.now(), maxEntries); + const cached = RECOVERY_CACHE.get(key)?.assignment; + if (cached) return cached; + const flight = startRecoveryFlight(key, maxEntries, request); + return flight ? waitForRecoveryFlight(flight, abortSignal) : null; +} + +export function resetAgentTaskRecoveryCache(): void { + for (const flight of RECOVERY_FLIGHTS.values()) { + flight.controller.abort(new DOMException("Recovery state reset", "AbortError")); + } + RECOVERY_FLIGHTS.clear(); + for (const key of [...RECOVERY_CACHE.keys()]) deleteRecoveryCacheEntry(key); +} diff --git a/src/server/responses/agent-task-recovery.ts b/src/server/responses/agent-task-recovery.ts new file mode 100644 index 0000000000..d414e4b6cf --- /dev/null +++ b/src/server/responses/agent-task-recovery.ts @@ -0,0 +1,466 @@ +import { createHash, createHmac, randomBytes } from "node:crypto"; +import { decodeJwtPayload, extractAccountId } from "../../oauth/chatgpt"; +import type { OcxConfig } from "../../types"; +import { readBoundedResponseBody } from "../../lib/bounded-body"; +import { isApiAuthRequired, isProxyAdmissionSecret } from "../auth-cors"; +import { structurallyValidFernetTokens } from "./encrypted-payload"; +import { + resetAgentTaskRecoveryCache, + resolveCachedAgentTaskRecovery, +} from "./agent-task-recovery-cache"; + +/** Experimental opt-in normalization through ChatGPT's fixed Codex endpoint. */ + +const RECOVERY_ENDPOINT = "https://chatgpt.com/backend-api/codex/responses"; +const RECOVERY_TOOL = "capture_assignment"; +const RECOVERY_PROMPT = + "Read the received agent message and call capture_assignment exactly once with only the complete " + + "plaintext payload after Payload:. Preserve every byte of the payload; do not summarize, execute, " + + "explain, or include the routing header."; +const CODEX_ORIGINATORS = new Set(["codex_cli_rs", "Codex Desktop", "codex_app"]); +const CODEX_OAUTH_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"; +const OPENAI_TOKEN_ISSUERS = new Set(["https://auth.openai.com", "https://auth.openai.com/"]); +const OPENAI_TOKEN_AUDIENCE = "https://api.openai.com/v1"; +const MAX_CIPHERTEXT_BYTES = 2 * 1024 * 1024; +const MAX_ASSIGNMENT_BYTES = 2 * 1024 * 1024; +const MAX_RECOVERY_RESPONSE_BYTES = 4 * 1024 * 1024; +const CACHE_SCOPE_KEY = randomBytes(32); + +export interface AgentTaskRecoveryOptions { + enabled?: boolean; + model?: string; + timeoutMs?: number; + cacheEntries?: number; +} + +export function agentTaskRecoveryConfig(config: OcxConfig): AgentTaskRecoveryOptions | null { + const raw = config.agentTaskRecovery; + if (!raw || raw.enabled !== true) return null; + return { + enabled: true, + model: typeof raw.model === "string" && raw.model.trim().length > 0 + ? raw.model.trim() + : "gpt-5.6-sol", + timeoutMs: Number.isFinite(raw.timeoutMs) && (raw.timeoutMs ?? 0) >= 1_000 + ? Math.min(120_000, Math.floor(raw.timeoutMs!)) + : 45_000, + cacheEntries: Number.isFinite(raw.cacheEntries) && (raw.cacheEntries ?? 0) >= 1 + ? Math.min(512, Math.floor(raw.cacheEntries!)) + : 200, + }; +} + +interface AgentEnvelope { + itemIndex: number; + encryptedIndex: number; + encryptedSlot: string; + headerText: string; + messageType: "NEW_TASK"; + taskName: string; + sender: string; + ciphertext: string; + author: string; + recipient: string; +} + +const ROUTING_HEADER = /(?:^|\n)Message Type\s*:\s*(NEW_TASK)\s*\nTask name\s*:\s*(\S+)\s*\nSender\s*:\s*(\S+)\s*\nPayload\s*:\s*(?:\n|$)/; + +function findEnvelope(input: unknown): AgentEnvelope | null { + if (!Array.isArray(input)) return null; + let itemIndex = input.length - 1; + while (itemIndex >= 0) { + const type = input[itemIndex] && typeof input[itemIndex] === "object" + ? (input[itemIndex] as { type?: unknown }).type + : undefined; + if (type !== "compaction_trigger" && type !== "additional_tools") break; + itemIndex -= 1; + } + + const item = input[itemIndex]; + if (!item || typeof item !== "object" || (item as { type?: unknown }).type !== "agent_message") { + return null; + } + const content = (item as { content?: unknown }).content; + if (!Array.isArray(content)) return null; + + let headerText: string | null = null; + let messageType: "NEW_TASK" | null = null; + let taskName: string | null = null; + let sender: string | null = null; + let encryptedIndex = -1; + let encryptedSlot = ""; + let ciphertext = ""; + let encryptedPartCount = 0; + let ciphertextCount = 0; + + for (let index = 0; index < content.length; index += 1) { + const part = content[index] as { type?: unknown; text?: unknown; encrypted_content?: unknown } | null; + if (!part) continue; + if ( + (part.type === "input_text" || part.type === "text") + && typeof part.text === "string" + ) { + const match = ROUTING_HEADER.exec(part.text); + if (match) { + if (headerText !== null) return null; + if ( + part.text.slice(0, match.index).trim().length > 0 + || part.text.slice(match.index + match[0].length).trim().length > 0 + ) return null; + headerText = match[0].startsWith("\n") ? match[0].slice(1) : match[0]; + messageType = "NEW_TASK"; + taskName = match[2]!; + sender = match[3]!; + } + } + if (part.type !== "encrypted_content" || typeof part.encrypted_content !== "string") continue; + encryptedPartCount += 1; + for (const token of structurallyValidFernetTokens(part.encrypted_content)) { + ciphertextCount += 1; + encryptedIndex = index; + encryptedSlot = part.encrypted_content; + ciphertext = token; + } + } + + if ( + !headerText + || !messageType + || !taskName + || !sender + || encryptedIndex < 0 + || encryptedPartCount !== 1 + || ciphertextCount !== 1 + || Buffer.byteLength(ciphertext) > MAX_CIPHERTEXT_BYTES + ) return null; + + const itemRecord = item as { author?: unknown; recipient?: unknown }; + if (typeof itemRecord.author !== "string" || typeof itemRecord.recipient !== "string") return null; + if (itemRecord.author !== sender || itemRecord.recipient !== taskName) return null; + + return { + itemIndex, + encryptedIndex, + encryptedSlot, + headerText, + messageType, + taskName, + sender, + ciphertext, + author: itemRecord.author, + recipient: itemRecord.recipient, + }; +} + +function stripMatchingEnvelope(assignment: string, envelope: AgentEnvelope): string | null { + const match = ROUTING_HEADER.exec(assignment); + if (!match) return assignment; + if (match.index !== 0) return null; + if ( + match[1] !== envelope.messageType + || match[2] !== envelope.taskName + || match[3] !== envelope.sender + ) return null; + return assignment.slice(match[0].length); +} + +function validateAssignment(assignment: unknown, envelope: AgentEnvelope): string | null { + if (typeof assignment !== "string") return null; + const payload = stripMatchingEnvelope(assignment, envelope); + if (payload === null || payload.trim().length === 0) return null; + if (Buffer.byteLength(payload) > MAX_ASSIGNMENT_BYTES) return null; + if (structurallyValidFernetTokens(payload).length > 0) return null; + return payload; +} + +function injectAssignment(input: unknown, envelope: AgentEnvelope, assignment: string): boolean { + if (!Array.isArray(input)) return false; + const item = input[envelope.itemIndex]; + if (!item || typeof item !== "object") return false; + const content = (item as { content?: unknown }).content; + if (!Array.isArray(content)) return false; + const part = content[envelope.encryptedIndex] as { type?: unknown; encrypted_content?: unknown } | undefined; + if ( + !part + || part.type !== "encrypted_content" + || part.encrypted_content !== envelope.encryptedSlot + ) return false; + + const text = envelope.encryptedSlot === envelope.ciphertext + ? assignment + : envelope.encryptedSlot.replace(envelope.ciphertext, assignment); + content[envelope.encryptedIndex] = { type: "input_text", text }; + const message = item as Record; + message.type = "message"; + message.role = "user"; + delete message.id; + delete message.author; + delete message.recipient; + return true; +} + +interface RecoveryAdmission { + headers: Headers; + cacheScope: string; +} + +function isNativeChatGptAccessToken(token: string): boolean { + const segments = token.split("."); + if (segments.length !== 3 || !segments[0] || !segments[1] || !segments[2]) return false; + let header: Record; + try { + header = JSON.parse(Buffer.from(segments[0], "base64url").toString("utf8")) as Record; + } catch { + return false; + } + if (header.alg !== "RS256" || header.typ !== "JWT" || typeof header.kid !== "string" || !header.kid) { + return false; + } + const payload = decodeJwtPayload(token); + if (!payload || !OPENAI_TOKEN_ISSUERS.has(payload.iss as string)) return false; + const audiences = Array.isArray(payload.aud) ? payload.aud : [payload.aud]; + if (!audiences.includes(OPENAI_TOKEN_AUDIENCE)) return false; + if (payload.client_id !== CODEX_OAUTH_CLIENT_ID && payload.azp !== CODEX_OAUTH_CLIENT_ID) return false; + const now = Math.floor(Date.now() / 1_000); + if (typeof payload.exp !== "number" || !Number.isFinite(payload.exp) || payload.exp <= now) return false; + if ( + payload.nbf !== undefined + && (typeof payload.nbf !== "number" || !Number.isFinite(payload.nbf) || payload.nbf > now + 60) + ) return false; + const auth = payload["https://api.openai.com/auth"]; + return !!auth && typeof auth === "object" && !Array.isArray(auth); +} + +function recoveryAdmission(req: Request, config: OcxConfig): RecoveryAdmission | null { + if (isApiAuthRequired(config)) return null; + if (!CODEX_ORIGINATORS.has(req.headers.get("originator") ?? "")) return null; + // Remote/shared proxy admission is intentionally unsupported: caller-controlled + // Codex metadata is not strong enough to authorize use of a stored ChatGPT session. + if (req.headers.has("x-opencodex-api-key") || req.headers.has("x-api-key")) return null; + + const authorization = req.headers.get("authorization")?.trim() ?? ""; + const match = /^Bearer\s+(\S+)$/i.exec(authorization); + if (!match) return null; + const token = match[1]!; + if (isProxyAdmissionSecret(token, config)) return null; + if (!isNativeChatGptAccessToken(token)) return null; + const accountId = extractAccountId(undefined, token); + const explicitAccountId = req.headers.get("chatgpt-account-id")?.trim(); + if (!accountId || !explicitAccountId || accountId !== explicitAccountId) return null; + + const headers = new Headers({ + authorization: `Bearer ${token}`, + "chatgpt-account-id": explicitAccountId, + "content-type": "application/json", + accept: "text/event-stream", + originator: req.headers.get("originator")!, + }); + for (const name of ["openai-beta", "user-agent"]) { + const value = req.headers.get(name); + if (value) headers.set(name, value); + } + const cacheScope = createHmac("sha256", CACHE_SCOPE_KEY) + .update(token) + .update("\0") + .update(explicitAccountId) + .digest("hex"); + return { headers, cacheScope }; +} + +function recoveryPayload(envelope: AgentEnvelope, model: string): string { + return JSON.stringify({ + model, + stream: true, + store: false, + instructions: RECOVERY_PROMPT, + tools: [{ + type: "function", + name: RECOVERY_TOOL, + description: "Return only the exact decrypted agent task payload.", + parameters: { + type: "object", + properties: { assignment: { type: "string" } }, + required: ["assignment"], + additionalProperties: false, + }, + strict: true, + }], + tool_choice: { type: "function", name: RECOVERY_TOOL }, + input: [{ + type: "agent_message", + author: envelope.author, + recipient: envelope.recipient, + content: [ + { type: "input_text", text: envelope.headerText }, + { type: "encrypted_content", encrypted_content: envelope.ciphertext }, + ], + }], + }); +} + +function sseDataPayloads(raw: string): string[] { + const payloads: string[] = []; + let data: string[] = []; + const dispatch = (): void => { + if (data.length > 0) payloads.push(data.join("\n")); + data = []; + }; + for (const line of raw.replace(/\r\n?/g, "\n").split("\n")) { + if (line === "") { + dispatch(); + continue; + } + if (line.startsWith(":")) continue; + if (line === "data") { + data.push(""); + continue; + } + if (!line.startsWith("data:")) continue; + const value = line.slice(5); + data.push(value.startsWith(" ") ? value.slice(1) : value); + } + dispatch(); + return payloads; +} + +function assignmentFromRecoverySse(raw: string, envelope: AgentEnvelope): string | null { + let assignment: string | null = null; + let completed = false; + let terminalFailure = false; + let conflictingAssignments = false; + let malformedEvent = false; + let invalidAssignment = false; + for (const data of sseDataPayloads(raw)) { + if (!data || data === "[DONE]") continue; + let event: any; + try { event = JSON.parse(data); } catch { + malformedEvent = true; + continue; + } + if ( + event?.type === "response.failed" + || event?.type === "response.incomplete" + || event?.type === "error" + ) terminalFailure = true; + if (event?.type === "response.completed" && event.response?.status === "completed") { + completed = true; + } + const items = event?.type === "response.output_item.done" + ? [event.item] + : event?.type === "response.function_call_arguments.done" + ? [{ type: "function_call", name: event.name, arguments: event.arguments }] + : event?.type === "response.completed" + ? (Array.isArray(event.response?.output) ? event.response.output : []).filter((candidate: any) => ( + candidate?.type === "function_call" && candidate?.name === RECOVERY_TOOL + )) + : []; + for (const item of items) { + if (item?.type !== "function_call" || item.name !== RECOVERY_TOOL) continue; + let args: unknown = item.arguments; + if (typeof args === "string") { + try { args = JSON.parse(args); } catch { + invalidAssignment = true; + continue; + } + } + if (!args || typeof args !== "object") { + invalidAssignment = true; + continue; + } + const candidate = validateAssignment((args as { assignment?: unknown }).assignment, envelope); + if (candidate === null) { + invalidAssignment = true; + continue; + } + if (assignment === null) assignment = candidate; + else if (assignment !== candidate) conflictingAssignments = true; + } + } + return completed && !terminalFailure && !conflictingAssignments && !malformedEvent && !invalidAssignment + ? assignment + : null; +} + +async function requestRecovery( + admission: RecoveryAdmission, + envelope: AgentEnvelope, + options: AgentTaskRecoveryOptions, + abortSignal?: AbortSignal, +): Promise { + const controller = new AbortController(); + const timeout = setTimeout( + () => controller.abort(new DOMException("Agent task recovery timed out", "TimeoutError")), + options.timeoutMs ?? 45_000, + ); + const signal = abortSignal + ? AbortSignal.any([abortSignal, controller.signal]) + : controller.signal; + try { + const response = await fetch(RECOVERY_ENDPOINT, { + method: "POST", + headers: admission.headers, + body: recoveryPayload(envelope, options.model ?? "gpt-5.6-sol"), + signal, + redirect: "error", + }); + if (!response.ok) { + try { await response.body?.cancel(); } catch { /* already closed */ } + return null; + } + const body = await readBoundedResponseBody(response, { + signal, + fatalUtf8: true, + maxBytes: MAX_RECOVERY_RESPONSE_BYTES, + totalTimeoutMs: options.timeoutMs ?? 45_000, + inactivityTimeoutMs: options.timeoutMs ?? 45_000, + firstByteTimeoutMs: options.timeoutMs ?? 45_000, + }); + if (body.truncated || body.oversized || body.timedOut || !body.displaySafe) return null; + return assignmentFromRecoverySse(body.text, envelope); + } catch { + return null; + } finally { + clearTimeout(timeout); + } +} + +export async function recoverEncryptedAgentTask( + req: Request, + input: unknown, + options: AgentTaskRecoveryOptions, + config: OcxConfig, + context: { parentThreadId?: string | null; abortSignal?: AbortSignal } = {}, +): Promise { + const envelope = findEnvelope(input); + if (!envelope) return false; + // Admission is deliberately checked before cache access. A cache hit must not + // turn this process into a plaintext oracle for an unauthenticated caller. + const admission = recoveryAdmission(req, config); + if (!admission) return false; + + const cacheKey = createHash("sha256") + .update(admission.cacheScope) + .update("\0") + .update(context.parentThreadId ?? "") + .update("\0") + .update(envelope.messageType) + .update("\0") + .update(envelope.taskName) + .update("\0") + .update(envelope.sender) + .update("\0") + .update(envelope.ciphertext) + .digest("hex"); + const assignment = await resolveCachedAgentTaskRecovery( + cacheKey, + options.cacheEntries ?? 200, + signal => requestRecovery(admission, envelope, options, signal), + context.abortSignal, + ); + if (!assignment || context.abortSignal?.aborted) return false; + return injectAssignment(input, envelope, assignment); +} + +export function resetAgentTaskRecoveryState(): void { + resetAgentTaskRecoveryCache(); +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 42e0e193b3..f6d2571466 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -177,6 +177,10 @@ import { relayWithAbort, sanitizePassthroughHeaders, } from "../relay"; +import { + agentTaskRecoveryConfig, + recoverEncryptedAgentTask, +} from "./agent-task-recovery"; import { relaySseEagerBounded } from "../relay-eager"; import { relayResponsesSseWithTerminalRepair, @@ -1484,6 +1488,7 @@ async function handleResponsesInner( // so an omitted value means a genuine Responses inbound. const inboundWire = options.inboundWire ?? "responses"; const translatorBudget = options.translatorBudget; + const agentTaskRecovery = agentTaskRecoveryConfig(config); let body: unknown; try { body = await readJsonRequestBody(req, translatorBudget); @@ -1494,7 +1499,7 @@ async function handleResponsesInner( if (comboId && Object.hasOwn(config.combos ?? {}, comboId)) { return handleComboResponses(req, body, comboId, config, logCtx, options); } - const unreadableEncryptedAgentTask = hasUnreadableEncryptedAgentTask( + let unreadableEncryptedAgentTask = hasUnreadableEncryptedAgentTask( (body as { input?: unknown } | undefined)?.input, ); const originalBody = body; @@ -1622,6 +1627,7 @@ async function handleResponsesInner( let selectedForwardHeaders = req.headers; let subagentFallbackAccountId = config.activeCodexAccountId ?? null; let subagentQuotaFailureModel = parsed.modelId; + const parentThreadId = req.headers.get("x-codex-parent-thread-id")?.trim() ?? null; try { if ( @@ -1683,6 +1689,59 @@ async function handleResponsesInner( previewSelectionAdmission?.release(); } + // Native fallback can consume ciphertext, so recover only after final route selection. + if ( + inboundWire === "responses" + && + threadSpawn + && unreadableEncryptedAgentTask + && agentTaskRecovery + && !isCanonicalOpenAiForwardProvider(route.provider) + && !options.comboAttempt + ) { + let recovered = false; + try { + recovered = await recoverEncryptedAgentTask( + req, + (body as { input?: unknown } | undefined)?.input, + agentTaskRecovery, + config, + { parentThreadId, abortSignal: options.abortSignal }, + ); + } catch { + recovered = false; + } + if (recovered) { + unreadableEncryptedAgentTask = hasUnreadableEncryptedAgentTask( + (body as { input?: unknown } | undefined)?.input, + ); + if (!unreadableEncryptedAgentTask) { + try { + const reparsed = parseRequest(body); + const kept: Array = [ + "_previousResponseInputExpanded", + "_providerContinuation", + "_cursorConversationId", + "_clientThreadId", + "_reasoningReplayScope", + "_cursorIsolateConversation", + ]; + for (const key of kept) { + if (parsed[key] !== undefined) { + (reparsed as unknown as Record)[key] = parsed[key]; + } + } + parsed = reparsed; + toolBridgeMaps = buildToolBridgeMaps(parsed, translatorBudget); + } catch { + unreadableEncryptedAgentTask = true; + } + } + } + } + + if (options.abortSignal?.aborted) return clientCancelledResponse(); + // Encrypted child tasks may only reach the canonical native backend. This check // runs against the FINAL route so native-only fallback can rescue a routed primary. if (!isCanonicalOpenAiForwardProvider(route.provider) && unreadableEncryptedAgentTask) { diff --git a/src/server/responses/encrypted-payload.ts b/src/server/responses/encrypted-payload.ts index 0466f8f151..70e4473c22 100644 --- a/src/server/responses/encrypted-payload.ts +++ b/src/server/responses/encrypted-payload.ts @@ -145,6 +145,10 @@ function isStructurallyValidFernetToken(token: string): boolean { return ciphertextLength >= 16 && ciphertextLength % 16 === 0; } +export function structurallyValidFernetTokens(payload: string): string[] { + return fernetTokenRuns(payload).map(run => run.token); +} + /** Maximal, boundary-delimited and structurally valid Fernet runs embedded in a slot. */ function fernetTokenRuns(payload: string): FernetTokenRun[] { const runs: FernetTokenRun[] = []; @@ -305,4 +309,3 @@ export function sanitizeEncryptedContentInPlace(input: unknown): number { visit(input); return rewritten; } - diff --git a/src/types.ts b/src/types.ts index 6de3fae29b..d0f9fcb21b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -777,6 +777,16 @@ export interface OcxConfig { * - "v2": force ALL models to v2 surface (override upstream pins) */ multiAgentMode?: "v1" | "default" | "v2"; + /** Experimental, default-off ChatGPT recovery for encrypted V2 routed tasks. */ + agentTaskRecovery?: { + enabled?: boolean; + /** ChatGPT model used by the recovery request. Default: gpt-5.6-sol. */ + model?: string; + /** Recovery request timeout in milliseconds. Default: 45000. */ + timeoutMs?: number; + /** Maximum in-memory ciphertext-to-assignment entries. Default: 200. */ + cacheEntries?: number; + }; /** Provider-level Codex-visible context caps. Values only lower known model context windows. */ providerContextCaps?: Record; /** Global Codex-visible context cap value (tokens). Falls back to DEFAULT_PROVIDER_CONTEXT_CAP. */ diff --git a/tests/agent-task-recovery-cache.test.ts b/tests/agent-task-recovery-cache.test.ts new file mode 100644 index 0000000000..4c255c8a01 --- /dev/null +++ b/tests/agent-task-recovery-cache.test.ts @@ -0,0 +1,95 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { + resetAgentTaskRecoveryCache, + resolveCachedAgentTaskRecovery, +} from "../src/server/responses/agent-task-recovery-cache"; + +const realDateNow = Date.now; + +describe("agent task recovery cache", () => { + beforeEach(() => resetAgentTaskRecoveryCache()); + + afterEach(() => { + Date.now = realDateNow; + resetAgentTaskRecoveryCache(); + }); + + test("expires recovered plaintext after fifteen minutes", async () => { + let now = 1_800_000_000_000; + Date.now = () => now; + let requests = 0; + const recover = async (): Promise => `assignment-${++requests}`; + + expect(await resolveCachedAgentTaskRecovery("task", 200, recover)).toBe("assignment-1"); + expect(await resolveCachedAgentTaskRecovery("task", 200, recover)).toBe("assignment-1"); + now += 15 * 60 * 1000 + 1; + expect(await resolveCachedAgentTaskRecovery("task", 200, recover)).toBe("assignment-2"); + expect(requests).toBe(2); + }); + + test("keeps a shared request alive while one authenticated waiter remains", async () => { + let release: (() => void) | undefined; + const gate = new Promise(resolve => { + release = resolve; + }); + let sharedSignal: AbortSignal | undefined; + let requests = 0; + const recover = async (signal: AbortSignal): Promise => { + requests += 1; + sharedSignal = signal; + await gate; + return "shared-assignment"; + }; + const firstController = new AbortController(); + const first = resolveCachedAgentTaskRecovery("shared", 200, recover, firstController.signal); + const second = resolveCachedAgentTaskRecovery("shared", 200, recover); + + firstController.abort(); + expect(await first).toBeNull(); + expect(sharedSignal?.aborted).toBe(false); + release?.(); + expect(await second).toBe("shared-assignment"); + expect(requests).toBe(1); + }); + + test("fails a thirty-third distinct recovery closed without starting it", async () => { + let release: (() => void) | undefined; + const gate = new Promise(resolve => { + release = resolve; + }); + let requests = 0; + const pending = Array.from({ length: 32 }, (_, index) => ( + resolveCachedAgentTaskRecovery(`task-${index}`, 200, async () => { + requests += 1; + await gate; + return `assignment-${index}`; + }) + )); + + const overflow = await resolveCachedAgentTaskRecovery("task-overflow", 200, async () => { + requests += 1; + return "must-not-run"; + }); + expect(overflow).toBeNull(); + expect(requests).toBe(32); + release?.(); + expect((await Promise.all(pending)).filter(Boolean)).toHaveLength(32); + }); + + test("evicts oldest recovered plaintext when the byte budget is exceeded", async () => { + const assignment = "x".repeat(2 * 1024 * 1024); + let requests = 0; + for (let index = 0; index < 5; index += 1) { + expect((await resolveCachedAgentTaskRecovery(`large-${index}`, 200, async () => { + requests += 1; + return assignment; + }))?.length).toBe(assignment.length); + } + + expect((await resolveCachedAgentTaskRecovery("large-0", 200, async () => { + requests += 1; + return assignment; + }))?.length).toBe(assignment.length); + expect(requests).toBe(6); + }); +}); diff --git a/tests/agent-task-recovery-security.test.ts b/tests/agent-task-recovery-security.test.ts new file mode 100644 index 0000000000..e7de516440 --- /dev/null +++ b/tests/agent-task-recovery-security.test.ts @@ -0,0 +1,393 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { resetAgentTaskRecoveryState } from "../src/server/responses/agent-task-recovery"; +import { + codexHeaders, + encryptedInput, + fakeChatGptJwt, + FERNET_TASK, + originalFetch, + post, + providerResponse, + recoverySse, + routedConfig, + ROUTING_ENVELOPE, +} from "./helpers/agent-task-recovery"; + +const realDateNow = Date.now; + +describe("agent task recovery security", () => { + beforeEach(() => resetAgentTaskRecoveryState()); + + afterEach(() => { + globalThis.fetch = originalFetch; + Date.now = realDateNow; + resetAgentTaskRecoveryState(); + }); + + test("uses only the fixed ChatGPT endpoint and forwards only allowlisted credentials", async () => { + const accountId = "acct-boundary"; + const token = fakeChatGptJwt(accountId); + const assignment = "Keep credentials on their owning transport."; + let recoveryUrl = ""; + let recoveryHeaders = new Headers(); + let recoveryBody = ""; + let recoveryMethod: string | undefined; + let recoveryRedirect: RequestRedirect | undefined; + let providerHeaders = new Headers(); + let providerBody = ""; + globalThis.fetch = (async (input, init) => { + if (String(input).includes("chatgpt.com")) { + recoveryUrl = String(input); + recoveryHeaders = new Headers(init?.headers); + recoveryBody = typeof init?.body === "string" ? init.body : ""; + recoveryMethod = init?.method; + recoveryRedirect = init?.redirect; + return new Response(recoverySse(assignment), { status: 200 }); + } + providerHeaders = new Headers(init?.headers); + providerBody = typeof init?.body === "string" ? init.body : ""; + return providerResponse(); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(accountId, { + "openai-beta": "responses=experimental", + "user-agent": "codex-test/1", + cookie: "private-cookie", + "x-private-caller-header": "private-value", + "x-codex-parent-thread-id": "parent-boundary", + }), + ); + + expect(response.status).toBe(200); + expect(recoveryUrl).toBe("https://chatgpt.com/backend-api/codex/responses"); + expect(recoveryMethod).toBe("POST"); + expect(recoveryRedirect).toBe("error"); + expect([...recoveryHeaders.keys()].sort()).toEqual([ + "accept", + "authorization", + "chatgpt-account-id", + "content-type", + "openai-beta", + "originator", + "user-agent", + ]); + expect(recoveryHeaders.get("authorization")).toBe(`Bearer ${token}`); + expect(recoveryHeaders.get("chatgpt-account-id")).toBe(accountId); + expect(recoveryHeaders.get("originator")).toBe("codex_cli_rs"); + expect(recoveryHeaders.get("openai-beta")).toBe("responses=experimental"); + expect(recoveryHeaders.get("user-agent")).toBe("codex-test/1"); + expect(recoveryHeaders.get("cookie")).toBeNull(); + expect(recoveryHeaders.get("x-private-caller-header")).toBeNull(); + expect(recoveryHeaders.get("x-codex-parent-thread-id")).toBeNull(); + expect(recoveryBody).toContain(FERNET_TASK); + expect(recoveryBody).not.toContain(token); + expect(recoveryBody).not.toContain(accountId); + expect(recoveryBody).not.toContain("private-cookie"); + expect(providerHeaders.get("authorization")).not.toContain(token); + expect(providerHeaders.get("chatgpt-account-id")).toBeNull(); + expect(providerHeaders.get("cookie")).toBeNull(); + expect(providerBody).toContain(assignment); + expect(providerBody).not.toContain(FERNET_TASK); + expect(providerBody).not.toContain(token); + expect(providerBody).not.toContain(accountId); + }); + + test("a cached recovery never bypasses caller authentication", async () => { + const assignment = `${ROUTING_ENVELOPE}Do not expose this cached task.`; + let recoveryFetches = 0; + let providerFetches = 0; + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) { + recoveryFetches += 1; + return new Response(recoverySse(assignment), { status: 200 }); + } + providerFetches += 1; + return providerResponse(); + }) as typeof fetch; + + expect((await post(routedConfig(), "xai/grok-4.5", encryptedInput(), codexHeaders())).status).toBe(200); + const unauthenticated = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + { originator: "codex_cli_rs", "x-openai-subagent": "collab_spawn" }, + ); + + expect(unauthenticated.status).toBe(400); + expect(await unauthenticated.json()).toMatchObject({ + error: { code: "unreadable_encrypted_agent_task" }, + }); + expect(recoveryFetches).toBe(1); + expect(providerFetches).toBe(1); + }); + + test("a proxy admission secret is never forwarded to ChatGPT", async () => { + let forwardedBody = ""; + globalThis.fetch = (async (input, init) => { + if (String(input).includes("chatgpt.com")) { + forwardedBody = typeof init?.body === "string" ? init.body : ""; + } + return new Response("event: error\ndata: {}\n\n", { status: 200 }); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + { + authorization: "Bearer ocx_data_testsecret", + "chatgpt-account-id": "acct-forged", + originator: "codex_cli_rs", + "x-openai-subagent": "collab_spawn", + }, + ); + + expect(response.status).toBe(400); + expect(forwardedBody).toBe(""); + }); + + for (const apiKeyHeader of ["x-opencodex-api-key", "x-api-key"] as const) { + test(`rejects callers admitted through ${apiKeyHeader}`, async () => { + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("recovery must stay unreachable"); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders("acct-api-key", { [apiKeyHeader]: "proxy-key" }), + ); + + expect(response.status).toBe(400); + expect(fetchCalls).toBe(0); + expect(await response.json()).toMatchObject({ + error: { code: "unreadable_encrypted_agent_task" }, + }); + }); + } + + test("rejects opaque bearer tokens and mismatched ChatGPT account headers", async () => { + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("recovery must stay unreachable"); + }) as typeof fetch; + + const opaque = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + { + authorization: "Bearer generic-api-token", + "chatgpt-account-id": "acct-generic", + originator: "codex_cli_rs", + "x-openai-subagent": "collab_spawn", + }, + ); + const mismatchedHeaders = codexHeaders("acct-token"); + mismatchedHeaders.set("chatgpt-account-id", "acct-other"); + const mismatched = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + mismatchedHeaders, + ); + + expect(opaque.status).toBe(400); + expect(mismatched.status).toBe(400); + expect(fetchCalls).toBe(0); + }); + + test("rejects expired ChatGPT tokens and non-loopback proxy binds", async () => { + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("recovery must stay unreachable"); + }) as typeof fetch; + + const expiredHeaders = codexHeaders("acct-expired"); + expiredHeaders.set("authorization", `Bearer ${fakeChatGptJwt("acct-expired", { + exp: Math.floor(Date.now() / 1000) - 1, + })}`); + const expired = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + expiredHeaders, + ); + const remoteConfig = routedConfig(); + remoteConfig.hostname = "0.0.0.0"; + const remote = await post( + remoteConfig, + "xai/grok-4.5", + encryptedInput(), + codexHeaders("acct-remote"), + ); + + expect(expired.status).toBe(400); + expect(remote.status).toBe(400); + expect(fetchCalls).toBe(0); + }); + + test("enforces token validity boundaries before recovery or cache access", async () => { + let now = 1_800_000_000_000; + Date.now = () => now; + const nowSeconds = Math.floor(now / 1_000); + let recoveryFetches = 0; + let providerFetches = 0; + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) { + recoveryFetches += 1; + return new Response(recoverySse("Credential-bound assignment."), { status: 200 }); + } + providerFetches += 1; + return providerResponse(); + }) as typeof fetch; + + const requestWithClaims = (accountId: string, claims: Record) => { + const headers = codexHeaders(accountId); + headers.set("authorization", `Bearer ${fakeChatGptJwt(accountId, claims)}`); + return post(routedConfig(), "xai/grok-4.5", encryptedInput(), headers); + }; + + expect((await requestWithClaims("acct-exp-now", { exp: nowSeconds })).status).toBe(400); + expect((await requestWithClaims("acct-valid", { exp: nowSeconds + 1 })).status).toBe(200); + expect((await requestWithClaims("acct-nbf-edge", { nbf: nowSeconds + 60 })).status).toBe(200); + expect((await requestWithClaims("acct-nbf-future", { nbf: nowSeconds + 61 })).status).toBe(400); + expect((await requestWithClaims("acct-nbf-invalid", { nbf: "tomorrow" })).status).toBe(400); + expect(recoveryFetches).toBe(2); + expect(providerFetches).toBe(2); + + const expiringHeaders = codexHeaders("acct-cache-expiry"); + expiringHeaders.set("authorization", `Bearer ${fakeChatGptJwt("acct-cache-expiry", { + exp: nowSeconds + 1, + })}`); + expect((await post(routedConfig(), "xai/grok-4.5", encryptedInput(), expiringHeaders)).status).toBe(200); + now += 2_000; + expect((await post(routedConfig(), "xai/grok-4.5", encryptedInput(), expiringHeaders)).status).toBe(400); + expect(recoveryFetches).toBe(3); + expect(providerFetches).toBe(3); + }); + + test("rejects ambiguous agent envelopes before authenticated recovery", async () => { + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("recovery must stay unreachable"); + }) as typeof fetch; + const ambiguous = encryptedInput() as Array<{ content: Array> }>; + ambiguous[0]!.content.push({ type: "encrypted_content", encrypted_content: FERNET_TASK }); + + const response = await post( + routedConfig(), + "xai/grok-4.5", + ambiguous, + codexHeaders("acct-ambiguous"), + ); + + expect(response.status).toBe(400); + expect(fetchCalls).toBe(0); + }); + + test("rejects JWTs outside the native Codex OAuth issuer, audience, or client", async () => { + const encodeJwt = (claims: Record): string => { + const header = Buffer.from(JSON.stringify({ + alg: "RS256", + typ: "JWT", + kid: "fixture-key", + })).toString("base64url"); + const payload = Buffer.from(JSON.stringify(claims)).toString("base64url"); + return `${header}.${payload}.fakesig`; + }; + const baseClaims = { + iss: "https://auth.openai.com/", + aud: "https://api.openai.com/v1", + client_id: "app_EMoamEEZ73f0CkXaXp7hrann", + exp: Math.floor(Date.now() / 1000) + 3_600, + "https://api.openai.com/auth": { chatgpt_account_id: "acct-forged" }, + }; + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("recovery must stay unreachable"); + }) as typeof fetch; + + for (const claims of [ + { ...baseClaims, iss: "https://issuer.example" }, + { ...baseClaims, aud: "https://api.example/v1" }, + { ...baseClaims, client_id: "third-party-client" }, + { ...baseClaims, "https://api.openai.com/auth": undefined, chatgpt_account_id: "acct-forged" }, + ]) { + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + { + authorization: `Bearer ${encodeJwt(claims)}`, + "chatgpt-account-id": "acct-forged", + originator: "codex_cli_rs", + "x-openai-subagent": "collab_spawn", + }, + ); + expect(response.status).toBe(400); + } + + expect(fetchCalls).toBe(0); + }); + + test("rejects unsigned JWT-shaped caller credentials", async () => { + const header = Buffer.from(JSON.stringify({ alg: "none", typ: "JWT" })).toString("base64url"); + const payload = Buffer.from(JSON.stringify({ + iss: "https://auth.openai.com/", + aud: "https://api.openai.com/v1", + client_id: "app_EMoamEEZ73f0CkXaXp7hrann", + exp: Math.floor(Date.now() / 1000) + 3_600, + "https://api.openai.com/auth": { chatgpt_account_id: "acct-unsigned" }, + })).toString("base64url"); + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("recovery must stay unreachable"); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + { + authorization: `Bearer ${header}.${payload}.unsigned`, + "chatgpt-account-id": "acct-unsigned", + originator: "codex_cli_rs", + "x-openai-subagent": "collab_spawn", + }, + ); + + expect(response.status).toBe(400); + expect(fetchCalls).toBe(0); + }); + + test("non-Codex originators keep the typed fail-fast error", async () => { + let recoveryFetches = 0; + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) recoveryFetches += 1; + return new Response("event: error\ndata: {}\n\n", { status: 200 }); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + { ...codexHeaders(), originator: "other" }, + ); + + expect(response.status).toBe(400); + expect(recoveryFetches).toBe(0); + }); +}); diff --git a/tests/agent-task-recovery.test.ts b/tests/agent-task-recovery.test.ts new file mode 100644 index 0000000000..57a3943dcd --- /dev/null +++ b/tests/agent-task-recovery.test.ts @@ -0,0 +1,494 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { resetAgentTaskRecoveryState } from "../src/server/responses/agent-task-recovery"; +import { + agentMessage, + codexHeaders, + encryptedInput, + FERNET_TASK, + originalFetch, + post, + providerResponse, + recoveryArgumentsDoneSse, + recoveryCompletedSse, + recoverySse, + routedConfig, + ROUTING_ENVELOPE, + SECOND_FERNET_TASK, +} from "./helpers/agent-task-recovery"; + +describe("agent task recovery (opt-in, default off)", () => { + beforeEach(() => { + resetAgentTaskRecoveryState(); + }); + + afterEach(() => { + globalThis.fetch = originalFetch; + resetAgentTaskRecoveryState(); + }); + + test("keeps the disabled fail-fast response byte-identical to the absent feature", async () => { + const snapshot = async (config: ReturnType) => { + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("recovery and provider dispatch must stay unreachable"); + }) as typeof fetch; + const response = await post(config, "xai/grok-4.5", encryptedInput(), codexHeaders()); + return { + status: response.status, + statusText: response.statusText, + headers: [...response.headers.entries()].sort(), + body: Buffer.from(await response.arrayBuffer()).toString("hex"), + fetchCalls, + }; + }; + + const absent = await snapshot(routedConfig(null)); + const disabled = await snapshot(routedConfig({ enabled: false })); + + expect(disabled).toEqual(absent); + expect(absent.status).toBe(400); + const raw = Buffer.from(absent.body, "hex").toString("utf8"); + expect(JSON.parse(raw)).toMatchObject({ + error: { code: "unreadable_encrypted_agent_task" }, + }); + expect(absent.fetchCalls).toBe(0); + expect(raw).not.toContain(FERNET_TASK); + expect(raw).not.toContain("acct-caller"); + }); + + test("keeps disabled normal routed requests behaviorally identical to the absent feature", async () => { + const snapshot = async (config: ReturnType) => { + let request: unknown = null; + globalThis.fetch = (async (input, init) => { + request = { + url: String(input), + method: init?.method, + headers: [...new Headers(init?.headers).entries()] + .filter(([name]) => name !== "x-grok-req-id") + .sort(), + body: typeof init?.body === "string" + ? Buffer.from(init.body).toString("hex") + : null, + }; + return providerResponse(); + }) as typeof fetch; + const response = await post( + config, + "xai/grok-4.5", + agentMessage([{ type: "input_text", text: "Ordinary routed request." }]), + codexHeaders(), + ); + const responseBody = await response.json() as Record; + delete responseBody.id; + delete responseBody.created_at; + return { + request, + response: { + status: response.status, + headers: [...response.headers.entries()].sort(), + body: responseBody, + }, + }; + }; + + expect(await snapshot(routedConfig({ enabled: false }))).toEqual(await snapshot(routedConfig(null))); + }); + + test("baseline: encrypted routed task still fails when recovery returns no assignment", async () => { + const fetchedUrls: string[] = []; + globalThis.fetch = (async (input) => { + fetchedUrls.push(String(input)); + return new Response("event: error\ndata: {}\n\n", { status: 200 }); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(), + ); + const json = await response.json() as { error?: { code?: string } }; + + expect(response.status).toBe(400); + expect(json.error?.code).toBe("unreadable_encrypted_agent_task"); + expect(fetchedUrls.length).toBeGreaterThan(0); + expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex"); + }); + + test("authenticated ChatGPT recovery accepts the decrypted payload without a duplicated routing envelope", async () => { + const assignment = "Implement the focused regression test."; + const fetchedUrls: string[] = []; + const forwardedBodies: string[] = []; + globalThis.fetch = (async (input, init) => { + fetchedUrls.push(String(input)); + const raw = typeof init?.body === "string" ? init.body : ""; + forwardedBodies.push(raw); + if (String(input).includes("chatgpt.com")) { + return new Response(recoverySse(assignment), { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); + } + return providerResponse(); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(), + ); + + expect(response.status).toBe(200); + expect(fetchedUrls).toHaveLength(2); + expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex"); + expect(forwardedBodies[0]).toContain("capture_assignment"); + expect(forwardedBodies[1]).toContain("Implement the focused regression test."); + expect(forwardedBodies[1]).not.toContain(FERNET_TASK); + expect(forwardedBodies[1].match(/Message Type: NEW_TASK/g)).toHaveLength(1); + }); + + test("accepts function-call-arguments SSE events", async () => { + const assignment = "Handle the recovered task."; + let providerBody = ""; + globalThis.fetch = (async (input, init) => { + if (String(input).includes("chatgpt.com")) { + return new Response(recoveryArgumentsDoneSse(assignment), { status: 200 }); + } + providerBody = typeof init?.body === "string" ? init.body : ""; + return providerResponse(); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(), + ); + + expect(response.status).toBe(200); + expect(providerBody).toContain("Message Type: NEW_TASK"); + expect(providerBody).toContain(assignment); + expect(providerBody).not.toContain(FERNET_TASK); + }); + + test("accepts an assignment carried only by the completed response snapshot", async () => { + const assignment = "Read the completed response output."; + let providerBody = ""; + globalThis.fetch = (async (input, init) => { + if (String(input).includes("chatgpt.com")) { + return new Response(recoveryCompletedSse(assignment), { status: 200 }); + } + providerBody = typeof init?.body === "string" ? init.body : ""; + return providerResponse(); + }) as typeof fetch; + + const response = await post(routedConfig(), "xai/grok-4.5", encryptedInput(), codexHeaders()); + + expect(response.status).toBe(200); + expect(providerBody).toContain(assignment); + }); + + test("fails closed when completed recovery events disagree", async () => { + let providerFetches = 0; + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) { + const first = recoverySse("First assignment.").split("data: {\"type\":\"response.completed\"")[0]!; + return new Response(`${first}${recoveryCompletedSse("Second assignment.")}`, { status: 200 }); + } + providerFetches += 1; + return providerResponse(); + }) as typeof fetch; + + const response = await post(routedConfig(), "xai/grok-4.5", encryptedInput(), codexHeaders()); + + expect(response.status).toBe(400); + expect(providerFetches).toBe(0); + }); + + test("fails closed on malformed recovery SSE without retrying or dispatching", async () => { + let recoveryFetches = 0; + let providerFetches = 0; + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) { + recoveryFetches += 1; + return new Response("data: {not-json}\n\ndata: [DONE]\n\n", { status: 200 }); + } + providerFetches += 1; + return providerResponse(); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(), + ); + const raw = await response.text(); + + expect(response.status).toBe(400); + expect(recoveryFetches).toBe(1); + expect(providerFetches).toBe(0); + expect(raw).not.toContain(FERNET_TASK); + expect(raw).not.toContain("acct-caller"); + }); + + test("rejects a plausible tool call when the recovery stream never completes", async () => { + let providerFetches = 0; + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) { + const partial = recoverySse("Never dispatch this partial result.") + .split("data: {\"type\":\"response.completed\"")[0]!; + return new Response(partial, { status: 200 }); + } + providerFetches += 1; + return providerResponse(); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(), + ); + + expect(response.status).toBe(400); + expect(providerFetches).toBe(0); + }); + + test("times out recovery without dispatching the encrypted task", async () => { + let recoveryFetches = 0; + let providerFetches = 0; + globalThis.fetch = ((input, init) => { + if (!String(input).includes("chatgpt.com")) { + providerFetches += 1; + return Promise.resolve(providerResponse()); + } + recoveryFetches += 1; + return new Promise((_resolve, reject) => { + const signal = init?.signal; + const rejectAbort = () => reject(signal?.reason ?? new DOMException("aborted", "AbortError")); + if (signal?.aborted) rejectAbort(); + else signal?.addEventListener("abort", rejectAbort, { once: true }); + }); + }) as typeof fetch; + + const response = await post( + routedConfig({ enabled: true, timeoutMs: 1_000 }), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(), + ); + + expect(response.status).toBe(400); + expect(recoveryFetches).toBe(1); + expect(providerFetches).toBe(0); + }); + + test("cancels recovery with the client and never reaches the routed provider", async () => { + const controller = new AbortController(); + let markRecoveryStarted: (() => void) | undefined; + const recoveryStarted = new Promise((resolve) => { + markRecoveryStarted = resolve; + }); + let providerFetches = 0; + globalThis.fetch = ((input, init) => { + if (!String(input).includes("chatgpt.com")) { + providerFetches += 1; + return Promise.resolve(providerResponse()); + } + markRecoveryStarted?.(); + return new Promise((_resolve, reject) => { + const signal = init?.signal; + const rejectAbort = () => reject(signal?.reason ?? new DOMException("aborted", "AbortError")); + if (signal?.aborted) rejectAbort(); + else signal?.addEventListener("abort", rejectAbort, { once: true }); + }); + }) as typeof fetch; + + const pending = post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders(), + controller.signal, + ); + await recoveryStarted; + controller.abort(new DOMException("client disconnected", "AbortError")); + const response = await pending; + + expect(response.status).toBe(499); + expect(providerFetches).toBe(0); + expect(await response.json()).toMatchObject({ + error: { code: "client_cancelled" }, + }); + }); + + test("scopes cache entries by parent thread and authenticated account", async () => { + let recoveryFetches = 0; + let providerFetches = 0; + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) { + recoveryFetches += 1; + return new Response(recoverySse("Scoped cached assignment."), { status: 200 }); + } + providerFetches += 1; + return providerResponse(); + }) as typeof fetch; + + const headers = codexHeaders("acct-one", { "x-codex-parent-thread-id": "parent-one" }); + expect((await post(routedConfig(), "xai/grok-4.5", encryptedInput(), headers)).status).toBe(200); + expect((await post(routedConfig(), "xai/grok-4.5", encryptedInput(), headers)).status).toBe(200); + expect((await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders("acct-one", { "x-codex-parent-thread-id": "parent-two" }), + )).status).toBe(200); + expect((await post( + routedConfig(), + "xai/grok-4.5", + encryptedInput(), + codexHeaders("acct-two", { "x-codex-parent-thread-id": "parent-one" }), + )).status).toBe(200); + + expect(recoveryFetches).toBe(3); + expect(providerFetches).toBe(4); + }); + + test("deduplicates concurrent recovery for the same scoped task", async () => { + let recoveryFetches = 0; + let providerFetches = 0; + let releaseRecovery: (() => void) | undefined; + const recoveryGate = new Promise((resolve) => { + releaseRecovery = resolve; + }); + globalThis.fetch = (async (input) => { + if (String(input).includes("chatgpt.com")) { + recoveryFetches += 1; + await recoveryGate; + return new Response(recoverySse("Shared recovery assignment."), { status: 200 }); + } + providerFetches += 1; + return providerResponse(); + }) as typeof fetch; + const headers = codexHeaders("acct-flight", { "x-codex-parent-thread-id": "parent-flight" }); + + const first = post(routedConfig(), "xai/grok-4.5", encryptedInput(), headers); + const second = post(routedConfig(), "xai/grok-4.5", encryptedInput(), headers); + await new Promise(resolve => setImmediate(resolve)); + releaseRecovery?.(); + const responses = await Promise.all([first, second]); + + expect(responses.map(response => response.status)).toEqual([200, 200]); + expect(recoveryFetches).toBe(1); + expect(providerFetches).toBe(2); + }); + + test("enforces the configured cache entry bound", async () => { + let recoveryFetches = 0; + globalThis.fetch = (async (input, init) => { + if (String(input).includes("chatgpt.com")) { + recoveryFetches += 1; + const body = typeof init?.body === "string" ? init.body : ""; + const assignment = body.includes(SECOND_FERNET_TASK) ? "Task B." : "Task A."; + return new Response(recoverySse(assignment), { status: 200 }); + } + return providerResponse(); + }) as typeof fetch; + const config = routedConfig({ enabled: true, cacheEntries: 1 }); + const headers = codexHeaders("acct-cache", { "x-codex-parent-thread-id": "parent-cache" }); + + expect((await post(config, "xai/grok-4.5", encryptedInput(), headers)).status).toBe(200); + expect((await post( + config, + "xai/grok-4.5", + encryptedInput({ ciphertext: SECOND_FERNET_TASK }), + headers, + )).status).toBe(200); + expect((await post(config, "xai/grok-4.5", encryptedInput(), headers)).status).toBe(200); + + expect(recoveryFetches).toBe(3); + }); + + test("keeps plaintext v1-style tasks on the normal routed path", async () => { + let recoveryFetches = 0; + let providerFetches = 0; + let providerBody = ""; + globalThis.fetch = (async (input, init) => { + if (String(input).includes("chatgpt.com")) { + recoveryFetches += 1; + throw new Error("recovery must stay unreachable"); + } + providerFetches += 1; + providerBody = typeof init?.body === "string" ? init.body : ""; + return providerResponse(); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "xai/grok-4.5", + agentMessage([ + { type: "input_text", text: ROUTING_ENVELOPE }, + { type: "encrypted_content", encrypted_content: "Readable task payload." }, + ]), + codexHeaders(), + ); + + expect(response.status).toBe(200); + expect(recoveryFetches).toBe(0); + expect(providerFetches).toBe(1); + expect(providerBody).toContain("Readable task payload."); + }); + + test("leaves native encrypted passthrough unchanged", async () => { + let fetchedUrl = ""; + let forwardedBody = ""; + globalThis.fetch = (async (input, init) => { + fetchedUrl = String(input); + forwardedBody = typeof init?.body === "string" ? init.body : ""; + return providerResponse(); + }) as typeof fetch; + + const response = await post( + routedConfig(), + "gpt-5.6-sol", + encryptedInput(), + codexHeaders(), + ); + + expect(response.status).toBe(200); + expect(fetchedUrl).toBe("https://chatgpt.com/backend-api/codex/responses"); + expect(forwardedBody).toContain(FERNET_TASK); + expect(forwardedBody).not.toContain("capture_assignment"); + }); + + test("does not enable recovery inside combo attempts", async () => { + const config = routedConfig(); + config.combos = { + routed: { + strategy: "failover", + targets: [{ provider: "xai", model: "grok-4.5" }], + }, + }; + let fetchCalls = 0; + globalThis.fetch = (async () => { + fetchCalls += 1; + throw new Error("combo must fail before dispatch"); + }) as typeof fetch; + + const response = await post( + config, + "combo/routed", + encryptedInput(), + codexHeaders(), + ); + + expect(response.status).toBe(400); + expect(fetchCalls).toBe(0); + expect(await response.json()).toMatchObject({ + error: { code: "unreadable_encrypted_agent_task" }, + }); + }); +}); diff --git a/tests/config.test.ts b/tests/config.test.ts index b527ff024c..2e74b37a58 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -439,6 +439,61 @@ describe("opencodex config defaults", () => { } }); + test("agentTaskRecovery is explicit, bounded, and degrades invalid hand edits", () => { + const base = { + port: 12345, + providers: { + custom: { + adapter: "openai-responses", + baseUrl: "https://example.test/v1", + }, + }, + defaultProvider: "custom", + }; + expect(getDefaultConfig().agentTaskRecovery).toBeUndefined(); + + const recovery = { + enabled: true, + model: "gpt-5.6-sol", + timeoutMs: 45_000, + cacheEntries: 200, + }; + writeConfig({ ...base, agentTaskRecovery: recovery }); + expect(loadConfig()).toMatchObject({ ...base, agentTaskRecovery: recovery }); + expect(validateConfigCandidate({ ...base, agentTaskRecovery: recovery })).toMatchObject({ + ok: true, + config: { agentTaskRecovery: recovery }, + }); + + for (const invalid of [ + true, + { enabled: "true" }, + { enabled: true, model: " " }, + { enabled: true, timeoutMs: 999 }, + { enabled: true, timeoutMs: 120_001 }, + { enabled: true, cacheEntries: 0 }, + { enabled: true, cacheEntries: 513 }, + { enabled: true, url: "https://attacker.example/responses" }, + ]) { + writeConfig({ ...base, agentTaskRecovery: invalid }); + const diagnostics = readConfigDiagnostics(); + expect(diagnostics).toMatchObject({ + source: "file", + error: null, + config: base, + }); + expect(diagnostics.config.agentTaskRecovery).toBeUndefined(); + expect(diagnostics.warnings?.some(warning => warning.startsWith("agentTaskRecovery"))).toBe(true); + expect(loadConfig()).toMatchObject(base); + expect(loadConfig().agentTaskRecovery).toBeUndefined(); + expect(validateConfigCandidate({ ...base, agentTaskRecovery: invalid })).toMatchObject({ + ok: false, + error: expect.stringContaining("agentTaskRecovery"), + }); + expect(backupNames()).toEqual([]); + } + }); + test("native subagent-default sync is opt-in and ignores malformed opt-ins without falling back", () => { const base = { port: 12345, diff --git a/tests/helpers/agent-task-recovery.ts b/tests/helpers/agent-task-recovery.ts new file mode 100644 index 0000000000..1e0e953694 --- /dev/null +++ b/tests/helpers/agent-task-recovery.ts @@ -0,0 +1,176 @@ +import { handleResponses } from "../../src/server/responses"; +import type { OcxConfig } from "../../src/types"; + +export const originalFetch = globalThis.fetch; + +export function fakeChatGptJwt(accountId: string, claimOverrides: Record = {}): string { + const header = Buffer.from(JSON.stringify({ + alg: "RS256", + typ: "JWT", + kid: "fixture-key", + })).toString("base64url"); + const payload = Buffer.from(JSON.stringify({ + iss: "https://auth.openai.com/", + aud: "https://api.openai.com/v1", + client_id: "app_EMoamEEZ73f0CkXaXp7hrann", + exp: Math.floor(Date.now() / 1000) + 3_600, + "https://api.openai.com/auth": { chatgpt_account_id: accountId }, + ...claimOverrides, + })).toString("base64url"); + return `${header}.${payload}.fakesig`; +} + +function fernetFixture(ciphertextBytes = 16, version = 0x80, fill = 0x5a): string { + const raw = Buffer.alloc(57 + ciphertextBytes, fill); + raw[0] = version; + raw.writeBigUInt64BE(1_720_000_000n, 1); + const unpadded = raw.toString("base64url"); + return `${unpadded}${"=".repeat((4 - (unpadded.length % 4)) % 4)}`; +} + +export const FERNET_TASK = fernetFixture(); +export const SECOND_FERNET_TASK = fernetFixture(16, 0x80, 0x4b); + +function routingEnvelope( + taskName = "/root/worker", + sender = "/root", +): string { + return [ + "Message Type: NEW_TASK", + `Task name: ${taskName}`, + `Sender: ${sender}`, + "Payload:", + "", + ].join("\n"); +} + +export const ROUTING_ENVELOPE = routingEnvelope(); + +export function agentMessage(content: Array>): unknown[] { + return [{ + type: "agent_message", + author: "/root", + recipient: "/root/worker", + content, + }]; +} + +export function routedConfig( + recovery: OcxConfig["agentTaskRecovery"] | null = { enabled: true }, +): OcxConfig { + const config = { + port: 0, + defaultProvider: "xai", + providers: { + xai: { + adapter: "openai-chat", + baseUrl: "https://api.x.ai/v1", + authMode: "key", + apiKey: "test-xai-key", + }, + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + }, + }, + } as OcxConfig; + if (recovery !== null) config.agentTaskRecovery = recovery; + return config; +} + +export function codexHeaders(accountId = "acct-caller", extra: HeadersInit = {}): Headers { + const headers = new Headers(extra); + headers.set("authorization", `Bearer ${fakeChatGptJwt(accountId)}`); + headers.set("chatgpt-account-id", accountId); + headers.set("originator", "codex_cli_rs"); + headers.set("x-openai-subagent", "collab_spawn"); + return headers; +} + +export function recoverySse(assignment: string): string { + const payload = JSON.stringify({ + type: "response.output_item.done", + item: { + type: "function_call", + name: "capture_assignment", + arguments: JSON.stringify({ assignment }), + }, + }); + return `data: ${payload}\n\ndata: ${JSON.stringify({ + type: "response.completed", + response: { status: "completed", output: [] }, + })}\n\n`; +} + +export function recoveryArgumentsDoneSse(assignment: string): string { + return `data: ${JSON.stringify({ + type: "response.function_call_arguments.done", + name: "capture_assignment", + arguments: JSON.stringify({ assignment }), + })}\n\ndata: ${JSON.stringify({ + type: "response.completed", + response: { status: "completed", output: [] }, + })}\n\n`; +} + +export function recoveryCompletedSse(assignment: string): string { + return `data: ${JSON.stringify({ + type: "response.completed", + response: { + status: "completed", + output: [{ + type: "function_call", + name: "capture_assignment", + arguments: JSON.stringify({ assignment }), + }], + }, + })}\n\n`; +} + +export function providerResponse(): Response { + return Response.json({ + id: "resp_routed", + object: "response", + status: "completed", + model: "grok-4.5", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); +} + +export async function post( + config: OcxConfig, + model: string, + input: unknown[], + headers: HeadersInit = {}, + abortSignal?: AbortSignal, +): Promise { + return handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { + "content-type": "application/json", + ...Object.fromEntries(new Headers(headers)), + }, + body: JSON.stringify({ model, input, stream: false }), + }), config, { model: "", provider: "" }, { abortSignal }); +} + +export function encryptedInput(options: { + ciphertext?: string; + taskName?: string; + sender?: string; +} = {}): unknown[] { + const taskName = options.taskName ?? "/root/worker"; + const sender = options.sender ?? "/root"; + return [{ + type: "agent_message", + author: sender, + recipient: taskName, + content: [ + { type: "input_text", text: routingEnvelope(taskName, sender) }, + { type: "encrypted_content", encrypted_content: options.ciphertext ?? FERNET_TASK }, + ], + }]; +} From fbb89f1fcade6e59177e790117499fd197fb52e9 Mon Sep 17 00:00:00 2001 From: Thierno Bah Date: Wed, 12 Aug 2026 16:14:08 +0200 Subject: [PATCH 2/4] fix(agents): tighten recovery review paths Document the exact credential boundary, cover startup warnings and shared-flight admission, and reject mixed encrypted slots before mutation. Refs #92 --- .../content/docs/guides/sub-agent-surface.md | 4 ++- .../docs/reference/configuration/agents.md | 13 +++++--- src/server/index.ts | 15 ++++++--- .../responses/agent-task-recovery-cache.ts | 6 ++++ src/server/responses/agent-task-recovery.ts | 12 ++----- tests/agent-task-recovery.test.ts | 32 ++++++++++++++++++- 6 files changed, 61 insertions(+), 21 deletions(-) diff --git a/docs-site/src/content/docs/guides/sub-agent-surface.md b/docs-site/src/content/docs/guides/sub-agent-surface.md index 526e0fd107..14bff535a6 100644 --- a/docs-site/src/content/docs/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/guides/sub-agent-surface.md @@ -132,7 +132,9 @@ v1 for heterogeneous-provider delegation, or resend the task as plaintext v2 `ag content when you control the caller. An experimental, disabled-by-default `agentTaskRecovery` option can recover this specific native- -to-routed shape through an additional authenticated request to ChatGPT before provider dispatch. +to-routed shape through a raw Responses passthrough to the fixed ChatGPT `/responses` endpoint using +forward-mode authentication. Only `authorization`, matching `chatgpt-account-id`, `originator`, and +optional `openai-beta`/`user-agent` metadata are forwarded; no other caller headers cross the boundary. It consumes quota, adds latency, briefly retains recovered plaintext in a bounded in-memory cache, and depends on undocumented ChatGPT backend behavior. Because a model returns the recovered text, byte-for-byte fidelity is not guaranteed. It rejects generic/API-key proxy callers and preserves diff --git a/docs-site/src/content/docs/reference/configuration/agents.md b/docs-site/src/content/docs/reference/configuration/agents.md index ba043ecacc..da179b4e75 100644 --- a/docs-site/src/content/docs/reference/configuration/agents.md +++ b/docs-site/src/content/docs/reference/configuration/agents.md @@ -123,10 +123,10 @@ fails instead of routing unreadable ciphertext elsewhere. `agentTaskRecovery` is an experimental compatibility path for a native ChatGPT parent spawning a routed v2 child. It is disabled by default. When explicitly enabled and the final routed child task -contains an otherwise unreadable Fernet payload, opencodex sends the isolated `agent_message` to -the fixed authenticated ChatGPT Codex Responses endpoint. ChatGPT returns the plaintext assignment -through a forced function call; opencodex then converts only that task item to a standard user -message before routed-provider dispatch. +contains an otherwise unreadable Fernet payload, opencodex uses a raw Responses passthrough request +to the fixed `https://chatgpt.com/backend-api/codex/responses` endpoint with forward-mode +authentication. ChatGPT returns the plaintext assignment through a forced function call; opencodex +then converts only that task item to a standard user message before routed-provider dispatch. This is not local decryption and does not fix the Codex wire protocol. It depends on undocumented ChatGPT backend behavior and may stop working after a backend change. The recovered assignment is @@ -143,6 +143,9 @@ Admission and retention are deliberately narrow: - raw ChatGPT credentials are sent only to the hard-coded ChatGPT endpoint and are never placed in the request body, logs, cache keys, or provider request; the in-memory cache scope uses only a process-random keyed digest of the caller credential and account; +- the recovery request forwards only `authorization`, the matching `chatgpt-account-id`, + `originator`, and optional `openai-beta` and `user-agent` metadata; opencodex sets `content-type` + and `accept` itself, and no other caller headers cross this boundary; - recovered plaintext is never logged or persisted; the process-local cache is credential-, parent- thread-, and ciphertext-scoped, expires after 15 minutes, and is bounded by both configured entry count (200 by default, 512 maximum) and 8 MiB total; @@ -179,7 +182,7 @@ Enable this only when the additional authenticated request, quota use, plaintext and private-backend dependency are acceptable. Prefer a native ChatGPT child or v1 heterogeneous delegation when they are not. -This recovery path applies to direct routed children. At most 32 recovery requests can be active at +This recovery path applies to direct-routed children. At most 32 recovery requests can be active at once; additional misses fail closed. Combo routing keeps its existing native-only filter for encrypted tasks and does not invoke recovery. diff --git a/src/server/index.ts b/src/server/index.ts index 905fbed2dd..509012454e 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -478,14 +478,19 @@ export function consumeStartupCacheInvalidationWrite(): boolean { return wrote; } +export function warnAgentTaskRecoveryStartup(config: { + agentTaskRecovery?: { enabled?: boolean }; +}): void { + if (config.agentTaskRecovery?.enabled !== true) return; + console.warn("⚠️ Experimental encrypted V2 task recovery is enabled."); + console.warn(" Each scoped cache miss sends an additional authenticated request to ChatGPT and may consume quota or add latency."); + console.warn(" Recovered model output is retained only in a bounded in-memory cache; exact fidelity is not guaranteed and the path depends on undocumented backend behavior."); +} + export function startServer(port?: number, deps: StartServerDeps = {}): Server { const localAttestationSecret = deps.localAttestationSecret ?? createLocalAttestationSecret(); const config = runAlibabaRegionStartupMigration(runOpenAiTierStartupMigration(loadConfig())); - if (config.agentTaskRecovery?.enabled === true) { - console.warn("⚠️ Experimental encrypted V2 task recovery is enabled."); - console.warn(" Each scoped cache miss sends an additional authenticated request to ChatGPT and may consume quota or add latency."); - console.warn(" Recovered model output is retained only in a bounded in-memory cache; exact fidelity is not guaranteed and the path depends on undocumented backend behavior."); - } + warnAgentTaskRecoveryStartup(config); setLiveStateStoreConfig(config); applyProxyEnv(config); assertServerAuthConfig(config); diff --git a/src/server/responses/agent-task-recovery-cache.ts b/src/server/responses/agent-task-recovery-cache.ts index 709bbe787b..b55767e18e 100644 --- a/src/server/responses/agent-task-recovery-cache.ts +++ b/src/server/responses/agent-task-recovery-cache.ts @@ -135,3 +135,9 @@ export function resetAgentTaskRecoveryCache(): void { RECOVERY_FLIGHTS.clear(); for (const key of [...RECOVERY_CACHE.keys()]) deleteRecoveryCacheEntry(key); } + +export function agentTaskRecoveryWaiterCountForTests(): number { + let count = 0; + for (const flight of RECOVERY_FLIGHTS.values()) count += flight.waiters; + return count; +} diff --git a/src/server/responses/agent-task-recovery.ts b/src/server/responses/agent-task-recovery.ts index d414e4b6cf..86b11b71b5 100644 --- a/src/server/responses/agent-task-recovery.ts +++ b/src/server/responses/agent-task-recovery.ts @@ -53,7 +53,6 @@ export function agentTaskRecoveryConfig(config: OcxConfig): AgentTaskRecoveryOpt interface AgentEnvelope { itemIndex: number; encryptedIndex: number; - encryptedSlot: string; headerText: string; messageType: "NEW_TASK"; taskName: string; @@ -88,7 +87,6 @@ function findEnvelope(input: unknown): AgentEnvelope | null { let taskName: string | null = null; let sender: string | null = null; let encryptedIndex = -1; - let encryptedSlot = ""; let ciphertext = ""; let encryptedPartCount = 0; let ciphertextCount = 0; @@ -118,7 +116,6 @@ function findEnvelope(input: unknown): AgentEnvelope | null { for (const token of structurallyValidFernetTokens(part.encrypted_content)) { ciphertextCount += 1; encryptedIndex = index; - encryptedSlot = part.encrypted_content; ciphertext = token; } } @@ -131,6 +128,7 @@ function findEnvelope(input: unknown): AgentEnvelope | null { || encryptedIndex < 0 || encryptedPartCount !== 1 || ciphertextCount !== 1 + || (content[encryptedIndex] as { encrypted_content?: unknown }).encrypted_content !== ciphertext || Buffer.byteLength(ciphertext) > MAX_CIPHERTEXT_BYTES ) return null; @@ -141,7 +139,6 @@ function findEnvelope(input: unknown): AgentEnvelope | null { return { itemIndex, encryptedIndex, - encryptedSlot, headerText, messageType, taskName, @@ -183,13 +180,10 @@ function injectAssignment(input: unknown, envelope: AgentEnvelope, assignment: s if ( !part || part.type !== "encrypted_content" - || part.encrypted_content !== envelope.encryptedSlot + || part.encrypted_content !== envelope.ciphertext ) return false; - const text = envelope.encryptedSlot === envelope.ciphertext - ? assignment - : envelope.encryptedSlot.replace(envelope.ciphertext, assignment); - content[envelope.encryptedIndex] = { type: "input_text", text }; + content[envelope.encryptedIndex] = { type: "input_text", text: assignment }; const message = item as Record; message.type = "message"; message.role = "user"; diff --git a/tests/agent-task-recovery.test.ts b/tests/agent-task-recovery.test.ts index 57a3943dcd..f0e1ae05c8 100644 --- a/tests/agent-task-recovery.test.ts +++ b/tests/agent-task-recovery.test.ts @@ -1,5 +1,7 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { warnAgentTaskRecoveryStartup } from "../src/server"; import { resetAgentTaskRecoveryState } from "../src/server/responses/agent-task-recovery"; +import { agentTaskRecoveryWaiterCountForTests } from "../src/server/responses/agent-task-recovery-cache"; import { agentMessage, codexHeaders, @@ -95,6 +97,31 @@ describe("agent task recovery (opt-in, default off)", () => { expect(await snapshot(routedConfig({ enabled: false }))).toEqual(await snapshot(routedConfig(null))); }); + test("warns at startup only for an explicit recovery opt-in without exposing credentials", () => { + const config = routedConfig(null); + const secret = "startup-secret-sentinel"; + config.providers.xai!.apiKey = secret; + const originalWarn = console.warn; + const capture = (recovery: typeof config.agentTaskRecovery): string[] => { + const warnings: string[] = []; + config.agentTaskRecovery = recovery; + console.warn = (...args: unknown[]) => { warnings.push(args.map(String).join(" ")); }; + warnAgentTaskRecoveryStartup(config); + return warnings; + }; + + try { + expect(capture(undefined)).toEqual([]); + expect(capture({ enabled: false })).toEqual([]); + const warnings = capture({ enabled: true }); + expect(warnings).toHaveLength(3); + expect(warnings.join("\n")).toContain("Experimental encrypted V2 task recovery is enabled"); + expect(warnings.join("\n")).not.toContain(secret); + } finally { + console.warn = originalWarn; + } + }); + test("baseline: encrypted routed task still fails when recovery returns no assignment", async () => { const fetchedUrls: string[] = []; globalThis.fetch = (async (input) => { @@ -377,7 +404,10 @@ describe("agent task recovery (opt-in, default off)", () => { const first = post(routedConfig(), "xai/grok-4.5", encryptedInput(), headers); const second = post(routedConfig(), "xai/grok-4.5", encryptedInput(), headers); - await new Promise(resolve => setImmediate(resolve)); + for (let turn = 0; turn < 200 && agentTaskRecoveryWaiterCountForTests() < 2; turn += 1) { + await new Promise(resolve => setImmediate(resolve)); + } + expect(agentTaskRecoveryWaiterCountForTests()).toBe(2); releaseRecovery?.(); const responses = await Promise.all([first, second]); From 5d3c2c50f7fab5df5d01f8f9c2f85bf10dad2f3a Mon Sep 17 00:00:00 2001 From: Thierno Bah Date: Wed, 12 Aug 2026 16:48:59 +0200 Subject: [PATCH 3/4] fix(agents): clarify recovery boundaries Make the loopback and credential constraints explicit in operator docs, and describe shared recovery flights accurately at startup. Refs #92 --- docs-site/src/content/docs/guides/sub-agent-surface.md | 7 +++++-- .../src/content/docs/reference/configuration/agents.md | 10 +++++++--- src/server/index.ts | 2 +- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/docs-site/src/content/docs/guides/sub-agent-surface.md b/docs-site/src/content/docs/guides/sub-agent-surface.md index 14bff535a6..32a689c560 100644 --- a/docs-site/src/content/docs/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/guides/sub-agent-surface.md @@ -133,8 +133,11 @@ content when you control the caller. An experimental, disabled-by-default `agentTaskRecovery` option can recover this specific native- to-routed shape through a raw Responses passthrough to the fixed ChatGPT `/responses` endpoint using -forward-mode authentication. Only `authorization`, matching `chatgpt-account-id`, `originator`, and -optional `openai-beta`/`user-agent` metadata are forwarded; no other caller headers cross the boundary. +the incoming credential shape used by the canonical `openai` provider with `authMode: "forward"`. +Recovery is available only while the proxy is bound to loopback. It never substitutes API-key +authentication, another provider credential, or another Codex account. Only `authorization`, matching +`chatgpt-account-id`, `originator`, and optional `openai-beta`/`user-agent` metadata are forwarded; +`content-type` and `accept` are generated locally, and no other caller headers cross the boundary. It consumes quota, adds latency, briefly retains recovered plaintext in a bounded in-memory cache, and depends on undocumented ChatGPT backend behavior. Because a model returns the recovered text, byte-for-byte fidelity is not guaranteed. It rejects generic/API-key proxy callers and preserves diff --git a/docs-site/src/content/docs/reference/configuration/agents.md b/docs-site/src/content/docs/reference/configuration/agents.md index da179b4e75..0b9ec7886d 100644 --- a/docs-site/src/content/docs/reference/configuration/agents.md +++ b/docs-site/src/content/docs/reference/configuration/agents.md @@ -131,13 +131,17 @@ then converts only that task item to a standard user message before routed-provi This is not local decryption and does not fix the Codex wire protocol. It depends on undocumented ChatGPT backend behavior and may stop working after a backend change. The recovered assignment is model output, not a cryptographically verified plaintext, so byte-for-byte fidelity is not -guaranteed. Each scoped cache miss adds one authenticated ChatGPT request, consumes account quota, -and adds latency before the routed request. Concurrent requests for the same scoped task share one +guaranteed. A scoped cache miss may add an authenticated ChatGPT request, consume account quota, and +add latency before the routed request. Concurrent requests for the same scoped task share one recovery request. Startup prints a warning whenever the feature is enabled. Admission and retention are deliberately narrow: -- only a native Codex caller with a matching ChatGPT bearer/account pair is eligible; +- recovery is available only while the proxy is bound to loopback; +- only a native Codex caller with a matching ChatGPT bearer/account pair is eligible. This is the + credential shape used by the canonical `openai` provider with `authMode: "forward"`; recovery uses + only the pair on the incoming request and never substitutes API-key authentication, another + provider credential, or another Codex account; - callers using `x-opencodex-api-key`, `x-api-key`, generic API credentials, or a proxy admission secret keep the existing `unreadable_encrypted_agent_task` failure; - raw ChatGPT credentials are sent only to the hard-coded ChatGPT endpoint and are never placed in diff --git a/src/server/index.ts b/src/server/index.ts index 509012454e..7f6e180339 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -483,7 +483,7 @@ export function warnAgentTaskRecoveryStartup(config: { }): void { if (config.agentTaskRecovery?.enabled !== true) return; console.warn("⚠️ Experimental encrypted V2 task recovery is enabled."); - console.warn(" Each scoped cache miss sends an additional authenticated request to ChatGPT and may consume quota or add latency."); + console.warn(" A scoped cache miss may send an additional authenticated request to ChatGPT and may consume quota or add latency; concurrent misses can share one request."); console.warn(" Recovered model output is retained only in a bounded in-memory cache; exact fidelity is not guaranteed and the path depends on undocumented backend behavior."); } From 68354a6b4afd7683f11e9f4361fa368dbc31fe07 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Thu, 13 Aug 2026 03:20:25 +0900 Subject: [PATCH 4/4] fix(agents): keep recovered task plaintext out of the persisted continuation cache Recovery mutates the request input in place, so the recovered plaintext becomes _rawBody. The non-streaming and streaming paths then hand _rawBody to rememberResponseState, which stores the input and schedules a snapshot write to responses-state.json. Decrypted task text therefore reached disk with no TTL, contradicting the in-memory 15-minute cache the recovery path documents. Bar the body from the continuation cache instead. The marker is a WeakSet keyed on the body object rather than a field, because _rawBody is serialized verbatim by the native passthrough and a field would be sent upstream. The check lives in rememberResponseState so every recording path inherits it. --- src/responses/state.ts | 22 +++++++++++++ src/server/responses/core.ts | 5 +++ tests/responses-state.test.ts | 61 +++++++++++++++++++++++++++++++++++ 3 files changed, 88 insertions(+) diff --git a/src/responses/state.ts b/src/responses/state.ts index 2e7c2abb63..a0ef3eb9dc 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -947,6 +947,27 @@ export function responseStateMetrics(): ResponseStateMetrics { * Cache completed output and max_output_tokens partial output for previous_response_id replay. * Content-filtered incomplete and failed output are not authoritative replay history. */ +/** + * Request bodies that must never enter the continuation cache. + * + * The cache is persisted to `responses-state.json`, so anything recorded here reaches disk. + * Encrypted-agent-task recovery decrypts task text into the request body and promises + * in-memory, TTL-bounded retention; recording that body would put the plaintext on disk with + * no TTL and break the promise. + * + * A WeakSet rather than a body field on purpose: `_rawBody` is serialized verbatim by the + * native passthrough, so any marker written into the body itself would be sent upstream. + * Marking is enforced once here rather than at each call site, because every recording path + * (streaming, non-streaming, passthrough, forced) funnels through `rememberResponseState` — + * a new call site cannot reintroduce the leak by forgetting a guard. + */ +const nonPersistableBodies = new WeakSet(); + +/** Bar this exact request body from the continuation cache, and therefore from disk. */ +export function markBodyNonPersistable(body: unknown): void { + if (body && typeof body === "object") nonPersistableBodies.add(body as object); +} + export function rememberResponseState( requestBody: unknown, response: { id?: unknown; output?: unknown; status?: unknown; incomplete_details?: unknown }, @@ -955,6 +976,7 @@ export function rememberResponseState( ): void { if (!requestBody || typeof requestBody !== "object" || Array.isArray(requestBody)) return; const request = requestBody as Record; + if (nonPersistableBodies.has(request)) return; // `force` bypasses only the store:false skip: Codex sends `store:false` on every non-Azure // HTTP request (and WS inherits it), yet its WS turns still chain with previous_response_id. // The passthrough branch records with force so those chains can be expanded locally; the diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index f6d2571466..7bd5cb1074 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -19,6 +19,7 @@ import { buildCompactV1Output, COMPACT_PROMPT, decodeCompactionSummary, extractC import { FORWARD_HEADERS, sanitizeReasoningInputContent } from "../../adapters/openai-responses"; import { expandPreviousResponseInput, + markBodyNonPersistable, previousResponseProviderState, previousResponseReplayFailure, rememberResponseState, @@ -1732,6 +1733,10 @@ async function handleResponsesInner( } } parsed = reparsed; + // The recovery mutated `body.input` in place, so `_rawBody` now carries decrypted task + // text. Bar it from the continuation cache before any recording path can reach it — + // that cache is persisted to disk, which would defeat the recovery cache's TTL. + markBodyNonPersistable(parsed._rawBody); toolBridgeMaps = buildToolBridgeMaps(parsed, translatorBudget); } catch { unreadableEncryptedAgentTask = true; diff --git a/tests/responses-state.test.ts b/tests/responses-state.test.ts index 9554ae616e..7bd3178a72 100644 --- a/tests/responses-state.test.ts +++ b/tests/responses-state.test.ts @@ -28,6 +28,7 @@ import { evictOldestResponseContinuationForBudget, expandPreviousResponseInput, flushResponseState, + markBodyNonPersistable, previousResponseConversationId, previousResponseProviderState, previousResponseReplayFailure, @@ -2062,4 +2063,64 @@ describe("Responses state admission boundary (oversized direct-spill)", () => { const raw = readFileSync(join(home, "responses-state.json"), "utf-8"); expect(raw).not.toContain("resp_multibyte"); }); + + /** + * Encrypted-agent-task recovery decrypts task text into the request body and promises + * in-memory retention bounded by a 15-minute TTL. The continuation cache persists request + * input to `responses-state.json`, so recording a recovered body would put that plaintext on + * disk with no TTL at all. The guard lives in `rememberResponseState` so every recording path + * inherits it. + */ + describe("bodies marked non-persistable never reach the continuation cache", () => { + test("a marked body is not stored and its text never reaches the snapshot", async () => { + const recovered = { + model: "m", + input: [{ type: "message", role: "user", content: "RECOVERED-PLAINTEXT-SENTINEL" }], + }; + markBodyNonPersistable(recovered); + + rememberResponseState(recovered, completedResponse("resp_recovered", "ok"), undefined, { force: true }); + await flushResponseState(); + + // Not in memory: a later turn replaying that id gets its request back UNEXPANDED, + // i.e. with no `input` grafted on from stored history. + const replay = expandPreviousResponseInput({ previous_response_id: "resp_recovered" }) as { + input?: unknown; + }; + expect(replay.input).toBeUndefined(); + // Not on disk, and neither is the response id that would have carried it. + const raw = existsSync(join(home, "responses-state.json")) + ? readFileSync(join(home, "responses-state.json"), "utf-8") + : ""; + expect(raw).not.toContain("RECOVERED-PLAINTEXT-SENTINEL"); + expect(raw).not.toContain("resp_recovered"); + }); + + test("an unmarked body with identical shape IS stored — the guard is the marker, not the shape", async () => { + const ordinary = { + model: "m", + input: [{ type: "message", role: "user", content: "ORDINARY-INPUT-SENTINEL" }], + }; + + rememberResponseState(ordinary, completedResponse("resp_ordinary", "ok"), undefined, { force: true }); + await flushResponseState(); + + const raw = readFileSync(join(home, "responses-state.json"), "utf-8"); + expect(raw).toContain("resp_ordinary"); + }); + + test("marking is per-object, so an unrelated body is unaffected", async () => { + const marked = { model: "m", input: "marked", store: true }; + const sibling = { model: "m", input: "sibling", store: true }; + markBodyNonPersistable(marked); + + rememberResponseState(marked, completedResponse("resp_marked", "ok"), undefined, { force: true }); + rememberResponseState(sibling, completedResponse("resp_sibling", "ok"), undefined, { force: true }); + await flushResponseState(); + + const raw = readFileSync(join(home, "responses-state.json"), "utf-8"); + expect(raw).not.toContain("resp_marked"); + expect(raw).toContain("resp_sibling"); + }); + }); });