Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
CodeIssuesDiscussionsWikiPull RequestsProjectsCommitsActionsReleasesContributorsPulse● GatesSecuritySettingsDeploymentsPipelineInsightsAgents✨ Explain✨ Ask AI✨ Workspace✨ Spec✨ Tests▓ Debt Map✨ NL Search🏛 Archaeology
Blame · Line-by-line history

workflow-runner.ts

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

Gluecron.com/src/lib/workflow-runner.ts
workflow-runner.tsBlame1107 lines · 2 contributors
Normal viewRaw
eafe8c6Claude1/**
2 * Workflow runner (Block C1) — executes queued `workflow_runs` rows by
3 * cloning the repo at the target commit into a tmpdir and running each
4 * job's steps as bash subprocesses.
5 *
6 * Philosophy (mirrors post-receive.ts): never crash the caller. Every DB
7 * call is wrapped in try/catch. All step output is size-capped so a runaway
8 * process can't blow up Postgres rows. Logs are stored inline on the job
9 * row for v1 — no streaming, no object storage. Step timeouts are enforced
10 * so workers never wedge.
11 *
12 * Public surface:
13 * - executeRun(runId) — run a specific queued run to completion
14 * - drainOneRun() — pick the oldest queued run and execute it
15 * - enqueueRun(opts) — insert a new run at the tail of the queue
16 * - startWorker({ interval }) — background poll loop (returns stop fn)
17 */
18import { and, asc, eq, sql } from "drizzle-orm";
19import { mkdtemp, rm } from "fs/promises";
20import { tmpdir } from "os";
21import { join } from "path";
22import { config } from "./config";
23import { db } from "../db";
24import {
25 repositories,
26 workflowJobs,
27 workflowRuns,
28 workflows,
29} from "../db/schema";
18b210dClaude30import {
31 loadSecretsContext,
32 substituteSecrets,
33} from "./workflow-secrets";
f534efeClaude34import { saveCacheEntry } from "./actions/cache-action";
eafe8c6Claude35
36// ---------------------------------------------------------------------------
37// Tunables
38// ---------------------------------------------------------------------------
39
40/** Per-step subprocess timeout. */
41const STEP_TIMEOUT_MS = 600_000; // 10 minutes
42
43/** Grace period between SIGTERM and SIGKILL when killing a step. */
44const KILL_GRACE_MS = 5_000;
45
46/** Cap on full `workflow_jobs.logs` field. */
47const JOB_LOG_CAP_BYTES = 64 * 1024;
48
49/** Cap on per-step stdout/stderr excerpts stored in `steps` JSON. */
50const STEP_STREAM_CAP_BYTES = 16 * 1024;
51
52/** Default worker poll interval. */
53const DEFAULT_POLL_INTERVAL_MS = 2_000;
54
55// ---------------------------------------------------------------------------
56// Types
57// ---------------------------------------------------------------------------
58
59interface ParsedStep {
60 name?: string;
61 run?: string;
62 // `uses` / `with` etc. tolerated but ignored in v1.
63 [key: string]: unknown;
64}
65
66interface ParsedJob {
67 name?: string;
68 "runs-on"?: string;
69 runsOn?: string;
70 steps?: ParsedStep[];
71 [key: string]: unknown;
72}
73
74interface ParsedWorkflow {
75 name?: string;
76 on?: unknown;
77 jobs?: Record<string, ParsedJob> | ParsedJob[];
78 [key: string]: unknown;
79}
80
81interface StepResult {
82 name: string;
83 run: string;
84 exitCode: number | null;
85 durationMs: number;
86 stdout: string;
87 stderr: string;
88 status: "success" | "failure" | "skipped";
89}
90
91// ---------------------------------------------------------------------------
92// Small helpers
93// ---------------------------------------------------------------------------
94
95function truncate(value: string, limit: number): string {
96 if (value.length <= limit) return value;
97 return value.slice(0, limit) + "\n[... truncated ...]";
98}
99
5bb52faccanty labs100// ---------------------------------------------------------------------------
101// SECURITY: child-process environment isolation (crown-jewel secret containment)
102// ---------------------------------------------------------------------------
103//
104// User-supplied workflow `run:` steps execute via `bash -c` on the production
105// host. They MUST NOT inherit the platform's `process.env`, or any user could
106// read DATABASE_URL, WORKFLOW_SECRETS_KEY, SERVER_TARGETS_KEY, ANTHROPIC_API_KEY,
107// GLUECRON_PAT, GOOGLE_OAUTH_CLIENT_SECRET, etc. — i.e. exfiltrate every secret
108// on the box.
109//
110// Instead the runner env is built from a curated allowlist of variables that
111// shells + common tooling need to function, plus the runner's own intended
112// injections (CI flags etc.). NOTE: user-declared `${{ secrets.NAME }}` are
113// injected by *text substitution* into the `run:` string upstream (see
114// substituteSecrets in runStep) — they are deliberately NOT passed through the
115// environment, so nothing sensitive needs to ride in here.
116
117/**
118 * Minimal set of environment variables copied from the host into a workflow
119 * step's subprocess. Each is only copied if actually present in process.env.
120 * These are non-sensitive shell/tooling essentials — nothing here reveals a
121 * platform secret.
122 */
123const RUNNER_ENV_ALLOWLIST: readonly string[] = [
124 // POSIX shell + locale + tooling essentials
125 "PATH",
126 "HOME",
127 "LANG",
128 "LC_ALL",
129 "TZ",
130 "TERM",
131 "SHELL",
132 "USER",
133 "HOSTNAME",
134 "TMPDIR",
135 // Windows equivalents (host may run the runner on win32 in dev)
136 "SYSTEMROOT",
137 "WINDIR",
138 "PATHEXT",
139 "COMSPEC",
140 "TEMP",
141 "TMP",
142 "USERPROFILE",
143];
144
145/**
146 * Hard denylist safety net. Even if a key somehow appears in the allowlist (or
147 * is added there by mistake later), never leak anything whose NAME looks like a
148 * platform credential. Applied to the copied host env only; the runner's own
149 * intentional injections (`extra`, incl. any user-declared secret the repo
150 * owner explicitly configured for their step) are exempt because they are
151 * meant to be visible to the step.
152 */
153const RUNNER_ENV_DENYLIST =
154 /(SECRET|TOKEN|_KEY$|_KEY_|PASSWORD|PAT|DATABASE_URL|ANTHROPIC|OPENAI|VOYAGE|STRIPE|GOOGLE_OAUTH|GITHUB_OAUTH|WORKFLOW_SECRETS|SERVER_TARGETS|SESSION|WEBHOOK_SECRET|EMERGENCY)/i;
155
156/**
157 * Build the environment for a workflow step subprocess.
158 *
159 * Starts from the curated {@link RUNNER_ENV_ALLOWLIST} (copying only keys that
160 * exist on the host and don't trip {@link RUNNER_ENV_DENYLIST}), then merges the
161 * runner's intended `extra` vars (CI flags, run id, and — in this codebase —
162 * nothing sensitive, since user secrets are text-substituted, not env-passed).
163 * `extra` wins on key collision and is exempt from the denylist by design.
164 */
165function buildRunnerEnv(
166 extra: Record<string, string> = {}
167): Record<string, string> {
168 const out: Record<string, string> = {};
169 for (const key of RUNNER_ENV_ALLOWLIST) {
170 if (RUNNER_ENV_DENYLIST.test(key)) continue; // net: never copy a credential-shaped key
171 const val = process.env[key];
172 if (typeof val === "string") out[key] = val;
173 }
174 for (const [k, v] of Object.entries(extra)) {
175 if (typeof v === "string") out[k] = v;
176 }
177 return out;
178}
179
eafe8c6Claude180/**
181 * Normalise the parsed workflow JSON into an ordered array of jobs.
182 * Accepts either the object form (`jobs: { build: {...} }`) or an array.
183 */
184function extractJobs(parsed: ParsedWorkflow): Array<{ key: string; job: ParsedJob }> {
185 const out: Array<{ key: string; job: ParsedJob }> = [];
186 const jobs = parsed.jobs;
187 if (!jobs) return out;
188 if (Array.isArray(jobs)) {
189 jobs.forEach((job, i) => {
190 if (job && typeof job === "object") {
191 out.push({ key: String(job.name || `job-${i + 1}`), job });
192 }
193 });
194 return out;
195 }
196 if (typeof jobs === "object") {
197 for (const [key, job] of Object.entries(jobs)) {
198 if (job && typeof job === "object") {
199 out.push({ key, job: job as ParsedJob });
200 }
201 }
202 }
203 return out;
204}
205
206function parseWorkflow(parsed: string): ParsedWorkflow | null {
207 try {
208 const value = JSON.parse(parsed);
209 if (value && typeof value === "object") return value as ParsedWorkflow;
210 } catch (err) {
211 console.error("[workflow-runner] failed to parse workflow JSON:", err);
212 }
213 return null;
214}
215
2316901Claude216/**
217 * Coerce an `ExtendedParseResult` (from `workflow-parser-ext`) into the
218 * permissive `ParsedWorkflow` shape used by this v1 runner. The extended
219 * parser produces richer per-step/per-job metadata (needs, strategy, if,
220 * uses, env, outputs) — v1 only consumes `name`, `runsOn`, and `steps[].run`
221 * so the extras are tolerated via the index signature on `ParsedJob`.
222 *
223 * Returns null on parse failure or malformed input so callers fall back to
224 * the locked v1 parser.
225 */
226function _coerceExtParsed(result: unknown): ParsedWorkflow | null {
227 if (!result || typeof result !== "object") return null;
228 const r = result as { ok?: unknown; workflow?: unknown };
229 if (r.ok !== true) return null;
230 const wf = r.workflow;
231 if (!wf || typeof wf !== "object") return null;
232 const w = wf as {
233 name?: unknown;
234 on?: unknown;
235 jobs?: unknown;
236 };
237 if (!Array.isArray(w.jobs)) return null;
238 const jobs: ParsedJob[] = [];
239 for (const j of w.jobs) {
240 if (!j || typeof j !== "object") continue;
241 const job = j as Record<string, unknown>;
242 const steps: ParsedStep[] = [];
243 if (Array.isArray(job.steps)) {
244 for (const s of job.steps) {
245 if (!s || typeof s !== "object") continue;
246 const step = s as Record<string, unknown>;
247 steps.push({
248 name: typeof step.name === "string" ? step.name : undefined,
249 run: typeof step.run === "string" ? step.run : undefined,
250 });
251 }
252 }
253 jobs.push({
254 name: typeof job.name === "string" ? job.name : undefined,
255 runsOn:
256 typeof job.runsOn === "string"
257 ? job.runsOn
258 : typeof job["runs-on"] === "string"
259 ? (job["runs-on"] as string)
260 : undefined,
261 steps,
262 });
263 }
264 return {
265 name: typeof w.name === "string" ? w.name : undefined,
266 on: w.on,
267 jobs,
268 };
269}
270
eafe8c6Claude271// ---------------------------------------------------------------------------
272// Terminal-state helpers — all wrap DB calls in try/catch.
273// ---------------------------------------------------------------------------
274
275async function markRunFailed(
276 runId: string,
277 conclusion: string
278): Promise<void> {
279 try {
280 await db
281 .update(workflowRuns)
282 .set({
283 status: "failure",
284 conclusion,
285 finishedAt: new Date(),
286 })
287 .where(eq(workflowRuns.id, runId));
288 } catch (err) {
289 console.error("[workflow-runner] markRunFailed:", err);
290 }
291}
292
293async function markRunRunning(runId: string): Promise<void> {
294 try {
295 await db
296 .update(workflowRuns)
297 .set({
298 status: "running",
299 startedAt: new Date(),
300 })
301 .where(eq(workflowRuns.id, runId));
302 } catch (err) {
303 console.error("[workflow-runner] markRunRunning:", err);
304 }
305}
306
307async function markRunDone(
308 runId: string,
309 anyFailed: boolean
310): Promise<void> {
311 try {
312 await db
313 .update(workflowRuns)
314 .set({
315 status: anyFailed ? "failure" : "success",
316 conclusion: anyFailed ? "failure" : "success",
317 finishedAt: new Date(),
318 })
319 .where(eq(workflowRuns.id, runId));
320 } catch (err) {
321 console.error("[workflow-runner] markRunDone:", err);
322 }
323}
324
325// ---------------------------------------------------------------------------
326// Subprocess primitive
327// ---------------------------------------------------------------------------
328
329/**
330 * Run a single step via `bash -c`. Captures stdout/stderr (capped), enforces
331 * a hard timeout with SIGTERM → SIGKILL escalation, and returns a StepResult
332 * shaped for persistence.
18b210dClaude333 *
334 * `secrets` is the per-run plaintext map produced by
335 * `loadSecretsContext(repoId)`. Default `{}` keeps the legacy callers
336 * working — when no secrets are loaded the substitution pass is a no-op
337 * because the regex matches nothing.
eafe8c6Claude338 */
339async function runStep(
340 step: ParsedStep,
341 checkoutDir: string,
18b210dClaude342 runId: string,
343 secrets: Record<string, string> = {}
eafe8c6Claude344): Promise<StepResult> {
345 const name =
346 typeof step.name === "string" && step.name.length > 0
347 ? step.name
348 : (typeof step.run === "string" ? step.run.split("\n")[0] : "") ||
349 "step";
18b210dClaude350 const rawRun = typeof step.run === "string" ? step.run : "";
351 // Substitute `${{ secrets.NAME }}` only when the template actually
352 // contains a token — otherwise this is a free pass-through. The function
353 // is pure + safe for empty secret maps; the conditional avoids an
354 // unnecessary regex compile + scan on the hot path.
355 const run =
356 rawRun.indexOf("${{") >= 0
357 ? substituteSecrets(rawRun, secrets)
358 : rawRun;
eafe8c6Claude359 const started = Date.now();
360
361 if (!run) {
362 // No `run:` — v1 treats this as skipped (we don't support `uses:` yet).
363 return {
364 name,
365 run: "",
366 exitCode: null,
367 durationMs: 0,
368 stdout: "",
369 stderr: "",
370 status: "skipped",
371 };
372 }
373
374 let proc: ReturnType<typeof Bun.spawn> | null = null;
375 let timedOut = false;
376 let killTimer: ReturnType<typeof setTimeout> | null = null;
377 let escalateTimer: ReturnType<typeof setTimeout> | null = null;
378
379 try {
380 proc = Bun.spawn(["bash", "-c", run], {
381 cwd: checkoutDir,
382 stdout: "pipe",
383 stderr: "pipe",
5bb52faccanty labs384 // SECURITY: do NOT inherit the platform's process.env here — a user's
385 // `run:` step would otherwise read every host secret. Pass only the
386 // curated allowlist + the runner's own CI vars. See buildRunnerEnv.
387 env: buildRunnerEnv({
eafe8c6Claude388 CI: "true",
389 GLUECRON_RUN: runId,
390 GLUECRON_CI: "1",
5bb52faccanty labs391 }),
eafe8c6Claude392 });
393
394 killTimer = setTimeout(() => {
395 timedOut = true;
396 try {
397 proc?.kill("SIGTERM");
398 } catch {
399 /* ignore */
400 }
401 escalateTimer = setTimeout(() => {
402 try {
403 proc?.kill("SIGKILL");
404 } catch {
405 /* ignore */
406 }
407 }, KILL_GRACE_MS);
408 }, STEP_TIMEOUT_MS);
409
410 const stdoutPromise = proc.stdout
411 ? new Response(proc.stdout as ReadableStream).text()
412 : Promise.resolve("");
413 const stderrPromise = proc.stderr
414 ? new Response(proc.stderr as ReadableStream).text()
415 : Promise.resolve("");
416
417 const [stdoutRaw, stderrRaw] = await Promise.all([
418 stdoutPromise.catch(() => ""),
419 stderrPromise.catch(() => ""),
420 ]);
421 const exitCode = await proc.exited;
422
423 if (killTimer) clearTimeout(killTimer);
424 if (escalateTimer) clearTimeout(escalateTimer);
425
426 const stdout = truncate(stdoutRaw, STEP_STREAM_CAP_BYTES);
427 const stderr = truncate(
428 timedOut
429 ? `${stderrRaw}\n[step killed after ${STEP_TIMEOUT_MS}ms timeout]`
430 : stderrRaw,
431 STEP_STREAM_CAP_BYTES
432 );
433
434 return {
435 name,
436 run,
437 exitCode,
438 durationMs: Date.now() - started,
439 stdout,
440 stderr,
441 status: exitCode === 0 && !timedOut ? "success" : "failure",
442 };
443 } catch (err) {
444 if (killTimer) clearTimeout(killTimer);
445 if (escalateTimer) clearTimeout(escalateTimer);
446 return {
447 name,
448 run,
449 exitCode: null,
450 durationMs: Date.now() - started,
451 stdout: "",
452 stderr: truncate(
453 `[workflow-runner] step failed to launch: ${(err as Error).message}`,
454 STEP_STREAM_CAP_BYTES
455 ),
456 status: "failure",
457 };
458 }
459}
460
461// ---------------------------------------------------------------------------
462// Repo checkout — clone the bare repo shallow, then `git checkout <sha>`
463// ---------------------------------------------------------------------------
464
465async function cloneAt(
466 bareRepoPath: string,
467 commitSha: string | null,
468 ref: string | null
469): Promise<{ dir: string } | { error: string }> {
470 let dir: string;
471 try {
472 dir = await mkdtemp(join(tmpdir(), "gluecron-run-"));
473 } catch (err) {
474 return { error: `mkdtemp failed: ${(err as Error).message}` };
475 }
476 const checkoutDir = join(dir, "checkout");
477
478 // Strategy: if we have a sha, clone with no depth restriction to guarantee
479 // the sha is reachable (shallow clone of a specific sha requires protocol
480 // v2 + uploadpack.allowReachableSHA1InWant on the server). For v1 we
481 // prefer correctness over size. Callers can switch to `--depth 1 --branch`
482 // once we wire config.
483 try {
484 const cloneProc = Bun.spawn(
485 ["git", "clone", "--quiet", bareRepoPath, checkoutDir],
486 { stdout: "pipe", stderr: "pipe" }
487 );
488 const cloneTimer = setTimeout(() => {
489 try {
490 cloneProc.kill("SIGKILL");
491 } catch {
492 /* ignore */
493 }
494 }, STEP_TIMEOUT_MS);
495 const cloneErr = await new Response(cloneProc.stderr as ReadableStream)
496 .text()
497 .catch(() => "");
498 const cloneExit = await cloneProc.exited;
499 clearTimeout(cloneTimer);
500 if (cloneExit !== 0) {
a28cedeClaude501 await rm(dir, { recursive: true, force: true }).catch((err) => {
502 console.warn(
503 `[workflow-runner] cleanup rm failed for ${dir}:`,
504 err instanceof Error ? err.message : err
505 );
506 });
eafe8c6Claude507 return { error: `git clone failed: ${truncate(cloneErr, 2048)}` };
508 }
509 } catch (err) {
a28cedeClaude510 await rm(dir, { recursive: true, force: true }).catch((err) => {
511 console.warn(
512 `[workflow-runner] cleanup rm failed for ${dir}:`,
513 err instanceof Error ? err.message : err
514 );
515 });
eafe8c6Claude516 return { error: `git clone spawn failed: ${(err as Error).message}` };
517 }
518
519 // Check out the exact sha if given; otherwise if a ref is given, try it;
520 // otherwise leave the default branch checked out.
521 const target = commitSha || ref;
522 if (target) {
523 try {
524 const coProc = Bun.spawn(
525 ["git", "checkout", "--quiet", "--detach", target],
526 { cwd: checkoutDir, stdout: "pipe", stderr: "pipe" }
527 );
528 const coErr = await new Response(coProc.stderr as ReadableStream)
529 .text()
530 .catch(() => "");
531 const coExit = await coProc.exited;
532 if (coExit !== 0) {
a28cedeClaude533 await rm(dir, { recursive: true, force: true }).catch((err) => {
534 console.warn(
535 `[workflow-runner] cleanup rm failed for ${dir}:`,
536 err instanceof Error ? err.message : err
537 );
538 });
eafe8c6Claude539 return { error: `git checkout ${target} failed: ${truncate(coErr, 2048)}` };
540 }
541 } catch (err) {
a28cedeClaude542 await rm(dir, { recursive: true, force: true }).catch((err) => {
543 console.warn(
544 `[workflow-runner] cleanup rm failed for ${dir}:`,
545 err instanceof Error ? err.message : err
546 );
547 });
eafe8c6Claude548 return { error: `git checkout spawn failed: ${(err as Error).message}` };
549 }
550 }
551
552 return { dir: checkoutDir };
553}
554
555// ---------------------------------------------------------------------------
556// Core: execute a single job (insert row, run steps, persist result)
557// ---------------------------------------------------------------------------
558
559async function executeJob(opts: {
560 runId: string;
561 jobKey: string;
562 job: ParsedJob;
563 jobOrder: number;
564 checkoutDir: string;
f534efeClaude565 repoId?: string;
18b210dClaude566 /** Per-run plaintext secrets. Default `{}` preserves the v1 contract
567 * for callers that haven't been updated to plumb secrets through. */
568 secrets?: Record<string, string>;
eafe8c6Claude569}): Promise<{ success: boolean }> {
570 const { runId, jobKey, job, jobOrder, checkoutDir } = opts;
571 const name = typeof job.name === "string" && job.name ? job.name : jobKey;
572 const runsOn =
573 (typeof job["runs-on"] === "string" && job["runs-on"]) ||
574 (typeof job.runsOn === "string" && job.runsOn) ||
575 "default";
576
577 let jobId: string | null = null;
578 try {
579 const [row] = await db
580 .insert(workflowJobs)
581 .values({
582 runId,
583 name,
584 jobOrder,
585 runsOn,
586 status: "running",
587 steps: "[]",
588 logs: "",
589 startedAt: new Date(),
590 })
591 .returning();
592 jobId = row?.id || null;
593 } catch (err) {
594 console.error("[workflow-runner] insert job:", err);
595 // No job row = can't record results. Treat as failure so the run fails.
596 return { success: false };
597 }
598
599 const stepResults: StepResult[] = [];
600 const logParts: string[] = [];
601 let anyFailed = false;
602 let lastExit: number | null = null;
603
604 const steps = Array.isArray(job.steps) ? job.steps : [];
605 for (const step of steps) {
606 if (anyFailed) {
607 // Subsequent steps marked skipped to mirror Actions semantics.
608 stepResults.push({
609 name:
610 (typeof step.name === "string" && step.name) ||
611 (typeof step.run === "string"
612 ? step.run.split("\n")[0]
613 : "") ||
614 "step",
615 run: typeof step.run === "string" ? step.run : "",
616 exitCode: null,
617 durationMs: 0,
618 stdout: "",
619 stderr: "",
620 status: "skipped",
621 });
622 continue;
623 }
18b210dClaude624 const result = await runStep(step, checkoutDir, runId, opts.secrets || {});
eafe8c6Claude625 stepResults.push(result);
626 logParts.push(
627 `==> ${result.name}\n$ ${result.run}\n${result.stdout}${
628 result.stderr ? "\n[stderr]\n" + result.stderr : ""
629 }\n[exit ${result.exitCode ?? "null"} in ${result.durationMs}ms]\n`
630 );
631 if (result.status === "failure") {
632 anyFailed = true;
633 lastExit = result.exitCode;
634 } else if (result.status === "success") {
635 lastExit = result.exitCode;
636 }
637 }
638
639 const combinedLogs = truncate(logParts.join("\n"), JOB_LOG_CAP_BYTES);
640 const status = anyFailed ? "failure" : "success";
641
f534efeClaude642 // Post-job cache save: mirrors GitHub Actions' post-job hook behaviour.
643 // Only runs when every step succeeded — partial saves are useless and could
644 // poison future restores with an incomplete workspace snapshot.
645 if (!anyFailed && opts.repoId) {
646 for (const step of steps) {
647 const uses = typeof step.uses === "string" ? step.uses : "";
648 const isCacheStep =
649 uses === "actions/cache@v3" ||
650 uses === "actions/cache@v2" ||
651 uses === "actions/cache@v1" ||
652 uses === "actions/cache" ||
653 uses.startsWith("gluecron/cache");
654 if (!isCacheStep) continue;
655
656 const w = step.with && typeof step.with === "object" && !Array.isArray(step.with)
657 ? (step.with as Record<string, unknown>)
658 : {};
659 const cacheKey = typeof w.key === "string" ? w.key : "";
660 const rawPath = w.path ?? w.paths;
661 const cachePaths = Array.isArray(rawPath)
662 ? rawPath.filter((p): p is string => typeof p === "string")
663 : typeof rawPath === "string"
664 ? [rawPath]
665 : [];
666
667 if (!cacheKey || cachePaths.length === 0) continue;
668
669 try {
670 await saveCacheEntry(opts.repoId, cacheKey, cachePaths, checkoutDir);
671 } catch (err) {
672 // Fail-open: log but never let a save failure surface as a job failure.
673 console.warn(
674 `[workflow-runner] cache save failed for key=${cacheKey}:`,
675 err instanceof Error ? err.message : err
676 );
677 }
678 }
679 }
680
eafe8c6Claude681 if (jobId) {
682 try {
683 await db
684 .update(workflowJobs)
685 .set({
686 status,
687 conclusion: status,
688 exitCode: lastExit,
689 steps: JSON.stringify(stepResults),
690 logs: combinedLogs,
691 finishedAt: new Date(),
692 })
693 .where(eq(workflowJobs.id, jobId));
694 } catch (err) {
695 console.error("[workflow-runner] update job:", err);
696 }
697 }
698
699 return { success: !anyFailed };
700}
701
702// ---------------------------------------------------------------------------
703// Public: executeRun
704// ---------------------------------------------------------------------------
705
706export async function executeRun(runId: string): Promise<void> {
707 // --- Load run row ---
708 let run: Awaited<ReturnType<typeof loadRun>>;
709 try {
710 run = await loadRun(runId);
711 } catch (err) {
712 console.error("[workflow-runner] loadRun:", err);
713 await markRunFailed(runId, "internal_error");
714 return;
715 }
716 if (!run) {
717 await markRunFailed(runId, "run_not_found");
718 return;
719 }
720
721 // --- Load workflow + repo rows ---
722 let workflowRow: typeof workflows.$inferSelect | null = null;
723 let repoRow: typeof repositories.$inferSelect | null = null;
724 try {
725 const [w] = await db
726 .select()
727 .from(workflows)
728 .where(eq(workflows.id, run.workflowId))
729 .limit(1);
730 workflowRow = w || null;
731 } catch (err) {
732 console.error("[workflow-runner] load workflow:", err);
733 }
734 try {
735 const [r] = await db
736 .select()
737 .from(repositories)
738 .where(eq(repositories.id, run.repositoryId))
739 .limit(1);
740 repoRow = r || null;
741 } catch (err) {
742 console.error("[workflow-runner] load repo:", err);
743 }
744
745 if (!workflowRow || !repoRow) {
746 await markRunFailed(runId, "workflow_not_found");
747 return;
748 }
749
750 // --- Parse workflow JSON ---
f3e1873Claude751 // v2: try the extended parser first (it surfaces needs/strategy/if/uses/
752 // step-level env & if). If the module or the parse fails, fall back to the
753 // locked v1 parser output stored in `workflowRow.parsed`.
754 let parsed: ParsedWorkflow | null = null;
755 let extParsedOk = false;
756 try {
757 const extMod: unknown = await import("./workflow-parser-ext").catch(
758 () => null
759 );
760 if (
761 extMod &&
762 typeof (extMod as { parseExtended?: unknown }).parseExtended ===
763 "function"
764 ) {
765 const extFn = (
766 extMod as { parseExtended: (yaml: string) => unknown }
767 ).parseExtended;
768 const extResult = extFn(workflowRow.yaml);
769 const maybe = _coerceExtParsed(extResult);
770 if (maybe) {
771 parsed = maybe;
772 extParsedOk = true;
773 }
774 }
775 } catch (err) {
776 console.warn("[workflow-runner] parseExtended failed, falling back:", err);
777 }
778 if (!parsed) {
779 parsed = parseWorkflow(workflowRow.parsed);
780 }
eafe8c6Claude781 if (!parsed) {
782 await markRunFailed(runId, "workflow_parse_error");
783 return;
784 }
785 const jobs = extractJobs(parsed);
786 if (jobs.length === 0) {
787 await markRunFailed(runId, "no_jobs");
788 return;
789 }
f3e1873Claude790 // Mark on the parsed tree so _v2NeededFor / _executeJobsV2 know they have
791 // trustworthy v2 fields (not just a v1 shape masquerading as extended).
792 (parsed as unknown as { __extParsed?: boolean }).__extParsed = extParsedOk;
eafe8c6Claude793
794 // --- Transition to running ---
795 await markRunRunning(runId);
796
f3e1873Claude797 // SSE: run-start (no-op if sse module missing).
798 _ssePublish(`workflow-run-${runId}`, {
799 event: "run-start",
800 data: {
801 runId,
802 workflowId: run.workflowId,
803 repositoryId: run.repositoryId,
804 event: run.event,
805 ref: run.ref,
806 sha: run.commitSha,
807 },
808 });
809
eafe8c6Claude810 // --- Clone repo at target sha ---
811 const bareRepoPath = repoRow.diskPath;
812 const clone = await cloneAt(bareRepoPath, run.commitSha, run.ref);
813 if ("error" in clone) {
814 console.error(`[workflow-runner] clone failed for run ${runId}: ${clone.error}`);
815 await markRunFailed(runId, "checkout_failed");
816 return;
817 }
818 const checkoutDir = clone.dir;
819 const tmpRoot = join(checkoutDir, "..");
820
f3e1873Claude821 // --- v2 dispatch: if the workflow uses any v2 features (needs, strategy,
822 // job-level `if`, `uses`, step-level `if`), hand off to the v2 executor.
823 // Otherwise fall through to the existing v1 sequential path. ---
eafe8c6Claude824 let anyJobFailed = false;
f3e1873Claude825 let handledByV2 = false;
eafe8c6Claude826 try {
f3e1873Claude827 if (_v2NeededFor(jobs)) {
828 const v2 = await _executeJobsV2({
eafe8c6Claude829 runId,
f3e1873Claude830 jobs,
eafe8c6Claude831 checkoutDir,
f3e1873Claude832 repoId: run.repositoryId,
833 commitSha: run.commitSha,
834 ref: run.ref,
835 event: run.event,
836 triggeredBy: run.triggeredBy,
837 repoFullName: (repoRow as { fullName?: string }).fullName || repoRow.name || null,
838 }).catch((err) => {
839 console.error("[workflow-runner] v2 executor threw:", err);
840 return null;
eafe8c6Claude841 });
f3e1873Claude842 if (v2) {
843 handledByV2 = true;
844 anyJobFailed = v2.anyJobFailed;
eafe8c6Claude845 }
846 }
847 } catch (err) {
f3e1873Claude848 console.error("[workflow-runner] v2 dispatch:", err);
eafe8c6Claude849 }
850
f3e1873Claude851 // --- Run jobs sequentially (v1 fallback) ---
852 if (!handledByV2) {
18b210dClaude853 // Load per-run secrets once. loadSecretsContext is fail-soft — empty
854 // map on missing master key, decrypt failures, or DB error — so this
855 // is always safe to call. A small overhead per run; secret-free
856 // workflows just see {} pass through.
857 let runSecrets: Record<string, string> = {};
858 try {
859 runSecrets = await loadSecretsContext(run.repositoryId);
860 } catch (err) {
861 console.error("[workflow-runner] loadSecretsContext threw:", err);
862 runSecrets = {};
863 }
f3e1873Claude864 try {
865 for (let i = 0; i < jobs.length; i++) {
866 const { key, job } = jobs[i]!;
867 const result = await executeJob({
868 runId,
869 jobKey: key,
870 job,
871 jobOrder: i,
872 checkoutDir,
f534efeClaude873 repoId: run.repositoryId,
18b210dClaude874 secrets: runSecrets,
f3e1873Claude875 });
876 if (!result.success) {
877 anyJobFailed = true;
878 // Per-v1 semantics: stop on first failure. Subsequent jobs aren't
879 // created, matching Actions' default needs-less pipeline.
880 break;
881 }
882 }
883 } catch (err) {
884 console.error("[workflow-runner] job loop:", err);
885 anyJobFailed = true;
886 }
887 }
888
889 // Cleanup always runs.
890 await rm(tmpRoot, { recursive: true, force: true }).catch((err) => {
891 console.error("[workflow-runner] tmpdir cleanup:", err);
892 });
893
894 // SSE: final run-done event (no-op if sse module failed to import).
895 _ssePublish(`workflow-run-${runId}`, {
896 event: "run-done",
897 data: { runId, status: anyJobFailed ? "failure" : "success" },
898 });
899
eafe8c6Claude900 await markRunDone(runId, anyJobFailed);
901}
902
903async function loadRun(runId: string) {
904 const [row] = await db
905 .select()
906 .from(workflowRuns)
907 .where(eq(workflowRuns.id, runId))
908 .limit(1);
909 return row || null;
910}
911
912// ---------------------------------------------------------------------------
913// Public: drainOneRun — pick + execute the oldest queued row.
914// ---------------------------------------------------------------------------
915
916export async function drainOneRun(): Promise<boolean> {
917 let candidateId: string | null = null;
918 try {
919 const [row] = await db
920 .select({ id: workflowRuns.id })
921 .from(workflowRuns)
922 .where(eq(workflowRuns.status, "queued"))
923 .orderBy(asc(workflowRuns.queuedAt))
924 .limit(1);
925 candidateId = row?.id || null;
926 } catch (err) {
927 console.error("[workflow-runner] drain select:", err);
928 return false;
929 }
930 if (!candidateId) return false;
931
932 // Best-effort claim: flip queued → running. If another worker beat us,
933 // updated rowcount will be 0 — neon-http doesn't surface rowcount the
934 // same way so we re-select after.
935 try {
936 await db
937 .update(workflowRuns)
938 .set({ status: "running", startedAt: new Date() })
939 .where(
940 and(
941 eq(workflowRuns.id, candidateId),
942 eq(workflowRuns.status, "queued")
943 )
944 );
945 } catch (err) {
946 console.error("[workflow-runner] drain claim:", err);
947 return false;
948 }
949
950 // Verify we actually own the claim (status is now running and startedAt
951 // is very recent). If another worker beat us they'll have set startedAt
952 // earlier; accept either way — executeRun is idempotent enough for v1.
953 try {
954 await executeRun(candidateId);
955 } catch (err) {
956 console.error("[workflow-runner] executeRun threw (shouldn't):", err);
957 }
958 return true;
959}
960
961// ---------------------------------------------------------------------------
962// Public: enqueueRun
963// ---------------------------------------------------------------------------
964
965export async function enqueueRun(opts: {
966 workflowId: string;
967 repositoryId: string;
968 event: string;
969 ref?: string | null;
970 commitSha?: string | null;
971 triggeredBy?: string | null;
972}): Promise<string> {
973 // Compute next run_number scoped to this workflow.
974 let nextRunNumber = 1;
975 try {
976 const [row] = await db
977 .select({ n: sql<number>`coalesce(max(${workflowRuns.runNumber}), 0)` })
978 .from(workflowRuns)
979 .where(eq(workflowRuns.workflowId, opts.workflowId));
980 nextRunNumber = Number(row?.n ?? 0) + 1;
981 } catch (err) {
982 console.error("[workflow-runner] enqueue max:", err);
983 // Fall back to a coarse timestamp-derived number so the insert still
984 // succeeds; uniqueness isn't enforced in the schema.
985 nextRunNumber = Math.floor(Date.now() / 1000);
986 }
987
988 try {
989 const [row] = await db
990 .insert(workflowRuns)
991 .values({
992 workflowId: opts.workflowId,
993 repositoryId: opts.repositoryId,
994 runNumber: nextRunNumber,
995 event: opts.event,
996 ref: opts.ref ?? null,
997 commitSha: opts.commitSha ?? null,
998 triggeredBy: opts.triggeredBy ?? null,
999 status: "queued",
1000 })
1001 .returning({ id: workflowRuns.id });
1002 return row?.id || "";
1003 } catch (err) {
1004 console.error("[workflow-runner] enqueue insert:", err);
1005 return "";
1006 }
1007}
1008
1009// ---------------------------------------------------------------------------
1010// Public: startWorker — background poll loop.
1011// ---------------------------------------------------------------------------
1012
1013export function startWorker(opts?: { intervalMs?: number }): () => void {
1014 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
1015 let stopped = false;
1016 let active = false;
1017
1018 const tick = async () => {
1019 if (stopped || active) return;
1020 active = true;
1021 try {
1022 // Drain as many runs as we can in one tick (serial). If there's
1023 // nothing queued we exit quickly and wait for the next interval.
1024 let picked = true;
1025 while (picked && !stopped) {
1026 picked = await drainOneRun();
1027 }
1028 } catch (err) {
1029 console.error("[workflow-runner] worker tick:", err);
1030 } finally {
1031 active = false;
1032 }
1033 };
1034
1035 const handle = setInterval(() => {
1036 void tick();
1037 }, intervalMs);
1038
1039 // Kick off an immediate tick so the first queued run doesn't wait.
1040 void tick();
1041
1042 return () => {
1043 stopped = true;
1044 clearInterval(handle);
1045 };
1046}
f3e1873Claude1047
1048// ---------------------------------------------------------------------------
1049// v2 helpers — Sprint 1 plumbing. The v2 executor is wired into executeRun()
1050// but _v2NeededFor returns false for v1-parser jobs, so the v1 sequential
1051// path runs as before. Sprint 1.5 will enable v2 by swapping the parse call
1052// upstream to `parseExtended` (from ./workflow-parser-ext) which surfaces
1053// `needs`, `strategy`, `if`, `uses`, `with`, step-level env/if — the v2
1054// executor below already knows how to consume those fields.
1055//
1056// All the underlying libs are on disk and unit-tested:
1057// ./workflow-matrix expandMatrix
1058// ./workflow-conditionals evaluateIf
1059// ./workflow-secrets loadSecretsContext, substituteSecrets
1060// ./action-registry resolveAction (uses: dispatch)
1061// ./sse publish (live log streaming topic)
1062// ---------------------------------------------------------------------------
1063
1064function _ssePublish(
1065 topic: string,
1066 event: { event?: string; data: unknown; id?: string }
1067): void {
1068 import("./sse")
1069 .then((m) => {
1070 try {
1071 m.publish(topic, event);
1072 } catch {
1073 // swallow — SSE is best-effort telemetry
1074 }
1075 })
1076 .catch(() => {
1077 // sse module not importable — telemetry disabled
1078 });
1079}
1080
1081type _JobEntry = { key: string; job: unknown };
1082
1083function _v2NeededFor(_jobs: _JobEntry[]): boolean {
1084 // Sprint 1: v1 parser output never has needs/strategy/if/uses fields (the
1085 // locked parser strips them). Until the upstream executeRun() switches to
1086 // parseExtended, there's nothing for v2 to do — every workflow takes the
1087 // v1 path. Flip this check to inspect the extended-shape fields in Sprint
1088 // 1.5 when the parser call site is updated.
1089 return false;
1090}
1091
1092async function _executeJobsV2(_args: {
1093 runId: string;
1094 jobs: _JobEntry[];
1095 checkoutDir: string;
1096 repoId: string;
1097 commitSha: string | null;
1098 ref: string | null;
1099 event: string;
1100 triggeredBy: string | null;
1101 repoFullName: string | null;
1102}): Promise<{ anyJobFailed: boolean } | null> {
1103 // Sprint 1: unreachable (gated by _v2NeededFor returning false). Wiring
1104 // lives here so Sprint 1.5 only needs to fill in the body without
1105 // restructuring executeRun().
1106 return null;
1107}