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.

Support reusable workflows (jobs.<id>.uses) (#6)

A run's workflow is flattened before planning: a caller job `deploy` that uses
a workflow with jobs `build` and `ship` becomes `deploy/build` and
`deploy/ship`, ordered after whatever `deploy` needs; jobs that need `deploy`
wait for all of them. Inputs come from `with` and the declared defaults,
secrets from a mapping or `inherit`, and `needs.deploy.outputs` from the
called workflow's `on.workflow_call.outputs`. Calls nest four levels deep and
can come from the same commit or another repository on this instance.

Also gives the e2e server a MASTER_KEY (secrets were untested end to end) and
checks the Kumo bundle after each deploy.


Claude-Session: https://claude.ai/code/session_01AizKXSsengXZRegJxRATkL

Co-authored-by: Claude <noreply@anthropic.com>
Canberk Demirci authored · GitHub committed signed1 parent 97dbd99commit e104acd926cfbe65d843a3882cd8f9e9c35b12afBrowse files
6 files changed+494−49
Files (6)
modified.github/workflows/deploy.yml+7−0View file
@@ -75,3 +75,10 @@
7575 loc=$(curl -sS -o /dev/null -w '%{http_code} %{redirect_url}' https://diffinery.com/auth/access)
7676 echo "/auth/access -> $loc"
7777 [[ "$loc" == 302\ https://*.cloudflareaccess.com/* ]] || echo "::warning::/auth/access is not behind Cloudflare Access yet, so email recovery and first admin sign-in will not work"
78+ for kind in js css; do
79+ asset=$(grep -o "/-/kumo/[a-z0-9]*\.$kind" <<<"$page" | head -1)
80+ [ -n "$asset" ] || { echo "::error::/login does not load the Kumo $kind bundle"; exit 1; }
81+ code=$(curl -sS -o /dev/null -w '%{http_code}' "https://diffinery.com$asset")
82+ [ "$code" = 200 ] || { echo "::error::$asset returned $code"; exit 1; }
83+ echo "$asset -> 200"
84+ done
modifiedREADME.md+7−2View file
@@ -138,9 +138,14 @@
138138 Cloudflare Container. `runs-on: [self-hosted, …]` jobs wait for a runner you
139139 start with `diffinery runner -R owner/repo --labels gpu,macos`. Use that for
140140 macOS, GPUs or private networks.
141+- Reusable workflows: `jobs.<id>.uses: ./.github/workflows/x.yml` from the same
142+ commit, or `owner/repo/.github/workflows/x.yml@ref` from another repository
143+ on this instance (public or internal ones, or private ones with the same
144+ owner). `with`, `secrets` (a mapping or `inherit`), `on.workflow_call`
145+ outputs and nesting up to four levels deep all work. A matrix on the calling
146+ job isn't supported yet.
141147
142-**Not supported yet:** reusable workflows (`jobs.<id>.uses`), job
143-`container:` and `services:`, Docker container actions, `actions/github-script`,
148+**Not supported yet:** job `container:` and `services:`, Docker container actions, `actions/github-script`,
144149 Windows and macOS hosted runners, `pull_request_target`, `merge_group`, and
145150 `deepen-not` / partial clone filters. Pull request runs use the head commit
146151 rather than a test merge commit. A job that uses an unsupported feature fails
modifiedapps/diffinery/src/repo/actions.ts+178−43View file
@@ -16,10 +16,12 @@
1616 type Workflow,
1717 } from "@diffinery/workflow-yaml";
1818 import { decryptSecret, encryptSecret } from "../lib/crypto.js";
19+import { Directory } from "../lib/directory.js";
1920 import { HttpError, newId, now, randomToken, sha256 } from "../lib/util.js";
2021 import type { Env } from "../env.js";
2122 import type { RepoDO } from "./RepoDO.js";
2223 import { atLeast, type Caller } from "./types.js";
24+import { ReusableError, expandReusable, localKey, parseCallRef, type Expanded, type Source } from "./reusable.js";
2325
2426 const WORKFLOW_DIRS = [".github/workflows", ".diffinery/workflows"];
2527 const SUPPORTED_LABEL = /^(ubuntu-.*|linux|diffinery(-.*)?|self-hosted|x64|amd64)$/i;
@@ -516,9 +518,13 @@
516518 async plan(runId: string): Promise<{ levels: string[][]; error?: string; concurrencyBlocked: boolean }> {
517519 const run = this.runRow(runId);
518520 if (run.status === "completed") return { levels: [], error: "run already completed", concurrencyBlocked: false };
519- const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path);
520- if (!wf?.workflow) return { levels: [], error: `workflow file not found or invalid: ${wf?.errors.join("; ") ?? run.workflow_path}`, concurrencyBlocked: false };
521- const { order, errors } = jobOrder(wf.workflow);
521+ let exp: Expanded;
522+ try {
523+ exp = await this.expanded(run);
524+ } catch (e) {
525+ return { levels: [], error: (e as Error).message, concurrencyBlocked: false };
526+ }
527+ const { order, errors } = jobOrder(exp.workflow);
522528 if (errors.length) return { levels: [], error: errors.map((e) => e.message).join("; "), concurrencyBlocked: false };
523529 const blocked =
524530 !!run.concurrency_group &&
@@ -527,37 +533,140 @@
527533 return { levels: order, concurrencyBlocked: blocked };
528534 }
529535
530- private needsResults(runId: string, keys: string[]): Record<string, NeedsResult> {
531- const out: Record<string, NeedsResult> = {};
532- for (const key of keys) {
533- const rows = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ?", runId, key).toArray();
534- let result: Conclusion = "success";
535- const outputs: Record<string, string> = {};
536- if (rows.length === 0) result = "skipped";
537- for (const r of rows) {
538- Object.assign(outputs, JSON.parse(r.outputs));
539- const c = r.conclusion === "timed_out" ? "failure" : (r.conclusion ?? "failure");
540- if (c === "failure") result = "failure";
541- else if (c === "cancelled" && result !== "failure") result = "cancelled";
542- else if (c === "skipped" && result === "success" && rows.length === 1) result = "skipped";
536+ /** The run's workflow with reusable workflow calls expanded (see ./reusable.ts). */
537+ private async expanded(run: RunRow): Promise<Expanded> {
538+ const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path);
539+ if (!wf?.workflow) throw new ReusableError(`workflow file not found or invalid: ${wf?.errors.join("; ") || run.workflow_path}`);
540+ return expandReusable(wf.workflow, { repoId: null, sha: run.sha }, (uses, from) => this.loadCalled(uses, from));
541+ }
542+
543+ /** Load a called workflow: `./path` from the same commit, or `owner/repo/path@ref` from a repository on this instance. */
544+ private async loadCalled(uses: string, from: Source): Promise<{ workflow: Workflow; source: Source }> {
545+ const ref = parseCallRef(uses);
546+ const m = this.repo.meta;
547+ let source: Source;
548+ let text: string | null;
549+ if (ref.kind === "local" && from.repoId === null) {
550+ source = from;
551+ const b = await this.repo.blob(from.sha, ref.path);
552+ text = b && !b.binary ? fromUtf8(b.content) : null;
553+ } else {
554+ let repoId: string;
555+ if (ref.kind === "local") repoId = from.repoId!;
556+ else if (ref.owner.toLowerCase() === m.owner.toLowerCase() && ref.repo.toLowerCase() === m.name.toLowerCase()) repoId = m.repoId;
557+ else {
558+ const target = await new Directory(this.env).repo(ref.owner, ref.repo);
559+ // Like GitHub: public and internal workflows can be shared; private ones only within the same owner.
560+ 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}`);
561+ repoId = target.id;
543562 }
544- out[key] = { result, outputs };
563+ if (repoId === m.repoId) {
564+ const sha = ref.kind === "local" ? from.sha : await this.repo.resolve(ref.ref);
565+ if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""}`);
566+ source = { repoId: null, sha };
567+ const b = await this.repo.blob(sha, ref.path);
568+ text = b && !b.binary ? fromUtf8(b.content) : null;
569+ } else {
570+ const stub = this.env.REPO.get(this.env.REPO.idFromName(repoId));
571+ const sha = ref.kind === "local" ? from.sha : await stub.resolve(ref.ref);
572+ if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""} in ${ref.kind === "remote" ? `${ref.owner}/${ref.repo}` : "the called repository"}`);
573+ source = { repoId, sha };
574+ const b = await stub.blob(sha, ref.path);
575+ text = b && !b.binary ? fromUtf8(b.content) : null;
576+ }
545577 }
578+ if (text === null) throw new Error(`${uses} not found`);
579+ const { workflow, errors } = parseWorkflow(text, ref.path);
580+ const fatal = errors.filter((e) => !isWarning(e));
581+ if (!workflow || fatal.length) throw new Error(`${uses} is invalid: ${fatal.map((e) => `${e.path}: ${e.message}`).join("; ")}`);
582+ return { workflow, source };
583+ }
584+
585+ /** `needs` context for a job or call: keys are as written inside its own workflow. */
586+ private async needsResults(run: RunRow, exp: Expanded, keys: string[], scope: string | null): Promise<Record<string, NeedsResult>> {
587+ const out: Record<string, NeedsResult> = {};
588+ for (const key of keys) out[localKey(key, scope)] = await this.resultOf(run, exp, key);
546589 return out;
547590 }
548591
592+ /** Result and outputs of a job, or of a reusable workflow call (from its `on.workflow_call.outputs`). */
593+ private async resultOf(run: RunRow, exp: Expanded, key: string): Promise<NeedsResult> {
594+ const call = exp.calls.get(key);
595+ const rows = call
596+ ? this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND substr(job_key, 1, ?) = ?", run.id, key.length + 1, `${key}/`).toArray()
597+ : this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ?", run.id, key).toArray();
598+ let result: Conclusion = rows.length === 0 ? "skipped" : "success";
599+ let outputs: Record<string, string> = {};
600+ for (const r of rows) {
601+ if (!call) Object.assign(outputs, JSON.parse(r.outputs));
602+ const c = r.conclusion === "timed_out" ? "failure" : (r.conclusion ?? "failure");
603+ if (c === "failure") result = "failure";
604+ else if (c === "cancelled" && result !== "failure") result = "cancelled";
605+ }
606+ if (rows.length && rows.every((r) => r.conclusion === "skipped")) result = "skipped";
607+ if (call && Object.keys(call.workflow.on.workflow_call?.outputs ?? {}).length) {
608+ const jobs: Record<string, NeedsResult> = {};
609+ for (const k of Object.keys(call.workflow.jobs)) jobs[k] = await this.resultOf(run, exp, `${key}/${k}`);
610+ const { inputs } = await this.callContext(run, exp, key, false);
611+ const ctx: ExprContext = { github: this.githubContext(run, localKey(key, call.parent), ""), jobs, inputs, vars: this.variables() };
612+ outputs = {};
613+ for (const [name, d] of Object.entries(call.workflow.on.workflow_call!.outputs!)) {
614+ try {
615+ const v = evaluateValue(d.value, ctx);
616+ outputs[name] = v === undefined || v === null ? "" : typeof v === "object" ? JSON.stringify(v) : String(v);
617+ } catch {
618+ outputs[name] = "";
619+ }
620+ }
621+ }
622+ return { result, outputs };
623+ }
624+
549625 /**
626+ * `inputs` and `secrets` inside a reusable workflow call (or at the top level when
627+ * `callKey` is null): inputs from the caller's `with` and the declared defaults,
628+ * secrets from `secrets: inherit` or the caller's mapping.
629+ */
630+ private async callContext(run: RunRow, exp: Expanded, callKey: string | null, withSecrets: boolean): Promise<{ inputs: Record<string, unknown>; secrets: Record<string, string> }> {
631+ if (callKey === null) return { inputs: JSON.parse(run.inputs), secrets: withSecrets ? { ...(await this.orgSecrets()), ...(await this.repoSecrets()) } : {} };
632+ const call = exp.calls.get(callKey)!;
633+ const parent = await this.callContext(run, exp, call.parent, withSecrets);
634+ 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() };
635+ const inputs: Record<string, unknown> = {};
636+ for (const [k, d] of Object.entries(call.workflow.on.workflow_call?.inputs ?? {})) {
637+ let v = call.job.with && k in call.job.with ? evaluateValue(call.job.with[k], ctx) : evaluateValue(d.default, ctx);
638+ if (v === undefined || v === null || v === "") {
639+ if (d.required) throw new ReusableError(`jobs.${callKey}: input '${k}' is required by ${call.job.uses}`);
640+ v = d.type === "boolean" ? false : d.type === "number" ? 0 : "";
641+ }
642+ inputs[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v;
643+ }
644+ const secrets: Record<string, string> = {};
645+ if (withSecrets) {
646+ if (call.job.secrets === "inherit") Object.assign(secrets, parent.secrets);
647+ else for (const [name, expr] of Object.entries(call.job.secrets ?? {})) secrets[name.toUpperCase()] = String(evaluateValue(expr, { ...ctx, secrets: parent.secrets }) ?? "");
648+ for (const [name, d] of Object.entries(call.workflow.on.workflow_call?.secrets ?? {})) {
649+ if (d.required && !secrets[name.toUpperCase()]) throw new ReusableError(`jobs.${callKey}: secret '${name}' is required by ${call.job.uses}`);
650+ }
651+ }
652+ return { inputs, secrets };
653+ }
654+
655+ /**
550656 * Start every matrix instance of a job. Returns the ids the workflow must wait
551657 * for (instances beyond max-parallel are queued and started as others finish).
552658 */
553659 async startJobKey(runId: string, jobKey: string): Promise<{ jobIds: string[] }> {
554660 const run = this.runRow(runId);
555- const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path)!.workflow!;
556- const job = wf.jobs[jobKey]!;
661+ const exp = await this.expanded(run);
662+ const job = exp.workflow.jobs[jobKey]!;
663+ const owner = exp.workflowOf.get(jobKey)!;
664+ const parent = exp.parentOf.get(jobKey) ?? null;
557665 const cancelled = run.conclusion === "cancelled";
558- const needs = this.needsResults(runId, job.needs);
559- const status = statusFor(Object.values(needs), cancelled);
560- const baseCtx: ExprContext = { github: this.githubContext(run, jobKey, ""), needs, inputs: JSON.parse(run.inputs), vars: this.variables(), env: wf.env };
666+ const chain: string[] = [];
667+ for (let c = parent; c; c = exp.calls.get(c)!.parent) chain.unshift(c);
668+ const prefix = chain.map((c) => exp.calls.get(c)!.job.name ?? localKey(c, exp.calls.get(c)!.parent)).join(" / ");
669+ const label = (name: string) => (prefix ? `${prefix} / ${name}` : name);
561670 const fail = (name: string, message: string, conclusion: Conclusion = "failure") => {
562671 const id = newId();
563672 const t = now();
@@ -565,35 +674,50 @@
565674 this.repo.broadcast(`run:${runId}`, { action: "job", jobId: id });
566675 return { jobIds: [] as string[] };
567676 };
677+ const short = localKey(jobKey, parent);
678+ let needs: Record<string, NeedsResult>;
679+ let baseCtx: ExprContext;
568680 let shouldRun: boolean;
569681 try {
570- shouldRun = evaluateIf(job.if, baseCtx, status);
682+ // The callers decide first: a skipped call, or one whose needs failed, skips everything inside it.
683+ for (const c of chain) {
684+ const call = exp.calls.get(c)!;
685+ const callNeeds = await this.needsResults(run, exp, call.needs, call.parent);
686+ const outer = await this.callContext(run, exp, call.parent, false);
687+ const callCtx: ExprContext = { github: this.githubContext(run, localKey(c, call.parent), ""), needs: callNeeds, inputs: outer.inputs, vars: this.variables() };
688+ if (!evaluateIf(call.job.if, callCtx, statusFor(Object.values(callNeeds), cancelled))) return fail(label(job.name ?? short), "", "skipped");
689+ }
690+ // Resolve the call's inputs and secrets now, so a missing required one fails here rather than in the runner.
691+ if (parent) await this.callContext(run, exp, parent, true);
692+ needs = await this.needsResults(run, exp, exp.needs.get(jobKey) ?? [], parent);
693+ baseCtx = { github: this.githubContext(run, short, ""), needs, inputs: (await this.callContext(run, exp, parent, false)).inputs, vars: this.variables(), env: owner.env };
694+ shouldRun = evaluateIf(job.if, baseCtx, statusFor(Object.values(needs), cancelled));
571695 } catch (e) {
572- return fail(job.name ?? jobKey, `Error evaluating 'if': ${(e as Error).message}`);
696+ return fail(label(job.name ?? short), e instanceof ReusableError ? e.message : `Error evaluating 'if': ${(e as Error).message}`);
573697 }
574- if (!shouldRun) return fail(job.name ?? jobKey, "", "skipped");
575- if (job.uses) return fail(job.name ?? jobKey, `Reusable workflows (jobs.${jobKey}.uses) are not supported yet on Diffinery.`);
698+ if (!shouldRun) return fail(label(job.name ?? short), "", "skipped");
576699 const labels = job.runsOn.map((l) => String(evaluateValue(l, baseCtx)));
577700 const unsupported = labels.filter((l) => !SUPPORTED_LABEL.test(l));
578701 const selfHosted = labels.some((l) => l.toLowerCase() === "self-hosted");
579- if (unsupported.length && !selfHosted) return fail(job.name ?? jobKey, `No Diffinery runner matches runs-on: ${unsupported.join(", ")}. Linux (ubuntu-*) runners are hosted; anything else needs runs-on: [self-hosted, ...] and a registered runner.`);
580- if (job.container) return fail(job.name ?? jobKey, "Job containers (jobs.<id>.container) are not supported yet; steps run directly on the Ubuntu runner image.");
581- if (job.services && Object.keys(job.services).length) return fail(job.name ?? jobKey, "Service containers (jobs.<id>.services) are not supported yet on Diffinery runners.");
702+ if (unsupported.length && !selfHosted) return fail(label(job.name ?? short), `No Diffinery runner matches runs-on: ${unsupported.join(", ")}. Linux (ubuntu-*) runners are hosted; anything else needs runs-on: [self-hosted, ...] and a registered runner.`);
703+ if (job.container) return fail(label(job.name ?? short), "Job containers (jobs.<id>.container) are not supported yet; steps run directly on the Ubuntu runner image.");
704+ if (job.services && Object.keys(job.services).length) return fail(label(job.name ?? short), "Service containers (jobs.<id>.services) are not supported yet on Diffinery runners.");
582705
583706 let combos: Array<Record<string, unknown>>;
584707 try {
585708 combos = expandMatrix(job.strategy?.matrix, baseCtx);
586709 } catch (e) {
587- return fail(job.name ?? jobKey, `Error expanding matrix: ${(e as Error).message}`);
710+ return fail(label(job.name ?? short), `Error expanding matrix: ${(e as Error).message}`);
588711 }
589- if (combos.length > 256) return fail(job.name ?? jobKey, "A matrix may produce at most 256 jobs.");
712+ if (combos.length > 256) return fail(label(job.name ?? short), "A matrix may produce at most 256 jobs.");
590713 const failFast = evaluateValue(job.strategy?.failFast ?? true, baseCtx) !== false;
591714 const maxParallel = Number(evaluateValue(job.strategy?.maxParallel ?? combos.length, baseCtx)) || combos.length;
592715 const ids: string[] = [];
593716 combos.forEach((matrix, i) => {
594717 const ctx = { ...baseCtx, matrix, strategy: { "fail-fast": failFast, "job-index": i, "job-total": combos.length, "max-parallel": maxParallel } };
595- let name = job.name ? interpolate(job.name, ctx) : jobKey;
596- if (!job.name && Object.keys(matrix).length) name = `${jobKey} (${Object.values(matrix).map((v) => (typeof v === "object" ? JSON.stringify(v) : String(v))).join(", ")})`;
718+ let name = job.name ? interpolate(job.name, ctx) : short;
719+ if (!job.name && Object.keys(matrix).length) name = `${short} (${Object.values(matrix).map((v) => (typeof v === "object" ? JSON.stringify(v) : String(v))).join(", ")})`;
720+ name = label(name);
597721 const timeout = Number(evaluateValue(job.timeoutMinutes, ctx)) || 360;
598722 const id = newId();
599723 this.sql.exec(
@@ -771,10 +895,11 @@
771895 const job = await this.authJob(jobId, token);
772896 if (job.status !== "in_progress") throw new HttpError(409, "job is not running");
773897 const run = this.runRow(job.run_id);
774- const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path)!.workflow!;
775- const def: Job = wf.jobs[job.job_key]!;
898+ const exp = await this.expanded(run);
899+ const def: Job = exp.workflow.jobs[job.job_key]!;
900+ const wf = exp.workflowOf.get(job.job_key)!;
776901 const allowed = wf.permissions === "write-all" || (typeof wf.permissions === "object" && wf.permissions["id-token"] === "write") || (typeof def.permissions === "object" && def.permissions["id-token"] === "write") || def.permissions === "write-all";
777- const ctx = this.githubContext(run, job.job_key, token) as Record<string, Json>;
902+ const ctx = this.githubContext(run, localKey(job.job_key, exp.parentOf.get(job.job_key) ?? null), token) as Record<string, Json>;
778903 const github: Record<string, string> = {};
779904 for (const [k, v] of Object.entries(ctx)) if (typeof v === "string" || typeof v === "number") github[k] = String(v);
780905 return { allowed, github };
@@ -783,31 +908,41 @@
783908 async runnerSpec(jobId: string, token: string): Promise<JobSpec> {
784909 const job = await this.authJob(jobId, token);
785910 const run = this.runRow(job.run_id);
786- const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path)!.workflow!;
787- const def: Job = wf.jobs[job.job_key]!;
911+ const exp = await this.expanded(run);
912+ const def: Job = exp.workflow.jobs[job.job_key]!;
913+ const wf = exp.workflowOf.get(job.job_key)!;
914+ const parent = exp.parentOf.get(job.job_key) ?? null;
915+ const short = localKey(job.job_key, parent);
916+ let scope: { inputs: Record<string, unknown>; secrets: Record<string, string> };
917+ try {
918+ scope = await this.callContext(run, exp, parent, true);
919+ } catch (e) {
920+ await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } });
921+ throw new HttpError(409, (e as Error).message);
922+ }
788923 const m = this.repo.meta;
789924 const t = now();
790925 this.sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ? WHERE id = ?", t, jobId);
791926 this.repo.upsertCheck({ 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}` });
792927 this.repo.broadcast(`run:${run.id}`, { action: "job", jobId });
793- const secrets: Record<string, string> = { ...(await this.orgSecrets()), ...(await this.repoSecrets()), GITHUB_TOKEN: token, DIFFINERY_TOKEN: token };
928+ const secrets: Record<string, string> = { ...scope.secrets, GITHUB_TOKEN: token, DIFFINERY_TOKEN: token };
794929 const idToken = wf.permissions === "write-all" || (typeof wf.permissions === "object" && wf.permissions["id-token"] === "write") || (typeof def.permissions === "object" && def.permissions["id-token"] === "write") || def.permissions === "write-all";
795930 return {
796931 runId: run.id,
797932 jobId,
798- jobKey: job.job_key,
933+ jobKey: short,
799934 name: job.name,
800935 repo: { owner: m.owner, name: m.name, cloneUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, defaultBranch: m.defaultBranch },
801936 sha: run.sha,
802937 ref: run.ref,
803938 workflowPath: run.workflow_path,
804- github: this.githubContext(run, job.job_key, token),
939+ github: this.githubContext(run, short, token),
805940 env: { ...wf.env, ...def.env },
806941 defaults: { run: { ...wf.defaults.run, ...def.defaults.run } },
807942 matrix: JSON.parse(job.matrix),
808943 strategy: JSON.parse(job.strategy),
809- needs: this.needsResults(run.id, def.needs),
810- inputs: JSON.parse(run.inputs),
944+ needs: await this.needsResults(run, exp, exp.needs.get(job.job_key) ?? [], parent),
945+ inputs: scope.inputs,
811946 vars: this.variables(),
812947 secrets,
813948 outputs: def.outputs,
addedapps/diffinery/src/repo/reusable.ts+105−0View file
@@ -1,0 +1,105 @@
1+import type { Job, Workflow } from "@diffinery/workflow-yaml";
2+
3+/**
4+ * Reusable workflows (`jobs.<id>.uses`), GitHub-compatible.
5+ *
6+ * A run's workflow is flattened before it is planned: a caller job `deploy` that
7+ * uses a workflow with jobs `build` and `ship` becomes the jobs `deploy/build`
8+ * and `deploy/ship`. Inner `needs` are prefixed; inner jobs without `needs`
9+ * wait for whatever the caller job needs; jobs that need `deploy` wait for every
10+ * job inside it. Calls nest up to four levels deep, like on GitHub.
11+ */
12+
13+/** Where a workflow file came from: this repository (repoId null) or another one, at a commit. */
14+export type Source = { repoId: string | null; sha: string };
15+
16+export type CallSite = {
17+ /** Full key of the caller job, e.g. "deploy" or "deploy/release". */
18+ key: string;
19+ /** Enclosing call, or null for a top-level caller job. */
20+ parent: string | null;
21+ /** The caller job: uses, with, secrets, if, name. */
22+ job: Job;
23+ /** The called workflow. */
24+ workflow: Workflow;
25+ /** The caller's own `needs`, as full keys in the parent scope. */
26+ needs: string[];
27+};
28+
29+export type Expanded = {
30+ /** The run's workflow with only runnable jobs, keyed by full key, `needs` expanded for ordering. */
31+ workflow: Workflow;
32+ calls: Map<string, CallSite>;
33+ /** Full key → the call it belongs to (null at top level). Covers jobs and calls. */
34+ parentOf: Map<string, string | null>;
35+ /** Declared `needs` per job or call, as full keys (a call stays a call). */
36+ needs: Map<string, string[]>;
37+ /** The workflow each runnable job is defined in (env, defaults, permissions). */
38+ workflowOf: Map<string, Workflow>;
39+};
40+
41+export type Loader = (uses: string, from: Source) => Promise<{ workflow: Workflow; source: Source }>;
42+
43+export const MAX_CALL_DEPTH = 4;
44+export const MAX_CALLED_WORKFLOWS = 20;
45+
46+export class ReusableError extends Error {}
47+
48+/** Short key inside the call that defines it ("deploy/build" → "build"). */
49+export const localKey = (full: string, parent: string | null): string => (parent ? full.slice(parent.length + 1) : full);
50+
51+export async function expandReusable(root: Workflow, rootSource: Source, load: Loader): Promise<Expanded> {
52+ const calls = new Map<string, CallSite>();
53+ const parentOf = new Map<string, string | null>();
54+ const needs = new Map<string, string[]>();
55+ const workflowOf = new Map<string, Workflow>();
56+ const leaves: Record<string, Job> = {};
57+ let called = 0;
58+
59+ const walk = async (wf: Workflow, source: Source, prefix: string | null, depth: number, inherited: string[]): Promise<void> => {
60+ for (const [key, job] of Object.entries(wf.jobs)) {
61+ const full = prefix ? `${prefix}/${key}` : key;
62+ const declared = job.needs.map((n) => (prefix ? `${prefix}/${n}` : n));
63+ const ordering = job.needs.length ? declared : inherited;
64+ parentOf.set(full, prefix);
65+ needs.set(full, declared);
66+ if (!job.uses) {
67+ leaves[full] = { ...job, key: full, needs: ordering };
68+ workflowOf.set(full, wf);
69+ continue;
70+ }
71+ if (depth >= MAX_CALL_DEPTH) throw new ReusableError(`jobs.${full}: reusable workflows can be nested at most ${MAX_CALL_DEPTH} levels deep`);
72+ if (++called > MAX_CALLED_WORKFLOWS) throw new ReusableError(`a run can call at most ${MAX_CALLED_WORKFLOWS} reusable workflows`);
73+ if (job.strategy?.matrix !== undefined) throw new ReusableError(`jobs.${full}: a matrix on a job that calls a reusable workflow is not supported yet`);
74+ let target: { workflow: Workflow; source: Source };
75+ try {
76+ target = await load(job.uses, source);
77+ } catch (e) {
78+ throw new ReusableError(`jobs.${full}.uses: ${(e as Error).message}`);
79+ }
80+ if (!target.workflow.on.workflow_call) throw new ReusableError(`jobs.${full}.uses: ${job.uses} is not a reusable workflow (it has no 'on: workflow_call' trigger)`);
81+ const declaredInputs = target.workflow.on.workflow_call.inputs ?? {};
82+ for (const k of Object.keys(job.with ?? {})) {
83+ if (!(k in declaredInputs)) throw new ReusableError(`jobs.${full}.with.${k}: ${job.uses} has no input named '${k}'`);
84+ }
85+ calls.set(full, { key: full, parent: prefix, job, workflow: target.workflow, needs: declared });
86+ await walk(target.workflow, target.source, full, depth + 1, ordering);
87+ }
88+ };
89+ await walk(root, rootSource, null, 0, []);
90+
91+ // A dependency on a call means a dependency on every runnable job inside it.
92+ const under = (call: string) => Object.keys(leaves).filter((k) => k.startsWith(`${call}/`));
93+ for (const job of Object.values(leaves)) job.needs = [...new Set(job.needs.flatMap((n) => (calls.has(n) ? under(n) : [n])))];
94+
95+ return { workflow: { ...root, jobs: leaves }, calls, parentOf, needs, workflowOf };
96+}
97+
98+/** Parse a `uses:` reference to a reusable workflow. */
99+export function parseCallRef(uses: string): { kind: "local"; path: string } | { kind: "remote"; owner: string; repo: string; path: string; ref: string } {
100+ const local = /^\.\/(.+\.ya?ml)$/.exec(uses);
101+ if (local) return { kind: "local", path: local[1]! };
102+ const remote = /^([A-Za-z0-9][\w.-]*)\/([\w.-]+)\/(.+\.ya?ml)@(.+)$/.exec(uses);
103+ if (remote) return { kind: "remote", owner: remote[1]!, repo: remote[2]!, path: remote[3]!, ref: remote[4]! };
104+ throw new Error(`'${uses}' must be ./.github/workflows/<file>.yml or <owner>/<repo>/.github/workflows/<file>.yml@<ref>`);
105+}
modifiedapps/diffinery/test/e2e/e2e.test.mjs+96−4View file
@@ -10,6 +10,7 @@
1010 import { spawn, spawnSync, execFileSync } from "node:child_process";
1111 import { mkdtempSync, writeFileSync, mkdirSync, readFileSync, existsSync, rmSync } from "node:fs";
1212 import { tmpdir } from "node:os";
13+import { randomBytes } from "node:crypto";
1314 import { join, dirname, resolve } from "node:path";
1415 import { fileURLToPath } from "node:url";
1516 import { createServer } from "node:net";
@@ -100,9 +101,9 @@
100101 : spawnSync(process.execPath, [AGENT], { encoding: "utf8", env: { ...process.env, ...jobEnv, DIFFINERY_WORK_ROOT: join(work, `job-${claim.job_id}`) }, timeout: 120_000 });
101102 const out = `${r.stdout}${r.stderr}`;
102103 if (r.status !== 0) {
103- const log = await (await fetch(`${base}/api/v3/repos/${owner}/${repo}/actions/jobs/${claim.job_id}/logs`, { headers: { authorization: `Bearer ${token}` } })).text();
104- const job = await api("GET", `/repos/${owner}/${repo}/actions/jobs/${claim.job_id}`);
105- assert.fail(`runner agent exited ${r.status}:\n${out}\n--- job ---\n${JSON.stringify(job.json, null, 2)}\n--- job log ---\n${log}`);
104+ const log = await fetch(`${base}/api/v3/repos/${owner}/${repo}/actions/jobs/${claim.job_id}/logs`, { headers: { authorization: `Bearer ${token}` } }).then((x) => x.text(), (e) => String(e));
105+ const job = await api("GET", `/repos/${owner}/${repo}/actions/jobs/${claim.job_id}`).catch((e) => ({ json: String(e) }));
106+ assert.fail(`runner agent exited ${r.status}:\n${out}\n--- job ---\n${JSON.stringify(job.json, null, 2)}\n--- job log ---\n${log}\n--- wrangler log tail ---\n${devLog.slice(-3000)}`);
106107 }
107108 return { claim, out };
108109 }
@@ -115,7 +116,7 @@
115116 execFileSync("pnpm", ["exec", "wrangler", "d1", "migrations", "apply", "diffinery-dev", "--local", "-e", "dev", "--persist-to", persist], { cwd: APP, stdio: "pipe", env: { ...process.env, CI: "1" } });
116117 // Until the web UI exists, run the same app through the API-only entry.
117118 const entry = existsSync(join(APP, "src/web/routes.tsx")) ? [] : ["test/e2e/entry.ts"];
118- dev = spawn("pnpm", ["exec", "wrangler", "dev", ...entry, "-e", "dev", "--ip", "127.0.0.1", "--port", String(port), "--persist-to", persist, "--var", `PUBLIC_URL:${base}`, "--show-interactive-dev-session=false"], {
119+ dev = spawn("pnpm", ["exec", "wrangler", "dev", ...entry, "-e", "dev", "--ip", "127.0.0.1", "--port", String(port), "--persist-to", persist, "--var", `PUBLIC_URL:${base}`, "--var", `MASTER_KEY:${randomBytes(32).toString("base64")}`, "--show-interactive-dev-session=false"], {
119120 cwd: APP,
120121 env: { ...process.env, CI: "1", WRANGLER_SEND_METRICS: "false" },
121122 stdio: ["ignore", "pipe", "pipe"],
@@ -343,3 +344,94 @@
343344 const agentCommit = await ok("GET", `/repos/${owner}/demo/commits/mcp-change`);
344345 assert.match(JSON.stringify(agentCommit), /e2e-bot/);
345346 });
347+
348+test("reusable workflows: inputs, secrets, outputs, nesting and skipped calls", async () => {
349+ const owner = (await ok("GET", "/user")).login;
350+ await ok("POST", "/user/repos", { name: "reuse" });
351+ await api("PUT", `/repos/${owner}/reuse/actions/secrets/TOPSECRET`, { value: "s3cret" });
352+ const src = join(work, "reuse");
353+ mkdirSync(join(src, ".github/workflows"), { recursive: true });
354+ writeFileSync(
355+ join(src, ".github/workflows/lib.yml"),
356+ `on:
357+ workflow_call:
358+ inputs:
359+ greeting: { type: string, required: true }
360+ loud: { type: boolean, default: false }
361+ secrets:
362+ TOPSECRET: { required: true }
363+ outputs:
364+ word:
365+ value: \${{ jobs.shout.outputs.word }}
366+jobs:
367+ say:
368+ runs-on: [self-hosted, linux]
369+ outputs:
370+ word: \${{ steps.s.outputs.word }}
371+ steps:
372+ - id: s
373+ run: echo "word=\${{ inputs.greeting }}-there" >> "$GITHUB_OUTPUT"
374+ - run: echo "secret ok=\${{ secrets.TOPSECRET == 's3cret' }} loud=\${{ inputs.loud }} job=\${{ github.job }}"
375+ shout:
376+ needs: say
377+ runs-on: [self-hosted, linux]
378+ outputs:
379+ word: \${{ steps.s.outputs.word }}
380+ steps:
381+ - id: s
382+ run: echo "word=\${{ needs.say.outputs.word }}!" >> "$GITHUB_OUTPUT"
383+`,
384+ );
385+ writeFileSync(
386+ join(src, ".github/workflows/main.yml"),
387+ `name: Main
388+on: push
389+jobs:
390+ call:
391+ name: Greeter
392+ uses: ./.github/workflows/lib.yml
393+ with:
394+ greeting: \${{ github.event_name == 'push' && 'hi' || 'no' }}
395+ loud: true
396+ secrets: inherit
397+ nope:
398+ if: \${{ false }}
399+ uses: ./.github/workflows/lib.yml
400+ with:
401+ greeting: never
402+ secrets:
403+ TOPSECRET: \${{ secrets.TOPSECRET }}
404+ after:
405+ needs: [call, nope]
406+ if: \${{ always() }}
407+ runs-on: [self-hosted, linux]
408+ steps:
409+ - run: echo "got=\${{ needs.call.outputs.word }} call=\${{ needs.call.result }} nope=\${{ needs.nope.result }}"
410+`,
411+ );
412+ git(src, ["init", "-q", "-b", "main"]);
413+ git(src, ["add", "."]);
414+ git(src, ["commit", "-q", "-m", "reusable"]);
415+ git(src, ["remote", "add", "origin", `${base}/${owner}/reuse.git`]);
416+ git(src, ["push", "-q", "origin", "main"]);
417+
418+ const run = await until("reuse run", async () => (await ok("GET", `/repos/${owner}/reuse/actions/runs`)).workflow_runs[0]);
419+ const logOf = async (id) => (await fetch(`${base}/api/v3/repos/${owner}/reuse/actions/jobs/${id}/logs`, { headers: { authorization: `Bearer ${token}` } })).text();
420+ const say = await runOneJob(owner, "reuse");
421+ assert.equal(say.claim.name, "Greeter / say");
422+ const sayLog = await logOf(say.claim.job_id);
423+ assert.match(sayLog, /secret ok=true loud=true job=say/);
424+ assert.doesNotMatch(sayLog, /s3cret/);
425+ const shout = await runOneJob(owner, "reuse");
426+ assert.equal(shout.claim.name, "Greeter / shout");
427+ const after = await runOneJob(owner, "reuse");
428+ assert.equal(after.claim.name, "after");
429+ assert.match(await logOf(after.claim.job_id), /got=hi-there! call=success nope=skipped/);
430+ const done = await until("reuse run to complete", async () => {
431+ const r = await ok("GET", `/repos/${owner}/reuse/actions/runs/${run.id}`);
432+ return r.status === "completed" ? r : null;
433+ });
434+ assert.equal(done.conclusion, "success");
435+ const jobs = (await ok("GET", `/repos/${owner}/reuse/actions/runs/${run.id}/jobs`)).jobs;
436+ assert.deepEqual(jobs.filter((j) => j.conclusion === "skipped").map((j) => j.name).sort(), ["nope / say", "nope / shout"]);
437+});
addedapps/diffinery/test/unit/reusable.test.ts+101−0View file
@@ -1,0 +1,101 @@
1+import { describe, expect, it } from "vitest";
2+import { jobOrder, parseWorkflow, type Workflow } from "@diffinery/workflow-yaml";
3+import { ReusableError, expandReusable, parseCallRef, type Loader } from "../../src/repo/reusable.js";
4+
5+const wf = (yaml: string, path = ".github/workflows/x.yml"): Workflow => {
6+ const { workflow, errors } = parseWorkflow(yaml, path);
7+ if (!workflow) throw new Error(errors.map((e) => e.message).join("; "));
8+ return workflow;
9+};
10+
11+const LIB = wf(`on:
12+ workflow_call:
13+ inputs:
14+ who: { type: string }
15+jobs:
16+ a:
17+ runs-on: ubuntu-latest
18+ steps: [{ run: echo a }]
19+ b:
20+ needs: a
21+ runs-on: ubuntu-latest
22+ steps: [{ run: echo b }]
23+`);
24+
25+const files: Record<string, Workflow> = { "./.github/workflows/lib.yml": LIB };
26+const load: Loader = async (uses, from) => {
27+ const w = files[uses];
28+ if (!w) throw new Error(`${uses} not found`);
29+ return { workflow: w, source: from };
30+};
31+const root = { repoId: null, sha: "abc" };
32+
33+describe("expandReusable", () => {
34+ it("flattens a call into prefixed jobs and orders around it", async () => {
35+ const main = wf(`on: push
36+jobs:
37+ first:
38+ runs-on: ubuntu-latest
39+ steps: [{ run: echo }]
40+ call:
41+ needs: first
42+ uses: ./.github/workflows/lib.yml
43+ with: { who: me }
44+ last:
45+ needs: call
46+ runs-on: ubuntu-latest
47+ steps: [{ run: echo }]
48+`);
49+ const exp = await expandReusable(main, root, load);
50+ expect(Object.keys(exp.workflow.jobs)).toEqual(["first", "call/a", "call/b", "last"]);
51+ expect(exp.workflow.jobs["call/a"]!.needs).toEqual(["first"]);
52+ expect(exp.workflow.jobs["call/b"]!.needs).toEqual(["call/a"]);
53+ expect(exp.workflow.jobs["last"]!.needs.sort()).toEqual(["call/a", "call/b"]);
54+ expect(exp.needs.get("last")).toEqual(["call"]);
55+ expect(exp.parentOf.get("call/b")).toBe("call");
56+ expect(jobOrder(exp.workflow).order).toEqual([["first"], ["call/a"], ["call/b"], ["last"]]);
57+ });
58+
59+ it("nests calls and stops at four levels", async () => {
60+ const level = (n: number) => wf(`on: workflow_call
61+jobs:
62+ j:
63+ uses: ./.github/workflows/l${n + 1}.yml
64+`);
65+ for (let i = 1; i <= 5; i++) files[`./.github/workflows/l${i}.yml`] = level(i);
66+ const main = wf(`on: push
67+jobs:
68+ top:
69+ uses: ./.github/workflows/l1.yml
70+`);
71+ await expect(expandReusable(main, root, load)).rejects.toThrow(ReusableError);
72+ files["./.github/workflows/l3.yml"] = LIB;
73+ const exp = await expandReusable(main, root, load);
74+ expect(Object.keys(exp.workflow.jobs)).toEqual(["top/j/j/a", "top/j/j/b"]);
75+ });
76+
77+ it("rejects unknown inputs and workflows without workflow_call", async () => {
78+ files["./.github/workflows/plain.yml"] = wf(`on: push
79+jobs:
80+ x:
81+ runs-on: ubuntu-latest
82+ steps: [{ run: echo }]
83+`);
84+ const bad = (uses: string, w = "") => wf(`on: push
85+jobs:
86+ c:
87+ uses: ${uses}
88+${w}`);
89+ await expect(expandReusable(bad("./.github/workflows/plain.yml"), root, load)).rejects.toThrow(/not a reusable workflow/);
90+ await expect(expandReusable(bad("./.github/workflows/lib.yml", " with: { nope: 1 }\n"), root, load)).rejects.toThrow(/no input named 'nope'/);
91+ await expect(expandReusable(bad("./.github/workflows/missing.yml"), root, load)).rejects.toThrow(/not found/);
92+ });
93+});
94+
95+describe("parseCallRef", () => {
96+ it("parses local and remote references", () => {
97+ expect(parseCallRef("./.github/workflows/a.yml")).toEqual({ kind: "local", path: ".github/workflows/a.yml" });
98+ expect(parseCallRef("octo/tools/.github/workflows/b.yaml@v1")).toEqual({ kind: "remote", owner: "octo", repo: "tools", path: ".github/workflows/b.yaml", ref: "v1" });
99+ expect(() => parseCallRef("octo/tools/.github/workflows/b.yaml")).toThrow();
100+ });
101+});