203 lines
8.7 KiB
JavaScript
203 lines
8.7 KiB
JavaScript
import fs from "node:fs";
|
|
import path from "node:path";
|
|
import { spawnSync } from "node:child_process";
|
|
import { fileURLToPath } from "node:url";
|
|
import { createFailureLedgerSession } from "./wbv2-failure-ledger.mjs";
|
|
|
|
const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../..");
|
|
const abs = (file) => path.join(root, file);
|
|
const supervisorRoot = "tests/golden/WBV2";
|
|
const supervisorLog = `${supervisorRoot}/continuous-supervisor.log`;
|
|
const checkpointPath = `${supervisorRoot}/continuous-supervisor.json`;
|
|
const lockPath = `${supervisorRoot}/continuous-supervisor.lock`;
|
|
const failureLedger = createFailureLedgerSession(root, { logPaths: [`${supervisorRoot}/continuous-supervisor.log`] });
|
|
|
|
const valueArg = (name, fallback) => {
|
|
const index = process.argv.indexOf(name);
|
|
return index >= 0 ? process.argv[index + 1] : fallback;
|
|
};
|
|
const maxIterations = Number(valueArg("--max-iterations", "0"));
|
|
const pollMs = Number(valueArg("--poll-ms", "1000"));
|
|
const pushRetries = Number(valueArg("--push-retries", "6"));
|
|
const commitEnabled = process.argv.includes("--commit");
|
|
const pushEnabled = process.argv.includes("--push");
|
|
if (!Number.isSafeInteger(maxIterations) || maxIterations < 0) throw new Error("--max-iterations must be a non-negative integer; 0 means unlimited");
|
|
if (!Number.isSafeInteger(pollMs) || pollMs < 0) throw new Error("--poll-ms must be a non-negative integer");
|
|
if (!Number.isSafeInteger(pushRetries) || pushRetries < 0) throw new Error("--push-retries must be a non-negative integer");
|
|
if (pushEnabled && !commitEnabled) throw new Error("--push requires --commit");
|
|
|
|
fs.mkdirSync(abs(supervisorRoot), { recursive: true });
|
|
let lockHandle;
|
|
const acquireLock = () => {
|
|
try {
|
|
return fs.openSync(abs(lockPath), "wx");
|
|
}
|
|
catch (error) {
|
|
if (error?.code !== "EEXIST") throw error;
|
|
try {
|
|
const lock = JSON.parse(fs.readFileSync(abs(lockPath), "utf8"));
|
|
if (Number.isSafeInteger(lock.pid)) {
|
|
try { process.kill(lock.pid, 0); }
|
|
catch (probeError) {
|
|
if (probeError?.code === "ESRCH") {
|
|
fs.unlinkSync(abs(lockPath));
|
|
return fs.openSync(abs(lockPath), "wx");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
catch {
|
|
// A live or unreadable lock is treated as owned to avoid concurrent writes.
|
|
}
|
|
throw new Error(`another WBV2 supervisor appears to be running; lock=${lockPath}`);
|
|
}
|
|
};
|
|
lockHandle = acquireLock();
|
|
fs.writeFileSync(lockHandle, `${JSON.stringify({ pid: process.pid, startedAt: new Date().toISOString() })}\n`);
|
|
|
|
let lockReleased = false;
|
|
const releaseLock = () => {
|
|
if (lockReleased || lockHandle === undefined) return;
|
|
lockReleased = true;
|
|
fs.closeSync(lockHandle);
|
|
try { fs.unlinkSync(abs(lockPath)); }
|
|
catch (error) { if (error?.code !== "ENOENT") throw error; }
|
|
};
|
|
let stopRequested = false;
|
|
let stopSignal;
|
|
let iteration = 0;
|
|
let terminalCompleted = false;
|
|
let fatalHandled = false;
|
|
const stopFromSignal = (signal) => {
|
|
stopRequested = true;
|
|
stopSignal = signal;
|
|
appendLog(`stop-requested signal=${signal}`);
|
|
process.exitCode = signal === "SIGINT" ? 130 : 143;
|
|
};
|
|
process.once("SIGINT", () => stopFromSignal("SIGINT"));
|
|
process.once("SIGTERM", () => stopFromSignal("SIGTERM"));
|
|
|
|
const stopOnFatal = (error, source) => {
|
|
if (fatalHandled) return;
|
|
fatalHandled = true;
|
|
try {
|
|
failureLedger.record({ task: currentTask()?.task ?? "UNKNOWN", phase: "supervisor", reason: source, detail: error?.message ?? String(error) });
|
|
writeCheckpoint({ state: "blocked", iteration, reason: source, error: error?.message ?? String(error) });
|
|
appendLog(`blocked reason=${source} error=${error?.message ?? String(error)}`);
|
|
}
|
|
finally {
|
|
failureLedger.flush();
|
|
releaseLock();
|
|
process.exit(1);
|
|
}
|
|
};
|
|
process.once("uncaughtException", (error) => stopOnFatal(error, "uncaught_exception"));
|
|
process.once("unhandledRejection", (error) => stopOnFatal(error, "unhandled_rejection"));
|
|
|
|
if (lockHandle === undefined) {
|
|
throw new Error(`unable to acquire supervisor lock=${lockPath}`);
|
|
}
|
|
|
|
function appendLog(text) {
|
|
fs.appendFileSync(abs(supervisorLog), `[${new Date().toISOString()}] ${text}\n`);
|
|
}
|
|
function writeCheckpoint(payload) {
|
|
const next = { schemaVersion: 1, pid: process.pid, updatedAt: new Date().toISOString(), ...payload };
|
|
const stable = (value) => { const { pid, updatedAt, ...rest } = value ?? {}; return rest; };
|
|
try {
|
|
const current = JSON.parse(fs.readFileSync(abs(checkpointPath), "utf8"));
|
|
if (JSON.stringify(stable(current)) === JSON.stringify(stable(next))) return;
|
|
}
|
|
catch { /* The first checkpoint or an interrupted write is not reusable. */ }
|
|
const temporaryPath = `${checkpointPath}.tmp-${process.pid}`;
|
|
fs.writeFileSync(abs(temporaryPath), `${JSON.stringify(next, null, 2)}\n`);
|
|
fs.renameSync(abs(temporaryPath), abs(checkpointPath));
|
|
}
|
|
const run = (command, args = []) => {
|
|
const result = spawnSync(command, args, { cwd: root, encoding: "utf8", maxBuffer: 16 * 1024 * 1024 });
|
|
return { code: result.status ?? 1, output: `${result.stdout ?? ""}${result.stderr ?? ""}` };
|
|
};
|
|
const currentTask = () => {
|
|
const queue = fs.readFileSync(abs("docs/EXECUTION_QUEUE.md"), "utf8");
|
|
const task = queue.match(/\| 当前任务 \| `(WBV2-P\d+-\d+)` \|/u)?.[1];
|
|
const parentManifest = queue.match(/\| parent manifest \| `([^`]+)` \|/u)?.[1];
|
|
if (!task || !parentManifest) throw new Error("queue pointer is incomplete");
|
|
return { task, parentManifest };
|
|
};
|
|
const terminalTask = (task) => {
|
|
try {
|
|
const index = JSON.parse(fs.readFileSync(abs("tests/golden/WBV2/task-index.json"), "utf8"));
|
|
const manifestPath = `tests/golden/${task}/manifest.json`;
|
|
const manifest = fs.existsSync(abs(manifestPath)) ? JSON.parse(fs.readFileSync(abs(manifestPath), "utf8")) : null;
|
|
return index.activeTask === null && manifest?.task === task && manifest.status === "done" && manifest.nextTask === null;
|
|
}
|
|
catch {
|
|
return false;
|
|
}
|
|
};
|
|
const childArgs = ["tools/web/run-wbv2-continuous.mjs", "--max-iterations", "1"];
|
|
if (commitEnabled) childArgs.push("--commit");
|
|
if (pushEnabled) childArgs.push("--push");
|
|
if (pushEnabled) childArgs.push("--push-retries", String(pushRetries));
|
|
|
|
appendLog(`started pid=${process.pid} maxIterations=${maxIterations} child=${childArgs.join(" ")}`);
|
|
writeCheckpoint({ state: "started", iteration: 0, current: currentTask(), child: childArgs.join(" ") });
|
|
|
|
let exitCode = 0;
|
|
try {
|
|
while (maxIterations === 0 || iteration < maxIterations) {
|
|
if (stopRequested) {
|
|
writeCheckpoint({ state: "stopped", iteration, current: currentTask(), reason: stopSignal });
|
|
break;
|
|
}
|
|
const before = currentTask();
|
|
writeCheckpoint({ state: "running", iteration: iteration + 1, before, child: childArgs.join(" ") });
|
|
const child = run(process.execPath, childArgs);
|
|
appendLog(`child task=${before.task} exit=${child.code}\n${child.output}`);
|
|
const after = currentTask();
|
|
if (stopRequested) {
|
|
writeCheckpoint({ state: "stopped", iteration, before, after, reason: stopSignal, childExit: child.code });
|
|
break;
|
|
}
|
|
if (child.code !== 0) {
|
|
const reason = after.task === before.task ? "child_failed_same_task" : "child_failed_after_queue_advance";
|
|
failureLedger.record({ task: before.task, phase: "supervisor", reason, childExit: child.code });
|
|
writeCheckpoint({ state: "blocked", iteration: iteration + 1, before, after, reason, childExit: child.code });
|
|
exitCode = 3;
|
|
break;
|
|
}
|
|
if (after.task === before.task && terminalTask(before.task)) {
|
|
writeCheckpoint({ state: "complete", iteration: iteration + 1, before, after, childExit: child.code, reason: "terminal_module_complete" });
|
|
terminalCompleted = true;
|
|
iteration += 1;
|
|
break;
|
|
}
|
|
if (after.task === before.task) {
|
|
failureLedger.record({ task: before.task, phase: "supervisor", reason: "child_exited_without_queue_advance", childExit: child.code });
|
|
writeCheckpoint({ state: "blocked", iteration: iteration + 1, before, after, reason: "child_exited_without_queue_advance", childExit: child.code });
|
|
exitCode = 4;
|
|
break;
|
|
}
|
|
iteration += 1;
|
|
writeCheckpoint({ state: "handed_off", iteration, before, after });
|
|
appendLog(`handoff iteration=${iteration} before=${before.task} after=${after.task}`);
|
|
if (pollMs > 0) {
|
|
const wait = new Int32Array(new SharedArrayBuffer(4));
|
|
Atomics.wait(wait, 0, 0, pollMs);
|
|
}
|
|
}
|
|
if (stopRequested) {
|
|
exitCode = stopSignal === "SIGINT" ? 130 : 143;
|
|
}
|
|
if (exitCode === 0 && !terminalCompleted) {
|
|
const current = currentTask();
|
|
writeCheckpoint({ state: maxIterations === 0 ? "running" : "complete", iteration, current });
|
|
appendLog(`stopped state=${maxIterations === 0 ? "running" : "complete"} iteration=${iteration} current=${current.task}`);
|
|
}
|
|
}
|
|
finally {
|
|
failureLedger.flush();
|
|
releaseLock();
|
|
}
|
|
process.exitCode = exitCode;
|