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.

0858d86

Branches

claude/community-replies claude/connector-bad-gateway-doc claude/contact-page claude/early-access-cta claude/github-discoverability claude/github-star-buttons claude/graphql-issues claude/mcp-links claude/merged-pr-diff claude/numeric-run-ids claude/readme-reorganize claude/readme-screenshots-up claude/showcase-agents-plugins community-replies docs-kvm-android docs-release-please-workflow-edit lists-archived main marketplace-builds mcp-registry-auth mcp-registry-docs plugins-any-instance privacy-page release-please-graphql

Tags

v0.1.0
diffinery/apps/diffinery/src/repo/actions.ts
2206 lines · 118.9 KBRaw History
1import { ZERO_OID, diffTrees, fromUtf8, isTreeMode, readCommit, readTree, resolvePath } from "@diffinery/git-core";
2import type { Actor, Conclusion, Json, JobCompletion, JobSpec, LogLine, NeedsResult, StepReport } from "@diffinery/shared";
3import {
4 evaluateIf,
5 evaluateValue,
6 expandMatrix,
7 interpolate,
8 isWarning,
9 jobOrder,
10 matchesTrigger,
11 parseWorkflow,
12 type ExprContext,
13 type Job,
14 type StatusFns,
15 type TriggerEvent,
16 type Workflow,
17} from "@diffinery/workflow-yaml";
18import { decryptSecret, encryptSecret } from "../lib/crypto.js";
19import { Directory } from "../lib/directory.js";
20import { getFirst, headFirst, legacyArtifactKeys, legacyCacheKeys, logKeys } from "../lib/storage-keys.js";
21import { permissionsForJob, type JobPermissions } from "../lib/job-permissions.js";
22import { signRuntimeToken } from "../lib/runtime-auth.js";
23import { ghRepo, numericId } from "../http/gh.js";
24import { HttpError, newId, now, randomToken, sha256 } from "../lib/util.js";
25import { localOnly } from "../auth/dev.js";
26import { endCompute, mayStart, startCompute } from "../lib/quotas.js";
27import { blockedBy, killSwitches, type Work } from "../lib/killswitch.js";
28import type { Env } from "../env.js";
29import type { RepoDO } from "./RepoDO.js";
30import { completePayload } from "./event-payloads.js";
31import { CACHE_IDLE_MS, CACHE_LIMIT_BYTES, artifactFileName, artifactObjectKeys } from "./runtime.js";
32import { atLeast, type Caller } from "./types.js";
33import { isJobCaller } from "./token-policy.js";
34import { ReusableError, expandReusable, localKey, parseCallRef, type Expanded, type Source } from "./reusable.js";
35import { ancestors, combineResults, jobStatus, runnerLabels, type JobNode } from "./job-graph.js";
36import { schedulerStub } from "../lib/schedules.js";
37import { APPROVAL_DAYS, type Deployment, type DeploymentReview, type Reviewer } from "./deployments.js";
38import { referencedSecrets, RUNNER_SECRETS, secretsFor } from "./job-secrets.js";
39
40/** How long a sandbox token works after the sandbox starts. */
41export const SANDBOX_TOKEN_TTL_MS = 24 * 3600_000;
42/** Sandboxes one person or agent may have open in a repository at once: their containers are shared by the whole instance. */
43export const MAX_OPEN_SANDBOXES = 5;
44
45const WORKFLOW_DIRS = [".github/workflows", ".diffinery/workflows"];
46/** Characters of text a job's log keeps; the rest is dropped after a line saying so. */
47export const MAX_JOB_LOG = 16 * 1048576;
48/** Every push reads and parses a commit's workflow files: at most this many, each at most this large. */
49export const MAX_WORKFLOW_FILES = 256;
50export const MAX_WORKFLOW_BYTES = 512 * 1024;
51/** Hosted jobs one repository runs at once (GitHub's free plan allows 20 per account); more wait their turn. */
52export const MAX_HOSTED_JOBS = 20;
53/**
54 * A hosted job whose container never got a machine (the platform had none free, usually when many
55 * jobs start at once or right after a deploy) goes back in line and starts again later, this many
56 * times, before it fails. Only jobs whose runner never reported in: nothing of theirs has run.
57 */
58export const MAX_PLACEMENT_TRIES = 8;
59/** Seconds before the next start of a job whose container got no machine: 30 s, doubling to 5 minutes. */
60export const placementDelay = (tries: number): number => Math.min(30 * 2 ** (tries - 1), 300);
61const NO_INSTANCE = "no container instance";
62/** Runs one repository may have unfinished at once; past it, new ones aren't started. */
63export const MAX_ACTIVE_RUNS = 500;
64/** Self-hosted runners one repository may have registered. */
65export const MAX_RUNNERS = 100;
66/** How long a runner registration token works, as on GitHub. */
67const REGISTRATION_TTL_MS = 3600_000;
68/** A runner that polled this recently shows as online. */
69const RUNNER_ONLINE_MS = 90_000;
70
71type RunnerRow = { id: number; name: string; labels: string; token_hash: string | null; ephemeral: number; created_by: string; created_at: string; last_seen_at: string | null };
72
73export interface RunnerInfo {
74 id: number;
75 name: string;
76 os: string;
77 status: "online" | "offline";
78 busy: boolean;
79 ephemeral: boolean;
80 labels: Array<{ id: number; name: string; type: "read-only" | "custom" }>;
81}
82const SUPPORTED_LABEL = /^(ubuntu-.*|linux|diffinery(-.*)?|self-hosted|x64|amd64)$/i;
83
84export type RunRow = {
85 id: string;
86 run_number: number;
87 workflow_path: string;
88 workflow_name: string;
89 event: string;
90 ref: string;
91 sha: string;
92 actor: string;
93 status: "queued" | "in_progress" | "completed" | "waiting";
94 conclusion: Conclusion | "timed_out" | null;
95 created_at: string;
96 updated_at: string;
97 pr_number: number | null;
98 attempt: number;
99 event_payload: string;
100 inputs: string;
101 concurrency_group: string | null;
102 display_title: string;
103 /** Commit the run reports checks on, when it differs from `sha` (pull request runs). */
104 head_sha: string | null;
105 /** 1 when the code the run executes came from an agent; its jobs then act as an agent. */
106 agent_origin?: number;
107 /** Account ids that started earlier attempts of this run (JSON list), for prevent_self_review. */
108 prior_actors?: string | null;
109};
110
111export type JobRow = {
112 id: string;
113 run_id: string;
114 job_key: string;
115 name: string;
116 /**
117 * pending: held back by its concurrency group until the group's running job finishes.
118 * approval: waiting on its environment's required reviewers or wait timer; nothing has started.
119 */
120 status: "queued" | "pending" | "approval" | "waiting" | "in_progress" | "completed";
121 conclusion: Conclusion | "timed_out" | null;
122 matrix: string;
123 strategy: string;
124 started_at: string | null;
125 completed_at: string | null;
126 outputs: string;
127 steps: string;
128 failure: string | null;
129 token_hash: string | null;
130 summary: string | null;
131 container: string | null;
132 timeout_minutes: number;
133 /** runs-on labels for self-hosted jobs (JSON array); null for Diffinery-hosted containers. */
134 labels: string | null;
135 /** `continue-on-error` as evaluated for this matrix leg (0 or 1). */
136 continue_on_error: number;
137 /** Job-level `concurrency` group as evaluated for this leg, and its cancel-in-progress (0 or 1). */
138 concurrency_group: string | null;
139 concurrency_cancel: number;
140 log_chars: number;
141 fleet_held: number;
142 runner_id: number | null;
143 /** `environment:` as evaluated at launch (null for jobs without one, and for jobs launched before it was stored). */
144 environment: string | null;
145 /** When the job started waiting on its environment's protection rules, and "released" once they passed. */
146 gate_since: string | null;
147 gate_state: string | null;
148 placement_tries: number;
149 runner_info: string | null;
150};
151
152/** Where a job ran, as its runner reported it: the runner image, the container's location and instance. */
153export type JobRunnerInfo = { image?: string; location?: string; instance?: string };
154
155/** Reads the runner's `x-diffinery-runner` header (JSON), keeping only short string fields. */
156export function parseRunnerInfo(header: string | null | undefined): JobRunnerInfo | null {
157 if (!header || header.length > 1000) return null;
158 try {
159 const raw = JSON.parse(header) as Record<string, unknown>;
160 const out: JobRunnerInfo = {};
161 for (const k of ["image", "location", "instance"] as const) if (typeof raw[k] === "string" && raw[k]) out[k] = (raw[k] as string).slice(0, 200);
162 return Object.keys(out).length ? out : null;
163 } catch {
164 return null;
165 }
166}
167
168/** An environment a run's jobs wait on (GitHub's pending deployment). */
169export type PendingDeployment = {
170 environment: { id: number; name: string };
171 waitTimer: number;
172 waitTimerStartedAt: string;
173 reviewers: Reviewer[];
174 preventSelfReview: boolean;
175 /** Still waiting for a required reviewer (false once approved, or when only a timer applies). */
176 needsReview: boolean;
177 currentUserCanApprove: boolean;
178 /** The viewer is listed but started the run, and the environment prevents self-review. */
179 selfBlocked: boolean;
180};
181
182export type RunInfo = {
183 id: string;
184 number: number;
185 attempt: number;
186 workflowPath: string;
187 workflowName: string;
188 title: string;
189 event: string;
190 ref: string;
191 sha: string;
192 /** The commit checks are reported on: the pull request head for pull_request runs. */
193 headSha: string;
194 /** Branch the run is about: the pull request's head branch for pull request runs. */
195 headBranch: string;
196 actor: Actor;
197 status: RunRow["status"];
198 conclusion: RunRow["conclusion"];
199 createdAt: string;
200 updatedAt: string;
201 prNumber: number | null;
202};
203
204export type JobInfo = {
205 id: string;
206 runId: string;
207 key: string;
208 name: string;
209 status: JobRow["status"];
210 conclusion: JobRow["conclusion"];
211 matrix: Record<string, Json>;
212 startedAt: string | null;
213 completedAt: string | null;
214 outputs: Record<string, string>;
215 steps: Array<StepReport & { index: number }>;
216 failure: JobCompletion["failure"] | null;
217 summary: string | null;
218 /** The runner that took the job: a self-hosted runner's name, or the hosted container's name. */
219 runnerName: string | null;
220 runner: JobRunnerInfo | null;
221};
222
223const toRun = (r: RunRow): RunInfo => ({
224 id: r.id,
225 number: r.run_number,
226 attempt: r.attempt,
227 workflowPath: r.workflow_path,
228 workflowName: r.workflow_name,
229 title: r.display_title,
230 event: r.event,
231 ref: r.ref,
232 sha: r.sha,
233 headSha: r.head_sha ?? r.sha,
234 headBranch: (r.pr_number ? (JSON.parse(r.event_payload) as { pull_request?: { head?: { ref?: string } } }).pull_request?.head?.ref : undefined) ?? r.ref.replace(/^refs\/(heads|tags)\//, ""),
235 actor: JSON.parse(r.actor),
236 status: r.status,
237 conclusion: r.conclusion,
238 createdAt: r.created_at,
239 updatedAt: r.updated_at,
240 prNumber: r.pr_number,
241});
242
243const toJob = (j: JobRow): JobInfo => ({
244 id: j.id,
245 runId: j.run_id,
246 key: j.job_key,
247 name: j.name,
248 status: j.status,
249 conclusion: j.conclusion,
250 matrix: JSON.parse(j.matrix),
251 startedAt: j.started_at,
252 completedAt: j.completed_at,
253 outputs: JSON.parse(j.outputs),
254 steps: JSON.parse(j.steps),
255 failure: j.failure ? JSON.parse(j.failure) : null,
256 summary: j.summary,
257 runnerName: j.container?.replace(/^self-hosted:/, "") ?? null,
258 runner: j.runner_info ? (JSON.parse(j.runner_info) as JobRunnerInfo) : null,
259});
260
261/**
262 * GitHub Actions compatible run orchestration state. The Cloudflare Workflow
263 * (`RunWorkflow`) drives a run through these methods; the runner agent inside
264 * each job's container reports back through the runner* methods.
265 */
266export class Actions {
267 constructor(
268 private repo: RepoDO,
269 private env: Env,
270 ) {}
271
272 private get sql() {
273 return this.repo.sql;
274 }
275
276 // ---------------------------------------------------------------- workflow discovery
277
278 async loadWorkflows(sha: string): Promise<Array<{ path: string; workflow?: Workflow; errors: string[] }>> {
279 const commit = await readCommit(this.repo.store, sha);
280 const out: Array<{ path: string; workflow?: Workflow; errors: string[] }> = [];
281 for (const dir of WORKFLOW_DIRS) {
282 const entry = await resolvePath(this.repo.store, commit.tree, dir);
283 if (!entry || !isTreeMode(entry.mode)) continue;
284 for (const f of await readTree(this.repo.store, entry.oid)) {
285 if (!/\.ya?ml$/.test(f.name) || isTreeMode(f.mode)) continue;
286 if (out.length >= MAX_WORKFLOW_FILES) break;
287 const path = `${dir}/${f.name}`;
288 if ((this.repo.store.size(f.oid) ?? 0) > MAX_WORKFLOW_BYTES) {
289 out.push({ path, errors: [`: the workflow file is larger than ${MAX_WORKFLOW_BYTES / 1024} KiB`] });
290 continue;
291 }
292 const o = await this.repo.store.read(f.oid);
293 if (!o) continue;
294 const { workflow, errors } = parseWorkflow(fromUtf8(o.content), path);
295 out.push({ path, workflow, errors: errors.filter((e) => !isWarning(e)).map((e) => `${e.path}${e.line ? `:${e.line}` : ""}: ${e.message}`) });
296 }
297 }
298 return out;
299 }
300
301 async listWorkflows(): Promise<Array<{ path: string; name: string; errors: string[]; triggers: string[] }>> {
302 const sha = await this.repo.resolve("HEAD");
303 if (!sha) return [];
304 return (await this.loadWorkflows(sha)).map((w) => ({ path: w.path, name: w.workflow?.name ?? w.path.split("/").pop()!, errors: w.errors, triggers: w.workflow ? Object.keys(w.workflow.on) : [] }));
305 }
306
307 /** The `repository` object of event payloads: GitHub's shape, with the repository's real visibility. */
308 async repositoryPayload(): Promise<Record<string, unknown>> {
309 const m = this.repo.meta;
310 const row = await new Directory(this.env).repoById(m.repoId);
311 if (row) return ghRepo(this.env.PUBLIC_URL, row);
312 return { id: numericId(m.repoId), name: m.name, full_name: `${m.owner}/${m.name}`, owner: { login: m.owner }, default_branch: m.defaultBranch, private: true, visibility: "private", html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}`, clone_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git` };
313 }
314
315 // ---------------------------------------------------------------- triggers
316
317 async onPush(ev: { ref: string; before: string; after: string; forced: boolean; created: boolean; deleted: boolean }, caller: Caller): Promise<void> {
318 if (ev.deleted) return;
319 const workflows = await this.loadWorkflows(ev.after);
320 let changedFiles: string[] | undefined;
321 if (ev.before !== ZERO_OID && this.repo.store.has(ev.before)) {
322 const [a, b] = await Promise.all([readCommit(this.repo.store, ev.before), readCommit(this.repo.store, ev.after)]);
323 changedFiles = (await diffTrees(this.repo.store, a.tree, b.tree)).map((c) => c.path);
324 }
325 const head = await readCommit(this.repo.store, ev.after);
326 const payload = {
327 ref: ev.ref,
328 before: ev.before,
329 after: ev.after,
330 created: ev.created,
331 deleted: ev.deleted,
332 forced: ev.forced,
333 repository: await this.repositoryPayload(),
334 pusher: { name: caller.login },
335 sender: { login: caller.login, type: caller.kind === "agent" ? "Bot" : "User" },
336 head_commit: { id: ev.after, message: head.message, timestamp: new Date(head.committer.timestamp * 1000).toISOString(), author: { name: head.author.name, email: head.author.email } },
337 commits: [{ id: ev.after, message: head.message }],
338 };
339 if (ev.ref === `refs/heads/${this.repo.meta.defaultBranch}`) await this.syncSchedules(workflows);
340 for (const w of workflows) {
341 if (!w.workflow) {
342 if (w.errors.length) await this.createFailedRun(w.path, "push", ev.ref, ev.after, caller, w.errors);
343 continue;
344 }
345 if (!matchesTrigger(w.workflow, { name: "push", ref: ev.ref, changedFiles })) continue;
346 await this.createRun(w.workflow, { event: "push", ref: ev.ref, sha: ev.after, payload, caller, title: head.message.split("\n")[0] ?? "" });
347 }
348 }
349
350 /**
351 * Pull request workflows, like GitHub's: `pull_request` runs the workflow from the test
352 * merge commit (`refs/pull/<n>/merge`, the head merged into the base; nothing runs while
353 * they conflict), and `pull_request_target` runs the base branch's own workflow on the
354 * base commit, so a pull request can never change what runs with the base's secrets.
355 * Both report their checks on the pull request's head commit.
356 */
357 async onPullRequest(action: string, number: number, caller: Caller): Promise<void> {
358 const { issue } = this.repo.collab.get(number);
359 const pull = issue.pull!;
360 // Once merged, the base already holds the changes: keep the refs as they were and let path filters pass.
361 const { baseSha, mergeSha } = pull.merged ? { baseSha: this.repo.getRef(`refs/heads/${pull.base}`), mergeSha: null } : await this.repo.collab.refreshPullRefs(number);
362 const changedFiles = pull.merged ? undefined : (await this.repo.collab.pullFiles(number)).files.map((f) => f.path);
363 const m = this.repo.meta;
364 const repository = await this.repositoryPayload();
365 const payload = {
366 action,
367 number,
368 pull_request: {
369 number,
370 id: number,
371 state: issue.state,
372 title: issue.title,
373 body: issue.body,
374 draft: pull.draft,
375 merged: pull.merged,
376 merged_at: pull.mergedAt,
377 merged_by: pull.mergedBy ? { login: pull.mergedBy } : null,
378 merge_commit_sha: pull.merged ? pull.mergeCommit : mergeSha,
379 user: { login: issue.author.login },
380 labels: issue.labels.map((name) => ({ name })),
381 assignees: issue.assignees.map((login) => ({ login })),
382 head: { ref: pull.head, sha: pull.headSha, repo: repository },
383 base: { ref: pull.base, sha: baseSha, repo: repository },
384 html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/pull/${number}`,
385 url: `${this.env.PUBLIC_URL}/api/v3/repos/${m.owner}/${m.name}/pulls/${number}`,
386 },
387 repository,
388 sender: { login: caller.login },
389 };
390 const common = { payload, caller, prNumber: number, headSha: pull.headSha, headRef: pull.head, baseRef: pull.base, title: issue.title };
391 // A merged pull request's runs use its merge commit, as on GitHub.
392 const prSha = pull.merged ? pull.mergeCommit : mergeSha;
393 if (prSha) {
394 for (const w of await this.loadWorkflows(prSha)) {
395 if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request", action, baseRef: pull.base, changedFiles })) continue;
396 await this.createRun(w.workflow, { ...common, event: "pull_request", ref: `refs/pull/${number}/merge`, sha: prSha });
397 }
398 }
399 if (baseSha) {
400 for (const w of await this.loadWorkflows(baseSha)) {
401 if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request_target", action, baseRef: pull.base, changedFiles })) continue;
402 await this.createRun(w.workflow, { ...common, event: "pull_request_target", ref: `refs/heads/${pull.base}`, sha: baseSha });
403 }
404 }
405 }
406
407 /** `merge_group` workflows for a merge queue group, run from the group's commit. */
408 async onMergeGroup(g: { number: number; base: string; ref: string; sha: string; parent: string; title: string; sender: Actor }): Promise<void> {
409 const head = await readCommit(this.repo.store, g.sha);
410 const payload = {
411 action: "checks_requested",
412 merge_group: {
413 head_sha: g.sha,
414 head_ref: g.ref,
415 base_sha: g.parent,
416 base_ref: `refs/heads/${g.base}`,
417 head_commit: {
418 id: g.sha,
419 tree_id: head.tree,
420 message: head.message,
421 timestamp: new Date(head.committer.timestamp * 1000).toISOString(),
422 author: { name: head.author.name, email: head.author.email },
423 committer: { name: head.committer.name, email: head.committer.email },
424 },
425 },
426 repository: await this.repositoryPayload(),
427 sender: { login: g.sender.login },
428 };
429 for (const w of await this.loadWorkflows(g.sha)) {
430 if (!w.workflow) {
431 if (w.errors.length) await this.createFailedRun(w.path, "merge_group", g.ref, g.sha, g.sender, w.errors);
432 continue;
433 }
434 if (!matchesTrigger(w.workflow, { name: "merge_group", action: "checks_requested", baseRef: g.base })) continue;
435 await this.createRun(w.workflow, { event: "merge_group", ref: g.ref, sha: g.sha, payload, caller: g.sender, title: `Merge group for #${g.number}: ${g.title}`, prNumbers: [g.number] });
436 }
437 }
438
439 /** Cancel unfinished runs on a ref (a merge group branch that is going away). */
440 async cancelRunsOnRef(ref: string, reason: string): Promise<void> {
441 for (const r of this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE ref = ? AND status != 'completed'", ref).toArray()) await this.cancelInternal(r.id, reason);
442 }
443
444 async dispatch(workflowFile: string, ref: string, inputs: Record<string, unknown>, caller: Caller): Promise<RunInfo> {
445 if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to run workflows");
446 await this.refuseIfBlocked("run");
447 const sha = await this.repo.resolve(ref);
448 if (!sha) throw new HttpError(422, `unknown ref ${ref}`);
449 const all = await this.loadWorkflows(sha);
450 const w = all.find((x) => x.path === workflowFile || x.path.endsWith(`/${workflowFile}`));
451 if (!w?.workflow) throw new HttpError(404, `workflow ${workflowFile} not found at ${ref}${w?.errors.length ? `: ${w.errors.join("; ")}` : ""}`);
452 if (!w.workflow.on.workflow_dispatch) throw new HttpError(422, "workflow does not have a workflow_dispatch trigger");
453 const defs = w.workflow.on.workflow_dispatch.inputs ?? {};
454 const resolved: Record<string, unknown> = {};
455 for (const [k, d] of Object.entries(defs)) {
456 const v = inputs[k] ?? d.default;
457 if (d.required && (v === undefined || v === "")) throw new HttpError(422, `input '${k}' is required`);
458 if (d.type === "choice" && d.options && v !== undefined && !d.options.includes(String(v))) throw new HttpError(422, `input '${k}' must be one of ${d.options.join(", ")}`);
459 resolved[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v;
460 }
461 const fullRef = ref.startsWith("refs/") ? ref : this.repo.getRef(`refs/heads/${ref}`) ? `refs/heads/${ref}` : `refs/tags/${ref}`;
462 const payload = { inputs: resolved, ref: fullRef, repository: await this.repositoryPayload(), sender: { login: caller.login }, workflow: w.path };
463 const m = this.repo.meta;
464 this.repo.emit({ type: "workflow_dispatch", repoId: m.repoId, owner: m.owner, repo: m.name, ref: fullRef, workflow: w.path, inputs: resolved as Record<string, Json>, actor: { id: caller.id, login: caller.login, kind: caller.kind, onBehalfOf: caller.onBehalfOf }, at: now() });
465 return (await this.createRun(w.workflow, { event: "workflow_dispatch", ref: fullRef, sha, payload, caller, inputs: resolved, title: w.workflow.name ?? w.path }))!;
466 }
467
468 async schedule(workflowPath: string, cron: string): Promise<void> {
469 const sha = await this.repo.resolve("HEAD");
470 if (!sha) return;
471 const w = (await this.loadWorkflows(sha)).find((x) => x.path === workflowPath);
472 if (!w?.workflow || !matchesTrigger(w.workflow, { name: "schedule", cron })) return;
473 const system: Caller = { id: "system", login: "diffinery-scheduler", kind: "system", role: "write" };
474 await this.createRun(w.workflow, { event: "schedule", ref: this.repo.headRef, sha, payload: { schedule: cron, repository: await this.repositoryPayload() }, caller: system, title: w.workflow.name ?? w.path });
475 }
476
477 private async syncSchedules(workflows: Array<{ path: string; workflow?: Workflow }>): Promise<void> {
478 const repoId = this.repo.meta.repoId;
479 const stmts = [this.env.DB.prepare("DELETE FROM schedules WHERE repo_id = ?").bind(repoId)];
480 for (const w of workflows) for (const s of w.workflow?.on.schedule ?? []) stmts.push(this.env.DB.prepare("INSERT OR IGNORE INTO schedules (repo_id, workflow_path, cron) VALUES (?, ?, ?)").bind(repoId, w.path, s.cron));
481 await this.env.DB.batch(stmts);
482 // The scheduler's alarm starts with the first schedule, without waiting for the cron trigger.
483 if (stmts.length > 1) await schedulerStub(this.env).ensure().catch((e) => console.error("scheduler alarm check failed", e));
484 }
485
486 // ---------------------------------------------------------------- runs
487
488 private nextRunNumber(): number {
489 this.sql.exec("INSERT INTO counters (name, value) VALUES ('run', 1) ON CONFLICT(name) DO UPDATE SET value = value + 1");
490 return this.sql.exec<{ value: number }>("SELECT value FROM counters WHERE name = 'run'").one().value;
491 }
492
493 private githubContext(run: RunRow, jobKey: string, token: string): Record<string, unknown> {
494 const m = this.repo.meta;
495 const payload = JSON.parse(run.event_payload) as Record<string, unknown>;
496 const actor = JSON.parse(run.actor) as Actor;
497 const refName = run.ref.replace(/^refs\/(heads|tags)\//, "").replace(/^refs\/pull\/(\d+)\/(head|merge)$/, "$1/merge");
498 const pr = payload.pull_request as { head?: { ref?: string }; base?: { ref?: string } } | undefined;
499 return {
500 event_name: run.event,
501 event: payload,
502 sha: run.sha,
503 ref: run.ref,
504 ref_name: refName,
505 ref_type: run.ref.startsWith("refs/tags/") ? "tag" : "branch",
506 ref_protected: false,
507 head_ref: pr?.head?.ref ?? "",
508 base_ref: pr?.base?.ref ?? "",
509 repository: `${m.owner}/${m.name}`,
510 repository_id: String(numericId(m.repoId)),
511 repository_owner: m.owner,
512 repositoryUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`,
513 actor: actor.login,
514 triggering_actor: actor.login,
515 // Numeric like GitHub's (actions parse it, e.g. download-artifact's run-id); the API accepts it as a run id.
516 run_id: String(run.run_number),
517 run_number: String(run.run_number),
518 run_attempt: String(run.attempt),
519 workflow: run.workflow_name,
520 workflow_ref: `${m.owner}/${m.name}/${run.workflow_path}@${run.ref}`,
521 job: jobKey,
522 server_url: this.env.PUBLIC_URL,
523 api_url: `${this.env.PUBLIC_URL}/api/v3`,
524 graphql_url: `${this.env.PUBLIC_URL}/api/graphql`,
525 token,
526 retention_days: "90",
527 secret_source: "Actions",
528 };
529 }
530
531 private async createFailedRun(path: string, event: string, ref: string, sha: string, caller: Actor, errors: string[]): Promise<void> {
532 // While Actions are suspended no run is recorded, not even a failed one.
533 if (await this.blocked("run")) return;
534 const id = newId();
535 const t = now();
536 this.sql.exec(
537 "INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, conclusion, created_at, updated_at, event_payload, display_title) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'completed', 'failure', ?, ?, ?, ?)",
538 id,
539 this.nextRunNumber(),
540 path,
541 path,
542 event,
543 ref,
544 sha,
545 JSON.stringify({ id: caller.id, login: caller.login, kind: caller.kind }),
546 t,
547 t,
548 "{}",
549 `Invalid workflow file: ${errors[0] ?? path}`,
550 );
551 const jobId = newId();
552 this.sql.exec(
553 "INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'invalid', 'Invalid workflow file', 'completed', 'failure', ?, ?, ?)",
554 jobId,
555 id,
556 t,
557 t,
558 JSON.stringify({ message: errors.join("\n") }),
559 );
560 }
561
562 async createRun(
563 wf: Workflow,
564 o: {
565 event: string;
566 ref: string;
567 sha: string;
568 headSha?: string;
569 payload: Record<string, unknown>;
570 caller: Caller | { id: string; login: string; kind: Actor["kind"] };
571 prNumber?: number;
572 inputs?: Record<string, unknown>;
573 headRef?: string;
574 baseRef?: string;
575 title: string;
576 /** Started by a run whose code came from an agent (a re-run, or a workflow_run follower). */
577 agentOrigin?: boolean;
578 /** Pull requests whose code the run includes (a merge group's). */
579 prNumbers?: number[];
580 },
581 ): Promise<RunInfo | null> {
582 // Like GitHub, what a job does with its GITHUB_TOKEN (pushes, pull requests) starts no new
583 // runs, so workflows cannot trigger themselves in a loop; workflow_dispatch and
584 // repository_dispatch are the exceptions.
585 if (isJobCaller(o.caller) && o.event !== "workflow_dispatch" && o.event !== "repository_dispatch") return null;
586 // A site administrator suspended Actions here (lib/killswitch.ts): events start nothing, and a
587 // dispatch, which someone asked for, is refused with the reason.
588 const block = await this.blocked("run");
589 if (block) {
590 if (o.event === "workflow_dispatch" || o.event === "repository_dispatch") throw new HttpError(403, block);
591 return null;
592 }
593 // A dispatch loop (or a flood of pushes) stops growing here.
594 const active = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE status != 'completed'").one().n;
595 if (active >= MAX_ACTIVE_RUNS) {
596 if (o.event === "workflow_dispatch" || o.event === "repository_dispatch") throw new HttpError(429, `this repository has ${active} unfinished workflow runs, the most allowed; wait for some to finish`);
597 console.warn(`not starting ${wf.path}: ${active} unfinished runs`);
598 return null;
599 }
600 const id = newId();
601 const t = now();
602 const run: RunRow = {
603 id,
604 run_number: this.nextRunNumber(),
605 workflow_path: wf.path,
606 workflow_name: wf.name ?? wf.path,
607 event: o.event,
608 ref: o.ref,
609 sha: o.sha,
610 actor: JSON.stringify({ id: o.caller.id, login: o.caller.login, kind: o.caller.kind }),
611 status: "queued",
612 conclusion: null,
613 created_at: t,
614 updated_at: t,
615 pr_number: o.prNumber ?? null,
616 attempt: 1,
617 event_payload: JSON.stringify(completePayload({ publicUrl: this.env.PUBLIC_URL, meta: this.repo.meta, visibility: (o.payload.repository as { visibility?: string } | undefined)?.visibility === "public" ? "public" : "private" }, o.payload, o.caller)),
618 inputs: JSON.stringify(o.inputs ?? {}),
619 concurrency_group: null,
620 display_title: o.title,
621 head_sha: o.headSha && o.headSha !== o.sha ? o.headSha : null,
622 agent_origin: this.repo.policy.agentOrigin({
623 caller: o.caller,
624 ref: o.ref,
625 prNumbers: [...(o.prNumber !== undefined ? [o.prNumber] : []), ...(o.prNumbers ?? [])],
626 inherited: o.agentOrigin,
627 })
628 ? 1
629 : 0,
630 };
631 const ctx: ExprContext = { github: this.githubContext(run, "", ""), inputs: o.inputs ?? {}, vars: this.variables() };
632 try {
633 if (wf.runName) run.display_title = interpolate(wf.runName, ctx);
634 if (wf.concurrency) run.concurrency_group = interpolate(wf.concurrency.group, ctx);
635 } catch (e) {
636 console.warn("run-name/concurrency evaluation failed", e);
637 }
638 this.sql.exec(
639 "INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, pr_number, attempt, event_payload, inputs, concurrency_group, display_title, head_sha, agent_origin) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
640 run.id,
641 run.run_number,
642 run.workflow_path,
643 run.workflow_name,
644 run.event,
645 run.ref,
646 run.sha,
647 run.actor,
648 run.status,
649 run.created_at,
650 run.updated_at,
651 run.pr_number,
652 run.attempt,
653 run.event_payload,
654 run.inputs,
655 run.concurrency_group,
656 run.display_title,
657 run.head_sha,
658 run.agent_origin ?? 0,
659 );
660 if (run.concurrency_group && wf.concurrency) {
661 const cancelInProgress = evaluateValue(wf.concurrency.cancelInProgress, ctx);
662 const others = this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE concurrency_group = ? AND status != 'completed' AND id != ?", run.concurrency_group, id).toArray();
663 for (const other of others) if (cancelInProgress === true || cancelInProgress === "true") await this.cancelInternal(other.id, "cancelled by a newer run in the same concurrency group");
664 }
665 await this.env.RUN_WORKFLOW.create({ id, params: { repoId: this.repo.meta.repoId, runId: id } });
666 const m = this.repo.meta;
667 this.repo.emit({ type: "workflow_run", action: "requested", repoId: m.repoId, owner: m.owner, repo: m.name, runId: id, actor: JSON.parse(run.actor), at: t });
668 this.repo.broadcast("runs", { action: "created", runId: id });
669 await this.onWorkflowRun(run, "requested");
670 return toRun(run);
671 }
672
673 /** A run by id, or by its number (github.run_id; the latest attempt). */
674 private runRow(id: string): RunRow {
675 const r = /^\d+$/.test(id)
676 ? this.sql.exec<RunRow>("SELECT * FROM runs WHERE run_number = ? ORDER BY attempt DESC, created_at DESC LIMIT 1", Number(id)).toArray()[0]
677 : this.sql.exec<RunRow>("SELECT * FROM runs WHERE id = ?", id).toArray()[0];
678 if (!r) throw new HttpError(404, "run not found");
679 return r;
680 }
681
682 private jobRow(id: string): JobRow {
683 const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", id).toArray()[0];
684 if (!j) throw new HttpError(404, "job not found");
685 return j;
686 }
687
688 listRuns(opts: { workflow?: string; branch?: string; status?: string; event?: string; sha?: string; prNumber?: number; limit?: number; offset?: number } = {}): { runs: RunInfo[]; total: number } {
689 const where: string[] = [];
690 const args: unknown[] = [];
691 if (opts.workflow) {
692 where.push("(workflow_path = ? OR workflow_path LIKE ?)");
693 args.push(opts.workflow, `%/${opts.workflow}`);
694 }
695 if (opts.branch) {
696 where.push("ref = ?");
697 args.push(`refs/heads/${opts.branch}`);
698 }
699 if (opts.status) {
700 where.push(opts.status === "completed" || opts.status === "queued" || opts.status === "in_progress" ? "status = ?" : "conclusion = ?");
701 args.push(opts.status);
702 }
703 if (opts.event) {
704 where.push("event = ?");
705 args.push(opts.event);
706 }
707 if (opts.sha) {
708 where.push("(sha = ? OR head_sha = ?)");
709 args.push(opts.sha, opts.sha);
710 }
711 if (opts.prNumber) {
712 where.push("pr_number = ?");
713 args.push(opts.prNumber);
714 }
715 const w = where.length ? `WHERE ${where.join(" AND ")}` : "";
716 const total = this.sql.exec<{ n: number }>(`SELECT count(*) AS n FROM runs ${w}`, ...args).one().n;
717 const rows = this.sql.exec<RunRow>(`SELECT * FROM runs ${w} ORDER BY created_at DESC LIMIT ? OFFSET ?`, ...args, Math.min(opts.limit ?? 25, 100), opts.offset ?? 0).toArray();
718 return { runs: rows.map(toRun), total };
719 }
720
721 getRun(id: string): RunInfo & { jobs: JobInfo[] } {
722 const run = this.runRow(id);
723 return { ...toRun(run), jobs: this.jobs(run.id) };
724 }
725
726 jobs(runId: string): JobInfo[] {
727 if (/^\d+$/.test(runId)) runId = this.runRow(runId).id;
728 return this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid", runId).toArray().map(toJob);
729 }
730
731 getJob(id: string): JobInfo {
732 return toJob(this.jobRow(id));
733 }
734
735 async logs(jobId: string, afterSeq = -1): Promise<{ lines: Array<LogLine & { seq: number }>; complete: boolean }> {
736 const j = this.jobRow(jobId);
737 const rows = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT 5000", jobId, afterSeq).toArray();
738 if (rows.length === 0 && j.status === "completed") {
739 const archived = await getFirst(this.env.BUCKET, logKeys(this.repo.meta.repoId, jobId));
740 if (archived && afterSeq < 0) return { lines: (await archived.json()) as Array<LogLine & { seq: number }>, complete: true };
741 }
742 return { lines: rows, complete: j.status === "completed" };
743 }
744
745 private logKey(jobId: string) {
746 return logKeys(this.repo.meta.repoId, jobId)[0];
747 }
748
749 // ---------------------------------------------------------------- orchestration (called by RunWorkflow)
750
751 /**
752 * The run's job graph: each job with the jobs it waits for (reusable workflow calls
753 * expanded). RunWorkflow starts a job as soon as its own needs have finished.
754 */
755 async plan(runId: string): Promise<{ jobs: JobNode[]; error?: string; concurrencyBlocked: boolean }> {
756 const run = this.runRow(runId);
757 if (run.status === "completed") return { jobs: [], error: "run already completed", concurrencyBlocked: false };
758 let exp: Expanded;
759 try {
760 exp = await this.expanded(run);
761 } catch (e) {
762 return { jobs: [], error: (e as Error).message, concurrencyBlocked: false };
763 }
764 const { order, errors } = jobOrder(exp.workflow);
765 if (errors.length) return { jobs: [], error: errors.map((e) => e.message).join("; "), concurrencyBlocked: false };
766 const blocked =
767 !!run.concurrency_group &&
768 this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE concurrency_group = ? AND status = 'in_progress' AND id != ?", run.concurrency_group, runId).one().n > 0;
769 if (!blocked) this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'queued'", now(), runId);
770 return { jobs: order.flat().map((key) => ({ key, needs: exp.workflow.jobs[key]!.needs })), concurrencyBlocked: blocked };
771 }
772
773 /** The run's workflow with reusable workflow calls expanded (see ./reusable.ts). */
774 private async expanded(run: RunRow): Promise<Expanded> {
775 const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path);
776 if (!wf?.workflow) throw new ReusableError(`workflow file not found or invalid: ${wf?.errors.join("; ") || run.workflow_path}`);
777 return expandReusable(wf.workflow, { repoId: null, sha: run.sha }, (uses, from) => this.loadCalled(uses, from));
778 }
779
780 /** Load a called workflow: `./path` from the same commit, or `owner/repo/path@ref` from a repository on this instance. */
781 private async loadCalled(uses: string, from: Source): Promise<{ workflow: Workflow; source: Source }> {
782 const ref = parseCallRef(uses);
783 const m = this.repo.meta;
784 let source: Source;
785 let text: string | null;
786 if (ref.kind === "local" && from.repoId === null) {
787 source = from;
788 const b = await this.repo.blob(from.sha, ref.path);
789 text = b && !b.binary ? fromUtf8(b.content) : null;
790 } else {
791 let repoId: string;
792 if (ref.kind === "local") repoId = from.repoId!;
793 else if (ref.owner.toLowerCase() === m.owner.toLowerCase() && ref.repo.toLowerCase() === m.name.toLowerCase()) repoId = m.repoId;
794 else {
795 const dir = new Directory(this.env);
796 // A transferred or renamed repository's workflows are found under the old name too.
797 const target = (await dir.repo(ref.owner, ref.repo)) ?? (await dir.movedRepo(ref.owner, ref.repo));
798 // Like GitHub: public workflows can be shared; private ones only within the same owner.
799 if (!target || (target.visibility === "private" && target.owner_id !== m.ownerId)) throw new Error(`repository ${ref.owner}/${ref.repo} not found, or its workflows are not shared with ${m.owner}/${m.name}`);
800 repoId = target.id;
801 }
802 if (repoId === m.repoId) {
803 const sha = ref.kind === "local" ? from.sha : await this.repo.resolve(ref.ref);
804 if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""}`);
805 source = { repoId: null, sha };
806 const b = await this.repo.blob(sha, ref.path);
807 text = b && !b.binary ? fromUtf8(b.content) : null;
808 } else {
809 const stub = this.env.REPO.get(this.env.REPO.idFromName(repoId));
810 const sha = ref.kind === "local" ? from.sha : await stub.resolve(ref.ref);
811 if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""} in ${ref.kind === "remote" ? `${ref.owner}/${ref.repo}` : "the called repository"}`);
812 source = { repoId, sha };
813 const b = await stub.blob(sha, ref.path);
814 text = b && !b.binary ? fromUtf8(b.content) : null;
815 }
816 }
817 if (text === null) throw new Error(`${uses} not found`);
818 const { workflow, errors } = parseWorkflow(text, ref.path);
819 const fatal = errors.filter((e) => !isWarning(e));
820 if (!workflow || fatal.length) throw new Error(`${uses} is invalid: ${fatal.map((e) => `${e.path}: ${e.message}`).join("; ")}`);
821 return { workflow, source };
822 }
823
824 /** `needs` context for a job or call: keys are as written inside its own workflow. */
825 private async needsResults(run: RunRow, exp: Expanded, keys: string[], scope: string | null): Promise<Record<string, NeedsResult>> {
826 const out: Record<string, NeedsResult> = {};
827 for (const key of keys) out[localKey(key, scope)] = await this.resultOf(run, exp, key);
828 return out;
829 }
830
831 /** Status functions for the `if:` of a job or call: from the results of all its ancestors, like GitHub. */
832 private statusOf(run: RunRow, exp: Expanded, key: string): StatusFns {
833 const results = ancestors(exp.needs, exp.needs.get(key) ?? []).map((k) => combineResults(this.rowsOf(run, exp, k)));
834 return jobStatus(results, run.conclusion === "cancelled");
835 }
836
837 /** The job rows of a job, or of every job inside a reusable workflow call. */
838 private rowsOf(run: RunRow, exp: Expanded, key: string): Array<JobRow & { continueOnError: boolean }> {
839 const rows = exp.calls.has(key)
840 ? this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND substr(job_key, 1, ?) = ?", run.id, key.length + 1, `${key}/`).toArray()
841 : this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ?", run.id, key).toArray();
842 return rows.map((r) => ({ ...r, continueOnError: !!r.continue_on_error }));
843 }
844
845 /** Result and outputs of a job, or of a reusable workflow call (from its `on.workflow_call.outputs`). */
846 private async resultOf(run: RunRow, exp: Expanded, key: string): Promise<NeedsResult> {
847 const call = exp.calls.get(key);
848 const rows = this.rowsOf(run, exp, key);
849 const result = combineResults(rows);
850 let outputs: Record<string, string> = {};
851 if (!call) for (const r of rows) Object.assign(outputs, JSON.parse(r.outputs));
852 if (call && Object.keys(call.workflow.on.workflow_call?.outputs ?? {}).length) {
853 const jobs: Record<string, NeedsResult> = {};
854 for (const k of Object.keys(call.workflow.jobs)) jobs[k] = await this.resultOf(run, exp, `${key}/${k}`);
855 const { inputs } = await this.callContext(run, exp, key, false);
856 const ctx: ExprContext = { github: this.githubContext(run, localKey(key, call.parent), ""), jobs, inputs, vars: this.variables() };
857 outputs = {};
858 for (const [name, d] of Object.entries(call.workflow.on.workflow_call!.outputs!)) {
859 try {
860 const v = evaluateValue(d.value, ctx);
861 outputs[name] = v === undefined || v === null ? "" : typeof v === "object" ? JSON.stringify(v) : String(v);
862 } catch {
863 outputs[name] = "";
864 }
865 }
866 }
867 return { result, outputs };
868 }
869
870 /**
871 * `inputs` and `secrets` inside a reusable workflow call (or at the top level when
872 * `callKey` is null): inputs from the caller's `with` and the declared defaults,
873 * secrets from `secrets: inherit` or the caller's mapping.
874 */
875 private async callContext(run: RunRow, exp: Expanded, callKey: string | null, withSecrets: boolean): Promise<{ inputs: Record<string, unknown>; secrets: Record<string, string> }> {
876 if (callKey === null) return { inputs: JSON.parse(run.inputs), secrets: withSecrets ? { ...(await this.orgSecrets()), ...(await this.repoSecrets()) } : {} };
877 const call = exp.calls.get(callKey)!;
878 const parent = await this.callContext(run, exp, call.parent, withSecrets);
879 const ctx: ExprContext = { github: this.githubContext(run, localKey(callKey, call.parent), ""), needs: await this.needsResults(run, exp, call.needs, call.parent), inputs: parent.inputs, vars: this.variables() };
880 const inputs: Record<string, unknown> = {};
881 for (const [k, d] of Object.entries(call.workflow.on.workflow_call?.inputs ?? {})) {
882 let v = call.job.with && k in call.job.with ? evaluateValue(call.job.with[k], ctx) : evaluateValue(d.default, ctx);
883 if (v === undefined || v === null || v === "") {
884 if (d.required) throw new ReusableError(`jobs.${callKey}: input '${k}' is required by ${call.job.uses}`);
885 v = d.type === "boolean" ? false : d.type === "number" ? 0 : "";
886 }
887 inputs[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v;
888 }
889 const secrets: Record<string, string> = {};
890 if (withSecrets) {
891 if (call.job.secrets === "inherit") Object.assign(secrets, parent.secrets);
892 else for (const [name, expr] of Object.entries(call.job.secrets ?? {})) secrets[name.toUpperCase()] = String(evaluateValue(expr, { ...ctx, secrets: parent.secrets }) ?? "");
893 for (const [name, d] of Object.entries(call.workflow.on.workflow_call?.secrets ?? {})) {
894 if (d.required && !secrets[name.toUpperCase()]) throw new ReusableError(`jobs.${callKey}: secret '${name}' is required by ${call.job.uses}`);
895 }
896 }
897 return { inputs, secrets };
898 }
899
900 /**
901 * Start every matrix instance of a job. Returns the ids the workflow must wait
902 * for (instances beyond max-parallel are queued and started as others finish).
903 */
904 async startJobKey(runId: string, jobKey: string): Promise<{ jobIds: string[] }> {
905 const run = this.runRow(runId);
906 const exp = await this.expanded(run);
907 const job = exp.workflow.jobs[jobKey]!;
908 const owner = exp.workflowOf.get(jobKey)!;
909 const parent = exp.parentOf.get(jobKey) ?? null;
910 const chain: string[] = [];
911 for (let c = parent; c; c = exp.calls.get(c)!.parent) chain.unshift(c);
912 const prefix = chain.map((c) => exp.calls.get(c)!.job.name ?? localKey(c, exp.calls.get(c)!.parent)).join(" / ");
913 const label = (name: string) => (prefix ? `${prefix} / ${name}` : name);
914 const fail = (name: string, message: string, conclusion: Conclusion = "failure") => {
915 const id = newId();
916 const t = now();
917 this.sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, ?, ?, 'completed', ?, ?, ?, ?)", id, runId, jobKey, name, conclusion, t, t, conclusion === "failure" ? JSON.stringify({ message }) : null);
918 this.repo.broadcast(`run:${runId}`, { action: "job", jobId: id });
919 return { jobIds: [] as string[] };
920 };
921 const short = localKey(jobKey, parent);
922 let needs: Record<string, NeedsResult>;
923 let baseCtx: ExprContext;
924 let shouldRun: boolean;
925 try {
926 // The callers decide first: a skipped call, or one whose ancestors failed or were skipped, skips everything inside it.
927 for (const c of chain) {
928 const call = exp.calls.get(c)!;
929 const callNeeds = await this.needsResults(run, exp, call.needs, call.parent);
930 const outer = await this.callContext(run, exp, call.parent, false);
931 const callCtx: ExprContext = { github: this.githubContext(run, localKey(c, call.parent), ""), needs: callNeeds, inputs: outer.inputs, vars: this.variables() };
932 if (!evaluateIf(call.job.if, callCtx, this.statusOf(run, exp, c))) return fail(label(job.name ?? short), "", "skipped");
933 }
934 // Resolve the call's inputs and secrets now, so a missing required one fails here rather than in the runner.
935 if (parent) await this.callContext(run, exp, parent, true);
936 needs = await this.needsResults(run, exp, exp.needs.get(jobKey) ?? [], parent);
937 baseCtx = { github: this.githubContext(run, short, ""), needs, inputs: (await this.callContext(run, exp, parent, false)).inputs, vars: this.variables(), env: owner.env };
938 shouldRun = evaluateIf(job.if, baseCtx, this.statusOf(run, exp, jobKey));
939 } catch (e) {
940 return fail(label(job.name ?? short), e instanceof ReusableError ? e.message : `Error evaluating 'if': ${(e as Error).message}`);
941 }
942 if (!shouldRun) return fail(label(job.name ?? short), "", "skipped");
943
944 let combos: Array<Record<string, unknown>>;
945 try {
946 combos = expandMatrix(job.strategy?.matrix, baseCtx);
947 } catch (e) {
948 return fail(label(job.name ?? short), `Error expanding matrix: ${(e as Error).message}`);
949 }
950 if (combos.length > 256) return fail(label(job.name ?? short), "A matrix may produce at most 256 jobs.");
951 const failFast = evaluateValue(job.strategy?.failFast ?? true, baseCtx) !== false;
952 const maxParallel = Number(evaluateValue(job.strategy?.maxParallel ?? combos.length, baseCtx)) || combos.length;
953 const ids: string[] = [];
954 combos.forEach((matrix, i) => {
955 // Like GitHub, name, runs-on, timeout-minutes, continue-on-error and concurrency are evaluated for each matrix leg.
956 const ctx = { ...baseCtx, matrix, strategy: { "fail-fast": failFast, "job-index": i, "job-total": combos.length, "max-parallel": maxParallel } };
957 let name = short;
958 if (!job.name && Object.keys(matrix).length) name = `${short} (${Object.values(matrix).map((v) => (typeof v === "object" ? JSON.stringify(v) : String(v))).join(", ")})`;
959 let leg: { labels: string[]; timeout: number; continueOnError: boolean; group: string | null; cancel: boolean; environment: string | null };
960 try {
961 if (job.name) name = interpolate(job.name, ctx);
962 const continueOnError = evaluateValue(job.continueOnError, ctx);
963 const cancel = job.concurrency ? evaluateValue(job.concurrency.cancelInProgress, ctx) : false;
964 leg = {
965 labels: runnerLabels(job.runsOn, ctx),
966 timeout: Number(evaluateValue(job.timeoutMinutes, ctx)) || 360,
967 continueOnError: continueOnError === true || continueOnError === "true",
968 group: (job.concurrency ? interpolate(job.concurrency.group, ctx).trim() : "") || null,
969 cancel: cancel === true || cancel === "true",
970 // Evaluated once, here: its protection rules gate the launch, and runnerSpec deploys to this same name.
971 environment: job.environment ? interpolate(job.environment.name, ctx).trim() : null,
972 };
973 } catch (e) {
974 fail(label(name), `Error evaluating the job for this matrix combination: ${(e as Error).message}`);
975 return;
976 }
977 const unsupported = leg.labels.filter((l) => !SUPPORTED_LABEL.test(l));
978 const selfHosted = leg.labels.some((l) => l.toLowerCase() === "self-hosted");
979 if (!leg.labels.length || (unsupported.length && !selfHosted)) {
980 fail(label(name), `No Diffinery runner matches runs-on: ${unsupported.join(", ") || "(no labels)"}. Linux (ubuntu-*) runners are hosted; anything else needs runs-on: [self-hosted, ...] and a registered runner.`);
981 return;
982 }
983 const id = newId();
984 this.sql.exec(
985 "INSERT INTO jobs (id, run_id, job_key, name, status, matrix, strategy, timeout_minutes, labels, continue_on_error, concurrency_group, concurrency_cancel, environment) VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?)",
986 id,
987 runId,
988 jobKey,
989 label(name),
990 JSON.stringify(matrix),
991 JSON.stringify(ctx.strategy),
992 Math.min(leg.timeout, 360),
993 selfHosted ? JSON.stringify(leg.labels.map((l) => l.toLowerCase())) : null,
994 leg.continueOnError ? 1 : 0,
995 leg.group,
996 leg.cancel ? 1 : 0,
997 leg.environment,
998 );
999 ids.push(id);
1000 });
1001 for (const id of ids.slice(0, maxParallel)) await this.launch(id);
1002 this.repo.broadcast(`run:${runId}`, { action: "jobs" });
1003 return { jobIds: ids };
1004 }
1005
1006 private checkName(run: RunRow, job: JobRow): string {
1007 return `${run.workflow_name} / ${job.name}`;
1008 }
1009
1010 private async launch(jobId: string): Promise<void> {
1011 const job = this.jobRow(jobId);
1012 if (job.status === "completed") return;
1013 // Kill switches (lib/killswitch.ts): a suspended repository or owner starts no job; a paused instance no hosted one.
1014 const block = await this.blocked(job.labels ? "self-hosted-job" : "hosted-job");
1015 if (block) return this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job was not started: ${block}.` } });
1016 if (!(await this.gate(job)) || !(await this.concurrencySlot(job))) return;
1017 const run = this.runRow(job.run_id);
1018 const m = this.repo.meta;
1019 if (job.labels) {
1020 // Self-hosted: park the job until a registered runner with matching labels claims it.
1021 this.sql.exec("UPDATE jobs SET status = 'waiting', container = NULL WHERE id = ?", jobId);
1022 this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` });
1023 return;
1024 }
1025 // The account that owns the repository pays (lib/quotas.ts): past its monthly minutes the job
1026 // fails; past its hosted jobs at once it waits (fleet_held = 2, retried by the alarm) for one to end.
1027 const quota = await mayStart(this.env, m.ownerId, "job").catch((e: unknown) => {
1028 console.error("quota check failed", e);
1029 return { ok: true } as const;
1030 });
1031 if (!quota.ok) {
1032 if (!quota.wait) return this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job was not started: ${quota.message}.` } });
1033 this.sql.exec("UPDATE jobs SET status = 'queued', fleet_held = 2 WHERE id = ?", jobId);
1034 this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` });
1035 this.repo.armAlarm(null);
1036 return;
1037 }
1038 // One repository can't take the whole fleet: past its share, the job waits for one of its own to finish.
1039 const busy = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM jobs WHERE labels IS NULL AND container IS NOT NULL AND status IN ('waiting', 'in_progress') AND id != ?", jobId).one().n;
1040 if (busy >= MAX_HOSTED_JOBS) {
1041 this.sql.exec("UPDATE jobs SET status = 'queued', fleet_held = 1 WHERE id = ?", jobId);
1042 this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` });
1043 return;
1044 }
1045 this.sql.exec("UPDATE jobs SET fleet_held = 0 WHERE id = ?", jobId);
1046 const token = `djt_${m.repoId}_${jobId}_${randomToken(24)}`;
1047 const container = `job-${jobId}`;
1048 this.sql.exec("UPDATE jobs SET status = 'waiting', token_hash = ?, container = ? WHERE id = ?", await sha256(token), container, jobId);
1049 await startCompute(this.env, { id: jobId, accountId: m.ownerId, repoId: m.repoId, kind: "job" }).catch((e: unknown) => console.error("compute usage failed", e));
1050 this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` });
1051 try {
1052 const stub = this.env.RUNNER.get(this.env.RUNNER.idFromName(container));
1053 await stub.startJob({ repoId: m.repoId, jobId, token, apiUrl: this.env.PUBLIC_URL });
1054 } catch (e) {
1055 const message = (e as Error).message;
1056 if (message.includes(NO_INSTANCE)) return this.placementFailed(jobId, message);
1057 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${message}` } });
1058 }
1059 }
1060
1061
1062 // ---------------------------------------------------------------- environment protection rules
1063
1064 private jobUrl(run: RunRow, jobId: string): string {
1065 const m = this.repo.meta;
1066 return `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`;
1067 }
1068
1069 /**
1070 * Whose run this is, for prevent_self_review, as account ids (logins can be renamed): who started
1071 * it and its earlier attempts, and for each of them that is an agent, the account that owns it.
1072 */
1073 async selfIds(run: RunRow): Promise<Set<string>> {
1074 const actor = JSON.parse(run.actor) as Actor;
1075 const ids = new Set([actor.id, ...((run.prior_actors ? JSON.parse(run.prior_actors) : []) as string[])].filter(Boolean));
1076 const dir = new Directory(this.env);
1077 for (const id of [...ids]) {
1078 const a = await dir.account(id).catch(() => null);
1079 if (a?.kind === "agent" && a.owner_id) ids.add(a.owner_id);
1080 }
1081 return ids;
1082 }
1083
1084 /**
1085 * A job whose environment has required reviewers or a wait timer waits here, before a token is
1086 * minted or a runner started, in the status `approval`: no concurrency slot, no hosted slot.
1087 * Returns whether the job may go on to start. The branch policy is checked first, so a ref that
1088 * may never deploy fails at once instead of asking anyone.
1089 */
1090 private async gate(job: JobRow): Promise<boolean> {
1091 if (!job.environment || job.gate_state === "released") return true;
1092 const rules = this.repo.deploys.protection(job.environment);
1093 if (!rules) {
1094 if (job.status === "approval") this.release(job);
1095 return true;
1096 }
1097 const run = this.runRow(job.run_id);
1098 const fail = async (message: string) => {
1099 await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message } });
1100 return false;
1101 };
1102 if (!this.repo.deploys.refAllowed(job.environment, run.ref)) return fail(`${run.ref.replace(/^refs\/(heads|tags)\//, "")} is not allowed to deploy to ${job.environment} due to environment protection rules.`);
1103 const decision = rules.reviewers.length ? this.repo.deploys.decision(run.id, job.environment) : null;
1104 if (decision === "rejected") {
1105 const r = this.repo.deploys.reviews(run.id).find((x) => x.environment.toLowerCase() === job.environment!.toLowerCase() && x.state === "rejected")!;
1106 return fail(`The deployment to ${job.environment} was rejected by ${r.user.login}${r.comment ? `: ${r.comment}` : "."}`);
1107 }
1108 let since = job.gate_since;
1109 if (job.status !== "approval") {
1110 since = now();
1111 this.sql.exec("UPDATE jobs SET status = 'approval', gate_since = ? WHERE id = ?", since, job.id);
1112 this.sql.exec("UPDATE runs SET status = 'waiting', updated_at = ? WHERE id = ? AND status != 'completed'", since, run.id);
1113 const actor = JSON.parse(run.actor) as Actor;
1114 this.repo.deploys.jobWaiting({ id: job.id, ref: run.ref, sha: run.sha, logUrl: this.jobUrl(run, job.id), actor: run.agent_origin ? { ...actor, kind: "agent" } : actor }, job.environment);
1115 this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId: job.id, detailsUrl: this.jobUrl(run, job.id) });
1116 // Reviewers hear about it once per run and environment, from its first job to wait.
1117 const first = !this.sql.exec("SELECT 1 FROM jobs WHERE run_id = ? AND environment = ? COLLATE NOCASE AND gate_since IS NOT NULL AND id != ?", run.id, job.environment, job.id).toArray().length;
1118 if (first && rules.reviewers.length && !decision) this.emitReview("requested", run, job.environment, rules.reviewers, actor, null, "");
1119 this.repo.broadcast(`run:${run.id}`, { action: "job", jobId: job.id });
1120 this.repo.broadcast("runs", { action: "waiting", runId: run.id });
1121 }
1122 const timerEnds = Date.parse(since!) + rules.waitTimer * timerMinute(this.env);
1123 if ((rules.reviewers.length && decision !== "approved") || Date.now() < timerEnds) return false;
1124 this.release({ ...job, status: "approval" });
1125 return true;
1126 }
1127
1128 /** The job's protection rules passed: it goes on to start, and its run stops waiting once no other job does. */
1129 private release(job: JobRow): void {
1130 this.sql.exec("UPDATE jobs SET gate_state = 'released', status = 'queued' WHERE id = ?", job.id);
1131 this.syncWaiting(job.run_id);
1132 }
1133
1134 private syncWaiting(runId: string): void {
1135 if (this.sql.exec("SELECT 1 FROM jobs WHERE run_id = ? AND status = 'approval'", runId).toArray().length) return;
1136 this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'waiting'", now(), runId);
1137 }
1138
1139 private async wakeGate(runId: string, jobId: string): Promise<void> {
1140 try {
1141 await (await this.env.RUN_WORKFLOW.get(runId)).sendEvent({ type: `gate-${jobId}`, payload: {} });
1142 } catch (e) {
1143 console.error("failed to wake a job waiting for approval", e);
1144 }
1145 }
1146
1147 private emitReview(action: "requested" | "approved" | "rejected", run: RunRow, environment: string, reviewers: Reviewer[], by: Actor, state: string | null, comment: string): void {
1148 const m = this.repo.meta;
1149 const url = `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}`;
1150 const ids = Object.fromEntries(this.repo.deploys.environments().map((e) => [e.name.toLowerCase(), e.id]));
1151 this.repo.emit({
1152 type: "deployment_review",
1153 action,
1154 repoId: m.repoId,
1155 owner: m.owner,
1156 repo: m.name,
1157 payload: {
1158 since: now(),
1159 environment: environment,
1160 environment_id: ids[environment.toLowerCase()] ?? null,
1161 comment,
1162 ...(state ? { state } : {}),
1163 workflow_run: { id: run.id, run_number: run.run_number, name: run.workflow_name, display_title: run.display_title, head_sha: run.head_sha ?? run.sha, head_branch: run.ref.replace(/^refs\/(heads|tags)\//, ""), html_url: url },
1164 reviewers: reviewers.map((r) => ({ type: "User", reviewer: { login: r.login, id: r.id } })),
1165 ...(action === "requested" ? { requestor: { login: by.login } } : { approver: { login: by.login } }),
1166 },
1167 actor: by,
1168 at: now(),
1169 });
1170 }
1171
1172 /**
1173 * Where a job waiting on its environment stands, for the run's Workflow: what it waits for and
1174 * until when. Re-checks the rules first, so a job whose time came starts here.
1175 */
1176 async jobGate(jobId: string): Promise<{ status: JobRow["status"]; wait: "review" | "timer" | null; until: string | null }> {
1177 let job = this.jobRow(jobId);
1178 if (job.status === "approval") await this.launch(jobId);
1179 job = this.jobRow(jobId);
1180 if (job.status !== "approval") return { status: job.status, wait: null, until: null };
1181 const rules = this.repo.deploys.protection(job.environment!) ?? { reviewers: [], waitTimer: 0, preventSelfReview: false };
1182 const since = Date.parse(job.gate_since!);
1183 const deadline = since + APPROVAL_DAYS * 86400_000;
1184 if (Date.now() >= deadline) {
1185 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The deployment to ${job.environment} wasn't approved within ${APPROVAL_DAYS} days.` } });
1186 return { status: "completed", wait: null, until: null };
1187 }
1188 const decided = this.repo.deploys.decision(job.run_id, job.environment!) === "approved";
1189 if (rules.reviewers.length && !decided) return { status: "approval", wait: "review", until: new Date(deadline).toISOString() };
1190 return { status: "approval", wait: "timer", until: new Date(since + rules.waitTimer * timerMinute(this.env)).toISOString() };
1191 }
1192
1193 /** The environments a run's jobs wait on, with who may approve them (GitHub's pending deployments). */
1194 async pendingDeployments(runId: string, viewer: { id: string; login: string; mayReview?: boolean } | null): Promise<PendingDeployment[]> {
1195 const run = this.runRow(runId);
1196 const waiting = this.sql.exec<{ environment: string; since: string }>("SELECT environment, MIN(gate_since) AS since FROM jobs WHERE run_id = ? AND status = 'approval' GROUP BY environment COLLATE NOCASE", run.id).toArray();
1197 const self = viewer ? await this.selfIds(run) : new Set<string>();
1198 return waiting.flatMap((w) => {
1199 const env = this.repo.deploys.environments().find((e) => e.name.toLowerCase() === w.environment.toLowerCase());
1200 if (!env) return [];
1201 const p = env.protection;
1202 const decided = this.repo.deploys.decision(run.id, env.name);
1203 // A listed reviewer who lost access to the repository no longer decides (the caller says so in mayReview).
1204 const listed = !!viewer && viewer.mayReview !== false && p.reviewers.some((r) => r.id === viewer.id);
1205 const selfBlocked = !!viewer && p.preventSelfReview && self.has(viewer.id);
1206 return [
1207 {
1208 environment: { id: env.id, name: env.name },
1209 waitTimer: p.waitTimer,
1210 waitTimerStartedAt: w.since,
1211 reviewers: p.reviewers,
1212 preventSelfReview: p.preventSelfReview,
1213 needsReview: p.reviewers.length > 0 && decided !== "approved",
1214 currentUserCanApprove: listed && !selfBlocked && p.reviewers.length > 0 && !decided,
1215 selfBlocked,
1216 },
1217 ];
1218 });
1219 }
1220
1221 /**
1222 * A required reviewer approves or rejects environments a run waits on. The caller checked that
1223 * `reviewer` is a person signed in or using a write token (never an agent or a job); here the
1224 * reviewer must be listed on every environment, and not whoever started the run when the
1225 * environment prevents self-review. One approval is enough. Approved jobs start once their
1226 * wait timer has also passed; rejected ones fail.
1227 */
1228 async reviewDeployments(runId: string, input: { environmentIds: number[]; state: "approved" | "rejected"; comment: string }, reviewer: { id: string; login: string; mayReview: boolean }): Promise<{ reviews: DeploymentReview[]; deployments: Deployment[] }> {
1229 const run = this.runRow(runId);
1230 if (!reviewer.mayReview) throw new HttpError(403, "only people with access to this repository review its deployments");
1231 if (!input.environmentIds.length) throw new HttpError(422, "environment_ids must name at least one environment");
1232 const pending = await this.pendingDeployments(run.id, reviewer);
1233 const chosen = input.environmentIds.map((id) => {
1234 const p = pending.find((x) => x.environment.id === id);
1235 if (!p || !p.needsReview) throw new HttpError(422, `environment ${id} isn't waiting for a review in this run`);
1236 if (p.selfBlocked) throw new HttpError(403, `${p.environment.name} doesn't let whoever started a run approve it`);
1237 if (!p.currentUserCanApprove) throw new HttpError(403, `you aren't a required reviewer for ${p.environment.name}`);
1238 return p;
1239 });
1240 const by: Actor = { id: reviewer.id, login: reviewer.login, kind: "user" };
1241 const out: DeploymentReview[] = [];
1242 const jobIds: string[] = [];
1243 for (const p of chosen) {
1244 out.push(this.repo.deploys.addReview(run.id, p.environment.name, input.state, input.comment, reviewer));
1245 this.emitReview(input.state, run, p.environment.name, p.reviewers, by, input.state, input.comment);
1246 const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND status = 'approval' AND environment = ? COLLATE NOCASE", run.id, p.environment.name).toArray();
1247 for (const j of jobs) {
1248 jobIds.push(j.id);
1249 await this.launch(j.id);
1250 await this.wakeGate(run.id, j.id);
1251 }
1252 }
1253 this.repo.broadcast(`run:${run.id}`, { action: "reviewed" });
1254 return { reviews: out, deployments: this.repo.deploys.jobDeployments(jobIds) };
1255 }
1256
1257 deploymentReviews(runId: string): DeploymentReview[] {
1258 return this.repo.deploys.reviews(this.runRow(runId).id);
1259 }
1260
1261 /**
1262 * The job's container got no machine (its start failed, or it stopped before the runner reported
1263 * in). Put the job back in line and have the run's workflow start it again after a pause, or fail it
1264 * once it has waited MAX_PLACEMENT_TRIES times.
1265 */
1266 private async placementFailed(jobId: string, why: string): Promise<void> {
1267 const job = this.jobRow(jobId);
1268 if (job.status === "completed") return;
1269 const tries = job.placement_tries + 1;
1270 if (tries > MAX_PLACEMENT_TRIES) {
1271 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `No hosted runner could be started: Diffinery tried ${MAX_PLACEMENT_TRIES} times over about half an hour (${why}). Re-run the job to try again.` } });
1272 return;
1273 }
1274 this.sql.exec("UPDATE jobs SET status = 'queued', container = NULL, token_hash = NULL, placement_tries = ? WHERE id = ?", tries, jobId);
1275 // Not running while it waits: it holds none of the account's hosted jobs and uses no minutes.
1276 await endCompute(this.env, jobId).catch((e: unknown) => console.error("compute usage failed", e));
1277 this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId });
1278 try {
1279 const instance = await this.env.RUN_WORKFLOW.get(job.run_id);
1280 await instance.sendEvent({ type: `job-${jobId}`, payload: { retryInSeconds: placementDelay(tries) } });
1281 } catch (e) {
1282 console.error("could not ask the run's workflow to start the job again", e);
1283 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${why}` } });
1284 }
1285 }
1286
1287 /** The run's workflow starts a job again after placementFailed put it back in line. */
1288 async relaunchJob(jobId: string): Promise<void> {
1289 if (this.jobRow(jobId).status === "queued") await this.launch(jobId);
1290 }
1291
1292 /**
1293 * Job-level `concurrency`, like GitHub: at most one job of a group runs at a time in
1294 * the repository. With cancel-in-progress the newcomer cancels the group's other jobs;
1295 * without it the newcomer waits as `pending` and replaces any older pending job.
1296 * Returns whether the job may start now.
1297 */
1298 private async concurrencySlot(job: JobRow): Promise<boolean> {
1299 if (!job.concurrency_group) return true;
1300 const others = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE concurrency_group = ? AND id != ? AND status IN ('pending', 'waiting', 'in_progress') ORDER BY rowid", job.concurrency_group, job.id).toArray();
1301 const pending = others.filter((o) => o.status === "pending");
1302 const running = others.filter((o) => o.status !== "pending");
1303 const failure = { message: `Canceling since a higher priority waiting request for '${job.concurrency_group}' exists` };
1304 if (running.length && !job.concurrency_cancel) {
1305 this.sql.exec("UPDATE jobs SET status = 'pending' WHERE id = ?", job.id);
1306 this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id });
1307 for (const o of pending) await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure });
1308 return false;
1309 }
1310 // Take the group before cancelling anything, so a job that starts meanwhile waits for this one.
1311 this.sql.exec("UPDATE jobs SET status = 'waiting' WHERE id = ?", job.id);
1312 for (const o of [...pending, ...running]) {
1313 await this.stopContainer(o);
1314 await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure });
1315 }
1316 return true;
1317 }
1318
1319 /** Whether jobs wait on the account's hosted-job quota (RepoDO retries them on its alarm). */
1320 hasAccountHeldJobs(): boolean {
1321 return this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM jobs WHERE fleet_held = 2 AND status = 'queued'").one().n > 0;
1322 }
1323
1324 /** Try again to start jobs waiting on the account's hosted-job quota, oldest first, until one waits again. */
1325 async releaseAccountHeld(): Promise<void> {
1326 for (const { id } of this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE fleet_held = 2 AND status = 'queued' ORDER BY rowid").toArray()) {
1327 await this.launch(id);
1328 if (this.jobRow(id).fleet_held === 2) return;
1329 }
1330 }
1331
1332 async jobTimedOut(jobId: string): Promise<void> {
1333 const j = this.jobRow(jobId);
1334 if (j.status === "completed") return;
1335 if (j.status === "queued" && j.fleet_held === 2) {
1336 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: "The job waited 6 hours for one of this account's hosted jobs to finish, and was not started." } }, "timed_out");
1337 return;
1338 }
1339 await this.stopContainer(j);
1340 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job exceeded its timeout of ${j.timeout_minutes} minutes.` } }, "timed_out");
1341 }
1342
1343 async containerStopped(jobId: string, exitCode: number | null): Promise<void> {
1344 const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", jobId).toArray()[0];
1345 if (!j || j.status === "completed") return;
1346 // Already back in line (placementFailed): a container given up on, which started after all,
1347 // finds its token revoked and exits. The job's next placement starts a new one.
1348 if (j.status === "queued") return;
1349 // Stopped before the runner fetched the job, so no step ran: place it again. During a
1350 // container rollout a new container can exit (or never get a machine) before it reports in.
1351 if (j.status === "waiting" && !j.labels) return this.placementFailed(jobId, exitCode ? `the runner exited with code ${exitCode} before it started the job` : "the container stopped before the runner started");
1352 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: exitCode ? `The runner stopped unexpectedly (exit code ${exitCode}).` : "The runner's container stopped before the job finished, without an exit code (the platform restarted or replaced it)." } });
1353 }
1354
1355 /** The oldest waiting job whose runs-on labels the runner all offers, handed to it with a fresh job token. */
1356 private async assignJob(runnerName: string, runnerLabels: string[], runnerId: number | null): Promise<{ jobId: string; token: string; runId: string; name: string } | null> {
1357 const offered = new Set(["self-hosted", ...runnerLabels.map((l) => l.toLowerCase())]);
1358 const waiting = this.sql
1359 .exec<JobRow>("SELECT j.* FROM jobs j JOIN runs r ON r.id = j.run_id WHERE j.status = 'waiting' AND j.labels IS NOT NULL AND j.container IS NULL AND r.status != 'completed' ORDER BY r.created_at, j.rowid")
1360 .toArray();
1361 const job = waiting.find((j) => (JSON.parse(j.labels!) as string[]).every((l) => offered.has(l)));
1362 if (!job) return null;
1363 // A job parked before Actions were suspended here isn't handed out.
1364 const block = await this.blocked("self-hosted-job");
1365 if (block) {
1366 await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message: `The job was not started: ${block}.` } });
1367 return null;
1368 }
1369 const m = this.repo.meta;
1370 const token = `djt_${m.repoId}_${job.id}_${randomToken(24)}`;
1371 const container = `self-hosted:${runnerName.slice(0, 64)}`;
1372 this.sql.exec("UPDATE jobs SET token_hash = ?, container = ?, runner_id = ? WHERE id = ? AND container IS NULL", await sha256(token), container, runnerId, job.id);
1373 this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id });
1374 return { jobId: job.id, token, runId: job.run_id, name: job.name };
1375 }
1376
1377 // ------------------------------------------------------------ registered self-hosted runners
1378
1379 /**
1380 * GitHub's runner registration token: a repository admin's credential, good for an hour, that a
1381 * machine trades for its own runner token. So the machine never holds a person's token, and the
1382 * jobs it runs (as the same OS user) can't read one.
1383 */
1384 async runnerRegistrationToken(caller: Caller): Promise<{ token: string; expires_at: string }> {
1385 if (caller.kind !== "user" || caller.permissions || !atLeast(caller.role, "admin")) throw new HttpError(403, "only repository admins can add self-hosted runners");
1386 const token = `drg_${randomToken(32)}`;
1387 const expires = new Date(Date.now() + REGISTRATION_TTL_MS).toISOString();
1388 const hash = await sha256(token);
1389 this.sql.exec("DELETE FROM runner_registrations WHERE expires_at < ?", now());
1390 this.sql.exec("INSERT INTO runner_registrations (token_hash, created_by, expires_at) VALUES (?, ?, ?)", hash, caller.login, expires);
1391 return { token, expires_at: expires };
1392 }
1393
1394 /** Register a runner with a registration token; returns its own token, shown once. */
1395 async registerRunner(registration: string, name: string, labels: string[], ephemeral: boolean): Promise<{ id: number; name: string; labels: string[]; ephemeral: boolean; token: string; registeredBy: string }> {
1396 const hash = await sha256(registration);
1397 const reg = this.sql.exec<{ created_by: string }>("SELECT created_by FROM runner_registrations WHERE token_hash = ? AND expires_at > ?", hash, now()).toArray()[0];
1398 if (!reg) throw new HttpError(401, "invalid or expired registration token");
1399 const clean = [...new Set(labels.map((l) => String(l).trim().toLowerCase()).filter((l) => l && l !== "self-hosted"))];
1400 if (clean.length > 20) throw new HttpError(422, "a runner can offer at most 20 labels");
1401 const bad = clean.find((l) => !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(l));
1402 if (bad) throw new HttpError(422, `invalid label ${JSON.stringify(bad.slice(0, 70))}: use letters, digits, '.', '_' and '-' (up to 64)`);
1403 const runnerName = name.trim().slice(0, 64) || "runner";
1404 const count = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runners").one().n;
1405 if (count >= MAX_RUNNERS) throw new HttpError(429, `this repository has ${count} self-hosted runners, the most allowed; remove one first`);
1406 const id = this.sql.exec<{ id: number }>("INSERT INTO runners (name, labels, ephemeral, created_by, created_at) VALUES (?, ?, ?, ?, ?) RETURNING id", runnerName, JSON.stringify(clean), ephemeral ? 1 : 0, reg.created_by, now()).one().id;
1407 const token = `drn_${this.repo.meta.repoId}_${id}_${randomToken(32)}`;
1408 this.sql.exec("UPDATE runners SET token_hash = ? WHERE id = ?", await sha256(token), id);
1409 return { id, name: runnerName, labels: clean, ephemeral, token, registeredBy: reg.created_by };
1410 }
1411
1412 /** A registered runner asks for work with its own token. An ephemeral runner's token ends with the job it takes. */
1413 async claimJobAsRunner(token: string): Promise<{ jobId: string; token: string; runId: string; name: string } | null> {
1414 const id = Number(token.split("_")[2]);
1415 const r = Number.isSafeInteger(id) ? this.sql.exec<RunnerRow>("SELECT * FROM runners WHERE id = ?", id).toArray()[0] : undefined;
1416 if (!r || !r.token_hash || r.token_hash !== (await sha256(token))) throw new HttpError(401, "this runner was removed, or its token is invalid; register it again");
1417 this.sql.exec("UPDATE runners SET last_seen_at = ? WHERE id = ?", now(), r.id);
1418 const job = await this.assignJob(r.name, JSON.parse(r.labels) as string[], r.id);
1419 if (job && r.ephemeral) this.sql.exec("UPDATE runners SET token_hash = NULL WHERE id = ?", r.id);
1420 return job;
1421 }
1422
1423 runners(caller: Caller): RunnerInfo[] {
1424 if (!atLeast(caller.role, "admin") || caller.permissions) throw new HttpError(403, "only repository admins can list self-hosted runners");
1425 const busy = new Set(this.sql.exec<{ runner_id: number }>("SELECT DISTINCT runner_id FROM jobs WHERE runner_id IS NOT NULL AND status != 'completed'").toArray().map((x) => x.runner_id));
1426 return this.sql.exec<RunnerRow>("SELECT * FROM runners ORDER BY id").toArray().map((r) => ({
1427 id: r.id,
1428 name: r.name,
1429 os: "unknown",
1430 status: r.last_seen_at && Date.now() - Date.parse(r.last_seen_at) < RUNNER_ONLINE_MS ? "online" : "offline",
1431 busy: busy.has(r.id),
1432 ephemeral: !!r.ephemeral,
1433 labels: ["self-hosted", ...(JSON.parse(r.labels) as string[])].map((name, i) => ({ id: i + 1, name, type: i === 0 ? "read-only" : "custom" })),
1434 }));
1435 }
1436
1437 /** Remove a runner: its token stops working at once. A job it is running carries on with its own job token. */
1438 deleteRunner(id: number, caller: Caller): void {
1439 if (caller.kind !== "user" || caller.permissions || !atLeast(caller.role, "admin")) throw new HttpError(403, "only repository admins can remove self-hosted runners");
1440 if (!this.sql.exec("DELETE FROM runners WHERE id = ? RETURNING id", id).toArray().length) throw new HttpError(404, "runner not found");
1441 }
1442
1443 private async stopContainer(j: JobRow): Promise<void> {
1444 if (!j.container || j.container.startsWith("self-hosted:")) return;
1445 try {
1446 await this.env.RUNNER.get(this.env.RUNNER.idFromName(j.container)).stopJob();
1447 } catch (e) {
1448 console.warn("stop container failed", e);
1449 }
1450 }
1451
1452 async finishRun(runId: string): Promise<RunInfo> {
1453 const run = this.runRow(runId);
1454 const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ?", runId).toArray();
1455 let conclusion: RunRow["conclusion"] = run.conclusion === "cancelled" ? "cancelled" : "success";
1456 if (conclusion !== "cancelled") {
1457 // A job with continue-on-error may fail without failing the run.
1458 if (jobs.some((j) => (j.conclusion === "failure" || j.conclusion === "timed_out") && !j.continue_on_error)) conclusion = "failure";
1459 else if (jobs.some((j) => j.conclusion === "cancelled")) conclusion = "cancelled";
1460 // As on GitHub, a run whose jobs were all skipped is skipped, not a success, so a
1461 // workflow_run follower that checks for success (a live test after a deploy) doesn't run.
1462 else if (jobs.length && jobs.every((j) => j.conclusion === "skipped")) conclusion = "skipped";
1463 }
1464 const t = now();
1465 this.sql.exec("UPDATE runs SET status = 'completed', conclusion = ?, updated_at = ? WHERE id = ?", conclusion, t, runId);
1466 const m = this.repo.meta;
1467 this.repo.emit({ type: "workflow_run", action: "completed", repoId: m.repoId, owner: m.owner, repo: m.name, runId, conclusion: conclusion ?? undefined, actor: JSON.parse(run.actor), at: t });
1468 this.repo.broadcast(`run:${runId}`, { action: "completed", conclusion });
1469 this.repo.broadcast("runs", { action: "completed", runId });
1470 await this.onWorkflowRun(this.runRow(runId), "completed");
1471 return toRun(this.runRow(runId));
1472 }
1473
1474 /**
1475 * Starts the default branch's `on: workflow_run` workflows that follow `run`. Like
1476 * GitHub, they run on the default branch, must name the workflows they follow, and
1477 * chains stop after three levels (A -> B -> C -> D; D's runs trigger nothing).
1478 */
1479 private async onWorkflowRun(run: RunRow, action: "requested" | "completed"): Promise<void> {
1480 if (this.chainDepth(run) >= 3) return;
1481 try {
1482 const sha = this.repo.getRef(this.repo.headRef);
1483 if (!sha) return;
1484 const followers = (await this.loadWorkflows(sha)).filter((w) => w.workflow?.on.workflow_run?.workflows?.length);
1485 if (!followers.length) return;
1486 const m = this.repo.meta;
1487 const payload = JSON.parse(run.event_payload) as { pull_request?: { head?: { ref?: string } } };
1488 const branch = payload.pull_request?.head?.ref ?? run.ref.replace(/^refs\/(heads|tags)\//, "");
1489 const actor = JSON.parse(run.actor) as Actor;
1490 const workflowRun = {
1491 id: run.run_number,
1492 name: run.workflow_name,
1493 display_title: run.display_title,
1494 path: run.workflow_path,
1495 run_number: run.run_number,
1496 run_attempt: run.attempt,
1497 event: run.event,
1498 status: run.status,
1499 conclusion: run.conclusion,
1500 head_branch: branch,
1501 head_sha: run.head_sha ?? run.sha,
1502 actor: { login: actor.login },
1503 triggering_actor: { login: actor.login },
1504 pull_requests: run.pr_number ? [{ number: run.pr_number }] : [],
1505 created_at: run.created_at,
1506 updated_at: run.updated_at,
1507 html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}`,
1508 };
1509 for (const w of followers) {
1510 if (!matchesTrigger(w.workflow!, { name: "workflow_run", action, workflow: run.workflow_name, branch })) continue;
1511 await this.createRun(w.workflow!, {
1512 event: "workflow_run",
1513 ref: this.repo.headRef,
1514 sha,
1515 payload: { action, workflow_run: workflowRun, workflow: { name: run.workflow_name, path: run.workflow_path }, repository: await this.repositoryPayload(), sender: { login: actor.login } },
1516 caller: actor,
1517 title: w.workflow!.name ?? w.path,
1518 // A follower acts on what the run it follows produced (artifacts, outputs), so it stays an agent's.
1519 agentOrigin: !!run.agent_origin,
1520 });
1521 }
1522 } catch (e) {
1523 console.warn("workflow_run trigger failed", e);
1524 }
1525 }
1526
1527 /** How many workflow_run links led to `run` (0 for a run started by any other event). */
1528 private chainDepth(run: RunRow): number {
1529 let depth = 0;
1530 for (let r: RunRow | undefined = run; r?.event === "workflow_run" && depth < 3; depth++) {
1531 const parent: string | number | undefined = (JSON.parse(r.event_payload) as { workflow_run?: { id?: string | number } }).workflow_run?.id;
1532 r = parent !== undefined ? this.sql.exec<RunRow>("SELECT * FROM runs WHERE id = ? OR run_number = ? ORDER BY attempt DESC LIMIT 1", String(parent), Number(parent) || -1).toArray()[0] : undefined;
1533 }
1534 return depth;
1535 }
1536
1537 async plannedError(runId: string, message: string): Promise<void> {
1538 const t = now();
1539 this.sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'setup', 'Set up workflow', 'completed', 'failure', ?, ?, ?)", newId(), runId, t, t, JSON.stringify({ message }));
1540 }
1541
1542 // ---------------------------------------------------------------- cancellation and re-runs
1543
1544 private async cancelInternal(runId: string, reason: string): Promise<void> {
1545 const run = this.runRow(runId);
1546 if (run.status === "completed") return;
1547 this.sql.exec("UPDATE runs SET conclusion = 'cancelled', updated_at = ? WHERE id = ?", now(), run.id);
1548 for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND status != 'completed'", run.id).toArray()) {
1549 await this.stopContainer(j);
1550 await this.completeJob(j.id, { conclusion: "cancelled", outputs: {}, failure: { message: reason } });
1551 }
1552 }
1553
1554 async cancel(runId: string, caller: Caller): Promise<void> {
1555 if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied");
1556 await this.cancelInternal(runId, `Cancelled by ${caller.login}`);
1557 }
1558
1559 async rerun(runId: string, caller: Caller): Promise<RunInfo> {
1560 if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied");
1561 if (caller.id.startsWith("job:")) throw new HttpError(403, "a workflow's GITHUB_TOKEN cannot re-run workflows");
1562 await this.refuseIfBlocked("run");
1563 const old = this.runRow(runId);
1564 const wf = (await this.loadWorkflows(old.sha)).find((w) => w.path === old.workflow_path)?.workflow;
1565 if (!wf) throw new HttpError(422, "workflow file no longer valid at this commit");
1566 const run = await this.createRun(wf, {
1567 event: old.event,
1568 ref: old.ref,
1569 sha: old.sha,
1570 headSha: old.head_sha ?? undefined,
1571 payload: JSON.parse(old.event_payload),
1572 caller,
1573 prNumber: old.pr_number ?? undefined,
1574 inputs: JSON.parse(old.inputs),
1575 title: old.display_title,
1576 // Re-running code an agent wrote keeps it acting as the agent, whoever re-runs it.
1577 agentOrigin: !!old.agent_origin,
1578 });
1579 // Not started after all (Actions suspended meanwhile, or too many unfinished runs).
1580 if (!run) throw new HttpError(409, "the run was not started; try again later");
1581 // Whoever started an earlier attempt still can't approve this one where an environment prevents self-review.
1582 const prior = [...new Set([...((old.prior_actors ? JSON.parse(old.prior_actors) : []) as string[]), (JSON.parse(old.actor) as Actor).id])];
1583 this.sql.exec("UPDATE runs SET attempt = ?, run_number = ?, prior_actors = ? WHERE id = ?", old.attempt + 1, old.run_number, JSON.stringify(prior), run.id);
1584 return toRun(this.runRow(run.id));
1585 }
1586
1587 // ---------------------------------------------------------------- runner protocol
1588
1589 /** Validate a job or sandbox token. Returns what it grants. */
1590 async runnerAuth(
1591 token: string,
1592 ): Promise<{ kind: "job"; jobId: string; runId: string; role: Caller["role"]; permissions: JobPermissions; actorKind: Actor["kind"] } | { kind: "sandbox"; sandboxId: string; caller: Caller; taskPerson?: string } | null> {
1593 const parts = token.split("_");
1594 if (parts.length < 4) return null;
1595 const hash = await sha256(token);
1596 if (parts[0] === "djt") {
1597 const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", parts[2]!).toArray()[0];
1598 if (!j || j.token_hash !== hash || j.status === "completed") return null;
1599 const grant = this.repo.runtime.grantFor(j.id);
1600 // Code an agent wrote, pushed or opened a pull request with runs as that agent (repo/token-policy.ts).
1601 const run = this.sql.exec<{ agent_origin: number }>("SELECT agent_origin FROM runs WHERE id = ?", j.run_id).toArray()[0];
1602 const actorKind: Actor["kind"] = run?.agent_origin ? "agent" : "system";
1603 return { kind: "job", jobId: j.id, runId: j.run_id, role: grant.role, permissions: grant.permissions, actorKind };
1604 }
1605 if (parts[0] === "dst") {
1606 const s = this.sql.exec<{ id: string; secret: string; created_by: string; status: string; created_at: string; task_person: string | null }>("SELECT * FROM sandboxes WHERE id = ?", parts[2]!).toArray()[0];
1607 if (!s || s.secret !== hash || s.status !== "running") return null;
1608 // A sandbox's token lasts a day at most, even if nobody closes the sandbox.
1609 if (Date.now() - Date.parse(s.created_at) > SANDBOX_TOKEN_TTL_MS) return null;
1610 return { kind: "sandbox", sandboxId: s.id, caller: JSON.parse(s.created_by) as Caller, ...(s.task_person ? { taskPerson: s.task_person } : {}) };
1611 }
1612 return null;
1613 }
1614
1615 private async authJob(jobId: string, token: string): Promise<JobRow> {
1616 const auth = await this.runnerAuth(token);
1617 if (!auth || auth.kind !== "job" || auth.jobId !== jobId) throw new HttpError(401, "invalid job token");
1618 return this.jobRow(jobId);
1619 }
1620
1621 /** Claims for an OIDC id token; side-effect free (unlike runnerSpec). */
1622 async oidcClaims(jobId: string, token: string): Promise<{ allowed: boolean; github: Record<string, string> }> {
1623 const job = await this.authJob(jobId, token);
1624 if (job.status !== "in_progress") throw new HttpError(409, "job is not running");
1625 const run = this.runRow(job.run_id);
1626 const exp = await this.expanded(run);
1627 const def: Job = exp.workflow.jobs[job.job_key]!;
1628 const wf = exp.workflowOf.get(job.job_key)!;
1629 const allowed = this.repo.runtime.grantFor(jobId).permissions["id-token"] === "write";
1630 const ctx = this.githubContext(run, localKey(job.job_key, exp.parentOf.get(job.job_key) ?? null), token) as Record<string, Json>;
1631 const github: Record<string, string> = {};
1632 for (const [k, v] of Object.entries(ctx)) if (typeof v === "string" || typeof v === "number") github[k] = String(v);
1633 // Ids don't change when a name is reused, so a cloud trust policy can pin them; the environment is the one
1634 // whose protection rules this job passed (runnerSpec records its deployment).
1635 github.repository_owner_id = String(numericId(this.repo.meta.ownerId));
1636 const actor = JSON.parse(run.actor) as Actor;
1637 if (actor.id) github.actor_id = String(numericId(actor.id));
1638 const env = this.sql.exec<{ environment: string }>("SELECT environment FROM deployments WHERE job_id = ?", jobId).toArray()[0];
1639 if (env) github.environment = env.environment;
1640 return { allowed, github };
1641 }
1642
1643 /**
1644 * What a job token may do with container packages (the `packages` permission).
1645 * A workflow without a permissions block gets the repository's default: read in
1646 * repositories with read-only tokens (the default for new ones), write in those that
1647 * keep GitHub's old permissive default. A permissions map that leaves `packages` out gets none.
1648 */
1649 async packagesPermission(token: string): Promise<"none" | "read" | "write"> {
1650 const auth = await this.runnerAuth(token);
1651 return auth?.kind === "job" ? auth.permissions.packages : "none";
1652 }
1653
1654 /**
1655 * A pull_request run for a pull request from another repository (a fork): its jobs
1656 * never get write or secrets. Decided from the pull request's row, not the event payload.
1657 */
1658 private untrusted(run: RunRow): boolean {
1659 return run.event === "pull_request" && this.repo.policy.isForkPull(run.pr_number);
1660 }
1661
1662 async runnerSpec(jobId: string, token: string, runner: JobRunnerInfo | null = null): Promise<JobSpec> {
1663 const job = await this.authJob(jobId, token);
1664 const run = this.runRow(job.run_id);
1665 const exp = await this.expanded(run);
1666 const def: Job = exp.workflow.jobs[job.job_key]!;
1667 const wf = exp.workflowOf.get(job.job_key)!;
1668 const parent = exp.parentOf.get(job.job_key) ?? null;
1669 const short = localKey(job.job_key, parent);
1670 let scope: { inputs: Record<string, unknown>; secrets: Record<string, string> };
1671 try {
1672 // A fork's code never sees the repository's secrets, as on GitHub, and an agent's only where an admin allows it.
1673 scope = await this.callContext(run, exp, parent, !this.untrusted(run) && (!run.agent_origin || this.repo.policy.settings().agentRunSecrets));
1674 } catch (e) {
1675 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } });
1676 throw new HttpError(409, (e as Error).message);
1677 }
1678 const m = this.repo.meta;
1679 // Deployment environment (`environment:`): its branch policy, secrets and variables, and a deployment record.
1680 let environment: { secrets: Record<string, string>; vars: Record<string, string> } = { secrets: {}, vars: {} };
1681 if (def.environment) {
1682 try {
1683 const ctx = { github: this.githubContext(run, short, ""), inputs: scope.inputs, vars: this.variables(), matrix: JSON.parse(job.matrix), strategy: JSON.parse(job.strategy), needs: await this.needsResults(run, exp, exp.needs.get(job.job_key) ?? [], parent) };
1684 // The name evaluated at launch, whose protection rules the job passed (older jobs evaluate it here).
1685 const name = job.environment ?? interpolate(def.environment.name, ctx);
1686 if (this.repo.deploys.protection(name) && job.gate_state !== "released") throw new Error(`The environment ${name} requires approval this job didn't get. Re-run the job to ask for it.`);
1687 const url = def.environment.url && !def.environment.url.includes("steps.") ? interpolate(def.environment.url, ctx) : undefined;
1688 const logUrl = `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`;
1689 // Code an agent wrote deploys as that agent (environment rules hold it to the agent's limits).
1690 const actor = JSON.parse(run.actor) as Actor;
1691 environment = await this.repo.deploys.jobStarted({ id: jobId, ref: run.ref, sha: run.sha, logUrl, actor: run.agent_origin ? { ...actor, kind: "agent" } : actor }, { name, url });
1692 } catch (e) {
1693 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } });
1694 throw new HttpError(409, (e as Error).message);
1695 }
1696 }
1697 const t = now();
1698 this.sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ?, runner_info = ? WHERE id = ?", t, runner ? JSON.stringify(runner) : null, jobId);
1699 this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "in_progress", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` });
1700 this.repo.broadcast(`run:${run.id}`, { action: "job", jobId });
1701 // Only the secrets the job's expressions name reach the runner, as on GitHub (plus the runner's own debug switches).
1702 const used = referencedSecrets(def, wf.env);
1703 if (used !== "all") for (const name of RUNNER_SECRETS) used.add(name);
1704 const secrets: Record<string, string> = { ...secretsFor({ ...scope.secrets, ...(this.untrusted(run) ? {} : environment.secrets) }, used), GITHUB_TOKEN: token, DIFFINERY_TOKEN: token };
1705 // The job token can do what the effective `permissions:` allow (job level wins, like GitHub).
1706 const permissions = permissionsForJob(exp, job.job_key, { untrusted: this.untrusted(run), repoDefault: this.repo.policy.settings().defaultPermissions });
1707 this.repo.runtime.grant(jobId, permissions);
1708 const idToken = permissions["id-token"] === "write";
1709 return {
1710 runId: run.id,
1711 jobId,
1712 jobKey: short,
1713 name: job.name,
1714 repo: { owner: m.owner, name: m.name, cloneUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, defaultBranch: m.defaultBranch },
1715 sha: run.sha,
1716 ref: run.ref,
1717 workflowPath: run.workflow_path,
1718 github: this.githubContext(run, short, token),
1719 env: { ...wf.env, ...def.env },
1720 defaults: { run: { ...wf.defaults.run, ...def.defaults.run } },
1721 matrix: JSON.parse(job.matrix),
1722 strategy: JSON.parse(job.strategy),
1723 needs: await this.needsResults(run, exp, exp.needs.get(job.job_key) ?? [], parent),
1724 inputs: scope.inputs,
1725 vars: { ...this.variables(), ...environment.vars },
1726 secrets,
1727 outputs: def.outputs,
1728 steps: def.steps.map((s) => ({ ...s, "timeout-minutes": s["timeout-minutes"] === undefined ? undefined : Number(s["timeout-minutes"]) })),
1729 container: def.container,
1730 services: def.services,
1731 timeoutMinutes: job.timeout_minutes,
1732 continueOnError: !!job.continue_on_error,
1733 idToken: !!idToken,
1734 apiUrl: this.env.PUBLIC_URL,
1735 token,
1736 runtime: await this.runtimeServices(run, job),
1737 };
1738 }
1739
1740 /** ACTIONS_RUNTIME_TOKEN and the service URLs; the token expires when the job would time out. */
1741 private async runtimeServices(run: RunRow, job: JobRow): Promise<JobSpec["runtime"]> {
1742 if (!this.env.MASTER_KEY) return undefined;
1743 const origin = new URL(this.env.PUBLIC_URL).origin;
1744 const exp = Math.floor(Date.now() / 1000) + (job.timeout_minutes + 30) * 60;
1745 const { scopes } = this.repo.runtime.job(job.id);
1746 const token = await signRuntimeToken(this.env.MASTER_KEY, origin, { repo: this.repo.meta.repoId, run: run.id, job: job.id, exp }, scopes);
1747 return { token, resultsUrl: `${origin}/`, cacheUrl: `${origin}/_runtime/`, runtimeUrl: `${origin}/_runtime/` };
1748 }
1749
1750 async runnerLogs(jobId: string, token: string, lines: LogLine[]): Promise<void> {
1751 const job = await this.authJob(jobId, token);
1752 let chars = job.log_chars ?? 0;
1753 if (chars > MAX_JOB_LOG) return;
1754 let seq = this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", jobId).one().n ?? -1;
1755 const out: Array<LogLine & { seq: number }> = [];
1756 for (const l of lines.slice(0, 5000)) {
1757 seq++;
1758 const t = String(l.t ?? "").slice(0, 40);
1759 const step = Number.isInteger(l.step) ? l.step : 0;
1760 let text = String(l.text ?? "");
1761 if (text.length > 16384) text = `${text.slice(0, 16384)}…`;
1762 // Past the cap, one last line says so and the rest is dropped: the whole log is read into
1763 // memory when the job completes, to archive it.
1764 if (chars + text.length > MAX_JOB_LOG) {
1765 text = `The log is cut off here: a job's log is limited to ${MAX_JOB_LOG / 1048576} MiB of text.`;
1766 chars = MAX_JOB_LOG + 1;
1767 } else chars += text.length;
1768 this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, ?, ?)", jobId, seq, t, step, text);
1769 out.push({ t, step, text, seq });
1770 if (chars > MAX_JOB_LOG) break;
1771 }
1772 this.sql.exec("UPDATE jobs SET log_chars = ? WHERE id = ?", chars, jobId);
1773 this.repo.broadcast(`job:${jobId}`, { action: "logs", lines: out });
1774 }
1775
1776 async runnerStep(jobId: string, token: string, index: number, report: StepReport): Promise<void> {
1777 const job = await this.authJob(jobId, token);
1778 const steps = JSON.parse(job.steps) as Array<StepReport & { index: number }>;
1779 const existing = steps.findIndex((s) => s.index === index);
1780 const next = { ...(existing >= 0 ? steps[existing] : {}), ...report, index };
1781 if (existing >= 0) steps[existing] = next;
1782 else steps.push(next);
1783 steps.sort((a, b) => a.index - b.index);
1784 this.sql.exec("UPDATE jobs SET steps = ? WHERE id = ?", JSON.stringify(steps), jobId);
1785 this.repo.broadcast(`job:${jobId}`, { action: "step", step: next });
1786 this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId });
1787 }
1788
1789 async runnerComplete(jobId: string, token: string, completion: JobCompletion): Promise<void> {
1790 await this.authJob(jobId, token);
1791 await this.completeJob(jobId, completion);
1792 }
1793
1794 private async completeJob(jobId: string, completion: JobCompletion, override?: "timed_out"): Promise<void> {
1795 const job = this.jobRow(jobId);
1796 if (job.status === "completed") return;
1797 const gated = job.status === "approval";
1798 const run = this.runRow(job.run_id);
1799 const t = now();
1800 const conclusion = override ?? completion.conclusion;
1801 this.sql.exec(
1802 "UPDATE jobs SET status = 'completed', conclusion = ?, completed_at = ?, outputs = ?, failure = ?, summary = ?, token_hash = NULL, started_at = COALESCE(started_at, ?) WHERE id = ?",
1803 conclusion,
1804 t,
1805 JSON.stringify(completion.outputs ?? {}),
1806 completion.failure ? JSON.stringify(completion.failure) : null,
1807 completion.summary ?? null,
1808 t,
1809 jobId,
1810 );
1811 this.repo.deploys.jobCompleted(jobId, conclusion);
1812 const m = this.repo.meta;
1813 this.repo.upsertCheck({
1814 sha: run.head_sha ?? run.sha,
1815 name: this.checkName(run, job),
1816 status: "completed",
1817 conclusion: conclusion === "timed_out" ? "timed_out" : conclusion,
1818 runId: run.id,
1819 jobId,
1820 detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`,
1821 output: completion.failure ? { title: "Failed", summary: completion.failure.message } : { title: conclusion },
1822 });
1823 // Archive logs to R2 and free the Durable Object's storage.
1824 const lines = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? ORDER BY seq", jobId).toArray();
1825 if (lines.length) {
1826 await this.env.BUCKET.put(this.logKey(jobId), JSON.stringify(lines), { httpMetadata: { contentType: "application/json" } });
1827 this.sql.exec("DELETE FROM logs WHERE job_id = ?", jobId);
1828 }
1829 // Uploads the job left unfinished stop counting against the budgets, and their bytes go.
1830 await this.repo.runtime.sweepJob(jobId).catch((e) => console.error("runtime sweep failed", e));
1831 this.repo.broadcast(`job:${jobId}`, { action: "completed", conclusion });
1832 this.repo.broadcast(`run:${run.id}`, { action: "job", jobId });
1833
1834 // fail-fast: cancel unfinished siblings of a failed matrix job (unless the failed leg may continue on error).
1835 const siblings = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ? AND id != ? AND status != 'completed'", run.id, job.job_key, jobId).toArray();
1836 const strategy = JSON.parse(job.strategy || "{}") as { "fail-fast"?: boolean };
1837 if ((conclusion === "failure" || conclusion === "timed_out") && !job.continue_on_error && strategy["fail-fast"] !== false && siblings.length) {
1838 for (const s of siblings) {
1839 await this.stopContainer(s);
1840 await this.completeJob(s.id, { conclusion: "cancelled", outputs: {}, failure: { message: `Cancelled because ${job.name} failed (fail-fast).` } });
1841 }
1842 } else {
1843 const queued = siblings.find((s) => s.status === "queued");
1844 if (queued) await this.launch(queued.id);
1845 }
1846 // An ephemeral runner is done once its one job is.
1847 if (job.runner_id) this.sql.exec("DELETE FROM runners WHERE id = ? AND ephemeral = 1 AND token_hash IS NULL", job.runner_id);
1848 // A hosted job hands its slot to the oldest job waiting for one (the repository's or the account's).
1849 if (!job.labels && job.container) {
1850 await endCompute(this.env, jobId).catch((e: unknown) => console.error("compute usage failed", e));
1851 const held = this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE fleet_held > 0 AND status = 'queued' ORDER BY rowid LIMIT 1").toArray()[0];
1852 if (held) await this.launch(held.id);
1853 }
1854 // A job that held its concurrency group hands it to the oldest pending job (of any run).
1855 if (job.concurrency_group && (job.status === "waiting" || job.status === "in_progress")) {
1856 const next = this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE concurrency_group = ? AND status = 'pending' ORDER BY rowid LIMIT 1", job.concurrency_group).toArray()[0];
1857 if (next) await this.launch(next.id);
1858 }
1859 try {
1860 const instance = await this.env.RUN_WORKFLOW.get(run.id);
1861 await instance.sendEvent({ type: `job-${jobId}`, payload: { conclusion } });
1862 } catch (e) {
1863 console.error("failed to notify workflow", e);
1864 }
1865 // The Workflow waits for a job held for approval on its own event.
1866 if (gated) {
1867 this.syncWaiting(run.id);
1868 await this.wakeGate(run.id, jobId);
1869 }
1870 }
1871
1872 // ---------------------------------------------------------------- secrets and variables
1873
1874 private async repoSecrets(): Promise<Record<string, string>> {
1875 const out: Record<string, string> = {};
1876 for (const r of this.sql.exec<{ name: string; ciphertext: string }>("SELECT name, ciphertext FROM secrets").toArray()) {
1877 out[r.name] = await decryptSecret(this.env.MASTER_KEY, `repo:${this.repo.meta.repoId}`, r.ciphertext);
1878 }
1879 return out;
1880 }
1881
1882 private async orgSecrets(): Promise<Record<string, string>> {
1883 const m = this.repo.meta;
1884 if (m.ownerKind !== "org") return {};
1885 // Only the secrets whose repository selection includes this repository (visibility all, private or selected).
1886 const rows = await this.env.DB.prepare(
1887 `SELECT s.name, s.ciphertext FROM org_secrets s WHERE s.org_id = ?1 AND (s.visibility = 'all'
1888 OR (s.visibility = 'private' AND EXISTS (SELECT 1 FROM repos r WHERE r.id = ?2 AND r.visibility != 'public'))
1889 OR (s.visibility = 'selected' AND EXISTS (SELECT 1 FROM org_secret_repos x WHERE x.org_id = s.org_id AND x.name = s.name AND x.repo_id = ?2)))`,
1890 )
1891 .bind(m.ownerId, m.repoId)
1892 .all<{ name: string; ciphertext: string }>();
1893 const out: Record<string, string> = {};
1894 for (const r of rows.results) out[r.name] = await decryptSecret(this.env.MASTER_KEY, `org:${m.ownerId}`, r.ciphertext);
1895 return out;
1896 }
1897
1898 async setSecret(name: string, value: string): Promise<void> {
1899 if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || /^GITHUB_/i.test(name)) throw new HttpError(422, "invalid secret name");
1900 this.sql.exec("INSERT OR REPLACE INTO secrets (name, ciphertext, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), await encryptSecret(this.env.MASTER_KEY, `repo:${this.repo.meta.repoId}`, value), now());
1901 }
1902
1903 deleteSecret(name: string): void {
1904 this.sql.exec("DELETE FROM secrets WHERE name = ?", name.toUpperCase());
1905 }
1906
1907 secretNames(): Array<{ name: string; updatedAt: string }> {
1908 return this.sql.exec<{ name: string; updated_at: string }>("SELECT name, updated_at FROM secrets ORDER BY name").toArray().map((r) => ({ name: r.name, updatedAt: r.updated_at }));
1909 }
1910
1911 setVariable(name: string, value: string | null): void {
1912 if (value === null) this.sql.exec("DELETE FROM variables WHERE name = ?", name.toUpperCase());
1913 else this.sql.exec("INSERT OR REPLACE INTO variables (name, value, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), value, now());
1914 }
1915
1916 /** Variables for `vars.*`: the organization's, overridden by the repository's. `withOrg = false` lists only the repository's own. */
1917 variables(withOrg = true): Record<string, string> {
1918 const own = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM variables").toArray().map((r) => [r.name, r.value]));
1919 if (!withOrg) return own;
1920 const org = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM org_variables").toArray().map((r) => [r.name, r.value]));
1921 return { ...org, ...own };
1922 }
1923
1924 // ---------------------------------------------------------------- cache and artifacts
1925
1926 /**
1927 * An older runner's cache (the runner API before the toolkit's cache service).
1928 * Entries are stored as `<ref>\n<key>` and scoped like the cache service's
1929 * (repo/runtime.ts): a job saves to its own ref and restores from its own ref,
1930 * then the pull request's base, then the default branch, so one branch or pull
1931 * request never sees another's caches. Entries saved before scoping (no ref)
1932 * are never restored, so they go a week after their last use.
1933 */
1934 async cacheLookup(jobId: string, token: string, key: string, restoreKeys: string[]): Promise<{ key: string; stored: string } | null> {
1935 await this.authJob(jobId, token);
1936 const { scopes } = this.repo.runtime.job(jobId);
1937 // An entry whose object is gone is forgotten, and the lookup moves on.
1938 for (let attempt = 0; attempt < 5; attempt++) {
1939 const hit = this.legacyCacheMatch(scopes, key, restoreKeys);
1940 if (!hit) return null;
1941 if (!(await headFirst(this.env.BUCKET, this.cacheR2Keys(hit)))) {
1942 this.sql.exec("DELETE FROM caches WHERE key = ?", hit);
1943 continue;
1944 }
1945 this.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", now(), hit);
1946 return { key: hit.slice(hit.indexOf("\n") + 1), stored: hit };
1947 }
1948 return null;
1949 }
1950
1951 /** In each scope in turn: the exact key, then the newest entry for each restore key prefix. */
1952 private legacyCacheMatch(scopes: string[], key: string, restoreKeys: string[]): string | null {
1953 for (const scope of scopes) {
1954 const exact = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE key = ?", `${scope}\n${key}`).toArray()[0];
1955 if (exact) return exact.key;
1956 for (const prefix of restoreKeys) {
1957 const p = `${scope}\n${prefix}`;
1958 const r = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE substr(key, 1, ?) = ? ORDER BY created_at DESC LIMIT 1", p.length, p).toArray()[0];
1959 if (r) return r.key;
1960 }
1961 }
1962 return null;
1963 }
1964
1965 /** Where a job saves `key` (its own ref's scope): the stored key and its R2 keys. */
1966 async cacheSaveKeys(jobId: string, token: string, key: string): Promise<{ stored: string; r2Keys: string[] }> {
1967 await this.authJob(jobId, token);
1968 const stored = `${this.repo.runtime.job(jobId).scopes[0]}\n${key}`;
1969 return { stored, r2Keys: this.cacheR2Keys(stored) };
1970 }
1971
1972 /** The R2 keys of entry `stored`, when the job may read its scope. */
1973 async cacheReadKeys(jobId: string, token: string, stored: string): Promise<string[] | null> {
1974 await this.authJob(jobId, token);
1975 const nl = stored.indexOf("\n");
1976 if (nl < 0 || !this.repo.runtime.job(jobId).scopes.includes(stored.slice(0, nl))) return null;
1977 return this.cacheR2Keys(stored);
1978 }
1979
1980 /** Where an older runner's cache entry is stored: the current key, then the old layout's. */
1981 cacheR2Keys(key: string): string[] {
1982 return legacyCacheKeys(this.repo.meta.repoId, key);
1983 }
1984
1985 /** Records a saved entry; returns the R2 keys of the entries it evicted, for the caller to delete. */
1986 async cacheSaved(jobId: string, token: string, key: string, size: number): Promise<string[]> {
1987 const { stored } = await this.cacheSaveKeys(jobId, token, key);
1988 this.sql.exec("INSERT OR REPLACE INTO caches (key, size, created_at, last_used) VALUES (?, ?, ?, ?)", stored, size, now(), now());
1989 // Like the cache service (repo/runtime.ts): entries unused for a week go, then the
1990 // least recently used ones above 10 GB per repository.
1991 const stale = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE last_used < ?", new Date(Date.now() - CACHE_IDLE_MS).toISOString()).toArray().map((r) => r.key);
1992 for (const k of stale) this.sql.exec("DELETE FROM caches WHERE key = ?", k);
1993 const evicted = stale.flatMap((k) => this.cacheR2Keys(k));
1994 let total = this.sql.exec<{ n: number | null }>("SELECT sum(size) AS n FROM caches").one().n ?? 0;
1995 while (total > CACHE_LIMIT_BYTES) {
1996 const old = this.sql.exec<{ key: string; size: number }>("SELECT key, size FROM caches ORDER BY last_used LIMIT 1").one();
1997 this.sql.exec("DELETE FROM caches WHERE key = ?", old.key);
1998 evicted.push(...this.cacheR2Keys(old.key));
1999 total -= old.size;
2000 }
2001 return evicted;
2002 }
2003
2004 /**
2005 * Where job `jobId` (authenticated by its token) stores artifact `name` of its run, and how many bytes
2006 * it may store there (what is left of the artifact quota). Refused when the run
2007 * already has an artifact of that name from the artifact service.
2008 */
2009 async artifactUploadKeys(jobId: string, token: string, name: string): Promise<{ r2Keys: string[]; room: number }> {
2010 const job = await this.authJob(jobId, token);
2011 this.repo.runtime.legacyArtifactCheck(job.run_id, name);
2012 return { r2Keys: this.artifactR2Keys(job.run_id, name), room: await this.repo.runtime.artifactRoom() };
2013 }
2014
2015 /** Records an older runner's artifact, stored at its legacy key, with the artifact service's (repo/runtime.ts). */
2016 async artifactSaved(jobId: string, token: string, name: string, size: number): Promise<{ id: number }> {
2017 const job = await this.authJob(jobId, token);
2018 return { id: this.repo.runtime.legacyArtifactSaved(job.run_id, jobId, name, size).id };
2019 }
2020
2021 /** Where job `jobId` (authenticated by its token) reads artifact `name` of its run. */
2022 async artifactKeysFor(jobId: string, token: string, name: string): Promise<string[]> {
2023 const job = await this.authJob(jobId, token);
2024 return this.artifactR2Keys(job.run_id, name);
2025 }
2026
2027 /** Where an older runner's artifact is stored: the current key, then the old layout's. */
2028 artifactR2Keys(runId: string, name: string): string[] {
2029 return legacyArtifactKeys(this.repo.meta.repoId, runId, name);
2030 }
2031
2032 /**
2033 * A run's artifacts: those uploaded through the artifact service (zip files, or
2034 * single unarchived files) and older runners' tarballs.
2035 */
2036 artifacts(runId: string): Array<{ name: string; size: number; createdAt: string; r2Keys: string[]; id: number; contentType: string; fileName: string }> {
2037 runId = this.runRow(runId).id;
2038 return this.repo.runtime
2039 .runArtifacts(runId)
2040 .map((a) => ({
2041 name: a.name,
2042 size: a.size,
2043 createdAt: a.createdAt,
2044 r2Keys: artifactObjectKeys(this.repo.meta.repoId, a),
2045 id: a.id,
2046 contentType: a.mimeType,
2047 fileName: artifactFileName(a),
2048 }))
2049 .sort((a, b) => a.name.localeCompare(b.name));
2050 }
2051
2052 // ---------------------------------------------------------------- sandboxes for agents
2053
2054 /**
2055 * `task` is set only by agentic/ (never from a request): the sandbox of an agent task, started for
2056 * the agent `caller` by the person `task.personId`. No public sandbox call reaches it. An agent
2057 * review's sandbox is `readOnly`: stored with a `read` caller, its git credential can't push.
2058 * A web terminal's sandbox (`kind: "terminal"`) is a task sandbox one person works in directly.
2059 */
2060 async createSandbox(caller: Caller, ref?: string, task?: { personId: string; readOnly?: boolean; kind?: "task" | "terminal" }): Promise<{ id: string; ref: string }> {
2061 if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to create a sandbox");
2062 // A review's sandbox acts with read access only: its git credential can clone but never push.
2063 if (task?.readOnly) caller = { ...caller, role: "read" };
2064 if (!task) {
2065 const since = new Date(Date.now() - SANDBOX_TOKEN_TTL_MS).toISOString();
2066 const open = this.sql
2067 .exec<{ n: number }>("SELECT count(*) AS n FROM sandboxes WHERE kind = 'api' AND status = 'running' AND created_at > ? AND json_extract(created_by, '$.id') = ?", since, caller.id)
2068 .one().n;
2069 if (open >= MAX_OPEN_SANDBOXES) throw new HttpError(429, `you have ${open} open sandboxes in this repository, the most allowed; close one first`);
2070 }
2071 await this.refuseIfBlocked("sandbox", false);
2072 // The account that owns the repository pays for sandboxes too (lib/quotas.ts).
2073 const quota = await mayStart(this.env, this.repo.meta.ownerId, "sandbox");
2074 if (!quota.ok) throw new HttpError(429, quota.message);
2075 const sha = await this.repo.resolve(ref ?? "HEAD");
2076 if (!sha) throw new HttpError(422, "unknown ref");
2077 const id = newId();
2078 const m = this.repo.meta;
2079 const token = `dst_${m.repoId}_${id}_${randomToken(24)}`;
2080 const secret = randomToken(32);
2081 this.sql.exec(
2082 "INSERT INTO sandboxes (id, secret, created_by, ref, created_at, status, kind, task_person) VALUES (?, ?, ?, ?, ?, 'running', ?, ?)",
2083 id,
2084 await sha256(token),
2085 JSON.stringify(caller),
2086 ref ?? m.defaultBranch,
2087 now(),
2088 task ? (task.kind ?? "task") : "api",
2089 task?.personId ?? null,
2090 );
2091 await startCompute(this.env, { id, accountId: m.ownerId, repoId: m.repoId, kind: "sandbox" });
2092 const stub = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id));
2093 try {
2094 await stub.boot({
2095 DIFFINERY_CLONE_URL: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`,
2096 DIFFINERY_REF: ref ?? m.defaultBranch,
2097 DIFFINERY_JOB_TOKEN: token,
2098 DIFFINERY_SANDBOX_SECRET: secret,
2099 DIFFINERY_API_URL: this.env.PUBLIC_URL,
2100 GIT_AUTHOR_NAME: caller.login,
2101 GIT_AUTHOR_EMAIL: `${caller.login}@users.noreply.diffinery.com`,
2102 GIT_COMMITTER_NAME: caller.login,
2103 GIT_COMMITTER_EMAIL: `${caller.login}@users.noreply.diffinery.com`,
2104 }, secret, id, { id, accountId: m.ownerId, repoId: m.repoId });
2105 } catch (e) {
2106 console.error("sandbox boot failed", e);
2107 this.sql.exec("UPDATE sandboxes SET status = 'failed' WHERE id = ?", id);
2108 await endCompute(this.env, id).catch(() => {});
2109 await stub.shutdown().catch(() => {});
2110 throw new HttpError(503, "the sandbox container did not start; try again");
2111 }
2112 return { id, ref: ref ?? m.defaultBranch };
2113 }
2114
2115 getSandbox(id: string, caller: Caller): { id: string; ref: string; status: string; createdAt: string } {
2116 const s = this.sql.exec<{ id: string; ref: string; status: string; created_at: string; created_by: string; kind: string }>("SELECT * FROM sandboxes WHERE id = ?", id).toArray()[0];
2117 // A task's sandbox holds a credential for the person's model access, and a terminal's may hold their
2118 // provider sign-in: nobody reaches either through the sandbox API, admins included.
2119 if (!s || s.kind !== "api") throw new HttpError(404, "sandbox not found");
2120 const owner = JSON.parse(s.created_by) as Caller;
2121 if (owner.id !== caller.id && !atLeast(caller.role, "admin")) throw new HttpError(403, "not your sandbox");
2122 return { id: s.id, ref: s.ref, status: s.status, createdAt: s.created_at };
2123 }
2124
2125 async closeSandbox(id: string, caller: Caller): Promise<void> {
2126 this.getSandbox(id, caller);
2127 await this.endSandbox(id);
2128 }
2129
2130 /** Close a sandbox: its token stops working, its time stops counting and its container stops. */
2131 private async endSandbox(id: string): Promise<void> {
2132 this.sql.exec("UPDATE sandboxes SET status = 'closed' WHERE id = ?", id);
2133 await endCompute(this.env, id);
2134 await this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)).shutdown();
2135 }
2136
2137 /** Close an agent task's or a web terminal's sandbox (agentic/), which revokes its git credential. */
2138 async closeTaskSandbox(id: string): Promise<void> {
2139 const s = this.sql.exec<{ kind: string }>("SELECT kind FROM sandboxes WHERE id = ?", id).toArray()[0];
2140 if (s?.kind !== "task" && s?.kind !== "terminal") return;
2141 await this.endSandbox(id);
2142 }
2143
2144 // ---------------------------------------------------------------- kill switches (lib/killswitch.ts)
2145
2146 /**
2147 * Why `work` may not start here now, or null. One D1 read. A failed read lets work start
2148 * (`failOpen`), as quotas do for jobs: a D1 outage shouldn't stop every repository's CI.
2149 */
2150 private async blocked(work: Work, failOpen = true): Promise<string | null> {
2151 try {
2152 return blockedBy(await killSwitches(this.env, this.repo.meta.repoId), work);
2153 } catch (e) {
2154 if (!failOpen) throw e;
2155 console.error("kill switch check failed", e);
2156 return null;
2157 }
2158 }
2159
2160 /** Refuse (403, with the reason) what someone asked to start while a kill switch stops it. */
2161 async refuseIfBlocked(work: Work, failOpen = true): Promise<void> {
2162 const block = await this.blocked(work, failOpen);
2163 if (block) throw new HttpError(403, block);
2164 }
2165
2166 /**
2167 * A site administrator just suspended Actions here or for the owner (`hostedOnly` false): every
2168 * unfinished run is cancelled and every open sandbox closed. Or paused hosted compute on the
2169 * instance (`hostedOnly`): running hosted jobs fail and open sandboxes close; self-hosted jobs go on.
2170 * `message` says why. Jobs still queued fail when they would start (launch).
2171 */
2172 async stopActions(o: { hostedOnly: boolean; message: string }): Promise<{ runs: number; jobs: number; sandboxes: number }> {
2173 let runs = 0;
2174 let jobs = 0;
2175 let sandboxes = 0;
2176 if (o.hostedOnly) {
2177 for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE labels IS NULL AND container IS NOT NULL AND status IN ('waiting', 'in_progress') ORDER BY rowid").toArray()) {
2178 await this.stopContainer(j);
2179 await this.completeJob(j.id, { conclusion: "failure", outputs: {}, failure: { message: `The job was stopped: ${o.message}.` } });
2180 jobs++;
2181 }
2182 } else {
2183 for (const r of this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE status != 'completed' ORDER BY rowid").toArray()) {
2184 jobs += this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM jobs WHERE run_id = ? AND status != 'completed'", r.id).one().n;
2185 await this.cancelInternal(r.id, `Cancelled: ${o.message}.`);
2186 runs++;
2187 }
2188 }
2189 for (const s of this.sql.exec<{ id: string }>("SELECT id FROM sandboxes WHERE status = 'running'").toArray()) {
2190 await this.endSandbox(s.id).catch((e: unknown) => console.error("closing a sandbox failed", e));
2191 sandboxes++;
2192 }
2193 return { runs, jobs, sandboxes };
2194 }
2195}
2196
2197/**
2198 * One minute of a wait timer: a real minute, except in the end-to-end tests (WAIT_TIMER_MINUTE_MS). The setting
2199 * shortens a protection rule, so it counts only on a Worker that runs on this machine (auth/dev.ts): a deployed
2200 * one ignores it.
2201 */
2202export function timerMinute(env: Pick<Env, "WAIT_TIMER_MINUTE_MS" | "PUBLIC_URL" | "ACCESS_AUD">): number {
2203 if (!env.WAIT_TIMER_MINUTE_MS || !localOnly(env)) return 60_000;
2204 const ms = Number(env.WAIT_TIMER_MINUTE_MS);
2205 return Number.isInteger(ms) && ms > 0 && ms <= 60_000 ? ms : 60_000;
2206}