From d0832cfa793e1179eeeae121942b8523cb61bdba Mon Sep 17 00:00:00 2001 From: pStrikeZ Date: Thu, 24 Sep 2026 12:06:48 +0900 Subject: [PATCH 1/2] fix(responses): reassemble output from closed items, not the terminal snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When Codex CLI triggers remote compaction over WebSocket, Floway failed during native compaction decryption with `OpenAI Responses compact decryption: native compaction returned no compaction output item`. The Codex upstream streams `response.created` (output []) → `response.in_progress` (output []) → `response.output_item.added` (item `{type: "compaction", ...}`, output_index 0) → two `response.compaction.compacting` events → `response.output_item.done` (same item) → `response.completed` with `response.output: []`. `reassembleOpenAIResponsesEvents` returned the terminal event's response verbatim and ignored `response.output_item.done`, so `collectOpenAIResponsesProtocolEventsToResult` yielded an empty output array. Upstream #530 (7a7b2c66a) widened the type guard to `compaction | compaction_summary`, but its tests fed a pre-filled terminal `output`, leaving this stream shape uncovered. The reassembler now follows the rule the client-facing egress already applies: the spec makes the item lifecycle the authority, so once any item closes, the closed items in `output_index` order replace the terminal `output`; a turn that closed nothing keeps the terminal snapshot. That also covers a Codex terminal that omits an assistant message it just closed, so `collectSummaryTurn` in `compact-shim.ts` drops its own copy of that collection. `OpenAIResponsesCompactionItem` types the wire metadata fields, and `response.compaction.compacting` is recognized as a stream event. A Responses Lite fixture that listed a reasoning item only in the terminal now streams its lifecycle like the other items, as Codex CLI itself reads items only from `response.output_item.done`. --- .../interceptors/compact-shim_test.ts | 134 +++++++++++++++++- .../interceptors/compact-shim.ts | 24 +--- .../openai-responses/reassemble_test.ts | 106 +++++++++++++- .../protocols/src/openai-responses/index.ts | 9 ++ .../src/openai-responses/reassemble.ts | 30 +++- .../provider-codex/__tests__/fetch_test.ts | 8 +- 6 files changed, 286 insertions(+), 25 deletions(-) diff --git a/packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/compact-shim_test.ts b/packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/compact-shim_test.ts index 30d7686dcf..d8b7cfd4a0 100644 --- a/packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/compact-shim_test.ts +++ b/packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/compact-shim_test.ts @@ -5,7 +5,7 @@ import type { OpenAIResponsesInvocation } from '../../../../../src/data-plane/ch import { encodeBase64UrlJson } from '../../../../../src/shared/base64url-json.ts'; import { mockChatGatewayCtx } from '../../../../test-utils/gateway-ctx.ts'; import { doneFrame, eventFrame, type ProtocolFrame } from '@floway-dev/protocols/common'; -import { collectOpenAIResponsesProtocolEventsToResult, type CanonicalOpenAIResponsesPayload, type OpenAIResponsesInputItem, type OpenAIResponsesPayload, type OpenAIResponsesResult, type OpenAIResponsesStreamEvent } from '@floway-dev/protocols/openai-responses'; +import { collectOpenAIResponsesProtocolEventsToResult, type CanonicalOpenAIResponsesPayload, type OpenAIResponsesInputItem, type OpenAIResponsesOutputItem, type OpenAIResponsesPayload, type OpenAIResponsesResult, type OpenAIResponsesStreamEvent } from '@floway-dev/protocols/openai-responses'; import { eventResult, type ExecuteResult, type FlagId } from '@floway-dev/provider'; import { assertEquals, stubModelCandidate, testTelemetryModelIdentity } from '@floway-dev/test-utils'; @@ -588,6 +588,138 @@ test('compact decrypt: preserves a generate response envelope for the compaction assertEquals(collected.output[0]?.type, 'compaction'); }); +test('compact decrypt: succeeds when native compaction stream has empty terminal output and intermediate compacting events', async () => { + const inv = makeInvocation( + { + input: [ + { type: 'message', role: 'user', content: 'prior turn' }, + { type: 'compaction_trigger' } as unknown as never, + ], + }, + { flagOn: false, decryptFlagOn: true }, + ); + + const nativeCompactionItem: OpenAIResponsesOutputItem = { + type: 'compaction', + id: 'cmp_native_capture', + encrypted_content: 'OPAQUE_CAPTURE_BLOB', + metadata: { turn_id: 'turn-123' }, + internal_chat_message_metadata_passthrough: { turn_id: 'turn-123' }, + }; + + const inProgressEnvelope: OpenAIResponsesResult = { + id: 'resp_capture', + object: 'response', + model: 'test-upstream-model', + status: 'in_progress', + output: [], + error: null, + incomplete_details: null, + }; + + const completedEnvelope: OpenAIResponsesResult = { + id: 'resp_capture', + object: 'response', + model: 'test-upstream-model', + status: 'completed', + output: [], // Empty terminal output, matching Codex upstream + error: null, + incomplete_details: null, + usage: { input_tokens: 150, output_tokens: 50, total_tokens: 200 }, + }; + + let calls = 0; + const result = await withOpenAIResponsesCompactShim(inv, stubCtx, async () => { + calls += 1; + if (calls === 1) { + return eventResult( + (async function* (): AsyncGenerator> { + yield eventFrame({ type: 'response.created', sequence_number: 0, response: inProgressEnvelope }); + yield eventFrame({ type: 'response.in_progress', sequence_number: 1, response: inProgressEnvelope }); + yield eventFrame({ + type: 'response.output_item.added', + sequence_number: 2, + output_index: 0, + item: nativeCompactionItem, + }); + yield eventFrame({ + type: 'response.compaction.compacting', + sequence_number: 3, + item_id: 'cmp_native_capture', + output_index: 0, + }); + yield eventFrame({ + type: 'response.compaction.compacting', + sequence_number: 4, + item_id: 'cmp_native_capture', + output_index: 0, + }); + yield eventFrame({ + type: 'response.output_item.done', + sequence_number: 5, + output_index: 0, + item: nativeCompactionItem, + }); + yield eventFrame({ type: 'response.completed', sequence_number: 6, response: completedEnvelope }); + yield doneFrame(); + })(), + testTelemetryModelIdentity, + { + finalMetadata: Promise.resolve({ + modelIdentity: testTelemetryModelIdentity, + billableUsage: { input: 150, cacheRead: 0, cacheWrite: 0, cacheWrite1h: 0, output: 50 }, + }), + }, + ); + } + + assertEquals(inv.action, 'generate'); + assertEquals(inv.payload.store, false); + assertEquals(inv.payload.input.length, 3); + const [, replayCompaction] = inv.payload.input; + assertEquals(replayCompaction, nativeCompactionItem); + + const replay = await fakeUpstreamRun('DECRYPTED CAPTURE SUMMARY')(); + if (replay.type !== 'events') throw new Error('expected replay events'); + return { + ...replay, + finalMetadata: Promise.resolve({ + modelIdentity: testTelemetryModelIdentity, + billableUsage: { input: 15, cacheRead: 0, cacheWrite: 0, cacheWrite1h: 0, output: 25 }, + }), + }; + }); + + assertEquals(calls, 2); + if (result.type !== 'events') throw new Error(`expected events branch, got ${result.type}`); + const collected = await collectOpenAIResponsesProtocolEventsToResult(result.events); + assertEquals(collected.output.length, 1); + const decryptedItem = collected.output[0] as unknown as { + type: string; + id: string; + encrypted_content: string; + metadata?: Record; + internal_chat_message_metadata_passthrough?: Record; + }; + assertEquals(decryptedItem.type, 'compaction'); + assertEquals(decryptedItem.id, 'cmp_native_capture'); + assertEquals(decryptedItem.metadata, { turn_id: 'turn-123' }); + assertEquals(decryptedItem.internal_chat_message_metadata_passthrough, { turn_id: 'turn-123' }); + + const expanded = expandShimCompactionItems({ + model: 'test-model', + input: [decryptedItem as unknown as OpenAIResponsesInputItem], + }); + const expandedUserMessage = expanded.input[0] as { + type: string; + role: string; + content: Array<{ type: string; text: string }>; + }; + assertEquals(expandedUserMessage.type, 'message'); + assertEquals(expandedUserMessage.role, 'user'); + assertEquals(expandedUserMessage.content[0].text, 'DECRYPTED CAPTURE SUMMARY'); +}); + // ── Bug 1 — engagement gating ──────────────────────────────────────────────── // // The shim engages when EITHER the per-upstream flag is on OR the candidate's diff --git a/packages/gateway/src/data-plane/chat/openai-responses/interceptors/compact-shim.ts b/packages/gateway/src/data-plane/chat/openai-responses/interceptors/compact-shim.ts index daa84930c8..ca6a01a4b9 100644 --- a/packages/gateway/src/data-plane/chat/openai-responses/interceptors/compact-shim.ts +++ b/packages/gateway/src/data-plane/chat/openai-responses/interceptors/compact-shim.ts @@ -251,15 +251,10 @@ const resultMetadata = async ( ...(result.performance !== undefined ? { performance: result.performance } : {}), }); -// The spec makes the item lifecycle the authority and requires nothing of the -// terminal's `output`; a Codex upstream states an `output` that omits the -// assistant message it just closed. A turn that closed nothing falls back to -// the terminal, as the client-facing egress does. +// The reassembler takes `output` from the closed items, since a Codex upstream +// states a terminal `output` that omits the assistant message it just closed. // https://github.com/openresponses/openresponses/blob/92c12d96d7b61d6d15e2214daa5e9c6000ab6e1c/src/specifications/2026-04-24.mdx#L237 -const summaryTextFrom = (closed: Map, stated: readonly OpenAIResponsesOutputItem[]): string => { - const items = closed.size === 0 - ? stated - : [...closed].sort(([left], [right]) => left - right).map(([, item]) => item); +const summaryTextFrom = (items: readonly OpenAIResponsesOutputItem[]): string => { const parts: string[] = []; for (const item of items) { if (item.type !== 'message') continue; @@ -273,17 +268,8 @@ const summaryTextFrom = (closed: Map, stated: const collectSummaryTurn = async ( result: Extract>, { type: 'events' }>, ): Promise<{ response: OpenAIResponsesResult; text: string }> => { - const closedItems = new Map(); - const observed = (async function* (): AsyncIterable> { - for await (const frame of result.events) { - if (frame.type === 'event' && frame.event.type === 'response.output_item.done') { - closedItems.set(frame.event.output_index, frame.event.item); - } - yield frame; - } - })(); - const response = await collectOpenAIResponsesProtocolEventsToResult(observed); - return { response, text: summaryTextFrom(closedItems, response.output) }; + const response = await collectOpenAIResponsesProtocolEventsToResult(result.events); + return { response, text: summaryTextFrom(response.output) }; }; const buildCompactionEnvelope = (cmpId: string, summaryText: string, upstream: OpenAIResponsesResult): OpenAIResponsesResult => { diff --git a/packages/protocols/__tests__/openai-responses/reassemble_test.ts b/packages/protocols/__tests__/openai-responses/reassemble_test.ts index 584e6d944d..1ec91fefab 100644 --- a/packages/protocols/__tests__/openai-responses/reassemble_test.ts +++ b/packages/protocols/__tests__/openai-responses/reassemble_test.ts @@ -1,6 +1,6 @@ import { test } from 'vitest'; -import type { OpenAIResponsesResult, OpenAIResponsesStreamEvent } from '../../src/openai-responses/index.ts'; +import type { OpenAIResponsesOutputItem, OpenAIResponsesResult, OpenAIResponsesStreamEvent } from '../../src/openai-responses/index.ts'; import { reassembleOpenAIResponsesEvents } from '../../src/openai-responses/reassemble.ts'; import { assertEquals, assertRejects } from '@floway-dev/test-utils'; @@ -113,3 +113,107 @@ test('reassembleOpenAIResponsesEvents throws when stream ends without terminal e await assertRejects(() => reassembleOpenAIResponsesEvents(body), Error, 'terminal'); }); + +test('reassembleOpenAIResponsesEvents reconstructs output in index order when terminal output is empty', async () => { + const item0: OpenAIResponsesOutputItem = { + type: 'message', + id: 'msg_0', + status: 'completed', + role: 'assistant', + content: [{ type: 'output_text', text: 'First item', annotations: [] }], + }; + const item1: OpenAIResponsesOutputItem = { + type: 'compaction', + id: 'cmp_1', + encrypted_content: 'BLOB_1', + }; + + const body = makeEvents([ + { + event: 'response.created', + data: { + type: 'response.created', + response: { id: 'resp_empty_term', object: 'response', model: 'gpt-test', status: 'in_progress', output: [], error: null, incomplete_details: null }, + }, + }, + // Emit out of order to verify sorting by output_index + { + event: 'response.output_item.done', + data: { type: 'response.output_item.done', output_index: 1, item: item1 }, + }, + { + event: 'response.output_item.done', + data: { type: 'response.output_item.done', output_index: 0, item: item0 }, + }, + { + event: 'response.completed', + data: { + type: 'response.completed', + response: { id: 'resp_empty_term', object: 'response', model: 'gpt-test', status: 'completed', output: [], error: null, incomplete_details: null }, + }, + }, + ]); + + const result = await reassembleOpenAIResponsesEvents(body); + assertEquals(result.output, [item0, item1]); +}); + +test('reassembleOpenAIResponsesEvents prefers closed items over a terminal that omits one', async () => { + const reasoning: OpenAIResponsesOutputItem = { type: 'reasoning', id: 'rs_0', summary: [], encrypted_content: 'BLOB_0' }; + const message: OpenAIResponsesOutputItem = { + type: 'message', + id: 'msg_1', + status: 'completed', + role: 'assistant', + content: [{ type: 'output_text', text: 'Closed but unstated', annotations: [] }], + }; + + // A Codex upstream states a terminal `output` that omits the message it just closed. + const body = makeEvents([ + { + event: 'response.output_item.done', + data: { type: 'response.output_item.done', output_index: 0, item: reasoning }, + }, + { + event: 'response.output_item.done', + data: { type: 'response.output_item.done', output_index: 1, item: message }, + }, + { + event: 'response.completed', + data: { + type: 'response.completed', + response: { id: 'resp_partial', object: 'response', model: 'gpt-test', status: 'completed', output: [reasoning], error: null, incomplete_details: null }, + }, + }, + ]); + + const result = await reassembleOpenAIResponsesEvents(body); + assertEquals(result.output, [reasoning, message]); +}); + +test('reassembleOpenAIResponsesEvents preserves terminal snapshot when no closed items observed', async () => { + const fallbackItem: OpenAIResponsesOutputItem = { + type: 'message', + id: 'msg_snap', + status: 'completed', + role: 'assistant', + content: [{ type: 'output_text', text: 'Snapshot only', annotations: [] }], + }; + + const body = makeEvents([ + { + event: 'response.output_text.delta', + data: { type: 'response.output_text.delta', delta: 'Snapshot only' }, + }, + { + event: 'response.completed', + data: { + type: 'response.completed', + response: { id: 'resp_snap', object: 'response', model: 'gpt-test', status: 'completed', output: [fallbackItem], error: null, incomplete_details: null }, + }, + }, + ]); + + const result = await reassembleOpenAIResponsesEvents(body); + assertEquals(result.output, [fallbackItem]); +}); diff --git a/packages/protocols/src/openai-responses/index.ts b/packages/protocols/src/openai-responses/index.ts index a20e3a0d03..92344f5b42 100644 --- a/packages/protocols/src/openai-responses/index.ts +++ b/packages/protocols/src/openai-responses/index.ts @@ -483,6 +483,8 @@ export interface OpenAIResponsesCompactionItem { id?: string | null; encrypted_content: string; created_by?: string; + internal_chat_message_metadata_passthrough?: Record; + metadata?: Record; } export const isOpenAIResponsesCompactionItem = (item: { type: string }): item is OpenAIResponsesCompactionItem => @@ -1373,6 +1375,13 @@ type OpenAIResponsesStreamEventVariant = output_index: number; diff: string; } + // Codex remote-compaction progress event. + // https://github.com/openai/codex/blob/0a2eb4696c/codex-rs/codex-api/src/sse/responses.rs + | { + type: 'response.compaction.compacting'; + item_id: string; + output_index: number; + } | { type: 'response.completed'; response: OpenAIResponsesResult } | { type: 'response.incomplete'; response: OpenAIResponsesResult } | { type: 'response.failed'; response: OpenAIResponsesResult } diff --git a/packages/protocols/src/openai-responses/reassemble.ts b/packages/protocols/src/openai-responses/reassemble.ts index 3e37a34dad..9ced526b2e 100644 --- a/packages/protocols/src/openai-responses/reassemble.ts +++ b/packages/protocols/src/openai-responses/reassemble.ts @@ -1,4 +1,4 @@ -import type { OpenAIResponsesResult, OpenAIResponsesStreamEvent } from './index.ts'; +import type { OpenAIResponsesOutputItem, OpenAIResponsesResult, OpenAIResponsesStreamEvent } from './index.ts'; type OpenAIResponsesReassembleEvent = | OpenAIResponsesStreamEvent @@ -7,7 +7,16 @@ type OpenAIResponsesReassembleEvent = message?: string; }; +// The spec makes the item lifecycle the authority and requires nothing of the +// terminal's `output`. A Codex upstream leaves it empty after closing a remote +// compaction item, and states one that omits the assistant message it just +// closed, so the closed items in `output_index` order replace it, as the +// client-facing egress does. A turn that closed nothing keeps the terminal. +// https://github.com/openai/codex/blob/0a2eb4696c/codex-rs/codex-api/src/sse/responses.rs +// https://github.com/openresponses/openresponses/blob/92c12d96d7b61d6d15e2214daa5e9c6000ab6e1c/src/specifications/2026-04-24.mdx#L313-L337 export async function reassembleOpenAIResponsesEvents(events: AsyncIterable): Promise { + const closedItems = new Map(); + for await (const event of events) { const rawEvent = event as unknown as Record; const type = rawEvent.type as string; @@ -17,8 +26,25 @@ export async function reassembleOpenAIResponsesEvents(events: AsyncIterable 0) { + const output = [...closedItems].sort(([left], [right]) => left - right).map(([, item]) => item); + return { + ...terminalResponse, + output, + }; + } + return terminalResponse; } } diff --git a/packages/provider-codex/__tests__/fetch_test.ts b/packages/provider-codex/__tests__/fetch_test.ts index 76b065f5b8..69dd628842 100644 --- a/packages/provider-codex/__tests__/fetch_test.ts +++ b/packages/provider-codex/__tests__/fetch_test.ts @@ -337,6 +337,8 @@ describe('Codex private Responses wire selection', () => { { type: 'response.output_item.added', output_index: 0, item: wireCall }, future, { type: 'response.output_item.done', output_index: 0, item: wireCall }, + { type: 'response.output_item.added', output_index: 1, item: opaque }, + { type: 'response.output_item.done', output_index: 1, item: opaque }, { type: 'response.completed', response: wireResponse }, ]); upstream.headers.delete('content-type'); @@ -357,7 +359,9 @@ describe('Codex private Responses wire selection', () => { expect(frames[0]).toMatchObject({ type: 'event', event: { item: expectedCall } }); expect(frames[1]).toEqual({ type: 'event', event: future }); expect(frames[2]).toMatchObject({ type: 'event', event: { item: expectedCall } }); - expect(frames[3]).toMatchObject({ + expect(frames[3]).toMatchObject({ type: 'event', event: { item: opaque } }); + expect(frames[4]).toMatchObject({ type: 'event', event: { item: opaque } }); + expect(frames[5]).toMatchObject({ type: 'event', event: { response: { @@ -366,7 +370,7 @@ describe('Codex private Responses wire selection', () => { }, }, }); - expect(frames[4]).toEqual({ type: 'done' }); + expect(frames[6]).toEqual({ type: 'done' }); } else { const collected = await collectOpenAIResponsesProtocolEventsToResult(result.events); expect(collected).toMatchObject({ From 1180c5d9aedde779fae2a10ba1b8e25519ea857b Mon Sep 17 00:00:00 2001 From: pStrikeZ Date: Thu, 24 Sep 2026 12:06:48 +0900 Subject: [PATCH 2/2] fix(responses): answer Codex WebSocket prewarms without an upstream call Codex opens every session with a WebSocket prewarm: a `response.create` carrying the session's instructions and tools with `generate: false`. It waits only for the terminal `response.completed`, then continues from that response's id and sends just the items the prewarm did not carry. Floway forwarded the field verbatim. The Codex HTTP backend rejects it with `{"detail":"Unsupported parameter: generate"}`, so every new Codex session recorded a failed 400 turn and lost its prewarm; a translated target would instead drop the field and run a full, billed generation. The gateway now answers the prewarm itself as the outermost OpenAI Responses interceptor, after serve preparation has resolved the model, expanded any `previous_response_id`, and staged the request's input. The empty completed response commits a snapshot of exactly that input, so the next turn's continuation replays the prewarmed instructions ahead of its own items. As with the Claude Code probe answer, the turn records zero usage and no latency sample. `generate` is now a typed request field. --- .../answer-websocket-warmup_test.ts | 60 ++++++++++++++ .../chat/openai-responses/websocket_test.ts | 79 +++++++++++++++++++ .../interceptors/answer-websocket-warmup.ts | 42 ++++++++++ .../openai-responses/interceptors/index.ts | 4 + .../protocols/src/openai-responses/index.ts | 5 ++ 5 files changed, 190 insertions(+) create mode 100644 packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup_test.ts create mode 100644 packages/gateway/src/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup.ts diff --git a/packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup_test.ts b/packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup_test.ts new file mode 100644 index 0000000000..9a06d175d5 --- /dev/null +++ b/packages/gateway/__tests__/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup_test.ts @@ -0,0 +1,60 @@ +import { test } from 'vitest'; + +import { answerWebSocketWarmup } from '../../../../../src/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup.ts'; +import type { OpenAIResponsesInvocation } from '../../../../../src/data-plane/chat/openai-responses/interceptors/types.ts'; +import { mockChatGatewayCtx } from '../../../../test-utils/gateway-ctx.ts'; +import { doneFrame, type ProtocolFrame } from '@floway-dev/protocols/common'; +import type { CanonicalOpenAIResponsesPayload, OpenAIResponsesStreamEvent } from '@floway-dev/protocols/openai-responses'; +import { eventResult } from '@floway-dev/provider'; +import { assertEquals, stubModelCandidate, testTelemetryModelIdentity } from '@floway-dev/test-utils'; + +const stubCtx = mockChatGatewayCtx(); + +const invocation = (payload: CanonicalOpenAIResponsesPayload): OpenAIResponsesInvocation => ({ + payload, + candidate: stubModelCandidate(), + targetApi: 'openaiResponses', + headers: new Headers(), + action: 'generate', +}); + +const payload = (generate?: boolean): CanonicalOpenAIResponsesPayload => ({ + model: 'gpt-test', + input: [{ type: 'message', role: 'developer', content: 'Base instructions' }], + ...(generate === undefined ? {} : { generate }), +}); + +test('answers a generate:false prewarm with an empty completed response and no upstream call', async () => { + let upstreamCalls = 0; + const result = await answerWebSocketWarmup(invocation(payload(false)), stubCtx, () => { + upstreamCalls++; + return Promise.resolve(eventResult((async function* () { yield doneFrame(); })(), testTelemetryModelIdentity)); + }); + + assertEquals(upstreamCalls, 0); + if (result.type !== 'events') throw new Error(`expected events, got ${result.type}`); + const frames: ProtocolFrame[] = []; + for await (const frame of result.events) frames.push(frame); + const types = frames.map(frame => frame.type === 'event' ? frame.event.type : frame.type); + assertEquals(types, ['response.created', 'response.in_progress', 'response.completed', 'done']); + const terminal = frames[2]; + if (terminal?.type !== 'event' || terminal.event.type !== 'response.completed') throw new Error('expected a completed terminal'); + assertEquals(terminal.event.response.status, 'completed'); + assertEquals(terminal.event.response.output, []); + assertEquals(terminal.event.response.model, 'gpt-test'); + assertEquals(terminal.event.response.usage, { input_tokens: 0, output_tokens: 0, total_tokens: 0 }); +}); + +for (const generate of [undefined, true]) { + test(`passes a generate:${generate} request through to the upstream`, async () => { + let upstreamCalls = 0; + const upstream = eventResult((async function* () { yield doneFrame(); })(), testTelemetryModelIdentity); + const result = await answerWebSocketWarmup(invocation(payload(generate)), stubCtx, () => { + upstreamCalls++; + return Promise.resolve(upstream); + }); + + assertEquals(upstreamCalls, 1); + assertEquals(result, upstream); + }); +} diff --git a/packages/gateway/__tests__/data-plane/chat/openai-responses/websocket_test.ts b/packages/gateway/__tests__/data-plane/chat/openai-responses/websocket_test.ts index 815fbfd597..60810c3157 100644 --- a/packages/gateway/__tests__/data-plane/chat/openai-responses/websocket_test.ts +++ b/packages/gateway/__tests__/data-plane/chat/openai-responses/websocket_test.ts @@ -852,6 +852,85 @@ test('OpenAI Responses WebSocket store:false keeps session snapshots without dur ); }); +test('OpenAI Responses WebSocket answers a Codex generate:false prewarm locally and continues from it', async () => { + const { apiKey } = await setupAppTest(); + const upstreamBodies: Record[] = []; + + await withMockedFetch( + async request => { + const url = new URL(request.url); + if (url.hostname === 'update.code.visualstudio.com') return jsonResponse(['1.110.1']); + if (url.pathname === '/copilot_internal/v2/token') { + return jsonResponse({ token: 'copilot-access-token', expires_at: 4102444800, refresh_in: 3600, endpoints: { api: 'https://api.individual.githubcopilot.com' } }); + } + if (url.pathname === '/models') { + return jsonResponse(copilotModels([{ id: 'gpt-direct-responses', supported_endpoints: ['/responses'] }])); + } + if (url.pathname === '/responses') { + upstreamBodies.push(JSON.parse(await request.text()) as Record); + return sseOpenAIResponsesResponse({ + id: 'resp_ws_after_prewarm', + object: 'response', + model: 'gpt-direct-responses', + status: 'completed', + output_text: 'answer', + output: [{ + id: 'assistant_ws_after_prewarm', + type: 'message', + role: 'assistant', + status: 'completed', + content: [{ type: 'output_text', text: 'answer', annotations: [] }], + }], + }); + } + throw new Error(`Unhandled fetch ${request.url}`); + }, + async () => await withWorkerWebSocketRuntime(async () => { + const client = await connectOpenAIResponsesWebSocket(apiKey.key); + const prewarmTerminal = waitForMessages(client, messages => messages.some(isTerminalResponseEvent)); + client.send(JSON.stringify({ + type: 'response.create', + response: { + model: 'gpt-direct-responses', + input: [{ type: 'message', role: 'developer', content: 'Base instructions' }], + store: false, + generate: false, + }, + })); + const prewarmMessages = await prewarmTerminal; + const prewarmResponseId = terminalResponseId(prewarmMessages); + const prewarmCompleted = prewarmMessages.find(isTerminalResponseEvent) as { type?: unknown; response?: { status?: unknown; output?: Array<{ type: string }> } }; + assertEquals(prewarmCompleted.type, 'response.completed'); + assertEquals(prewarmCompleted.response?.status, 'completed'); + // Nothing was generated; the only item is the affinity carrier every + // routed turn states. + assertEquals(prewarmCompleted.response?.output?.map(item => item.type), ['reasoning']); + assertEquals(upstreamBodies.length, 0); + + const turnTerminal = waitForMessages(client, messages => messages.some(isTerminalResponseEvent)); + client.send(JSON.stringify({ + type: 'response.create', + response: { + model: 'gpt-direct-responses', + previous_response_id: prewarmResponseId, + input: 'first question', + store: false, + }, + })); + await turnTerminal; + + assertEquals(upstreamBodies.length, 1); + const body = upstreamBodies[0] as { generate?: unknown; previous_response_id?: unknown; input: Array<{ type: string; role?: string; content?: unknown }> }; + assertEquals(Object.hasOwn(body, 'generate'), false); + assertEquals(body.previous_response_id, undefined); + assertEquals(body.input.map(item => [item.type, item.role, item.content]), [ + ['message', 'developer', 'Base instructions'], + ['message', 'user', 'first question'], + ]); + }), + ); +}); + test('OpenAI Responses WebSocket evicts a failed continuation target so the next attempt reports previous_response_not_found', async () => { const { apiKey } = await setupAppTest(); let responseCalls = 0; diff --git a/packages/gateway/src/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup.ts b/packages/gateway/src/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup.ts new file mode 100644 index 0000000000..e27ea946cc --- /dev/null +++ b/packages/gateway/src/data-plane/chat/openai-responses/interceptors/answer-websocket-warmup.ts @@ -0,0 +1,42 @@ +import type { OpenAIResponsesInterceptor } from './types.ts'; +import { telemetryModelIdentity } from '../../../shared/telemetry/attribution.ts'; +import { syntheticEventsFromResult } from '../items/output.ts'; +import type { OpenAIResponsesResult } from '@floway-dev/protocols/openai-responses'; +import { eventResult, providerModelOf } from '@floway-dev/provider'; + +// Codex opens every session with a WebSocket prewarm: a `response.create` that +// carries the session's instructions and tools with `generate: false`. It is +// connection setup rather than inference, so Codex waits only for the terminal +// `response.completed` and then continues from that response's id, sending +// just the items the prewarm did not already carry. +// https://github.com/openai/codex/blob/6989c6548b3737f108e2bb5ae1171b1d2032e30c/codex-rs/core/src/client.rs#L17-L18 +// https://github.com/openai/codex/blob/6989c6548b3737f108e2bb5ae1171b1d2032e30c/codex-rs/core/src/client.rs#L2025 +// https://github.com/openai/codex/blob/6989c6548b3737f108e2bb5ae1171b1d2032e30c/codex-rs/core/src/client.rs#L2181-L2184 +// +// No upstream call can stand in for it. The Codex HTTP backend rejects the +// field with `{"detail":"Unsupported parameter: generate"}`, and a translated +// target drops it and runs a full, billed generation. The gateway answers the +// prewarm itself: by now serve preparation has resolved the model, expanded +// any `previous_response_id`, and staged this request's input, so the empty +// completed response commits a snapshot of exactly that input and the next +// turn's continuation replays it. An id no upstream serves still fails before +// this runs. +// +// No `performance` context on the result: a turn that never dialed the +// upstream has no latency to report. The usage row still lands, at zero, so the +// request stays visible in the dashboard. +export const answerWebSocketWarmup: OpenAIResponsesInterceptor = async (ctx, _gatewayCtx, run) => { + if (ctx.payload.generate !== false) return await run(); + const result: OpenAIResponsesResult = { + // Replaced by the client-output boundary's own response id. + id: '', + object: 'response', + model: ctx.payload.model, + status: 'completed', + output: [], + error: null, + incomplete_details: null, + usage: { input_tokens: 0, output_tokens: 0, total_tokens: 0 }, + }; + return eventResult(syntheticEventsFromResult(result), telemetryModelIdentity(ctx.candidate, providerModelOf(ctx.candidate).id)); +}; diff --git a/packages/gateway/src/data-plane/chat/openai-responses/interceptors/index.ts b/packages/gateway/src/data-plane/chat/openai-responses/interceptors/index.ts index d08ddc412c..bdd374f045 100644 --- a/packages/gateway/src/data-plane/chat/openai-responses/interceptors/index.ts +++ b/packages/gateway/src/data-plane/chat/openai-responses/interceptors/index.ts @@ -1,3 +1,4 @@ +import { answerWebSocketWarmup } from './answer-websocket-warmup.ts'; import { withRoleCompatibilityApplied } from './apply-role-compatibility.ts'; import { withOpenAIResponsesCollaborationShim } from './collaboration-shim.ts'; import { withOpenAIResponsesCompactShim } from './compact-shim.ts'; @@ -21,6 +22,8 @@ import { withVendorQwenOpenAIResponsesNormalize } from './vendor-qwen-normalize. // after pairwise translation has finished. // // Order matters: earlier entries wrap later ones. +// - answerWebSocketWarmup: runs outermost so a `generate: false` prewarm is +// answered before any shim or upstream call can turn it into a generation. // - withOpenAIResponsesCompactShim: runs outermost so the action pivot // ('compact' → 'generate' for the inner summarization turn) is visible // to every downstream interceptor + the provider terminal. Also @@ -53,6 +56,7 @@ import { withVendorQwenOpenAIResponsesNormalize } from './vendor-qwen-normalize. // the role-compatibility entry so each gets the final say on the outbound wire // body. export const openaiResponsesInterceptors: readonly OpenAIResponsesInterceptor[] = [ + answerWebSocketWarmup, withOpenAIResponsesCompactShim, withOpenAIResponsesCollaborationShim, withOpenAIResponsesServerToolShim([ diff --git a/packages/protocols/src/openai-responses/index.ts b/packages/protocols/src/openai-responses/index.ts index 92344f5b42..cafba4e273 100644 --- a/packages/protocols/src/openai-responses/index.ts +++ b/packages/protocols/src/openai-responses/index.ts @@ -33,6 +33,11 @@ export interface OpenAIResponsesPayload { metadata?: Record | null; stream?: boolean | null; store?: boolean | null; + // `false` asks for a prewarm: a response that records this request's + // context without generating, which the next request continues from via + // `previous_response_id`. Codex sends it on its WebSocket transport. + // https://github.com/openai/codex/blob/6989c6548b3737f108e2bb5ae1171b1d2032e30c/codex-rs/codex-api/src/common.rs#L355 + generate?: boolean | null; parallel_tool_calls?: boolean | null; reasoning?: { effort?: string;