Phase 2: job processor + BullMQ worker wiring (verified against real bullmq)

The processor orchestration and the BullMQ Worker construction, both with the wiring the
research confirmed live against a real Valkey.

processor.mjs -- the money-safety ORDER as a pure-ish function over injected deps, so it is
testable without GitHub/Docker/Redis. The order IS the contract: mint token -> refuse an
unprotected default branch -> prepare (resolve SHA, clone, materialise .pi/, write prompt) ->
reserve budget -> ONLY THEN run the container. Budget is reserved as late as possible but
strictly before the container, so a refusal from a free earlier gate (unprotected repo, clone
failure) never consumes a daily slot. 9 tests assert the order directly: an unprotected repo
never prepares or spends; over-budget prepares but never runs a container; exit 0 -> return,
2 -> return, 1 -> throw InfraRetry (retryable); cleanup runs on every path; a local-folder
job skips minting and branch-check.

index.mjs -- the Worker, with every load-bearing fact from the wiring research:
- connection.maxRetriesPerRequest: null (REQUIRED, or bullmq throws at construction)
- maxStalledCount: 0 (a stalled paid job FAILS, never silently re-runs -- confirmed live)
- the processor declares EXACTLY 3 params (job, token, signal). bullmq only allocates an
  AbortController when processor.length >= 3, so dropping the unused token would silently
  disable the 30-minute timeout AND shutdown abort. A test asserts arity===3 precisely
  because the failure is silent -- verified passing with bullmq present.
- the 30-min timeout is ours: a setTimeout fires worker.cancelJob(id), which raises the
  AbortSignal; the abort handler docker-stops the container. The same abort path serves
  SIGTERM/SIGINT shutdown -- one mechanism, two triggers. Verified: the timeout fires
  cancelJob, and an abort stops the right container.

While testing I found and fixed a real robustness gap: runContainer must handle an
already-aborted signal (the timeout can fire during a slow prepare), now stated in its
contract. The remaining GitHub-dependent deps -- mintToken, isDefaultBranchProtected,
prepareWorkspace (clone at SHA via git fetch --depth) -- need a GitHub App to verify end to
end rather than guess, and are the injected seams the processor is already tested against.

Author: edgehero

worker/src/index.mjs

