Phase 2 (local loop): config + real runContainer

config.mjs: the worker's env config, validated and fail-loud (configError-tagged, like the
runner). Conservative money defaults -- PI_DAILY_CAP=25 bounds container starts/day, cap 0
rejected (would fail closed, likelier a typo), PI_MAX_TURNS=30 (pi has none of its own),
concurrency 3. Model default is the DATED claude-sonnet-4-5-20250929 for determinism per
CONST-PI-VERSION-PINNED, not the floating alias.

run-container.mjs: the real runContainer the processor injects. spawn (not execFile) because
a non-zero exit is NORMAL -- exit 1/2 are outcomes, not errors to reject on; the code comes
from the close event. It reuses buildContainerEnv (closed provider-key allowlist) and
buildDockerRunArgs (isolation flags). An already-aborted signal returns 137 and never spawns
a container (the timeout can fire during a slow prepare). Output streams to the operator's
console so they watch the agent work on their own folder. async so a sync throw (unconfigured
provider) surfaces as an awaitable rejection.

Verified against the REAL image end-to-end: with a fake key forwarded, runContainer launched
the container, the runner read the mounted prompt, found the model, called Anthropic's real
API, got a genuine 401, and the pi-never-throws -> stopReason -> exit-code chain classified it
as error -> exit 1 (retryable) with turns:1. That is the entire local execution path proven
against a real provider round-trip, with a fake key so no spend. The already-aborted path
returns 137, no container. CI unit tests use an injected spawn (no docker needed): argv,
mounts, exit-code passthrough, and the pre-spend refusal when the provider is unconfigured.

Author: edgehero

worker/src/config.mjs

@@ -0,0 +1,50 @@
+/**
+ * Worker configuration, from the environment. Validated and fail-loud: a misconfigured worker
+ * should refuse to start with a clear message, not launch and fail per-job.
+ *
+ * Errors are tagged `piDispatchConfig` so the CLI/entry can print them cleanly and exit non-zero.
+ */
+
+export function configError(message) {
+	const error = new Error(message);
+	error.piDispatchConfig = true;
+	return error;
+}
+
+function positiveInt(env, name, fallback) {
+	const raw = env[name];
+	if (raw === undefined || raw === "") {
+		if (fallback !== undefined) return fallback;
+		throw configError(`missing required env: ${name}`);
+	}
+	const n = Number.parseInt(raw, 10);
+	if (!Number.isInteger(n) || n < 1 || String(n) !== String(raw).trim()) {
+		throw configError(`invalid ${name}: ${JSON.stringify(raw)} (want a positive integer)`);
+	}
+	return n;
+}
+
+/**
+ * Parse the worker's config from `env` (default process.env). All defaults are conservative:
+ * spend controls (`PI_DAILY_CAP`, `PI_MAX_TURNS`) exist to bound money, so they default low, and a
+ * cap of 0 would fail closed (budget.mjs refuses every job) rather than mean "unlimited".
+ */
+export function loadConfig(env = process.env) {
+	const model = env.PI_MODEL ?? "claude-sonnet-4-5-20250929"; // dated snapshot; deterministic per CONST-PI-VERSION-PINNED
+	return {
+		valkeyUrl: env.VALKEY_URL ?? "redis://127.0.0.1:6379",
+		concurrency: positiveInt(env, "PI_CONCURRENCY", 3), // DES-CONCURRENCY-3
+		dailyCap: positiveInt(env, "PI_DAILY_CAP", 25), // bounds container STARTS per day (money)
+		provider: env.PI_PROVIDER ?? "anthropic",
+		model,
+		maxTurns: positiveInt(env, "PI_MAX_TURNS", 30), // pi has no turn limit; we impose one
+		jobImage: env.PI_JOB_IMAGE ?? "pi-job:latest",
+		jobsDir: env.PI_JOBS_DIR ?? defaultJobsDir(),
+	};
+}
+
+function defaultJobsDir() {
+	// Under the OS temp dir by default. Holds only the read-only /job inputs (prompt + .pi/); the
+	// workspace for a local job is the operator's own folder, not here.
+	return `${process.env.TMPDIR ?? process.env.TEMP ?? "/tmp"}/pi-dispatch/jobs`.replace(/\\/g, "/");
+}

worker/src/run-container.mjs

