// 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 { LOG_OP_RE, PRIVILEGED, SEMANTICS, UNRESOLVED_STATES, VERSION, applyEntry, attemptOf, buildDoc, canonArgs, checkCallerOpId, checkName, classifyView, countHeading, describeUnshown, genesisReceipt, genesisRows, gitBlobId, loadDoc, logDigest, manifestEntries, nextFor, parseBriefSpec, parseManifest, parseMigrationMap, render, rowsArray, sameJson, serialize, sha256, splitView, } from "./queue.mjs"; import { DEFAULT_DEADLINE_MS, REPO_API, bodyTooLong, callTool, checkComment, checkUser, classifyPost, credCheck, loginFor, markers, requestBody, } from "./review.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, deadlineMs: opts.deadlineMs ?? DEFAULT_DEADLINE_MS, }; } // --- 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 and review-outcome, which only a // request writes. `yes` confirms accept-history and review-abandon; 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 { if (verb === "review-outcome") throw refuse("a request records its own outcome; resolve or abandon the attempt instead"); if (verb === "review-resolve") checkResolve(ctx, { op, args, by }); let res = mutateAs(ctx, { verb, op, args, by, yes }); res = verb === "move" || verb === "review-request" ? requestStep(ctx, res) : strip(res); if (mismatch) res.err.unshift(mismatch); return res; } catch (err) { if (mismatch && err instanceof Error) err.message += `\n${mismatch}`; throw err; } } // `derived` is the outcome write: its op is the attempt's op plus .outcome. function mutateAs(ctx, { verb, op, args, by, yes = false, derived = false }) { if (derived) { if (!LOG_OP_RE.test(op) || op !== `${args.attempt}.outcome`) throw refuse(`outcome op ${op} is not its attempt's op plus .outcome`); } else { 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})`); Object.assign(res, { fresh: false, entry: prior, rows: cur.state.rows, top: loc.top }); return; } if (verb === "review-abandon" && !yes) { throw refuse("abandoning a request means a request comment may exist twice on the issue; re-run with --yes to record that"); } 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); Object.assign(res, { fresh: true, entry, rows: applied.state.rows, top: loc.top }); }); } // --- the request comment (8.9) --- const STATE_CODE = { posted: 0, failed: 1, abandoned: 1 }; function stateCode(state) { return STATE_CODE[state] ?? 3; } function strip(res) { return { out: res.out, err: res.err, code: res.code }; } function retryHint(row, round, attempt) { const where = `row ${row.id} round ${round.n} on #${round.issue}`; if (attempt.state === "posted") return `request ${attempt.op} is posted (${where}), comment ${attempt.comment}; nothing was sent again`; if (!UNRESOLVED_STATES.includes(attempt.state)) return `request ${attempt.op} is ${attempt.state} (${where}); nothing was sent again`; return `request ${attempt.op} is ${attempt.state} (${where}); nothing was sent again.\n${settleHint(row, round, attempt)}`; } // What to do about a request that may or may not be on the issue. function settleHint(row, round, attempt) { return [ `Look on #${round.issue} for a comment carrying ${markers(attempt.op, row.id, round.n, round.candidate.digest)[0]}.`, `If it is there: ${FIX} review resolve ${row.id} ${attempt.op} --comment ID --op OP. If not: a privileged actor runs ${FIX} review abandon ${row.id} ${attempt.op} --reason TEXT --yes --op OP.`, ].join("\n"); } // After a move or review-request is logged: post the request comment once, // then log what the transport said. A retried op never posts again. function requestStep(ctx, res) { const e = res.entry; const row = res.rows.get(e.args.id); const found = row ? attemptOf(row, e.op) : null; if (!found) return strip(res); const { round, attempt } = found; if (!res.fresh) { res.err.push(retryHint(row, round, attempt)); return { ...strip(res), code: stateCode(attempt.state) }; } const actor = e.by; ctx.hook("pre-send"); const body = requestBody(row, round, e.op); let why = credCheck(ctx.env, actor); if (why === null) why = checkUser(callTool(ctx, res.top, ["GET", "user"]), loginFor(actor)); let t; if (why !== null) { res.err.push(`request ${e.op} not sent: ${why}`); t = { outcome: "failed", status: null, comment: null, detail: "pre-send: the credential or account check failed" }; } else if (bodyTooLong(body)) { res.err.push(`request ${e.op} not sent: the comment would be longer than the limit`); t = { outcome: "failed", status: null, comment: null, detail: "pre-send: the request comment is too long" }; } else { const r = callTool(ctx, res.top, ["POST", `${REPO_API}/issues/${round.issue}/comments`, JSON.stringify({ body })]); ctx.hook("posted"); t = classifyPost(r); } ctx.hook("outcome"); let out; try { out = mutateAs(ctx, { verb: "review-outcome", op: `${e.op}.outcome`, args: { id: row.id, attempt: e.op, ...t }, by: actor, derived: true }); } catch (err) { const said = t.outcome === "posted" ? `posted comment ${t.comment}` : `${t.outcome} (${t.detail})`; throw new QueueError(`${[...res.out, ...res.err].join("\n")}\nuncertain ${e.op}: the transport said ${said}; that outcome is not recorded: ${err.message}`, 3); } const now = attemptOf(out.rows.get(row.id), e.op); const err = [...res.err, ...out.err]; if (UNRESOLVED_STATES.includes(now.attempt.state)) err.push(settleHint(out.rows.get(row.id), now.round, now.attempt)); return { out: [...res.out, ...out.out], err, code: stateCode(now.attempt.state) }; } // review resolve checks the comment it names before logging anything: on // the round's issue, by the requester's account, with both markers. function checkResolve(ctx, { op, args, by }) { checkCallerOp(op); const actor = actorOf(ctx, by); const a = canonArgs("review-resolve", args); const st = readStateWith(ctx); if (st.doc.log.some((e) => e.op === op)) return; const row = st.state.rows.get(a.id); const found = row ? attemptOf(row, a.attempt) : null; // Anything the log would refuse anyway is refused there, with no request. if (!found || !UNRESOLVED_STATES.includes(found.attempt.state) || row.state === "done") return; if (actor !== row.owner && !PRIVILEGED.has(actor)) return; const why = credCheck(ctx.env, actor); if (why) throw refuse(`cannot check comment ${a.comment}: ${why}`); const r = callTool(ctx, st.loc.top, ["GET", `${REPO_API}/issues/comments/${a.comment}`]); const { round } = found; const bad = checkComment(r, { id: a.comment, issue: round.issue, op: a.attempt, row: row.id, round: round.n, digest: round.candidate.digest, author: loginFor(found.attempt.by), }); if (bad) throw new QueueError(bad.reason, bad.code); } // review verify-commit: does REF's tree hold exactly the candidate? export function verifyCommit(opts, id, ref) { const st = readState(opts); const { ctx, loc } = st; const row = st.state.rows.get(id); if (!row) throw refuse(`no row ${id}`); const cur = row.review?.rounds.at(-1); if (!cur) throw refuse(`row ${id} has no review round`); const tree = git(ctx, loc.top, ["rev-parse", "--verify", "--quiet", "--end-of-options", `${ref}^{tree}`], { allowFail: true }); if (tree === null) throw refuse(`${ref} is not a commit or tree here`); const have = lsTree(ctx, loc.top, tree.toString().trim()); const c = cur.candidate; const bad = []; let count = 0; if (c.kind === "manifest") { for (const { digest, path } of manifestEntries(c.text)) { count++; const h = have.get(path); if (!h || h.type !== "blob") bad.push(`${path}: missing`); else if (sha256(git(ctx, loc.top, ["cat-file", "blob", h.oid])) !== digest) bad.push(`${path}: content differs`); } } else { if (git(ctx, loc.top, ["cat-file", "-e", `${c.digest}^{commit}`], { allowFail: true }) === null) { throw refuse(`candidate commit ${c.digest} is no longer in this repository; a commit candidate stays checkable only while a ref keeps it`); } const parent = git(ctx, loc.top, ["rev-parse", "--verify", "--quiet", `${c.digest}^1`], { allowFail: true }); const base = parent === null ? git(ctx, loc.top, ["hash-object", "-t", "tree", "/dev/null"]).toString().trim() : parent.toString().trim(); const raw = git(ctx, loc.top, ["diff-tree", "-r", "--no-renames", "-z", base, c.digest]).toString().split("\0"); for (let i = 0; i + 1 < raw.length; i += 2) { const m = /^:(\d{6}) (\d{6}) ([0-9a-f]{40}) ([0-9a-f]{40}) ([A-Z])/.exec(raw[i]); if (!m) throw new QueueError(`git diff-tree printed a line this check does not read: ${JSON.stringify(raw[i].slice(0, 80))}`, 1); const path = raw[i + 1]; count++; const h = have.get(path); if (m[5] === "D") { if (h) bad.push(`${path}: deleted in the candidate, present here`); } else if (!h) { bad.push(`${path}: missing`); } else if (h.oid !== m[4] || h.mode !== m[2]) { bad.push(`${path}: ${h.oid !== m[4] ? "content" : "mode"} differs`); } } } if (bad.length) throw refuse(`${ref} does not match row ${id} round ${cur.n}'s candidate:\n${bad.join("\n")}`); return { out: [`ok row ${id} round ${cur.n}: ${ref} matches the ${c.kind} candidate (${count} paths)`], err: st.notes, code: 0 }; } function lsTree(ctx, top, tree) { const map = new Map(); for (const e of git(ctx, top, ["ls-tree", "-r", "-z", "--full-tree", tree]).toString().split("\0")) { const m = /^(\d{6}) (\w+) ([0-9a-f]{40})\t(.*)$/s.exec(e); if (m) map.set(m[4], { mode: m[1], type: m[2], oid: m[3] }); } return map; } // 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) { return readStateWith(makeCtx(opts)); } function readStateWith(ctx) { 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 }; }