// Trusted host journal. No SQL, transition or effect interface is exposed to workers.
import { createHash } from 'node:crypto';
import { isAbsolute, normalize } from 'node:path';
import { ID_PATTERN, parseGiteaUrl } from '../../organization/src/vocabulary.mjs';
import { checkVars } from '../../organization/src/vars.mjs';

export class RequestError extends Error {
  constructor(code) { super(code); this.code = code; }
}
export const refuse = code => { throw new RequestError(code); };
export const object = x => x !== null && typeof x === 'object' && !Array.isArray(x);
export const text = (x, max = 32768) => typeof x === 'string' && x.length > 0 && x.length <= max && !x.includes('\0');
export const id = x => text(x, 160) && /^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$/.test(x);
export const key = x => typeof x === 'string' && ID_PATTERN.test(x);
const hash = x => typeof x === 'string' && /^[a-f0-9]{64}$/.test(x);
const oid = x => typeof x === 'string' && /^[a-f0-9]{40}([a-f0-9]{24})?$/.test(x);
export const positive = x => Number.isSafeInteger(x) && x > 0;
export function exact(x, keys, optional = [], code = 'invalid-request') {
  if (!object(x) || keys.some(k => !Object.hasOwn(x, k)) || Object.keys(x).some(k => !keys.includes(k) && !optional.includes(k))) refuse(code);
}
const canonical = x => Array.isArray(x) ? x.map(canonical) : object(x)
  ? Object.fromEntries(Object.keys(x).sort().map(k => [k, canonical(x[k])])) : x;
export const digest = x => createHash('sha256').update(typeof x === 'string' ? x : JSON.stringify(canonical(x))).digest('hex');
const equal = (a, b) => digest(a) === digest(b);
const nullableId = x => x === null || id(x);
// A question belongs to the PM's discovery, PRD review and planning work.
export const QUESTION_PHASES = ['discovery','prd-review','planning'];
export const PICKER_CODES = ['request-existing-project','request-attachment-selected'];
export const optionList = x => Array.isArray(x) && x.length <= 8 && x.every(o => object(o) && Object.keys(o).length === 2 && id(o.key) && text(o.text, 200)) &&
  new Set(x.map(o => o.key)).size === x.length;
