Phase 2 (local loop): connection, queue, prepare dispatcher, entry, CLI -- loop CLOSED

The glue that turns the tested modules into a running system. The full local loop is now
verified end-to-end against real Valkey + the real image + the real provider API.

- connection.mjs: BullMQ connection opts + a raw ioredis client (for the budget), both from
  one VALKEY_URL. maxRetriesPerRequest: null (required by BullMQ's blocking connections).
- job-id.mjs: the local dedup key (content hash), kept free of any bullmq import so the dedup
  logic tests everywhere. NUL-delimited so field-concatenation cannot collide.
- queue.mjs: makeQueue + enqueueLocalJob. Verified against real Valkey: enqueue, dedup (a
  same-minute double-invoke is ignored), and the exact data shape runJob consumes.
- prepare.mjs: the prepareWorkspace dispatcher (local -> prepareLocalWorkspace; github throws
  "not in this slice"); the flow becomes a prompt hint, the skill comes from the project's
  materialised .pi/skills. cleanup removes the per-job dir, never the operator's folder.
- start.mjs: the runnable worker -- reads config, connects, wires every REAL dep, calls
  createWorker (which already has the timeout, abort->docker-stop, and SIGTERM shutdown).
- cli.mjs (bin: pi-dispatch): `run <folder> --task ... [--flow --provider --model
  --max-turns --force]` and `worker`. Refuses a dirty git working tree unless --force, because
  a local job edits in place with no undo (SECURITY.md). Validation returns before any queue
  contact.

Verified: `pi-dispatch run <clean git folder> --task tidy` enqueued (exit 0); the worker
picked it up (ACTIVE), materialised .pi/, RESERVED BUDGET (counter=1, proving
CONST-BUDGET-BEFORE-TOKENS in the assembled system), launched the real container; the runner
read the mounted prompt, found the model, called Anthropic's real API, got a genuine 401,
mapped it to exit 1 (error), and the worker classified it InfraRetry -> retry. A valid key
turns that 401 into a real edit. Fake key throughout -- no spend.

ioredis 5.11.1 pinned (bullmq's version). CLI validation + pure dedup run everywhere; the
Valkey enqueue integration test runs when VALKEY_TEST_URL is set (CI service, next commit).

Author: edgehero

package-lock.json

@@ -6,15 +6,20 @@
   "description": "BullMQ worker: drains the queue, mints scoped tokens, runs one container per job.",
   "license": "MIT",
   "main": "src/index.mjs",
+  "bin": {
+    "pi-dispatch": "src/cli.mjs"
+  },
   "engines": {
     "node": ">=22.19.0"
   },
   "scripts": {
-    "test": "node --test \"test/*.test.mjs\""
+    "test": "node --test \"test/*.test.mjs\"",
+    "start": "node src/cli.mjs worker"
   },
   "dependencies": {
     "@earendil-works/pi-ai": "0.80.7",
     "@octokit/auth-app": "8.2.0",
-    "bullmq": "5.80.4"
+    "bullmq": "5.80.4",
+    "ioredis": "5.11.1"
   }
 }

worker/package.json

@@ -0,0 +1,95 @@
+#!/usr/bin/env node
+import { execFileSync } from "node:child_process";
+import { existsSync } from "node:fs";
+import { resolve } from "node:path";
+import { parseArgs } from "node:util";
+import { loadConfig } from "./config.mjs";
+
+const USAGE = `pi-dispatch — run pi coding-agent flows on your own folders
+
+  pi-dispatch run <folder> --task "<what to do>" [--flow <name>]
+                           [--provider <p>] [--model <m>] [--max-turns <n>] [--force]
+  pi-dispatch worker       drain the queue (run this in another terminal, or as a service)
+
+Config comes from the environment (see .env.example); flags override it per run.`;
+
+export async function main(argv = process.argv.slice(2), env = process.env) {
+	const cmd = argv[0];
+
+	if (cmd === "worker") {
+		const { startWorker } = await import("./start.mjs");
+		startWorker(env);
+		return 0; // the worker keeps the process alive until SIGTERM
+	}
+
+	if (cmd === "run") {
+		const { values, positionals } = parseArgs({
+			args: argv.slice(1),
+			allowPositionals: true,
+			options: {
+				task: { type: "string" },
+				flow: { type: "string" },
+				provider: { type: "string" },
+				model: { type: "string" },
+				"max-turns": { type: "string" },
+				force: { type: "boolean", default: false },
+			},
+		});
+		const folder = positionals[0] && resolve(positionals[0]);
+		if (!folder || !existsSync(folder)) return fail(`folder not found: ${positionals[0] ?? "(none given)"}`);
+		if (!values.task) return fail("a --task is required");
+
+		// A local job edits the folder IN PLACE with no undo (SECURITY.md). Refuse a dirty working
+		// tree unless --force, so a bad run cannot mix with uncommitted work the operator can't
+		// cleanly separate. A non-git folder is caught later by prepare (v1 requires a git repo).
+		if (existsSync(`${folder}/.git`) && !values.force) {
+			const dirty = gitDirty(folder);
+			if (dirty === null) return fail(`${folder} is not a usable git repository`);
+			if (dirty) return fail(`${folder} has uncommitted changes. Commit or stash them, or pass --force.`);
+		}
+
+		const config = loadConfig(env);
+		const { parseConnection } = await import("./connection.mjs");
+		const { makeQueue, enqueueLocalJob } = await import("./queue.mjs");
+		const queue = makeQueue(parseConnection(config.valkeyUrl));
+		try {
+			const jobId = await enqueueLocalJob(queue, {
+				folder,
+				task: values.task,
+				flow: values.flow,
+				provider: values.provider ?? config.provider,
+				model: values.model ?? config.model,
+				maxTurns: values["max-turns"] ? Number(values["max-turns"]) : config.maxTurns,
+			});
+			process.stdout.write(`queued ${jobId} — folder ${folder}\nrun \`pi-dispatch worker\` to process it.\n`);
+		} finally {
+			await queue.close();
+		}
+		return 0;
+	}
+
+	process.stdout.write(`${USAGE}\n`);
+	return cmd ? 1 : 0;
+}
+
+function fail(message) {
+	process.stderr.write(`error: ${message}\n`);
+	return 1;
+}
+
+/** true = dirty, false = clean, null = not a working git repo. */
+function gitDirty(folder) {
+	try {
+		const out = execFileSync("git", ["-C", folder, "status", "--porcelain"], { encoding: "utf8" });
+		return out.trim().length > 0;
+	} catch {
+		return null;
+	}
+}
+
+// Entry point when run as a bin. Kept out of the exported main so tests can call main() directly.
+if (import.meta.url === `file://${process.argv[1]}` || process.argv[1]?.endsWith("cli.mjs")) {
+	main().then((code) => {
+		if (code) process.exitCode = code;
+	});
+}

worker/src/cli.mjs

@@ -0,0 +1,27 @@
+import { Redis } from "ioredis";
+
+/**
+ * Connection helpers for BullMQ and the budget's raw Redis client, both from one VALKEY_URL.
+ *
+ * `maxRetriesPerRequest: null` is REQUIRED by BullMQ for its blocking connections (the Worker
+ * uses BRPOPLPUSH); without it BullMQ throws at construction. It is harmless on the Queue and the
+ * budget client, so it is set consistently.
+ */
+
+/** BullMQ connection options parsed from a redis:// URL. */
+export function parseConnection(url) {
+	const u = new URL(url);
+	return {
+		host: u.hostname || "127.0.0.1",
+		port: Number(u.port || 6379),
+		...(u.password ? { password: u.password } : {}),
+		...(u.username ? { username: u.username } : {}),
+		...(u.pathname && u.pathname !== "/" ? { db: Number(u.pathname.slice(1)) } : {}),
+		maxRetriesPerRequest: null,
+	};
+}
+
+/** A raw ioredis client for the budget's INCR/EXPIRE. */
+export function makeRedisClient(url) {
+	return new Redis(url, { maxRetriesPerRequest: null });
+}

worker/src/connection.mjs

@@ -0,0 +1,16 @@
+import { createHash } from "node:crypto";
+
+/**
+ * A deterministic jobId for a local job. BullMQ's dedup is `EXISTS jobId`, so a double-invoke of
+ * the same task within the same minute produces the same id and the duplicate is ignored -- the
+ * local equivalent of REQ-DEDUP-BY-DELIVERY-GUID, guarding against a hasty second Enter
+ * double-spending, without blocking a deliberate re-run a minute later.
+ *
+ * Kept free of any bullmq import so the dedup logic is testable everywhere, not only where the
+ * queue's dependencies are installed.
+ */
+export function localJobId({ folder, flow, task, minute }) {
+	// NUL-delimited so {folder:'a',task:'bc'} and {folder:'ab',task:'c'} cannot collide.
+	const digest = createHash("sha256").update([folder, flow ?? "", task ?? "", minute].join("\0")).digest("hex");
+	return `local-${digest.slice(0, 16)}`;
+}

worker/src/job-id.mjs

@@ -0,0 +1,31 @@
+import { mkdirSync, mkdtempSync } from "node:fs";
+import { rm } from "node:fs/promises";
+import { join } from "node:path";
+import { prepareLocalWorkspace } from "./prepare-local.mjs";
+
+/**
+ * The `prepareWorkspace` dispatcher the processor injects. Creates a per-job dir under `jobsDir`
+ * (holding the read-only /job inputs) and routes by job kind.
+ *
+ * The `flow` becomes a prompt hint; the actual skill is provided by the project's materialised
+ * .pi/skills. GitHub jobs are not implemented in this slice.
+ */
+export function makePrepareWorkspace({ jobsDir }) {
+	mkdirSync(jobsDir, { recursive: true });
+	return async function prepareWorkspace(job) {
+		const jobDir = mkdtempSync(join(jobsDir, "job-"));
+		if (job.kind === "local") {
+			const task = job.flow ? `Use the "${job.flow}" skill for this task.\n\n${job.task ?? ""}` : (job.task ?? "");
+			return await prepareLocalWorkspace({ folder: job.folder, task, jobDir });
+		}
+		if (job.kind === "github") {
+			throw new Error("github jobs are not implemented in this slice (needs a GitHub App)");
+		}
+		throw new Error(`unknown job kind: ${job.kind}`);
+	};
+}
+
+/** Remove a per-job dir after the run. The workspace (the operator's folder) is never touched here. */
+export async function cleanup(prepared) {
+	if (prepared?.jobDir) await rm(prepared.jobDir, { recursive: true, force: true });
+}

worker/src/prepare.mjs

@@ -0,0 +1,30 @@
+import { Queue } from "bullmq";
+import { localJobId } from "./job-id.mjs";
+
+export const QUEUE = "pi-jobs";
+export { localJobId };
+
+export function makeQueue(connection) {
+	return new Queue(QUEUE, { connection });
+}
+
+/**
+ * Enqueue a local-folder job. Returns the jobId. The data shape is what the processor's runJob
+ * consumes (kind/folder/flow/task/provider/model/maxTurns).
+ *
+ * removeOnComplete keeps the dedup window ~= the retention. Unlike webhooks, local jobs are not
+ * redelivered, so a modest window is enough.
+ */
+export async function enqueueLocalJob(queue, { folder, flow, task, provider, model, maxTurns, now = new Date() }) {
+	const minute = now.toISOString().slice(0, 16); // YYYY-MM-DDTHH:MM -- the dedup window
+	const jobId = localJobId({ folder, flow, task, minute });
+	const data = { kind: "local", folder, flow, task, provider, model, maxTurns };
+	await queue.add("local", data, {
+		jobId,
+		attempts: 2,
+		backoff: { type: "exponential", delay: 60_000 },
+		removeOnComplete: { age: 24 * 3600 },
+		removeOnFail: { age: 7 * 24 * 3600 },
+	});
+	return jobId;
+}

worker/src/queue.mjs

@@ -0,0 +1,47 @@
+import { loadConfig } from "./config.mjs";
+import { makeRedisClient, parseConnection } from "./connection.mjs";
+import { createWorker } from "./index.mjs";
+import { cleanup, makePrepareWorkspace } from "./prepare.mjs";
+import { makeRunContainer } from "./run-container.mjs";
+
+/**
+ * The runnable worker. Reads config, connects to Valkey, wires every REAL dependency the processor
+ * needs, and starts draining the queue. `createWorker` already installs the timeout, the
+ * abort->docker-stop, and the SIGTERM/SIGINT graceful shutdown.
+ *
+ * GitHub deps (mintToken, isDefaultBranchProtected) are wired to throw: a github job reaching this
+ * worker in this slice is a clear error, not a silent no-op.
+ */
+export function startWorker(env = process.env) {
+	const config = loadConfig(env);
+	const log = (event, fields = {}) => process.stdout.write(`${JSON.stringify({ event, ...fields })}\n`);
+
+	const worker = createWorker({
+		connection: parseConnection(config.valkeyUrl),
+		concurrency: config.concurrency,
+		cap: config.dailyCap,
+		redis: makeRedisClient(config.valkeyUrl),
+		deps: {
+			runContainer: makeRunContainer({ image: config.jobImage, hostEnv: env }),
+			prepareWorkspace: makePrepareWorkspace({ jobsDir: config.jobsDir }),
+			cleanup,
+			comment: (job, text) => log("comment", { jobId: job?.id, text }),
+			log,
+			mintToken: async () => {
+				throw new Error("github jobs are not implemented in this slice");
+			},
+			isDefaultBranchProtected: async () => {
+				throw new Error("github jobs are not implemented in this slice");
+			},
+		},
+	});
+
+	log("worker_started", {
+		queue: "pi-jobs",
+		concurrency: config.concurrency,
+		dailyCap: config.dailyCap,
+		image: config.jobImage,
+		valkey: config.valkeyUrl,
+	});
+	return worker;
+}

worker/src/start.mjs

@@ -0,0 +1,68 @@
+import assert from "node:assert/strict";
+import { execFileSync } from "node:child_process";
+import { mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { test } from "node:test";
+import { main } from "../src/cli.mjs";
+
+// cli.mjs dynamic-imports bullmq (in the `run` enqueue and `worker` paths), so the VALIDATION
+// paths -- which return before any enqueue -- run everywhere. That is exactly the safety surface
+// worth testing: nothing should reach the queue if the inputs are bad.
+
+const env = { VALKEY_URL: "redis://127.0.0.1:6399" };
+
+test("no args prints usage and exits 0", async () => {
+	assert.equal(await main([], env), 0);
+});
+
+test("an unknown command exits 1", async () => {
+	assert.equal(await main(["frobnicate"], env), 1);
+});
+
+test("run with no folder fails", async () => {
+	assert.equal(await main(["run"], env), 1);
+});
+
+test("run with a missing folder fails before touching the queue", async () => {
+	assert.equal(await main(["run", "/no/such/folder", "--task", "x"], env), 1);
+});
+
+test("run with no --task fails", async () => {
+	const dir = mkdtempSync(join(tmpdir(), "cli-"));
+	assert.equal(await main(["run", dir, "--flow", "tidy"], env), 1);
+});
+
+function gitRepo({ dirty }) {
+	const dir = mkdtempSync(join(tmpdir(), "cli-git-"));
+	const g = (args) =>
+		execFileSync("git", ["-C", dir, ...args], {
+			env: { ...process.env, GIT_AUTHOR_NAME: "t", GIT_AUTHOR_EMAIL: "t@t", GIT_COMMITTER_NAME: "t", GIT_COMMITTER_EMAIL: "t@t" },
+		});
+	g(["init", "-q"]);
+	g(["config", "core.autocrlf", "false"]);
+	writeFileSync(join(dir, "f.txt"), "one\n");
+	g(["add", "-A"]);
+	g(["commit", "-qm", "init"]);
+	if (dirty) writeFileSync(join(dir, "f.txt"), "one\ntwo\n"); // uncommitted change
+	return dir;
+}
+
+test("run refuses a dirty git working tree (edits are in place, no undo)", async () => {
+	const dir = gitRepo({ dirty: true });
+	assert.equal(await main(["run", dir, "--task", "x"], env), 1);
+});
+
+test("--force overrides the dirty-tree refusal (reaches the enqueue, which needs Valkey)", async () => {
+	// Without Valkey the enqueue will throw; we only assert the guard did NOT stop it (it got past
+	// validation). If a Valkey is present it returns 0. Either way it must not return 1-from-guard.
+	const dir = gitRepo({ dirty: true });
+	let reachedEnqueue = false;
+	try {
+		const code = await main(["run", dir, "--task", "x", "--force"], env);
+		reachedEnqueue = code === 0; // Valkey present
+	} catch {
+		reachedEnqueue = true; // tried to connect => passed the guard
+	}
+	assert.ok(reachedEnqueue, "--force must let a dirty tree through to the enqueue");
+});