From e5adb72c92d19a05d02ed4dcab467a01f3ac03ae Mon Sep 17 00:00:00 2001 From: moseoridev Date: Sat, 26 Sep 2026 18:42:11 +0900 Subject: [PATCH 1/5] fix(adapters): remint duplicate tool call ids on the openai-chat lane (#5914) Carried from #5914 as one squashed commit. Co-authored-by: moseoridev --- .../000_overview.md | 9 + .../260926_unique_tool_call_ids/010_remint.md | 50 ++++ scripts/test-layout/layout.json | 1 + .../openai-chat/tool-call-id-remint.ts | 65 ++++++ src/adapters/registry.ts | 3 +- src/adapters/unique-tool-call-ids.ts | 63 +++++ structure/providers-and-adapters.md | 3 +- .../openai-chat-tool-call-id-remint.test.ts | 216 ++++++++++++++++++ tests/fixtures/test-layout-expected.json | 1 + 9 files changed, 409 insertions(+), 2 deletions(-) create mode 100644 devlog/_plan/260926_unique_tool_call_ids/000_overview.md create mode 100644 devlog/_plan/260926_unique_tool_call_ids/010_remint.md create mode 100644 src/adapters/openai-chat/tool-call-id-remint.ts create mode 100644 src/adapters/unique-tool-call-ids.ts create mode 100644 tests/adapters/openai/openai-chat-tool-call-id-remint.test.ts diff --git a/devlog/_plan/260926_unique_tool_call_ids/000_overview.md b/devlog/_plan/260926_unique_tool_call_ids/000_overview.md new file mode 100644 index 00000000000..5f45d247e38 --- /dev/null +++ b/devlog/_plan/260926_unique_tool_call_ids/000_overview.md @@ -0,0 +1,9 @@ +# Unique tool-call ids for positionally-minting chat upstreams + +Unit opened 2026-09-26. + +- [010_remint.md](010_remint.md) — the `openai-chat` lane. **DONE.** + +Origin: a DeepSeek V4.1 Flash conversation through this proxy looped with a thinking-only turn that +never terminated, while the same client against the same gateway on a GLM model was unaffected. +The two models differ in the id they mint (positional vs random), which is the whole defect. diff --git a/devlog/_plan/260926_unique_tool_call_ids/010_remint.md b/devlog/_plan/260926_unique_tool_call_ids/010_remint.md new file mode 100644 index 00000000000..9956a5c4381 --- /dev/null +++ b/devlog/_plan/260926_unique_tool_call_ids/010_remint.md @@ -0,0 +1,50 @@ +# Unique tool-call ids for positionally-minting chat upstreams + +Defect: `src/adapters/registry.ts` `openai-chat` entry. The adapter forwards +`tool_calls[].id` upstream-verbatim, which is correct for an upstream that mints a fresh random id +per call (measured: `zai-org/glm-5.3-flash` mints `call_<24 hex>`, never repeating). An upstream +that derives the id from the call's **position in its response** instead mints the same `call-0-0` +on every turn of a conversation (measured: `deepseek-ai/deepseek-v4.1-flash` behind the same +gateway — three sequential turns, `call-0-0` every time; non-streaming mints `call-`). +A Messages client has already paired that id with an earlier call, drops the duplicate, and is left +with a tool call carrying no result: the turn folds to an assistant message with no content, the +model re-issues the same call, and the conversation loops without ever erroring. The client's own +debug log confirms it sees no duplicate — `tool_uses=[call-0-0]` / `tool_results=[call-0-0]` with +zero `api_retry` — so the loop is not a retry storm but a silently-dropped pairing. + +Reproduced with real Claude Code against a mock upstream that always mints `call-0-0`: +**unpatched, 2186 stream events, 1 unique id, exit 124 (loop); patched, 3 unique ids +(`call-0-0`, `call-0-0-2`, `call-0-0-3`), `subtype: success`, exit 0.** + +Change: + +- New leaf `src/adapters/openai-chat/tool-call-id-remint.ts`: `createToolCallIdReminter(reserved)` and + `reservedToolCallIdsFromHistory(messages)`. First occurrence of an id is emitted byte-identical — + prompt-cache keys, reasoning-replay lookups and already-unique upstreams are untouched; only a + **repeat** is rewritten, to the smallest unused suffix that fits the 64-char Anthropic id bound. + The suffix is `-`, never `_`: an id extending another as `_` is read by the + client as batch sub-call N of ``, which pairs the second call's result to the first. That + shape was measured separately: with an `_` remint the same harness accumulated 10 placeholder + results; `-` produced none. +- New wrapper `src/adapters/unique-tool-call-ids.ts`: remints `tool_call_start` on both the + streaming and buffered paths, seeded from `parsed.context.messages` in `buildRequest` — the only + point that sees the caller's history, which is the authority on which ids are taken because it is + the side that discards duplicates. Emission-only: ingest-time rewriting would strip a pending + streamed call of the identity its own delta fragments match against. +- `src/adapters/registry.ts`: the `openai-chat` factory now wraps in `withUniqueToolCallIds`, the + same shape as the existing `withClinePassDeepSeekV4ToolReplayCompatibility` wrapper. +- `src/adapters/openai-chat.ts` is **untouched**: it sits at its 822-line ratchet cap with zero + headroom, and the remedy is a sibling file, not a raised number (`AGENTS.md:255-271`). + +Tests (new `tests/adapters/openai/openai-chat-tool-call-id-remint.test.ts`, registered in both +`scripts/test-layout/layout.json` and `tests/fixtures/test-layout-expected.json`): a positional +upstream driven through three real turns emits three distinct ids; a first turn with no history +emits the upstream id byte-identical; the suffix shape is asserted directly; repeated occurrences +within one response stay distinct; a reserved suffix is skipped; every rewrite stays conforming and +within the length bound; a non-conforming id is sanitized rather than dropped; the history scan +reads both the assistant call and the tool result. Verified to fail against a pass-through wrapper +(`["call-0-0","call-0-0","call-0-0"]`) and pass with the fix. + +Docs: `structure/providers-and-adapters.md` gains the `src/adapters/unique-tool-call-ids.ts` row and +the `tool-call-id-remint.ts` leaf. `structure/providers/chat-compat.md` would have been the natural +home but sits at exactly its 600-line budget. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 26aad251318..c3997a747e8 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -171,6 +171,7 @@ "deepseek-artifact-tool-schema.test.ts": "providers", "client-config-export-output-limit.test.ts": "config", "openai-chat-serialized-tool-call-scaling.test.ts": "adapters/openai", + "openai-chat-tool-call-id-remint.test.ts": "adapters/openai", "coding-agent-json-lines-scaling.test.ts": "providers", "usage-snapshot-digest-reuse.test.ts": "usage", "release-desktop-scripts.test.ts": "ci-workflows", diff --git a/src/adapters/openai-chat/tool-call-id-remint.ts b/src/adapters/openai-chat/tool-call-id-remint.ts new file mode 100644 index 00000000000..bddfb4cb6a2 --- /dev/null +++ b/src/adapters/openai-chat/tool-call-id-remint.ts @@ -0,0 +1,65 @@ +import { isConformingToolCallId, MAX_TOOL_CALL_ID_LENGTH } from "../tool-call-id"; +import type { OcxMessage } from "../../types"; + +/** + * Remint a tool-call id the client's history already carries. + * + * Some upstreams mint tool-call ids deterministically per RESPONSE rather than per call: the same + * `call-0-0` on every turn of a conversation. Anthropic's wire requires `tool_use.id` to identify + * one call, and a client that has already stored that id for an earlier call cannot pair the new + * result to the new call — it drops one of them, the turn becomes an assistant message whose + * tool call has no result, and the model re-issues the same call forever. + * + * The first occurrence of an id is emitted byte-identical, so prompt-cache keys, reasoning-replay + * lookups, and every upstream that already mints unique ids stay untouched. Only a repeat — against + * the history the caller seeded, or against a call already emitted in this response — is rewritten, + * to the smallest unused suffix that still fits Anthropic's id bound. + */ +export function createToolCallIdReminter(reservedIds: Iterable): (rawId: string) => string { + const occupied = new Set(reservedIds); + return rawId => { + if (!occupied.has(rawId)) { + occupied.add(rawId); + return rawId; + } + // A non-conforming source is sanitized, never dropped: the wire still needs an id, and the + // occupied check below covers a sanitized form that now equals some other call's id. + const base = isConformingToolCallId(rawId) ? rawId : rawId.replace(/[^a-zA-Z0-9_-]/g, "_"); + for (let n = 2; ; n++) { + // Hyphen, not underscore: an id that extends another id as `_` is parsed by + // at least one client as a batch sub-call of ``, which pairs the second call's + // result to the first call. A `-` suffix is in the same id family without that reading. + const suffix = `-${n}`; + const candidate = base.slice(0, Math.max(1, MAX_TOOL_CALL_ID_LENGTH - suffix.length)) + suffix; + if (!occupied.has(candidate)) { + occupied.add(candidate); + return candidate; + } + } + }; +} + +/** + * Tool-call ids the client's own history has already fixed: every assistant tool call it kept, plus + * every tool result that answered one. A response repeating any of them is the collision this + * module exists for. + * + * Read from the client's history rather than from earlier responses because the client is the + * authority on uniqueness here: it is the side that drops duplicates, so the proxy cannot observe + * the ids it discarded — a dropped id only ever exists as the absence it caused. A Messages client + * sends the whole conversation on every turn, which is why one turn's history is a complete picture + * of the ids that may not be reused. + */ +export function reservedToolCallIdsFromHistory(messages: readonly OcxMessage[]): Set { + const ids = new Set(); + for (const message of messages) { + if (message.role === "assistant") { + for (const part of message.content) { + if (part.type === "toolCall") ids.add(part.id); + } + continue; + } + if (message.role === "toolResult") ids.add(message.toolCallId); + } + return ids; +} diff --git a/src/adapters/registry.ts b/src/adapters/registry.ts index a99a9fe2071..c8c43748402 100644 --- a/src/adapters/registry.ts +++ b/src/adapters/registry.ts @@ -3,6 +3,7 @@ import { createAzureAdapter } from "./azure"; import type { ProviderAdapter } from "./base"; import { createClaudeCliAdapter } from "./claude-cli/adapter"; import { withClinePassDeepSeekV4ToolReplayCompatibility } from "./cline-pass-deepseek-v4-tool-replay"; +import { withUniqueToolCallIds } from "./unique-tool-call-ids"; import { createCodeBuddyAdapter } from "./codebuddy/adapter"; import { createQoderAdapter } from "./qoder/adapter"; import { createCommandCodeAdapter } from "./command-code"; @@ -84,7 +85,7 @@ export const ADAPTER_REGISTRY = { wire: "openai-chat", mutation: "codex-owned", create: (provider: OcxProviderConfig, _context: AdapterFactoryContext) => - withClinePassDeepSeekV4ToolReplayCompatibility(createOpenAIChatAdapter(provider)), + withUniqueToolCallIds(withClinePassDeepSeekV4ToolReplayCompatibility(createOpenAIChatAdapter(provider))), }, "ollama-native": { wire: "ollama-native", diff --git a/src/adapters/unique-tool-call-ids.ts b/src/adapters/unique-tool-call-ids.ts new file mode 100644 index 00000000000..28540bd6249 --- /dev/null +++ b/src/adapters/unique-tool-call-ids.ts @@ -0,0 +1,63 @@ +import type { ProviderAdapter } from "./base"; +import type { AdapterEvent, OcxMessage } from "../types"; +import { createToolCallIdReminter, reservedToolCallIdsFromHistory } from "./openai-chat/tool-call-id-remint"; + +/** + * Give every tool call in a conversation a wire id the client has not already stored. + * + * The openai-chat adapter forwards `tool_calls[].id` upstream-verbatim, which is correct for an + * upstream that mints a fresh random id per call. An upstream that derives the id from the call's + * position in its response mints the same `call-0-0` on every turn. The client has already paired + * that id with an earlier call, so it drops the duplicate; the dropped call has no result, the turn + * reads as an assistant message with no content, and the model re-issues the same call forever. + * + * Applied as a wrapper rather than inside the adapter so the adapter's own id handling stays + * untouched, matching how this repository already scopes a wire-compatibility policy + * (`withClinePassDeepSeekV4ToolReplayCompatibility`). + * + * Reminting happens at emission — on the events the adapter yields — never at ingestion: ingestion + * matches streamed deltas and continuation fragments against the id the upstream sent, so rewriting + * there would strip a pending call of its own identity mid-stream. The ids to avoid come from the + * caller's own history, captured where the request is built because that is the only point that sees + * the inbound conversation; the client is the authority on which ids are taken, since it is the side + * that discards the duplicates the proxy would otherwise never observe. + * + * The first occurrence of an id is emitted byte-identical, so prompt-cache keys, reasoning-replay + * lookups, and upstreams that already mint unique ids are unaffected. + */ +export function withUniqueToolCallIds(adapter: ProviderAdapter): ProviderAdapter { + // Set where the request is built and consumed by the two emission paths below — the same + // build-into-parse handoff the openai-chat adapter already uses for its requested model id. The + // identity default means a parse that runs without a build cannot fail: it has no history to + // collide with. + let remintToolCallId: (rawId: string) => string = id => id; + + const remintEvents = (events: AdapterEvent[]): AdapterEvent[] => + events.map(event => event.type === "tool_call_start" ? { ...event, id: remintToolCallId(event.id) } : event); + + return { + ...adapter, + + async buildRequest(parsed, incoming) { + const history: OcxMessage[] | undefined = parsed.context?.messages; + remintToolCallId = createToolCallIdReminter( + Array.isArray(history) ? reservedToolCallIdsFromHistory(history) : [], + ); + return adapter.buildRequest(parsed, incoming); + }, + + async *parseStream(response, budget, tierMetadata): AsyncGenerator { + for await (const event of adapter.parseStream(response, budget, tierMetadata)) { + yield event.type === "tool_call_start" ? { ...event, id: remintToolCallId(event.id) } : event; + } + }, + + ...(adapter.parseResponse + ? { + async parseResponse(response, budget, tierMetadata) { + return remintEvents(await adapter.parseResponse!(response, budget, tierMetadata)); + }, + } + : {}), + }; +} diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index d09433afca1..9871c973ec4 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -79,9 +79,10 @@ rewrite rules and the routed-id settlement. | `src/combos/request.ts` | Clones each selected combo target request and applies the existing target capability ladder: adaptive unknown targets and explicit empty ladders receive no unsupported reasoning/thinking controls, while known ladders retain per-target resolution. | | `src/adapters/openai-responses.ts` | Native OpenAI/ChatGPT Responses passthrough. | | `src/responses/muse-tool-name-alias.ts` | Host-gated Meta Muse 64-char tool-name alias/restore used by the Responses passthrough. | -| `src/adapters/openai-chat.ts`, `src/adapters/openai-chat/` | OpenAI-compatible Chat Completions bridge, split into leaves (`wire.ts`, `messages.ts`, `response-events.ts`, `passthrough.ts`, `parallel-tool-calls.ts`, `reasoning-wire.ts`, `serialized-tool-call-content.ts`, `tool-call-validation.ts`, `tool-schema.ts`, `errors.ts`). `parallel-tool-calls.ts` owns the `parallel_tool_calls` wire value for both the translated and native builders, so the three provider states — configured opt-out, configured opt-in, and the unset default that forwards only a caller's explicit `false` — cannot drift between them. `reasoning-wire.ts` applies explicit gateway-object and tool-bearing effort-omission declarations to both builders; absent declarations leave native raw forwarding unchanged. Its client delivery shapes in `src/chat/outbound.ts` and `src/server/chat-native-sse.ts` relay the upstream `service_tier` echo on non-stream, folded-stream, and synthesized-SSE bodies, never inventing the key when the upstream omits it. | +| `src/adapters/openai-chat.ts`, `src/adapters/openai-chat/` | OpenAI-compatible Chat Completions bridge, split into leaves (`wire.ts`, `messages.ts`, `response-events.ts`, `passthrough.ts`, `parallel-tool-calls.ts`, `reasoning-wire.ts`, `serialized-tool-call-content.ts`, `tool-call-validation.ts`, `tool-call-id-remint.ts`, `tool-schema.ts`, `errors.ts`). `parallel-tool-calls.ts` owns the `parallel_tool_calls` wire value for both the translated and native builders, so the three provider states — configured opt-out, configured opt-in, and the unset default that forwards only a caller's explicit `false` — cannot drift between them. `reasoning-wire.ts` applies explicit gateway-object and tool-bearing effort-omission declarations to both builders; absent declarations leave native raw forwarding unchanged. Its client delivery shapes in `src/chat/outbound.ts` and `src/server/chat-native-sse.ts` relay the upstream `service_tier` echo on non-stream, folded-stream, and synthesized-SSE bodies, never inventing the key when the upstream omits it. | | `src/adapters/anthropic.ts` | Anthropic Messages bridge. A `refusal` or `content_filter` stop reason yields an explicit `incomplete` event with `retryable: false` rather than `done` with that stopReason (#4312); `max_tokens` remains `done`. It is the wire that defines `tools[*].strict` and `tools[*].allowed_callers`, so a rebuilt declaration carries both: an explicit `strict: true` and any `allowed_callers` the caller declared. An absent `strict` stays absent, because the Messages inbound records it as `false` and a `false` on the wire would read as an opt-out nobody asked for. Anthropic Fast uses the native `anthropic-speed` FastWire: a set decision sends `speed: "fast"` with `fast-mode-2026-02-01` in one case-insensitively merged, deduplicated `anthropic-beta` header that preserves OAuth betas. Stream and buffered `usage.speed` echoes confirm fast or downgrade to standard; no echo leaves the request assumed. `tests/adapters/anthropic/anthropic-fast-speed.test.ts` pins the wire and echoes. Anthropic Fast is opt-in: the registry marks both Anthropic entries `fastOptIn`, and `src/providers/fast-opt-in.ts` (`providerFastSwitchOff`) keeps Fast off until `providers..fastEnabled` is `true`. An off switch is provider capability `false`, applied in the FastPolicy authority (`service-tier.ts`), `resolveModelPolicy`, and router registry enrichment, so no model-level Fast toggle, `--fast` row, or proxy-generated `speed` field is produced. Native Claude Messages passthrough still forwards a `speed` field the caller sends itself, outside the proxy Fast policy. `tests/adapters/anthropic/anthropic-fast-opt-in.test.ts` pins the default, the switch, and the management PATCH/GET. | | `src/adapters/google.ts` | Gemini bridge. The final wire compiler owns [endpoint-scoped tool-schema loss policy](providers/google.md#google-tool-schema-loss-reporting): compatible mode changes no request bytes, strict initial loss creates no physical send, and strict non-direct repair creates no changed repair send. A caller-declared strict tool selects `functionCallingConfig.mode: "VALIDATED"` in place of the absent-choice default; `NONE`, `ANY` and a forced-name choice are stronger constraints the caller asked for and are never overwritten. | +| `src/adapters/unique-tool-call-ids.ts` | Request-scoped tool-call-id uniqueness for every `openai-chat` provider. An upstream that mints an id from the call's position in its response repeats `call-0-0` on every turn; a Messages client has already paired that id, drops the duplicate, and is left with a call that has no result, so the turn reads as empty and the model re-issues it indefinitely. Only a **repeat** is rewritten — the first occurrence stays byte-identical, leaving prompt-cache keys, reasoning-replay lookups and already-unique upstreams untouched. The ids to avoid come from the caller's history, captured in `buildRequest` (the only point that sees it) and applied at emission, never at ingestion: ingestion matches streamed deltas against the id upstream sent, so rewriting there would strip a pending call of its identity mid-stream. A repeat takes a `-` suffix, never `_`, because `_` reads as a batch sub-call of ``. Covered by `tests/adapters/openai/openai-chat-tool-call-id-remint.test.ts`. | | `src/adapters/declaration-carrier.ts`, `src/adapters/input-media-guard.ts` | Default-deny allowlists for constraints the normalized request carries but a wire may not be able to express: `tools[*].allowed_callers`, which fences a tool off from callers, and inline document bytes. Both are refused with a 400 at the single guard every registered adapter passes through, rather than left to each adapter, because an adapter that never learned about the carrier rebuilds without it and answers normally. `allowed_callers` reaches the `anthropic` wire; document bytes reach `anthropic`, `openai-chat` and `google`; the `openai-responses` wire is exempt from the whole guard because it forwards the original body. Adding an `AdapterWire` member makes the omission visible in these lists instead of at a customer's upstream. The unrestricted `["direct"]` caller default is not a restriction. | | `src/adapters/azure.ts` | Azure OpenAI bridge. | | `src/adapters/cursor.ts`, `src/adapters/cursor/` | Cursor protobuf transport: discovery, request builder, event decoding, MCP, thread continuity, native-exec policy. | diff --git a/tests/adapters/openai/openai-chat-tool-call-id-remint.test.ts b/tests/adapters/openai/openai-chat-tool-call-id-remint.test.ts new file mode 100644 index 00000000000..c638fd86127 --- /dev/null +++ b/tests/adapters/openai/openai-chat-tool-call-id-remint.test.ts @@ -0,0 +1,216 @@ +import { describe, expect, test } from "bun:test"; +import { createOpenAIChatAdapter } from "../../../src/adapters/openai-chat"; +import { createToolCallIdReminter, reservedToolCallIdsFromHistory } from "../../../src/adapters/openai-chat/tool-call-id-remint"; +import { MAX_TOOL_CALL_ID_LENGTH } from "../../../src/adapters/tool-call-id"; +import { withUniqueToolCallIds } from "../../../src/adapters/unique-tool-call-ids"; +import { createTestTranslatorBudget, withTestTranslatorBudget } from "../../helpers/translator-budget"; +import type { OcxMessage } from "../../../src/types"; + +/** + * Some upstreams derive a tool-call id from the call's position in the response and therefore mint + * the same `call-0-0` on every turn. A client that already paired that id with an earlier call drops + * the duplicate, the turn collapses to an assistant message with no content, and the model re-issues + * the call forever. The reminter makes the second and later occurrences unique. + */ +describe("createToolCallIdReminter", () => { + test("keeps the first occurrence byte-identical", () => { + const remint = createToolCallIdReminter(["call_already_stored"]); + + expect(remint("call_fresh")).toBe("call_fresh"); + expect(remint("call_already_stored")).not.toBe("call_already_stored"); + }); + + test("deduplicates the positional id an upstream repeats every turn", () => { + const emitted: string[] = []; + const history: string[] = []; + for (let turn = 0; turn < 4; turn++) { + const remint = createToolCallIdReminter(history); + const wire = remint("call-0-0"); + emitted.push(wire); + history.push(wire); + } + + expect(emitted).toEqual(["call-0-0", "call-0-0-2", "call-0-0-3", "call-0-0-4"]); + expect(new Set(emitted).size).toBe(4); + }); + + test("avoids a suffix that would read as a batch sub-call of an earlier id", () => { + const remint = createToolCallIdReminter(["call-0-0"]); + + // `_` is parsed by at least one client as sub-call N of ``, which pairs the + // result to the wrong call. The rewrite must not land in that family. + expect(remint("call-0-0")).toBe("call-0-0-2"); + }); + + test("repeated occurrences within one response stay distinct", () => { + const remint = createToolCallIdReminter([]); + const ids = [remint("call-0-0"), remint("call-0-0"), remint("call-0-0")]; + + expect(new Set(ids).size).toBe(3); + }); + + test("skips a suffix the reserved set already occupies", () => { + const remint = createToolCallIdReminter(["call-0-0", "call-0-0-2"]); + + expect(remint("call-0-0")).toBe("call-0-0-3"); + }); + + test("every rewritten id is conforming and within the Anthropic length bound", () => { + const long = "c".repeat(MAX_TOOL_CALL_ID_LENGTH); + const remint = createToolCallIdReminter([long]); + + const rewritten = remint(long); + expect(rewritten).toMatch(/^[a-zA-Z0-9_-]+$/); + expect(rewritten.length).toBeLessThanOrEqual(MAX_TOOL_CALL_ID_LENGTH); + }); + + test("sanitizes a non-conforming id rather than dropping it", () => { + const remint = createToolCallIdReminter(["call:0:0"]); + + const rewritten = remint("call:0:0"); + expect(rewritten).toMatch(/^[a-zA-Z0-9_-]+$/); + expect(rewritten).not.toBe("call:0:0"); + }); +}); + +describe("withUniqueToolCallIds", () => { + const baseProvider = { adapter: "openai-chat", baseUrl: "https://openrouter.ai/api/v1", apiKey: "key" } as const; + + /** An upstream that mints the id from the call's position, so every response repeats `call-0-0`. */ + const positionalUpstream = () => Response.json({ + choices: [{ + message: { + content: "", + tool_calls: [{ id: "call-0-0", type: "function", function: { name: "Bash", arguments: "{}" } }], + }, + finish_reason: "tool_calls", + }], + }); + + const buildParsed = (history: OcxMessage[]) => ({ + modelId: "mock-model", + stream: false, + options: {}, + context: { + messages: [ + { role: "user", content: "go", timestamp: 0 }, + ...history, + ], + tools: [{ name: "Bash", description: "run", parameters: { type: "object", properties: {} } }], + }, + }) as never; + + test("a repeated upstream id is rewritten so the client never sees the same id twice", async () => { + const adapter = withTestTranslatorBudget(withUniqueToolCallIds(createOpenAIChatAdapter(baseProvider))); + const emitted: string[] = []; + const history: OcxMessage[] = []; + + // Each turn: build (sees the history), parse (emits the call), then the client stores it. + for (let turn = 0; turn < 3; turn++) { + adapter.buildRequest(buildParsed(history)); + const events = await adapter.parseResponse!(positionalUpstream(), createTestTranslatorBudget()); + const started = events.find(event => event.type === "tool_call_start"); + if (started?.type !== "tool_call_start") throw new Error("no tool call emitted"); + emitted.push(started.id); + history.push({ role: "assistant", content: [{ type: "toolCall", id: started.id, name: "Bash", arguments: "{}" }], timestamp: turn }); + } + + expect(emitted).toEqual(["call-0-0", "call-0-0-2", "call-0-0-3"]); + expect(new Set(emitted).size).toBe(3); + }); + + test("a first turn with no history emits the upstream id byte-identical", async () => { + const adapter = withTestTranslatorBudget(withUniqueToolCallIds(createOpenAIChatAdapter(baseProvider))); + adapter.buildRequest(buildParsed([])); + + const events = await adapter.parseResponse!(positionalUpstream(), createTestTranslatorBudget()); + const started = events.find(event => event.type === "tool_call_start"); + + expect(started?.type === "tool_call_start" && started.id).toBe("call-0-0"); + }); + + /** The same positional upstream, but framed as an SSE stream: the stream path remints separately. */ + const positionalUpstreamStream = (id: string) => new Response( + `data: ${JSON.stringify({ + choices: [{ + delta: { tool_calls: [{ index: 0, id, function: { name: "Bash", arguments: '{"command":' } }] }, + }], + })}\n\n` + + `data: ${JSON.stringify({ + choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: '"ls"}' } }] } }], + })}\n\n` + + `data: ${JSON.stringify({ + choices: [{ delta: {}, finish_reason: "tool_calls" }], + })}\n\ndata: [DONE]\n\n`, + ); + + test("the stream path remints a repeated id and keeps each call's event sequence intact", async () => { + const adapter = withTestTranslatorBudget(withUniqueToolCallIds(createOpenAIChatAdapter(baseProvider))); + const emitted: string[] = []; + const history: OcxMessage[] = []; + + // Each turn: build (sees the history), stream (emits the call), then the client stores it. + for (let turn = 0; turn < 2; turn++) { + adapter.buildRequest(buildParsed(history)); + const events = []; + for await (const event of adapter.parseStream(positionalUpstreamStream("call-0-0"), createTestTranslatorBudget())) { + events.push(event); + } + + const started = events.find(event => event.type === "tool_call_start"); + if (started?.type !== "tool_call_start") throw new Error("no tool call emitted"); + emitted.push(started.id); + + // The remint must not disturb the deltas or the terminator around it. + const startIndex = events.indexOf(started); + const endIndex = events.findIndex(event => event.type === "tool_call_end"); + expect(startIndex).toBeGreaterThanOrEqual(0); + expect(endIndex).toBeGreaterThan(startIndex); + const argumentsParts: string[] = []; + for (const event of events) { + if (event.type === "tool_call_delta") argumentsParts.push(event.arguments); + } + const arguments_ = argumentsParts.join(""); + expect(arguments_).toBe('{"command":"ls"}'); + + history.push({ role: "assistant", content: [{ type: "toolCall", id: started.id, name: "Bash", arguments: arguments_ }], timestamp: turn }); + } + + expect(emitted).toEqual(["call-0-0", "call-0-0-2"]); + expect(new Set(emitted).size).toBe(2); + }); + + test("a first streamed turn with no history emits the upstream id byte-identical", async () => { + const adapter = withTestTranslatorBudget(withUniqueToolCallIds(createOpenAIChatAdapter(baseProvider))); + adapter.buildRequest(buildParsed([])); + + let id: string | undefined; + for await (const event of adapter.parseStream(positionalUpstreamStream("call-0-0"), createTestTranslatorBudget())) { + if (event.type === "tool_call_start") id = event.id; + } + + expect(id).toBe("call-0-0"); + }); +}); + +describe("reservedToolCallIdsFromHistory", () => { + test("collects both sides of every prior call", () => { + const history = [ + { role: "assistant", content: [{ type: "toolCall", id: "call-0-0", name: "Bash", arguments: "{}" }], timestamp: 1 }, + { role: "toolResult", toolCallId: "call-0-0", content: "ok", timestamp: 2 }, + ] as unknown as OcxMessage[]; + + const reserved = reservedToolCallIdsFromHistory(history); + expect(reserved.has("call-0-0")).toBe(true); + expect([...reserved]).toEqual(["call-0-0"]); + }); + + test("ignores a call the client never stored a result for", () => { + const history = [ + { role: "user", content: "hello", timestamp: 1 }, + { role: "assistant", content: [{ type: "text", text: "hi" }], timestamp: 2 }, + ] as unknown as OcxMessage[]; + + expect(reservedToolCallIdsFromHistory(history).size).toBe(0); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 2eed0eac3af..9e86e0ca857 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -2,6 +2,7 @@ "deepseek-artifact-tool-schema.test.ts": "providers", "client-config-export-output-limit.test.ts": "config", "openai-chat-serialized-tool-call-scaling.test.ts": "adapters/openai", + "openai-chat-tool-call-id-remint.test.ts": "adapters/openai", "coding-agent-json-lines-scaling.test.ts": "providers", "usage-snapshot-digest-reuse.test.ts": "usage", "release-desktop-scripts.test.ts": "ci-workflows", From 95813aac1f0fffd6ca5071cfb143604dc8b0ed48 Mon Sep 17 00:00:00 2001 From: Yum-wu <1172989563@qq.com> Date: Sat, 26 Sep 2026 18:42:11 +0900 Subject: [PATCH 2/5] fix(chat-native): refetch on zero-output mid-stream socket reset (#5882) Carried from #5882 as one squashed commit. Co-authored-by: Yum-wu <1172989563@qq.com> --- scripts/test-layout/layout.json | 1 + src/lib/upstream-retry.ts | 57 ++++- src/server/chat-native.ts | 203 ++++++++++++------ src/server/responses/reset-replay.ts | 32 +++ tests/fixtures/test-layout-expected.json | 1 + tests/lib/upstream-retry-zero-output.test.ts | 195 +++++++++++++++++ tests/responses/chat-native-spend.test.ts | 200 ++++++++++++++++- .../responses/responses-reset-replay.test.ts | 43 ++++ 8 files changed, 661 insertions(+), 71 deletions(-) create mode 100644 tests/lib/upstream-retry-zero-output.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index c3997a747e8..2eae1c41a3a 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1804,6 +1804,7 @@ "upstream-http-version.test.ts": "server", "upstream-reachability.test.ts": "codex-integration", "upstream-retry.test.ts": "lib", + "upstream-retry-zero-output.test.ts": "lib", "upstream-transient-retry.test.ts": "providers", "url-normalization.test.ts": "config", "usage-aggregate-cache.test.ts": "usage", diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index 2b2b546388c..d6a2be5257a 100644 --- a/src/lib/upstream-retry.ts +++ b/src/lib/upstream-retry.ts @@ -847,6 +847,61 @@ export async function refetchAfterProtocolSafeReset( console.warn("[upstream-retry] protocol-safe refetch rejected" + label + "; preserving original stream error"); return null; } - console.warn("[upstream-retry] pre-output Responses reset" + label + "; using one replacement stream"); + console.warn("[upstream-retry] pre-output stream reset" + label + "; using one replacement stream"); return replacement; } + +/** + * Wrap a streamed body so a reset that arrives before the downstream reader has consumed a single + * byte swaps in ONE replacement body. + * + * The zero-byte gate is the whole reason this wrapper exists: the caller observed nothing, which is + * the stage where a replacement may even be considered. Every other question -- whether the operator + * granted one, whether the request is replayable, whether the replacement is a fresh unlocked body + * that matches the contract already promised to the client -- belongs to + * {@link refetchAfterProtocolSafeReset}. Delegating rather than re-deciding is what keeps the chat + * lane from drifting away from the one the Responses stream already uses. + * + * Partial output is never masked: once a byte has reached the caller, the original failure stands. + */ +export function wrapWithZeroOutputRefetch( + body: ReadableStream, + doFetch: ProtocolSafeRefetch, + opts: ProtocolSafeRefetchOptions = {}, +): ReadableStream { + let reader = body.getReader(); + let bytesRead = 0; + let retried = false; + return new ReadableStream({ + async pull(controller) { + for (;;) { + try { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + return; + } + bytesRead += value.byteLength; + controller.enqueue(value); + return; + } catch (err) { + if (!retried && bytesRead === 0 && !opts.abortSignal?.aborted) { + retried = true; + const replacement = await refetchAfterProtocolSafeReset(doFetch, err, opts); + if (replacement?.body) { + try { void reader.cancel().catch(() => {}); } catch { /* broken reader; the replacement won */ } + reader = replacement.body.getReader(); + continue; + } + } + try { controller.error(err); } catch { /* already torn down */ } + return; + } + } + }, + cancel(reason) { + try { void reader.cancel(reason).catch(() => {}); } catch { /* already torn down */ } + }, + }); +} + diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts index 04e143a3b82..fe5896a7758 100644 --- a/src/server/chat-native.ts +++ b/src/server/chat-native.ts @@ -38,11 +38,14 @@ import { UpstreamRetryEvidenceError, type UpstreamSendRecovery, UPSTREAM_RESET_REPLAY_REFUSED_CODE, + wrapWithZeroOutputRefetch, } from "../lib/upstream-retry"; import { isTranslatorBudgetExceededError, type TranslatorBudget, } from "../lib/translator-budget"; +import { authorizeResendForRecovery } from "../lib/request-resend-gate"; +import { ambiguousResendAllowanceFor, selfContainedChatBody } from "./responses/reset-replay"; import { hasKeyPoolFailover, selectProactiveApiKeyTransport, @@ -369,7 +372,37 @@ export async function runNativeChatAttempt( ); const transientSendAvailable = (): boolean => remainingTransientSends() > 0; - const send = async (request: AdapterRequest, recovery?: "rate-limit-429" | "key-429"): Promise => { + /** + * The operator's replacement grant for THIS logical request, read per leg because + * `activeProvider` is reassigned by credential rotation inside the send loop. + * + * A combo child already holds the request's send ledger, so the grant comes from it and every + * ambiguous stage of the request draws on the same counter. A direct Chat request opens no such + * ledger -- it accounts through a spend tracker instead -- so it keeps the single grant locally, + * which is the same ceiling for the same one attempt. + */ + const requestIsSelfContained = (() => { + let memo: boolean | undefined; + return (): boolean => memo ??= selfContainedChatBody(execution.chatBody); + })(); + let localAmbiguousResendSpent = false; + const claimAmbiguousResend = (limit: number): boolean => { + if (sendBudget?.claimAmbiguousResend) return sendBudget.claimAmbiguousResend(limit); + if (localAmbiguousResendSpent || limit <= 0) return false; + localAmbiguousResendSpent = true; + return true; + }; + const ambiguousResend = () => ambiguousResendAllowanceFor( + activeProvider, + requestIsSelfContained, + claimAmbiguousResend, + ); + + const send = async ( + request: AdapterRequest, + recovery?: "rate-limit-429" | "key-429" | UpstreamSendRecovery, + singleShot = false, + ): Promise => { try { // #2643: opted-in key-auth openai-chat providers retry pre-stream transient statuses on // the native chat lane too; everyone else keeps reset-only semantics. @@ -378,74 +411,81 @@ export async function runNativeChatAttempt( if (requestTransientPolicy && remaining <= 0) { throw new Error("native Chat transient send budget exhausted before recovery dispatch"); } + const dispatch = (transportRecovery?: UpstreamSendRecovery) => { + const effectiveRecovery = transportRecovery + ?? (recovery === "connection-reset" ? "connection-reset" : undefined); + return fetchWithHeaderTimeout( + request.url, + applyUpstreamRecoveryInit({ + method: request.method, + headers: request.headers, + body: request.body, + }, effectiveRecovery), + upstream.signal, + connectMs, + requestedStream, + providerFetch(activeProvider, undefined, { + providerName: route.providerName, + modelId: route.modelId, + dispatchOverride: async (_input, init, execute) => { + if (!providerApiKeySelectionIsCurrent(config, route.providerName, activeProvider)) { + const current = resolveCurrentProviderApiKeyTransport(config, route.providerName, activeProvider); + if (!current || !isNativeChatRouteEligible({ ...route, provider: current }, execution.chatBody, config)) { + throw new Error("Provider key selection is no longer available for native Chat"); + } + activeProvider = current; + stampApiKeyAccountLabel(logCtx, route.providerName, activeProvider); + activeAdapter = createOpenAIChatAdapter(current); + activeRequest.releaseBodyObservation?.(); + releaseRetainedRequest(); + activeRequest = buildActiveRequest(); + try { retainRequest(activeRequest); } + catch (error) { activeRequest.releaseBodyObservation?.(); throw error; } + } + // The retry closure may still hold a pre-pacing request. Replace its entire + // wire shape, not just Authorization, and retain transport recovery flags. + request = activeRequest; + const headers = new Headers(request.headers); + const encoding = new Headers(init.headers).get("accept-encoding"); + if (!headers.has("accept-encoding") && encoding) headers.set("accept-encoding", encoding); + if (init.signal?.aborted) throw init.signal.reason; + if (sendBudget) { + // Backstop for sends the helper cannot see coming (a reset replay). The first + // report settles the combo's booking; each later one is charged and booked. + if (physicalSends > 0 && sendBudget.remainingBaseSends(sharedSendCap) <= 0) { + throw new SendBudgetExhaustedError(safeHostLabel(request.url)); + } + physicalSends += 1; + sendBudget.used += 1; + } else if (!spendTracker?.charge()) throw new NativeChatSpendRefusal(); + noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery); + // A reselected provider transport is still a physical send: the connection policy + // and manual-redirect ownership wrap the selected implementation (#4992). + const dispatched = await sendWithConnectionPolicy( + (activeProvider as OcxProviderTransport).fetch ?? execute, + request.url, + applyUpstreamRecoveryInit({ + ...init, method: request.method, headers, body: request.body, + }, transportRecovery), + // Reselection can replace the provider transport and the wire shape, so the + // egress route is bound to the provider this send actually uses. Omitting it + // here would let a provider transport bypass its configured route entirely, + // because that transport wins over the executor that carries the binding. + { providerName: route.providerName, provider: activeProvider }, + ); + if (!dispatched.ok) await recordKeyAttemptFailure(logCtx, dispatched, init.signal ?? upstream.signal); + return dispatched; + }, + }), + ); + }; + // A replacement bought by an ambiguous reset is ONE send. Routing it back through a retry + // helper would let that helper spend sends the operator allowance never granted, which is + // the duplicated inference the allowance exists to bound. + if (singleShot) return await dispatch("connection-reset"); const fetchWithPolicy = requestTransientPolicy ? fetchWithTransientRetry : fetchWithResetRetry; return await fetchWithPolicy( - (transportRecovery?: UpstreamSendRecovery) => { - return fetchWithHeaderTimeout( - request.url, - applyUpstreamRecoveryInit({ - method: request.method, - headers: request.headers, - body: request.body, - }, transportRecovery), - upstream.signal, - connectMs, - requestedStream, - providerFetch(activeProvider, undefined, { - providerName: route.providerName, - modelId: route.modelId, - dispatchOverride: async (_input, init, execute) => { - if (!providerApiKeySelectionIsCurrent(config, route.providerName, activeProvider)) { - const current = resolveCurrentProviderApiKeyTransport(config, route.providerName, activeProvider); - if (!current || !isNativeChatRouteEligible({ ...route, provider: current }, execution.chatBody, config)) { - throw new Error("Provider key selection is no longer available for native Chat"); - } - activeProvider = current; - stampApiKeyAccountLabel(logCtx, route.providerName, activeProvider); - activeAdapter = createOpenAIChatAdapter(current); - activeRequest.releaseBodyObservation?.(); - releaseRetainedRequest(); - activeRequest = buildActiveRequest(); - try { retainRequest(activeRequest); } - catch (error) { activeRequest.releaseBodyObservation?.(); throw error; } - } - // The retry closure may still hold a pre-pacing request. Replace its entire - // wire shape, not just Authorization, and retain transport recovery flags. - request = activeRequest; - const headers = new Headers(request.headers); - const encoding = new Headers(init.headers).get("accept-encoding"); - if (!headers.has("accept-encoding") && encoding) headers.set("accept-encoding", encoding); - if (init.signal?.aborted) throw init.signal.reason; - if (sendBudget) { - // Backstop for sends the helper cannot see coming (a reset replay). The first - // report settles the combo's booking; each later one is charged and booked. - if (physicalSends > 0 && sendBudget.remainingBaseSends(sharedSendCap) <= 0) { - throw new SendBudgetExhaustedError(safeHostLabel(request.url)); - } - physicalSends += 1; - sendBudget.used += 1; - } else if (!spendTracker?.charge()) throw new NativeChatSpendRefusal(); - noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery); - // A reselected provider transport is still a physical send: the connection policy - // and manual-redirect ownership wrap the selected implementation (#4992). - const dispatched = await sendWithConnectionPolicy( - (activeProvider as OcxProviderTransport).fetch ?? execute, - request.url, - applyUpstreamRecoveryInit({ - ...init, method: request.method, headers, body: request.body, - }, transportRecovery), - // Reselection can replace the provider transport and the wire shape, so the - // egress route is bound to the provider this send actually uses. Omitting it - // here would let a provider transport bypass its configured route entirely, - // because that transport wins over the executor that carries the binding. - { providerName: route.providerName, provider: activeProvider }, - ); - if (!dispatched.ok) await recordKeyAttemptFailure(logCtx, dispatched, init.signal ?? upstream.signal); - return dispatched; - }, - }), - ); - }, + dispatch, { abortSignal: upstream.signal, label: safeHostLabel(request.url), @@ -630,7 +670,34 @@ export async function runNativeChatAttempt( if (contentType.includes("text/event-stream") && response.body) { if (requestedStream) transferTurnToStream(); let terminalStatus: number | undefined; - const stream = nativeChatSse(response.body, { + const resilientBody = wrapWithZeroOutputRefetch( + response.body, + // One physical send, not one more trip through a retry helper: the allowance below buys a + // single replacement, so the send must not be able to spend more than it granted. + async () => { + try { + return await send(activeRequest, "connection-reset", true); + } finally { + // Reselection can charge the rebuilt request after the initial send settled. + releaseRetainedRequest(); + } + }, + { + abortSignal: upstream.signal, + label: safeHostLabel(activeRequest.url), + attempts: remainingTransientSends(), + // The origin returned a head and may already be running the turn, so this is the + // ambiguous row of the stage table. Only the operator allowance the Responses stream + // also draws on can authorise it; without one the original failure stands. + authorize: () => authorizeResendForRecovery("headers-only", "connection-reset", ambiguousResend()).allowed, + // A replacement must satisfy the contract already promised to the client. A 200 that is + // not an event stream would reach the SSE translator as a non-SSE body and surface as a + // malformed-stream error instead of the reset that actually happened. + acceptResponse: replacement => + (replacement.headers.get("content-type") ?? "").toLowerCase().includes("text/event-stream"), + }, + ); + const stream = nativeChatSse(resilientBody, { requestedModel, translatorBudget, signal: upstream.signal, diff --git a/src/server/responses/reset-replay.ts b/src/server/responses/reset-replay.ts index b8e47d44b93..00c3f7a963d 100644 --- a/src/server/responses/reset-replay.ts +++ b/src/server/responses/reset-replay.ts @@ -78,6 +78,38 @@ export function selfContainedResponsesBody(body: unknown): boolean { }); } +/** + * Chat Completions tools are executed by the client by spec -- the origin only emits the call. + * Anything else (`web_search`, `code_interpreter`, a vendor hosted tool) runs on the origin during + * the turn, so an unknown or hosted entry fails the whole catalog rather than being skipped. + */ +function clientExecutedChatTools(tools: unknown, budget: { remaining: number }): boolean { + if (!Array.isArray(tools) || tools.length > budget.remaining) return false; + budget.remaining -= tools.length; + return tools.every(tool => record(tool) && tool.type === "function"); +} + +/** + * A Chat Completions body whose second send can only repeat the inference. + * + * The lane is stateless by construction -- the proxy rebuilds the whole `messages` array on every + * turn, and `previous_response_id` is not part of the Chat wire -- so the only hazards left are + * server-side storage, which would record a second completion, and a hosted tool the origin would + * run a second time. + */ +export function selfContainedChatBody(body: unknown): boolean { + if (!record(body)) return false; + if (body.store === true) return false; + if (body.previous_response_id != null) return false; + // Hosted execution requested outside the `tools` catalog. Judged on the inbound body like every + // other hazard here, which is conservative for the outbound request by design: a body that asks + // for a hosted search is refused rather than assumed harmless because a later stage might drop + // the field. + if (body.web_search_options !== undefined) return false; + if (!Array.isArray(body.messages)) return false; + return body.tools === undefined || clientExecutedChatTools(body.tools, { remaining: MAX_TOOL_ENTRIES }); +} + /** * The allowance for ONE logical request, or nothing when the provider did not opt in. * diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 9e86e0ca857..d5272cbc96e 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1630,6 +1630,7 @@ "upstream-http-version.test.ts": "server", "upstream-reachability.test.ts": "codex-integration", "upstream-retry.test.ts": "lib", + "upstream-retry-zero-output.test.ts": "lib", "upstream-transient-retry.test.ts": "providers", "url-normalization.test.ts": "config", "usage-aggregate-cache.test.ts": "usage", diff --git a/tests/lib/upstream-retry-zero-output.test.ts b/tests/lib/upstream-retry-zero-output.test.ts new file mode 100644 index 00000000000..4c669aba1c8 --- /dev/null +++ b/tests/lib/upstream-retry-zero-output.test.ts @@ -0,0 +1,195 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import { wrapWithZeroOutputRefetch } from "../../src/lib/upstream-retry"; + +function resetError(): Error { + // Shape of Bun's fetch rejection on a stale pooled socket. + const err = new Error("The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the second argument to fetch()"); + (err as Error & { code: string }).code = "ECONNRESET"; + return err; +} + +const encoder = new TextEncoder(); + +function streamOf(chunks: Uint8Array[]): ReadableStream { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) controller.enqueue(chunks[i++]!); + else controller.close(); + }, + }); +} + +function failingStream(err: Error): ReadableStream { + return new ReadableStream({ + pull(controller) { + controller.error(err); + }, + }); +} + +async function collect(stream: ReadableStream): Promise { + const out: Uint8Array[] = []; + const reader = stream.getReader(); + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + out.push(value); + } + return out; +} + +function sseResponse(body: ReadableStream): Response { + return new Response(body, { headers: { "Content-Type": "text/event-stream" } }); +} + +const allow = { + authorize: () => true, + acceptResponse: (r: Response) => r.headers.get("content-type") === "text/event-stream", +}; + +const warnSpies: Array> = []; +function silenceWarn(): void { + warnSpies.push(spyOn(console, "warn").mockImplementation(() => {})); +} + +afterEach(() => { + for (const spy of warnSpies.splice(0)) spy.mockRestore(); +}); + +describe("wrapWithZeroOutputRefetch", () => { + test("swaps in the replacement stream on a zero-byte reset", async () => { + silenceWarn(); + let calls = 0; + const wrapped = wrapWithZeroOutputRefetch( + failingStream(resetError()), + async () => { calls += 1; return sseResponse(streamOf([encoder.encode("ok")])); }, + { ...allow, attempts: 1 }, + ); + const chunks = await collect(wrapped); + expect(new TextDecoder().decode(chunks[0])).toBe("ok"); + expect(calls).toBe(1); + }); + + test("refuses the replacement when the operator granted no allowance", async () => { + silenceWarn(); + let calls = 0; + const wrapped = wrapWithZeroOutputRefetch( + failingStream(resetError()), + async () => { calls += 1; return sseResponse(streamOf([encoder.encode("never")])); }, + { ...allow, authorize: () => false, attempts: 1 }, + ); + await expect(collect(wrapped)).rejects.toThrow(/socket connection was closed/i); + expect(calls).toBe(0); + }); + + test("refuses the replacement when the send budget is spent", async () => { + silenceWarn(); + let calls = 0; + const wrapped = wrapWithZeroOutputRefetch( + failingStream(resetError()), + async () => { calls += 1; return sseResponse(streamOf([encoder.encode("never")])); }, + { ...allow, attempts: 0 }, + ); + await expect(collect(wrapped)).rejects.toThrow(/socket connection was closed/i); + expect(calls).toBe(0); + }); + + test("rejects a replacement that is not the event stream the client was promised", async () => { + silenceWarn(); + const wrapped = wrapWithZeroOutputRefetch( + failingStream(resetError()), + async () => new Response(streamOf([encoder.encode("{}")]), { headers: { "Content-Type": "application/json" } }), + { ...allow, attempts: 1 }, + ); + await expect(collect(wrapped)).rejects.toThrow(/socket connection was closed/i); + }); + + test("rejects a non-OK replacement", async () => { + silenceWarn(); + const wrapped = wrapWithZeroOutputRefetch( + failingStream(resetError()), + async () => new Response("nope", { status: 502, headers: { "Content-Type": "text/event-stream" } }), + { ...allow, attempts: 1 }, + ); + await expect(collect(wrapped)).rejects.toThrow(/socket connection was closed/i); + }); + + test("does not refetch after bytes were already consumed", async () => { + silenceWarn(); + let delivered = false; + const stream = new ReadableStream({ + pull(controller) { + if (!delivered) { + delivered = true; + controller.enqueue(encoder.encode("partial")); + return; + } + controller.error(resetError()); + }, + }); + let calls = 0; + const wrapped = wrapWithZeroOutputRefetch( + stream, + async () => { calls += 1; return sseResponse(streamOf([encoder.encode("never")])); }, + { ...allow, attempts: 1 }, + ); + const reader = wrapped.getReader(); + const first = await reader.read(); + // Partial output reached the caller, so the original failure must stand: masking it with a + // replay would deliver a second turn's bytes after the first turn's. + expect(new TextDecoder().decode(first.value)).toBe("partial"); + await expect(reader.read()).rejects.toThrow(/socket connection was closed/i); + expect(calls).toBe(0); + }); + + test("propagates a non-reset failure without asking for a replacement", async () => { + let calls = 0; + const wrapped = wrapWithZeroOutputRefetch( + failingStream(new Error("boom")), + async () => { calls += 1; return sseResponse(streamOf([])); }, + { ...allow, attempts: 1 }, + ); + await expect(collect(wrapped)).rejects.toThrow("boom"); + expect(calls).toBe(0); + }); + + test("propagates the original reset when the replacement send itself fails", async () => { + silenceWarn(); + const wrapped = wrapWithZeroOutputRefetch( + failingStream(resetError()), + async () => { throw new Error("refetch failed"); }, + { ...allow, attempts: 1 }, + ); + await expect(collect(wrapped)).rejects.toThrow(/socket connection was closed/i); + }); + + test("retries at most once: a second zero-byte reset on the replacement propagates", async () => { + silenceWarn(); + let calls = 0; + const wrapped = wrapWithZeroOutputRefetch( + failingStream(resetError()), + async () => { calls += 1; return sseResponse(failingStream(resetError())); }, + { ...allow, attempts: 2 }, + ); + await expect(collect(wrapped)).rejects.toThrow(/socket connection was closed/i); + expect(calls).toBe(1); + }); + + test("forwards cancellation to the active reader", async () => { + const cancelled: string[] = []; + const original = new ReadableStream({ + pull(controller) { + controller.enqueue(encoder.encode("x")); + }, + cancel(reason) { + cancelled.push(String(reason)); + }, + }); + const wrapped = wrapWithZeroOutputRefetch(original, async () => sseResponse(streamOf([])), allow); + const reader = wrapped.getReader(); + await reader.read(); + await reader.cancel("stop"); + expect(cancelled.length).toBe(1); + }); +}); diff --git a/tests/responses/chat-native-spend.test.ts b/tests/responses/chat-native-spend.test.ts index 7e847c42972..967d520630a 100644 --- a/tests/responses/chat-native-spend.test.ts +++ b/tests/responses/chat-native-spend.test.ts @@ -1,5 +1,6 @@ -import { afterEach, beforeEach, expect, test } from "bun:test"; +import { afterEach, beforeEach, expect, spyOn, test } from "bun:test"; import { mkdtempSync } from "node:fs"; +import { createServer } from "node:net"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { saveConfig } from "../../src/config"; @@ -14,23 +15,32 @@ import { removeTreeWithRetry } from "../helpers/remove-tree"; import { resetProviderRequestPacingForTest } from "../../src/providers/request-pacing"; import { estimateTokens } from "../../src/lib/token-estimate"; import { getRequestLogEntries } from "../../src/server/request-log"; +import * as stateStores from "../../src/lib/state-store-registrations"; let previousHome: string | undefined; let testDir = ""; let isolatedCodexHome: IsolatedCodexHome | null = null; let activeServer: ReturnType | undefined; let activeUpstream: ReturnType | undefined; +let activeRawUpstream: ReturnType | undefined; let stopping: Promise | undefined; function stopFixtureServers(): Promise { // A timed-out body and its afterEach join one owner instead of racing two stops. return stopping ??= (async () => { try { await activeServer?.stop(true); } - finally { await activeUpstream?.stop(true); } + finally { + await activeUpstream?.stop(true); + if (activeRawUpstream) { + const raw = activeRawUpstream; + await new Promise(resolve => raw.close(() => resolve())); + } + } })(); } beforeEach(() => { activeServer = undefined; activeUpstream = undefined; + activeRawUpstream = undefined; stopping = undefined; previousHome = process.env.OPENCODEX_HOME; isolatedCodexHome = installIsolatedCodexHome("ocx-chat-spend-"); @@ -166,3 +176,189 @@ test("native Chat includes tool definitions in its pre-dispatch spend reservatio await stopFixtureServers(); } }); + +/** + * A raw TCP upstream, because `Bun.serve` turns a body error into a CLEAN EOF: the client would + * read an empty 200 instead of the reset under test. Only a real socket close after the head + * produces the `ECONNRESET` the zero-output wrapper is built for, so the fixture writes the head + * by hand and destroys the socket before the first chunk. + */ +async function mockResettingChatUpstream( + opts: { + onSend?: (sendIndex: number, headers: string) => void; + replacementFrames?: string; + } = {}, +): Promise<{ baseUrl: string; sends: () => number }> { + let sends = 0; + const server = createServer(socket => { + let buffered = Buffer.alloc(0); + socket.on("error", () => {}); + socket.on("data", chunk => { + buffered = Buffer.concat([buffered, chunk]); + const headerEnd = buffered.indexOf("\r\n\r\n"); + if (headerEnd < 0) return; + const headers = buffered.subarray(0, headerEnd).toString("latin1"); + const declared = /content-length:\s*(\d+)/i.exec(headers); + if (buffered.length < headerEnd + 4 + (declared ? Number(declared[1]) : 0)) return; + sends += 1; + opts.onSend?.(sends, headers); + if (sends === 1) { + // A real 200 head, chunked, then a hard close before the first chunk: the head is + // genuine and the body never carried a byte, which is the stage under test. + socket.write("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n"); + socket.destroy(); + return; + } + const body = opts.replacementFrames ?? recoveredChatFrames(); + socket.end(`HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: ${Buffer.byteLength(body)}\r\nConnection: close\r\n\r\n${body}`); + }); + }); + activeRawUpstream = server; + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address() as { port: number }; + return { baseUrl: `http://127.0.0.1:${address.port}/v1`, sends: () => sends }; +} + +function recoveredChatFrames(): string { + return [ + `data: ${JSON.stringify({ choices: [{ index: 0, delta: { role: "assistant", content: "Recovered" } }] })}\n\n`, + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 5, completion_tokens: 2 } })}\n\n`, + "data: [DONE]\n\n", + ].join(""); +} + +async function postStreamingChat( + server: ReturnType, + extra: Record = {}, +): Promise<{ status: number; text: string }> { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + messages: [{ role: "user", content: "hello" }], + stream: true, + ...extra, + }), + }); + return { status: response.status, text: await response.text() }; +} + +test("native Chat replaces a zero-output mid-stream reset exactly once under the operator opt-in", async () => { + const upstream = await mockResettingChatUpstream(); + saveConfig(mockConfig(upstream.baseUrl, { retryOnReset: {} })); + const server = startServer(0); + activeServer = server; + try { + const { status, text } = await postStreamingChat(server); + expect(status).toBe(200); + expect(text).toContain("Recovered"); + // The allowance buys ONE replacement send, not one more retry ladder. + expect(upstream.sends()).toBe(2); + } finally { + await stopFixtureServers(); + } +}); + +test("native Chat leaves a zero-output mid-stream reset alone without the operator opt-in", async () => { + const upstream = await mockResettingChatUpstream(); + saveConfig(mockConfig(upstream.baseUrl)); + const server = startServer(0); + activeServer = server; + try { + const { text } = await postStreamingChat(server); + expect(upstream.sends()).toBe(1); + expect(text).not.toContain("Recovered"); + } finally { + await stopFixtureServers(); + } +}); + +test("native Chat refuses the replacement when the tool catalog cannot be replayed", async () => { + const upstream = await mockResettingChatUpstream(); + saveConfig(mockConfig(upstream.baseUrl, { retryOnReset: {} })); + const server = startServer(0); + activeServer = server; + try { + // A malformed catalog still routes native -- eligibility judges only the Responses-only + // fields -- so this is the case that actually reaches selfContainedChatBody. + const { text } = await postStreamingChat(server, { tools: "not a list" }); + expect(upstream.sends()).toBe(1); + expect(text).not.toContain("Recovered"); + } finally { + await stopFixtureServers(); + } +}); + +test("native Chat refuses the replacement when the body asks for hosted search", async () => { + const upstream = await mockResettingChatUpstream(); + saveConfig(mockConfig(upstream.baseUrl, { retryOnReset: {} })); + const server = startServer(0); + activeServer = server; + try { + const { text } = await postStreamingChat(server, { web_search_options: {} }); + expect(upstream.sends()).toBe(1); + expect(text).not.toContain("Recovered"); + } finally { + await stopFixtureServers(); + } +}); + +test("a store-enabled turn leaves the native lane, so its reset is not replaced", async () => { + const upstream = await mockResettingChatUpstream(); + saveConfig(mockConfig(upstream.baseUrl, { retryOnReset: {} })); + const server = startServer(0); + activeServer = server; + try { + // `store: true` is a Responses-only feature, so eligibility declines the native lane before + // selfContainedChatBody is consulted. The replacement is the native lane's, so none is made. + const { text } = await postStreamingChat(server, { store: true }); + expect(upstream.sends()).toBe(1); + expect(text).not.toContain("Recovered"); + } finally { + await stopFixtureServers(); + } +}); + +test("native Chat releases retained request bytes after key reselection on replacement send", async () => { + let liveConfig: OcxConfig | null = null; + const realSetLive = stateStores.setLiveStateStoreConfig; + const liveSpy = spyOn(stateStores, "setLiveStateStoreConfig").mockImplementation(cfg => { + liveConfig = cfg; + return realSetLive(cfg); + }); + const seenAuth: string[] = []; + const largeDelta = "X".repeat(64 * 1024); + const largeFrames = [ + `data: ${JSON.stringify({ choices: [{ index: 0, delta: { role: "assistant", content: largeDelta } }] })}\n\n`, + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 5, completion_tokens: 100 } })}\n\n`, + "data: [DONE]\n\n", + ].join(""); + + const upstream = await mockResettingChatUpstream({ + onSend: (sendIndex, headers) => { + const auth = /authorization:\s*([^\r\n]+)/i.exec(headers)?.[1]?.trim(); + if (auth) seenAuth.push(auth); + if (sendIndex === 1 && liveConfig?.providers.mock) { + liveConfig.providers.mock.apiKey = "k-rotated"; + } + }, + replacementFrames: largeFrames, + }); + + saveConfig(mockConfig(upstream.baseUrl, { retryOnReset: {} })); + const server = startServer(0); + activeServer = server; + try { + const { status, text } = await postStreamingChat(server, { + messages: [{ role: "user", content: "hello ".repeat(500) }], + }); + expect(status).toBe(200); + expect(text).toContain(largeDelta); + expect(upstream.sends()).toBe(2); + expect(seenAuth).toEqual(["Bearer k", "Bearer k-rotated"]); + } finally { + liveSpy.mockRestore(); + await stopFixtureServers(); + } +}); diff --git a/tests/responses/responses-reset-replay.test.ts b/tests/responses/responses-reset-replay.test.ts index 81b252738d6..3413782ba4f 100644 --- a/tests/responses/responses-reset-replay.test.ts +++ b/tests/responses/responses-reset-replay.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import { ambiguousResendAllowanceFor, + selfContainedChatBody, selfContainedResponsesBody, } from "../../src/server/responses/reset-replay"; import { authorizeResendForRecovery } from "../../src/lib/request-resend-gate"; @@ -78,6 +79,48 @@ describe("selfContainedResponsesBody", () => { }); }); +describe("selfContainedChatBody", () => { + const chatTurn = { + model: "mock/test-model", + messages: [{ role: "user", content: "hi" }], + tools: [{ type: "function", function: { name: "read" } }], + }; + + test("accepts a Chat turn whose second send can only repeat the inference", () => { + expect(selfContainedChatBody(chatTurn)).toBe(true); + expect(selfContainedChatBody({ model: "m", messages: [] })).toBe(true); + }); + + test("refuses anything that leaves state behind or continues someone else's turn", () => { + for (const override of [ + { store: true }, + { previous_response_id: "resp_1" }, + { messages: undefined }, + { messages: "not a list" }, + ]) { + expect(selfContainedChatBody({ ...chatTurn, ...override })).toBe(false); + } + expect(selfContainedChatBody(null)).toBe(false); + expect(selfContainedChatBody([chatTurn])).toBe(false); + }); + + test("refuses hosted execution requested outside the tool catalog", () => { + expect(selfContainedChatBody({ ...chatTurn, web_search_options: {} })).toBe(false); + expect(selfContainedChatBody({ ...chatTurn, web_search_options: undefined })).toBe(true); + }); + + test("refuses a catalog carrying a tool the origin would execute", () => { + for (const tools of [ + [{ type: "web_search" }], + [{ type: "function", function: { name: "read" } }, { type: "code_interpreter" }], + [{ function: { name: "read" } }], + "not a list", + ]) { + expect(selfContainedChatBody({ ...chatTurn, tools })).toBe(false); + } + }); +}); + describe("ambiguousResendAllowanceFor", () => { test("absent or disabled policy grants nothing", () => { const claim = () => true; From bde7a6d46b4e232abf1f9e1738df79bec4aa4f28 Mon Sep 17 00:00:00 2001 From: Zhaofeng Li Date: Sat, 26 Sep 2026 18:42:16 +0900 Subject: [PATCH 3/5] test: stabilize full-suite isolation and integration budgets (#5849) Carried from #5849 as one squashed commit. The tests/service/service-claim.test.ts hunk is dropped: dev already sandboxes that case with a homedir spy and a stricter assertion. Co-authored-by: Zhaofeng Li --- scripts/test.ts | 6 ++++++ .../remote-workspace-command-runner.test.ts | 18 ++++++++++++++---- tests/codex-integration/codex-shim.test.ts | 2 +- .../issue-702-expired-replay-state.test.ts | 2 +- .../chat-conversation-affinity.test.ts | 4 ++-- .../responses-compaction-routing.test.ts | 5 +++-- tests/server/server-auth.test.ts | 6 ++++-- tests/service/shutdown-launcher.test.ts | 10 ++++++++-- 8 files changed, 39 insertions(+), 14 deletions(-) diff --git a/scripts/test.ts b/scripts/test.ts index 6db9075224a..1b06865dc72 100644 --- a/scripts/test.ts +++ b/scripts/test.ts @@ -371,6 +371,9 @@ export const SERIAL_FULL_SUITE_FILES = [ // Synchronous injection subprocesses can wedge the long-lived macOS isolate // parent while reaping a history Worker; contain them in a fresh bounded lane. "codex-integration/codex-inject-write-lock.test.ts", + // Its management API import stalled the long-lived macOS isolate pool before + // any case ran; the complete file finishes in under a second in a fresh process. + "routing/subagent-roster-retention.test.ts", "update/update-stop-first.test.ts", // Relays a 50 MiB WebSocket frame end to end against a 15s deadline, so its result is a // measurement of the whole process, not of the relay. On a healthy 3-CPU macOS runner the @@ -386,6 +389,9 @@ export const SERIAL_FULL_SUITE_FILES = [ "service/service-ownership-state.test.ts", "service/service-sqlite-home.test.ts", "service/service.test.ts", + "service/service-claim.test.ts", + "service/service-wsl-home-ownership.test.ts", + "codex-integration/native-codex-toggle.test.ts", "codex-integration/native-grok-toggle.test.ts", ] as const; diff --git a/tests/clients/remote-workspace-command-runner.test.ts b/tests/clients/remote-workspace-command-runner.test.ts index fc8170c41ac..e4d8304ac55 100644 --- a/tests/clients/remote-workspace-command-runner.test.ts +++ b/tests/clients/remote-workspace-command-runner.test.ts @@ -1,6 +1,6 @@ import { afterEach, describe, expect, test } from "bun:test"; import { randomUUID } from "node:crypto"; -import { chmodSync, existsSync, linkSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, symlinkSync, writeFileSync } from "node:fs"; +import { chmodSync, existsSync, linkSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, statSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import { @@ -32,9 +32,19 @@ function fixture() { } function privateBubblewrapFixture(): string { - // The production guard checks every ancestor, so tmpdir's shared /tmp parent - // is deliberately ineligible. Own a disposable sibling under the trusted - // interpreter directory without chmod'ing the interpreter or shared parents. + // These argv tests never execute bubblewrap. On Unix, use a system executable + // whose ancestors are trusted even when the test's Bun binary lives under /tmp. + // Minimal images may hard-link env, so choose a single-link executable. + if (process.platform !== "win32") { + for (const candidate of ["/usr/bin/env", "/bin/true", "/usr/bin/true", "/bin/cat", "/bin/ls"]) { + try { + const canonical = realpathSync(candidate); + const file = statSync(canonical); + if (file.isFile() && file.nlink === 1) return canonical; + } catch { /* Candidate is absent on this image. */ } + } + throw new Error("no single-link system executable for the bubblewrap argv fixture"); + } const root = mkdtempSync(join(dirname(realpathSync(process.execPath)), "ocx-bwrap-fixture-")); roots.push(root); const path = join(root, "bwrap"); diff --git a/tests/codex-integration/codex-shim.test.ts b/tests/codex-integration/codex-shim.test.ts index efc971030b4..4c4c8fc728c 100644 --- a/tests/codex-integration/codex-shim.test.ts +++ b/tests/codex-integration/codex-shim.test.ts @@ -641,7 +641,7 @@ os._exit(0) try { process.env.PATH = prependPath(binDir, oldPath); process.env.OPENCODEX_HOME = home; - setCodexShimProbeObservationMsForTests(1_500); + setCodexShimProbeObservationMsForTests(4_000); writeFileSync(codexPath, original, "utf8"); chmodSync(codexPath, 0o755); diff --git a/tests/codex-integration/issue-702-expired-replay-state.test.ts b/tests/codex-integration/issue-702-expired-replay-state.test.ts index 59657cfd870..a682fa6db88 100644 --- a/tests/codex-integration/issue-702-expired-replay-state.test.ts +++ b/tests/codex-integration/issue-702-expired-replay-state.test.ts @@ -787,7 +787,7 @@ describe("Issue #702 expired forward replay state", () => { expect(serialized).toContain(HISTORICAL_USER_SENTINEL); expect(serialized).toContain(HISTORICAL_ASSISTANT_SENTINEL); expect(serialized).toContain(CURRENT_USER_SENTINEL); - }); + }, SERVER_BUDGET_MS); test("a task-scope mismatch refuses the delta before ordinary HTTP upstream I/O", async () => { const scenario = await runForwardScenario("fresh", { "x-codex-parent-thread-id": "other-task" }); diff --git a/tests/responses/chat-conversation-affinity.test.ts b/tests/responses/chat-conversation-affinity.test.ts index 3254c709042..98cef3e41ac 100644 --- a/tests/responses/chat-conversation-affinity.test.ts +++ b/tests/responses/chat-conversation-affinity.test.ts @@ -90,7 +90,7 @@ describe("Chat conversation identity at canonical Responses outbound boundary", } expect(JSON.stringify(seen[1]!.body.input).length).toBeGreaterThan(JSON.stringify(seen[0]!.body.input).length); expect(seen[0]!.body.input).toEqual(seen[2]!.body.input); - }); + }, 20_000); } test(`identity absent, key=${keyPresent}: no session is synthesized`, async () => { @@ -101,6 +101,6 @@ describe("Chat conversation identity at canonical Responses outbound boundary", for (const name of identityHeaders) expect(wire.headers.has(name)).toBe(false); expect(wire.body.prompt_cache_key).toBe(keyPresent ? "shared-cache-cohort" : undefined); } - }); + }, 20_000); } }); diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index bf2a4555991..52540f1dd4a 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -44,6 +44,7 @@ import { captureConfigGeneration } from "../../src/lib/state-store-sweeper"; import { removeTreeWithRetry } from "../helpers/remove-tree"; import { baseCompactionBody, compactionRequest, completedPayload, jsonResponse, keyProviderConfig, nativePoolConfig, sseResponse, twoAccountPoolConfig } from "../helpers/compaction-routing-fixtures"; import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { SERVER_BUDGET_MS } from "../helpers/test-budget"; const originalFetch = globalThis.fetch; @@ -981,7 +982,7 @@ describe("compact alternate-account attempt (#913)", () => { clearComboTargetCooldowns(); } }); - }); + }, SERVER_BUDGET_MS); } for (const [model, account] of [["gpt-5.5", "pool-a"], ["side/gpt-5.5", "pool-b"]] as const) { @@ -1122,7 +1123,7 @@ describe("compact alternate-account attempt (#913)", () => { expect(JSON.stringify(calls.at(-1)!.body.input)).toContain(OPAQUE_COMPACTION_NOTE); expect(calls).toHaveLength(4); }); - }); + }, 20_000); test("native compact headers followed by a stalled body return 504 without retry and release account cleanup", async () => { await withPoolEnv("ocx-compact-body-deadline-", async config => { diff --git a/tests/server/server-auth.test.ts b/tests/server/server-auth.test.ts index 8bccda10382..55680463391 100644 --- a/tests/server/server-auth.test.ts +++ b/tests/server/server-auth.test.ts @@ -1515,7 +1515,7 @@ describe("server local API auth", () => { url.protocol = "ws:"; const ws = new WebSocket(url, { headers: { "x-opencodex-api-key": "local-secret", ...(headers ?? {}) } } as unknown as string[]); return new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error("tier websocket timeout")), watchdogMs(5_000)); + const timer = setTimeout(() => reject(new Error("tier websocket timeout")), watchdogMs(20_000)); ws.addEventListener("open", () => { ws.send(JSON.stringify({ type: "response.create", model, input: "hello" })); }, { once: true }); @@ -3035,7 +3035,9 @@ describe("server local API auth", () => { model, headers: { "chatgpt-account-id": "acct-caller-main" }, }); - expect(response.status).toBe(200); + if (response.status !== 200) { + throw new Error(`caller-main retry returned ${response.status}: ${await response.text()}; dispatches=${JSON.stringify(harness.dispatches)}; observed=${JSON.stringify(observed)}`); + } expect((await response.json() as { id: string }).id).toBe("caller-main-success"); expect(observed).toEqual([ { authorization: "Bearer pool-a-token", accountId: "acct-pool-a" }, diff --git a/tests/service/shutdown-launcher.test.ts b/tests/service/shutdown-launcher.test.ts index c8cf5b2d98b..9f7dca0b8c2 100644 --- a/tests/service/shutdown-launcher.test.ts +++ b/tests/service/shutdown-launcher.test.ts @@ -142,12 +142,18 @@ describe.skipIf(!runnable)("ocx launcher graceful shutdown", () => { child.stderr?.on("data", chunk => { output += String(chunk); }); // 1. Proxy comes up + injected the Codex config (Design B root override on loopback). - const up = await waitUntil(() => healthy(port), STARTUP_BUDGET_MS); + // The health listener may answer before the launcher completes injection. + let healthSeen = false; + const up = await waitUntil(async () => { + if (!(await healthy(port))) return false; + healthSeen = true; + return readFileSync(codexConfig, "utf8").includes(OCX_ROUTING_MARKER_LINE); + }, STARTUP_BUDGET_MS); if (!up) { // Name what actually went wrong instead of asserting a bare boolean. const died = exited ? ` The launcher EXITED (code ${exitCode}, signal ${exitSignal}).` : " The launcher was still running."; throw new Error( - `The proxy never answered /healthz on port ${port} within ${STARTUP_BUDGET_MS}ms.${died}` + `The proxy ${healthSeen ? "answered /healthz but did not inject Codex config" : "never answered /healthz"} on port ${port} within ${STARTUP_BUDGET_MS}ms.${died}` + ` Launcher output:\n${output.trim() || "(none)"}`, ); } From 5f336b78e4dc383cb2859cdb52400b448e7d698f Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 26 Sep 2026 18:43:09 +0900 Subject: [PATCH 4/5] test(chat-native): prove the replacement request copy is released mid-relay The #5882 regression streamed a 64 KiB delta against a 32 MiB turn budget, so it passed with or without the release. The new case rotates the key between sends, holds the replacement body after its first frame, and reads the live translator charge while the stream is relayed: 1x the request size with the release, 2x without it (verified red by reverting 318520ebd6). Also drops a trailing blank line in src/lib/upstream-retry.ts and adds the batch plan. --- devlog/_plan/260926_bug_train_6/000_plan.md | 34 +++++++++ src/lib/upstream-retry.ts | 1 - tests/responses/chat-native-spend.test.ts | 78 +++++++++++++++++++++ 3 files changed, 112 insertions(+), 1 deletion(-) create mode 100644 devlog/_plan/260926_bug_train_6/000_plan.md diff --git a/devlog/_plan/260926_bug_train_6/000_plan.md b/devlog/_plan/260926_bug_train_6/000_plan.md new file mode 100644 index 00000000000..19b2b73925e --- /dev/null +++ b/devlog/_plan/260926_bug_train_6/000_plan.md @@ -0,0 +1,34 @@ +# Bug-PR merge train batch 6 — plan + +Branch `codex/bug-train-6` from `origin/dev` at `e807e1e27b`. One PR to `dev`; each carried PR is one +squashed commit with the contributor as author and a `Co-authored-by` trailer. Integration fixes are +separate commits after the carried ones. + +## Carried + +| PR | Change | Why it is in | +|---|---|---| +| #5914 (@moseoridev) | `withUniqueToolCallIds` wraps the `openai-chat` adapter and remints only tool-call ids that repeat the caller's history or an id already emitted in the response (`-` suffix). | Real infinite-loop bug with Claude Code behind upstreams that mint positional ids (`call-0-0`). Adapter-local, no credential path. | +| #5882 (@Yum-wu) | Native Chat refetches once on a zero-output mid-stream socket reset, gated by the ambiguous-resend allowance; replacement send releases its retained request copy. | Fixes dropped native Chat turns on reset. Maintainer blocker (retained request bytes after reselection) answered by `318520ebd6`; audit must confirm. | +| #5849 (@lzfxxx) | Test-only isolation and budget fixes (serial lane additions, fixture executable, timeouts, launcher wait for config injection). | Removes known flakes on macOS/Linux runners; no runtime change. Needs conflict resolution against current `dev`. | + +## Left out, with reason + +- #5539 — flips deliberate "preserve caller spelling" tests for unpinned native Chat; a policy change the author marked `[WRONG BRANCH]`. +- #5916, #5831, #5911, #5915 — OAuth / main-account credential paths; they need a written security review (batch 7 candidate). +- #5782, #5800, #5497, #4222 — large feature-sized or conflicting, hygiene failures. + +## Build steps + +1. `git merge --squash pr-` per PR in order 5914, 5882, 5849; resolve conflicts against `dev`. +2. Check file-size ratchet, test-layout registries, structure docs for each carried file set. +3. Integration commits only if a gate fails. + +## Check + +- `bun x tsc --noEmit`, `bun run structure:check`, `bun run privacy:scan`. +- Focused: `tests/adapters/openai/openai-chat-tool-call-id-remint.test.ts`, `tests/responses/chat-native-spend.test.ts`, + `tests/responses/responses-reset-replay.test.ts`, `tests/lib/upstream-retry-zero-output.test.ts`, test-layout guards, + file-size ratchet, the files #5849 touches. +- Exact-head hosted CI on the batch PR, then `gh pr merge --squash --admin --match-head-commit`. + diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index d6a2be5257a..605bbc3fd46 100644 --- a/src/lib/upstream-retry.ts +++ b/src/lib/upstream-retry.ts @@ -904,4 +904,3 @@ export function wrapWithZeroOutputRefetch( }, }); } - diff --git a/tests/responses/chat-native-spend.test.ts b/tests/responses/chat-native-spend.test.ts index 967d520630a..7ee598d32b3 100644 --- a/tests/responses/chat-native-spend.test.ts +++ b/tests/responses/chat-native-spend.test.ts @@ -16,6 +16,7 @@ import { resetProviderRequestPacingForTest } from "../../src/providers/request-p import { estimateTokens } from "../../src/lib/token-estimate"; import { getRequestLogEntries } from "../../src/server/request-log"; import * as stateStores from "../../src/lib/state-store-registrations"; +import { translatorAggregateCurrentBytesForTests } from "../../src/lib/translator-budget"; let previousHome: string | undefined; let testDir = ""; @@ -187,6 +188,8 @@ async function mockResettingChatUpstream( opts: { onSend?: (sendIndex: number, headers: string) => void; replacementFrames?: string; + /** Replacement streams its first frame, then waits on this before finishing the body. */ + holdReplacement?: { firstFrame: string; rest: string; release: Promise }; } = {}, ): Promise<{ baseUrl: string; sends: () => number }> { let sends = 0; @@ -209,6 +212,14 @@ async function mockResettingChatUpstream( socket.destroy(); return; } + const hold = opts.holdReplacement; + if (hold) { + const chunk = (text: string) => `${Buffer.byteLength(text).toString(16)}\r\n${text}\r\n`; + socket.write("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n"); + socket.write(chunk(hold.firstFrame)); + void hold.release.then(() => socket.end(`${chunk(hold.rest)}0\r\n\r\n`)); + return; + } const body = opts.replacementFrames ?? recoveredChatFrames(); socket.end(`HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: ${Buffer.byteLength(body)}\r\nConnection: close\r\n\r\n${body}`); }); @@ -362,3 +373,70 @@ test("native Chat releases retained request bytes after key reselection on repla await stopFixtureServers(); } }); + +test("the replacement send's request copy is released before the replacement stream is relayed", async () => { + // Rotating the key between the sends makes the replacement rebuild and re-charge the request + // copy. The upstream then holds the replacement body after its first frame, so the live + // translator charge is read while the stream is being relayed, not after the turn disposed it. + let liveConfig: OcxConfig | null = null; + const realSetLive = stateStores.setLiveStateStoreConfig; + const liveSpy = spyOn(stateStores, "setLiveStateStoreConfig").mockImplementation(cfg => { + liveConfig = cfg; + return realSetLive(cfg); + }); + let release!: () => void; + const released = new Promise(resolve => { release = resolve; }); + const upstream = await mockResettingChatUpstream({ + onSend: sendIndex => { + if (sendIndex === 1 && liveConfig?.providers.mock) liveConfig.providers.mock.apiKey = "k-rotated"; + }, + holdReplacement: { + firstFrame: `data: ${JSON.stringify({ choices: [{ index: 0, delta: { role: "assistant", content: "HeldFrame" } }] })}\n\n`, + rest: [ + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 5, completion_tokens: 1 } })}\n\n`, + "data: [DONE]\n\n", + ].join(""), + release: released, + }, + }); + saveConfig(mockConfig(upstream.baseUrl, { retryOnReset: {} })); + const server = startServer(0); + activeServer = server; + const requestBytes = 1024 * 1024; + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "mock/test-model", + messages: [{ role: "user", content: "x".repeat(requestBytes) }], + stream: true, + }), + }); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let text = ""; + while (!text.includes("HeldFrame")) { + const { value, done } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + expect(text).toContain("HeldFrame"); + expect(upstream.sends()).toBe(2); + // The accepted inbound body stays observed for the whole turn (~1x requestBytes). Without the + // release, the rebuilt request copy is charged on top of it for the whole relay (~2x). + expect(translatorAggregateCurrentBytesForTests()).toBeLessThan(requestBytes * 1.5); + release(); + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + expect(text).toContain("[DONE]"); + } finally { + release(); + liveSpy.mockRestore(); + await stopFixtureServers(); + } +}); From 5ce30b3b2d791f37fc4d16fd92511ee4a36ab7df Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 26 Sep 2026 18:55:33 +0900 Subject: [PATCH 5/5] fix(upstream-retry): require the resend gate on the zero-output wrapper wrapWithZeroOutputRefetch forwarded its options object, so the source guard that proves every post-header replacement is authorized could not see a gate at the new call. The wrapper now requires authorize in its type and passes it explicitly, and the guard scans wrapWithZeroOutputRefetch call sites as well. Fixes the red tests/lib/ambiguous-resend-composition.test.ts on test 1/4. --- src/lib/upstream-retry.ts | 6 ++++-- tests/lib/ambiguous-resend-composition.test.ts | 4 +++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index 605bbc3fd46..62bd9d963a4 100644 --- a/src/lib/upstream-retry.ts +++ b/src/lib/upstream-retry.ts @@ -867,7 +867,9 @@ export async function refetchAfterProtocolSafeReset( export function wrapWithZeroOutputRefetch( body: ReadableStream, doFetch: ProtocolSafeRefetch, - opts: ProtocolSafeRefetchOptions = {}, + // `authorize` is optional on the shared options but required here: a zero-output replacement + // is always a post-header resend, so every caller must name the gate that weighs it. + opts: ProtocolSafeRefetchOptions & { authorize: () => boolean }, ): ReadableStream { let reader = body.getReader(); let bytesRead = 0; @@ -887,7 +889,7 @@ export function wrapWithZeroOutputRefetch( } catch (err) { if (!retried && bytesRead === 0 && !opts.abortSignal?.aborted) { retried = true; - const replacement = await refetchAfterProtocolSafeReset(doFetch, err, opts); + const replacement = await refetchAfterProtocolSafeReset(doFetch, err, { ...opts, authorize: opts.authorize }); if (replacement?.body) { try { void reader.cancel().catch(() => {}); } catch { /* broken reader; the replacement won */ } reader = replacement.body.getReader(); diff --git a/tests/lib/ambiguous-resend-composition.test.ts b/tests/lib/ambiguous-resend-composition.test.ts index 4b0fd0838ac..f82bf06bcd2 100644 --- a/tests/lib/ambiguous-resend-composition.test.ts +++ b/tests/lib/ambiguous-resend-composition.test.ts @@ -327,7 +327,9 @@ describe("one resend budget across composed recovery legs", () => { let callSites = 0; for await (const relative of new Bun.Glob("**/*.ts").scan({ cwd: srcDir })) { const source = readFileSync(join(srcDir, relative), "utf8"); - for (const match of source.matchAll(/refetchAfterProtocolSafeReset\(/g)) { + // The zero-output wrapper forwards its caller's options to the helper, so its own call + // sites are post-header replacements too and must carry the gate themselves. + for (const match of source.matchAll(/(?:refetchAfterProtocolSafeReset|wrapWithZeroOutputRefetch)\(/g)) { const start = match.index ?? 0; // The declaration itself is not a call site, and the import names it without one. if (/\bfunction\s+$/.test(source.slice(Math.max(0, start - 24), start))) continue;