Files
pikasTech-agentrun/scripts/src/cli.ts
T
2026-06-02 10:04:36 +08:00

487 lines
24 KiB
TypeScript

import { mkdir, readFile, rm, writeFile } from "node:fs/promises";
import { spawn } from "node:child_process";
import { closeSync, existsSync, openSync } from "node:fs";
import path from "node:path";
import { startManagerServer } from "../../src/mgr/server.js";
import { MemoryAgentRunStore } from "../../src/mgr/store.js";
import { ManagerClient } from "../../src/mgr/client.js";
import { runOnce } from "../../src/runner/run-once.js";
import { renderRunnerJobDryRun } from "../../src/runner/k8s-job.js";
import { renderCodexProviderSecretPlan } from "./secret-render.js";
import type { JsonRecord, JsonValue, RunRecord } from "../../src/common/types.js";
import { AgentRunError, errorToJson } from "../../src/common/errors.js";
import type { RunnerOnceOptions } from "../../src/runner/run-once.js";
import { isBackendProfile } from "../../src/common/backend-profiles.js";
interface ParsedArgs {
positional: string[];
flags: Map<string, string | boolean>;
}
export async function runCli(argv: string[]): Promise<void> {
try {
const result = await dispatch(parseArgs(argv));
print({ ok: true, data: result });
} catch (error) {
const status = error instanceof AgentRunError ? error.httpStatus : 1;
print({ ok: false, ...(error instanceof AgentRunError ? { failureKind: error.failureKind, message: error.message } : { failureKind: "infra-failed", message: error instanceof Error ? error.message : String(error) }), error: errorToJson(error) });
process.exitCode = status === 0 ? 1 : status > 255 ? 1 : status;
}
}
async function dispatch(args: ParsedArgs): Promise<JsonValue> {
const [group, command, id] = args.positional;
if (!group || group === "help") return help();
if (group === "server" && command === "start") return startServer(args);
if (group === "server" && command === "status") return serverStatus(args);
if (group === "server" && command === "logs") return serverLogs(args);
if (group === "server" && command === "stop") return stopServer(args);
if (group === "backends" && command === "list") return client(args).get("/api/v1/backends");
if (group === "secrets" && command === "codex" && id === "render") return renderCodexSecret(args);
if (group === "queue" && command === "submit") return submitQueueTask(args);
if (group === "queue" && command === "list") return listQueueTasks(args);
if (group === "queue" && command === "show" && id) return client(args).get(`/api/v1/queue/tasks/${encodeURIComponent(id)}`);
if (group === "queue" && command === "stats") return client(args).get(`/api/v1/queue/stats${queueQuery(args)}`);
if (group === "queue" && command === "commander") return client(args).get(`/api/v1/queue/commander${queueQuery(args)}`);
if (group === "queue" && command === "read" && id) return client(args).post(`/api/v1/queue/tasks/${encodeURIComponent(id)}/read`, { readerId: optionalFlag(args, "reader-id") ?? "cli" });
if (group === "queue" && command === "cancel" && id) return client(args).post(`/api/v1/queue/tasks/${encodeURIComponent(id)}/cancel`, cancelBody(args));
if (group === "queue" && command === "dispatch" && id) return dispatchQueueTask(args, id);
if (group === "queue" && command === "refresh" && id) return client(args).post(`/api/v1/queue/tasks/${encodeURIComponent(id)}/refresh`, {});
if (group === "runs" && command === "create") return client(args).post("/api/v1/runs", await jsonFile(args));
if (group === "runs" && command === "show" && id) return client(args).get(`/api/v1/runs/${encodeURIComponent(id)}`);
if (group === "runs" && command === "events" && id) return client(args).get(`/api/v1/runs/${encodeURIComponent(id)}/events?afterSeq=${flag(args, "after-seq", "0")}&limit=${flag(args, "limit", "100")}`);
if (group === "runs" && command === "result" && id) {
const commandId = optionalFlag(args, "command-id");
return client(args).get(`/api/v1/runs/${encodeURIComponent(id)}/result${commandId ? `?commandId=${encodeURIComponent(commandId)}` : ""}`);
}
if (group === "runs" && command === "cancel" && id) return client(args).post(`/api/v1/runs/${encodeURIComponent(id)}/cancel`, cancelBody(args));
if (group === "commands" && command === "create" && id) {
const body = await jsonFile(args);
if (!body.type) body.type = flag(args, "type", "turn");
const idempotencyKey = optionalFlag(args, "idempotency-key");
if (idempotencyKey) body.idempotencyKey = idempotencyKey;
return client(args).post(`/api/v1/runs/${encodeURIComponent(id)}/commands`, body);
}
if (group === "commands" && command === "show" && id) {
const runId = flag(args, "run-id", "");
if (!runId) throw new AgentRunError("schema-invalid", "commands show requires --run-id", { httpStatus: 2 });
return client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}/commands/${encodeURIComponent(id)}`);
}
if (group === "commands" && command === "result" && id) {
const runId = flag(args, "run-id", "");
if (!runId) throw new AgentRunError("schema-invalid", "commands result requires --run-id", { httpStatus: 2 });
return client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}/commands/${encodeURIComponent(id)}/result`);
}
if (group === "commands" && command === "cancel" && id) return client(args).post(`/api/v1/commands/${encodeURIComponent(id)}/cancel`, cancelBody(args));
if (group === "runner" && command === "start") {
const runId = flag(args, "run-id", "");
if (!runId) throw new AgentRunError("schema-invalid", "runner start requires --run-id", { httpStatus: 2 });
const options: RunnerOnceOptions = {
managerUrl: managerUrl(args),
runId,
};
const runnerId = optionalFlag(args, "runner-id");
const backend = optionalFlag(args, "backend");
const codexCommand = optionalFlag(args, "codex-command");
const codexHome = optionalFlag(args, "codex-home") ?? process.env.CODEX_HOME;
if (runnerId) options.runnerId = runnerId;
if (backend) {
if (!isBackendProfile(backend)) throw new AgentRunError("schema-invalid", `runner start --backend ${backend} is not supported in v0.1`, { httpStatus: 2 });
options.backendProfile = backend;
}
if (codexCommand) options.codexCommand = codexCommand;
if (codexHome) options.codexHome = codexHome;
const idleTimeoutMs = optionalFlag(args, "idle-timeout-ms");
const pollIntervalMs = optionalFlag(args, "poll-interval-ms");
if (idleTimeoutMs) options.idleTimeoutMs = Number(idleTimeoutMs);
if (pollIntervalMs) options.pollIntervalMs = Number(pollIntervalMs);
if (args.flags.get("one-shot") === true) options.oneShot = true;
return runOnce(options) as unknown as JsonValue;
}
if (group === "runner" && command === "job") return renderRunnerJob(args);
if (group === "runner" && command === "jobs") return listRunnerJobs(args);
if (group === "runner" && command === "job-status") return showRunnerJobStatus(args);
throw new AgentRunError("schema-invalid", `unsupported command: ${args.positional.join(" ")}`, { httpStatus: 2 });
}
async function listRunnerJobs(args: ParsedArgs): Promise<JsonValue> {
const runId = flag(args, "run-id", "");
if (!runId) throw new AgentRunError("schema-invalid", "runner jobs requires --run-id", { httpStatus: 2 });
const commandId = optionalFlag(args, "command-id");
return client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}/runner-jobs${commandId ? `?commandId=${encodeURIComponent(commandId)}` : ""}`);
}
async function submitQueueTask(args: ParsedArgs): Promise<JsonValue> {
const body = await jsonFile(args);
const idempotencyKey = optionalFlag(args, "idempotency-key");
if (idempotencyKey) body.idempotencyKey = idempotencyKey;
return client(args).post("/api/v1/queue/tasks", body);
}
async function listQueueTasks(args: ParsedArgs): Promise<JsonValue> {
const params = new URLSearchParams();
const queue = optionalFlag(args, "queue");
const state = optionalFlag(args, "state");
const cursor = optionalFlag(args, "cursor");
const limit = optionalFlag(args, "limit");
const updatedAfter = optionalFlag(args, "updated-after");
if (queue) params.set("queue", queue);
if (state) params.set("state", state);
if (cursor) params.set("cursor", cursor);
if (limit) params.set("limit", limit);
if (updatedAfter) params.set("updatedAfter", updatedAfter);
const query = params.toString();
return client(args).get(`/api/v1/queue/tasks${query ? `?${query}` : ""}`);
}
function queueQuery(args: ParsedArgs): string {
const queue = optionalFlag(args, "queue");
return queue ? `?queue=${encodeURIComponent(queue)}` : "";
}
async function dispatchQueueTask(args: ParsedArgs, taskId: string): Promise<JsonValue> {
const body = await optionalJsonFile(args);
const copy = (flagName: string, key = flagName.replace(/-([a-z])/gu, (_, letter: string) => letter.toUpperCase())): void => {
const value = optionalFlag(args, flagName);
if (value) body[key] = value;
};
copy("idempotency-key", "idempotencyKey");
copy("image");
copy("namespace");
copy("attempt-id", "attemptId");
copy("runner-id", "runnerId");
copy("source-commit", "sourceCommit");
copy("runner-manager-url", "managerUrl");
copy("service-account-name", "serviceAccountName");
return client(args).post(`/api/v1/queue/tasks/${encodeURIComponent(taskId)}/dispatch`, body);
}
async function showRunnerJobStatus(args: ParsedArgs): Promise<JsonValue> {
const runId = flag(args, "run-id", "");
if (!runId) throw new AgentRunError("schema-invalid", "runner job-status requires --run-id", { httpStatus: 2 });
const runnerJobId = args.positional[2] ?? optionalFlag(args, "runner-job-id");
if (!runnerJobId) return listRunnerJobs(args);
return client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}/runner-jobs/${encodeURIComponent(runnerJobId)}`);
}
async function renderRunnerJob(args: ParsedArgs): Promise<JsonRecord> {
const runId = flag(args, "run-id", "");
const commandId = flag(args, "command-id", "");
if (!runId) throw new AgentRunError("schema-invalid", "runner job requires --run-id", { httpStatus: 2 });
if (!commandId) throw new AgentRunError("schema-invalid", "runner job requires --command-id", { httpStatus: 2 });
const image = optionalFlag(args, "image");
if (args.flags.get("dry-run") !== true) {
const body: JsonRecord = { commandId };
if (image) body.image = image;
const namespace = optionalFlag(args, "namespace");
const attemptId = optionalFlag(args, "attempt-id");
const runnerId = optionalFlag(args, "runner-id");
const sourceCommit = optionalFlag(args, "source-commit");
const runnerManagerUrl = optionalFlag(args, "runner-manager-url");
const idempotencyKey = optionalFlag(args, "idempotency-key");
if (namespace) body.namespace = namespace;
if (attemptId) body.attemptId = attemptId;
if (runnerId) body.runnerId = runnerId;
if (sourceCommit) body.sourceCommit = sourceCommit;
if (runnerManagerUrl) body.managerUrl = runnerManagerUrl;
if (idempotencyKey) body.idempotencyKey = idempotencyKey;
return await client(args).post(`/api/v1/runs/${encodeURIComponent(runId)}/runner-jobs`, body) as JsonRecord;
}
if (!image) throw new AgentRunError("schema-invalid", "runner job --dry-run requires --image", { httpStatus: 2 });
const run = await client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}`) as RunRecord;
const options = {
run,
commandId,
image,
managerUrl: managerUrl(args),
namespace: optionalFlag(args, "namespace") ?? "agentrun-v01",
};
const attemptId = optionalFlag(args, "attempt-id");
const runnerId = optionalFlag(args, "runner-id");
const sourceCommit = optionalFlag(args, "source-commit");
return renderRunnerJobDryRun({
...options,
...(attemptId ? { attemptId } : {}),
...(runnerId ? { runnerId } : {}),
...(sourceCommit ? { sourceCommit } : {}),
});
}
async function renderCodexSecret(args: ParsedArgs): Promise<JsonRecord> {
if (args.flags.get("dry-run") !== true) {
throw new AgentRunError("schema-invalid", "secrets codex render requires --dry-run", { httpStatus: 2 });
}
const options: Parameters<typeof renderCodexProviderSecretPlan>[0] = { dryRun: true };
const profile = optionalFlag(args, "profile");
const codexHome = optionalFlag(args, "codex-home");
const authFile = optionalFlag(args, "auth-file");
const configFile = optionalFlag(args, "config-file");
const namespace = optionalFlag(args, "namespace");
const secretName = optionalFlag(args, "secret-name");
if (profile) options.profile = profile;
if (codexHome) options.codexHome = codexHome;
if (authFile) options.authFile = authFile;
if (configFile) options.configFile = configFile;
if (namespace) options.namespace = namespace;
if (secretName) options.secretName = secretName;
return renderCodexProviderSecretPlan(options);
}
async function startServer(args: ParsedArgs): Promise<JsonRecord> {
if (args.flags.get("foreground") === true) return startServerForeground(args);
const port = Number(flag(args, "port", "8080"));
const host = flag(args, "host", "0.0.0.0");
const state = await readServerState(port);
if (state.pidAlive || state.portListening) {
throw new AgentRunError("infra-failed", `agentrun-mgr already appears to be running on port ${port}; use server status or server stop first`, { httpStatus: 409, details: state as JsonRecord });
}
await ensureDir(stateDir());
await ensureDir(logDir());
const logPath = serverLogPath(port);
const argsForChild = [process.argv[1] ?? "scripts/agentrun-cli.ts", "server", "start", "--foreground", "--host", host, "--port", String(port)];
const store = optionalFlag(args, "store");
if (store) argsForChild.push("--store", store);
const stdoutFd = openSync(logPath, "a");
const stderrFd = openSync(logPath, "a");
const child = spawn(process.execPath, argsForChild, {
cwd: process.cwd(),
env: process.env,
detached: true,
stdio: ["ignore", stdoutFd, stderrFd],
});
closeSync(stdoutFd);
closeSync(stderrFd);
child.unref();
const pidFile = pidFilePath(port);
await writeFile(pidFile, JSON.stringify({ pid: child.pid, port, host, logPath, startedAt: new Date().toISOString() }) + "\n", "utf8");
const localBaseUrl = `http://${host === "0.0.0.0" ? "127.0.0.1" : host}:${port}`;
return { action: "server-start", mode: "background", serviceId: "agentrun-mgr", pid: child.pid ?? null, port, host, baseUrl: localBaseUrl, pidFile, logPath, pollCommands: { status: `./scripts/agentrun server status --port ${port}`, logs: `./scripts/agentrun server logs --port ${port}`, stop: `./scripts/agentrun server stop --port ${port}` } };
}
async function startServerForeground(args: ParsedArgs): Promise<JsonRecord> {
const port = Number(flag(args, "port", "8080"));
const host = flag(args, "host", "0.0.0.0");
const storeMode = optionalFlag(args, "store") ?? process.env.AGENTRUN_STORE ?? process.env.AGENTRUN_MGR_STORE;
const started = await startManagerServer({ port, host, ...(storeMode === "memory" ? { store: new MemoryAgentRunStore() } : {}) });
const database = await started.store.health();
return { serviceId: "agentrun-mgr", baseUrl: started.baseUrl, pid: process.pid, database, mode: "foreground", note: "foreground process; use server start without --foreground for local background mode" };
}
async function serverStatus(args: ParsedArgs): Promise<JsonRecord> {
const explicitPort = optionalFlag(args, "port");
const baseUrl = explicitPort ? `http://127.0.0.1:${Number(explicitPort)}` : managerUrl(args);
const port = Number(explicitPort ?? portFromUrl(baseUrl));
const state = await readServerState(port);
let readiness: JsonValue = null;
let readinessFailure: JsonRecord | null = null;
if (explicitPort ? state.portListening : true) {
try {
readiness = await new ManagerClient(baseUrl).get("/health/readiness");
} catch (error) {
readinessFailure = errorToJson(error);
}
}
return { action: "server-status", serviceId: "agentrun-mgr", port, baseUrl, local: state as JsonRecord, readiness, readinessFailure, pollCommands: { status: `./scripts/agentrun server status --port ${port}`, logs: `./scripts/agentrun server logs --port ${port}`, stop: `./scripts/agentrun server stop --port ${port}` } };
}
async function serverLogs(args: ParsedArgs): Promise<JsonRecord> {
const port = Number(flag(args, "port", portFromManagerUrl(args)));
const state = await readServerState(port);
const logPath = optionalFlag(args, "log-file") ?? (typeof state.logPath === "string" ? state.logPath : null);
const tailBytes = Number(flag(args, "tail-bytes", "12000"));
if (!logPath) return { action: "server-logs", serviceId: "agentrun-mgr", port, logPath: null, exists: false, tail: "", bytes: 0, truncated: false, message: "no log file recorded for this port" };
if (!existsSync(logPath)) return { action: "server-logs", serviceId: "agentrun-mgr", port, logPath, exists: false, tail: "", bytes: 0, truncated: false, message: "log file does not exist" };
const bytes = await readFile(logPath);
const start = Math.max(0, bytes.byteLength - tailBytes);
return { action: "server-logs", serviceId: "agentrun-mgr", port, logPath, exists: true, bytes: bytes.byteLength, tailBytes, truncated: start > 0, tail: bytes.subarray(start).toString("utf8") };
}
async function stopServer(args: ParsedArgs): Promise<JsonRecord> {
const port = Number(flag(args, "port", portFromManagerUrl(args)));
const before = await readServerState(port);
let signalSent = false;
const beforePid = typeof before.pid === "number" ? before.pid : null;
const beforePortPid = typeof before.portPid === "number" ? before.portPid : null;
if (before.pidAlive === true && beforePid !== null) {
process.kill(beforePid, "SIGTERM");
signalSent = true;
} else if (beforePortPid !== null) {
process.kill(beforePortPid, "SIGTERM");
signalSent = true;
}
await sleep(500);
const after = await readServerState(port);
if (!after.pidAlive && !after.portListening && existsSync(pidFilePath(port))) await rm(pidFilePath(port), { force: true });
return { action: "server-stop", serviceId: "agentrun-mgr", port, signalSent, before: before as JsonRecord, after: after as JsonRecord, stopped: !after.pidAlive && !after.portListening };
}
function client(args: ParsedArgs): ManagerClient {
return new ManagerClient(managerUrl(args));
}
function managerUrl(args: ParsedArgs): string {
return optionalFlag(args, "manager-url") ?? process.env.AGENTRUN_MGR_URL ?? "http://127.0.0.1:8080";
}
function portFromManagerUrl(args: ParsedArgs): string {
return portFromUrl(managerUrl(args));
}
function portFromUrl(value: string): string {
try {
const url = new URL(value);
return url.port || (url.protocol === "https:" ? "443" : "80");
} catch {
return "8080";
}
}
function stateDir(): string {
return path.join(process.cwd(), ".state");
}
function logDir(): string {
return path.join(process.cwd(), "logs", new Date().toISOString().slice(0, 10).replace(/-/gu, ""));
}
function pidFilePath(port: number): string {
return path.join(stateDir(), `agentrun-mgr-${port}.pid.json`);
}
function serverLogPath(port: number): string {
return path.join(logDir(), `agentrun-mgr-${port}-${new Date().toISOString().replace(/[:.]/gu, "-")}.jsonl`);
}
async function ensureDir(dir: string): Promise<void> {
await mkdir(dir, { recursive: true });
}
async function readServerState(port: number): Promise<JsonRecord> {
const pidFile = pidFilePath(port);
const pidRecord = await readPidFile(pidFile);
const pid = typeof pidRecord?.pid === "number" ? pidRecord.pid : null;
const pidAlive = pid !== null && isPidAlive(pid);
const portPid = await pidForPort(port);
return { pidFile, pid, pidAlive, port, portListening: portPid !== null, portPid, logPath: typeof pidRecord?.logPath === "string" ? pidRecord.logPath : null, startedAt: typeof pidRecord?.startedAt === "string" ? pidRecord.startedAt : null };
}
async function readPidFile(pidFile: string): Promise<JsonRecord | null> {
try {
return JSON.parse(await readFile(pidFile, "utf8")) as JsonRecord;
} catch {
return null;
}
}
function isPidAlive(pid: number): boolean {
try {
process.kill(pid, 0);
return true;
} catch {
return false;
}
}
async function pidForPort(port: number): Promise<number | null> {
const proc = spawn("sh", ["-c", `command -v ss >/dev/null 2>&1 && ss -ltnp 'sport = :${port}' || true`], { stdio: ["ignore", "pipe", "ignore"] });
const chunks: Buffer[] = [];
proc.stdout.on("data", (chunk: Buffer) => chunks.push(chunk));
await new Promise<void>((resolve) => proc.on("close", () => resolve()));
const text = Buffer.concat(chunks).toString("utf8");
const match = text.match(/pid=(\d+)/u);
return match ? Number(match[1]) : null;
}
async function sleep(ms: number): Promise<void> {
await new Promise((resolve) => setTimeout(resolve, ms));
}
async function jsonFile(args: ParsedArgs): Promise<JsonRecord> {
const file = optionalFlag(args, "json-file");
if (!file) throw new AgentRunError("schema-invalid", "--json-file is required", { httpStatus: 2 });
const value = JSON.parse(await readFile(file, "utf8")) as unknown;
if (typeof value === "object" && value !== null && !Array.isArray(value)) return value as JsonRecord;
throw new AgentRunError("schema-invalid", "json file must contain an object", { httpStatus: 2 });
}
async function optionalJsonFile(args: ParsedArgs): Promise<JsonRecord> {
const file = optionalFlag(args, "json-file");
if (!file) return {};
return jsonFile(args);
}
function parseArgs(argv: string[]): ParsedArgs {
const positional: string[] = [];
const flags = new Map<string, string | boolean>();
for (let index = 0; index < argv.length; index += 1) {
const item = argv[index] ?? "";
if (!item.startsWith("--")) {
positional.push(item);
continue;
}
const key = item.slice(2);
const next = argv[index + 1];
if (next === undefined || next.startsWith("--")) flags.set(key, true);
else {
flags.set(key, next);
index += 1;
}
}
return { positional, flags };
}
function flag(args: ParsedArgs, name: string, fallback: string): string {
const value = args.flags.get(name);
return typeof value === "string" ? value : fallback;
}
function optionalFlag(args: ParsedArgs, name: string): string | null {
const value = args.flags.get(name);
return typeof value === "string" && value.length > 0 ? value : null;
}
function cancelBody(args: ParsedArgs): JsonRecord {
const reason = optionalFlag(args, "reason");
return reason ? { reason } : {};
}
function help(): JsonRecord {
return {
commands: [
"runs create --json-file <run.json>",
"runs show <runId>",
"runs events <runId> --after-seq <n> --limit <n>",
"runs result <runId> [--command-id <commandId>]",
"runs cancel <runId> [--reason <text>]",
"commands create <runId> --type turn|steer|interrupt --json-file <payload.json>",
"commands show <commandId> --run-id <runId>",
"commands result <commandId> --run-id <runId>",
"commands cancel <commandId> [--reason <text>]",
"runner start --run-id <runId> [--backend codex|deepseek|minimax-m3]",
"runner job --run-id <runId> --command-id <commandId> [--image <image>] [--runner-manager-url <url>] [--idempotency-key <key>]",
"runner job --dry-run --run-id <runId> --command-id <commandId> --image <image>",
"runner jobs --run-id <runId> [--command-id <commandId>]",
"runner job-status [runnerJobId] --run-id <runId>",
"queue submit --json-file <task.json> [--idempotency-key <key>]",
"queue list [--queue <queue>] [--state <state>] [--cursor <cursor>] [--limit <limit>] [--updated-after <version>]",
"queue show <taskId>",
"queue stats [--queue <queue>]",
"queue commander [--queue <queue>]",
"queue read <taskId> [--reader-id <reader>]",
"queue cancel <taskId> [--reason <text>]",
"queue dispatch <taskId> [--json-file <dispatch.json>] [--idempotency-key <key>] [--image <image>] [--namespace <namespace>]",
"queue refresh <taskId>",
"secrets codex render --dry-run [--profile codex|deepseek|minimax-m3] [--codex-home <dir>] [--namespace agentrun-v01] [--secret-name <name>]",
"backends list",
"server start [--port <port>] [--host <host>] [--foreground]",
"server status [--port <port>]",
"server logs [--port <port>] [--tail-bytes <bytes>] [--log-file <path>]",
"server stop [--port <port>]",
],
};
}
function print(value: JsonRecord): void {
process.stdout.write(`${JSON.stringify(value)}\n`);
}