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/.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/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/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 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 ee3c340..334a2ab 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 @@ -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; } @@ -176,6 +181,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 // ────────────────────────────────────────── @@ -275,6 +334,8 @@ async function handleChatCompletions( apiBase: config.ccApiBase, apiKey, ccVersion: config.ccVersion, + timeoutMs: config.upstreamTimeoutMs, + idleTimeoutMs: config.idleTimeoutMs, }, abort.signal, ); @@ -288,25 +349,36 @@ 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)), + // 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. + // `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(); - }); - 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); + } + // 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); const response = buildNonStreamingResponse(events, model, encoder.id); sendJson(res, 200, response); @@ -357,7 +429,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; @@ -370,26 +448,32 @@ 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(); - } - }); - destroyStreamOnClientDisconnect(req, stream); + 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.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); 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..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), @@ -249,11 +250,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 +269,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,8 +315,18 @@ export class AnthropicStreamEncoder { } private handleFinish(event: CCEvent): AnthropicSSERecord[] { + 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"; @@ -389,6 +406,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 +515,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 ── @@ -526,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({ @@ -534,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 5d2d720..2a4c2b5 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,35 +392,53 @@ export class OpenAIStreamEncoder { } case "error": { - const errMsg = + this.sawFinish = true; + 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, @@ -392,6 +446,33 @@ 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 }], + }, + ]; + } + + /** + * 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); + } } // ────────────────────────────────────────── @@ -434,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/src/upstream.ts b/src/upstream.ts index 129893b..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,22 +227,100 @@ function combineSignals(...signals: AbortSignal[]): AbortSignal { function nodeReaderToStream( reader: ReadableStreamDefaultReader, + opts: { idleTimeoutMs?: number; abortSignal?: AbortSignal } = {}, ): 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; - return new Readable({ + // 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; + if (typeof cancel === "function") { + cancel.call(reader).catch(() => { + /* already closed */ + }); + } + }; + + const stream = 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; + } + + armIdle(); const { done, value } = await reader.read(); + disarmIdle(); 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,19 +329,34 @@ 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); } }, }); + + // 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/e2e.test.ts b/tests/e2e.test.ts index 9fc67c1..c48d6d4 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,93 @@ 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); + }); + + // 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); + }); }); // ────────────────────────────────────────── @@ -307,6 +431,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..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"); @@ -404,6 +425,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", () => { @@ -439,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 2296f0b..2bcca86 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); + }); }); // ────────────────────────────────────────── @@ -377,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"}'); + }); }); // ────────────────────────────────────────── diff --git a/tests/upstream.test.ts b/tests/upstream.test.ts index e2e02fc..d697fd2 100644 --- a/tests/upstream.test.ts +++ b/tests/upstream.test.ts @@ -164,4 +164,184 @@ 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); + }); + + // 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(); + }); });