From 645180ceaf123c954ab5306969cf82da83566648 Mon Sep 17 00:00:00 2001 From: Maple Date: Fri, 4 Sep 2026 01:55:10 +0800 Subject: [PATCH] fix(responses): repair sparse Grok terminal output --- src/server/responses-snapshot-repair.ts | 327 ++++++++++++++++ src/server/responses/core.ts | 10 + structure/04_transports-and-sidecars.md | 29 ++ .../responses-snapshot-repair-server.test.ts | 145 ++++++- .../responses-snapshot-repair.test.ts | 359 ++++++++++++++++++ 5 files changed, 866 insertions(+), 4 deletions(-) diff --git a/src/server/responses-snapshot-repair.ts b/src/server/responses-snapshot-repair.ts index 9ae137888e..109ef861ac 100644 --- a/src/server/responses-snapshot-repair.ts +++ b/src/server/responses-snapshot-repair.ts @@ -597,6 +597,333 @@ export function createResponsesSnapshotBlockRewrite( return rewrite; } +type SparseTerminalOpenItem = { + type: string; + id?: string; + sourceBytes: number; +}; + +type SparseTerminalCompletedItem = RetainedOutputItem & { + visibleToGrok: boolean; +}; + +const MAX_GROK_OPEN_ITEM_IDENTITY_BYTES = MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES; + +const GROK_TERMINAL_OUTPUT_ITEM_TYPES = new Set([ + "message", + "reasoning", + "function_call", + "custom_tool_call", + "web_search_call", + "code_interpreter_call", + "mcp_call", +]); + +function hasValidOptionalId(item: Record): boolean { + return !("id" in item) + || (typeof item.id === "string" && item.id.trim().length > 0); +} + +function hasCompletedStatusWhenPresent(item: Record): boolean { + return !("status" in item) || item.status === "completed"; +} + +function isNullableString(value: unknown): boolean { + return value === null || typeof value === "string"; +} + +function isValidOutputMessagePart(part: unknown): boolean { + if (!isPlainObject(part)) return false; + if (part.type === "output_text") { + return typeof part.text === "string" + && (!("annotations" in part) || Array.isArray(part.annotations)) + && (!("logprobs" in part) || part.logprobs === null || Array.isArray(part.logprobs)); + } + return part.type === "refusal" && typeof part.refusal === "string"; +} + +function isValidReasoningPart(part: unknown, type: "summary_text" | "reasoning_text"): boolean { + return isPlainObject(part) && part.type === type && typeof part.text === "string"; +} + +function isValidWebSearchAction(value: unknown): boolean { + if (!isPlainObject(value)) return false; + if (value.type === "search") { + return typeof value.query === "string" + && (!("sources" in value) || value.sources === null || (Array.isArray(value.sources) + && value.sources.every(source => isPlainObject(source) + && typeof source.type === "string" && typeof source.url === "string"))); + } + if (value.type === "open_page") { + return !("url" in value) || isNullableString(value.url); + } + if (value.type === "find" || value.type === "find_in_page") { + return typeof value.url === "string" && typeof value.pattern === "string"; + } + return false; +} + +function isValidCodeInterpreterOutput(value: unknown): boolean { + return isPlainObject(value) + && ((value.type === "logs" && typeof value.logs === "string") + || (value.type === "image" && typeof value.url === "string")); +} + +/** + * Validate the pre-field-backfill item carried by a real output_item.done. + * Missing ids, message status, and output-text annotations are allowed because + * the always-on field backfill safely supplies only those schema defaults. + * Contradictory values and semantic content repairs are never accepted as + * proof that an empty terminal snapshot was sparse. + */ +function trustedGrokCompletedItem( + item: Record, +): { visibleToGrok: boolean } | null { + if (!hasValidOptionalId(item) || !hasCompletedStatusWhenPresent(item)) return null; + + if (item.type === "message") { + if (item.role !== "assistant" || !Array.isArray(item.content)) return null; + if (!(item.content as unknown[]).every(isValidOutputMessagePart)) return null; + if ("phase" in item && item.phase !== "commentary" && item.phase !== "final_answer") return null; + return { + // grok-build currently turns only output_text parts into final Assistant + // content; refusal parts do not satisfy its visible-content gate. + visibleToGrok: item.content.some(part => isPlainObject(part) + && part.type === "output_text" && typeof part.text === "string" && part.text.length > 0), + }; + } + + if (item.type === "reasoning") { + if (!Array.isArray(item.summary) + || !item.summary.every(part => isValidReasoningPart(part, "summary_text"))) return null; + if ("content" in item && item.content !== null + && (!Array.isArray(item.content) + || !item.content.every(part => isValidReasoningPart(part, "reasoning_text")))) return null; + if ("encrypted_content" in item && !isNullableString(item.encrypted_content)) return null; + return { visibleToGrok: false }; + } + + if (item.type === "function_call") { + if (typeof item.call_id !== "string" || item.call_id.trim().length === 0 + || typeof item.name !== "string" || item.name.trim().length === 0 + || typeof item.arguments !== "string") return null; + return { visibleToGrok: true }; + } + + if (item.type === "custom_tool_call") { + if (typeof item.call_id !== "string" || item.call_id.trim().length === 0 + || typeof item.name !== "string" || item.name.trim().length === 0 + || typeof item.input !== "string") return null; + return { visibleToGrok: false }; + } + + if (item.type === "web_search_call") { + if (item.status !== "completed" || !isValidWebSearchAction(item.action)) return null; + return { visibleToGrok: false }; + } + + if (item.type === "code_interpreter_call") { + if (item.status !== "completed" + || typeof item.container_id !== "string" || item.container_id.trim().length === 0 + || ("code" in item && !isNullableString(item.code)) + || ("outputs" in item && item.outputs !== null + && (!Array.isArray(item.outputs) || !item.outputs.every(isValidCodeInterpreterOutput)))) return null; + return { visibleToGrok: false }; + } + + if (item.type === "mcp_call") { + if (typeof item.arguments !== "string" + || typeof item.name !== "string" || item.name.trim().length === 0 + || typeof item.server_label !== "string" || item.server_label.trim().length === 0 + || ("approval_request_id" in item && !isNullableString(item.approval_request_id)) + || ("error" in item && !isNullableString(item.error)) + || ("output" in item && !isNullableString(item.output))) return null; + return { visibleToGrok: false }; + } + + return null; +} + +function plausibleGrokOpenItem( + item: Record, +): Omit | null { + const type = typeof item.type === "string" ? item.type : ""; + if (!GROK_TERMINAL_OUTPUT_ITEM_TYPES.has(type) || !hasValidOptionalId(item)) return null; + if ("status" in item && item.status !== "in_progress") return null; + if (type === "message") { + if ("role" in item && item.role !== "assistant") return null; + if ("content" in item && !Array.isArray(item.content)) return null; + } + return { + type, + ...(typeof item.id === "string" ? { id: item.id } : {}), + }; +} + +/** + * Narrow client repair for grok-build's Responses consumer. + * + * grok-build streams text deltas but builds its durable Assistant item only + * from response.completed.response.output. Some native Responses streams put + * the durable items in output_item.done and finish with a missing or explicit + * empty output. Reconstruct only from real, unique, contiguous, bounded done + * events whose raw semantics are already valid. Any ambiguity stays byte-level + * fail-closed; the provider-opt-in snapshot repair above is unchanged. + */ +export function createGrokResponsesSparseTerminalBlockRewrite( + budget?: TranslatorBudget, +): SseBlockRewrite { + const openItems = new Map(); + const completedItems = new Map(); + let aggregateItemBytes = 0; + let aggregateOpenItemBytes = 0; + let tainted = false; + let hasVisibleOutput = false; + + const clearRetained = (): void => { + const retainedBytes = aggregateItemBytes + aggregateOpenItemBytes; + if (retainedBytes > 0) { + budget?.releaseRetained(retainedBytes, { kind: "retained_collectors" }); + } + openItems.clear(); + completedItems.clear(); + aggregateItemBytes = 0; + aggregateOpenItemBytes = 0; + hasVisibleOutput = false; + }; + + const reset = (): void => { + clearRetained(); + tainted = false; + }; + + const taintAndRelease = (): void => { + clearRetained(); + tainted = true; + }; + + const retainCompletedItem = ( + index: number, + item: Record, + visibleToGrok: boolean, + ): void => { + if (tainted) return; + const sourceBytes = Buffer.byteLength(JSON.stringify(item), "utf8"); + if (sourceBytes > MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES + || completedItems.size >= MAX_COMPLETED_OUTPUT_ITEMS + || aggregateItemBytes + sourceBytes > MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES) { + taintAndRelease(); + return; + } + budget?.chargeRetained(sourceBytes, { kind: "retained_collectors" }); + completedItems.set(index, { item, sourceBytes, visibleToGrok }); + aggregateItemBytes += sourceBytes; + hasVisibleOutput = hasVisibleOutput || visibleToGrok; + }; + + const closeOpenItem = (index: number): void => { + const open = openItems.get(index); + if (!open) return; + openItems.delete(index); + aggregateOpenItemBytes -= open.sourceBytes; + budget?.releaseRetained(open.sourceBytes, { kind: "retained_collectors" }); + }; + + const rewrite: SseBlockRewrite = (block: string): readonly string[] => { + const payload = sseDataPayload(block); + if (payload === null) return [block]; + if (payload === "[DONE]") { + reset(); + return [block]; + } + + let parsed: unknown; + try { + parsed = JSON.parse(payload); + } catch { + taintAndRelease(); + return [block]; + } + if (!isPlainObject(parsed) || typeof parsed.type !== "string") { + taintAndRelease(); + return [block]; + } + + const type = parsed.type; + const outputIndex = Number.isInteger(parsed.output_index) && (parsed.output_index as number) >= 0 + ? parsed.output_index as number + : undefined; + + if (type === "response.output_item.added") { + const open = isPlainObject(parsed.item) ? plausibleGrokOpenItem(parsed.item) : null; + if (outputIndex === undefined || !open + || openItems.has(outputIndex) || completedItems.has(outputIndex) + || openItems.size >= MAX_OPEN_ITEMS) { + taintAndRelease(); + } else if (!tainted) { + const sourceBytes = Buffer.byteLength(JSON.stringify(open), "utf8"); + if (sourceBytes > MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES + || aggregateOpenItemBytes + sourceBytes > MAX_GROK_OPEN_ITEM_IDENTITY_BYTES) { + taintAndRelease(); + } else { + budget?.chargeRetained(sourceBytes, { kind: "retained_collectors" }); + openItems.set(outputIndex, { ...open, sourceBytes }); + aggregateOpenItemBytes += sourceBytes; + } + } + return [block]; + } + + if (type === "response.output_item.done") { + const item = isPlainObject(parsed.item) ? parsed.item : null; + const proof = item ? trustedGrokCompletedItem(item) : null; + if (outputIndex === undefined || !proof || completedItems.has(outputIndex)) { + taintAndRelease(); + return [block]; + } + const open = openItems.get(outputIndex); + const doneId = typeof item!.id === "string" ? item!.id : undefined; + if (open && (open.type !== item!.type || open.id !== doneId)) { + taintAndRelease(); + return [block]; + } + closeOpenItem(outputIndex); + retainCompletedItem(outputIndex, item!, proof.visibleToGrok); + return [block]; + } + + const isTerminal = type === "response.completed" + || type === "response.failed" + || type === "response.incomplete"; + if (!isTerminal) return [block]; + + let out = block; + if (type === "response.completed" && !tainted && isPlainObject(parsed.response)) { + const response = parsed.response; + const output = response.output; + const terminalStatusConsistent = !("status" in response) || response.status === "completed"; + const outputIsAuthoritative = Array.isArray(output) && output.length > 0; + const outputIsSparse = !("output" in response) + || (Array.isArray(output) && output.length === 0); + if (!outputIsAuthoritative && outputIsSparse && terminalStatusConsistent + && completedItems.size > 0 && openItems.size === 0 && hasVisibleOutput) { + const ordered = [...completedItems.entries()].sort(([left], [right]) => left - right); + if (ordered.every(([index], position) => index === position)) { + out = jsonBlock({ + ...parsed, + response: { ...response, output: ordered.map(([, retained]) => retained.item) }, + }); + } + } + } + reset(); + return [out]; + }; + + rewrite.dispose = reset; + return rewrite; +} + /** Repair a non-streaming Responses JSON object without changing raw inspection state. */ export function repairResponsesSnapshotJson(payload: string, requestBody?: unknown): string { let response: unknown; diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 366074b49a..67392e56a5 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -360,6 +360,7 @@ import { type UpstreamHostAdmissionLease, } from "../../codex/upstream-host-health"; import { + createGrokResponsesSparseTerminalBlockRewrite, createResponsesSnapshotBlockRewrite, hasResponsesSnapshotRepair, repairResponsesSnapshotJson, @@ -4807,6 +4808,12 @@ async function handleResponsesInner( ) : upstreamResponse.body; const repairConfig = route.provider.responsesItemIdRepair; + // Grok Build renders deltas live but reconstructs its durable assistant + // turn from the completed response snapshot. Native Responses streams + // may instead carry the complete items in output_item.done, so the + // generated Grok marker enables a separate, strict terminal-only repair. + // The provider's broader snapshot/lifecycle repair remains opt-in. + const grokClientSnapshotRepairEnabled = logCtx.surface === "grok"; const snapshotRepairEnabled = hasResponsesSnapshotRepair(route.provider.responsesSnapshotRepair); const githubCopilotRepairEnabled = route.providerName === "github-copilot"; const responseModelRewrite = parsed._responseModelId !== undefined @@ -4855,6 +4862,9 @@ async function handleResponsesInner( githubCopilotRepairEnabled ? createGithubCopilotResponsesBlockRewrite(translatorBudget) : undefined, + grokClientSnapshotRepairEnabled + ? createGrokResponsesSparseTerminalBlockRewrite(translatorBudget) + : undefined, snapshotRepairEnabled ? createResponsesSnapshotBlockRewrite(outboundRequestBody, translatorBudget) : undefined, diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index e1068b68ae..24768aaf64 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -1060,6 +1060,35 @@ Grounded in the open-sourced official client (xai-org/grok-build); unit + eviden `fetchWithHeaderTimeout` takes an executor so provider fetch wrappers stay inside the timeout race. +The generated Grok client marker also enables a client-facing sparse-terminal repair for native +Responses streams. Grok Build renders text deltas immediately but derives its durable assistant +turn from `response.completed.response.output`; an OpenAI-compatible stream may instead place the +complete items in `response.output_item.done` and finish with an explicit empty output array. For +that marked client only, OpenCodex uses a terminal-only tracker: it retains bounded, contiguous, +unique and semantically valid raw completed items, then backfills a missing or empty terminal +snapshot. It never promotes locally synthesized or merely repaired items. Unmarked callers continue +to treat an explicit empty array as authoritative. Within this marked client-facing repair, +malformed, gapped, oversized, contradictory, failed, or incomplete streams stay fail-closed. + +[Decision Log] +- 목적과 의도: Prevent Grok Build from classifying a visibly streamed answer as empty and replaying + the same billable turn when the terminal snapshot is sparse. +- 기존 구현 및 제약 조건: OpenCodex already reconstructed missing terminal output for provider + opt-ins, but preserved explicit empty arrays; Grok Build discarded ordinary completed-item events + when constructing its final conversation response. +- 검토한 주요 대안: Change every caller's empty-array semantics; accept a turn merely because a + text delta was visible; reuse the provider's broader lifecycle synthesis; add a strict repair at + the generated Grok client boundary. +- 선택한 방식: Use the existing generated client marker to opt Grok into a terminal-only repair and + backfill only from unique, contiguous, bounded real done items whose raw semantics are valid. +- 다른 대안 대신 이 방식을 선택한 이유: A global rewrite would alter valid provider semantics, + while accepting deltas without durable items would leave persistence and continuation empty. The + marker is already the client-specific compatibility boundary; keeping the provider repair separate + also prevents synthesized or permissively normalized items from overriding an explicit empty terminal. +- 장점, 단점 및 영향: Grok receives one durable completed answer without a paid retry; ordinary + clients remain byte-semantics compatible. The proxy retains bounded item state for marked streams + and intentionally refuses ambiguous reconstruction. + ## Kiro client parallel-tool hint Kiro's wire remains serialized even when an OpenAI Responses client sends diff --git a/tests/responses/responses-snapshot-repair-server.test.ts b/tests/responses/responses-snapshot-repair-server.test.ts index 57916e8f4e..aea85638db 100644 --- a/tests/responses/responses-snapshot-repair-server.test.ts +++ b/tests/responses/responses-snapshot-repair-server.test.ts @@ -23,11 +23,46 @@ const SPARSE_EVENTS = [ { type: "response.completed", response: { id: "resp_sparse" } }, ]; -function sparseSseBody(): ReadableStream { +const EXPLICIT_EMPTY_TERMINAL_EVENTS = [ + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", + id: "msg_sparse", + role: "assistant", + status: "completed", + phase: "final_answer", + content: [{ type: "output_text", text: "hello", annotations: [] }], + }, + }, + { + type: "response.completed", + response: { id: "resp_sparse", status: "completed", output: [] }, + }, +]; + +const CODEX_SPARSE_TERMINAL_EVENTS = [ + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "hello" }], + }, + }, + { + type: "response.completed", + response: { id: "resp_sparse", status: "completed" }, + }, +]; + +function sparseSseBody(events: readonly Record[] = SPARSE_EVENTS): ReadableStream { return new ReadableStream({ start(controller) { const encoder = new TextEncoder(); - for (const event of SPARSE_EVENTS) { + for (const event of events) { controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)); } controller.enqueue(encoder.encode("data: [DONE]\n\n")); @@ -36,7 +71,10 @@ function sparseSseBody(): ReadableStream { }); } -function stubSparseGateway(origin: string): void { +function stubSparseGateway( + origin: string, + events: readonly Record[] = SPARSE_EVENTS, +): void { globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { const requestUrl = typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; const url = new URL(requestUrl); @@ -44,7 +82,7 @@ function stubSparseGateway(origin: string): void { return Response.json({ data: [] }); } if (url.origin === origin && url.pathname.endsWith("/responses")) { - return new Response(sparseSseBody(), { + return new Response(sparseSseBody(events), { status: 200, headers: { "content-type": "text/event-stream" }, }); @@ -189,4 +227,103 @@ describe("responsesSnapshotRepair through /v1/responses", () => { await server.stop(true); } }); + + test("the Grok client marker alone repairs an explicit empty completed snapshot", async () => { + const gateway = "https://grok-sparse-terminal.example.test"; + stubSparseGateway(gateway, EXPLICIT_EMPTY_TERMINAL_EVENTS); + saveConfig({ + port: 0, + defaultProvider: "sparse", + providers: { + sparse: { + adapter: "openai-responses", + baseUrl: `${gateway}/v1`, + authMode: "key", + apiKey: "test-key", + }, + }, + } as OcxConfig); + + const server = startServer(0); + try { + const request = (grokMarker: boolean) => originalFetch(new URL("/v1/responses", server.url), { + method: "POST", + headers: { + "content-type": "application/json", + ...(grokMarker ? { "x-opencodex-grok": "1" } : {}), + }, + body: JSON.stringify({ model: "sparse-model", input: "hi", stream: true }), + }); + + const grokResponse = await request(true); + expect(grokResponse.status).toBe(200); + const grokText = await grokResponse.text(); + const grokCompletedLine = grokText.split("\n") + .find(line => line.includes('"response.completed"')); + expect(grokCompletedLine).toBeDefined(); + const grokCompleted = JSON.parse(grokCompletedLine!.replace(/^data: /, "")) as { + response: { output: { id: string }[] }; + }; + expect(grokCompleted.response.output[0]?.id).toBe("msg_sparse"); + + const ordinaryResponse = await request(false); + expect(ordinaryResponse.status).toBe(200); + const ordinaryText = await ordinaryResponse.text(); + const ordinaryCompletedLine = ordinaryText.split("\n") + .find(line => line.includes('"response.completed"')); + expect(ordinaryCompletedLine).toBeDefined(); + const ordinaryCompleted = JSON.parse(ordinaryCompletedLine!.replace(/^data: /, "")) as { + response: { output: unknown[] }; + }; + expect(ordinaryCompleted.response.output).toEqual([]); + } finally { + await server.stop(true); + } + }); + + test("the Grok marker repairs Codex-style done items plus a sparse completed response", async () => { + const gateway = "https://grok-codex-sparse.example.test"; + stubSparseGateway(gateway, CODEX_SPARSE_TERMINAL_EVENTS); + saveConfig({ + port: 0, + defaultProvider: "sparse", + providers: { + sparse: { + adapter: "openai-responses", + baseUrl: `${gateway}/v1`, + authMode: "key", + apiKey: "test-key", + }, + }, + } as OcxConfig); + + const server = startServer(0); + try { + const response = await originalFetch(new URL("/v1/responses", server.url), { + method: "POST", + headers: { + "content-type": "application/json", + "x-opencodex-grok": "1", + }, + body: JSON.stringify({ model: "sparse-model", input: "hi", stream: true }), + }); + expect(response.status).toBe(200); + const text = await response.text(); + const completedLine = text.split("\n").find(line => line.includes('"response.completed"')); + expect(completedLine).toBeDefined(); + const completed = JSON.parse(completedLine!.replace(/^data: /, "")) as { + response: { output: Array> }; + }; + expect(completed.response.output).toHaveLength(1); + expect(completed.response.output[0]).toMatchObject({ + id: "msg_ocx_0", + type: "message", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "hello", annotations: [] }], + }); + } finally { + await server.stop(true); + } + }); }); diff --git a/tests/responses/responses-snapshot-repair.test.ts b/tests/responses/responses-snapshot-repair.test.ts index 0231fdb77f..14171a69b2 100644 --- a/tests/responses/responses-snapshot-repair.test.ts +++ b/tests/responses/responses-snapshot-repair.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test"; import { + createGrokResponsesSparseTerminalBlockRewrite, createResponsesSnapshotBlockRewrite, hasResponsesSnapshotRepair, repairResponsesSnapshotJson, @@ -9,6 +10,10 @@ import { payloadRewriteAsBlockRewrite, relaySseWithBlockRewrite, } from "../../src/server/sse-payload-rewrite"; +import { + MAX_COMPLETED_OUTPUT_ITEMS, + MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES, +} from "../../src/server/relay"; import { createTestTranslatorBudget } from "../helpers/translator-budget"; @@ -85,7 +90,361 @@ describe("createResponsesSnapshotBlockRewrite", () => { const terminal = eventsOf(out).find(event => event.type === "response.completed")!; expect((terminal.response as Record).output).toEqual([]); }); +}); + +describe("createGrokResponsesSparseTerminalBlockRewrite", () => { + test("Grok compatibility backfills explicit empty output from completed items", () => { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + const item = { + type: "message", + id: "msg_1", + role: "assistant", + status: "completed", + phase: "final_answer", + content: [{ type: "output_text", text: "hello", annotations: [] }], + }; + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", status: "completed", output: [] }, + })); + const terminal = eventsOf(out).find(event => event.type === "response.completed")!; + expect((terminal.response as Record).output).toEqual([item]); + }); + + test("Grok compatibility preserves explicit empty output when completed items are gapped", () => { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + rewrite(dataBlock({ + type: "response.output_item.done", + output_index: 1, + item: { + type: "message", + id: "msg_2", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "partial", annotations: [] }], + }, + })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", status: "completed", output: [] }, + })); + const terminal = eventsOf(out).find(event => event.type === "response.completed")!; + expect((terminal.response as Record).output).toEqual([]); + }); + + test("Grok compatibility also backfills a missing terminal output", () => { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + const item = { + type: "message", + id: "msg_1", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "hello", annotations: [] }], + }; + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item })); + const out = rewrite(dataBlock({ type: "response.completed", response: { id: "resp_1" } })); + const terminal = eventsOf(out)[0]!.response as Record; + expect(terminal.output).toEqual([item]); + }); + + test("Grok compatibility trusts missing ids/status only when semantic content is already valid", () => { + // The always-on field backfill that follows this rewrite supplies the id, + // message status, and annotations. The official Codex SSE parser also + // accepts done items that omit id/status, so absence alone is not a + // contradiction; malformed values still are. + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + const item = { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "hello" }], + }; + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + })); + const terminal = eventsOf(out)[0]!.response as Record; + expect(terminal.output).toEqual([item]); + }); + + test("Grok compatibility never promotes repaired or contradictory done items", () => { + const invalidItems = [ + { + type: "message", + id: "msg_user", + role: "user", + status: "completed", + content: [{ type: "output_text", text: "bad", annotations: [] }], + }, + { + type: "message", + id: "msg_failed", + role: "assistant", + status: "failed", + content: [{ type: "output_text", text: "bad", annotations: [] }], + }, + { + type: "message", + id: "msg_content", + role: "assistant", + status: "completed", + content: "bad", + }, + { + type: "message", + id: "", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "bad", annotations: [] }], + }, + { + type: "message", + id: 42, + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "bad", annotations: [] }], + }, + ]; + for (const item of invalidItems) { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + })); + const terminal = eventsOf(out)[0]!.response as Record; + expect(terminal.output).toEqual([]); + } + }); + + test("Grok compatibility treats every duplicate done index as contradictory", () => { + for (const conflicting of [false, true]) { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + const first = { + type: "message", + id: "msg_1", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "first", annotations: [] }], + }; + const second = conflicting + ? { ...first, id: "msg_2", content: [{ type: "output_text", text: "second", annotations: [] }] } + : first; + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item: first })); + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item: second })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + })); + const terminal = eventsOf(out)[0]!.response as Record; + expect(terminal.output).toEqual([]); + } + }); + + test("Grok compatibility requires a real done item, not deltas or an open item", () => { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + rewrite(dataBlock({ + type: "response.output_item.added", + output_index: 0, + item: { type: "message", id: "msg_1", role: "assistant", status: "in_progress", content: [] }, + })); + rewrite(dataBlock({ + type: "response.output_text.delta", + output_index: 0, + item_id: "msg_1", + delta: "visible but not durable", + })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + })); + const terminal = eventsOf(out)[0]!.response as Record; + expect(terminal.output).toEqual([]); + }); + + test("Grok compatibility reconstructs a contiguous reasoning-plus-message snapshot", () => { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + const reasoning = { + type: "reasoning", + id: "rs_1", + status: "completed", + summary: [{ type: "summary_text", text: "summary" }], + content: [{ type: "reasoning_text", text: "reasoning" }], + encrypted_content: null, + }; + const message = { + type: "message", + id: "msg_1", + role: "assistant", + status: "completed", + phase: "final_answer", + content: [{ type: "output_text", text: "answer", annotations: [] }], + }; + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item: reasoning })); + rewrite(dataBlock({ type: "response.output_item.done", output_index: 1, item: message })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + })); + const terminal = eventsOf(out)[0]!.response as Record; + expect(terminal.output).toEqual([reasoning, message]); + }); + + test("Grok compatibility reconstructs a valid function call", () => { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + const call = { + type: "function_call", + id: "fc_1", + status: "completed", + call_id: "call_1", + name: "search", + arguments: "{}", + }; + rewrite(dataBlock({ type: "response.output_item.done", output_index: 0, item: call })); + const out = rewrite(dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + })); + const terminal = eventsOf(out)[0]!.response as Record; + expect(terminal.output).toEqual([call]); + }); + + test("Grok compatibility preserves a non-empty terminal snapshot as authoritative", () => { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + rewrite(dataBlock({ + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", id: "msg_done", role: "assistant", status: "completed", + content: [{ type: "output_text", text: "done", annotations: [] }], + }, + })); + const canonical = [{ + type: "message", id: "msg_canonical", role: "assistant", status: "completed", + content: [{ type: "output_text", text: "canonical", annotations: [] }], + }]; + const terminalBlock = dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: canonical }, + }); + const out = rewrite(terminalBlock); + expect(out).toEqual([terminalBlock]); + }); + + test("Grok compatibility leaves explicit malformed terminal output fail-closed", () => { + for (const malformed of [null, "bad", 42, { bad: true }]) { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + rewrite(dataBlock({ + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", id: "msg_1", role: "assistant", status: "completed", + content: [{ type: "output_text", text: "answer", annotations: [] }], + }, + })); + const terminalBlock = dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: malformed }, + }); + expect(rewrite(terminalBlock)).toEqual([terminalBlock]); + } + }); + + test("Grok compatibility bounds and releases open-item identity state", () => { + const budget = createTestTranslatorBudget(); + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(budget); + for (let outputIndex = 0; outputIndex <= MAX_COMPLETED_OUTPUT_ITEMS; outputIndex++) { + rewrite(dataBlock({ + type: "response.output_item.added", + output_index: outputIndex, + item: { type: "message", id: `msg_${outputIndex}` }, + })); + } + // The first item beyond the count bound taints and immediately refunds all + // retained identities; an empty terminal remains authoritative. + expect(budget.snapshot().currentBytes).toBe(0); + const terminalBlock = dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + }); + expect(rewrite(terminalBlock)).toEqual([terminalBlock]); + }); + test("Grok compatibility rejects an oversized retained open identity", () => { + const budget = createTestTranslatorBudget(); + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(budget); + rewrite(dataBlock({ + type: "response.output_item.added", + output_index: 0, + item: { type: "message", id: "x".repeat(MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES + 1) }, + })); + expect(budget.snapshot().currentBytes).toBe(0); + const terminalBlock = dataBlock({ + type: "response.completed", + response: { id: "resp_1", output: [] }, + }); + expect(rewrite(terminalBlock)).toEqual([terminalBlock]); + }); + + test("Grok compatibility dispose releases an unfinished open identity", () => { + const budget = createTestTranslatorBudget(); + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(budget); + rewrite(dataBlock({ + type: "response.output_item.added", + output_index: 0, + item: { type: "message", id: "msg_1" }, + })); + expect(budget.snapshot().currentBytes).toBeGreaterThan(0); + rewrite.dispose?.(); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("Grok compatibility never rewrites failed, incomplete, or contradictory completed terminals", () => { + for (const terminal of [ + { type: "response.failed", response: { id: "resp_1", output: [] } }, + { type: "response.incomplete", response: { id: "resp_1", output: [] } }, + { type: "response.completed", response: { id: "resp_1", status: "failed", output: [] } }, + ]) { + const rewrite = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()); + rewrite(dataBlock({ + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", id: "msg_1", role: "assistant", status: "completed", + content: [{ type: "output_text", text: "answer", annotations: [] }], + }, + })); + const terminalBlock = dataBlock(terminal); + expect(rewrite(terminalBlock)).toEqual([terminalBlock]); + } + }); + + test("Grok sparse repair still fills an empty terminal when composed ahead of provider snapshot repair", () => { + const done = { + type: "message", + id: "msg_1", + role: "assistant", + status: "completed", + phase: "final_answer", + content: [{ type: "output_text", text: "answer", annotations: [] }], + }; + const chain = composeSseBlockRewrites( + createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget()), + createResponsesSnapshotBlockRewrite(undefined, createTestTranslatorBudget()), + ); + chain(dataBlock({ type: "response.output_item.done", output_index: 0, item: done })); + const out = chain(dataBlock({ + type: "response.completed", + response: { id: "resp_1", status: "completed", output: [] }, + })); + const completed = eventsOf(out).find(event => event.type === "response.completed"); + expect(completed).toBeDefined(); + expect((completed!.response as Record).output).toEqual([done]); + }); +}); + +describe("createResponsesSnapshotBlockRewrite", () => { test("already-canonical streams pass through byte-identical", () => { const rewrite = createResponsesSnapshotBlockRewrite(); const canonical = [