@@ -0,0 +1,54 @@
+import { spawn } from "node:child_process";
+import { buildDockerRunArgs } from "./docker-run.mjs";
+import { buildContainerEnv } from "./env-allowlist.mjs";
+
+/**
+ * The real `runContainer` the processor injects. Launches one job container and returns its exit
+ * code, which INT-RUNNER-EXIT-CODE-PROTOCOL turns into retry-vs-success.
+ *
+ * `spawn` (not execFile) because a non-zero exit is NORMAL here: exit 1 (infra) and 2 (policy) are
+ * expected outcomes, not errors to reject on. The exit code comes from the `close` event.
+ *
+ * The container is stopped on abort by the worker wiring (index.mjs onAbort -> docker stop), which
+ * causes `docker run` to exit and this promise to resolve. We only handle the entry case here: if
+ * the signal is ALREADY aborted (the 30-min timeout fired during a slow prepare), do not start a
+ * container at all.
+ *
+ * Output is streamed to `onOutput` (default: the worker's stdout) so the operator watches the agent
+ * work on their own machine -- the natural local UX. This is the operator's own console for their
+ * own folder; it is not a persistent PII log.
+ */
+export function makeRunContainer({ image, hostEnv = process.env, onOutput = (c) => process.stdout.write(c), spawnFn = spawn }) {
+	// async so a synchronous throw (e.g. buildContainerEnv on an unconfigured provider) surfaces as
+	// a rejection, uniformly awaitable by the processor and by tests.
+	return async function runContainer({ job, token, prepared, name, signal }) {
+		if (signal?.aborted) return 137; // killed before it could start
+
+		// Closed env allowlist: only the provider key + the declared PI_* vars. Throws (config) if
+		// the provider is unconfigured -- the processor turns that into a pre-spend refusal.
+		const env = buildContainerEnv({
+			provider: job.provider,
+			model: job.model,
+			maxTurns: job.maxTurns,
+			jobId: name,
+			githubToken: token ?? undefined,
+			hostEnv,
+		});
+
+		const args = buildDockerRunArgs({
+			image,
+			env,
+			jobDir: prepared.jobDir,
+			workspace: prepared.workspace,
+			name,
+		});
+
+		return await new Promise((resolve, reject) => {
+			const child = spawnFn("docker", args, { stdio: ["ignore", "pipe", "pipe"] });
+			child.stdout?.on("data", onOutput);
+			child.stderr?.on("data", onOutput);
+			child.on("error", reject); // docker not found / cannot spawn -- a real infra failure
+			child.on("close", (code) => resolve(code ?? 1));
+		});
+	};
+}

worker/test/config.test.mjs

@@ -0,0 +1,47 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import { configError, loadConfig } from "../src/config.mjs";
+
+test("loads conservative defaults with an empty-ish env", () => {
+	const c = loadConfig({});
+	assert.equal(c.concurrency, 3);
+	assert.equal(c.dailyCap, 25);
+	assert.equal(c.provider, "anthropic");
+	assert.equal(c.model, "claude-sonnet-4-5-20250929"); // dated, deterministic
+	assert.equal(c.maxTurns, 30);
+	assert.equal(c.valkeyUrl, "redis://127.0.0.1:6379");
+	assert.equal(c.jobImage, "pi-job:latest");
+	assert.ok(c.jobsDir.length > 0);
+});
+
+test("env overrides every field", () => {
+	const c = loadConfig({
+		VALKEY_URL: "redis://valkey:6379",
+		PI_CONCURRENCY: "6",
+		PI_DAILY_CAP: "100",
+		PI_PROVIDER: "openai",
+		PI_MODEL: "gpt-x",
+		PI_MAX_TURNS: "50",
+		PI_JOB_IMAGE: "pi-job:0.1.0",
+		PI_JOBS_DIR: "/srv/jobs",
+	});
+	assert.equal(c.concurrency, 6);
+	assert.equal(c.dailyCap, 100);
+	assert.equal(c.provider, "openai");
+	assert.equal(c.model, "gpt-x");
+	assert.equal(c.jobsDir, "/srv/jobs");
+});
+
+test("a malformed integer is a config error, not a silent NaN", () => {
+	for (const bad of ["0", "-1", "3.5", "abc", "3x"]) {
+		assert.throws(() => loadConfig({ PI_CONCURRENCY: bad }), (e) => e.piDispatchConfig === true, `PI_CONCURRENCY=${bad}`);
+	}
+});
+
+test("cap 0 is rejected -- it would fail closed, and is more likely a typo than intent", () => {
+	assert.throws(() => loadConfig({ PI_DAILY_CAP: "0" }), (e) => e.piDispatchConfig === true);
+});
+
+test("configError is tagged for clean CLI reporting", () => {
+	assert.equal(configError("x").piDispatchConfig, true);
+});

