import crypto from "node:crypto";
import { Readable } from "node:stream";
import { parseCCLine } from "@/stream.js";
import type { CCEvent, CCRequestBody } from "@/translate/types.js";
import { logger } from "@/logger.js";
interface UpstreamOptions {
apiBase: string;
apiKey: string;
ccVersion: string;
/** Per-attempt deadline for headers and any non-2xx error body. */
timeoutMs?: number;
/** Max ms allowed between consecutive data chunks during streaming. */
idleTimeoutMs?: number;
}
/**
* Build the header set the official Command Code CLI sends. CC's server
* inspects these and rejects requests that look like a proxy ("Proxy use
* detected") if any of the CLI-identifying headers are missing/stale.
*/
export function buildHeaders(
apiKey: string,
ccVersion: string,
body: CCRequestBody,
): Record {
const sessionId = body.threadId;
logger.debug(`Sending Authorization header (key length: ${apiKey.length})`);
return {
"Content-Type": "application/json",
Accept: "application/json, */*",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
Connection: "keep-alive",
"User-Agent": `commandcode-cli/${ccVersion} Node.js/${process.version}`,
Authorization: `Bearer ${apiKey}`,
"x-cli-environment": "production",
"x-command-code-version": ccVersion,
"x-session-id": sessionId,
"x-co-flag": "false",
"x-taste-learning": "false",
"x-project-slug": slugifyWorkingDir(body.config.workingDir as string),
traceparent: generateTraceparent(),
};
}
function slugifyWorkingDir(workingDir: string): string {
const base =
(workingDir || process.cwd()).split(/[/\\]/).filter(Boolean).pop() ?? "commandcode-proxy";
return (
base
.toLowerCase()
.replace(/[^a-z0-9-]/g, "-")
.slice(0, 40) || "commandcode-proxy"
);
}
function generateTraceparent(): string {
const traceId = crypto.randomBytes(16).toString("hex");
const parentId = crypto.randomBytes(8).toString("hex");
return `00-${traceId}-${parentId}-01`;
}
/**
* Send a request to the Command Code /alpha/generate endpoint and parse
* the NDJSON response into CCEvent objects.
*
* CC's upstream is always streaming; we force `params.stream = true` here
* regardless of the downstream client's `stream` flag. For non-streaming
* downstream requests, the caller drains the returned `stream` into events.
*
* Retryable failures (HTTP 5xx/429, timeouts, network errors) are retried up
* to MAX_RETRIES times with linear backoff but ONLY before the stream starts.
* A client-initiated abort (caller already disconnected) is never retried.
*
* Returns a Readable of parsed CCEvents. Callers MUST consume or destroy it.
*/
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) => {
const t = setTimeout(resolve, ms);
signal?.addEventListener(
"abort",
() => {
clearTimeout(t);
resolve();
},
{ once: true },
);
});
}
export async function sendToCC(
body: CCRequestBody,
options: UpstreamOptions,
signal?: AbortSignal,
): Promise {
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.
body.params.stream = true;
let lastError: UpstreamError | null = null;
for (let attempt = 1; attempt {
if (idleMs {
const err = new Error(`CC upstream idle timeout: no data for ${idleMs}ms`);
err.name = "IdleTimeoutError";
// 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?.();
};
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 = (reason?: Error | null): void => {
disarmIdle();
if (readerReleased) return;
readerReleased = true;
const cancel = (reader as { cancel?: (reason?: unknown) => Promise }).cancel;
if (typeof cancel === "function") {
cancel.call(reader, reason).catch(() => {
/* already closed */
});
}
};
const stream = new Readable({
objectMode: true,
emitClose: true,
destroy(err, cb) {
releaseReader(err);
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 (this.destroyed) return;
if (done) {
upstreamDone = true;
releaseReader();
// Flush trailing partial line (no newline terminator).
if (buffer.trim()) {
const result = parseCCLine(buffer);
buffer = "";
if (result.type === "event" && result.event) {
if (!this.push(result.event)) return; // backpressured; null next read
}
}
this.push(null);
return;
}
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() ?? "";
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) {
stream.destroy(new Error("Client disconnected"));
} else {
sig.addEventListener(
"abort",
() => {
disarmIdle();
stream.destroy(new Error("Client disconnected"));
},
{ once: true },
);
}
}
return stream;
}