feat(bus): approve and activate an exact product PRD version (#1839)

BUILD-LOG.md

@@ -6007,3 +6007,5 @@ Owner directed static-only fallback after actual nativepriority drift: no new pr
 2026-11-04T02:45Z | Claude Sonnet 5.5 implementation sub-agent, t3/lifecycle-bootstrap-fix-round | #797 / Gitea #1839 lifecycle Task 2 fix round | One fix round on c55849806 for review findings C1, C2, I1, I2, I3, I4; Minor items, Tasks 3-8, roles/*.json, ROUTED and all migrations untouched (no schema change). C1: seeding no longer follows a committed .mosaic link. Linked ancestors and non-regular metadata are refused in the source and in the clone, and the seed and the marker temp are written with O_EXCL|O_NOFOLLOW and renamed, so a planted link is never followed. C2: the clone goes into an exclusively created code.tmp backed by a code.tmp.claim receipt; an unclaimed or foreign-content code.tmp is kept untouched with bootstrap-uncertain, a claimed clone left by a kill is adopted exactly once, and no deterministic path is removed recursively. I1: the #800 freshness check runs before every effect including recovery effects; a mismatch appends setup wait request-plan-stale and does nothing further, evidence kept. I2: managedState classifies by layout (entry root and repo match the managed paths) and reads readiness from the marker and the committed activate receipt, so a missing bootstrap.json no longer admits an unfinished product; the launcher and trackerFor already call managedState and needed no change. I3: new packages/bus/src/request-host.mjs wires preparePlans and admitPlan in the real host from the organization document and revision, the booted tracker config and the staff the organization names; a missing tracker root or staff is a named setup wait (request-tracker-unavailable, request-staff-unavailable, request-organization-unavailable, request-organization-changed) and creates nothing; process.mjs now forwards trackers to startBroker. A normal boot with a tracker root and named staff reaches a bootstrap selection and bootstraps the project. I4: the Task 1 existing-project comparison was reversed and is fixed in localPlan with regressions in request-intake. RED/GREEN: each regression was failing on c55849806 and passes now; the regression set was mutation-checked by reverting each guard, with two exceptions recorded honestly: the executor-level metadataProblem mutant (c1-exec) still passes because git status is a second guard, and the i3 admitPlan unit test was not mutation-checked. Corrections: the committed project-file link test first failed at boot because the config loader refuses the link (test now commits the link after boot); the c2-claim mutant first survived because clone verification is a second guard (two unclaimed-code.tmp tests added); boot recovery rewrote a complete marker for a deleted reserved one, fixed by requiring the marker to prove ownership before every post-reserve effect; the i2 assertion on admission revision null was dropped because that Minor item is out of scope. First full bus run had one failure, model-final-completion 656 FIFO niceness (startNice 16 !== 19), unrelated to this change; the file passed 21/21 three times alone and the full rerun was 1477/1477. GREEN: brief trio 79/79, kept-green trio 49/49, bus 1477/1477 (was 1456), business 305/305 (was 304), project+seat 267/267, stack-release 225/225, zero skips. Limits: the selected tracker board is unverified until the first task is created; after a post-effect stale wait nothing resolves automatically, the request waits and each resume re-checks; a kill mid-clone leaves code.tmp kept and uncertain, never deleted; the prepareExisting/listProjects wiring is out of scope; admission revision stays null. Logs /mnt/storage/scratch/tmp/fix797/. Not executed on any running instance; local commit only, no push/merge/PR/deployment.
 
 2026-11-04T02:51:19Z | Lead (Claude Opus 5.5, T3 c5c1e8ba) | #797 / Gitea #1839 lifecycle Task 2 landing | Task 2 c55849806 (Sonnet 5.5 xhigh): exact-add organization writer, journaled bootstrap steps, 3o->3p append-only migration, receipt-backed runtime admission. Review R1 (GPT-6.1 Sol): NEEDS FIXES, C1 seed write through a committed .mosaic link, C2 recursive code.tmp delete, I1 #800 freshness only at reservation, I2 missing marker admitted, I3 no host preparation/admission wiring, I4 Task 1 existing-project root/repo comparison reversed. Fix round 883003041 (Sonnet 5.5 xhigh): O_EXCL|O_NOFOLLOW publish, claimed code.tmp, per-effect fresh(), layout-based managedState, request-host.mjs wiring, corrected localPlan. Gates on 883003041: focused 79/79, kept-green 49/49, bus 1477/1477 (first run had one model-final-completion startNice flake, reviewer found no causal path), business 305/305, project+seat 267/267, stack-release 225/225, zero skips (/mnt/storage/scratch/tmp/fix797). Review R2: C1 C2 I3 I4 resolved; I1 (org write during intent step Git reads still lets a stale intent write), I2 (failed final marker rename or later marker removal leaves admission ready) and new N1 (preexisting <dataRoot>/working ancestor link sends the working repo outside dataRoot) open. Lead decision: land under the MVP one-round rule; all three need write access to dataRoot, a concurrent organization edit or an I/O fault, and none is reachable by a sandboxed worker. They form one hardening follow-up that must land before bootstrap is exercised on .53. Minors are tracker #801.
+
+2026-11-04T04:09:35Z | Codex implementation sub-agent, t3/lifecycle-prd-approval-activation | #797 / Gitea #1839 lifecycle Task 4 core | Before: parent 6dd4f92b0 has intake/bootstrap and schema 3p, with no exact product PRD activation authority. After: bounded committed candidate reads (64 requirements, 32 KiB PRD, 16 KiB requirement text, 256 KiB aggregate), strict records, server-derived blob IDs, selected-path HEAD equality and current owner spec; PM-only record.write submission raises the existing nonblocking prd.approve action. Autonomous routing requires a distinct authorized technical arbiter instance and definition; unusable/self routing records setup waits. Guided approval uses the same action without a routing citation, while retaining its spec/mode binding. Host-only catalog activation rereads candidate/spec/mode/current role authority and the exact approving resolution, independent of run liveness; snapshot, activation event, continuation and receipt share one SQLite transaction, with deterministic idempotent replay. Broker.requirementsFor keeps the agreed authenticated consumer signature; no tasks are manufactured. Mode/spec changes invalidate admissions and revoke snapshots; owner-only prd.revoke.console is revision/op checked and replayable, and request.draft.console reads only the bound committed candidate. Runtime recovers from journal/records alone and drains queued lifecycle work on close. New append-only migration 3p to 3q widens request event kinds, preserving rows, sequences and all guards; current version/fixture pins updated, table/guard counts stay 30/114. RED on 6dd4f92b0 with the final new tests: 0/27 pass, 27 expected missing-feature failures (baseline-new-tests-red.log). Corrections: the first fixture tried to initialize an already bootstrapped records repo, fixed before feature RED; journal composition initially attempted nested transactions, now composed inside one host transaction; new automatic closure continuation needed shutdown draining; one revoke assertion needed awaited rejection. First full bus run 1496/1498: the old two-verb whitelist pin was updated to the authorized lifecycle verbs, and unchanged #656 FIFO child measured niceness 16 rather than 19. Both files pass 38/38 on exact parent 6dd4f92b0 and on this branch; the FIFO failure did not reproduce in isolation. Final GREEN: new suites 27/27, requested kept-green trio 21/21, bus 1498/1498, project 50/50, business 305/305, all zero skips; import-closure checks 3/3, git diff --check clean. Full logs under /mnt/storage/scratch/tmp/wZ4am1 (focused-final.log, kept-green-final.log, bus-final.log, project.log, business.log; earlier runs retained). Boundaries: Console HTTP/browser files, revoke route and request-panel UI deferred until Task 3 integration; tracker/task consumer wiring belongs to Task 5; built-in legacy binding is deliberately unset here. The Stack PRD amendment remains proposed. Queue command refused this linked worktree, so queue movement stays with the canonical coordinator. One local commit authorized, no push, merge, deployment, live instance changes, role/contract edits or secret access.

docs/SESSIONS.md

@@ -1997,3 +1997,4 @@ are never rewritten or removed; corrections are new entries.
 
 2026-11-04 | Claude Sonnet 5.5 implementation sub-agent, t3/lifecycle-bootstrap-fix-round | #797 / Gitea #1839 lifecycle Task 2 fix round | C1 link-confined seed and marker writes, C2 receipt-backed code.tmp kept as bootstrap-uncertain when unproven, I1 freshness recheck before every effect, I2 layout-based managed classification, I3 host preparePlans/admitPlan wired (named setup waits), I4 existing-project comparison. Brief 79/79, kept-green 49/49, bus 1477/1477, business 305/305, project+seat 267/267, stack-release 225/225, zero skips; no schema change; local commit only, no push.
 2026-11-04 | Lead (Claude Opus 5.5, T3 c5c1e8ba) | #797 lifecycle Task 2 review, fix round and landing | R1 needs fixes, fix 883003041, R2 left I1/I2/N1 residuals as a pre-.53 hardening follow-up; landed on refactor
+2026-11-04 | Codex implementation sub-agent, t3/lifecycle-prd-approval-activation | #797 / Gitea #1839 lifecycle Task 4 core | Exact committed product PRD approval/activation, authenticated requirementsFor, owner revoke/mode/draft broker controls, atomic append-only snapshots/receipts and shutdown-safe recovery; migration 3p to 3q. New 27/27, kept-green 21/21, bus 1498/1498, project 50/50, business 305/305, zero skips; logs /mnt/storage/scratch/tmp/wZ4am1. Console HTTP/browser slice deferred for Task 3 integration. Local commit only, no push.

packages/bus/schema.sql

@@ -535,7 +535,7 @@ CREATE TABLE request_events (
   business TEXT NOT NULL,
   request TEXT NOT NULL REFERENCES events(id),
   revision INTEGER NOT NULL CHECK (revision >= 1),
-  kind TEXT NOT NULL CHECK (kind IN ('opened','selection.proposed','selection.staged','selection.confirmed','setup.wait','bootstrap.stale','bootstrap.completed','cancelled')),
+  kind TEXT NOT NULL CHECK (kind IN ('opened','selection.proposed','selection.staged','selection.confirmed','setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','cancelled')),
   actor_role TEXT, actor_run TEXT,
   at TEXT NOT NULL,
   body TEXT NOT NULL CHECK (json_valid(body) AND json_type(body) = 'object'),

packages/bus/src/broker.mjs

@@ -123,6 +123,10 @@ export const HOST_STDIN_MAX = 65536;
 export const HOST_TIMEOUT_DEFAULT = 60;
 export const HOST_TIMEOUT_MAX = 900;
 const sha256 = (x) => createHash('sha256').update(x).digest('hex');
+// Canonical candidate bindings remain pure so the broker's sandbox import closure stays fixed.
+const orderedPrd = value => Array.isArray(value) ? value.map(orderedPrd) : object(value)
+  ? Object.fromEntries(Object.keys(value).sort().map(k => [k, orderedPrd(value[k])])) : value;
+const digestPrd = value => sha256(JSON.stringify(orderedPrd(value)));
 // D4: what a host.sudo decision binds, from the raise's args and again from the call's.
 const hostRun = (host, argv, stdin) => ({
   host,
@@ -156,7 +160,7 @@ export const USER_VERBS = Object.freeze(['project.list', 'thread.create', 'threa
   'thread.share', 'thread.unshare', 'thread.shares']);
 // Row 252 (#1753, design sections 2-3): the only verbs a Console owner cap reaches. Every other verb
 // refuses it human-required before its own checks, and these refuse every other cap.
-export const CONSOLE_VERBS = Object.freeze(['decision.preview.console', 'decision.resolve.console']);
+export const CONSOLE_VERBS = Object.freeze(['decision.preview.console', 'decision.resolve.console', 'prd.revoke.console', 'request.mode.console', 'request.draft.console']);
 // Row 275 (193b1b, #1780, row 193 design 4.3 and 9): the user verbs a viewer grant reaches. They
 // read, so a live grant at any level covers them. thread.create, thread.open and thread.send need a
 // collaborator grant on the project and instance, so a viewer gets access-refused on each.
@@ -572,6 +576,12 @@ export class Broker {
         },
         projectPaths: business => Object.values(this.#business(business).projects ?? {}).flatMap(p => [p.root, p.repo]).filter(p => typeof p === 'string'),
         raiseDecision: (identity, args) => this.#dispatch(identity, 'decision.raise', args),
+        raisePrdDecision: (identity, args, spec) => this.#dispatch(identity, 'decision.raise', args, spec),
+        prdBusiness: business => structuredClone(this.#business(business)),
+        prdAuthority: (state, binding) => this.#prdAuthority(state, binding),
+        prdApproval: (state, binding) => this.#prdApproval(state, binding),
+        prdDecision: (business, decision) => this.#decisionView(this.#store.get('SELECT * FROM decisions WHERE business=? AND id=?', business, decision)),
+        recordPrd: (identity, project, decision) => this.#consumeAuthority(identity, 'record.write', { target: project }, { operation: 'prd.submit', decision }),
         readDecision: (business, decision) => {
           const d = this.#store.get('SELECT * FROM decisions WHERE business=? AND id=?', business, decision);
           if (!d || d.action !== 'project.bootstrap' || d.route_to !== 'human') fail('request-selection-refused');
@@ -592,18 +602,68 @@ export class Broker {
     if (!s || s.business !== state.business) fail('request-access-refused');
     if (s.reader) fail('read-only');
     if (s.tester) fail('tester-refused');
-    if (s.user || s.console) fail('human-required');
+    if (s.user || (s.console && !['mode','draft'].includes(operation))) fail('human-required');
+    if (['mode','draft'].includes(operation) && !s.console) fail('human-required');
     if (s.human) {
-      this.#human(s);
+      if (!s.console) this.#human(s);
       if (s.human !== state.owner || s.human !== this.#business(s.business).human) fail('request-access-refused');
-      if (operation === 'propose') fail('agent-required');
+      if (['propose','prd.submit'].includes(operation)) fail('agent-required');
     } else {
       this.#agent(s);
       if (s.role !== state.pm || this.#business(s.business).roles[s.role]?.definition !== 'pm') fail('request-access-refused');
       // No separate authority: intake preparation consumes the PM's existing message.send grant.
-      if (operation !== 'get') this.#checkAuthority(s, 'message.send', { target: state.pm });
+      if (operation === 'prd.submit') this.#checkAuthority(s, 'record.write', { target: state.project });
+      else if (operation !== 'get') this.#checkAuthority(s, 'message.send', { target: state.pm });
     }
   }
+  #prdAuthority(state, binding) {
+    if (!state || !binding) fail('prd-authority-refused');
+    const b = this.#business(state.business), pm = b.roles[state.pm];
+    if (b.human !== state.owner || pm?.definition !== 'pm' || !pm.authority?.withinRole.includes('record.write')) fail('prd-authority-refused');
+    for (const author of binding.authors ?? []) {
+      const r = b.roles[author.instance];
+      if (r?.definition !== author.definition || !r.authority?.withinRole.includes('record.write')) fail('prd-authority-refused');
+    }
+    let route = { instance: 'human', definition: 'human' };
+    if (state.mode === 'autonomous') {
+      const instance = b.arbiters?.technical, arbiter = b.roles[instance];
+      if (!instance || !arbiter || !arbiter.authority?.withinRole.includes('decision.resolve.technical')) fail('prd-arbiter-unavailable');
+      if (instance === state.pm || arbiter.definition === pm.definition || binding.authors?.some(a => a.instance === instance || a.definition === arbiter.definition)) fail('prd-self-approval');
+      route = { instance, definition: arbiter.definition };
+    }
+    if (binding.route && digestPrd(binding.route) !== digestPrd(route)) fail('prd-authority-refused');
+    return route;
+  }
+  #prdApproval(state, binding) {
+    const route = this.#prdAuthority(state, binding), d = this.#store.get('SELECT * FROM decisions WHERE business=? AND id=?', state.business, binding.decision);
+    const r = d && this.#closed(d.id), raised = d && this.#raised(state.business, d.id);
+    if (!d || d.action !== 'prd.approve' || d.project !== state.project || d.raised_by_role !== state.pm || d.route_to !== route.instance ||
+        d.blocking !== 0 || d.task_ref !== null || d.requirement_ref !== null || d.class !== 'gated' ||
+        raised?.target !== `prd-${digestPrd(binding.candidate)}` || raised?.approvalChoice !== 'yes' ||
+        (state.mode === 'autonomous' ? d.spec_commit !== binding.candidate.spec_commit || d.spec_path !== SPEC_PATH : d.spec_commit !== null)) fail('prd-authority-refused');
+    if (!r) fail('prd-not-approved');
+    if (r.op !== 'resolved') fail('prd-decision-closed');
+    if (r.choice !== raised.approvalChoice) fail('prd-changes-requested');
+    if (route.instance === 'human') {
+      if (r.by !== state.owner || !['cli','console'].includes(r.via) || (r.via === 'console' && !this.#consoleInput(state.business, d.id, r))) fail('prd-authority-refused');
+    } else if (r.via !== 'broker' || !(this.#store.get("SELECT 1 FROM role_claims WHERE business=? AND role=? AND holder_run=? AND op='claim'", state.business, route.instance, r.by) ||
+      this.#store.get("SELECT 1 FROM thread_claims c JOIN threads t ON t.id=c.thread WHERE c.business=? AND t.instance=? AND c.holder_run=? AND c.op='claim'", state.business, route.instance, r.by))) fail('prd-authority-refused');
+    return { decision: d.id, resolution_seq: r.seq, by: r.by, via: r.via };
+  }
+  async requirementsFor(cap, args) {
+    const s = this.#session(cap);
+    try {
+      keys(args, ['project_key','requirement_snapshot','requirement'], ['project_key']);
+      this.#agent(s); this.#processingGate(s, 'requirements');
+      if (!this.#business(s.business).projects?.[args.project_key]) fail('request-project-refused');
+      if (!this.#requests?.catalog) fail('requirement-snapshot-required');
+      const result = await this.#requests.catalog.resolve({ business: s.business, project: args.project_key,
+        ...(args.requirement_snapshot === undefined ? {} : { snapshot: args.requirement_snapshot }) });
+      this.#agent(s); this.#processingGate(s, 'requirements');
+      if (args.requirement !== undefined && !result.requirements.some(r => r.id === args.requirement)) fail('unknown-requirement');
+      return result;
+    } catch (e) { throw this.#refused(s, new BusError(e.code ?? 'requirement-snapshot-refused'), { args }); }
+  }
   #requestOrigin(s, request) {
     const b = this.#business(s.business);
     const e = this.#store.get("SELECT * FROM events WHERE business=? AND id=? AND kind='human.input' AND actor_role IS NULL AND actor_run IS NULL", s.business, request);
@@ -4976,11 +5036,18 @@ export class Broker {
         return { ended: ended.map((h) => ({ thread: h.thread, session: h.session, run: h.holder_run })) };
       }
       this.#processingGate(s, request.verb);
-      if (['request.open','request.get','request.projects','request.propose'].includes(request.verb)) {
+      if (['prd.revoke.console','request.mode.console','request.draft.console'].includes(request.verb)) {
+        if (!s.console || s.human !== this.#business(s.business).human) fail('human-required');
+        if (!this.#requests?.catalog) fail('request-not-configured');
+        return Promise.resolve().then(() => request.verb === 'request.draft.console' ? this.#requests.draftConsole(s, structuredClone(args)) : request.verb === 'prd.revoke.console'
+          ? this.#requests.catalog.revoke(s, structuredClone(args)) : this.#requests.modeConsole(s, structuredClone(args)))
+          .catch(e => { throw this.#refused(s, new BusError(e.code ?? 'request-failed'), request); });
+      }
+      if (['request.open','request.get','request.projects','request.propose','prd.submit'].includes(request.verb)) {
         if (s.reader) fail('read-only');
         if (s.tester) fail('tester-refused');
         if (!this.#requests) fail('request-not-configured');
-        const method = request.verb.slice('request.'.length);
+        const method = request.verb === 'prd.submit' ? 'submit' : request.verb.slice('request.'.length);
         return Promise.resolve().then(() => this.#requests[method](s, structuredClone(args)))
           .then(result => { this.#secretCheck(result); return result; })
           .catch(e => { throw this.#refused(s, e instanceof BusError ? e : new BusError(e.code ?? 'request-failed'), request); });
@@ -6293,6 +6360,8 @@ export class Broker {
           );
           this.#route(s.business, mid, d.raised_by_role, session);
         }
+        if (op !== 'seen' && d.action === 'prd.approve' && this.#requests)
+          setImmediate(() => { Promise.resolve().then(() => this.#requests.decisionClosed?.(s.business, d.id)).catch(() => {}); });
         const answer = viaConsole ? { id: d.id, op, choice: a.choice } : { id: d.id, op };
         return grant ? { ...answer, grant: 'started' } : answer;
       }
@@ -7474,6 +7543,7 @@ export class Broker {
   // The notice sits in no thread, so the user's thread view shows none of this traffic. Nothing is
   // launched here: a notice to a role no run holds waits for a claim and does not tell launch.by.
   #specNotice(business, project, commit, base) {
+    this.#requests?.catalog?.invalidate(business, project, 'requirement-snapshot-revoked', true);
     const b = this.#business(business);
     const roles = Object.keys(b.roles).filter((r) => SPEC_REVIEWERS.includes(b.roles[r].definition));
     if (!roles.length) return;

packages/bus/src/migrations/3p-3q.sql

@@ -0,0 +1,45 @@
+-- Task 4: preserve all evidence and guards, widen only lifecycle event kinds.
+DROP TRIGGER request_events_origin;
+DROP TRIGGER request_events_revision;
+DROP TRIGGER request_effects_binding;
+DROP TRIGGER request_events_no_update;
+DROP TRIGGER request_events_no_delete;
+DROP TRIGGER request_events_no_replace;
+DROP INDEX request_events_request;
+ALTER TABLE request_events RENAME TO request_events_3p;
+CREATE TABLE request_events (
+  seq INTEGER PRIMARY KEY AUTOINCREMENT,
+  business TEXT NOT NULL,
+  request TEXT NOT NULL REFERENCES events(id),
+  revision INTEGER NOT NULL CHECK (revision >= 1),
+  kind TEXT NOT NULL CHECK (kind IN ('opened','selection.proposed','selection.staged','selection.confirmed','setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','cancelled')),
+  actor_role TEXT, actor_run TEXT,
+  at TEXT NOT NULL,
+  body TEXT NOT NULL CHECK (json_valid(body) AND json_type(body) = 'object'),
+  CHECK ((actor_role IS NULL) = (actor_run IS NULL)),
+  UNIQUE (business,request,revision)
+) STRICT;
+INSERT INTO request_events(seq,business,request,revision,kind,actor_role,actor_run,at,body)
+  SELECT seq,business,request,revision,kind,actor_role,actor_run,at,body FROM request_events_3p ORDER BY seq;
+DELETE FROM sqlite_sequence WHERE name='request_events';
+INSERT INTO sqlite_sequence(name,seq) SELECT 'request_events', seq FROM sqlite_sequence WHERE name='request_events_3p';
+DROP TABLE request_events_3p;
+CREATE INDEX request_events_request ON request_events (business,request,seq);
+CREATE TRIGGER request_events_origin BEFORE INSERT ON request_events
+  WHEN NOT EXISTS (SELECT 1 FROM events e JOIN messages m ON m.id=e.subject AND m.id=json_extract(e.body,'$.message')
+    WHERE e.id=NEW.request AND e.business=NEW.business AND m.business=e.business AND e.kind='human.input'
+      AND e.actor_role IS NULL AND e.actor_run IS NULL AND json_extract(e.body,'$.kind')='instruction' AND m.from_role='human')
+  BEGIN SELECT RAISE(ABORT, 'request-origin-refused'); END;
+CREATE TRIGGER request_events_revision BEFORE INSERT ON request_events
+  WHEN NEW.revision <> 1 + coalesce((SELECT max(revision) FROM request_events WHERE business=NEW.business AND request=NEW.request),0)
+    OR (NEW.revision=1) <> (NEW.kind='opened')
+  BEGIN SELECT RAISE(ABORT, 'request-revision-conflict'); END;
+CREATE TRIGGER request_effects_binding BEFORE INSERT ON request_effects
+  WHEN NOT EXISTS (SELECT 1 FROM request_events WHERE business=NEW.business AND request=NEW.request)
+    OR EXISTS (SELECT 1 FROM request_effects WHERE business=NEW.business AND request=NEW.request AND op=NEW.op AND (digest<>NEW.digest OR kind<>NEW.kind))
+    OR (NEW.stage<>'prepared' AND NOT EXISTS (SELECT 1 FROM request_effects WHERE business=NEW.business AND request=NEW.request AND op=NEW.op AND stage='prepared'))
+    OR EXISTS (SELECT 1 FROM request_effects WHERE business=NEW.business AND request=NEW.request AND op=NEW.op AND stage IN ('committed','refused'))
+  BEGIN SELECT RAISE(ABORT, 'request-effect-conflict'); END;
+CREATE TRIGGER request_events_no_update BEFORE UPDATE ON request_events BEGIN SELECT RAISE(ABORT, 'request_events is append-only'); END;
+CREATE TRIGGER request_events_no_delete BEFORE DELETE ON request_events BEGIN SELECT RAISE(ABORT, 'request_events is append-only'); END;
+CREATE TRIGGER request_events_no_replace BEFORE INSERT ON request_events WHEN EXISTS (SELECT 1 FROM request_events WHERE seq=NEW.seq OR (business=NEW.business AND request=NEW.request AND revision=NEW.revision)) BEGIN SELECT RAISE(ABORT, 'request_events is append-only'); END;

packages/bus/src/request-lifecycle.mjs

@@ -6,6 +6,7 @@ import { exact, object, id, key, text, digest, validatePlan, refuse } from './re
 
 export function createRequestLifecycle({ journal, services }) {
   if (!journal || !services || !['origin','authorize','raiseDecision','readDecision','project'].every(k => typeof services[k] === 'function')) refuse('request-services-unavailable');
+  let closing = false;
   const pending = new Map();
   const serial = (business, request, fn) => {
     const k = business + '/' + request;
@@ -200,9 +201,79 @@ export function createRequestLifecycle({ journal, services }) {
     return append(state, 'bootstrap.completed', { digest: binding.digest, project: done.project, chat_project: done.chat_project,
       spec_commit: done.spec_commit, organization_revision: done.organization_revision });
   }
+  async function submit(identity, args) {
+    exact(args, ['request','revision','op','prd_version','records_commit','prd_path','requirement_ids']);
+    if (!id(args.request) || !id(args.op) || !Number.isSafeInteger(args.revision) || args.revision < 1) refuse('invalid-request');
+    return serial(identity.business, args.request, async () => {
+      let state = stateOf(identity.business, args.request); authorize(identity, state, 'prd.submit');
+      const proposal_digest = digest(args), old = journal.proposal(state.business, state.request, args.op);
+      if (old) {
+        if (old.proposal_digest !== proposal_digest) refuse('request-operation-conflict');
+        if (!old.candidate) return state;
+        return { candidate: old.candidate, decision: services.prdDecision(state.business, old.decision) };
+      }
+      if (state.revision !== args.revision) refuse('request-revision-conflict');
+      if (!['discovery','prd-review','planning'].includes(state.phase) || state.wait?.code === 'prd-approval') refuse('request-phase-refused');
+      const { revision, op, ...input } = args;
+      const material = await services.readCandidate({ ...input, project: state.project, mode_revision: state.mode_revision }, state.business);
+      const recheck = () => {
+        const now = stateOf(state.business, state.request); authorize(identity, now, 'prd.submit');
+        if (now.revision !== args.revision) refuse('request-revision-conflict');
+        if (journal.projectState(state.business, state.project)?.revision !== state.mode_revision) refuse('prd-admission-revoked');
+        return now;
+      };
+      return journal.transaction(tx => {
+        state = recheck();
+        let route;
+        try { route = services.prdAuthority(state, { authors: material.authors }); }
+        catch (e) { if (!['prd-arbiter-unavailable','prd-self-approval'].includes(e.code)) throw e; return wait(state, e.code, identity, {}, tx.appendRequest); }
+        const candidate = material.candidate;
+        const decision = services.raisePrdDecision(identity, { action: 'prd.approve', domain: 'technical', project: state.project,
+          target: `prd-${digest(candidate)}`, question: `Approve product PRD ${candidate.prd_version} for ${state.project}, exact candidate ${digest(candidate)} at ${candidate.records_commit}?`,
+          options: [{ key: 'yes', text: 'Approve this exact product PRD version' }, { key: 'no', text: 'Request changes' }], recommendation: 'yes', approvalChoice: 'yes', blocking: false,
+          ...(state.mode === 'autonomous' ? { spec_path: 'spec/north-star.md', spec_commit: candidate.spec_commit } : {}) },
+          state.mode === 'autonomous' ? { path: 'spec/north-star.md', commit: candidate.spec_commit } : null);
+        const header = { version: 1, business: state.business, request: state.request, op, kind: 'prd-submit', digest: proposal_digest };
+        journal.prepareEffect({ ...header, stage: 'prepared', result: null, code: null });
+        // A replacement retires existing admissions without changing their immutable evidence.
+        if (state.snapshot) services.catalog.invalidate(state.business, state.project, 'prd-replaced');
+        state = stateOf(state.business, state.request);
+        append(state, 'prd.submitted', { candidate, decision: decision.id, op, proposal_digest, authors: material.authors, route }, identity, tx.appendRequest);
+        journal.finishEffect({ ...header, stage: 'committed', result: { decision: decision.id }, code: null });
+        services.recordPrd?.(identity, state.project, decision.id);
+        return { candidate, decision };
+      });
+    });
+  }
+  async function modeConsole(identity, args) {
+    exact(args, ['request','revision','mode_revision','mode','op']);
+    if (!id(args.request) || !id(args.op) || !Number.isSafeInteger(args.revision) || !Number.isSafeInteger(args.mode_revision) || !['guided','autonomous'].includes(args.mode)) refuse('invalid-request');
+    return serial(identity.business, args.request, () => journal.transaction(() => {
+      let state = stateOf(identity.business, args.request); authorize(identity, state, 'mode');
+      const proposal_digest = digest(args), old = journal.proposal(state.business, state.request, args.op);
+      if (old) { if (old.proposal_digest !== proposal_digest) refuse('request-operation-conflict'); return state; }
+      if (state.revision !== args.revision) refuse('request-revision-conflict');
+      if (!state.project || args.mode_revision !== state.mode_revision) refuse('request-mode-revision-conflict');
+      const mode_revision = journal.changeMode({ business: state.business, project: state.project, revision: args.mode_revision, mode: args.mode, op: args.op, request: args.request });
+      services.catalog.invalidate(state.business, state.project, 'request-mode-changed');
+      for (const row of journal.listRequests()) {
+        if (row.business !== state.business) continue;
+        const attached = stateOf(row.business, row.request);
+        if (attached.project === state.project && attached.phase !== 'cancelled') append(attached, 'mode.changed', { mode: args.mode, mode_revision, op: args.op, proposal_digest });
+      }
+      return stateOf(state.business, state.request);
+    }));
+  }
   async function resume(business, request) {
     return serial(business, request, async () => {
       const state = stateOf(business, request);
+      if (state.wait?.code === 'prd-approval') {
+        try { await services.catalog.activate({ request, decision: state.wait.subject }); return stateOf(business, request); }
+        catch (e) {
+          if (e.code === 'prd-not-approved') return state;
+          return append(stateOf(business, request), 'prd.invalidated', { code: e.code ?? 'prd-activation-failed' });
+        }
+      }
       if (state.wait?.kind !== 'decision') return runBootstrap(state);
       const close = services.readDecision(business, state.selection.decision);
       if (!close || close.op === 'seen') return state;
@@ -231,6 +302,21 @@ export function createRequestLifecycle({ journal, services }) {
       return runBootstrap(confirmed);
     });
   }
-  return Object.freeze({ open, get, projects, propose, resume,
+  return Object.freeze({ open, get, projects, propose, submit, modeConsole,
+    draftConsole: async (identity, args) => {
+      exact(args, ['request']); if (!id(args.request)) refuse('invalid-request');
+      const state = stateOf(identity.business, args.request); authorize(identity, state, 'draft');
+      if (!state.candidate) refuse('prd-candidate-required');
+      const material = await services.readCandidate(state.candidate, state.business);
+      const current = stateOf(state.business, state.request); authorize(identity, current, 'draft');
+      if (current.revision !== state.revision) refuse('request-revision-conflict');
+      return { candidate: material.candidate, prd: material.prd, requirements: material.requirements };
+    }, resume, catalog: services.catalog,
+    decisionClosed: (business, decision) => {
+      if (closing) return Promise.resolve();
+      for (const row of journal.listRequests()) if (row.business === business && journal.readRequest(business, row.request).wait?.subject === decision)
+        return resume(business, row.request);
+    },
+    close: async () => { closing = true; await Promise.allSettled([...pending.values()]); await services.catalog?.drain(); },
     recover: async () => { const states = []; for (const { business, request } of journal.listRequests()) states.push(await resume(business, request)); return states; } });
 }

packages/bus/src/request-store.mjs

@@ -48,6 +48,21 @@ export function validatePlan(p) {
   try { checkVars(p.project_vars, 'project', 'request resource plan'); } catch { refuse(code); }
   return p;
 }
+export function validateCandidate(c) {
+  const code = 'invalid-prd-candidate';
+  exact(c, ['version','request','project','prd_version','records_commit','prd_path','prd_blob','requirements','spec_commit','mode_revision'], [], code);
+  if (c.version !== 1 || !id(c.request) || !key(c.project) || typeof c.prd_version !== 'string' || !/^\d+\.\d+(?:\.\d+)?$/.test(c.prd_version) ||
+      !oid(c.records_commit) || c.prd_path !== `docs/prd-${c.prd_version}.md` || !oid(c.prd_blob) || !oid(c.spec_commit) || !positive(c.mode_revision) ||
+      !Array.isArray(c.requirements) || c.requirements.length < 1 || c.requirements.length > 64) refuse(code);
+  const seen = new Set();
+  for (const r of c.requirements) {
+    exact(r, ['id','title','path','blob'], [], code);
+    if (typeof r.id !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(r.id) || !text(r.title, 512) || !r.title.trim() ||
+        r.path !== `requirements/${r.id}.json` || !oid(r.blob) || seen.has(r.path)) refuse(code);
+    seen.add(r.path);
+  }
+  return c;
+}
 function initial(business, request, b) {
   exact(b, ['source_message','thread','session','owner','pm','intent'], [], 'invalid-request-event');
   if (!id(b.source_message) || !nullableId(b.thread) || !nullableId(b.session) || !id(b.owner) || !key(b.pm)) refuse('invalid-request-event');
@@ -107,6 +122,32 @@ function reduce(state, e) {
     if (s.phase !== 'bootstrap' || s.plan_digest !== b.digest || b.project !== s.project || !id(b.chat_project) || !oid(b.spec_commit) || !text(b.organization_revision, 256)) refuse(code);
     s.chat_project = b.chat_project; s.spec_commit = b.spec_commit; s.phase = 'discovery';
     s.wait = { kind: 'setup', code: 'request-discovery-unavailable', subject: s.request, resume_phase: 'discovery', observed_revision: s.revision };
+  } else if (e.kind === 'prd.submitted') {
+    exact(b, ['candidate','decision','op','proposal_digest','authors','route'], [], code);
+    validateCandidate(b.candidate);
+    exact(b.route, ['instance','definition'], [], code);
+    if (!key(b.route.instance) || !key(b.route.definition)) refuse(code);
+    if (!['discovery','prd-review','planning'].includes(s.phase) || !id(b.decision) || !id(b.op) || !hash(b.proposal_digest) ||
+        b.candidate?.request !== s.request || b.candidate.project !== s.project || b.candidate.mode_revision !== s.mode_revision || !Array.isArray(b.authors) || !b.authors.length) refuse(code);
+    for (const a of b.authors) { exact(a, ['definition','instance','commit'], [], code); if (!key(a.definition) || !key(a.instance) || !oid(a.commit)) refuse(code); }
+    s.candidate = structuredClone(b.candidate); s.phase = 'prd-review'; s.snapshot = null; s.spec_commit = b.candidate.spec_commit;
+    s.wait = { kind: 'decision', code: 'prd-approval', subject: b.decision, resume_phase: 'prd-review', observed_revision: s.revision };
+  } else if (e.kind === 'prd.activated') {
+    exact(b, ['decision','snapshot'], [], code);
+    if (s.phase !== 'prd-review' || s.wait?.subject !== b.decision || !id(b.snapshot)) refuse(code);
+    s.snapshot = b.snapshot; s.phase = 'planning';
+    s.wait = { kind: 'setup', code: 'request-planning-unavailable', subject: s.request, resume_phase: 'planning', observed_revision: s.revision };
+  } else if (e.kind === 'prd.invalidated') {
+    exact(b, ['code'], [], code);
+    if (!text(b.code, 128)) refuse(code);
+    s.snapshot = null; s.candidate = null; s.phase = 'discovery';
+    s.wait = { kind: 'setup', code: b.code, subject: s.request, resume_phase: s.phase, observed_revision: s.revision };
+  } else if (e.kind === 'mode.changed') {
+    exact(b, ['mode','mode_revision','op','proposal_digest'], [], code);
+    if (!s.project || !['guided','autonomous'].includes(b.mode) || !positive(b.mode_revision) || !id(b.op) || !hash(b.proposal_digest)) refuse(code);
+    s.mode = b.mode; s.mode_revision = b.mode_revision;
+    s.candidate = null; s.snapshot = null; s.phase = 'discovery';
+    s.wait = { kind: 'setup', code: 'request-mode-changed', subject: s.request, resume_phase: s.phase, observed_revision: s.revision };
   } else if (e.kind === 'cancelled') {
     exact(b, ['decision'], [], code);
     if (s.wait?.kind !== 'decision' || s.selection?.decision !== b.decision) refuse(code);
@@ -115,6 +156,12 @@ function reduce(state, e) {
   return s;
 }
 export function createRequestStore(store) {
+  let composing = false;
+  const compose = fn => {
+    const before = composing; composing = true;
+    try { return fn(); } finally { composing = before; }
+  };
+  const atomic = fn => composing ? fn() : store.transaction(() => compose(fn));
   const validateIntent = (event, origin) => {
     if (event.kind === 'selection.proposed' && (!Array.isArray(event.body.bindings) ||
         event.body.bindings.some(b => !object(b?.plan?.intent) || !equal(b.plan.intent, origin?.intent)))) refuse('invalid-request-event');
@@ -150,6 +197,18 @@ export function createRequestStore(store) {
       if (!d || d.action !== 'project.bootstrap' || d.route_to !== 'human' || d.blocking !== 0 || d.task_ref !== null || d.requirement_ref !== null ||
           d.raised_by_role !== state.pm || store.get("SELECT 1 FROM decision_events WHERE decision=? AND op IN ('resolved','withdrawn','expired')", d.id)) refuse('invalid-request-event');
     }
+    if (e.kind === 'prd.submitted') {
+      const d = store.get('SELECT * FROM decisions WHERE business=? AND id=?', e.business, e.body.decision);
+      if (!d || d.action !== 'prd.approve' || d.project !== state.project || d.raised_by_role !== state.pm || d.raised_by_role !== e.actor.role ||
+          d.blocking !== 0 || d.task_ref !== null || d.requirement_ref !== null || d.route_to !== e.body.route.instance) refuse('invalid-request-event');
+    }
+    if (e.kind === 'prd.activated') {
+      const row = store.get('SELECT body FROM request_snapshots WHERE id=? AND business=? AND project=?', e.body.snapshot, e.business, state.project);
+      if (!row || JSON.parse(row.body).approval.decision !== e.body.decision) refuse('invalid-request-event');
+    }
+    if (e.kind === 'selection.confirmed' && !store.get('SELECT 1 FROM request_project_events WHERE business=? AND project=?', e.business, next.project))
+      store.run("INSERT INTO request_project_events(at,business,project,revision,kind,body) VALUES(?,?,?,1,'attached',?)",
+        new Date().toISOString(), e.business, next.project, JSON.stringify({ mode: next.mode, request: e.request }));
     // A bootstrap outcome follows only from the receipts it names, so the journal cannot claim a project
     // that was not activated, or retreat from one whose mapping already landed.
     if (e.kind === 'bootstrap.completed' || e.kind === 'bootstrap.stale') {
@@ -170,7 +229,7 @@ export function createRequestStore(store) {
         !['bootstrap','prd-submit','prd-activate','task-create','continuation','delivery'].includes(r.kind) ||
         !(prepare ? r.stage === 'prepared' && r.result === null && r.code === null : ['committed','uncertain','refused'].includes(r.stage)) ||
         !(r.result === null || object(r.result)) || !(r.code === null || text(r.code, 128))) refuse('invalid-request-effect');
-    return store.transaction(() => {
+    return atomic(() => {
       if (!readRequest(r.business, r.request)) refuse('request-not-found');
       const rows = store.all('SELECT version,business,request,op,kind,digest,stage,result,code FROM request_effects WHERE business=? AND request=? AND op=? ORDER BY seq', r.business, r.request, r.op)
         .map(x => ({ ...x, result: x.result === null ? null : JSON.parse(x.result) }));
@@ -186,9 +245,19 @@ export function createRequestStore(store) {
       return structuredClone(r);
     });
   }
-  return Object.freeze({ readRequest, appendRequest: e => store.transaction(() => appendInTransaction(e)),
+  const projectState = (business, project) => {
+    const row = store.get('SELECT * FROM request_project_events WHERE business=? AND project=? ORDER BY revision DESC LIMIT 1', business, project);
+    if (row) return { revision: row.revision, ...JSON.parse(row.body) };
+    // Task 1/2 databases predate mode writers. Their confirmed selection is the durable initial mode.
+    for (const r of store.all('SELECT DISTINCT request FROM request_events WHERE business=?', business)) {
+      const state = readRequest(business, r.request);
+      if (state.project === project && state.mode_revision > 0) return { revision: state.mode_revision, mode: state.mode, request: state.request };
+    }
+    return null;
+  };
+  return Object.freeze({ compose, readRequest, appendRequest: e => atomic(() => appendInTransaction(e)),
     prepareEffect: r => receipt(r, true), finishEffect: r => receipt(r, false),
-    transaction: fn => store.transaction(() => fn({ appendRequest: appendInTransaction, readRequest })),
+    transaction: fn => atomic(() => fn({ appendRequest: appendInTransaction, readRequest })),
     // Receipts read back, parsed. Hosts use these to resume a bootstrap and to rebuild readiness.
     effects: (business, request) => {
       if (!key(business) || !id(request)) refuse('invalid-request');
@@ -201,6 +270,36 @@ export function createRequestStore(store) {
         .map(x => ({ ...x, result: x.result === null ? null : JSON.parse(x.result) }));
     },
     listRequests: () => store.all('SELECT DISTINCT business,request FROM request_events ORDER BY business,request'),
+    stateForRequest: request => {
+      const row = store.get('SELECT business FROM request_events WHERE request=? LIMIT 1', request);
+      return row ? readRequest(row.business, request) : null;
+    },
+    submission: (business, request, decision) => {
+      const row = store.get("SELECT body FROM request_events WHERE business=? AND request=? AND kind='prd.submitted' AND json_extract(body,'$.decision')=?", business, request, decision);
+      return row ? JSON.parse(row.body) : null;
+    },
+    projectState,
+    changeMode: ({ business, project, revision, mode, op, request }) => atomic(() => {
+      const last = projectState(business, project);
+      if ((last?.revision ?? 0) !== revision) refuse('request-mode-revision-conflict');
+      store.run("INSERT INTO request_project_events(at,business,project,revision,kind,body) VALUES(?,?,?,?,'mode.changed',?)",
+        new Date().toISOString(), business, project, revision + 1, JSON.stringify({ mode, op, request }));
+      return revision + 1;
+    }),
+    snapshotRequest: snapshot => store.get("SELECT request FROM request_effects WHERE kind='prd-activate' AND stage='committed' AND json_extract(result,'$.snapshot')=?", snapshot)?.request,
+    snapshot: snapshot => {
+      const row = store.get('SELECT body FROM request_snapshots WHERE id=?', snapshot);
+      return row ? JSON.parse(row.body) : null;
+    },
+    snapshotEvents: (business, project) => store.all('SELECT * FROM request_snapshot_events WHERE business=? AND project=? ORDER BY revision', business, project),
+    publishSnapshot: snapshot => store.run('INSERT INTO request_snapshots(id,at,business,project,body) VALUES(?,?,?,?,?)',
+      snapshot.id, new Date().toISOString(), snapshot.business, snapshot.project, JSON.stringify(snapshot)),
+    snapshotEvent: ({ business, project, snapshot, kind, cause }) => {
+      const revision = 1 + (store.get('SELECT max(revision) n FROM request_snapshot_events WHERE business=? AND project=?', business, project).n ?? 0);
+      store.run('INSERT INTO request_snapshot_events(at,business,project,revision,snapshot,kind,cause) VALUES(?,?,?,?,?,?,?)',
+        new Date().toISOString(), business, project, revision, snapshot, kind, cause);
+      return { snapshot, revision, kind, cause };
+    },
     proposal: (business, request, op) => {
       const row = store.get("SELECT body FROM request_events WHERE business=? AND request=? AND json_extract(body,'$.op')=? ORDER BY seq DESC LIMIT 1", business, request, op);
       return row ? JSON.parse(row.body) : null;

packages/bus/src/requirement-catalog.mjs

@@ -0,0 +1,106 @@
+// Host-only admission. The journal's readCandidate/authority/approval adapters read current host state.
+import { exact, id, key, text, digest, refuse } from './request-store.mjs';
+export function createRequirementCatalog({ journal, readSpec, builtinBinding }) {
+  const append = (state, kind, body) => journal.appendRequest({ business: state.business, request: state.request,
+    expectedRevision: state.revision, kind, body, actor: { role: null, run: null } });
+  const eventFor = snap => journal.snapshotEvents(snap.business, snap.project).filter(e => e.snapshot === snap.id).at(-1);
+  function invalidate(business, project, code, inTransaction = false) {
+    return (inTransaction ? journal.compose : journal.transaction)(() => {
+      const latest = new Map(journal.snapshotEvents(business, project).map(e => [e.snapshot, e]));
+      for (const e of latest.values()) if (e.kind === 'activated') journal.snapshotEvent({ business, project, snapshot: e.snapshot, kind: 'revoked', cause: code });
+      for (const row of journal.listRequests()) {
+        if (row.business !== business) continue;
+        const state = journal.readRequest(business, row.request);
+        if (state.project === project && (state.candidate || state.snapshot)) append(state, 'prd.invalidated', { code });
+      }
+    });
+  }
+  async function resolve(args) {
+    exact(args, ['business','project'], ['snapshot']);
+    if (!key(args.business) || !key(args.project) || (args.snapshot !== undefined && !id(args.snapshot))) refuse('invalid-request');
+    const { business, project } = args;
+    const events = journal.snapshotEvents(business, project);
+    const snapshot = args.snapshot ?? [...events].reverse().find(e => e.kind === 'activated')?.snapshot;
+    if (!snapshot) {
+      const builtin = await builtinBinding?.({ business, project });
+      if (builtin && builtin.origin === 'builtin' && builtin.business === business && builtin.project === project) return builtin;
+      refuse('requirement-snapshot-required');
+    }
+    const snap = journal.snapshot(snapshot);
+    if (!snap || snap.business !== business || snap.project !== project) refuse('requirement-snapshot-refused');
+    if (eventFor(snap)?.kind !== 'activated') refuse('requirement-snapshot-revoked');
+    const spec = await readSpec({ business, project });
+    const mode = journal.projectState(business, project);
+    if (!spec.confirmed || spec.commit !== snap.spec_commit || mode?.revision !== snap.mode_revision) {
+      invalidate(business, project, 'requirement-snapshot-revoked'); refuse('requirement-snapshot-revoked');
+    }
+    const state = journal.stateForRequest(journal.snapshotRequest(snap.id));
+    const binding = state && journal.submission(business, state.request, snap.approval.decision);
+    const authority = journal.authority(state, binding);
+    if (digest(journal.approval(state, binding)) !== digest(snap.approval)) refuse('prd-authority-refused');
+    // Reread immutable records, including selected-path equality at HEAD, before admitting new work.
+    const material = await journal.readCandidate(binding.candidate, business);
+    if (digest(material.candidate) !== digest(binding.candidate) || digest(material.authors) !== digest(snap.authors) ||
+        digest(material.candidate.requirements.map((r, i) => ({ ...r, text: material.requirements[i].text }))) !== digest(snap.requirements)) refuse('prd-candidate-stale');
+    if (digest(journal.authority(journal.stateForRequest(state.request), binding)) !== digest(authority) || eventFor(snap)?.kind !== 'activated' || journal.projectState(business, project)?.revision !== snap.mode_revision) refuse('requirement-snapshot-revoked');
+    return structuredClone(snap);
+  }
+  async function activateOnce(args) {
+    exact(args, ['request','decision']); if (!id(args.request) || !id(args.decision)) refuse('invalid-request');
+    const state = journal.stateForRequest(args.request) ?? refuse('request-not-found');
+    const binding = journal.submission(state.business, state.request, args.decision) ?? refuse('prd-candidate-required');
+    const old = journal.committed(state.business, 'prd-activate').find(e => e.request === state.request && e.result?.decision === args.decision);
+    if (old) return structuredClone(journal.snapshot(old.result.snapshot));
+    if (state.phase !== 'prd-review' || state.wait?.subject !== args.decision || digest(state.candidate) !== digest(binding.candidate)) refuse('prd-admission-revoked');
+    const authority = journal.authority(state, binding), approval = journal.approval(state, binding);
+    const material = await journal.readCandidate(binding.candidate, state.business);
+    if (digest(material.candidate) !== digest(binding.candidate) || digest(material.authors) !== digest(binding.authors)) refuse('prd-candidate-stale');
+    const spec = await readSpec({ business: state.business, project: state.project });
+    if (!spec.confirmed || spec.commit !== binding.candidate.spec_commit) refuse('stale-spec');
+    return journal.transaction(() => {
+      const prior = journal.committed(state.business, 'prd-activate').find(e => e.request === state.request && e.result?.decision === args.decision);
+      if (prior) return structuredClone(journal.snapshot(prior.result.snapshot));
+      const current = journal.readRequest(state.business, state.request);
+      if (current.revision !== state.revision || current.wait?.subject !== args.decision || journal.projectState(state.business, state.project)?.revision !== binding.candidate.mode_revision) refuse('prd-admission-revoked');
+      if (digest(journal.authority(current, binding)) !== digest(authority) || digest(journal.approval(current, binding)) !== digest(approval)) refuse('prd-admission-revoked');
+      const c = binding.candidate, snapshot = { version: 1, id: `prd-${digest({ business: state.business, candidate: c, decision: args.decision })}`,
+        business: state.business, project: state.project, prd_version: c.prd_version, records_commit: c.records_commit, prd_path: c.prd_path,
+        prd_blob: c.prd_blob, spec_commit: c.spec_commit, requirements: c.requirements.map((r, i) => ({ ...r, text: material.requirements[i].text })),
+        authors: material.authors, approval, mode_revision: c.mode_revision, origin: 'product' };
+      const header = { version: 1, business: state.business, request: state.request, op: `prd-activate:${args.decision}`, kind: 'prd-activate', digest: digest(c) };
+      journal.prepareEffect({ ...header, stage: 'prepared', result: null, code: null });
+      journal.publishSnapshot(snapshot);
+      journal.snapshotEvent({ business: state.business, project: state.project, snapshot: snapshot.id, kind: 'activated', cause: args.decision });
+      append(current, 'prd.activated', { decision: args.decision, snapshot: snapshot.id });
+      journal.finishEffect({ ...header, stage: 'committed', result: { snapshot: snapshot.id, decision: args.decision }, code: null });
+      return snapshot;
+    });
+  }
+  function revoke(identity, args) {
+    exact(args, ['project','snapshot','revision','reason','op']);
+    if (!key(args.project) || !id(args.snapshot) || !Number.isSafeInteger(args.revision) || args.revision < 1 || !text(args.reason, 1024) || !id(args.op)) refuse('invalid-request');
+    return journal.transaction(() => {
+      const snap = journal.snapshot(args.snapshot);
+      if (!snap || snap.business !== identity.business || snap.project !== args.project) refuse('requirement-snapshot-refused');
+      const cause = JSON.stringify({ op: args.op, reason: args.reason, by: identity.human, digest: digest(args) });
+      const events = journal.snapshotEvents(identity.business, args.project), old = events.find(e => e.kind === 'revoked' && e.cause === cause);
+      if (old) return { snapshot: old.snapshot, revision: old.revision, kind: old.kind, cause: old.cause };
+      if (events.some(e => { try { return JSON.parse(e.cause).op === args.op; } catch { return false; } })) refuse('request-operation-conflict');
+      if ((events.at(-1)?.revision ?? 0) !== args.revision) refuse('request-revision-conflict');
+      const result = journal.snapshotEvent({ business: identity.business, project: args.project, snapshot: snap.id, kind: 'revoked', cause });
+      for (const row of journal.listRequests()) {
+        const state = journal.readRequest(row.business, row.request);
+        if (row.business === identity.business && state.project === args.project && state.snapshot === snap.id) append(state, 'prd.invalidated', { code: 'requirement-snapshot-revoked' });
+      }
+      return result;
+    });
+  }
+  const pending = new Map();
+  function activate(args) {
+    const operation = `${args?.request}/${args?.decision}`;
+    if (pending.has(operation)) return pending.get(operation);
+    const job = activateOnce(args).finally(() => { if (pending.get(operation) === job) pending.delete(operation); });
+    pending.set(operation, job); return job;
+  }
+  return Object.freeze({ resolve, activate, invalidate, revoke, drain: () => Promise.allSettled([...pending.values()]) });
+}

packages/bus/src/runtime.mjs

@@ -1,3 +1,5 @@
+import { createRequirementCatalog } from './requirement-catalog.mjs';
+import { readPrdMaterial } from '../../project/src/prd.mjs';
 import { lstatSync, readdirSync, readFileSync } from 'node:fs';
 import { join } from 'node:path';
 import { fileURLToPath } from 'node:url';
@@ -564,11 +566,21 @@ export async function startBroker({
       return no(declared ? 'project-pending' : 'project-unknown');
     }
     broker = new Broker({
-      requestLifecycle: services => (requestLifecycle = createRequestLifecycle({ journal: requestJournal, services: { ...(writer ? hostPreparation({ dataRoot, organization: writer, trackers }) : {}), ...requestServices, ...services,
+      requestLifecycle: services => {
+        const readCandidate = (candidate, business) => {
+          const { request, project, prd_version, records_commit, prd_path, mode_revision } = candidate;
+          return readPrdMaterial({ business: services.prdBusiness(business), dataRoot, request, project, prd_version, records_commit, prd_path, mode_revision,
+            requirement_ids: candidate.requirement_ids ?? candidate.requirements.map(r => r.id) });
+        };
+        const catalog = createRequirementCatalog({ journal: { ...requestJournal, readCandidate, authority: services.prdAuthority, approval: services.prdApproval },
+          readSpec: ({ business, project }) => projectModule.readSpec({ business: services.prdBusiness(business), project, dataRoot }),
+          builtinBinding: null });
+        return (requestLifecycle = createRequestLifecycle({ journal: requestJournal, services: { catalog, readCandidate, ...(writer ? hostPreparation({ dataRoot, organization: writer, trackers }) : {}), ...requestServices, ...services,
         // The new-project executor needs the organization writer; without one a confirmed selection stays at its setup wait.
         ...(writer ? { bootstrap: ({ binding, state }) => processing.work(() => executeBootstrap({ binding, owner: state.owner, journal: requestJournal, services: {
           dataRoot, organization: writer, activate: activateManagedProject, barrier: requestServices.barrier,
-          chat: { state: args => broker.chatProjectState(args), associate: args => broker.associateChat(args) } } })) } : {}) } })),
+          chat: { state: args => broker.chatProjectState(args), associate: args => broker.associateChat(args) } } })) } : {}) } }));
+      },
       // FED R3: the access file is read per user request; a refusal leaves the user no grant.
       access: access ? () => loadAccess(access.file, access.context) : null,
       // FED R4 (#1801, design 3.7 trigger 1): a request stats the file and reads it only when this
@@ -963,6 +975,7 @@ export async function startBroker({
     return {
       broker,
       requestLifecycle,
+      requirementCatalog: requestLifecycle.catalog,
       admission,
       credentials,
       // P2 consumes this inside the bus only. No process IPC/verb returns it.
@@ -1033,6 +1046,7 @@ export async function startBroker({
         await stopSync();
         await server.close();
         await adapter?.close?.();
+        await requestLifecycle.close();
         await processing?.close();
         store.close();
         credentials.close();

packages/bus/src/store.mjs

@@ -13,8 +13,8 @@ import { join, isAbsolute } from 'node:path';
 import { createHash } from 'node:crypto';
 
 const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
-export const SCHEMA_SHA256 = 'ae56fbdcb5ea9df3a15a47c7978f5c22516cf1c85dd9a1c9885a63bf7f80d252';
+export const SCHEMA_SHA256 = '2a6a161b3d671418e690942f02203b4a014e939b46ae8f651da11f091dac5455';
-export const SCHEMA_VERSION = '3p';
+export const SCHEMA_VERSION = '3q';
 // Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned.
 const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8');
 const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65';
@@ -93,6 +93,9 @@ const migrations = [{
   from: '3o',
   to: '3p',
   sql: readFileSync(new URL('./migrations/3o-3p.sql', import.meta.url), 'utf8'),
+}, {
+  from: '3p', to: '3q',
+  sql: readFileSync(new URL('./migrations/3p-3q.sql', import.meta.url), 'utf8'),
 }];
 const timedTables = [
   'events',

packages/bus/tests/account-migration.test.mjs

@@ -90,6 +90,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) {
   copyFileSync(here('../src/migrations/3m-3n.sql'), join(moduleRoot, 'src/migrations/3m-3n.sql'));
   copyFileSync(here('../src/migrations/3n-3o.sql'), join(moduleRoot, 'src/migrations/3n-3o.sql'));
   copyFileSync(here('../src/migrations/3o-3p.sql'), join(moduleRoot, 'src/migrations/3o-3p.sql'));
+  copyFileSync(here('../src/migrations/3p-3q.sql'), join(moduleRoot, 'src/migrations/3p-3q.sql'));
   return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
 }
 
@@ -97,7 +98,7 @@ test('REQ-LIMIT-4 3h copy migrates to 3i: every row keeps its seq, the sequence
   const { root, path } = source3h(t), before = snapshot(path), was = master(path);
   assert.throws(() => new Store(root), /schema-mismatch/);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   // Every table's rows as they were, usage_records with its seq values and sqlite_sequence with 50.
   assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows);
   assert.deepEqual(before.rows.usage_records.map(r => r.seq), [1, 2, 3]);
@@ -147,7 +148,7 @@ test('REQ-LIMIT-4 a 3h→3i step that fails partway leaves the store at 3h with
   assert.deepEqual(snapshot(path, before.columns).rows, before.rows);
   // Still a whole 3h store: the real step migrates it.
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows);
 });
 
@@ -159,14 +160,14 @@ test('REQ-LIMIT-4 the 3h engine refuses a migrated store, migrating or not, and
     schema: ddl3h,
     store: (source) => source
       .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3h).digest('hex')}';`)
-      .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3h';")
+      .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3h';")
       .replace(/\}, \{\n {2}\/\/ Row 126 \(#1620\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'),
   });
   const pinned = readFileSync(join(root, 'engine-3h/src/store.mjs'), 'utf8');
   assert.ok(pinned.includes("export const SCHEMA_VERSION = '3h';") && !pinned.includes('3h-3i'), 'the 3h engine has no 3h→3i step');
   assert.throws(() => new Store3h(root), /schema-mismatch/);
   assert.throws(() => new Store3h(root, { migrateSchema: true }), /schema-mismatch/);
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(master(path), was);
   assert.deepEqual(snapshot(path, after.columns).rows, after.rows);
 });

packages/bus/tests/chief-of-staff.test.mjs

@@ -614,7 +614,7 @@ async function database3c(t) {
   let source = readFileSync(here('../src/store.mjs'), 'utf8');
   const pinned = source
     .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3c).digest('hex')}';`)
-    .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3c';")
+    .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3c';")
     .replace(/\}, \{\n {2}from: '3c',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
   assert.notEqual(pinned, source);
   assert.ok(!pinned.includes('3c-3d'), 'the 3c engine has no 3c→3d step');
@@ -638,7 +638,7 @@ test('REQ-ROLE-2, REQ-COS-1: the 3c→3d migration lets the chief of staff end a
   assert.throws(() => new Store(root), /schema-mismatch/);
   const store = new Store(root, { migrateSchema: true });
   t.after(() => store.close());
-  assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3p');
+  assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
   assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}');
   assert.equal(store.get("SELECT holder_run FROM role_claims WHERE role='coder'").holder_run, 'coder-run');
   const b = new Broker({ store, businesses: demo(), organization: null });

packages/bus/tests/comment-migration.test.mjs

@@ -33,7 +33,7 @@ test('explicit 3b copy migration preserves evidence and opens as 3g; normal star
   const original = readFileSync(source);
   assert.throws(() => new Store(root), /schema-mismatch/);
   const store = new Store(root, { migrateSchema: true });
-  assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3p');
+  assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
   assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}');
   assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
   const epoch = store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value;
@@ -75,15 +75,15 @@ test('named migration registry chains 3b through 3g to a fixture-only future ver
   const sha = createHash('sha256').update(futureDDL).digest('hex');
   let source = readFileSync(new URL('../src/store.mjs', import.meta.url), 'utf8');
   source = source.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${sha}';`)
-    .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3q';")
+    .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3r';")
-    .replace('}];\nconst timedTables', `}, { from: '3p', to: '3q', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
+    .replace('}];\nconst timedTables', `}, { from: '3q', to: '3r', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
   writeFileSync(join(moduleRoot, 'src/store.mjs'), source);
   writeFileSync(join(moduleRoot, 'schema.sql'), futureDDL);
-  for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql', '3i-3j.sql', '3j-3k.sql', '3k-3l.sql', '3l-3m.sql', '3m-3n.sql', '3n-3o.sql', '3o-3p.sql'])
+  for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql', '3i-3j.sql', '3j-3k.sql', '3k-3l.sql', '3l-3m.sql', '3m-3n.sql', '3n-3o.sql', '3o-3p.sql', '3p-3q.sql'])
     copyFileSync(new URL('../src/migrations/' + name, import.meta.url), join(moduleRoot, 'src/migrations', name));
   const { Store: FutureStore } = await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')));
   const future = new FutureStore(root, { migrateSchema: true });
-  assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
+  assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r');
   assert.equal(future.get("SELECT name FROM sqlite_master WHERE name='future_fixture'").name, 'future_fixture');
   assert.equal(future.get("SELECT count(*) n FROM task_comment_cursors").n, 0);
   assert.equal(future.get("SELECT id FROM events").id, 'preserved');
@@ -109,7 +109,7 @@ test('offline command migrates only the supplied fixture root and reports usage
   const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname;
   const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' });
   assert.equal(migrated.status, 0, migrated.stderr);
-  assert.match(migrated.stdout, /bus schema 3p ready/);
+  assert.match(migrated.stdout, /bus schema 3q ready/);
   new Store(root).close();
   const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' });
   assert.equal(usage.status, 2);

packages/bus/tests/console-decide.test.mjs

@@ -86,8 +86,8 @@ function setup(t) {
   return { store, broker, make, caps, call, raise, events, grantEvents, settle, closes, enrolls, grant };
 }
 
-test('N4: the console verbs are exactly the preview and the console resolve, and decision.resolve is not one', () => {
+test('N4: owner Console verbs have an explicit immutable whitelist and exclude decision.resolve', () => {
-  assert.deepEqual([...CONSOLE_VERBS], ['decision.preview.console', 'decision.resolve.console']);
+  assert.deepEqual([...CONSOLE_VERBS], ['decision.preview.console', 'decision.resolve.console', 'prd.revoke.console', 'request.mode.console', 'request.draft.console']);
   assert.equal(CONSOLE_VERBS.includes('decision.resolve'), false);
   assert.ok(Object.isFrozen(CONSOLE_VERBS));
 });

packages/bus/tests/gitea-migration.test.mjs

@@ -80,6 +80,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) {
   copyFileSync(here('../src/migrations/3m-3n.sql'), join(moduleRoot, 'src/migrations/3m-3n.sql'));
   copyFileSync(here('../src/migrations/3n-3o.sql'), join(moduleRoot, 'src/migrations/3n-3o.sql'));
   copyFileSync(here('../src/migrations/3o-3p.sql'), join(moduleRoot, 'src/migrations/3o-3p.sql'));
+  copyFileSync(here('../src/migrations/3p-3q.sql'), join(moduleRoot, 'src/migrations/3p-3q.sql'));
   return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
 }
 function fresh(t) {
@@ -101,7 +102,7 @@ test('3j copy migrates to 3k: every row kept, and sqlite_master gains only the t
   const { root, path } = source3j(t), before = rows(path), was = master(path);
   assert.throws(() => new Store(root), /schema-mismatch/);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(migrated3l(rows(path)), before);
   const now = master(path);
   assert.deepEqual(now.filter(r => !NEW.includes(r.name) && !later3l(r)), was.filter(r => !later3l(r)));
@@ -127,7 +128,7 @@ test('a 3j→3k step that fails partway leaves the store at 3j with every row an
   assert.deepEqual(master(path), was);
   assert.deepEqual(rows(path), before);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
 });
 
 test('the 3j engine refuses a migrated store, migrating or not, and leaves it as it was', async t => {
@@ -138,14 +139,14 @@ test('the 3j engine refuses a migrated store, migrating or not, and leaves it as
     schema: ddl3j,
     store: (source) => source
       .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3j).digest('hex')}';`)
-      .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3j';")
+      .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3j';")
       .replace(/\}, \{\n {2}\/\/ Row 236 \(#1742[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'),
   });
   const pinned = readFileSync(join(root, 'engine-3j/src/store.mjs'), 'utf8');
   assert.ok(pinned.includes("export const SCHEMA_VERSION = '3j';") && !pinned.includes('3j-3k'), 'the 3j engine has no 3j→3k step');
   assert.throws(() => new Store3j(root), /schema-mismatch/);
   assert.throws(() => new Store3j(root, { migrateSchema: true }), /schema-mismatch/);
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(master(path), was);
   assert.deepEqual(rows(path), after);
 });

packages/bus/tests/grant-migration.test.mjs

@@ -83,6 +83,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) {
   copyFileSync(here('../src/migrations/3m-3n.sql'), join(moduleRoot, 'src/migrations/3m-3n.sql'));
   copyFileSync(here('../src/migrations/3n-3o.sql'), join(moduleRoot, 'src/migrations/3n-3o.sql'));
   copyFileSync(here('../src/migrations/3o-3p.sql'), join(moduleRoot, 'src/migrations/3o-3p.sql'));
+  copyFileSync(here('../src/migrations/3p-3q.sql'), join(moduleRoot, 'src/migrations/3p-3q.sql'));
   return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
 }
 function fresh(t) {
@@ -106,7 +107,7 @@ test('3i copy migrates to 3j: every row kept, and sqlite_master gains only the t
   const { root, path } = source3i(t), before = rows(path), was = master(path);
   assert.throws(() => new Store(root), /schema-mismatch/);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(migrated3l(rows(path)), before);
   const now = master(path).filter(r => !LATER.includes(r.name));
   assert.deepEqual(now.filter(r => !NEW.includes(r.name) && !later3l(r)), was.filter(r => !later3l(r)));
@@ -131,7 +132,7 @@ test('a 3i→3j step that fails partway leaves the store at 3i with every row an
   assert.deepEqual(master(path), was);
   assert.deepEqual(rows(path), before);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
 });
 
 test('the 3i engine refuses a migrated store, migrating or not, and leaves it as it was', async t => {
@@ -142,14 +143,14 @@ test('the 3i engine refuses a migrated store, migrating or not, and leaves it as
     schema: ddl3i,
     store: (source) => source
       .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3i).digest('hex')}';`)
-      .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3i';")
+      .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3i';")
       .replace(/\}, \{\n {2}\/\/ Row 234 \(#1740\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'),
   });
   const pinned = readFileSync(join(root, 'engine-3i/src/store.mjs'), 'utf8');
   assert.ok(pinned.includes("export const SCHEMA_VERSION = '3i';") && !pinned.includes('3i-3j'), 'the 3i engine has no 3i→3j step');
   assert.throws(() => new Store3i(root), /schema-mismatch/);
   assert.throws(() => new Store3i(root, { migrateSchema: true }), /schema-mismatch/);
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(master(path), was);
   assert.deepEqual(rows(path), after);
 });

packages/bus/tests/name-migration.test.mjs

@@ -65,7 +65,7 @@ test('3f copy migrates to 3g preserving every old column/row, with an empty inst
   assert.throws(() => new Store(root), /schema-mismatch/);
   const store = new Store(root, { migrateSchema: true });
   try {
-    assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3p');
+    assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
     assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null });
     assert.equal(store.get('SELECT count(*) AS n FROM instance_names').n, 0);
     assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: 'docs/spec.md', spec_commit: 'c'.repeat(40) });

packages/bus/tests/prd-activation.test.mjs

@@ -0,0 +1,215 @@
+import test from 'node:test';
+import { DatabaseSync } from 'node:sqlite';
+import assert from 'node:assert/strict';
+import { mkdirSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { startBroker } from '../src/runtime.mjs';
+import { bootConfig, loadSystem } from '../../cli/src/config.mjs';
+import { fixture, tmp } from '../../cli/tests/helpers.mjs';
+import { prdFixture, git } from '../../project/tests/prd-fixture.mjs';
+import { createRequestStore } from '../src/request-store.mjs';
+
+async function setup(t, mode = 'autonomous') {
+  const root = tmp(t, 'activation-'), f = fixture(root, 'demo');
+  const source = join(root, 'project'); mkdirSync(join(source, '.mosaic'), { recursive: true });
+  writeFileSync(join(source, '.mosaic/project.json'), JSON.stringify({ projectVersion: 1, id: 'stack', vars: { 'tracker.project': 1 } }));
+  git(source, 'init', '-qb', 'main'); git(source, 'add', '-A'); git(source, 'commit', '-qm', 'Factory');
+  const system = loadSystem({ env: f.env });
+  const boot = async () => {
+    const config = bootConfig({ system, businessId: 'demo', env: f.env, warn: () => {} });
+    const bus = await startBroker(config); t.after(() => bus.close().catch(() => {})); return bus;
+  };
+  const bus = await boot(), caps = {};
+  const db = new DatabaseSync(join(f.dataRoot, 'bus/bus.sqlite'), { readOnly: true }); t.after(() => db.close());
+  const store = { get: (sql, ...a) => db.prepare(sql).get(...a), all: (sql, ...a) => db.prepare(sql).all(...a) };
+  const bind = b => {
+    caps.human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
+    caps.console = b.bindConsole({ business: 'demo', human: 'jason', subject: { provider: 'test', iss: 'test', sub: 'owner' } });
+    caps.reader = b.bindReader({ business: 'demo' });
+    for (const role of ['pm','cto','coder']) { caps[role] = b.bindLaunch({ business: 'demo', role, run: role + '-run', harness: 'pi' }); b.request(caps[role], { verb: 'role.claim' }); }
+  };
+  bind(bus.broker);
+  const call = (who, verb, args = {}) => bus.broker.request(caps[who], { verb, args });
+  const sent = call('human', 'message.send', { to: 'pm', body: 'Build a timer.\n' });
+  let state = await call('pm', 'request.open', { request: sent.request });
+  state = await call('pm', 'request.propose', { request: sent.request, revision: state.revision, op: 'plan', source_project: 'stack', project: 'timer', name: 'Timer', tracker_project: 2 });
+  call('human', 'decision.resolve', { id: state.selection.decision, choice: mode });
+  state = await bus.requestLifecycle.resume('demo', sent.request);
+  const context = { business: { id: 'demo', human: 'jason', projects: { timer: { root: join(f.dataRoot, 'products/demo/timer/code') } } }, project: 'timer', dataRoot: f.dataRoot };
+  const p = await prdFixture(t, context);
+  const args = { request: sent.request, revision: state.revision, op: 'prd1', prd_version: '1.0', records_commit: p.args.records_commit, prd_path: p.args.prd_path, requirement_ids: p.args.requirement_ids };
+  const submit = () => call('pm', 'prd.submit', args);
+  const approve = (submission, who = mode === 'guided' ? 'human' : 'cto') => call(who, 'decision.resolve', { id: submission.decision.id, choice: 'yes' });
+  const activate = submission => bus.requirementCatalog.activate({ request: sent.request, decision: submission.decision.id });
+  return { ...p, f, bus, caps, call, state, request: sent.request, args, submit, approve, activate, boot, bind, config: bootConfig({ system, businessId: 'demo', env: f.env, warn: () => {} }), store, journal: createRequestStore(store) };
+}
+
+test('autonomous PM submits technical PRD to distinct CTO, no task before atomic idempotent activation', async t => {
+  const f = await setup(t), submission = await f.submit();
+  assert.equal(submission.decision.action, 'prd.approve'); assert.equal(submission.decision.route_to, 'cto');
+  assert.equal(submission.decision.authorization.target.startsWith('prd-'), true);
+  assert.equal(submission.decision.spec_commit, submission.candidate.spec_commit); assert.equal(submission.decision.blocking, false);
+  assert.equal(f.store.get('SELECT count(*) n FROM task_snapshots').n, 0);
+  await assert.rejects(f.activate(submission), { code: 'prd-not-approved' });
+  assert.throws(() => f.call('pm', 'decision.resolve', { id: submission.decision.id, choice: 'yes' }), { code: 'not-resolver' });
+  f.approve(submission);
+  f.call('cto', 'role.release');
+  const snap = await f.activate(submission);
+  assert.equal(snap.requirements[0].text, 'A working timer.'); assert.equal(snap.origin, 'product');
+  assert.deepEqual(await f.activate(submission), snap);
+  assert.equal(f.store.get('SELECT count(*) n FROM request_snapshots').n, 1);
+  assert.equal(f.journal.readRequest('demo', f.request).phase, 'planning');
+  assert.equal(f.journal.effects('demo', f.request).filter(e => e.kind === 'prd-activate' && e.stage === 'committed').length, 1);
+  assert.equal((await f.bus.broker.requirementsFor(f.caps.pm, { project_key: 'timer', requirement_snapshot: snap.id, requirement: 'REQ-POMO-1' })).id, snap.id);
+});
+test('guided routes the same PRD action to owner without a routing citation', async t => {
+  const f = await setup(t, 'guided'), sub = await f.submit();
+  assert.equal(sub.decision.route_to, 'human'); assert.equal(sub.decision.spec_commit, null);
+  assert.throws(() => f.approve(sub, 'cto'), { code: 'human-required' });
+  f.approve(sub); assert.equal((await f.activate(sub)).approval.by, 'jason');
+});
+test('submit is PM-only, strict, revision checked, idempotent and size rejected before a decision', async t => {
+  const f = await setup(t);
+  for (const who of ['coder','cto','human','reader']) await assert.rejects(Promise.resolve().then(() => f.call(who, 'prd.submit', f.args)));
+  await assert.rejects(f.call('pm', 'prd.submit', { ...f.args, prd_blob: 'forged' }));
+  await assert.rejects(f.call('pm', 'prd.submit', { ...f.args, revision: 1 }), { code: 'request-revision-conflict' });
+  const count = f.store.get('SELECT count(*) n FROM decisions').n;
+  const records_commit = f.save('docs/prd-1.0.md', 'x'.repeat(32769));
+  await assert.rejects(f.call('pm', 'prd.submit', { ...f.args, records_commit }), { code: 'prd-candidate-size' });
+  assert.equal(f.store.get('SELECT count(*) n FROM decisions').n, count);
+});
+for (const path of ['docs/prd-1.0.md','requirements/REQ-POMO-1.json','spec/north-star.md']) test(`activation rereads current selected content and intent: ${path}`, async t => {
+  const f = await setup(t), sub = await f.submit(); f.approve(sub);
+  f.save(path, path.endsWith('.json') ? JSON.stringify({ recordVersion: 1, id: 'REQ-POMO-1', title: 'Other', text: 'Other', status: 'accepted' }) : 'Changed\n');
+  await assert.rejects(f.activate(sub));
+  assert.equal(f.store.get('SELECT count(*) n FROM request_snapshots').n, 0);
+});
+test('forged accepted status is no authority, wrong project and business cannot borrow a snapshot', async t => {
+  const f = await setup(t);
+  f.args.records_commit = f.save('requirements/REQ-POMO-1.json', JSON.stringify({ recordVersion: 1, id: 'REQ-POMO-1', title: 'Timer', text: 'A working timer.', status: 'accepted' }), 'pm.pm');
+  await assert.rejects(f.bus.broker.requirementsFor(f.caps.pm, { project_key: 'timer' }), { code: 'requirement-snapshot-required' });
+  const sub = await f.submit(); f.approve(sub); const snap = await f.activate(sub);
+  await assert.rejects(f.bus.broker.requirementsFor(f.caps.pm, { project_key: 'stack', requirement_snapshot: snap.id }));
+  await assert.rejects(f.bus.requirementCatalog.resolve({ business: 'other', project: 'timer', snapshot: snap.id }));
+});
+test('fresh runtime resolves identical approved snapshot from committed records and journal alone', async t => {
+  const f = await setup(t), sub = await f.submit(); f.approve(sub); const snap = await f.activate(sub);
+  await f.bus.close(); const next = await f.boot();
+  assert.deepEqual(await next.requirementCatalog.resolve({ business: 'demo', project: 'timer', snapshot: snap.id }), snap);
+});
+test('owner-only explicit revocation rejects stale revisions and makes pending work wait for replacement', async t => {
+  const f = await setup(t), sub = await f.submit(); f.approve(sub); const snap = await f.activate(sub);
+  const args = { project: 'timer', snapshot: snap.id, revision: 1, reason: 'Replace requirements', op: 'revoke1' };
+  for (const who of ['pm','human','reader']) assert.throws(() => f.call(who, 'prd.revoke.console', args), { code: 'human-required' });
+  await assert.rejects(f.call('console', 'prd.revoke.console', { ...args, revision: 0 }));
+  const revoked = await f.call('console', 'prd.revoke.console', args);
+  assert.deepEqual(await f.call('console', 'prd.revoke.console', args), revoked);
+  await assert.rejects(f.bus.broker.requirementsFor(f.caps.pm, { project_key: 'timer', requirement_snapshot: snap.id }), { code: 'requirement-snapshot-revoked' });
+  assert.equal(f.journal.readRequest('demo', f.request).wait.code, 'requirement-snapshot-revoked');
+});
+test('mode and spec changes invalidate admissions and active snapshots', async t => {
+  const f = await setup(t), sub = await f.submit(); f.approve(sub);
+  let state = f.journal.readRequest('demo', f.request);
+  await f.call('console', 'request.mode.console', { request: f.request, revision: state.revision, mode_revision: state.mode_revision, mode: 'guided', op: 'mode1' });
+  await assert.rejects(f.activate(sub));
+  state = f.journal.readRequest('demo', f.request);
+  const next = await f.call('pm', 'prd.submit', { ...f.args, revision: state.revision, op: 'prd2' });
+  f.approve(next, 'human'); const snap = await f.activate(next);
+  const spec = await f.call('human', 'project.file', { project: 'timer', path: 'spec/north-star.md' });
+  await f.call('human', 'project.file.save', { project: 'timer', path: spec.path, base: spec.commit, content: 'Changed intent\n' });
+  await assert.rejects(f.bus.requirementCatalog.resolve({ business: 'demo', project: 'timer', snapshot: snap.id }), { code: 'requirement-snapshot-revoked' });
+});
+
+for (const fault of ['missing-authority','self-instance','same-definition']) test(`autonomous arbiter ${fault} is a setup wait without a decision`, async t => {
+  const f = await setup(t), b = structuredClone(f.config.businesses.demo);
+  if (fault === 'missing-authority') b.roles.cto.authority.withinRole = b.roles.cto.authority.withinRole.filter(a => a !== 'decision.resolve.technical');
+  if (fault === 'self-instance') b.arbiters.technical = 'pm';
+  if (fault === 'same-definition') b.roles.cto.definition = 'pm';
+  f.bus.broker.reloadBusiness('demo', b);
+  const count = f.store.get('SELECT count(*) n FROM decisions').n, result = await f.submit();
+  assert.equal(result.wait.kind, 'setup'); assert.match(result.wait.code, /^prd-(arbiter-unavailable|self-approval)$/);
+  assert.equal(f.store.get('SELECT count(*) n FROM decisions').n, count);
+});
+test('activation rechecks current role authority and cannot use another candidate decision', async t => {
+  const f = await setup(t), sub = await f.submit(); f.approve(sub);
+  await assert.rejects(f.bus.requirementCatalog.activate({ request: f.request, decision: f.state.selection.decision }), { code: 'prd-candidate-required' });
+  const b = structuredClone(f.config.businesses.demo); b.roles.cto.authority.withinRole = b.roles.cto.authority.withinRole.filter(a => a !== 'decision.resolve.technical');
+  f.bus.broker.reloadBusiness('demo', b);
+  await assert.rejects(f.activate(sub), { code: 'prd-arbiter-unavailable' });
+  assert.equal(f.store.get('SELECT count(*) n FROM request_snapshots').n, 0);
+});
+test('submission replay has one decision and rejects a changed operation binding', async t => {
+  const f = await setup(t), sub = await f.submit();
+  assert.deepEqual(await f.submit(), sub);
+  await assert.rejects(f.call('pm', 'prd.submit', { ...f.args, prd_version: '2.0' }), { code: 'request-operation-conflict' });
+});
+test('changed current intent is refused before submission creates a decision', async t => {
+  const f = await setup(t), count = f.store.get('SELECT count(*) n FROM decisions').n;
+  f.save('spec/north-star.md', 'Changed spec\n');
+  await assert.rejects(f.submit(), { code: 'stale-spec' });
+  assert.equal(f.store.get('SELECT count(*) n FROM decisions').n, count);
+});
+test('activation rollback leaves no snapshot, receipt or continuation after an insert failure', async t => {
+  const f = await setup(t), sub = await f.submit();
+  // A host fixture trigger injects a failure after the snapshot insert, before its activation event.
+  const db = new DatabaseSync(join(f.f.dataRoot, 'bus/bus.sqlite'));
+  db.exec("CREATE TRIGGER reject_activation BEFORE INSERT ON request_snapshot_events BEGIN SELECT RAISE(ABORT,'fixture-interruption'); END");
+  f.approve(sub);
+  await assert.rejects(f.activate(sub));
+  assert.equal(f.store.get('SELECT count(*) n FROM request_snapshots').n, 0);
+  assert.equal(f.journal.effects('demo', f.request).filter(e => e.kind === 'prd-activate').length, 0);
+  assert.notEqual(f.journal.readRequest('demo', f.request).phase, 'planning');
+  db.exec('DROP TRIGGER reject_activation'); db.close();
+});
+test('owner draft read shows exact bound PRD and requirements without a generic file endpoint', async t => {
+  const f = await setup(t), sub = await f.submit();
+  const draft = await f.call('console', 'request.draft.console', { request: f.request });
+  assert.deepEqual(draft.candidate, sub.candidate); assert.equal(draft.prd, '# Timer 1.0\nA timer.\n'); assert.equal(draft.requirements[0].id, 'REQ-POMO-1');
+  for (const who of ['pm','human','reader']) assert.throws(() => f.call(who, 'request.draft.console', { request: f.request }), { code: 'human-required' });
+});
+
+test('3p to 3q migration preserves existing request rows and the append-only guards', async t => {
+  const { readFileSync, chmodSync } = await import('node:fs');
+  const { Store, schemaDigest } = await import('../src/store.mjs');
+  const root = tmp(t, 'prd-migration-'); mkdirSync(join(root, 'bus'), { mode: 0o700 });
+  const path = join(root, 'bus/bus.sqlite'), ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8')
+    .replace(",'prd.submitted','prd.activated','prd.invalidated','mode.changed'", '');
+  const db = new DatabaseSync(path); db.exec(ddl);
+  const at = '2026-11-03T00:00:00.000Z';
+  db.prepare('INSERT INTO meta VALUES(?,?)').run('schema_version', '3p');
+  db.prepare('INSERT INTO meta VALUES(?,?)').run('schema_digest', schemaDigest(db));
+  db.prepare('INSERT INTO meta VALUES(?,?)').run('comment_cursor_epoch', at);
+  db.prepare("INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES('m',?,'demo','human','human','pm','INFO','Timer')").run(at);
+  db.prepare("INSERT INTO events(id,at,business,kind,subject,body) VALUES('r',?,'demo','human.input','m',?)").run(at, JSON.stringify({ kind: 'instruction', message: 'm' }));
+  db.prepare("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',1,'opened',?,'{}')").run(at);
+  const before = db.prepare('SELECT * FROM request_events').all();
+  assert.throws(() => db.prepare("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',2,'prd.submitted',?,'{}')").run(at));
+  db.close(); chmodSync(path, 0o600);
+  assert.throws(() => new Store(root), /schema-mismatch/);
+  const store = new Store(root, { migrateSchema: true }); t.after(() => store.close());
+  assert.deepEqual(store.all('SELECT * FROM request_events'), before);
+  assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
+  store.run("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',2,'prd.submitted',?,'{}')", at);
+  assert.throws(() => store.run('DELETE FROM request_events'), /append-only/);
+  assert.throws(() => store.run("UPDATE request_events SET body='{}'"), /append-only/);
+  assert.throws(() => store.run("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',4,'prd.activated',?,'{}')", at), /revision-conflict/);
+});
+
+test('guided Console approval uses the existing confirmation digest and attributed resolution evidence', async t => {
+  const f = await setup(t, 'guided'), sub = await f.submit();
+  const preview = f.call('console', 'decision.preview.console', { id: sub.decision.id });
+  f.call('console', 'decision.resolve.console', { id: sub.decision.id, choice: 'yes', digest: preview.digest });
+  const snap = await f.activate(sub); assert.equal(snap.approval.via, 'console');
+  assert.equal(f.store.get("SELECT count(*) n FROM messages WHERE decision=? AND class='DECISION'", sub.decision.id).n, 1);
+});
+test('declined draft returns to discovery and mode change revokes an active snapshot', async t => {
+  const f = await setup(t), sub = await f.submit();
+  f.call('cto', 'decision.resolve', { id: sub.decision.id, choice: 'no' });
+  const discovery = await f.bus.requestLifecycle.resume('demo', f.request);
+  assert.equal(discovery.phase, 'discovery'); assert.equal(discovery.wait.code, 'prd-changes-requested');
+  const next = await f.call('pm', 'prd.submit', { ...f.args, revision: discovery.revision, op: 'revision2' });
+  f.approve(next); const snap = await f.activate(next);
+  const state = f.journal.readRequest('demo', f.request);
+  await f.call('console', 'request.mode.console', { request: f.request, revision: state.revision, mode_revision: state.mode_revision, mode: 'guided', op: 'mode2' });
+  await assert.rejects(f.bus.requirementCatalog.resolve({ business: 'demo', project: 'timer', snapshot: snap.id }), { code: 'requirement-snapshot-revoked' });
+});

packages/bus/tests/request-intake.test.mjs

@@ -274,7 +274,7 @@ test('3n_to_3o_is_explicit_offline_preserves_rows_and_old_engine_refuses_new_sch
   const root = mkdtempSync(join(tmpdir(), 'request-migration-'));
   t.after(() => rmSync(root, { recursive: true, force: true })); mkdirSync(join(root, 'bus'), { mode: 0o700 });
   // Task 2's 3o→3p step widens only request_events.kind, so the 3o schema is the current one with the old list.
-  const path = join(root, 'bus/bus.sqlite'), currentDDL = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','cancelled'", "'setup.wait','cancelled'");
+  const path = join(root, 'bus/bus.sqlite'), currentDDL = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','cancelled'", "'setup.wait','cancelled'");
   const newDDL = readFileSync(new URL('../src/migrations/3n-3o.sql', import.meta.url), 'utf8');
   assert.ok(currentDDL.endsWith(newDDL)); const oldDDL = currentDDL.slice(0, -newDDL.length);
   const { schemaDigest, SCHEMA_VERSION } = awaitImportStore;
@@ -286,7 +286,7 @@ test('3n_to_3o_is_explicit_offline_preserves_rows_and_old_engine_refuses_new_sch
   db.close(); chmodSync(path, 0o600); const before = readFileSync(path);
   assert.throws(() => new Store(root), /schema-mismatch/); assert.deepEqual(readFileSync(path), before);
   const cli = spawnSync(process.execPath, [new URL('../src/migrate-schema.mjs', import.meta.url).pathname, root], { encoding: 'utf8' });
-  assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3p ready\n'); assert.equal(SCHEMA_VERSION, '3p');
+  assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3q ready\n'); assert.equal(SCHEMA_VERSION, '3q');
   const store = new Store(root); assert.equal(store.get("SELECT id FROM events").id, 'preserved');
   for (const table of ['request_events','request_effects','request_snapshots','request_snapshot_events','request_project_events']) assert.equal(store.get(`SELECT count(*) n FROM ${table}`).n, 0);
   store.close(); assert.equal(spawnSync(process.execPath, [new URL('../src/migrate-schema.mjs', import.meta.url).pathname, root]).status, 0);
@@ -307,7 +307,7 @@ test('3o_to_3p_widens_only_the_event_kinds_and_keeps_rows_seq_and_guards', t =>
   t.after(() => rmSync(root, { recursive: true, force: true })); mkdirSync(join(root, 'bus'), { mode: 0o700 });
   const path = join(root, 'bus/bus.sqlite');
   // The 3o schema is the current one with the narrower kind list.
-  const ddl3o = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','cancelled'", "'setup.wait','cancelled'");
+  const ddl3o = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','cancelled'", "'setup.wait','cancelled'");
   const { schemaDigest } = awaitImportStore, AT = '2026-11-03T00:00:00.000Z';
   const db = new DatabaseSync(path); db.exec(ddl3o);
   db.prepare('INSERT INTO meta VALUES(?,?)').run('schema_version', '3o');
@@ -323,7 +323,7 @@ test('3o_to_3p_widens_only_the_event_kinds_and_keeps_rows_seq_and_guards', t =>
   db.close(); chmodSync(path, 0o600);
   assert.throws(() => new Store(root), /schema-mismatch/);
   const cli = spawnSync(process.execPath, [new URL('../src/migrate-schema.mjs', import.meta.url).pathname, root], { encoding: 'utf8' });
-  assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3p ready\n');
+  assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3q ready\n');
   const store = new Store(root);
   assert.deepEqual(store.all('SELECT * FROM request_events ORDER BY seq'), before, 'every row keeps its seq and bytes');
   assert.deepEqual(store.all('SELECT * FROM request_effects ORDER BY seq'), effects);

packages/bus/tests/schema.test.mjs

@@ -525,6 +525,7 @@ test('v3b prototype refusals, views and append-only mutations', (t) => {
   // Row 206 (#1704) adds thread_shares and its 3 append-only triggers; thread-shares.test.mjs tests them.
   // Row 271 (#1776) adds session_ops, message_cancels and 7 triggers; session-clear.test.mjs tests them.
   // #797 / #1839 Task 1 adds five append-only lifecycle tables and eighteen guards.
+  // Task 4 (3p to 3q) widens request_events.kind in place; all 30 tables and 114 guards remain.
   assert.equal(count('table') - 1, 30);
   assert.equal(count('trigger'), 114);
   assert.equal(count('view'), 5);

packages/bus/tests/session-clear.test.mjs

@@ -530,7 +530,7 @@ test('T4 schema: a 3m store migrates to 3n, session_ops and message_cancels are
   const before = tables(path);
   assert.throws(() => new Store(root), /schema-mismatch/);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   const after = tables(path);
   for (const name of ['session_ops', 'message_cancels', 'request_events', 'request_effects', 'request_snapshots', 'request_snapshot_events', 'request_project_events']) {
     assert.deepEqual(after[name], [], name);
@@ -571,7 +571,7 @@ async function engine3m(root) {
   const source = readFileSync(here('../src/store.mjs'), 'utf8');
   const edited = source
     .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3m).digest('hex')}';`)
-    .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3m';")
+    .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3m';")
     .replace(/\}, \{\n {2}\/\/ Row 271 \(#1776\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
   assert.ok(edited.includes("export const SCHEMA_VERSION = '3m';") && edited.includes('3l-3m') && !edited.includes('3m-3n'), 'the 3m engine has no 3m→3n step');
   writeFileSync(join(moduleRoot, 'src/store.mjs'), edited);
@@ -588,7 +588,7 @@ test('T4 schema: the 3m engine refuses a 3n store, migrating or not, and leaves
   const Store3m = await engine3m(root);
   assert.throws(() => new Store3m(root), /schema-mismatch/);
   assert.throws(() => new Store3m(root, { migrateSchema: true }), /schema-mismatch/);
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(master(path), was);
   assert.deepEqual(tables(path), before);
   // The same engine opens a fresh 3m store, so the refusal is the version's and not a broken copy.

packages/bus/tests/sessions.test.mjs

@@ -374,7 +374,7 @@ test('T7: a 3k store migrates to 3l, each thread\'s default session has the thre
   const { root, path } = source3k(t);
   assert.throws(() => new Store(root), /schema-mismatch/);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   // The migrated schema is the fresh one, object for object.
   const freshRoot = mkdtempSync(join(tmpdir(), 'sessions-fresh-'));
   t.after(() => rmSync(freshRoot, { recursive: true, force: true }));
@@ -430,7 +430,7 @@ async function engine3k(root) {
   const source = readFileSync(here('../src/store.mjs'), 'utf8');
   const edited = source
     .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3k).digest('hex')}';`)
-    .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3k';")
+    .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3k';")
     .replace(/\}, \{\n {2}\/\/ Row 264a \(#1769\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
   assert.ok(edited.includes("export const SCHEMA_VERSION = '3k';") && edited.includes('3j-3k') && !edited.includes('3k-3l'), 'the 3k engine has no 3k→3l step');
   writeFileSync(join(moduleRoot, 'src/store.mjs'), edited);
@@ -447,7 +447,7 @@ test('T7 D9: the 3k engine refuses a 3l store, migrating or not, and leaves it a
   const Store3k = await engine3k(root);
   assert.throws(() => new Store3k(root), /schema-mismatch/);
   assert.throws(() => new Store3k(root, { migrateSchema: true }), /schema-mismatch/);
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(master(path), was);
   assert.deepEqual(rows(path), before);
   // The same engine opens a fresh 3k store, so the refusal is the version's and not a broken copy.

packages/bus/tests/spec-migration.test.mjs

@@ -63,7 +63,7 @@ test('3e copy migrates through 3f to 3g preserving every old column/row and addi
   assert.throws(() => new Store(root), /schema-mismatch/);
   const store = new Store(root, { migrateSchema: true });
   try {
-    assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3p');
+    assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
     assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: null, spec_commit: null });
     assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok');
     assert.deepEqual(store.all('PRAGMA foreign_key_check'), []);

packages/bus/tests/thread-migration.test.mjs

@@ -49,7 +49,7 @@ async function database3d(t) {
   const source = readFileSync(here('../src/store.mjs'), 'utf8');
   const pinned = source
     .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3d).digest('hex')}';`)
-    .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3d';")
+    .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3d';")
     .replace(/\}, \{\n {2}from: '3d',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
   assert.ok(!pinned.includes('3d-3e'), 'the 3d engine has no 3d→3e step');
   writeFileSync(join(moduleRoot, 'src/store.mjs'), pinned);
@@ -101,7 +101,7 @@ test('REQ-THREAD-1 REQ-ORG-2: the 3d→3e migration adds the thread tables and l
   assert.throws(() => new Store(root), /schema-mismatch/);
   const store = new Store(root, { migrateSchema: true });
   t.after(() => store.close());
-  assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3p');
+  assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
   for (const table of ['projects', 'threads', 'thread_claims', 'message_threads', 'thread_opens'])
     assert.equal(store.get(`SELECT count(*) n FROM ${table}`).n, 0, table);
   // Parked: the undelivered ones are counted per business and role, and none is changed or given a thread.
@@ -161,5 +161,5 @@ test('REQ-THREAD-1: the offline command prints the kept-message counts before th
   const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname;
   const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' });
   assert.equal(migrated.status, 0, migrated.stderr);
-  assert.equal(migrated.stdout, [...REPORT, 'bus schema 3p ready', ''].join('\n'));
+  assert.equal(migrated.stdout, [...REPORT, 'bus schema 3q ready', ''].join('\n'));
 });

packages/bus/tests/thread-shares.test.mjs

@@ -403,7 +403,7 @@ test('a 3l store migrates to 3m: thread_shares is new and empty, and every other
   const before = tables(path);
   assert.throws(() => new Store(root), /schema-mismatch/);
   new Store(root, { migrateSchema: true }).close();
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   const after = tables(path);
   assert.deepEqual(after.thread_shares, []);
   delete after.thread_shares;
@@ -439,7 +439,7 @@ async function engine3l(root) {
   const source = readFileSync(here('../src/store.mjs'), 'utf8');
   const edited = source
     .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3l).digest('hex')}';`)
-    .replace("export const SCHEMA_VERSION = '3p';", "export const SCHEMA_VERSION = '3l';")
+    .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3l';")
     .replace(/\}, \{\n {2}\/\/ Row 206 \(193b2[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
   assert.ok(edited.includes("export const SCHEMA_VERSION = '3l';") && edited.includes('3k-3l') && !edited.includes('3l-3m'), 'the 3l engine has no 3l→3m step');
   writeFileSync(join(moduleRoot, 'src/store.mjs'), edited);
@@ -456,7 +456,7 @@ test('the 3l engine refuses a 3m store, migrating or not, and leaves it as it wa
   const Store3l = await engine3l(root);
   assert.throws(() => new Store3l(root), /schema-mismatch/);
   assert.throws(() => new Store3l(root, { migrateSchema: true }), /schema-mismatch/);
-  assert.equal(version(path), '3p');
+  assert.equal(version(path), '3q');
   assert.deepEqual(master(path), was);
   assert.deepEqual(tables(path), before);
   // The same engine opens a fresh 3l store, so the refusal is the version's and not a broken copy.

packages/bus/tests/usage-migration.test.mjs

@@ -68,7 +68,7 @@ test('REQ-LIMIT-4 3g copy migrates through 3h, 3i, 3j and 3k to 3l preserving ev
   assert.throws(() => new Store(root), /schema-mismatch/);
   const store = new Store(root, { migrateSchema: true });
   try {
-    assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3p');
+    assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q');
     assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null });
     assert.equal(store.get('SELECT count(*) AS n FROM usage_records').n, 0);
     assert.equal(store.get('SELECT name FROM instance_names').name, 'Builder');

packages/project/src/index.mjs

@@ -261,3 +261,5 @@ export async function writeHumanFile({ path, content, base, ...context }) {
 }
 
 export { readProjectFile, readSpec, listProjectFiles } from './read.mjs';
+
+export { readPrdCandidate } from './prd.mjs';

packages/project/src/prd.mjs

@@ -0,0 +1,55 @@
+// Private committed-blob access does not extend the public document editor's path vocabulary.
+import { ProjectError, ID } from './paths.mjs';
+import { openRevision, committedBlob, readSpec } from './read.mjs';
+import { git } from './git.mjs';
+import { validateRecord } from './validate.mjs';
+const fail = code => { throw new ProjectError(code, code); };
+const KEYS = ['business','project','dataRoot','request','prd_version','records_commit','prd_path','requirement_ids','mode_revision'];
+export async function readPrdMaterial(args) {
+  if (!args || typeof args !== 'object' || KEYS.some(k => !Object.hasOwn(args, k)) || Object.keys(args).some(k => !KEYS.includes(k)) ||
+      typeof args.request !== 'string' || !args.request.length || !/^\d+\.\d+(?:\.\d+)?$/.test(args.prd_version) ||
+      args.prd_path !== `docs/prd-${args.prd_version}.md` || !Number.isSafeInteger(args.mode_revision) || args.mode_revision < 1 ||
+      !Array.isArray(args.requirement_ids) || !args.requirement_ids.length || args.requirement_ids.length > 64 ||
+      args.requirement_ids.some(id => typeof id !== 'string' || !ID.test(id)) || new Set(args.requirement_ids).size !== args.requirement_ids.length)
+    fail('prd-candidate-invalid');
+  const { records_commit, prd_path, requirement_ids, mode_revision, request, project, prd_version, ...context } = args;
+  const ctx = { ...context, project }, opened = await openRevision({ ...ctx, commit: records_commit });
+  const { repo, head } = opened;
+  let total = 0;
+  const selected = async path => {
+    const item = await committedBlob(repo, records_commit, path, 256 * 1024);
+    const current = await committedBlob(repo, head, path, 256 * 1024);
+    if (item.blob !== current.blob) fail('prd-candidate-stale');
+    total += item.bytes;
+    if (total > 256 * 1024) fail('prd-candidate-size');
+    return item;
+  };
+  const prd = await selected(prd_path);
+  if (!prd.content.trim() || prd.bytes > 32768) fail('prd-candidate-size');
+  const requirements = [], records = [], authors = [];
+  const authored = async path => {
+    const commit = await git(repo, ['log', '-1', '--format=%H', records_commit, '--', path]);
+    const name = await git(repo, ['show', '-s', '--format=%an', commit]);
+    const split = name.indexOf('.');
+    if (split < 1 || !ID.test(name.slice(0, split)) || !ID.test(name.slice(split + 1))) fail('prd-author-invalid');
+    const author = { definition: name.slice(0, split), instance: name.slice(split + 1), commit };
+    if (!authors.some(a => a.commit === commit && a.instance === author.instance)) authors.push(author);
+  };
+  await authored(prd_path);
+  for (const id of requirement_ids) {
+    const path = `requirements/${id}.json`, item = await selected(path);
+    let record; try { record = JSON.parse(item.content); } catch { fail('prd-requirement-invalid'); }
+    if (validateRecord('requirement', record).length || record.id !== id || !record.title.trim() || !record.text.trim()) fail('prd-requirement-invalid');
+    if (Buffer.byteLength(record.text) > 16384) fail('prd-candidate-size');
+    requirements.push({ id, title: record.title, path, blob: item.blob }); records.push(record);
+    await authored(path);
+  }
+  const spec = await readSpec(ctx), previousSpec = await readSpec({ ...ctx, commit: records_commit });
+  if (!spec.confirmed || previousSpec.commit !== spec.commit) fail('stale-spec');
+  // Detect a concurrent records write, even when it happened during the bounded reads.
+  if (await git(repo, ['rev-parse', 'HEAD']) !== head) fail('prd-candidate-stale');
+  const candidate = { version: 1, request, project, prd_version, records_commit, prd_path, prd_blob: prd.blob,
+    requirements, spec_commit: spec.commit, mode_revision };
+  return { candidate, prd: prd.content, requirements: records, authors };
+}
+export async function readPrdCandidate(args) { return (await readPrdMaterial(args)).candidate; }

packages/project/src/read.mjs

@@ -15,19 +15,22 @@ function text(bytes) {
     return value;
   } catch { fail('project-read-content', 'expected bounded UTF-8 without NUL'); }
 }
-async function blob(repo, revision, path) {
+export async function committedBlob(repo, revision, path, maxBytes = MAX_BYTES) {
   const entry = await git(repo, ['ls-tree', '-z', revision, '--', path]);
   if (!entry) fail('project-read-missing', 'file absent at revision');
   const match = /^(100644|100755) blob ([0-9a-f]{40})\t([^\0]+)\0$/.exec(entry);
   if (!match || match[3] !== path) fail('project-read-content', 'expected regular Git blob');
   const size = Number(await git(repo, ['cat-file', '-s', match[2]]));
-  if (!Number.isSafeInteger(size) || size < 0 || size > MAX_BYTES)
+  if (!Number.isSafeInteger(size) || size < 0 || size > maxBytes)
     fail('project-read-content', 'blob exceeds size bound');
-  return text(await gitBytes(repo, ['cat-file', 'blob', match[2]], MAX_BYTES));
+  return { blob: match[2], bytes: size, content: text(await gitBytes(repo, ['cat-file', 'blob', match[2]], maxBytes)) };
+}
+async function blob(repo, revision, path) {
+  return (await committedBlob(repo, revision, path)).content;
 }
 // The repository, its HEAD and the revision to read, with the committed project identity checked
 // against the caller's business and project. Shared by the file read and the file list.
-async function openRevision({ business, project, dataRoot, commit }) {
+export async function openRevision({ business, project, dataRoot, commit }) {
   const repo = projectRepo({ business, project, dataRoot });
   await inspectGit(repo);
   let head;

packages/project/tests/prd-fixture.mjs

@@ -0,0 +1,24 @@
+import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { execFileSync } from 'node:child_process';
+import { initProject, writeRecord } from '../src/index.mjs';
+export const git = (repo, ...args) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', '-c', 'maintenance.auto=false', '-c', 'gc.auto=0', '-C', repo, ...args], { encoding: 'utf8', env: { ...process.env, GIT_CONFIG_GLOBAL: '/dev/null', GIT_CONFIG_SYSTEM: '/dev/null', GIT_AUTHOR_NAME: 'jason', GIT_AUTHOR_EMAIL: '[email protected]', GIT_COMMITTER_NAME: 'jason', GIT_COMMITTER_EMAIL: '[email protected]' } }).trim();
+export async function prdFixture(t, context) {
+  const root = context?.dataRoot ?? mkdtempSync(join(tmpdir(), 'prd-'));
+  if (!context) t.after(() => rmSync(root, { recursive: true, force: true }));
+  context ??= { dataRoot: root, project: 'timer', business: { id: 'demo', human: 'jason', projects: { timer: { root: join(root, 'code') } } } };
+  const repo = context.business.projects[context.project].repo ?? join(context.dataRoot, 'projects', context.business.id, context.project);
+  if (!existsSync(join(repo, '.git'))) await initProject(context);
+  const save = (path, bytes, author = 'jason') => {
+    writeFileSync(join(repo, path), bytes);
+    git(repo, 'add', '--', path); git(repo, 'commit', '--author', `${author} <[email protected]>`, '-qm', 'Save ' + path);
+    return git(repo, 'rev-parse', 'HEAD');
+  };
+  if (git(repo, 'show', 'HEAD:spec/north-star.md') !== 'Build a timer.') save('spec/north-star.md', 'Build a timer.\n');
+  const author = { definition: 'pm', instance: 'pm' };
+  await writeRecord({ ...context, author, path: 'docs/prd-1.0.md', record: '# Timer 1.0\nA timer.\n' });
+  await writeRecord({ ...context, author, path: 'requirements/REQ-POMO-1.json', record: { recordVersion: 1, id: 'REQ-POMO-1', title: 'Timer', text: 'A working timer.', status: 'draft' } });
+  const args = { ...context, request: 'request-1', prd_version: '1.0', records_commit: git(repo, 'rev-parse', 'HEAD'), prd_path: 'docs/prd-1.0.md', requirement_ids: ['REQ-POMO-1'], mode_revision: 1 };
+  return { context, repo, save, args };
+}

packages/project/tests/prd.test.mjs

@@ -0,0 +1,51 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { readFileSync, rmSync, symlinkSync } from 'node:fs';
+import { join } from 'node:path';
+import * as api from '../src/index.mjs';
+import { git, prdFixture } from './prd-fixture.mjs';
+
+test('candidate reads exact committed blobs and tolerates unrelated HEAD commits', async t => {
+  const f = await prdFixture(t), candidate = await api.readPrdCandidate(f.args);
+  assert.equal(candidate.prd_blob, git(f.repo, 'rev-parse', `${f.args.records_commit}:docs/prd-1.0.md`));
+  assert.equal(candidate.requirements[0].blob, git(f.repo, 'rev-parse', `${f.args.records_commit}:requirements/REQ-POMO-1.json`));
+  assert.equal(candidate.spec_commit, git(f.repo, 'log', '-1', '--format=%H', '--', 'spec/north-star.md'));
+  assert.deepEqual(Object.keys(candidate).sort(), ['version','request','project','prd_version','records_commit','prd_path','prd_blob','requirements','spec_commit','mode_revision'].sort());
+  f.save('docs/unrelated.md', 'Unrelated');
+  assert.deepEqual(await api.readPrdCandidate(f.args), candidate);
+});
+for (const path of ['docs/prd-1.0.md', 'requirements/REQ-POMO-1.json']) test(`changed selected blob is stale: ${path}`, async t => {
+  const f = await prdFixture(t);
+  f.save(path, path.endsWith('.md') ? 'Edited' : JSON.stringify({ recordVersion: 1, id: 'REQ-POMO-1', title: 'Edited', text: 'Changed', status: 'accepted' }));
+  await assert.rejects(api.readPrdCandidate(f.args), { code: 'prd-candidate-stale' });
+});
+test('candidate rejects extra assertions, duplicate IDs, wrong version and unreachable commits', async t => {
+  const f = await prdFixture(t);
+  for (const patch of [{ prd_blob: 'forged' }, { requirement_ids: ['REQ-POMO-1','REQ-POMO-1'] }, { prd_version: '2.0' }, { records_commit: 'a'.repeat(40) }])
+    await assert.rejects(api.readPrdCandidate({ ...f.args, ...patch }));
+  await assert.rejects(api.readProjectFile({ ...f.context, path: 'requirements/REQ-POMO-1.json' }), { code: 'project-read-path' });
+});
+test('candidate rejects malformed requirements, mismatched IDs and nonregular blobs', async t => {
+  const f = await prdFixture(t);
+  const original = JSON.parse(readFileSync(join(f.repo, 'requirements/REQ-POMO-1.json')));
+  for (const record of [{ ...original, extra: true }, { ...original, id: 'OTHER' }, { ...original, title: '  ' }]) {
+    const records_commit = f.save('requirements/REQ-POMO-1.json', JSON.stringify(record), 'pm.pm');
+    await assert.rejects(api.readPrdCandidate({ ...f.args, records_commit }));
+  }
+  rmSync(join(f.repo, 'docs/prd-1.0.md')); symlinkSync('../spec/north-star.md', join(f.repo, 'docs/prd-1.0.md'));
+  git(f.repo, 'add', '-A'); git(f.repo, 'commit', '-qm', 'Link');
+  await assert.rejects(api.readPrdCandidate({ ...f.args, records_commit: git(f.repo, 'rev-parse', 'HEAD') }));
+});
+test('candidate enforces 64 requirements, 32 KiB PRD, 16 KiB requirement and 256 KiB aggregate', async t => {
+  const f = await prdFixture(t);
+  await assert.rejects(api.readPrdCandidate({ ...f.args, requirement_ids: Array.from({ length: 65 }, (_, i) => `R${i}`) }));
+  let records_commit = f.save('docs/prd-1.0.md', 'x'.repeat(32769));
+  await assert.rejects(api.readPrdCandidate({ ...f.args, records_commit }), { code: 'prd-candidate-size' });
+  f.save('docs/prd-1.0.md', 'x'.repeat(32768), 'pm.pm');
+  const record = { recordVersion: 1, id: 'REQ-POMO-1', title: 'Timer', text: 'x'.repeat(16385), status: 'draft' };
+  records_commit = f.save('requirements/REQ-POMO-1.json', JSON.stringify(record), 'pm.pm');
+  await assert.rejects(api.readPrdCandidate({ ...f.args, records_commit }), { code: 'prd-candidate-size' });
+  const ids = Array.from({ length: 16 }, (_, i) => `R${i}`);
+  for (const id of ids) f.save(`requirements/${id}.json`, JSON.stringify({ ...record, id, text: 'x'.repeat(16384) }), 'pm.pm');
+  await assert.rejects(api.readPrdCandidate({ ...f.args, records_commit: git(f.repo, 'rev-parse', 'HEAD'), requirement_ids: ids }), { code: 'prd-candidate-size' });
+});