mosaicstack/stack
25
0 Fork 1
Code Issues 404 Pull Requests 13 Packages Projects Releases 15 Wiki Activity
Files
sp2
stack/packages/bus/src/request-store.mjs
T
jason.woltje a8d9eecf65 organization: move packages/business to packages/organization (rename r2 stage B1) 
git mv packages/business -> packages/organization (@mosaic/organization). The
old package stays as one-line re-export shims of each moved module, with
src/business.mjs re-exporting src/organization-config.mjs and src/cli.mjs
forwarding to the moved executable. BusinessError is OrganizationConfigError
(one constructor, name kept BusinessError); OrganizationError stays distinct.
The bus adapter gains organization-context.mjs (busOrganization), with
business.mjs (busBusiness) as the same function.

Import consumers, manifests, the lockfile, scripts/mosaic, stack-copy.sh
COPY_PATHS (both roots), seat sandbox RUNNER_FILES and the copy closure
walker (organization-imports.mjs, both roots required) move to the new path.
v1 serialization, v1 writers and the stage A fixtures are unchanged; the stage
A contract tests stay where they were and pass as before.
2026-11-09 11:47:48 -06:00

395 lines
32 KiB
JavaScript
Raw Permalink Blame History

// 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;
} });
}
Reference in new issue View Git Blame Copy Permalink