perf(web): centralize batch handoff rotation
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
import fs from "node:fs";
|
||||
import crypto from "node:crypto";
|
||||
import path from "node:path";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
@@ -79,6 +80,57 @@ function read(file) {
|
||||
return fs.readFileSync(abs(file), "utf8");
|
||||
}
|
||||
|
||||
function refreshTaskContext(task) {
|
||||
const result = spawnSync(process.execPath, [abs("tools/web/check-wbv2-context-cache.mjs"), "--task", task], { cwd: root, encoding: "utf8", maxBuffer: 8 * 1024 * 1024 });
|
||||
if ((result.status ?? 1) !== 0) throw new Error(`context rotation failed for ${task}: ${result.stderr ?? result.stdout ?? "unknown error"}`);
|
||||
}
|
||||
|
||||
function refreshContinuation(task) {
|
||||
const continuationPath = abs("nextTask.md");
|
||||
const markerPath = abs("tests/golden/WBV2/continuation-snapshot.json");
|
||||
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;
|
||||
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 {
|
||||
const marker = JSON.parse(fs.readFileSync(markerPath, "utf8"));
|
||||
if (marker.task === task && marker.fingerprint === fingerprint) return false;
|
||||
}
|
||||
catch { /* First rotation or an interrupted marker write. */ }
|
||||
const tick = String.fromCharCode(96);
|
||||
let next = fs.readFileSync(continuationPath, "utf8");
|
||||
const snapshotTask = task;
|
||||
const snapshotState = terminal ? "done" : "in_progress";
|
||||
const snapshotNext = nextTask ?? "NONE";
|
||||
const snapshotCommand = nextTask ? `node tools/web/run-wbv2-module-task.mjs --task ${nextTask} --write` : "node tools/web/check-wbv2-module-execution.mjs";
|
||||
next = next.replace(/\| Current task when this file was written \| `[^`]+` \|/u, `| Current task when this file was written | ${tick}${snapshotTask}${tick} |`).replace(/\| Current task state \| `[^`]+` \|/u, `| Current task state | ${tick}${snapshotState}${tick} |`).replace(/\| Parent manifest \| `[^`]+` \|/u, `| Parent manifest | ${tick}tests/golden/${task}/manifest.json${tick} |`).replace(/\| Current task card \| `[^`]+` \|/u, `| Current task card | ${tick}docs/tasks/${snapshotTask}.md${tick} |`).replace(/\| Current task-context nextTask \| `[^`]+` \|/u, `| Current task-context nextTask | ${tick}${snapshotNext}${tick} |`).replace(/\| Focused command \| `[^`]+` \|/u, `| Focused command | ${tick}${snapshotCommand}${tick} |`);
|
||||
fs.writeFileSync(continuationPath, next);
|
||||
const marker = { schemaVersion: 1, task, fingerprint, refreshedAt: new Date().toISOString(), nextTask: snapshotNext };
|
||||
const temporary = `${markerPath}.tmp-${process.pid}`;
|
||||
fs.writeFileSync(temporary, `${JSON.stringify(marker, null, 2)}\n`);
|
||||
fs.renameSync(temporary, markerPath);
|
||||
return true;
|
||||
}
|
||||
|
||||
function commitContinuation() {
|
||||
const add = spawnSync("git", ["add", "-f", "nextTask.md", "tests/golden/WBV2/continuation-snapshot.json"], { cwd: root, encoding: "utf8" });
|
||||
if ((add.status ?? 1) !== 0) throw new Error(`continuation staging failed: ${add.stderr ?? add.stdout ?? "unknown error"}`);
|
||||
const pending = spawnSync("git", ["diff", "--cached", "--quiet"], { cwd: root, encoding: "utf8" });
|
||||
if ((pending.status ?? 0) === 1) {
|
||||
const commit = spawnSync("git", ["commit", "-m", "chore(web): rotate continuation snapshot"], { cwd: root, encoding: "utf8" });
|
||||
if ((commit.status ?? 1) !== 0) throw new Error(`continuation commit failed: ${commit.stderr ?? commit.stdout ?? "unknown error"}`);
|
||||
if (pushEnabled) {
|
||||
const push = spawnSync("git", ["push", "origin", "main"], { cwd: root, encoding: "utf8" });
|
||||
if ((push.status ?? 1) !== 0) throw new Error(`continuation push failed: ${push.stderr ?? push.stdout ?? "unknown error"}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function currentTask() {
|
||||
const queue = read("docs/EXECUTION_QUEUE.md");
|
||||
const task = queue.match(/\| 当前任务 \| `(WBV2-P\d+-\d+)` \|/u)?.[1];
|
||||
@@ -250,6 +302,16 @@ try {
|
||||
process.stdout.write(`scheduler-preflight-ready task=${before.task} class=${readiness.taskClass}\n`);
|
||||
break;
|
||||
}
|
||||
try {
|
||||
refreshTaskContext(before.task);
|
||||
}
|
||||
catch (error) {
|
||||
recordFailure(root, { 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;
|
||||
break;
|
||||
}
|
||||
writeCheckpoint({ state: "running", iteration: iteration + 1, before, readiness });
|
||||
const child = runOne(before);
|
||||
const after = currentTask();
|
||||
@@ -268,6 +330,16 @@ try {
|
||||
exitCode = 3;
|
||||
break;
|
||||
}
|
||||
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) });
|
||||
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;
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user