From bb41ac3727b501d38992c47acf22780db2f948ec Mon Sep 17 00:00:00 2001 From: HeiTuz Date: Sun, 18 Oct 2026 01:42:31 +0900 Subject: [PATCH 1/3] =?UTF-8?q?fix(opengateway):=20=EB=8F=84=EA=B5=AC=20?= =?UTF-8?q?=EA=B2=B0=EA=B3=BC=20=EB=B0=B0=EC=97=B4=EA=B3=BC=20=EC=9D=B4?= =?UTF-8?q?=EB=AF=B8=EC=A7=80=20=EC=9E=85=EB=A0=A5=20=ED=98=B8=ED=99=98?= =?UTF-8?q?=EC=84=B1=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/content/docs/guides/providers.md | 2 + .../src/content/docs/ko/guides/providers.md | 2 + .../openai-responses/opengateway-input.ts | 78 ++++++++++++++++++ src/adapters/openai-responses/passthrough.ts | 2 + structure/providers-and-adapters.md | 1 + structure/transports/responses-wire-shapes.md | 1 + tests/providers/opengateway-provider.test.ts | 79 +++++++++++++++++++ 7 files changed, 165 insertions(+) create mode 100644 src/adapters/openai-responses/opengateway-input.ts diff --git a/docs-site/src/content/docs/guides/providers.md b/docs-site/src/content/docs/guides/providers.md index bbbf9a61325..a091941c357 100644 --- a/docs-site/src/content/docs/guides/providers.md +++ b/docs-site/src/content/docs/guides/providers.md @@ -736,6 +736,8 @@ Create a key in the [OpenGateway dashboard](https://opengateway.ai/api-keys), th use the configured Bearer key; the public model list does not validate that key. On the Responses wire OpenGateway rejects replayed native custom tool calls, so the preset lowers Codex custom tools such as code-mode `exec` to function tools before sending them. +Tool-result content arrays are sent as text with images preserved in adjacent messages. +Image detail `original` maps to `high`, which the gateway accepts; image bytes remain unchanged. **TokenLab** ([sponsor](https://github.com/lidge-jun/opencodex/blob/main/SPONSORS.md)) is an OpenAI-compatible API gateway at [tokenlab.sh](https://tokenlab.sh/r/OPENCODEX), diff --git a/docs-site/src/content/docs/ko/guides/providers.md b/docs-site/src/content/docs/ko/guides/providers.md index 71e0fcb6d1f..c54c8e816d0 100644 --- a/docs-site/src/content/docs/ko/guides/providers.md +++ b/docs-site/src/content/docs/ko/guides/providers.md @@ -278,6 +278,8 @@ Chat Completions 모델(그리고 Responses 전용이라 Responses로 보내는 [OpenGateway 대시보드](https://opengateway.ai/api-keys)에서 키를 발급한 뒤 `ocx provider add opengateway`를 실행하거나 대시보드에서 **OpenGateway**를 선택하세요. Chat 요청은 설정한 Bearer 키를 사용하며 공개 모델 목록은 키의 유효성을 검증하지 않습니다. +Responses 도구 결과 배열은 텍스트로 변환하고 이미지는 인접 메시지에 보존합니다. +이미지의 `original` 상세도는 게이트웨이가 지원하는 `high`로 바꾸며 이미지 바이트는 그대로 유지합니다. **OpenCode Zen**(`opencode-zen`)과 키 없는 **OpenCode Free** 프리셋은 `https://opencode.ai/zen/v1`을 공유합니다. 그 게이트웨이의 무료 모델은 종종 분당 약 15–20회 요청의 짧은 창 속도 제한에 걸립니다(커뮤니티 측정; OpenCode는 RPM을 공개하지 않음). Zen은 `Retry-After` / `X-RateLimit-*` 헤더 없는 일반 429를 반환할 수 있습니다. 이는 키 없는 데스크톱 할당량(`opencode-free`에서 약 5시간당 Big Pickle/무료 모델 200회)과 별개입니다. Zen이 그런 429에서 `Retry-After`를 생략하면 opencodex는 클라이언트 오류에 안내를 더하고 합성 `Retry-After`를 붙입니다(업스트림 `Retry-After`가 있으면 그것이 우선). 동일 키 대기 재시도는 [`retryOn429`](/ko/reference/configuration/)로 선택합니다. diff --git a/src/adapters/openai-responses/opengateway-input.ts b/src/adapters/openai-responses/opengateway-input.ts new file mode 100644 index 00000000000..694b21f19dc --- /dev/null +++ b/src/adapters/openai-responses/opengateway-input.ts @@ -0,0 +1,78 @@ +import { isPlainObject } from "./internal"; +import { normalizeResponsesToolResultAdjacency } from "./tool-output-recovery"; + +function normalizeImageDetail(part: unknown): unknown { + return isPlainObject(part) && part.type === "input_image" && part.detail === "original" + ? { ...part, detail: "high" } + : part; +} + +/** OpenGateway requires string tool results and auto/low/high image detail. */ +export function normalizeOpenGatewayResponsesInput(body: unknown, url: string): unknown { + const destination = new URL(url); + if (destination.origin !== "https://apis.opengateway.ai" || destination.pathname !== "/v1/responses" + || !isPlainObject(body) || !Array.isArray(body.input)) return body; + + let changed = false; + const input: unknown[] = []; + const pendingImages: unknown[] = []; + for (const item of body.input) { + if (!isPlainObject(item) || item.type !== "function_call_output") { + input.push(...pendingImages); + pendingImages.length = 0; + } + if (!isPlainObject(item)) { input.push(item); continue; } + if (item.type === "function_call_output" && Array.isArray(item.output)) { + const content: Record[] = []; + const text: string[] = []; + let supported = true; + let hasImage = false; + for (const part of item.output) { + if (!isPlainObject(part)) { supported = false; break; } + switch (part.type) { + case "text": + case "input_text": + case "output_text": + if (typeof part.text !== "string") { supported = false; break; } + text.push(part.text); + content.push({ type: "input_text", text: part.text }); + break; + case "refusal": + if (typeof part.refusal !== "string") { supported = false; break; } + text.push(`[refusal] ${part.refusal}`); + content.push({ type: "input_text", text: `[refusal] ${part.refusal}` }); + break; + case "input_image": + if (typeof part.image_url !== "string" && typeof part.file_id !== "string") { supported = false; break; } + content.push({ ...part, ...(part.detail === "original" ? { detail: "high" } : {}) }); + hasImage = true; + break; + default: + supported = false; + } + if (!supported) break; + } + // Opaque or unsupported parts must remain visible to validation, never disappear. + if (supported) { + changed = true; + input.push({ ...item, output: text.join("\n") }); + if (hasImage) pendingImages.push({ type: "message", role: "user", content: [ + { type: "input_text", text: `[tool output for ${item.call_id}]` }, ...content, + ] }); + continue; + } + } + if (Array.isArray(item.content)) { + const originalContent = item.content; + const content = originalContent.map(normalizeImageDetail); + if (content.some((part, index) => part !== originalContent[index])) { + changed = true; + input.push({ ...item, content }); + continue; + } + } + input.push(item); + } + input.push(...pendingImages); + return changed ? normalizeResponsesToolResultAdjacency({ ...body, input }) : body; +} diff --git a/src/adapters/openai-responses/passthrough.ts b/src/adapters/openai-responses/passthrough.ts index 68f04a3f948..7e7a709220b 100644 --- a/src/adapters/openai-responses/passthrough.ts +++ b/src/adapters/openai-responses/passthrough.ts @@ -55,6 +55,7 @@ import { observeOutbound } from "../../usage/cache-diagnostic"; import { normalizeMuseToolChoice } from "./muse-tool-choice"; import { renderCompactionSearchHistoryItems } from "./compaction-search-history"; import { stripKimiEmptyAssistantMessages } from "./empty-assistant"; +import { normalizeOpenGatewayResponsesInput } from "./opengateway-input"; /** * Identifies DeepSeek's strict Responses replay contract: tool-bearing continuations need @@ -503,6 +504,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): && item.type === "item_reference" && typeof item.id === "string" ? [item.id] : [])); finalBody = { ...finalBody, input: renderCompactionSearchHistoryItems(finalBody.input, referencedIds) }; } + finalBody = normalizeOpenGatewayResponsesInput(finalBody, url); observeOutbound(parsed._rawBody, finalBody, headers); if (!isCanonicalOpenAiForwardProvider(provider)) { validateFinalCustomToolCompatibility(finalBody, provider.supportsResponsesCustomTools); diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index aab02c118d2..8a7fd7bbe2f 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -323,6 +323,7 @@ Responses (`openai/o3-pro`); public discovery does not establish key validity. S `deepseek/deepseek-v4.1-flash-ultrafast` and `z-ai/glm-5.3-flash-ultrafast` are listed first. A custom model that replaces one of the discovered rows keeps that row's slot (`src/codex/catalog/routed-gather.ts`). Keys are created at `https://opengateway.ai/api-keys`. +`src/adapters/openai-responses/opengateway-input.ts`는 OpenGateway의 `/v1/responses` 전송 직전에 텍스트 도구 결과 배열을 문자열로 바꾼다. `call_id`와 텍스트 순서는 유지한다. 이미지가 섞인 결과는 원본 이미지와 내용 순서를 사용자 메시지로 보존하며 병렬 호출의 결과 묶음 뒤에 둔다. 이미지의 `detail: original`은 upstream이 수용하는 `high`로 바꾸되 이미지 바이트는 바꾸지 않는다. 알 수 없는 형식과 암호화된 결과 배열은 조용히 삭제하지 않고 기존 검증에 남긴다. 다른 목적지와 원본 요청 객체는 바꾸지 않는다. `tests/providers/opengateway-provider.test.ts`가 실제 adapter 직렬화 경계에서 이 계약을 확인한다. ## TokenLab chat provider diff --git a/structure/transports/responses-wire-shapes.md b/structure/transports/responses-wire-shapes.md index b171c2e1884..b584f90dbb8 100644 --- a/structure/transports/responses-wire-shapes.md +++ b/structure/transports/responses-wire-shapes.md @@ -528,6 +528,7 @@ request-log accounting without promoting a truncated repair candidate. Terminal > Decision record: [ADR-0044](../decisions/ADR-0044-responses-http-sse.md) ## Inbound history and code-mode shell wire repairs +OpenGateway 도구 결과 배열과 이미지 상세도 변환은 [OpenGateway 전송 계약](../providers-and-adapters.md#opengateway-chat-provider)을 따른다. Inbound function-call history with a missing JSON object prefix is repaired for every provider when restoring it produces an object. A terminal MiMo `` suffix is removed only when the remaining text parses as an object unchanged or after restoring a closing brace or quote and brace. Valid JSON containing these tags stays opaque; unrelated malformed arguments replay as `{}` without logging their content (`src/responses/parser.ts`). Regression coverage lives in `tests/responses/responses-mimo-envelope.test.ts`. Function-wrapper restoration supports `default.`-prefixed `exec` and `apply_patch` aliases while diff --git a/tests/providers/opengateway-provider.test.ts b/tests/providers/opengateway-provider.test.ts index 0e771486929..65c4df38371 100644 --- a/tests/providers/opengateway-provider.test.ts +++ b/tests/providers/opengateway-provider.test.ts @@ -112,3 +112,82 @@ describe("OpenGateway Responses custom tools", () => { expect(body.input[2]).toMatchObject({ call_id: "call_exec" }); }); }); + +describe("OpenGateway Responses tool result arrays", () => { + function build(input: Record[], baseUrl = "https://apis.opengateway.ai/v1") { + const raw = { model: "z-ai/glm-5.3-flash-ultrafast", input }; + const snapshot = JSON.stringify(raw); + const routed: OcxProviderConfig = { ...provider, baseUrl }; + enrichProviderFromRegistry("opengateway", routed); + const adapter = withTestTranslatorBudget(createResponsesPassthroughAdapter({ ...routed, adapter: "openai-responses" })); + const request = adapter.buildRequest({ + modelId: raw.model, context: { messages: [] }, stream: false, options: {}, _rawBody: raw, + } as Parameters[0], { headers: new Headers() }); + expect(JSON.stringify(raw)).toBe(snapshot); + const body: { input: Record[] } = JSON.parse(String(request.body)); + return body.input; + } + + for (const kind of ["function", "custom_tool"] as const) { + test(`${kind} text result arrays reach OpenGateway as strings`, () => { + const input = build([ + { type: `${kind}_call`, call_id: "call_array", name: "exec", ...(kind === "function" ? { arguments: "{}" } : { input: "text(1)" }) }, + { type: `${kind}_call_output`, call_id: "call_array", output: [{ type: "input_text", text: "first" }, { type: "input_text", text: "second" }] }, + ]); + expect(input).toContainEqual(expect.objectContaining({ type: "function_call_output", call_id: "call_array", output: "first\nsecond" })); + }); + } + + test("parallel image results retain image bytes and all paired outputs before image messages", () => { + const image = { type: "input_image", image_url: "data:image/png;base64,cHJvYmU=", detail: "original" }; + const input = build([ + { type: "function_call", call_id: "a", name: "echo", arguments: "{}" }, + { type: "function_call", call_id: "b", name: "echo", arguments: "{}" }, + { type: "function_call_output", call_id: "a", output: [{ type: "input_text", text: "before" }, image, { type: "input_text", text: "after" }] }, + { type: "function_call_output", call_id: "b", output: [] }, + ]); + expect(input.map(item => item.type)).toEqual(["function_call", "function_call", "function_call_output", "function_call_output", "message"]); + expect(input[2]).toMatchObject({ call_id: "a", output: "before\nafter" }); + expect(input[3]).toMatchObject({ call_id: "b", output: "" }); + expect(input[4]).toMatchObject({ role: "user", content: expect.arrayContaining([{ ...image, detail: "high" }]) }); + }); + + test("OpenGateway user images normalize original detail without changing image bytes", () => { + const image = { type: "input_image", image_url: "data:image/png;base64,cHJvYmU=", detail: "original" }; + const input = build([{ type: "message", role: "user", content: [image] }]); + expect(input[0]).toMatchObject({ content: [{ ...image, detail: "high" }] }); + }); + + test("reversed parallel results stay together before their image messages", () => { + const image = { type: "input_image", image_url: "data:image/png;base64,cHJvYmU=", detail: "high" }; + const input = build([ + { type: "function_call", call_id: "a", name: "echo", arguments: "{}" }, + { type: "function_call", call_id: "b", name: "echo", arguments: "{}" }, + { type: "function_call_output", call_id: "b", output: [image] }, + { type: "function_call_output", call_id: "a", output: "done" }, + ]); + expect(input.map(item => item.type)).toEqual(["function_call", "function_call", "function_call_output", "function_call_output", "message"]); + expect(input[2]?.call_id).toBe("b"); + expect(input[3]?.call_id).toBe("a"); + expect(input[4]).toMatchObject({ role: "user", content: expect.arrayContaining([image]) }); + }); + + test("other destinations keep arrays and original image detail", () => { + const output = [{ type: "input_text", text: "unchanged" }, { type: "input_image", image_url: "data:image/png;base64,cHJvYmU=", detail: "original" }]; + const input = build([ + { type: "function_call", call_id: "a", name: "echo", arguments: "{}" }, + { type: "function_call_output", call_id: "a", output }, + ], "https://api.openai.com/v1"); + expect(input.find(item => item.type === "function_call_output")?.output).toEqual(output); + }); + + test("string and unsupported opaque result arrays are preserved", () => { + for (const output of ["exact string", [{ type: "encrypted_content", encrypted_content: "opaque" }]]) { + const input = build([ + { type: "function_call", call_id: "a", name: "echo", arguments: "{}" }, + { type: "function_call_output", call_id: "a", output }, + ]); + expect(input.find(item => item.type === "function_call_output")?.output).toEqual(output); + } + }); +}); From d337e6ecfba90325e48ba21e3abceedf50e7027f Mon Sep 17 00:00:00 2001 From: HeiTuz Date: Sun, 18 Oct 2026 02:57:57 +0900 Subject: [PATCH 2/3] =?UTF-8?q?fix(opengateway):=20=EC=B6=9C=EB=A0=A5=20?= =?UTF-8?q?=EC=97=86=EB=8A=94=20=EC=99=84=EB=A3=8C=20=EC=9D=91=EB=8B=B5?= =?UTF-8?q?=EC=9D=84=20=EC=95=88=EC=A0=84=ED=95=98=EA=B2=8C=20=EC=B2=98?= =?UTF-8?q?=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/content/docs/guides/providers.md | 8 + .../src/content/docs/ko/guides/providers.md | 8 + .../responses/opengateway-completion-guard.ts | 204 ++++++++++++++ .../responses/passthrough-completion.ts | 41 +++ src/server/responses/passthrough-dispatch.ts | 18 +- structure/providers-and-adapters.md | 4 + tests/responses/empty-completion-core.test.ts | 155 ++++++++++ .../empty-completion-hardening.test.ts | 266 ++++++++++++++++++ 8 files changed, 693 insertions(+), 11 deletions(-) create mode 100644 src/server/responses/opengateway-completion-guard.ts create mode 100644 src/server/responses/passthrough-completion.ts diff --git a/docs-site/src/content/docs/guides/providers.md b/docs-site/src/content/docs/guides/providers.md index a091941c357..ae5e593f7ab 100644 --- a/docs-site/src/content/docs/guides/providers.md +++ b/docs-site/src/content/docs/guides/providers.md @@ -738,6 +738,14 @@ On the Responses wire OpenGateway rejects replayed native custom tool calls, so lowers Codex custom tools such as code-mode `exec` to function tools before sending them. Tool-result content arrays are sent as text with images preserved in adjacent messages. Image detail `original` maps to `high`, which the gateway accepts; image bytes remain unchanged. +The native Responses route also checks text-only empty completions before exposing them as success. +Exhausted output budgets produce `incomplete` with `max_output_tokens`; the proxy does not increase +the requested budget. A replay-safe empty completion with budget remaining may retry once, within +the existing send limit. Explicit `emptyCompletionRetry: false` or `OCX_EMPTY_COMPLETION_RETRY=0` +disables that retry. Text, refusals, tool activity, and unknown stream events prevent replay and +retain the original stream. The protection does not change other providers or their retry defaults. +Replay requires explicit `store: false`, complete input, and client-executed tools only; a hosted +tool such as `web_search` prevents automatic replay even if no tool event was observed. **TokenLab** ([sponsor](https://github.com/lidge-jun/opencodex/blob/main/SPONSORS.md)) is an OpenAI-compatible API gateway at [tokenlab.sh](https://tokenlab.sh/r/OPENCODEX), diff --git a/docs-site/src/content/docs/ko/guides/providers.md b/docs-site/src/content/docs/ko/guides/providers.md index c54c8e816d0..c747989698f 100644 --- a/docs-site/src/content/docs/ko/guides/providers.md +++ b/docs-site/src/content/docs/ko/guides/providers.md @@ -280,6 +280,14 @@ Chat Completions 모델(그리고 Responses 전용이라 Responses로 보내는 Chat 요청은 설정한 Bearer 키를 사용하며 공개 모델 목록은 키의 유효성을 검증하지 않습니다. Responses 도구 결과 배열은 텍스트로 변환하고 이미지는 인접 메시지에 보존합니다. 이미지의 `original` 상세도는 게이트웨이가 지원하는 `high`로 바꾸며 이미지 바이트는 그대로 유지합니다. +네이티브 Responses 경로는 텍스트가 없는 완료 응답도 검사합니다. 출력 예산이 소진됐으면 +`incomplete / max_output_tokens`로 알리며 요청한 예산을 임의로 늘리지 않습니다. +예산이 남고 안전하게 재전송할 수 있는 빈 완료는 기존 호출 한도 안에서 한 번만 재시도합니다. +`emptyCompletionRetry: false` 또는 `OCX_EMPTY_COMPLETION_RETRY=0`이면 이 재시도를 끕니다. +텍스트·거절·도구 동작·알 수 없는 스트림 이벤트가 관측되면 재전송하지 않고 원본을 유지합니다. +다른 제공자의 동작과 재시도 기본값에는 적용하지 않습니다. +재전송은 명시적인 `store: false`, 완전한 입력, 클라이언트 실행 도구만 있는 요청에 한정합니다. +`web_search` 같은 서버 실행 도구가 선언돼 있으면 도구 이벤트가 없어도 자동 재전송하지 않습니다. **OpenCode Zen**(`opencode-zen`)과 키 없는 **OpenCode Free** 프리셋은 `https://opencode.ai/zen/v1`을 공유합니다. 그 게이트웨이의 무료 모델은 종종 분당 약 15–20회 요청의 짧은 창 속도 제한에 걸립니다(커뮤니티 측정; OpenCode는 RPM을 공개하지 않음). Zen은 `Retry-After` / `X-RateLimit-*` 헤더 없는 일반 429를 반환할 수 있습니다. 이는 키 없는 데스크톱 할당량(`opencode-free`에서 약 5시간당 Big Pickle/무료 모델 200회)과 별개입니다. Zen이 그런 429에서 `Retry-After`를 생략하면 opencodex는 클라이언트 오류에 안내를 더하고 합성 `Retry-After`를 붙입니다(업스트림 `Retry-After`가 있으면 그것이 우선). 동일 키 대기 재시도는 [`retryOn429`](/ko/reference/configuration/)로 선택합니다. diff --git a/src/server/responses/opengateway-completion-guard.ts b/src/server/responses/opengateway-completion-guard.ts new file mode 100644 index 00000000000..216930064fc --- /dev/null +++ b/src/server/responses/opengateway-completion-guard.ts @@ -0,0 +1,204 @@ +import { guardResponseBodyInactivity } from "../../lib/response-body-inactivity"; +import { isNonReplayableResponse } from "../../lib/upstream-retry"; +import { isPlainObject } from "../responses-snapshot-codec"; +import { BoundedSseFrameBuffer, joinSseFrameBytes } from "../sse-frame-buffer"; +import { sseDataPayload } from "../sse-payload-rewrite"; +import { EMPTY_COMPLETION_MAX_BUFFERED_BYTES, EMPTY_COMPLETION_MAX_BUFFERED_EVENTS, EMPTY_COMPLETION_RETRY_ENV } from "./empty-completion-guard"; +import { selfContainedResponsesBody } from "./reset-replay"; +import { carryResponseToolOwnership } from "./response-tool-ownership"; + +const MAX_BYTES = EMPTY_COMPLETION_MAX_BUFFERED_BYTES; +const MAX_EVENTS = EMPTY_COMPLETION_MAX_BUFFERED_EVENTS; +const MAX_HOLD_MS = 30_000; +const decoder = new TextDecoder(); +const reasoningEvents = new Set([ + "response.reasoning_summary_part.added", "response.reasoning_summary_part.done", + "response.reasoning_summary_text.delta", "response.reasoning_summary_text.done", + "response.reasoning_text.delta", "response.reasoning_text.done", +]); +type RecordValue = Record; + +function partCommits(part: unknown): boolean { + return !isPlainObject(part) || part.type !== "output_text" || part.text !== "" + || (part.annotations !== undefined && (!Array.isArray(part.annotations) || part.annotations.length > 0)); +} +function outputCommits(output: unknown): boolean { + if (output === undefined) return false; + return !Array.isArray(output) || output.some(item => !isPlainObject(item) + || (item.type !== "reasoning" && (item.type !== "message" || !Array.isArray(item.content) || item.content.some(partCommits)))); +} +function wellFormedEmpty(snapshot: unknown): snapshot is RecordValue { + return isPlainObject(snapshot) && typeof snapshot.id === "string" && snapshot.id.trim().length > 0 + && snapshot.object === "response" && snapshot.status === "completed" && Array.isArray(snapshot.output) + && snapshot.error == null && snapshot.incomplete_details == null + && snapshot.output.every(item => isPlainObject(item) && typeof item.id === "string" && item.id.length > 0 + && (item.type === "reasoning" || (item.type === "message" && item.role === "assistant" + && item.status === "completed" && Array.isArray(item.content) && !item.content.some(partCommits)))); +} +function eventCommits(event: RecordValue): boolean { + if (event.replayUnsafe === true) return true; + switch (event.type) { + case "response.created": case "response.in_progress": case "response.queued": case "response.heartbeat": + return isPlainObject(event.response) && outputCommits(event.response.output); + case "response.output_item.added": case "response.output_item.done": return outputCommits([event.item]); + case "response.content_part.added": case "response.content_part.done": return partCommits(event.part); + case "response.output_text.delta": return event.delta !== ""; + case "response.output_text.done": return event.text !== ""; + default: return !reasoningEvents.has(String(event.type)); + } +} + +export interface OpenGatewayCompletionGuardOptions { + readonly url: string; + readonly body: unknown; + readonly signal: AbortSignal; + readonly retry: () => Promise; + readonly recordUsage: (usage: unknown) => void; + readonly retryEnabled?: boolean; + readonly stallTimeoutMs?: number; + /** Expiry releases original bytes; it never aborts a healthy upstream. */ + readonly holdTimeoutMs?: number; +} + +type Inspection = { kind: "pass"; response: Response } + | { kind: "empty" | "invalid"; snapshot: RecordValue; streaming: boolean }; + +async function inspect(response: Response, options: OpenGatewayCompletionGuardOptions): Promise { + const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; + const streaming = contentType.includes("text/event-stream") + || (!contentType && isPlainObject(options.body) && options.body.stream === true); + if (!response.body || (!streaming && !contentType.includes("application/json"))) return { kind: "pass", response }; + const guarded = guardResponseBodyInactivity(response, options.signal, streaming ? options.stallTimeoutMs ?? 0 : 0); + const reader = guarded.response.body?.getReader(); + if (!reader) return { kind: "pass", response }; + const framer = new BoundedSseFrameBuffer(MAX_BYTES); + const held: Uint8Array[] = []; + let bytes = 0, events = 0, transferred = false, cleaned = false; + let pending: ReturnType | undefined; + const holdMs = options.holdTimeoutMs ?? MAX_HOLD_MS; + const deadline = performance.now() + holdMs; + let timer: ReturnType | undefined; + const expired = new Promise<"expired">(resolve => { timer = setTimeout(() => resolve("expired"), holdMs); }); + const cleanup = () => { + if (cleaned) return; + cleaned = true; + guarded.dispose(); + reader.releaseLock(); + }; + const pass = (): Inspection => { + transferred = true; + let index = 0; + const body = new ReadableStream({ + async pull(controller) { + if (index < held.length) { controller.enqueue(held[index++]); return; } + try { + const next = await (pending ?? reader.read()); + pending = undefined; + if (next.done) { cleanup(); controller.close(); } + else controller.enqueue(next.value); + } catch (error) { cleanup(); controller.error(error); } + }, + cancel() { cleanup(); }, + }, { highWaterMark: 0 }); + return { kind: "pass", response: carryResponseToolOwnership(response, new Response(body, { + status: response.status, statusText: response.statusText, headers: response.headers, + })) }; + }; + const terminal = (snapshot: unknown): Inspection => ({ + kind: wellFormedEmpty(snapshot) ? "empty" : "invalid", + snapshot: isPlainObject(snapshot) ? snapshot : {}, streaming, + }); + try { + for (;;) { + if (performance.now() >= deadline) return pass(); + pending = reader.read(); + let next: Awaited | "expired"; + try { next = await Promise.race([pending, expired]); } + catch { return pass(); } + if (next === "expired") return pass(); + pending = undefined; + if (!next.done) { + if (bytes + next.value.byteLength > MAX_BYTES || held.length >= MAX_EVENTS) { + held.push(next.value); + return pass(); + } + held.push(next.value.slice()); + bytes += next.value.byteLength; + } + if (!streaming) { + if (!next.done) continue; + let snapshot: unknown; + try { snapshot = JSON.parse(decoder.decode(joinSseFrameBytes(held))); } + catch { return terminal(undefined); } + if (isPlainObject(snapshot) && (outputCommits(snapshot.output) + || snapshot.error != null || ["failed", "incomplete", "queued", "in_progress"].includes(String(snapshot.status)))) return pass(); + return terminal(snapshot); + } + let frames; + try { frames = next.done ? [{ block: framer.finish() }] : framer.feed(next.value); } + catch { return pass(); } + for (const raw of frames) { + if (raw.block.byteLength === 0) continue; + if (++events >= MAX_EVENTS) return pass(); + const block = decoder.decode(raw.block), data = sseDataPayload(block); + if (data === null) { if (/^event:/m.test(block)) return pass(); continue; } + let event: unknown; + try { event = JSON.parse(data); } catch { return pass(); } + if (!isPlainObject(event)) return pass(); + if (event.replayUnsafe === true) return pass(); + if (event.type === "response.completed") { + if (isPlainObject(event.response) && outputCommits(event.response.output)) return pass(); + return terminal(event.response); + } + if (eventCommits(event)) return pass(); + } + if (next.done) return pass(); + } + } finally { + clearTimeout(timer); framer.dispose(); + if (!transferred) cleanup(); + } +} + +function terminalResponse(source: Response, inspection: Exclude, reason: string): Response { + const incomplete = reason === "max_output_tokens"; + const snapshot = { ...inspection.snapshot, status: incomplete ? "incomplete" : "failed", + ...(incomplete ? { incomplete_details: { reason } } : { error: { + type: "upstream_error", code: reason, message: "OpenGateway returned no usable completion before output.", + } }), + }; + const headers = new Headers(source.headers); + headers.delete("content-length"); headers.delete("content-encoding"); + if (inspection.streaming) headers.set("content-type", "text/event-stream"); + const text = inspection.streaming + ? `event: response.${snapshot.status}\ndata: ${JSON.stringify({ type: `response.${snapshot.status}`, response: snapshot })}\n\n` + : JSON.stringify(snapshot); + return new Response(text, { status: inspection.streaming || incomplete ? 200 : 502, headers }); +} + +export async function guardOpenGatewayCompletion(response: Response, options: OpenGatewayCompletionGuardOptions): Promise { + if (options.url !== "https://apis.opengateway.ai/v1/responses" || response.status !== 200) return response; + let body: unknown; + try { body = typeof options.body === "string" ? JSON.parse(options.body) : options.body; } + catch { return response; } + if (!isPlainObject(body)) return response; + const canReplay = options.retryEnabled !== false && process.env[EMPTY_COMPLETION_RETRY_ENV] !== "0" + && selfContainedResponsesBody(body) && !isNonReplayableResponse(response); + let selected = response; + for (let attempt = 0; ; attempt++) { + const inspected = await inspect(selected, options); + if (inspected.kind === "pass") return inspected.response; + if (inspected.kind === "invalid") return terminalResponse(selected, inspected, "invalid_completion"); + const usage = inspected.snapshot.usage; + if (typeof body.max_output_tokens === "number" && body.max_output_tokens > 0 + && isPlainObject(usage) && typeof usage.output_tokens === "number" && usage.output_tokens >= body.max_output_tokens) { + return terminalResponse(selected, inspected, "max_output_tokens"); + } + if (attempt > 0 || !canReplay || options.signal.aborted) { + return terminalResponse(selected, inspected, attempt > 0 ? "empty_completion_retry_failed" : "empty_completion"); + } + options.recordUsage(usage); + selected = await options.retry(); + if (selected.status !== 200) return selected; + } +} diff --git a/src/server/responses/passthrough-completion.ts b/src/server/responses/passthrough-completion.ts new file mode 100644 index 00000000000..85ccac8af4a --- /dev/null +++ b/src/server/responses/passthrough-completion.ts @@ -0,0 +1,41 @@ +import type { AdapterRequest } from "../../adapters/base"; +import { formatErrorResponse } from "../../bridge"; +import { resolveStallTimeoutMs } from "../../stall-timeout"; +import type { OcxConfig } from "../../types"; +import { recordKeyWireAttemptUsage, usageFromResponsesPayload, type RequestLogContext } from "../request-log"; +import { isPlainObject } from "../responses-snapshot-codec"; +import { guardOpenGatewayCompletion } from "./opengateway-completion-guard"; + +export function parseOutboundRequestBody(bodyText: string): Record | undefined { + try { + const body: unknown = JSON.parse(bodyText); + return isPlainObject(body) ? body : undefined; + } catch { + return undefined; + } +} + +export function guardPassthroughCompletion( + response: Response, + request: AdapterRequest, + config: Pick, + signal: AbortSignal, + logCtx: RequestLogContext, + rebuildAndRefetch: (kind: "empty-completion") => Promise, +): Promise { + if (request.url !== "https://apis.opengateway.ai/v1/responses") return Promise.resolve(response); + return guardOpenGatewayCompletion(response, { + url: request.url, + body: parseOutboundRequestBody(request.body), + signal, + retryEnabled: config.emptyCompletionRetry, + stallTimeoutMs: resolveStallTimeoutMs(config.stallTimeoutSec), + recordUsage: usage => { recordKeyWireAttemptUsage(logCtx, usageFromResponsesPayload(usage)); }, + retry: async () => { + const result = await rebuildAndRefetch("empty-completion"); + if (!("failed" in result)) return result; + return result.failed instanceof Response ? result.failed + : formatErrorResponse(502, "upstream_error", "Native recovery changed the provider wire"); + }, + }); +} diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 9eee8ea699f..94ae188ba04 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -171,6 +171,7 @@ import { upstreamErrorMessageFromPayload, ENCRYPTED_FUNCTION_OUTPUT_REJECTION } import { isTransientConsoleGoUploadRejection } from "../../providers/opencode-zen-rate-limit"; import { planReasoningEffortDowngrade } from "../../providers/reasoning-metadata"; import { unboundPoolSpendRefusalResponse } from "../workflow-refusal"; +import { guardPassthroughCompletion, parseOutboundRequestBody } from "./passthrough-completion"; /** Transfer an uncommitted Copilot exchange to the existing translated pipeline without new admission. */ export interface PassthroughAdapterHandoff { kind: "adapter-handoff" } @@ -432,16 +433,6 @@ export async function preparePassthroughExchange( // `{model, input, stream}` with no tools and Copilot answers with a `custom_tool_call` for // `apply_patch`. Policing an absent catalog truncates that turn. An unreadable body lands there // too because the proxy cannot establish the caller's declared authorization boundary. - const parseOutboundRequestBody = (bodyText: string): Record | undefined => { - try { - const body = JSON.parse(bodyText) as unknown; - return body && typeof body === "object" && !Array.isArray(body) - ? body as Record - : undefined; - } catch { - return undefined; - } - }; let outboundRequestBody: Record | undefined; const declaredWireToolNames = new Set(); const declaredBareWireToolNames = new Set(); @@ -1165,6 +1156,7 @@ export async function preparePassthroughExchange( `${route.providerName}|${route.modelId}|${recovery}`, { allowFinalRecoveryReserve: transientSendPolicy() === null }, ); + if (recovery === "empty-completion") allowance.attempts = Math.min(1, allowance.attempts); try { return await fetchWithTransientRetry( innerRecovery => { @@ -1195,7 +1187,8 @@ export async function preparePassthroughExchange( .then(adoptObservedResponse); }, { abortSignal: upstream.signal, label: safeHostLabel(request.url), attempts: allowance.attempts, - onSendsConsumed: transientSendReporter(allowance.permit), claimAmbiguousResend: claimPreHeaderResend }, + onSendsConsumed: transientSendReporter(allowance.permit), + claimAmbiguousResend: recovery === "empty-completion" ? undefined : claimPreHeaderResend }, ); } catch (err) { return { failed: transportFailureResponse(err) }; @@ -1945,6 +1938,9 @@ export async function preparePassthroughExchange( break; } + upstreamResponse = await guardPassthroughCompletion( + upstreamResponse, request, config, upstream.signal, logCtx, rebuildAndRefetch, + ); // Where this relay dials upstream. Local infrastructure (loopback / private / `.local` / `.lan`) // is operator-trusted and its silent phases are normal, so an unset stall budget resolves to // disabled for it. Provider rotation above keeps the same endpoint origin, so the routed diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index 8a7fd7bbe2f..911401f884a 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -325,6 +325,10 @@ that row's slot (`src/codex/catalog/routed-gather.ts`). Keys are created at `https://opengateway.ai/api-keys`. `src/adapters/openai-responses/opengateway-input.ts`는 OpenGateway의 `/v1/responses` 전송 직전에 텍스트 도구 결과 배열을 문자열로 바꾼다. `call_id`와 텍스트 순서는 유지한다. 이미지가 섞인 결과는 원본 이미지와 내용 순서를 사용자 메시지로 보존하며 병렬 호출의 결과 묶음 뒤에 둔다. 이미지의 `detail: original`은 upstream이 수용하는 `high`로 바꾸되 이미지 바이트는 바꾸지 않는다. 알 수 없는 형식과 암호화된 결과 배열은 조용히 삭제하지 않고 기존 검증에 남긴다. 다른 목적지와 원본 요청 객체는 바꾸지 않는다. `tests/providers/opengateway-provider.test.ts`가 실제 adapter 직렬화 경계에서 이 계약을 확인한다. +`src/server/responses/opengateway-completion-guard.ts`는 정확한 `https://apis.opengateway.ai/v1/responses`의 HTTP 200 SSE/JSON에서 출력 전 빈 완료만 검사한다. 첫 텍스트, refusal(빈 refusal 포함), action 또는 unknown 이벤트부터는 원본 바이트를 그대로 전달하며 snapshot을 재구성하지 않는다. 정상 형식의 빈 terminal에서 reported `output_tokens >= outbound max_output_tokens`이면 재송신 없이 `incomplete/max_output_tokens`, 형식이 불완전하면 `invalid_completion` 실패다. Reasoning은 답변이 아니다. +미소진 빈 완료는 실제 outbound body가 `selfContainedResponsesBody`를 통과할 때만 한 번 재시도한다. 명시적 `store:false`와 client-only tools가 필요하며 store 생략/true, continuation, background, hosted/unknown tool은 `empty_completion` 실패로 끝난다. `web_search`가 남아 있는 Codex 요청도 재생하지 않는다. 이 목적지에서는 설정 미지정 시 복구하되 `emptyCompletionRetry:false`와 `OCX_EMPTY_COMPLETION_RETRY=0`을 존중하며, 다른 목적지와 translated opt-in은 바뀌지 않는다. +`src/server/responses/passthrough-completion.ts`가 기존 `rebuildAndRefetch("empty-completion")`의 pacing·accounting·abort·send budget에 연결하며 복구 물리 전송은 한 번이다. 첫 usage는 새 전송 전에 기존 attempt baseline에 기록한다. 두 번째 정상 형식의 빈 200만 `empty_completion_retry_failed`이며 재시도 HTTP 실패와 send-budget 거절은 원래 Response를 delivery에 넘긴다. 1 MiB/1,024개 이벤트·청크/30초 보관 한도에 닿으면 원본 prefix와 진행 중인 read를 전달하고 재시도를 닫는다. 보관 한도는 upstream을 취소하지 않으며 SSE stall은 기존 `resolveStallTimeoutMs`와 0-disable 정책을 따른다. timer·framer·reader cleanup 및 원본 바이트 보존은 `tests/responses/empty-completion-core.test.ts`, `tests/responses/empty-completion-hardening.test.ts`가 검증한다. + ## TokenLab chat provider The `tokenlab` key preset uses the existing OpenAI Chat adapter at diff --git a/tests/responses/empty-completion-core.test.ts b/tests/responses/empty-completion-core.test.ts index 6a645c8da54..4e7428d82c9 100644 --- a/tests/responses/empty-completion-core.test.ts +++ b/tests/responses/empty-completion-core.test.ts @@ -502,3 +502,158 @@ describe("empty-completion core integration", () => { expect(body.match(/event: response\.completed/g)).toHaveLength(1); }); }); + +describe("OpenGateway native empty completion", () => { + const endpoint = "https://apis.opengateway.ai/v1/responses"; + const empty = (tokens = 1) => ({ + id: "resp_empty", object: "response", status: "completed", + output: [{ id: "msg_1", type: "message", role: "assistant", status: "completed", + content: [{ type: "output_text", text: "", annotations: [] }] }], + usage: { input_tokens: 3, output_tokens: tokens, output_tokens_details: { reasoning_tokens: tokens } }, + }); + const answer = () => ({ ...empty(2), id: "resp_answer", output: [ + { id: "msg_1", type: "message", role: "assistant", status: "completed", + content: [{ type: "output_text", text: "정상 답변", annotations: [] }] }, + ] }); + const sse = (snapshot: unknown, prefix: unknown[] = []) => new Response( + [...prefix, { type: "response.completed", response: snapshot }] + .map(event => `data: ${JSON.stringify(event)}\n\n`).join(""), + { headers: { "content-type": "text/event-stream" } }, + ); + function fixture(replies: (() => Response)[], url = endpoint, providerExtra: Partial = {}) { + const sends: { url: string; body: unknown }[] = []; + const provider: OcxProviderConfig & { fetch?: typeof globalThis.fetch } = { + adapter: "openai-responses", baseUrl: url.replace(/\/responses$/, ""), + apiKey: "fixture-key", authMode: "key", models: ["model"], ...providerExtra, + fetch: async (input, init) => { + sends.push({ url: String(input), body: init?.body }); + return replies[Math.min(sends.length - 1, replies.length - 1)]?.() ?? Response.error(); + }, + }; + const cfg = { port: 0, defaultProvider: "fixture", providers: { fixture: provider } } satisfies OcxConfig; + return { cfg, sends }; + } + + for (const streaming of [false, true]) { + test(`native ${streaming ? "SSE" : "JSON"} exhausted reasoning becomes incomplete without replay`, async () => { + const f = fixture([() => streaming ? sse(empty(16)) : Response.json(empty(16))]); + takeSpendHome(); + const response = await handleResponses(request(streaming, "질문", { store: false, max_output_tokens: 16 }), f.cfg, { model: "", provider: "" }); + const text = await response.text(); + expect(text).toContain('"status":"incomplete"'); + expect(text).toContain('"reason":"max_output_tokens"'); + expect(text).not.toContain('"type":"response.completed"'); + expect(f.sends).toHaveLength(1); + }); + + test(`native ${streaming ? "SSE" : "JSON"} safe empty retries once and preserves first usage`, async () => { + const f = fixture([ + () => streaming ? sse(empty(), [ + { type: "response.created", response: { id: "resp_empty", status: "in_progress", output: [] } }, + { type: "response.output_item.added", output_index: 0, item: { id: "msg_1", type: "message", role: "assistant", content: [] } }, + { type: "response.content_part.added", output_index: 0, content_index: 0, item_id: "msg_1", part: { type: "output_text", text: "", annotations: [] } }, + { type: "response.output_text.done", output_index: 0, content_index: 0, item_id: "msg_1", text: "" }, + { type: "response.content_part.done", output_index: 0, content_index: 0, item_id: "msg_1", part: { type: "output_text", text: "", annotations: [] } }, + { type: "response.output_item.done", output_index: 0, item: empty().output[0] }, + ]) : Response.json(empty()), + () => streaming ? sse(answer()) : Response.json(answer()), + ]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + takeSpendHome(); + const response = await handleResponses(request(streaming, "질문", { store: false, max_output_tokens: 1024 }), f.cfg, logCtx); + expect(await response.text()).toContain("정상 답변"); + expect(f.sends).toHaveLength(2); + expect(f.sends[1]).toEqual(f.sends[0]); + expect(logCtx.activeAttempt).toMatchObject({ sendCount: 2, recoveryKinds: ["empty-completion"], + usage: { inputTokens: 6, outputTokens: 3 } }); + }); + + test(`native ${streaming ? "SSE" : "JSON"} second empty is an explicit failure`, async () => { + const f = fixture([() => streaming ? sse(empty()) : Response.json(empty())]); + takeSpendHome(); + const response = await handleResponses(request(streaming, "질문", { store: false }), f.cfg, { model: "", provider: "" }); + const text = await response.text(); + expect(text).toContain("empty_completion_retry_failed"); + if (streaming) expect(text).not.toContain('"type":"response.completed"'); + else expect(JSON.parse(text).status).toBe("failed"); + expect(f.sends).toHaveLength(2); + }); + } + + test("native SSE leaves empty terminal snapshots untouched after the first text", async () => { + const f = fixture([() => sse(empty(), [ + { type: "response.output_item.added", output_index: 0, item: { id: "msg_1", type: "message", role: "assistant", content: [] } }, + { type: "response.output_text.delta", output_index: 0, content_index: 0, item_id: "msg_1", delta: "남아야 할 답변" }, + { type: "response.output_text.done", output_index: 0, content_index: 0, item_id: "msg_1", text: "" }, + { type: "response.output_item.done", output_index: 0, item: empty().output[0] }, + ])]); + takeSpendHome(); + const response = await handleResponses(request(true, "질문", { store: false }), f.cfg, { model: "", provider: "" }); + const text = await response.text(); + const events = text.split("\n").filter(line => line.startsWith("data: {")).map(line => JSON.parse(line.slice(6))); + expect(events.find(event => event.type === "response.output_text.delta")?.delta).toBe("남아야 할 답변"); + expect(events.find(event => event.type === "response.output_text.done")?.text).toBe(""); + expect(events.find(event => event.type === "response.completed")?.response.output[0].content[0].text).toBe(""); + expect(f.sends).toHaveLength(1); + }); + + test("native retry failure consumes only one replacement even on HTTP 503", async () => { + const f = fixture([() => sse(empty()), () => new Response("unavailable", { status: 503 })]); + takeSpendHome(); + const response = await handleResponses(request(true, "질문", { store: false }), f.cfg, { model: "", provider: "" }); + expect(response.status).toBe(503); + expect(await response.text()).not.toContain("empty_completion_retry_failed"); + expect(f.sends).toHaveLength(2); + }); + + test("native recovery cannot exceed a provider's explicit one-send budget", async () => { + const f = fixture([() => sse(empty())], endpoint, { transientRetryOn5xx: { enabled: true, attempts: 1 } }); + takeSpendHome(); + const response = await handleResponses(request(true, "질문", { store: false }), f.cfg, { model: "", provider: "" }); + expect(response.status).toBeGreaterThanOrEqual(400); + expect(await response.text()).not.toContain("empty_completion_retry_failed"); + expect(f.sends).toHaveLength(1); + }); + + test("native JSON retry HTTP failure retains first usage exactly once", async () => { + const f = fixture([() => Response.json(empty()), () => new Response("unavailable", { status: 503 })]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + takeSpendHome(); + const response = await handleResponses(request(false, "질문", { store: false }), f.cfg, logCtx); + expect(response.status).toBe(503); + expect(await response.text()).not.toContain("empty_completion_retry_failed"); + expect(f.sends).toHaveLength(2); + expect(logCtx.activeAttempt?.usage).toMatchObject({ inputTokens: 3, outputTokens: 1 }); + }); + + test("native recovery preserves upstream SSE even when the caller requested JSON", async () => { + const f = fixture([() => sse(empty()), () => sse(answer())]); + takeSpendHome(); + const response = await handleResponses(request(false, "질문", { store: false }), f.cfg, { model: "", provider: "" }); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + expect(await response.text()).toContain("정상 답변"); + expect(f.sends).toHaveLength(2); + }); + + for (const optOut of ["config", "environment"]) { + test(`native ${optOut} opt-out prevents replay but still rejects empty success`, async () => { + const f = fixture([() => sse(empty())]); + if (optOut === "environment") process.env.OCX_EMPTY_COMPLETION_RETRY = "0"; + takeSpendHome(); + const response = await handleResponses(request(true, "질문"), + { ...f.cfg, ...(optOut === "config" ? { emptyCompletionRetry: false } : {}) }, { model: "", provider: "" }); + expect(await response.text()).toContain('"type":"response.failed"'); + expect(f.sends).toHaveLength(1); + }); + } + + for (const url of ["https://other.test/v1/responses", "http://apis.opengateway.ai/v1/responses", "https://apis.opengateway.ai/other/responses"]) { + test(`native empty guard does not expand to ${url}`, async () => { + const f = fixture([() => Response.json(empty())], url); + takeSpendHome(); + const response = await handleResponses(request(false, "질문", { store: false }), f.cfg, { model: "", provider: "" }); + expect(await response.text()).toContain('"status":"completed"'); + expect(f.sends).toHaveLength(1); + }); + } +}); diff --git a/tests/responses/empty-completion-hardening.test.ts b/tests/responses/empty-completion-hardening.test.ts index 41f4cf4f874..124abc6aade 100644 --- a/tests/responses/empty-completion-hardening.test.ts +++ b/tests/responses/empty-completion-hardening.test.ts @@ -5,6 +5,7 @@ import { guardEmptyCompletionEventStream, } from "../../src/server/responses/empty-completion-guard"; import type { AdapterEvent } from "../../src/types"; +import { guardOpenGatewayCompletion } from "../../src/server/responses/opengateway-completion-guard"; async function collect(source: AsyncIterable): Promise { const events: AdapterEvent[] = []; @@ -16,6 +17,271 @@ function eventsOf(...items: AdapterEvent[]): AsyncIterable { return (async function* () { yield* items; })(); } +describe("OpenGateway native replay safety", () => { + const empty = { id: "resp_1", object: "response", status: "completed", output: [], usage: { input_tokens: 2, output_tokens: 1 } }; + const encode = (event: unknown) => new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`); + const terminal = { type: "response.completed", response: empty }; + const response = (...events: unknown[]) => new Response(new ReadableStream({ + start(controller) { for (const event of events) controller.enqueue(encode(event)); controller.close(); }, + }), { headers: { "content-type": "text/event-stream" } }); + async function run(source: Response, extra: Record = {}, signal = new AbortController().signal, timeoutMs = 1000) { + let retries = 0; + const result = await guardOpenGatewayCompletion(source, { + url: "https://apis.opengateway.ai/v1/responses", body: { stream: true, store: false, input: "question", ...extra }, + signal, holdTimeoutMs: timeoutMs, recordUsage() {}, retry: async () => { retries++; return response(terminal); }, + }); + return { result, retries: () => retries }; + } + + for (const output of [ + [{ type: "message", content: [{ type: "output_text", text: "answer" }] }], + [{ type: "message", content: [{ type: "refusal", refusal: "declined" }] }], + [{ type: "function_call", call_id: "call_1", name: "test", arguments: "{}" }], + [{ type: "web_search_call", status: "completed" }], + [{ type: "mcp_call", name: "external_action" }], + ]) { + test(`meaningful ${output[0]?.type} output remains byte-identical without retry`, async () => { + const event = { type: "response.completed", response: { ...empty, output } }; + const source = response(event), expected = await source.clone().text(); + const f = await run(source); + expect(await f.result.text()).toBe(expected); + expect(f.retries()).toBe(0); + }); + } + + for (const event of [ + { type: "response.future_action", value: "unknown" }, + { type: "response.heartbeat", replayUnsafe: true }, + { type: "response.output_item.added", output_index: 0, item: { type: "future_tool" } }, + { type: "response.created", response: { ...empty, output: [{ type: "future_tool" }] } }, + ]) { + test(`unknown or unsafe ${event.type} cannot authorize replay`, async () => { + const f = await run(response(event, terminal)); + const text = await f.result.text(); + expect(text).toContain(JSON.stringify(event)); + expect(text).toContain('"type":"response.completed"'); + expect(f.retries()).toBe(0); + }); + } + + test("a tool already emitted remains meaningful even when the final output is empty", async () => { + const event = { type: "response.output_item.done", output_index: 0, item: { type: "function_call", name: "test", call_id: "call_1", arguments: "{}" } }; + const source = response(event, terminal), expected = await source.clone().text(); + const f = await run(source); + expect(await f.result.text()).toBe(expected); + expect(f.retries()).toBe(0); + }); + + for (const extra of [{ tools: [{ type: "mcp" }] }, { tools: [{ type: "web_search" }] }, { store: true }, { store: undefined }, { previous_response_id: "resp_old" }, { background: true }]) { + test(`request-side replay hazard ${Object.keys(extra)[0]} refuses an empty completion`, async () => { + const f = await run(response(terminal), extra); + expect(await f.result.text()).toContain('"type":"response.failed"'); + expect(f.retries()).toBe(0); + }); + } + + test("reasoning alone is not a delivered answer", async () => { + const f = await run(response({ type: "response.reasoning_summary_text.delta", delta: "reasoning" }, terminal)); + expect(await f.result.text()).toContain("empty_completion_retry_failed"); + expect(f.retries()).toBe(1); + }); + + test("the event bound releases the original stream and forbids replay", async () => { + const f = await run(response(...Array.from({ length: EMPTY_COMPLETION_MAX_BUFFERED_EVENTS + 1 }, () => ({ type: "response.heartbeat" })), terminal)); + expect(await f.result.text()).toContain('"type":"response.completed"'); + expect(f.retries()).toBe(0); + }); + + test("the byte bound releases the prefix and forbids replay", async () => { + const reasoning = { type: "response.reasoning_summary_text.delta", delta: "x".repeat(600_000) }; + const f = await run(response(reasoning, reasoning, terminal)); + const text = await f.result.text(); + expect(text).toContain(reasoning.delta); + expect(text).toContain('"type":"response.completed"'); + expect(f.retries()).toBe(0); + }); + + test("the hold deadline flushes a quiet prefix without cancelling a healthy source", async () => { + let cancelled = false; + const sourceController = Promise.withResolvers>(); + const source = new Response(new ReadableStream({ + start(controller) { sourceController.resolve(controller); controller.enqueue(encode({ type: "response.created" })); }, + cancel() { cancelled = true; return new Promise(() => {}); }, + }), { headers: { "content-type": "text/event-stream" } }); + const f = await run(source, {}, undefined, 10); + const reader = f.result.body?.getReader(); + if (!reader) throw new Error("fixture body missing"); + expect(new TextDecoder().decode((await reader.read()).value)).toContain("response.created"); + expect(cancelled).toBe(false); + const controller = await sourceController.promise; + controller.enqueue(encode(terminal)); + controller.close(); + const tail: Uint8Array[] = []; + for (;;) { const next = await reader.read(); if (next.done) break; tail.push(next.value); } + expect(new TextDecoder().decode(Buffer.concat(tail))).toContain('"type":"response.completed"'); + reader.releaseLock(); + expect(f.retries()).toBe(0); + }); + + test("abort during a pending read cancels upstream and never retries", async () => { + let cancelled = false; + const reading = Promise.withResolvers(); + const source = new Response(new ReadableStream({ + pull() { reading.resolve(); }, cancel() { cancelled = true; }, + }), { headers: { "content-type": "text/event-stream" } }); + const abort = new AbortController(); + const guarded = run(source, {}, abort.signal); + await reading.promise; + abort.abort(new DOMException("cancel", "AbortError")); + const f = await guarded; + const caught = f.result.text().catch(error => error); + expect(await caught).toBe(abort.signal.reason); + expect(cancelled).toBe(true); + expect(f.retries()).toBe(0); + }); + + test("cancellation before the first pull releases the untouched source", async () => { + let cancelled = false; + const source = new Response(new ReadableStream({ cancel() { cancelled = true; } }), + { headers: { "content-type": "text/event-stream" } }); + const f = await run(source); + await f.result.body?.cancel(); + expect(cancelled).toBe(true); + expect(f.retries()).toBe(0); + }); + + for (const output of [[], [{ id: "msg_1", type: "message", role: "assistant", content: [] }]]) { + test(`an empty ${output.length ? "message" : "output array"} stays byte-identical after text`, async () => { + const source = response( + { type: "response.output_text.delta", output_index: 0, content_index: 0, item_id: "msg_1", delta: "원문" }, + { ...terminal, response: { ...empty, output } }, + ); + const expected = await source.clone().text(); + const f = await run(source); + const text = await f.result.text(); + expect(text).toBe(expected); + expect(f.retries()).toBe(0); + }); + } + + test("contradictory item identity stays byte-identical after the first delta", async () => { + const source = response( + { type: "response.output_text.delta", output_index: 0, content_index: 0, item_id: "msg_1", delta: "원문" }, + { ...terminal, response: { ...empty, output: [{ id: "different", type: "message", content: [{ type: "output_text", text: "" }] }] } }, + ); + const expected = await source.clone().text(); + const f = await run(source); + const text = await f.result.text(); + expect(text).toBe(expected); + expect(f.retries()).toBe(0); + }); + + test("fragmented UTF-8 and CRLF framing preserves normal text", async () => { + const wire = `data: ${JSON.stringify({ ...terminal, response: { ...empty, output: [{ type: "message", content: [{ type: "output_text", text: "한글" }] }] } })}\r\n\r\n`; + const source = new Response(new ReadableStream({ + start(controller) { for (const byte of new TextEncoder().encode(wire)) controller.enqueue(Uint8Array.of(byte)); controller.close(); }, + }), { headers: { "content-type": "text/event-stream" } }); + const f = await run(source); + expect(await f.result.text()).toContain('"text":"한글"'); + expect(f.retries()).toBe(0); + }); + + for (const snapshot of [{}, { status: "completed" }, { ...empty, id: "" }, { ...empty, object: undefined }, { ...empty, output: undefined }]) { + test(`malformed terminal ${JSON.stringify(snapshot)} never retries`, async () => { + const f = await run(response({ type: "response.completed", response: snapshot })); + expect(await f.result.text()).toContain('"type":"response.failed"'); + expect(f.retries()).toBe(0); + }); + } + + test("an empty refusal is a replay barrier", async () => { + const event = { ...terminal, response: { ...empty, output: [{ type: "message", content: [{ type: "refusal", refusal: "" }] }] } }; + const source = response(event), expected = await source.clone().text(); + const f = await run(source); + expect(await f.result.text()).toBe(expected); + expect(f.retries()).toBe(0); + }); + + test("partial terminal output is byte-identical after two text parts", async () => { + const source = response( + { type: "response.output_text.delta", output_index: 0, content_index: 0, item_id: "msg_1", delta: "first" }, + { type: "response.output_text.delta", output_index: 0, content_index: 1, item_id: "msg_1", delta: "second" }, + { ...terminal, response: { ...empty, output: [{ id: "msg_1", type: "message", content: [{ type: "output_text", text: "first" }] }] } }, + ); + const expected = await source.clone().text(), f = await run(source); + expect(await f.result.text()).toBe(expected); + expect(f.retries()).toBe(0); + }); + + for (const status of [401, 403, 429, 503]) { + test(`retry HTTP ${status} keeps the original response and headers`, async () => { + const refusal = Response.json({ error: { code: "original_failure", message: "fixture" } }, + { status, headers: { "retry-after": "9", "x-should-retry": "false" } }); + let retries = 0; + const result = await guardOpenGatewayCompletion(response(terminal), { + url: "https://apis.opengateway.ai/v1/responses", body: { stream: true, store: false, input: "question" }, + signal: new AbortController().signal, recordUsage() {}, retry: async () => { retries++; return refusal; }, + }); + expect(result).toBe(refusal); + expect(result.status).toBe(status); + expect(result.headers.get("retry-after")).toBe("9"); + expect((await result.json()).error.code).toBe("original_failure"); + expect(retries).toBe(1); + }); + } + + test("post-content silence is not governed by the prefix hold deadline", async () => { + const control = Promise.withResolvers>(); + let cancelled = false; + const delta = { type: "response.output_text.delta", delta: "answer" }; + const source = new Response(new ReadableStream({ + start(controller) { control.resolve(controller); controller.enqueue(encode(delta)); }, + cancel() { cancelled = true; }, + }), { headers: { "content-type": "text/event-stream" } }); + const f = await run(source, {}, undefined, 10); + const reader = f.result.body?.getReader(); + if (!reader) throw new Error("fixture body missing"); + expect(new TextDecoder().decode((await reader.read()).value)).toBe(new TextDecoder().decode(encode(delta))); + const tail = reader.read(); + const controller = await control.promise; + const timer = setTimeout(() => { controller.enqueue(encode(terminal)); controller.close(); }, 25); + try { + expect(new TextDecoder().decode((await tail).value)).toBe(new TextDecoder().decode(encode(terminal))); + expect((await reader.read()).done).toBe(true); + expect(cancelled).toBe(false); + expect(f.retries()).toBe(0); + } finally { clearTimeout(timer); reader.releaseLock(); } + }); + + test("configured stall timeout remains active and cancels a hanging source", async () => { + let cancelled = false; + const source = new Response(new ReadableStream({ + cancel() { cancelled = true; return new Promise(() => {}); }, + }), { headers: { "content-type": "text/event-stream" } }); + const result = await guardOpenGatewayCompletion(source, { + url: "https://apis.opengateway.ai/v1/responses", body: { stream: true, store: false, input: "question" }, + signal: new AbortController().signal, holdTimeoutMs: 1000, stallTimeoutMs: 10, + recordUsage() {}, retry: async () => { throw new Error("unexpected retry"); }, + }); + await expect(result.text()).rejects.toHaveProperty("name", "TimeoutError"); + expect(cancelled).toBe(true); + }); + + test("a replay-unsafe terminal passes through without replay", async () => { + const source = response({ ...terminal, replayUnsafe: true }), expected = await source.clone().text(); + const f = await run(source); + expect(await f.result.text()).toBe(expected); + expect(f.retries()).toBe(0); + }); + + test("JSON above the inspection byte cap is returned unchanged", async () => { + const source = Response.json({ ...empty, output: [{ type: "message", content: [{ type: "output_text", text: "x".repeat(EMPTY_COMPLETION_MAX_BUFFERED_BYTES + 1) }] }] }); + const expected = await source.clone().text(), f = await run(source); + expect(await f.result.text()).toBe(expected); + expect(f.retries()).toBe(0); + }); +}); + describe("empty-completion guard hardening", () => { // #6876: Vertex fails a turn with MALFORMED_FUNCTION_CALL after dropping the model's only call. const malformed = (usage?: { inputTokens: number; outputTokens: number }): AdapterEvent => ({ From 16263b787fb7ddf8e42db368a192d039ec9cc62b Mon Sep 17 00:00:00 2001 From: HeiTuz Date: Mon, 19 Oct 2026 02:26:34 +0900 Subject: [PATCH 3/3] =?UTF-8?q?fix(opengateway):=20=EB=8F=84=EA=B5=AC=20?= =?UTF-8?q?=EA=B2=B0=EA=B3=BC=20=ED=8E=98=EC=96=B4=EB=A7=81=EA=B3=BC=20?= =?UTF-8?q?=EB=B9=88=20=EC=99=84=EB=A3=8C=20=EA=B2=80=EC=82=AC=20=EA=B3=84?= =?UTF-8?q?=EC=95=BD=EC=9D=84=20=EB=B6=84=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 입력 정규화는 호출 쌍을 유지하고 미지원 part를 성공 문자열로 바꾸지 않는다. 빈 완료 검사는 원본 바이트를 넘긴 뒤 참조를 놓고, terminal/cancel에서만 upstream read를 취소한다. 재작성 응답은 도구 소유권과 시도별 usage를 유지하며 물리 재시도는 한 번이다. --- .../src/content/docs/guides/providers.md | 2 +- .../src/content/docs/ko/guides/providers.md | 2 +- scripts/test-layout/layout.json | 2 + .../openai-responses/opengateway-input.ts | 137 ++++++++---- .../responses/opengateway-completion-guard.ts | 64 ++++-- structure/providers-and-adapters.md | 4 +- .../adapters/openai/opengateway-input.test.ts | 113 ++++++++++ tests/fixtures/test-layout-expected.json | 4 +- tests/providers/opengateway-provider.test.ts | 2 +- .../opengateway-completion-receipts.test.ts | 210 ++++++++++++++++++ 10 files changed, 471 insertions(+), 69 deletions(-) create mode 100644 tests/adapters/openai/opengateway-input.test.ts create mode 100644 tests/responses/opengateway-completion-receipts.test.ts diff --git a/docs-site/src/content/docs/guides/providers.md b/docs-site/src/content/docs/guides/providers.md index ae5e593f7ab..c814f37ce66 100644 --- a/docs-site/src/content/docs/guides/providers.md +++ b/docs-site/src/content/docs/guides/providers.md @@ -736,7 +736,7 @@ Create a key in the [OpenGateway dashboard](https://opengateway.ai/api-keys), th use the configured Bearer key; the public model list does not validate that key. On the Responses wire OpenGateway rejects replayed native custom tool calls, so the preset lowers Codex custom tools such as code-mode `exec` to function tools before sending them. -Tool-result content arrays are sent as text with images preserved in adjacent messages. +Tool-result content arrays are sent as text, with images preserved after the paired tool-result group. Unsupported parts, including refusals, stay on the original array instead of being rewritten into a successful string. Image detail `original` maps to `high`, which the gateway accepts; image bytes remain unchanged. The native Responses route also checks text-only empty completions before exposing them as success. Exhausted output budgets produce `incomplete` with `max_output_tokens`; the proxy does not increase diff --git a/docs-site/src/content/docs/ko/guides/providers.md b/docs-site/src/content/docs/ko/guides/providers.md index c747989698f..e407610b489 100644 --- a/docs-site/src/content/docs/ko/guides/providers.md +++ b/docs-site/src/content/docs/ko/guides/providers.md @@ -278,7 +278,7 @@ Chat Completions 모델(그리고 Responses 전용이라 Responses로 보내는 [OpenGateway 대시보드](https://opengateway.ai/api-keys)에서 키를 발급한 뒤 `ocx provider add opengateway`를 실행하거나 대시보드에서 **OpenGateway**를 선택하세요. Chat 요청은 설정한 Bearer 키를 사용하며 공개 모델 목록은 키의 유효성을 검증하지 않습니다. -Responses 도구 결과 배열은 텍스트로 변환하고 이미지는 인접 메시지에 보존합니다. +Responses 도구 결과 배열은 텍스트로 변환하고 이미지는 짝이 맞는 도구 결과 묶음 뒤의 메시지에 보존합니다. refusal을 포함한 지원하지 않는 part는 성공 문자열로 바꾸지 않고 원래 배열에 남깁니다. 이미지의 `original` 상세도는 게이트웨이가 지원하는 `high`로 바꾸며 이미지 바이트는 그대로 유지합니다. 네이티브 Responses 경로는 텍스트가 없는 완료 응답도 검사합니다. 출력 예산이 소진됐으면 `incomplete / max_output_tokens`로 알리며 요청한 예산을 임의로 늘리지 않습니다. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index edab985cdac..c2bf6197724 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -924,6 +924,7 @@ "empty-completion-core.test.ts": "responses", "empty-completion-guard.test.ts": "responses", "empty-completion-hardening.test.ts": "responses", + "opengateway-completion-receipts.test.ts": "responses", "empty-tool-output-annotation.test.ts": "adapters", "ensure-desired-integrations-race.test.ts": "cli", "error-fidelity.test.ts": "server", @@ -1350,6 +1351,7 @@ "opencode-zen-deepseek-reasoning.test.ts": "providers", "opencode-zen-gemini-wire.test.ts": "providers", "opencode-zen-rate-limit.test.ts": "providers", "opengateway-provider.test.ts": "providers", + "opengateway-input.test.ts": "adapters/openai", "opengateway-key-validation.test.ts": "providers", "openrouter-provider-routing.test.ts": "providers", "openrouter-quota-reset-cooldown-4024.test.ts": "providers", diff --git a/src/adapters/openai-responses/opengateway-input.ts b/src/adapters/openai-responses/opengateway-input.ts index 694b21f19dc..4c7e101a1ef 100644 --- a/src/adapters/openai-responses/opengateway-input.ts +++ b/src/adapters/openai-responses/opengateway-input.ts @@ -1,64 +1,97 @@ import { isPlainObject } from "./internal"; import { normalizeResponsesToolResultAdjacency } from "./tool-output-recovery"; +const TOOL_ROUND_TYPES = new Set([ + "function_call", + "function_call_output", + "custom_tool_call", + "custom_tool_call_output", + "local_shell_call", + "local_shell_call_output", +]); + function normalizeImageDetail(part: unknown): unknown { return isPlainObject(part) && part.type === "input_image" && part.detail === "original" ? { ...part, detail: "high" } : part; } -/** OpenGateway requires string tool results and auto/low/high image detail. */ -export function normalizeOpenGatewayResponsesInput(body: unknown, url: string): unknown { - const destination = new URL(url); - if (destination.origin !== "https://apis.opengateway.ai" || destination.pathname !== "/v1/responses" - || !isPlainObject(body) || !Array.isArray(body.input)) return body; +function isToolRound(item: unknown): boolean { + return isPlainObject(item) && typeof item.type === "string" && TOOL_ROUND_TYPES.has(item.type); +} +function pairedCallId(item: Record): string | undefined { + return typeof item.call_id === "string" && item.call_id.length > 0 ? item.call_id : undefined; +} + +type ProjectedInput = { readonly changed: boolean; readonly input: unknown[] }; + +function projectOpenGatewayInput(input: readonly unknown[]): ProjectedInput { let changed = false; - const input: unknown[] = []; + const projected: unknown[] = []; const pendingImages: unknown[] = []; - for (const item of body.input) { - if (!isPlainObject(item) || item.type !== "function_call_output") { - input.push(...pendingImages); - pendingImages.length = 0; + const flushImages = (): void => { + if (pendingImages.length === 0) return; + projected.push(...pendingImages); + pendingImages.length = 0; + }; + for (const item of input) { + if (!isToolRound(item)) flushImages(); + if (!isPlainObject(item)) { + projected.push(item); + continue; } - if (!isPlainObject(item)) { input.push(item); continue; } if (item.type === "function_call_output" && Array.isArray(item.output)) { + const callId = pairedCallId(item); const content: Record[] = []; const text: string[] = []; - let supported = true; + let supported = callId !== undefined; let hasImage = false; - for (const part of item.output) { - if (!isPlainObject(part)) { supported = false; break; } - switch (part.type) { - case "text": - case "input_text": - case "output_text": - if (typeof part.text !== "string") { supported = false; break; } - text.push(part.text); - content.push({ type: "input_text", text: part.text }); - break; - case "refusal": - if (typeof part.refusal !== "string") { supported = false; break; } - text.push(`[refusal] ${part.refusal}`); - content.push({ type: "input_text", text: `[refusal] ${part.refusal}` }); - break; - case "input_image": - if (typeof part.image_url !== "string" && typeof part.file_id !== "string") { supported = false; break; } - content.push({ ...part, ...(part.detail === "original" ? { detail: "high" } : {}) }); - hasImage = true; - break; - default: + if (supported) { + for (const part of item.output) { + if (!isPlainObject(part)) { supported = false; + break; + } + switch (part.type) { + case "text": + case "input_text": + case "output_text": + if (typeof part.text !== "string") { + supported = false; + break; + } + text.push(part.text); + content.push({ type: "input_text", text: part.text }); + break; + case "input_image": + if (typeof part.image_url !== "string" && typeof part.file_id !== "string") { + supported = false; + break; + } + content.push({ ...part, ...(part.detail === "original" ? { detail: "high" } : {}) }); + hasImage = true; + break; + default: + supported = false; + } + if (!supported) break; } - if (!supported) break; } - // Opaque or unsupported parts must remain visible to validation, never disappear. - if (supported) { + if (supported && callId !== undefined) { changed = true; - input.push({ ...item, output: text.join("\n") }); - if (hasImage) pendingImages.push({ type: "message", role: "user", content: [ - { type: "input_text", text: `[tool output for ${item.call_id}]` }, ...content, - ] }); + const marker = `[tool output for ${callId}]`; + projected.push({ + ...item, + output: text.length > 0 ? text.join("\n") : hasImage ? marker : "", + }); + if (hasImage) { + pendingImages.push({ + type: "message", + role: "user", + content: [{ type: "input_text", text: marker }, ...content], + }); + } continue; } } @@ -67,12 +100,28 @@ export function normalizeOpenGatewayResponsesInput(body: unknown, url: string): const content = originalContent.map(normalizeImageDetail); if (content.some((part, index) => part !== originalContent[index])) { changed = true; - input.push({ ...item, content }); + projected.push({ ...item, content }); continue; } } - input.push(item); + projected.push(item); + } + flushImages(); + return { changed, input: projected }; +} + +/** OpenGateway requires string tool results and auto/low/high image detail. */ +export function normalizeOpenGatewayResponsesInput(body: unknown, url: string): unknown { + const destination = new URL(url); + if (destination.origin !== "https://apis.opengateway.ai" || destination.pathname !== "/v1/responses" + || !isPlainObject(body) || !Array.isArray(body.input)) return body; + + const direct = projectOpenGatewayInput(body.input); + if (!direct.changed) return body; + const aligned = normalizeResponsesToolResultAdjacency(body); + if (!isPlainObject(aligned) || !Array.isArray(aligned.input) || aligned === body) { + return { ...body, input: direct.input }; } - input.push(...pendingImages); - return changed ? normalizeResponsesToolResultAdjacency({ ...body, input }) : body; + const paired = projectOpenGatewayInput(aligned.input); + return { ...aligned, input: paired.changed ? paired.input : direct.input }; } diff --git a/src/server/responses/opengateway-completion-guard.ts b/src/server/responses/opengateway-completion-guard.ts index 216930064fc..f403b48d819 100644 --- a/src/server/responses/opengateway-completion-guard.ts +++ b/src/server/responses/opengateway-completion-guard.ts @@ -72,33 +72,51 @@ async function inspect(response: Response, options: OpenGatewayCompletionGuardOp const reader = guarded.response.body?.getReader(); if (!reader) return { kind: "pass", response }; const framer = new BoundedSseFrameBuffer(MAX_BYTES); - const held: Uint8Array[] = []; + const held: Array = []; let bytes = 0, events = 0, transferred = false, cleaned = false; let pending: ReturnType | undefined; const holdMs = options.holdTimeoutMs ?? MAX_HOLD_MS; const deadline = performance.now() + holdMs; let timer: ReturnType | undefined; const expired = new Promise<"expired">(resolve => { timer = setTimeout(() => resolve("expired"), holdMs); }); - const cleanup = () => { + const dropHeld = (): void => { + for (let index = 0; index < held.length; index += 1) held[index] = undefined; + }; + const finishOwned = (): void => { + if (cleaned) return; + cleaned = true; + guarded.dispose(); + try { reader.releaseLock(); } + catch (error) { if (!(error instanceof TypeError)) throw error; } + }; + const cancelOwned = (reason?: unknown): void => { if (cleaned) return; cleaned = true; + dropHeld(); + try { void reader.cancel(reason).catch(() => {}); } + catch (error) { if (!(error instanceof TypeError)) throw error; } guarded.dispose(); - reader.releaseLock(); }; const pass = (): Inspection => { transferred = true; let index = 0; const body = new ReadableStream({ async pull(controller) { - if (index < held.length) { controller.enqueue(held[index++]); return; } + if (index < held.length) { + const chunk = held[index]; + held[index] = undefined; + index += 1; + if (chunk) controller.enqueue(chunk); + return; + } try { const next = await (pending ?? reader.read()); pending = undefined; - if (next.done) { cleanup(); controller.close(); } + if (next.done) { finishOwned(); controller.close(); } else controller.enqueue(next.value); - } catch (error) { cleanup(); controller.error(error); } + } catch (error) { cancelOwned(error); controller.error(error); } }, - cancel() { cleanup(); }, + cancel(reason) { dropHeld(); cancelOwned(reason); }, }, { highWaterMark: 0 }); return { kind: "pass", response: carryResponseToolOwnership(response, new Response(body, { status: response.status, statusText: response.statusText, headers: response.headers, @@ -128,7 +146,7 @@ async function inspect(response: Response, options: OpenGatewayCompletionGuardOp if (!streaming) { if (!next.done) continue; let snapshot: unknown; - try { snapshot = JSON.parse(decoder.decode(joinSseFrameBytes(held))); } + try { snapshot = JSON.parse(decoder.decode(joinSseFrameBytes(held.flatMap(chunk => chunk ? [chunk] : [])))); } catch { return terminal(undefined); } if (isPlainObject(snapshot) && (outputCommits(snapshot.output) || snapshot.error != null || ["failed", "incomplete", "queued", "in_progress"].includes(String(snapshot.status)))) return pass(); @@ -156,24 +174,31 @@ async function inspect(response: Response, options: OpenGatewayCompletionGuardOp } } finally { clearTimeout(timer); framer.dispose(); - if (!transferred) cleanup(); + if (!transferred) cancelOwned(); } } function terminalResponse(source: Response, inspection: Exclude, reason: string): Response { const incomplete = reason === "max_output_tokens"; - const snapshot = { ...inspection.snapshot, status: incomplete ? "incomplete" : "failed", - ...(incomplete ? { incomplete_details: { reason } } : { error: { + const snapshot: RecordValue = { ...inspection.snapshot, status: incomplete ? "incomplete" : "failed" }; + if (incomplete) { + snapshot.incomplete_details = { reason }; + delete snapshot.error; + } else { + snapshot.error = { type: "upstream_error", code: reason, message: "OpenGateway returned no usable completion before output.", - } }), - }; + }; + delete snapshot.incomplete_details; + } const headers = new Headers(source.headers); headers.delete("content-length"); headers.delete("content-encoding"); if (inspection.streaming) headers.set("content-type", "text/event-stream"); const text = inspection.streaming ? `event: response.${snapshot.status}\ndata: ${JSON.stringify({ type: `response.${snapshot.status}`, response: snapshot })}\n\n` : JSON.stringify(snapshot); - return new Response(text, { status: inspection.streaming || incomplete ? 200 : 502, headers }); + return carryResponseToolOwnership(source, new Response(text, { + status: inspection.streaming || incomplete ? 200 : 502, headers, + })); } export async function guardOpenGatewayCompletion(response: Response, options: OpenGatewayCompletionGuardOptions): Promise { @@ -185,19 +210,22 @@ export async function guardOpenGatewayCompletion(response: Response, options: Op const canReplay = options.retryEnabled !== false && process.env[EMPTY_COMPLETION_RETRY_ENV] !== "0" && selfContainedResponsesBody(body) && !isNonReplayableResponse(response); let selected = response; + let physicalRetries = 0; for (let attempt = 0; ; attempt++) { const inspected = await inspect(selected, options); if (inspected.kind === "pass") return inspected.response; - if (inspected.kind === "invalid") return terminalResponse(selected, inspected, "invalid_completion"); const usage = inspected.snapshot.usage; + options.recordUsage(usage); + if (inspected.kind === "invalid") return terminalResponse(selected, inspected, "invalid_completion"); if (typeof body.max_output_tokens === "number" && body.max_output_tokens > 0 && isPlainObject(usage) && typeof usage.output_tokens === "number" && usage.output_tokens >= body.max_output_tokens) { return terminalResponse(selected, inspected, "max_output_tokens"); } - if (attempt > 0 || !canReplay || options.signal.aborted) { - return terminalResponse(selected, inspected, attempt > 0 ? "empty_completion_retry_failed" : "empty_completion"); + if (attempt > 0 || physicalRetries > 0 || !canReplay || options.signal.aborted) { + return terminalResponse(selected, inspected, attempt > 0 || physicalRetries > 0 + ? "empty_completion_retry_failed" : "empty_completion"); } - options.recordUsage(usage); + physicalRetries += 1; selected = await options.retry(); if (selected.status !== 200) return selected; } diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index 911401f884a..fa1dc1ffa61 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -323,11 +323,11 @@ Responses (`openai/o3-pro`); public discovery does not establish key validity. S `deepseek/deepseek-v4.1-flash-ultrafast` and `z-ai/glm-5.3-flash-ultrafast` are listed first. A custom model that replaces one of the discovered rows keeps that row's slot (`src/codex/catalog/routed-gather.ts`). Keys are created at `https://opengateway.ai/api-keys`. -`src/adapters/openai-responses/opengateway-input.ts`는 OpenGateway의 `/v1/responses` 전송 직전에 텍스트 도구 결과 배열을 문자열로 바꾼다. `call_id`와 텍스트 순서는 유지한다. 이미지가 섞인 결과는 원본 이미지와 내용 순서를 사용자 메시지로 보존하며 병렬 호출의 결과 묶음 뒤에 둔다. 이미지의 `detail: original`은 upstream이 수용하는 `high`로 바꾸되 이미지 바이트는 바꾸지 않는다. 알 수 없는 형식과 암호화된 결과 배열은 조용히 삭제하지 않고 기존 검증에 남긴다. 다른 목적지와 원본 요청 객체는 바꾸지 않는다. `tests/providers/opengateway-provider.test.ts`가 실제 adapter 직렬화 경계에서 이 계약을 확인한다. +`src/adapters/openai-responses/opengateway-input.ts`는 OpenGateway의 `/v1/responses` 전송 직전에 텍스트 도구 결과 배열을 문자열로 바꾼다. `call_id`와 텍스트 순서는 유지한다. 이미지가 섞인 결과는 원본 이미지와 내용 순서를 사용자 메시지로 보존하며, 인접성 정렬 뒤의 도구 결과 묶음 뒤에 두어 call과 output 사이를 가르지 않는다. 이미지만 있는 결과는 빈 성공 문자열이 아니라 같은 `call_id` 표시를 남긴다. 이미지의 `detail: original`은 upstream이 수용하는 `high`로 바꾸되 이미지 바이트는 바꾸지 않는다. refusal을 포함한 지원하지 않는 part와 암호화된 결과 배열은 성공 문자열로 바꾸지 않고 기존 검증에 남긴다. 다른 목적지와 원본 요청 객체는 바꾸지 않는다. `tests/adapters/openai/opengateway-input.test.ts`가 입력 정규화를 빈 완료 복구와 분리해 검증하고, `tests/providers/opengateway-provider.test.ts`가 adapter 직렬화 경계에서 같은 계약을 확인한다. `src/server/responses/opengateway-completion-guard.ts`는 정확한 `https://apis.opengateway.ai/v1/responses`의 HTTP 200 SSE/JSON에서 출력 전 빈 완료만 검사한다. 첫 텍스트, refusal(빈 refusal 포함), action 또는 unknown 이벤트부터는 원본 바이트를 그대로 전달하며 snapshot을 재구성하지 않는다. 정상 형식의 빈 terminal에서 reported `output_tokens >= outbound max_output_tokens`이면 재송신 없이 `incomplete/max_output_tokens`, 형식이 불완전하면 `invalid_completion` 실패다. Reasoning은 답변이 아니다. 미소진 빈 완료는 실제 outbound body가 `selfContainedResponsesBody`를 통과할 때만 한 번 재시도한다. 명시적 `store:false`와 client-only tools가 필요하며 store 생략/true, continuation, background, hosted/unknown tool은 `empty_completion` 실패로 끝난다. `web_search`가 남아 있는 Codex 요청도 재생하지 않는다. 이 목적지에서는 설정 미지정 시 복구하되 `emptyCompletionRetry:false`와 `OCX_EMPTY_COMPLETION_RETRY=0`을 존중하며, 다른 목적지와 translated opt-in은 바뀌지 않는다. -`src/server/responses/passthrough-completion.ts`가 기존 `rebuildAndRefetch("empty-completion")`의 pacing·accounting·abort·send budget에 연결하며 복구 물리 전송은 한 번이다. 첫 usage는 새 전송 전에 기존 attempt baseline에 기록한다. 두 번째 정상 형식의 빈 200만 `empty_completion_retry_failed`이며 재시도 HTTP 실패와 send-budget 거절은 원래 Response를 delivery에 넘긴다. 1 MiB/1,024개 이벤트·청크/30초 보관 한도에 닿으면 원본 prefix와 진행 중인 read를 전달하고 재시도를 닫는다. 보관 한도는 upstream을 취소하지 않으며 SSE stall은 기존 `resolveStallTimeoutMs`와 0-disable 정책을 따른다. timer·framer·reader cleanup 및 원본 바이트 보존은 `tests/responses/empty-completion-core.test.ts`, `tests/responses/empty-completion-hardening.test.ts`가 검증한다. +`src/server/responses/passthrough-completion.ts`가 기존 `rebuildAndRefetch("empty-completion")`의 pacing·accounting·abort·send budget에 연결하며 복구 물리 전송은 한 번이다. 첫 usage는 새 전송 전에 기존 attempt baseline에 기록한다. 두 번째 정상 형식의 빈 200만 `empty_completion_retry_failed`이며 재시도 HTTP 실패와 send-budget 거절은 원래 Response를 delivery에 넘긴다. 1 MiB/1,024개 이벤트·청크/30초 보관 한도에 닿으면 원본 prefix와 진행 중인 read를 전달하고 재시도를 닫는다. 보관 한도는 upstream을 취소하지 않으며 SSE stall은 기존 `resolveStallTimeoutMs`와 0-disable 정책을 따른다. 보관 한도와 hold 만료는 원본 청크를 소비자에게 넘긴 뒤 검사 참조를 놓고 upstream을 취소하지 않는다. 진정 terminal과 소비자 cancel만 소유한 upstream read를 취소하며 그 promise를 기다리지 않는다. 재작성된 응답은 도구 소유권을 유지하고, 소비한 시도의 usage는 시도마다 한 번씩 기록한다. 물리 재시도는 한 번이다. `tests/responses/opengateway-completion-receipts.test.ts`가 SSE/JSON, reason-only/빈 출력, max-output incomplete, stalled/cancelled read, replay-unsafe를 exact head로 검증하고, `tests/responses/empty-completion-core.test.ts`와 `tests/responses/empty-completion-hardening.test.ts`가 통합 경로를 검증한다. ## TokenLab chat provider diff --git a/tests/adapters/openai/opengateway-input.test.ts b/tests/adapters/openai/opengateway-input.test.ts new file mode 100644 index 00000000000..15716d2226e --- /dev/null +++ b/tests/adapters/openai/opengateway-input.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, test } from "bun:test"; +import { normalizeOpenGatewayResponsesInput } from "../../../src/adapters/openai-responses/opengateway-input"; + +const url = "https://apis.opengateway.ai/v1/responses"; +const image = (id: string, detail = "original") => ({ + type: "input_image", + image_url: `data:image/png;base64,${id}`, + detail, +}); + +function project(input: unknown[], destination = url): unknown[] { + const body = { model: "m", input }; + const snapshot = JSON.stringify(body); + const normalized = normalizeOpenGatewayResponsesInput(body, destination) as { input: unknown[] }; + expect(JSON.stringify(body)).toBe(snapshot); + return normalized.input; +} + +describe("OpenGateway input normalization", () => { + test("text projection keeps call ids and does not depend on completion recovery", () => { + const input = project([ + { type: "function_call", call_id: "call_a", name: "echo", arguments: "{}" }, + { type: "function_call_output", call_id: "call_a", output: [ + { type: "input_text", text: "first" }, + { type: "output_text", text: "second" }, + ] }, + ]); + expect(input.map(item => (item as { type: string }).type)).toEqual(["function_call", "function_call_output"]); + expect(input[1]).toEqual({ + type: "function_call_output", + call_id: "call_a", + output: "first\nsecond", + }); + }); + + test("image messages stay after the aligned tool group, not between a call and its output", () => { + const input = project([ + { type: "reasoning", id: "rs_1", summary: [] }, + { type: "function_call", call_id: "a", name: "echo", arguments: "{}" }, + { type: "function_call", call_id: "b", name: "echo", arguments: "{}" }, + { type: "message", role: "developer", content: [{ type: "input_text", text: "hook" }] }, + { type: "function_call_output", call_id: "a", output: [{ type: "input_text", text: "A" }, image("AAA")] }, + { type: "function_call_output", call_id: "b", output: [{ type: "input_text", text: "B" }, image("BBB", "high")] }, + ]); + expect(input.map(item => (item as { type: string }).type)).toEqual([ + "reasoning", "function_call", "function_call", "function_call_output", "function_call_output", + "message", "message", "message", + ]); + expect(input[1]).toMatchObject({ call_id: "a" }); + expect(input[2]).toMatchObject({ call_id: "b" }); + expect(input[3]).toMatchObject({ call_id: "a", output: "A" }); + expect(input[4]).toMatchObject({ call_id: "b", output: "B" }); + expect(input[5]).toMatchObject({ + role: "user", + content: [ + { type: "input_text", text: "[tool output for a]" }, + { type: "input_text", text: "A" }, + { ...image("AAA"), detail: "high" }, + ], + }); + expect(input[6]).toMatchObject({ role: "user", content: expect.arrayContaining([image("BBB", "high")]) }); + expect(input[7]).toMatchObject({ role: "developer" }); + }); + + test("a later tool round is not split by the previous image message", () => { + const input = project([ + { type: "function_call", call_id: "a", name: "echo", arguments: "{}" }, + { type: "function_call_output", call_id: "a", output: [image("AAA")] }, + { type: "function_call", call_id: "b", name: "echo", arguments: "{}" }, + { type: "function_call_output", call_id: "b", output: [{ type: "input_text", text: "done" }] }, + ]); + expect(input.map(item => (item as { type?: string; call_id?: string }).call_id ?? (item as { type: string }).type)).toEqual([ + "a", "a", "b", "b", "message", + ]); + expect(input[1]).toMatchObject({ call_id: "a", output: "[tool output for a]" }); + expect(input[3]).toMatchObject({ call_id: "b", output: "done" }); + }); + + test("unsupported parts stay visible instead of becoming a successful string", () => { + const refusal = [{ type: "refusal", refusal: "no" }]; + const mixed = [ + { type: "input_text", text: "keep-me" }, + { type: "encrypted_content", encrypted_content: "opaque" }, + ]; + const missingImage = [ + { type: "input_text", text: "visible" }, + { type: "input_image", detail: "high" }, + ]; + const unpaired = [{ type: "input_text", text: "orphan" }]; + expect(project([ + { type: "function_call_output", call_id: "a", output: refusal }, + ])[0]).toMatchObject({ output: refusal }); + expect(project([ + { type: "function_call_output", call_id: "a", output: mixed }, + ])[0]).toMatchObject({ output: mixed }); + expect(project([ + { type: "function_call_output", call_id: "a", output: missingImage }, + ])[0]).toMatchObject({ output: missingImage }); + expect(project([ + { type: "function_call_output", output: unpaired }, + ])[0]).toMatchObject({ output: unpaired }); + }); + + test("an empty array is a string, and another destination is untouched", () => { + expect(project([ + { type: "function_call_output", call_id: "a", output: [] }, + ])[0]).toMatchObject({ output: "" }); + const output = [{ type: "input_text", text: "stay" }, image("AAA")]; + expect(project([ + { type: "function_call_output", call_id: "a", output }, + ], "https://api.openai.com/v1")[0]).toMatchObject({ output }); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index b29ec798a94..ff04e6c4b82 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -674,7 +674,7 @@ "droid-client.test.ts": "clients", "dsh-path-contract.test.ts": "ci-workflows", "dsh-rc6-compat-script.test.ts": "ci-workflows", "dsh-writer-lock.test.ts": "ci-workflows", "effort-policy.test.ts": "codex-integration", "empty-completion-core.test.ts": "responses", - "empty-completion-guard.test.ts": "responses", "empty-completion-hardening.test.ts": "responses", + "empty-completion-guard.test.ts": "responses", "empty-completion-hardening.test.ts": "responses", "opengateway-completion-receipts.test.ts": "responses", "empty-tool-output-annotation.test.ts": "adapters", "ensure-desired-integrations-race.test.ts": "cli", "error-fidelity.test.ts": "server", "errors-adapter-failure.test.ts": "server", "eventstream-decoder.test.ts": "responses", "exa-web-search.test.ts": "providers", @@ -949,7 +949,7 @@ "opencode-go-muse-vision.test.ts": "providers", "opencode-go-quota.test.ts": "providers", "opencode-go-session-header.test.ts": "providers", "opencode-management-transport.test.ts": "providers", "opencode-zen-deepseek-reasoning.test.ts": "providers", "opencode-zen-gemini-wire.test.ts": "providers", "opencode-zen-rate-limit.test.ts": "providers", - "opengateway-provider.test.ts": "providers", "opengateway-key-validation.test.ts": "providers", + "opengateway-provider.test.ts": "providers", "opengateway-key-validation.test.ts": "providers", "opengateway-input.test.ts": "adapters/openai", "openrouter-provider-routing.test.ts": "providers", "openrouter-quota-reset-cooldown-4024.test.ts": "providers", "opper-provider.test.ts": "providers", "tokenlab-protocols.test.ts": "providers", "modelflare-provider.test.ts": "providers", diff --git a/tests/providers/opengateway-provider.test.ts b/tests/providers/opengateway-provider.test.ts index 65c4df38371..e74f2780ea8 100644 --- a/tests/providers/opengateway-provider.test.ts +++ b/tests/providers/opengateway-provider.test.ts @@ -182,7 +182,7 @@ describe("OpenGateway Responses tool result arrays", () => { }); test("string and unsupported opaque result arrays are preserved", () => { - for (const output of ["exact string", [{ type: "encrypted_content", encrypted_content: "opaque" }]]) { + for (const output of ["exact string", [{ type: "encrypted_content", encrypted_content: "opaque" }], [{ type: "refusal", refusal: "no" }], [{ type: "input_text", text: "keep" }, { type: "encrypted_content", encrypted_content: "opaque" }]]) { const input = build([ { type: "function_call", call_id: "a", name: "echo", arguments: "{}" }, { type: "function_call_output", call_id: "a", output }, diff --git a/tests/responses/opengateway-completion-receipts.test.ts b/tests/responses/opengateway-completion-receipts.test.ts new file mode 100644 index 00000000000..739e17b3559 --- /dev/null +++ b/tests/responses/opengateway-completion-receipts.test.ts @@ -0,0 +1,210 @@ +import { describe, expect, test } from "bun:test"; +import { EMPTY_COMPLETION_MAX_BUFFERED_BYTES } from "../../src/server/responses/empty-completion-guard"; +import { guardOpenGatewayCompletion } from "../../src/server/responses/opengateway-completion-guard"; +import { attachResponseToolOwnership, responseToolOwnership } from "../../src/server/responses/response-tool-ownership"; + +const url = "https://apis.opengateway.ai/v1/responses"; +const encode = (event: unknown) => new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`); + +function snapshot(usage = { input_tokens: 2, output_tokens: 1 }, output: unknown[] = []) { + return { id: "resp_1", object: "response", status: "completed", output, usage, error: null, incomplete_details: null }; +} + +function sse(events: unknown[], extra: Uint8Array[] = []): Response { + return new Response(new ReadableStream({ + start(controller) { + for (const event of events) controller.enqueue(encode(event)); + for (const chunk of extra) controller.enqueue(chunk); + controller.close(); + }, + }), { headers: { "content-type": "text/event-stream" } }); +} + +async function run( + source: Response, + extra: Record = {}, + retrySource?: () => Response, + signal = new AbortController().signal, + timeouts: { holdTimeoutMs?: number; stallTimeoutMs?: number } = {}, +) { + const usages: unknown[] = []; + let retries = 0; + const result = await guardOpenGatewayCompletion(source, { + url, + body: { stream: source.headers.get("content-type")?.includes("event-stream") === true, store: false, input: "question", ...extra }, + signal, + recordUsage(usage) { usages.push(usage); }, + retry: async () => { + retries += 1; + return retrySource?.() ?? Response.json(snapshot({ input_tokens: 9, output_tokens: 8 })); + }, + ...timeouts, + }); + return { result, usages, retries }; +} + +function head(text: string): { event: string; data: { type: string; response: Record } } { + const event = text.match(/^event: (\S+)/)?.[1]; + const dataLine = text.match(/^data: (.+)$/m)?.[1]; + if (!event || !dataLine) throw new Error(`missing SSE head: ${text.slice(0, 180)}`); + return { event, data: JSON.parse(dataLine) }; +} + +describe("OpenGateway completion exact-head receipts", () => { + test("SSE empty failure head is response.failed after one physical retry", async () => { + const first = snapshot(); + const second = snapshot({ input_tokens: 4, output_tokens: 3 }); + const f = await run(sse([{ type: "response.completed", response: first }]), {}, () => sse([{ type: "response.completed", response: second }])); + const text = await f.result.text(); + const parsed = head(text); + expect(f.result.status).toBe(200); + expect(f.result.headers.get("content-type")).toBe("text/event-stream"); + expect(parsed.event).toBe("response.failed"); + expect(parsed.data.type).toBe("response.failed"); + expect(parsed.data.response.status).toBe("failed"); + expect(parsed.data.response.error).toEqual({ + type: "upstream_error", + code: "empty_completion_retry_failed", + message: "OpenGateway returned no usable completion before output.", + }); + expect(parsed.data.response.incomplete_details).toBeUndefined(); + expect(text).not.toContain("response.completed"); + expect(f.retries).toBe(1); + expect(f.usages).toEqual([first.usage, second.usage]); + }); + + test("JSON empty failure head is 502 and records each attempt once", async () => { + const first = snapshot(); + const second = snapshot({ input_tokens: 4, output_tokens: 3 }); + const owned = new Set([{ itemType: "web_search_call", callIdPrefix: "ws_" }]); + const retry = Response.json(second); + attachResponseToolOwnership(retry, owned); + const f = await run(Response.json(first), {}, () => retry); + const body = await f.result.json() as Record; + expect(f.result.status).toBe(502); + expect(f.result.headers.get("content-type")).toContain("application/json"); + expect(body.status).toBe("failed"); + expect(body.error).toEqual({ + type: "upstream_error", + code: "empty_completion_retry_failed", + message: "OpenGateway returned no usable completion before output.", + }); + expect(body.incomplete_details).toBeUndefined(); + expect(f.retries).toBe(1); + expect(f.usages).toEqual([first.usage, second.usage]); + expect(responseToolOwnership(f.result)).toEqual(owned); + }); + + test("reason-only JSON is not a delivered answer", async () => { + const reasonOnly = snapshot({ input_tokens: 3, output_tokens: 2 }, [ + { id: "rs_1", type: "reasoning", summary: [] }, + ]); + const answer = { + ...snapshot({ input_tokens: 3, output_tokens: 2 }), + id: "resp_answer", + output: [{ id: "msg_1", type: "message", role: "assistant", status: "completed", content: [{ type: "output_text", text: "answer" }] }], + }; + const f = await run(Response.json(reasonOnly), {}, () => Response.json(answer)); + expect(f.retries).toBe(1); + expect(f.usages).toEqual([reasonOnly.usage]); + expect(await f.result.json()).toMatchObject({ id: "resp_answer", status: "completed" }); + }); + + test("an empty output array is the same retry class as reason-only", async () => { + const empty = snapshot({ input_tokens: 1, output_tokens: 0 }, []); + const f = await run(Response.json(empty), {}, () => Response.json(empty)); + const body = await f.result.json() as { status: string; error: { code: string } }; + expect(body.status).toBe("failed"); + expect(body.error.code).toBe("empty_completion_retry_failed"); + expect(f.retries).toBe(1); + expect(f.usages).toEqual([empty.usage, empty.usage]); + }); + + for (const streaming of [false, true]) { + test(`${streaming ? "SSE" : "JSON"} max-output incomplete does not retry`, async () => { + const body = snapshot({ input_tokens: 6, output_tokens: 16 }); + const source = streaming + ? sse([{ type: "response.completed", response: body }]) + : Response.json(body); + const f = await run(source, { max_output_tokens: 16 }, () => { throw new Error("retry"); }); + expect(f.retries).toBe(0); + expect(f.usages).toEqual([body.usage]); + expect(f.result.status).toBe(200); + if (streaming) { + const text = await f.result.text(); + const parsed = head(text); + expect(f.result.headers.get("content-type")).toBe("text/event-stream"); + expect(parsed.event).toBe("response.incomplete"); + expect(parsed.data.response.status).toBe("incomplete"); + expect(parsed.data.response.incomplete_details).toEqual({ reason: "max_output_tokens" }); + expect(parsed.data.response.error).toBeUndefined(); + expect(text).not.toContain("response.completed"); + } else { + expect(f.result.headers.get("content-type")).toContain("application/json"); + const parsed = await f.result.json() as { status: string; incomplete_details: unknown; error?: unknown }; + expect(parsed).toMatchObject({ + status: "incomplete", + incomplete_details: { reason: "max_output_tokens" }, + }); + expect(parsed.error).toBeUndefined(); + } + }); + } + + test("a stalled read cancels the owned source and does not retry", async () => { + let cancelled = false; + const source = new Response(new ReadableStream({ + cancel() { cancelled = true; return new Promise(() => {}); }, + }), { headers: { "content-type": "text/event-stream" } }); + const f = await run(source, {}, undefined, new AbortController().signal, { stallTimeoutMs: 10, holdTimeoutMs: 1_000 }); + await expect(f.result.text()).rejects.toEqual(expect.objectContaining({ name: "TimeoutError" })); + expect(cancelled).toBe(true); + expect(f.retries).toBe(0); + }); + + test("consumer cancel resolves without waiting on a hanging upstream cancel", async () => { + let cancelled = false; + const source = new Response(new ReadableStream({ + start(controller) { controller.enqueue(encode({ type: "response.created" })); }, + cancel() { cancelled = true; return new Promise(() => {}); }, + }), { headers: { "content-type": "text/event-stream" } }); + const f = await run(source, {}, undefined, new AbortController().signal, { holdTimeoutMs: 10 }); + const outcome = await Promise.race([ + f.result.body?.cancel().then(() => "resolved"), + Bun.sleep(50).then(() => "hung"), + ]); + expect(outcome).toBe("resolved"); + expect(cancelled).toBe(true); + expect(f.retries).toBe(0); + }); + + test("a replay-unsafe control is an exact pass-through with no retry", async () => { + const source = sse([{ type: "response.completed", response: snapshot(), replayUnsafe: true }]); + const expected = await source.clone().text(); + const f = await run(source); + expect(f.result.status).toBe(200); + expect(f.result.headers.get("content-type")).toBe("text/event-stream"); + expect(await f.result.text()).toBe(expected); + expect(f.retries).toBe(0); + expect(f.usages).toEqual([]); + }); + + test("size expiry forwards the original chunk and does not cancel a healthy source", async () => { + const original = new Uint8Array(EMPTY_COMPLETION_MAX_BUFFERED_BYTES + 8); + original[0] = 7; + let cancelled = false; + const source = new Response(new ReadableStream({ + start(controller) { controller.enqueue(original); controller.close(); }, + cancel() { cancelled = true; }, + }), { headers: { "content-type": "application/json" } }); + const f = await run(source, { stream: false }); + const reader = f.result.body?.getReader(); + if (!reader) throw new Error("missing body"); + const first = await reader.read(); + expect(first.value).toBe(original); + expect((await reader.read()).done).toBe(true); + reader.releaseLock(); + expect(cancelled).toBe(false); + expect(f.retries).toBe(0); + }); +});