diff --git a/open-sse/services/compression/engines/cavemanAdapter.ts b/open-sse/services/compression/engines/cavemanAdapter.ts index 0fa8e5bb9e..d07e0b0c91 100644 --- a/open-sse/services/compression/engines/cavemanAdapter.ts +++ b/open-sse/services/compression/engines/cavemanAdapter.ts @@ -262,6 +262,10 @@ export const liteEngine: CompressionEngine = { }, apply(body, options) { const adapter = adaptBodyForCompression(body); + // stepConfig is Record, so its compressToolResults is `unknown`. + // Only an explicit boolean counts as a step override — anything else falls through + // to global config.lite, then the default (keeps the type `boolean`, and a malformed + // step value can no longer leak through the `??` chain as `{}`). const stepCompressToolResults = options?.stepConfig?.compressToolResults; const result = applyLiteCompression(adapter.body, { ...options, diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index e79c858b13..11a6f4e779 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -507,6 +507,31 @@ export function buildStreamErrorChunks( return encodeSseEvent(errorEvent, { includeDone: true }); } +/** + * Synthesized terminal frames for a graceful truncation (#7699): the upstream + * ended without a terminal marker AFTER content was already forwarded to the + * client. Instead of an `event: error` frame (which would discard the partial + * content and report a mid-response failure), emit a clean Claude completion — + * `message_delta` carrying `stop_reason: "max_tokens"` followed by + * `message_stop` — so Anthropic SDK / Claude Code treat the response as a + * budget-limited finish and keep everything already received. + */ +export function buildGracefulTruncationChunks(clientResponseFormat?: string | null): Uint8Array[] { + if (clientResponseFormat !== FORMATS.CLAUDE) return []; + + return [ + ...encodeSseEvent( + { + type: "message_delta", + delta: { stop_reason: "max_tokens", stop_sequence: null }, + usage: { input_tokens: 0, output_tokens: 0 }, + }, + { event: "message_delta" } + ), + ...encodeSseEvent({ type: "message_stop" }, { event: "message_stop" }), + ]; +} + /** * Minimal `writable` half used by `pipeWithDisconnect`. The real writable is * driven entirely by the upstream-piped readable, so the writer only needs an @@ -534,10 +559,13 @@ export function createNoopAbortWritable(): { * - **#7699, no terminal marker.** Scoped to Claude (`/v1/messages`), which is * the issue's real scope: Anthropic's SSE spec permits a mid-stream * `event: error`, and Claude clients treat a stream ending without - * `message_stop` as an error. For every other format (plain OpenAI chat - * completions included) a done-without-recognized-marker close is NOT - * necessarily a drop — many formats have no `[DONE]` equivalent — so - * synthesising an error there would be a false positive. + * `message_stop` as an error. When content already reached the client this is + * NOT a provider failure — the partial response is valid and must be kept — so + * it resolves to a graceful truncation (`stop_reason: max_tokens`). For every + * other format (plain OpenAI chat completions included) a + * done-without-recognized-marker close is NOT necessarily a drop — many + * formats have no `[DONE]` equivalent — so synthesising an error there would + * be a false positive. * * - **#8649, no content at all.** The stream terminated properly and carried no * model output. Unlike the marker case this is not format-dependent: a @@ -547,17 +575,26 @@ export function createNoopAbortWritable(): { * emptiness is legitimate (length / tool_calls / content_filter / max_tokens / * tool_use) are excluded by the watcher. */ -function resolveSilentCloseReason(input: { +type SilentCloseOutcome = { kind: "truncated" } | { kind: "error"; reason: string }; + +function resolveSilentCloseOutcome(input: { bytesWereForwarded: boolean; clientTerminalSeen: boolean; clientResponseFormat?: string | null; contentWatcher: StreamContentWatcher; -}): string | null { +}): SilentCloseOutcome | null { if (!input.bytesWereForwarded) return null; if (!input.clientTerminalSeen) { - if (input.clientResponseFormat === FORMATS.CLAUDE) { - return "Upstream stream ended without a terminal marker"; + if ( + input.clientResponseFormat === FORMATS.CLAUDE && + input.contentWatcher.sawContent() + ) { + // #7699 — upstream dropped after content reached the client on a Claude + // stream. Keep the partial response: emit a clean max_tokens completion + // instead of an error frame so Anthropic SDK / Claude Code don't report + // a mid-response break. + return { kind: "truncated" }; } // #10443: every known path that produces OpenAI chat chunks emits a // terminal — the response translators (gemini/claude/kiro/cursor-to-openai) @@ -569,13 +606,13 @@ function resolveSilentCloseReason(input: { // legitimate end. Guard on sawContent() so the #8649 empty-content // verdict below keeps its more precise shape for content-free closes. if (input.clientResponseFormat === FORMATS.OPENAI && input.contentWatcher.sawContent()) { - return "Upstream stream ended without a terminal marker"; + return { kind: "error", reason: "Upstream stream ended without a terminal marker" }; } } const watcher = input.contentWatcher; if (watcher.sawSseFrame() && !watcher.sawContent() && !watcher.sawLegitEmptyTerminal()) { - return "Provider returned empty content"; + return { kind: "error", reason: "Provider returned empty content" }; } return null; @@ -659,20 +696,35 @@ export function createDisconnectAwareStream(transformStream, streamController) { const { done, value } = await reader.read(); if (done) { contentWatcher.finish(); - const silentCloseReason = resolveSilentCloseReason({ + const silentClose = resolveSilentCloseOutcome({ bytesWereForwarded, clientTerminalSeen, clientResponseFormat: streamController.clientResponseFormat, contentWatcher, }); - if (silentCloseReason) { + if (silentClose?.kind === "truncated") { + // #7699 — the upstream dropped without a terminal marker after + // content reached the client. Keep the partial response: emit a + // clean `max_tokens` completion instead of an error frame so + // Anthropic SDK / Claude Code don't report a mid-response break. + streamController.handleComplete(); + try { + for (const chunk of buildGracefulTruncationChunks( + streamController.clientResponseFormat + )) { + controller.enqueue(chunk); + } + } catch { + // downstream may have closed; stream already marked complete + } + } else if (silentClose) { streamController.handleError( - Object.assign(new Error(silentCloseReason), { statusCode: 502 }) + Object.assign(new Error(silentClose.reason), { statusCode: 502 }) ); try { for (const chunk of buildStreamErrorChunks( - silentCloseReason, + silentClose.reason, 502, streamController.clientResponseFormat )) { diff --git a/tests/unit/compression/lite.test.ts b/tests/unit/compression/lite.test.ts index 6decf519c5..74a87307d8 100644 --- a/tests/unit/compression/lite.test.ts +++ b/tests/unit/compression/lite.test.ts @@ -263,6 +263,26 @@ describe("stacked Lite precedence (global config vs explicit step)", () => { assert.ok(!result.stats?.techniquesUsed.includes("tool-compress")); }); + it("non-boolean step compressToolResults falls through to global config", () => { + // stepConfig is Record — a malformed (non-boolean) step value must + // not override; it falls through to global config.lite (false → no truncation). + const result = applyCompression( + { messages: [{ role: "tool", content: toolContent }] }, + "stacked", + { + config: { + ...baseConfig, + lite: { compressToolResults: false }, + stackedPipeline: [{ engine: "lite", config: { compressToolResults: "yes" } }], + }, + } + ); + const messages = result.body.messages as Array<{ content: string }>; + assert.equal(messages[0].content, toolContent.trimEnd()); + assert.doesNotMatch(messages[0].content, /\[truncated\]/); + assert.ok(!result.stats?.techniquesUsed.includes("tool-compress")); + }); + it("stacked default (no lite config) keeps truncation enabled", () => { const result = applyCompression( { messages: [{ role: "tool", content: toolContent }] }, diff --git a/tests/unit/silent-sse-close-7699.test.ts b/tests/unit/silent-sse-close-7699.test.ts index a24a6670d2..e6dd15d32f 100644 --- a/tests/unit/silent-sse-close-7699.test.ts +++ b/tests/unit/silent-sse-close-7699.test.ts @@ -120,7 +120,7 @@ async function drainStream(stream: ReadableStream): Promise ); } -test("#7699 createDisconnectAwareStream emits synthetic error when upstream ends without terminal marker (Claude)", async () => { +test("#7699 createDisconnectAwareStream gracefully truncates (max_tokens) when upstream ends without terminal marker (Claude)", async () => { // Upstream sends some partial content then ends (done=true) without // ever emitting message_stop — reproduces the silent mid-stream close. const upstream = new ReadableStream({ @@ -160,10 +160,52 @@ test("#7699 createDisconnectAwareStream emits synthetic error when upstream ends // Must contain the partial content that was forwarded... assert.match(text, /content_block_delta/); - // ...AND the synthetic terminal frames (error + message_stop) — NOT a silent close. + // ...AND a clean max_tokens completion (message_delta + message_stop) so the + // client keeps the partial response instead of reporting a mid-stream error. + assert.match(text, /event: message_delta\r?\n/); + assert.match(text, /"stop_reason":\s*"max_tokens"/); + assert.match(text, /event: message_stop\r?\n/); + // Graceful truncation must NOT surface an error frame. + assert.doesNotMatch(text, /event: error\r?\n/); + assert.doesNotMatch(text, /"type":\s*"error"/); +}); + +test("#7699 createDisconnectAwareStream still errors when upstream ends with NO content (Claude)", async () => { + // A stream that terminates with an SSE frame but no content is a real failure + // (#8649) and must keep surfacing an error frame — not a fake max_tokens stop. + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"type":"message_start"}\n\n')); + controller.close(); + }, + }); + + const transform = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk); + }, + }); + + const transformedBody = upstream.pipeThrough(transform); + + const sc = createStreamController({ + provider: "test", + model: "test-model", + clientResponseFormat: FORMATS.CLAUDE, + }); + + const wrapped = createDisconnectAwareStream( + { readable: transformedBody, writable: createNoopAbortWritableStream() }, + sc + ); + + const text = await drainStream(wrapped); + assert.match(text, /event: error\r?\n/); assert.match(text, /event: message_stop\r?\n/); - assert.match(text, /Upstream stream ended without a terminal marker/); + assert.match(text, /Provider returned empty content/); + // Empty-content failure must NOT be masked as a max_tokens truncation. + assert.doesNotMatch(text, /"stop_reason":\s*"max_tokens"/); }); // Minimal noop writable for the test wiring (mirrors createNoopAbortWritable).