@@ -0,0 +1,85 @@
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
+import { UnrecoverableError, Worker } from "bullmq";
+import { InfraRetry, runJob } from "./processor.mjs";
+
+const exec = promisify(execFile);
+
+export const QUEUE = "pi-jobs";
+export const JOB_TIMEOUT_MS = 30 * 60 * 1000; // REQ-JOB-TIMEOUT-30M
+
+/**
+ * Build the BullMQ processor.
+ *
+ * It MUST declare exactly three parameters (job, token, signal). BullMQ only allocates an
+ * AbortController when `processor.length >= 3` (it inspects the function's arity at construction),
+ * so dropping the unused `token` would silently disable BOTH the 30-minute timeout and the shutdown
+ * abort -- with no error. A test asserts the arity precisely because the failure is silent.
+ *
+ * Dependencies are injected so this is testable without a live queue: `cancelJob` (fired by the
+ * timeout), `stopContainer` (fired by the abort), and the orchestration deps.
+ */
+export function makeProcessor({ cancelJob, stopContainer, redis, cap, deps, timeoutMs = JOB_TIMEOUT_MS }) {
+	return async function processor(job, token, signal) {
+		const name = `pi-job-${job.id}`;
+		const timer = setTimeout(() => {
+			// BullMQ has no per-job kill timer; this is ours. cancelJob raises the AbortSignal.
+			Promise.resolve(cancelJob(job.id, "job-timeout-30m")).catch(() => {});
+		}, timeoutMs);
+
+		// Abort (timeout OR shutdown) => stop the container. docker stop sends SIGTERM then SIGKILL
+		// after the grace period; the runner exits and runContainer returns/throws.
+		const onAbort = () => {
+			Promise.resolve(stopContainer(name)).catch(() => {});
+		};
+		signal.addEventListener("abort", onAbort, { once: true });
+
+		try {
+			return await runJob(job.data, {
+				redis,
+				cap,
+				...deps,
+				runContainer: (ctx) => deps.runContainer({ ...ctx, name, signal }),
+			});
+		} catch (error) {
+			if (error instanceof InfraRetry) throw error; // retryable: BullMQ retries per attempts
+			// A non-retryable, non-infra error (our bug) must not retry forever. UnrecoverableError
+			// records it as failed-and-distinct on the dashboard without a retry.
+			throw new UnrecoverableError(error.message);
+		} finally {
+			clearTimeout(timer);
+			signal.removeEventListener("abort", onAbort);
+		}
+	};
+}
+
+export function createWorker({ connection, concurrency, cap, redis, deps, limiter }) {
+	let worker; // referenced by cancelJob before assignment; only called later, so the TDZ is fine
+	const processor = makeProcessor({
+		cancelJob: (id, reason) => worker.cancelJob(id, reason),
+		stopContainer: (name) => exec("docker", ["stop", "-t", "5", name]),
+		redis,
+		cap,
+		deps,
+	});
+
+	worker = new Worker(QUEUE, processor, {
+		// maxRetriesPerRequest: null is REQUIRED for BullMQ's blocking connections, or it throws.
+		connection: { ...connection, maxRetriesPerRequest: null },
+		concurrency,
+		maxStalledCount: 0, // a stalled paid job FAILS, never silently re-runs (verified live)
+		...(limiter ? { limiter } : {}),
+	});
+
+	const shutdown = async () => {
+		// Abort active jobs (=> docker stop via onAbort), then close. Without the cancel,
+		// worker.close() would wait up to 30 minutes for the container.
+		await Promise.resolve(worker.cancelAllJobs?.("shutdown")).catch(() => {});
+		await worker.close();
+		process.exit(0);
+	};
+	process.once("SIGTERM", shutdown);
+	process.once("SIGINT", shutdown);
+
+	return worker;
+}

worker/src/processor.mjs

