| 1 | import { spawn, type ChildProcess } from "node:child_process"; |
| 2 | import { existsSync } from "node:fs"; |
| 3 | import { mkdir } from "node:fs/promises"; |
| 4 | import { homedir, hostname } from "node:os"; |
| 5 | import path from "node:path"; |
| 6 | import { createInterface } from "node:readline"; |
| 7 | import { ApiError, type Api } from "../api.js"; |
| 8 | import { list, parseRepoArg, repoPath, str, type Command, type Ctx, type RepoRef } from "../cli.js"; |
| 9 | import { CliError, type Io } from "../io.js"; |
| 10 | |
| 11 | export interface ClaimedJob { |
| 12 | job_id: string; |
| 13 | run_id: string; |
| 14 | name: string; |
| 15 | token: string; |
| 16 | api_url: string; |
| 17 | } |
| 18 | |
| 19 | const AGENT_HELP = `The runner needs the Diffinery runner agent (a single Node.js file). Build it from the |
| 20 | Diffinery repository with: |
| 21 | |
| 22 | pnpm --filter @diffinery/runner build # produces containers/runner/dist/agent.mjs |
| 23 | |
| 24 | then pass --agent /path/to/agent.mjs or set DIFFINERY_RUNNER_AGENT.`; |
| 25 | |
| 26 | export function resolveAgentPath(io: Io, flag: string | undefined): string { |
| 27 | const p = flag ?? io.env.DIFFINERY_RUNNER_AGENT; |
| 28 | if (!p) throw new CliError(`no runner agent given.\n\n${AGENT_HELP}`); |
| 29 | const abs = path.resolve(io.cwd, p); |
| 30 | if (!existsSync(abs)) throw new CliError(`runner agent not found at ${abs}.\n\n${AGENT_HELP}`); |
| 31 | return abs; |
| 32 | } |
| 33 | |
| 34 | /** POST .../actions/runners/claim: a job, or null when nothing is waiting (204). */ |
| 35 | export async function claim(api: Api, repo: RepoRef, name: string, labels: string[]): Promise<ClaimedJob | null> { |
| 36 | const res = await api.raw<ClaimedJob | null>("POST", `${repoPath(repo)}/actions/runners/claim`, { body: { name, labels } }); |
| 37 | if (res.status === 204 || !res.data || typeof res.data !== "object") return null; |
| 38 | return res.data; |
| 39 | } |
| 40 | |
| 41 | /** Environment for the agent: the job's credentials, minus the CLI's own token. */ |
| 42 | export function agentEnv(base: NodeJS.ProcessEnv, job: ClaimedJob, workRoot: string, echo: boolean): NodeJS.ProcessEnv { |
| 43 | const env: NodeJS.ProcessEnv = { ...base }; |
| 44 | for (const k of ["DIFFINERY_TOKEN", "DIFFINERY_URL", "DIFFINERY_RUNNER_AGENT", "GH_TOKEN", "GITHUB_TOKEN"]) delete env[k]; |
| 45 | env.DIFFINERY_API_URL = job.api_url; |
| 46 | env.DIFFINERY_JOB_ID = job.job_id; |
| 47 | env.DIFFINERY_JOB_TOKEN = job.token; |
| 48 | env.DIFFINERY_WORK_ROOT ??= workRoot; |
| 49 | if (echo) env.DIFFINERY_ECHO_LOGS = "1"; |
| 50 | return env; |
| 51 | } |
| 52 | |
| 53 | function pipeLines(stream: NodeJS.ReadableStream, prefix: string, write: (s: string) => void): Promise<void> { |
| 54 | return new Promise((resolve) => { |
| 55 | const rl = createInterface({ input: stream, crlfDelay: Infinity }); |
| 56 | rl.on("line", (l) => write(`${prefix}${l}\n`)); |
| 57 | rl.on("close", () => resolve()); |
| 58 | }); |
| 59 | } |
| 60 | |
| 61 | async function runJob(ctx: Ctx, agent: string, job: ClaimedJob, repo: RepoRef, workRoot: string, echo: boolean, onChild: (c: ChildProcess | null) => void): Promise<number> { |
| 62 | const c = ctx.c; |
| 63 | const prefix = c.dim(`[${job.name}] `); |
| 64 | ctx.out(`${c.cyan("→")} ${repo.owner}/${repo.name}: running job ${c.bold(job.name)} ${c.dim(`(run ${job.run_id}, job ${job.job_id})`)}`); |
| 65 | await mkdir(workRoot, { recursive: true }); |
| 66 | const child = spawn(process.execPath, [agent], { cwd: workRoot, env: agentEnv(ctx.io.env, job, workRoot, echo), stdio: ["ignore", "pipe", "pipe"] }); |
| 67 | onChild(child); |
| 68 | const done = Promise.all([pipeLines(child.stdout!, prefix, (s) => ctx.io.stdout.write(s)), pipeLines(child.stderr!, prefix, (s) => ctx.io.stderr.write(s))]); |
| 69 | const code = await new Promise<number>((resolve) => { |
| 70 | child.on("error", (e) => { |
| 71 | ctx.err(`${c.red("error:")} could not start the agent: ${e.message}`); |
| 72 | resolve(1); |
| 73 | }); |
| 74 | child.on("exit", (code, signal) => resolve(code ?? (signal ? 128 + (signal === "SIGKILL" ? 9 : 15) : 1))); |
| 75 | }); |
| 76 | await done; |
| 77 | onChild(null); |
| 78 | ctx.out(`${code === 0 ? c.green("✓") : c.red("X")} job ${job.name} finished${code === 0 ? "" : ` with exit code ${code}`}`); |
| 79 | return code; |
| 80 | } |
| 81 | |
| 82 | export const runnerCommand: Command = { |
| 83 | name: "runner", |
| 84 | summary: "Run self-hosted workflow jobs on this machine", |
| 85 | description: `Polls the given repositories for jobs with 'runs-on: [self-hosted, ...]' whose labels this |
| 86 | runner offers, and runs each with the Diffinery runner agent. Requires admin access to the |
| 87 | repositories. Jobs run directly on this machine with your user's permissions. |
| 88 | |
| 89 | ${AGENT_HELP}`, |
| 90 | options: { |
| 91 | repo: { type: "string", short: "R", multiple: true, desc: "Repository to take jobs from (default: the git origin)", arg: "OWNER/NAME" }, |
| 92 | labels: { type: "string", short: "l", desc: "Extra labels this runner offers, comma separated ('self-hosted' is implied)", arg: "a,b" }, |
| 93 | name: { type: "string", short: "n", desc: "Runner name (default: host name)", arg: "NAME" }, |
| 94 | once: { type: "boolean", desc: "Run at most one job, then exit with its exit code" }, |
| 95 | agent: { type: "string", short: "a", desc: "Path to the runner agent (agent.mjs); default DIFFINERY_RUNNER_AGENT", arg: "PATH" }, |
| 96 | "work-dir": { type: "string", short: "w", desc: "Working directory for jobs (default ~/.cache/diffinery/runner/<name>)", arg: "DIR" }, |
| 97 | interval: { type: "string", desc: "Seconds between polls when idle (default 5)", arg: "SECONDS" }, |
| 98 | quiet: { type: "boolean", short: "q", desc: "Do not echo step output locally (logs still go to the server)" }, |
| 99 | }, |
| 100 | maxArgs: 0, |
| 101 | async run(ctx, o) { |
| 102 | const agent = resolveAgentPath(ctx.io, str(o, "agent")); |
| 103 | const repos = list(o, "repo").length ? list(o, "repo").map(parseRepoArg) : [await ctx.repo()]; |
| 104 | const labels = (str(o, "labels") ?? "").split(",").map((s) => s.trim()).filter(Boolean); |
| 105 | const name = str(o, "name") ?? hostname(); |
| 106 | const intervalMs = Math.max(0, Number(str(o, "interval") ?? 5)) * 1000; |
| 107 | if (!Number.isFinite(intervalMs)) throw new CliError("--interval must be a number", 2); |
| 108 | const cacheHome = ctx.io.env.XDG_CACHE_HOME && path.isAbsolute(ctx.io.env.XDG_CACHE_HOME) ? ctx.io.env.XDG_CACHE_HOME : path.join(ctx.io.env.HOME || homedir(), ".cache"); |
| 109 | const workRoot = path.resolve(ctx.io.cwd, str(o, "work-dir") ?? path.join(cacheHome, "diffinery", "runner", name.replace(/[^A-Za-z0-9._-]/g, "_"))); |
| 110 | const api = await ctx.authedApi(); |
| 111 | const c = ctx.c; |
| 112 | |
| 113 | let stopping = false; |
| 114 | let child: ChildProcess | null = null; |
| 115 | const abort = new AbortController(); |
| 116 | let signals = 0; |
| 117 | const onSignal = () => { |
| 118 | signals++; |
| 119 | stopping = true; |
| 120 | abort.abort(); |
| 121 | if (child && child.exitCode === null) { |
| 122 | if (signals === 1) { |
| 123 | ctx.err(c.yellow("Stopping: asking the running job to cancel (press Ctrl-C again to kill it)…")); |
| 124 | child.kill("SIGTERM"); |
| 125 | } else child.kill("SIGKILL"); |
| 126 | } |
| 127 | }; |
| 128 | process.on("SIGINT", onSignal); |
| 129 | process.on("SIGTERM", onSignal); |
| 130 | |
| 131 | try { |
| 132 | ctx.out(`Runner ${c.bold(name)} offering [${["self-hosted", ...labels].join(", ")}] for ${repos.map((r) => `${r.owner}/${r.name}`).join(", ")}`); |
| 133 | if (!o.once) ctx.out(c.dim("Waiting for jobs. Press Ctrl-C to stop.")); |
| 134 | let lastError = ""; |
| 135 | while (!stopping) { |
| 136 | let ran = false; |
| 137 | for (const repo of repos) { |
| 138 | if (stopping) break; |
| 139 | let job: ClaimedJob | null; |
| 140 | try { |
| 141 | job = await claim(api, repo, name, labels); |
| 142 | lastError = ""; |
| 143 | } catch (e) { |
| 144 | if (o.once || (e instanceof ApiError && [401, 403, 404].includes(e.status))) throw e; |
| 145 | const msg = (e as Error).message; |
| 146 | if (msg !== lastError) ctx.err(`${c.yellow("warning:")} ${msg} (will retry)`); |
| 147 | lastError = msg; |
| 148 | continue; |
| 149 | } |
| 150 | if (!job) continue; |
| 151 | ran = true; |
| 152 | const code = await runJob(ctx, agent, job, repo, workRoot, !o.quiet, (ch) => (child = ch)); |
| 153 | if (o.once) return code; |
| 154 | if (stopping) break; |
| 155 | } |
| 156 | if (o.once) { |
| 157 | ctx.out("No jobs waiting."); |
| 158 | return 0; |
| 159 | } |
| 160 | if (!ran && !stopping) await ctx.io.sleep(intervalMs, abort.signal); |
| 161 | } |
| 162 | ctx.out("Runner stopped."); |
| 163 | return 0; |
| 164 | } finally { |
| 165 | process.off("SIGINT", onSignal); |
| 166 | process.off("SIGTERM", onSignal); |
| 167 | } |
| 168 | }, |
| 169 | }; |