import fs from "node:fs"; import path from "node:path"; import { spawnSync } from "node:child_process"; import { fileURLToPath } from "node:url"; 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 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 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 { writeCheckpoint({ state: "blocked", iteration, reason: source, error: error?.message ?? String(error) }); appendLog(`blocked reason=${source} error=${error?.message ?? String(error)}`); } finally { 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) { fs.writeFileSync(abs(checkpointPath), `${JSON.stringify({ schemaVersion: 1, pid: process.pid, updatedAt: new Date().toISOString(), ...payload }, null, 2)}\n`); } 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 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(); const print = run(process.execPath, ["tools/web/print-task-context.mjs"]); const check = run(process.execPath, ["tools/web/check-task-context.mjs"]); appendLog(`preflight task=${before.task} printExit=${print.code} checkExit=${check.code}`); if (print.code !== 0 || check.code !== 0) { writeCheckpoint({ state: "blocked", iteration, current: before, reason: "preflight_failed", printExit: print.code, checkExit: check.code }); exitCode = 2; break; } 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"; writeCheckpoint({ state: "blocked", iteration: iteration + 1, before, after, reason, childExit: child.code }); exitCode = 3; break; } if (after.task === before.task) { 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) { const current = currentTask(); writeCheckpoint({ state: maxIterations === 0 ? "running" : "complete", iteration, current }); appendLog(`stopped state=${maxIterations === 0 ? "running" : "complete"} iteration=${iteration} current=${current.task}`); } } finally { releaseLock(); } process.exitCode = exitCode;