@@ -0,0 +1,99 @@
+import { reserveBudget } from "./budget.mjs";
+import { EXIT_COMPLETED, EXIT_INFRA, EXIT_POLICY } from "./exit-code.mjs";
+
+/**
+ * The job orchestration. Deliberately a pure-ish function over INJECTED side-effecting deps, so
+ * the money-safety ORDER can be tested without GitHub, Docker, or Redis.
+ *
+ * The order is the contract, and every step before `runContainer` must be free of provider spend:
+ *
+ *   1. mint a scoped token (GitHub jobs)          -- CONST-TOKEN-SCOPED-PER-JOB
+ *   2. REFUSE an unprotected default branch       -- REQ-BRANCH-PROTECTION-PRECONDITION
+ *   3. resolve the default-branch SHA (fresh API), clone at it, materialise .pi/, write the prompt
+ *   4. reserve a budget slot                      -- CONST-BUDGET-BEFORE-TOKENS
+ *   5. ONLY NOW run the container (the only step that spends provider tokens)
+ *   6. map the container exit code to retry-vs-success
+ *
+ * Budget is reserved as late as possible but strictly before the container, so a refusal from an
+ * earlier free gate (unprotected repo, clone failure) never consumes a daily slot. The container
+ * is the only thing that spends money, so "before tokens" means "before this line".
+ *
+ * Returns a result object on a non-retryable outcome; THROWS on a retryable (infra) one so BullMQ
+ * retries per `attempts`. The caller (the BullMQ processor) turns the thrown/returned distinction
+ * into the queue's retry behaviour -- that is INT-RUNNER-EXIT-CODE-PROTOCOL.
+ */
+export async function runJob(job, deps) {
+	const {
+		redis,
+		cap,
+		mintToken, // (repo) => scoped 1h token  | null for local-folder jobs
+		isDefaultBranchProtected, // (repo, token) => boolean
+		prepareWorkspace, // (job, token) => { workspaceDir, jobDir }  (clone+materialise+prompt)
+		// runContainer({ job, token, prepared, name, signal }) => exitCode. It MUST honour `signal`:
+		// stop the container on abort, and reject/exit promptly if `signal.aborted` is already true
+		// at entry (the timeout can fire during a slow prepare). The wiring injects name + signal.
+		runContainer,
+		cleanup, // (dirs) => void
+		comment, // (job, text) => void   (issue status; no-op for local jobs)
+		log = () => {},
+		now = new Date(),
+	} = deps;
+
+	const isGitHub = job.kind === "github";
+	let token = null;
+	let prepared = null;
+	let reserved = false;
+
+	try {
+		if (isGitHub) {
+			token = await mintToken(job.repo);
+
+			// REQ-BRANCH-PROTECTION-PRECONDITION. The agent's token can merge (contents:write covers
+			// push AND merge), so branch protection is the only technical barrier to a self-merge.
+			// Refuse before spending anything.
+			if (!(await isDefaultBranchProtected(job.repo, token))) {
+				await comment(job, "Refused: the default branch is not protected. See SECURITY.md.");
+				log("refused_unprotected", { repo: job.repo });
+				return { outcome: "policy", reason: "unprotected-branch" }; // return => not retried
+			}
+		}
+
+		prepared = await prepareWorkspace(job, token); // resolves SHA, clones, materialises .pi/, writes prompt
+
+		// Budget last-but-before-container. A refusal here spends nothing (no container starts).
+		const budget = await reserveBudget(redis, { cap, now });
+		reserved = true;
+		if (!budget.allowed) {
+			await comment(job, `Over the daily budget cap (${budget.cap}). Not run.`);
+			log("over_budget", { reserved: budget.reserved, cap: budget.cap });
+			return { outcome: "policy", reason: "over-budget" }; // return => not retried
+		}
+
+		const exitCode = await runContainer({ job, token, prepared });
+		log("container_exit", { exitCode });
+
+		switch (exitCode) {
+			case EXIT_COMPLETED:
+				return { outcome: "completed" };
+			case EXIT_POLICY:
+				return { outcome: "policy", reason: "runner-policy" };
+			case EXIT_INFRA:
+				throw new InfraRetry(`infra failure, container exit ${exitCode}`);
+			default:
+				throw new InfraRetry(`unknown container exit ${exitCode}`);
+		}
+	} finally {
+		if (prepared) await cleanup(prepared).catch(() => {});
+	}
+}
+
+/** Thrown for the retryable (infra) class only. The BullMQ processor lets this propagate to retry. */
+export class InfraRetry extends Error {
+	constructor(message) {
+		super(message);
+		this.name = "InfraRetry";
+		this.piDispatchRetry = true;
+	}
+}
+
+export { EXIT_COMPLETED, EXIT_INFRA, EXIT_POLICY };

worker/test/processor.test.mjs

