perf(web): add idempotent batch verification fast paths
This commit is contained in:
@@ -3,7 +3,7 @@ import crypto from "node:crypto";
|
||||
import path from "node:path";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { recordFailure } from "./wbv2-failure-ledger.mjs";
|
||||
import { createFailureLedgerSession } from "./wbv2-failure-ledger.mjs";
|
||||
|
||||
const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../..");
|
||||
const controlRoot = "tests/golden/WBV2";
|
||||
@@ -11,6 +11,10 @@ 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 failureLedger = createFailureLedgerSession(root, {
|
||||
logPaths: [`${controlRoot}/continuous-scheduler.log`, `${controlRoot}/continuous-supervisor.log`, `${controlRoot}/continuous-guardian.log`],
|
||||
});
|
||||
let latestBatch = null;
|
||||
|
||||
const argValue = (name, fallback) => {
|
||||
const index = process.argv.indexOf(name);
|
||||
@@ -60,6 +64,7 @@ function writeCheckpoint(payload) {
|
||||
pid: process.pid,
|
||||
updatedAt: new Date().toISOString(),
|
||||
...payload,
|
||||
latestBatch: payload.latestBatch ?? latestBatch,
|
||||
};
|
||||
const stable = (value) => { const { pid, updatedAt, ...rest } = value ?? {}; return rest; };
|
||||
try {
|
||||
@@ -72,6 +77,43 @@ function writeCheckpoint(payload) {
|
||||
fs.renameSync(abs(temporaryPath), abs(checkpointPath));
|
||||
}
|
||||
|
||||
function batchInfo(pointer) {
|
||||
if (!pointer?.task) return null;
|
||||
let manifest = {};
|
||||
try { manifest = JSON.parse(read(`tests/golden/${pointer.task}/manifest.json`)); }
|
||||
catch { /* The active batch may not have produced a manifest yet. */ }
|
||||
let card = "";
|
||||
try { card = read(`docs/tasks/${pointer.task}.md`); }
|
||||
catch { /* Readiness reports the missing card separately. */ }
|
||||
const batchId = manifest.batchId ?? card.match(/^- batchId: `?([^`\n]+)`?/mu)?.[1] ?? null;
|
||||
let selectedRecordIds = Array.isArray(manifest.selectedRecordIds) ? manifest.selectedRecordIds : [];
|
||||
if (selectedRecordIds.length === 0) {
|
||||
try {
|
||||
const selected = card.match(/^- selectedRecordIds:\s*(\[[^\n]+\])/mu)?.[1];
|
||||
selectedRecordIds = selected ? JSON.parse(selected) : [];
|
||||
} catch { selectedRecordIds = []; }
|
||||
}
|
||||
return {
|
||||
task: pointer.task,
|
||||
batchId,
|
||||
selectedRecordIds,
|
||||
completed: manifest.status === "done" && manifest.nextTask === null,
|
||||
};
|
||||
}
|
||||
|
||||
function planNextBatch(pointer) {
|
||||
const result = spawnSync("python3", [abs("my-tools/authorize_next_wbv2_task.py"), "--write"], {
|
||||
cwd: root,
|
||||
encoding: "utf8",
|
||||
maxBuffer: 8 * 1024 * 1024,
|
||||
});
|
||||
const output = `${result.stdout ?? ""}${result.stderr ?? ""}`;
|
||||
if (output) appendLog(`planner task=${pointer?.task ?? "NONE"} exit=${result.status ?? 1}\n${output}`);
|
||||
if ((result.status ?? 1) !== 0 && /no configured unresolved Blender source batch remains/u.test(output)) return { terminal: true };
|
||||
if ((result.status ?? 1) !== 0) throw new Error(`next batch planner failed: ${output}`);
|
||||
return { terminal: false };
|
||||
}
|
||||
|
||||
function appendLog(message) {
|
||||
fs.appendFileSync(abs(logPath), `[${new Date().toISOString()}] ${message}\n`);
|
||||
}
|
||||
@@ -91,10 +133,10 @@ function refreshContinuation(task) {
|
||||
if (!fs.existsSync(continuationPath)) return false;
|
||||
const pointer = currentTask();
|
||||
const index = JSON.parse(read("tests/golden/WBV2/task-index.json"));
|
||||
const plan = JSON.parse(read(index.source.path));
|
||||
const taskIndex = plan.tasks.findIndex((candidate) => candidate.id === task);
|
||||
const terminal = terminalTask(task);
|
||||
const nextTask = terminal ? null : plan.tasks[taskIndex + 1]?.id ?? null;
|
||||
// The compact task index already contains the direct successor. Reading the
|
||||
// full plan here adds no evidence and used to dominate short handoff cycles.
|
||||
const nextTask = terminal ? null : index.entries?.[task]?.next ?? null;
|
||||
const inputFiles = ["docs/EXECUTION_QUEUE.md", `docs/tasks/${task}.md`, pointer.parentManifest, `tests/golden/${task}/task-context.json`];
|
||||
const fingerprint = crypto.createHash("sha256").update(inputFiles.map((file) => `${file}\0${crypto.createHash("sha256").update(fs.readFileSync(abs(file))).digest("hex")}\n`).join("")).digest("hex");
|
||||
try {
|
||||
@@ -246,7 +288,9 @@ 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 });
|
||||
const initial = currentTask();
|
||||
latestBatch = batchInfo(initial);
|
||||
writeCheckpoint({ state: "started", iteration, current: initial, maxIterations, commit: commitEnabled, push: pushEnabled });
|
||||
while (maxIterations === 0 || iteration < maxIterations) {
|
||||
if (stopRequested) {
|
||||
writeCheckpoint({ state: "stopped", iteration, current: currentTask(), reason: stopSignal });
|
||||
@@ -254,6 +298,7 @@ try {
|
||||
break;
|
||||
}
|
||||
const before = currentTask();
|
||||
latestBatch = batchInfo(before);
|
||||
if (!before) {
|
||||
writeCheckpoint({ state: "complete", iteration, current: null, reason: "queue_closed" });
|
||||
appendLog(`complete queue_closed iteration=${iteration}`);
|
||||
@@ -266,14 +311,24 @@ try {
|
||||
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" });
|
||||
const completedBatchInfo = { ...batchInfo(before), completed: true };
|
||||
latestBatch = completedBatchInfo;
|
||||
const planned = preflightOnly ? { terminal: true } : planNextBatch(before);
|
||||
if (!planned.terminal) {
|
||||
const afterPlan = currentTask();
|
||||
latestBatch = batchInfo(afterPlan);
|
||||
writeCheckpoint({ state: "planned_next_batch", iteration, current: afterPlan, before, reason: "completed_batch_planned" });
|
||||
appendLog(`planned next batch before=${before.task} after=${afterPlan.task}`);
|
||||
continue;
|
||||
}
|
||||
writeCheckpoint({ state: "complete", iteration, current: before, reason: "v2_plan_terminal", latestBatch: completedBatchInfo });
|
||||
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) });
|
||||
failureLedger.record({ 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`);
|
||||
@@ -290,7 +345,7 @@ try {
|
||||
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 });
|
||||
failureLedger.record({ 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`);
|
||||
@@ -306,7 +361,7 @@ try {
|
||||
refreshTaskContext(before.task);
|
||||
}
|
||||
catch (error) {
|
||||
recordFailure(root, { task: before.task, phase: "scheduler", reason: "context_rotation_failed", detail: error?.message ?? String(error) });
|
||||
failureLedger.record({ task: before.task, phase: "scheduler", reason: "context_rotation_failed", detail: error?.message ?? String(error) });
|
||||
writeCheckpoint({ state: "blocked", iteration, before, reason: "context_rotation_failed", error: error?.message ?? String(error) });
|
||||
process.stderr.write(`scheduler-blocked task=${before.task} reason=context_rotation_failed\n`);
|
||||
exitCode = 2;
|
||||
@@ -315,6 +370,7 @@ try {
|
||||
writeCheckpoint({ state: "running", iteration: iteration + 1, before, readiness });
|
||||
const child = runOne(before);
|
||||
const after = currentTask();
|
||||
latestBatch = batchInfo(after);
|
||||
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 });
|
||||
@@ -323,7 +379,7 @@ try {
|
||||
}
|
||||
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 });
|
||||
failureLedger.record({ 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`);
|
||||
@@ -335,7 +391,7 @@ try {
|
||||
if (refreshContinuation(after?.task ?? before.task) && commitEnabled) commitContinuation();
|
||||
}
|
||||
catch (error) {
|
||||
recordFailure(root, { task: before.task, phase: "scheduler", reason: "continuation_rotation_failed", detail: error?.message ?? String(error) });
|
||||
failureLedger.record({ task: before.task, phase: "scheduler", reason: "continuation_rotation_failed", detail: error?.message ?? String(error) });
|
||||
writeCheckpoint({ state: "blocked", iteration, before, after, reason: "continuation_rotation_failed", error: error?.message ?? String(error) });
|
||||
process.stderr.write(`scheduler-blocked task=${before.task} reason=continuation_rotation_failed\n`);
|
||||
exitCode = 5;
|
||||
@@ -348,7 +404,7 @@ try {
|
||||
break;
|
||||
}
|
||||
if (!after || after.task === before.task) {
|
||||
recordFailure(root, { task: before.task, phase: "scheduler", reason: "child_exited_without_queue_advance", childExit: child.code });
|
||||
failureLedger.record({ 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`);
|
||||
@@ -372,12 +428,13 @@ try {
|
||||
}
|
||||
catch (error) {
|
||||
exitCode = 5;
|
||||
try { recordFailure(root, { task: currentTask()?.task ?? "UNKNOWN", phase: "scheduler", reason: "scheduler_error", detail: error?.message ?? String(error) }); }
|
||||
try { failureLedger.record({ 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 {
|
||||
failureLedger.flush();
|
||||
releaseLock();
|
||||
}
|
||||
process.exitCode = exitCode;
|
||||
|
||||
Reference in New Issue
Block a user