9router/open-sse/executors/commandcode.js

import { randomUUID } from "crypto";
import { BaseExecutor } from "./base.js";
import { PROVIDERS } from "../config/providers.js";
import { HTTP_STATUS } from "../config/runtimeConfig.js";
import { commandCodeToOpenAIResponse } from "../translator/response/commandcode-to-openai.js";
import { SSE_DONE } from "../utils/sseConstants.js";

/**
 * CommandCodeExecutor — talks to https://api.commandcode.ai/alpha/generate
 *
 * Auth: Bearer <user_xxx> API key (stored as the connection's apiKey).
 * Adds the per-request `x-session-id` header expected by CommandCode upstream.
 *
 * Upstream returns AI SDK v5 NDJSON (one JSON event per line, no `data:` prefix).
 * We translate each event to an OpenAI chat.completion.chunk and emit it as SSE so
 * both the streaming and non-streaming (forced SSE → JSON) downstream handlers in
 * 9router can consume it without further format translation.
 *
 * Terminal upstream failures arrive as `{"type":"error"}` events inside the HTTP
 * 200 stream, so a plain `response.ok` check cannot see them. We peek the first
 * events before committing the response (see peekForUpstreamError) so a stream
 * that starts with an error fails fast — the normal `!response.ok` path then
 * triggers account/model fallback instead of streaming fake success content.
 */
export class CommandCodeExecutor extends BaseExecutor {
	constructor() {
		super("commandcode", PROVIDERS.commandcode);
	}

	transformRequest(model, body, stream, credentials) {
		body.stream = true;
		return body;
	}

	buildHeaders(credentials, stream = true) {
		const headers = {
			"Content-Type": "application/json",
			...(this.config.headers || {}),
			"x-session-id": randomUUID(),
		};

		const token = credentials?.apiKey || credentials?.accessToken;
		if (token) headers["Authorization"] = `Bearer ${token}`;

		if (stream) headers["Accept"] = "text/event-stream";
		return headers;
	}

	async execute(opts) {
		const result = await super.execute(opts);
		if (!result?.response?.ok || !result.response.body) return result;
		result.response = await peekForUpstreamError(result.response, opts.model, {
			signal: opts.signal,
		});
		return result;
	}
}

// How long to hold the response open while peeking the first upstream events.
// An upstream error event ("Network connection lost") is emitted at stream
// start, so the peek is fast; the bound just prevents a slow-started stream
// from being held hostage. Env: COMMANDCODE_PEEK_TIMEOUT_MS.
const PEEK_TIMEOUT_MS = (() => {
	const raw = process.env.COMMANDCODE_PEEK_TIMEOUT_MS;
	const n = raw ? parseInt(raw, 10) : NaN;
	return Number.isFinite(n) && n > 0 ? n : 10 * 1000;
})();

// Event types that count as "the stream has started producing". Everything
// else (start, start-step, reasoning-start, text-start, ...) is metadata and
// does not end the peek.
const MEANINGFUL_EVENT_TYPES = new Set([
	"text-delta",
	"reasoning-delta",
	"tool-input-start",
	"tool-input-delta",
	"tool-input-end",
	"tool-call",
	"finish-step",
	"finish",
]);

function makeAbortError(reason) {
	const error = new Error(reason?.message || reason || "Request aborted");
	error.name = "AbortError";
	return error;
}

function tryParseEvent(line) {
	const trimmed = line.trim();
	if (!trimmed) return null;
	const json = trimmed.startsWith("data:") ? trimmed.slice(5).trim() : trimmed;
	if (!json || json === "[DONE]") return null;
	try {
		return JSON.parse(json);
	} catch {
		return null;
	}
}

function formatErrorValue(errVal) {
	const errStr =
		typeof errVal === "string"
			? errVal
			: typeof errVal?.message === "string"
				? errVal.message
				: JSON.stringify(errVal);
	const errType =
		typeof errVal === "string"
			? "upstream_error"
			: errVal?.type || "upstream_error";
	return { message: errStr, type: errType };
}

/**
 * Read the first upstream events before committing the response.
 *
 * - `{"type":"error"}` as the first meaningful event → return a 502 Response so
 *   chatCore's `!response.ok` path parses the error and triggers fallback.
 * - Otherwise → re-emit the buffered bytes + the rest of the stream through the
 *   normal NDJSON → OpenAI SSE wrapper and return it untouched in spirit.
 *
 * Bounded by `timeoutMs` (default PEEK_TIMEOUT_MS): if no meaningful event
 * arrives in time, or the request signal aborts, we commit whatever we have and
 * let the regular stream pipeline (stall detection, abort handling) take over.
 */