worker/test/run-container.test.mjs

@@ -0,0 +1,72 @@
+import assert from "node:assert/strict";
+import { EventEmitter } from "node:events";
+import { test } from "node:test";
+
+// run-container imports env-allowlist -> @earendil-works/pi-ai, so this skips below the node floor
+// and runs in CI (PI_DISPATCH_REQUIRE_WORKER_TESTS=1 makes a skip a hard failure).
+let mod;
+let importError;
+try {
+	mod = await import("../src/run-container.mjs");
+} catch (error) {
+	importError = error;
+}
+if (!mod && process.env.PI_DISPATCH_REQUIRE_WORKER_TESTS === "1") {
+	throw new Error(`run-container tests are REQUIRED here but pi-ai could not import.\n${importError}`);
+}
+const skip = mod ? false : `pi-ai not installed (node ${process.version} < 22.19.0); CI runs these`;
+
+const HOST = { ANTHROPIC_API_KEY: "sk-real" };
+const JOB = { kind: "local", provider: "anthropic", model: "m", maxTurns: 5 };
+const PREPARED = { workspace: "/host/folder", jobDir: "/host/jobs/j1" };
+
+/** A fake `docker` child: records argv, lets the test drive its exit. */
+function fakeSpawn(recorder, exitCode = 0) {
+	return (cmd, args) => {
+		recorder.cmd = cmd;
+		recorder.args = args;
+		const child = new EventEmitter();
+		child.stdout = new EventEmitter();
+		child.stderr = new EventEmitter();
+		queueMicrotask(() => child.emit("close", exitCode));
+		return child;
+	};
+}
+
+test("an already-aborted signal returns 137 and NEVER spawns docker", { skip }, async () => {
+	const rec = {};
+	const runContainer = mod.makeRunContainer({ image: "pi-job:x", hostEnv: HOST, spawnFn: fakeSpawn(rec) });
+	const ac = new AbortController();
+	ac.abort();
+	const code = await runContainer({ job: JOB, prepared: PREPARED, name: "j1", signal: ac.signal });
+	assert.equal(code, 137);
+	assert.equal(rec.cmd, undefined, "no container may start once the timeout has fired");
+});
+
+test("launches docker with the isolation argv and returns the container's exit code", { skip }, async () => {
+	const rec = {};
+	const runContainer = mod.makeRunContainer({ image: "pi-job:x", hostEnv: HOST, spawnFn: fakeSpawn(rec, 2) });
+	const code = await runContainer({ job: JOB, prepared: PREPARED, name: "j1", signal: new AbortController().signal });
+	assert.equal(code, 2, "exit 2 (policy) is a normal outcome, not an error to reject on");
+	assert.equal(rec.cmd, "docker");
+	assert.ok(rec.args.includes("--cap-drop=ALL"), "isolation flags present");
+	assert.ok(rec.args.includes("/host/jobs/j1:/job:ro"), "whole /job mounted read-only");
+	assert.ok(rec.args.includes("/host/folder:/workspace"), "the folder is the workspace");
+	assert.ok(rec.args.includes("ANTHROPIC_API_KEY=sk-real"), "the provider key is forwarded");
+});
+
+test("exit 1 (infra) is returned, not thrown -- it is retryable, not a spawn error", { skip }, async () => {
+	const runContainer = mod.makeRunContainer({ image: "pi-job:x", hostEnv: HOST, spawnFn: fakeSpawn({}, 1) });
+	const code = await runContainer({ job: JOB, prepared: PREPARED, name: "j1", signal: new AbortController().signal });
+	assert.equal(code, 1);
+});
+
+test("refuses before spawning if the provider is unconfigured (pre-spend guard)", { skip }, async () => {
+	const rec = {};
+	const runContainer = mod.makeRunContainer({ image: "pi-job:x", hostEnv: {}, spawnFn: fakeSpawn(rec) });
+	await assert.rejects(
+		() => runContainer({ job: JOB, prepared: PREPARED, name: "j1", signal: new AbortController().signal }),
+		(e) => e.piDispatchConfig === true,
+	);
+	assert.equal(rec.cmd, undefined, "no container for an unconfigured provider");
+});