Add serialized two-lane chat campaign coordinator

tests/runner-e2e/chat-campaign-coordinator.test.ts

@@ -0,0 +1,255 @@
+import { createHash } from "node:crypto";
+import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync, existsSync, unlinkSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { afterEach, expect, it } from "vitest";
+import { ChatCampaignCoordinator, type CampaignCoordinatorOptions, type CampaignLaunchPermit } from "./chat-campaign-coordinator.js";
+import type { CampaignAttempt, CampaignScenario, ChatCampaign } from "./chat-campaign.js";
+
+const roots: string[] = [];
+afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); });
+function fixture(fault?: CampaignCoordinatorOptions["fault"]) {
+  const root = mkdtempSync(path.join(tmpdir(), "chat-two-lane-")); roots.push(root);
+  const baseline = "a".repeat(40), startedAt = "2026-10-27T10:00:00.000Z";
+  const selections = ["fixture.first", "fixture.second", "fixture.third", "fixture.fourth"].map(scenarioId => {
+    const attempt: CampaignAttempt = {
+      ordinal: 1, outcome: "running", executionId: `chat-features.codex.local.${scenarioId}`, startedAt, finishedAt: null,
+      sourceRevision: baseline, workingTreeDigest: "bytes", definitionDigest: `definition-${scenarioId}`, suiteDefinitionDigest: "suite",
+      graderDigest: `grader-${scenarioId}`, model: "synthetic", profile: "synthetic", environment: "local", automaticRetries: 0,
+      correction: null, classification: null, observedFailure: null, evidencePaths: ["tests/runner-e2e/results/synthetic/selection.json"],
+      usage: { costUsd: null, tokens: null, coverage: "not run" }, cleanup: "pending", corePrDependencies: [], databaseLane: { lane: "embedded-postgres" },
+      productIdentity: { kind: "source-build", productionSourceRevision: baseline, snapshotSourceRevision: baseline,
+        nativeBinarySha256: "b".repeat(64), workspaceBuildDigest: "c".repeat(64), receiptPaths: ["tests/runner-e2e/results/synthetic/build.json"], coreCandidate: false },
+    };
+    return { scenarioId, attempt };
+  });
+  const scenarios: CampaignScenario[] = selections.map(({ scenarioId, attempt }) => ({
+    id: scenarioId, feature: "fixture", sourceFile: "fixture.yaml", sourceSha256: "source", scenarioSha256: "scenario", capabilities: [],
+    context: "synthetic only", query: "synthetic", expected_outcome: "synthetic", owner: "fixture", readiness: "implemented", authoringStatus: "complete",
+    qualifyingDefinitionDigest: attempt.definitionDigest, qualifyingGraderDigest: attempt.graderDigest, executionId: attempt.executionId, blocker: null,
+    productPolicy: { lane: "source-build", sourceBaselineRevision: baseline }, attempts: [],
+  }));
+  const manifest: ChatCampaign = { version: 1, campaignId: "fixture", frozenAt: startedAt, baselineRevision: baseline, inventoryDigest: "inventory",
+    defaultsSha256: "defaults", denominator: 4, areas: 1, capabilities: 0, requiredPasses: 4, corePrs: [], evalPr: null, scenarios };
+  const manifestPath = path.join(root, "manifest.json"); writeFileSync(manifestPath, JSON.stringify(manifest));
+  const options: CampaignCoordinatorOptions = { manifestPath, controlDirectory: path.join(root, "control"), batchId: "synthetic-batch", selections, fault, evidenceRoot: root, now: () => startedAt };
+  return { root, options, ledger: () => JSON.parse(readFileSync(manifestPath, "utf8")) as ChatCampaign, open: () => ChatCampaignCoordinator.open(options) };
+}
+function completion() {
+  return { outcome: "failed" as const, finishedAt: "2026-10-27T10:00:01.000Z", definitionDigest: "", graderDigest: "",
+    evidencePaths: ["tests/runner-e2e/results/synthetic/result.json"], cleanup: "passed" as const,
+    usage: { costUsd: null, tokens: 123, coverage: "synthetic test receipt; no model" }, classification: "synthetic_behavior_failure",
+    observedFailure: "Synthetic substantive failure", corePrDependencies: [] };
+}
+function result(f: ReturnType<typeof fixture>, p: CampaignLaunchPermit) {
+  const plan = f.options.selections.find(x => x.scenarioId === p.scenarioId)!;
+  return { ...completion(), definitionDigest: plan.attempt.definitionDigest, graderDigest: plan.attempt.graderDigest };
+}
+const retainFailure = async (_permit: unknown, error: unknown) => ({ evidencePath: "tests/runner-e2e/results/synthetic/failure.json", summary: error instanceof Error ? error.message : "Synthetic failure" });
+function deferred() { let release!: () => void; const promise = new Promise<void>(resolve => { release = resolve; }); return { promise, release }; }
+
+it("serializes racing reservation/completion writes, preserves both rows and exactly matching receipts", async () => {
+  const f = fixture(), c = f.open();
+  const [a, b, duplicate] = await Promise.all([c.reserve("fixture.first"), c.reserve("fixture.second"), c.reserve("fixture.first")]);
+  expect(a).toEqual(duplicate); expect(a.reservationId).not.toBe(b.reservationId);
+  expect(f.ledger().scenarios.slice(0, 2).map(x => x.attempts.length)).toEqual([1, 1]);
+  await expect(c.reserve("fixture.third")).rejects.toThrow("Two lanes");
+  let launches = 0;
+  await Promise.all([a, b].map(p => c.launchOnce(p, async permit => {
+    launches++; expect(permit.args).toEqual(["--id", p.executionId, "--max-parallel", "1", "--max-automatic-retries", "0"]);
+  })));
+  await expect(c.launchOnce(duplicate, async () => { launches++; })).rejects.toThrow("not admissible");
+  await Promise.all([c.complete(a, result(f, a)), c.complete(b, result(f, b)), c.complete(a, result(f, a))]);
+  expect(launches).toBe(2);
+  expect(f.ledger().scenarios.slice(0, 2).map(x => x.attempts[0]?.usage.tokens)).toEqual([123, 123]);
+  await expect(c.complete(a, { ...result(f, a), observedFailure: "wrong bytes" })).rejects.toThrow("content conflict");
+  await c.close();
+});
+
+it("rejects a second owner even with a different batch directory and never steals stale PID locks", async () => {
+  const f = fixture(), c = f.open();
+  expect(() => ChatCampaignCoordinator.open({ ...f.options, controlDirectory: path.join(f.root, "other") })).toThrow();
+  const filename = `${f.options.manifestPath}.coordinator-lock`, lock = JSON.parse(readFileSync(filename, "utf8"));
+  writeFileSync(filename, JSON.stringify({ ...lock, pid: 2147483647, createdAt: "1900-01-01" }));
+  expect(() => f.open()).toThrow();
+  // Restoring this test's exact owner does not model authorizing recovery of real processes.
+  writeFileSync(filename, JSON.stringify(lock)); await c.close();
+});
+
+it.each(["executionId", "definitionDigest", "graderDigest", "ordinal", "productIdentity", "databaseLane"] as const)("rejects changed reviewed %s before admission", async field => {
+  const f = fixture();
+  const manifest = f.ledger();
+  if (field === "definitionDigest") manifest.scenarios[0]!.qualifyingDefinitionDigest = "changed";
+  else if (field === "graderDigest") manifest.scenarios[0]!.qualifyingGraderDigest = "changed";
+  else if (field === "executionId") manifest.scenarios[0]!.executionId = "foreign";
+  else if (field === "ordinal") f.options.selections[0]!.attempt.ordinal = 2;
+  else if (field === "productIdentity") f.options.selections[0]!.attempt.productIdentity = { ...f.options.selections[0]!.attempt.productIdentity!, kind: "source-build", productionSourceRevision: "f".repeat(40) } as CampaignAttempt["productIdentity"];
+  else f.options.selections[0]!.attempt.databaseLane = { lane: "owned-docker-postgres", engine: "foreign" } as unknown as CampaignAttempt["databaseLane"];
+  writeFileSync(f.options.manifestPath, JSON.stringify(manifest));
+  const c = f.open(); await expect(c.reserve("fixture.first")).rejects.toThrow();
+  expect(f.ledger().scenarios.every(x => !x.attempts.length)).toBe(true); await c.close();
+});
+
+it("uses fresh ledger reads and rejects changed reserved identity or tampered matching-completion ledger", async () => {
+  const f = fixture(), c = f.open(), p = await c.reserve("fixture.first");
+  await c.launchOnce(p, async () => undefined);
+  const m = f.ledger(); m.scenarios[1]!.blocker = "independent metadata preserved"; writeFileSync(f.options.manifestPath, JSON.stringify(m));
+  await c.complete(p, result(f, p));
+  expect(f.ledger().scenarios[1]!.blocker).toBe("independent metadata preserved");
+  const changed = f.ledger(); changed.scenarios[0]!.attempts[0]!.usage.tokens = 999; writeFileSync(f.options.manifestPath, JSON.stringify(changed));
+  await expect(c.complete(p, result(f, p))).rejects.toThrow("differs from receipt"); await c.close();
+});
+
+it.each(["intent", "manifest", "state"] as const)("fails closed at the real filesystem %s crash boundary without replaying a launch", async stage => {
+  const f = fixture(actual => { if (actual === stage) throw new Error("synthetic crash"); }), c = f.open();
+  await expect(c.reserve("fixture.first")).rejects.toThrow("synthetic crash");
+  expect(() => f.open()).toThrow(); // crash lock is retained
+  expect(() => JSON.parse(readFileSync(f.options.manifestPath, "utf8"))).not.toThrow();
+  // Synthetic owner-death reconciliation: no child was invoked in this test.
+  unlinkSync(`${f.options.manifestPath}.coordinator-lock`); f.options.fault = undefined;
+  const restarted = f.open(), state = await restarted.snapshot();
+  expect(state.stop?.reason).toBe("coordinator_failure");
+  await expect(restarted.reserve("fixture.second")).rejects.toThrow("stopped");
+  expect(f.ledger().scenarios[0]!.attempts.length).toBe(stage === "intent" ? 0 : 1);
+  await restarted.close();
+});
+
+it("retains launch uncertainty both before spawn and after a child returns, never automatically relaunches", async () => {
+  const f = fixture(), c = f.open(), p = await c.reserve("fixture.first");
+  let launches = 0;
+  await expect(c.launchOnce(p, async () => { launches++; throw new Error("spawn boundary uncertain"); })).rejects.toThrow("uncertain");
+  await expect(c.launchOnce(p, async () => { launches++; })).rejects.toThrow();
+  await c.close();
+  const restarted = f.open(); expect((await restarted.snapshot()).stop?.reason).toBe("coordinator_failure");
+  await expect(restarted.launchOnce(p, async () => { launches++; })).rejects.toThrow();
+  expect(launches).toBe(1); await restarted.close();
+});
+
+it("runs at most two independent children and drains the entire wave before stop-before-refill", async () => {
+  const f = fixture(), c = f.open(), gate = deferred(), secondStarted = deferred();
+  const launched: string[] = []; let active = 0, maximum = 0;
+  const run = c.run({
+    retainFailure, launchOnly: async p => { launched.push(p.scenarioId); maximum = Math.max(maximum, ++active);
+      if (p.scenarioId === "fixture.first") await secondStarted.promise;
+      if (p.scenarioId === "fixture.second") { secondStarted.release(); await gate.promise; } active--; return p.scenarioId; },
+    readCompletion: async (p, id) => ({ completion: result(f, p), ...(id === "fixture.first" ? { stop: { reason: "secret_leak" as const, detail: "actual synthetic secret match" } } : {}) }),
+  });
+  await secondStarted.promise;
+  await expect(c.close()).rejects.toThrow("in flight");
+  gate.release(); const state = await run;
+  expect(maximum).toBe(2); expect(launched.sort()).toEqual(["fixture.first", "fixture.second"]);
+  expect(Object.values(state.entries).every(x => x.phase === "completed")).toBe(true);
+  expect(state.stop?.reason).toBe("secret_leak"); expect(f.ledger().scenarios[2]!.attempts).toEqual([]);
+  await c.close(); const restarted = f.open(); await expect(restarted.reserve("fixture.third")).rejects.toThrow("stopped"); await restarted.close();
+});
+
+it("does not classify a caller-verified safe bound-only ZIP omission as a secret leak", async () => {
+  const f = fixture(), c = f.open(); let launches = 0;
+  const state = await c.run({ retainFailure, launchOnly: async () => { launches++; }, readCompletion: async p => ({ completion: { ...result(f, p), observedFailure: "Reporter privacy violation: independently verified safe bound-only ZIP omission, no secret match" } }) });
+  expect(launches).toBe(4); expect(state.stop).toBeNull(); await c.close();
+});
+
+it("durably stops on cleanup failure before completion commit or admission, including a crash", async () => {
+  let fail = false;
+  const f = fixture(stage => { if (fail && stage === "intent") throw new Error("completion crash"); });
+  const c = f.open(), p = await c.reserve("fixture.first"); await c.launchOnce(p, async () => undefined); fail = true;
+  await expect(c.complete(p, { ...result(f, p), cleanup: "failed" })).rejects.toThrow("completion crash");
+  expect(JSON.parse(readFileSync(path.join(f.options.controlDirectory, "state.json"), "utf8")).stop.reason).toBe("cleanup_failure");
+  expect(existsSync(path.join(f.options.controlDirectory, "pending.json"))).toBe(true);
+});
+
+it("does not relaunch completed entries when a clean parent resumes remaining selections", async () => {
+  const f = fixture(), c = f.open(), p = await c.reserve("fixture.first");
+  await c.launchOnce(p, async () => undefined); await c.complete(p, result(f, p)); await c.close();
+  const next = f.open(), launched: string[] = [];
+  await next.run({ retainFailure, launchOnly: async permit => { launched.push(permit.scenarioId); }, readCompletion: async permit => ({ completion: result(f, permit) }) });
+  expect(launched.sort()).toEqual(["fixture.fourth", "fixture.second", "fixture.third"]); await next.close();
+});
+
+it("rejects an outside writer during commit and retains the recovery lock instead of overwriting", async () => {
+  const f = fixture(stage => { if (stage === "intent") {
+    const manifest = f.ledger(); manifest.evalPr = "unexpected writer"; writeFileSync(f.options.manifestPath, JSON.stringify(manifest));
+  } });
+  const c = f.open(); await expect(c.reserve("fixture.first")).rejects.toThrow("outside coordinator");
+  expect(f.ledger().evalPr).toBe("unexpected writer"); expect(f.ledger().scenarios[0]!.attempts.length).toBe(0);
+  unlinkSync(`${f.options.manifestPath}.coordinator-lock`); f.options.fault = undefined;
+  expect(() => f.open()).toThrow("Unrecognized manifest");
+  expect(existsSync(`${f.options.manifestPath}.coordinator-lock`)).toBe(true);
+});
+
+it("rejects a forged permit and modified native reservation before accepting result evidence", async () => {
+  const f = fixture(), c = f.open(), p = await c.reserve("fixture.first");
+  await expect(c.launchOnce({ ...p, args: [...p.args, "--all"] }, async () => undefined)).rejects.toThrow("not admissible");
+  await c.launchOnce(p, async () => undefined);
+  const m = f.ledger(); m.scenarios[0]!.attempts[0]!.model = "wrong model"; writeFileSync(f.options.manifestPath, JSON.stringify(m));
+  await expect(c.complete(p, result(f, p))).rejects.toThrow("Reserved attempt changed"); await c.close();
+});
+
+it("persists a queued explicit stop before admitting the next queued reservation", async () => {
+  const f = fixture(), c = f.open();
+  const stopped = c.stop("shared_infrastructure", "synthetic real host failure");
+  await expect(c.reserve("fixture.first")).rejects.toThrow("stopped"); await stopped;
+  expect(f.ledger().scenarios.every(x => !x.attempts.length)).toBe(true); await c.close();
+});
+
+it("retains the original passing measurement while invalidating only qualification from a hash-bound closure review", async () => {
+  const f = fixture(), c = f.open(), p = await c.reserve("fixture.first"); await c.launchOnce(p, async () => undefined);
+  const original = f.options.selections[0]!.attempt;
+  const previous = { definitionDigest: original.definitionDigest, graderDigest: original.graderDigest };
+  const next = { definitionDigest: "new-normalized-selected-closure", graderDigest: "new-normalized-grader" };
+  const evidencePath = "tests/runner-e2e/results/synthetic/identity-review.json";
+  const filename = path.join(f.root, evidencePath); mkdirSync(path.dirname(filename), { recursive: true });
+  const bytes = JSON.stringify({ id: p.scenarioId, ordinal: p.ordinal, previous, next }); writeFileSync(filename, bytes);
+  const review = { evidencePath, receiptSha256: createHash("sha256").update(bytes).digest("hex"), previous, next };
+  const passed = { ...result(f, p), outcome: "passed" as const, evidencePaths: [evidencePath] };
+  await expect(c.complete(p, passed, undefined, { ...review, next: previous })).rejects.toThrow("Invalid qualifying");
+  await expect(c.complete(p, passed, undefined, { ...review, receiptSha256: "wrong" })).rejects.toThrow("receipt mismatch");
+  await c.complete(p, passed, undefined, review);
+  const row = f.ledger().scenarios[0]!;
+  expect(row.qualifyingDefinitionDigest).toBeNull(); expect(row.qualifyingGraderDigest).toBeNull();
+  expect(row.attempts[0]).toMatchObject({ outcome: "passed", definitionDigest: original.definitionDigest, graderDigest: original.graderDigest });
+  await c.close();
+});
+
+it("recovers a committed completion after interruption without duplicating the attempt or restarting any child", async () => {
+  let fault = false;
+  const f = fixture(stage => { if (fault && stage === "manifest") throw new Error("after completion commit"); });
+  const c = f.open(), p = await c.reserve("fixture.first"); await c.launchOnce(p, async () => undefined); fault = true;
+  await expect(c.complete(p, result(f, p))).rejects.toThrow("completion commit");
+  unlinkSync(`${f.options.manifestPath}.coordinator-lock`); f.options.fault = undefined;
+  const restarted = f.open(); await restarted.complete(p, result(f, p));
+  expect(f.ledger().scenarios[0]!.attempts).toHaveLength(1);
+  expect((await restarted.snapshot()).stop?.reason).toBe("coordinator_failure");
+  await expect(restarted.launchOnce(p, async () => { throw new Error("must not launch"); })).rejects.toThrow("not admissible");
+  await restarted.close();
+});
+
+it("stamps each real reservation time rather than preparation time, retaining exact later-wave identity", async () => {
+  const f = fixture(); let clock = "2026-10-27T10:00:00.100Z";
+  f.options.now = () => clock;
+  const c = f.open(), first = await c.reserve("fixture.first");
+  expect(f.ledger().scenarios[0]!.attempts[0]!.startedAt).toBe(clock);
+  await c.launchOnce(first, async () => undefined); await c.complete(first, result(f, first));
+  clock = "2026-10-27T10:15:00.000Z";
+  const later = await c.reserve("fixture.third");
+  expect(f.ledger().scenarios[2]!.attempts[0]!.startedAt).toBe(clock);
+  const entry = (await c.snapshot()).entries[later.reservationId]!;
+  expect(entry.reservedAttempt.startedAt).toBe(clock);
+  await c.launchOnce(later, async () => undefined);
+  await c.complete(later, { ...result(f, later), finishedAt: "2026-10-27T10:16:00.000Z" }); await c.close();
+  // Regenerated preparation timestamps are not a new reviewed selection identity.
+  for (const plan of f.options.selections) plan.attempt.startedAt = "2026-10-27T09:00:00.000Z";
+  const restarted = f.open(); expect((await restarted.snapshot()).entries[later.reservationId]!.reservedAttempt.startedAt).toBe(clock);
+  await restarted.close();
+});
+
+it("retains actionable sanitized completion failure evidence while draining the peer and stopping refill", async () => {
+  const f = fixture(), c = f.open(); const retained: string[] = [];
+  const state = await c.run({ launchOnly: async p => p.scenarioId,
+    readCompletion: async (p, id) => { if (id === "fixture.first") throw new Error("Synthetic result source digest differs from reserved digest"); return { completion: result(f, p) }; },
+    retainFailure: async (p, error, phase) => { retained.push(`${p?.scenarioId}:${phase}`); return retainFailure(p, error); },
+  });
+  expect(retained).toEqual(["fixture.first:completion"]);
+  expect(state.failures[0]).toMatchObject({ scenarioId: "fixture.first", phase: "completion", summary: "Synthetic result source digest differs from reserved digest" });
+  expect(state.stop?.detail).toContain("failure.json"); expect(f.ledger().scenarios[2]!.attempts).toEqual([]); await c.close();
+});