@@ -0,0 +1,88 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import { InfraRetry, runJob } from "../src/processor.mjs";
+
+/** A fake redis whose counter we can preset, to force over/under budget. */
+function fakeRedis(start = 0) {
+	let n = start;
+	return { async incr() { return ++n; }, async decr() { return --n; }, async expire() {} };
+}
+
+/** Deps with call-order tracking, so tests can assert the money-safety ORDER, not just outcomes. */
+function deps(overrides = {}) {
+	const calls = [];
+	const base = {
+		redis: fakeRedis(),
+		cap: 10,
+		mintToken: async (repo) => (calls.push(`mint:${repo}`), "tok"),
+		isDefaultBranchProtected: async () => (calls.push("branch-check"), true),
+		prepareWorkspace: async () => (calls.push("prepare"), { workspaceDir: "/w", jobDir: "/j" }),
+		runContainer: async () => (calls.push("run-container"), 0),
+		cleanup: async () => (calls.push("cleanup"), undefined),
+		comment: async (_j, t) => calls.push(`comment:${t.slice(0, 12)}`),
+		now: new Date("2026-07-16T10:00:00Z"),
+	};
+	return { deps: { ...base, ...overrides }, calls };
+}
+
+const ghJob = { kind: "github", repo: "org/repo", provider: "anthropic", model: "m", maxTurns: 20 };
+
+test("happy path: mint -> branch-check -> prepare -> budget -> container, in that order", async () => {
+	const { deps: d, calls } = deps();
+	const r = await runJob(ghJob, d);
+	assert.equal(r.outcome, "completed");
+	assert.deepEqual(calls, ["mint:org/repo", "branch-check", "prepare", "run-container", "cleanup"]);
+});
+
+test("an unprotected branch refuses BEFORE any container -- and never prepares/spends", async () => {
+	const { deps: d, calls } = deps({ isDefaultBranchProtected: async () => false });
+	const r = await runJob(ghJob, d);
+	assert.equal(r.outcome, "policy");
+	assert.equal(r.reason, "unprotected-branch");
+	assert.ok(!calls.includes("run-container"), "must not spend on an unprotected repo");
+	assert.ok(!calls.includes("prepare"), "must not even clone an unprotected repo");
+});
+
+test("over budget refuses AFTER prepare but BEFORE the container -- no provider spend", async () => {
+	// counter starts at cap, so the reservation lands over-cap.
+	const { deps: d, calls } = deps({ redis: fakeRedis(10), cap: 10 });
+	const r = await runJob(ghJob, d);
+	assert.equal(r.reason, "over-budget");
+	assert.ok(calls.includes("prepare"), "prepared (free work) before the budget gate");
+	assert.ok(!calls.includes("run-container"), "over budget => NO container, no money spent");
+});
+
+test("container exit 0 => success, no retry", async () => {
+	const { deps: d } = deps({ runContainer: async () => 0 });
+	assert.equal((await runJob(ghJob, d)).outcome, "completed");
+});
+
+test("container exit 2 => policy, RETURNS (not retried)", async () => {
+	const { deps: d } = deps({ runContainer: async () => 2 });
+	assert.equal((await runJob(ghJob, d)).outcome, "policy");
+});
+
+test("container exit 1 => THROWS InfraRetry (BullMQ will retry)", async () => {
+	const { deps: d } = deps({ runContainer: async () => 1 });
+	await assert.rejects(() => runJob(ghJob, d), (e) => e instanceof InfraRetry && e.piDispatchRetry === true);
+});
+
+test("an unknown exit code throws (retry-then-visible), never silent success", async () => {
+	const { deps: d } = deps({ runContainer: async () => 137 });
+	await assert.rejects(() => runJob(ghJob, d), InfraRetry);
+});
+
+test("cleanup runs even when the container throws", async () => {
+	const { deps: d, calls } = deps({ runContainer: async () => 1 });
+	await runJob(ghJob, d).catch(() => {});
+	assert.ok(calls.includes("cleanup"), "the job dir must be cleaned up on the infra path too");
+});
+
+test("a local-folder job skips minting and branch-check entirely", async () => {
+	const { deps: d, calls } = deps();
+	const localJob = { kind: "local", folder: "/home/rob/proj", provider: "anthropic", model: "m", maxTurns: 5 };
+	const r = await runJob(localJob, d);
+	assert.equal(r.outcome, "completed");
+	assert.ok(!calls.some((c) => c.startsWith("mint")), "no token for a local job");
+	assert.ok(!calls.includes("branch-check"), "no branch check for a non-git folder");
+});

worker/test/wiring.test.mjs

