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;