feat: add runner tran passthrough
This commit is contained in:
Codex
+174
-28
@@ -1,9 +1,12 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { type UniDeskConfig } from "./config";
|
||||
import { type DebugDispatchCommand, isDebugDispatchCommand } from "./debug";
|
||||
import { summarizeMicroserviceHealthResponse, summarizeMicroserviceObservation, summarizeMicroserviceProxyResponse } from "./microservices";
|
||||
import { parseNetworkPerfOptions, runNetworkPerf } from "./network-perf";
|
||||
import { formatSshFailureHint, isSshSkillDiscoveryArgs, parseSshArgs, sshFailureHint } from "./ssh";
|
||||
import { formatSshFailureHint, isSshSkillDiscoveryArgs, parseSshInvocation, sshFailureHint } from "./ssh";
|
||||
import { codexJudgeQueryAsync, codexOutputQueryAsync, codexPrPreflightQueryAsync, codexQueuesQueryAsync, codexTaskQueryAsync, codexTasksQueryAsync, codexUnreadTriageAsync } from "./code-queue";
|
||||
import { runDecisionCenterCommandAsync } from "./decision-center";
|
||||
import {
|
||||
@@ -44,6 +47,7 @@ interface FetchJsonResult {
|
||||
status?: number;
|
||||
body?: unknown;
|
||||
error?: string;
|
||||
responseHeaders?: Record<string, string>;
|
||||
responseTruncated?: boolean;
|
||||
responseBytesRead?: number;
|
||||
responseContentLength?: string | null;
|
||||
@@ -99,6 +103,13 @@ function normalizeRemoteHostHint(raw: string | undefined): string | null {
|
||||
return value.replace(/\/+$/u, "");
|
||||
}
|
||||
|
||||
function remoteHttpClientMode(env: NodeJS.ProcessEnv = process.env): "curl" | "fetch" {
|
||||
const explicit = env.UNIDESK_REMOTE_HTTP_CLIENT?.trim().toLowerCase();
|
||||
if (explicit === "fetch") return "fetch";
|
||||
if (explicit === "curl") return "curl";
|
||||
return isCodeQueueRunnerEnv(env) ? "curl" : "fetch";
|
||||
}
|
||||
|
||||
function isCodeQueueRunnerEnv(env: NodeJS.ProcessEnv): boolean {
|
||||
return Boolean(env.CODE_QUEUE_SERVICE_ROLE || env.CODE_QUEUE_INSTANCE_ID || env.CODE_QUEUE_DEV_CONTAINER_MASTER_HOST || env.KUBERNETES_SERVICE_HOST);
|
||||
}
|
||||
@@ -293,6 +304,7 @@ function frontendBaseUrl(host: string, config: UniDeskConfig): string {
|
||||
}
|
||||
|
||||
async function readJson(url: string, init?: RequestInit, timeoutMs = 8000, maxResponseBytes = 5_000_000): Promise<FetchJsonResult> {
|
||||
if (remoteHttpClientMode() === "curl") return readJsonWithCurl(url, init, timeoutMs, maxResponseBytes);
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
@@ -339,7 +351,7 @@ async function readJson(url: string, init?: RequestInit, timeoutMs = 8000, maxRe
|
||||
if (responseTruncated) {
|
||||
body = { _unideskResponseTruncated: true, maxResponseBytes, bytesRead: bytes, contentLength: res.headers.get("content-length"), textPreview: text };
|
||||
}
|
||||
return { ok: res.ok, status: res.status, body, responseTruncated, responseBytesRead: bytes, responseContentLength: res.headers.get("content-length") };
|
||||
return { ok: res.ok, status: res.status, body, responseHeaders: responseHeadersRecord(res.headers), responseTruncated, responseBytesRead: bytes, responseContentLength: res.headers.get("content-length") };
|
||||
} catch (error) {
|
||||
return { ok: false, error: error instanceof Error ? error.message : String(error) };
|
||||
} finally {
|
||||
@@ -347,30 +359,145 @@ async function readJson(url: string, init?: RequestInit, timeoutMs = 8000, maxRe
|
||||
}
|
||||
}
|
||||
|
||||
function responseHeadersRecord(headers: Headers): Record<string, string> {
|
||||
const record: Record<string, string> = {};
|
||||
headers.forEach((value, key) => {
|
||||
record[key.toLowerCase()] = value;
|
||||
});
|
||||
return record;
|
||||
}
|
||||
|
||||
function requestHeaders(init?: RequestInit): Array<[string, string]> {
|
||||
const headers = new Headers(init?.headers);
|
||||
const output: Array<[string, string]> = [];
|
||||
headers.forEach((value, key) => output.push([key, value]));
|
||||
return output;
|
||||
}
|
||||
|
||||
async function runCurl(args: string[], timeoutMs: number): Promise<{ status: number | null; stdout: string; stderr: string; timedOut: boolean; error?: string }> {
|
||||
const child = spawn("curl", args, { stdio: ["ignore", "pipe", "pipe"] });
|
||||
const stdoutChunks: Buffer[] = [];
|
||||
const stderrChunks: Buffer[] = [];
|
||||
let timedOut = false;
|
||||
const timer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
child.kill("SIGTERM");
|
||||
}, timeoutMs + 1000);
|
||||
child.stdout?.on("data", (chunk) => stdoutChunks.push(Buffer.from(chunk)));
|
||||
child.stderr?.on("data", (chunk) => stderrChunks.push(Buffer.from(chunk)));
|
||||
return await new Promise((resolve) => {
|
||||
child.on("error", (error) => {
|
||||
clearTimeout(timer);
|
||||
resolve({ status: null, stdout: "", stderr: "", timedOut, error: error.message });
|
||||
});
|
||||
child.on("close", (status) => {
|
||||
clearTimeout(timer);
|
||||
resolve({
|
||||
status,
|
||||
stdout: Buffer.concat(stdoutChunks).toString("utf8"),
|
||||
stderr: Buffer.concat(stderrChunks).toString("utf8"),
|
||||
timedOut,
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function parseCurlResponseHeaders(raw: string): Record<string, string> {
|
||||
const blocks = raw.split(/\r?\n\r?\n/u).map((block) => block.trim()).filter(Boolean);
|
||||
const latest = blocks.at(-1) ?? "";
|
||||
const headers: Record<string, string> = {};
|
||||
for (const line of latest.split(/\r?\n/u).slice(1)) {
|
||||
const splitAt = line.indexOf(":");
|
||||
if (splitAt <= 0) continue;
|
||||
headers[line.slice(0, splitAt).trim().toLowerCase()] = line.slice(splitAt + 1).trim();
|
||||
}
|
||||
return headers;
|
||||
}
|
||||
|
||||
function decodeBoundedBody(buffer: Buffer, maxResponseBytes: number): { text: string; truncated: boolean; bytesRead: number } {
|
||||
if (buffer.byteLength <= maxResponseBytes) return { text: buffer.toString("utf8"), truncated: false, bytesRead: buffer.byteLength };
|
||||
return { text: buffer.subarray(0, maxResponseBytes).toString("utf8"), truncated: true, bytesRead: maxResponseBytes };
|
||||
}
|
||||
|
||||
async function readJsonWithCurl(url: string, init?: RequestInit, timeoutMs = 8000, maxResponseBytes = 5_000_000): Promise<FetchJsonResult> {
|
||||
const dir = await mkdtemp(path.join(tmpdir(), "unidesk-remote-http-"));
|
||||
const headersFile = path.join(dir, "headers.txt");
|
||||
const bodyFile = path.join(dir, "body.bin");
|
||||
const requestBodyFile = path.join(dir, "request-body.bin");
|
||||
try {
|
||||
const method = init?.method ?? (init?.body === undefined ? "GET" : "POST");
|
||||
const args = [
|
||||
"-sS",
|
||||
"--max-time", String(Math.max(1, Math.ceil(timeoutMs / 1000))),
|
||||
"-D", headersFile,
|
||||
"-o", bodyFile,
|
||||
"-w", "%{http_code}",
|
||||
"-X", method,
|
||||
];
|
||||
for (const [key, value] of requestHeaders(init)) args.push("-H", `${key}: ${value}`);
|
||||
if (init?.body !== undefined) {
|
||||
await writeFile(requestBodyFile, typeof init.body === "string" ? init.body : String(init.body));
|
||||
args.push("--data-binary", `@${requestBodyFile}`);
|
||||
}
|
||||
args.push(url);
|
||||
const curl = await runCurl(args, timeoutMs);
|
||||
if (curl.status !== 0) {
|
||||
const curlError = curl.error ?? curl.stderr.trim();
|
||||
return {
|
||||
ok: false,
|
||||
status: curl.stdout.trim().match(/^\d{3}$/u) ? Number(curl.stdout.trim()) : undefined,
|
||||
error: curl.timedOut ? `curl timed out after ${timeoutMs}ms` : (curlError.length > 0 ? curlError : `curl exited with ${curl.status}`),
|
||||
};
|
||||
}
|
||||
const status = Number(curl.stdout.trim());
|
||||
const headersText = await readFile(headersFile, "utf8").catch(() => "");
|
||||
const responseHeaders = parseCurlResponseHeaders(headersText);
|
||||
const bodyBuffer = await readFile(bodyFile).catch(() => Buffer.alloc(0));
|
||||
const decoded = decodeBoundedBody(bodyBuffer, maxResponseBytes);
|
||||
let body: unknown = null;
|
||||
try {
|
||||
body = decoded.text.length > 0 && !decoded.truncated ? JSON.parse(decoded.text) : null;
|
||||
} catch {
|
||||
body = { text: decoded.text };
|
||||
}
|
||||
if (decoded.truncated) {
|
||||
body = {
|
||||
_unideskResponseTruncated: true,
|
||||
maxResponseBytes,
|
||||
bytesRead: decoded.bytesRead,
|
||||
contentLength: responseHeaders["content-length"] ?? null,
|
||||
textPreview: decoded.text,
|
||||
};
|
||||
}
|
||||
return {
|
||||
ok: status >= 200 && status < 300,
|
||||
status,
|
||||
body,
|
||||
responseHeaders,
|
||||
responseTruncated: decoded.truncated,
|
||||
responseBytesRead: decoded.bytesRead,
|
||||
responseContentLength: responseHeaders["content-length"] ?? null,
|
||||
};
|
||||
} catch (error) {
|
||||
return { ok: false, error: error instanceof Error ? error.message : String(error) };
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
async function loginFrontend(host: string, config: UniDeskConfig): Promise<FrontendSession> {
|
||||
const baseUrl = frontendBaseUrl(host, config);
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), 8_000);
|
||||
let res: Response;
|
||||
try {
|
||||
res = await fetch(`${baseUrl}/login`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ username: config.auth.username, password: config.auth.password }),
|
||||
signal: controller.signal,
|
||||
});
|
||||
} catch (error) {
|
||||
throw new RemoteCliFailure("remote-proxy-missing", `frontend login request failed via ${baseUrl}: ${error instanceof Error ? error.message : String(error)}`, { baseUrl });
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
const body = await res.text();
|
||||
const res = await readJson(`${baseUrl}/login`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ username: config.auth.username, password: config.auth.password }),
|
||||
}, 8_000, 120_000);
|
||||
if (!res.ok) {
|
||||
const failureClassification: RemoteFailureClassification = res.status === 401 || res.status === 403 ? "auth-missing" : "remote-proxy-missing";
|
||||
throw new RemoteCliFailure(failureClassification, `frontend login failed via ${baseUrl}: status=${res.status} body=${body.slice(0, 300)}`, { baseUrl, status: res.status });
|
||||
throw new RemoteCliFailure(failureClassification, `frontend login failed via ${baseUrl}: status=${res.status ?? "unknown"} body=${JSON.stringify(res.body ?? res.error).slice(0, 300)}`, { baseUrl, status: res.status ?? null });
|
||||
}
|
||||
const cookie = res.headers.get("set-cookie")?.split(";")[0] ?? "";
|
||||
if (cookie.length === 0) throw new RemoteCliFailure("auth-missing", `frontend login via ${baseUrl} did not return a session cookie`, { baseUrl, status: res.status });
|
||||
const cookie = res.responseHeaders?.["set-cookie"]?.split(";")[0] ?? "";
|
||||
if (cookie.length === 0) throw new RemoteCliFailure("auth-missing", `frontend login via ${baseUrl} did not return a session cookie`, { baseUrl, status: res.status ?? null });
|
||||
return { baseUrl, cookie };
|
||||
}
|
||||
|
||||
@@ -824,11 +951,24 @@ async function remoteNetworkPerf(options: RemoteCliOptions, config: UniDeskConfi
|
||||
};
|
||||
}
|
||||
|
||||
async function runRemoteSshOverFrontend(session: FrontendSession, providerId: string | undefined, args: string[]): Promise<number> {
|
||||
if (!providerId) throw new Error("remote ssh requires provider id, for example: bun scripts/cli.ts --main-server-ip 74.48.78.17 ssh D601 hostname");
|
||||
const parsed = parseSshArgs(args);
|
||||
export function remoteSshFrontendPlanForTest(target: string, args: string[]): Record<string, unknown> {
|
||||
const invocation = parseSshInvocation(target, args);
|
||||
return {
|
||||
providerId: invocation.providerId,
|
||||
route: invocation.route,
|
||||
remoteCommand: invocation.parsed.remoteCommand,
|
||||
requiresStdin: invocation.parsed.requiresStdin,
|
||||
invocationKind: invocation.parsed.invocationKind,
|
||||
payloadCwd: invocation.route.plane === "host" ? invocation.route.workspace : null,
|
||||
};
|
||||
}
|
||||
|
||||
async function runRemoteSshOverFrontend(session: FrontendSession, target: string | undefined, args: string[]): Promise<number> {
|
||||
if (!target) throw new Error("remote ssh requires a route, for example: bun scripts/cli.ts --main-server-ip 74.48.78.17 ssh D601 hostname");
|
||||
const invocation = parseSshInvocation(target, args);
|
||||
const parsed = invocation.parsed;
|
||||
if (parsed.requiresStdin) {
|
||||
process.stderr.write("remote frontend transport does not stream stdin for ssh helper subcommands such as apply-patch or py; run the command on the main server or use --main-server-transport ssh\n");
|
||||
process.stderr.write("remote frontend transport does not stream stdin for ssh helper subcommands such as script, apply-patch or py; run the command on the main server, use --main-server-transport ssh, or use an argv/pod-route operation that does not need stdin\n");
|
||||
return 255;
|
||||
}
|
||||
if (parsed.remoteCommand === null) {
|
||||
@@ -843,9 +983,15 @@ async function runRemoteSshOverFrontend(session: FrontendSession, providerId: st
|
||||
const dispatch = await frontendJson(session, "/api/dispatch", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({
|
||||
providerId,
|
||||
providerId: invocation.providerId,
|
||||
command: "host.ssh",
|
||||
payload: { source: "cli-remote-ssh", mode: "exec", command: remoteCommand, timeoutMs: isSshSkillDiscoveryArgs(args) ? 30000 : 15000 },
|
||||
payload: {
|
||||
source: "cli-remote-ssh",
|
||||
mode: "exec",
|
||||
command: remoteCommand,
|
||||
...(invocation.route.plane === "host" && invocation.route.workspace !== null ? { cwd: invocation.route.workspace } : {}),
|
||||
timeoutMs: isSshSkillDiscoveryArgs(args) ? 30000 : 15000,
|
||||
},
|
||||
}),
|
||||
});
|
||||
const taskId = (dispatch as { body?: { taskId?: string } }).body?.taskId ?? "";
|
||||
@@ -863,7 +1009,7 @@ async function runRemoteSshOverFrontend(session: FrontendSession, providerId: st
|
||||
if (task?.status !== "succeeded") {
|
||||
if (stdout.length === 0 && stderr.length === 0) process.stderr.write(`${JSON.stringify({ taskId, task }, null, 2)}\n`);
|
||||
const exitCode = typeof result.exitCode === "number" ? result.exitCode : 255;
|
||||
const hint = sshFailureHint(providerId, parsed, exitCode, stderr.length > 0 ? stderr : String(task?.message ?? ""));
|
||||
const hint = sshFailureHint(invocation.providerId, parsed, exitCode, stderr.length > 0 ? stderr : String(task?.message ?? ""));
|
||||
if (hint !== null) process.stderr.write(formatSshFailureHint(hint));
|
||||
return exitCode;
|
||||
}
|
||||
|
||||
@@ -1012,6 +1012,7 @@ function parseK3sTargetOperation(route: ParsedSshRoute, args: string[]): ParsedS
|
||||
if (operation === "apply-patch" || operation === "patch") return buildK3sApplyPatchCommand([...targetArgs, ...operationArgs]);
|
||||
if (operation === "script") return { remoteCommand: buildK3sScriptCommand([...targetArgs, ...operationArgs]), requiresStdin: true, invocationKind: "helper" };
|
||||
if (operation === "logs") return { remoteCommand: buildK3sLogsCommand([...targetArgs, ...operationArgs]), requiresStdin: false, invocationKind: "helper" };
|
||||
if (operation === "argv") return { remoteCommand: buildK3sExecCommand([...targetArgs, ...k3sRouteCommandArgs(operationArgs)]), requiresStdin: false, invocationKind: "argv" };
|
||||
if (operation === "get" || operation === "describe") {
|
||||
return { remoteCommand: buildK3sTargetObjectCommand(operation, route, operationArgs), requiresStdin: false, invocationKind: "helper" };
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user