From 39072c7a629ea4f7cc8081bf3201077650052e2a Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 01:41:32 +0900 Subject: [PATCH 1/7] docs: refresh upstream WebSocket stack layer --- devlog/_plan/260906_a_runtime_stack/021_ws_refresh.md | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 devlog/_plan/260906_a_runtime_stack/021_ws_refresh.md diff --git a/devlog/_plan/260906_a_runtime_stack/021_ws_refresh.md b/devlog/_plan/260906_a_runtime_stack/021_ws_refresh.md new file mode 100644 index 0000000000..f6de6471c3 --- /dev/null +++ b/devlog/_plan/260906_a_runtime_stack/021_ws_refresh.md @@ -0,0 +1,3 @@ +# WebSocket layer P refresh + +Consume 020 above prepared SSE parent 4b34cbb8d, with source #3679 b05cccf264b4ab61db5d8dee8232c2f89bb1b541. Public author updated the old head and resolved the three original review threads. Retain Clive Rosfield attribution and -x source identity. Existing shared proxy-formats documentation contains SSE paragraph; preserve both sections. B owns concurrent providerContextLimits config changes; A updates only applyProxyEnv. This layer stays independently verified and draft while full CI runs; main merges only after full required gates. No local project checks. From 10fbda2e031230d57873b5afb828324c18ae090f Mon Sep 17 00:00:00 2001 From: Clive Rosfield <64878945+S0RYUASUKA@users.noreply.github.com> Date: Sat, 5 Sep 2026 00:02:29 +0800 Subject: [PATCH 2/7] fix(responses): honor proxy routing for upstream websocket (cherry picked from commit b05cccf264b4ab61db5d8dee8232c2f89bb1b541) --- .../src/content/docs/guides/providers.md | 6 + .../src/content/docs/reference/adapters.md | 5 + .../content/docs/reference/proxy-formats.md | 13 ++ src/config.ts | 11 +- src/lib/provider-outbound.ts | 47 +----- src/lib/proxy-env.ts | 67 +++++++++ src/server/responses/codex-ws-pool.ts | 8 +- src/server/responses/codex-ws-session.ts | 4 +- src/server/responses/ws-upstream.ts | 11 +- structure/04_transports-and-sidecars.md | 8 +- tests/responses/ws-upstream-reuse.test.ts | 29 +++- tests/responses/ws-upstream.test.ts | 82 +++++++++-- tests/server/proxy-env.test.ts | 135 +++++++++++++++++- 13 files changed, 350 insertions(+), 76 deletions(-) diff --git a/docs-site/src/content/docs/guides/providers.md b/docs-site/src/content/docs/guides/providers.md index 6a37cf8a70..255c0d8dc4 100644 --- a/docs-site/src/content/docs/guides/providers.md +++ b/docs-site/src/content/docs/guides/providers.md @@ -620,6 +620,12 @@ A provider is included when opencodex has a matching wire adapter, **not** based (AI Studio, Vertex, and Antigravity/Cloud Code Assist modes), `azure` / `azure-openai`, `kiro`, and `cursor`. A proprietary API without one of these implementations, such as native Amazon Bedrock, is not supported directly. + +Provider configuration selects the adapter; upstream transport selection is separate. Eligible +Responses traffic can use WSS with [explicit proxy routing](/reference/proxy-formats/#json-and-sse-output). +Invalid or unsupported WebSocket proxy settings fall back to HTTP/SSE, which uses Bun's HTTP +proxy rules rather than the WSS-specific `ALL_PROXY` fallback. + **GitHub Copilot** is an OAuth provider (`ocx login github-copilot`) that exchanges a GitHub device-flow login for a short-lived Copilot API token — not a pasted API key. **GitLab Duo** remains a key/subscription-token gateway on its OpenAI-compatible endpoint. **Cloudflare AI diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index 1db98357d3..e2a24c67df 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -95,6 +95,11 @@ body and response, with narrow compatibility rewrites for routed gateways. `forward` uses configured static headers without relaying caller authorization; `key` uses the configured provider key. +Adapter selection does not select the upstream transport. Eligible requests can use the +[upstream WebSocket proxy route](/reference/proxy-formats/#json-and-sse-output); invalid or unsupported +WebSocket proxy settings fall back to HTTP/SSE. HTTP fetch-based Responses handling uses Bun's +HTTP proxy rules and does not inherit the WSS-specific `ALL_PROXY` fallback. + Noncanonical Responses gateways receive Codex's client-executed `tool_search` declaration as a collision-safe public function tool. Matching request history and JSON/SSE function calls are translated back to the private `tool_search` lifecycle for the client. Canonical OpenAI forward diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 19049e87f5..59e4e5f10e 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -118,6 +118,19 @@ the raw JSON frame and its SSE envelope at 4 MiB, and closes the upstream when i would overflow. That overflow emits a terminal downstream `response.failed` event followed by `[DONE]`. +The upstream WebSocket checks `NO_PROXY`/`no_proxy` first. Otherwise it uses the first non-empty +`HTTPS_PROXY`, `https_proxy`, `ALL_PROXY`, or `all_proxy` value; `HTTP_PROXY` alone does not proxy a +WSS connection. HTTP and HTTPS proxy URLs are passed to Bun. If the selected value is invalid or +uses an unsupported protocol, opencodex skips the WebSocket attempt and uses HTTP/SSE instead of +dialing the upstream directly. + +These rules belong to the upstream WebSocket transport, independently of the selected provider +adapter. HTTP fetch-based Responses requests, including SSE fallback, use Bun's HTTP proxy rules +and do not use `ALL_PROXY`. `config.proxy` fills missing `HTTP_PROXY`/`HTTPS_PROXY` values; the +resulting scheme-specific value also takes precedence over an existing `ALL_PROXY` for WebSocket. +For an HTTPS upstream that requires a proxy, set `HTTPS_PROXY` or `config.proxy`; `HTTP_PROXY` +alone leaves both WSS and its HTTPS fallback without a scheme-matched proxy. + Every terminal Responses usage object includes both detail objects, even when the provider did not report those details: diff --git a/src/config.ts b/src/config.ts index 5d67275dce..72da455382 100644 --- a/src/config.ts +++ b/src/config.ts @@ -3738,11 +3738,12 @@ function warnProxyConfigDiscardOnce(kind: "proxy" | "noProxy" | "noProxyElements } /** - * Mirror `config.proxy` into HTTP(S)_PROXY env vars so Bun's native fetch routes every outbound - * provider call through the proxy — no per-callsite changes (verified: Bun honors these plus - * NO_PROXY). User-set env vars always win; localhost/127.0.0.1 are appended to NO_PROXY so the - * CLI's own health checks and running-proxy API calls stay direct. Call once per process entry - * that makes outbound provider requests (server start, catalog sync). + * Mirror `config.proxy` into HTTP(S)_PROXY env vars. Bun fetch consumes them natively; transports + * such as the ChatGPT upstream WebSocket select the same environment explicitly. User-set HTTP(S)_PROXY + * variables win; config fills missing scheme proxies, which take precedence over ALL_PROXY for WS. + * localhost/127.0.0.1 are appended to NO_PROXY so the CLI's own health checks and + * running-proxy API calls stay direct. Call once per process entry that makes outbound provider + * requests (server start, catalog sync). */ export function applyProxyEnv(config: OcxConfig): void { applyProxyEnvWith(config); diff --git a/src/lib/provider-outbound.ts b/src/lib/provider-outbound.ts index 495fef0b8b..02bdbc2077 100644 --- a/src/lib/provider-outbound.ts +++ b/src/lib/provider-outbound.ts @@ -7,7 +7,7 @@ import { resolvePublicAddresses, } from "./destination-policy"; import { pinnedHttpGet, pinnedHttpPost } from "./pinned-http"; -import { effectiveProxyFor, outboundProxyConfigured } from "./proxy-env"; +import { effectiveProxyFor, noProxyMatches, normalizeProxyHostname, outboundProxyConfigured } from "./proxy-env"; import { publicProviderBaseUrl } from "./provider-url"; type ProviderGetInit = Omit; @@ -37,10 +37,6 @@ function pickPinnedAddress(addresses: Array<{ address: string; family: number }> return addresses.find(address => address.family === 4) ?? addresses[0]!; } -function configuredProxyFor(): boolean { - return outboundProxyConfigured(); -} - /** * Registry-owned fake-IP transparency exception (Clash/Surge/Mihomo TUN mode). * @@ -76,45 +72,6 @@ function transparentFakeIpException( return isCanonicalUrl(name, url); } -function normalizeProxyHostname(hostname: string): string { - const normalized = hostname.trim().toLowerCase().replace(/\.+$/, ""); - return normalized.startsWith("[") && normalized.endsWith("]") - ? normalized.slice(1, -1) - : normalized; -} - -function noProxyMatches(url: URL): boolean { - const raw = process.env.NO_PROXY ?? process.env.no_proxy ?? ""; - const hostname = normalizeProxyHostname(url.hostname); - const port = url.port || (url.protocol === "https:" ? "443" : "80"); - for (const rawEntry of raw.split(",")) { - let entry = rawEntry.trim().toLowerCase(); - if (!entry) continue; - if (entry === "*") return true; - entry = entry.replace(/^https?:\/\//, "").split("/", 1)[0]!; - - let entryHost = entry; - let entryPort = ""; - const bracketed = /^\[([^\]]+)](?::(\d+))?$/.exec(entry); - if (bracketed) { - entryHost = bracketed[1]!; - entryPort = bracketed[2] ?? ""; - } else if ((entry.match(/:/g)?.length ?? 0) === 1) { - const separator = entry.lastIndexOf(":"); - const possiblePort = entry.slice(separator + 1); - if (/^\d+$/.test(possiblePort)) { - entryHost = entry.slice(0, separator); - entryPort = possiblePort; - } - } - if (entryPort && entryPort !== port) continue; - entryHost = normalizeProxyHostname(entryHost.replace(/^\*?\./, "")); - if (!entryHost) continue; - if (hostname === entryHost || hostname.endsWith(`.${entryHost}`)) return true; - } - return false; -} - let proxyBoundaryWarned = false; let proxyDnsDegradationWarned = false; @@ -181,7 +138,7 @@ async function providerOutboundRequest( return provider.fetch(url, { ...init, method, redirect: "manual" }); } const parsed = postUrl ?? new URL(url); - const proxyConfigured = configuredProxyFor(); + const proxyConfigured = outboundProxyConfigured(); // Snapshot the scheme-matched proxy once, before the DNS await, so admission and transport // below reason about the same value. `null` here means "no proxy fetch would actually use", // even if some other proxy variable is set. diff --git a/src/lib/proxy-env.ts b/src/lib/proxy-env.ts index 46df592689..0ac9ed735c 100644 --- a/src/lib/proxy-env.ts +++ b/src/lib/proxy-env.ts @@ -3,6 +3,73 @@ export const PROXY_ENV_KEYS = [...OUTBOUND_PROXY_ENV_KEYS, "NO_PROXY"] as const; export type ProxyEnvKey = typeof PROXY_ENV_KEYS[number]; export type ProxyEnvMap = Record; +export type ProxyRoute = + | { kind: "direct" } + | { kind: "proxy"; proxy: string } + | { kind: "fallback" }; + +export function normalizeProxyHostname(hostname: string): string { + const normalized = hostname.trim().toLowerCase().replace(/\.+$/, ""); + return normalized.startsWith("[") && normalized.endsWith("]") + ? normalized.slice(1, -1) + : normalized; +} + +export function noProxyMatches( + url: URL, + env: ProxyEnvMap = process.env, +): boolean { + const raw = env.NO_PROXY ?? env.no_proxy ?? ""; + const hostname = normalizeProxyHostname(url.hostname); + const port = url.port || (url.protocol === "https:" || url.protocol === "wss:" ? "443" : "80"); + for (const rawEntry of raw.split(",")) { + let entry = rawEntry.trim().toLowerCase(); + if (!entry) continue; + if (entry === "*") return true; + entry = entry.replace(/^(?:https?|wss?):\/\//, "").split("/", 1)[0]!; + + let entryHost = entry; + let entryPort = ""; + const bracketed = /^\[([^\]]+)](?::(\d+))?$/.exec(entry); + if (bracketed) { + entryHost = bracketed[1]!; + entryPort = bracketed[2] ?? ""; + } else if ((entry.match(/:/g)?.length ?? 0) === 1) { + const separator = entry.lastIndexOf(":"); + const possiblePort = entry.slice(separator + 1); + if (/^\d+$/.test(possiblePort)) { + entryHost = entry.slice(0, separator); + entryPort = possiblePort; + } + } + if (entryPort && entryPort !== port) continue; + entryHost = normalizeProxyHostname(entryHost.replace(/^\*?\./, "")); + if (entryHost && (hostname === entryHost || hostname.endsWith(`.${entryHost}`))) return true; + } + return false; +} + +export function resolveProxyRoute( + url: URL, + env: ProxyEnvMap = process.env, +): ProxyRoute { + if (noProxyMatches(url, env)) return { kind: "direct" }; + const key = url.protocol === "https:" || url.protocol === "wss:" + ? "HTTPS_PROXY" + : "HTTP_PROXY"; + const proxy = [key, key.toLowerCase(), "ALL_PROXY", "all_proxy"] + .map(candidate => env[candidate]?.trim()) + .find(Boolean); + if (!proxy) return { kind: "direct" }; + try { + const protocol = new URL(proxy).protocol; + return protocol === "http:" || protocol === "https:" + ? { kind: "proxy", proxy } + : { kind: "fallback" }; + } catch { + return { kind: "fallback" }; + } +} export function proxyEnvPresent( key: ProxyEnvKey, diff --git a/src/server/responses/codex-ws-pool.ts b/src/server/responses/codex-ws-pool.ts index 378cf2d4a3..5d406bee4f 100644 --- a/src/server/responses/codex-ws-pool.ts +++ b/src/server/responses/codex-ws-pool.ts @@ -25,7 +25,7 @@ function digest(input: unknown): string { } /** Identity comes from the selected outgoing request, never a model label or caller hint. */ -export function codexWsReuseIdentity(url: string, headers: Record, frameText: string): CodexWsReuseIdentity | null { +export function codexWsReuseIdentity(url: string, headers: Record, frameText: string, proxy?: string): CodexWsReuseIdentity | null { if (url !== CODEX_RESPONSES_HTTP_URL) return null; let body: unknown; try { body = JSON.parse(frameText); } catch { return null; } @@ -52,7 +52,7 @@ export function codexWsReuseIdentity(url: string, headers: Record): CodexWsSession | null { + acquire(identity: CodexWsReuseIdentity, url: string, headers: Record, proxy?: string): CodexWsSession | null { this.sweep(); for (const entry of this.entries.values()) { if (entry.identity.scope !== identity.scope || entry.identity.key === identity.key) continue; @@ -94,7 +94,7 @@ export class CodexWsPool { this.remove(oldest); } const createdAt = this.now(); - const session = new CodexWsSession(url, headers, true, () => this.changed(entry)); + const session = new CodexWsSession(url, headers, true, () => this.changed(entry), proxy); const entry: Entry = { identity, session, createdAt, idleAt: createdAt, retired: false }; session.reserve(); this.entries.set(identity.key, entry); diff --git a/src/server/responses/codex-ws-session.ts b/src/server/responses/codex-ws-session.ts index bbf62f8137..32716a5297 100644 --- a/src/server/responses/codex-ws-session.ts +++ b/src/server/responses/codex-ws-session.ts @@ -10,8 +10,8 @@ export class CodexWsSession { private readonly completedIds = new Set(); constructor(url: string, headers: Record, readonly retainable = false, - private readonly changed: () => void = () => {}) { - this.socket = new WebSocket(url, { headers } as unknown as string[]); + private readonly changed: () => void = () => {}, proxy?: string) { + this.socket = new WebSocket(url, { headers, ...(proxy ? { proxy } : {}) } as unknown as string[]); this.socket.addEventListener("open", this.onOpen); this.socket.addEventListener("message", this.onIdleMessage); this.socket.addEventListener("close", this.onClose); diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index e9773d02a3..87b3767d2b 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -13,6 +13,7 @@ // (passthrough relay, adapter parsers, usage sniffing) is unchanged. import { compareBunVersions } from "../../lib/bun-stream-caps"; +import { resolveProxyRoute } from "../../lib/proxy-env"; import type { CodexWsQuotaObserver } from "./codex-ws-metadata"; import { CODEX_RESPONSES_HTTP_URL, CODEX_RESPONSES_WS_URL, prepareCodexHttpInit, prepareCodexWsRequest } from "./codex-ws-request"; import { codexWsExchange } from "./codex-ws-exchange"; @@ -150,6 +151,10 @@ export function codexWsUpstreamFetch( return sseFallback(url, init); } + const wsUrl = wsUpstreamUrlFor(url); + const proxyRoute = resolveProxyRoute(new URL(wsUrl)); + if (proxyRoute.kind === "fallback") return sseFallback(url, init); + const proxy = proxyRoute.kind === "proxy" ? proxyRoute.proxy : undefined; // A genuine caller `originator` is already in these headers via the forward // set. Never fabricate one here: pool/forward traffic must not impersonate // Codex CLI, per the metadata-integrity contract. (The backend's fast lane @@ -164,9 +169,9 @@ export function codexWsUpstreamFetch( } let session: CodexWsSession; try { - const identity = codexWsReuseIdentity(url, headers, frameText); - session = (identity ? codexWsPool.acquire(identity, wsUpstreamUrlFor(url), headers) : null) - ?? new CodexWsSession(wsUpstreamUrlFor(url), headers); + const identity = codexWsReuseIdentity(url, headers, frameText, proxy); + session = (identity ? codexWsPool.acquire(identity, wsUrl, headers, proxy) : null) + ?? new CodexWsSession(wsUrl, headers, false, undefined, proxy); if (!session.busy && !session.reserve()) { session.dispose(); return sseFallback(url, init); diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 4ee22c114c..a45a98c87b 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -435,7 +435,7 @@ These are transport-fidelity guarantees, not a provider-billing guarantee. Eligible complete-input creates can retain a canonical upstream socket within one selected account, credential, thread and turn. Model/tier and immutable -handshake headers must also match. Turn-state and turn-metadata headers are +handshake headers and the selected outbound proxy must also match. Turn-state and turn-metadata headers are projected into their same-name per-frame metadata slots; explicit body values win. The pool retains at most 32 sockets, expires idle sockets after 30 seconds, and retires a socket after five minutes or 32 successful exchanges (after active work @@ -644,7 +644,11 @@ the upgrade with 426 so Codex falls back to HTTP cleanly. That setting controls the client-facing upgrade only. The transparent upstream ChatGPT WS optimization described above is selected independently and still -returns the same downstream SSE contract. +returns the same downstream SSE contract. Its WSS route checks NO_PROXY first, then selects the +first non-empty HTTPS_PROXY, https_proxy, ALL_PROXY, or all_proxy value. HTTP_PROXY alone does not +route WSS. Unsupported or malformed selected proxy values skip the WebSocket attempt and use the +existing SSE path immediately; they never fall through to a lower-priority proxy or direct WebSocket +egress. HTTP/SSE fallback retains Bun fetch's own proxy rules, which do not consult ALL_PROXY. The endpoint handles `response.create`, ignores `response.processed`, supports warmup `generate: false`, and feeds the same request pipeline as HTTP/SSE. diff --git a/tests/responses/ws-upstream-reuse.test.ts b/tests/responses/ws-upstream-reuse.test.ts index fd0a8fb5a1..b957fdb317 100644 --- a/tests/responses/ws-upstream-reuse.test.ts +++ b/tests/responses/ws-upstream-reuse.test.ts @@ -6,6 +6,8 @@ import { prepareCodexWsRequest } from "../../src/server/responses/codex-ws-reque const URL = "https://chatgpt.com/backend-api/codex/responses"; const realWebSocket = globalThis.WebSocket; +const proxyEnvKeys = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy"]; +let savedProxyEnv: Record; let sequence = 0; class Socket extends EventTarget { @@ -13,7 +15,7 @@ class Socket extends EventTarget { static onSend: (socket: Socket, frame: Record) => void = (socket) => socket.complete(); readyState = 0; frames: Record[] = []; - constructor(readonly url: string) { + constructor(readonly url: string, readonly options?: { proxy?: string }) { super(); Socket.all.push(this); queueMicrotask(() => { if (this.readyState === 0) { this.readyState = 1; this.dispatchEvent(new Event("open")); } }); @@ -58,7 +60,11 @@ function bodyWith(fields: Record) { options.body = JSON.stringify({ ...JSON.parse(options.body as string), ...fields }); return options; } -beforeEach(() => { globalThis.WebSocket = Socket as unknown as typeof WebSocket; }); +beforeEach(() => { + globalThis.WebSocket = Socket as unknown as typeof WebSocket; + savedProxyEnv = Object.fromEntries(proxyEnvKeys.map(key => [key, process.env[key]])); + for (const key of proxyEnvKeys) delete process.env[key]; +}); afterEach(() => { runOptionalShutdownHooks(); @@ -67,6 +73,25 @@ afterEach(() => { Socket.onSend = socket => socket.complete(); sequence = 0; globalThis.WebSocket = realWebSocket; + for (const key of proxyEnvKeys) delete process.env[key]; + for (const key of proxyEnvKeys) { + if (savedProxyEnv[key] !== undefined) process.env[key] = savedProxyEnv[key]; + } +}); + +test("proxy changes and NO_PROXY retire the old route while unchanged routes reuse", async () => { + for (const proxy of ["http://proxy-a.example:8080", "http://proxy-b.example:8080"]) { + process.env.HTTPS_PROXY = proxy; + await drain(); + await drain(); + } + process.env.NO_PROXY = "chatgpt.com:443"; + await drain(); + await drain(); + expect(Socket.all.map(socket => socket.options?.proxy)) + .toEqual(["http://proxy-a.example:8080", "http://proxy-b.example:8080", undefined]); + expect(Socket.all.map(socket => socket.frames.length)).toEqual([2, 2, 2]); + expect(Socket.all.map(socket => socket.readyState)).toEqual([3, 3, 1]); }); test("same account/thread/turn reuses one socket without trimming either HTTP input", async () => { diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts index fd09513070..cfb087a4bb 100644 --- a/tests/responses/ws-upstream.test.ts +++ b/tests/responses/ws-upstream.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, jest, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, jest, test } from "bun:test"; import { providerFetch } from "../../src/server/responses/fetch-helpers"; import { handleResponses } from "../../src/server/responses"; import { isEagerRelaySseResponse } from "../../src/server/relay"; @@ -162,18 +162,24 @@ describe("shouldUseCodexWsUpstream", () => { }); type Listener = (event: unknown) => void; +type FakeWebSocketOptions = { + headers?: Record; + proxy?: string; +}; /** Minimal scriptable stand-in for Bun's WebSocket. */ class FakeWebSocket { static instances: FakeWebSocket[] = []; static script: (ws: FakeWebSocket) => void = () => {}; url: string; + options?: FakeWebSocketOptions; sent: string[] = []; closed = false; listeners = new Map(); - constructor(url: string) { + constructor(url: string, options?: FakeWebSocketOptions) { this.url = url; + this.options = options; FakeWebSocket.instances.push(this); queueMicrotask(() => FakeWebSocket.script(this)); } @@ -205,12 +211,23 @@ class FakeWebSocket { const RealWebSocket = globalThis.WebSocket; const RealFetch = globalThis.fetch; +const PROXY_ENV_KEYS = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy"] as const; +let savedProxyEnv: Record; + +beforeEach(() => { + savedProxyEnv = Object.fromEntries(PROXY_ENV_KEYS.map(key => [key, process.env[key]])); + for (const key of PROXY_ENV_KEYS) delete process.env[key]; +}); afterEach(() => { globalThis.WebSocket = RealWebSocket; globalThis.fetch = RealFetch; FakeWebSocket.instances = []; FakeWebSocket.script = () => {}; + for (const key of PROXY_ENV_KEYS) delete process.env[key]; + for (const key of PROXY_ENV_KEYS) { + if (savedProxyEnv[key] !== undefined) process.env[key] = savedProxyEnv[key]; + } }); function installFake(script: (ws: FakeWebSocket) => void) { @@ -525,6 +542,41 @@ describe("codexWsUpstreamFetch", () => { expect(text).not.toContain("must-not-leak"); }); + test("passes the selected proxy without changing handshake headers", async () => { + process.env.HTTPS_PROXY = "http://proxy.example:8080"; + installFake(ws => { + ws.emit("open", {}); + ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: {} }) }); + }); + + await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => { + throw new Error("fallback must not run"); + }) as unknown as typeof fetch); + + const options = FakeWebSocket.instances[0]!.options; + expect(options?.proxy).toBe("http://proxy.example:8080"); + expect(options?.headers?.authorization).toBe("Bearer test"); + expect(options?.headers?.["openai-beta"]).toContain("responses_websockets"); + expect(options?.headers?.["content-type"]).toBeUndefined(); + }); + + test.each([ + ["unsupported protocol", "socks5://proxy.example:1080"], + ["invalid URL", "not a proxy URL"], + ])("falls back once without dialing for an %s", async (_label, proxy) => { + process.env.HTTPS_PROXY = proxy; + const sentinel = new Response("sse-fallback"); + let fallbackCalls = 0; + const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (async () => { + fallbackCalls += 1; + return sentinel; + }) as typeof fetch); + + expect(response).toBe(sentinel); + expect(fallbackCalls).toBe(1); + expect(FakeWebSocket.instances).toHaveLength(0); + }); + test("relays event frames as an SSE response and sends one response.create frame", async () => { installFake(ws => { ws.emit("open", {}); @@ -654,6 +706,7 @@ describe("codexWsUpstreamFetch", () => { }); test("falls back to the HTTP fetch when the upgrade is rejected before open", async () => { + process.env.HTTPS_PROXY = "http://proxy.example:8080"; installFake(ws => ws.close()); const sentinel = new Response("sse-fallback", { status: 429 }); let fallbackCalls = 0; @@ -666,6 +719,7 @@ describe("codexWsUpstreamFetch", () => { expect(response).toBe(sentinel); expect(isCodexWsUpstreamResponse(response)).toBe(false); expect(fallbackCalls).toBe(1); + expect(FakeWebSocket.instances[0]!.options?.proxy).toBe("http://proxy.example:8080"); }); test("falls back to the HTTP fetch when the upgrade deadline elapses without open or close", async () => { @@ -800,15 +854,17 @@ describe("codexWsUpstreamFetch", () => { }); test("preserves caller headers on the handshake without fabricating an originator", async () => { - const seen: Record[] = []; + process.env.HTTPS_PROXY = "http://proxy.example:8080"; + process.env.NO_PROXY = "chatgpt.com:443"; + const seen: FakeWebSocketOptions[] = []; FakeWebSocket.script = ws => { ws.emit("open", {}); ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: {} }) }); }; class HeaderCapturingWebSocket extends FakeWebSocket { - constructor(url: string, options?: { headers?: Record }) { - super(url); - seen.push(options?.headers ?? {}); + constructor(url: string, options?: FakeWebSocketOptions) { + super(url, options); + seen.push(options ?? {}); } } globalThis.WebSocket = HeaderCapturingWebSocket as unknown as typeof WebSocket; @@ -817,18 +873,19 @@ describe("codexWsUpstreamFetch", () => { await codexWsUpstreamFetch(CODEX_URL, streamingInit(), fallback); // Without a caller originator none is invented: pool/forward traffic must // not impersonate Codex CLI (metadata-integrity contract). - expect(seen[0].originator).toBeUndefined(); - expect(seen[0]["openai-beta"]).toContain("responses_websockets"); - expect(seen[0].authorization).toBe("Bearer test"); + expect(seen[0].proxy).toBeUndefined(); + expect(seen[0].headers?.originator).toBeUndefined(); + expect(seen[0].headers?.["openai-beta"]).toContain("responses_websockets"); + expect(seen[0].headers?.authorization).toBe("Bearer test"); // HTTP body-framing headers do not belong on a WS handshake. - expect(seen[0]["content-type"]).toBeUndefined(); + expect(seen[0].headers?.["content-type"]).toBeUndefined(); // A genuine caller originator is forwarded verbatim. await codexWsUpstreamFetch(CODEX_URL, { ...streamingInit(), headers: { ...streamingInit().headers as Record, originator: "codex_cli_rs" }, }, fallback); - expect(seen[1].originator).toBe("codex_cli_rs"); + expect(seen[1].headers?.originator).toBe("codex_cli_rs"); }); test("aborting before open rejects like an aborted fetch", async () => { @@ -1194,6 +1251,8 @@ describe("oversized Codex create frames", () => { }); test("dials the configured provider's own wss URL for an opt-in upstream", async () => { + process.env.HTTPS_PROXY = "http://proxy.example:8080"; + process.env.NO_PROXY = "sub2api.example.com:443"; installFake(ws => { ws.emit("open", {}); ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: { id: "r-ws" } }) }); @@ -1206,6 +1265,7 @@ describe("oversized Codex create frames", () => { ); expect(FakeWebSocket.instances).toHaveLength(1); expect(FakeWebSocket.instances[0]!.url).toBe("wss://sub2api.example.com/v1/responses"); + expect(FakeWebSocket.instances[0]!.options?.proxy).toBeUndefined(); expect(response.headers.get("content-type")).toContain("text/event-stream"); expect(await response.text()).toContain("response.completed"); }); diff --git a/tests/server/proxy-env.test.ts b/tests/server/proxy-env.test.ts index e43ad2d9be..c795c6cf2d 100644 --- a/tests/server/proxy-env.test.ts +++ b/tests/server/proxy-env.test.ts @@ -1,8 +1,10 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { createServer } from "node:http"; import { applyProxyEnv } from "../../src/config"; +import { resolveProxyRoute } from "../../src/lib/proxy-env"; import type { OcxConfig } from "../../src/types"; -const PROXY_ENV_KEYS = ["HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "no_proxy", "OCX_TEST_PROXY_REF", "OCX_TEST_NO_PROXY_REF"] as const; +const PROXY_ENV_KEYS = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy", "OCX_TEST_PROXY_REF", "OCX_TEST_NO_PROXY_REF"] as const; let saved: Record; beforeEach(() => { @@ -30,6 +32,128 @@ function configWithRawProxy(proxy: unknown, noProxy?: unknown): OcxConfig { return { proxy, noProxy, providers: {} } as unknown as OcxConfig; } +describe("resolveProxyRoute", () => { + test("wss uses HTTPS_PROXY and never HTTP_PROXY", () => { + const target = new URL("wss://chatgpt.com/backend-api/codex/responses"); + expect(resolveProxyRoute(target, { + HTTPS_PROXY: "http://secure-proxy.example:8443", + HTTP_PROXY: "http://plain-proxy.example:8080", + })).toEqual({ kind: "proxy", proxy: "http://secure-proxy.example:8443" }); + expect(resolveProxyRoute(target, { + HTTP_PROXY: "http://plain-proxy.example:8080", + })).toEqual({ kind: "direct" }); + }); + + test.each([ + ["exact host", "wss://chatgpt.com/path", "chatgpt.com", "direct"], + ["domain suffix", "wss://api.chatgpt.com/path", ".chatgpt.com", "direct"], + ["wildcard suffix", "wss://api.chatgpt.com/path", "*.chatgpt.com", "direct"], + ["wss default port", "wss://chatgpt.com/path", "chatgpt.com:443", "direct"], + ["ws default port", "ws://chatgpt.com/path", "chatgpt.com:80", "direct"], + ["port mismatch", "wss://chatgpt.com/path", "chatgpt.com:80", "proxy"], + ["bracketed IPv6", "wss://[2001:db8::1]/path", "[2001:db8::1]:443", "direct"], + ["URL-style entry", "wss://chatgpt.com/path", "https://chatgpt.com/ignored", "direct"], + ] as const)("honors NO_PROXY for %s", (_label, target, noProxy, expectedKind) => { + expect(resolveProxyRoute(new URL(target), { + HTTPS_PROXY: "http://secure-proxy.example:8443", + NO_PROXY: noProxy, + }).kind).toBe(expectedKind); + }); + + test("uses stable proxy precedence and fails closed on the first unusable proxy", () => { + const target = new URL("wss://chatgpt.com/backend-api/codex/responses"); + const route = (env: Record) => resolveProxyRoute(target, env); + expect([ + route({ HTTPS_PROXY: "http://upper-https:1", https_proxy: "http://lower-https:2", ALL_PROXY: "http://upper-all:3", all_proxy: "http://lower-all:4" }), + route({ HTTPS_PROXY: " ", https_proxy: "http://lower-https:2", ALL_PROXY: "http://upper-all:3" }), + route({ ALL_PROXY: "http://upper-all:3", all_proxy: "http://lower-all:4" }), + route({ all_proxy: "https://lower-all:4" }), + route({ HTTPS_PROXY: "socks5://unsupported:1080", ALL_PROXY: "http://must-not-win:3" }), + route({ HTTPS_PROXY: "not a proxy URL", ALL_PROXY: "http://must-not-win:3" }), + route({}), + ]).toEqual([ + { kind: "proxy", proxy: "http://upper-https:1" }, + { kind: "proxy", proxy: "http://lower-https:2" }, + { kind: "proxy", proxy: "http://upper-all:3" }, + { kind: "proxy", proxy: "https://lower-all:4" }, + { kind: "fallback" }, + { kind: "fallback" }, + { kind: "direct" }, + ]); + }); + + test("preserves uppercase NO_PROXY precedence when it is explicitly empty", () => { + expect(resolveProxyRoute(new URL("wss://chatgpt.com/path"), { + HTTPS_PROXY: "http://secure-proxy.example:8443", + NO_PROXY: "", + no_proxy: "chatgpt.com", + })).toEqual({ kind: "proxy", proxy: "http://secure-proxy.example:8443" }); + }); + + test("Bun WebSocket sends WSS through an HTTP CONNECT proxy", async () => { + let resolveConnect!: (target: string) => void; + const connected = new Promise(resolve => { resolveConnect = resolve; }); + const proxy = createServer(); + proxy.on("connect", (request, socket) => { + resolveConnect(request.url ?? ""); + socket.end("HTTP/1.1 502 Probe Complete\r\nContent-Length: 0\r\n\r\n"); + }); + await new Promise((resolve, reject) => { + proxy.once("error", reject); + proxy.listen(0, "127.0.0.1", resolve); + }); + const address = proxy.address(); + if (!address || typeof address === "string") throw new Error("proxy did not bind a TCP port"); + const socket = new WebSocket("wss://proxy-probe.invalid/backend-api/codex/responses", { + proxy: `http://127.0.0.1:${address.port}`, + } as unknown as string[]); + try { + expect(await Promise.race([ + connected, + new Promise((_, reject) => setTimeout(() => reject(new Error("CONNECT was not observed")), 5_000)), + ])).toBe("proxy-probe.invalid:443"); + } finally { + try { socket.close(); } catch { /* probe is already complete */ } + await new Promise(resolve => proxy.close(() => resolve())); + } + }, 10_000); + + test.skipIf(process.platform !== "win32")("Bun fetch honors NO_PROXY on Windows", async () => { + let providerRequests = 0; + let proxyRequests = 0; + const provider = createServer((_request, response) => { + providerRequests += 1; + response.end("direct"); + }); + const proxy = createServer((_request, response) => { + proxyRequests += 1; + response.end("proxied"); + }); + const listen = async (server: typeof provider): Promise => { + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("server did not bind a TCP port"); + return address.port; + }; + const [providerPort, proxyPort] = await Promise.all([listen(provider), listen(proxy)]); + process.env.HTTP_PROXY = `http://127.0.0.1:${proxyPort}`; + process.env.NO_PROXY = "127.0.0.1"; + try { + expect(await (await fetch(`http://127.0.0.1:${providerPort}/models`)).text()).toBe("direct"); + expect(providerRequests).toBe(1); + expect(proxyRequests).toBe(0); + } finally { + await Promise.all([ + new Promise(resolve => provider.close(() => resolve())), + new Promise(resolve => proxy.close(() => resolve())), + ]); + } + }); +}); + describe("applyProxyEnv with values the schema does not constrain", () => { test("warns once per discarded proxy setting without exposing its raw value", () => { const secret = "raw-proxy-credential-sentinel-2947"; @@ -122,6 +246,14 @@ describe("applyProxyEnv", () => { expect(process.env.HTTP_PROXY).toBe("http://proxy.corp:8080"); }); + test.each(["ALL_PROXY", "all_proxy"])("config fills a scheme proxy ahead of %s for WSS", key => { + process.env[key] = "http://fallback-proxy.example:8081"; + applyProxyEnv(configWithProxy("http://configured-proxy.example:8080")); + expect(process.env[key]).toBe("http://fallback-proxy.example:8081"); + expect(resolveProxyRoute(new URL("wss://chatgpt.com/backend-api/codex/responses"))) + .toEqual({ kind: "proxy", proxy: "http://configured-proxy.example:8080" }); + }); + test("appends loopback entries to an existing NO_PROXY without duplicating", () => { process.env.NO_PROXY = "internal.corp,localhost"; applyProxyEnv(configWithProxy("http://proxy.corp:8080")); @@ -217,4 +349,3 @@ describe("applyProxyEnv with proxy: \"auto\" (#1525)", () => { expect(process.env.HTTP_PROXY).toBeUndefined(); }); }); - From b0f0a94855196126ed30db868730d6b26927d2cc Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 01:47:46 +0900 Subject: [PATCH 3/7] docs: refresh native MESSAGE recovery layer --- devlog/_plan/260906_a_runtime_stack/031_recovery_refresh.md | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 devlog/_plan/260906_a_runtime_stack/031_recovery_refresh.md diff --git a/devlog/_plan/260906_a_runtime_stack/031_recovery_refresh.md b/devlog/_plan/260906_a_runtime_stack/031_recovery_refresh.md new file mode 100644 index 0000000000..9112a2c8d7 --- /dev/null +++ b/devlog/_plan/260906_a_runtime_stack/031_recovery_refresh.md @@ -0,0 +1,3 @@ +# Recovery layer P refresh + +Consume 030 above prepared WS parent10fbda2e0. Original #3568 remains open at036a9321788464fdf33a387c9f44a834a844bdc1; carry all three voiys commits in order. No catalog/effort hunks from #3571. Add the planned cache exact-expiry/no-TTL-extension and canonical-Go conversion negatives, with a scoped inherited worker owning only the named three regression files after original carry. Main owns production integration, author commits and review. Runtime correction: isolated checks now invoke repository node_modules/.bin/bun and assert package.json dependencies.bun=1.4.0 before any execution. Full per-head CI remains mandatory before landing. #3661 remains partial, with no automatic close reference. From 106e004f98e199869b94ba0c8c34f45e0776af12 Mon Sep 17 00:00:00 2001 From: voiys Date: Sat, 5 Sep 2026 03:17:03 +0200 Subject: [PATCH 4/7] fix(responses): recover native messages for Go subagents (cherry picked from commit e8f8726040dbc45b1e946d59db6b9c477459b8d7) --- .../docs/reference/configuration/providers.md | 25 +++ scripts/test-layout/layout.json | 2 + src/adapters/openai-responses.ts | 2 + src/adapters/opencode-go.ts | 35 +++++ .../responses/agent-task-recovery-cache.ts | 11 ++ src/server/responses/agent-task-recovery.ts | 28 +++- src/server/responses/core.ts | 11 +- tests/fixtures/test-layout-expected.json | 2 + .../opencode-go-agent-messages.test.ts | 56 +++++++ .../server-agent-task-recovery-replay.test.ts | 142 ++++++++++++++++++ 10 files changed, 307 insertions(+), 7 deletions(-) create mode 100644 src/adapters/opencode-go.ts create mode 100644 tests/providers/opencode-go-agent-messages.test.ts create mode 100644 tests/server/server-agent-task-recovery-replay.test.ts diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index d6adbdac9c..7282a21bd3 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -810,3 +810,28 @@ ids with context `922000` and max input `922000`; OpenRouter seeds `openai/gpt-5 "visionSidecar": { "enabled": true } } ``` + +## OpenCode Go session and agent messages + +On Go Responses routes, plaintext Codex `agent_message` items become user messages. +Author and recipient remain explicit text metadata, and the content parts are preserved. +Encrypted and unknown content is not normalized; native encrypted tasks still require the +separate opt-in [task recovery](/reference/configuration/agents/#encrypted-v2-task-recovery). + +With task recovery enabled, replayed `NEW_TASK` and `MESSAGE` items reuse a cached assignment only +after validating the caller and matching the parent-thread scope. Replay restoration +does not make a new recovery request or extend cache expiry. Expired or unseen +ciphertext is not replaced. Fresh encrypted `NEW_TASK` and `MESSAGE` items use the same +opt-in recovery path, including native-parent `send_message` delivery. Message type, +sender, recipient, parent scope and caller credentials remain part of validation or cache identity. + +When a request contains several agent messages, cached replay restoration checks each +message independently. The cache separates message type, sender, recipient and ciphertext +within the admitted caller/account and parent scope. Fresh recovery only handles the +current tail message (ignoring trailing `compaction_trigger` or `additional_tools` metadata). +It does not batch-recover unseen historical messages; those remain unchanged. A cache miss +or expiry does not extend the history-recovery contract. + +Sender and recipient on Go Responses are context for the receiving model, not a new +machine-readable routing protocol. Tool routing continues to use the existing collaboration +contracts. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 29dd2c5f1c..b740dcbf20 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -889,6 +889,7 @@ "openai-responses-passthrough.test.ts": "responses", "opencode-cli.test.ts": "providers", "opencode-free-provider.test.ts": "providers", + "opencode-go-agent-messages.test.ts": "providers", "opencode-go-deepseek.test.ts": "providers", "opencode-go-grok46-responses.test.ts": "providers", "opencode-go-luna-wire.test.ts": "providers", @@ -1064,6 +1065,7 @@ "selected-models.test.ts": "codex-integration", "self-launch-argv.test.ts": "lib", "server-403-permission-e2e.test.ts": "server", + "server-agent-task-recovery-replay.test.ts": "server", "server-auth.test.ts": "server", "server-background-lifecycle.test.ts": "server", "server-clickjacking-headers.test.ts": "server", diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index a007e21a4a..c4bbae4a09 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -1,3 +1,4 @@ +import { isOpenCodeGo, normalizeOpenCodeGoAgentMessages } from "./opencode-go"; import { createHash } from "node:crypto"; import type { IncomingMeta, ProviderAdapter } from "./base"; import { namespacedToolName, type AdapterEvent, type OcxParsedRequest, type OcxProviderConfig, type OcxUsage, type TierDecision } from "../types"; @@ -2355,6 +2356,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): parsed._rawBody, forward || parsed._previousResponseInputExpanded === true, ); + if (!forward && isOpenCodeGo(provider.baseUrl)) outBody = normalizeOpenCodeGoAgentMessages(outBody); outBody = mapRoutedResponsesReasoningEffort(outBody, provider, parsed.modelId); // stripPreviousResponseId() intentionally returns its input on a no-op. Detach before the // tier write so a force-fast/default decision can never mutate parsed._rawBody. diff --git a/src/adapters/opencode-go.ts b/src/adapters/opencode-go.ts new file mode 100644 index 0000000000..94055a292a --- /dev/null +++ b/src/adapters/opencode-go.ts @@ -0,0 +1,35 @@ +/** Match the Go destination, including user-renamed provider entries. */ +export function isOpenCodeGo(baseUrl: string): boolean { + try { + const url = new URL(baseUrl); + return url.origin === "https://opencode.ai" && url.pathname.replace(/\/+$/, "") === "/zen/go/v1"; + } catch { return false; } +} + +/** Public Responses rejects Codex's private agent_message variant, even with plaintext content. */ +export function normalizeOpenCodeGoAgentMessages(body: unknown): unknown { + if (!body || typeof body !== "object" || Array.isArray(body)) return body; + const record = body as Record; + if (!Array.isArray(record.input)) return body; + let changed = false; + const input = record.input.map((item: unknown) => { + if (!item || typeof item !== "object" || Array.isArray(item)) return item; + const message = item as Record; + if (message.type !== "agent_message" || !Array.isArray(message.content) || message.content.length === 0) return item; + // Genuine ciphertext and unknown part types must retain their existing fail-closed path. + if (!message.content.every(part => part && typeof part === "object" + && ["input_text", "input_image", "input_file"].includes(part.type))) return item; + const identities = Object.fromEntries(["author", "recipient"] + .filter(key => typeof message[key] === "string") + .map(key => [key, message[key]])); + changed = true; + return { + type: "message", role: "user", + content: [ + ...(Object.keys(identities).length ? [{ type: "input_text", text: `Agent message ${JSON.stringify(identities)}` }] : []), + ...message.content, + ], + }; + }); + return changed ? { ...record, input } : body; +} diff --git a/src/server/responses/agent-task-recovery-cache.ts b/src/server/responses/agent-task-recovery-cache.ts index 44e32c62bb..93d0c1778b 100644 --- a/src/server/responses/agent-task-recovery-cache.ts +++ b/src/server/responses/agent-task-recovery-cache.ts @@ -149,3 +149,14 @@ export function agentTaskRecoveryWaiterCountForTests(): number { export function agentTaskRecoveryCacheSnapshotForTests(): { entries: number; bytes: number } { return { entries: RECOVERY_CACHE.size, bytes: recoveryCacheBytes }; } + +/** Read an existing recovery without starting a request or extending its lifetime. */ +export function cachedAgentTaskRecovery(key: string): string | null { + const entry = RECOVERY_CACHE.get(key); + if (!entry) return null; + if (entry.expiresAt <= Date.now()) { + deleteRecoveryCacheEntry(key, entry); + return null; + } + return entry.assignment; +} diff --git a/src/server/responses/agent-task-recovery.ts b/src/server/responses/agent-task-recovery.ts index e1c35932ff..89af59a1ef 100644 --- a/src/server/responses/agent-task-recovery.ts +++ b/src/server/responses/agent-task-recovery.ts @@ -5,6 +5,7 @@ import { readBoundedResponseBody } from "../../lib/bounded-body"; import { isApiAuthRequired, isProxyAdmissionSecret } from "../auth-cors"; import { structurallyValidFernetTokens } from "./encrypted-payload"; import { + cachedAgentTaskRecovery, discardCachedAgentTaskRecovery, resetAgentTaskRecoveryCache, resolveCachedAgentTaskRecovery, @@ -61,7 +62,7 @@ interface AgentEnvelope { itemIndex: number; encryptedIndex: number; headerText: string; - messageType: "NEW_TASK"; + messageType: "NEW_TASK" | "MESSAGE"; taskName: string; sender: string; ciphertext: string; @@ -69,7 +70,7 @@ interface AgentEnvelope { 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|$)/; +const ROUTING_HEADER = /(?:^|\n)Message Type\s*:\s*(NEW_TASK|MESSAGE)\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; @@ -90,7 +91,7 @@ function findEnvelope(input: unknown): AgentEnvelope | null { if (!Array.isArray(content)) return null; let headerText: string | null = null; - let messageType: "NEW_TASK" | null = null; + let messageType: "NEW_TASK" | "MESSAGE" | null = null; let taskName: string | null = null; let sender: string | null = null; let encryptedIndex = -1; @@ -113,7 +114,7 @@ function findEnvelope(input: unknown): AgentEnvelope | null { || 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"; + messageType = match[1] as "NEW_TASK" | "MESSAGE"; taskName = match[2]!; sender = match[3]!; } @@ -496,3 +497,22 @@ export function discardEncryptedAgentTaskRecovery( export function resetAgentTaskRecoveryState(): void { resetAgentTaskRecoveryCache(); } + +/** Codex replays the original encrypted agent messages after tool calls. Reuse only an admitted cache hit. */ +export function restoreCachedEncryptedAgentTasks( + req: Request, input: unknown, config: OcxConfig, + context: { parentThreadId?: string | null } = {}, +): number { + if (!Array.isArray(input)) return 0; + let restored = 0; + for (const item of input) { + if (!item || typeof item !== "object" || item.type !== "agent_message") continue; + const single = [item]; + // Revalidates caller credentials and the exact supported agent envelope before cache access. + const admitted = admittedRecovery(req, single, config, context.parentThreadId); + if (!admitted) continue; + const assignment = cachedAgentTaskRecovery(admitted.cacheKey); + if (assignment && injectAssignment(single, admitted.envelope, assignment)) restored += 1; + } + return restored; +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 9e087bcc90..aa813e9c51 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -318,6 +318,7 @@ import { agentTaskRecoveryConfig, discardEncryptedAgentTaskRecovery, recoverEncryptedAgentTask, + restoreCachedEncryptedAgentTasks, } from "./agent-task-recovery"; import { relaySseEagerBounded } from "../relay-eager"; import { @@ -3234,14 +3235,18 @@ async function handleResponsesInner( inboundWire === "responses" && threadSpawn - && unreadableEncryptedAgentTask && agentTaskRecovery && !isCanonicalOpenAiForwardProvider(route.provider) && !options.comboAttempt && !canPassThroughEncryptedV2AgentTask(route, inboundWire) ) { - let recovered = false; - try { + let recovered = restoreCachedEncryptedAgentTasks( + req, (body as { input?: unknown } | undefined)?.input, config, { parentThreadId }, + ) > 0; + unreadableEncryptedAgentTask = hasUnreadableEncryptedAgentTask( + (body as { input?: unknown } | undefined)?.input, + ); + if (unreadableEncryptedAgentTask) try { recovered = await recoverEncryptedAgentTask( req, (body as { input?: unknown } | undefined)?.input, diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 114c699eaf..0b79ffe21d 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -726,6 +726,7 @@ "openai-responses-passthrough.test.ts": "responses", "opencode-cli.test.ts": "providers", "opencode-free-provider.test.ts": "providers", + "opencode-go-agent-messages.test.ts": "providers", "opencode-go-deepseek.test.ts": "providers", "opencode-go-grok46-responses.test.ts": "providers", "opencode-go-luna-wire.test.ts": "providers", @@ -901,6 +902,7 @@ "selected-models.test.ts": "codex-integration", "self-launch-argv.test.ts": "lib", "server-403-permission-e2e.test.ts": "server", + "server-agent-task-recovery-replay.test.ts": "server", "server-auth.test.ts": "server", "server-background-lifecycle.test.ts": "server", "server-clickjacking-headers.test.ts": "server", diff --git a/tests/providers/opencode-go-agent-messages.test.ts b/tests/providers/opencode-go-agent-messages.test.ts new file mode 100644 index 0000000000..e12860d0e0 --- /dev/null +++ b/tests/providers/opencode-go-agent-messages.test.ts @@ -0,0 +1,56 @@ +import { expect, test } from "bun:test"; +import { createResponsesPassthroughAdapter } from "../../src/adapters/openai-responses"; +import { normalizeOpenCodeGoAgentMessages } from "../../src/adapters/opencode-go"; +import { parseRequest } from "../../src/responses/parser"; +import { createTranslatorBudget } from "../../src/lib/translator-budget"; +import type { OcxProviderConfig } from "../../src/types"; + +const base: OcxProviderConfig = { adapter: "openai-responses", baseUrl: "https://opencode.ai/zen/go/v1", authMode: "key", apiKey: "synthetic-key" }; +const body = () => ({ model: "muse-spark-1.3-contributor", input: [{ type: "agent_message", id: "amsg_test", author: "/root/reader", recipient: "/root/checker", content: [{ type: "input_text", text: "Exact assignment\nwith lines." }] }], stream: true }); + +test("Responses converts plaintext task and peer messages without mutating replay or losing routing identities", async () => { + const raw = body(); const original = structuredClone(raw); const budget = createTranslatorBudget(); + const request = await createResponsesPassthroughAdapter(base).buildRequest(parseRequest(raw), { headers: new Headers(), translatorBudget: budget }); + const sent = JSON.parse(request.body as string); + expect(sent.input[0].type).toBe("message"); + expect(sent.input[0].role).toBe("user"); + expect(sent.input[0].content[0].text).toContain('"author":"/root/reader"'); + expect(sent.input[0].content[0].text).toContain('"recipient":"/root/checker"'); + expect(sent.input[0].content[1]).toEqual(raw.input[0]!.content[0]); + expect(sent.input[0].id).toBeUndefined(); + expect(raw).toEqual(original); + budget.dispose(); +}); + +test("ciphertext and unknown content are never reclassified as plaintext", () => { + for (const part of [{ type: "encrypted_content", encrypted_content: "opaque" }, { type: "future_type", text: "opaque" }]) { + const raw = { input: [{ type: "agent_message", content: [part] }] }; + expect(normalizeOpenCodeGoAgentMessages(raw)).toBe(raw); + } +}); + +test("image parts stay intact beside the assignment", () => { + const image = { type: "input_image", image_url: "data:image/png;base64,AAAA", detail: "high" }; + const raw = { input: [{ type: "agent_message", content: [{ type: "input_text", text: "Inspect image" }, image] }] }; + const result = normalizeOpenCodeGoAgentMessages(raw) as typeof raw; + expect(result.input[0]!.content[1]).toBe(image); +}); + +test("native forward keeps agent_message and auth/session headers unchanged", async () => { + const budget = createTranslatorBudget(); + const provider = { ...base, baseUrl: "https://chatgpt.com/backend-api/codex", authMode: "forward" as const }; + const request = await createResponsesPassthroughAdapter(provider).buildRequest(parseRequest(body()), { headers: new Headers({ "session-id": "native-id", authorization: "Bearer native-test" }), translatorBudget: budget }); + expect(JSON.parse(request.body as string).input[0].type).toBe("agent_message"); + expect(new Headers(request.headers).get("x-opencode-session")).toBeNull(); + expect(new Headers(request.headers).get("session-id")).toBe("native-id"); + expect(new Headers(request.headers).get("authorization")).toBe("Bearer native-test"); + budget.dispose(); +}); + +test("other destinations do not get Go normalization or session identity", async () => { + const budget = createTranslatorBudget(); + const request = await createResponsesPassthroughAdapter({ ...base, baseUrl: "https://example.test/v1" }).buildRequest(parseRequest(body()), { headers: new Headers({ "session-id": "child-id" }), translatorBudget: budget }); + expect(JSON.parse(request.body as string).input[0].type).toBe("agent_message"); + expect(new Headers(request.headers).get("x-opencode-session")).toBeNull(); + budget.dispose(); +}); diff --git a/tests/server/server-agent-task-recovery-replay.test.ts b/tests/server/server-agent-task-recovery-replay.test.ts new file mode 100644 index 0000000000..cfb735af01 --- /dev/null +++ b/tests/server/server-agent-task-recovery-replay.test.ts @@ -0,0 +1,142 @@ +import { afterEach, expect, test } from "bun:test"; +import { recoverEncryptedAgentTask, resetAgentTaskRecoveryState, restoreCachedEncryptedAgentTasks } from "../../src/server/responses/agent-task-recovery"; +import { codexHeaders, encryptedInput, FERNET_TASK, SECOND_FERNET_TASK, originalFetch, recoverySse, routedConfig } from "../helpers/agent-task-recovery"; +afterEach(() => { globalThis.fetch = originalFetch; resetAgentTaskRecoveryState(); }); + +test("replay reuses admitted recovery after a tool result without another network call", async () => { + let calls = 0; + globalThis.fetch = (async () => { calls++; return new Response(recoverySse("Read nonce.txt exactly.")); }) as typeof fetch; + const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); + const config = routedConfig({ enabled: true }); + expect(await recoverEncryptedAgentTask(req, encryptedInput(), {}, config, { parentThreadId: "parent" })).toBe(true); + const replay = [...encryptedInput(), { type: "function_call_output", call_id: "tool", output: "result" }]; + expect(restoreCachedEncryptedAgentTasks(req, replay, config, { parentThreadId: "parent" })).toBe(1); + expect(JSON.stringify(replay)).toContain("Read nonce.txt exactly."); + expect(JSON.stringify(replay)).not.toContain(FERNET_TASK); + expect(calls).toBe(1); +}); + +test("replay does not recover unseen envelopes, other parents, or other callers", async () => { + let calls = 0; + globalThis.fetch = (async () => { calls++; return new Response(recoverySse("Private assignment.")); }) as typeof fetch; + const config = routedConfig({ enabled: true }); + const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); + expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config, { parentThreadId: "parent" })).toBe(0); + expect(calls).toBe(0); + expect(await recoverEncryptedAgentTask(req, encryptedInput(), {}, config, { parentThreadId: "parent" })).toBe(true); + for (const [request, parent] of [[req, "another-parent"], [new Request("http://localhost/v1/responses", { headers: codexHeaders("another-account") }), "parent"], [new Request("http://localhost/v1/responses"), "parent"]] as const) { + const input = encryptedInput(); + expect(restoreCachedEncryptedAgentTasks(request, input, config, { parentThreadId: parent })).toBe(0); + expect(JSON.stringify(input)).toContain(FERNET_TASK); + } + expect(calls).toBe(1); +}); + +test("Responses handler restores a cached task in a continued child turn", async () => { + const { post, providerResponse } = await import("../helpers/agent-task-recovery"); + let recoveries = 0; + const bodies: string[] = []; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + if (String(url).includes("chatgpt.com")) { + recoveries++; + return new Response(recoverySse("Read nonce.txt exactly.")); + } + bodies.push(String(init?.body)); + return providerResponse(); + }) as typeof fetch; + const config = routedConfig({ enabled: true }); + expect((await post(config, "xai/grok-4.5", encryptedInput(), codexHeaders())).status).toBe(200); + expect((await post(config, "xai/grok-4.5", [...encryptedInput(), { type: "message", role: "user", content: "Continue the original task." }], codexHeaders())).status).toBe(200); + expect(recoveries).toBe(1); + expect(bodies).toHaveLength(2); + expect(bodies[1]).toContain("Read nonce.txt exactly."); + expect(bodies[1]).not.toContain(FERNET_TASK); +}); + +function encryptedMessage(): unknown[] { + return JSON.parse(JSON.stringify(encryptedInput()).replace("Message Type: NEW_TASK", "Message Type: MESSAGE")); +} + +test("MESSAGE recovery reaches the provider and survives tool-result replay", async () => { + const { post, providerResponse } = await import("../helpers/agent-task-recovery"); + let recoveries = 0; + const bodies: string[] = []; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + if (String(url).includes("chatgpt.com")) { + expect(String(init?.body)).toContain("Message Type: MESSAGE"); + recoveries++; + return new Response(recoverySse("Stop waiting and report your result.")); + } + bodies.push(String(init?.body)); + return providerResponse(); + }) as typeof fetch; + const config = routedConfig({ enabled: true }); + expect((await post(config, "xai/grok-4.5", encryptedMessage(), codexHeaders())).status).toBe(200); + expect((await post(config, "xai/grok-4.5", [...encryptedMessage(), { + type: "message", role: "user", content: "Continue after the tool result.", + }], codexHeaders())).status).toBe(200); + expect(recoveries).toBe(1); + expect(bodies).toHaveLength(2); + for (const body of bodies) { + expect(body).toContain("Stop waiting and report your result."); + expect(body).not.toContain(FERNET_TASK); + } +}); + +test("MESSAGE cache remains isolated by message type, account, parent and sender", async () => { + let calls = 0; + globalThis.fetch = (async () => { calls++; return new Response(recoverySse("Private message.")); }) as typeof fetch; + const config = routedConfig({ enabled: true }); + const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); + expect(await recoverEncryptedAgentTask(req, encryptedMessage(), {}, config, { parentThreadId: "parent" })).toBe(true); + expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config, { parentThreadId: "parent" })).toBe(0); + for (const [request, parent] of [[req, "other-parent"], [new Request("http://localhost/v1/responses", { headers: codexHeaders("other-account") }), "parent"]] as const) { + expect(restoreCachedEncryptedAgentTasks(request, encryptedMessage(), config, { parentThreadId: parent })).toBe(0); + } + const malformed = JSON.parse(JSON.stringify(encryptedMessage())); + malformed[0].author = "/root/wrong-sender"; + expect(await recoverEncryptedAgentTask(req, malformed, {}, config)).toBe(false); + const unknown = JSON.parse(JSON.stringify(encryptedMessage()).replace("Message Type: MESSAGE", "Message Type: UNKNOWN")); + expect(await recoverEncryptedAgentTask(req, unknown, {}, config)).toBe(false); + expect(calls).toBe(1); +}); + + +test("mixed history restores cached NEW_TASK and MESSAGE separately before recovering only the new tail", async () => { + let calls = 0; + const payloads = ["Initial assignment.", "First message.", "Second message."]; + globalThis.fetch = (async () => new Response(recoverySse(payloads[calls++]!))) as typeof fetch; + const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); + const config = routedConfig({ enabled: true }); + const scope = { parentThreadId: "parent" }; + const nextMessage = () => JSON.parse(JSON.stringify(encryptedMessage()).replace(FERNET_TASK, SECOND_FERNET_TASK)); + + expect(await recoverEncryptedAgentTask(req, encryptedInput(), {}, config, scope)).toBe(true); + expect(await recoverEncryptedAgentTask(req, encryptedMessage(), {}, config, scope)).toBe(true); + const input = [...encryptedInput(), ...encryptedMessage(), ...nextMessage()]; + expect(restoreCachedEncryptedAgentTasks(req, input, config, scope)).toBe(2); + expect(calls).toBe(2); + expect(await recoverEncryptedAgentTask(req, input, {}, config, scope)).toBe(true); + expect(calls).toBe(3); + for (const payload of payloads) expect(JSON.stringify(input)).toContain(payload); + expect(JSON.stringify(input)).not.toContain(SECOND_FERNET_TASK); + + const replay = [...encryptedInput(), ...encryptedMessage(), ...nextMessage(), { + type: "function_call_output", call_id: "tool", output: "done", + }]; + expect(restoreCachedEncryptedAgentTasks(req, replay, config, scope)).toBe(3); + expect(calls).toBe(3); +}); + +test("fresh recovery only handles the current tail, leaving uncached history unchanged", async () => { + let calls = 0; + globalThis.fetch = (async () => { calls++; return new Response(recoverySse("Current message.")); }) as typeof fetch; + const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); + const config = routedConfig({ enabled: true }); + const historical = encryptedInput(); + const input = [...historical, ...encryptedMessage()]; + expect(await recoverEncryptedAgentTask(req, input, {}, config)).toBe(true); + expect(input[0]).toEqual(encryptedInput()[0]); + expect(JSON.stringify(input[1])).toContain("Current message."); + expect(calls).toBe(1); +}); From c279e3166af5421e7323a06ca1de0947d7dee7f2 Mon Sep 17 00:00:00 2001 From: voiys Date: Sat, 5 Sep 2026 04:09:31 +0200 Subject: [PATCH 5/7] docs: clarify Go Responses agent-message destination (cherry picked from commit 4464892336c75b8861ee4caeddfd97d6c4e0e6ab) --- docs-site/src/content/docs/reference/adapters.md | 6 ++++++ .../src/content/docs/reference/configuration/providers.md | 5 ++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index e2a24c67df..36b5b70b0a 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -105,6 +105,12 @@ collision-safe public function tool. Matching request history and JSON/SSE funct translated back to the private `tool_search` lifecycle for the client. Canonical OpenAI forward keeps the native private type unchanged. +For OpenCode Go at `https://opencode.ai/zen/go/v1`, plaintext Codex `agent_message` +items become public user messages, preserving content parts and readable author/recipient +metadata. This conversion leaves encrypted or unknown content unchanged and does not apply +to other destinations. See [Go agent messages](/reference/configuration/providers/#opencode-go-session-and-agent-messages) +for the separate opt-in encrypted-task recovery behavior. + The canonical ChatGPT Codex forward destination also normalizes two public Responses shapes that its stricter backend rejects: fully textual `system` messages inside `input` are appended to the top-level `instructions` string in request order, and the top-level `truncation` field is removed. diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 7282a21bd3..8e551f05c9 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -813,7 +813,10 @@ ids with context `922000` and max input `922000`; OpenRouter seeds `openai/gpt-5 ## OpenCode Go session and agent messages -On Go Responses routes, plaintext Codex `agent_message` items become user messages. +With the [`openai-responses` adapter](/reference/adapters/#openai-responses) and +base URL `https://opencode.ai/zen/go/v1`, plaintext Codex `agent_message` items +become user messages. This conversion is scoped to that destination, including +renamed provider entries; other Responses destinations keep their input unchanged. Author and recipient remain explicit text metadata, and the content parts are preserved. Encrypted and unknown content is not normalized; native encrypted tasks still require the separate opt-in [task recovery](/reference/configuration/agents/#encrypted-v2-task-recovery). From fc031d4e1b1ee92891489d8b9999cb5b18c359eb Mon Sep 17 00:00:00 2001 From: voiys Date: Sat, 5 Sep 2026 11:48:03 +0200 Subject: [PATCH 6/7] docs: describe Go agent-message forward-auth exception (cherry picked from commit 036a9321788464fdf33a387c9f44a834a844bdc1) --- docs-site/src/content/docs/reference/adapters.md | 7 ++++--- .../src/content/docs/reference/configuration/providers.md | 3 ++- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index 36b5b70b0a..2848fed564 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -105,10 +105,11 @@ collision-safe public function tool. Matching request history and JSON/SSE funct translated back to the private `tool_search` lifecycle for the client. Canonical OpenAI forward keeps the native private type unchanged. -For OpenCode Go at `https://opencode.ai/zen/go/v1`, plaintext Codex `agent_message` -items become public user messages, preserving content parts and readable author/recipient +For OpenCode Go at `https://opencode.ai/zen/go/v1`, requests with `authMode` other +than `"forward"` convert plaintext Codex `agent_message` items into public user messages, preserving content parts and readable author/recipient metadata. This conversion leaves encrypted or unknown content unchanged and does not apply -to other destinations. See [Go agent messages](/reference/configuration/providers/#opencode-go-session-and-agent-messages) +to other destinations. Providers using `authMode: "forward"` retain these items unchanged. +See [Go agent messages](/reference/configuration/providers/#opencode-go-session-and-agent-messages) for the separate opt-in encrypted-task recovery behavior. The canonical ChatGPT Codex forward destination also normalizes two public Responses shapes that diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 8e551f05c9..51e71fa1f4 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -815,7 +815,8 @@ ids with context `922000` and max input `922000`; OpenRouter seeds `openai/gpt-5 With the [`openai-responses` adapter](/reference/adapters/#openai-responses) and base URL `https://opencode.ai/zen/go/v1`, plaintext Codex `agent_message` items -become user messages. This conversion is scoped to that destination, including +become user messages when `authMode` is not `"forward"` (for example, `"key"`). +Providers using `authMode: "forward"` retain these items unchanged. This conversion is scoped to that destination, including renamed provider entries; other Responses destinations keep their input unchanged. Author and recipient remain explicit text metadata, and the content parts are preserved. Encrypted and unknown content is not normalized; native encrypted tasks still require the From 332a30e6d3a6cb3941acbf68df0656df3778c731 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 6 Sep 2026 01:52:17 +0900 Subject: [PATCH 7/7] fix(responses): preserve termination scope across recovery replay --- src/server/responses/core.ts | 1 + .../opencode-go-agent-messages.test.ts | 126 ++++++++++++++++- .../server/agent-task-recovery-cache.test.ts | 58 ++++++++ .../server-agent-task-recovery-replay.test.ts | 130 +++++++++++++++++- 4 files changed, 313 insertions(+), 2 deletions(-) diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index aa813e9c51..69e10a4a93 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -3280,6 +3280,7 @@ async function handleResponsesInner( (reparsed as unknown as Record)[key] = parsed[key]; } } + bindTurnTerminationScope(reparsed, resolvedConversationId); 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 — diff --git a/tests/providers/opencode-go-agent-messages.test.ts b/tests/providers/opencode-go-agent-messages.test.ts index e12860d0e0..f79f529a5e 100644 --- a/tests/providers/opencode-go-agent-messages.test.ts +++ b/tests/providers/opencode-go-agent-messages.test.ts @@ -1,7 +1,8 @@ import { expect, test } from "bun:test"; import { createResponsesPassthroughAdapter } from "../../src/adapters/openai-responses"; -import { normalizeOpenCodeGoAgentMessages } from "../../src/adapters/opencode-go"; +import { isOpenCodeGo, normalizeOpenCodeGoAgentMessages } from "../../src/adapters/opencode-go"; import { parseRequest } from "../../src/responses/parser"; +import { routeModel } from "../../src/router"; import { createTranslatorBudget } from "../../src/lib/translator-budget"; import type { OcxProviderConfig } from "../../src/types"; @@ -54,3 +55,126 @@ test("other destinations do not get Go normalization or session identity", async expect(new Headers(request.headers).get("x-opencode-session")).toBeNull(); budget.dispose(); }); + +test("canonical Go forward auth preserves private agent messages and the raw replay body", async () => { + const raw = body(); + const original = structuredClone(raw); + const parsed = parseRequest(raw); + const budget = createTranslatorBudget(); + try { + const request = await createResponsesPassthroughAdapter({ ...base, authMode: "forward" }).buildRequest(parsed, { + headers: new Headers(), translatorBudget: budget, + }); + expect(request.url).toBe("https://opencode.ai/zen/go/v1/responses"); + expect(JSON.parse(request.body as string).input[0]).toMatchObject({ + type: "agent_message", author: "/root/reader", recipient: "/root/checker", + content: original.input[0]!.content, + }); + expect(parsed._rawBody).toBe(raw); + expect(raw).toEqual(original); + } finally { + budget.dispose(); + } +}); + +test.each(["https://opencode.ai/zen/go/v1", "https://opencode.ai/zen/go/v1/"])( + "a renamed provider at %s still converts plaintext agent messages", + async baseUrl => { + const raw = body(); + const original = structuredClone(raw); + const route = routeModel({ + port: 0, defaultProvider: "my-go", providers: { "my-go": { ...base, baseUrl, models: [raw.model] } }, + }, `my-go/${raw.model}`); + const parsed = parseRequest(raw); + const budget = createTranslatorBudget(); + try { + const request = await createResponsesPassthroughAdapter(route.provider).buildRequest(parsed, { + headers: new Headers(), translatorBudget: budget, + }); + const sent = JSON.parse(request.body as string); + expect(request.url).toBe("https://opencode.ai/zen/go/v1/responses"); + expect(sent.input[0]).toMatchObject({ type: "message", role: "user" }); + expect(sent.input[0].content.slice(1)).toEqual(original.input[0]!.content); + expect(parsed._rawBody).toBe(raw); + expect(raw).toEqual(original); + } finally { + budget.dispose(); + } + }, +); + +test.each([ + "https://opencode.ai.evil.test/zen/go/v1", + "http://opencode.ai/zen/go/v1", + "https://opencode.ai/zen/v1", + "https://opencode.ai/zen/go/v10", +])("Go-like destination %s preserves private agent messages", async baseUrl => { + const raw = body(); + const original = structuredClone(raw); + const parsed = parseRequest(raw); + const budget = createTranslatorBudget(); + try { + const request = await createResponsesPassthroughAdapter({ ...base, baseUrl }).buildRequest(parsed, { + headers: new Headers(), translatorBudget: budget, + }); + expect(JSON.parse(request.body as string).input[0]).toMatchObject({ + type: "agent_message", content: original.input[0]!.content, + }); + expect(parsed._rawBody).toBe(raw); + expect(raw).toEqual(original); + } finally { + budget.dispose(); + } +}); + +test.each(["not a URL", "https://", "/zen/go/v1"])( + "malformed destination %s is not classified as Go", + baseUrl => expect(isOpenCodeGo(baseUrl)).toBe(false), +); + +test("Go conversion preserves file payloads beside text without mutating raw replay", async () => { + const file = { type: "input_file", filename: "assignment.txt", file_data: "data:text/plain;base64,SGVsbG8=" }; + const message = body().input[0]!; + const raw = { ...body(), input: [{ ...message, content: [...message.content, file] }] }; + const original = structuredClone(raw); + const parsed = parseRequest(raw); + const budget = createTranslatorBudget(); + try { + const request = await createResponsesPassthroughAdapter(base).buildRequest(parsed, { + headers: new Headers(), translatorBudget: budget, + }); + const sent = JSON.parse(request.body as string); + expect(sent.input[0]).toMatchObject({ type: "message", role: "user" }); + expect(sent.input[0].content.slice(1)).toEqual(original.input[0]!.content); + expect(parsed._rawBody).toBe(raw); + expect(raw).toEqual(original); + } finally { + budget.dispose(); + } +}); + +for (const { name, content } of [ + { name: "empty content", content: [] }, + { name: "text mixed with an unknown part", content: [ + { type: "input_text", text: "Known prefix" }, { type: "future_type", text: "Do not lose this" }, + ] }, + { name: "text mixed with ciphertext", content: [ + { type: "input_text", text: "Routing header" }, { type: "encrypted_content", encrypted_content: "opaque" }, + ] }, +]) test(`Go preserves ${name} without partially converting it`, async () => { + const raw = { ...body(), input: [{ ...body().input[0]!, content }] }; + const original = structuredClone(raw); + expect(normalizeOpenCodeGoAgentMessages(raw)).toBe(raw); + const parsed = parseRequest(raw); + const budget = createTranslatorBudget(); + try { + const request = await createResponsesPassthroughAdapter(base).buildRequest(parsed, { + headers: new Headers(), translatorBudget: budget, + }); + expect(JSON.parse(request.body as string).input[0]).toMatchObject({ type: "agent_message", content }); + expect(parsed._rawBody).toBe(raw); + expect(raw).toEqual(original); + } finally { + budget.dispose(); + } +}); diff --git a/tests/server/agent-task-recovery-cache.test.ts b/tests/server/agent-task-recovery-cache.test.ts index 3107bff1a1..3a7324340d 100644 --- a/tests/server/agent-task-recovery-cache.test.ts +++ b/tests/server/agent-task-recovery-cache.test.ts @@ -1,5 +1,8 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { + agentTaskRecoveryCacheSnapshotForTests, + agentTaskRecoveryWaiterCountForTests, + cachedAgentTaskRecovery, resetAgentTaskRecoveryCache, resolveCachedAgentTaskRecovery, } from "../../src/server/responses/agent-task-recovery-cache"; @@ -14,6 +17,61 @@ describe("agent task recovery cache", () => { resetAgentTaskRecoveryCache(); }); + test("read-only hits retain the original expiry and exact-expiry reads release UTF-8 bytes", async () => { + const insertedAt = 1_800_000_000_000; + let now = insertedAt; + Date.now = () => now; + let requests = 0; + expect(cachedAgentTaskRecovery("missing")).toBeNull(); + expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 0, bytes: 0 }); + expect(await resolveCachedAgentTaskRecovery("task", 200, async () => { + requests++; + return "한😀"; // Three UTF-8 bytes plus four, rather than three UTF-16 code units. + })).toBe("한😀"); + + for (const elapsed of [0, 60_000, 15 * 60 * 1000 - 1]) { + now = insertedAt + elapsed; + expect(cachedAgentTaskRecovery("task")).toBe("한😀"); + expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 1, bytes: 7 }); + } + now = insertedAt + 15 * 60 * 1000; + expect(cachedAgentTaskRecovery("task")).toBeNull(); + expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 0, bytes: 0 }); + expect(cachedAgentTaskRecovery("task")).toBeNull(); + expect(requests).toBe(1); + + // Repeated expiry reads must not subtract bytes belonging to a later entry. + await resolveCachedAgentTaskRecovery("later", 200, async () => "ok"); + expect(cachedAgentTaskRecovery("task")).toBeNull(); + expect(cachedAgentTaskRecovery("later")).toBe("ok"); + expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 1, bytes: 2 }); + }); + + test("read-only misses do not join or restart an in-flight recovery", async () => { + let release: (() => void) | undefined; + const gate = new Promise(resolve => { release = resolve; }); + let requests = 0; + const pending = resolveCachedAgentTaskRecovery("pending", 200, async () => { + requests++; + await gate; + return "recovered"; + }); + try { + expect(cachedAgentTaskRecovery("pending")).toBeNull(); + expect(cachedAgentTaskRecovery("unknown")).toBeNull(); + expect(cachedAgentTaskRecovery("pending")).toBeNull(); + expect(agentTaskRecoveryWaiterCountForTests()).toBe(1); + expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 0, bytes: 0 }); + expect(requests).toBe(1); + } finally { + release?.(); + await pending; + } + expect(cachedAgentTaskRecovery("pending")).toBe("recovered"); + expect(agentTaskRecoveryWaiterCountForTests()).toBe(0); + expect(requests).toBe(1); + }); + test("expires recovered plaintext after fifteen minutes", async () => { let now = 1_800_000_000_000; Date.now = () => now; diff --git a/tests/server/server-agent-task-recovery-replay.test.ts b/tests/server/server-agent-task-recovery-replay.test.ts index cfb735af01..fd572e9fd8 100644 --- a/tests/server/server-agent-task-recovery-replay.test.ts +++ b/tests/server/server-agent-task-recovery-replay.test.ts @@ -1,4 +1,10 @@ -import { afterEach, expect, test } from "bun:test"; +import { afterEach, expect, spyOn, test } from "bun:test"; +import { createKiroAdapter } from "../../src/adapters/kiro"; +import { ADAPTER_REGISTRY } from "../../src/adapters/registry"; +import { parseRequest } from "../../src/responses/parser"; +import { bindTurnTerminationScope, rememberDeliveredFinalAnswer } from "../../src/responses/turn-termination"; +import { conversationIdFromResponsesRequest } from "../../src/server/request-log-conversation"; +import type { OcxParsedRequest } from "../../src/types"; import { recoverEncryptedAgentTask, resetAgentTaskRecoveryState, restoreCachedEncryptedAgentTasks } from "../../src/server/responses/agent-task-recovery"; import { codexHeaders, encryptedInput, FERNET_TASK, SECOND_FERNET_TASK, originalFetch, recoverySse, routedConfig } from "../helpers/agent-task-recovery"; afterEach(() => { globalThis.fetch = originalFetch; resetAgentTaskRecoveryState(); }); @@ -128,6 +134,128 @@ test("mixed history restores cached NEW_TASK and MESSAGE separately before recov expect(calls).toBe(3); }); +test("Responses handler restores known history and recovers only the new MESSAGE tail", async () => { + const { post, providerResponse } = await import("../helpers/agent-task-recovery"); + const assignments = ["Initial assignment.", "First message.", "Second message."]; + const recoveryBodies: string[] = []; + const providerBodies: string[] = []; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + const requestBody = String(init?.body); + if (String(url).includes("chatgpt.com")) { + recoveryBodies.push(requestBody); + return new Response(recoverySse(assignments[recoveryBodies.length - 1] ?? "Unexpected extra recovery.")); + } + providerBodies.push(requestBody); + return providerResponse(); + }) as typeof fetch; + const config = routedConfig({ enabled: true }); + const headers = codexHeaders(); + const nextMessage = () => JSON.parse(JSON.stringify(encryptedMessage()).replace(FERNET_TASK, SECOND_FERNET_TASK)); + const turns = [ + encryptedInput(), + [...encryptedInput(), ...encryptedMessage()], + [...encryptedInput(), ...encryptedMessage(), ...nextMessage()], + ]; + + for (const [index, input] of turns.entries()) { + const response = await post(config, "xai/grok-4.5", input, headers); + expect(response.status).toBe(200); + await response.text(); + expect(recoveryBodies).toHaveLength(index + 1); + expect(providerBodies).toHaveLength(index + 1); + const sent = providerBodies[index]!; + let previousPosition = -1; + for (const assignment of assignments.slice(0, index + 1)) { + const position = sent.indexOf(assignment); + expect(position).toBeGreaterThan(previousPosition); + previousPosition = position; + } + expect(sent).not.toContain(FERNET_TASK); + expect(sent).not.toContain(SECOND_FERNET_TASK); + } + // Recovery may receive only the fresh tail, never a batch of cached history. + expect(JSON.parse(recoveryBodies[2]!).input).toEqual(nextMessage()); + + const response = await post(config, "xai/grok-4.5", [ + ...encryptedInput(), ...encryptedMessage(), ...nextMessage(), + { type: "message", role: "user", content: "Continue with all three instructions." }, + ], headers); + expect(response.status).toBe(200); + await response.text(); + expect(recoveryBodies).toHaveLength(3); + expect(providerBodies).toHaveLength(4); + for (const assignment of assignments) expect(providerBodies[3]).toContain(assignment); + expect(providerBodies[3]).toContain("Continue with all three instructions."); + expect(providerBodies[3]).not.toContain(FERNET_TASK); + expect(providerBodies[3]).not.toContain(SECOND_FERNET_TASK); +}); + +test("cached-history reparse preserves recorded final-answer scope without suppressing a user follow-up", async () => { + const { post, providerResponse } = await import("../helpers/agent-task-recovery"); + const sessionId = `recovery-final-replay-${crypto.randomUUID()}`; + const headers = codexHeaders("acct-caller", { session_id: sessionId }); + const config = routedConfig({ enabled: true }); + const deliveredAnswer = "The assignment is complete."; + const recorded = parseRequest({ model: "xai/grok-4.5", input: "Earlier turn" }); + bindTurnTerminationScope(recorded, conversationIdFromResponsesRequest({ sessionIdHeader: sessionId })); + rememberDeliveredFinalAnswer(recorded, { output: [{ + type: "message", role: "assistant", phase: "final_answer", + content: [{ type: "output_text", text: deliveredAnswer }], + }] }); + + let recoveries = 0; + const providerBodies: string[] = []; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + if (String(url).includes("chatgpt.com")) { + recoveries++; + return new Response(recoverySse("Read the assignment.")); + } + providerBodies.push(String(init?.body)); + return providerResponse(); + }) as typeof fetch; + const req = new Request("http://localhost/v1/responses", { headers }); + expect(await recoverEncryptedAgentTask(req, encryptedInput(), {}, config)).toBe(true); + + // Keep the ordinary transport fixture, but exercise Kiro's real pre-send termination hook. + // The remembered record above belongs to a different parsed object: only core can bind + // the new object produced by recovery reparse to the same conversation. + const kiro = createKiroAdapter({ adapter: "kiro", baseUrl: "https://kiro.test", authMode: "key", apiKey: "synthetic-key" }); + const createChat = ADAPTER_REGISTRY["openai-chat"].create; + const inspectedBodies: string[] = []; + const factory = spyOn(ADAPTER_REGISTRY["openai-chat"], "create").mockImplementation((provider, context) => ({ + ...createChat(provider, context), + localTerminal(parsed: OcxParsedRequest) { + inspectedBodies.push(JSON.stringify(parsed._rawBody)); + return kiro.localTerminal?.(parsed); + }, + })); + const finalMessage = { type: "message", role: "assistant", content: [{ type: "output_text", text: deliveredAnswer }] }; + try { + for (let attempt = 0; attempt < 2; attempt++) { + const response = await post(config, "xai/grok-4.5", [...encryptedInput(), finalMessage], headers); + expect(response.status).toBe(200); + expect((await response.json() as { output: unknown[] }).output).toEqual([]); + expect(providerBodies).toHaveLength(0); + } + const followUp = await post(config, "xai/grok-4.5", [ + ...encryptedInput(), finalMessage, + { type: "message", role: "user", content: "Now explain your result." }, + ], headers); + expect(followUp.status).toBe(200); + await followUp.text(); + expect(providerBodies).toHaveLength(1); + expect(providerBodies[0]).toContain("Now explain your result."); + expect(inspectedBodies).toHaveLength(3); + for (const inspected of inspectedBodies) { + expect(inspected).toContain("Read the assignment."); + expect(inspected).not.toContain(FERNET_TASK); + } + expect(recoveries).toBe(1); + } finally { + factory.mockRestore(); + } +}); + test("fresh recovery only handles the current tail, leaving uncached history unchanged", async () => { let calls = 0; globalThis.fetch = (async () => { calls++; return new Response(recoverySse("Current message.")); }) as typeof fetch;