| 1 | import { spawn, type ChildProcess } from "node:child_process"; |
| 2 | import { existsSync } from "node:fs"; |
| 3 | import { chmod, mkdir, readFile, writeFile } 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 { Api, ApiError } from "../api.js"; |
| 8 | import { configDir } from "../config.js"; |
| 9 | import { list, parseRepoArg, repoPath, str, type Command, type Ctx, type RepoRef } from "../cli.js"; |
| 10 | import { CliError, type Io } from "../io.js"; |
| 11 | |
| 12 | export interface ClaimedJob { |
| 13 | job_id: string; |
| 14 | run_id: string; |
| 15 | name: string; |
| 16 | token: string; |
| 17 | api_url: string; |
| 18 | } |
| 19 | |
| 20 | const AGENT_HELP = `The runner needs the Diffinery runner agent (a single Node.js file). Build it from the |
| 21 | Diffinery repository with: |
| 22 | |
| 23 | pnpm --filter @diffinery/runner build # produces containers/runner/dist/agent.mjs |
| 24 | |
| 25 | then pass --agent /path/to/agent.mjs or set DIFFINERY_RUNNER_AGENT.`; |
| 26 | |
| 27 | export function resolveAgentPath(io: Io, flag: string | undefined): string { |
| 28 | const p = flag ?? io.env.DIFFINERY_RUNNER_AGENT; |
| 29 | if (!p) throw new CliError(`no runner agent given.\n\n${AGENT_HELP}`); |
| 30 | const abs = path.resolve(io.cwd, p); |
| 31 | if (!existsSync(abs)) throw new CliError(`runner agent not found at ${abs}.\n\n${AGENT_HELP}`); |
| 32 | return abs; |
| 33 | } |
| 34 | |
| 35 | /** POST .../actions/runners/claim: a job, or null when nothing is waiting (204). A registered runner sends an empty body. */ |
| 36 | export async function claim(api: Api, repo: RepoRef, body: { name?: string; labels?: string[] }): Promise<ClaimedJob | null> { |
| 37 | const res = await api.raw<ClaimedJob | null>("POST", `${repoPath(repo)}/actions/runners/claim`, { body }); |
| 38 | if (res.status === 204 || !res.data || typeof res.data !== "object") return null; |
| 39 | return res.data; |
| 40 | } |
| 41 | |
| 42 | /** A registered runner's own credentials: the token claims this repository's jobs and nothing else. */ |
| 43 | export interface RunnerCredentials { |
| 44 | url: string; |
| 45 | owner: string; |
| 46 | repo: string; |
| 47 | id: number; |
| 48 | name: string; |
| 49 | labels: string[]; |
| 50 | token: string; |
| 51 | } |
| 52 | |
| 53 | const safe = (s: string) => s.replace(/[^A-Za-z0-9._-]/g, "_"); |
| 54 | |
| 55 | /** Where a registered runner's credentials are kept (mode 0600). */ |
| 56 | export function runnerFile(env: NodeJS.ProcessEnv, url: string, repo: RepoRef, name: string): string { |
| 57 | return path.join(configDir(env), "runners", safe(new URL(url).host), safe(repo.owner), safe(repo.name), `${safe(name)}.json`); |
| 58 | } |
| 59 | |
| 60 | async function readRunner(file: string): Promise<RunnerCredentials | null> { |
| 61 | try { |
| 62 | const v = JSON.parse(await readFile(file, "utf8")) as RunnerCredentials; |
| 63 | return typeof v.token === "string" && v.token.startsWith("drn_") ? v : null; |
| 64 | } catch (e) { |
| 65 | if ((e as NodeJS.ErrnoException).code === "ENOENT") return null; |
| 66 | throw new CliError(`${file} is not a valid runner registration; delete it and register again`); |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | async function saveRunner(file: string, creds: RunnerCredentials): Promise<void> { |
| 71 | await mkdir(path.dirname(file), { recursive: true, mode: 0o700 }); |
| 72 | await writeFile(file, JSON.stringify(creds, null, 2) + "\n", { mode: 0o600 }); |
| 73 | await chmod(file, 0o600); |
| 74 | } |
| 75 | |
| 76 | /** POST .../actions/runners/register: trade a registration token for the runner's own token. */ |
| 77 | export async function register(url: string, repo: RepoRef, registrationToken: string, name: string, labels: string[], ephemeral: boolean): Promise<RunnerCredentials> { |
| 78 | const res = await new Api(url, null).raw<{ id: number; name: string; labels: string[]; token: string }>("POST", `${repoPath(repo)}/actions/runners/register`, { |
| 79 | body: { registration_token: registrationToken, name, labels, ephemeral }, |
| 80 | }); |
| 81 | return { url, owner: repo.owner, repo: repo.name, id: res.data.id, name: res.data.name, labels: res.data.labels, token: res.data.token }; |
| 82 | } |
| 83 | |
| 84 | /** How the runner claims from one repository: with its own registered token, or (older servers) the signed-in person's. */ |
| 85 | interface Source { |
| 86 | repo: RepoRef; |
| 87 | api: Api; |
| 88 | body: { name?: string; labels?: string[] }; |
| 89 | /** The saved registration, when the token came from one. */ |
| 90 | file?: string; |
| 91 | } |
| 92 | |
| 93 | /** Environment for the agent: the job's credentials, minus the CLI's own token. */ |
| 94 | export function agentEnv(base: NodeJS.ProcessEnv, job: ClaimedJob, workRoot: string, echo: boolean): NodeJS.ProcessEnv { |
| 95 | const env: NodeJS.ProcessEnv = { ...base }; |
| 96 | for (const k of ["DIFFINERY_TOKEN", "DIFFINERY_URL", "DIFFINERY_RUNNER_AGENT", "GH_TOKEN", "GITHUB_TOKEN"]) delete env[k]; |
| 97 | env.DIFFINERY_API_URL = job.api_url; |
| 98 | env.DIFFINERY_JOB_ID = job.job_id; |
| 99 | env.DIFFINERY_JOB_TOKEN = job.token; |
| 100 | env.DIFFINERY_WORK_ROOT ??= workRoot; |
| 101 | if (echo) env.DIFFINERY_ECHO_LOGS = "1"; |
| 102 | return env; |
| 103 | } |
| 104 | |
| 105 | function pipeLines(stream: NodeJS.ReadableStream, prefix: string, write: (s: string) => void): Promise<void> { |
| 106 | return new Promise((resolve) => { |
| 107 | const rl = createInterface({ input: stream, crlfDelay: Infinity }); |
| 108 | rl.on("line", (l) => write(`${prefix}${l}\n`)); |
| 109 | rl.on("close", () => resolve()); |
| 110 | }); |
| 111 | } |
| 112 | |
| 113 | async function runJob(ctx: Ctx, agent: string, job: ClaimedJob, repo: RepoRef, workRoot: string, echo: boolean, onChild: (c: ChildProcess | null) => void): Promise<number> { |
| 114 | const c = ctx.c; |
| 115 | const prefix = c.dim(`[${job.name}] `); |
| 116 | ctx.out(`${c.cyan("→")} ${repo.owner}/${repo.name}: running job ${c.bold(job.name)} ${c.dim(`(run ${job.run_id}, job ${job.job_id})`)}`); |
| 117 | await mkdir(workRoot, { recursive: true }); |
| 118 | const child = spawn(process.execPath, [agent], { cwd: workRoot, env: agentEnv(ctx.io.env, job, workRoot, echo), stdio: ["ignore", "pipe", "pipe"] }); |
| 119 | onChild(child); |
| 120 | const done = Promise.all([pipeLines(child.stdout!, prefix, (s) => ctx.io.stdout.write(s)), pipeLines(child.stderr!, prefix, (s) => ctx.io.stderr.write(s))]); |
| 121 | const code = await new Promise<number>((resolve) => { |
| 122 | child.on("error", (e) => { |
| 123 | ctx.err(`${c.red("error:")} could not start the agent: ${e.message}`); |
| 124 | resolve(1); |
| 125 | }); |
| 126 | child.on("exit", (code, signal) => resolve(code ?? (signal ? 128 + (signal === "SIGKILL" ? 9 : 15) : 1))); |
| 127 | }); |
| 128 | await done; |
| 129 | onChild(null); |
| 130 | ctx.out(`${code === 0 ? c.green("✓") : c.red("X")} job ${job.name} finished${code === 0 ? "" : ` with exit code ${code}`}`); |
| 131 | return code; |
| 132 | } |
| 133 | |
| 134 | export const runnerCommand: Command = { |
| 135 | name: "runner", |
| 136 | summary: "Run self-hosted workflow jobs on this machine", |
| 137 | description: `Polls the given repositories for jobs with 'runs-on: [self-hosted, ...]' whose labels this |
| 138 | runner offers, and runs each with the Diffinery runner agent. Jobs run directly on this machine |
| 139 | with your user's permissions. |
| 140 | |
| 141 | The runner registers with each repository and claims jobs with its own token, which can't do |
| 142 | anything else. Register it with a registration token from a repository admin |
| 143 | (POST /repos/OWNER/NAME/actions/runners/registration-token, valid for an hour): |
| 144 | |
| 145 | diffinery runner -R OWNER/NAME --registration-token drg_... --labels linux |
| 146 | |
| 147 | so the machine never holds a personal token, which the jobs it runs could read. The runner's |
| 148 | credentials are saved under ~/.config/diffinery/runners and used on the next start. Signed in |
| 149 | as a repository admin instead, the runner registers itself with your token. |
| 150 | |
| 151 | --ephemeral registers a runner for one job: its token stops working once it takes the job, and |
| 152 | the runner exits when the job finishes. |
| 153 | |
| 154 | ${AGENT_HELP}`, |
| 155 | options: { |
| 156 | repo: { type: "string", short: "R", multiple: true, desc: "Repository to take jobs from (default: the git origin)", arg: "OWNER/NAME" }, |
| 157 | labels: { type: "string", short: "l", desc: "Extra labels this runner offers, comma separated ('self-hosted' is implied)", arg: "a,b" }, |
| 158 | name: { type: "string", short: "n", desc: "Runner name (default: host name)", arg: "NAME" }, |
| 159 | once: { type: "boolean", desc: "Run at most one job, then exit with its exit code" }, |
| 160 | agent: { type: "string", short: "a", desc: "Path to the runner agent (agent.mjs); default DIFFINERY_RUNNER_AGENT", arg: "PATH" }, |
| 161 | "work-dir": { type: "string", short: "w", desc: "Working directory for jobs (default ~/.cache/diffinery/runner/<name>)", arg: "DIR" }, |
| 162 | interval: { type: "string", desc: "Seconds between polls when idle (default 5)", arg: "SECONDS" }, |
| 163 | quiet: { type: "boolean", short: "q", desc: "Do not echo step output locally (logs still go to the server)" }, |
| 164 | "registration-token": { type: "string", desc: "Register this runner with a repository admin's registration token (one repository)", arg: "TOKEN" }, |
| 165 | ephemeral: { type: "boolean", desc: "Register for a single job, then exit when it finishes" }, |
| 166 | }, |
| 167 | maxArgs: 0, |
| 168 | async run(ctx, o) { |
| 169 | const agent = resolveAgentPath(ctx.io, str(o, "agent")); |
| 170 | const repos = list(o, "repo").length ? list(o, "repo").map(parseRepoArg) : [await ctx.repo()]; |
| 171 | const labels = (str(o, "labels") ?? "").split(",").map((s) => s.trim()).filter(Boolean); |
| 172 | const name = str(o, "name") ?? hostname(); |
| 173 | const intervalMs = Math.max(0, Number(str(o, "interval") ?? 5)) * 1000; |
| 174 | if (!Number.isFinite(intervalMs)) throw new CliError("--interval must be a number", 2); |
| 175 | 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"); |
| 176 | const workRoot = path.resolve(ctx.io.cwd, str(o, "work-dir") ?? path.join(cacheHome, "diffinery", "runner", name.replace(/[^A-Za-z0-9._-]/g, "_"))); |
| 177 | const c = ctx.c; |
| 178 | const ephemeral = o.ephemeral === true; |
| 179 | const registration = str(o, "registration-token") ?? ctx.io.env.DIFFINERY_RUNNER_REGISTRATION_TOKEN; |
| 180 | if (registration && repos.length > 1) throw new CliError("a registration token is for one repository: pass a single -R OWNER/NAME", 2); |
| 181 | const url = (await ctx.config()).url; |
| 182 | const sources: Source[] = []; |
| 183 | for (const repo of repos) { |
| 184 | const file = runnerFile(ctx.io.env, url, repo, name); |
| 185 | const saved = ephemeral ? null : await readRunner(file); |
| 186 | let creds: RunnerCredentials | null = null; |
| 187 | if (registration) creds = await register(url, repo, registration, name, labels, ephemeral); |
| 188 | else if (saved) { |
| 189 | creds = saved; |
| 190 | if (labels.length && labels.join(",") !== saved.labels.join(",")) ctx.err(`${c.yellow("warning:")} ${name} was registered with ${repo.owner}/${repo.name} offering [${saved.labels.join(", ")}]; to change its labels, delete ${file} and register it again`); |
| 191 | } else { |
| 192 | // Signed in: register with the person's token, so jobs are claimed with the runner's own. |
| 193 | const user = await ctx.authedApi(); |
| 194 | const t = await user.raw<{ token: string }>("POST", `${repoPath(repo)}/actions/runners/registration-token`, { allow: [404, 405] }); |
| 195 | if (t.status === 201 && t.data?.token) creds = await register(url, repo, t.data.token, name, labels, ephemeral); |
| 196 | else sources.push({ repo, api: user, body: { name, labels } }); // an older server without registration |
| 197 | } |
| 198 | if (!creds) continue; |
| 199 | if (!saved && !ephemeral) { |
| 200 | await saveRunner(file, creds); |
| 201 | ctx.out(`Registered runner ${c.bold(creds.name)} with ${repo.owner}/${repo.name} ${c.dim(`(saved to ${file})`)}`); |
| 202 | } |
| 203 | sources.push({ repo, api: new Api(url, creds.token), body: {}, file: ephemeral ? undefined : file }); |
| 204 | } |
| 205 | |
| 206 | let stopping = false; |
| 207 | let child: ChildProcess | null = null; |
| 208 | const abort = new AbortController(); |
| 209 | let signals = 0; |
| 210 | const onSignal = () => { |
| 211 | signals++; |
| 212 | stopping = true; |
| 213 | abort.abort(); |
| 214 | if (child && child.exitCode === null) { |
| 215 | if (signals === 1) { |
| 216 | ctx.err(c.yellow("Stopping: asking the running job to cancel (press Ctrl-C again to kill it)…")); |
| 217 | child.kill("SIGTERM"); |
| 218 | } else child.kill("SIGKILL"); |
| 219 | } |
| 220 | }; |
| 221 | process.on("SIGINT", onSignal); |
| 222 | process.on("SIGTERM", onSignal); |
| 223 | |
| 224 | try { |
| 225 | ctx.out(`Runner ${c.bold(name)} offering [${["self-hosted", ...labels].join(", ")}] for ${repos.map((r) => `${r.owner}/${r.name}`).join(", ")}`); |
| 226 | if (!o.once) ctx.out(c.dim("Waiting for jobs. Press Ctrl-C to stop.")); |
| 227 | let lastError = ""; |
| 228 | while (!stopping) { |
| 229 | let ran = false; |
| 230 | for (const { repo, api, body, file } of sources) { |
| 231 | if (stopping) break; |
| 232 | let job: ClaimedJob | null; |
| 233 | try { |
| 234 | job = await claim(api, repo, body); |
| 235 | lastError = ""; |
| 236 | } catch (e) { |
| 237 | if (file && e instanceof ApiError && e.status === 401) throw new CliError(`runner ${name} was removed from ${repo.owner}/${repo.name}; delete ${file} and register it again`); |
| 238 | if (o.once || ephemeral || (e instanceof ApiError && [401, 403, 404].includes(e.status))) throw e; |
| 239 | const msg = (e as Error).message; |
| 240 | if (msg !== lastError) ctx.err(`${c.yellow("warning:")} ${msg} (will retry)`); |
| 241 | lastError = msg; |
| 242 | continue; |
| 243 | } |
| 244 | if (!job) continue; |
| 245 | ran = true; |
| 246 | const code = await runJob(ctx, agent, job, repo, workRoot, !o.quiet, (ch) => (child = ch)); |
| 247 | if (o.once || ephemeral) return code; |
| 248 | if (stopping) break; |
| 249 | } |
| 250 | if (o.once) { |
| 251 | ctx.out("No jobs waiting."); |
| 252 | return 0; |
| 253 | } |
| 254 | if (!ran && !stopping) await ctx.io.sleep(intervalMs, abort.signal); |
| 255 | } |
| 256 | ctx.out("Runner stopped."); |
| 257 | return 0; |
| 258 | } finally { |
| 259 | process.off("SIGINT", onSignal); |
| 260 | process.off("SIGTERM", onSignal); |
| 261 | } |
| 262 | }, |
| 263 | }; |