Diffinery

codician/diffinerypublic

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

1631fd0

Branches

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

Tags

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