perf(test): speed up workspace test suite (#172)

Author: HenryQW

package.json

@@ -38,7 +38,7 @@
   "scripts": {
     "build": "npm run build --workspace @henryqw/pi-herdr --workspace @henryqw/pi-task-models --workspace @henryqw/pi-subagent",
     "postinstall": "npm run build",
-    "test": "npm test --workspaces --if-present",
+    "test": "node scripts/test-workspaces.mjs",
     "test:live": "npm run test:live --workspace @henryqw/pi-auto-compact",
     "typecheck": "npm run typecheck --workspaces --if-present",
     "pack:check": "npm run pack:check --workspaces --if-present"

packages/pi-auto-dag/test/execution.test.ts

@@ -521,44 +521,6 @@ test("process interruption after claiming leaves inspectable execution state and
 	assert.deepEqual((await readdir(stateRoot(project.root))).filter((name) => name.startsWith(".lifecycle")), []);
 });
 
-test("abort retains an undelivered blocked notification until acknowledgement", async (t) => {
-	const project = await setupProject(t);
-	const graph = parseDeliveryGraph({ ...graphInput, issues: [graphInput.issues[2]] });
-	const initial = createInitialRunState({
-		run_id: RUN_ID,
-		graph,
-		source_commit: await git(project.root, "rev-parse", "HEAD"),
-		main_worktree: project.root,
-		integration_branch: "integration",
-		default_branch: "main",
-		created_at: "2026-08-09T00:00:00.000Z",
-		main_pane: "main-pane",
-		workspace_id: "main-workspace",
-	});
-	await createRun(project.root, initial, () => "create");
-	await writeRunState(project.root, {
-		...initial,
-		phase: "blocked",
-		block_reason: "API needs a decision.",
-		tasks: {
-			...initial.tasks,
-			api: { status: "blocked", attempts: 1, block_reason: "Choose a protocol.", blocked_role: "implementer" },
-		},
-	}, () => "blocked");
-	const pending = (await readRunState(project.root, RUN_ID))!.notifications[0];
-	const lifecycle = createCoreLifecycle({ uuid: () => "lifecycle", now: () => "2026-08-09T01:00:00.000Z" });
-	const aborted = await lifecycle.abort(project.root, "Cancelled by user");
-	assert.equal(aborted.phase, "aborted");
-	assert.equal(aborted.abort_cleanup_complete, true);
-	assert.equal(await readActiveRunId(project.root), RUN_ID);
-	const resumed = await lifecycle.resume(project.root);
-	assert.equal(resumed.phase, "aborted");
-	assert.equal(resumed.notifications.find(({ event_id }) => event_id === pending.event_id)?.delivered_at, undefined);
-	assert.equal(await readActiveRunId(project.root), RUN_ID);
-	await lifecycle.acknowledgeNotification(project.root, pending.event_id);
-	assert.equal(await readActiveRunId(project.root), undefined);
-});
-
 test("terminal settlement retains an aborted lock without cleanup proof", async (t) => {
 	const project = await setupProject(t);
 	const graph = parseDeliveryGraph({ ...graphInput, issues: [graphInput.issues[2]] });
@@ -579,49 +541,7 @@ test("terminal settlement retains an aborted lock without cleanup proof", async
 	assert.equal(await readActiveRunId(project.root), RUN_ID);
 });
 
-test("terminal settlement releases a completed lock after acknowledgement persistence", async (t) => {
-	const project = await setupProject(t);
-	const graph = parseDeliveryGraph({ ...graphInput, issues: [graphInput.issues[2]] });
-	const initial = createInitialRunState({
-		run_id: RUN_ID,
-		graph,
-		source_commit: await git(project.root, "rev-parse", "HEAD"),
-		main_worktree: project.root,
-		integration_branch: "integration",
-		default_branch: "main",
-		created_at: "2026-08-09T00:00:00.000Z",
-		main_pane: "main-pane",
-		workspace_id: "main-workspace",
-	});
-	await createRun(project.root, initial, () => "create");
-	await writeRunState(project.root, {
-		...initial,
-		phase: "completed",
-		tasks: Object.fromEntries(Object.entries(initial.tasks).map(([id, task]) => [id, { status: "completed", attempts: task.attempts }])),
-		pr: {
-			number: 42,
-			url: "https://example.test/pull/42",
-			head_ref: "integration",
-			base_ref: "main",
-			head_oid: initial.integration_head,
-		},
-	}, () => "completed");
-	const completed = (await readRunState(project.root, RUN_ID))!;
-	await assert.rejects(createCoreLifecycle().abort(project.root, "Too late"), /Cannot abort a completed run/);
-	assert.equal((await readRunState(project.root, RUN_ID))!.phase, "completed");
-	await writeRunState(project.root, {
-		...completed,
-		notifications: completed.notifications.map((notification) => ({
-			...notification,
-			delivered_at: "2026-08-09T01:00:00.000Z",
-		})),
-	}, () => "delivered");
-	assert.equal(await readActiveRunId(project.root), RUN_ID);
-	await createCoreLifecycle().settleTerminal(project.root);
-	assert.equal(await readActiveRunId(project.root), undefined);
-});
-
-test("durable blocked/completed notifications retain stable IDs and release completion only after ack", async (t) => {
+test("durable blocked/completed notifications retain stable IDs and settle delivered completion", async (t) => {
 	const project = await setupProject(t);
 	const graph = parseDeliveryGraph({ ...graphInput, issues: [graphInput.issues[2]] });
 	const initial = createInitialRunState({
@@ -703,10 +623,17 @@ test("durable blocked/completed notifications retain stable IDs and release comp
 	const completedRoundTrip = (await readRunState(project.root, RUN_ID))!;
 	assert.equal(completedRoundTrip.notifications.find(({ kind }) => kind === "completed")!.event_id, completedId);
 	assert.equal(completedRoundTrip.notifications.length, 3);
-
-	const delivered = await lifecycle.acknowledgeNotification(project.root, completedId);
+	await assert.rejects(lifecycle.abort(project.root, "Too late"), /Cannot abort a completed run/);
+	assert.deepEqual(await readRunState(project.root, RUN_ID), completedRoundTrip);
+	await writeRunState(project.root, {
+		...completedRoundTrip,
+		notifications: completedRoundTrip.notifications.map((notification) => notification.kind === "completed"
+			? { ...notification, delivered_at: "2026-08-09T01:00:00.000Z" }
+			: notification),
+	}, () => "completed-delivered");
+	assert.equal(await readActiveRunId(project.root), RUN_ID);
+	await lifecycle.settleTerminal(project.root);
 	assert.equal(await readActiveRunId(project.root), undefined);
-	assert.deepEqual(await lifecycle.acknowledgeNotification(project.root, completedId), delivered);
 });
 
 test("followUp delivery is fire-and-forget: failed or unacknowledged dispatch stays pending until explicit acknowledgement", async (t) => {
@@ -731,8 +658,15 @@ test("followUp delivery is fire-and-forget: failed or unacknowledged dispatch st
 		tasks: { ...initial.tasks, api: { status: "blocked", attempts: 1, block_reason: "Choose a protocol.", blocked_role: "implementer" } },
 	}, () => "blocked");
 	const lifecycle = createCoreLifecycle({ uuid: () => "lifecycle", now: () => "2026-08-09T01:00:00.000Z" });
-	await lifecycle.abort(project.root, "Cancelled by user");
 	const eventId = (await readRunState(project.root, RUN_ID))!.notifications[0].event_id;
+	const aborted = await lifecycle.abort(project.root, "Cancelled by user");
+	assert.equal(aborted.phase, "aborted");
+	assert.equal(aborted.abort_cleanup_complete, true);
+	assert.equal(await readActiveRunId(project.root), RUN_ID);
+	assert.equal((await lifecycle.resume(project.root)).phase, "aborted");
+	const pending = (await readRunState(project.root, RUN_ID))!.notifications[0];
+	assert.equal(pending.event_id, eventId);
+	assert.equal(pending.delivered_at, undefined);
 
 	let failSend = true;
 	const sent: string[] = [];

packages/pi-auto-dag/test/handoff.test.ts

@@ -177,6 +177,7 @@ test("failed terminal delivery releases the prompted-turn latch for retry", asyn
 	const tool = tools.find((candidate) => candidate.name === WORKER_TOOLS.request_review)!;
 
 	await assert.rejects(tool.execute("first", { summary: "finished" }), /acceptance receipt/);
+	await assert.rejects(readFile(action.value.receipt_path), /ENOENT/);
 	const result = await tool.execute("retry", { summary: "finished" });
 
 	assert.equal(result.details.status, "accepted");
@@ -337,27 +338,6 @@ test("worker redelivers an active rejected ticket for lifecycle recovery", async
 	assert.deepEqual(deliveries, [action.value.event_id, replacement.event_id]);
 });
 
-test("delivery without lifecycle acceptance does not report Sent", async (t) => {
-	const root = await mkdtemp(join(tmpdir(), "pi-auto-dag-handoff-"));
-	t.after(async () => await rm(root, { recursive: true, force: true }));
-	const action = await ticket(root);
-	const tools: Array<{ name: string; execute: Function }> = [];
-	createWorkerExtension({
-		environment: environment(action.path),
-		deliveryAttempts: 1,
-		delay: async () => {},
-		runner: async (command) => command === "git"
-			? { code: 0, stdout: `${HEAD}\n`, stderr: "" }
-			: { code: 0, stdout: "", stderr: "" },
-	})({ on() {}, registerTool(tool: { name: string; execute: Function }) { tools.push(tool); } } as never);
-
-	await assert.rejects(
-		tools.find((tool) => tool.name === WORKER_TOOLS.request_review)!.execute("call", { summary: "finished" }),
-		/acceptance receipt/,
-	);
-	await assert.rejects(readFile(action.value.receipt_path), /ENOENT/);
-});
-
 test("worker requests compaction before high-context submission", async (t) => {
 	const root = await mkdtemp(join(tmpdir(), "pi-auto-dag-handoff-"));
 	t.after(async () => await rm(root, { recursive: true, force: true }));

packages/pi-auto-dag/test/pr-lifecycle.test.ts

@@ -43,6 +43,11 @@ test("the frozen final check prepares its disposable checkout and opens one exac
 
 	assert.equal(state.phase, "completed");
 	assert.equal(state.tasks["final-check"].status, "completed");
+	const notification = state.notifications[0];
+	assert.equal(state.notifications.length, 1);
+	assert.ok(notification?.kind === "completed");
+	assert.equal(notification.delivered_at, undefined);
+	assert.equal(notification.payload.pr.url, "https://example.test/pr/42");
 	assert.equal(await readActiveRunId(project.root), RUN_ID);
 	assert.equal(gh.pr?.headRefName, "dag");
 	assert.equal(gh.pr?.baseRefName, "main");
@@ -60,43 +65,11 @@ test("the frozen final check prepares its disposable checkout and opens one exac
 		.map((call) => JSON.parse(call.args[3]))
 		.find((value) => value.type === "auto_dag_review" && value.kind === "final_check");
 	await assertReviewPatch(finalPacket, project.root, state.source_commit, state.integration_head, { type: "integration_head" });
-});
-
-test("completion notification retains the active lock until idempotent acknowledgement", async (t) => {
-	const project = await makeProject(t);
-	const lifecycle = makeLifecycle(combinedRunner(fakeHerdr(), fakeGh(project.root)));
-	await finishInitialRun(project.root, lifecycle);
-	const state = (await lifecycle.status(project.root, RUN_ID))!;
-	const notification = state.notifications[0];
-
-	assert.equal(state.notifications.length, 1);
-	assert.ok(notification?.kind === "completed");
-	assert.equal(notification.delivered_at, undefined);
-	assert.equal(notification.payload.pr.url, "https://example.test/pr/42");
-	assert.equal(await readActiveRunId(project.root), RUN_ID);
 	const acknowledged = await lifecycle.acknowledgeNotification(project.root, notification.event_id);
 	assert.equal(await readActiveRunId(project.root), undefined);
 	assert.deepEqual(await lifecycle.acknowledgeNotification(project.root, notification.event_id), acknowledged);
 });
 
-test("a retained completed run resumes unfinished terminal cleanup without revalidating its PR", async (t) => {
-	const project = await makeProject(t);
-	const gh = fakeGh(project.root);
-	const lifecycle = makeLifecycle(combinedRunner(fakeHerdr(), gh));
-	await finishInitialRun(project.root, lifecycle);
-	const completed = (await lifecycle.status(project.root, RUN_ID))!;
-
-	// Simulate an interrupted cleanup save plus a PR merged after completion.
-	await writeRunState(project.root, replaceTask(completed, "alpha", { ...completed.tasks.alpha, branch_cleanup_done: undefined }), () => "interrupted");
-	gh.state = "MERGED";
-	const ghCalls = gh.calls.length;
-
-	const resumed = await lifecycle.resume(project.root);
-	assert.equal(resumed.phase, "completed");
-	assert.equal(resumed.tasks.alpha.branch_cleanup_done, true);
-	assert.equal(gh.calls.length, ghCalls);
-});
-
 test("a failed terminal cleanup keeps completion durable and retries after the PR merges", async (t) => {
 	const project = await makeProject(t);
 	const gh = fakeGh(project.root);
@@ -117,12 +90,14 @@ test("a failed terminal cleanup keeps completion durable and retries after the P
 	await git(project.root, "branch", completed.tasks.alpha.branch!, completed.tasks.alpha.commit!);
 	failBranchDelete = true;
 	gh.state = "MERGED";
+	let ghCalls = gh.calls.length;
 	let resumed = await lifecycle.resume(project.root);
 	assert.equal(resumed.phase, "completed");
 	assert.equal(resumed.cleanup_blocks?.[0]?.operation, "branch");
+	assert.equal(gh.calls.length, ghCalls);
 
 	failBranchDelete = false;
-	const ghCalls = gh.calls.length;
+	ghCalls = gh.calls.length;
 	resumed = await lifecycle.resume(project.root);
 	assert.equal(resumed.phase, "completed");
 	assert.equal(resumed.cleanup_blocks, undefined);

packages/pi-auto-dag/test/run.mjs

@@ -11,7 +11,7 @@ if (!files.length || files.some((file) => !existsSync(file))) {
 	throw new Error(`Unknown pi-auto-dag test suite: ${suite ?? "(none)"}`);
 }
 
-const child = spawn(process.execPath, ["--test", ...files], { stdio: "inherit" });
+const child = spawn(process.execPath, ["--test", "--test-concurrency=2", ...files], { stdio: "inherit" });
 child.on("exit", (code) => {
 	process.exitCode = code ?? 1;
 });

packages/pi-subagent/test/subagent.test.ts

@@ -1843,16 +1843,20 @@ Do bounded work.
 `);
 		const runner = join(agentDir, "fake-pi.mjs");
 		await writeFile(runner, `const event = (value) => console.log(JSON.stringify(value));
-setTimeout(() => event({ type: "tool_execution_start", toolCallId: "tool-1", toolName: "bash", args: {} }), 80);
+setTimeout(() => event({ type: "tool_execution_start", toolCallId: "tool-1", toolName: "bash", args: {} }), 200);
+setTimeout(() => event({ type: "message_update", usage: { totalTokens: 1 } }), 600);
 setTimeout(() => {
 	event({ type: "tool_execution_end", toolCallId: "tool-1", toolName: "bash", result: {}, isError: false });
 	event({ type: "message_end", message: { role: "assistant", content: [{ type: "text", text: "done" }], stopReason: "end" } });
-}, 180);
+}, 1_000);
 `);
 		process.argv[1] = runner;
-		const app = harness({ timeoutPolicy: { idleMs: 120, maxMs: 270 } });
+		const timeoutPolicy = { idleMs: 800, maxMs: 1_500 };
+		const app = harness({ timeoutPolicy });
+		const startedAt = Date.now();
 		const result = await app.tool.execute("call-1", { role: "worker", task: "work" }, undefined, undefined, app.ctx);
 		assert.equal(singleOutput(result), "done");
+		assert.ok(Date.now() - startedAt > timeoutPolicy.idleMs, "completion did not outlast the original idle deadline");
 	});
 });
 
@@ -1866,7 +1870,7 @@ test("workflow transport retains executor rejection Usage when no child result e
 		const runner = join(agentDir, "fake-pi.mjs");
 		await writeFile(runner, `console.log(JSON.stringify({ type: "message_update", usage: ${JSON.stringify(observedUsage)} })); setInterval(() => {}, 1_000);`);
 		process.argv[1] = runner;
-		const app = harness({ timeoutPolicy: { idleMs: 40, maxMs: 100 } });
+		const app = harness({ timeoutPolicy: { idleMs: 500, maxMs: 800 } });
 		const error = await app.tool.execute("usage-rejection", { role: "worker", task: "work" }, undefined, undefined, app.ctx).then(
 			() => assert.fail("expected workflow failure"),
 			(reason) => reason,

scripts/test-workspaces.mjs

@@ -0,0 +1,65 @@
+import { spawn, spawnSync } from "node:child_process";
+
+const npmExecPath = process.env.npm_execpath;
+if (!npmExecPath) throw new Error("npm_execpath is required to run workspace tests.");
+
+const concurrency = 2;
+const testArgs = process.argv.slice(2);
+
+const discovery = spawnSync(process.execPath, [npmExecPath, "pkg", "get", "name", "scripts.test", "--workspaces", "--json"], {
+	encoding: "utf8",
+	stdio: ["ignore", "pipe", "inherit"],
+});
+
+if (discovery.error) throw discovery.error;
+if (discovery.status !== 0) process.exit(discovery.status ?? 1);
+
+const manifests = JSON.parse(discovery.stdout);
+if (!manifests || typeof manifests !== "object" || Array.isArray(manifests)) {
+	throw new Error("npm did not return workspace package metadata.");
+}
+
+const workspaces = Object.entries(manifests)
+	.filter(([, manifest]) => manifest && typeof manifest === "object" && Object.hasOwn(manifest, "scripts.test"))
+	.map(([workspace]) => workspace);
+
+if (!workspaces.length) throw new Error("No workspace test scripts found.");
+
+const prioritizedWorkspaces = ["@henryqw/pi-auto-dag", "@henryqw/pi-subagent"];
+const scheduledWorkspaces = [
+	...prioritizedWorkspaces.filter((workspace) => workspaces.includes(workspace)),
+	...workspaces.filter((workspace) => !prioritizedWorkspaces.includes(workspace)),
+];
+const failures = [];
+
+function run(workspace) {
+	return new Promise((resolve) => {
+		const args = ["run", "test", "--workspace", workspace];
+		if (workspace === "@henryqw/pi-subagent") args.push("--", "--test-concurrency=2", ...testArgs);
+		else if (testArgs.length) args.push("--", ...testArgs);
+		const child = spawn(process.execPath, [npmExecPath, ...args], { stdio: "inherit" });
+		child.once("error", (error) => {
+			console.error(`Failed to start tests for ${workspace}: ${error.message}`);
+			resolve(1);
+		});
+		child.once("exit", (code) => resolve(code ?? 1));
+	});
+}
+
+let next = 0;
+async function worker() {
+	while (next < scheduledWorkspaces.length) {
+		const workspace = scheduledWorkspaces[next++];
+		const code = await run(workspace);
+		if (code !== 0) failures.push({ workspace, code });
+	}
+}
+
+await Promise.all(Array.from({ length: Math.min(concurrency, scheduledWorkspaces.length) }, worker));
+
+if (failures.length) {
+	for (const { workspace, code } of failures) {
+		console.error(`Workspace test failed: ${workspace} (exit ${code})`);
+	}
+	process.exitCode = 1;
+}