From 6bf579510babdb2e2807e9bafd99bd5a46867048 Mon Sep 17 00:00:00 2001 From: Omar <13569609+omar391@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:37:06 +0600 Subject: [PATCH 1/6] feat(responses): complete gracefully on max_output_tokens When an upstream model hits max_output_tokens / max_tokens, Codex CLI and consumers can fail (e.g. 'Error running remote compact task: Incomplete response returned, reason: max_output_tokens', retry loop exhaustion, or parsing errors on open markdown code fences). - Adds closeOpenMarkdownBlocks to close dangling fences and paired tags. - Introduces completeOnMaxOutputTokens config option (defaults to true). - In bridge SSE, response JSON, and codex-ws exchange, gracefully completes truncated turns with status completed, repairs markdown fences, and synthesizes compaction items when compaction was requested. --- src/bridge/response-json.ts | 33 ++- src/bridge/sse.ts | 70 ++++- src/config/schema/config-schema.ts | 1 + src/responses/markdown-block-repair.ts | 39 +++ src/server/responses/adapter-delivery.ts | 2 + src/server/responses/codex-ws-exchange.ts | 75 ++++- src/server/responses/run-turn-execution.ts | 2 + src/server/responses/ws-upstream.ts | 2 + src/types/config.ts | 1 + .../graceful-max-output-tokens.test.ts | 268 ++++++++++++++++++ 10 files changed, 470 insertions(+), 23 deletions(-) create mode 100644 src/responses/markdown-block-repair.ts create mode 100644 tests/responses/graceful-max-output-tokens.test.ts diff --git a/src/bridge/response-json.ts b/src/bridge/response-json.ts index a699947161c..72f4684ff78 100644 --- a/src/bridge/response-json.ts +++ b/src/bridge/response-json.ts @@ -35,6 +35,7 @@ import { } from "../responses/citation-markers"; import { declaresCodeModeExec, normalizeDeclaredToolName } from "../types"; import { appendSafeWebSearchSource, safeWebSearchSources } from "../web-search/sources"; +import { closeOpenMarkdownBlocks } from "../responses/markdown-block-repair"; import { isTranslatorBudgetExceededError, releaseTranslatedEvent, @@ -117,6 +118,7 @@ function buildResponseJSONWithBudget( * so a block that fit live delivery could otherwise overflow here and lose the cache write. */ omitHiddenReasoningEnvelope?: boolean; + completeOnMaxOutputTokens?: boolean; }, ): Record { const responseId = `resp_${uuid()}`; @@ -619,30 +621,47 @@ function buildResponseJSONWithBudget( batchKiroRedacted = undefined; batchKiroRedactedBytes = 0; } + const gracefulComplete = Boolean(options?.completeOnMaxOutputTokens) + && (stopReason === "max_tokens" || (incompleteEvent && (truncationReasonFor(incompleteEvent.reason) === "max_output_tokens" || incompleteEvent.reason === "max_output_tokens"))); + // A truncated turn must never be installed as replacement history: emit the // compaction item only when the turn actually completed (#422). if ( options?.compaction && !errorEvent - && !incompleteEvent + && (!incompleteEvent || gracefulComplete) // A stream that stopped without any terminal did not complete either. The original guard // could only see explicit failure events, so an adapter EOF slipped past it and installed a // truncated summary as replacement history — the exact #422 hazard, reached by a route that // did not exist when the guard was written. && sawTerminal - && !isTruncatedStopReason(rawStopReason) + && (!isTruncatedStopReason(rawStopReason) || gracefulComplete) ) { - const item = { + const summary = joinChunks(batchCompaction); + const fallbackSummary = summary.trim().length > 0 ? summary : "[earlier conversation was compacted]"; + const item = { type: "compaction", id: `cmp_${uuid()}`, - encrypted_content: compactionEncryptedContent ?? (options?.compactionRetention?.encode(joinChunks(batchCompaction)) ?? encodeCompactionSummary(joinChunks(batchCompaction))), + encrypted_content: compactionEncryptedContent ?? (options?.compactionRetention?.encode(fallbackSummary) ?? encodeCompactionSummary(fallbackSummary)), }; pushOutput(item, compactionEncryptedContent ? 0 : batchCompaction.bytes); } + if (gracefulComplete) { + for (const item of output) { + if (typeof item === "object" && item !== null && (item as any).type === "message" && Array.isArray((item as any).content)) { + for (const part of (item as any).content) { + if (typeof part === "object" && part !== null && part.type === "output_text" && typeof part.text === "string") { + part.text = closeOpenMarkdownBlocks(part.text); + } + } + } + } + } + const failure = errorEvent ? adapterFailureFromEvent(errorEvent) : undefined; const status = errorEvent ? "failed" - : incompleteEvent || stopReason === "max_tokens" || stopReason === "content_filter" + : ((incompleteEvent || stopReason === "max_tokens" || stopReason === "content_filter") && !gracefulComplete) ? "incomplete" : sawTerminal ? "completed" @@ -662,13 +681,13 @@ function buildResponseJSONWithBudget( ...(failure && isCyberPolicyCode(failure.error.code) ? { retryable: false } : errorEvent?.retryable !== undefined ? { retryable: errorEvent.retryable } : {}), - ...(incompleteEvent ? { + ...(!gracefulComplete && incompleteEvent ? { incomplete_details: { reason: truncationReasonFor(incompleteEvent.reason) ?? incompleteEvent.reason, ...(incompleteEvent.message ? { message: incompleteEvent.message } : {}), ...(incompleteEvent.retryable !== undefined ? { retryable: incompleteEvent.retryable } : {}), }, - } : stopReason === "max_tokens" ? { + } : (!gracefulComplete && stopReason === "max_tokens") ? { incomplete_details: { reason: "max_output_tokens" }, } : stopReason === "content_filter" ? { incomplete_details: { reason: "content_filter" }, diff --git a/src/bridge/sse.ts b/src/bridge/sse.ts index fbbeefcd002..2378dd17f65 100644 --- a/src/bridge/sse.ts +++ b/src/bridge/sse.ts @@ -43,6 +43,7 @@ import { } from "../responses/citation-markers"; import { declaresCodeModeExec, normalizeDeclaredToolName } from "../types"; import { appendSafeWebSearchSource, safeWebSearchSources } from "../web-search/sources"; +import { closeOpenMarkdownBlocks } from "../responses/markdown-block-repair"; import { isTranslatorBudgetExceededError, releaseTranslatedEvent, @@ -162,6 +163,7 @@ export function bridgeToResponsesSSE( setInterval: (handler: () => void, ms: number) => unknown; clearInterval: (id: unknown) => void; }; + completeOnMaxOutputTokens?: boolean; }, ): ReadableStream { const replayCacheScope = options?.replayCacheScope; @@ -502,7 +504,7 @@ export function bridgeToResponsesSSE( return anns; }; - const closeCurrentMessage = (inferredPhase?: OcxMessagePhase) => { + const closeCurrentMessage = (inferredPhase?: OcxMessagePhase, gracefulRepair?: boolean) => { if (!currentMsg) return; // Release anything the citation filter was holding for this message, then strip the // accumulated text: closeCurrentMessage re-sends it in output_text.done and @@ -514,7 +516,18 @@ export function bridgeToResponsesSSE( content_index: 0, delta: trailing, }); } - const messageText = stripCitationMarkers(joinChunks(currentMsg.text)); + let messageText = stripCitationMarkers(joinChunks(currentMsg.text)); + if (gracefulRepair) { + const repaired = closeOpenMarkdownBlocks(messageText); + if (repaired !== messageText) { + const delta = repaired.slice(messageText.length); + emit("response.output_text.delta", { + item_id: currentMsg.itemId, output_index: currentMsg.outputIndex, + content_index: 0, delta, + }); + messageText = repaired; + } + } // Chat Completions has no message-phase field. Keep its live item provisional, then // classify it only when the next adapter event proves whether this text led into more // work or completed the turn. Explicit adapter phases always outrank this inference. @@ -1256,17 +1269,21 @@ export function bridgeToResponsesSSE( break; } case "done": { - if (currentMsg) closeCurrentMessage(isTruncatedStopReason(event.stopReason) ? undefined : "final_answer"); + const isTruncated = isTruncatedStopReason(event.stopReason); + const truncationReason = truncationReasonFor(event.stopReason); + const gracefulComplete = Boolean(options?.completeOnMaxOutputTokens) && truncationReason === "max_output_tokens"; + + if (currentMsg) closeCurrentMessage(isTruncated && !gracefulComplete ? undefined : "final_answer", gracefulComplete); if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); if (currentToolCall) { - if (isTruncatedStopReason(event.stopReason)) failCurrentToolCall(); + if (isTruncated) failCurrentToolCall(); else closeCurrentToolCall(); } // A search still in flight when upstream truncates never returned results, so it // takes the same "failed" status as the error/incomplete terminals below. - if (currentWebSearch) closeCurrentWebSearch(isTruncatedStopReason(event.stopReason) ? "failed" : "completed", []); + if (currentWebSearch) closeCurrentWebSearch(isTruncated ? "failed" : "completed", []); releasePendingWebSources(); // Redacted-only turns (or hidden thinking without a trailing signature event) still // need their envelope-only reasoning item so the blocks replay next turn. @@ -1274,15 +1291,14 @@ export function bridgeToResponsesSSE( // After every close above, so the blob lands AFTER the assistant message it belongs // to and the parser's backwards pairing finds it. flushKiroRedactedReasoning(); - // Truncated turns must never install replacement history (#422). The buffered path - // has always checked this; streaming emitted the item BEFORE reading stopReason, so - // a max_tokens/content_filter turn shipped a half-written summary and then declared - // itself incomplete — the same hazard, one branch over. - if (options?.compaction && !isTruncatedStopReason(event.stopReason)) { + + if (options?.compaction && (!isTruncated || gracefulComplete)) { // Exactly one compaction item per turn; codex-rs takes the first and fatals on 0. + const summary = joinChunks(compaction); + const fallbackSummary = summary.trim().length > 0 ? summary : "[earlier conversation was compacted]"; const item = { type: "compaction", id: `cmp_${uuid()}`, - encrypted_content: event.compactionEncryptedContent ?? (options?.compactionRetention?.encode(joinChunks(compaction)) ?? encodeCompactionSummary(joinChunks(compaction))), + encrypted_content: event.compactionEncryptedContent ?? (options?.compactionRetention?.encode(fallbackSummary) ?? encodeCompactionSummary(fallbackSummary)), }; emit("response.output_item.done", { output_index: outputIndex, item }); retainFinishedItem(item as OutputItem, event.compactionEncryptedContent ? 0 : compaction.bytes); @@ -1293,7 +1309,7 @@ export function bridgeToResponsesSSE( // Suppression and terminal status must agree: withholding the compaction item while // still reporting success hands codex-rs a completed response with zero compaction // items, which it treats as fatal. - if (truncationReasonFor(event.stopReason)) { + if (truncationReason && !gracefulComplete) { // Upstream stopped before a normal completion. Surface as incomplete so the // client can distinguish a truncated/filtered turn from a finished one. // #1926 gap 2: bound the window in which a handed-out thought signature is @@ -1303,7 +1319,7 @@ export function bridgeToResponsesSSE( ...responseSnapshot("incomplete", finishedItems, event.endTurn), usage: responsesUsage(event.usage), incomplete_details: { - reason: truncationReasonFor(event.stopReason) ?? "content_filter", + reason: truncationReason ?? "content_filter", }, }; // Cache max-output partials so previous_response_id replay can continue them; @@ -1327,7 +1343,10 @@ export function bridgeToResponsesSSE( break; } case "incomplete": { - if (currentMsg) closeCurrentMessage(); + const truncReason = truncationReasonFor(event.reason) ?? event.reason; + const gracefulComplete = Boolean(options?.completeOnMaxOutputTokens) && truncReason === "max_output_tokens"; + + if (currentMsg) closeCurrentMessage(gracefulComplete ? "final_answer" : undefined, gracefulComplete); if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); @@ -1337,6 +1356,29 @@ export function bridgeToResponsesSSE( flushHiddenReasoningEnvelope(); options?.onUsage?.(event.usage); await awaitThoughtSignatureDurability(); + if (gracefulComplete) { + if (options?.compaction) { + const summary = joinChunks(compaction); + const fallbackSummary = summary.trim().length > 0 ? summary : "[earlier conversation was compacted]"; + const item = { + type: "compaction", id: `cmp_${uuid()}`, + encrypted_content: options?.compactionRetention?.encode(fallbackSummary) ?? encodeCompactionSummary(fallbackSummary), + }; + emit("response.output_item.done", { output_index: outputIndex, item }); + retainFinishedItem(item as OutputItem, compaction.bytes); + outputIndex++; + if (!closed && !terminated) options.compactionRetention?.commit(); + } + const response = { + ...responseSnapshot("completed", finishedItems, event.endTurn), + usage: responsesUsage(event.usage), + }; + options?.onCompletedResponse?.(response); + emit("response.completed", { response }); + reportTerminal("completed"); + terminalEvent = true; + break; + } const refusal = options?.terminalRefusalAsFailed ? terminalRefusalFailure(event) : undefined; if (refusal) { emit("response.failed", { diff --git a/src/config/schema/config-schema.ts b/src/config/schema/config-schema.ts index 0565f9efc35..8cf36e14648 100644 --- a/src/config/schema/config-schema.ts +++ b/src/config/schema/config-schema.ts @@ -262,6 +262,7 @@ export const configSchema = z.object({ emptyCompletionRetry: z.boolean().optional().catch(false), // Header suppression changes what Codex sees, so absence and malformed edits stay off. dropCodexSafetyBuffering: z.boolean().optional().catch(false), + completeOnMaxOutputTokens: z.boolean().optional().catch(true), // A malformed hand edit must not silently stop opening the browser: fall back // to undefined, which resolves to the historical auto-open behavior. oauthOpenBrowser: z.boolean().optional().catch(undefined), diff --git a/src/responses/markdown-block-repair.ts b/src/responses/markdown-block-repair.ts new file mode 100644 index 00000000000..c7e7be86c71 --- /dev/null +++ b/src/responses/markdown-block-repair.ts @@ -0,0 +1,39 @@ +/** + * Closes unclosed markdown code fences and common paired markup in text that was cut short + * by token limit truncation (max_output_tokens). + */ +export function closeOpenMarkdownBlocks(text: string): string { + if (!text) return text; + let repaired = text; + + // 1. Close unclosed code fences (``` or ~~~) + const fenceRegex = /(?:^|\n)(```+|~~~+)[^\n]*(?:\n|$)/g; + const fences: string[] = []; + let match: RegExpExecArray | null; + while ((match = fenceRegex.exec(text)) !== null) { + const marker = match[1].slice(0, 3); + if (fences.length > 0 && fences[fences.length - 1] === marker) { + fences.pop(); + } else { + fences.push(marker); + } + } + if (fences.length > 0) { + if (!repaired.endsWith("\n")) repaired += "\n"; + for (let i = fences.length - 1; i >= 0; i--) { + repaired += `${fences[i]}\n`; + } + } + + // 2. Close common unclosed paired tags if opened + for (const tag of ["details", "thought"]) { + const openCount = (repaired.match(new RegExp(`<${tag}(?:\\s[^>]*)?>`, "gi")) || []).length; + const closeCount = (repaired.match(new RegExp(``, "gi")) || []).length; + if (openCount > closeCount) { + if (!repaired.endsWith("\n")) repaired += "\n"; + repaired += `\n`.repeat(openCount - closeCount); + } + } + + return repaired; +} diff --git a/src/server/responses/adapter-delivery.ts b/src/server/responses/adapter-delivery.ts index 986bd4c0a45..ac9837d8ff2 100644 --- a/src/server/responses/adapter-delivery.ts +++ b/src/server/responses/adapter-delivery.ts @@ -269,6 +269,7 @@ export async function deliverAdapterResponse( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, // Same grok-surface split as the runTurn branch above. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), onUsage: usage => { @@ -378,6 +379,7 @@ export async function deliverAdapterResponse( freeformToolNames, toolSearchToolNames, ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, onProviderState: state => { providerState = state; }, onUsage: usage => { transportState.bindKeyUsageFromBridge(usage); diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 11e0c8b5422..d7343f9725b 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -6,6 +6,8 @@ import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer"; import { isSafeResponseHeader } from "../safe-response-headers"; import { CodexWsMetadata, type CodexWsQuotaObserver } from "./codex-ws-metadata"; import { CODEX_RESPONSES_HTTP_URL, type PreparedCodexWsRequest } from "./codex-ws-request"; +import { encodeCompactionSummary } from "../../responses/compaction"; +import { randomUUID } from "node:crypto"; import { CodexWsCorrelation } from "./codex-ws-correlation"; import type { CodexWsSession } from "./codex-ws-session"; import { UPGRADE_DEADLINE_MS, CODEX_WS_LIVENESS_PING_INTERVAL_MS, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, @@ -36,6 +38,7 @@ interface ExchangeOptions { * control exchanges and on transports without the metadata channel. */ preOutputRefusalHold?: CodexWsPreOutputRefusalHold; + completeOnMaxOutputTokens?: boolean; } const HTTP_HEADER_TOKEN = /^[!#$%&'*+.^_`|~0-9a-z-]+$/i; @@ -44,6 +47,18 @@ function record(value: unknown): value is Record { return value !== null && typeof value === "object" && !Array.isArray(value); } +function isCompactionWsRequest(prepared: PreparedCodexWsRequest): boolean { + try { + const parsed = JSON.parse(prepared.frameText); + const body = record(parsed.payload) ? parsed.payload : parsed; + if (body.compaction_trigger !== undefined) return true; + if (Array.isArray(body.input) && body.input.some((i: unknown) => record(i) && i.type === "compaction_trigger")) return true; + return false; + } catch { + return false; + } +} + /** Rebuild only permitted metadata: upstream framing describes a different body. */ function rejectionHeaders(source: Record, prelude: Headers): Headers { const connectionHeaders = new Set(); @@ -117,7 +132,7 @@ function refusalErrorBody(source: unknown): Record { /** The sole SSE exchange state machine for both one-shot and retained sockets. */ export function codexWsExchange(options: ExchangeOptions): Promise { const { session, url, init, prepared, sseFallback, onQuota, beforeDispatch, bunVersion, nativeControl, beforeContinuation, - preOutputRefusalHold } = options; + preOutputRefusalHold, completeOnMaxOutputTokens } = options; const { frameText, headers } = prepared; const signal = init.signal ?? undefined; return new Promise((resolve, reject) => { @@ -129,6 +144,9 @@ export function codexWsExchange(options: ExchangeOptions): Promise { let received = false; let responseCommitted = false; let terminal = false; + const isCompaction = isCompactionWsRequest(prepared); + let sawCompactionOutput = false; + let maxOutputIndex = 0; // #4191: the counters behind the failure classification. A user whose long // thread died here could not tell an unanswered socket from one that carried // only quota frames, because both arrived as the same one-line message. @@ -615,7 +633,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { } const normalized = normalizeResponsesWsRelayEvent(text); if (!normalized) return; - const { type } = normalized; + let { type } = normalized; let relayText = normalized.text; let controlFrame = false; if (metadata) { @@ -636,6 +654,59 @@ export function codexWsExchange(options: ExchangeOptions): Promise { failStream("codex websocket frame exceeds the response size limit"); return; } + if (record(normalized.payload)) { + const item = normalized.payload.item; + if (record(item) && item.type === "compaction") { + sawCompactionOutput = true; + } + const idx = normalized.payload.output_index; + if (typeof idx === "number" && idx >= maxOutputIndex) { + maxOutputIndex = idx + 1; + } + } + if (type === "response.incomplete" && completeOnMaxOutputTokens) { + const resp = record(normalized.payload.response) ? normalized.payload.response : null; + const details = resp && record(resp.incomplete_details) ? resp.incomplete_details : null; + if (details?.reason === "max_output_tokens") { + if (isCompaction && !sawCompactionOutput) { + const summary = "[earlier conversation was compacted]"; + const cmpItem = { + type: "compaction", + id: `cmp_${randomUUID()}`, + status: "completed", + encrypted_content: encodeCompactionSummary(summary), + }; + const cmpDone = { + type: "response.output_item.done", + output_index: maxOutputIndex++, + item: cmpItem, + }; + const cmpDoneText = encoder.encode(JSON.stringify(cmpDone)); + const cmpPrefix = encoder.encode("event: response.output_item.done\ndata: "); + const cmpSuffix = encoder.encode("\n\n"); + const cmpFrame = new Uint8Array(cmpPrefix.byteLength + cmpDoneText.byteLength + cmpSuffix.byteLength); + cmpFrame.set(cmpPrefix); + cmpFrame.set(cmpDoneText, cmpPrefix.byteLength); + cmpFrame.set(cmpSuffix, cmpPrefix.byteLength + cmpDoneText.byteLength); + controller.enqueue(cmpFrame); + sawCompactionOutput = true; + } + const completedResp: Record = { + ...resp, + status: "completed", + }; + delete completedResp.incomplete_details; + const completedPayload = { + ...normalized.payload, + type: "response.completed", + response: completedResp, + }; + relayText = JSON.stringify(completedPayload); + normalized.text = relayText; + normalized.payload = completedPayload; + type = "response.completed"; + } + } if (!controlFrame && !type.startsWith("response.") && type !== "error") return; let steeringEnded = false; let refusalStatus: number | undefined; diff --git a/src/server/responses/run-turn-execution.ts b/src/server/responses/run-turn-execution.ts index e2f6277004f..284be9a5f47 100644 --- a/src/server/responses/run-turn-execution.ts +++ b/src/server/responses/run-turn-execution.ts @@ -911,6 +911,7 @@ export async function executeResponsesRunTurn( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, // grok-build's strict decoder dies on the typed response.heartbeat frame; its // eventsource layer tolerates comment keep-alives. Codex needs the opposite. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), @@ -1072,6 +1073,7 @@ export async function executeResponsesRunTurn( bareCustomToolNames, toolSearchToolNames, ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, onProviderState: state => { providerState = state; }, onUsage: usage => { transportState.bindKeyUsageFromBridge(usage); diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index b7a7b98a1ca..77b1355126a 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -16,6 +16,7 @@ import type { NativeResponseControl } from "./native-response-control"; // (passthrough relay, adapter parsers, usage sniffing) is unchanged. import { codexWsReuseAcrossTurnsEnabled } from "../../config/codex-ws-reuse-setting"; +import { loadConfig } from "../../config"; import { compareBunVersions } from "../../lib/bun-stream-caps"; import { resolveProxyRoute, socks5ProxyFromEnv } from "../../lib/proxy-env"; import type { CodexWsQuotaObserver } from "./codex-ws-metadata"; @@ -252,5 +253,6 @@ export async function codexWsUpstreamFetch( beforeContinuation, onPhysicalDispatch, preOutputRefusalHold, bunVersion: typeof runtime === "string" ? runtime : runtime.version, + completeOnMaxOutputTokens: loadConfig().completeOnMaxOutputTokens ?? true, }); } diff --git a/src/types/config.ts b/src/types/config.ts index 30da5d1a18a..cfe79dee259 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -571,6 +571,7 @@ export interface OcxConfig { emptyCompletionRetry?: boolean; /** Suppress allowlisted client-facing Codex transport hints; provider enforcement is unchanged. */ dropCodexSafetyBuffering?: boolean; + completeOnMaxOutputTokens?: boolean; /** * Whether a login may open a browser on the machine running the proxy. * diff --git a/tests/responses/graceful-max-output-tokens.test.ts b/tests/responses/graceful-max-output-tokens.test.ts new file mode 100644 index 00000000000..59bab39616b --- /dev/null +++ b/tests/responses/graceful-max-output-tokens.test.ts @@ -0,0 +1,268 @@ +import { describe, expect, test } from "bun:test"; +import { closeOpenMarkdownBlocks } from "../../src/responses/markdown-block-repair"; +import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge"; +import type { AdapterEvent } from "../../src/types"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { codexWsExchange } from "../../src/server/responses/codex-ws-exchange"; +import type { CodexWsSession } from "../../src/server/responses/codex-ws-session"; +import type { PreparedCodexWsRequest } from "../../src/server/responses/codex-ws-request"; + +describe("closeOpenMarkdownBlocks", () => { + test("leaves normal and empty text unchanged", () => { + expect(closeOpenMarkdownBlocks("")).toBe(""); + expect(closeOpenMarkdownBlocks("Hello world!")).toBe("Hello world!"); + expect(closeOpenMarkdownBlocks("```ts\nconst x = 1;\n```\n")).toBe("```ts\nconst x = 1;\n```\n"); + }); + + test("closes unclosed code fences", () => { + const truncated = "Here is some code:\n```typescript\nfunction greet() {\n return 'hi';"; + const repaired = closeOpenMarkdownBlocks(truncated); + expect(repaired).toBe("Here is some code:\n```typescript\nfunction greet() {\n return 'hi';\n```\n"); + }); + + test("closes nested or multiple open code fences", () => { + const truncated = "```markdown\nInside markdown:\n~~~python\nprint(1)"; + const repaired = closeOpenMarkdownBlocks(truncated); + expect(repaired).toContain("~~~\n"); + expect(repaired).toContain("```\n"); + }); + + test("closes unclosed paired tags", () => { + const truncated = "
More info\nSome body text"; + const repaired = closeOpenMarkdownBlocks(truncated); + expect(repaired).toBe("
More info\nSome body text\n
\n"); + }); +}); + +describe("graceful completion on max_output_tokens: bridgeToResponsesSSE", () => { + async function streamEvents(events: AdapterEvent[], options: Parameters[7]) { + const budget = createTestTranslatorBudget(); + async function* gen() { + for (const e of events) yield e; + } + const sseStream = bridgeToResponsesSSE( + gen(), + "test-model", + undefined, + undefined, + undefined, + () => {}, + 100, + { ...options, translatorBudget: budget }, + ); + const text = await new Response(sseStream).text(); + const parsed: any[] = []; + for (const block of text.split("\n\n")) { + for (const line of block.split("\n")) { + if (line.startsWith("data: ") && line !== "data: [DONE]") { + try { + parsed.push(JSON.parse(line.slice(6))); + } catch {} + } + } + } + return parsed; + } + + test("emits response.completed and repairs markdown fences in normal turn", async () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Here is partial code:\n```python\ndef run():\n pass" }, + { type: "done", stopReason: "max_tokens", usage: { inputTokens: 10, outputTokens: 100 } }, + ]; + const sse = await streamEvents(events, { + completeOnMaxOutputTokens: true, + }); + + const terminal = sse.find(e => e.type === "response.completed"); + expect(terminal).toBeDefined(); + expect(terminal.response.status).toBe("completed"); + + // Markdown fence delta emitted + const textDeltas = sse.filter(e => e.type === "response.output_text.delta").map(e => e.delta); + const fullText = textDeltas.join(""); + expect(fullText).toContain("```\n"); + }); + + test("emits compaction item and commits retention on compaction max_output_tokens", async () => { + let committed = false; + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Summary up to limit..." }, + { type: "done", stopReason: "max_tokens", usage: { inputTokens: 10, outputTokens: 100 } }, + ]; + const sse = await streamEvents(events, { + completeOnMaxOutputTokens: true, + compaction: true, + compactionRetention: { + encode: s => `encoded:${s}`, + commit: () => { committed = true; }, + }, + }); + + const terminal = sse.find(e => e.type === "response.completed"); + expect(terminal).toBeDefined(); + expect(terminal.response.status).toBe("completed"); + + const compactionDone = sse.find(e => e.type === "response.output_item.done" && e.item?.type === "compaction"); + expect(compactionDone).toBeDefined(); + expect(committed).toBe(true); + }); + + test("recovers from explicit incomplete event with max_output_tokens", async () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Accumulated response before cut off" }, + { type: "incomplete", reason: "max_output_tokens", usage: { inputTokens: 5, outputTokens: 50 } }, + ]; + const sse = await streamEvents(events, { + completeOnMaxOutputTokens: true, + }); + + const terminal = sse.find(e => e.type === "response.completed"); + expect(terminal).toBeDefined(); + expect(terminal.response.status).toBe("completed"); + expect(sse.some(e => e.type === "response.incomplete")).toBe(false); + }); + + test("keeps response.incomplete when completeOnMaxOutputTokens is false", async () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Some text" }, + { type: "done", stopReason: "max_tokens" }, + ]; + const sse = await streamEvents(events, { + completeOnMaxOutputTokens: false, + }); + + const terminal = sse.find(e => e.type === "response.incomplete"); + expect(terminal).toBeDefined(); + expect(terminal.response.incomplete_details?.reason).toBe("max_output_tokens"); + expect(sse.some(e => e.type === "response.completed")).toBe(false); + }); + + test("keeps response.incomplete for content_filter even if completeOnMaxOutputTokens is true", async () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Some text" }, + { type: "done", stopReason: "content_filter" }, + ]; + const sse = await streamEvents(events, { + completeOnMaxOutputTokens: true, + }); + + const terminal = sse.find(e => e.type === "response.incomplete"); + expect(terminal).toBeDefined(); + expect(terminal.response.incomplete_details?.reason).toBe("content_filter"); + }); +}); + +describe("graceful completion on max_output_tokens: buildResponseJSON", () => { + test("returns status completed and repairs markdown in buffered response", () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Hello\n```js\nconsole.log(1);" }, + { type: "done", stopReason: "max_tokens", usage: { inputTokens: 5, outputTokens: 20 } }, + ]; + const json = buildResponseJSON(events, "test-model", { + completeOnMaxOutputTokens: true, + }); + + expect(json.status).toBe("completed"); + expect((json as any).incomplete_details).toBeUndefined(); + + const output = json.output as any[]; + const msg = output.find(item => item.type === "message"); + expect(msg).toBeDefined(); + expect(msg.content[0].text).toBe("Hello\n```js\nconsole.log(1);\n```\n"); + }); + + test("emits compaction item and commits retention on compaction max_output_tokens in buffered response", () => { + let committed = false; + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Compacted summary text" }, + { type: "done", stopReason: "max_tokens", usage: { inputTokens: 5, outputTokens: 20 } }, + ]; + const json = buildResponseJSON(events, "test-model", { + completeOnMaxOutputTokens: true, + compaction: true, + compactionRetention: { + encode: s => `encoded:${s}`, + commit: () => { committed = true; }, + }, + }); + + expect(json.status).toBe("completed"); + expect(committed).toBe(true); + const output = json.output as any[]; + expect(output.some(item => item.type === "compaction")).toBe(true); + }); + + test("returns status incomplete when completeOnMaxOutputTokens is false", () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "Hello" }, + { type: "done", stopReason: "max_tokens" }, + ]; + const json = buildResponseJSON(events, "test-model", { + completeOnMaxOutputTokens: false, + }); + expect(json.status).toBe("incomplete"); + expect((json as any).incomplete_details?.reason).toBe("max_output_tokens"); + }); +}); + +describe("graceful completion on max_output_tokens: codexWsExchange", () => { + test("transforms response.incomplete into response.completed and injects compaction item", async () => { + let messageHandler: ((event: MessageEvent) => void) | undefined; + const fakeSocket = { + send: () => {}, + close: () => {}, + addEventListener: (type: string, handler: any) => { + if (type === "message") messageHandler = handler; + }, + removeEventListener: () => {}, + }; + + const session: CodexWsSession = { + socket: fakeSocket as any, + dispose: () => {}, + metrics: () => ({} as any), + on: () => {}, + off: () => {}, + bindOwner: () => () => {}, + release: () => {}, + } as any; + + const prepared: PreparedCodexWsRequest = { + frameText: JSON.stringify({ payload: { compaction_trigger: "auto" } }), + headers: new Headers(), + } as any; + + const responsePromise = codexWsExchange({ + session, + url: "https://api.openai.com/v1/responses", + init: {}, + prepared, + completeOnMaxOutputTokens: true, + sseFallback: async () => new Response("fallback"), + }); + + // Simulate WS emitting response.incomplete due to max_output_tokens + setTimeout(() => { + messageHandler?.({ + data: JSON.stringify({ + type: "response.incomplete", + response: { + id: "resp_123", + status: "incomplete", + incomplete_details: { reason: "max_output_tokens" }, + }, + }), + } as MessageEvent); + }, 10); + + const res = await responsePromise; + expect(res.status).toBe(200); + + const text = await res.text(); + expect(text).toContain("event: response.output_item.done"); + expect(text).toContain('"type":"compaction"'); + expect(text).toContain("event: response.completed"); + expect(text).toContain('"status":"completed"'); + expect(text).not.toContain("event: response.incomplete"); + }); +}); From 5f15e261fc1db7b30c134fb81422e16d4a4be81c Mon Sep 17 00:00:00 2001 From: Omar <13569609+omar391@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:38:54 +0600 Subject: [PATCH 2/6] test(responses): remove empty catch in graceful max output tokens test --- tests/responses/graceful-max-output-tokens.test.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/responses/graceful-max-output-tokens.test.ts b/tests/responses/graceful-max-output-tokens.test.ts index 59bab39616b..65057cef046 100644 --- a/tests/responses/graceful-max-output-tokens.test.ts +++ b/tests/responses/graceful-max-output-tokens.test.ts @@ -55,9 +55,7 @@ describe("graceful completion on max_output_tokens: bridgeToResponsesSSE", () => for (const block of text.split("\n\n")) { for (const line of block.split("\n")) { if (line.startsWith("data: ") && line !== "data: [DONE]") { - try { - parsed.push(JSON.parse(line.slice(6))); - } catch {} + parsed.push(JSON.parse(line.slice(6))); } } } From eb0b09b418ebbba4de1ccfba514e99251208487b Mon Sep 17 00:00:00 2001 From: Omar <13569609+omar391@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:47:24 +0600 Subject: [PATCH 3/6] fix(config): default completeOnMaxOutputTokens to false to preserve strict unconfigured contracts --- src/config/schema/config-schema.ts | 2 +- src/server/responses/adapter-delivery.ts | 4 ++-- src/server/responses/run-turn-execution.ts | 4 ++-- src/server/responses/ws-upstream.ts | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/config/schema/config-schema.ts b/src/config/schema/config-schema.ts index 8cf36e14648..2ace63dd844 100644 --- a/src/config/schema/config-schema.ts +++ b/src/config/schema/config-schema.ts @@ -262,7 +262,7 @@ export const configSchema = z.object({ emptyCompletionRetry: z.boolean().optional().catch(false), // Header suppression changes what Codex sees, so absence and malformed edits stay off. dropCodexSafetyBuffering: z.boolean().optional().catch(false), - completeOnMaxOutputTokens: z.boolean().optional().catch(true), + completeOnMaxOutputTokens: z.boolean().optional().catch(false), // A malformed hand edit must not silently stop opening the browser: fall back // to undefined, which resolves to the historical auto-open behavior. oauthOpenBrowser: z.boolean().optional().catch(undefined), diff --git a/src/server/responses/adapter-delivery.ts b/src/server/responses/adapter-delivery.ts index ac9837d8ff2..05c776aba5c 100644 --- a/src/server/responses/adapter-delivery.ts +++ b/src/server/responses/adapter-delivery.ts @@ -269,7 +269,7 @@ export async function deliverAdapterResponse( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, // Same grok-surface split as the runTurn branch above. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), onUsage: usage => { @@ -379,7 +379,7 @@ export async function deliverAdapterResponse( freeformToolNames, toolSearchToolNames, ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, onProviderState: state => { providerState = state; }, onUsage: usage => { transportState.bindKeyUsageFromBridge(usage); diff --git a/src/server/responses/run-turn-execution.ts b/src/server/responses/run-turn-execution.ts index 284be9a5f47..78d66b1cd04 100644 --- a/src/server/responses/run-turn-execution.ts +++ b/src/server/responses/run-turn-execution.ts @@ -911,7 +911,7 @@ export async function executeResponsesRunTurn( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, // grok-build's strict decoder dies on the typed response.heartbeat frame; its // eventsource layer tolerates comment keep-alives. Codex needs the opposite. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), @@ -1073,7 +1073,7 @@ export async function executeResponsesRunTurn( bareCustomToolNames, toolSearchToolNames, ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? true, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, onProviderState: state => { providerState = state; }, onUsage: usage => { transportState.bindKeyUsageFromBridge(usage); diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index 77b1355126a..8b62065b40a 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -253,6 +253,6 @@ export async function codexWsUpstreamFetch( beforeContinuation, onPhysicalDispatch, preOutputRefusalHold, bunVersion: typeof runtime === "string" ? runtime : runtime.version, - completeOnMaxOutputTokens: loadConfig().completeOnMaxOutputTokens ?? true, + completeOnMaxOutputTokens: loadConfig().completeOnMaxOutputTokens ?? false, }); } From 52d2abd0ae9315b08164a62fd1c0e27df4c395fa Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:09:02 +0900 Subject: [PATCH 4/6] fix(responses): narrow opted-in token-limit completion to plain text Preserve the contributor history while removing compaction exceptions, native WebSocket rewrites, and output repair. Exclude structured output and all non-text operations before bridge cleanup; retain provider continuation state and usage. Register bridge and real delivery regressions in the existing layout shards. Co-authored-by: Omar <13569609+omar391@users.noreply.github.com> --- scripts/test-layout/layout-responses.json | 2 + src/bridge/max-output-completion.ts | 41 ++ src/bridge/response-json.ts | 38 +- src/bridge/sse.ts | 73 +-- src/responses/markdown-block-repair.ts | 39 -- src/server/responses/adapter-delivery.ts | 6 +- src/server/responses/codex-ws-exchange.ts | 75 +-- src/server/responses/run-turn-execution.ts | 6 +- src/server/responses/ws-upstream.ts | 2 - src/types/config.ts | 1 + tests/fixtures/test-layout-expected.json | 2 + ...ceful-max-output-tokens-exclusions.test.ts | 168 +++++++ .../graceful-max-output-tokens.test.ts | 445 ++++++++---------- 13 files changed, 454 insertions(+), 444 deletions(-) create mode 100644 src/bridge/max-output-completion.ts delete mode 100644 src/responses/markdown-block-repair.ts create mode 100644 tests/responses/graceful-max-output-tokens-exclusions.test.ts diff --git a/scripts/test-layout/layout-responses.json b/scripts/test-layout/layout-responses.json index 18f7d3828b1..f2c05d33b03 100644 --- a/scripts/test-layout/layout-responses.json +++ b/scripts/test-layout/layout-responses.json @@ -1,4 +1,6 @@ { + "graceful-max-output-tokens.test.ts": "responses", + "graceful-max-output-tokens-exclusions.test.ts": "responses", "passthrough-media-type.test.ts": "responses", "passthrough-override.test.ts": "responses", "passthrough-headers.test.ts": "responses", diff --git a/src/bridge/max-output-completion.ts b/src/bridge/max-output-completion.ts new file mode 100644 index 00000000000..3ec9e1408f6 --- /dev/null +++ b/src/bridge/max-output-completion.ts @@ -0,0 +1,41 @@ +import type { AdapterEvent, OcxRequestOptions } from "../types"; + +export interface MaxOutputCompletionOptions { + /** Set only for an explicitly opted-in Responses inbound request. */ + completeOnMaxOutputTokens?: boolean; + textFormat?: OcxRequestOptions["textFormat"]; + compaction?: boolean; +} + +/** Track the original event stream: bridge cleanup must not erase an unfinished operation. */ +export function createMaxOutputCompletion(options?: MaxOutputCompletionOptions) { + let textOnly = options?.completeOnMaxOutputTokens === true && !options.compaction && !options.textFormat; + let hasText = false; + let completed = false; + return { + observe(event: AdapterEvent): void { + switch (event.type) { + case "text_delta": + if (event.anthropicThinkingSignature !== undefined) textOnly = false; + if (event.text.length > 0) hasText = true; + break; + case "heartbeat": + case "assistant_boundary": + break; + case "done": + case "incomplete": { + const reason = event.type === "done" ? event.stopReason : event.reason; + completed = textOnly && hasText && event.endTurn !== false + && (event.type !== "done" || event.compactionEncryptedContent === undefined) + && (reason === "max_tokens" || reason === "length" || reason === "max_output_tokens"); + break; + } + default: + // All non-text output and unknown future operations are excluded, even if hidden or closed. + textOnly = false; + completed = false; + } + }, + get completed(): boolean { return completed; }, + }; +} diff --git a/src/bridge/response-json.ts b/src/bridge/response-json.ts index 72f4684ff78..f4bdc5a902b 100644 --- a/src/bridge/response-json.ts +++ b/src/bridge/response-json.ts @@ -35,7 +35,6 @@ import { } from "../responses/citation-markers"; import { declaresCodeModeExec, normalizeDeclaredToolName } from "../types"; import { appendSafeWebSearchSource, safeWebSearchSources } from "../web-search/sources"; -import { closeOpenMarkdownBlocks } from "../responses/markdown-block-repair"; import { isTranslatorBudgetExceededError, releaseTranslatedEvent, @@ -46,6 +45,7 @@ import { import { adapterFailureFromEvent, emptyChunks, joinChunks, responsesUsage, toolCallArgumentsUsable, uuid, webSearchAction, webSearchCellItemId } from "./internal"; import type { OutputItem, StringChunks } from "./internal"; import { bridgeToResponsesSSE } from "./sse"; +import { createMaxOutputCompletion, type MaxOutputCompletionOptions } from "./max-output-completion"; /** Build a buffered Responses result within a caller-owned or temporary translator budget. */ export function buildResponseJSON( @@ -85,7 +85,7 @@ export function buildResponseJSON( function buildResponseJSONWithBudget( events: AdapterEvent[], modelId: string, - options?: { + options?: MaxOutputCompletionOptions & { hideThinkingSummary?: boolean; /** Internal Chat delivery override for raw reasoning only; summaries keep their policy. */ chatRawReasoningVisible?: boolean; @@ -118,7 +118,6 @@ function buildResponseJSONWithBudget( * so a block that fit live delivery could otherwise overflow here and lose the cache write. */ omitHiddenReasoningEnvelope?: boolean; - completeOnMaxOutputTokens?: boolean; }, ): Record { const responseId = `resp_${uuid()}`; @@ -204,6 +203,7 @@ function buildResponseJSONWithBudget( // which is only true for a `done` without a stop reason. A buffered turn whose adapter simply // stopped emitting has no terminal at all, and must not be reported as a success. let sawTerminal = false; + const maxOutputCompletion = createMaxOutputCompletion(options); let batchCompaction = emptyChunks(); let compactionEncryptedContent: string | undefined; @@ -373,6 +373,7 @@ function buildResponseJSONWithBudget( }; for (const e of events) { + maxOutputCompletion.observe(e); if (errorEvent) { // Match streaming: once the turn fails, later parallel calls must not become executable // completed output. Still release every retained event in order and preserve terminal usage. @@ -572,6 +573,7 @@ function buildResponseJSONWithBudget( break; case "incomplete": incompleteEvent = e; + usage = e.usage ?? usage; sawTerminal = true; endTurn = e.endTurn; if (e.providerState) options?.onProviderState?.(e.providerState); @@ -621,47 +623,31 @@ function buildResponseJSONWithBudget( batchKiroRedacted = undefined; batchKiroRedactedBytes = 0; } - const gracefulComplete = Boolean(options?.completeOnMaxOutputTokens) - && (stopReason === "max_tokens" || (incompleteEvent && (truncationReasonFor(incompleteEvent.reason) === "max_output_tokens" || incompleteEvent.reason === "max_output_tokens"))); - // A truncated turn must never be installed as replacement history: emit the // compaction item only when the turn actually completed (#422). if ( options?.compaction && !errorEvent - && (!incompleteEvent || gracefulComplete) + && !incompleteEvent // A stream that stopped without any terminal did not complete either. The original guard // could only see explicit failure events, so an adapter EOF slipped past it and installed a // truncated summary as replacement history — the exact #422 hazard, reached by a route that // did not exist when the guard was written. && sawTerminal - && (!isTruncatedStopReason(rawStopReason) || gracefulComplete) + && !isTruncatedStopReason(rawStopReason) ) { - const summary = joinChunks(batchCompaction); - const fallbackSummary = summary.trim().length > 0 ? summary : "[earlier conversation was compacted]"; - const item = { + const item = { type: "compaction", id: `cmp_${uuid()}`, - encrypted_content: compactionEncryptedContent ?? (options?.compactionRetention?.encode(fallbackSummary) ?? encodeCompactionSummary(fallbackSummary)), + encrypted_content: compactionEncryptedContent ?? (options?.compactionRetention?.encode(joinChunks(batchCompaction)) ?? encodeCompactionSummary(joinChunks(batchCompaction))), }; pushOutput(item, compactionEncryptedContent ? 0 : batchCompaction.bytes); } - if (gracefulComplete) { - for (const item of output) { - if (typeof item === "object" && item !== null && (item as any).type === "message" && Array.isArray((item as any).content)) { - for (const part of (item as any).content) { - if (typeof part === "object" && part !== null && part.type === "output_text" && typeof part.text === "string") { - part.text = closeOpenMarkdownBlocks(part.text); - } - } - } - } - } - const failure = errorEvent ? adapterFailureFromEvent(errorEvent) : undefined; + const gracefulComplete = maxOutputCompletion.completed && !errorEvent; const status = errorEvent ? "failed" - : ((incompleteEvent || stopReason === "max_tokens" || stopReason === "content_filter") && !gracefulComplete) + : !gracefulComplete && (incompleteEvent || stopReason === "max_tokens" || stopReason === "content_filter") ? "incomplete" : sawTerminal ? "completed" @@ -687,7 +673,7 @@ function buildResponseJSONWithBudget( ...(incompleteEvent.message ? { message: incompleteEvent.message } : {}), ...(incompleteEvent.retryable !== undefined ? { retryable: incompleteEvent.retryable } : {}), }, - } : (!gracefulComplete && stopReason === "max_tokens") ? { + } : !gracefulComplete && stopReason === "max_tokens" ? { incomplete_details: { reason: "max_output_tokens" }, } : stopReason === "content_filter" ? { incomplete_details: { reason: "content_filter" }, diff --git a/src/bridge/sse.ts b/src/bridge/sse.ts index 2378dd17f65..99e4463f29b 100644 --- a/src/bridge/sse.ts +++ b/src/bridge/sse.ts @@ -43,7 +43,6 @@ import { } from "../responses/citation-markers"; import { declaresCodeModeExec, normalizeDeclaredToolName } from "../types"; import { appendSafeWebSearchSource, safeWebSearchSources } from "../web-search/sources"; -import { closeOpenMarkdownBlocks } from "../responses/markdown-block-repair"; import { isTranslatorBudgetExceededError, releaseTranslatedEvent, @@ -54,6 +53,7 @@ import { import { adapterFailureFromEvent, emptyChunks, joinChunks, ownedBudgetAbandonedMs, responsesUsage, toolCallArgumentsCouldBeJson, toolCallArgumentsUsable, uuid, webSearchAction, webSearchCellItemId } from "./internal"; import type { OutputItem, StringChunks } from "./internal"; import { terminalRefusalFailure } from "./terminal-refusal"; +import { createMaxOutputCompletion, type MaxOutputCompletionOptions } from "./max-output-completion"; function sseEvent(name: string, data: Record): string { return `event: ${name}\ndata: ${JSON.stringify(data)}\n\n`; @@ -74,7 +74,7 @@ export function bridgeToResponsesSSE( toolSearchToolNames?: Set, onCancel?: () => void, heartbeatMs = 2_000, - options?: { + options?: MaxOutputCompletionOptions & { responseId?: string; /** Trusted structural grouping only; never provider state or execution authority. */ outputItemIds?: { next(kind: "message" | "reasoning" | "function_call" | "custom_tool_call" | "tool_search_call"): string }; @@ -163,10 +163,10 @@ export function bridgeToResponsesSSE( setInterval: (handler: () => void, ms: number) => unknown; clearInterval: (id: unknown) => void; }; - completeOnMaxOutputTokens?: boolean; }, ): ReadableStream { const replayCacheScope = options?.replayCacheScope; + const maxOutputCompletion = createMaxOutputCompletion(options); const outputItemId = (kind: "message" | "reasoning" | "function_call" | "custom_tool_call" | "tool_search_call", prefix: string): string => options?.outputItemIds?.next(kind) ?? `${prefix}_${uuid()}`; @@ -504,7 +504,7 @@ export function bridgeToResponsesSSE( return anns; }; - const closeCurrentMessage = (inferredPhase?: OcxMessagePhase, gracefulRepair?: boolean) => { + const closeCurrentMessage = (inferredPhase?: OcxMessagePhase) => { if (!currentMsg) return; // Release anything the citation filter was holding for this message, then strip the // accumulated text: closeCurrentMessage re-sends it in output_text.done and @@ -516,18 +516,7 @@ export function bridgeToResponsesSSE( content_index: 0, delta: trailing, }); } - let messageText = stripCitationMarkers(joinChunks(currentMsg.text)); - if (gracefulRepair) { - const repaired = closeOpenMarkdownBlocks(messageText); - if (repaired !== messageText) { - const delta = repaired.slice(messageText.length); - emit("response.output_text.delta", { - item_id: currentMsg.itemId, output_index: currentMsg.outputIndex, - content_index: 0, delta, - }); - messageText = repaired; - } - } + const messageText = stripCitationMarkers(joinChunks(currentMsg.text)); // Chat Completions has no message-phase field. Keep its live item provisional, then // classify it only when the next adapter event proves whether this text led into more // work or completed the turn. Explicit adapter phases always outrank this inference. @@ -865,6 +854,7 @@ export function bridgeToResponsesSSE( } if (next.done) { upstreamDone = true; break; } const event = next.value; + maxOutputCompletion.observe(event); let terminalEvent = false; // Invisible adapter heartbeats (and buffered web-search progress) count as upstream // liveness only — they must not suppress wire keepalives that re-arm Codex idle timers. @@ -1269,21 +1259,17 @@ export function bridgeToResponsesSSE( break; } case "done": { - const isTruncated = isTruncatedStopReason(event.stopReason); - const truncationReason = truncationReasonFor(event.stopReason); - const gracefulComplete = Boolean(options?.completeOnMaxOutputTokens) && truncationReason === "max_output_tokens"; - - if (currentMsg) closeCurrentMessage(isTruncated && !gracefulComplete ? undefined : "final_answer", gracefulComplete); + if (currentMsg) closeCurrentMessage(isTruncatedStopReason(event.stopReason) ? undefined : "final_answer"); if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); if (currentToolCall) { - if (isTruncated) failCurrentToolCall(); + if (isTruncatedStopReason(event.stopReason)) failCurrentToolCall(); else closeCurrentToolCall(); } // A search still in flight when upstream truncates never returned results, so it // takes the same "failed" status as the error/incomplete terminals below. - if (currentWebSearch) closeCurrentWebSearch(isTruncated ? "failed" : "completed", []); + if (currentWebSearch) closeCurrentWebSearch(isTruncatedStopReason(event.stopReason) ? "failed" : "completed", []); releasePendingWebSources(); // Redacted-only turns (or hidden thinking without a trailing signature event) still // need their envelope-only reasoning item so the blocks replay next turn. @@ -1291,14 +1277,15 @@ export function bridgeToResponsesSSE( // After every close above, so the blob lands AFTER the assistant message it belongs // to and the parser's backwards pairing finds it. flushKiroRedactedReasoning(); - - if (options?.compaction && (!isTruncated || gracefulComplete)) { + // Truncated turns must never install replacement history (#422). The buffered path + // has always checked this; streaming emitted the item BEFORE reading stopReason, so + // a max_tokens/content_filter turn shipped a half-written summary and then declared + // itself incomplete — the same hazard, one branch over. + if (options?.compaction && !isTruncatedStopReason(event.stopReason)) { // Exactly one compaction item per turn; codex-rs takes the first and fatals on 0. - const summary = joinChunks(compaction); - const fallbackSummary = summary.trim().length > 0 ? summary : "[earlier conversation was compacted]"; const item = { type: "compaction", id: `cmp_${uuid()}`, - encrypted_content: event.compactionEncryptedContent ?? (options?.compactionRetention?.encode(fallbackSummary) ?? encodeCompactionSummary(fallbackSummary)), + encrypted_content: event.compactionEncryptedContent ?? (options?.compactionRetention?.encode(joinChunks(compaction)) ?? encodeCompactionSummary(joinChunks(compaction))), }; emit("response.output_item.done", { output_index: outputIndex, item }); retainFinishedItem(item as OutputItem, event.compactionEncryptedContent ? 0 : compaction.bytes); @@ -1309,7 +1296,7 @@ export function bridgeToResponsesSSE( // Suppression and terminal status must agree: withholding the compaction item while // still reporting success hands codex-rs a completed response with zero compaction // items, which it treats as fatal. - if (truncationReason && !gracefulComplete) { + if (truncationReasonFor(event.stopReason) && !maxOutputCompletion.completed) { // Upstream stopped before a normal completion. Surface as incomplete so the // client can distinguish a truncated/filtered turn from a finished one. // #1926 gap 2: bound the window in which a handed-out thought signature is @@ -1319,7 +1306,7 @@ export function bridgeToResponsesSSE( ...responseSnapshot("incomplete", finishedItems, event.endTurn), usage: responsesUsage(event.usage), incomplete_details: { - reason: truncationReason ?? "content_filter", + reason: truncationReasonFor(event.stopReason) ?? "content_filter", }, }; // Cache max-output partials so previous_response_id replay can continue them; @@ -1343,10 +1330,7 @@ export function bridgeToResponsesSSE( break; } case "incomplete": { - const truncReason = truncationReasonFor(event.reason) ?? event.reason; - const gracefulComplete = Boolean(options?.completeOnMaxOutputTokens) && truncReason === "max_output_tokens"; - - if (currentMsg) closeCurrentMessage(gracefulComplete ? "final_answer" : undefined, gracefulComplete); + if (currentMsg) closeCurrentMessage(); if (currentReasoning) closeCurrentReasoning(); if (currentRawReasoning) closeCurrentRawReasoning(); flushHiddenRawReasoning(); @@ -1356,24 +1340,9 @@ export function bridgeToResponsesSSE( flushHiddenReasoningEnvelope(); options?.onUsage?.(event.usage); await awaitThoughtSignatureDurability(); - if (gracefulComplete) { - if (options?.compaction) { - const summary = joinChunks(compaction); - const fallbackSummary = summary.trim().length > 0 ? summary : "[earlier conversation was compacted]"; - const item = { - type: "compaction", id: `cmp_${uuid()}`, - encrypted_content: options?.compactionRetention?.encode(fallbackSummary) ?? encodeCompactionSummary(fallbackSummary), - }; - emit("response.output_item.done", { output_index: outputIndex, item }); - retainFinishedItem(item as OutputItem, compaction.bytes); - outputIndex++; - if (!closed && !terminated) options.compactionRetention?.commit(); - } - const response = { - ...responseSnapshot("completed", finishedItems, event.endTurn), - usage: responsesUsage(event.usage), - }; - options?.onCompletedResponse?.(response); + if (maxOutputCompletion.completed) { + const response = { ...responseSnapshot("completed", finishedItems, event.endTurn), usage: responsesUsage(event.usage) }; + options?.onCompletedResponse?.(response, event.providerState); emit("response.completed", { response }); reportTerminal("completed"); terminalEvent = true; diff --git a/src/responses/markdown-block-repair.ts b/src/responses/markdown-block-repair.ts deleted file mode 100644 index c7e7be86c71..00000000000 --- a/src/responses/markdown-block-repair.ts +++ /dev/null @@ -1,39 +0,0 @@ -/** - * Closes unclosed markdown code fences and common paired markup in text that was cut short - * by token limit truncation (max_output_tokens). - */ -export function closeOpenMarkdownBlocks(text: string): string { - if (!text) return text; - let repaired = text; - - // 1. Close unclosed code fences (``` or ~~~) - const fenceRegex = /(?:^|\n)(```+|~~~+)[^\n]*(?:\n|$)/g; - const fences: string[] = []; - let match: RegExpExecArray | null; - while ((match = fenceRegex.exec(text)) !== null) { - const marker = match[1].slice(0, 3); - if (fences.length > 0 && fences[fences.length - 1] === marker) { - fences.pop(); - } else { - fences.push(marker); - } - } - if (fences.length > 0) { - if (!repaired.endsWith("\n")) repaired += "\n"; - for (let i = fences.length - 1; i >= 0; i--) { - repaired += `${fences[i]}\n`; - } - } - - // 2. Close common unclosed paired tags if opened - for (const tag of ["details", "thought"]) { - const openCount = (repaired.match(new RegExp(`<${tag}(?:\\s[^>]*)?>`, "gi")) || []).length; - const closeCount = (repaired.match(new RegExp(``, "gi")) || []).length; - if (openCount > closeCount) { - if (!repaired.endsWith("\n")) repaired += "\n"; - repaired += `\n`.repeat(openCount - closeCount); - } - } - - return repaired; -} diff --git a/src/server/responses/adapter-delivery.ts b/src/server/responses/adapter-delivery.ts index 05c776aba5c..968ea8d3119 100644 --- a/src/server/responses/adapter-delivery.ts +++ b/src/server/responses/adapter-delivery.ts @@ -269,7 +269,8 @@ export async function deliverAdapterResponse( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens === true && options.inboundWire !== "chat" && options.inboundWire !== "anthropic", + textFormat: parsed.options.textFormat, // Same grok-surface split as the runTurn branch above. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), onUsage: usage => { @@ -379,7 +380,8 @@ export async function deliverAdapterResponse( freeformToolNames, toolSearchToolNames, ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens === true && options.inboundWire !== "chat" && options.inboundWire !== "anthropic", + textFormat: parsed.options.textFormat, onProviderState: state => { providerState = state; }, onUsage: usage => { transportState.bindKeyUsageFromBridge(usage); diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index d7343f9725b..11e0c8b5422 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -6,8 +6,6 @@ import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer"; import { isSafeResponseHeader } from "../safe-response-headers"; import { CodexWsMetadata, type CodexWsQuotaObserver } from "./codex-ws-metadata"; import { CODEX_RESPONSES_HTTP_URL, type PreparedCodexWsRequest } from "./codex-ws-request"; -import { encodeCompactionSummary } from "../../responses/compaction"; -import { randomUUID } from "node:crypto"; import { CodexWsCorrelation } from "./codex-ws-correlation"; import type { CodexWsSession } from "./codex-ws-session"; import { UPGRADE_DEADLINE_MS, CODEX_WS_LIVENESS_PING_INTERVAL_MS, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, @@ -38,7 +36,6 @@ interface ExchangeOptions { * control exchanges and on transports without the metadata channel. */ preOutputRefusalHold?: CodexWsPreOutputRefusalHold; - completeOnMaxOutputTokens?: boolean; } const HTTP_HEADER_TOKEN = /^[!#$%&'*+.^_`|~0-9a-z-]+$/i; @@ -47,18 +44,6 @@ function record(value: unknown): value is Record { return value !== null && typeof value === "object" && !Array.isArray(value); } -function isCompactionWsRequest(prepared: PreparedCodexWsRequest): boolean { - try { - const parsed = JSON.parse(prepared.frameText); - const body = record(parsed.payload) ? parsed.payload : parsed; - if (body.compaction_trigger !== undefined) return true; - if (Array.isArray(body.input) && body.input.some((i: unknown) => record(i) && i.type === "compaction_trigger")) return true; - return false; - } catch { - return false; - } -} - /** Rebuild only permitted metadata: upstream framing describes a different body. */ function rejectionHeaders(source: Record, prelude: Headers): Headers { const connectionHeaders = new Set(); @@ -132,7 +117,7 @@ function refusalErrorBody(source: unknown): Record { /** The sole SSE exchange state machine for both one-shot and retained sockets. */ export function codexWsExchange(options: ExchangeOptions): Promise { const { session, url, init, prepared, sseFallback, onQuota, beforeDispatch, bunVersion, nativeControl, beforeContinuation, - preOutputRefusalHold, completeOnMaxOutputTokens } = options; + preOutputRefusalHold } = options; const { frameText, headers } = prepared; const signal = init.signal ?? undefined; return new Promise((resolve, reject) => { @@ -144,9 +129,6 @@ export function codexWsExchange(options: ExchangeOptions): Promise { let received = false; let responseCommitted = false; let terminal = false; - const isCompaction = isCompactionWsRequest(prepared); - let sawCompactionOutput = false; - let maxOutputIndex = 0; // #4191: the counters behind the failure classification. A user whose long // thread died here could not tell an unanswered socket from one that carried // only quota frames, because both arrived as the same one-line message. @@ -633,7 +615,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { } const normalized = normalizeResponsesWsRelayEvent(text); if (!normalized) return; - let { type } = normalized; + const { type } = normalized; let relayText = normalized.text; let controlFrame = false; if (metadata) { @@ -654,59 +636,6 @@ export function codexWsExchange(options: ExchangeOptions): Promise { failStream("codex websocket frame exceeds the response size limit"); return; } - if (record(normalized.payload)) { - const item = normalized.payload.item; - if (record(item) && item.type === "compaction") { - sawCompactionOutput = true; - } - const idx = normalized.payload.output_index; - if (typeof idx === "number" && idx >= maxOutputIndex) { - maxOutputIndex = idx + 1; - } - } - if (type === "response.incomplete" && completeOnMaxOutputTokens) { - const resp = record(normalized.payload.response) ? normalized.payload.response : null; - const details = resp && record(resp.incomplete_details) ? resp.incomplete_details : null; - if (details?.reason === "max_output_tokens") { - if (isCompaction && !sawCompactionOutput) { - const summary = "[earlier conversation was compacted]"; - const cmpItem = { - type: "compaction", - id: `cmp_${randomUUID()}`, - status: "completed", - encrypted_content: encodeCompactionSummary(summary), - }; - const cmpDone = { - type: "response.output_item.done", - output_index: maxOutputIndex++, - item: cmpItem, - }; - const cmpDoneText = encoder.encode(JSON.stringify(cmpDone)); - const cmpPrefix = encoder.encode("event: response.output_item.done\ndata: "); - const cmpSuffix = encoder.encode("\n\n"); - const cmpFrame = new Uint8Array(cmpPrefix.byteLength + cmpDoneText.byteLength + cmpSuffix.byteLength); - cmpFrame.set(cmpPrefix); - cmpFrame.set(cmpDoneText, cmpPrefix.byteLength); - cmpFrame.set(cmpSuffix, cmpPrefix.byteLength + cmpDoneText.byteLength); - controller.enqueue(cmpFrame); - sawCompactionOutput = true; - } - const completedResp: Record = { - ...resp, - status: "completed", - }; - delete completedResp.incomplete_details; - const completedPayload = { - ...normalized.payload, - type: "response.completed", - response: completedResp, - }; - relayText = JSON.stringify(completedPayload); - normalized.text = relayText; - normalized.payload = completedPayload; - type = "response.completed"; - } - } if (!controlFrame && !type.startsWith("response.") && type !== "error") return; let steeringEnded = false; let refusalStatus: number | undefined; diff --git a/src/server/responses/run-turn-execution.ts b/src/server/responses/run-turn-execution.ts index 78d66b1cd04..d6a607ba15d 100644 --- a/src/server/responses/run-turn-execution.ts +++ b/src/server/responses/run-turn-execution.ts @@ -911,7 +911,8 @@ export async function executeResponsesRunTurn( toolParameterSchemas, ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens === true && enforceDeclaredToolNames, + textFormat: parsed.options.textFormat, // grok-build's strict decoder dies on the typed response.heartbeat frame; its // eventsource layer tolerates comment keep-alives. Codex needs the opposite. ...(logCtx.surface === "grok" ? { heartbeatStyle: "comment" as const } : {}), @@ -1073,7 +1074,8 @@ export async function executeResponsesRunTurn( bareCustomToolNames, toolSearchToolNames, ...(routedCompaction ? { compaction: true, compactionRetention: parsed._compactionRetention } : {}), - completeOnMaxOutputTokens: config.completeOnMaxOutputTokens ?? false, + completeOnMaxOutputTokens: config.completeOnMaxOutputTokens === true && enforceDeclaredToolNames, + textFormat: parsed.options.textFormat, onProviderState: state => { providerState = state; }, onUsage: usage => { transportState.bindKeyUsageFromBridge(usage); diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index 8b62065b40a..b7a7b98a1ca 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -16,7 +16,6 @@ import type { NativeResponseControl } from "./native-response-control"; // (passthrough relay, adapter parsers, usage sniffing) is unchanged. import { codexWsReuseAcrossTurnsEnabled } from "../../config/codex-ws-reuse-setting"; -import { loadConfig } from "../../config"; import { compareBunVersions } from "../../lib/bun-stream-caps"; import { resolveProxyRoute, socks5ProxyFromEnv } from "../../lib/proxy-env"; import type { CodexWsQuotaObserver } from "./codex-ws-metadata"; @@ -253,6 +252,5 @@ export async function codexWsUpstreamFetch( beforeContinuation, onPhysicalDispatch, preOutputRefusalHold, bunVersion: typeof runtime === "string" ? runtime : runtime.version, - completeOnMaxOutputTokens: loadConfig().completeOnMaxOutputTokens ?? false, }); } diff --git a/src/types/config.ts b/src/types/config.ts index cfe79dee259..c8d5d06db16 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -571,6 +571,7 @@ export interface OcxConfig { emptyCompletionRetry?: boolean; /** Suppress allowlisted client-facing Codex transport hints; provider enforcement is unchanged. */ dropCodexSafetyBuffering?: boolean; + /** Opt in to completing truncated plain-text translated Responses turns; default off. */ completeOnMaxOutputTokens?: boolean; /** * Whether a login may open a browser on the machine running the proxy. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 0b5ed542af8..191b885859c 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1,4 +1,6 @@ { "update-reclaim-authorizer.test.ts": "update", "update-capability-policy.test.ts": "update", "update-service-home.test.ts": "update", "update-service-home-policy.test.ts": "update", "update-home-recovery.test.ts": "update", "update-recovery-authority.test.ts": "update", "update-recovery-boundaries.test.ts": "update", "update-home-launcher.test.ts": "update", "update-home-worker.test.ts": "update", "update-mutation-boundaries.test.ts": "update", "update-mutation-guard.test.ts": "update", "update-guarded-ops.test.ts": "update", "update-effect-authority.test.ts": "update", "update-windows-wrapper-authority.test.ts": "update", "update-mutation-scanner.test.ts": "update", "refreshed-upstream-deadlines.test.ts": "responses", "local-upstream-deadlines.test.ts": "responses", "service-desktop-recovery.test.ts": "service", "provider-input-modalities.test.ts": "providers", "turn-membership.test.ts": "responses", "turn-membership-ingress.test.ts": "responses", "workbuddy-quota.test.ts": "providers", "workbuddy-login.test.ts": "oauth", "workbuddy-adapter.test.ts": "adapters", "workbuddy-provider.test.ts": "providers", "link-tunnel-history.test.ts": "clients", "link-readiness.test.ts": "clients", "link-diagnostics.test.ts": "clients", "combo-description.test.ts": "routing", "combo-description-catalog.test.ts": "codex-integration", "catalog-auto-review-override.test.ts": "codex-integration", "management-auto-review-settings.test.ts": "server", "management-auto-review-auth.test.ts": "server", "prefer-model-alias-ids.test.ts": "server", "responses-mimo-envelope.test.ts": "responses", "responses-call-id-linkage.test.ts": "responses", "responses-phase-inference.test.ts": "responses", + "graceful-max-output-tokens.test.ts": "responses", + "graceful-max-output-tokens-exclusions.test.ts": "responses", "windows-config-root-async-writers.test.ts": "windows", "t3code-client.test.ts": "clients", "t3code-client-lifecycle.test.ts": "clients", diff --git a/tests/responses/graceful-max-output-tokens-exclusions.test.ts b/tests/responses/graceful-max-output-tokens-exclusions.test.ts new file mode 100644 index 00000000000..4c1e60313d0 --- /dev/null +++ b/tests/responses/graceful-max-output-tokens-exclusions.test.ts @@ -0,0 +1,168 @@ +import { describe, expect, test } from "bun:test"; +import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge"; +import type { AdapterEvent, OcxProviderContinuationState } from "../../src/types"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; + +type Options = NonNullable[7]>; +type Frame = { type: string; response?: Record; delta?: string; text?: string }; +const text: AdapterEvent = { type: "text_delta", text: "Partial answer\n```ts\nconst x =" }; +const terminals: AdapterEvent[] = [ + { type: "done", stopReason: "max_tokens", usage: { inputTokens: 5, outputTokens: 20 } }, + { type: "incomplete", reason: "max_output_tokens", usage: { inputTokens: 5, outputTokens: 20 } }, +]; + +async function deliver(events: AdapterEvent[], options: Options, streaming: boolean) { + if (!streaming) { + let providerState: OcxProviderContinuationState | undefined; + const response = buildResponseJSON(events, "test-model", { + ...options, onProviderState: state => { providerState = state; }, + }); + return { response, frames: [] as Frame[], providerState }; + } + const budget = createTestTranslatorBudget(); + let providerState: OcxProviderContinuationState | undefined; + async function* source() { yield* events; } + try { + const body = await new Response(bridgeToResponsesSSE( + source(), "test-model", undefined, undefined, undefined, undefined, 2_000, + { ...options, translatorBudget: budget, onCompletedResponse: (_response, state) => { providerState = state; } }, + )).text(); + const frames: Frame[] = body.split("\n").filter(line => line.startsWith("data: {")).map(line => JSON.parse(line.slice(6))); + const terminal = frames.filter(frame => ["response.completed", "response.incomplete", "response.failed"].includes(frame.type)); + expect(terminal).toHaveLength(1); + expect(terminal[0]!.type).toBe(`response.${terminal[0]!.response!.status}`); + return { response: terminal[0]!.response!, frames, providerState }; + } finally { budget.dispose(); } +} + +for (const streaming of [false, true]) { + for (const terminal of terminals) { + describe(`${streaming ? "SSE" : "JSON"} ${terminal.type} max-output completion`, () => { + test("explicit opt-in completes text without changing deltas, done text, or usage", async () => { + const { response, frames } = await deliver([text, terminal], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("completed"); + expect(response.incomplete_details).toBeUndefined(); + expect(response.usage).toMatchObject({ input_tokens: 5, output_tokens: 20 }); + expect((response.output as any[])[0].content[0].text).toBe(text.text); + if (streaming) { + expect(frames.filter(frame => frame.type === "response.output_text.delta").map(frame => frame.delta).join("")).toBe(text.text); + expect(frames.find(frame => frame.type === "response.output_text.done")!.text).toBe(text.text); + } + }); + + for (const enabled of [undefined, false]) { + test(`setting ${enabled} preserves incomplete`, async () => { + const { response } = await deliver([text, terminal], { completeOnMaxOutputTokens: enabled }, streaming); + expect(response.status).toBe("incomplete"); + expect(response.incomplete_details).toMatchObject({ reason: "max_output_tokens" }); + }); + } + + for (const type of ["json_schema", "json_object"] as const) { + test(`structured ${type} stays incomplete`, async () => { + const options = { completeOnMaxOutputTokens: true, textFormat: { type } }; + const { response } = await deliver([text, terminal], options, streaming); + expect(response.status).toBe("incomplete"); + expect(response.incomplete_details).toMatchObject({ reason: "max_output_tokens" }); + }); + } + + const exclusions: [string, AdapterEvent[]][] = [ + ["unfinished function", [{ type: "tool_call_start", id: "call_a", name: "lookup" }, { type: "tool_call_delta", arguments: '{"q":' }]], + ["unfinished function before later text", [{ type: "tool_call_start", id: "call_a", name: "lookup" }, { type: "tool_call_delta", arguments: "{}" }, text]], + ["explicitly incomplete tool end", [{ type: "tool_call_start", id: "call_a", name: "lookup" }, { type: "tool_call_delta", arguments: "{}" }, { type: "tool_call_end", incomplete: true }]], + ["completed tool output", [{ type: "tool_call_start", id: "call_a", name: "lookup" }, { type: "tool_call_delta", arguments: "{}" }, { type: "tool_call_end" }, text]], + ["orphan tool delta", [{ type: "tool_call_delta", arguments: "{}" }, text]], + ["unfinished hosted search", [{ type: "web_search_call_begin", id: "search_a" }, text]], + ["completed hosted search", [{ type: "web_search_call_end", id: "search_a", queries: ["fixture query"] }, text]], + ["reasoning output", [{ type: "thinking_delta", thinking: "pending thought" }, text]], + ["hidden raw reasoning", [{ type: "reasoning_raw_delta", text: "pending thought" }, text]], + ["signed progress", [{ type: "text_delta", text: "progress", anthropicThinkingSignature: "fixture-signature" }, text]], + ]; + for (const [name, events] of exclusions) { + test(`${name} stays incomplete`, async () => { + const { response } = await deliver([text, ...events, terminal], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("incomplete"); + expect(response.incomplete_details).toMatchObject({ reason: "max_output_tokens" }); + }); + } + + test("empty output stays incomplete", async () => { + const { response } = await deliver([terminal], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("incomplete"); + }); + + test("compaction truncation emits no replacement and commits no retention", async () => { + let commits = 0; + const { response } = await deliver([text, terminal], { + completeOnMaxOutputTokens: true, compaction: true, + compactionRetention: { encode: summary => `encoded:${summary}`, commit: () => { commits++; } }, + }, streaming); + expect(response.status).toBe("incomplete"); + expect(response.output).toEqual([]); + expect(commits).toBe(0); + }); + + test("provider state and end_turn survive the converted terminal", async () => { + const state = { cursor: { conversationId: "fixture-conversation", checkpointRef: "fixture-checkpoint" } }; + const { response, providerState } = await deliver([text, { ...terminal, providerState: state, endTurn: true }], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("completed"); + expect(providerState).toEqual(state); + expect(response.end_turn).toBe(true); + }); + + test("explicitly unfinished turn stays incomplete", async () => { + const { response } = await deliver([text, { ...terminal, endTurn: false }], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("incomplete"); + expect(response.end_turn).toBe(false); + }); + }); + } + + describe(`${streaming ? "SSE" : "JSON"} other exclusions and output integrity`, () => { + for (const reason of ["length", "max_output_tokens"]) { + test(`raw done ${reason} qualifies`, async () => { + const { response } = await deliver([text, { type: "done", stopReason: reason }], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("completed"); + }); + } + for (const stopReason of ["pause_turn", "model_context_window_exceeded", "model_context_window_exceeded_exception"]) { + test(`raw done ${stopReason} stays incomplete`, async () => { + const { response } = await deliver([text, { type: "done", stopReason }], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("incomplete"); + }); + } + for (const reason of ["content_filter", "pause_turn", "model_context_window_exceeded", "model_context_window_exceeded_exception", "adapter_eof"]) { + test(`${reason} stays incomplete`, async () => { + const { response } = await deliver([text, { type: "incomplete", reason }], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("incomplete"); + }); + } + test("adapter EOF stays incomplete", async () => { + const { response } = await deliver([text], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("incomplete"); + }); + test("errors stay failed", async () => { + const { response } = await deliver([text, { type: "error", message: "fixture failure", status: 502 }], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("failed"); + }); + test("earlier completed messages and valid empty fences remain exact", async () => { + const first = "```\n```"; + const { response } = await deliver([ + { type: "text_delta", text: first }, { type: "assistant_boundary" }, text, terminals[0]!, + ], { completeOnMaxOutputTokens: true }, streaming); + expect(response.status).toBe("completed"); + expect((response.output as any[]).map(item => item.content[0].text)).toEqual([first, text.text]); + }); + for (const enabled of [false, true]) { + test(`clean empty compaction never substitutes history (setting ${enabled})`, async () => { + const encoded: string[] = []; + await deliver([{ type: "done" }], { + completeOnMaxOutputTokens: enabled, compaction: true, + compactionRetention: { encode: summary => { encoded.push(summary); return "fixture-ciphertext"; }, commit: () => {} }, + }, streaming); + expect(encoded).toEqual([""]); + }); + } + }); +} diff --git a/tests/responses/graceful-max-output-tokens.test.ts b/tests/responses/graceful-max-output-tokens.test.ts index 65057cef046..608c97bae74 100644 --- a/tests/responses/graceful-max-output-tokens.test.ts +++ b/tests/responses/graceful-max-output-tokens.test.ts @@ -1,266 +1,215 @@ -import { describe, expect, test } from "bun:test"; -import { closeOpenMarkdownBlocks } from "../../src/responses/markdown-block-repair"; -import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge"; -import type { AdapterEvent } from "../../src/types"; -import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { afterAll, beforeAll, describe, expect, spyOn, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { ADAPTER_REGISTRY } from "../../src/adapters/registry"; +import { waitForNativeMainStartupGate } from "../../src/codex/native-profile-startup"; +import { saveConfig } from "../../src/config"; +import * as configModule from "../../src/config"; +import { configSchema } from "../../src/config/schema/config-schema"; +import { startServer } from "../../src/server"; import { codexWsExchange } from "../../src/server/responses/codex-ws-exchange"; -import type { CodexWsSession } from "../../src/server/responses/codex-ws-session"; -import type { PreparedCodexWsRequest } from "../../src/server/responses/codex-ws-request"; - -describe("closeOpenMarkdownBlocks", () => { - test("leaves normal and empty text unchanged", () => { - expect(closeOpenMarkdownBlocks("")).toBe(""); - expect(closeOpenMarkdownBlocks("Hello world!")).toBe("Hello world!"); - expect(closeOpenMarkdownBlocks("```ts\nconst x = 1;\n```\n")).toBe("```ts\nconst x = 1;\n```\n"); +import { CODEX_RESPONSES_HTTP_URL, prepareCodexWsRequest } from "../../src/server/responses/codex-ws-request"; +import { CodexWsSession } from "../../src/server/responses/codex-ws-session"; +import { codexWsUpstreamFetch } from "../../src/server/responses/ws-upstream"; +import type { OcxConfig } from "../../src/types"; +import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { installCompactionRoutingAclFixture } from "../helpers/compaction-routing-fixtures"; + +installCompactionRoutingAclFixture(); + +const TEXT = "Partial code\n```ts\nconst x ="; +const oldHome = process.env.OPENCODEX_HOME; +const originalCreate = ADAPTER_REGISTRY.anthropic.create; +let home: string; +let isolated: IsolatedCodexHome; +let upstream: ReturnType; +let server: ReturnType; +let runTurnCalls = 0; + +function frames(body: string): { name?: string; data: Record }[] { + return body.split("\n\n").flatMap(block => { + const lines = block.split("\n"); + const data = lines.find(line => line.startsWith("data: "))?.slice(6); + return !data || data === "[DONE]" ? [] : [{ + name: lines.find(line => line.startsWith("event: "))?.slice(7), data: JSON.parse(data), + }]; }); - - test("closes unclosed code fences", () => { - const truncated = "Here is some code:\n```typescript\nfunction greet() {\n return 'hi';"; - const repaired = closeOpenMarkdownBlocks(truncated); - expect(repaired).toBe("Here is some code:\n```typescript\nfunction greet() {\n return 'hi';\n```\n"); +} + +beforeAll(async () => { + home = mkdtempSync(join(tmpdir(), "ocx-graceful-output-")); + isolated = installIsolatedCodexHome("ocx-graceful-output-codex-"); + process.env.OPENCODEX_HOME = home; + upstream = Bun.serve({ + hostname: "127.0.0.1", port: 0, + async fetch(request) { + const body = await request.json() as { stream?: boolean }; + if (!body.stream) return Response.json({ + id: "msg_fixture", type: "message", role: "assistant", model: "fixture-model", + content: [{ type: "text", text: TEXT }], stop_reason: "max_tokens", + usage: { input_tokens: 5, output_tokens: 20 }, + }); + const frame = (type: string, data: Record) => `event: ${type}\ndata: ${JSON.stringify({ type, ...data })}\n\n`; + return new Response([ + frame("message_start", { message: { id: "msg_fixture", model: "fixture-model", usage: { input_tokens: 5, output_tokens: 0 } } }), + frame("content_block_start", { index: 0, content_block: { type: "text", text: "" } }), + frame("content_block_delta", { index: 0, delta: { type: "text_delta", text: TEXT } }), + frame("content_block_stop", { index: 0 }), + frame("message_delta", { delta: { stop_reason: "max_tokens" }, usage: { output_tokens: 20 } }), + frame("message_stop", {}), + ].join(""), { headers: { "content-type": "text/event-stream" } }); + }, }); - - test("closes nested or multiple open code fences", () => { - const truncated = "```markdown\nInside markdown:\n~~~python\nprint(1)"; - const repaired = closeOpenMarkdownBlocks(truncated); - expect(repaired).toContain("~~~\n"); - expect(repaired).toContain("```\n"); + ADAPTER_REGISTRY.anthropic.create = (provider, context) => { + const adapter = originalCreate(provider, context); + if (provider.defaultModel !== "runturn-model") return adapter; + return { ...adapter, async runTurn(_parsed, _incoming, emit) { + runTurnCalls++; + emit({ type: "text_delta", text: TEXT }); + emit({ type: "incomplete", reason: "max_output_tokens", usage: { inputTokens: 5, outputTokens: 20 } }); + } }; + }; + const provider = (model: string) => ({ + adapter: "anthropic", baseUrl: `http://127.0.0.1:${upstream.port}`, apiKey: "sk-test", + allowPrivateNetwork: true, liveModels: false, defaultModel: model, models: [model], }); + saveConfig({ + hostname: "127.0.0.1", port: 0, defaultProvider: "fixture", completeOnMaxOutputTokens: true, + providers: { fixture: provider("fixture-model"), runturn: provider("runturn-model") }, + } as OcxConfig); + server = startServer(0); + await waitForNativeMainStartupGate(); +}); - test("closes unclosed paired tags", () => { - const truncated = "
More info\nSome body text"; - const repaired = closeOpenMarkdownBlocks(truncated); - expect(repaired).toBe("
More info\nSome body text\n
\n"); - }); +afterAll(async () => { + await server?.stop(true); + await upstream?.stop(true); + ADAPTER_REGISTRY.anthropic.create = originalCreate; + if (oldHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = oldHome; + isolated?.restore(); + if (home) removeTreeWithRetry(home); }); -describe("graceful completion on max_output_tokens: bridgeToResponsesSSE", () => { - async function streamEvents(events: AdapterEvent[], options: Parameters[7]) { - const budget = createTestTranslatorBudget(); - async function* gen() { - for (const e of events) yield e; - } - const sseStream = bridgeToResponsesSSE( - gen(), - "test-model", - undefined, - undefined, - undefined, - () => {}, - 100, - { ...options, translatorBudget: budget }, - ); - const text = await new Response(sseStream).text(); - const parsed: any[] = []; - for (const block of text.split("\n\n")) { - for (const line of block.split("\n")) { - if (line.startsWith("data: ") && line !== "data: [DONE]") { - parsed.push(JSON.parse(line.slice(6))); - } +async function post(path: string, body: Record) { + const response = await fetch(new URL(path, server.url), { + method: "POST", headers: { "content-type": "application/json", "anthropic-version": "2023-06-01" }, + body: JSON.stringify(body), + }); + expect(response.status).toBe(200); + return response; +} + +describe("opt-in delivery stays on Responses text only", () => { + for (const model of ["fixture/fixture-model", "runturn/runturn-model"]) { + for (const stream of [false, true]) { + test(`${model} ${stream ? "SSE" : "JSON"} completes text`, async () => { + const before = runTurnCalls; + const response = await post("/v1/responses", { model, input: "hi", stream }); + const result = stream ? frames(await response.text()).find(frame => frame.name === "response.completed")?.data.response : await response.json(); + expect(result?.status).toBe("completed"); + expect(result.output[0].content[0].text).toBe(TEXT); + if (model.startsWith("runturn/")) expect(runTurnCalls - before).toBe(1); + }); + for (const type of ["json_schema", "json_object"]) { + test(`${model} ${stream ? "SSE" : "JSON"} excludes text.format ${type}`, async () => { + const response = await post("/v1/responses", { + model, input: "hi", stream, text: { format: { type, name: "fixture", schema: { type: "object" } } }, + }); + const result = stream ? frames(await response.text()).find(frame => frame.name === "response.incomplete")?.data.response : await response.json(); + expect(result?.status).toBe("incomplete"); + expect(result.incomplete_details.reason).toBe("max_output_tokens"); + }); } + test(`${model} ${stream ? "SSE" : "JSON"} Chat preserves length`, async () => { + const response = await post("/v1/chat/completions", { model, stream, messages: [{ role: "user", content: "hi" }] }); + const result = stream ? frames(await response.text()).find(frame => frame.data.choices?.[0]?.finish_reason)?.data : await response.json(); + expect(result.choices[0].finish_reason).toBe("length"); + }); + test(`${model} ${stream ? "SSE" : "JSON"} Messages preserves max_tokens`, async () => { + const response = await post("/v1/messages", { model, stream, max_tokens: 20, messages: [{ role: "user", content: "hi" }] }); + const result = stream ? frames(await response.text()).find(frame => frame.data.delta?.stop_reason)?.data.delta : await response.json(); + expect(result.stop_reason).toBe("max_tokens"); + }); } - return parsed; } - - test("emits response.completed and repairs markdown fences in normal turn", async () => { - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Here is partial code:\n```python\ndef run():\n pass" }, - { type: "done", stopReason: "max_tokens", usage: { inputTokens: 10, outputTokens: 100 } }, - ]; - const sse = await streamEvents(events, { - completeOnMaxOutputTokens: true, - }); - - const terminal = sse.find(e => e.type === "response.completed"); - expect(terminal).toBeDefined(); - expect(terminal.response.status).toBe("completed"); - - // Markdown fence delta emitted - const textDeltas = sse.filter(e => e.type === "response.output_text.delta").map(e => e.delta); - const fullText = textDeltas.join(""); - expect(fullText).toContain("```\n"); - }); - - test("emits compaction item and commits retention on compaction max_output_tokens", async () => { - let committed = false; - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Summary up to limit..." }, - { type: "done", stopReason: "max_tokens", usage: { inputTokens: 10, outputTokens: 100 } }, - ]; - const sse = await streamEvents(events, { - completeOnMaxOutputTokens: true, - compaction: true, - compactionRetention: { - encode: s => `encoded:${s}`, - commit: () => { committed = true; }, - }, - }); - - const terminal = sse.find(e => e.type === "response.completed"); - expect(terminal).toBeDefined(); - expect(terminal.response.status).toBe("completed"); - - const compactionDone = sse.find(e => e.type === "response.output_item.done" && e.item?.type === "compaction"); - expect(compactionDone).toBeDefined(); - expect(committed).toBe(true); - }); - - test("recovers from explicit incomplete event with max_output_tokens", async () => { - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Accumulated response before cut off" }, - { type: "incomplete", reason: "max_output_tokens", usage: { inputTokens: 5, outputTokens: 50 } }, - ]; - const sse = await streamEvents(events, { - completeOnMaxOutputTokens: true, - }); - - const terminal = sse.find(e => e.type === "response.completed"); - expect(terminal).toBeDefined(); - expect(terminal.response.status).toBe("completed"); - expect(sse.some(e => e.type === "response.incomplete")).toBe(false); - }); - - test("keeps response.incomplete when completeOnMaxOutputTokens is false", async () => { - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Some text" }, - { type: "done", stopReason: "max_tokens" }, - ]; - const sse = await streamEvents(events, { - completeOnMaxOutputTokens: false, - }); - - const terminal = sse.find(e => e.type === "response.incomplete"); - expect(terminal).toBeDefined(); - expect(terminal.response.incomplete_details?.reason).toBe("max_output_tokens"); - expect(sse.some(e => e.type === "response.completed")).toBe(false); - }); - - test("keeps response.incomplete for content_filter even if completeOnMaxOutputTokens is true", async () => { - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Some text" }, - { type: "done", stopReason: "content_filter" }, - ]; - const sse = await streamEvents(events, { - completeOnMaxOutputTokens: true, - }); - - const terminal = sse.find(e => e.type === "response.incomplete"); - expect(terminal).toBeDefined(); - expect(terminal.response.incomplete_details?.reason).toBe("content_filter"); + test("missing, false, and malformed settings do not enable the policy", () => { + const base = { defaultProvider: "fixture", providers: { fixture: { + adapter: "anthropic", baseUrl: "http://127.0.0.1:1", allowPrivateNetwork: true, defaultModel: "fixture-model", models: ["fixture-model"], + } } }; + for (const value of [undefined, false, "true", 1, null]) { + const parsed = configSchema.parse({ ...base, completeOnMaxOutputTokens: value }); + expect(parsed.completeOnMaxOutputTokens === true).toBe(false); + } + expect(configSchema.parse({ ...base, completeOnMaxOutputTokens: true }).completeOnMaxOutputTokens).toBe(true); }); }); -describe("graceful completion on max_output_tokens: buildResponseJSON", () => { - test("returns status completed and repairs markdown in buffered response", () => { - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Hello\n```js\nconsole.log(1);" }, - { type: "done", stopReason: "max_tokens", usage: { inputTokens: 5, outputTokens: 20 } }, - ]; - const json = buildResponseJSON(events, "test-model", { - completeOnMaxOutputTokens: true, - }); - - expect(json.status).toBe("completed"); - expect((json as any).incomplete_details).toBeUndefined(); - - const output = json.output as any[]; - const msg = output.find(item => item.type === "message"); - expect(msg).toBeDefined(); - expect(msg.content[0].text).toBe("Hello\n```js\nconsole.log(1);\n```\n"); - }); - - test("emits compaction item and commits retention on compaction max_output_tokens in buffered response", () => { - let committed = false; - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Compacted summary text" }, - { type: "done", stopReason: "max_tokens", usage: { inputTokens: 5, outputTokens: 20 } }, - ]; - const json = buildResponseJSON(events, "test-model", { - completeOnMaxOutputTokens: true, - compaction: true, - compactionRetention: { - encode: s => `encoded:${s}`, - commit: () => { committed = true; }, - }, - }); - - expect(json.status).toBe("completed"); - expect(committed).toBe(true); - const output = json.output as any[]; - expect(output.some(item => item.type === "compaction")).toBe(true); - }); - - test("returns status incomplete when completeOnMaxOutputTokens is false", () => { - const events: AdapterEvent[] = [ - { type: "text_delta", text: "Hello" }, - { type: "done", stopReason: "max_tokens" }, - ]; - const json = buildResponseJSON(events, "test-model", { - completeOnMaxOutputTokens: false, - }); - expect(json.status).toBe("incomplete"); - expect((json as any).incomplete_details?.reason).toBe("max_output_tokens"); +test("native WebSocket preserves incomplete framing with global opt-in enabled", async () => { + const payload = { type: "response.incomplete", response: { + id: "resp_fixture", status: "incomplete", output: [], incomplete_details: { reason: "max_output_tokens" }, + } }; + const listener = Bun.serve({ + hostname: "127.0.0.1", port: 0, + fetch(req, server) { if (!server.upgrade(req)) return new Response(null, { status: 400 }); }, + websocket: { message(socket) { socket.send(JSON.stringify(payload)); } }, }); + const init: RequestInit = { method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "fixture-model", stream: true, input: [] }) }; + const prepared = prepareCodexWsRequest(CODEX_RESPONSES_HTTP_URL, init)!; + const session = new CodexWsSession(`ws://127.0.0.1:${listener.port}`, prepared.headers); + try { + expect(session.reserve()).toBe(true); + const response = await codexWsExchange({ session, prepared, init, url: CODEX_RESPONSES_HTTP_URL, + sseFallback: (async () => { throw new Error("unexpected fallback"); }) as typeof fetch }); + const all = frames(await response.text()); + expect(all).toEqual([{ name: "response.incomplete", data: payload }]); + } finally { session.dispose(); await listener.stop(true); } }); -describe("graceful completion on max_output_tokens: codexWsExchange", () => { - test("transforms response.incomplete into response.completed and injects compaction item", async () => { - let messageHandler: ((event: MessageEvent) => void) | undefined; - const fakeSocket = { - send: () => {}, - close: () => {}, - addEventListener: (type: string, handler: any) => { - if (type === "message") messageHandler = handler; - }, - removeEventListener: () => {}, - }; - - const session: CodexWsSession = { - socket: fakeSocket as any, - dispose: () => {}, - metrics: () => ({} as any), - on: () => {}, - off: () => {}, - bindOwner: () => () => {}, - release: () => {}, - } as any; - - const prepared: PreparedCodexWsRequest = { - frameText: JSON.stringify({ payload: { compaction_trigger: "auto" } }), - headers: new Headers(), - } as any; - - const responsePromise = codexWsExchange({ - session, - url: "https://api.openai.com/v1/responses", - init: {}, - prepared, - completeOnMaxOutputTokens: true, - sseFallback: async () => new Response("fallback"), - }); - - // Simulate WS emitting response.incomplete due to max_output_tokens - setTimeout(() => { - messageHandler?.({ - data: JSON.stringify({ - type: "response.incomplete", - response: { - id: "resp_123", - status: "incomplete", - incomplete_details: { reason: "max_output_tokens" }, - }, - }), - } as MessageEvent); - }, 10); - - const res = await responsePromise; - expect(res.status).toBe(200); - - const text = await res.text(); - expect(text).toContain("event: response.output_item.done"); - expect(text).toContain('"type":"compaction"'); - expect(text).toContain("event: response.completed"); - expect(text).toContain('"status":"completed"'); - expect(text).not.toContain("event: response.incomplete"); +test("WS upstream does not reload config after reserving its session", async () => { + const realSocket = globalThis.WebSocket; + const realLoad = configModule.loadConfig; + const reserve = CodexWsSession.prototype.reserve; + const proxyKeys = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy"]; + const proxyEnv = Object.fromEntries(proxyKeys.map(key => [key, process.env[key]])); + let reserved = false; + let lateLoads = 0; + const loadSpy = spyOn(configModule, "loadConfig").mockImplementation((...args) => { + if (reserved) lateLoads++; + return realLoad(...args); + }); + const reserveSpy = spyOn(CodexWsSession.prototype, "reserve").mockImplementation(function (this: CodexWsSession) { + const result = reserve.call(this); + if (result) reserved = true; + return result; }); + class Socket extends EventTarget { + readyState = 1; + constructor() { super(); queueMicrotask(() => this.dispatchEvent(new Event("open"))); } + send() { queueMicrotask(() => this.dispatchEvent(new MessageEvent("message", { data: JSON.stringify({ + type: "response.incomplete", response: { id: "r_no_reload", status: "incomplete", output: [], incomplete_details: { reason: "max_output_tokens" } }, + }) }))); } + close() { this.dispatchEvent(new Event("close")); } + } + try { + for (const key of proxyKeys) delete process.env[key]; + globalThis.WebSocket = Socket as unknown as typeof WebSocket; + const response = await codexWsUpstreamFetch(CODEX_RESPONSES_HTTP_URL, { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "fixture-model", stream: true, input: [] }), + }, (async () => { throw new Error("unexpected HTTP fallback"); }) as typeof fetch, "1.4.0"); + const all = frames(await response.text()); + expect(reserved).toBe(true); + expect(lateLoads).toBe(0); + expect(all.map(frame => frame.name)).toEqual(["response.incomplete"]); + } finally { + loadSpy.mockRestore(); reserveSpy.mockRestore(); globalThis.WebSocket = realSocket; + for (const key of proxyKeys) { + if (proxyEnv[key] === undefined) delete process.env[key]; + else process.env[key] = proxyEnv[key]; + } + } }); From 6ddde6dac4adcad25c12fa23448ad4ccdbc42d60 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:09:32 +0900 Subject: [PATCH 5/6] docs(responses): document narrowed token-limit completion policy --- .../docs/reference/configuration/server.md | 9 ++++++ structure/INDEX.md | 8 ++++-- structure/config.md | 2 +- structure/manifest.json | 13 +++++++++ structure/transports/max-output-completion.md | 28 +++++++++++++++++++ structure/transports/responses.md | 1 + 6 files changed, 58 insertions(+), 3 deletions(-) create mode 100644 structure/transports/max-output-completion.md diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 8c9f113168b..f3aba6cd42e 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -8,6 +8,15 @@ runs helper features around provider requests. ## Server fields +`completeOnMaxOutputTokens` is an optional boolean, default `false`. Set it to `true` to +report a token-limited translated Responses text answer as `completed` rather than `incomplete`. +The answer remains truncated: OpenCodex preserves its exact text, including unfinished Markdown. +Only nonempty plain-text output qualifies. Tool calls and hosted operations (even if completed), +reasoning, `text.format` with `json_schema` or `json_object`, compaction, explicitly unfinished +turns, errors and missing terminal events remain excluded. Context-window exhaustion and +`pause_turn` remain incomplete. Chat Completions, Anthropic Messages and native Responses +passthrough, including native WebSocket output, retain their existing stop behavior. + | Field | Type | Default | Meaning | | --- | --- | --- | --- | | `port` | `number` | `10100` | Proxy listen port. | diff --git a/structure/INDEX.md b/structure/INDEX.md index 909e73fc0df..44a73e48d1a 100644 --- a/structure/INDEX.md +++ b/structure/INDEX.md @@ -52,6 +52,7 @@ The wire surfaces a client actually talks to. | [`transports/anthropic-thinking-recovery.md`](transports/anthropic-thinking-recovery.md) | One-shot exact signature-binding recovery before output, and successful-retry memo boundaries. | | [`transports/response-turn-membership.md`](transports/response-turn-membership.md) | Embedding-owned signing material, dedicated authenticated caller/own-thread scope and pre-parser structural normalization. | | [`transports/response-citations.md`](transports/response-citations.md) | OpenAI-operated outbound citation repair and generated JSON/SSE annotation titles. | +| [`transports/max-output-completion.md`](transports/max-output-completion.md) | Default-off translated Responses text-only token-limit completion and exclusions. | | [`transports/responses.md`](transports/responses.md) | The Responses HTTP/SSE endpoint, dispatch, credential and upload boundaries, and core module ownership. | | [`transports/native-reasoning-retention.md`](transports/native-reasoning-retention.md) | Optional native reasoning ciphertext forwarding on model/account switches, configuration and management controls. | | [`transports/responses-citations.md`](transports/responses-citations.md) | Destination-gated citation repair for ordinary Responses and every native compact send. | @@ -217,6 +218,7 @@ A source area can be described by more than one doc, because these docs are orga | `src/advisor/` | [`advisor.md`](advisor.md) | | `src/bridge.ts` | [`transports/responses.md`](transports/responses.md) | | `src/bridge/` | [`transports/response-citations.md`](transports/response-citations.md)
[`transports/responses.md`](transports/responses.md)
[`transports/responses-wire-shapes.md`](transports/responses-wire-shapes.md)
[`providers/anthropic-progress.md`](providers/anthropic-progress.md) | +| `src/bridge/max-output-completion.ts` | [`transports/max-output-completion.md`](transports/max-output-completion.md) | | `src/bridge/sse.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md) | | `src/chat/` | [`runtime.md`](runtime.md)
[`transports/byte-accounting.md`](transports/byte-accounting.md)
[`transports/inventory.md`](transports/inventory.md)
[`data-planes/inbound-compat.md`](data-planes/inbound-compat.md)
[`providers-and-adapters.md`](providers-and-adapters.md)
[`providers/chat-compat.md`](providers/chat-compat.md) | | `src/chat/reasoning-details.ts` | [`data-planes/chat-reasoning-details.md`](data-planes/chat-reasoning-details.md) | @@ -291,6 +293,7 @@ A source area can be described by more than one doc, because these docs are orga | `src/config/live-reconcile.ts` | [`pricing.md`](pricing.md) | | `src/config/persisted-mutation.ts` | [`runtime.md`](runtime.md) | | `src/config/schema/account-auto-pause.ts` | [`account-auto-pause.md`](account-auto-pause.md) | +| `src/config/schema/config-schema.ts` | [`transports/max-output-completion.md`](transports/max-output-completion.md) | | `src/config/schema/leaf-validators.ts` | [`transports/timeouts.md`](transports/timeouts.md)
[`pricing.md`](pricing.md) | | `src/generated/` | [`runtime.md`](runtime.md) | | `src/github/` | [`runtime.md`](runtime.md) | @@ -370,7 +373,7 @@ A source area can be described by more than one doc, because these docs are orga | `src/server/management/quota-pool-routes.ts` | [`quota-pool.md`](quota-pool.md) | | `src/server/responses/` | [`data-planes/claude-tool-loading.md`](data-planes/claude-tool-loading.md) | | `src/server/responses/account-auto-pause-recovery.ts` | [`account-auto-pause.md`](account-auto-pause.md) | -| `src/server/responses/adapter-delivery.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md) | +| `src/server/responses/adapter-delivery.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md)
[`transports/max-output-completion.md`](transports/max-output-completion.md) | | `src/server/responses/adapter-dispatch.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md) | | `src/server/responses/advisor-plan-slot.ts` | [`advisor.md`](advisor.md) | | `src/server/responses/advisor-slot.ts` | [`advisor.md`](advisor.md) | @@ -381,6 +384,7 @@ A source area can be described by more than one doc, because these docs are orga | `src/server/responses/passthrough-dispatch.ts` | [`providers/opengateway.md`](providers/opengateway.md) | | `src/server/responses/request-prepare.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md) | | `src/server/responses/request-transport.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md)
[`account-auto-pause.md`](account-auto-pause.md) | +| `src/server/responses/run-turn-execution.ts` | [`transports/max-output-completion.md`](transports/max-output-completion.md) | | `src/server/responses/turn-membership.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md) | | `src/service-manager-probe.ts` | [`ops/service-and-sidecars.md`](ops/service-and-sidecars.md) | | `src/service.ts` | [`runtime.md`](runtime.md)
[`ops/docs-and-release.md`](ops/docs-and-release.md) | @@ -409,7 +413,7 @@ A source area can be described by more than one doc, because these docs are orga | `src/types.ts` | [`runtime.md`](runtime.md)
[`config.md`](config.md) | | `src/types/` | [`providers/generic-account-pool.md`](providers/generic-account-pool.md)
[`transports/native-reasoning-retention.md`](transports/native-reasoning-retention.md)
[`providers/anthropic-progress.md`](providers/anthropic-progress.md)
[`providers-and-adapters.md`](providers-and-adapters.md)
[`fallback-policy.md`](fallback-policy.md) | | `src/types/account-auto-pause.ts` | [`account-auto-pause.md`](account-auto-pause.md) | -| `src/types/config.ts` | [`transports/timeouts.md`](transports/timeouts.md) | +| `src/types/config.ts` | [`transports/timeouts.md`](transports/timeouts.md)
[`transports/max-output-completion.md`](transports/max-output-completion.md) | | `src/types/provider.ts` | [`pricing.md`](pricing.md) | | `src/update/` | [`runtime.md`](runtime.md)
[`ops/service-and-sidecars.md`](ops/service-and-sidecars.md) | | `src/update/guarded-port-reclaim.ts` | [`ops/update-supervised-recovery.md`](ops/update-supervised-recovery.md) | diff --git a/structure/config.md b/structure/config.md index 5b3301556a4..74c31237622 100644 --- a/structure/config.md +++ b/structure/config.md @@ -1,5 +1,5 @@ # Config Surface -See [local timeout policy and OpenCode export](transports/timeouts.md). Pasted GCP JSON persists only as `gcp-sa:` references in one pool transaction in `src/providers/api-keys.ts`; post-commit deletion follows [Vertex credential markers](providers/google.md#vertex-credential-markers). +See [local timeout policy and OpenCode export](transports/timeouts.md) and default-off [`completeOnMaxOutputTokens`](transports/max-output-completion.md). Pasted GCP JSON persists only as `gcp-sa:` references in one pool transaction in `src/providers/api-keys.ts`; post-commit deletion follows [Vertex credential markers](providers/google.md#vertex-credential-markers). `src/config/schema/leaf-validators.ts` accepts only boolean `apiKeys[].allowMeteredComboTargets`; malformed entries are dropped rather than promoted. `src/types/config.ts` and combo validation define optional boolean target `metered`, normalized false when omitted, and optional per-combo `allowMeteredTargets` (default false, literal boolean validation shared by config schema and management API). Explicit consent is required under the [combo runtime contract](runtime.md#request-local-target-compatibility). Model-price overrides follow the [pricing contract](pricing.md). Quota activation reuses the existing next-reset fields without adding a polling configuration key. See the [quota activation contract](providers/openai-tiers.md#public-provider-contract). `src/types/config.ts` defines `anthropicAccountPool.routes` as ordered model rules. `src/oauth/anthropic-model-routes.ts` validates bounded names, full case-sensitive globs and stored account IDs; `src/config/diagnostics.ts` rejects malformed candidate writes. Load retains a malformed hand edit so unrelated providers survive, while the enabled Anthropic selector refuses it locally before dispatch. Saved rules remain inert when the pool is disabled; affinity is process-local. Request logs use only the rule’s 1-based `route:#` position, never its configured name. `src/cli/hub-disable.ts` removes interdependent hub/loopback/role/hostname settings through `mutatePersistedConfig` and `validateConfigCandidate` in one commit. Top-level deletion intent uses `src/config/rebase-provenance.ts` and authoritative disk provenance in `src/config/live-reconcile.ts` to preserve removed hub bindings through later live saves, `src/config/live-topology.ts` keeps an independent active-topology baseline and merges desired next-start settings into a separate persistence projection, retaining actual socket identity and live hub/client topology during guarded saves and OAuth disk adoption. Client connections and runtimeRole remain one topology unit through connect/disconnect, including fresh disk-only connections. Disk additions/value updates and deletions all survive repeated saves without changing active topology. `src/config/api-key-reconcile.ts` merges key rows by stable ID under the existing save lock; either side's revocation defeats stale edits while disjoint names and rotations merge; unchanged live key membership/order adopts persisted ordering. `tests/config/hub-disable-live-save.test.ts` exercises disable followed by live key creation, rename, rotation, unrelated settings and repeated saves. Saved API keys have no immutable enrollment marker; explicit exact-ID revocation preserves all unselected keys. The WorkBuddy AI account row pins its adapter and origin under the [WorkBuddy contract](providers/workbuddy.md). JEV `decisionQuotaSignals` and partial `decisionQuotaTiers` use the shared validator and sparse preserve/clear/strategy-switch contract in [loaded advisory quota](providers/jev-decision.md#loaded-advisory-quota); the opt-in JEV `sessionPin` follows the same contract under [session pin](providers/jev-decision.md#session-pin). Native reasoning retention follows [its optional configuration contract](transports/native-reasoning-retention.md#configuration-and-management). Native function-result injection follows [the separate opt-in control contract](transports/streaming-health.md#experimental-native-function-result-injection); this surface does not infer upstream support or alter its defaults. diff --git a/structure/manifest.json b/structure/manifest.json index 9fa23803f1e..25cc4c07c14 100644 --- a/structure/manifest.json +++ b/structure/manifest.json @@ -350,6 +350,19 @@ "scope": "OpenAI-operated outbound citation repair and generated JSON/SSE annotation titles.", "documents": ["src/adapters/openai-responses/", "src/bridge/"] }, + { + "path": "transports/max-output-completion.md", + "tier": 3, + "title": "Opt-in Responses Text Completion", + "scope": "Default-off translated Responses text-only token-limit completion and exclusions.", + "documents": [ + "src/bridge/max-output-completion.ts", + "src/config/schema/config-schema.ts", + "src/types/config.ts", + "src/server/responses/adapter-delivery.ts", + "src/server/responses/run-turn-execution.ts" + ] + }, { "path": "transports/responses.md", "tier": 3, diff --git a/structure/transports/max-output-completion.md b/structure/transports/max-output-completion.md new file mode 100644 index 00000000000..0207d7c5f83 --- /dev/null +++ b/structure/transports/max-output-completion.md @@ -0,0 +1,28 @@ +# Opt-in Responses text completion + +`completeOnMaxOutputTokens` is an optional boolean in `src/config/schema/config-schema.ts` +and `src/types/config.ts`. Absence, false and malformed values leave it off. +`src/server/responses/adapter-delivery.ts` and `src/server/responses/run-turn-execution.ts` +pass request-local opt-in and parsed `textFormat` only to translated Responses delivery. +Chat and Messages retain their existing stop semantics. Native Responses passthrough, +including native WebSocket delivery, does not participate or reload configuration for this policy. + +`src/bridge/max-output-completion.ts` observes the original adapter events before bridge cleanup. +Only nonempty plain text, optional message boundaries and heartbeats qualify. Any tool or +hosted operation, including one later closed by bridge cleanup or explicitly completed, +excludes the turn. Reasoning (including hidden and signed progress), structured `textFormat`, +compaction and explicit `endTurn: false` also exclude it. The allowlist excludes future event +types until they are explicitly reviewed. Errors and missing terminals retain their failure +or incomplete shape. Raw `max_tokens`, `length` and `max_output_tokens` stops qualify; +`pause_turn` and context-window exhaustion remain incomplete even though the general +truncation classifier maps them to the same public incomplete reason. + +The eligible terminal becomes `completed` and omits `incomplete_details`. The model's text +is unchanged, including earlier messages, open Markdown and valid empty fenced blocks. +Usage, `end_turn` and provider continuation state survive both buffered and streamed delivery. +Compaction retains its existing history and retention-commit rules, without placeholders. + +`tests/responses/graceful-max-output-tokens-exclusions.test.ts` covers both bridge shapes +and terminal forms. `tests/responses/graceful-max-output-tokens.test.ts` exercises the +real server's adapter and runTurn delivery, both structured formats, Chat/Messages stop +semantics, malformed/default config, native WS framing and no config reload after reservation. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 810c8d782e4..82585590f56 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -573,6 +573,7 @@ reuses `beginInferenceAttempt` and `createFinalRequestLog` with its own 401/429 ## Adapter-to-Responses bridge +Opt-in translated Responses text-only token-limit completion follows [the text completion policy](max-output-completion.md). `src/bridge.ts` is a re-export facade; the implementation lives in `src/bridge/`. `src/bridge/sse.ts` (`bridgeToResponsesSSE`) turns adapter events into the Responses SSE stream; its once-only first-output observer includes nonempty `tool_call_delta` arguments (including custom-tool input), but not tool-start scaffolding or empty arguments, and `src/bridge/response-json.ts` (`buildResponseJSON`) builds the non-streaming Responses body From 8b26d105da9ca3f1a7597004c55e79adb1d58051 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Thu, 29 Oct 2026 06:34:08 +0900 Subject: [PATCH 6/6] docs(responses): index token-limit setting and bridge ownership Add the English server-field index entry and map the streaming and buffered bridge builders to their completion-policy contract. Preserve the detailed eligibility explanation and runtime behavior. Co-authored-by: Omar <13569609+omar391@users.noreply.github.com> --- docs-site/src/content/docs/reference/configuration/server.md | 1 + structure/INDEX.md | 3 ++- structure/manifest.json | 2 ++ structure/transports/max-output-completion.md | 3 ++- 4 files changed, 7 insertions(+), 2 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index f3aba6cd42e..57475797363 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -19,6 +19,7 @@ passthrough, including native WebSocket output, retain their existing stop behav | Field | Type | Default | Meaning | | --- | --- | --- | --- | +| `completeOnMaxOutputTokens?` | `boolean` | `false` | Opt in to completing eligible token-limited translated Responses text output. Text stays truncated; see the eligibility and exclusions above. | | `port` | `number` | `10100` | Proxy listen port. | | `hostname?` | `string` | `"127.0.0.1"` | Bind address. A non-loopback bind requires a data-admission token, resolved from `OPENCODEX_API_AUTH_TOKEN`, then `OCX_API_TOKEN_FILE`, then the installed owner-only `service-api-token` — nothing has to be exported by hand. See [Remote access](#remote-access). | | `proxy?` | `string` | — | Outbound HTTP(S) or SOCKS5 proxy URL (`socks5://host:port`), `${ENV_VAR}`, or `"auto"`. HTTP URLs apply to `HTTP_PROXY` / `HTTPS_PROXY` when those are unset. SOCKS5 URLs use OpenCodex's real SOCKS5 transport and are also exposed through `ALL_PROXY` (`ocx start --socks5`); inherited `HTTP(S)_PROXY` is cleared in this process. Loopback stays in `NO_PROXY`. `"auto"` reads Windows WinINET or macOS static HTTP/HTTPS settings once at startup. Inherited HTTP(S) proxy variables skip discovery; on macOS, inherited `ALL_PROXY`/`all_proxy` also skips it. Windows keeps separate `http=` and `https=` entries; a bare `ProxyServer` applies to both. macOS translates IP literals, `*`, and a valid `*.` glob to Bun's `.` bypass. That glob also bypasses the bare apex ``. The exact link-local ranges `169.254/16`, `169.254.0.0/16`, and `fe80::/10` are omitted with a diagnostic: link-local IP literals use the proxy. Other CIDRs, globs, and simple-host exceptions refuse discovery without changing the proxy environment. Disabled, malformed, PAC/WPAD, SOCKS-only, and live changes are not followed. Restart after changing system settings. | diff --git a/structure/INDEX.md b/structure/INDEX.md index d0e0f659fa0..8b5d91749b9 100644 --- a/structure/INDEX.md +++ b/structure/INDEX.md @@ -219,7 +219,8 @@ A source area can be described by more than one doc, because these docs are orga | `src/bridge.ts` | [`transports/responses.md`](transports/responses.md) | | `src/bridge/` | [`transports/response-citations.md`](transports/response-citations.md)
[`transports/responses.md`](transports/responses.md)
[`transports/responses-wire-shapes.md`](transports/responses-wire-shapes.md)
[`providers/anthropic-progress.md`](providers/anthropic-progress.md) | | `src/bridge/max-output-completion.ts` | [`transports/max-output-completion.md`](transports/max-output-completion.md) | -| `src/bridge/sse.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md) | +| `src/bridge/response-json.ts` | [`transports/max-output-completion.md`](transports/max-output-completion.md) | +| `src/bridge/sse.ts` | [`transports/response-turn-membership.md`](transports/response-turn-membership.md)
[`transports/max-output-completion.md`](transports/max-output-completion.md) | | `src/chat/` | [`runtime.md`](runtime.md)
[`transports/byte-accounting.md`](transports/byte-accounting.md)
[`transports/inventory.md`](transports/inventory.md)
[`data-planes/inbound-compat.md`](data-planes/inbound-compat.md)
[`providers-and-adapters.md`](providers-and-adapters.md)
[`providers/chat-compat.md`](providers/chat-compat.md) | | `src/chat/reasoning-details.ts` | [`data-planes/chat-reasoning-details.md`](data-planes/chat-reasoning-details.md) | | `src/chatgpt/` | [`clients/chatgpt-desktop.md`](clients/chatgpt-desktop.md) | diff --git a/structure/manifest.json b/structure/manifest.json index 33ca7456620..798469a9e1f 100644 --- a/structure/manifest.json +++ b/structure/manifest.json @@ -361,6 +361,8 @@ "scope": "Default-off translated Responses text-only token-limit completion and exclusions.", "documents": [ "src/bridge/max-output-completion.ts", + "src/bridge/sse.ts", + "src/bridge/response-json.ts", "src/config/schema/config-schema.ts", "src/types/config.ts", "src/server/responses/adapter-delivery.ts", diff --git a/structure/transports/max-output-completion.md b/structure/transports/max-output-completion.md index 0207d7c5f83..fd5114dfccf 100644 --- a/structure/transports/max-output-completion.md +++ b/structure/transports/max-output-completion.md @@ -17,7 +17,8 @@ or incomplete shape. Raw `max_tokens`, `length` and `max_output_tokens` stops qu `pause_turn` and context-window exhaustion remain incomplete even though the general truncation classifier maps them to the same public incomplete reason. -The eligible terminal becomes `completed` and omits `incomplete_details`. The model's text +The builders in `src/bridge/sse.ts` and `src/bridge/response-json.ts` turn an eligible +terminal into `completed` and omit `incomplete_details`. The model's text is unchanged, including earlier messages, open Markdown and valid empty fenced blocks. Usage, `end_turn` and provider continuation state survive both buffered and streamed delivery. Compaction retains its existing history and retention-commit rules, without placeholders.