export async function peekForUpstreamError(
	originalResponse,
	model,
	{ signal = null, timeoutMs = PEEK_TIMEOUT_MS } = {},
) {
	const reader = originalResponse.body.getReader();
	const decoder = new TextDecoder();
	const encoder = new TextEncoder();
	const abortController = new AbortController();
	const forwardAbort = () => abortController.abort(signal?.reason);
	if (signal?.aborted) abortController.abort(signal?.reason);
	else if (signal)
		signal.addEventListener("abort", forwardAbort, { once: true });

	let peeked = "";
	let errorEvent = null;
	let committed = false;

	const readWithTimeout = (ms) => {
		if (abortController.signal.aborted) {
			return Promise.reject(makeAbortError(abortController.signal.reason));
		}
		const timeoutPromise = new Promise((_, reject) => {
			const t = setTimeout(() => reject(new Error("peek timeout")), ms);
			t.unref?.();
		});
		const abortPromise = new Promise((_, reject) => {
			abortController.signal.addEventListener(
				"abort",
				() => reject(makeAbortError(abortController.signal.reason)),
				{ once: true },
			);
		});
		return Promise.race([reader.read(), timeoutPromise, abortPromise]);
	};

	try {
		const deadline = Date.now() + timeoutMs;
		while (!errorEvent && !committed && Date.now() < deadline) {
			const { done, value } = await readWithTimeout(
				Math.max(deadline - Date.now(), 1),
			);
			if (done) break;
			peeked += decoder.decode(value, { stream: true });
			const lines = peeked.split("\n");
			// The last segment may be a partial line — only parse complete ones.
			for (const line of lines.slice(0, -1)) {
				const event = tryParseEvent(line);
				if (!event?.type) continue;
				if (event.type === "error") {
					errorEvent = event;
					break;
				}
				if (MEANINGFUL_EVENT_TYPES.has(event.type)) {
					committed = true;
					break;
				}
			}
		}
	} catch {
		// timeout / abort / read failure during the peek → commit whatever we have;
		// the downstream stream pipeline (stall detection, abort handling) takes over.
	}

	// Flush any partial multi-byte UTF-8 sequence held by the decoder so the
	// re-encoded peeked bytes round-trip losslessly.
	peeked += decoder.decode();

	if (signal) signal.removeEventListener("abort", forwardAbort);

	if (errorEvent) {
		await reader.cancel("commandcode early error detected").catch(() => {});
		const { message, type } = formatErrorValue(
			errorEvent.error ?? errorEvent.message ?? "unknown",
		);
		return new Response(JSON.stringify({ error: { message, type } }), {
			status: HTTP_STATUS.BAD_GATEWAY,
			statusText: message.slice(0, 200),
			headers: { "Content-Type": "application/json" },
		});
	}

	const remaining = new ReadableStream({
		start(controller) {
			(async () => {
				try {
					if (peeked) controller.enqueue(encoder.encode(peeked));
					while (true) {
						const { done, value } = await reader.read();
						if (done) break;
						controller.enqueue(value);
					}
					controller.close();
				} catch (err) {
					controller.error(err);
				}
			})();
		},
		cancel() {
			reader.cancel("commandcode stream cancelled").catch(() => {});
		},
	});

	const combined = new Response(remaining, {
		status: originalResponse.status,
		statusText: originalResponse.statusText,
		headers: originalResponse.headers,
	});
	return wrapNdjsonAsOpenAISse(combined, model);
}

function wrapNdjsonAsOpenAISse(originalResponse, model) {
	const decoder = new TextDecoder();
	const encoder = new TextEncoder();
	let buffer = "";
	const state = { model };

	const emitChunks = (chunks, controller) => {
		if (!chunks) return;
		const list = Array.isArray(chunks) ? chunks : [chunks];
		for (const c of list) {
			if (c == null) continue;
			controller.enqueue(encoder.encode(`data: ${JSON.stringify(c)}\n\n`));
		}
	};

	const transform = new TransformStream({
		transform(chunk, controller) {
			buffer += decoder.decode(chunk, { stream: true });
			const lines = buffer.split("\n");
			buffer = lines.pop() || "";
			for (const line of lines) {
				const trimmed = line.trim();
				if (!trimmed) continue;
				// Translate AI SDK v5 NDJSON line to one or more OpenAI chunks
				emitChunks(commandCodeToOpenAIResponse(trimmed, state), controller);
			}
		},
		flush(controller) {
			const trimmed = buffer.trim();
			if (trimmed) {
				emitChunks(commandCodeToOpenAIResponse(trimmed, state), controller);
			}
			controller.enqueue(encoder.encode(SSE_DONE));
		},
	});

	const newBody = originalResponse.body.pipeThrough(transform);
	return new Response(newBody, {
		status: originalResponse.status,
		statusText: originalResponse.statusText,
		headers: originalResponse.headers,
	});
}

export default CommandCodeExecutor;