@@ -0,0 +1,90 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+
+// index.mjs imports bullmq, so this skips below the node floor / without deps and runs in CI,
+// where PI_DISPATCH_REQUIRE_WORKER_TESTS=1 turns a skip into a hard failure.
+let mod;
+let importError;
+try {
+	mod = await import("../src/index.mjs");
+} catch (error) {
+	importError = error;
+}
+if (!mod && process.env.PI_DISPATCH_REQUIRE_WORKER_TESTS === "1") {
+	throw new Error(`worker wiring tests are REQUIRED here but bullmq could not import.\n${importError}`);
+}
+const skip = mod ? false : `bullmq not installed (node ${process.version} < 22.19.0); CI runs these`;
+
+test("the processor declares arity 3 -- the silent trap that would disable the timeout", { skip }, () => {
+	// BullMQ only allocates an AbortController when processor.length >= 3. If a refactor drops the
+	// unused `token` param, the 30-minute timeout and shutdown abort silently stop working. This is
+	// the single most important assertion in the worker's wiring, because nothing at runtime reports
+	// its failure -- the container just runs unbounded.
+	const processor = mod.makeProcessor({
+		cancelJob: () => {},
+		stopContainer: () => {},
+		redis: {},
+		cap: 10,
+		deps: {},
+	});
+	assert.equal(processor.length, 3, "processor must declare (job, token, signal) or the abort dies");
+});
+
+test("the timeout fires cancelJob after timeoutMs", { skip }, async () => {
+	let cancelled = null;
+	const processor = mod.makeProcessor({
+		cancelJob: (id, reason) => (cancelled = { id, reason }),
+		stopContainer: () => {},
+		redis: { async incr() { return 1; }, async expire() {} },
+		cap: 10,
+		timeoutMs: 20,
+		deps: {
+			mintToken: async () => "t",
+			isDefaultBranchProtected: async () => true,
+			prepareWorkspace: async () => ({}),
+			// a real container exits when docker stop runs; mirror that -- reject on abort.
+			runContainer: ({ signal }) =>
+				new Promise((_, reject) => signal.addEventListener("abort", () => reject(new Error("stopped")), { once: true })),
+			cleanup: async () => {},
+			comment: async () => {},
+		},
+	});
+
+	const ac = new AbortController();
+	const job = { id: "j1", data: { kind: "github", repo: "o/r" } };
+	const running = processor(job, "tok", ac.signal).catch(() => {});
+	await new Promise((r) => setTimeout(r, 60));
+	assert.equal(cancelled?.id, "j1");
+	assert.equal(cancelled?.reason, "job-timeout-30m");
+	ac.abort(); // let the hung runContainer's abort path settle
+	await running;
+});
+
+test("an abort stops the container", { skip }, async () => {
+	let stopped = null;
+	const processor = mod.makeProcessor({
+		cancelJob: () => {},
+		stopContainer: (name) => (stopped = name),
+		redis: { async incr() { return 1; }, async expire() {} },
+		cap: 10,
+		timeoutMs: 100000,
+		deps: {
+			mintToken: async () => "t",
+			isDefaultBranchProtected: async () => true,
+			prepareWorkspace: async () => ({}),
+			runContainer: ({ signal }) =>
+				new Promise((_, reject) => signal.addEventListener("abort", () => reject(new Error("stopped")), { once: true })),
+			cleanup: async () => {},
+			comment: async () => {},
+		},
+	});
+	const ac = new AbortController();
+	const running = processor({ id: "j2", data: { kind: "github", repo: "o/r" } }, "tok", ac.signal).catch(() => {});
+	// Let the job reach the running container before aborting -- in reality the container has been
+	// up for minutes when the 30-min timeout fires.
+	await new Promise((r) => setTimeout(r, 10));
+	ac.abort();
+	await new Promise((r) => setTimeout(r, 10));
+	assert.equal(stopped, "pi-job-j2");
+	await running;
+});