export function validatePlan(p) {
  const code = 'invalid-request-plan', shape = (x, keys) => exact(x, keys, [], code);
  shape(p, ['version','business','request','project','name','chat_project','mode','organization_revision','source','code_root','records_root','tracker','staff','intent','project_vars','existing']);
  if (p.version !== 1 || !key(p.business) || !id(p.request) || !key(p.project) || !text(p.name, 200) || /[\r\n]/.test(p.name) ||
      !nullableId(p.chat_project) || !['autonomous','guided'].includes(p.mode) || !text(p.organization_revision, 256) || typeof p.existing !== 'boolean') refuse(code);
  shape(p.source, ['project','commit','remote','forks']);
  if (!key(p.source.project) || !oid(p.source.commit) || (p.source.remote !== null && !parseGiteaUrl(p.source.remote, { repo: true }).url) || !Array.isArray(p.source.forks)) refuse(code);
  const instances = new Set();
  for (const f of p.source.forks) {
    shape(f, ['instance','url','grant','state']);
    if (!key(f.instance) || instances.has(f.instance) || !parseGiteaUrl(f.url, { repo: true }).url || !nullableId(f.grant) || !['verified','not-granted'].includes(f.state)) refuse(code);
    instances.add(f.instance);
  }
  for (const path of [p.code_root, p.records_root]) if (!text(path, 4096) || !isAbsolute(path) || normalize(path) !== path) refuse(code);
  if (p.code_root === p.records_root || p.code_root.startsWith(p.records_root + '/') || p.records_root.startsWith(p.code_root + '/')) refuse(code);
  shape(p.tracker, ['root','project']);
  if (!positive(p.tracker.root) || !positive(p.tracker.project)) refuse(code);
  shape(p.staff, ['pm','technical_reviewer','coder','reviewer','chief']);
  if (!Object.values(p.staff).every(key) || p.staff.pm === p.staff.technical_reviewer) refuse(code);
  shape(p.intent, ['text','sha256','source_message','generated']);
  if (!text(p.intent.text) || !hash(p.intent.sha256) || digest(p.intent.text) !== p.intent.sha256 || !id(p.intent.source_message) || typeof p.intent.generated !== 'boolean') refuse(code);
  if (!object(p.project_vars) || Object.hasOwn(p.project_vars, 'tracker.project')) refuse(code);
  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');
  exact(b.intent, ['text','sha256','source_message','generated'], [], 'invalid-request-event');
  if (!text(b.intent.text) || digest(b.intent.text) !== b.intent.sha256 || b.intent.source_message !== b.source_message || b.intent.generated !== false) refuse('invalid-request-event');
  return { version: 1, request, business, revision: 1, source_message: b.source_message, thread: b.thread, session: b.session,
    owner: b.owner, pm: b.pm, phase: 'selection', project: null, chat_project: null, mode: null, mode_revision: 0,
    plan_digest: null, spec_commit: null, snapshot: null, tasks: [], delivery: null, question: null, candidate: null, selection: null,
    // The owner's chosen route survives every setup wait; a wait code never says which route was chosen.
    route: null, wait: null, launch_refusal: null };
}
function reduce(state, e) {
  const code = 'invalid-request-event', b = e.body;
  if (e.kind === 'opened') { if (state || e.revision !== 1) refuse(code); return initial(e.business, e.request, b); }
  if (!state || e.revision !== state.revision + 1 || state.phase === 'cancelled') refuse(code);
  const s = structuredClone(state); s.revision = e.revision;
  if (e.kind === 'setup.wait') {
    exact(b, ['code','subject'], ['op','proposal_digest'], code);
    if (!text(b.code, 128) || !id(b.subject) || ('op' in b && (!id(b.op) || !hash(b.proposal_digest)))) refuse(code);
    s.wait = { kind: 'setup', code: b.code, subject: b.subject, resume_phase: s.phase, observed_revision: s.revision };
  } else if (e.kind === 'selection.proposed') {
    exact(b, ['op','proposal_digest','decision','bindings','kind'], [], code);
    if (s.phase !== 'selection' || s.wait?.kind === 'decision' || !id(b.op) || !hash(b.proposal_digest) || !id(b.decision) || !['mode','existing'].includes(b.kind) || !Array.isArray(b.bindings)) refuse(code);
    const choices = b.kind === 'mode' ? ['autonomous','guided'] : ['confirm'];
    if (b.bindings.length !== choices.length) refuse(code);
    for (let i = 0; i < choices.length; i++) {
      const binding = b.bindings[i]; exact(binding, ['plan','digest','decision','choice'], [], code); validatePlan(binding.plan);
      if (binding.choice !== choices[i] || binding.decision !== b.decision || binding.digest !== digest(binding.plan) ||
          binding.plan.business !== s.business || binding.plan.request !== s.request || binding.plan.staff.pm !== s.pm ||
          binding.plan.intent.source_message !== s.source_message || binding.plan.existing !== (b.kind === 'existing') ||
          (b.kind === 'mode' && binding.plan.mode !== binding.choice)) refuse(code);
    }
    s.selection = { decision: b.decision, bindings: structuredClone(b.bindings) };
    s.wait = { kind: 'decision', code: b.kind === 'mode' ? 'request-mode' : 'request-existing-confirmation', subject: b.decision, resume_phase: 'selection', observed_revision: s.revision };
  } else if (e.kind === 'selection.staged') {
    exact(b, ['decision','choice'], [], code);
    if (s.wait?.kind !== 'decision' || s.selection?.decision !== b.decision || !['existing','back'].includes(b.choice) ||
        (b.choice === 'existing') !== (s.wait.code === 'request-mode')) refuse(code);
    s.selection = null; s.route = 'existing';
    s.wait = { kind: 'setup', code: 'request-existing-project', subject: s.request, resume_phase: 'selection', observed_revision: s.revision };
  } else if (e.kind === 'selection.confirmed') {
    exact(b, ['decision','choice','digest'], [], code);
    const binding = s.selection?.bindings.find(p => p.decision === b.decision && p.choice === b.choice && p.digest === b.digest);
    if (s.wait?.kind !== 'decision' || !binding) refuse(code);
    const p = binding.plan;
    s.project = p.project; s.chat_project = p.chat_project; s.mode = p.mode; s.mode_revision = 1; s.plan_digest = binding.digest;
    s.phase = p.existing ? 'discovery' : 'bootstrap';
    s.wait = { kind: 'setup', code: p.existing ? 'request-attachment-unavailable' : 'request-bootstrap-unavailable', subject: s.request, resume_phase: s.phase, observed_revision: s.revision };
  } else if (e.kind === 'bootstrap.stale') {
    // The confirmed plan no longer matches the world (#800). Nothing was bound; the owner gets a fresh preview.
    exact(b, ['digest','code'], [], code);
    if (s.phase !== 'bootstrap' || s.plan_digest !== b.digest || b.code !== 'request-plan-stale') refuse(code);
    s.project = null; s.chat_project = null; s.mode = null; s.mode_revision = 0; s.plan_digest = null; s.selection = null; s.phase = 'selection';
    s.wait = { kind: 'setup', code: b.code, subject: s.request, resume_phase: 'selection', observed_revision: s.revision };
  } else if (e.kind === 'bootstrap.completed') {
    exact(b, ['digest','project','chat_project','spec_commit','organization_revision'], ['mode_revision','snapshot','candidate'], code);
    const existing = s.selection?.bindings.find(p => p.digest === s.plan_digest)?.plan.existing;
    if ((existing ? s.phase !== 'discovery' || s.attachment?.completed : s.phase !== 'bootstrap') ||
        s.plan_digest !== b.digest || b.project !== s.project || !(existing ? nullableId(b.chat_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 };
    if (existing) {
      if (!positive(b.mode_revision) || !nullableId(b.snapshot) || (b.snapshot === null) !== (b.candidate === null)) refuse(code);
      s.mode_revision = b.mode_revision;
      if (s.attachment) s.attachment.completed = true;
      if (b.snapshot) {
        validateCandidate(b.candidate);
        if (b.candidate.project !== s.project || b.candidate.mode_revision !== s.mode_revision || b.candidate.spec_commit !== s.spec_commit) refuse(code);
        s.candidate = structuredClone(b.candidate); 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 (['mode_revision','snapshot','candidate'].some(k => k in b)) refuse(code);
  } else if (e.kind === 'question.asked') {
    // The PM's one outstanding question. The id is part of the body so a reader never recomputes it.
    exact(b, ['op','proposal_digest','id','text','options'], [], code);
    if (!id(b.op) || !hash(b.proposal_digest) || !id(b.id) || !text(b.text, 2000) || !optionList(b.options)) refuse(code);
    if (!QUESTION_PHASES.includes(s.phase) || s.wait?.kind === 'decision' || (s.question && s.question.answer === null)) refuse(code);
    s.question = { id: b.id, text: b.text, options: structuredClone(b.options), answer: null };
    s.wait = { kind: 'question', code: 'question', subject: b.id, resume_phase: s.phase, observed_revision: s.revision };
  } else if (e.kind === 'question.answered') {
    // `answer` is the owner's human.input event. appendInTransaction checks that event; text never approves anything.
    exact(b, ['op','proposal_digest','question','answer'], [], code);
    if (!id(b.op) || !hash(b.proposal_digest) || !id(b.question) || !id(b.answer)) refuse(code);
    if (!s.question || s.question.answer !== null || s.question.id !== b.question) refuse(code);
    s.question.answer = b.answer;
    if (s.wait?.kind === 'question' && s.wait.subject === b.question) s.wait = null;
  } else if (e.kind === 'attach.selected') {
    // The owner's pick of an existing project. It is a pick only: no mapping, effect or decision exists until the
    // PM proposes exactly this project and the owner confirms the decision that proposal raises.
    exact(b, ['op','proposal_digest','project'], ['mapping_revision'], code);
    if (!id(b.op) || !hash(b.proposal_digest) || !key(b.project)) refuse(code);
    if (s.phase !== 'selection' || s.route !== 'existing' || s.wait?.kind !== 'setup' || !PICKER_CODES.includes(s.wait.code)) refuse(code);
    s.wait = { kind: 'setup', code: 'request-attachment-selected', subject: b.project, resume_phase: 'selection', observed_revision: s.revision };
    if (b.mapping_revision !== undefined) {
      if (!hash(b.mapping_revision)) refuse(code);
      s.attachment = { project: b.project, mapping_revision: b.mapping_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 === 'advanced') {
    exact(b, ['op','proposal_digest','from','phase','tasks','evidence'], [], code);
    const next = { planning: ['implementation'], implementation: ['review'], review: ['delivery','implementation'], delivery: ['complete'] };
    if (b.from !== s.phase || !next[s.phase]?.includes(b.phase) || !id(b.op) || !hash(b.proposal_digest) ||
        !Array.isArray(b.tasks) || !b.tasks.length || !b.tasks.every(x => /^vikunja:[1-9][0-9]*\/[1-9][0-9]*$/.test(x)) ||
        !Array.isArray(b.evidence) || !b.evidence.length || !b.evidence.every(id)) refuse(code);
    s.tasks = [...b.tasks]; s.phase = b.phase; s.evidence = [...b.evidence]; s.wait = null;
  } else if (e.kind === 'cancelled') {
    // `decision` is the open selection decision the cancellation closes, or null when none is open. A project or
    // mapping that already exists stays where it is; the journal never undoes an effect.
    exact(b, ['decision'], ['op','proposal_digest'], code);
    if (('op' in b) !== ('proposal_digest' in b) || ('op' in b && (!id(b.op) || !hash(b.proposal_digest)))) refuse(code);
    if (s.phase === 'delivered') refuse(code);
    if (s.wait?.kind === 'decision' ? s.wait.subject !== b.decision : b.decision !== null) refuse(code);
    s.phase = 'cancelled'; s.wait = null; s.selection = null;
  } else refuse(code);
  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');
  };
  const readRequest = (business, request) => {
    if (!key(business) || !id(request)) refuse('invalid-request');
    const events = store.all('SELECT * FROM request_events WHERE business=? AND request=? ORDER BY revision', business, request)
      .map(e => ({ ...e, body: JSON.parse(e.body) }));
    const state = events.reduce((s, e) => { validateIntent(e, events[0].body); return reduce(s, e); }, null);
    if (!state) return state;
    const launches = store.all("SELECT result FROM request_effects WHERE business=? AND request=? AND kind='continuation' AND stage='committed' AND json_extract(result,'$.type')='launch' ORDER BY seq", business, request).map(r => JSON.parse(r.result));
    const active = new Set(store.all("SELECT json_extract(result,'$.intent.key') key FROM request_effects WHERE business=? AND request=? AND kind='continuation' AND stage='uncertain' AND json_extract(result,'$.intent.revision')=?", business, request, state.revision).map(r => r.key));
    const latest = new Map();
    for (const launch of launches) if (active.has(launch.key)) latest.set(launch.instance, launch);
    const last = launches.findLast(launch => launch.code && latest.get(launch.instance) === launch);
    const delivery = store.get("SELECT result FROM request_effects WHERE business=? AND request=? AND kind='delivery' AND stage='committed' ORDER BY seq LIMIT 1", business, request);
    if (delivery && state.phase === 'complete') { const value = JSON.parse(delivery.result); state.delivery = value.id; state.delivery_evidence = value; }
    state.launch_refusal = last ? { instance: last.instance, code: last.code, key: last.key } : null;
    return state;
  };
  // appendInTransaction is only for composing the decision and journal atomically on the host.
  const appendInTransaction = e => {
    exact(e, ['business','request','expectedRevision','kind','body','actor']);
    exact(e.actor, ['role','run']);
    if (!key(e.business) || !id(e.request) || !Number.isSafeInteger(e.expectedRevision) || e.expectedRevision < 0 ||
        !((e.actor.role === null && e.actor.run === null) || (key(e.actor.role) && id(e.actor.run)))) refuse('invalid-request-event');
    const state = readRequest(e.business, e.request);
    if ((state?.revision ?? 0) !== e.expectedRevision) refuse('request-revision-conflict');
    const next = reduce(state, { ...e, revision: e.expectedRevision + 1 });
    if (e.kind === 'opened') {
      const source = store.get(`SELECT m.id,m.body,mt.thread,mt.session FROM events e JOIN messages m ON m.id=e.subject
        AND m.id=json_extract(e.body,'$.message') AND m.business=e.business
        LEFT JOIN message_threads mt ON mt.message=m.id
        WHERE e.business=? AND e.id=? 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'`, e.business, e.request);
      if (!source || source.id !== e.body.source_message || source.body !== e.body.intent?.text ||
          (source.thread ?? null) !== e.body.thread || (source.session ?? null) !== e.body.session) refuse('request-origin-refused');
    }
    if (e.kind === 'selection.proposed') {
      const origin = JSON.parse(store.get("SELECT body FROM request_events WHERE business=? AND request=? AND kind='opened'", e.business, e.request).body);
      validateIntent(e, origin);
      const d = store.get('SELECT * FROM decisions WHERE business=? AND id=?', e.business, e.body.decision);
      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 === 'advanced') {
      if (e.actor.role !== state.pm || e.body.evidence.some(eid => !store.get('SELECT 1 FROM events WHERE business=? AND id=?', e.business, eid))) 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 }));
    // An answer follows only from the owner's own console human.input for that question and op, so the journal
    // cannot record text the owner never sent or the origin instruction as an answer.
    if (e.kind === 'question.answered') {
      const a = store.get("SELECT body FROM events WHERE business=? AND id=? AND kind='human.input' AND actor_role IS NULL AND actor_run IS NULL", e.business, e.body.answer);
      const x = a && JSON.parse(a.body);
      if (!x || x.kind !== 'answer' || x.via !== 'console' || x.request !== e.request || x.question !== e.body.question || x.op !== e.body.op) refuse('invalid-request-event');
    }
    // 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') {
      const op = step => store.get("SELECT result FROM request_effects WHERE business=? AND request=? AND op=? AND kind='bootstrap' AND stage='committed'",
        e.business, e.request, `bootstrap:${e.body.digest}:${step}`);
      if (e.kind === 'bootstrap.completed') {
        const existing = state.selection?.bindings.find(b => b.digest === state.plan_digest)?.plan.existing;
        const done = op(existing ? 'attach' : 'activate');
        if (!done || JSON.parse(done.result).project !== e.body.project) refuse('invalid-request-event');
        if (existing) {
          const { project, ...outcome } = JSON.parse(done.result);
          if (!equal(outcome, Object.fromEntries(['spec_commit','mode_revision','snapshot','candidate'].map(k => [k, e.body[k]])))) refuse('invalid-request-event');
          if (e.body.snapshot) {
            const snap = store.get('SELECT body FROM request_snapshots WHERE business=? AND project=? AND id=?', e.business, project, e.body.snapshot);
            if (!snap || JSON.parse(snap.body).records_commit !== e.body.candidate.records_commit) refuse('invalid-request-event');
          }
        }
      } else if (op('mapping')) refuse('invalid-request-event');
    }
    store.run('INSERT INTO request_events(business,request,revision,kind,actor_role,actor_run,at,body) VALUES(?,?,?,?,?,?,?,?)',
      e.business, e.request, next.revision, e.kind, e.actor.role, e.actor.run, new Date().toISOString(), JSON.stringify(e.body));
    return next;
  };
  function receipt(r, prepare) {
    exact(r, ['version','business','request','op','kind','digest','stage','result','code'], [], 'invalid-request-effect');
    if (r.version !== 1 || !key(r.business) || !id(r.request) || !id(r.op) || !hash(r.digest) ||
        !['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 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) }));
      if (rows.some(x => x.digest !== r.digest || x.kind !== r.kind)) refuse('request-effect-conflict');
      // A retry must see committed/uncertain/refused evidence instead of executing again.
      if (prepare && rows.length) return structuredClone(rows.at(-1));
      const old = rows.find(x => x.stage === r.stage);
      if (old) { if (!equal(old, r)) refuse('request-effect-conflict'); return structuredClone(r); }
      if (!prepare && !rows.some(x => x.stage === 'prepared')) refuse('request-effect-unprepared');
      if (rows.some(x => ['committed','refused'].includes(x.stage))) refuse('request-effect-conflict');
      store.run('INSERT INTO request_effects(at,version,business,request,op,kind,digest,stage,result,code) VALUES(?,?,?,?,?,?,?,?,?,?)',
        new Date().toISOString(), r.version, r.business, r.request, r.op, r.kind, r.digest, r.stage, r.result === null ? null : JSON.stringify(r.result), r.code);
      return structuredClone(r);
    });
  }
  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 => 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');
      return store.all('SELECT version,business,request,op,kind,digest,stage,result,code FROM request_effects WHERE business=? AND request=? ORDER BY seq', business, request)
        .map(x => ({ ...x, result: x.result === null ? null : JSON.parse(x.result) }));
    },
    committed: (business, kind) => {
      if (!key(business)) refuse('invalid-request');
      return store.all("SELECT request,op,digest,result FROM request_effects WHERE business=? AND kind=? AND stage='committed' ORDER BY seq", business, kind)
        .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,
    intent: (business, request) => {
      const row = store.get("SELECT body FROM request_events WHERE business=? AND request=? AND kind='opened'", business, request);
      return row ? JSON.parse(row.body).intent : null;
    },
    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;
    } });
}