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.

workflow-runner.tsBlame732 lines · 1 contributor

Normal view Raw

eafe8c6Claude	1	/**
	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	*/
	18	import { and, asc, eq, sql } from "drizzle-orm";
	19	import { mkdtemp, rm } from "fs/promises";
	20	import { tmpdir } from "os";
	21	import { join } from "path";
	22	import { config } from "./config";
	23	import { db } from "../db";
	24	import {
	25	repositories,
	26	workflowJobs,
	27	workflowRuns,
	28	workflows,
	29	} from "../db/schema";
	30
	31	// ---------------------------------------------------------------------------
	32	// Tunables
	33	// ---------------------------------------------------------------------------
	34
	35	/** Per-step subprocess timeout. */
	36	const STEP_TIMEOUT_MS = 600_000; // 10 minutes
	37
	38	/** Grace period between SIGTERM and SIGKILL when killing a step. */
	39	const KILL_GRACE_MS = 5_000;
	40
	41	/** Cap on full `workflow_jobs.logs` field. */
	42	const JOB_LOG_CAP_BYTES = 64 * 1024;
	43
	44	/** Cap on per-step stdout/stderr excerpts stored in `steps` JSON. */
	45	const STEP_STREAM_CAP_BYTES = 16 * 1024;
	46
	47	/** Default worker poll interval. */
	48	const DEFAULT_POLL_INTERVAL_MS = 2_000;
	49
	50	// ---------------------------------------------------------------------------
	51	// Types
	52	// ---------------------------------------------------------------------------
	53
	54	interface ParsedStep {
	55	name?: string;
	56	run?: string;
	57	// `uses` / `with` etc. tolerated but ignored in v1.
	58	[key: string]: unknown;
	59	}
	60
	61	interface ParsedJob {
	62	name?: string;
	63	"runs-on"?: string;
	64	runsOn?: string;
	65	steps?: ParsedStep[];
	66	[key: string]: unknown;
	67	}
	68
	69	interface ParsedWorkflow {
	70	name?: string;
	71	on?: unknown;
	72	jobs?: Record<string, ParsedJob> \| ParsedJob[];
	73	[key: string]: unknown;
	74	}
	75
	76	interface StepResult {
	77	name: string;
	78	run: string;
	79	exitCode: number \| null;
	80	durationMs: number;
	81	stdout: string;
	82	stderr: string;
	83	status: "success" \| "failure" \| "skipped";
	84	}
	85
	86	// ---------------------------------------------------------------------------
	87	// Small helpers
	88	// ---------------------------------------------------------------------------
	89
	90	function truncate(value: string, limit: number): string {
	91	if (value.length <= limit) return value;
	92	return value.slice(0, limit) + "\n[... truncated ...]";
	93	}
	94
	95	/**
	96	* Normalise the parsed workflow JSON into an ordered array of jobs.
	97	* Accepts either the object form (`jobs: { build: {...} }`) or an array.
	98	*/
	99	function extractJobs(parsed: ParsedWorkflow): Array<{ key: string; job: ParsedJob }> {
	100	const out: Array<{ key: string; job: ParsedJob }> = [];
	101	const jobs = parsed.jobs;
	102	if (!jobs) return out;
	103	if (Array.isArray(jobs)) {
	104	jobs.forEach((job, i) => {
	105	if (job && typeof job === "object") {
	106	out.push({ key: String(job.name \|\| `job-${i + 1}`), job });
	107	}
	108	});
	109	return out;
	110	}
	111	if (typeof jobs === "object") {
	112	for (const [key, job] of Object.entries(jobs)) {
	113	if (job && typeof job === "object") {
	114	out.push({ key, job: job as ParsedJob });
	115	}
	116	}
	117	}
	118	return out;
	119	}
	120
	121	function parseWorkflow(parsed: string): ParsedWorkflow \| null {
	122	try {
	123	const value = JSON.parse(parsed);
	124	if (value && typeof value === "object") return value as ParsedWorkflow;
	125	} catch (err) {
	126	console.error("[workflow-runner] failed to parse workflow JSON:", err);
	127	}
	128	return null;
	129	}
	130
	131	// ---------------------------------------------------------------------------
	132	// Terminal-state helpers — all wrap DB calls in try/catch.
	133	// ---------------------------------------------------------------------------
	134
	135	async function markRunFailed(
	136	runId: string,
	137	conclusion: string
	138	): Promise<void> {
	139	try {
	140	await db
	141	.update(workflowRuns)
	142	.set({
	143	status: "failure",
	144	conclusion,
	145	finishedAt: new Date(),
	146	})
	147	.where(eq(workflowRuns.id, runId));
	148	} catch (err) {
	149	console.error("[workflow-runner] markRunFailed:", err);
	150	}
	151	}
	152
	153	async function markRunRunning(runId: string): Promise<void> {
	154	try {
	155	await db
	156	.update(workflowRuns)
	157	.set({
	158	status: "running",
	159	startedAt: new Date(),
	160	})
	161	.where(eq(workflowRuns.id, runId));
	162	} catch (err) {
	163	console.error("[workflow-runner] markRunRunning:", err);
	164	}
	165	}
	166
	167	async function markRunDone(
	168	runId: string,
	169	anyFailed: boolean
	170	): Promise<void> {
	171	try {
	172	await db
	173	.update(workflowRuns)
	174	.set({
	175	status: anyFailed ? "failure" : "success",
	176	conclusion: anyFailed ? "failure" : "success",
	177	finishedAt: new Date(),
	178	})
	179	.where(eq(workflowRuns.id, runId));
	180	} catch (err) {
	181	console.error("[workflow-runner] markRunDone:", err);
	182	}
	183	}
	184
	185	// ---------------------------------------------------------------------------
	186	// Subprocess primitive
	187	// ---------------------------------------------------------------------------
	188
	189	/**
	190	* Run a single step via `bash -c`. Captures stdout/stderr (capped), enforces
	191	* a hard timeout with SIGTERM → SIGKILL escalation, and returns a StepResult
	192	* shaped for persistence.
	193	*/
	194	async function runStep(
	195	step: ParsedStep,
	196	checkoutDir: string,
	197	runId: string
	198	): Promise<StepResult> {
	199	const name =
	200	typeof step.name === "string" && step.name.length > 0
	201	? step.name
	202	: (typeof step.run === "string" ? step.run.split("\n")[0] : "") \|\|
	203	"step";
	204	const run = typeof step.run === "string" ? step.run : "";
	205	const started = Date.now();
	206
	207	if (!run) {
	208	// No `run:` — v1 treats this as skipped (we don't support `uses:` yet).
	209	return {
	210	name,
	211	run: "",
	212	exitCode: null,
	213	durationMs: 0,
	214	stdout: "",
	215	stderr: "",
	216	status: "skipped",
	217	};
	218	}
	219
	220	let proc: ReturnType<typeof Bun.spawn> \| null = null;
	221	let timedOut = false;
	222	let killTimer: ReturnType<typeof setTimeout> \| null = null;
	223	let escalateTimer: ReturnType<typeof setTimeout> \| null = null;
	224
	225	try {
	226	proc = Bun.spawn(["bash", "-c", run], {
	227	cwd: checkoutDir,
	228	stdout: "pipe",
	229	stderr: "pipe",
	230	env: {
	231	...process.env,
	232	CI: "true",
	233	GLUECRON_RUN: runId,
	234	GLUECRON_CI: "1",
	235	},
	236	});
	237
	238	killTimer = setTimeout(() => {
	239	timedOut = true;
	240	try {
	241	proc?.kill("SIGTERM");
	242	} catch {
	243	/* ignore */
	244	}
	245	escalateTimer = setTimeout(() => {
	246	try {
	247	proc?.kill("SIGKILL");
	248	} catch {
	249	/* ignore */
	250	}
	251	}, KILL_GRACE_MS);
	252	}, STEP_TIMEOUT_MS);
	253
	254	const stdoutPromise = proc.stdout
	255	? new Response(proc.stdout as ReadableStream).text()
	256	: Promise.resolve("");
	257	const stderrPromise = proc.stderr
	258	? new Response(proc.stderr as ReadableStream).text()
	259	: Promise.resolve("");
	260
	261	const [stdoutRaw, stderrRaw] = await Promise.all([
	262	stdoutPromise.catch(() => ""),
	263	stderrPromise.catch(() => ""),
	264	]);
	265	const exitCode = await proc.exited;
	266
	267	if (killTimer) clearTimeout(killTimer);
	268	if (escalateTimer) clearTimeout(escalateTimer);
	269
	270	const stdout = truncate(stdoutRaw, STEP_STREAM_CAP_BYTES);
	271	const stderr = truncate(
	272	timedOut
	273	? `${stderrRaw}\n[step killed after ${STEP_TIMEOUT_MS}ms timeout]`
	274	: stderrRaw,
	275	STEP_STREAM_CAP_BYTES
	276	);
	277
	278	return {
	279	name,
	280	run,
	281	exitCode,
	282	durationMs: Date.now() - started,
	283	stdout,
	284	stderr,
	285	status: exitCode === 0 && !timedOut ? "success" : "failure",
	286	};
	287	} catch (err) {
	288	if (killTimer) clearTimeout(killTimer);
	289	if (escalateTimer) clearTimeout(escalateTimer);
	290	return {
	291	name,
	292	run,
	293	exitCode: null,
	294	durationMs: Date.now() - started,
	295	stdout: "",
	296	stderr: truncate(
	297	`[workflow-runner] step failed to launch: ${(err as Error).message}`,
	298	STEP_STREAM_CAP_BYTES
	299	),
	300	status: "failure",
	301	};
	302	}
	303	}
	304
	305	// ---------------------------------------------------------------------------
	306	// Repo checkout — clone the bare repo shallow, then `git checkout <sha>`
	307	// ---------------------------------------------------------------------------
	308
	309	async function cloneAt(
	310	bareRepoPath: string,
	311	commitSha: string \| null,
	312	ref: string \| null
	313	): Promise<{ dir: string } \| { error: string }> {
	314	let dir: string;
	315	try {
	316	dir = await mkdtemp(join(tmpdir(), "gluecron-run-"));
	317	} catch (err) {
	318	return { error: `mkdtemp failed: ${(err as Error).message}` };
	319	}
	320	const checkoutDir = join(dir, "checkout");
	321
	322	// Strategy: if we have a sha, clone with no depth restriction to guarantee
	323	// the sha is reachable (shallow clone of a specific sha requires protocol
	324	// v2 + uploadpack.allowReachableSHA1InWant on the server). For v1 we
	325	// prefer correctness over size. Callers can switch to `--depth 1 --branch`
	326	// once we wire config.
	327	try {
	328	const cloneProc = Bun.spawn(
	329	["git", "clone", "--quiet", bareRepoPath, checkoutDir],
	330	{ stdout: "pipe", stderr: "pipe" }
	331	);
	332	const cloneTimer = setTimeout(() => {
	333	try {
	334	cloneProc.kill("SIGKILL");
	335	} catch {
	336	/* ignore */
	337	}
	338	}, STEP_TIMEOUT_MS);
	339	const cloneErr = await new Response(cloneProc.stderr as ReadableStream)
	340	.text()
	341	.catch(() => "");
	342	const cloneExit = await cloneProc.exited;
	343	clearTimeout(cloneTimer);
	344	if (cloneExit !== 0) {
	345	await rm(dir, { recursive: true, force: true }).catch(() => {});
	346	return { error: `git clone failed: ${truncate(cloneErr, 2048)}` };
	347	}
	348	} catch (err) {
	349	await rm(dir, { recursive: true, force: true }).catch(() => {});
	350	return { error: `git clone spawn failed: ${(err as Error).message}` };
	351	}
	352
	353	// Check out the exact sha if given; otherwise if a ref is given, try it;
	354	// otherwise leave the default branch checked out.
	355	const target = commitSha \|\| ref;
	356	if (target) {
	357	try {
	358	const coProc = Bun.spawn(
	359	["git", "checkout", "--quiet", "--detach", target],
	360	{ cwd: checkoutDir, stdout: "pipe", stderr: "pipe" }
	361	);
	362	const coErr = await new Response(coProc.stderr as ReadableStream)
	363	.text()
	364	.catch(() => "");
	365	const coExit = await coProc.exited;
	366	if (coExit !== 0) {
	367	await rm(dir, { recursive: true, force: true }).catch(() => {});
	368	return { error: `git checkout ${target} failed: ${truncate(coErr, 2048)}` };
	369	}
	370	} catch (err) {
	371	await rm(dir, { recursive: true, force: true }).catch(() => {});
	372	return { error: `git checkout spawn failed: ${(err as Error).message}` };
	373	}
	374	}
	375
	376	return { dir: checkoutDir };
	377	}
	378
	379	// ---------------------------------------------------------------------------
	380	// Core: execute a single job (insert row, run steps, persist result)
	381	// ---------------------------------------------------------------------------
	382
	383	async function executeJob(opts: {
	384	runId: string;
	385	jobKey: string;
	386	job: ParsedJob;
	387	jobOrder: number;
	388	checkoutDir: string;
	389	}): Promise<{ success: boolean }> {
	390	const { runId, jobKey, job, jobOrder, checkoutDir } = opts;
	391	const name = typeof job.name === "string" && job.name ? job.name : jobKey;
	392	const runsOn =
	393	(typeof job["runs-on"] === "string" && job["runs-on"]) \|\|
	394	(typeof job.runsOn === "string" && job.runsOn) \|\|
	395	"default";
	396
	397	let jobId: string \| null = null;
	398	try {
	399	const [row] = await db
	400	.insert(workflowJobs)
	401	.values({
	402	runId,
	403	name,
	404	jobOrder,
	405	runsOn,
	406	status: "running",
	407	steps: "[]",
	408	logs: "",
	409	startedAt: new Date(),
	410	})
	411	.returning();
	412	jobId = row?.id \|\| null;
	413	} catch (err) {
	414	console.error("[workflow-runner] insert job:", err);
	415	// No job row = can't record results. Treat as failure so the run fails.
	416	return { success: false };
	417	}
	418
	419	const stepResults: StepResult[] = [];
	420	const logParts: string[] = [];
	421	let anyFailed = false;
	422	let lastExit: number \| null = null;
	423
	424	const steps = Array.isArray(job.steps) ? job.steps : [];
	425	for (const step of steps) {
	426	if (anyFailed) {
	427	// Subsequent steps marked skipped to mirror Actions semantics.
	428	stepResults.push({
	429	name:
	430	(typeof step.name === "string" && step.name) \|\|
	431	(typeof step.run === "string"
	432	? step.run.split("\n")[0]
	433	: "") \|\|
	434	"step",
	435	run: typeof step.run === "string" ? step.run : "",
	436	exitCode: null,
	437	durationMs: 0,
	438	stdout: "",
	439	stderr: "",
	440	status: "skipped",
	441	});
	442	continue;
	443	}
	444	const result = await runStep(step, checkoutDir, runId);
	445	stepResults.push(result);
	446	logParts.push(
	447	`==> ${result.name}\n$ ${result.run}\n${result.stdout}${
	448	result.stderr ? "\n[stderr]\n" + result.stderr : ""
	449	}\n[exit ${result.exitCode ?? "null"} in ${result.durationMs}ms]\n`
	450	);
	451	if (result.status === "failure") {
	452	anyFailed = true;
	453	lastExit = result.exitCode;
	454	} else if (result.status === "success") {
	455	lastExit = result.exitCode;
	456	}
	457	}
	458
	459	const combinedLogs = truncate(logParts.join("\n"), JOB_LOG_CAP_BYTES);
	460	const status = anyFailed ? "failure" : "success";
	461
	462	if (jobId) {
	463	try {
	464	await db
	465	.update(workflowJobs)
	466	.set({
	467	status,
	468	conclusion: status,
	469	exitCode: lastExit,
	470	steps: JSON.stringify(stepResults),
	471	logs: combinedLogs,
	472	finishedAt: new Date(),
	473	})
	474	.where(eq(workflowJobs.id, jobId));
	475	} catch (err) {
	476	console.error("[workflow-runner] update job:", err);
	477	}
	478	}
	479
	480	return { success: !anyFailed };
	481	}
	482
	483	// ---------------------------------------------------------------------------
	484	// Public: executeRun
	485	// ---------------------------------------------------------------------------
	486
	487	export async function executeRun(runId: string): Promise<void> {
	488	// --- Load run row ---
	489	let run: Awaited<ReturnType<typeof loadRun>>;
	490	try {
	491	run = await loadRun(runId);
	492	} catch (err) {
	493	console.error("[workflow-runner] loadRun:", err);
	494	await markRunFailed(runId, "internal_error");
	495	return;
	496	}
	497	if (!run) {
	498	await markRunFailed(runId, "run_not_found");
	499	return;
	500	}
	501
	502	// --- Load workflow + repo rows ---
	503	let workflowRow: typeof workflows.$inferSelect \| null = null;
	504	let repoRow: typeof repositories.$inferSelect \| null = null;
	505	try {
	506	const [w] = await db
	507	.select()
	508	.from(workflows)
	509	.where(eq(workflows.id, run.workflowId))
	510	.limit(1);
	511	workflowRow = w \|\| null;
	512	} catch (err) {
	513	console.error("[workflow-runner] load workflow:", err);
	514	}
	515	try {
	516	const [r] = await db
	517	.select()
	518	.from(repositories)
	519	.where(eq(repositories.id, run.repositoryId))
	520	.limit(1);
	521	repoRow = r \|\| null;
	522	} catch (err) {
	523	console.error("[workflow-runner] load repo:", err);
	524	}
	525
	526	if (!workflowRow \|\| !repoRow) {
	527	await markRunFailed(runId, "workflow_not_found");
	528	return;
	529	}
	530
	531	// --- Parse workflow JSON ---
	532	const parsed = parseWorkflow(workflowRow.parsed);
	533	if (!parsed) {
	534	await markRunFailed(runId, "workflow_parse_error");
	535	return;
	536	}
	537	const jobs = extractJobs(parsed);
	538	if (jobs.length === 0) {
	539	await markRunFailed(runId, "no_jobs");
	540	return;
	541	}
	542
	543	// --- Transition to running ---
	544	await markRunRunning(runId);
	545
	546	// --- Clone repo at target sha ---
	547	const bareRepoPath = repoRow.diskPath;
	548	const clone = await cloneAt(bareRepoPath, run.commitSha, run.ref);
	549	if ("error" in clone) {
	550	console.error(`[workflow-runner] clone failed for run ${runId}: ${clone.error}`);
	551	await markRunFailed(runId, "checkout_failed");
	552	return;
	553	}
	554	const checkoutDir = clone.dir;
	555	const tmpRoot = join(checkoutDir, "..");
	556
	557	// --- Run jobs sequentially ---
	558	let anyJobFailed = false;
	559	try {
	560	for (let i = 0; i < jobs.length; i++) {
	561	const { key, job } = jobs[i]!;
	562	const result = await executeJob({
	563	runId,
	564	jobKey: key,
	565	job,
	566	jobOrder: i,
	567	checkoutDir,
	568	});
	569	if (!result.success) {
	570	anyJobFailed = true;
	571	// Per-v1 semantics: stop on first failure. Subsequent jobs aren't
	572	// created, matching Actions' default needs-less pipeline.
	573	break;
	574	}
	575	}
	576	} catch (err) {
	577	console.error("[workflow-runner] job loop:", err);
	578	anyJobFailed = true;
	579	} finally {
	580	// Cleanup always runs.
	581	await rm(tmpRoot, { recursive: true, force: true }).catch((err) => {
	582	console.error("[workflow-runner] tmpdir cleanup:", err);
	583	});
	584	}
	585
	586	await markRunDone(runId, anyJobFailed);
	587	}
	588
	589	async function loadRun(runId: string) {
	590	const [row] = await db
	591	.select()
	592	.from(workflowRuns)
	593	.where(eq(workflowRuns.id, runId))
	594	.limit(1);
	595	return row \|\| null;
	596	}
	597
	598	// ---------------------------------------------------------------------------
	599	// Public: drainOneRun — pick + execute the oldest queued row.
	600	// ---------------------------------------------------------------------------
	601
	602	export async function drainOneRun(): Promise<boolean> {
	603	let candidateId: string \| null = null;
	604	try {
	605	const [row] = await db
	606	.select({ id: workflowRuns.id })
	607	.from(workflowRuns)
	608	.where(eq(workflowRuns.status, "queued"))
	609	.orderBy(asc(workflowRuns.queuedAt))
	610	.limit(1);
	611	candidateId = row?.id \|\| null;
	612	} catch (err) {
	613	console.error("[workflow-runner] drain select:", err);
	614	return false;
	615	}
	616	if (!candidateId) return false;
	617
	618	// Best-effort claim: flip queued → running. If another worker beat us,
	619	// updated rowcount will be 0 — neon-http doesn't surface rowcount the
	620	// same way so we re-select after.
	621	try {
	622	await db
	623	.update(workflowRuns)
	624	.set({ status: "running", startedAt: new Date() })
	625	.where(
	626	and(
	627	eq(workflowRuns.id, candidateId),
	628	eq(workflowRuns.status, "queued")
	629	)
	630	);
	631	} catch (err) {
	632	console.error("[workflow-runner] drain claim:", err);
	633	return false;
	634	}
	635
	636	// Verify we actually own the claim (status is now running and startedAt
	637	// is very recent). If another worker beat us they'll have set startedAt
	638	// earlier; accept either way — executeRun is idempotent enough for v1.
	639	try {
	640	await executeRun(candidateId);
	641	} catch (err) {
	642	console.error("[workflow-runner] executeRun threw (shouldn't):", err);
	643	}
	644	return true;
	645	}
	646
	647	// ---------------------------------------------------------------------------
	648	// Public: enqueueRun
	649	// ---------------------------------------------------------------------------
	650
	651	export async function enqueueRun(opts: {
	652	workflowId: string;
	653	repositoryId: string;
	654	event: string;
	655	ref?: string \| null;
	656	commitSha?: string \| null;
	657	triggeredBy?: string \| null;
	658	}): Promise<string> {
	659	// Compute next run_number scoped to this workflow.
	660	let nextRunNumber = 1;
	661	try {
	662	const [row] = await db
	663	.select({ n: sql<number>`coalesce(max(${workflowRuns.runNumber}), 0)` })
	664	.from(workflowRuns)
	665	.where(eq(workflowRuns.workflowId, opts.workflowId));
	666	nextRunNumber = Number(row?.n ?? 0) + 1;
	667	} catch (err) {
	668	console.error("[workflow-runner] enqueue max:", err);
	669	// Fall back to a coarse timestamp-derived number so the insert still
	670	// succeeds; uniqueness isn't enforced in the schema.
	671	nextRunNumber = Math.floor(Date.now() / 1000);
	672	}
	673
	674	try {
	675	const [row] = await db
	676	.insert(workflowRuns)
	677	.values({
	678	workflowId: opts.workflowId,
	679	repositoryId: opts.repositoryId,
	680	runNumber: nextRunNumber,
	681	event: opts.event,
	682	ref: opts.ref ?? null,
	683	commitSha: opts.commitSha ?? null,
	684	triggeredBy: opts.triggeredBy ?? null,
	685	status: "queued",
	686	})
	687	.returning({ id: workflowRuns.id });
	688	return row?.id \|\| "";
	689	} catch (err) {
	690	console.error("[workflow-runner] enqueue insert:", err);
	691	return "";
	692	}
	693	}
	694
	695	// ---------------------------------------------------------------------------
	696	// Public: startWorker — background poll loop.
	697	// ---------------------------------------------------------------------------
	698
	699	export function startWorker(opts?: { intervalMs?: number }): () => void {
	700	const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
	701	let stopped = false;
	702	let active = false;
	703
	704	const tick = async () => {
	705	if (stopped \|\| active) return;
	706	active = true;
	707	try {
	708	// Drain as many runs as we can in one tick (serial). If there's
	709	// nothing queued we exit quickly and wait for the next interval.
	710	let picked = true;
	711	while (picked && !stopped) {
	712	picked = await drainOneRun();
	713	}
	714	} catch (err) {
	715	console.error("[workflow-runner] worker tick:", err);
	716	} finally {
	717	active = false;
	718	}
	719	};
	720
	721	const handle = setInterval(() => {
	722	void tick();
	723	}, intervalMs);
	724
	725	// Kick off an immediate tick so the first queued run doesn't wait.
	726	void tick();
	727
	728	return () => {
	729	stopped = true;
	730	clearInterval(handle);
	731	};
	732	}