From 49e06605d88dcddfac7b727f7f4f2d928097727a Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 27 Oct 2026 04:42:56 -0500 Subject: [PATCH] Add serialized two-lane chat campaign coordinator --- .../chat-campaign-coordinator.test.ts | 255 ++++++++++++++++ tests/runner-e2e/chat-campaign-coordinator.ts | 280 ++++++++++++++++++ 2 files changed, 535 insertions(+) create mode 100644 tests/runner-e2e/chat-campaign-coordinator.test.ts create mode 100644 tests/runner-e2e/chat-campaign-coordinator.ts diff --git a/tests/runner-e2e/chat-campaign-coordinator.test.ts b/tests/runner-e2e/chat-campaign-coordinator.test.ts new file mode 100644 index 00000000000..5873713bb05 --- /dev/null +++ b/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, 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(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(); +}); diff --git a/tests/runner-e2e/chat-campaign-coordinator.ts b/tests/runner-e2e/chat-campaign-coordinator.ts new file mode 100644 index 00000000000..286090a0114 --- /dev/null +++ b/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[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; + 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(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 = 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(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(operation: () => T): Promise { + const result = this.tail.then(() => { this.assertOwner(); return operation(); }); + this.tail = result.catch(() => undefined); + return result; + } + private recover() { + let state = read(this.statePath); + if (existsSync(this.intentPath)) { + const pending = read(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 { + return this.serial(() => { + const state = read(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(permit: CampaignLaunchPermit, launchOnly: (permit: CampaignLaunchPermit) => Promise): Promise { + await this.serial(() => { + const state = read(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 { + return this.serial(() => { + const state = read(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(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 { + return this.serial(() => { const state = read(this.statePath); state.stop ??= { reason, detail }; atomicWrite(this.statePath, json(state)); }); + } + snapshot(): Promise> { return this.serial(() => read(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(input: { + launchOnly: (permit: CampaignLaunchPermit) => Promise; + /** 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(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 { + 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; + }); + } +}