diff --git a/src/config.ts b/src/config.ts index b7982ed..73c9e90 100644 --- a/src/config.ts +++ b/src/config.ts @@ -15,7 +15,7 @@ export interface Config { ccVersion: string; logLevel: string; corsOrigin: string; - /** Max wall-clock ms for upstream to send response headers + first byte. */ + /** Per-attempt deadline for upstream headers and any non-2xx error body. */ upstreamTimeoutMs: number; /** Max ms between consecutive chunks during streaming. 0 = disabled. */ idleTimeoutMs: number; @@ -92,7 +92,7 @@ export function loadConfig(): Config { const corsOrigin = process.env.CORS_ORIGIN ?? "*"; // Upstream timeouts. The connection timeout covers the wall-clock time - // until the upstream returns response headers + first byte — bump it for + // until the upstream returns headers (and consumes any error body) — 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 diff --git a/src/server.ts b/src/server.ts index 0972e1a..87f5964 100644 --- a/src/server.ts +++ b/src/server.ts @@ -17,6 +17,7 @@ import { getProxyVersion } from "@/version.js"; import { validateOpenAIChatRequest, validateAnthropicRequest, + validateCountTokensRequest, ValidationError, } from "@/translate/validation.js"; import type { AnthropicRequest, AnthropicSSERecord } from "@/translate/anthropic-types.js"; @@ -159,24 +160,17 @@ function corsHeaders(): Record { // Helpers // ────────────────────────────────────────── -function abortOnClientDisconnect( - req: http.IncomingMessage, - res: http.ServerResponse, -): AbortController { +function abortOnClientDisconnect(res: http.ServerResponse): AbortController { const abort = new AbortController(); - req.on("close", () => { + // IncomingMessage.close marks a completed upload, not a lost response client. + const onClose = (): void => { if (!res.writableEnded) abort.abort(); - }); + }; + res.once("close", onClose); + if (res.destroyed) onClose(); return abort; } -function destroyStreamOnClientDisconnect( - req: http.IncomingMessage, - stream: NodeJS.ReadableStream, -): void { - 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 @@ -186,7 +180,17 @@ 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)); + const settle = (writable: boolean): void => { + res.off("drain", onDrain); + res.off("close", onClose); + res.off("error", onClose); + resolve(writable); + }; + const onDrain = (): void => settle(!res.writableEnded && !res.destroyed); + const onClose = (): void => settle(false); + res.once("drain", onDrain); + res.once("close", onClose); + res.once("error", onClose); }); } @@ -214,7 +218,7 @@ async function pumpStream( // 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; + throw err; } for (const chunk of chunks) { if (!(await writable(chunk))) return; @@ -322,7 +326,7 @@ async function handleChatCompletions( const ccBody = toCCRequest(openAIReq); - const abort = abortOnClientDisconnect(req, res); + const abort = abortOnClientDisconnect(res); try { const result = await sendToCC( @@ -366,13 +370,8 @@ async function handleChatCompletions( res.write(formatSSEDone()); res.end(); } - // 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). + // The response abort signal covers both streaming and JSON clients. } else { - destroyStreamOnClientDisconnect(req, stream); const events = await collectEvents(stream); const response = buildNonStreamingResponse(events, model, encoder.id); sendJson(res, 200, response); @@ -418,7 +417,7 @@ async function handleMessages(req: http.IncomingMessage, res: http.ServerRespons const encoder = new AnthropicStreamEncoder(model); const ccBody = anToCCRequest(anthropicReq); - const abort = abortOnClientDisconnect(req, res); + const abort = abortOnClientDisconnect(res); try { const result = await sendToCC( @@ -462,10 +461,8 @@ async function handleMessages(req: http.IncomingMessage, res: http.ServerRespons }, ); if (!res.writableEnded && !res.destroyed) res.end(); - // No destroyStreamOnClientDisconnect here — see OpenAI streaming path - // for rationale (abort signal already covers mid-stream disconnect). + // The response abort signal already covers mid-stream disconnects. } else { - destroyStreamOnClientDisconnect(req, stream); const events = await collectEvents(stream); const response = buildAnthropicResponse(events, model, encoder.messageId); res.writeHead(200, { "Content-Type": "application/json", ...corsHeaders() }); @@ -494,7 +491,17 @@ async function handleCountTokens( ); } - const body = rawBody as Record; + let body: Record; + try { + body = validateCountTokensRequest(rawBody); + } catch (err) { + return sendAnthropicError( + res, + 400, + "invalid_request_error", + err instanceof ValidationError ? err.message : "Invalid request body", + ); + } const parts: string[] = []; if (typeof body.system === "string") parts.push(body.system); diff --git a/src/translate/anthropic.ts b/src/translate/anthropic.ts index 726585d..243e7f2 100644 --- a/src/translate/anthropic.ts +++ b/src/translate/anthropic.ts @@ -1,4 +1,5 @@ import crypto from "node:crypto"; +import { toolArgumentSuffix } from "./tool-arguments.js"; import type { AnthropicRequest, AnthropicContentBlock, @@ -245,8 +246,12 @@ export function toCCRequest( export class AnthropicStreamEncoder { readonly messageId: string; private blockIndex = 0; + private currentBlockIndex = 0; private currentBlockType: "text" | "thinking" | "tool_use" | null = null; - private currentToolCallId: string | null = null; + private readonly toolBlocks = new Map< + string, + { index: number; arguments: string; closed: boolean } + >(); private pendingStart: CCEvent | null = null; private started = false; private pinged = false; @@ -281,6 +286,7 @@ export class AnthropicStreamEncoder { records.push(this.makeMessageStart(0)); } this.closeCurrentBlock(records); + this.closeToolBlocks(records); records.push({ event: "error", data: { type: "error", error: { type: "api_error", message: msg } }, @@ -328,6 +334,7 @@ export class AnthropicStreamEncoder { } this.closeCurrentBlock(records); + this.closeToolBlocks(records); const finishReason = (event.data.finishReason as string) ?? "stop"; const usage = extractUsage(event.data as Record); @@ -381,21 +388,18 @@ export class AnthropicStreamEncoder { case "tool-call-delta": { const tcId = (event.data.toolCallId as string) ?? ""; const tcName = (event.data.name as string) ?? ""; - if (this.currentBlockType !== "tool_use" || this.currentToolCallId !== tcId) { - this.closeCurrentBlock(records); - this.ensureBlockOpenWith(records, "tool_use", { - type: "tool_use", - id: tcId, - name: tcName, - input: {}, - }); - this.currentToolCallId = tcId; - } + const block = this.ensureToolBlock(records, tcId, tcName); + if (block.closed) throw new Error("Inconsistent upstream tool arguments"); + const args = (event.data.arguments as string) ?? ""; + block.arguments += args; records.push( - this.makeDelta({ - type: "input_json_delta", - partial_json: (event.data.arguments as string) ?? "", - }), + this.makeDelta( + { + type: "input_json_delta", + partial_json: args, + }, + block.index, + ), ); break; } @@ -406,26 +410,16 @@ 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", - id: tcId, - name: tcName, - input: {}, - }); - if (argsStr) { - records.push(this.makeDelta({ type: "input_json_delta", partial_json: argsStr })); + const block = this.ensureToolBlock(records, tcId, tcName); + const suffix = toolArgumentSuffix(block.arguments, argsStr); + if (suffix) { + if (block.closed) throw new Error("Inconsistent upstream tool arguments"); + records.push( + this.makeDelta({ type: "input_json_delta", partial_json: suffix }, block.index), + ); + block.arguments += suffix; } - this.closeCurrentBlock(records); + this.closeToolBlock(records, block); break; } } @@ -433,6 +427,34 @@ export class AnthropicStreamEncoder { return records; } + private ensureToolBlock(records: AnthropicSSERecord[], id: string, name: string) { + const existing = this.toolBlocks.get(id); + if (existing) return existing; + this.closeCurrentBlock(records); + this.ensureBlockOpenWith(records, "tool_use", { type: "tool_use", id, name, input: {} }); + const block = { index: this.currentBlockIndex, arguments: "", closed: false }; + this.toolBlocks.set(id, block); + // Tool blocks have independent lifetimes: interleaved deltas retain indices. + this.currentBlockType = null; + return block; + } + + private closeToolBlock( + records: AnthropicSSERecord[], + block: { index: number; closed: boolean }, + ): void { + if (block.closed) return; + records.push({ + event: "content_block_stop", + data: { type: "content_block_stop", index: block.index }, + }); + block.closed = true; + } + + private closeToolBlocks(records: AnthropicSSERecord[]): void { + for (const block of this.toolBlocks.values()) this.closeToolBlock(records, block); + } + private ensureBlockOpen( records: AnthropicSSERecord[], type: "text" | "thinking" | "tool_use", @@ -450,11 +472,12 @@ export class AnthropicStreamEncoder { block: ContentBlockStartShape, ): void { this.currentBlockType = type; + this.currentBlockIndex = this.blockIndex++; records.push({ event: "content_block_start", data: { type: "content_block_start", - index: this.blockIndex, + index: this.currentBlockIndex, content_block: block, }, }); @@ -477,17 +500,16 @@ export class AnthropicStreamEncoder { records.push({ event: "content_block_stop", - data: { type: "content_block_stop", index: this.blockIndex }, + data: { type: "content_block_stop", index: this.currentBlockIndex }, }); - this.blockIndex++; this.currentBlockType = null; } - private makeDelta(delta: DeltaShape): AnthropicSSERecord { + private makeDelta(delta: DeltaShape, index = this.currentBlockIndex): AnthropicSSERecord { return { event: "content_block_delta", - data: { type: "content_block_delta", index: this.blockIndex, delta }, + data: { type: "content_block_delta", index, delta }, }; } @@ -529,6 +551,7 @@ export class AnthropicStreamEncoder { this.started = true; } this.closeCurrentBlock(records); + this.closeToolBlocks(records); records.push({ event: "message_delta", data: { @@ -555,6 +578,9 @@ export function buildAnthropicResponse( for (const event of events) { switch (event.type) { + case "error": + // Do not turn failed generations (or private diagnostics) into content. + throw new Error("CC upstream generation failed"); case "text-delta": textContent += (event.data.text as string) ?? ""; break; diff --git a/src/translate/openai.ts b/src/translate/openai.ts index 3c501a2..14f8b82 100644 --- a/src/translate/openai.ts +++ b/src/translate/openai.ts @@ -1,4 +1,5 @@ import crypto from "node:crypto"; +import { toolArgumentSuffix } from "./tool-arguments.js"; import type { OpenAIChatRequest, OpenAIMessage, @@ -227,6 +228,26 @@ export class OpenAIStreamEncoder { // `index` field would all be assigned index 0 and the client would merge // them into a single tool call. private readonly toolCallIdToIndex = new Map(); + private readonly toolArguments = new Map(); + private readonly toolMetadata = new Map(); + + private toolMetadataDelta(index: number, id?: string, name?: string) { + const seen = this.toolMetadata.get(index) ?? {}; + const delta: { id?: string; type?: string; name?: string } = {}; + for (const [field, value] of [["id", id], ["name", name]] as const) { + if (!value) continue; + if (seen[field] && seen[field] !== value) { + throw new Error("Inconsistent upstream tool metadata"); + } + if (!seen[field]) { + seen[field] = value; + delta[field] = value; + if (field === "id") delta.type = "function"; + } + } + this.toolMetadata.set(index, seen); + return delta; + } constructor(private readonly model: string) { this.id = crypto.randomUUID(); @@ -314,13 +335,18 @@ export class OpenAIStreamEncoder { index: this.resolveToolCallIndex(toolCallId, upstreamIndex), function: { arguments: (event.data.arguments as string) ?? "" }, }; - if (toolCallId) { - tc.id = toolCallId; - tc.type = "function"; + const metadata = this.toolMetadataDelta(tc.index, toolCallId, event.data.name as string | undefined); + if (metadata.id) { + tc.id = metadata.id; + tc.type = metadata.type; } - if (event.data.name) { - tc.function.name = event.data.name as string; + if (metadata.name) { + tc.function.name = metadata.name; } + this.toolArguments.set( + tc.index, + (this.toolArguments.get(tc.index) ?? "") + tc.function.arguments, + ); chunks.push({ id, object: "chat.completion.chunk", @@ -343,6 +369,10 @@ export class OpenAIStreamEncoder { toolCallId || undefined, (event.data.index as number) ?? undefined, ); + const emitted = this.toolArguments.get(index) ?? ""; + const suffix = toolArgumentSuffix(emitted, args); + this.toolArguments.set(index, emitted + suffix); + const { name, ...identity } = this.toolMetadataDelta(index, toolCallId, toolName); chunks.push({ id, object: "chat.completion.chunk", @@ -355,9 +385,8 @@ export class OpenAIStreamEncoder { tool_calls: [ { index, - id: toolCallId, - type: "function", - function: { name: toolName, arguments: args }, + ...identity, + function: { ...(name ? { name } : {}), arguments: suffix }, }, ], }, @@ -495,6 +524,9 @@ export function buildNonStreamingResponse(events: CCEvent[], model: string, id: for (const event of events) { switch (event.type) { + case "error": + // Do not turn failed generations (or private diagnostics) into content. + throw new Error("CC upstream generation failed"); case "text-delta": content += (event.data.text as string) ?? ""; break; diff --git a/src/translate/tool-arguments.ts b/src/translate/tool-arguments.ts new file mode 100644 index 0000000..7dfe49c --- /dev/null +++ b/src/translate/tool-arguments.ts @@ -0,0 +1,36 @@ +import { isDeepStrictEqual } from "node:util"; + +/** Reconcile canonical arguments with bytes already sent to an SDK. */ +export function toolArgumentSuffix(emitted: string, canonical: string): string { + if (canonical.startsWith(emitted)) return canonical.slice(emitted.length); + try { + // Canonical objects may be serialized with different whitespace/key order. + if (isDeepStrictEqual(JSON.parse(emitted), JSON.parse(canonical))) return ""; + } catch { + // Incomplete JSON may still match after ignoring insignificant whitespace. + } + let offset = 0; + let inString = false; + let escaped = false; + for (const char of emitted) { + if (!inString) { + if (/[ \t\r\n]/.test(char)) continue; + while (offset < canonical.length && /[ \t\r\n]/.test(canonical[offset])) offset++; + } + if (!canonical.startsWith(char, offset)) { + throw new Error("Inconsistent upstream tool arguments"); + } + offset += char.length; + if (escaped) escaped = false; + else if (inString && char === "\\") escaped = true; + else if (char === '"') inString = !inString; + } + const suffix = canonical.slice(offset); + try { + // Skipping whitespace must not splice a number/literal token, e.g. `1 2`. + if (isDeepStrictEqual(JSON.parse(emitted + suffix), JSON.parse(canonical))) return suffix; + } catch { + // The original bytes plus the suffix must still form valid JSON. + } + throw new Error("Inconsistent upstream tool arguments"); +} diff --git a/src/translate/validation.ts b/src/translate/validation.ts index 73754b6..bc62457 100644 --- a/src/translate/validation.ts +++ b/src/translate/validation.ts @@ -43,6 +43,53 @@ export function validateOpenAIChatRequest(body: unknown): OpenAIChatRequest { // ── Anthropic validation ── +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +/** Validate only fields consumed by the local estimator, not generation fields. */ +export function validateCountTokensRequest(body: unknown): Record { + if (!isRecord(body)) throw new ValidationError("Request body must be a JSON object"); + if (body.system !== undefined && typeof body.system !== "string") { + if ( + !Array.isArray(body.system) || + !body.system.every((b) => isRecord(b) && (b.text === undefined || typeof b.text === "string")) + ) { + throw new ValidationError("Field 'system' must be a string or array of text blocks"); + } + } + if (body.messages !== undefined) { + if ( + !Array.isArray(body.messages) || + !body.messages.every( + (m) => + isRecord(m) && + (typeof m.content === "string" || + (Array.isArray(m.content) && m.content.every(isRecord))), + ) + ) { + throw new ValidationError( + "Field 'messages' must contain objects with string or block-array content", + ); + } + } + if (body.tools !== undefined) { + if ( + !Array.isArray(body.tools) || + !body.tools.every( + (t) => + isRecord(t) && + (t.name === undefined || typeof t.name === "string") && + (t.description === undefined || typeof t.description === "string") && + (t.input_schema === undefined || isRecord(t.input_schema)), + ) + ) { + throw new ValidationError("Field 'tools' must be an array of tool objects"); + } + } + return body; +} + const UNSUPPORTED_CONTENT_TYPES = new Set([ "document", "search_result", diff --git a/src/upstream.ts b/src/upstream.ts index f4a45ef..25c22e4 100644 --- a/src/upstream.ts +++ b/src/upstream.ts @@ -8,7 +8,7 @@ interface UpstreamOptions { apiBase: string; apiKey: string; ccVersion: string; - /** Wall-clock timeout for receiving response headers + first byte. */ + /** Per-attempt deadline for headers and any non-2xx error body. */ timeoutMs?: number; /** Max ms allowed between consecutive data chunks during streaming. */ idleTimeoutMs?: number; @@ -77,6 +77,45 @@ function generateTraceparent(): string { */ const MAX_RETRIES = 2; const RETRY_BACKOFF_MS = 500; +const MAX_ERROR_BODY_BYTES = 16 * 1024; + +/** Bound diagnostics independently of how (or whether) the peer ends its body. */ +async function readErrorBody(response: Response, signal: AbortSignal): Promise { + if (!response.body) return ""; + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let size = 0; + let onAbort = (): void => {}; + const aborted = new Promise((_resolve, reject) => { + onAbort = () => reject(new Error("Error body read aborted")); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + }); + try { + while (true) { + const { done, value } = await Promise.race([reader.read(), aborted]); + if (done) return Buffer.concat(chunks).toString("utf8"); + if (size + value.byteLength >= MAX_ERROR_BODY_BYTES) { + // Do not return a prefix that could end halfway through a secret. + return "[error body truncated]"; + } + chunks.push(value); + size += value.byteLength; + } + } catch { + return "[error body unavailable]"; + } finally { + signal.removeEventListener("abort", onAbort); + // Never wait on an uncooperative underlying cancel implementation. + void reader.cancel?.().catch(() => {}); + reader.releaseLock?.(); + } +} + +function sanitizeErrorText(text: string, apiKey: string): string { + const redacted = apiKey ? text.replaceAll(apiKey, "[redacted]") : text; + return redacted.replace(/[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/g, ""); +} function sleep(ms: number, signal?: AbortSignal): Promise { return new Promise((resolve) => { @@ -106,6 +145,7 @@ export async function sendToCC( let lastError: UpstreamError | null = null; for (let attempt = 1; attempt <= MAX_RETRIES + 1; attempt++) { + if (signal?.aborted) throw new UpstreamError("Request aborted", 0, false); // Per-attempt timeout so one dead connection can't burn the whole budget. const controller = new AbortController(); const combinedSignal = signal ? combineSignals(signal, controller.signal) : controller.signal; @@ -118,13 +158,13 @@ export async function sendToCC( body: JSON.stringify(body), signal: combinedSignal, }); - clearTimeout(timeout); - if (!response.ok) { - const errorText = await response.text().catch(() => ""); + const errorText = await readErrorBody(response, combinedSignal); + clearTimeout(timeout); + if (signal?.aborted) throw new UpstreamError("Request aborted", 0, false); const retryable = response.status >= 500 || response.status === 429; lastError = new UpstreamError( - `CC API ${response.status}: ${errorText || response.statusText}`, + `CC API ${response.status}: ${sanitizeErrorText(errorText || response.statusText, apiKey)}`, response.status, retryable, ); @@ -136,6 +176,8 @@ export async function sendToCC( throw lastError; } + // Successful generation remains governed by the separate idle timeout. + clearTimeout(timeout); if (!response.body) { throw new UpstreamError("CC API returned no body", 0, true); } @@ -148,14 +190,13 @@ export async function sendToCC( }; } catch (err) { clearTimeout(timeout); + if (signal?.aborted) throw new UpstreamError("Request aborted", 0, false); if (err instanceof UpstreamError) throw err; // Distinguish a client-initiated abort (caller is gone — never retry, it // only wastes a request) from a timeout/network blip (retryable). const aborted = (err as Error).name === "AbortError"; - if (aborted && signal?.aborted) { - throw new UpstreamError("Request aborted", 0, true); - } + lastError = new UpstreamError( aborted ? "Upstream timeout" : `Upstream request failed: ${(err as Error).message}`, 0, @@ -245,11 +286,9 @@ function nodeReaderToStream( 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(() => {}); - } + // Native reader.cancel() resolves pending reads as EOF. Destroy the + // Node stream explicitly so consumers see an error, not silent success. + stream.destroy(err); }, idleMs); // Don't keep the event loop alive just for the idle timer. idleTimer.unref?.(); @@ -264,13 +303,13 @@ function nodeReaderToStream( // 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 => { + const releaseReader = (reason?: Error | null): void => { disarmIdle(); if (readerReleased) return; readerReleased = true; - const cancel = (reader as { cancel?: () => Promise }).cancel; + const cancel = (reader as { cancel?: (reason?: unknown) => Promise }).cancel; if (typeof cancel === "function") { - cancel.call(reader).catch(() => { + cancel.call(reader, reason).catch(() => { /* already closed */ }); } @@ -280,7 +319,7 @@ function nodeReaderToStream( objectMode: true, emitClose: true, destroy(err, cb) { - releaseReader(); + releaseReader(err); cb(err); }, async read() { @@ -304,6 +343,7 @@ function nodeReaderToStream( armIdle(); const { done, value } = await reader.read(); disarmIdle(); + if (this.destroyed) return; if (done) { upstreamDone = true; releaseReader(); @@ -337,7 +377,7 @@ function nodeReaderToStream( if (opts.abortSignal) { const sig = opts.abortSignal; if (sig.aborted) { - releaseReader(); + stream.destroy(new Error("Client disconnected")); } else { sig.addEventListener( "abort", diff --git a/tests/reliability-http.test.ts b/tests/reliability-http.test.ts new file mode 100644 index 0000000..810276b --- /dev/null +++ b/tests/reliability-http.test.ts @@ -0,0 +1,279 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import http from "node:http"; +import { once } from "node:events"; +import { createServer } from "@/server.js"; +import type { CCEvent } from "@/translate/types.js"; + +// Synthetic upstream fixtures only. Real HTTP clients use ephemeral loopback ports. +const fixtureBase = "https://upstream.invalid"; +const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); +const servers: http.Server[] = []; +afterEach(async () => { + vi.restoreAllMocks(); + for (const server of servers.splice(0)) { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + } +}); +async function listen() { + const server = createServer({ + host: "127.0.0.1", + port: 0, + apiKey: null, + ccApiBase: fixtureBase, + ccVersion: "0.0.0", + logLevel: "error", + corsOrigin: "", + upstreamTimeoutMs: 1000, + idleTimeoutMs: 0, + }); + servers.push(server); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + return (server.address() as { port: number }).port; +} +function post(port: number, path: string, body: unknown) { + const req = http.request({ + host: "127.0.0.1", + port, + path, + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: "Bearer synthetic-fixture-only", + }, + }); + req.end(JSON.stringify(body)); + return req; +} +async function jsonPost(port: number, path: string, body: unknown) { + const [res] = (await once(post(port, path, body), "response")) as [http.IncomingMessage]; + let text = ""; + for await (const chunk of res) text += chunk; + return { status: res.statusCode, body: JSON.parse(text) }; +} +const paths = ["/v1/chat/completions", "/v1/messages"]; +const requestBody = (stream: boolean) => ({ + model: "fixture-model", + max_tokens: 16, + messages: [{ role: "user", content: "fixture" }], + stream, +}); +function fixtureFetch(body: ReadableStream, capture?: (signal: AbortSignal) => void) { + return vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => { + if (String(url) !== `${fixtureBase}/alpha/generate`) throw new Error("Unexpected fixture URL"); + capture?.(init!.signal!); + return new Response(body); + }); +} +const bytes = (events: CCEvent[]) => + new TextEncoder().encode(events.map((e) => JSON.stringify(e)).join("\n") + "\n"); + +describe.each(paths)("real HTTP reliability: %s", (path) => { + it.each([true, false])( + "cancels upstream on a disconnected client (stream=%s)", + async (stream) => { + let controller!: ReadableStreamDefaultController; + const cancel = vi.fn(); + let signal!: AbortSignal; + const upstream = new ReadableStream({ + start(c) { + controller = c; + }, + cancel, + }); + const fetchSpy = fixtureFetch(upstream, (s) => { + signal = s; + }); + const port = await listen(); + const req = post(port, path, requestBody(stream)); + req.on("error", () => {}); // Expected client-side socket hangup in JSON mode. + try { + await vi.waitFor(() => expect(fetchSpy).toHaveBeenCalledOnce()); + // The uploaded request is complete, but generation must remain alive. + await delay(30); + expect(signal.aborted).toBe(false); + expect(cancel).not.toHaveBeenCalled(); + if (stream) { + const response = once(req, "response"); + controller.enqueue(bytes([{ type: "text-delta", data: { text: "fixture" } }])); + const [res] = (await response) as [http.IncomingMessage]; + await once(res, "data"); + res.destroy(); + } else { + req.destroy(); + } + await vi.waitFor( + () => { + expect(signal.aborted).toBe(true); + expect(cancel).toHaveBeenCalledOnce(); + }, + { timeout: 500 }, + ); + expect(fetchSpy).toHaveBeenCalledOnce(); + } finally { + req.destroy(); + if (!cancel.mock.calls.length) controller.close(); + } + }, + ); + + it("settles a backpressured SSE write when the client closes", async () => { + let controller!: ReadableStreamDefaultController; + const cancel = vi.fn(); + fixtureFetch( + new ReadableStream({ + start(c) { + controller = c; + }, + cancel, + }), + ); + const port = await listen(); + let response!: http.ServerResponse; + servers[servers.length - 1].prependListener("request", (_req, res) => { + response = res; + const write = res.write.bind(res); + // Synthetic backpressure; actual bytes and disconnect use real HTTP. + res.write = ((chunk: string) => { + write(chunk); + return false; + }) as typeof res.write; + }); + const req = post(port, path, requestBody(true)); + req.on("error", () => {}); + try { + const pending = once(req, "response"); + controller.enqueue(bytes([{ type: "text-delta", data: { text: "fixture" } }])); + const [res] = (await pending) as [http.IncomingMessage]; + await once(res, "data"); + expect(response.listenerCount("drain")).toBe(1); + res.destroy(); + await vi.waitFor(() => expect(response.destroyed).toBe(true)); + expect(response.listenerCount("drain")).toBe(0); + } finally { + req.destroy(); + if (!cancel.mock.calls.length) controller.close(); + } + }); + + it("surfaces inconsistent canonical tool arguments as a streaming error", async () => { + fixtureFetch( + new ReadableStream({ + start(c) { + c.enqueue( + bytes([ + { + type: "tool-call-delta", + data: { toolCallId: "a", name: "fixture_tool", arguments: '{"x":1}' }, + }, + { + type: "tool-call", + data: { toolCallId: "a", toolName: "fixture_tool", input: { x: 2 } }, + }, + ]), + ); + c.close(); + }, + }), + ); + const [res] = (await once(post(await listen(), path, requestBody(true)), "response")) as [ + http.IncomingMessage, + ]; + let text = ""; + for await (const chunk of res) text += chunk; + expect(text).toContain("Inconsistent upstream tool arguments"); + expect(text).toContain(path === "/v1/messages" ? "event: error" : "[upstream error]"); + }); + + it.each([true, false])( + "does not abort a normally completed response (stream=%s)", + async (stream) => { + let signal!: AbortSignal; + fixtureFetch( + new ReadableStream({ + start(c) { + c.enqueue( + bytes([ + { type: "text-delta", data: { text: "fixture" } }, + { type: "finish", data: { finishReason: "stop" } }, + ]), + ); + c.close(); + }, + }), + (s) => { + signal = s; + }, + ); + const port = await listen(); + const [res] = (await once(post(port, path, requestBody(stream)), "response")) as [ + http.IncomingMessage, + ]; + let text = ""; + for await (const chunk of res) text += chunk; + await delay(10); + expect(res.statusCode).toBe(200); + expect(text).toContain("fixture"); + expect(signal.aborted).toBe(false); + }, + ); + + it("returns a protocol-shaped 502 for a nonstream upstream error event", async () => { + fixtureFetch( + new ReadableStream({ + start(c) { + c.enqueue( + bytes([ + { type: "text-delta", data: { text: "partial" } }, + { type: "error", data: { message: "synthetic-private-diagnostic" } }, + ]), + ); + c.close(); + }, + }), + ); + const response = await jsonPost(await listen(), path, requestBody(false)); + expect(response.status).toBe(502); + expect(response.body.error.message).toBe("CC upstream generation failed"); + expect(JSON.stringify(response.body)).not.toContain("synthetic-private-diagnostic"); + if (path === "/v1/messages") expect(response.body.type).toBe("error"); + }); +}); + +describe("count_tokens input shapes", () => { + it.each([ + null, + [], + "text", + 1, + { system: null }, + { system: [null] }, + { system: [{ text: 1 }] }, + { messages: {} }, + { messages: [null] }, + { messages: [{ content: null }] }, + { messages: [{ content: [null] }] }, + { tools: {} }, + { tools: [null] }, + { tools: [{ name: 1 }] }, + { tools: [{ description: [] }] }, + { tools: [{ input_schema: [] }] }, + ])("returns Anthropic 400 for invalid shape %#", async (body) => { + const response = await jsonPost(await listen(), "/v1/messages/count_tokens", body); + expect(response.status).toBe(400); + expect(response.body).toMatchObject({ + type: "error", + error: { type: "invalid_request_error" }, + }); + }); + + it.each([{}, { system: "abcd", messages: [{ content: "abcd" }], tools: [] }])( + "preserves local estimates without requiring generation-only fields %#", + async (body) => { + const response = await jsonPost(await listen(), "/v1/messages/count_tokens", body); + expect(response.status).toBe(200); + expect(response.body.input_tokens).toBe("system" in body ? 2 : 0); + }, + ); +}); diff --git a/tests/reliability-translate.test.ts b/tests/reliability-translate.test.ts new file mode 100644 index 0000000..864bb35 --- /dev/null +++ b/tests/reliability-translate.test.ts @@ -0,0 +1,158 @@ +import { describe, expect, it } from "vitest"; +import { OpenAIStreamEncoder, buildNonStreamingResponse } from "@/translate/openai.js"; +import { AnthropicStreamEncoder, buildAnthropicResponse } from "@/translate/anthropic.js"; +import type { CCEvent } from "@/translate/types.js"; + +// Synthetic protocol fixtures. These reducers model SDK concatenation by index. +const delta = (id: string, args: string): CCEvent => ({ + type: "tool-call-delta", + data: { toolCallId: id, name: "fixture_tool", arguments: args }, +}); +const final = (id: string, input: unknown): CCEvent => ({ + type: "tool-call", + data: { toolCallId: id, toolName: "fixture_tool", input }, +}); +function encode(protocol: string, events: CCEvent[]) { + const encoder = + protocol === "OpenAI" + ? new OpenAIStreamEncoder("fixture") + : new AnthropicStreamEncoder("fixture"); + const records = [ + ...events, + { type: "finish", data: { finishReason: "tool-calls" } } as CCEvent, + ].flatMap((e) => encoder.emit(e)) as any[]; + const calls = new Map(); + for (const r of records) { + if (protocol === "OpenAI") { + for (const tc of r.choices?.[0]?.delta?.tool_calls ?? []) { + const call = calls.get(tc.index) ?? { id: tc.id, args: "" }; + call.args += tc.function?.arguments ?? ""; + calls.set(tc.index, call); + } + } else { + if (r.event === "content_block_start" && r.data.content_block.type === "tool_use") { + expect(calls.has(r.data.index)).toBe(false); + calls.set(r.data.index, { id: r.data.content_block.id, args: "", closed: false }); + } + if (r.event === "content_block_delta" && r.data.delta.type === "input_json_delta") { + const call = calls.get(r.data.index)!; + expect(call.closed).toBe(false); + call.args += r.data.delta.partial_json; + } + if (r.event === "content_block_stop" && calls.has(r.data.index)) { + const call = calls.get(r.data.index)!; + expect(call.closed).toBe(false); + call.closed = true; + } + } + } + if (protocol === "Anthropic") for (const call of calls.values()) expect(call.closed).toBe(true); + return [...calls.values()].map(({ id, args }) => ({ id, args })); +} + +describe("OpenAI SDK metadata accumulation", () => { + it.each([ + [delta("call-a", '{"x":1}'), final("call-a", { x: 1 })], + [delta("call-a", '{"x":'), delta("call-a", "1}"), final("call-a", { x: 1 })], + [final("call-a", { x: 1 }), final("call-a", { x: 1 })], + ])("emits complete id/name metadata only once per tool index (%#)", (...events) => { + const encoder = new OpenAIStreamEncoder("fixture"); + const accumulated = { id: "", name: "", arguments: "" }; + for (const event of events) { + for (const chunk of encoder.emit(event) as any[]) { + for (const call of chunk.choices?.[0]?.delta?.tool_calls ?? []) { + accumulated.id += call.id ?? ""; + accumulated.name += call.function?.name ?? ""; + accumulated.arguments += call.function?.arguments ?? ""; + } + } + } + expect(accumulated).toEqual({ id: "call-a", name: "fixture_tool", arguments: '{"x":1}' }); + }); +}); + +describe.each(["OpenAI", "Anthropic"])("%s tool argument reliability", (protocol) => { + it("does not duplicate complete streamed JSON in the canonical event", () => { + const calls = encode(protocol, [delta("a", '{"x":1}'), final("a", { x: 1 })]); + expect(calls).toEqual([{ id: "a", args: '{"x":1}' }]); + expect(JSON.parse(calls[0].args)).toEqual({ x: 1 }); + }); + it("emits only the missing canonical suffix", () => { + expect(encode(protocol, [delta("a", '{"x":'), final("a", { x: 1 })])).toEqual([ + { id: "a", args: '{"x":1}' }, + ]); + }); + it.each([ + ['{"x": ', { x: 1 }, '1}'], + [' \t{\r\n "x" \t: [ 1 , ', { x: [1, true] }, 'true]}'], + ['{"x": 1 \t', { x: 1 }, '}'], + ['{"x": tr', { x: true }, 'ue}'], + ['{"x": 1e', { x: 1e21 }, '+21}'], + ['{"x": "a ', { x: "a b" }, 'b"}'], + ['{"x": "a\\" ', { x: 'a" b' }, 'b"}'], + ['{"x": "a\\', { x: 'a" b' }, '" b"}'], + ['{"x": "a\\\\', { x: "a\\ b" }, ' b"}'], + ])("reconciles partial whitespace without changing emitted bytes (%#)", (prefix, input, suffix) => { + const calls = encode(protocol, [delta("a", prefix as string), final("a", input)]); + expect(calls).toEqual([{ id: "a", args: `${prefix}${suffix}` }]); + expect(JSON.parse(calls[0].args)).toEqual(input); + }); + it.each([ + ['{ "x": 2', { x: 1 }], + ['{"x": "a ', { x: "ab" }], + ['{"x": "a\\" ', { x: 'a"b' }], + ['{"x": 1 ', { x: 12 }], + ['{"x": 1 .', { x: 1.5 }], + ['{"x": 1e ', { x: 1e21 }], + ['{"x": tr ', { x: true }], + ['{"x": f al', { x: false }], + ['{"x": n u', { x: null }], + ['{"x":\u00a0', { x: 1 }], + ])("rejects inconsistent values or whitespace splitting JSON tokens (%#)", (prefix, input) => { + expect(() => encode(protocol, [delta("a", prefix as string), final("a", input)])).toThrow( + "Inconsistent upstream tool arguments", + ); + }); + it("preserves final-only tool calls", () => { + expect(encode(protocol, [final("a", { x: 1 })])).toEqual([{ id: "a", args: '{"x":1}' }]); + }); + it("keeps interleaved tool arguments on their original SDK indices", () => { + expect( + encode(protocol, [ + delta("a", '{"x":'), + delta("b", '{"y":'), + delta("a", "1}"), + final("a", { x: 1 }), + final("b", { y: 2 }), + ]), + ).toEqual([ + { id: "a", args: '{"x":1}' }, + { id: "b", args: '{"y":2}' }, + ]); + }); + it("rejects a final payload inconsistent with already emitted arguments", () => { + expect(() => encode(protocol, [delta("a", '{"x":1}'), final("a", { x: 2 })])).toThrow( + "Inconsistent upstream tool arguments", + ); + }); + it("accepts semantically equal complete JSON with different whitespace", () => { + expect(encode(protocol, [delta("a", '{ "x": 1 }'), final("a", { x: 1 })])).toEqual([ + { id: "a", args: '{ "x": 1 }' }, + ]); + }); + it("does not duplicate a repeated canonical event", () => { + expect(encode(protocol, [final("a", { x: 1 }), final("a", { x: 1 })])).toEqual([ + { id: "a", args: '{"x":1}' }, + ]); + }); + it("rejects error events when building nonstream responses", () => { + const build = protocol === "OpenAI" ? buildNonStreamingResponse : buildAnthropicResponse; + expect(() => + build( + [{ type: "error", data: { message: "synthetic-private-diagnostic" } }], + "fixture", + "id", + ), + ).toThrow("CC upstream generation failed"); + }); +}); diff --git a/tests/reliability-upstream.test.ts b/tests/reliability-upstream.test.ts new file mode 100644 index 0000000..181589e --- /dev/null +++ b/tests/reliability-upstream.test.ts @@ -0,0 +1,166 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { Readable } from "node:stream"; +import { collectEvents, sendToCC } from "@/upstream.js"; +import { toCCRequest } from "@/translate/openai.js"; + +// All fetches are synthetic fixtures; no provider requests or real keys. +const body = () => toCCRequest({ model: "fixture", messages: [] }); +const options = { + apiBase: "https://upstream.invalid", + apiKey: "synthetic-fixture-key", + ccVersion: "0.0.0", +}; +afterEach(() => vi.restoreAllMocks()); + +describe("native upstream stream reliability", () => { + it("propagates an idle timeout as an error, not native reader cancellation EOF", async () => { + const cancel = vi.fn(); + vi.spyOn(globalThis, "fetch").mockResolvedValue(new Response(new ReadableStream({ cancel }))); + const { stream } = await sendToCC(body(), { ...options, idleTimeoutMs: 30 }); + await expect(collectEvents(stream)).rejects.toMatchObject({ name: "IdleTimeoutError" }); + expect(cancel).toHaveBeenCalledOnce(); + expect(cancel.mock.calls[0][0]).toMatchObject({ name: "IdleTimeoutError" }); + }); + + it("bounds the non-2xx body wait by the per-attempt deadline", async () => { + let controller!: ReadableStreamDefaultController; + const cancel = vi.fn(); + vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + new ReadableStream({ + start(c) { + controller = c; + }, + cancel, + }), + { status: 400 }, + ), + ); + const result = sendToCC(body(), { ...options, timeoutMs: 30 }).catch((err) => err); + try { + const value = await Promise.race([ + result, + new Promise((r) => setTimeout(() => r("still pending"), 200)), + ]); + expect(value).toMatchObject({ statusCode: 400 }); + expect(cancel).toHaveBeenCalledOnce(); + } finally { + if (!cancel.mock.calls.length) controller.close(); + await result; + } + }); + + it("caps an oversized non-2xx body and cancels the unread remainder", async () => { + let controller!: ReadableStreamDefaultController; + const cancel = vi.fn(); + vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + new ReadableStream({ + start(c) { + controller = c; + c.enqueue(new TextEncoder().encode("x".repeat(128 * 1024))); + }, + cancel, + }), + { status: 400 }, + ), + ); + const result = sendToCC(body(), { ...options, timeoutMs: 1000 }).catch((err) => err); + try { + const value = await Promise.race([ + result, + new Promise((r) => setTimeout(() => r("still pending"), 200)), + ]); + expect(value).toMatchObject({ statusCode: 400 }); + expect(value.message.length).toBeLessThanOrEqual(16 * 1024 + 80); + expect(value.message).toContain("truncated"); + expect(cancel).toHaveBeenCalledOnce(); + } finally { + if (!cancel.mock.calls.length) controller.close(); + await result; + } + }); + + it("redacts the supplied fixture key and control characters in HTTP errors", async () => { + vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response(`bad\u001b[31m ${options.apiKey}`, { status: 400 }), + ); + const error = await sendToCC(body(), options).catch((err) => err); + expect(error.statusCode).toBe(400); + expect(error.message).not.toContain(options.apiKey); + expect(error.message).not.toContain("\u001b"); + expect(error.message).toContain("bad"); + }); + + it("does not start an upstream request for an already disconnected caller", async () => { + const fetchSpy = vi + .spyOn(globalThis, "fetch") + .mockRejectedValue(new Error("fixture must not run")); + const abort = new AbortController(); + abort.abort(); + await expect(sendToCC(body(), options, abort.signal)).rejects.toThrow("Request aborted"); + expect(fetchSpy).not.toHaveBeenCalled(); + }); + + it("does not retry after disconnect while reading a retryable error body", async () => { + const abort = new AbortController(); + const cancel = vi.fn(); + const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { + return new Response(new ReadableStream({ cancel }), { status: 503 }); + }); + const result = sendToCC(body(), { ...options, timeoutMs: 100 }, abort.signal).catch( + (err) => err, + ); + await new Promise((r) => setTimeout(r, 10)); + abort.abort(); + expect(await result).toMatchObject({ message: "Request aborted" }); + expect(fetchSpy).toHaveBeenCalledOnce(); + expect(cancel).toHaveBeenCalledOnce(); + }); + + it("does not retry a caller abort with a custom Error reason", async () => { + const abort = new AbortController(); + const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { + abort.abort(new Error("synthetic disconnect")); + throw abort.signal.reason; + }); + await expect(sendToCC(body(), options, abort.signal)).rejects.toThrow("Request aborted"); + expect(fetchSpy).toHaveBeenCalledOnce(); + }); + + it("rejects instead of returning silent EOF if the caller aborts as headers arrive", async () => { + const abort = new AbortController(); + const cancel = vi.fn(); + vi.spyOn(globalThis, "fetch").mockImplementation(async () => { + abort.abort(); + return new Response(new ReadableStream({ cancel })); + }); + const result = sendToCC(body(), options, abort.signal).then(({ stream }) => + collectEvents(stream), + ); + await expect(result).rejects.toThrow(/Request aborted|Client disconnected/); + expect(cancel).toHaveBeenCalledOnce(); + }); + + it("does not apply the header deadline to a successful generation body", async () => { + let controller!: ReadableStreamDefaultController; + vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + new ReadableStream({ + start(c) { + controller = c; + }, + }), + ), + ); + const { stream } = await sendToCC(body(), { ...options, timeoutMs: 10, idleTimeoutMs: 0 }); + const result = collectEvents(stream); + await new Promise((r) => setTimeout(r, 40)); + controller.enqueue( + new TextEncoder().encode('{"type":"finish","data":{"finishReason":"stop"}}\n'), + ); + controller.close(); + expect(await result).toMatchObject([{ type: "finish" }]); + expect((stream as Readable).destroyed).toBe(true); + }); +});