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

2026-11-04 10:49:15 -06:00
1 parent ea0e885bc9
commit c098fa022c
@@ -6015,3 +6015,4 @@ Owner directed static-only fallback after actual nativepriority drift: no new pr
 2026-11-04T03:23:23Z | Claude Sonnet 5.5 implementation sub-agent, t3/harden-project-bootstrap-executor | #802 bootstrap executor hardening after review R2 | One commit on 6dd4f92b0 for the three open R2 findings; nothing else touched (Minors #801, Tasks 3-8, roles/*.json, ROUTED, migrations out of scope; no schema change). N1: the working repository path is confined before and inside the helper. working.mjs ensure() now walks every directory between the data root and the repository with lstat, creates missing ones one at a time and refuses a link or a non-directory (working-repo-invalid); push() goes through the same ensure. The executor refuses a linked or non-directory ancestor as bootstrap-uncertain, keeps a repository that is already there on a first attempt, and on a retry adopts only a bare repository whose refs are root and main at the seed commit. I1: request-bootstrap.mjs gains guarded(effect): the freshness checks and the start of the effect run in one synchronous run with the organization lock held, after every asynchronous read the effect depends on. It wraps the clone, claim, remote removal and rename, the seed publish, add and commit, the working init/rename/fetch/ref writes (initializeWorking and project commitFile take an optional `around` that runs each write), the records init and the intent file publish. The old lock comment, which claimed more than fresh() gave, is replaced with what each form proves. I2: activate publishes the complete marker first and calls services.activate second, so a failed marker rename ends the step before the broker is touched. Readiness needs both a complete marker and a committed activate receipt at boot (pendingProject), at reload (activateManagedProject refuses bootstrap-marker-incomplete) and in admission() (project-pending), where a receipt alone used to be enough. RED/GREEN: run against 6dd4f92b0 source with the new tests copied in, 7 of 39 bus tests and 3 of 31 seat tests fail (n1 working link, n1 business-directory link, n1 foreign ref on retry, i1 intent interleaving, i2 EIO at the marker rename, i2 runtime marker fence, i2 marker removal live and after reopen; seat: both link refusals and the five-write around test); all pass now. Honest notes: five new tests also pass on the old source and are guards, not reproducers: the working-file ancestor (one bus test, one seat test), the preexisting repository on a first attempt, the kill-and-adopt retry, and the i1 seed interleaving (the old code ran fresh() directly before the publish on the default path, where origin is already removed); my first business-directory test asserted the link at the wrong path and was corrected. GREEN: brief trio 90/90 (was 79), bus 1488/1488 (was 1477), business 305/305, project+seat 271/271 (was 267), launcher+stack-release 261/261, zero skips. Limits: a git child's own running time after it starts is not covered by the lock, and the next step's check catches a change made meanwhile; between services.activate and the receipt commit there is one microtask where the broker already serves the project, and admission() fences that but tool serving is not fenced; push now refuses a symlinked working ancestor it used to follow. Logs /mnt/storage/scratch/tmp/V7Ps0O/. Not executed on any running instance; local commit only, no push/merge/PR/deployment.

 2026-11-04T16:35:38Z | GPT-6.1 Sol implementation sub-agent, t3/bootstrap-hardening-fix-802-1 | #802 bootstrap hardening fix round 1 before/after | Before: fast-forwarded the assigned worktree to reviewed 52b283c8a; C1 foreign working-repo retry adoption/internal object links, C2 stale records and intent commits, C3 broker serving before the activate receipt. After: a first-attempt foreign repository gets a durable refused working receipt; recovery requires this plan's ownership claim, created exclusively inside the init directory before rename, as well as the existing ref checks. Every working-repository Git child checks for internal links and alternates; bootstrap also rechecks between its recovery reads. The existing kill-and-adopt recovery remains green. Organization holdAsync keeps the existing on-disk lock through asynchronous effects and releases it on success or failure; records initialization, intent publication/add/commit and intent rollback stay protected, while the bootstrap's exact-add mapping keeps its own transaction. The complete marker is verified and the activate receipt commits before the broker reload; runtime no longer overrides pending admission, and receipt I/O failure stays fenced until a fault-free retry. RED on 52b283c8a with final tests overlaid: 79 existing tests pass and all 16 new regressions fail as expected across request-bootstrap, project-bootstrap and working (95 total, zero skips); all five supplied reproducers were ported with corrected assertions. Final gates, all zero failures/skips/cancellations: focused 103/103 (baseline 90), bus 1499/1499 (1488), business 307/307 (305), project+seat 274/274 (271), launcher+stack-release including external-release cases 286/286 (286). Gate 5 uses node --test packages/cli/tests/launcher.test.mjs packages/cli/tests/stack-release*.test.mjs because these are CLI test files, not separate packages. Corrections: an initial gate-5 command named nonexistent package directories and ran no tests; an intermediate bus run failed the newly added per-child C1 regression, also reproduced on the base and fixed before the final green rerun. An earlier candidate gate-5 run was superseded and only its own descendant processes stopped; the final run completed. No preexisting test failure was found. Evidence: /mnt/storage/scratch/tmp/qRkZh1/ (before-1..4.log, before-5-cli.log, after-1..5.log, red-final-snapshot.log, verified-files.sha256). Provider interruption recovered from complete logs and matching source/test hashes. The queue refuses linked worktrees, so the explicit controller assignment was followed without changing the canonical queue. One local commit only; no push, live-instance action, role/contract/migration edit or residual C1/C2/C3 finding.
