diff --git a/fixtures/codex-sse/README.md b/fixtures/codex-sse/README.md index 7f652e6f9..bd0c581c5 100644 --- a/fixtures/codex-sse/README.md +++ b/fixtures/codex-sse/README.md @@ -15,6 +15,7 @@ or user content — placeholders only. | `failed.json` | `response.failed` with error message | | `error.json` | Top-level stream `error` event | | `lifecycle-ignored.json` | `response.created` / `in_progress` / content_part / `*.done` envelopes | +| `parallel-tools-arguments-on-done.json` | Parallel function_calls whose arguments arrive only on `*.done` | Loaded by `src/provider/codex-responses-sse.test.ts`. diff --git a/fixtures/codex-sse/parallel-tools-arguments-on-done.json b/fixtures/codex-sse/parallel-tools-arguments-on-done.json new file mode 100644 index 000000000..9df8a74e6 --- /dev/null +++ b/fixtures/codex-sse/parallel-tools-arguments-on-done.json @@ -0,0 +1,83 @@ +[ + { + "type": "response.created", + "response": { + "id": "resp_fixture_parallel", + "status": "in_progress", + "model": "gpt-fixture-codex" + } + }, + { + "type": "response.output_item.added", + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_fixture_a", + "call_id": "call_fixture_a", + "name": "read_file", + "arguments": "", + "status": "in_progress" + } + }, + { + "type": "response.function_call_arguments.done", + "item_id": "fc_fixture_a", + "output_index": 0, + "arguments": "{\"path\":\"a.txt\"}" + }, + { + "type": "response.output_item.done", + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_fixture_a", + "call_id": "call_fixture_a", + "name": "read_file", + "arguments": "{\"path\":\"a.txt\"}", + "status": "completed" + } + }, + { + "type": "response.output_item.added", + "output_index": 1, + "item": { + "type": "function_call", + "id": "fc_fixture_b", + "call_id": "call_fixture_b", + "name": "read_file", + "arguments": "", + "status": "in_progress" + } + }, + { + "type": "response.function_call_arguments.done", + "item_id": "fc_fixture_b", + "output_index": 1, + "arguments": "{\"path\":\"b.txt\"}" + }, + { + "type": "response.output_item.done", + "output_index": 1, + "item": { + "type": "function_call", + "id": "fc_fixture_b", + "call_id": "call_fixture_b", + "name": "read_file", + "arguments": "{\"path\":\"b.txt\"}", + "status": "completed" + } + }, + { + "type": "response.completed", + "response": { + "id": "resp_fixture_parallel", + "status": "completed", + "usage": { + "input_tokens": 80, + "output_tokens": 20, + "input_tokens_details": { "cached_tokens": 0 }, + "output_tokens_details": { "reasoning_tokens": 0 } + } + } + } +] diff --git a/src/provider/codex-responses-sse.test.ts b/src/provider/codex-responses-sse.test.ts index db5ff95e5..1e13e9369 100644 --- a/src/provider/codex-responses-sse.test.ts +++ b/src/provider/codex-responses-sse.test.ts @@ -212,6 +212,26 @@ describe("codex-sse fixtures (golden parse)", () => { expect(String(error)).toMatch(/fixture stream error \(sanitized\)/); }); + test("parallel calls whose arguments arrive only on done still carry their arguments", () => { + const out = parseFixture("parallel-tools-arguments-on-done.json"); + + expect(eventTypes(out)).toEqual([ + "inference.tool_call.start", + "inference.tool_call.delta", + "inference.tool_call.start", + "inference.tool_call.delta", + "inference.usage", + ]); + expect(out[1]).toMatchObject({ + type: "inference.tool_call.delta", + data: { callId: "0", argumentFragment: '{"path":"a.txt"}', index: 0 }, + }); + expect(out[3]).toMatchObject({ + type: "inference.tool_call.delta", + data: { callId: "1", argumentFragment: '{"path":"b.txt"}', index: 1 }, + }); + }); + test("lifecycle envelopes alone produce no InferenceEvents", () => { const out = parseFixture("lifecycle-ignored.json"); expect(out).toEqual([]); diff --git a/src/provider/codex-responses.ts b/src/provider/codex-responses.ts index 609bea5eb..c913b7127 100644 --- a/src/provider/codex-responses.ts +++ b/src/provider/codex-responses.ts @@ -5,6 +5,7 @@ import { CODEX_SESSION_ID_OPTION, withCodexContentTypeRepair, } from "@corbits/codex-provider"; +import { type } from "arktype"; import type { AdapterFactory, ProviderAdapter } from "@intx/inference"; import type { InferenceOptions } from "@intx/types/runtime"; import { ENVIRONMENT_TAG_NAME, PRODUCT_NAME } from "../branding.js"; @@ -73,11 +74,70 @@ export function withParallelToolCalls( }; } +const FunctionCallItemDone = type({ + type: "'response.output_item.done'", + item: { type: "'function_call'", call_id: "string", arguments: "string" }, +}); + +// With parallel calls on, some Codex backends send a call's arguments only on +// output_item.done and never as deltas. The packaged parser reads deltas +// alone, so without this replay the harness assembles those calls as `{}` +// (CL-10582). +function withArgumentsFromDone(adapter: ProviderAdapter): ProviderAdapter { + let indexByCallId = new Map(); + let streamedIndices = new Set(); + return { + ...adapter, + buildRequest: (messages, model, options) => { + indexByCallId = new Map(); + streamedIndices = new Set(); + return adapter.buildRequest(messages, model, options); + }, + parseResponse: (sseData) => { + const events = adapter.parseResponse(sseData); + for (const event of events) { + if (event.type === "inference.tool_call.start") { + if (event.data.index !== undefined) { + indexByCallId.set(event.data.callId, event.data.index); + } + } else if (event.type === "inference.tool_call.delta") { + if (event.data.index !== undefined) { + streamedIndices.add(event.data.index); + } + } + } + if (!sseData.includes('"response.output_item.done"')) return events; + const done = FunctionCallItemDone(JSON.parse(sseData)); + if (done instanceof type.errors || done.item.arguments.length === 0) { + return events; + } + const index = indexByCallId.get(done.item.call_id); + if (index === undefined || streamedIndices.has(index)) return events; + streamedIndices.add(index); + return [ + ...events, + { + type: "inference.tool_call.delta", + seq: 0, + data: { + callId: String(index), + argumentFragment: done.item.arguments, + partial: { text: "" }, + index, + }, + }, + ]; + }, + }; +} + // CodexQuirks are required by the package factory; Responses option keys are // not part of that bag, so host reasoning_effort is aliased at buildRequest. export const createCodexResponsesAdapter: AdapterFactory = (source) => withHostReasoningEffort( withParallelToolCalls( - createPackageCodexResponsesAdapter(source, CODEX_QUIRKS), + withArgumentsFromDone( + createPackageCodexResponsesAdapter(source, CODEX_QUIRKS), + ), ), );