From 22bf3de083c8df1e8de1689d1362af3b0b9c481b Mon Sep 17 00:00:00 2001 From: thaolaptrinh Date: Tue, 21 Jul 2026 19:40:24 +0700 Subject: [PATCH 1/4] fix(stream): prevent tool-call truncation, premature end, and upstream leaks Multiple bugs caused tool-call responses to be cut off mid-stream or end abruptly with no finish_reason / message_stop. Each was reproduced with a script before fixing and is covered by a regression test. Critical fixes - upstream: preserve NDJSON lines across backpressure. nodeReaderToStream returned out of read() when push() returned false, dropping the rest of the chunk's parsed lines forever. Tool-call deltas often arrive bundled in a single TCP packet, so this was the primary cause of "tool call truncated mid-stream". - stream: synthesize a finish chunk (OpenAI) / message_delta + message_stop (Anthropic) when the upstream closes without a finish event (network drop, upstream crash mid-tool-call). Previously the client saw a clean end-of-stream with no terminal event. - translate: set sawFinish=true on error events. Error events already emit a terminal chunk; without the flag, the server synthesized a second one on stream end (duplicate finish_reason chunks / duplicate message_stop, which the Anthropic SDK rejects). Tool-call correctness - openai: stable per-toolCallId streaming index. tool-call-delta events without an explicit index all defaulted to 0, so parallel tool calls were merged into one. Now allocate via Map and reuse for subsequent deltas + final tool-call. - anthropic: reuse the open tool_use block when a final tool-call arrives for the same toolCallId after deltas, instead of opening a second block (which produced duplicate tool_use blocks). Resource hygiene - upstream: cancel the upstream reader when the consumer stream is destroyed (client disconnect). Without this, CC kept generating tokens nobody read, burning the user's quota until upstream timeout. - server: wire destroyStreamOnClientDisconnect on non-streaming paths too, so client disconnects abort the upstream fetch instead of silently draining into a discarded response. - server: replace the ad-hoc stream.on('data'/'end'/'error') handlers with a pumpStream helper that applies client-side backpressure via pause-on-drain, wraps encoder.emit in try/catch (isolates encoder bugs instead of crashing the process), and consistently checks res.writableEnded. Tests - 4 new upstream tests (backpressure, reader-cancel-on-destroy). - 5 new e2e tests (truncated OpenAI stream, truncated Anthropic stream, upstream error without duplicate finish, duplicate message_stop, stable tool-call indices). - 3 new unit tests for the encoders (sawFinish on error, stable indices, tool-call reuses open block). - All 144 tests pass; tsc --noEmit clean. --- src/server.ts | 130 ++++++++++++++++++++-------- src/translate/anthropic.ts | 41 +++++++++ src/translate/openai.ts | 63 +++++++++++++- src/upstream.ts | 59 ++++++++++--- tests/e2e.test.ts | 137 ++++++++++++++++++++++++++++++ tests/translate-anthropic.test.ts | 37 ++++++++ tests/translate.test.ts | 55 ++++++++++++ tests/upstream.test.ts | 86 +++++++++++++++++++ 8 files changed, 560 insertions(+), 48 deletions(-) diff --git a/src/server.ts b/src/server.ts index ee3c340..35efc2a 100644 --- a/src/server.ts +++ b/src/server.ts @@ -18,7 +18,7 @@ import { validateAnthropicRequest, ValidationError, } from "@/translate/validation.js"; -import type { AnthropicRequest } from "@/translate/anthropic-types.js"; +import type { AnthropicRequest, AnthropicSSERecord } from "@/translate/anthropic-types.js"; // ────────────────────────────────────────── // Mutable server state @@ -176,6 +176,60 @@ function destroyStreamOnClientDisconnect( req.on("close", () => (stream as Readable).destroy()); } +/** + * Write a chunk to `res`, returning a Promise that resolves once the + * underlying socket has drained (when backpressure applies). Returns + * `false` if the response is no longer writable. + */ +function writeSSE(res: http.ServerResponse, chunk: string): Promise { + if (res.writableEnded || res.destroyed) return Promise.resolve(false); + if (res.write(chunk)) return Promise.resolve(true); + return new Promise((resolve) => { + res.once("drain", () => resolve(!res.writableEnded && !res.destroyed)); + }); +} + +/** + * Drive the upstream CC stream through `encoder`, writing formatted SSE + * records to `res`. Applies client-side backpressure (pauses the upstream + * when `res` buffers fill), and isolates encoder errors so they terminate + * the stream cleanly instead of crashing the process. + */ +async function pumpStream( + stream: NodeJS.ReadableStream, + res: http.ServerResponse, + encode: (event: CCEvent) => string[], + onEnd: () => string[], + onError: (err: Error) => string[], +): Promise { + const writable = (chunk: string): Promise => writeSSE(res, chunk); + + try { + for await (const event of stream) { + let chunks: string[]; + try { + chunks = encode(event as unknown as CCEvent); + } catch (err) { + // Encoder blew up — turn it into a stream error so the catch below + // handles it uniformly instead of crashing the proxy. + (stream as Readable).destroy(err as Error); + break; + } + for (const chunk of chunks) { + if (!(await writable(chunk))) return; + } + } + for (const chunk of onEnd()) { + if (!(await writable(chunk))) return; + } + } catch (err) { + logger.error("[stream] upstream streaming error:", (err as Error).message); + for (const chunk of onError(err as Error)) { + if (!(await writable(chunk))) return; + } + } +} + // ────────────────────────────────────────── // Route handlers // ────────────────────────────────────────── @@ -288,25 +342,28 @@ async function handleChatCompletions( ...corsHeaders(), }); - stream.on("data", (event: CCEvent) => { - for (const chunk of encoder.emit(event)) { - res.write(formatSSE(chunk)); - } - }); - stream.on("end", () => { + await pumpStream( + stream, + res, + (event) => encoder.emit(event).map((c) => formatSSE(c)), + () => + encoder.finished + ? [] + : encoder.finishChunks("stop").map((c) => formatSSE(c)), + (err) => { + const chunks: object[] = [encoder.errorChunk(err)]; + if (!encoder.finished) chunks.push(...encoder.finishChunks("stop")); + return chunks.map((c) => formatSSE(c)); + }, + ); + // After pump completes, emit the [DONE] sentinel if we still can. + if (!res.writableEnded) { res.write(formatSSEDone()); res.end(); - }); - stream.on("error", (err: Error) => { - logger.error("[stream] OpenAI streaming error:", err.message); - if (!res.destroyed) { - res.write(formatSSE(encoder.errorChunk(err))); - res.write(formatSSEDone()); - res.end(); - } - }); + } destroyStreamOnClientDisconnect(req, stream); } else { + destroyStreamOnClientDisconnect(req, stream); const events = await collectEvents(stream); const response = buildNonStreamingResponse(events, model, encoder.id); sendJson(res, 200, response); @@ -370,26 +427,31 @@ async function handleMessages(req: http.IncomingMessage, res: http.ServerRespons ...corsHeaders(), }); - stream.on("data", (event: CCEvent) => { - for (const record of encoder.emit(event)) { - res.write(formatAnthropicSSE(record.event, record.data)); - } - }); - stream.on("end", () => res.end()); - stream.on("error", (err: Error) => { - logger.error("[stream] Anthropic streaming error:", err.message); - if (!res.destroyed) { - res.write( - formatAnthropicSSE("error", { - type: "error", - error: { type: "api_error", message: err.message }, - }), - ); - res.end(); - } - }); + await pumpStream( + stream, + res, + (event) => encoder.emit(event).map((r) => formatAnthropicSSE(r.event, r.data)), + () => + encoder.finished + ? [] + : encoder + .finishRecords("end_turn") + .map((r) => formatAnthropicSSE(r.event, r.data)), + (err) => { + const records: AnthropicSSERecord[] = [ + { + event: "error", + data: { type: "error", error: { type: "api_error", message: err.message } }, + }, + ]; + if (!encoder.finished) records.push(...encoder.finishRecords("end_turn")); + return records.map((r) => formatAnthropicSSE(r.event, r.data)); + }, + ); + if (!res.writableEnded) res.end(); destroyStreamOnClientDisconnect(req, stream); } else { + destroyStreamOnClientDisconnect(req, stream); const events = await collectEvents(stream); const response = buildAnthropicResponse(events, model, encoder.messageId); res.writeHead(200, { "Content-Type": "application/json", ...corsHeaders() }); diff --git a/src/translate/anthropic.ts b/src/translate/anthropic.ts index 11be46d..aaf83db 100644 --- a/src/translate/anthropic.ts +++ b/src/translate/anthropic.ts @@ -249,11 +249,16 @@ export class AnthropicStreamEncoder { private pendingStart: CCEvent | null = null; private started = false; private pinged = false; + private sawFinish = false; constructor(private readonly model: string) { this.messageId = `msg_${crypto.randomUUID()}`; } + get finished(): boolean { + return this.sawFinish; + } + emit(event: CCEvent): AnthropicSSERecord[] { if (event.type === "start") { this.blockIndex = 0; @@ -263,6 +268,7 @@ export class AnthropicStreamEncoder { } if (event.type === "error") { + this.sawFinish = true; const msg = (event.data.message as string) ?? (event.data.error as { message?: string } | undefined)?.message ?? @@ -308,6 +314,7 @@ export class AnthropicStreamEncoder { } private handleFinish(event: CCEvent): AnthropicSSERecord[] { + this.sawFinish = true; const records: AnthropicSSERecord[] = []; this.closeCurrentBlock(records); @@ -389,6 +396,15 @@ export class AnthropicStreamEncoder { const input = event.data.input ?? event.data.arguments; const argsStr = typeof input === "string" ? input : input != null ? JSON.stringify(input) : ""; + // If the upstream streamed deltas for this tool call first and then + // sent the final `tool-call` event with the same id, reuse the block + // it already opened instead of creating a duplicate `tool_use`. + if (this.currentBlockType === "tool_use" && this.currentToolCallId === tcId) { + if (argsStr) { + records.push(this.makeDelta({ type: "input_json_delta", partial_json: argsStr })); + } + break; + } this.closeCurrentBlock(records); this.ensureBlockOpenWith(records, "tool_use", { type: "tool_use", @@ -489,6 +505,31 @@ export class AnthropicStreamEncoder { }, }; } + + /** + * Build the closing records for a stream that ended without a `finish` + * event from the upstream (e.g. upstream connection dropped mid-response). + * Emits synthetic message_delta + message_stop so the Anthropic SDK sees + * a well-formed end-of-stream instead of a truncated response. + */ + finishRecords(stopReason: AnthropicStopReason = "end_turn"): AnthropicSSERecord[] { + const records: AnthropicSSERecord[] = []; + if (!this.started) { + records.push(this.makeMessageStart(0)); + this.started = true; + } + this.closeCurrentBlock(records); + records.push({ + event: "message_delta", + data: { + type: "message_delta", + delta: { stop_reason: stopReason, stop_sequence: null }, + usage: { input_tokens: 0, output_tokens: 0 }, + }, + }); + records.push({ event: "message_stop", data: {} }); + return records; + } } // ── Non-streaming response builder ── diff --git a/src/translate/openai.ts b/src/translate/openai.ts index 5d2d720..6fec7f8 100644 --- a/src/translate/openai.ts +++ b/src/translate/openai.ts @@ -221,12 +221,38 @@ export class OpenAIStreamEncoder { readonly id: string; private readonly created: number; private toolCallIndex = 0; + private sawFinish = false; + // Map a CC toolCallId to the OpenAI streaming `index` we assigned it. + // Without this, parallel tool-call-delta streams without an explicit + // `index` field would all be assigned index 0 and the client would merge + // them into a single tool call. + private readonly toolCallIdToIndex = new Map(); constructor(private readonly model: string) { this.id = crypto.randomUUID(); this.created = Math.floor(Date.now() / 1000); } + get finished(): boolean { + return this.sawFinish; + } + + /** + * Resolve the OpenAI streaming `index` for a given tool-call id, allocating + * a new one on first sighting. Falls back to the upstream-provided index + * (if any) so we don't fight a bridge that already numbers them. + */ + private resolveToolCallIndex(toolCallId: string | undefined, upstreamIndex: number | undefined): number { + if (toolCallId) { + const known = this.toolCallIdToIndex.get(toolCallId); + if (known !== undefined) return known; + const idx = upstreamIndex ?? this.toolCallIndex++; + this.toolCallIdToIndex.set(toolCallId, idx); + return idx; + } + return upstreamIndex ?? this.toolCallIndex++; + } + emit(event: CCEvent): object[] { const chunks: object[] = []; const id = this.id; @@ -274,17 +300,19 @@ export class OpenAIStreamEncoder { } case "tool-call-delta": { + const toolCallId = (event.data.toolCallId as string) ?? undefined; + const upstreamIndex = (event.data.index as number) ?? undefined; const tc: { index: number; id?: string; type?: string; function: { name?: string; arguments: string }; } = { - index: (event.data.index as number) ?? 0, + index: this.resolveToolCallIndex(toolCallId, upstreamIndex), function: { arguments: (event.data.arguments as string) ?? "" }, }; - if (event.data.toolCallId) { - tc.id = event.data.toolCallId as string; + if (toolCallId) { + tc.id = toolCallId; tc.type = "function"; } if (event.data.name) { @@ -305,6 +333,13 @@ export class OpenAIStreamEncoder { const toolName = (event.data.toolName as string) ?? (event.data.name as string) ?? ""; const input = event.data.input ?? event.data.arguments; const args = typeof input === "string" ? input : input != null ? JSON.stringify(input) : ""; + // Reuse the index allocated by any prior tool-call-delta for this id + // so the client merges the final tool-call with the streamed deltas + // instead of seeing it as a brand-new tool call. + const index = this.resolveToolCallIndex( + toolCallId || undefined, + (event.data.index as number) ?? undefined, + ); chunks.push({ id, object: "chat.completion.chunk", @@ -316,7 +351,7 @@ export class OpenAIStreamEncoder { delta: { tool_calls: [ { - index: this.toolCallIndex++, + index, id: toolCallId, type: "function", function: { name: toolName, arguments: args }, @@ -331,6 +366,7 @@ export class OpenAIStreamEncoder { } case "finish": { + this.sawFinish = true; const usage = extractUsage(event.data as Record); const finishReason = (event.data.finishReason as string) ?? "stop"; chunks.push({ @@ -356,6 +392,7 @@ export class OpenAIStreamEncoder { } case "error": { + this.sawFinish = true; const errMsg = (event.data.message as string) ?? (event.data.error as { message?: string } | undefined)?.message ?? @@ -392,6 +429,24 @@ export class OpenAIStreamEncoder { }, }; } + + /** + * Build the closing chunks for a stream that ended without a `finish` + * event from the upstream (e.g. upstream connection dropped mid-response). + * Emits a synthetic finish_reason so the client SDK doesn't see a + * truncated response. + */ + finishChunks(reason: string = "stop"): object[] { + return [ + { + id: this.id, + object: "chat.completion.chunk", + created: this.created, + model: this.model, + choices: [{ index: 0, delta: {}, finish_reason: reason }], + }, + ]; + } } // ────────────────────────────────────────── diff --git a/src/upstream.ts b/src/upstream.ts index 129893b..c596d38 100644 --- a/src/upstream.ts +++ b/src/upstream.ts @@ -216,19 +216,63 @@ function nodeReaderToStream( ): NodeJS.ReadableStream { const decoder = new TextDecoder(); let buffer = ""; + // Lines parsed from the current upstream chunk that haven't been pushed yet. + // Kept in closure scope so backpressure mid-chunk doesn't drop them: when + // push() returns false we return out of read(), and resume here on the next + // read() call instead of starting a fresh reader.read(). + let pendingLines: string[] = []; + let upstreamDone = false; + let readerReleased = false; + + // Release the underlying reader when the consumer destroys this stream + // (e.g. client disconnected). Otherwise CC keeps generating tokens nobody + // will read, burning the user's quota until upstream's own timeout fires. + const releaseReader = (): void => { + if (readerReleased) return; + readerReleased = true; + const cancel = (reader as { cancel?: () => Promise }).cancel; + if (typeof cancel === "function") { + cancel.call(reader).catch(() => { + /* already closed */ + }); + } + }; return new Readable({ objectMode: true, + emitClose: true, + destroy(err, cb) { + releaseReader(); + cb(err); + }, async read() { try { while (true) { + // Drain anything left over from a previous chunk that was + // interrupted by backpressure before we read more from upstream. + while (pendingLines.length > 0) { + const line = pendingLines.shift() as string; + const result = parseCCLine(line); + if (result.type === "event" && result.event) { + if (!this.push(result.event)) return; // still backpressured + } + } + + if (upstreamDone) { + this.push(null); + return; + } + const { done, value } = await reader.read(); if (done) { - // Flush remaining buffer + upstreamDone = true; + releaseReader(); + // Flush trailing partial line (no newline terminator). if (buffer.trim()) { const result = parseCCLine(buffer); + buffer = ""; if (result.type === "event" && result.event) { - this.push(result.event); + if (!this.push(result.event)) return; // backpressured; null next read } } this.push(null); @@ -237,17 +281,12 @@ function nodeReaderToStream( buffer += decoder.decode(value, { stream: true }); const lines = buffer.split("\n"); + // Last segment is the partial line awaiting its newline; keep it. buffer = lines.pop() ?? ""; - - for (const line of lines) { - const result = parseCCLine(line); - if (result.type === "event" && result.event) { - const canContinue = this.push(result.event); - if (!canContinue) return; - } - } + pendingLines = lines; } } catch (err) { + releaseReader(); this.destroy(err as Error); } }, diff --git a/tests/e2e.test.ts b/tests/e2e.test.ts index 9fc67c1..e6faef3 100644 --- a/tests/e2e.test.ts +++ b/tests/e2e.test.ts @@ -137,6 +137,43 @@ describe("E2E: OpenAI /v1/chat/completions", () => { expect(finishChunk.choices[0].finish_reason).toBe("stop"); }); + // Regression: when the upstream stream ends without a `finish` event + // (network drop, CC restart mid-tool-call), the proxy must synthesize a + // finish_reason chunk so the OpenAI SDK sees a well-formed end-of-stream + // instead of an abruptly truncated response. + it("streaming: synthesizes a finish chunk when upstream ends without finish", async () => { + const truncatedEvents: CCEvent[] = [ + { type: "start", data: {} }, + { type: "text-delta", data: { text: "partial response" } }, + // NO finish event — simulates upstream connection drop + ]; + sendToCCSpy.mockResolvedValue({ stream: Readable.from(truncatedEvents) }); + + const res = await fetch(`${baseUrl}/v1/chat/completions`, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer test-key", + }, + body: JSON.stringify({ + model: "deepseek/deepseek-v4-flash", + messages: [{ role: "user", content: "Hi" }], + stream: true, + }), + }); + + expect(res.status).toBe(200); + const text = await res.text(); + // Even with no upstream finish, the proxy must close the SSE stream + // cleanly with a synthesized finish_reason and the [DONE] sentinel. + expect(text).toContain("[DONE]"); + const lines = text.split("\n").filter((l) => l.startsWith("data: ") && !l.includes("[DONE]")); + const parsed = lines.map((l) => JSON.parse(l.replace("data: ", ""))); + const finishChunk = parsed.find((c) => c.choices?.[0]?.finish_reason); + expect(finishChunk).toBeDefined(); + expect(finishChunk.choices[0].finish_reason).toBe("stop"); + }); + it("returns 400 for invalid JSON body", async () => { const res = await fetch(`${baseUrl}/v1/chat/completions`, { method: "POST", @@ -150,6 +187,40 @@ describe("E2E: OpenAI /v1/chat/completions", () => { const body = (await res.json()) as any; expect(body.error.message).toBe("Invalid JSON body"); }); + + // Regression: an upstream `error` event already emits a terminal chunk with + // finish_reason:"stop". The server MUST NOT synthesize a second finish + // chunk on stream end (which would emit duplicate finish_reason chunks). + it("streaming: upstream error event does not produce duplicate finish chunks", async () => { + const errorEvents: CCEvent[] = [ + { type: "start", data: {} }, + { type: "error", data: { message: "CC upstream exploded" } }, + // No finish event; stream just ends after the error. + ]; + sendToCCSpy.mockResolvedValue({ stream: Readable.from(errorEvents) }); + + const res = await fetch(`${baseUrl}/v1/chat/completions`, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer test-key", + }, + body: JSON.stringify({ + model: "deepseek/deepseek-v4-flash", + messages: [{ role: "user", content: "Hi" }], + stream: true, + }), + }); + + expect(res.status).toBe(200); + const text = await res.text(); + expect(text).toContain("[DONE]"); + const lines = text.split("\n").filter((l) => l.startsWith("data: ") && !l.includes("[DONE]")); + const parsed = lines.map((l) => JSON.parse(l.replace("data: ", ""))); + // Exactly ONE chunk with finish_reason set (from the error case). + const finishChunks = parsed.filter((c) => c.choices?.[0]?.finish_reason); + expect(finishChunks).toHaveLength(1); + }); }); // ────────────────────────────────────────── @@ -307,6 +378,72 @@ describe("E2E: Anthropic /v1/messages", () => { expect(text).toContain("event: message_stop"); }); + // Regression: when the upstream stream ends without a `finish` event + // (network drop, CC restart mid-tool-call), the proxy must synthesize the + // closing message_delta + message_stop so the Anthropic SDK sees a + // well-formed end-of-stream instead of a truncated response. + it("streaming: synthesizes message_stop when upstream ends without finish", async () => { + const truncatedEvents: CCEvent[] = [ + { type: "start", data: {} }, + { type: "text-delta", data: { text: "partial" } }, + // NO finish event — simulates upstream connection drop + ]; + sendToCCSpy.mockResolvedValue({ stream: Readable.from(truncatedEvents) }); + + const res = await fetch(`${baseUrl}/v1/messages`, { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-api-key": "test-key", + }, + body: JSON.stringify({ + model: "claude-sonnet-4-5-20250929", + max_tokens: 100, + messages: [{ role: "user", content: "Hi" }], + stream: true, + }), + }); + + expect(res.status).toBe(200); + const text = await res.text(); + // Even with no upstream finish, the proxy must close the SSE stream + // cleanly with a synthesized message_delta + message_stop. + expect(text).toContain("event: message_start"); + expect(text).toContain("event: message_delta"); + expect(text).toContain("event: message_stop"); + expect(text).toContain('"stop_reason"'); + }); + + // Regression: an upstream `error` event already emits message_stop. The + // server MUST NOT synthesize a second one (Anthropic SDK throws on + // duplicate message_stop events). + it("streaming: upstream error event does not produce duplicate message_stop", async () => { + const errorEvents: CCEvent[] = [ + { type: "start", data: {} }, + { type: "error", data: { message: "CC upstream exploded" } }, + ]; + sendToCCSpy.mockResolvedValue({ stream: Readable.from(errorEvents) }); + + const res = await fetch(`${baseUrl}/v1/messages`, { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-api-key": "test-key", + }, + body: JSON.stringify({ + model: "claude-sonnet-4-5-20250929", + max_tokens: 100, + messages: [{ role: "user", content: "Hi" }], + stream: true, + }), + }); + + expect(res.status).toBe(200); + const text = await res.text(); + const stopCount = (text.match(/event: message_stop/g) ?? []).length; + expect(stopCount).toBe(1); + }); + it("returns 400 for invalid JSON body with Anthropic error shape", async () => { const res = await fetch(`${baseUrl}/v1/messages`, { method: "POST", diff --git a/tests/translate-anthropic.test.ts b/tests/translate-anthropic.test.ts index 7265076..0685872 100644 --- a/tests/translate-anthropic.test.ts +++ b/tests/translate-anthropic.test.ts @@ -404,6 +404,43 @@ describe("AnthropicStreamEncoder", () => { ); expect(events).toContain("content_block_stop"); }); + + // Regression: error events must set `finished` so the server doesn't + // synthesize a duplicate message_stop on stream end (Anthropic SDK + // throws on a double message_stop). + it("marks the encoder as finished after an error event", () => { + const encoder = new AnthropicStreamEncoder("m"); + expect(encoder.finished).toBe(false); + encoder.emit({ type: "start", data: {} }); + encoder.emit({ type: "error", data: { message: "boom" } }); + expect(encoder.finished).toBe(true); + }); + + // Regression: when CC sends tool-call-delta events followed by a final + // tool-call for the same toolCallId, the encoder must reuse the already- + // open tool_use block instead of opening a second one (which would + // produce duplicate tool_use blocks for the same id). + it("tool-call after tool-call-delta for the same id reuses the open block", () => { + const encoder = new AnthropicStreamEncoder("m"); + encoder.emit({ type: "start", data: {} }); + + // Stream partial args via deltas. + const d1 = encoder.emit({ + type: "tool-call-delta", + data: { toolCallId: "tc_X", name: "search", arguments: '{"q":' }, + }); + // Final tool-call for the same id. + const d2 = encoder.emit({ + type: "tool-call", + data: { toolCallId: "tc_X", toolName: "search", input: { q: "hi" } }, + }); + + const allRecords = [...d1, ...d2]; + const blockStarts = allRecords.filter((r) => r.event === "content_block_start"); + // Exactly one tool_use block — not two. + expect(blockStarts).toHaveLength(1); + expect((blockStarts[0].data.content_block as { type: string }).type).toBe("tool_use"); + }); }); describe("buildAnthropicResponse", () => { diff --git a/tests/translate.test.ts b/tests/translate.test.ts index 2296f0b..e879af3 100644 --- a/tests/translate.test.ts +++ b/tests/translate.test.ts @@ -338,6 +338,61 @@ describe("OpenAIStreamEncoder", () => { expect((chunks[0] as any).choices[0].finish_reason).toBe(to); } }); + + // Regression: error events emit a finish_reason chunk but must also flip + // `finished` to true, otherwise the server would synthesize ANOTHER + // finish chunk on stream end and the client would see a duplicate. + it("marks the encoder as finished after an error event", () => { + expect(encoder.finished).toBe(false); + encoder.emit({ type: "error", data: { message: "boom" } }); + expect(encoder.finished).toBe(true); + }); + + // Regression: parallel tool-call-delta streams without an explicit `index` + // would all be assigned index 0 and the client would merge them into a + // single tool call. Each toolCallId must get its own stable index. + it("assigns stable tool-call indices per toolCallId", () => { + encoder.emit({ type: "start", data: {} }); + + // First tool, two deltas. + const a1 = encoder.emit({ + type: "tool-call-delta", + data: { toolCallId: "call_A", name: "search", arguments: "{" }, + }); + const a2 = encoder.emit({ + type: "tool-call-delta", + data: { toolCallId: "call_A", arguments: '"q":"hi"}' }, + }); + // Second tool, parallel. + const b1 = encoder.emit({ + type: "tool-call-delta", + data: { toolCallId: "call_B", name: "calc", arguments: "{}" }, + }); + + const idxA1 = (a1[0] as any).choices[0].delta.tool_calls[0].index; + const idxA2 = (a2[0] as any).choices[0].delta.tool_calls[0].index; + const idxB1 = (b1[0] as any).choices[0].delta.tool_calls[0].index; + + expect(idxA1).toBe(0); + expect(idxA2).toBe(0); // same id → same index + expect(idxB1).toBe(1); // different id → new index + }); + + // Regression: a final `tool-call` event after deltas for the same id must + // reuse the index allocated by the deltas so the client merges them. + it("tool-call after tool-call-delta for the same id reuses the index", () => { + encoder.emit({ type: "start", data: {} }); + encoder.emit({ + type: "tool-call-delta", + data: { toolCallId: "call_X", name: "search", arguments: '{"q":"a"' }, + }); + const finalChunks = encoder.emit({ + type: "tool-call", + data: { toolCallId: "call_X", toolName: "search", input: '{"q":"a"}' }, + }); + const idx = (finalChunks[0] as any).choices[0].delta.tool_calls[0].index; + expect(idx).toBe(0); + }); }); // ────────────────────────────────────────── diff --git a/tests/upstream.test.ts b/tests/upstream.test.ts index e2e02fc..2bad866 100644 --- a/tests/upstream.test.ts +++ b/tests/upstream.test.ts @@ -164,4 +164,90 @@ describe("sendToCC retry", () => { // 1 initial + 2 retries = 3 attempts. expect(mock).toHaveBeenCalledTimes(3); }); + + // Regression: when a single upstream chunk carries many NDJSON lines and the + // consumer applies backpressure (push() returns false), the proxy used to + // drop all remaining lines in that chunk. Tool-call deltas typically arrive + // bundled this way, so the symptom was "tool calls truncated mid-stream". + it("does not drop events when the consumer applies backpressure", async () => { + // 50 events packed into ONE upstream chunk. + const bodyLines: string[] = []; + for (let i = 0; i < 50; i++) { + bodyLines.push(JSON.stringify({ type: "text-delta", data: { text: `t${i}` } }) + "\n"); + } + bodyLines.push(JSON.stringify({ type: "finish", data: { finishReason: "stop" } }) + "\n"); + + const mock = vi.fn().mockResolvedValue( + fakeResponse({ ok: true, status: 200, bodyLines }), + ); + globalThis.fetch = mock as unknown as typeof fetch; + + const { stream } = await sendToCC(sampleBody(), { + apiBase: "https://example.test", + apiKey: "k", + ccVersion: "0.0.0", + }); + + // Consume slowly to force the Readable's internal buffer to fill. + // pause()/resume() gives us deterministic backpressure. + const events: CCEvent[] = []; + stream.on("data", (e: CCEvent) => events.push(e)); + (stream as Readable).pause(); + + // Allow a tick for the producer to attempt stuffing the internal buffer. + await new Promise((r) => setImmediate(r)); + (stream as Readable).resume(); + await new Promise((resolve) => stream.on("end", () => resolve())); + + const textDeltas = events.filter((e) => e.type === "text-delta"); + expect(textDeltas.length).toBe(50); + // First and last deltas specifically — these are the ones most likely to + // be lost at the boundary. + expect(textDeltas[0].data.text).toBe("t0"); + expect(textDeltas[49].data.text).toBe("t49"); + expect(events.some((e) => e.type === "finish")).toBe(true); + }); + + // Regression: when the consumer stream is destroyed (client disconnect), + // the underlying upstream reader must be cancelled so the upstream TCP + // connection closes and CC stops generating tokens nobody will read. + it("cancels the upstream reader when the consumer is destroyed", async () => { + // Reader fake that supports cancel() and produces an unbounded stream. + const enc = new TextEncoder(); + let cancelCalled = false; + const reader = { + read: async () => ({ + done: false, + value: enc.encode('{"type":"text-delta","data":{"text":"x"}}\n'), + }), + cancel: async () => { + cancelCalled = true; + }, + }; + const mock = vi.fn().mockResolvedValue({ + ok: true, + status: 200, + statusText: "", + text: async () => "", + body: { getReader: () => reader }, + } as unknown as Response); + globalThis.fetch = mock as unknown as typeof fetch; + + const { stream } = await sendToCC(sampleBody(), { + apiBase: "https://example.test", + apiKey: "k", + ccVersion: "0.0.0", + }); + + // Read one event so the producer is up, then destroy (client disconnect). + await new Promise((resolve) => { + stream.once("data", () => { + (stream as Readable).destroy(); + resolve(); + }); + }); + + // cancel() is called synchronously from destroy(). + expect(cancelCalled).toBe(true); + }); }); From 6727c84ce2cc355d2cadd35d2e9b91f88456af2f Mon Sep 17 00:00:00 2001 From: thaolaptrinh Date: Tue, 21 Jul 2026 19:59:23 +0700 Subject: [PATCH 2/4] fix(upstream): add streaming idle timeout + make timeouts configurable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous 5-minute timeout only protected the time-to-first-byte phase: clearTimeout() was called the moment fetch() resolved (response headers), leaving the entire streaming phase with no protection at all. Two new failure modes resulted: - A stalled upstream (TCP open, no chunks arriving mid-tool-call) would hang the consumer forever. The client's own timeout would eventually fire, but the proxy kept the upstream connection open and the request slot occupied until CC's own server-side timeout (if any) released it. - The 5-minute connect timeout was hardcoded — slow reasoning models with long initial processing would be killed at the boundary with no way for the operator to bump it. Fixes - config: add CC_UPSTREAM_TIMEOUT_MS (default 600000 / 10 min, up from 5 min) and CC_IDLE_TIMEOUT_MS (default 120000 / 2 min, 0 disables). Both parse-positive-int guarded. - upstream: nodeReaderToStream now arms an idle timer before each reader.read() and disarms it on chunk arrival. If no data arrives within idleTimeoutMs, the reader is cancelled with an IdleTimeoutError that propagates through the stream's error path (which the existing pumpStream turns into a clean finish for the client). The timer is unref'd so it never keeps the event loop alive. - upstream: caller's AbortSignal is now plumbed into nodeReaderToStream so a client disconnect during a stalled read immediately destroys the stream instead of waiting for the idle timer. - server: both /v1/chat/completions and /v1/messages pass the new timeout options through. Docs - README + .env.example document the two new env vars. Tests - upstream: idle timeout fires after the configured interval and surfaces an IdleTimeoutError; idleTimeoutMs=0 disables it. - config: defaults, env-var override, 0-disabled, and invalid-fallback cases. - All 149 tests pass; tsc --noEmit clean. --- .env.example | 11 +++++ README.md | 20 +++++---- src/config.ts | 35 +++++++++++++++- src/server.ts | 10 ++++- src/upstream.ts | 74 +++++++++++++++++++++++++++++++-- tests/config.test.ts | 37 +++++++++++++++++ tests/upstream.test.ts | 94 ++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 267 insertions(+), 14 deletions(-) diff --git a/.env.example b/.env.example index fff5d79..4cf41cf 100644 --- a/.env.example +++ b/.env.example @@ -5,5 +5,16 @@ CC_API_KEY= HOST=127.0.0.1 PORT=8787 +# Upstream timeouts (milliseconds) +# Max wall-clock time for CC to return response headers + first byte. +# Bump this for slow reasoning models that take a while to start streaming. +# CC_UPSTREAM_TIMEOUT_MS=600000 # default: 10 minutes + +# Max gap between consecutive data chunks during streaming. If CC opens the +# connection but stops sending data mid-response (stalled tool call, dropped +# TCP, etc.), the proxy aborts the stream and the client gets a clean +# error/finish instead of hanging forever. Set to 0 to disable. +# CC_IDLE_TIMEOUT_MS=120000 # default: 2 minutes + # Anthropic model mapping (optional — prefer --setup-claude-code) # ANTHROPIC_DEFAULT_MODEL=deepseek/deepseek-v4-pro diff --git a/README.md b/README.md index 177186c..f3ffb27 100644 --- a/README.md +++ b/README.md @@ -73,15 +73,17 @@ commandcode-api-proxy auth logout Equivalent env vars (lower priority than CLI flags): -| Env var | Description | -| ---------------- | ----------------------------------------------------------------------------------------------------------------------- | -| `HOST` | Bind address | -| `PORT` | Port | -| `CC_API_KEY` | Command Code API key | -| `CC_API_BASE` | Upstream API base URL | -| `CC_CLI_VERSION` | CLI version sent upstream | -| `LOG_LEVEL` | Log level (`info`, `debug`, etc.) | -| `CORS_ORIGIN` | `Access-Control-Allow-Origin` value. `*` by default; empty string disables CORS. Restrict before exposing on a network. | +| Env var | Description | +| ------------------------ | ----------------------------------------------------------------------------------------------------------------------- | +| `HOST` | Bind address | +| `PORT` | Port | +| `CC_API_KEY` | Command Code API key | +| `CC_API_BASE` | Upstream API base URL | +| `CC_CLI_VERSION` | CLI version sent upstream | +| `CC_UPSTREAM_TIMEOUT_MS` | Max ms for upstream to return response headers + first byte (default `600000` / 10 min). Bump for slow reasoning models | +| `CC_IDLE_TIMEOUT_MS` | Max ms between consecutive stream chunks (default `120000` / 2 min). `0` disables — detects stalled upstreams | +| `LOG_LEVEL` | Log level (`info`, `debug`, etc.) | +| `CORS_ORIGIN` | `Access-Control-Allow-Origin` value. `*` by default; empty string disables CORS. Restrict before exposing on a network. | > **Security:** the proxy forwards your paid Command Code key upstream and > accepts any auth token from clients (`proxy-managed`), so it is designed for diff --git a/src/config.ts b/src/config.ts index bdbf8b5..fb74cda 100644 --- a/src/config.ts +++ b/src/config.ts @@ -15,6 +15,10 @@ export interface Config { ccVersion: string; logLevel: string; corsOrigin: string; + /** Max wall-clock ms for upstream to send response headers + first byte. */ + upstreamTimeoutMs: number; + /** Max ms between consecutive chunks during streaming. 0 = disabled. */ + idleTimeoutMs: number; } /** @@ -86,5 +90,34 @@ export function loadConfig(): Config { // empty to disable) before exposing the proxy on a network. const corsOrigin = process.env.CORS_ORIGIN ?? "*"; - return { host, port, apiKey, ccApiBase, ccVersion, logLevel, corsOrigin }; + // Upstream timeouts. The connection timeout covers the wall-clock time + // until the upstream returns response headers + first byte — bump it for + // slow reasoning models. The idle timeout catches stalled streams where + // the upstream opened the connection but stopped sending chunks + // mid-response (e.g. tool call hung on the upstream side). Set + // CC_IDLE_TIMEOUT_MS=0 to disable idle detection entirely. + const upstreamTimeoutMs = parsePositiveInt( + process.env.CC_UPSTREAM_TIMEOUT_MS, + 600_000, // 10 minutes — covers high-effort reasoning models + ); + const idleTimeoutMs = parsePositiveInt(process.env.CC_IDLE_TIMEOUT_MS, 120_000); // 2 minutes + + return { + host, + port, + apiKey, + ccApiBase, + ccVersion, + logLevel, + corsOrigin, + upstreamTimeoutMs, + idleTimeoutMs, + }; +} + +function parsePositiveInt(raw: string | undefined, fallback: number): number { + if (raw == null || raw === "") return fallback; + const n = Number(raw); + if (!Number.isFinite(n) || n < 0) return fallback; + return Math.floor(n); } diff --git a/src/server.ts b/src/server.ts index 35efc2a..c2a4bb6 100644 --- a/src/server.ts +++ b/src/server.ts @@ -329,6 +329,8 @@ async function handleChatCompletions( apiBase: config.ccApiBase, apiKey, ccVersion: config.ccVersion, + timeoutMs: config.upstreamTimeoutMs, + idleTimeoutMs: config.idleTimeoutMs, }, abort.signal, ); @@ -414,7 +416,13 @@ async function handleMessages(req: http.IncomingMessage, res: http.ServerRespons try { const result = await sendToCC( ccBody, - { apiBase: config.ccApiBase, apiKey, ccVersion: config.ccVersion }, + { + apiBase: config.ccApiBase, + apiKey, + ccVersion: config.ccVersion, + timeoutMs: config.upstreamTimeoutMs, + idleTimeoutMs: config.idleTimeoutMs, + }, abort.signal, ); const stream = result.stream; diff --git a/src/upstream.ts b/src/upstream.ts index c596d38..ec8f02c 100644 --- a/src/upstream.ts +++ b/src/upstream.ts @@ -8,7 +8,10 @@ interface UpstreamOptions { apiBase: string; apiKey: string; ccVersion: string; + /** Wall-clock timeout for receiving response headers + first byte. */ timeoutMs?: number; + /** Max ms allowed between consecutive data chunks during streaming. */ + idleTimeoutMs?: number; } /** @@ -94,7 +97,13 @@ export async function sendToCC( options: UpstreamOptions, signal?: AbortSignal, ): Promise<{ stream: NodeJS.ReadableStream }> { - const { apiBase, apiKey, ccVersion, timeoutMs = 300_000 } = options; + const { + apiBase, + apiKey, + ccVersion, + timeoutMs = 600_000, + idleTimeoutMs = 120_000, + } = options; const url = `${apiBase}/alpha/generate`; // CC's API is always streaming — force it on so the upstream stays a stream. @@ -137,7 +146,12 @@ export async function sendToCC( throw new UpstreamError("CC API returned no body", 0, true); } - return { stream: nodeReaderToStream(response.body.getReader()) }; + return { + stream: nodeReaderToStream(response.body.getReader(), { + idleTimeoutMs, + abortSignal: signal, + }), + }; } catch (err) { clearTimeout(timeout); if (err instanceof UpstreamError) throw err; @@ -213,6 +227,7 @@ function combineSignals(...signals: AbortSignal[]): AbortSignal { function nodeReaderToStream( reader: ReadableStreamDefaultReader, + opts: { idleTimeoutMs?: number; abortSignal?: AbortSignal } = {}, ): NodeJS.ReadableStream { const decoder = new TextDecoder(); let buffer = ""; @@ -224,10 +239,41 @@ function nodeReaderToStream( let upstreamDone = false; let readerReleased = false; + // Idle timeout: detect a stalled upstream (TCP open, no chunks arriving). + // Reset on every successful read(). If it fires we abort the reader so + // pumpStream's error path synthesizes a clean finish for the client + // instead of hanging forever waiting on a dead connection. + const idleMs = opts.idleTimeoutMs ?? 0; + let idleTimer: NodeJS.Timeout | null = null; + const armIdle = (): void => { + if (idleMs <= 0) return; + disarmIdle(); + idleTimer = setTimeout(() => { + const err = new Error( + `CC upstream idle timeout: no data for ${idleMs}ms`, + ); + err.name = "IdleTimeoutError"; + // Cancel the reader — pending read() will reject with this reason. + const cancel = (reader as { cancel?: (reason?: unknown) => Promise }).cancel; + if (typeof cancel === "function") { + cancel.call(reader, err).catch(() => {}); + } + }, idleMs); + // Don't keep the event loop alive just for the idle timer. + idleTimer.unref?.(); + }; + const disarmIdle = (): void => { + if (idleTimer) { + clearTimeout(idleTimer); + idleTimer = null; + } + }; + // Release the underlying reader when the consumer destroys this stream // (e.g. client disconnected). Otherwise CC keeps generating tokens nobody // will read, burning the user's quota until upstream's own timeout fires. const releaseReader = (): void => { + disarmIdle(); if (readerReleased) return; readerReleased = true; const cancel = (reader as { cancel?: () => Promise }).cancel; @@ -238,7 +284,7 @@ function nodeReaderToStream( } }; - return new Readable({ + const stream = new Readable({ objectMode: true, emitClose: true, destroy(err, cb) { @@ -263,7 +309,9 @@ function nodeReaderToStream( return; } + armIdle(); const { done, value } = await reader.read(); + disarmIdle(); if (done) { upstreamDone = true; releaseReader(); @@ -291,4 +339,24 @@ function nodeReaderToStream( } }, }); + + // If the caller aborts (client disconnect), make sure a pending read() + // wakes up. The reader.cancel() in destroy() handles the converse. + if (opts.abortSignal) { + const sig = opts.abortSignal; + if (sig.aborted) { + releaseReader(); + } else { + sig.addEventListener( + "abort", + () => { + disarmIdle(); + stream.destroy(new Error("Client disconnected")); + }, + { once: true }, + ); + } + } + + return stream; } diff --git a/tests/config.test.ts b/tests/config.test.ts index 20fe6c5..de9c5f7 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -30,6 +30,43 @@ describe("loadConfig", () => { expect(config.apiKey).toBeNull(); expect(config.ccApiBase).toBe("https://api.commandcode.ai"); expect(config.ccVersion).toBe("0.40.3"); + // Defaults: 10-minute connect, 2-minute idle. + expect(config.upstreamTimeoutMs).toBe(600_000); + expect(config.idleTimeoutMs).toBe(120_000); + }); + + it("reads upstream + idle timeouts from env vars", async () => { + vi.stubEnv("CC_API_KEY", ""); + vi.stubEnv("CC_UPSTREAM_TIMEOUT_MS", "1800000"); // 30 min + vi.stubEnv("CC_IDLE_TIMEOUT_MS", "300000"); // 5 min + + const { loadConfig } = await import("@/config.js"); + const config = loadConfig(); + + expect(config.upstreamTimeoutMs).toBe(1_800_000); + expect(config.idleTimeoutMs).toBe(300_000); + }); + + it("allows disabling the idle timeout by setting it to 0", async () => { + vi.stubEnv("CC_API_KEY", ""); + vi.stubEnv("CC_IDLE_TIMEOUT_MS", "0"); + + const { loadConfig } = await import("@/config.js"); + const config = loadConfig(); + + expect(config.idleTimeoutMs).toBe(0); + }); + + it("falls back to defaults on invalid timeout values", async () => { + vi.stubEnv("CC_API_KEY", ""); + vi.stubEnv("CC_UPSTREAM_TIMEOUT_MS", "not-a-number"); + vi.stubEnv("CC_IDLE_TIMEOUT_MS", "-5"); + + const { loadConfig } = await import("@/config.js"); + const config = loadConfig(); + + expect(config.upstreamTimeoutMs).toBe(600_000); + expect(config.idleTimeoutMs).toBe(120_000); }); it("reads from environment variables", async () => { diff --git a/tests/upstream.test.ts b/tests/upstream.test.ts index 2bad866..d697fd2 100644 --- a/tests/upstream.test.ts +++ b/tests/upstream.test.ts @@ -250,4 +250,98 @@ describe("sendToCC retry", () => { // cancel() is called synchronously from destroy(). expect(cancelCalled).toBe(true); }); + + // Regression: a stalled upstream (TCP open, no chunks arriving) used to + // hang the consumer forever — no idle timeout was wired during streaming. + // Now nodeReaderToStream aborts the reader after `idleTimeoutMs` of no data. + it("aborts the upstream reader after the idle timeout elapses with no data", async () => { + // Reader that never produces data and never returns done. + let cancelReason: unknown = undefined; + let cancelCallCount = 0; + const pendingReadRejectors: Array<(e: unknown) => void> = []; + const reader = { + // read() never resolves on its own — simulates a stalled upstream. + read: () => + new Promise<{ done: boolean; value?: Uint8Array }>((_resolve, reject) => { + pendingReadRejectors.push(reject); + }), + // cancel(reason) rejects the pending read() — matches the + // ReadableStreamDefaultReader spec, which propagates the cancel + // reason through any in-flight read(). + cancel: async (reason?: unknown) => { + // Only capture the FIRST cancel call's reason — the proxy calls + // cancel() again from releaseReader() with no arg, which would + // otherwise overwrite our assertion. + if (cancelCallCount === 0) cancelReason = reason; + cancelCallCount++; + for (const reject of pendingReadRejectors.splice(0)) { + reject(reason); + } + }, + }; + const mock = vi.fn().mockResolvedValue({ + ok: true, + status: 200, + statusText: "", + text: async () => "", + body: { getReader: () => reader }, + } as unknown as Response); + globalThis.fetch = mock as unknown as typeof fetch; + + const { stream } = await sendToCC(sampleBody(), { + apiBase: "https://example.test", + apiKey: "k", + ccVersion: "0.0.0", + idleTimeoutMs: 50, // very short for the test + }); + + // Resume so the Readable actually starts calling read() — otherwise the + // idle timer never gets a chance to arm. + (stream as Readable).resume(); + + const error: Error | undefined = await new Promise((resolve) => { + stream.once("error", (e: Error) => resolve(e)); + }); + + expect(cancelCallCount).toBeGreaterThan(0); + expect((cancelReason as Error)?.name).toBe("IdleTimeoutError"); + // The stream should surface an error to the consumer so the SSE handler + // synthesizes a clean finish instead of hanging the client. + expect(error).toBeInstanceOf(Error); + expect(error?.message).toContain("idle timeout"); + }); + + // Regression: idle timeout is disabled when idleTimeoutMs=0. + it("does not fire idle timeout when idleTimeoutMs is 0", async () => { + // Same stalled reader, but idle disabled. + const reader = { + read: () => new Promise<{ done: boolean; value?: Uint8Array }>(() => {}), + cancel: async () => {}, + }; + const mock = vi.fn().mockResolvedValue({ + ok: true, + status: 200, + statusText: "", + text: async () => "", + body: { getReader: () => reader }, + } as unknown as Response); + globalThis.fetch = mock as unknown as typeof fetch; + + const { stream } = await sendToCC(sampleBody(), { + apiBase: "https://example.test", + apiKey: "k", + ccVersion: "0.0.0", + idleTimeoutMs: 0, + }); + + (stream as Readable).resume(); + + // Wait long enough that an idle timer WOULD have fired. + const errored = await Promise.race([ + new Promise((resolve) => stream.once("error", resolve)), + new Promise((r) => setTimeout(() => r(undefined), 80)), + ]); + expect(errored).toBeUndefined(); + (stream as Readable).destroy(); + }); }); From c887b9f3f8a3dbe57eb77dcb3ccabbd1c546ebb3 Mon Sep 17 00:00:00 2001 From: thaolaptrinh Date: Tue, 21 Jul 2026 20:26:33 +0700 Subject: [PATCH 3/4] fix(translate): non-streaming tool-call dedup, content order, message_start on empty stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final-pass audit surfaced eight remaining correctness and resource hygiene issues across the OpenAI/Anthropic translation layers and the streaming server. None of them are the original 'tool call truncated' class — they're separate contract violations and edge cases. Critical - openai: buildNonStreamingResponse now deduplicates tool-call-delta + final tool-call with the same id (the streaming encoder was fixed in the previous commit but the non-streaming builder was not). Without this, clients received two tool_calls entries with the same id, which some clients call twice and some dedupe wrong. - anthropic: handleFinish now emits message_start when no content event arrived first (empty response, max_tokens=0, immediate refusal). The Anthropic SDK requires message_start as the first event of a stream and throws on its absence — the error and finishRecords paths already guarded this, but handleFinish did not. High - anthropic: non-streaming response now orders content blocks [thinking, text, tool_use] per the extended-thinking contract. Previously emitted [text, thinking, tool_use], which broke Claude Code's thinking-block continuation logic. Existing test asserted the wrong order; fixed. - server: post-pumpStream [DONE] write now guards res.destroyed, not just res.writableEnded, preventing ERR_STREAM_DESTROYED on a socket torn down by mid-stream client disconnect. - server: parseBody now calls req.destroy() on 413 so the client socket is freed immediately instead of lingering until the upload completes (was holding the connection open for the full oversized body even after rejecting). - openai/server: pumpStream onError no longer mixes a non-chunk {error:...} envelope with valid chunks — both error paths (encoder error event + stream-level error) now emit uniform content+finish chat.completion.chunk records. Some clients were parsing the bare envelope as a tool call named 'error'. Medium - server: removed dead destroyStreamOnClientDisconnect calls in the streaming paths (mid-stream disconnect is already handled via the abort signal plumbed into nodeReaderToStream; the call after pumpStream returns is a no-op since the stream has ended/errored). Low - anthropic: top_p is now propagated into ccBody.params (was silently dropped, affecting Anthropic clients that tune sampling). Tests - translate: tool-call-delta + tool-call same-id merge in non-streaming builder. - translate-anthropic: message_start ordering when finish arrives with no prior content; content block ordering thinking-before-text. - e2e: stream-level error produces uniform chunks (no out-of-band envelope). - All 152 tests pass; tsc --noEmit clean. --- src/server.ts | 32 ++++++++--- src/translate/anthropic.ts | 22 ++++++- src/translate/openai.ts | 95 +++++++++++++++++++++---------- tests/e2e.test.ts | 53 +++++++++++++++++ tests/translate-anthropic.test.ts | 28 ++++++++- tests/translate.test.ts | 21 +++++++ 6 files changed, 207 insertions(+), 44 deletions(-) diff --git a/src/server.ts b/src/server.ts index c2a4bb6..334a2ab 100644 --- a/src/server.ts +++ b/src/server.ts @@ -58,6 +58,11 @@ function parseBody(req: http.IncomingMessage): Promise { size += chunk.length; if (size > MAX_BODY_BYTES) { tooLarge = true; + // Tear down the underlying socket so the client stops uploading the + // rest of an oversized body. Without this the connection lingers + // until the client finishes (or its own timeout fires) — wasting + // bandwidth and a request slot. + req.destroy(); reject(new BodyParseError(413, "Request body too large")); return; } @@ -352,18 +357,26 @@ async function handleChatCompletions( encoder.finished ? [] : encoder.finishChunks("stop").map((c) => formatSSE(c)), - (err) => { - const chunks: object[] = [encoder.errorChunk(err)]; - if (!encoder.finished) chunks.push(...encoder.finishChunks("stop")); - return chunks.map((c) => formatSSE(c)); - }, + // Stream-level error (TCP failure, idle timeout, encoder throw). + // Always emit a uniform content+finish chunk pair via streamErrorChunks + // — mixing a non-chunk `{error:...}` envelope with valid chunks + // confused some clients (treating the envelope as a tool call named + // "error", or failing JSON parse). + (err) => encoder.streamErrorChunks(err).map((c) => formatSSE(c)), ); // After pump completes, emit the [DONE] sentinel if we still can. - if (!res.writableEnded) { + // `writableEnded` only flips when end() is called — `res.destroyed` + // catches the case where the client disconnected mid-stream and the + // socket was torn down underneath us. + if (!res.writableEnded && !res.destroyed) { res.write(formatSSEDone()); res.end(); } - destroyStreamOnClientDisconnect(req, stream); + // No destroyStreamOnClientDisconnect here — by the time pumpStream + // returns the stream has already ended or errored, so the call would + // be a no-op. Mid-stream disconnects are handled by the abort signal + // (see abortOnClientDisconnect + nodeReaderToStream's abortSignal + // listener). } else { destroyStreamOnClientDisconnect(req, stream); const events = await collectEvents(stream); @@ -456,8 +469,9 @@ async function handleMessages(req: http.IncomingMessage, res: http.ServerRespons return records.map((r) => formatAnthropicSSE(r.event, r.data)); }, ); - if (!res.writableEnded) res.end(); - destroyStreamOnClientDisconnect(req, stream); + if (!res.writableEnded && !res.destroyed) res.end(); + // No destroyStreamOnClientDisconnect here — see OpenAI streaming path + // for rationale (abort signal already covers mid-stream disconnect). } else { destroyStreamOnClientDisconnect(req, stream); const events = await collectEvents(stream); diff --git a/src/translate/anthropic.ts b/src/translate/anthropic.ts index aaf83db..726585d 100644 --- a/src/translate/anthropic.ts +++ b/src/translate/anthropic.ts @@ -223,6 +223,7 @@ export function toCCRequest( stream: req.stream ?? false, max_tokens: req.max_tokens, temperature: req.temperature, + top_p: req.top_p, stop: req.stop_sequences, tools: ccTools, tool_choice: resolveToolChoice(req), @@ -317,6 +318,15 @@ export class AnthropicStreamEncoder { this.sawFinish = true; const records: AnthropicSSERecord[] = []; + // If we never emitted a content event, `started` is still false and no + // message_start was sent. The Anthropic SDK requires message_start as + // the first event — synthesize one before the closing records so we + // don't deliver a stream that starts with message_delta. + if (!this.started) { + records.push(this.makeMessageStart(0)); + this.started = true; + } + this.closeCurrentBlock(records); const finishReason = (event.data.finishReason as string) ?? "stop"; @@ -567,6 +577,10 @@ export function buildAnthropicResponse( } } + // Block ordering follows Anthropic's extended-thinking contract: + // thinking blocks must precede the text they reason about, and tool_use + // blocks come last. Mixing this up confuses strict clients (Claude Code + // uses thinking-block position to continue reasoning across turns). const content: OutputContentBlock[] = []; if (thinkingContent) { content.push({ @@ -575,10 +589,12 @@ export function buildAnthropicResponse( signature: "_cc_proxy_placeholder", }); } + if (textContent) content.push({ type: "text", text: textContent }); content.push(...toolUseBlocks); - if (textContent || content.length === 0) { - content.unshift({ type: "text", text: textContent }); - } + // Anthropic requires content to be non-empty — if there was no text, no + // thinking, and no tool calls (e.g. empty refusal), synthesize an empty + // text block rather than sending an empty array. + if (content.length === 0) content.push({ type: "text", text: "" }); const finishEvent = events.find((e) => e.type === "finish"); const finishReason = (finishEvent?.data.finishReason as string) ?? "stop"; diff --git a/src/translate/openai.ts b/src/translate/openai.ts index 6fec7f8..2a4c2b5 100644 --- a/src/translate/openai.ts +++ b/src/translate/openai.ts @@ -393,35 +393,52 @@ export class OpenAIStreamEncoder { case "error": { this.sawFinish = true; - const errMsg = + return this.errorChunks( (event.data.message as string) ?? - (event.data.error as { message?: string } | undefined)?.message ?? - JSON.stringify(event.data); - logger.error(`[CC upstream error] ${errMsg}`); - chunks.push({ - id, - object: "chat.completion.chunk", - created, - model: this.model, - choices: [ - { index: 0, delta: { content: `[upstream error] ${errMsg}` }, finish_reason: null }, - ], - }); - chunks.push({ - id, - object: "chat.completion.chunk", - created, - model: this.model, - choices: [{ index: 0, delta: {}, finish_reason: "stop" }], - }); - break; + (event.data.error as { message?: string } | undefined)?.message ?? + JSON.stringify(event.data), + ); } } return chunks; } + /** + * Build a uniform error→content+finish chunk pair. Used both by `emit()` + * for upstream error events and by the server's pumpStream catch for + * stream-level errors (TCP failure, idle timeout, encoder throw). Both + * paths surface errors as valid chat.completion.chunk records so the + * client always sees a uniform stream shape instead of a mix of valid + * chunks and an ad-hoc `{error:...}` envelope. + */ + private errorChunks(message: string): object[] { + const id = this.id; + const created = this.created; + logger.error(`[CC upstream error] ${message}`); + return [ + { + id, + object: "chat.completion.chunk", + created, + model: this.model, + choices: [ + { index: 0, delta: { content: `[upstream error] ${message}` }, finish_reason: null }, + ], + }, + { + id, + object: "chat.completion.chunk", + created, + model: this.model, + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + }, + ]; + } + errorChunk(err: Error): object { + // Retained for backwards-compat with tests; new code should call + // errorChunks() instead so the error appears inside a valid chunk. return { error: { message: err.message, @@ -447,6 +464,15 @@ export class OpenAIStreamEncoder { }, ]; } + + /** + * Public wrapper around errorChunks() for stream-level errors caught by + * pumpStream (TCP failure, idle timeout, encoder throw). Emits a uniform + * content+finish chunk pair instead of an out-of-band error envelope. + */ + streamErrorChunks(err: Error): object[] { + return this.errorChunks(err.message); + } } // ────────────────────────────────────────── @@ -489,16 +515,27 @@ export function buildNonStreamingResponse(events: CCEvent[], model: string, id: break; } case "tool-call": { + // CC bridges commonly emit tool-call-delta * N then a final tool-call + // with the same id. If we already accumulated args from deltas, + // replace that entry with the canonical final payload (matches what + // the streaming encoder does). Otherwise this is a one-shot tool + // call with no preceding deltas — push it. + const id = (event.data.toolCallId as string) ?? ""; const input = event.data.input ?? event.data.arguments; - toolCalls.push({ - id: (event.data.toolCallId as string) ?? "", + const argsStr = + typeof input === "string" ? input : input != null ? JSON.stringify(input) : ""; + const name = (event.data.toolName as string) ?? (event.data.name as string) ?? ""; + const existingIdx = toolCalls.findIndex((tc) => tc.id === id); + const entry: ToolCall = { + id, type: "function", - function: { - name: (event.data.toolName as string) ?? (event.data.name as string) ?? "", - arguments: - typeof input === "string" ? input : input != null ? JSON.stringify(input) : "", - }, - }); + function: { name, arguments: argsStr }, + }; + if (existingIdx >= 0) { + toolCalls[existingIdx] = entry; + } else { + toolCalls.push(entry); + } break; } case "finish": diff --git a/tests/e2e.test.ts b/tests/e2e.test.ts index e6faef3..c48d6d4 100644 --- a/tests/e2e.test.ts +++ b/tests/e2e.test.ts @@ -221,6 +221,59 @@ describe("E2E: OpenAI /v1/chat/completions", () => { const finishChunks = parsed.filter((c) => c.choices?.[0]?.finish_reason); expect(finishChunks).toHaveLength(1); }); + + // Regression: a stream-level error (TCP failure, idle timeout, encoder + // throw) used to emit a non-chunk `{error:{...}}` envelope alongside a + // valid chunk, which some clients parsed as a tool call named "error". + // Now pumpStream emits only uniform valid chunks. + it("streaming: stream-level error surfaces as a uniform chunk (no out-of-band error envelope)", async () => { + // Build a stream that emits one event, then errors out. Attach an + // error listener up-front so destroy(err) doesn't trigger Node's + // unhandled-exception path (pumpStream's for-await handles the read + // rejection, but the bare 'error' event still propagates). + const stream = new Readable({ objectMode: true, read() {} }); + stream.on("error", () => { + /* swallowed — pumpStream catches the read() rejection */ + }); + process.nextTick(() => { + stream.push({ type: "start", data: {} }); + stream.push({ type: "text-delta", data: { text: "partial" } }); + stream.destroy(new Error("simulated TCP RST")); + }); + sendToCCSpy.mockResolvedValue({ stream }); + + const res = await fetch(`${baseUrl}/v1/chat/completions`, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer test-key", + }, + body: JSON.stringify({ + model: "deepseek/deepseek-v4-flash", + messages: [{ role: "user", content: "Hi" }], + stream: true, + }), + }); + + expect(res.status).toBe(200); + const text = await res.text(); + const lines = text.split("\n").filter((l) => l.startsWith("data: ") && !l.includes("[DONE]")); + const parsed = lines.map((l) => JSON.parse(l.replace("data: ", ""))); + // Every emitted data record must be a valid chat.completion.chunk — no + // bare `{error:...}` envelope mixed in. + for (const chunk of parsed) { + expect(chunk.object).toBe("chat.completion.chunk"); + } + // The error text should appear as delta.content (visible to the user) + // and the stream should still terminate with finish_reason:"stop". + const errorChunk = parsed.find((c) => + typeof c.choices?.[0]?.delta?.content === "string" && + c.choices[0].delta.content.includes("simulated TCP RST"), + ); + expect(errorChunk).toBeDefined(); + const finishChunks = parsed.filter((c) => c.choices?.[0]?.finish_reason); + expect(finishChunks).toHaveLength(1); + }); }); // ────────────────────────────────────────── diff --git a/tests/translate-anthropic.test.ts b/tests/translate-anthropic.test.ts index 0685872..27e3015 100644 --- a/tests/translate-anthropic.test.ts +++ b/tests/translate-anthropic.test.ts @@ -350,6 +350,27 @@ describe("AnthropicStreamEncoder", () => { expect(usage.service_tier).toBe("standard"); }); + // Regression: if CC sends start + finish with no content events in + // between (empty response, immediate refusal, max_tokens=0, etc.), + // handleFinish must synthesize message_start — the Anthropic SDK + // requires message_start as the first event of the stream. + it("emits message_start when finish arrives with no prior content", () => { + const encoder = new AnthropicStreamEncoder("m"); + encoder.emit({ type: "start", data: {} }); + const finishRecords = encoder.emit({ + type: "finish", + data: { finishReason: "stop" }, + }); + const eventTypes = finishRecords.map((r) => r.event); + // message_start must come BEFORE message_delta/message_stop. + const msIdx = eventTypes.indexOf("message_start"); + const mdIdx = eventTypes.indexOf("message_delta"); + const stopIdx = eventTypes.indexOf("message_stop"); + expect(msIdx).toBeGreaterThanOrEqual(0); + expect(mdIdx).toBeGreaterThan(msIdx); + expect(stopIdx).toBeGreaterThan(mdIdx); + }); + it("two encoders are independent (concurrency fix)", () => { const a = new AnthropicStreamEncoder("a"); const b = new AnthropicStreamEncoder("b"); @@ -476,9 +497,10 @@ describe("buildAnthropicResponse", () => { ]; const resp = buildAnthropicResponse(events, "m", "id"); expect(resp.content).toHaveLength(2); - expect(resp.content[0].type).toBe("text"); - expect(resp.content[1].type).toBe("thinking"); - expect((resp.content[1] as { signature: string }).signature).toBe("_cc_proxy_placeholder"); + // Thinking must precede text per Anthropic's extended-thinking contract. + expect(resp.content[0].type).toBe("thinking"); + expect(resp.content[1].type).toBe("text"); + expect((resp.content[0] as { signature: string }).signature).toBe("_cc_proxy_placeholder"); }); it("includes tool_use blocks (no text means no empty text block)", () => { diff --git a/tests/translate.test.ts b/tests/translate.test.ts index e879af3..2bcca86 100644 --- a/tests/translate.test.ts +++ b/tests/translate.test.ts @@ -432,6 +432,27 @@ describe("buildNonStreamingResponse", () => { expect(resp.choices[0].message.reasoning_content).toBe("thinking"); expect(resp.choices[0].message.content).toBe("Answer"); }); + + // Regression: when CC streams tool-call-delta * N then a final tool-call + // with the same id, the non-streaming builder used to push a duplicate + // entry — producing two tool_calls with the same id, which most clients + // either dedupe wrongly or call the tool twice. + it("merges tool-call-delta + tool-call with the same id into one entry", () => { + const events: CCEvent[] = [ + { type: "start", data: {} }, + { type: "tool-call-delta", data: { toolCallId: "call_X", name: "search", arguments: '{"q":' } }, + { type: "tool-call-delta", data: { toolCallId: "call_X", arguments: '"hi"}' } }, + { type: "tool-call", data: { toolCallId: "call_X", toolName: "search", input: { q: "hi" } } }, + { type: "finish", data: { finishReason: "tool-call" } }, + ]; + const resp = buildNonStreamingResponse(events, "m", "id") as any; + const toolCalls = resp.choices[0].message.tool_calls as any[]; + expect(toolCalls).toHaveLength(1); + expect(toolCalls[0].id).toBe("call_X"); + expect(toolCalls[0].function.name).toBe("search"); + // Final tool-call payload overrides accumulated deltas. + expect(toolCalls[0].function.arguments).toBe('{"q":"hi"}'); + }); }); // ────────────────────────────────────────── From 54c831064cebcb66c82615b103adcec660c2712b Mon Sep 17 00:00:00 2001 From: thaolaptrinh Date: Tue, 21 Jul 2026 20:48:10 +0700 Subject: [PATCH 4/4] chore: remove Docker support (unnecessary for a localhost proxy) --- .github/workflows/ci.yml | 16 ---------------- DEVELOPMENT.md | 20 -------------------- Dockerfile | 17 ----------------- Makefile | 17 +---------------- docker-compose.yml | 19 ------------------- 5 files changed, 1 insertion(+), 88 deletions(-) delete mode 100644 Dockerfile delete mode 100644 docker-compose.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d25ff8b..3917e56 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,19 +34,3 @@ jobs: - name: Build run: pnpm build - docker: - runs-on: ubuntu-latest - needs: quality - if: github.ref == 'refs/heads/main' && github.event_name == 'push' - - steps: - - uses: actions/checkout@v4 - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 - - - name: Build Docker image - uses: docker/build-push-action@v6 - with: - push: false - tags: commandcode-api-proxy:latest diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index af4c0df..9d0c24a 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -72,26 +72,6 @@ tests/ └── e2e.test.ts # End-to-end integration tests ``` -## Docker - -```bash -# Build -docker build -t commandcode-api-proxy . - -# Run with env var -docker run --rm -p 8787:8787 \ - -e CC_API_KEY=user_xxx \ - commandcode-api-proxy - -# Or mount auth.json -docker run --rm -p 8787:8787 \ - -v ~/.config/commandcode-api-proxy:/home/node/.config/commandcode-api-proxy:ro \ - commandcode-api-proxy - -# Using docker compose -docker compose up -d -``` - ## Tech stack - **Runtime:** Node.js (zero runtime dependencies) diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index 5b4aaa4..0000000 --- a/Dockerfile +++ /dev/null @@ -1,17 +0,0 @@ -FROM node:24-alpine AS builder -RUN corepack enable pnpm -WORKDIR /app -COPY package.json pnpm-lock.yaml ./ -RUN pnpm install --frozen-lockfile -COPY src/ src/ -COPY tsconfig.json ./ -RUN pnpm build - -FROM node:24-alpine AS runner -WORKDIR /app -COPY --from=builder /app/dist ./dist -COPY --from=builder /app/package.json ./ -EXPOSE 8787 -ENV HOST=0.0.0.0 -ENV PORT=8787 -ENTRYPOINT ["node", "dist/proxy.js"] diff --git a/Makefile b/Makefile index 2ef3359..33d1b96 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: build dev start test test-watch test-coverage fmt lint check clean docker-build docker-run docker-run-detached release help +.PHONY: build dev start test test-watch test-coverage fmt lint check clean release help APP_NAME := commandcode-api-proxy VERSION := $(shell node -p "require('./package.json').version") @@ -32,21 +32,6 @@ clean: ## Clean build artifacts rm -rf dist/ rm -rf coverage/ -docker-build: ## Build Docker image - docker build -t $(APP_NAME):$(VERSION) . - docker tag $(APP_NAME):$(VERSION) $(APP_NAME):latest - -docker-run: ## Run Docker container - docker run --rm -p 8787:8787 \ - -e CC_API_KEY=$(or $(CC_API_KEY),) \ - $(APP_NAME):latest - -docker-run-detached: ## Run Docker container in background - docker run -d --name $(APP_NAME) \ - -p 8787:8787 \ - -e CC_API_KEY=$(or $(CC_API_KEY),) \ - $(APP_NAME):latest - release: build test ## Build and test for release @echo "Ready for release: pnpm publish" diff --git a/docker-compose.yml b/docker-compose.yml deleted file mode 100644 index 306e76d..0000000 --- a/docker-compose.yml +++ /dev/null @@ -1,19 +0,0 @@ -services: - commandcode-api-proxy: - build: . - image: commandcode-api-proxy:latest - container_name: commandcode-api-proxy - restart: unless-stopped - ports: - # Bind host-side to localhost only so the proxy (which forwards your paid - # CC key with no real auth) is not exposed to the LAN. Remove the - # 127.0.0.1: prefix only if you intentionally want network access. - - "127.0.0.1:8787:8787" - environment: - - HOST=0.0.0.0 - - PORT=8787 - - CC_API_KEY=${CC_API_KEY:-} - - LOG_LEVEL=${LOG_LEVEL:-info} - volumes: - # auth.json lives at ~/.config/commandcode-api-proxy/auth.json (see auth.ts). - - ~/.config/commandcode-api-proxy:/home/node/.config/commandcode-api-proxy:ro