Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

904 lines · 37.3 KBRaw History
1import type { ChildProcess } from "node:child_process";
2import { randomUUID } from "node:crypto";
3import { existsSync } from "node:fs";
4import { mkdir, readFile, writeFile } from "node:fs/promises";
5import path from "node:path";
6import type { Readable } from "node:stream";
7import type { Conclusion, ContainerSpec, JobCompletion, JobSpec, StepReport, StepSpec } from "@diffinery/shared";
8import { evaluate, evaluateIf, interpolate, type StatusFns } from "@diffinery/workflow-yaml";
9import { cpuTimes, runnerInfo, type RunnerApi } from "./api.js";
10import { parseKeyValueFile, parsePathFile } from "./commands.js";
11import { Docker, type Wrap } from "./docker.js";
12import { JobLogger, OutputProcessor, type Annotation, type OutputSink } from "./log.js";
13import { Masker } from "./mask.js";
14import { ToolPrefetch } from "./prefetch.js";
15import { RuntimeClient } from "./runtime.js";
16import { runAction } from "./steps/actions.js";
17import { runScript } from "./steps/run.js";
18import { exec, hashFiles, killTree, nowIso, runnerArch, type ExecResult } from "./util.js";
19
20export type Outcome = "success" | "failure" | "cancelled" | "skipped";
21
22/**
23 * After a cancel or a job timeout, how long the steps that still run (`if: always()`, `if: cancelled()`,
24 * post steps) get before they are stopped too: GitHub's runner gives them 5 minutes.
25 */
26export const CANCEL_GRACE_MS = 5 * 60_000;
27
28export interface StepContextEntry {
29 outputs: Record<string, string>;
30 outcome: Outcome;
31 conclusion: Outcome;
32}
33
34/** Execution scope: the job itself, or the inside of a composite action. */
35export interface Scope {
36 steps: Record<string, StepContextEntry>;
37 inputs: Record<string, unknown>;
38 /** A step in this scope failed (without continue-on-error). */
39 failed: boolean;
40 /** Env inherited from the calling step (composite actions). */
41 inheritedEnv: Record<string, string>;
42 composite: boolean;
43 action?: { path: string; repository: string; ref: string };
44 depth: number;
45}
46
47export interface StepIO extends OutputSink {
48 /** Short reason recorded when the step fails (e.g. exit code). */
49 reason?: string;
50}
51
52export interface StepResult {
53 outcome: Outcome;
54 conclusion: Outcome;
55 outputs: Record<string, string>;
56}
57
58export interface PostStep {
59 name: string;
60 /** Expression evaluated with job status (e.g. `always()`, `success()`). */
61 condition: string;
62 run: (logIndex: number) => Promise<void>;
63}
64
65export interface RunnerOptions {
66 workRoot: string;
67 /** Tool cache dir; defaults to `${workRoot}/_tool` (or RUNNER_TOOL_CACHE). */
68 toolCache?: string;
69 echoLogs?: boolean;
70 /** Base env for child processes (defaults to process.env). */
71 baseEnv?: NodeJS.ProcessEnv;
72 /** Start the Docker daemon as the job begins when the job will use Docker (hosted runners). */
73 prewarmDocker?: boolean;
74 /** Hosted runners: start installing the job's Go or Java toolchain as the job begins (prefetch.ts). */
75 prefetchTools?: boolean;
76}
77
78export class StepFailure extends Error {}
79
80/**
81 * Whether a job will use Docker: job or service containers, `docker://` actions, or a step that names
82 * docker (a `run:` script, or an action such as docker/build-push-action). A guess, so only for a head start.
83 */
84export function jobUsesDocker(spec: JobSpec): boolean {
85 if (spec.container?.image || Object.keys(spec.services ?? {}).length) return true;
86 return spec.steps.some((s) => /\bdocker\b|testcontainers/i.test(`${s.uses ?? ""}\n${s.run ?? ""}`));
87}
88
89/** The runner's own credentials: never in a step's environment. */
90const SENSITIVE_ENV = ["DIFFINERY_JOB_TOKEN", "DIFFINERY_SANDBOX_SECRET", "DIFFINERY_GITHUB_TOKEN"];
91
92export class JobRunner {
93 readonly masker = new Masker();
94 readonly logger: JobLogger;
95 readonly workspace: string;
96 readonly tempDir: string;
97 readonly toolCache: string;
98 readonly actionsDir: string;
99 readonly fileCmdDir: string;
100 readonly eventPath: string;
101 readonly debug: boolean;
102
103 /** Accumulated GITHUB_ENV values. */
104 envAccum: Record<string, string> = {};
105 /** Accumulated GITHUB_PATH entries, most recent first. */
106 pathAccum: string[] = [];
107 jobEnv: Record<string, string> = {};
108 readonly summaries: string[] = [];
109 readonly posts: PostStep[] = [];
110 readonly children = new Set<ChildProcess>();
111 /**
112 * Aborted when the job is cancelled or times out, for work the runner does itself (downloads). A new
113 * one takes its place for the steps that still run after a cancel.
114 */
115 private aborter = new AbortController();
116 readonly stepErrors = new Map<number, Annotation[]>();
117 readonly stepReasons = new Map<number, string>();
118 /** Job and service containers, and Docker actions (self-hosted runners with Docker). */
119 readonly docker: Docker;
120 readonly prefetch = new ToolPrefetch(this);
121 /** The Actions runtime services client (null when the server sent none). */
122 readonly runtime: RuntimeClient | null;
123
124 /** Why the job is ending early: the job's status for status functions and its conclusion. */
125 private stopping: null | "cancelled" | "timeout" = null;
126 /**
127 * Whether the work under way must stop now: set by a cancel, cleared once the steps that still
128 * run after it begin (resumeAfterCancel), and set for good when their grace runs out (`forced`).
129 */
130 private halting = false;
131 private forced = false;
132 private graceTimer: NodeJS.Timeout | undefined;
133 /** Set when the platform signals that it is replacing this runner (a container rollout). */
134 replacing = false;
135 private firstFailure: { step?: number; message: string } | null = null;
136 /** The vCPUs' time when a hosted job started, to say at the end how much of it the host took (steal). */
137 private cpuAtStart: { total: number; steal: number } | null = null;
138 private githubCtx: Record<string, unknown> = {};
139 private defaultEnv: Record<string, string> = {};
140 /**
141 * The Actions runtime services' variables. Like GitHub, only JavaScript and
142 * Docker actions get them (and the built-ins, which call the services directly);
143 * `run:` steps do not.
144 */
145 runtimeEnv: Record<string, string> = {};
146 private baseEnv: NodeJS.ProcessEnv;
147 private secrets: Record<string, string>;
148
149 constructor(
150 readonly spec: JobSpec,
151 readonly api: RunnerApi,
152 logger: JobLogger | null,
153 readonly opts: RunnerOptions,
154 batcher: import("./api.js").LogBatcher | null = null,
155 ) {
156 this.logger = logger ?? new JobLogger(this.masker, batcher, !!opts.echoLogs);
157 this.workspace = path.join(opts.workRoot, spec.repo.name);
158 this.tempDir = path.join(opts.workRoot, "_temp");
159 this.toolCache = opts.toolCache ?? process.env.RUNNER_TOOL_CACHE ?? path.join(opts.workRoot, "_tool");
160 this.actionsDir = path.join(opts.workRoot, "_actions");
161 this.fileCmdDir = path.join(this.tempDir, "_runner_file_commands");
162 this.eventPath = path.join(this.tempDir, "_github_workflow", "event.json");
163 this.secrets = { ...spec.secrets, GITHUB_TOKEN: spec.token };
164 this.debug = this.secrets.ACTIONS_STEP_DEBUG === "true" || spec.vars?.ACTIONS_STEP_DEBUG === "true";
165 const base = { ...(opts.baseEnv ?? process.env) };
166 for (const k of SENSITIVE_ENV) delete base[k];
167 this.baseEnv = base;
168 for (const v of Object.values(this.secrets)) this.masker.add(v);
169 if (spec.runtime) this.masker.add(spec.runtime.token);
170 this.runtime = RuntimeClient.fromSpec(spec);
171 this.docker = new Docker(this);
172 }
173
174 /** The runner's own environment (for internal tools such as the docker CLI). */
175 hostEnv(): NodeJS.ProcessEnv {
176 return { ...this.baseEnv };
177 }
178
179 // ------------------------------------------------------------ public
180
181 get signal(): AbortSignal {
182 return this.aborter.signal;
183 }
184
185 /** Whether the step under way must stop (the job was cancelled or timed out while it ran). */
186 get isStopping(): boolean {
187 return this.halting;
188 }
189
190 /**
191 * SIGINT / SIGTERM / external cancellation, or the job's timeout. Like GitHub, the step under way
192 * stops, and the steps that still run after a cancel (`if: always()`, `if: cancelled()`, post steps)
193 * get `graceMs` before they are stopped too. A second cancel stops them at once.
194 */
195 cancel(reason: "cancelled" | "timeout" = "cancelled", graceMs = CANCEL_GRACE_MS): void {
196 if (this.stopping) return this.force();
197 this.stopping = reason;
198 this.halt();
199 if (graceMs <= 0) return this.force();
200 this.graceTimer = setTimeout(() => {
201 this.logger.line(-1, `##[error]The steps that run after a cancel took longer than ${graceMs / 60_000} minutes; stopping them.`);
202 this.force();
203 }, graceMs);
204 this.graceTimer.unref();
205 }
206
207 private force(): void {
208 if (this.forced) return;
209 this.forced = true;
210 if (this.graceTimer) clearTimeout(this.graceTimer);
211 this.halt();
212 }
213
214 private halt(): void {
215 this.halting = true;
216 this.aborter.abort();
217 for (const c of this.children) killTree(c);
218 }
219
220 /** A process the job started: stopped with the job, or at once when it started after the stop. */
221 private track(c: ChildProcess): void {
222 this.children.add(c);
223 c.once("close", () => this.children.delete(c));
224 if (this.halting) killTree(c);
225 }
226
227 /** Once the cancelled step has stopped: let the steps that still run after a cancel do their work. */
228 private resumeAfterCancel(): void {
229 if (!this.halting || this.forced) return;
230 this.halting = false;
231 this.aborter = new AbortController();
232 }
233
234 async run(): Promise<JobCompletion> {
235 const spec = this.spec;
236 let jobTimer: NodeJS.Timeout | undefined;
237 if (spec.timeoutMinutes > 0) {
238 jobTimer = setTimeout(() => {
239 this.logger.line(-1, `##[error]The job running on runner has exceeded the maximum execution time of ${spec.timeoutMinutes} minutes.`);
240 this.cancel("timeout");
241 }, spec.timeoutMinutes * 60_000);
242 jobTimer.unref();
243 }
244 const top: Scope = { steps: {}, inputs: spec.inputs ?? {}, failed: false, inheritedEnv: {}, composite: false, depth: 0 };
245 if (this.opts.prewarmDocker && jobUsesDocker(spec)) this.docker.prewarm();
246 if (this.opts.prefetchTools) this.prefetch.start(spec);
247 try {
248 await this.setup();
249 await this.initContainers(top);
250 for (let i = 0; i < spec.steps.length; i++) {
251 const step = spec.steps[i]!;
252 const fallbackName = defaultStepName(step);
253 if (this.forced) {
254 await this.report(i, { status: "completed", conclusion: "cancelled", name: safeName(step, fallbackName), completedAt: nowIso() });
255 continue;
256 }
257 this.resumeAfterCancel();
258 const startedAt = nowIso();
259 let started = false;
260 const res = await this.executeStep(step, top, i, async (name) => {
261 started = true;
262 await this.report(i, { status: "in_progress", name, startedAt });
263 });
264 await this.report(i, {
265 status: "completed",
266 // After a cancel, a step whose condition kept it from running shows as cancelled, as before.
267 conclusion: this.stopping && res.conclusion === "skipped" ? "cancelled" : res.conclusion,
268 name: res.name ?? fallbackName,
269 ...(started ? { startedAt } : {}),
270 completedAt: nowIso(),
271 });
272 }
273 if (!this.forced) await this.runPosts(top);
274 } catch (e) {
275 const msg = e instanceof Error ? e.message : String(e);
276 this.logger.line(-1, `##[error]${msg}`);
277 top.failed = true;
278 this.firstFailure ??= { message: msg };
279 } finally {
280 if (jobTimer) clearTimeout(jobTimer);
281 if (this.graceTimer) clearTimeout(this.graceTimer);
282 await this.prefetch.stop();
283 try {
284 await this.docker.cleanup({ logIndex: -1, outputs: {}, state: {} });
285 } catch (e) {
286 this.logger.line(-1, `##[warning]Could not remove the job's containers: ${errMsg(e)}`);
287 }
288 }
289
290 let conclusion: Conclusion = top.failed ? "failure" : "success";
291 if (this.stopping === "cancelled") conclusion = "cancelled";
292 if (this.stopping === "timeout") {
293 conclusion = "failure";
294 this.firstFailure = { ...(this.firstFailure ?? {}), message: `The job exceeded the maximum execution time of ${spec.timeoutMinutes} minutes.` };
295 }
296 const outputs = this.jobOutputs(top);
297 const completion: JobCompletion = { conclusion, outputs };
298 const summary = this.summaries.filter((s) => s.trim() !== "").join("\n");
299 if (summary) completion.summary = this.masker.mask(summary);
300 if (conclusion === "failure" && this.firstFailure) completion.failure = { ...this.firstFailure, message: this.masker.mask(this.firstFailure.message) };
301 if (conclusion === "cancelled") completion.failure = { message: "The job was cancelled." };
302 if (this.replacing && conclusion !== "success") completion.replaced = true;
303 const stolen = this.stolenShare();
304 if (stolen !== null) this.logger.line(-1, `Host: ${stolen}% of the vCPUs' time went to other work on the host (steal)`);
305 this.logger.line(-1, `Job completed with conclusion: ${conclusion}`);
306 return completion;
307 }
308
309 /** The share of the vCPUs' time since the job started that the hypervisor gave to other work, in percent. */
310 private stolenShare(): number | null {
311 const end = this.cpuAtStart && cpuTimes();
312 if (!this.cpuAtStart || !end || end.total <= this.cpuAtStart.total) return null;
313 return Math.round((100 * (end.steal - this.cpuAtStart.steal)) / (end.total - this.cpuAtStart.total));
314 }
315
316 // ------------------------------------------------------------ setup
317
318 private async setup(): Promise<void> {
319 const spec = this.spec;
320 await mkdir(this.workspace, { recursive: true });
321 await mkdir(this.fileCmdDir, { recursive: true });
322 await mkdir(this.toolCache, { recursive: true });
323 await mkdir(this.actionsDir, { recursive: true });
324 await mkdir(path.dirname(this.eventPath), { recursive: true });
325 const g = spec.github ?? {};
326 const refName = String(g.ref_name ?? spec.ref.replace(/^refs\/(heads|tags)\//, ""));
327 const refType = String(g.ref_type ?? (spec.ref.startsWith("refs/tags/") ? "tag" : "branch"));
328 const server = spec.apiUrl.replace(/\/$/, "");
329 const repoFull = `${spec.repo.owner}/${spec.repo.name}`;
330 this.githubCtx = {
331 repository: repoFull,
332 repository_owner: spec.repo.owner,
333 sha: spec.sha,
334 ref: spec.ref,
335 ref_name: refName,
336 ref_type: refType,
337 run_id: spec.runId,
338 run_number: g.run_number ?? "1",
339 run_attempt: g.run_attempt ?? "1",
340 job: spec.jobKey,
341 workflow: spec.workflowPath,
342 event_name: "push",
343 event: {},
344 actor: "diffinery",
345 server_url: server,
346 api_url: `${server}/api/v3`,
347 graphql_url: `${server}/api/graphql`,
348 ...g,
349 workspace: this.workspace,
350 token: spec.token,
351 action: "",
352 action_path: "",
353 };
354 await writeFile(this.eventPath, JSON.stringify(this.githubCtx.event ?? {}, null, 2));
355 const str = (v: unknown) => (v === undefined || v === null ? "" : String(v));
356 const d: Record<string, string> = {
357 CI: "true",
358 GITHUB_ACTIONS: "true",
359 DIFFINERY: "true",
360 GITHUB_REPOSITORY: repoFull,
361 GITHUB_REPOSITORY_OWNER: spec.repo.owner,
362 GITHUB_SHA: spec.sha,
363 GITHUB_REF: spec.ref,
364 GITHUB_REF_NAME: refName,
365 GITHUB_REF_TYPE: refType,
366 GITHUB_HEAD_REF: str(g.head_ref),
367 GITHUB_BASE_REF: str(g.base_ref),
368 GITHUB_RUN_ID: str(this.githubCtx.run_id),
369 GITHUB_RUN_NUMBER: str(this.githubCtx.run_number),
370 GITHUB_RUN_ATTEMPT: str(this.githubCtx.run_attempt),
371 GITHUB_JOB: spec.jobKey,
372 GITHUB_ACTOR: str(this.githubCtx.actor),
373 GITHUB_TRIGGERING_ACTOR: str(g.triggering_actor ?? this.githubCtx.actor),
374 GITHUB_EVENT_NAME: str(this.githubCtx.event_name),
375 GITHUB_EVENT_PATH: this.eventPath,
376 GITHUB_SERVER_URL: server,
377 GITHUB_API_URL: `${server}/api/v3`,
378 GITHUB_GRAPHQL_URL: `${server}/api/graphql`,
379 // GitHub Enterprise's names for the same two URLs. Tools such as super-linter build api.github.com from a
380 // default domain unless these are set, and would send the job's token there.
381 GITHUB_CUSTOM_API_URL: `${server}/api/v3`,
382 GITHUB_CUSTOM_SERVER_URL: server,
383 GITHUB_WORKFLOW: str(g.workflow ?? spec.workflowPath),
384 GITHUB_WORKFLOW_REF: str(g.workflow_ref ?? `${repoFull}/${spec.workflowPath}@${spec.ref}`),
385 GITHUB_WORKSPACE: this.workspace,
386 GITHUB_RETENTION_DAYS: "90",
387 RUNNER_OS: "Linux",
388 RUNNER_ARCH: runnerArch(),
389 RUNNER_NAME: "diffinery",
390 RUNNER_ENVIRONMENT: "diffinery-hosted",
391 RUNNER_TEMP: this.tempDir,
392 RUNNER_TOOL_CACHE: this.toolCache,
393 RUNNER_WORKSPACE: path.dirname(this.workspace),
394 AGENT_TOOLSDIRECTORY: this.toolCache,
395 // What hosted ubuntu-24.04 runners report; setup-* actions key tool and dependency caches on it.
396 ImageOS: process.env.ImageOS ?? "ubuntu24",
397 ImageVersion: process.env.ImageVersion ?? process.env.DIFFINERY_IMAGE_VERSION ?? "20250101.1",
398 };
399 if (this.debug || this.secrets.ACTIONS_RUNNER_DEBUG === "true" || spec.vars?.ACTIONS_RUNNER_DEBUG === "true") d.RUNNER_DEBUG = "1";
400 if (spec.runtime) {
401 // The Actions runtime services (cache, artifacts), scoped to this job and expiring with it.
402 this.runtimeEnv = {
403 ACTIONS_RUNTIME_TOKEN: spec.runtime.token,
404 ACTIONS_RESULTS_URL: spec.runtime.resultsUrl,
405 ACTIONS_CACHE_URL: spec.runtime.cacheUrl,
406 ACTIONS_RUNTIME_URL: spec.runtime.runtimeUrl,
407 ACTIONS_CACHE_SERVICE_V2: "true",
408 };
409 }
410 if (spec.idToken && this.api.online) {
411 // @actions/core appends `&audience=...`, so the URL must already carry a query string.
412 d.ACTIONS_ID_TOKEN_REQUEST_URL = `${this.api.url("/oidc")}?api-version=2.0`;
413 d.ACTIONS_ID_TOKEN_REQUEST_TOKEN = spec.token;
414 }
415 this.defaultEnv = d;
416 this.logger.line(-1, `Diffinery runner: job ${spec.jobKey} (${spec.name}) for ${repoFull}@${spec.sha.slice(0, 12)}`);
417 const where = runnerInfo();
418 if (Object.keys(where).length) this.logger.line(-1, `Runner: ${[where.image && `image ${where.image}`, where.location && `in ${where.location}`, where.instance && `container ${where.instance}`].filter(Boolean).join(", ")}`);
419 if (spec.runnerSize) this.logger.line(-1, spec.runnerSize);
420 if (where.cpu) {
421 this.logger.line(-1, `Host: ${where.cpu}`);
422 this.cpuAtStart = cpuTimes();
423 }
424 this.logger.line(-1, `Workspace: ${this.workspace}`);
425 // job-level env
426 const ctx = this.contexts({ steps: {}, inputs: spec.inputs ?? {}, failed: false, inheritedEnv: {}, composite: false, depth: 0 }, {});
427 for (const [k, v] of Object.entries(spec.env ?? {})) {
428 this.jobEnv[k] = this.interp(String(v ?? ""), ctx);
429 }
430 }
431
432 // ------------------------------------------------------------ containers
433
434 private evalContainer(spec: ContainerSpec, ctx: Record<string, unknown>): ContainerSpec {
435 const i = (v: unknown) => this.interp(v === undefined || v === null ? "" : String(v), ctx);
436 const out: ContainerSpec = { image: i(spec.image).trim() };
437 if (spec.credentials) out.credentials = { username: i(spec.credentials.username), password: i(spec.credentials.password) };
438 if (spec.env) out.env = Object.fromEntries(Object.entries(spec.env).map(([k, v]) => [k, i(v)]));
439 if (spec.ports) out.ports = spec.ports.map((p) => i(p)).filter(Boolean);
440 if (spec.volumes) out.volumes = spec.volumes.map((v) => i(v)).filter(Boolean);
441 if (spec.options) out.options = i(spec.options);
442 return out;
443 }
444
445 /** Starts service containers and the job container, like GitHub's "Initialize containers". */
446 private async initContainers(top: Scope): Promise<void> {
447 const spec = this.spec;
448 if (!spec.container && !Object.keys(spec.services ?? {}).length) return;
449 const ctx = this.contexts(top, { ...this.jobEnv });
450 const container = spec.container ? this.evalContainer(spec.container, ctx) : null;
451 // A service whose image evaluates to an empty string is skipped (GitHub does the same).
452 const services = Object.entries(spec.services ?? {})
453 .map(([k, s]) => [k, this.evalContainer(s, ctx)] as const)
454 .filter(([, s]) => s.image);
455 if (!container?.image && !services.length) return;
456 const io: StepIO = { logIndex: -1, outputs: {}, state: {} };
457 this.logger.line(-1, "##[group]Initialize containers");
458 try {
459 await this.docker.ensure(io, container?.image ? "Job containers (jobs.<id>.container)" : "Service containers (jobs.<id>.services)");
460 for (const [key, s] of services) await this.docker.startService(io, key, s);
461 if (container?.image) await this.docker.startJobContainer(io, container);
462 } finally {
463 this.logger.line(-1, "##[endgroup]");
464 }
465 }
466
467 /** Wraps a step process so it runs in the job container, when there is one. */
468 containerWrap(): Wrap | undefined {
469 return this.docker.execWrap();
470 }
471
472 /** The node binary JavaScript actions run with (the runner's own, mounted into a job container). */
473 nodePath(): string {
474 return this.docker.container ? "/__e/node" : process.execPath;
475 }
476
477 // ------------------------------------------------------------ contexts
478
479 statusFns(scope: Scope): StatusFns {
480 return {
481 success: () => !scope.failed && this.stopping === null,
482 // GitHub: once the job is cancelled its status is cancelled, so failure() is false.
483 failure: () => scope.failed && this.stopping === null,
484 cancelled: () => this.stopping !== null,
485 always: () => true,
486 };
487 }
488
489 extra(): { hashFiles: (...p: string[]) => string } {
490 return { hashFiles: (...patterns: string[]) => hashFiles(this.workspace, patterns) };
491 }
492
493 /** Env context visible to a step before its own `env:` is applied. */
494 baseEnvContext(scope: Scope): Record<string, string> {
495 return { ...this.jobEnv, ...this.envAccum, ...scope.inheritedEnv };
496 }
497
498 contexts(scope: Scope, env: Record<string, string>): Record<string, unknown> {
499 const spec = this.spec;
500 const github = { ...this.githubCtx };
501 if (scope.action) {
502 github.action_path = scope.action.path;
503 github.action_repository = scope.action.repository;
504 github.action_ref = scope.action.ref;
505 }
506 return {
507 github,
508 env,
509 vars: spec.vars ?? {},
510 secrets: this.secrets,
511 matrix: spec.matrix ?? {},
512 needs: spec.needs ?? {},
513 steps: scope.steps,
514 runner: {
515 os: "Linux",
516 arch: runnerArch(),
517 name: "diffinery",
518 temp: this.tempDir,
519 tool_cache: this.toolCache,
520 workspace: path.dirname(this.workspace),
521 debug: this.debug ? "1" : "",
522 environment: "diffinery-hosted",
523 },
524 inputs: scope.inputs,
525 job: {
526 status: this.stopping ? "cancelled" : scope.failed ? "failure" : "success",
527 container: this.docker.container ? { id: this.docker.container.id, network: this.docker.container.network } : {},
528 services: Object.fromEntries(Object.entries(this.docker.services).map(([k, v]) => [k, { id: v.id, network: v.network, ports: v.ports }])),
529 },
530 strategy: spec.strategy ?? {},
531 };
532 }
533
534 interp(template: string, ctx: Record<string, unknown>, scope?: Scope): string {
535 if (!template.includes("${{")) return template;
536 const status = scope ? this.statusFns(scope) : undefined;
537 return interpolate(template, ctx, status, this.extra());
538 }
539
540 evalIf(cond: string | undefined, ctx: Record<string, unknown>, scope: Scope): boolean {
541 return evaluateIf(cond, ctx, this.statusFns(scope), this.extra());
542 }
543
544 evalExpr(expr: string, ctx: Record<string, unknown>, scope: Scope): unknown {
545 return evaluate(expr, ctx, this.statusFns(scope), this.extra());
546 }
547
548 /**
549 * Layers workflow env over the runner's defaults. Like GitHub, workflows can
550 * override defaults such as CI or DIFFINERY, but not GITHUB_* or RUNNER_*.
551 */
552 withDefaults(envCtx: Record<string, string>): Record<string, string> {
553 const env = { ...this.defaultEnv, ...envCtx };
554 for (const [k, v] of Object.entries(this.defaultEnv)) if (/^(GITHUB|RUNNER)_/.test(k)) env[k] = v;
555 return env;
556 }
557
558 /** Builds the environment for a child process. */
559 processEnv(envCtx: Record<string, string>, extra: Record<string, string> = {}): NodeJS.ProcessEnv {
560 const env: NodeJS.ProcessEnv = { ...this.baseEnv, ...this.withDefaults(envCtx), ...extra };
561 const basePath = envCtx.PATH ?? this.baseEnv.PATH ?? "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";
562 env.PATH = [...this.pathAccum, basePath].filter(Boolean).join(path.delimiter);
563 return env;
564 }
565
566 // ------------------------------------------------------------ steps
567
568 /**
569 * Evaluates `if`, env, name; runs the step; updates the scope's steps
570 * context and failure state.
571 */
572 async executeStep(step: StepSpec, scope: Scope, logIndex: number, onStart?: (name: string) => Promise<void>): Promise<StepResult & { name: string }> {
573 const baseEnv = this.baseEnvContext(scope);
574 let ctx = this.contexts(scope, baseEnv);
575 let name = defaultStepName(step);
576 try {
577 name = step.name !== undefined ? this.interp(String(step.name), ctx, scope) : name;
578 } catch {
579 /* keep default */
580 }
581 const id = step.id;
582 const record = (outcome: Outcome, conclusion: Outcome, outputs: Record<string, string>) => {
583 if (id) scope.steps[id] = { outputs, outcome, conclusion };
584 };
585
586 let run: boolean;
587 try {
588 run = evaluateIf(step.if, ctx, this.statusFns(scope), this.extra());
589 } catch (e) {
590 this.logger.line(logIndex, `##[error]Error evaluating 'if' condition: ${errMsg(e)}`);
591 this.noteFailure(logIndex, `Error evaluating 'if' condition of step '${name}': ${errMsg(e)}`);
592 scope.failed = true;
593 record("failure", "failure", {});
594 return { outcome: "failure", conclusion: "failure", outputs: {}, name };
595 }
596 if (!run) {
597 if (scope.depth > 0) this.logger.line(logIndex, `Skipping step '${name}' (condition not met)`);
598 record("skipped", "skipped", {});
599 return { outcome: "skipped", conclusion: "skipped", outputs: {}, name };
600 }
601 await onStart?.(name);
602 if (scope.depth === 0) await this.prefetch.wait(logIndex, (m) => this.logger.line(logIndex, m));
603
604 const io: StepIO = { logIndex, outputs: {}, state: {} };
605 let outcome: Outcome = "success";
606 let continueOnError = false;
607 try {
608 continueOnError = this.evalBoolish(step["continue-on-error"], ctx, scope);
609 const stepEnv: Record<string, string> = {};
610 const envCtx = { ...baseEnv };
611 ctx = this.contexts(scope, envCtx);
612 for (const [k, v] of Object.entries(step.env ?? {})) {
613 stepEnv[k] = this.interp(String(v ?? ""), ctx, scope);
614 envCtx[k] = stepEnv[k]!;
615 }
616 const timeoutMin = this.evalNumber(step["timeout-minutes"], ctx, scope);
617 const timeoutMs = timeoutMin > 0 ? timeoutMin * 60_000 : 0;
618 if (step.run !== undefined) {
619 await runScript(this, { step, scope, ctx, envCtx, io, timeoutMs, name });
620 } else if (step.uses) {
621 await runAction(this, { step, scope, ctx, envCtx, io, timeoutMs, name });
622 } else {
623 throw new StepFailure("Step has neither `run` nor `uses`.");
624 }
625 if (this.halting) outcome = "cancelled";
626 } catch (e) {
627 if (this.halting) {
628 outcome = "cancelled";
629 } else {
630 outcome = "failure";
631 const msg = errMsg(e);
632 if (!(e instanceof StepFailure && (e as StepFailure & { logged?: boolean }).logged)) {
633 this.logger.line(logIndex, `##[error]${msg}`);
634 this.addAnnotation(logIndex, { level: "error", message: msg });
635 }
636 io.reason ??= msg;
637 }
638 }
639
640 let conclusion: Outcome = outcome;
641 if (outcome === "failure") {
642 if (continueOnError) {
643 conclusion = "success";
644 this.logger.line(logIndex, "##[warning]Step failed but continue-on-error is set; continuing.");
645 } else {
646 scope.failed = true;
647 this.noteFailure(logIndex, io.reason);
648 }
649 }
650 record(outcome, conclusion, io.outputs);
651 return { outcome, conclusion, outputs: io.outputs, name };
652 }
653
654 private evalBoolish(v: unknown, ctx: Record<string, unknown>, scope: Scope): boolean {
655 if (v === undefined || v === null || v === false) return false;
656 if (v === true) return true;
657 let s = String(v).trim();
658 if (s.includes("${{")) s = this.interp(s, ctx, scope);
659 return s === "true";
660 }
661
662 private evalNumber(v: unknown, ctx: Record<string, unknown>, scope: Scope): number {
663 if (v === undefined || v === null || v === "") return 0;
664 if (typeof v === "number") return v;
665 let s = String(v);
666 if (s.includes("${{")) s = this.interp(s, ctx, scope);
667 const n = Number(s);
668 return Number.isFinite(n) ? n : 0;
669 }
670
671 addAnnotation(logIndex: number, a: Annotation): void {
672 let list = this.stepErrors.get(logIndex);
673 if (!list) this.stepErrors.set(logIndex, (list = []));
674 list.push(a);
675 }
676
677 private noteFailure(logIndex: number, reason?: string): void {
678 if (this.firstFailure) return;
679 const errs = (this.stepErrors.get(logIndex) ?? []).filter((a) => a.level === "error");
680 // prefer annotations emitted by the step itself over our generic "exit code" message
681 const userErr = [...errs].reverse().find((a) => a.message !== reason);
682 let message: string;
683 if (userErr) {
684 const loc = userErr.file ? `${userErr.file}${userErr.line ? `:${userErr.line}` : ""}: ` : "";
685 message = `${loc}${userErr.title ? `${userErr.title}: ` : ""}${userErr.message}`;
686 } else {
687 const tail = this.logger.tail(logIndex, 20).filter((l) => !l.startsWith("##[error]"));
688 message = [reason ?? "Step failed", ...(tail.length ? ["", ...tail] : [])].join("\n");
689 }
690 this.firstFailure = { step: logIndex >= 0 ? logIndex : undefined, message };
691 if (this.firstFailure.step === undefined) delete this.firstFailure.step;
692 }
693
694 /**
695 * Runs a process for a step with fresh file-command files, interpreting
696 * workflow commands, then applies GITHUB_ENV/PATH/OUTPUT/STATE/SUMMARY.
697 */
698 async runStepProcess(
699 argv: string[],
700 o: {
701 cwd: string;
702 envCtx: Record<string, string>;
703 io: StepIO;
704 timeoutMs?: number;
705 extraEnv?: Record<string, string>;
706 timeoutLabel?: string;
707 /** Run the process somewhere else (the job container, or a Docker action's container). */
708 wrap?: Wrap;
709 },
710 ): Promise<ExecResult> {
711 const id = randomUUID();
712 const files = {
713 GITHUB_ENV: path.join(this.fileCmdDir, `set_env_${id}`),
714 GITHUB_PATH: path.join(this.fileCmdDir, `add_path_${id}`),
715 GITHUB_OUTPUT: path.join(this.fileCmdDir, `set_output_${id}`),
716 GITHUB_STATE: path.join(this.fileCmdDir, `save_state_${id}`),
717 GITHUB_STEP_SUMMARY: path.join(this.fileCmdDir, `step_summary_${id}`),
718 };
719 await mkdir(this.fileCmdDir, { recursive: true });
720 await Promise.all(Object.values(files).map((f) => writeFile(f, "")));
721 const proc = this.processor(o.io);
722 if (!existsSync(o.cwd)) {
723 throw new StepFailure(`An error occurred trying to start process '${argv[0]}' with working directory '${o.cwd}'. No such file or directory`);
724 }
725 let cmd = argv;
726 let cwd = o.cwd;
727 let env = this.processEnv(o.envCtx, { ...(o.extraEnv ?? {}), ...files });
728 if (o.wrap) ({ argv: cmd, cwd, env } = o.wrap(argv, { ...this.withDefaults(o.envCtx), ...(o.extraEnv ?? {}), ...files }, o.cwd));
729 const res = await exec(cmd[0]!, cmd.slice(1), {
730 cwd,
731 env,
732 timeoutMs: o.timeoutMs,
733 onLine: (l) => proc.handle(l),
734 onSpawn: (c) => this.track(c),
735 });
736 await this.applyFileCommands(files, o.io);
737 if (res.signal && this.replacing && !this.stopping) {
738 // The platform's SIGTERM (a rollout, or a move) reaches every process in the container, not only the agent. The job
739 // can't succeed now, so stop it and report at once: Diffinery starts it again on a new runner.
740 const msg = "The platform stopped this step's processes while stopping the runner.";
741 this.logger.line(o.io.logIndex, `##[error]${msg}`);
742 this.addAnnotation(o.io.logIndex, { level: "error", message: msg });
743 // No steps run after this cancel: the platform is stopping the container, and the job may start again elsewhere.
744 this.cancel("cancelled", 0);
745 }
746 if (res.timedOut) {
747 const mins = (o.timeoutMs ?? 0) / 60000;
748 const msg = `The ${o.timeoutLabel ?? "step"} has timed out after ${Number(mins.toFixed(2))} minutes.`;
749 this.logger.line(o.io.logIndex, `##[error]${msg}`);
750 this.addAnnotation(o.io.logIndex, { level: "error", message: msg });
751 o.io.reason = msg;
752 }
753 return res;
754 }
755
756 processor(io: StepIO): OutputProcessor {
757 return new OutputProcessor(this.logger, this.masker, io, {
758 debug: this.debug,
759 onAnnotation: (a) => this.addAnnotation(io.logIndex, a),
760 });
761 }
762
763 private async applyFileCommands(files: Record<string, string>, io: StepIO): Promise<void> {
764 const read = async (f: string) => {
765 try {
766 return await readFile(f, "utf8");
767 } catch {
768 return "";
769 }
770 };
771 const [envC, pathC, outC, stateC, sumC] = await Promise.all([
772 read(files.GITHUB_ENV!),
773 read(files.GITHUB_PATH!),
774 read(files.GITHUB_OUTPUT!),
775 read(files.GITHUB_STATE!),
776 read(files.GITHUB_STEP_SUMMARY!),
777 ]);
778 const safe = (label: string, fn: () => void) => {
779 try {
780 fn();
781 } catch (e) {
782 this.logger.line(io.logIndex, `##[error]Unable to process file command '${label}' successfully: ${errMsg(e)}`);
783 this.addAnnotation(io.logIndex, { level: "error", message: `Unable to process file command '${label}': ${errMsg(e)}` });
784 }
785 };
786 safe("env", () => {
787 for (const [k, v] of parseKeyValueFile(envC)) {
788 if (k === "NODE_OPTIONS" && /--require|-r\b/.test(v)) continue;
789 this.envAccum[k] = v;
790 }
791 });
792 safe("path", () => {
793 for (const p of parsePathFile(pathC)) this.addPath(p);
794 });
795 safe("output", () => {
796 for (const [k, v] of parseKeyValueFile(outC)) io.outputs[k] = v;
797 });
798 safe("state", () => {
799 for (const [k, v] of parseKeyValueFile(stateC)) io.state[k] = v;
800 });
801 if (sumC.trim()) this.summaries.push(sumC);
802 }
803
804 addPath(p: string): void {
805 this.pathAccum = [p, ...this.pathAccum.filter((x) => x !== p)];
806 }
807
808 /** Runs an internal tool (git, tar, ...) streaming its output into the step log. */
809 async tool(cmd: string, args: string[], o: { io: StepIO; cwd?: string; env?: NodeJS.ProcessEnv; capture?: boolean; silent?: boolean; input?: string | Readable }): Promise<ExecResult> {
810 return exec(cmd, args, {
811 cwd: o.cwd ?? this.workspace,
812 env: o.env ?? this.processEnv(this.baseEnvContext({ steps: {}, inputs: {}, failed: false, inheritedEnv: {}, composite: false, depth: 0 })),
813 capture: o.capture,
814 input: o.input,
815 onLine: o.silent ? undefined : (l) => this.logger.line(o.io.logIndex, l),
816 onSpawn: (c) => this.track(c),
817 });
818 }
819
820 registerPost(p: PostStep): void {
821 this.posts.push(p);
822 }
823
824 private async runPosts(top: Scope): Promise<void> {
825 const posts = [...this.posts].reverse();
826 let index = this.spec.steps.length;
827 for (const p of posts) {
828 const logIndex = index++;
829 if (this.forced) break;
830 this.resumeAfterCancel();
831 let ok: boolean;
832 try {
833 ok = evaluateIf(p.condition, this.contexts(top, this.baseEnvContext(top)), this.statusFns(top), this.extra());
834 } catch {
835 ok = false;
836 }
837 if (!ok) continue;
838 const startedAt = nowIso();
839 await this.report(logIndex, { status: "in_progress", name: p.name, startedAt });
840 let conclusion: Conclusion = "success";
841 try {
842 await p.run(logIndex);
843 } catch (e) {
844 conclusion = "failure";
845 const msg = errMsg(e);
846 if (!(e instanceof StepFailure && (e as StepFailure & { logged?: boolean }).logged)) {
847 this.logger.line(logIndex, `##[error]${msg}`);
848 this.addAnnotation(logIndex, { level: "error", message: msg });
849 }
850 top.failed = true;
851 this.noteFailure(logIndex, msg);
852 }
853 if (this.halting) conclusion = "cancelled";
854 await this.report(logIndex, { status: "completed", conclusion, name: p.name, startedAt, completedAt: nowIso() });
855 }
856 }
857
858 private jobOutputs(top: Scope): Record<string, string> {
859 const out: Record<string, string> = {};
860 const ctx = this.contexts(top, this.baseEnvContext(top));
861 for (const [k, expr] of Object.entries(this.spec.outputs ?? {})) {
862 try {
863 const v = this.interp(String(expr), ctx, top);
864 if (this.masker.containsSecret(v)) {
865 this.logger.line(-1, `##[warning]Skip output '${k}' since it may contain secret.`);
866 continue;
867 }
868 out[k] = v;
869 } catch (e) {
870 this.logger.line(-1, `##[warning]Failed to evaluate job output '${k}': ${errMsg(e)}`);
871 }
872 }
873 return out;
874 }
875
876 private async report(index: number, r: StepReport): Promise<void> {
877 try {
878 await this.api.postStep(index, { ...r, name: this.masker.mask(r.name) });
879 } catch (e) {
880 this.logger.line(-1, `##[warning]Failed to report step ${index}: ${errMsg(e)}`);
881 }
882 }
883}
884
885export function errMsg(e: unknown): string {
886 return e instanceof Error ? e.message : String(e);
887}
888
889export function failure(message: string, logged = false): StepFailure {
890 const f = new StepFailure(message) as StepFailure & { logged?: boolean };
891 f.logged = logged;
892 return f;
893}
894
895export function defaultStepName(step: StepSpec): string {
896 if (step.name) return String(step.name);
897 if (step.uses) return `Run ${step.uses}`;
898 if (step.run) return `Run ${String(step.run).trim().split("\n")[0]}`;
899 return "Step";
900}
901
902function safeName(step: StepSpec, fallback: string): string {
903 return step.name && !String(step.name).includes("${{") ? String(step.name) : fallback;
904}