tests/runner-e2e/chat-campaign-coordinator.ts

@@ -0,0 +1,280 @@
+import { createHash, randomUUID } from "node:crypto";
+import { closeSync, existsSync, fsyncSync, mkdirSync, openSync, readFileSync, realpathSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
+import path from "node:path";
+import { campaignProductIdentityEligible, completeCampaignAttempt, reserveCampaignAttempt, type CampaignAttempt, type ChatCampaign } from "./chat-campaign.js";
+
+type Completion = Parameters<typeof completeCampaignAttempt>[1];
+export type CoordinatorStopReason = "cleanup_failure" | "secret_leak" | "shared_infrastructure" | "coordinator_failure";
+export interface CampaignQualificationReview {
+  evidencePath: string; receiptSha256: string;
+  previous: { definitionDigest: string; graderDigest: string };
+  next: { definitionDigest: string; graderDigest: string };
+}
+export interface ReviewedCampaignSelection { scenarioId: string; attempt: CampaignAttempt }
+export interface CampaignLaunchPermit {
+  reservationId: string; scenarioId: string; ordinal: 1 | 2; executionId: string;
+  campaignId: string; args: readonly string[]; maxParallel: 1; automaticRetries: 0;
+}
+interface Entry {
+  permit: CampaignLaunchPermit; reservationDigest: string; reservedAttempt: CampaignAttempt;
+  phase: "reserved" | "launch_uncertain" | "completed";
+  completionDigest?: string;
+}
+interface State {
+  version: 1; selectionDigest: string; entries: Record<string, Entry>;
+  stop: { reason: CoordinatorStopReason; detail: string } | null;
+  failures: Array<{ scenarioId: string; phase: "reserve" | "launch" | "completion"; evidencePath: string; summary: string }>;
+}
+interface Pending { before: string; after: string; manifest: string; state: State }
+export interface CampaignCoordinatorOptions {
+  manifestPath: string; controlDirectory: string; batchId: string;
+  /** Defaults to the repository root above tests/runner-e2e. */
+  evidenceRoot?: string;
+  selections: readonly ReviewedCampaignSelection[];
+  /** Controlled filesystem fault points for offline crash tests only. */
+  now?: () => string;
+  fault?: (stage: "intent" | "manifest" | "state") => void;
+}
+const hash = (value: string) => createHash("sha256").update(value).digest("hex");
+const json = (value: unknown) => `${JSON.stringify(value, null, 2)}\n`;
+function reviewedIdentity(selections: readonly ReviewedCampaignSelection[]) {
+  return selections.map(({ scenarioId, attempt }) => ({ scenarioId, attempt: { ...attempt, startedAt: null } }));
+}
+function read<T>(filename: string): T { return JSON.parse(readFileSync(filename, "utf8")); }
+function atomicWrite(filename: string, bytes: string) {
+  const temporary = `${filename}.${randomUUID()}.tmp`;
+  const fd = openSync(temporary, "wx", 0o600);
+  try { writeFileSync(fd, bytes); fsyncSync(fd); } finally { closeSync(fd); }
+  renameSync(temporary, filename);
+  const directory = openSync(path.dirname(filename), "r");
+  try { fsyncSync(directory); } finally { closeSync(directory); }
+}
+
+/** One ledger owner, two independent one-case launchers. Never steals a stale lock.
+ * A crashed owner requires independent process/resource reconciliation before an
+ * operator removes its lock. Restart never replays a launch, even if no PID was saved.
+ * Reconciliation/manual ledger writers must remain paused for this owner's lifetime.
+ */
+export class ChatCampaignCoordinator {
+  private tail: Promise<unknown> = Promise.resolve();
+  private readonly plans: ReviewedCampaignSelection[];
+  private readonly statePath: string;
+  private readonly intentPath: string;
+  private readonly lockPath: string;
+  private readonly owner = randomUUID();
+  private poisoned = false;
+  private closed = false;
+  private inFlight = 0;
+  private constructor(private readonly options: CampaignCoordinatorOptions) {
+    this.plans = structuredClone([...options.selections]);
+    this.statePath = path.join(options.controlDirectory, "state.json");
+    this.intentPath = path.join(options.controlDirectory, "pending.json");
+    // Lock belongs to the manifest, not a caller-chosen batch directory.
+    this.lockPath = `${options.manifestPath}.coordinator-lock`;
+  }
+  static open(options: CampaignCoordinatorOptions) {
+    if (!/^[a-zA-Z0-9_-]{1,100}$/.test(options.batchId) || !options.selections.length
+      || new Set(options.selections.map(x => x.scenarioId)).size !== options.selections.length)
+      throw new Error("Invalid reviewed campaign selection");
+    for (const { scenarioId, attempt } of options.selections) {
+      if (!/^[a-z0-9-]+\.[a-z0-9-]+$/.test(scenarioId) || attempt.environment !== "local"
+        || attempt.automaticRetries !== 0 || attempt.outcome !== "running" || !attempt.productIdentity
+        || attempt.productIdentity.kind !== "source-build" || !attempt.databaseLane)
+        throw new Error("Two-lane coordinator requires reviewed local source-build provenance");
+    }
+    mkdirSync(options.controlDirectory, { recursive: true, mode: 0o700 });
+    options = { ...options, manifestPath: realpathSync(options.manifestPath), controlDirectory: realpathSync(options.controlDirectory) };
+    const coordinator = new ChatCampaignCoordinator(options);
+    const fd = openSync(coordinator.lockPath, "wx", 0o600);
+    try { writeFileSync(fd, json({ version: 1, owner: coordinator.owner, pid: process.pid,
+      batchId: options.batchId, controlDirectory: path.resolve(options.controlDirectory), createdAt: new Date().toISOString() })); fsyncSync(fd); }
+    finally { closeSync(fd); }
+    try {
+      const selectionDigest = hash(json({ batchId: options.batchId, manifestPath: path.resolve(options.manifestPath), selections: reviewedIdentity(coordinator.plans) }));
+      if (!existsSync(coordinator.statePath)) atomicWrite(coordinator.statePath, json({ version: 1, selectionDigest, entries: {}, stop: null, failures: [] } satisfies State));
+      const state = read<State>(coordinator.statePath);
+      if (state.version !== 1 || state.selectionDigest !== selectionDigest) throw new Error("Coordinator selection changed");
+      coordinator.recover();
+      return coordinator;
+    } catch (error) {
+      // Retain ownership on recovery failure: no second writer may guess what happened.
+      coordinator.poisoned = true;
+      throw error;
+    }
+  }
+  private assertOwner() {
+    if (this.closed || this.poisoned || read<{ owner: string }>(this.lockPath).owner !== this.owner)
+      throw new Error("Coordinator ownership unavailable");
+  }
+  private serial<T>(operation: () => T): Promise<T> {
+    const result = this.tail.then(() => { this.assertOwner(); return operation(); });
+    this.tail = result.catch(() => undefined);
+    return result;
+  }
+  private recover() {
+    let state = read<State>(this.statePath);
+    if (existsSync(this.intentPath)) {
+      const pending = read<Pending>(this.intentPath), actual = hash(readFileSync(this.options.manifestPath, "utf8"));
+      if (actual !== pending.before && actual !== pending.after) throw new Error("Unrecognized manifest after interrupted transaction");
+      // Never replay an uncommitted reservation or completion on restart. If the
+      // manifest commit landed, recover its corresponding state, then stop.
+      if (actual === pending.after) state = pending.state;
+      state.stop ??= { reason: "coordinator_failure", detail: "Interrupted ledger transaction; independent reconciliation required" };
+      atomicWrite(this.statePath, json(state)); unlinkSync(this.intentPath);
+    }
+    if (Object.values(state.entries).some(x => x.phase !== "completed")) {
+      state.stop ??= { reason: "coordinator_failure", detail: "Reserved or possibly launched execution; no automatic recovery launch" };
+      atomicWrite(this.statePath, json(state));
+    }
+  }
+  private commit(manifestBefore: string, manifest: ChatCampaign, state: State) {
+    const bytes = json(manifest);
+    try {
+      atomicWrite(this.intentPath, json({ before: hash(manifestBefore), after: hash(bytes), manifest: bytes, state } satisfies Pending));
+      this.options.fault?.("intent");
+      if (hash(readFileSync(this.options.manifestPath, "utf8")) !== hash(manifestBefore)) throw new Error("Manifest changed outside coordinator");
+      atomicWrite(this.options.manifestPath, bytes); this.options.fault?.("manifest");
+      atomicWrite(this.statePath, json(state)); this.options.fault?.("state");
+      unlinkSync(this.intentPath);
+    } catch (error) { this.poisoned = true; throw error; }
+  }
+  reserve(scenarioId: string): Promise<CampaignLaunchPermit> {
+    return this.serial(() => {
+      const state = read<State>(this.statePath), plan = this.plans.find(x => x.scenarioId === scenarioId);
+      if (!plan) throw new Error("Scenario not reviewed");
+      if (state.stop) throw new Error("Admissions stopped");
+      const reservationId = hash(json({ batchId: this.options.batchId, scenarioId, ordinal: plan.attempt.ordinal }));
+      const reservationDigest = hash(json(reviewedIdentity([plan])));
+      if (state.entries[reservationId]) {
+        if (state.entries[reservationId]!.reservationDigest !== reservationDigest) throw new Error("Reservation identity conflict");
+        return structuredClone(state.entries[reservationId]!.permit);
+      }
+      if (Object.values(state.entries).filter(x => x.phase !== "completed").length >= 2) throw new Error("Two lanes already reserved");
+      const before = readFileSync(this.options.manifestPath, "utf8"), manifest = JSON.parse(before) as ChatCampaign;
+      const row = manifest.scenarios.find(x => x.id === scenarioId);
+      if (!row || row.executionId !== plan.attempt.executionId || row.qualifyingDefinitionDigest !== plan.attempt.definitionDigest
+        || row.qualifyingGraderDigest !== plan.attempt.graderDigest
+        || !campaignProductIdentityEligible(row, plan.attempt, manifest.baselineRevision)) throw new Error("Reviewed selection differs from authoritative ledger");
+      const reservedAttempt = { ...structuredClone(plan.attempt), startedAt: this.options.now?.() ?? new Date().toISOString() };
+      if (!Number.isFinite(Date.parse(reservedAttempt.startedAt))) throw new Error("Invalid reservation clock");
+      reserveCampaignAttempt(row, reservedAttempt);
+      const permit: CampaignLaunchPermit = { reservationId, scenarioId, ordinal: plan.attempt.ordinal, executionId: plan.attempt.executionId,
+        campaignId: `${this.options.batchId}-${reservationId.slice(0, 20)}`, maxParallel: 1, automaticRetries: 0,
+        args: ["--id", plan.attempt.executionId, "--max-parallel", "1", "--max-automatic-retries", "0"] };
+      state.entries[reservationId] = { permit, reservationDigest, reservedAttempt: structuredClone(reservedAttempt), phase: "reserved" };
+      this.commit(before, manifest, state);
+      return structuredClone(permit);
+    });
+  }
+  async launchOnce<T>(permit: CampaignLaunchPermit, launchOnly: (permit: CampaignLaunchPermit) => Promise<T>): Promise<T> {
+    await this.serial(() => {
+      const state = read<State>(this.statePath), entry = state.entries[permit.reservationId];
+      if (state.stop || !entry || entry.phase !== "reserved" || json(entry.permit) !== json(permit)) throw new Error("Launch is not admissible");
+      // Persist uncertainty BEFORE calling spawn. A crash in this gap consumes
+      // the permit; it can never be replayed based on absent PID alone.
+      entry.phase = "launch_uncertain"; atomicWrite(this.statePath, json(state)); this.inFlight++;
+    });
+    try { return await launchOnly(structuredClone(permit)); }
+    finally { this.inFlight--; }
+  }
+  complete(permit: CampaignLaunchPermit, completion: Completion, stop?: { reason: CoordinatorStopReason; detail: string }, qualificationReview?: CampaignQualificationReview): Promise<void> {
+    return this.serial(() => {
+      const state = read<State>(this.statePath), entry = state.entries[permit.reservationId];
+      if (!entry || json(entry.permit) !== json(permit)) throw new Error("Completion identity conflict");
+      const completionDigest = hash(json({ completion, stop: stop ?? null, qualificationReview: qualificationReview ?? null }));
+      if (entry.phase === "completed") {
+        if (entry.completionDigest !== completionDigest) throw new Error("Completion content conflict");
+        const current = read<ChatCampaign>(this.options.manifestPath).scenarios.find(x => x.id === permit.scenarioId)?.attempts.at(-1);
+        if (!current || current.ordinal !== permit.ordinal || current.executionId !== permit.executionId
+          || Object.entries(completion).some(([key, value]) => json(current[key as keyof CampaignAttempt]) !== json(value)))
+          throw new Error("Completed ledger differs from receipt");
+        return;
+      }
+      if (entry.phase !== "launch_uncertain") throw new Error("No admitted launcher for completion");
+      const before = readFileSync(this.options.manifestPath, "utf8"), manifest = JSON.parse(before) as ChatCampaign;
+      const row = manifest.scenarios.find(x => x.id === permit.scenarioId), plan = this.plans.find(x => x.scenarioId === permit.scenarioId)!;
+      const actual = row?.attempts.at(-1);
+      if (!row || !actual || actual.ordinal !== permit.ordinal || json(actual) !== json(entry.reservedAttempt)) throw new Error("Reserved attempt changed");
+      completeCampaignAttempt(row, structuredClone(completion));
+      if (qualificationReview) {
+        const review = qualificationReview;
+        if (completion.outcome !== "passed" || !completion.evidencePaths.includes(review.evidencePath)
+          || !review.evidencePath.startsWith("tests/runner-e2e/results/") || review.evidencePath.split("/").includes("..")
+          || review.previous.definitionDigest !== plan.attempt.definitionDigest || review.previous.graderDigest !== plan.attempt.graderDigest
+          || json(review.previous) === json(review.next) || !review.next.definitionDigest || !review.next.graderDigest)
+          throw new Error("Invalid qualifying identity review");
+        const bytes = readFileSync(path.join(this.options.evidenceRoot ?? path.resolve(path.dirname(this.options.manifestPath), "../.."), review.evidencePath), "utf8");
+        const retained = JSON.parse(bytes);
+        if (hash(bytes) !== review.receiptSha256 || retained.id !== permit.scenarioId || retained.ordinal !== permit.ordinal
+          || json(retained.previous) !== json(review.previous) || json(retained.next) !== json(review.next))
+          throw new Error("Qualifying identity review receipt mismatch");
+        row.qualifyingDefinitionDigest = null; row.qualifyingGraderDigest = null;
+      }
+      state.stop ??= stop ?? (completion.cleanup !== "passed" ? { reason: "cleanup_failure", detail: "Measured cleanup did not pass" } : null);
+      // Stop is durable before the completion/summary commit or any lane refill.
+      if (state.stop) atomicWrite(this.statePath, json(state));
+      entry.phase = "completed"; entry.completionDigest = completionDigest;
+      this.commit(before, manifest, state);
+    });
+  }
+  stop(reason: CoordinatorStopReason, detail: string): Promise<void> {
+    return this.serial(() => { const state = read<State>(this.statePath); state.stop ??= { reason, detail }; atomicWrite(this.statePath, json(state)); });
+  }
+  snapshot(): Promise<Readonly<State>> { return this.serial(() => read<State>(this.statePath)); }
+  /** Finite waves avoid refilling while an already settled peer awaits grading.
+   * readCompletion owns authentic result/usage and normalized selected-closure
+   * fingerprint validation (not unrelated raw descriptor/manifest bytes). It must
+   * distinguish actual secret matches from safe, bound-only ZIP omissions.
+   */
+  async run<T>(input: {
+    launchOnly: (permit: CampaignLaunchPermit) => Promise<T>;
+    /** Parent redacts and retains the real failure privately before returning a safe reference. */
+    retainFailure: (permit: CampaignLaunchPermit | null, error: unknown, phase: "reserve" | "launch" | "completion") => Promise<{ evidencePath: string; summary: string }>;
+    readCompletion: (permit: CampaignLaunchPermit, result: T) => Promise<{ completion: Completion; stop?: { reason: CoordinatorStopReason; detail: string }; qualificationReview?: CampaignQualificationReview }>;
+  }) {
+    const unretained: unknown[] = [];
+    const existing = await this.snapshot();
+    const remaining = this.plans.filter(plan => !Object.values(existing.entries).some(entry => entry.permit.scenarioId === plan.scenarioId && entry.phase === "completed"));
+    for (let index = 0; index < remaining.length; index += 2) {
+      if ((await this.snapshot()).stop) break;
+      const settled = await Promise.allSettled(remaining.slice(index, index + 2).map(async plan => {
+        let permit: CampaignLaunchPermit | null = null;
+        let phase: "reserve" | "launch" | "completion" = "reserve";
+        try {
+          permit = await this.reserve(plan.scenarioId); phase = "launch";
+          const result = await this.launchOnce(permit, input.launchOnly); phase = "completion";
+          const observed = await input.readCompletion(permit, result);
+          await this.complete(permit, observed.completion, observed.stop, observed.qualificationReview);
+        } catch (error) {
+          try {
+            const retained = await input.retainFailure(permit, error, phase);
+            if (!retained.evidencePath.startsWith("tests/runner-e2e/results/") || retained.evidencePath.split("/").includes("..") || !retained.summary.trim())
+              throw new Error("Failure requires a sanitized retained evidence reference");
+            if (!this.poisoned) await this.serial(() => {
+              const state = read<State>(this.statePath);
+              state.failures.push({ scenarioId: plan.scenarioId, phase, ...retained });
+              state.stop ??= { reason: "coordinator_failure", detail: `${retained.summary} (${retained.evidencePath})` };
+              atomicWrite(this.statePath, json(state));
+            });
+            else unretained.push(error); // WAL/lock remain authoritative; expose original error to owner.
+          } catch (retentionError) {
+            unretained.push(error, retentionError);
+            if (!this.poisoned) await this.stop("coordinator_failure", "Failure evidence retention failed; inspect returned errors");
+          }
+        }
+      }));
+      for (const result of settled) if (result.status === "rejected") unretained.push(result.reason);
+      if (unretained.length) throw new AggregateError(unretained, "Coordinator failure requires independent reconciliation; no further admission");
+      if (this.poisoned) throw new Error("Coordinator ledger transaction interrupted");
+      if ((await this.snapshot()).stop) break;
+    }
+    return this.snapshot();
+  }
+  close(): Promise<void> {
+    return this.serial(() => {
+      if (this.inFlight) throw new Error("Cannot release coordinator with launchers in flight");
+      if (existsSync(this.intentPath)) throw new Error("Cannot release interrupted transaction");
+      unlinkSync(this.lockPath); this.closed = true;
+    });
+  }
+}