diff --git a/my-tools/watch_blender_web_scheduler.py b/my-tools/watch_blender_web_scheduler.py index 98c0f7fc2..807ad1201 100755 --- a/my-tools/watch_blender_web_scheduler.py +++ b/my-tools/watch_blender_web_scheduler.py @@ -32,6 +32,14 @@ def now() -> str: def write_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) + try: + current = json.loads(path.read_text(encoding="utf-8")) + stable_current = {key: item for key, item in current.items() if key != "updatedAt"} + stable_next = {key: item for key, item in value.items() if key != "updatedAt"} + if stable_current == stable_next: + return + except (OSError, json.JSONDecodeError): + pass temporary = path.with_suffix(path.suffix + f".tmp-{os.getpid()}") temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") temporary.replace(path) diff --git a/tools/web/check-wbv2-context-cache.mjs b/tools/web/check-wbv2-context-cache.mjs new file mode 100644 index 000000000..a7b5ba1fe --- /dev/null +++ b/tools/web/check-wbv2-context-cache.mjs @@ -0,0 +1,44 @@ +import crypto from "node:crypto"; +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 cachePath = abs("tests/golden/WBV2/context-cache.json"); +const read = (file) => fs.readFileSync(abs(file), "utf8"); +const digest = (value) => crypto.createHash("sha256").update(value).digest("hex"); +const queue = read("docs/EXECUTION_QUEUE.md"); +const task = process.argv.includes("--task") + ? process.argv[process.argv.indexOf("--task") + 1] + : queue.match(/\| 当前任务 \| `([^`]+)` \|/u)?.[1]; +if (!task) throw new Error("unable to resolve current WBV2 task"); +const parentManifest = queue.match(/\| parent manifest \| `([^`]+)` \|/u)?.[1]; +if (!parentManifest) throw new Error("queue parent manifest is missing"); +const inputPaths = [ + "docs/EXECUTION_QUEUE.md", + `docs/tasks/${task}.md`, + parentManifest, + `tests/golden/${task}/task-context.json`, +]; +const inputHash = digest(inputPaths.map((file) => `${file}\0${digest(read(file))}\n`).join("")); +let cached = null; +try { cached = JSON.parse(fs.readFileSync(cachePath, "utf8")); } +catch { /* First use or an interrupted write. */ } +if (!process.argv.includes("--force") && cached?.status === "PASS" && cached.task === task && cached.inputHash === inputHash) { + process.stdout.write(`wbv2-context-cache-hit task=${task} inputHash=${inputHash}\n`); + process.exit(0); +} +const result = spawnSync(process.execPath, [abs("tools/web/print-task-context.mjs"), "--task", task], { cwd: root, encoding: "utf8", maxBuffer: 8 * 1024 * 1024 }); +const check = spawnSync(process.execPath, [abs("tools/web/check-task-context.mjs"), "--task", task], { cwd: root, encoding: "utf8", maxBuffer: 8 * 1024 * 1024 }); +if ((result.status ?? 1) !== 0 || (check.status ?? 1) !== 0) { + process.stderr.write(`${result.stdout ?? ""}${result.stderr ?? ""}${check.stdout ?? ""}${check.stderr ?? ""}`); + process.exit(Math.max(result.status ?? 1, check.status ?? 1)); +} +fs.mkdirSync(path.dirname(cachePath), { recursive: true }); +const payload = { schemaVersion: 1, status: "PASS", task, inputHash, refreshedAt: new Date().toISOString() }; +const temporary = `${cachePath}.tmp-${process.pid}`; +fs.writeFileSync(temporary, `${JSON.stringify(payload, null, 2)}\n`); +fs.renameSync(temporary, cachePath); +process.stdout.write(`wbv2-context-cache-refresh task=${task} inputHash=${inputHash}\n`); diff --git a/tools/web/run-wbv2-continuous.mjs b/tools/web/run-wbv2-continuous.mjs index aac21369e..ac436a746 100644 --- a/tools/web/run-wbv2-continuous.mjs +++ b/tools/web/run-wbv2-continuous.mjs @@ -59,7 +59,7 @@ function taskCommand(task) { } function assertContext() { - const result = command("node tools/web/print-task-context.mjs && node tools/web/check-task-context.mjs", "tests/golden/WBV2/continuous-context.log"); + const result = command("node tools/web/check-wbv2-context-cache.mjs", "tests/golden/WBV2/continuous-context.log"); if (result.code !== 0) throw new Error(`context gate failed (exit ${result.code})`); } diff --git a/tools/web/run-wbv2-guardian.mjs b/tools/web/run-wbv2-guardian.mjs index 9e55ca7e5..094c1d779 100644 --- a/tools/web/run-wbv2-guardian.mjs +++ b/tools/web/run-wbv2-guardian.mjs @@ -2,12 +2,14 @@ import fs from "node:fs"; import path from "node:path"; import { spawn } from "node:child_process"; import { fileURLToPath } from "node:url"; +import { recordFailure } from "./wbv2-failure-ledger.mjs"; const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../.."); const abs = (file) => path.join(root, file); const controlRoot = "tests/golden/WBV2"; const logPath = `${controlRoot}/continuous-guardian.log`; const checkpointPath = `${controlRoot}/continuous-guardian.json`; +const supervisorCheckpointPath = `${controlRoot}/continuous-supervisor.json`; const lockPath = `${controlRoot}/continuous-guardian.lock`; const argv = process.argv.slice(2); @@ -73,8 +75,15 @@ const releaseLock = () => { }; const appendLog = (text) => fs.appendFileSync(abs(logPath), `[${new Date().toISOString()}] ${text}\n`); const 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({ schemaVersion: 1, pid: process.pid, updatedAt: new Date().toISOString(), ...payload }, null, 2)}\n`); + fs.writeFileSync(abs(temporaryPath), `${JSON.stringify(next, null, 2)}\n`); fs.renameSync(abs(temporaryPath), abs(checkpointPath)); }; const readJson = (file) => { @@ -94,23 +103,6 @@ const sleep = (milliseconds) => new Promise((resolve) => setTimeout(resolve, mil let child; let stopRequested = false; let stopSignal; -let heartbeatTimer; -const writeHeartbeat = () => { - if (stopRequested || released) return; - try { - writeCheckpoint({ - state: "running", - restartCount, - current: currentTask(), - supervisorPid: child?.pid ?? null, - heartbeatAt: new Date().toISOString(), - supervisor: supervisorCommand, - }); - } - catch (error) { - appendLog(`heartbeat-failed error=${error?.message ?? String(error)}`); - } -}; const signalChild = (signal) => { if (!child?.pid) return; try { process.kill(-child.pid, signal); } @@ -158,7 +150,6 @@ let restartCount = 0; try { appendLog(`started pid=${process.pid} supervisor=${supervisorCommand} maxRestarts=${maxRestarts} restartDelayMs=${restartDelayMs}`); writeCheckpoint({ state: "started", restartCount, supervisor: supervisorCommand, current: currentTask() }); - heartbeatTimer = setInterval(writeHeartbeat, 2000); while (true) { if (stopRequested) { writeCheckpoint({ state: "stopped", restartCount, reason: stopSignal, current: currentTask() }); @@ -168,7 +159,7 @@ try { const before = currentTask(); writeCheckpoint({ state: "running", restartCount, before, supervisor: supervisorCommand }); const result = await runSupervisor(); - const checkpoint = readJson(checkpointPath); + const checkpoint = readJson(supervisorCheckpointPath); const after = currentTask(); appendLog(`supervisor-exit code=${result.code} signal=${result.signal ?? "none"} task=${before.task} queue=${after.task} state=${checkpoint?.state ?? "missing"}`); if (stopRequested) { @@ -182,6 +173,7 @@ try { } const decision = safeRestart(checkpoint, after); if (!decision.ok) { + recordFailure(root, { task: before.task, phase: "guardian", reason: decision.reason, childExit: result.code }); writeCheckpoint({ state: "blocked", restartCount, before, after, reason: decision.reason, supervisorExit: result.code }); appendLog(`blocked reason=${decision.reason}; no automatic replay`); guardianExit = 2; @@ -189,6 +181,7 @@ try { } restartCount += 1; if (maxRestarts > 0 && restartCount > maxRestarts) { + recordFailure(root, { task: before.task, phase: "guardian", reason: "max_safe_restarts_exceeded", childExit: result.code }); writeCheckpoint({ state: "blocked", restartCount, before, after, reason: "max_safe_restarts_exceeded", supervisorExit: result.code }); appendLog("blocked reason=max_safe_restarts_exceeded"); guardianExit = 3; @@ -201,11 +194,12 @@ try { } catch (error) { guardianExit = 4; + try { recordFailure(root, { task: currentTask()?.task ?? "UNKNOWN", phase: "guardian", reason: "guardian_error", detail: error?.message ?? String(error) }); } + catch { /* Preserve the original guardian failure. */ } writeCheckpoint({ state: "blocked", restartCount, reason: "guardian_error", error: error?.message ?? String(error) }); appendLog(`blocked reason=guardian_error error=${error?.message ?? String(error)}`); } finally { - if (heartbeatTimer !== undefined) clearInterval(heartbeatTimer); signalChild("SIGTERM"); releaseLock(); } diff --git a/tools/web/run-wbv2-scheduler.mjs b/tools/web/run-wbv2-scheduler.mjs new file mode 100644 index 000000000..aa612f619 --- /dev/null +++ b/tools/web/run-wbv2-scheduler.mjs @@ -0,0 +1,310 @@ +import fs from "node:fs"; +import path from "node:path"; +import { spawnSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { recordFailure } from "./wbv2-failure-ledger.mjs"; + +const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../.."); +const controlRoot = "tests/golden/WBV2"; +const checkpointPath = `${controlRoot}/continuous-scheduler.json`; +const lockPath = `${controlRoot}/continuous-scheduler.lock`; +const logPath = `${controlRoot}/continuous-scheduler.log`; +const abs = (file) => path.join(root, file); + +const argValue = (name, fallback) => { + const index = process.argv.indexOf(name); + return index >= 0 ? process.argv[index + 1] : fallback; +}; +const maxIterations = Number(argValue("--max-iterations", "0")); +const pollMs = Number(argValue("--poll-ms", "1000")); +const pushRetries = Number(argValue("--push-retries", "6")); +const commitEnabled = process.argv.includes("--commit"); +const pushEnabled = process.argv.includes("--push"); +const preflightOnly = process.argv.includes("--preflight-only"); +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(controlRoot), { 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 { + // An unreadable lock is treated as owned to prevent concurrent writers. + } + throw new Error(`another WBV2 scheduler appears to be running; lock=${lockPath}`); + } +}; + +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(read(checkpointPath)); + 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)); +} + +function appendLog(message) { + fs.appendFileSync(abs(logPath), `[${new Date().toISOString()}] ${message}\n`); +} + +function read(file) { + return fs.readFileSync(abs(file), "utf8"); +} + +function currentTask() { + const queue = read("docs/EXECUTION_QUEUE.md"); + const task = queue.match(/\| 当前任务 \| `(WBV2-P\d+-\d+)` \|/u)?.[1]; + const parentManifest = queue.match(/\| parent manifest \| `([^`]+)` \|/u)?.[1]; + if (!task && /队列.*(闭合|完成)|queue.*(closed|complete)/iu.test(queue)) return null; + if (!task || !parentManifest) throw new Error("queue pointer is incomplete"); + return { task, parentManifest }; +} +function terminalTask(task) { + try { + const index = JSON.parse(read("tests/golden/WBV2/task-index.json")); + const manifestPath = `tests/golden/${task}/manifest.json`; + const manifest = fs.existsSync(abs(manifestPath)) ? JSON.parse(read(manifestPath)) : null; + return index.activeTask === null && manifest?.task === task && manifest.status === "done" && manifest.nextTask === null; + } + catch { + return false; + } +} + +function parseTask(task) { + const match = task.match(/^WBV2-P(\d+)-(00[0-3])$/u); + if (!match) return { phase: null, index: null, taskClass: "UNKNOWN" }; + const index = Number(match[2]); + return { + phase: Number(match[1]), + index, + taskClass: index === 1 ? "QUEUE_ENABLEMENT" : index === 2 ? "SOURCE_IMPLEMENTATION" : index === 3 ? "INTEGRATION_BUILD" : "QUEUE_ENABLEMENT", + }; +} + +function parentManifest(pointer) { + const manifest = JSON.parse(read(pointer.parentManifest)); + if (manifest.status !== "done" || manifest.nextTask !== pointer.task) { + throw new Error(`parent manifest does not authorize task=${pointer.task}: ${pointer.parentManifest}`); + } + return manifest; +} + +function sourceMappingReadiness(pointer) { + const { phase, index, taskClass } = parseTask(pointer.task); + const cardPath = `docs/tasks/${pointer.task}.md`; + if (!fs.existsSync(abs(cardPath))) return { ok: false, reason: "task_card_missing", taskClass }; + const card = read(cardPath); + const implementation = card.match(/- Implementation entrypoint: `([^`]+)`/u)?.[1]; + if (!implementation) return { ok: false, reason: "implementation_entrypoint_missing", taskClass }; + + // Continuation source tasks must have an explicit, non-generic phase mapping. + // This prevents a long-running loop from replaying an older marker forever. + if (taskClass === "SOURCE_IMPLEMENTATION" && implementation.includes("generate-wbv2-capability-continuation-slice.mjs")) { + const generator = read("tools/web/generate-wbv2-capability-continuation-slice.mjs"); + const phaseBlock = generator.match(new RegExp(`(?:^|\\n)(?:else )?if \\(phase === ${phase}\\) \\{([\\s\\S]*?)(?=\\n(?:else )?if \\(phase ===|\\nelse if \\(phase >=|\\nfor \\(const file)`, "m"))?.[1]; + if (!phaseBlock) return { ok: false, reason: `source_mapping_missing_phase_${phase}`, taskClass, phase }; + const marker = phaseBlock.match(/markers\[2\] = "([^"]+)"/u)?.[1]; + if (!marker) return { ok: false, reason: `source_mapping_missing_marker_${phase}`, taskClass, phase }; + const genericMarkers = new Set([ + "capability-family-workbench", + "web_engine_blend_main_set_world_lightgroup", + "web_engine_blend_main_set_object_lightgroup", + "web_engine_blend_main_set_scene_view_layer_lightgroups", + "WEB_ENGINE_COMMAND_CAPABILITIES", + ]); + if (genericMarkers.has(marker)) return { ok: false, reason: `source_mapping_generic_marker_${marker}`, taskClass, phase, marker }; + const firstSource = phaseBlock.match(/files\[2\] = \[\s*"([^"]+)"/u)?.[1]; + if (!firstSource || !fs.existsSync(abs(firstSource))) return { ok: false, reason: "source_mapping_first_file_missing", taskClass, phase, marker, firstSource }; + if (!read(firstSource).includes(marker)) return { ok: false, reason: "source_behavior_marker_missing", taskClass, phase, marker, firstSource }; + return { ok: true, taskClass, phase, marker, firstSource, implementation }; + } + + if (index === 3) { + const parent = parentManifest(pointer); + if (parent.taskClass !== "SOURCE_IMPLEMENTATION" || parent.assertions?.sourceBacked !== true) { + return { ok: false, reason: "integration_parent_not_source_backed", taskClass, parentTask: parent.task }; + } + } + return { ok: true, taskClass, implementation }; +} + +function runOne(pointer) { + const outputPath = `tests/golden/${pointer.task}/scheduler-child.log`; + fs.mkdirSync(abs(`tests/golden/${pointer.task}`), { recursive: true }); + const args = ["tools/web/run-wbv2-guardian.mjs", "--max-iterations", "1", "--max-restarts", "0"]; + if (commitEnabled) args.push("--commit"); + if (pushEnabled) args.push("--push", "--push-retries", String(pushRetries)); + const result = spawnSync(process.execPath, args, { + cwd: root, + encoding: "utf8", + maxBuffer: 32 * 1024 * 1024, + }); + const output = `${result.stdout ?? ""}${result.stderr ?? ""}`; + fs.writeFileSync(abs(outputPath), output); + return { code: result.status ?? 1, signal: result.signal, outputPath, args }; +} + +let stopRequested = false; +let stopSignal; +process.once("SIGINT", () => { stopRequested = true; stopSignal = "SIGINT"; }); +process.once("SIGTERM", () => { stopRequested = true; stopSignal = "SIGTERM"; }); + +lockHandle = acquireLock(); +fs.writeFileSync(lockHandle, `${JSON.stringify({ pid: process.pid, startedAt: new Date().toISOString() })}\n`); +let released = false; +const releaseLock = () => { + if (released) return; + released = true; + fs.closeSync(lockHandle); + try { fs.unlinkSync(abs(lockPath)); } + catch (error) { if (error?.code !== "ENOENT") throw error; } +}; + +let exitCode = 0; +let iteration = 0; +let completed = false; +try { + appendLog(`started pid=${process.pid} maxIterations=${maxIterations} commit=${commitEnabled} push=${pushEnabled}`); + writeCheckpoint({ state: "started", iteration, current: currentTask(), maxIterations, commit: commitEnabled, push: pushEnabled }); + while (maxIterations === 0 || iteration < maxIterations) { + if (stopRequested) { + writeCheckpoint({ state: "stopped", iteration, current: currentTask(), reason: stopSignal }); + exitCode = stopSignal === "SIGINT" ? 130 : 143; + break; + } + const before = currentTask(); + if (!before) { + writeCheckpoint({ state: "complete", iteration, current: null, reason: "queue_closed" }); + appendLog(`complete queue_closed iteration=${iteration}`); + completed = true; + break; + } + // A terminal V2 plan keeps the last completed task as an auditable queue + // pointer. Close cleanly instead of replaying that module through the + // guardian/supervisor chain. + try { + const index = JSON.parse(read("tests/golden/WBV2/task-index.json")); + if (index.activeTask === null && index.entries?.[before.task]) { + writeCheckpoint({ state: "complete", iteration, current: before, reason: "v2_plan_terminal" }); + appendLog(`complete v2_plan_terminal task=${before.task} iteration=${iteration}`); + completed = true; + break; + } + } + catch (error) { + recordFailure(root, { task: before.task, phase: "scheduler", reason: "terminal_plan_check_failed", detail: error?.message ?? String(error) }); + writeCheckpoint({ state: "blocked", iteration, before, reason: "terminal_plan_check_failed", error: error?.message ?? String(error) }); + appendLog(`blocked task=${before.task} reason=terminal_plan_check_failed error=${error?.message ?? String(error)}`); + process.stderr.write(`scheduler-blocked task=${before.task} reason=terminal_plan_check_failed\n`); + exitCode = 2; + break; + } + let readiness; + try { + readiness = sourceMappingReadiness(before); + } + catch (error) { + readiness = { ok: false, reason: "readiness_error", error: error?.message ?? String(error) }; + } + writeCheckpoint({ state: "preflight", iteration, before, readiness }); + appendLog(`preflight task=${before.task} class=${readiness.taskClass ?? "UNKNOWN"} ok=${readiness.ok} reason=${readiness.reason ?? "ready"}`); + if (!readiness.ok) { + recordFailure(root, { task: before.task, phase: "scheduler", reason: readiness.reason }); + writeCheckpoint({ state: "blocked", iteration, before, readiness }); + appendLog(`blocked task=${before.task} reason=${readiness.reason}`); + process.stderr.write(`scheduler-blocked task=${before.task} reason=${readiness.reason}\n`); + exitCode = 2; + break; + } + if (preflightOnly) { + writeCheckpoint({ state: "preflight_ready", iteration, before, readiness }); + process.stdout.write(`scheduler-preflight-ready task=${before.task} class=${readiness.taskClass}\n`); + break; + } + writeCheckpoint({ state: "running", iteration: iteration + 1, before, readiness }); + const child = runOne(before); + const after = currentTask(); + appendLog(`child task=${before.task} exit=${child.code} signal=${child.signal ?? "none"} after=${after?.task ?? "NONE"} log=${child.outputPath}`); + if (stopRequested) { + writeCheckpoint({ state: "stopped", iteration, before, after, reason: stopSignal, childExit: child.code }); + exitCode = stopSignal === "SIGINT" ? 130 : 143; + break; + } + if (child.code !== 0) { + const reason = after?.task !== before.task ? "child_failed_after_queue_advance" : "child_failed_same_task"; + recordFailure(root, { task: before.task, phase: "scheduler", reason, childExit: child.code }); + writeCheckpoint({ state: "blocked", iteration, before, after, reason, childExit: child.code, childLog: child.outputPath }); + appendLog(`blocked task=${before.task} reason=${reason}`); + process.stderr.write(`scheduler-blocked task=${before.task} reason=${reason} childExit=${child.code}\n`); + exitCode = 3; + break; + } + if (after && after.task === before.task && terminalTask(before.task)) { + writeCheckpoint({ state: "complete", iteration: iteration + 1, before, after, childExit: child.code, reason: "terminal_module_complete", childLog: child.outputPath }); + completed = true; + iteration += 1; + break; + } + if (!after || after.task === before.task) { + recordFailure(root, { task: before.task, phase: "scheduler", reason: "child_exited_without_queue_advance", childExit: child.code }); + writeCheckpoint({ state: "blocked", iteration, before, after, reason: "child_exited_without_queue_advance", childExit: child.code, childLog: child.outputPath }); + appendLog(`blocked task=${before.task} reason=child_exited_without_queue_advance`); + process.stderr.write(`scheduler-blocked task=${before.task} reason=child_exited_without_queue_advance\n`); + exitCode = 4; + break; + } + iteration += 1; + writeCheckpoint({ state: "handed_off", iteration, before, after, childLog: child.outputPath }); + process.stdout.write(`scheduler-iteration=${iteration} task=${before.task} next=${after.task}\n`); + if (pollMs > 0) { + const wait = new Int32Array(new SharedArrayBuffer(4)); + Atomics.wait(wait, 0, 0, pollMs); + } + } + if (exitCode === 0 && !preflightOnly && !stopRequested) { + const current = currentTask(); + const state = completed || maxIterations !== 0 ? "complete" : "running"; + writeCheckpoint({ state, iteration, current }); + appendLog(`stopped state=${state} iteration=${iteration} current=${current?.task ?? "NONE"}`); + } +} +catch (error) { + exitCode = 5; + try { recordFailure(root, { task: currentTask()?.task ?? "UNKNOWN", phase: "scheduler", reason: "scheduler_error", detail: error?.message ?? String(error) }); } + catch { /* Preserve the original scheduler failure. */ } + writeCheckpoint({ state: "blocked", iteration, reason: "scheduler_error", error: error?.message ?? String(error) }); + appendLog(`blocked reason=scheduler_error error=${error?.message ?? String(error)}`); +} +finally { + releaseLock(); +} +process.exitCode = exitCode; diff --git a/tools/web/run-wbv2-supervisor.mjs b/tools/web/run-wbv2-supervisor.mjs index d8c1fcc13..3b2e25207 100644 --- a/tools/web/run-wbv2-supervisor.mjs +++ b/tools/web/run-wbv2-supervisor.mjs @@ -2,6 +2,7 @@ import fs from "node:fs"; import path from "node:path"; import { spawnSync } from "node:child_process"; import { fileURLToPath } from "node:url"; +import { recordFailure } from "./wbv2-failure-ledger.mjs"; const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../.."); const abs = (file) => path.join(root, file); @@ -64,6 +65,7 @@ const releaseLock = () => { let stopRequested = false; let stopSignal; let iteration = 0; +let terminalCompleted = false; let fatalHandled = false; const stopFromSignal = (signal) => { stopRequested = true; @@ -97,8 +99,15 @@ 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({ schemaVersion: 1, pid: process.pid, updatedAt: new Date().toISOString(), ...payload }, null, 2)}\n`); + fs.writeFileSync(abs(temporaryPath), `${JSON.stringify(next, null, 2)}\n`); fs.renameSync(abs(temporaryPath), abs(checkpointPath)); } const run = (command, args = []) => { @@ -112,6 +121,17 @@ const currentTask = () => { 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"); @@ -128,11 +148,11 @@ try { 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 }); + const context = run(process.execPath, ["tools/web/check-wbv2-context-cache.mjs", "--task", before.task]); + appendLog(`preflight task=${before.task} contextExit=${context.code}`); + if (context.code !== 0) { + recordFailure(root, { task: before.task, phase: "supervisor", reason: "preflight_failed", childExit: context.code }); + writeCheckpoint({ state: "blocked", iteration, current: before, reason: "preflight_failed", contextExit: context.code }); exitCode = 2; break; } @@ -146,11 +166,19 @@ try { } if (child.code !== 0) { const reason = after.task === before.task ? "child_failed_same_task" : "child_failed_after_queue_advance"; + recordFailure(root, { 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) { + recordFailure(root, { 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; @@ -166,7 +194,7 @@ try { if (stopRequested) { exitCode = stopSignal === "SIGINT" ? 130 : 143; } - if (exitCode === 0) { + 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}`); diff --git a/tools/web/wbv2-failure-ledger.mjs b/tools/web/wbv2-failure-ledger.mjs new file mode 100644 index 000000000..8be17e6ac --- /dev/null +++ b/tools/web/wbv2-failure-ledger.mjs @@ -0,0 +1,21 @@ +import crypto from "node:crypto"; +import fs from "node:fs"; +import path from "node:path"; + +const ledgerPath = (root) => path.join(root, "tests/golden/WBV2/failure-ledger.json"); +const keyFor = (entry) => crypto.createHash("sha256").update(JSON.stringify({ task: entry.task, phase: entry.phase, reason: entry.reason, childExit: entry.childExit ?? null })).digest("hex"); + +export function recordFailure(root, entry) { + const file = ledgerPath(root); + let ledger = { schemaVersion: 1, entries: [] }; + try { ledger = JSON.parse(fs.readFileSync(file, "utf8")); } + catch { /* First failure or an interrupted write. */ } + const key = keyFor(entry); + if (ledger.entries.some((candidate) => candidate.key === key)) return false; + ledger.entries = [...ledger.entries, { ...entry, key, recordedAt: new Date().toISOString() }].slice(-200); + fs.mkdirSync(path.dirname(file), { recursive: true }); + const temporary = `${file}.tmp-${process.pid}`; + fs.writeFileSync(temporary, `${JSON.stringify(ledger, null, 2)}\n`); + fs.renameSync(temporary, file); + return true; +}