UniDesk Codex
467 lines
18 KiB
JavaScript
467 lines
18 KiB
JavaScript
#!/usr/bin/env node
|
|
import { randomBytes } from "node:crypto";
|
|
import { createServer, request as httpRequest } from "node:http";
|
|
import { request as httpsRequest } from "node:https";
|
|
import { Transform } from "node:stream";
|
|
|
|
const serviceId = process.env.OTEL_SERVICE_NAME || process.env.HWLAB_SERVICE_ID || "opencode-provider-proxy";
|
|
const port = parsePositiveInteger(process.env.PORT || process.env.HWLAB_OPENCODE_PROVIDER_PROXY_PORT, 4097);
|
|
const upstreamBaseUrl = new URL(requiredEnv("HWLAB_OPENCODE_PROVIDER_PROXY_UPSTREAM_BASE_URL"));
|
|
const publicBasePath = normalizeBasePath(process.env.HWLAB_OPENCODE_PROVIDER_PROXY_PUBLIC_BASE_PATH || "/v1");
|
|
const timeoutMs = parsePositiveInteger(process.env.HWLAB_OPENCODE_PROVIDER_PROXY_TIMEOUT_MS, 600000);
|
|
const otelEndpoint = otelTracesEndpoint(process.env);
|
|
const startedAt = new Date();
|
|
|
|
const hopByHopHeaders = new Set(["connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade"]);
|
|
|
|
const server = createServer((clientReq, clientRes) => {
|
|
const url = new URL(clientReq.url || "/", "http://opencode-provider-proxy.local");
|
|
if (url.pathname === "/health" || url.pathname === "/health/live" || url.pathname === "/health/readiness") {
|
|
sendJson(clientRes, 200, healthPayload());
|
|
return;
|
|
}
|
|
proxyProviderRequest(clientReq, clientRes, url).catch((error) => {
|
|
if (!clientRes.headersSent) {
|
|
sendJson(clientRes, 502, {
|
|
error: "opencode_provider_proxy_failed",
|
|
message: "OpenCode provider proxy failed",
|
|
reason: error instanceof Error ? error.message : String(error)
|
|
});
|
|
} else {
|
|
clientRes.destroy(error);
|
|
}
|
|
});
|
|
});
|
|
|
|
server.listen(port, "0.0.0.0", () => {
|
|
process.stdout.write(JSON.stringify({ serviceId, status: "listening", port, upstreamHost: upstreamBaseUrl.host, publicBasePath, valuesPrinted: false }) + "\n");
|
|
});
|
|
|
|
async function proxyProviderRequest(clientReq, clientRes, url) {
|
|
const trace = traceContextFromHeaders(clientReq.headers);
|
|
const startSpanId = randomOtelSpanId();
|
|
const started = process.hrtime.bigint();
|
|
const startTime = Date.now();
|
|
const target = providerTargetUrl(url);
|
|
const route = providerRoute(url.pathname);
|
|
const attrs = {
|
|
"http.request.method": clientReq.method || "GET",
|
|
"http.route": route,
|
|
"url.scheme": target.protocol.replace(":", ""),
|
|
"server.address": target.hostname,
|
|
"server.port": Number(target.port || (target.protocol === "https:" ? 443 : 80)),
|
|
"opencode.provider.upstream_host": upstreamBaseUrl.host,
|
|
"opencode.provider.public_base_path": publicBasePath,
|
|
valuesPrinted: false
|
|
};
|
|
|
|
emitOtelSpanAsync({
|
|
name: "opencode.provider.request.start",
|
|
traceId: trace.traceId,
|
|
spanId: startSpanId,
|
|
parentSpanId: trace.parentSpanId,
|
|
kind: 1,
|
|
startTimeMs: startTime,
|
|
endTimeMs: startTime,
|
|
attributes: { ...attrs, "opencode.provider.phase": "start" }
|
|
});
|
|
|
|
clientRes.setHeader("traceparent", traceparent(trace.traceId, startSpanId));
|
|
clientRes.setHeader("x-hwlab-otel-trace-id", trace.traceId);
|
|
|
|
await new Promise((resolve) => {
|
|
let settled = false;
|
|
let statusCode = 0;
|
|
let errorCode = "";
|
|
let timeout = null;
|
|
let streamStats = createOpenAIStreamStats(false);
|
|
const settle = () => {
|
|
if (settled) return;
|
|
settled = true;
|
|
if (timeout) clearTimeout(timeout);
|
|
const elapsedMs = Number((process.hrtime.bigint() - started) / 1000000n);
|
|
emitOtelSpanAsync({
|
|
name: "opencode.provider.request",
|
|
traceId: trace.traceId,
|
|
spanId: randomOtelSpanId(),
|
|
parentSpanId: startSpanId,
|
|
kind: 3,
|
|
startTimeMs: startTime,
|
|
endTimeMs: startTime + Math.max(elapsedMs, 0),
|
|
statusCode,
|
|
errorCode,
|
|
attributes: {
|
|
...attrs,
|
|
"http.response.status_code": statusCode || undefined,
|
|
"http.response.status_class": statusCode ? `${Math.floor(statusCode / 100)}xx` : "unknown",
|
|
"opencode.provider.elapsed_ms": elapsedMs,
|
|
"opencode.provider.timeout_ms": timeoutMs,
|
|
"opencode.provider.phase": errorCode ? "failed" : "complete",
|
|
...openAIStreamStatsAttributes(streamStats)
|
|
}
|
|
});
|
|
resolve();
|
|
};
|
|
|
|
const requestImpl = target.protocol === "https:" ? httpsRequest : httpRequest;
|
|
const upstreamReq = requestImpl({
|
|
protocol: target.protocol,
|
|
hostname: target.hostname,
|
|
port: target.port || (target.protocol === "https:" ? 443 : 80),
|
|
method: clientReq.method,
|
|
path: `${target.pathname}${target.search}`,
|
|
headers: providerRequestHeaders(clientReq.headers, target, trace, startSpanId)
|
|
}, (upstreamRes) => {
|
|
statusCode = upstreamRes.statusCode || 502;
|
|
const filterReasoning = shouldFilterOpenAIReasoning(route, upstreamRes.headers);
|
|
streamStats = createOpenAIStreamStats(filterReasoning);
|
|
clientRes.writeHead(statusCode, {
|
|
...providerResponseHeaders(upstreamRes.headers),
|
|
traceparent: traceparent(trace.traceId, startSpanId),
|
|
"x-hwlab-otel-trace-id": trace.traceId
|
|
});
|
|
const responseStream = filterReasoning ? upstreamRes.pipe(openAIReasoningFilterStream(streamStats)) : upstreamRes;
|
|
responseStream.pipe(clientRes);
|
|
upstreamRes.on("end", settle);
|
|
upstreamRes.on("close", settle);
|
|
upstreamRes.on("error", (error) => {
|
|
errorCode = error?.code || "upstream_response_error";
|
|
if (!clientRes.writableEnded) clientRes.destroy(error);
|
|
settle();
|
|
});
|
|
});
|
|
|
|
timeout = setTimeout(() => {
|
|
errorCode = "upstream_timeout";
|
|
upstreamReq.destroy(new Error(`opencode provider proxy timed out after ${timeoutMs}ms`));
|
|
}, timeoutMs);
|
|
|
|
upstreamReq.on("error", (error) => {
|
|
if (!errorCode) errorCode = error?.code || "upstream_request_error";
|
|
if (!clientRes.headersSent) {
|
|
statusCode = errorCode === "upstream_timeout" ? 504 : 502;
|
|
sendJson(clientRes, statusCode, {
|
|
error: "opencode_provider_upstream_error",
|
|
message: "OpenCode provider upstream request failed",
|
|
reason: error instanceof Error ? error.message : String(error),
|
|
traceId: trace.traceId,
|
|
valuesPrinted: false
|
|
});
|
|
} else if (!clientRes.writableEnded) {
|
|
clientRes.destroy(error);
|
|
}
|
|
settle();
|
|
});
|
|
|
|
clientReq.on("error", (error) => {
|
|
errorCode = error?.code || "client_request_error";
|
|
upstreamReq.destroy(error);
|
|
});
|
|
clientReq.pipe(upstreamReq);
|
|
});
|
|
}
|
|
|
|
function healthPayload() {
|
|
return { serviceId, status: "ok", port, upstreamHost: upstreamBaseUrl.host, publicBasePath, startedAt: startedAt.toISOString(), uptimeSeconds: Math.round((Date.now() - startedAt.getTime()) / 1000), otelEnabled: Boolean(otelEndpoint), valuesPrinted: false };
|
|
}
|
|
|
|
function providerTargetUrl(url) {
|
|
const suffix = stripBasePath(url.pathname, publicBasePath);
|
|
const basePath = upstreamBaseUrl.pathname.replace(/\/+$/u, "");
|
|
const nextPath = suffix ? `${basePath}/${suffix.replace(/^\/+/, "")}` : basePath || "/";
|
|
const target = new URL(upstreamBaseUrl.toString());
|
|
target.pathname = nextPath;
|
|
target.search = url.search;
|
|
return target;
|
|
}
|
|
|
|
function stripBasePath(pathname, basePath) {
|
|
if (!basePath || basePath === "/") return pathname;
|
|
if (pathname === basePath) return "/";
|
|
if (pathname.startsWith(`${basePath}/`)) return pathname.slice(basePath.length) || "/";
|
|
return pathname;
|
|
}
|
|
|
|
function providerRoute(pathname) {
|
|
const stripped = stripBasePath(pathname, publicBasePath);
|
|
if (/^\/chat\/completions\/?$/u.test(stripped)) return `${publicBasePath}/chat/completions`;
|
|
if (/^\/models\/?$/u.test(stripped)) return `${publicBasePath}/models`;
|
|
return `${publicBasePath}${stripped.startsWith("/") ? stripped : `/${stripped}`}`;
|
|
}
|
|
|
|
function shouldFilterOpenAIReasoning(route, headers) {
|
|
return route === `${publicBasePath}/chat/completions` && /^text\/event-stream(?:\s*;|$)/iu.test(firstHeaderValue(headers?.["content-type"]));
|
|
}
|
|
|
|
function createOpenAIStreamStats(filterEnabled) {
|
|
return {
|
|
filterEnabled: Boolean(filterEnabled),
|
|
dataLines: 0,
|
|
outputDataLines: 0,
|
|
doneLines: 0,
|
|
jsonErrors: 0,
|
|
choices: 0,
|
|
contentChunks: 0,
|
|
contentChars: 0,
|
|
reasoningChunks: 0,
|
|
reasoningChars: 0,
|
|
reasoningOnlyChoicesDropped: 0,
|
|
emptyDeltaChoicesDropped: 0,
|
|
finishChoices: 0,
|
|
usageChunks: 0
|
|
};
|
|
}
|
|
|
|
function openAIStreamStatsAttributes(stats) {
|
|
return {
|
|
"opencode.provider.sse.filter_enabled": stats.filterEnabled,
|
|
"opencode.provider.sse.data_lines": stats.dataLines,
|
|
"opencode.provider.sse.output_data_lines": stats.outputDataLines,
|
|
"opencode.provider.sse.done_lines": stats.doneLines,
|
|
"opencode.provider.sse.json_errors": stats.jsonErrors,
|
|
"opencode.provider.sse.choice_count": stats.choices,
|
|
"opencode.provider.sse.content_chunks": stats.contentChunks,
|
|
"opencode.provider.sse.content_chars": stats.contentChars,
|
|
"opencode.provider.sse.reasoning_chunks": stats.reasoningChunks,
|
|
"opencode.provider.sse.reasoning_chars": stats.reasoningChars,
|
|
"opencode.provider.sse.reasoning_only_choices_dropped": stats.reasoningOnlyChoicesDropped,
|
|
"opencode.provider.sse.empty_delta_choices_dropped": stats.emptyDeltaChoicesDropped,
|
|
"opencode.provider.sse.finish_choices": stats.finishChoices,
|
|
"opencode.provider.sse.usage_chunks": stats.usageChunks
|
|
};
|
|
}
|
|
|
|
function openAIReasoningFilterStream(stats = createOpenAIStreamStats(true)) {
|
|
let pending = "";
|
|
return new Transform({
|
|
transform(chunk, _encoding, callback) {
|
|
pending += chunk.toString("utf8");
|
|
let newlineIndex = pending.indexOf("\n");
|
|
while (newlineIndex >= 0) {
|
|
const line = pending.slice(0, newlineIndex + 1);
|
|
pending = pending.slice(newlineIndex + 1);
|
|
const next = filterOpenAISseLine(line, stats);
|
|
if (next) {
|
|
stats.outputDataLines += isOpenAIDataLine(next) ? 1 : 0;
|
|
this.push(next);
|
|
}
|
|
newlineIndex = pending.indexOf("\n");
|
|
}
|
|
callback();
|
|
},
|
|
flush(callback) {
|
|
if (pending) {
|
|
const next = filterOpenAISseLine(pending, stats);
|
|
if (next) {
|
|
stats.outputDataLines += isOpenAIDataLine(next) ? 1 : 0;
|
|
this.push(next);
|
|
}
|
|
}
|
|
callback();
|
|
}
|
|
});
|
|
}
|
|
|
|
function isOpenAIDataLine(line) {
|
|
return /^data:\s*/u.test(line);
|
|
}
|
|
|
|
function filterOpenAISseLine(line, stats = createOpenAIStreamStats(true)) {
|
|
const match = line.match(/^(data:\s*)(.*?)(\r?\n)?$/u);
|
|
if (!match) return line;
|
|
const [, prefix, payload, lineEnding = ""] = match;
|
|
const text = payload.trim();
|
|
if (!text) return line;
|
|
stats.dataLines += 1;
|
|
if (text === "[DONE]") {
|
|
stats.doneLines += 1;
|
|
return line;
|
|
}
|
|
try {
|
|
const parsed = JSON.parse(text);
|
|
if (parsed.usage != null) stats.usageChunks += 1;
|
|
const choices = Array.isArray(parsed.choices) ? parsed.choices.map((choice) => filterOpenAIChoiceReasoning(choice, stats)).filter(Boolean) : parsed.choices;
|
|
if (Array.isArray(parsed.choices)) {
|
|
if (choices.length === 0 && parsed.usage === null) return "";
|
|
parsed.choices = choices;
|
|
}
|
|
return `${prefix}${JSON.stringify(parsed)}${lineEnding}`;
|
|
} catch {
|
|
stats.jsonErrors += 1;
|
|
return line;
|
|
}
|
|
}
|
|
|
|
function filterOpenAIChoiceReasoning(choice, stats = createOpenAIStreamStats(true)) {
|
|
if (!choice || typeof choice !== "object") return choice;
|
|
const next = { ...choice };
|
|
stats.choices += 1;
|
|
if (next.finish_reason != null) stats.finishChoices += 1;
|
|
if (next.delta && typeof next.delta === "object") {
|
|
const delta = { ...next.delta };
|
|
const reasoningText = [delta.reasoning_content, delta.reasoning, delta.reasoning_details]
|
|
.filter((value) => typeof value === "string")
|
|
.join("");
|
|
if (reasoningText) {
|
|
stats.reasoningChunks += 1;
|
|
stats.reasoningChars += reasoningText.length;
|
|
}
|
|
if (typeof delta.content === "string" && delta.content.length > 0) {
|
|
stats.contentChunks += 1;
|
|
stats.contentChars += delta.content.length;
|
|
}
|
|
delete delta.reasoning_content;
|
|
delete delta.reasoning;
|
|
delete delta.reasoning_details;
|
|
if (delta.content === null) delete delta.content;
|
|
next.delta = delta;
|
|
}
|
|
if (next.message && typeof next.message === "object") {
|
|
next.message = { ...next.message };
|
|
delete next.message.reasoning_content;
|
|
delete next.message.reasoning;
|
|
delete next.message.reasoning_details;
|
|
}
|
|
const deltaEmpty = next.delta && typeof next.delta === "object" && Object.keys(next.delta).length === 0;
|
|
if (deltaEmpty && next.finish_reason == null && next.logprobs == null) {
|
|
stats.emptyDeltaChoicesDropped += 1;
|
|
stats.reasoningOnlyChoicesDropped += 1;
|
|
return null;
|
|
}
|
|
return next;
|
|
}
|
|
|
|
function providerRequestHeaders(headers, target, trace, parentSpanId) {
|
|
const result = {};
|
|
for (const [name, value] of Object.entries(headers)) {
|
|
const lower = name.toLowerCase();
|
|
if (hopByHopHeaders.has(lower) || lower === "host") continue;
|
|
if (value !== undefined) result[lower] = value;
|
|
}
|
|
result.host = target.host;
|
|
result.traceparent = traceparent(trace.traceId, parentSpanId);
|
|
result["x-hwlab-otel-trace-id"] = trace.traceId;
|
|
result["x-source-service-id"] = serviceId;
|
|
return result;
|
|
}
|
|
|
|
function providerResponseHeaders(headers) {
|
|
const result = {};
|
|
for (const [name, value] of Object.entries(headers)) {
|
|
const lower = name.toLowerCase();
|
|
if (hopByHopHeaders.has(lower) || lower === "content-length") continue;
|
|
if (value !== undefined) result[name] = value;
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function sendJson(response, statusCode, body) {
|
|
const payload = JSON.stringify(body);
|
|
response.writeHead(statusCode, { "content-type": "application/json; charset=utf-8", "content-length": Buffer.byteLength(payload) });
|
|
response.end(payload);
|
|
}
|
|
|
|
function emitOtelSpanAsync(span) {
|
|
if (!otelEndpoint) return;
|
|
setTimeout(() => {
|
|
emitOtelSpan(span).catch((error) => {
|
|
process.stderr.write(JSON.stringify({ event: "opencode-provider-proxy-otel-emit-failed", serviceId, error: error instanceof Error ? error.message : String(error), valuesPrinted: false }) + "\n");
|
|
});
|
|
}, 0);
|
|
}
|
|
|
|
async function emitOtelSpan(span) {
|
|
const controller = new AbortController();
|
|
const timeout = setTimeout(() => controller.abort(), 1500);
|
|
try {
|
|
await fetch(otelEndpoint, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(otelPayload(span)), signal: controller.signal });
|
|
} finally {
|
|
clearTimeout(timeout);
|
|
}
|
|
}
|
|
|
|
function otelPayload(span) {
|
|
return { resourceSpans: [{ resource: { attributes: otelAttributes({ "service.name": serviceId, "deployment.environment": process.env.HWLAB_ENVIRONMENT || "v03" }) }, scopeSpans: [{ scope: { name: "hwlab.opencode.provider_proxy", version: "1" }, spans: [{ traceId: span.traceId, spanId: span.spanId, parentSpanId: span.parentSpanId || undefined, name: span.name, kind: span.kind, startTimeUnixNano: unixNano(span.startTimeMs), endTimeUnixNano: unixNano(span.endTimeMs), attributes: otelAttributes({ ...span.attributes, "otel.trace_id": span.traceId }), status: span.errorCode || (span.statusCode && span.statusCode >= 500) ? { code: 2, message: span.errorCode || `http_${span.statusCode}` } : { code: 1 } }] }] }] };
|
|
}
|
|
|
|
function otelAttributes(values) {
|
|
const result = [];
|
|
for (const [key, value] of Object.entries(values || {})) {
|
|
if (value === undefined || value === null || value === "") continue;
|
|
if (typeof value === "number" && Number.isFinite(value)) result.push({ key, value: Number.isInteger(value) ? { intValue: String(value) } : { doubleValue: value } });
|
|
else if (typeof value === "boolean") result.push({ key, value: { boolValue: value } });
|
|
else result.push({ key, value: { stringValue: String(value) } });
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function unixNano(ms) {
|
|
return String(BigInt(Math.max(0, Math.trunc(ms))) * 1000000n);
|
|
}
|
|
|
|
function traceContextFromHeaders(headers) {
|
|
const parsed = parseTraceparent(headers.traceparent);
|
|
return { traceId: parsed?.traceId || normalizedTraceId(headers["x-hwlab-otel-trace-id"]) || randomOtelTraceId(), parentSpanId: parsed?.spanId || "" };
|
|
}
|
|
|
|
function parseTraceparent(value) {
|
|
const match = firstHeaderValue(value).trim().match(/^00-([0-9a-f]{32})-([0-9a-f]{16})-[0-9a-f]{2}$/u);
|
|
if (!match) return null;
|
|
const traceId = normalizedTraceId(match[1]);
|
|
const spanId = normalizedSpanId(match[2]);
|
|
return traceId && spanId ? { traceId, spanId } : null;
|
|
}
|
|
|
|
function traceparent(traceId, spanId) {
|
|
return `00-${traceId}-${spanId}-01`;
|
|
}
|
|
|
|
function randomOtelTraceId() {
|
|
const value = randomBytes(16).toString("hex");
|
|
return value === "00000000000000000000000000000000" ? "00000000000000000000000000000001" : value;
|
|
}
|
|
|
|
function randomOtelSpanId() {
|
|
const value = randomBytes(8).toString("hex");
|
|
return value === "0000000000000000" ? "0000000000000001" : value;
|
|
}
|
|
|
|
function normalizedTraceId(value) {
|
|
const text = firstHeaderValue(value).trim().toLowerCase();
|
|
return /^[0-9a-f]{32}$/u.test(text) && text !== "00000000000000000000000000000000" ? text : "";
|
|
}
|
|
|
|
function normalizedSpanId(value) {
|
|
const text = firstHeaderValue(value).trim().toLowerCase();
|
|
return /^[0-9a-f]{16}$/u.test(text) && text !== "0000000000000000" ? text : "";
|
|
}
|
|
|
|
function firstHeaderValue(value) {
|
|
if (Array.isArray(value)) return String(value[0] ?? "");
|
|
return String(value ?? "");
|
|
}
|
|
|
|
function otelTracesEndpoint(env) {
|
|
const explicit = String(env.HWLAB_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT || env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT || "").trim();
|
|
if (explicit) return explicit;
|
|
const base = String(env.HWLAB_OTEL_EXPORTER_OTLP_ENDPOINT || env.OTEL_EXPORTER_OTLP_ENDPOINT || "").trim();
|
|
return base ? `${base.replace(/\/+$/u, "")}/v1/traces` : "";
|
|
}
|
|
|
|
function normalizeBasePath(value) {
|
|
const text = String(value || "/").trim();
|
|
const normalized = `/${text.replace(/^\/+|\/+$/gu, "")}`;
|
|
return normalized === "/" ? "/" : normalized;
|
|
}
|
|
|
|
function parsePositiveInteger(value, fallback) {
|
|
const parsed = Number.parseInt(String(value ?? ""), 10);
|
|
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
|
|
}
|
|
|
|
function requiredEnv(name) {
|
|
const value = String(process.env[name] || "").trim();
|
|
if (!value) throw new Error(`${name} is required`);
|
|
return value;
|
|
}
|