jason.woltjeandClaude Opus 5.5
6ca116b7ba
Filbert approved round 1 (f167b85e). Manifest 782bcb62, 21 files, plus the QUEUE.md markers and the TOOLS.md section. Lead decision 35. Co-Authored-By: Claude Opus 5.5 <[email protected]>
800 lines
35 KiB
JavaScript
800 lines
35 KiB
JavaScript
// The queue on disk: canonical checks (8.3), brief checks (8.13), the write
|
||
// path (8.5) and every verb. The CLI passes only its parsed arguments. Tests
|
||
// may also pass `io`, `proc`, `hook`, `now` and `readOrder` through this API
|
||
// to inject faults and pause at named steps; no flag or environment variable
|
||
// reaches them.
|
||
import { spawnSync } from "node:child_process";
|
||
import { lstatSync, readFileSync, readdirSync, realpathSync } from "node:fs";
|
||
import { isAbsolute, join, resolve as resolvePath } from "node:path";
|
||
import { fileURLToPath } from "node:url";
|
||
import { QueueError } from "./errors.mjs";
|
||
import { checkPlatform, errno, fsyncFile, lstatOrNull, readOrNull, realIo, unlinkQuiet, writeTemp } from "./io.mjs";
|
||
import { acquire, checkGate, realProc, releaseOrWarn, unlock as unlockLock } from "./lock.mjs";
|
||
import {
|
||
PRIVILEGED, SEMANTICS, VERSION, applyEntry, buildDoc, canonArgs, checkCallerOpId, checkName, classifyView, countHeading,
|
||
describeUnshown, genesisReceipt, genesisRows, gitBlobId, loadDoc, logDigest, nextFor, parseBriefSpec, parseManifest,
|
||
parseMigrationMap, render, rowsArray, sameJson, serialize, sha256, splitView,
|
||
} from "./queue.mjs";
|
||
|
||
export const QUEUE_REL = "docs/plans/queue.json";
|
||
export const VIEW_REL = "docs/plans/QUEUE.md";
|
||
export const WITNESS_NAME = "mosaic-queue.head";
|
||
const CODE_FILE = fileURLToPath(import.meta.url);
|
||
const FIX = "scripts/mosaic queue";
|
||
|
||
function refuse(message) {
|
||
return new QueueError(message, 2);
|
||
}
|
||
|
||
function makeCtx(opts = {}) {
|
||
return {
|
||
cwd: opts.cwd ?? process.cwd(),
|
||
env: opts.env ?? process.env,
|
||
io: opts.io ?? realIo,
|
||
proc: opts.proc ?? realProc,
|
||
hook: opts.hook ?? (() => {}),
|
||
now: opts.now ?? (() => new Date().toISOString()),
|
||
readOrder: opts.readOrder ?? "witness-first",
|
||
lockWaitMs: opts.lockWaitMs ?? 10000,
|
||
lockStepMs: opts.lockStepMs ?? 100,
|
||
};
|
||
}
|
||
|
||
// --- git and the canonical checks (8.3) ---
|
||
|
||
function git(ctx, cwd, args, { allowFail = false } = {}) {
|
||
const r = spawnSync("git", ["-C", cwd, ...args], { env: ctx.env, maxBuffer: 64 << 20 });
|
||
if (r.error) throw new QueueError(`cannot run git: ${errno(r.error)}`, 1);
|
||
if (r.status !== 0) {
|
||
if (allowFail) return null;
|
||
throw new QueueError(`git ${args.join(" ")} failed: ${r.stderr.toString().trim()}`, 1);
|
||
}
|
||
return r.stdout;
|
||
}
|
||
|
||
function locate(ctx) {
|
||
for (const k of ["GIT_DIR", "GIT_WORK_TREE", "GIT_COMMON_DIR"]) {
|
||
if (ctx.env[k] !== undefined) throw refuse(`${k} is set; the queue locates its repository only from the working directory`);
|
||
}
|
||
const raw = git(ctx, ctx.cwd, ["rev-parse", "--show-toplevel"], { allowFail: true });
|
||
if (raw === null) throw refuse(`${ctx.cwd} is not inside a git checkout`);
|
||
const top = realpathSync(raw.toString().trim());
|
||
const dirs = git(ctx, top, ["rev-parse", "--path-format=absolute", "--git-dir", "--git-common-dir"]).toString().split("\n");
|
||
const gitDir = realpathSync(dirs[0]);
|
||
if (gitDir !== realpathSync(dirs[1])) throw refuse(`${top} is a linked worktree; the queue runs only in the canonical checkout`);
|
||
if (gitDir !== join(top, ".git")) throw refuse(`${top}'s git directory is ${gitDir}, not ${top}/.git`);
|
||
const ref = git(ctx, top, ["symbolic-ref", "-q", "HEAD"], { allowFail: true });
|
||
return {
|
||
top, gitDir, ref: ref === null ? null : ref.toString().trim(),
|
||
queuePath: join(top, QUEUE_REL), viewPath: join(top, VIEW_REL), docsDir: join(top, "docs/plans"), witnessPath: join(gitDir, WITNESS_NAME),
|
||
};
|
||
}
|
||
|
||
function checkBranch(loc, branch) {
|
||
if (loc.ref === null) throw refuse(`HEAD is detached; the queue runs only on branch ${branch}`);
|
||
if (loc.ref !== `refs/heads/${branch}`) throw refuse(`HEAD is ${loc.ref}; the queue runs only on branch ${branch}`);
|
||
}
|
||
|
||
function checkCode(root) {
|
||
if (!realpathSync(CODE_FILE).startsWith(`${root}/`)) throw refuse(`this queue code (${CODE_FILE}) is not under the canonical root ${root}`);
|
||
}
|
||
|
||
function checkCanonical(loc, root, branch) {
|
||
if (loc.top !== root) throw refuse(`this checkout is ${loc.top}; the queue's canonical root is ${root}`);
|
||
checkBranch(loc, branch);
|
||
checkCode(root);
|
||
}
|
||
|
||
function headHas(ctx, loc, path) {
|
||
return git(ctx, loc.top, ["cat-file", "-e", `HEAD:${path}`], { allowFail: true }) !== null;
|
||
}
|
||
|
||
function actorOf(ctx, by) {
|
||
const name = by ?? ctx.env.MOSAIC_AGENT_NAME;
|
||
if (name === undefined || name === "") throw refuse("no actor: pass --by NAME or set MOSAIC_AGENT_NAME");
|
||
return checkName(name, "actor");
|
||
}
|
||
|
||
// N12: --by wins over MOSAIC_AGENT_NAME, and a difference is worth a line on
|
||
// stderr. Both are self-asserted (J2), so nothing is logged.
|
||
function actorMismatch(ctx, by) {
|
||
const env = ctx.env.MOSAIC_AGENT_NAME;
|
||
if (by === null || by === undefined || env === undefined || env === "" || by === env) return null;
|
||
return `warning: --by ${JSON.stringify(by)} differs from MOSAIC_AGENT_NAME=${JSON.stringify(env)}`;
|
||
}
|
||
|
||
function isSeatDir(top, name) {
|
||
try {
|
||
return lstatSync(join(top, "agents", name)).isDirectory();
|
||
} catch {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
function checkActorSeat(top, name) {
|
||
if (name !== "jason" && !isSeatDir(top, name)) throw refuse(`actor ${name} is not jason or a seat under agents/`);
|
||
}
|
||
|
||
function checkSeat(top, name, what, owner) {
|
||
if (name === "jason" || isSeatDir(top, name)) return;
|
||
if (owner && (name === "coordinator" || name === "unassigned")) return;
|
||
throw refuse(`${what} ${name} is not jason${owner ? ", coordinator, unassigned" : ""} or a seat under agents/`);
|
||
}
|
||
|
||
// Seat names checked when a write sets them; replay never looks at agents/.
|
||
function checkNewSeats(top, before, after) {
|
||
for (const row of after.values()) {
|
||
const old = before.get(row.id);
|
||
if (!old || old.owner !== row.owner) checkSeat(top, row.owner, `row ${row.id} owner`, true);
|
||
if (!old || old.gateOwner !== row.gateOwner) checkSeat(top, row.gateOwner, `row ${row.id} gate owner`, true);
|
||
if (!old || !sameJson(old.reviewers, row.reviewers)) row.reviewers.forEach((r) => checkSeat(top, r, `row ${row.id} reviewer`, false));
|
||
}
|
||
}
|
||
|
||
// --- brief and candidate checks (8.13, 8.9) ---
|
||
|
||
function headEntry(ctx, loc, path) {
|
||
const out = git(ctx, loc.top, ["ls-tree", "-z", "--full-tree", "HEAD", "--", path], { allowFail: true });
|
||
if (out === null) return null;
|
||
const entries = out.toString().split("\0").filter(Boolean).map((e) => {
|
||
const m = /^(\d{6}) (\w+) ([0-9a-f]{40})\t(.*)$/s.exec(e);
|
||
return m ? { mode: m[1], type: m[2], oid: m[3], path: m[4] } : null;
|
||
});
|
||
return entries.find((e) => e && e.path === path) ?? null;
|
||
}
|
||
|
||
export function briefCheck(ctx, loc, path, anchor) {
|
||
const abs = join(loc.top, path);
|
||
let st;
|
||
try {
|
||
st = lstatSync(abs);
|
||
} catch {
|
||
throw refuse(`brief ${path} does not exist in the working tree`);
|
||
}
|
||
if (st.isSymbolicLink() || !st.isFile()) throw refuse(`brief ${path} must be a regular file, not a symlink`);
|
||
if (!realpathSync(abs).startsWith(`${loc.top}/`)) throw refuse(`brief ${path} resolves outside ${loc.top}`);
|
||
const entry = headEntry(ctx, loc, path);
|
||
if (entry === null) throw refuse(`brief ${path} is not committed in HEAD; commit it first (a staged brief is refused)`);
|
||
if (entry.type !== "blob" || entry.mode === "120000") throw refuse(`brief ${path} is a ${entry.mode === "120000" ? "symlink" : entry.type} in HEAD, not a file`);
|
||
const text = git(ctx, loc.top, ["cat-file", "blob", entry.oid]).toString("utf8");
|
||
const n = countHeading(text, anchor);
|
||
if (n !== 1) throw refuse(`brief ${path} has ${n} headings "${anchor}" in HEAD; exactly one is required`);
|
||
return { path, anchor, blob: entry.oid };
|
||
}
|
||
|
||
// The working copy a seat reads (8.13): its git blob id, computed in-process.
|
||
function workingBriefMatches(loc, brief) {
|
||
try {
|
||
const abs = join(loc.top, brief.path);
|
||
if (!lstatSync(abs).isFile()) return false;
|
||
return gitBlobId(readFileSync(abs)) === brief.blob;
|
||
} catch {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
function resolveCandidate(ctx, loc, spec) {
|
||
const path = isAbsolute(spec) ? spec : resolvePath(ctx.cwd, spec);
|
||
let st = null;
|
||
try { st = lstatSync(path); } catch { /* not a file: try it as a commit */ }
|
||
if (st !== null) {
|
||
if (!st.isFile()) throw refuse(`candidate ${spec} is not a regular file`);
|
||
let text;
|
||
try {
|
||
text = new TextDecoder("utf-8", { fatal: true }).decode(readFileSync(path));
|
||
} catch {
|
||
throw refuse(`candidate manifest ${spec} is not valid UTF-8`);
|
||
}
|
||
parseManifest(text);
|
||
return { kind: "manifest", digest: sha256(text), text };
|
||
}
|
||
const out = git(ctx, loc.top, ["rev-parse", "--verify", "--quiet", "--end-of-options", `${spec}^{commit}`], { allowFail: true });
|
||
if (out === null) throw refuse(`candidate ${spec} is neither a manifest file nor a commit`);
|
||
const sha = out.toString().trim();
|
||
const refs = git(ctx, loc.top, ["for-each-ref", "--contains", sha, "--format=%(refname)", "refs/heads", "refs/tags"]).toString().trim();
|
||
if (refs === "") throw refuse(`candidate commit ${sha} is not reachable from any local branch or tag`);
|
||
return { kind: "commit", digest: sha, text: null };
|
||
}
|
||
|
||
// --- files: queue.json, the witness, the view ---
|
||
|
||
function statKey(st) {
|
||
return { dev: st.dev, ino: st.ino, size: st.size, mtimeNs: st.mtimeNs };
|
||
}
|
||
|
||
function sameStat(a, b) {
|
||
return a.dev === b.dev && a.ino === b.ino && a.size === b.size && a.mtimeNs === b.mtimeNs;
|
||
}
|
||
|
||
// 8.5 step 1: bytes, stat, schema, byte-for-byte form, replay, canonical checks.
|
||
function readLocked(ctx, loc) {
|
||
const st = lstatOrNull(ctx.io, loc.queuePath);
|
||
if (st === null) throw refuse(`no ${QUEUE_REL}; genesis has not run`);
|
||
const bytes = ctx.io.readFile(loc.queuePath);
|
||
const d = loadDoc(bytes);
|
||
checkCanonical(loc, d.doc.canonicalRoot, d.branch);
|
||
return { bytes, stat: statKey(st), sha: sha256(bytes), doc: d.doc, state: d.state, branch: d.branch };
|
||
}
|
||
|
||
function parseWitness(bytes) {
|
||
if (bytes === null) return null;
|
||
let w;
|
||
try { w = JSON.parse(bytes.toString("utf8")); } catch { return { invalid: true }; }
|
||
const ok = w !== null && typeof w === "object" && Object.keys(w).join() === "revision,logDigest,fileSha,at"
|
||
&& Number.isInteger(w.revision) && w.revision >= 0 && /^[0-9a-f]{64}$/.test(w.logDigest) && /^[0-9a-f]{64}$/.test(w.fileSha)
|
||
&& typeof w.at === "string";
|
||
return ok ? w : { invalid: true };
|
||
}
|
||
|
||
function readWitness(ctx, loc) {
|
||
return parseWitness(readOrNull(ctx.io, loc.witnessPath));
|
||
}
|
||
|
||
// match, tail (the file extends the witness), lost, or absent. An invalid
|
||
// witness counts as absent.
|
||
export function compareWitness(doc, fileSha, w) {
|
||
if (w === null || w.invalid) return { state: "absent", invalid: Boolean(w?.invalid) };
|
||
if (w.revision > doc.revision || logDigest(doc.log, w.revision) !== w.logDigest) return { state: "lost", w };
|
||
if (w.revision === doc.revision) return w.fileSha === fileSha ? { state: "match", w } : { state: "lost", w };
|
||
return { state: "tail", w, from: w.revision + 1 };
|
||
}
|
||
|
||
function writeWitness(ctx, loc, doc, bytes) {
|
||
const rec = { revision: doc.revision, logDigest: logDigest(doc.log, doc.revision), fileSha: sha256(bytes), at: ctx.now() };
|
||
const tmp = `${loc.witnessPath}.tmp`;
|
||
unlinkQuiet(ctx.io, tmp);
|
||
writeTemp(ctx.io, tmp, Buffer.from(JSON.stringify(rec) + "\n"));
|
||
try {
|
||
ctx.io.rename(tmp, loc.witnessPath);
|
||
} catch (err) {
|
||
unlinkQuiet(ctx.io, tmp);
|
||
throw err;
|
||
}
|
||
try {
|
||
ctx.io.fsyncDir(loc.gitDir);
|
||
} catch (err) {
|
||
throw Object.assign(new Error(errno(err)), { code: err.code, renamed: true });
|
||
}
|
||
}
|
||
|
||
function confirmTail(ctx, loc, cur) {
|
||
try {
|
||
fsyncFile(ctx.io, loc.queuePath);
|
||
ctx.io.fsyncDir(loc.docsDir);
|
||
writeWitness(ctx, loc, cur.doc, cur.bytes);
|
||
} catch (err) {
|
||
throw new QueueError(`cannot confirm rev ${cur.doc.revision} durable (${errno(err)}); nothing changed`, 1);
|
||
}
|
||
}
|
||
|
||
// The refusal for lost history or a missing witness. A file holding genesis
|
||
// alone has one more way out: `sync` confirms it (8.5 step 2).
|
||
function lostRefusal(cmp, doc) {
|
||
const way = cmp.state === "absent" && doc.log.length === 1
|
||
? `run \`${FIX} sync\` (the file holds genesis alone)`
|
||
: `every verb refuses except \`${FIX} accept-history\` (see the README's manual recovery)`;
|
||
return refuse(`${lostMessage(cmp, doc)}; ${way}`);
|
||
}
|
||
|
||
function lostMessage(cmp, doc) {
|
||
if (cmp.state === "absent") {
|
||
return `the witness .git/${WITNESS_NAME} is ${cmp.invalid ? "invalid" : "missing"}; this file holds revs 0..${doc.revision}, and ops recorded after them may be lost`;
|
||
}
|
||
return `history lost: the witness recorded rev ${cmp.w.revision} (logDigest ${cmp.w.logDigest.slice(0, 12)}…, at ${cmp.w.at}); this file holds revs 0..${doc.revision} and does not extend it`;
|
||
}
|
||
|
||
// 8.5 step 2 for a locked caller. Returns the comparison, with `confirmed`
|
||
// listing the entries this call made durable.
|
||
function witnessStep(ctx, loc, cur, { verb, op = null, args = null }) {
|
||
const cmp = compareWitness(cur.doc, cur.sha, readWitness(ctx, loc));
|
||
const log = cur.doc.log;
|
||
const isRetry = (e) => e.op === op && e.verb === verb && sameJson(e.args, args);
|
||
if (cmp.state === "match") return { ...cmp, confirmed: [] };
|
||
if (cmp.state === "tail") {
|
||
const tail = log.slice(cmp.from);
|
||
if (verb === "sync" || tail.some(isRetry)) {
|
||
confirmTail(ctx, loc, cur);
|
||
return { ...cmp, confirmed: tail };
|
||
}
|
||
throw refuse(`unconfirmed tail: ${describeUnshown(tail)} visible but not confirmed durable; run \`${FIX} sync\` or retry that op`);
|
||
}
|
||
if (cmp.state === "absent" && log.length === 1 && (verb === "sync" || isRetry(log[0]))) {
|
||
confirmTail(ctx, loc, cur);
|
||
return { ...cmp, confirmed: log };
|
||
}
|
||
if (verb === "accept-history") return { ...cmp, confirmed: [] };
|
||
throw lostRefusal(cmp, cur.doc);
|
||
}
|
||
|
||
function readView(ctx, loc) {
|
||
const bytes = readOrNull(ctx.io, loc.viewPath);
|
||
return { bytes, text: bytes === null ? null : bytes.toString("utf8") };
|
||
}
|
||
|
||
function staleMessage(view, log) {
|
||
const last = log.length - 1;
|
||
const lines = view.unshown.map((e) => `rev ${e.rev} (op ${e.op} by ${e.by} at ${e.at})`);
|
||
return `view stale: QUEUE.md shows rev ${view.shown}; ${lines.join(", ")} ${view.unshown.length === 1 ? "is" : "are"} recorded but the table shows rev ${view.shown}`
|
||
+ ` and may never have been acknowledged. Tell ${[...new Set(view.unshown.map((e) => e.by))].join(", ")}, then run \`${FIX} render\` (the file is at rev ${last})`;
|
||
}
|
||
|
||
function unknownMessage(view) {
|
||
return `view unknown: ${view.why}; restore the table with git or re-apply the edit as queue ops, then run \`${FIX} render\``;
|
||
}
|
||
|
||
// 8.5 step 11. Returns null, or the warning that the view was not written.
|
||
function writeView(ctx, loc, before, parts, body, tag) {
|
||
const stale = `QUEUE.md changed since it was read; the view was not written and is stale. Check it, then run \`${FIX} render\``;
|
||
const now = readOrNull(ctx.io, loc.viewPath);
|
||
if (now === null || !now.equals(before)) return stale;
|
||
const tmp = `${loc.viewPath}.${tag}.tmp`;
|
||
let renamed = false;
|
||
try {
|
||
const mode = Number(ctx.io.stat(loc.viewPath).mode & 0o777n);
|
||
unlinkQuiet(ctx.io, tmp);
|
||
writeTemp(ctx.io, tmp, Buffer.from(parts.head + body + parts.tail), mode);
|
||
const again = readOrNull(ctx.io, loc.viewPath);
|
||
if (again === null || !again.equals(before)) {
|
||
unlinkQuiet(ctx.io, tmp);
|
||
return stale;
|
||
}
|
||
ctx.io.rename(tmp, loc.viewPath);
|
||
renamed = true;
|
||
ctx.io.fsyncDir(loc.docsDir);
|
||
return null;
|
||
} catch (err) {
|
||
if (renamed) return `the view is written but not confirmed durable (${errno(err)}); the op stands; after a host crash, check the table with \`${FIX} verify\``;
|
||
unlinkQuiet(ctx.io, tmp);
|
||
return `the view write failed (${errno(err)}); the op stands and the view is stale; run \`${FIX} render\``;
|
||
}
|
||
}
|
||
|
||
// 8.5 steps 6–9. `exclusive` (genesis) links instead of renaming, so an
|
||
// existing queue.json is never replaced.
|
||
function writeQueue(ctx, loc, cur, bytes, op, rev, exclusive = false) {
|
||
const { io } = ctx;
|
||
const tmp = `${loc.queuePath}.${op}.tmp`;
|
||
// Only the lock holder writes these; one left here is from a killed writer.
|
||
unlinkQuiet(io, tmp);
|
||
try {
|
||
writeTemp(io, tmp, bytes, 0o644);
|
||
} catch (err) {
|
||
throw new QueueError(`cannot write ${QUEUE_REL}.${op}.tmp (${errno(err)}); nothing changed`, 1);
|
||
}
|
||
ctx.hook("temp-written");
|
||
let unchanged;
|
||
try {
|
||
if (exclusive) {
|
||
unchanged = lstatOrNull(io, loc.queuePath) === null;
|
||
} else {
|
||
const st = lstatOrNull(io, loc.queuePath);
|
||
const now = st === null ? null : io.readFile(loc.queuePath);
|
||
unchanged = st !== null && sameStat(statKey(st), cur.stat) && now.equals(cur.bytes);
|
||
}
|
||
} catch (err) {
|
||
unlinkQuiet(io, tmp);
|
||
throw new QueueError(`cannot recheck ${QUEUE_REL} (${errno(err)}); nothing changed`, 1);
|
||
}
|
||
if (!unchanged) {
|
||
unlinkQuiet(io, tmp);
|
||
throw refuse(`${QUEUE_REL} changed outside the queue lock (git?) since it was read; nothing changed; retry the same op`);
|
||
}
|
||
try {
|
||
if (exclusive) {
|
||
io.link(tmp, loc.queuePath);
|
||
unlinkQuiet(io, tmp);
|
||
} else {
|
||
io.rename(tmp, loc.queuePath);
|
||
}
|
||
} catch (err) {
|
||
unlinkQuiet(io, tmp);
|
||
throw new QueueError(`cannot replace ${QUEUE_REL} (${errno(err)}); nothing changed`, 1);
|
||
}
|
||
ctx.hook("renamed");
|
||
try {
|
||
io.fsyncDir(loc.docsDir);
|
||
} catch (err) {
|
||
throw new QueueError(`uncertain ${op} rev ${rev}: visible, durability not confirmed (${errno(err)})`, 3);
|
||
}
|
||
ctx.hook("dir-synced");
|
||
}
|
||
|
||
function commitWrite(ctx, loc, cur, doc, bytes, op, view, body, exclusive) {
|
||
const rev = doc.revision;
|
||
writeQueue(ctx, loc, cur, bytes, op, rev, exclusive);
|
||
try {
|
||
writeWitness(ctx, loc, doc, bytes);
|
||
} catch (err) {
|
||
const what = err.renamed ? "witness written, its directory fsync failed" : "witness not updated";
|
||
throw new QueueError(`uncertain ${op} rev ${rev}: durable, ${what} (${errno(err)})`, 3);
|
||
}
|
||
ctx.hook("witnessed");
|
||
const warn = writeView(ctx, loc, view.bytes, view.parts, body, op);
|
||
ctx.hook("viewed");
|
||
return warn;
|
||
}
|
||
|
||
function withLock(ctx, loc, { op = null, verb }, fn) {
|
||
checkPlatform(ctx.io, [loc.docsDir, loc.gitDir]);
|
||
const handle = acquire({ gitDir: loc.gitDir, io: ctx.io, proc: ctx.proc, op, verb, waitMs: ctx.lockWaitMs, stepMs: ctx.lockStepMs, hook: ctx.hook });
|
||
const res = { out: [], err: [], code: 0 };
|
||
let failure = null;
|
||
try {
|
||
ctx.hook("locked");
|
||
fn(res);
|
||
} catch (err) {
|
||
failure = err;
|
||
}
|
||
const msg = releaseOrWarn(handle, ctx.io);
|
||
// A refusal still reports what release found (8.4).
|
||
if (msg && failure instanceof Error) failure.message += `\nwarning: ${msg}`;
|
||
else if (msg) res.err.push(`warning: ${msg}`);
|
||
if (failure) throw failure;
|
||
return res;
|
||
}
|
||
|
||
function checkCallerOp(op) {
|
||
if (typeof op !== "string" || op === "") throw new QueueError("--op ID is required; choose it before the first attempt and reuse it on every retry", 4);
|
||
checkCallerOpId(op);
|
||
}
|
||
|
||
// --- mutations ---
|
||
|
||
function resolveFor(ctx, loc, cur, verb, args, cmp) {
|
||
if (verb === "add") {
|
||
const { path, anchor } = parseBriefSpec(args.brief);
|
||
return { brief: briefCheck(ctx, loc, path, anchor) };
|
||
}
|
||
if (verb === "set" && args.field === "brief") {
|
||
const { path, anchor } = parseBriefSpec(args.value);
|
||
return { brief: briefCheck(ctx, loc, path, anchor) };
|
||
}
|
||
if (verb === "move") {
|
||
const row = cur.state.rows.get(args.id);
|
||
if (args.candidate !== null && row?.state === "in-progress" && args.to === "in-review") return { candidate: resolveCandidate(ctx, loc, args.candidate) };
|
||
return { candidate: null };
|
||
}
|
||
if (verb === "accept-history") return { oldWitness: cmp.w ? { revision: cmp.w.revision, logDigest: cmp.w.logDigest } : null };
|
||
return {};
|
||
}
|
||
|
||
// Every logged verb except genesis. `yes` is accept-history's confirmation;
|
||
// it is not part of the op's identity.
|
||
export function mutate(opts, { verb, op, args, by, yes = false }) {
|
||
const ctx = makeCtx(opts);
|
||
const mismatch = actorMismatch(ctx, by);
|
||
try {
|
||
const res = mutateAs(ctx, { verb, op, args, by, yes });
|
||
if (mismatch) res.err.unshift(mismatch);
|
||
return res;
|
||
} catch (err) {
|
||
if (mismatch && err instanceof Error) err.message += `\n${mismatch}`;
|
||
throw err;
|
||
}
|
||
}
|
||
|
||
function mutateAs(ctx, { verb, op, args, by, yes }) {
|
||
checkCallerOp(op);
|
||
const actor = actorOf(ctx, by);
|
||
const cargs = canonArgs(verb, args);
|
||
if (verb === "genesis") return genesis(ctx, op, cargs, actor);
|
||
const loc = locate(ctx);
|
||
checkActorSeat(loc.top, actor);
|
||
if (!headHas(ctx, loc, QUEUE_REL)) throw refuse(`genesis not committed: HEAD has no ${QUEUE_REL}; commit genesis with \`scripts/queue-commit.sh --genesis\` first`);
|
||
return withLock(ctx, loc, { op, verb }, (res) => {
|
||
const cur = readLocked(ctx, loc);
|
||
const cmp = witnessStep(ctx, loc, cur, { verb, op, args: cargs });
|
||
for (const e of cmp.confirmed) res.err.push(`durable now, never acknowledged: ${e.op} by ${e.by} at ${e.at}`);
|
||
// Step 3: a recorded op answers with its receipt, before any view check.
|
||
const prior = cur.doc.log.find((e) => e.op === op);
|
||
if (prior) {
|
||
if (prior.verb !== verb || !sameJson(prior.args, cargs)) {
|
||
throw refuse(`op ${op} is recorded at rev ${prior.rev} as ${prior.verb} with other arguments; a new operation needs a new op id`);
|
||
}
|
||
const view = classifyView(readView(ctx, loc).text, cur.doc.log);
|
||
if (view.state === "stale") res.err.push(`warning: ${staleMessage(view, cur.doc.log)}`);
|
||
if (view.state === "unknown") res.err.push(`warning: ${unknownMessage(view)}`);
|
||
res.out.push(`${prior.result.receipt} (already recorded at rev ${prior.rev})`);
|
||
return;
|
||
}
|
||
if (verb === "accept-history") {
|
||
if (cmp.state !== "lost" && cmp.state !== "absent") throw refuse("history is not lost and the witness is present; accept-history has nothing to accept");
|
||
const range = `${lostMessage(cmp, cur.doc)}. ops in that range are no longer deduplicated`;
|
||
if (!yes) throw refuse(`${range}. Re-run with --yes to accept this file as the queue's history`);
|
||
res.err.push(range);
|
||
}
|
||
// Step 4.
|
||
const viewRead = readView(ctx, loc);
|
||
const view = classifyView(viewRead.text, cur.doc.log);
|
||
if (view.state === "unknown") throw refuse(unknownMessage(view));
|
||
if (view.state === "stale" && verb !== "accept-history") throw refuse(staleMessage(view, cur.doc.log));
|
||
// Step 5.
|
||
const resolved = resolveFor(ctx, loc, cur, verb, cargs, cmp);
|
||
const rev = cur.doc.revision + 1;
|
||
const entry = { rev, op, verb, args: cargs, by: actor, at: ctx.now(), semantics: SEMANTICS, result: null, viewSha: null };
|
||
const applied = applyEntry({ ...cur.state, revision: cur.doc.revision }, entry, resolved);
|
||
if (verb === "move" && applied.result.from === "briefed" && applied.result.to === "in-progress") {
|
||
const row = cur.state.rows.get(cargs.id);
|
||
if (!workingBriefMatches(loc, row.brief)) throw refuse(`row ${row.id}: brief differs from pinned blob; ask the lead to re-pin (${row.brief.path}, pinned ${row.brief.blob})`);
|
||
}
|
||
checkNewSeats(loc.top, cur.state.rows, applied.state.rows);
|
||
entry.result = applied.result;
|
||
const body = render(rowsArray(applied.state), rev);
|
||
entry.viewSha = sha256(body);
|
||
const doc = buildDoc(cur.doc.canonicalRoot, applied.state, [...cur.doc.log, entry]);
|
||
const bytes = Buffer.from(serialize(doc));
|
||
loadDoc(bytes);
|
||
const warn = commitWrite(ctx, loc, cur, doc, bytes, op, { bytes: viewRead.bytes, parts: view.parts }, body, false);
|
||
if (warn) res.err.push(`warning: ${warn}`);
|
||
res.out.push(entry.result.receipt);
|
||
});
|
||
}
|
||
|
||
// log[0] (8.2). Runs before canonicalRoot exists; its arguments are checked
|
||
// against this checkout instead.
|
||
function genesis(ctx, op, args, actor) {
|
||
if (!PRIVILEGED.has(actor)) throw refuse("only a privileged actor (jason or sage) may run genesis");
|
||
const loc = locate(ctx);
|
||
checkActorSeat(loc.top, actor);
|
||
if (args.root !== loc.top) throw refuse(`--root ${args.root} is not this checkout's toplevel ${loc.top}`);
|
||
checkBranch(loc, args.branch);
|
||
checkCode(loc.top);
|
||
return withLock(ctx, loc, { op, verb: "genesis" }, (res) => {
|
||
if (lstatOrNull(ctx.io, loc.queuePath) !== null) {
|
||
const cur = readLocked(ctx, loc);
|
||
const g = cur.doc.log[0];
|
||
if (g.op !== op || !sameJson(g.args, args)) throw refuse(`${QUEUE_REL} exists; genesis runs once`);
|
||
const cmp = witnessStep(ctx, loc, cur, { verb: "genesis", op, args });
|
||
for (const e of cmp.confirmed) res.err.push(`durable now, never acknowledged: ${e.op} by ${e.by} at ${e.at}`);
|
||
res.out.push(`${g.result.receipt} (already recorded at rev 0)`);
|
||
return;
|
||
}
|
||
if (headHas(ctx, loc, QUEUE_REL)) throw refuse(`HEAD already has ${QUEUE_REL}; genesis runs once`);
|
||
if (lstatOrNull(ctx.io, loc.witnessPath) !== null) throw refuse(`the witness .git/${WITNESS_NAME} exists without ${QUEUE_REL}; diagnose by hand`);
|
||
const mapEntry = headEntry(ctx, loc, args.map);
|
||
if (mapEntry === null || mapEntry.type !== "blob" || mapEntry.mode === "120000") throw refuse(`the migration map ${args.map} is not a committed file in HEAD`);
|
||
const map = parseMigrationMap(git(ctx, loc.top, ["cat-file", "blob", mapEntry.oid]).toString("utf8"));
|
||
const blobs = new Map();
|
||
for (const r of map.rows) {
|
||
if (r.brief !== null) blobs.set(r.id, briefCheck(ctx, loc, r.brief.path, r.brief.anchor).blob);
|
||
}
|
||
const viewRead = readView(ctx, loc);
|
||
const parts = viewRead.text === null ? null : splitView(viewRead.text);
|
||
if (parts === null) throw refuse(`${VIEW_REL} needs the two queue markers, once each and in order`);
|
||
const at = ctx.now();
|
||
const rows = genesisRows(map, blobs, op, at, actor);
|
||
checkNewSeats(loc.top, new Map(), new Map(rows.map((r) => [r.id, r])));
|
||
const body = render(rows, 0);
|
||
const result = {
|
||
mapBlob: mapEntry.oid, highWater: map.highWater, retired: [...map.retired].sort((a, b) => a - b), rows,
|
||
legacyView: parts.body, receipt: genesisReceipt(op, rows.length),
|
||
};
|
||
const entry = { rev: 0, op, verb: "genesis", args, by: actor, at, semantics: SEMANTICS, result, viewSha: sha256(body) };
|
||
const doc = { version: VERSION, canonicalRoot: args.root, revision: 0, rows, log: [entry] };
|
||
const bytes = Buffer.from(serialize(doc));
|
||
loadDoc(bytes);
|
||
const warn = commitWrite(ctx, loc, null, doc, bytes, op, { bytes: viewRead.bytes, parts }, body, true);
|
||
if (warn) res.err.push(`warning: ${warn}`);
|
||
res.out.push(result.receipt);
|
||
});
|
||
}
|
||
|
||
// --- reads without the lock (8.4) ---
|
||
|
||
function readUnlocked(ctx, loc) {
|
||
const { io } = ctx;
|
||
let w;
|
||
let bytes;
|
||
if (ctx.readOrder === "file-first") {
|
||
bytes = readOrNull(io, loc.queuePath);
|
||
ctx.hook("reader-between");
|
||
w = parseWitness(readOrNull(io, loc.witnessPath));
|
||
} else {
|
||
w = parseWitness(readOrNull(io, loc.witnessPath));
|
||
ctx.hook("reader-between");
|
||
bytes = readOrNull(io, loc.queuePath);
|
||
}
|
||
const notes = [];
|
||
if (bytes !== null) {
|
||
let d = null;
|
||
try { d = loadDoc(bytes); } catch (err) { if (!(err instanceof QueueError)) throw err; }
|
||
if (d !== null) {
|
||
checkCanonical(loc, d.doc.canonicalRoot, d.branch);
|
||
const cmp = compareWitness(d.doc, sha256(bytes), w);
|
||
if (cmp.state === "match" || cmp.state === "tail") {
|
||
if (cmp.state === "tail") notes.push(`rev ${d.doc.revision} visible, not confirmed durable`);
|
||
return { doc: d.doc, state: d.state, notes };
|
||
}
|
||
}
|
||
}
|
||
// Adverse: recheck under the lock before reporting anything.
|
||
let out = null;
|
||
const res = withLock(ctx, loc, { verb: "read" }, () => {
|
||
const cur = readLocked(ctx, loc);
|
||
const cmp = compareWitness(cur.doc, cur.sha, readWitness(ctx, loc));
|
||
if (cmp.state === "tail") notes.push(`unconfirmed tail: ${describeUnshown(cur.doc.log.slice(cmp.from))} visible but not confirmed durable; run \`${FIX} sync\``);
|
||
else if (cmp.state !== "match") throw lostRefusal(cmp, cur.doc);
|
||
out = { doc: cur.doc, state: cur.state };
|
||
});
|
||
return { ...out, notes: [...notes, ...res.err] };
|
||
}
|
||
|
||
function viewNotes(ctx, loc, log) {
|
||
const view = classifyView(readView(ctx, loc).text, log);
|
||
if (view.state === "stale") return [`warning: ${staleMessage(view, log)}`];
|
||
if (view.state === "unknown") return [`warning: ${unknownMessage(view)}`];
|
||
return [];
|
||
}
|
||
|
||
function readState(opts) {
|
||
const ctx = makeCtx(opts);
|
||
const loc = locate(ctx);
|
||
const r = readUnlocked(ctx, loc);
|
||
return { ctx, loc, ...r, notes: [...r.notes, ...viewNotes(ctx, loc, r.doc.log)] };
|
||
}
|
||
|
||
export function list(opts) {
|
||
const r = readState(opts);
|
||
const out = rowsArray(r.state).map((row) => `${row.id}\t${row.state}\t${row.owner}\t${row.piece}`);
|
||
return { out, err: r.notes, code: 0 };
|
||
}
|
||
|
||
export function show(opts, id) {
|
||
const r = readState(opts);
|
||
const row = r.state.rows.get(id);
|
||
if (!row) throw refuse(`no row ${id}`);
|
||
return { out: [JSON.stringify(row, null, 2)], err: r.notes, code: 0 };
|
||
}
|
||
|
||
export function next(opts, seat) {
|
||
const ctx = makeCtx(opts);
|
||
const name = seat ?? ctx.env.MOSAIC_AGENT_NAME;
|
||
if (name === undefined || name === "") throw refuse("next needs a seat: pass SEAT or set MOSAIC_AGENT_NAME");
|
||
checkName(name, "seat");
|
||
const r = readState(opts);
|
||
const n = nextFor(r.state.rows, name, (row) => workingBriefMatches(r.loc, row.brief));
|
||
if (n.action === "nothing") return { out: ["nothing"], err: r.notes, code: 0 };
|
||
const row = n.row;
|
||
const brief = row.brief ? `; brief ${row.brief.path} § ${row.brief.anchor}` : "";
|
||
const flag = n.briefDiffers ? "; brief differs from pinned blob; ask the lead to re-pin" : "";
|
||
return { out: [`${n.action} row ${row.id}: ${row.piece}${brief}${flag}`], err: r.notes, code: 0 };
|
||
}
|
||
|
||
// --- locked verbs that log nothing ---
|
||
|
||
function lockedRead(opts, verb, fn) {
|
||
const ctx = makeCtx(opts);
|
||
const loc = locate(ctx);
|
||
return withLock(ctx, loc, { verb }, (res) => {
|
||
const cur = readLocked(ctx, loc);
|
||
fn(ctx, loc, cur, res);
|
||
});
|
||
}
|
||
|
||
export function sync(opts, op = null) {
|
||
if (op !== null) checkCallerOp(op);
|
||
return lockedRead(opts, "sync", (ctx, loc, cur, res) => {
|
||
const cmp = witnessStep(ctx, loc, cur, { verb: "sync" });
|
||
if (cmp.confirmed.length === 0) res.out.push(`nothing to confirm: rev ${cur.doc.revision} is durable and witnessed`);
|
||
for (const e of cmp.confirmed) res.out.push(`durable now, never acknowledged: ${e.op} by ${e.by} at ${e.at}`);
|
||
res.err.push(...viewNotes(ctx, loc, cur.doc.log));
|
||
});
|
||
}
|
||
|
||
export function renderView(opts, { check = false } = {}) {
|
||
return lockedRead(opts, "render", (ctx, loc, cur, res) => {
|
||
witnessStep(ctx, loc, cur, { verb: "render" });
|
||
const log = cur.doc.log;
|
||
const viewRead = readView(ctx, loc);
|
||
const view = classifyView(viewRead.text, log);
|
||
if (view.state === "unknown") throw refuse(unknownMessage(view));
|
||
if (view.state === "current") {
|
||
res.out.push(`view current at rev ${cur.doc.revision}${check ? "" : "; nothing written"}`);
|
||
return;
|
||
}
|
||
if (check) throw refuse(staleMessage(view, log));
|
||
const body = render(rowsArray(cur.state), cur.doc.revision);
|
||
const warn = writeView(ctx, loc, viewRead.bytes, view.parts, body, "render");
|
||
if (warn) throw new QueueError(warn, 1);
|
||
res.out.push(`rendered rev ${cur.doc.revision} over rev ${view.shown}; newly shown: ${describeUnshown(view.unshown)}`);
|
||
});
|
||
}
|
||
|
||
function briefDrift(ctx, loc, rows) {
|
||
const drift = [];
|
||
for (const row of rowsArray({ rows })) {
|
||
if (row.brief === null || row.state === "done" || row.state === "parked") continue;
|
||
const e = headEntry(ctx, loc, row.brief.path);
|
||
if (e === null || e.type !== "blob" || e.mode === "120000") {
|
||
drift.push(`row ${row.id}: brief ${row.brief.path} is no longer a file in HEAD`);
|
||
} else if (e.oid !== row.brief.blob) {
|
||
drift.push(`row ${row.id}: brief ${row.brief.path} changed in HEAD (pinned ${row.brief.blob}, HEAD ${e.oid})`);
|
||
}
|
||
}
|
||
return drift;
|
||
}
|
||
|
||
export function verify(opts, { current = false } = {}) {
|
||
return lockedRead(opts, "verify", (ctx, loc, cur, res) => {
|
||
witnessStep(ctx, loc, cur, { verb: "verify" });
|
||
const view = classifyView(readView(ctx, loc).text, cur.doc.log);
|
||
if (view.state === "unknown") throw refuse(unknownMessage(view));
|
||
if (view.state === "stale") throw refuse(staleMessage(view, cur.doc.log));
|
||
if (current) {
|
||
const drift = briefDrift(ctx, loc, cur.state.rows);
|
||
if (drift.length) throw refuse(`brief drift against HEAD:\n${drift.join("\n")}`);
|
||
}
|
||
res.out.push(`ok verify rev ${cur.doc.revision}: file valid, witness matches, view current${current ? ", briefs match HEAD" : ""}`);
|
||
});
|
||
}
|
||
|
||
// 8.12 step 4. Reads only DIR/queue.json, DIR/QUEUE.md and the base: no git,
|
||
// no working directory, no lock. It does not certify witness continuity.
|
||
export function verifySnapshot(dir, { baseFile = null, baseAbsent = false } = {}) {
|
||
if (baseAbsent === (baseFile !== null)) throw new QueueError("verify --snapshot needs exactly one of --base-file F or --base-absent", 4);
|
||
const bytes = readFileOrRefuse(join(dir, "queue.json"));
|
||
const viewText = readFileOrRefuse(join(dir, "QUEUE.md")).toString("utf8");
|
||
const { doc } = loadDoc(bytes);
|
||
const view = classifyView(viewText, doc.log);
|
||
if (view.state !== "current") throw refuse(`snapshot QUEUE.md is not the render of rev ${doc.revision} (${view.state})`);
|
||
if (baseAbsent) {
|
||
if (doc.log.length !== 1) throw refuse(`the base is absent, so the snapshot must hold genesis alone; it holds revs 0..${doc.revision}`);
|
||
} else {
|
||
const base = loadDoc(readFileOrRefuse(baseFile)).doc;
|
||
if (base.canonicalRoot !== doc.canonicalRoot) throw refuse("the snapshot's canonicalRoot differs from the base's");
|
||
if (doc.revision < base.revision || logDigest(doc.log, base.revision) !== logDigest(base.log, base.revision)) {
|
||
throw refuse(`the snapshot's log does not extend the base's (base rev ${base.revision}, snapshot rev ${doc.revision})`);
|
||
}
|
||
}
|
||
return { out: [`ok verify --snapshot rev ${doc.revision}: pair valid, view current, ${baseAbsent ? "genesis alone" : "extends the base"}`], err: [], code: 0 };
|
||
}
|
||
|
||
function readFileOrRefuse(path) {
|
||
try {
|
||
return readFileSync(path);
|
||
} catch (err) {
|
||
throw refuse(`cannot read ${path} (${errno(err)})`);
|
||
}
|
||
}
|
||
|
||
export function snapshot(opts, outDir) {
|
||
const ctx = makeCtx(opts);
|
||
const dir = resolvePath(ctx.cwd, outDir);
|
||
let real;
|
||
try { real = realpathSync(dir); } catch { throw refuse(`--out ${dir} does not exist`); }
|
||
if (!lstatSync(real).isDirectory()) throw refuse(`--out ${dir} is not a directory`);
|
||
if (readdirSync(real).length !== 0) throw refuse(`--out ${dir} is not empty`);
|
||
return lockedRead(opts, "snapshot", (c, loc, cur, res) => {
|
||
if (real === loc.top || real.startsWith(`${loc.top}/`)) throw refuse(`--out ${dir} is inside the repository`);
|
||
witnessStep(c, loc, cur, { verb: "snapshot" });
|
||
const viewRead = readView(c, loc);
|
||
const view = classifyView(viewRead.text, cur.doc.log);
|
||
if (view.state !== "current") throw refuse(view.state === "stale" ? staleMessage(view, cur.doc.log) : unknownMessage(view));
|
||
writeTemp(c.io, join(real, "queue.json"), cur.bytes);
|
||
writeTemp(c.io, join(real, "QUEUE.md"), viewRead.bytes);
|
||
res.out.push(`snapshot rev ${cur.doc.revision}: queue.json ${cur.sha}, QUEUE.md ${sha256(viewRead.bytes)}`);
|
||
});
|
||
}
|
||
|
||
// --- the lock by hand ---
|
||
|
||
function unlockLoc(ctx) {
|
||
const loc = locate(ctx);
|
||
const bytes = readOrNull(ctx.io, loc.queuePath);
|
||
if (bytes !== null) {
|
||
let d = null;
|
||
try { d = loadDoc(bytes); } catch { /* an invalid file does not keep a dead lock in place */ }
|
||
if (d !== null) checkCanonical(loc, d.doc.canonicalRoot, d.branch);
|
||
}
|
||
checkPlatform(ctx.io, [loc.gitDir]);
|
||
return loc;
|
||
}
|
||
|
||
export function unlock(opts, { checkGateOnly = false } = {}) {
|
||
const ctx = makeCtx(opts);
|
||
const loc = unlockLoc(ctx);
|
||
if (checkGateOnly) return { out: [checkGate({ gitDir: loc.gitDir, io: ctx.io, proc: ctx.proc }).line], err: [], code: 0 };
|
||
const { result, warning } = unlockLock({ gitDir: loc.gitDir, io: ctx.io, proc: ctx.proc, hook: ctx.hook });
|
||
return { out: [result], err: warning ? [`warning: ${warning}`] : [], code: 0 };
|
||
}
|