apps/diffinery/src/repo/actions.ts at 6074ea77cf4560239acc053623dd5f8a1a1db9d5 · codician/diffinery · Diffinery

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