+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.
@@ -2003,3 +2003,4 @@ are never rewritten or removed; corrections are new entries.
 2026-11-03 | GPT-6.1 Sol implementation sub-agent, t3/lifecycle-task-5a-create-recovery | #797 / Gitea #1839 lifecycle Task 5a | Optional journaled create operations, sync-only complete-scan adoption, host-only recovery with original attribution and PM follow-up resume; no migration. Recovery 21/21, required 61/61, tasks 121/121, bus 1477/1477, zero skips; diagnostics and parent reruns in /mnt/storage/scratch/tmp/CkjqAm/. Task 5b deferred, queue left to lead after linked-worktree refusal; one local commit only, no push or live changes.
 2026-11-04 | Claude Sonnet 5.5 implementation sub-agent, t3/harden-project-bootstrap-executor | #802 bootstrap executor hardening | N1 working path confined before and inside the helper, I1 guarded() makes the freshness check and every protected write one synchronous run under the organization lock, I2 complete marker published before activation and readiness needs marker plus committed receipt at boot, reload and admission. Brief trio 90/90, bus 1488/1488, business 305/305, project+seat 271/271, launcher+stack-release 261/261, zero skips; no schema change; local commit only, no push.
 2026-11-04 | GPT-6.1 Sol implementation sub-agent, t3/bootstrap-hardening-fix-802-1 | #802 bootstrap hardening fix round 1 | C1 permanent foreign-repo refusal, exclusive ownership claim and per-child internal-link/alternates checks; C2 async-held organization lock through records init, intent commit and rollback; C3 committed activate receipt before broker reload, fenced receipt failure and successful retry. All 16 added regressions fail on 52b283c8a; final focused 103/103, bus 1499/1499, business 307/307, project+seat 274/274, launcher+release including external cases 286/286, zero skips. Logs /mnt/storage/scratch/tmp/qRkZh1/; one local commit, no push, no role/contract/migration/live changes.
+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.
@@ -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'),
@@ -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.
@@ -593,6 +597,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');
@@ -613,18 +623,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);
@@ -5070,11 +5130,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); });
@@ -6387,6 +6454,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;
      }
@@ -7573,6 +7642,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;
@@ -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;
@@ -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; } });
 }
@@ -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;
@@ -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()]) });
+}
@@ -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';
@@ -579,11 +581,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
@@ -978,6 +990,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.
@@ -1048,6 +1061,7 @@ export async function startBroker({
        await stopSync();
        await server.close();
        await adapter?.close?.();
+        await requestLifecycle.close();
        await processing?.close();
        store.close();
        credentials.close();
@@ -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_VERSION = '3p';
+export const SCHEMA_SHA256 = '2a6a161b3d671418e690942f02203b4a014e939b46ae8f651da11f091dac5455';
+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',
@@ -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);
 });
@@ -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 });
@@ -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('}];\nconst timedTables', `}, { from: '3p', to: '3q', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
+    .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3r';")
+    .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);
@@ -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', () => {
-  assert.deepEqual([...CONSOLE_VERBS], ['decision.preview.console', 'decision.resolve.console']);
+test('N4: owner Console verbs have an explicit immutable whitelist and exclude decision.resolve', () => {
+  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));
 });
@@ -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);
 });
@@ -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);
 });
@@ -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) });
@@ -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' });
+});
@@ -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);
@@ -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);
@@ -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.
@@ -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.
@@ -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'), []);
@@ -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'));
 });
@@ -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.
@@ -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');
@@ -268,3 +268,5 @@ export async function writeHumanFile({ path, content, base, around, ...context }
 }

 export { readProjectFile, readSpec, listProjectFiles } from './read.mjs';
+
+export { readPrdCandidate } from './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; }
@@ -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;
@@ -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 };
+}
@@ -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' });
+});