import { randomUUID, randomBytes } from 'node:crypto'; // Kept closed while S1 lands; its vocabulary is the integration authority. export const ACTIONS = Object.freeze([ 'task.create', 'task.assign', 'task.schedule', 'task.update.assigned', 'task.close', 'task.reassign', 'task.scope.change', 'task.priority.change', 'git.push.working', 'git.push.protected', 'git.merge.protected', 'review.request', 'review.verdict', 'message.send', 'message.external', 'role.launch', 'role.revoke', 'credential.mint', 'spend', 'deploy', 'policy.change', 'prd.approve', 'decision.resolve.technical', ]); const GATED = new Set([ 'credential.mint', 'git.merge.protected', 'git.push.protected', 'deploy', 'spend', 'message.external', 'policy.change', 'prd.approve', 'role.revoke', ]); const CLOSED = "('resolved','withdrawn','expired')"; export const TASK_VERBS = Object.freeze(ACTIONS.filter((a) => a.startsWith('task.'))); const SELF_KINDS = new Set(['task.created', 'task.assigned', 'task.state', 'task.closed', 'task.conflict']); const POLL_KINDS = new Set([ 'task.changed.external', 'task.missing', 'credential.expiring', 'credential.expired', 'credential.changed', ]); const canonical = (x) => typeof x === 'string' && /^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z$/.test(x) && Number.isFinite(Date.parse(x)) && new Date(x).toISOString() === x; export class BusError extends Error { constructor(code) { super(code); this.code = code; } } const fail = (code) => { throw new BusError(code); }; const object = (x) => x !== null && typeof x === 'object' && !Array.isArray(x); const text = (x, max = 4096) => typeof x === 'string' && x.length > 0 && x.length <= max && !x.includes('\0'); const id = (x) => text(x, 160) && /^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$/.test(x); function keys(x, allowed, required = []) { if (!object(x) || Object.keys(x).some((k) => !allowed.includes(k)) || required.some((k) => !(k in x))) fail('invalid-request'); } function string(x, max) { if (!text(x, max)) fail('invalid-request'); return x; } function identifier(x) { if (!id(x)) fail('invalid-request'); return x; } const taskRef = (x) => typeof x === 'string' && /^vikunja:[1-9][0-9]*\/[1-9][0-9]*$/.test(x); export class Broker { #store; #businesses; #sessions = new Map(); #runs = new Map(); #ended = new Set(); #secretCheck; #clock = 0; constructor({ store, businesses, secretCheck = () => {} }) { if (!object(businesses) || !Object.keys(businesses).length) fail('invalid-business'); this.#businesses = structuredClone(businesses); this.#store = store; this.#secretCheck = secretCheck; for (const table of [ 'events', 'role_claims', 'decisions', 'decision_events', 'messages', 'deliveries', 'task_snapshots', ]) { const at = store.get(`SELECT max(at) at FROM ${table}`).at; if (at) { const ms = Date.parse(at); if (!Number.isFinite(ms)) fail('invalid-timestamp'); this.#clock = Math.max(this.#clock, ms); } } for (const [name, b] of Object.entries(this.#businesses)) { if (!id(name) || b.id !== name || !id(b.human) || !object(b.roles) || !object(b.arbiters)) fail('invalid-business'); for (const role of Object.keys(b.roles)) { if (!id(role) || role === 'human') fail('invalid-business'); const a = b.roles[role].authority; if (!object(a) || !Array.isArray(a.withinRole) || !Array.isArray(a.crossRole)) fail('invalid-authority'); const all = [...a.withinRole, ...a.crossRole]; if (new Set(all).size !== all.length || all.some((v) => !ACTIONS.includes(v) || GATED.has(v))) fail('invalid-authority'); } if (!Object.hasOwn(b.roles, b.arbiters.technical) || !Object.hasOwn(b.roles, b.arbiters.delivery)) fail('invalid-arbiter'); } } #now() { this.#clock = Math.max(Date.now(), this.#clock + 1); return new Date(this.#clock).toISOString(); } #business(name) { if (!Object.hasOwn(this.#businesses, name)) fail('unknown-business'); return this.#businesses[name]; } #session(cap) { return this.#sessions.get(cap) ?? fail('unauthenticated'); } #holder(business, role) { return this.#store.get( 'SELECT * FROM role_claims WHERE business=? AND role=? ORDER BY seq DESC LIMIT 1', business, role, ); } #agent(s) { if (s.reader) fail('read-only'); if (s.human) fail('agent-required'); const h = this.#holder(s.business, s.role); if (h?.op !== 'claim' || h.holder_run !== s.run) fail('not-holder'); return h; } #human(s) { if (s.reader) fail('read-only'); if (!s.human || s.via !== 'cli' || s.outsideAgent !== true) fail('human-required'); } #cap(s) { const cap = randomBytes(32).toString('hex'); this.#sessions.set(cap, Object.freeze(s)); return cap; } // Trusted launcher API, deliberately NOT a socket verb. S6 supplies immutable launch records. bindLaunch(record) { keys( record, ['business', 'role', 'run', 'harness', 'address', 'pid', 'startTime'], ['business', 'role', 'run', 'harness'], ); const b = this.#business(identifier(record.business)); if (!Object.hasOwn(b.roles, record.role)) fail('unknown-role'); identifier(record.run); if (!['pi', 'claude-code'].includes(record.harness)) fail('invalid-harness'); if (record.address !== undefined) string(record.address, 1024); const key = record.business + ':' + record.run; if (this.#runs.has(key)) fail('duplicate-run'); const session = { ...record, address: record.address ?? null, human: false }; this.#secretCheck(record); const old = this.#store.get( "SELECT actor_role,body FROM events WHERE business=? AND kind='session.launched' AND subject=? ORDER BY seq LIMIT 1", record.business, record.run, ); const body = { harness: record.harness, address: record.address ?? null, pid: record.pid ?? null, startTime: record.startTime ?? null, }; if (old) { if (old.actor_role !== record.role || old.body !== JSON.stringify(body)) fail('launch-record-mismatch'); // An ended run stays ended across broker restarts. if (this.#store.get("SELECT 1 FROM events WHERE business=? AND kind='session.ended' AND subject=?", record.business, record.run)) fail('run-ended'); } else this.#store.transaction(() => this.#event(session, 'session.launched', body, record.run)); this.#runs.set(key, session); return this.#cap(session); } // Trusted launcher API, the counterpart of bindLaunch and likewise NOT a socket verb. The S6 host // calls it once when a launched run's process exits: session.ended is written, the run's claim is // released if it still holds one, and every capability of the run stops working. endLaunch(record) { keys(record, ['business', 'run', 'reason', 'exitCode'], ['business', 'run', 'reason']); this.#business(identifier(record.business)); const key = record.business + ':' + identifier(record.run); const session = this.#runs.get(key) ?? fail('unknown-run'); if (this.#ended.has(key)) fail('run-ended'); identifier(record.reason); const exitCode = record.exitCode ?? null; if (exitCode !== null && !Number.isSafeInteger(exitCode)) fail('invalid-request'); this.#ended.add(key); for (const [cap, s] of this.#sessions) if (!s.human && !s.reader && s.business === record.business && s.run === record.run) this.#sessions.delete(cap); return this.#store.transaction(() => { const h = this.#holder(session.business, session.role); const released = h?.op === 'claim' && h.holder_run === session.run; if (released) this.#claimEnd(session, h, 'release', null); this.#event(session, 'session.ended', { reason: record.reason, exitCode, released }, session.run); return { released }; }); } // Trusted launcher API: records a refusal the host decided itself (instance list, capacity, // credentials) as action.refused against the caller, the way request() records its own. refuse(cap, code) { const s = this.#session(cap); if (!/^[a-z-]{1,64}$/.test(code)) fail('invalid-request'); return this.#refused(s, new BusError(code), null).code; } // Trusted human transport invokes only AFTER checking its CLI process and launch ancestry. bindHuman(record) { keys(record, ['business', 'human', 'via', 'outsideAgent'], ['business', 'human', 'via', 'outsideAgent']); const b = this.#business(record.business); if (record.human !== b.human || record.via !== 'cli' || record.outsideAgent !== true) fail('human-required'); return this.#cap({ ...record, role: null, run: null, human: b.human }); } bindReader({ business }) { this.#business(business); return this.#cap({ business, reader: true, role: null, run: null, human: false }); } disconnect(cap) { this.#sessions.delete(cap); } identity(cap) { const s = this.#session(cap); return { business: s.business, role: s.role, run: s.run, human: s.human || null }; } #event(s, kind, body, subject = null) { this.#secretCheck({ kind, body, subject }); if ( (kind.startsWith('action.') || kind.startsWith('review.')) && taskRef(body.task_ref ?? body.target) && subject !== (body.task_ref ?? body.target) ) fail('task-subject-required'); const eid = randomUUID(); this.#store.run( 'INSERT INTO events(id,at,business,kind,actor_role,actor_run,subject,body) VALUES(?,?,?,?,?,?,?,?)', eid, this.#now(), s.business, kind, s.role, s.run, subject, JSON.stringify(body), ); return eid; } #classification(s, action) { if (!ACTIONS.includes(action)) fail('unknown-action'); if (GATED.has(action) || (action === 'role.launch' && this.#business(s.business).launch?.by !== s.role)) return 'gated'; const a = this.#business(s.business).roles[s.role]?.authority; return a?.withinRole.includes(action) ? 'within-role' : a?.crossRole.includes(action) ? 'cross-role' : 'gated'; } #decision(s, id) { const d = this.#store.get('SELECT * FROM decisions WHERE id=? AND business=?', id, s.business); if (!d) fail('decision-not-found'); return d; } #decisionView(d) { const row = this.#store.get( "SELECT body FROM events WHERE business=? AND kind='action.allowed' AND json_extract(body,'$.operation')='decision.raise' AND json_extract(body,'$.decision')=? ORDER BY seq LIMIT 1", d.business, d.id, ); const context = row ? JSON.parse(row.body) : null; return { ...d, options: JSON.parse(d.options), blocking: !!d.blocking, authorization: context ? { action: d.action, target: context.target, approvalChoice: context.approvalChoice } : null, }; } #closed(id) { return this.#store.get( `SELECT * FROM decision_events WHERE decision=? AND op IN ${CLOSED} ORDER BY seq DESC LIMIT 1`, id, ); } #checkAuthority(s, action, { decision = null, target = null } = {}) { this.#agent(s); if ( action === 'role.launch' && this.#store.get('SELECT state FROM launch_state WHERE business=?', s.business)?.state === 'revoked' ) fail('launch-revoked'); const cls = this.#classification(s, action); if (cls === 'within-role' && !decision) return { class: cls }; if (!decision) fail('decision-required'); const d = this.#decision(s, decision), r = this.#closed(d.id); const evidence = this.#store.get( "SELECT body FROM events WHERE business=? AND json_extract(body,'$.decision')=? AND json_extract(body,'$.operation')='decision.raise' AND kind='action.allowed' ORDER BY seq LIMIT 1", s.business, d.id, ); const context = evidence ? JSON.parse(evidence.body) : null; if ( d.class !== cls || d.action !== action || d.raised_by_role !== s.role || d.raised_by_run !== s.run || !context || context.target !== target ) fail('decision-mismatch'); if (r?.op !== 'resolved' || r.choice !== context.approvalChoice) fail('decision-not-approved'); if (action === 'role.revoke' && context.holderRun !== this.#holder(s.business, target)?.holder_run) fail('decision-mismatch'); return { class: cls, decision: d.id }; } // Caller owns the transaction: authority, consumption and effect commit together. #consumeAuthority(s, action, context = {}) { const result = this.#checkAuthority(s, action, context); // decision.raise records context, not permission consumed by an executor. if ( result.decision && this.#store.get( "SELECT 1 FROM events WHERE business=? AND kind='action.allowed' AND json_extract(body,'$.decision')=? AND json_extract(body,'$.operation') IS NOT 'decision.raise' LIMIT 1", s.business, result.decision, ) ) fail('decision-consumed'); this.#event( s, 'action.allowed', { action, ...result, target: context.target ?? null }, context.target ?? null, ); return result; } // Trusted S3 handler calls inside its own operation; not a generic socket action executor. authorize(cap, action, context = {}) { const s = this.#session(cap); return this.#store.transaction(() => this.#consumeAuthority(s, action, context)); } // Trusted adapters only. No agent socket route reaches this method. recordEvent(cap, { kind, body, subject = null }) { const s = this.#session(cap); return this.#store.transaction(() => { this.#agent(s); if (['human.input', 'launch.revoked', 'launch.restored'].includes(kind)) fail('reserved-event'); return this.#event(s, kind, body, subject); }); } // Trusted S3 adapter only. With a cap it records a role's own completed write: the external call // already happened, so this records and does not re-authorize. Without one it records sync reads. recordTask({ cap = null, business, snapshots = [], events = [] }) { const s = cap === null ? { business, role: null, run: null } : this.#session(cap); this.#business(business); if (cap !== null && (s.human || s.reader || s.business !== business)) fail('agent-required'); if (!Array.isArray(snapshots) || !Array.isArray(events)) fail('invalid-request'); const kinds = cap === null ? POLL_KINDS : SELF_KINDS; this.#secretCheck({ snapshots, events }); return this.#store.transaction(() => ({ snapshots: snapshots.map((x) => this.#snapshot(s, x)), events: events.map((e) => { keys(e, ['kind', 'body', 'subject'], ['kind', 'body']); if (!kinds.has(e.kind)) fail('reserved-event'); if (!object(e.body)) fail('invalid-request'); return this.#event(s, e.kind, e.body, e.subject ?? null); }), })); } #snapshot(s, x) { const poll = s.role === null; keys( x, ['task_ref', 'updated', 'etag', 'digest', 'fields', ...(poll ? ['via', 'read_at'] : [])], ['task_ref', 'updated', 'digest', 'fields', ...(poll ? ['via', 'read_at'] : [])], ); if (!taskRef(x.task_ref) || !object(x.fields) || !/^[0-9a-f]{64}$/.test(x.digest ?? '')) fail('invalid-snapshot'); string(x.updated, 64); if (x.etag !== undefined && x.etag !== null) string(x.etag, 256); if (poll && (!['board', 'cursor', 'task', 'reconcile'].includes(x.via) || !canonical(x.read_at))) fail('invalid-snapshot'); return this.#store.run( 'INSERT INTO task_snapshots(at,business,task_ref,updated,etag,digest,fields,source,via,read_at,role,run) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)', this.#now(), s.business, x.task_ref, x.updated, x.etag ?? null, x.digest, JSON.stringify(x.fields), poll ? 'poll' : 'self', poll ? x.via : null, poll ? x.read_at : null, s.role, s.run, ).lastInsertRowid; } // Trusted S3 reads. Agents read the same rows through the 'tasks' view verb. taskView(business, view, arg = null) { this.#business(business); const parse = (r) => ({ ...r, fields: JSON.parse(r.fields) }); if (view === 'current' && arg === null) return this.#store.all('SELECT * FROM task_current WHERE business=? ORDER BY task_ref', business).map(parse); if (view === 'current') { if (!taskRef(arg)) fail('invalid-request'); const r = this.#store.get('SELECT * FROM task_current WHERE business=? AND task_ref=?', business, arg); return r ? parse(r) : null; } if (view === 'open') return this.#store.all('SELECT task_ref,bucket FROM tasks_open WHERE business=? ORDER BY task_ref', business); if (view === 'input') return Boolean( typeof arg === 'string' && this.#store.get("SELECT 1 FROM events WHERE business=? AND kind='human.input' AND id=?", business, arg), ); fail('invalid-request'); } #refused(s, e, request, other = 'storage-refused') { const error = e instanceof BusError ? e : new BusError(other); // No request content or raw exception text in refusal evidence. try { this.#store.transaction(() => this.#event( s, 'action.refused', { code: error.code }, taskRef(request?.args?.task_ref) ? request.args.task_ref : taskRef(request?.args?.target) ? request.args.target : null, ), ); } catch { return new BusError('storage-unavailable'); } return error; } request(cap, request) { const s = this.#session(cap); try { keys(request, ['verb', 'args'], ['verb']); string(request.verb, 64); const args = request.args ?? {}; if (!object(args)) fail('invalid-request'); this.#secretCheck(args); return this.#store.transaction(() => { if (s.reader && !['inbox', 'agents', 'tasks', 'trail'].includes(request.verb)) fail('read-only'); const result = this.#dispatch(s, request.verb, args); this.#secretCheck(result); return result; }); } catch (e) { throw this.#refused(s, e, request); } } // The eight task verbs reach the trusted S3 handler here. It runs outside any transaction, gets the // cap to authorize and record through authorize() and recordTask(); refusals are recorded as in request(). async requestTask(cap, request, handler) { const s = this.#session(cap); try { keys(request, ['verb', 'args'], ['verb']); if (!TASK_VERBS.includes(request.verb)) fail('unknown-verb'); const args = request.args ?? {}; if (!object(args)) fail('invalid-request'); this.#secretCheck(args); this.#store.transaction(() => this.#agent(s)); const result = await handler(cap, request.verb, args); this.#secretCheck(result); return result; } catch (e) { throw this.#refused(s, e, request, 'adapter-failed'); } } #dispatch(s, verb, a) { const b = this.#business(s.business), by = s.human || s.run; switch (verb) { case 'role.claim': { keys(a, []); if (s.human) fail('agent-required'); if ( this.#store.get( "SELECT 1 FROM role_claims WHERE business=? AND role=? AND holder_run=? AND op='revoke' LIMIT 1", s.business, s.role, s.run, ) ) fail('run-revoked'); if (this.#holder(s.business, s.role)?.op === 'claim') fail('role-already-held'); this.#store.run( 'INSERT INTO role_claims(at,business,role,op,holder_run,harness,address,by) VALUES(?,?,?,?,?,?,?,?)', this.#now(), s.business, s.role, 'claim', s.run, s.harness, s.address, by, ); this.#routeWaiting(s.business, s.role); return { role: s.role, run: s.run }; } case 'role.release': { keys(a, []); const h = this.#agent(s); this.#claimEnd(s, h, 'release', null); return { released: true }; } case 'role.revoke': { keys(a, ['role', 'decision'], ['role', 'decision']); identifier(a.role); identifier(a.decision); this.#consumeAuthority(s, 'role.revoke', { decision: a.decision, target: a.role }); const h = this.#holder(s.business, a.role); if (h?.op !== 'claim') fail('not-held'); this.#claimEnd(s, h, 'revoke', a.decision); return { revoked: true }; } case 'decision.raise': { keys( a, [ 'action', 'domain', 'target', 'project', 'question', 'options', 'recommendation', 'blocking', 'task_ref', 'requirement_ref', 'supersedes', 'choice', 'approvalChoice', ], ['action', 'question', 'options', 'recommendation', 'blocking'], ); this.#agent(s); const cls = this.#classification(s, a.action); string(a.question); if (typeof a.blocking !== 'boolean') fail('invalid-request'); if (a.blocking && !a.task_ref) fail('task-required'); if (a.task_ref !== undefined && !/^vikunja:[1-9][0-9]*\/[1-9][0-9]*$/.test(a.task_ref)) fail('invalid-task-ref'); for (const k of ['target', 'project', 'requirement_ref', 'supersedes']) if (a[k] !== undefined) identifier(a[k]); if (!Array.isArray(a.options) || a.options.length < 2 || a.options.length > 9) fail('invalid-options'); for (const o of a.options) { keys(o, ['key', 'text'], ['key', 'text']); if (!id(o.key) || !text(o.text, 1024)) fail('invalid-options'); } const choices = a.options.map((o) => o.key); if (new Set(choices).size !== choices.length || !choices.includes(a.recommendation)) fail('invalid-options'); const approvalChoice = a.approvalChoice ?? 'yes'; if (!choices.includes(approvalChoice)) fail('invalid-approval-choice'); const domain = a.domain ?? 'delivery'; if (!['technical', 'delivery'].includes(domain)) fail('invalid-domain'); const proposedRoute = cls === 'gated' ? 'human' : cls === 'cross-role' ? b.arbiters[domain] : s.role; const route = cls === 'cross-role' && proposedRoute === s.role ? 'human' : proposedRoute; if (a.supersedes) { const old = this.#decision(s, a.supersedes); if (old.raised_by_run !== s.run || this.#closed(old.id)) fail('supersede-refused'); this.#store.run( 'INSERT INTO decision_events(decision,at,op,by) VALUES(?,?,?,?)', old.id, this.#now(), 'withdrawn', by, ); } const did = randomUUID(); this.#store.run( 'INSERT INTO decisions(id,at,business,project,raised_by_role,raised_by_run,class,action,route_to,question,options,recommendation,task_ref,requirement_ref,blocking,supersedes) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', did, this.#now(), s.business, a.project ?? null, s.role, s.run, cls, a.action, route, a.question, JSON.stringify(a.options), a.recommendation, a.task_ref ?? null, a.requirement_ref ?? null, Number(a.blocking), a.supersedes ?? null, ); this.#event( s, 'action.allowed', { decision: did, action: a.action, operation: 'decision.raise', task_ref: a.task_ref ?? null, target: a.target ?? null, approvalChoice, holderRun: a.action === 'role.revoke' ? (this.#holder(s.business, a.target)?.holder_run ?? null) : null, }, a.task_ref ?? (taskRef(a.target) ? a.target : did), ); if (cls === 'within-role') { if (!choices.includes(a.choice)) fail('invalid-choice'); this.#store.run( 'INSERT INTO decision_events(decision,at,op,by,choice,via) VALUES(?,?,?,?,?,?)', did, this.#now(), 'resolved', by, a.choice, 'broker', ); } return this.#decisionView(this.#decision(s, did)); } case 'decision.resolve': case 'decision.seen': case 'decision.withdraw': { keys(a, ['id', 'choice', 'note'], ['id']); identifier(a.id); if (a.note !== undefined) string(a.note); const d = this.#decision(s, a.id); if (this.#closed(d.id)) fail('decision-closed'); if (verb === 'decision.withdraw') { if (s.human) { this.#human(s); } else { this.#agent(s); if (d.raised_by_run !== s.run) fail('not-resolver'); } } else if (d.route_to === 'human') this.#human(s); else { this.#agent(s); if (s.role !== d.route_to) fail('not-resolver'); } const op = verb === 'decision.resolve' ? 'resolved' : verb === 'decision.seen' ? 'seen' : 'withdrawn'; if (op === 'resolved' && !JSON.parse(d.options).some((o) => o.key === a.choice)) fail('invalid-choice'); this.#store.run( 'INSERT INTO decision_events(decision,at,op,by,choice,note,via) VALUES(?,?,?,?,?,?,?)', d.id, this.#now(), op, by, op === 'resolved' ? a.choice : null, a.note ?? null, s.human ? 'cli' : 'broker', ); if (s.human) this.#event( s, 'human.input', { kind: 'answer', decision: d.id, op, choice: a.choice ?? null }, d.id, ); return { id: d.id, op }; } case 'message.send': { keys(a, ['to', 'body', 'class', 'in_reply_to', 'corrects', 'decision'], ['to', 'body']); if (!s.human) { const withinRole = this.#classification(s, 'message.send') === 'within-role'; this.#consumeAuthority(s, 'message.send', { decision: withinRole ? null : (a.decision ?? null), target: a.to, }); } else this.#human(s); if (a.to !== 'human' && !Object.hasOwn(b.roles, a.to)) fail('unknown-role'); string(a.body, 32768); if ( a.class !== undefined && ![ 'REQUEST', 'ASSIGNMENT', 'REVIEW-REQUEST', 'REVIEW-RESULT', 'RESULT', 'INFO', 'DECISION', 'REACTION', ].includes(a.class) ) fail('invalid-class'); for (const k of ['in_reply_to', 'corrects']) { if (a[k] !== undefined) { identifier(a[k]); if (!this.#store.get('SELECT id FROM messages WHERE id=? AND business=?', a[k], s.business)) fail('message-not-found'); } } if (a.decision) this.#decision(s, a.decision); const mid = randomUUID(); this.#store.run( 'INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,in_reply_to,decision,corrects,body) VALUES(?,?,?,?,?,?,?,?,?,?,?)', mid, this.#now(), s.business, s.human ? 'human' : s.role, s.human ? 'human-cli' : s.run, a.to, a.class ?? 'INFO', a.in_reply_to ?? null, a.decision ?? null, a.corrects ?? null, a.body, ); const h = this.#holder(s.business, a.to); this.#delivery(mid, 'routed', h?.op === 'claim' ? h : null); const request = s.human ? this.#event(s, 'human.input', { kind: 'instruction', message: mid }, mid) : null; return { id: mid, request }; } case 'message.receive': { keys(a, []); if (s.human) this.#human(s); else this.#agent(s); const role = s.human ? 'human' : s.role; const rows = this.#store.all( "SELECT * FROM messages m WHERE business=? AND to_role=? AND NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op IN ('delivered','read')) ORDER BY seq LIMIT 100", s.business, role, ); for (const m of rows) this.#delivery( m.id, 'delivered', s.human ? null : this.#holder(s.business, s.role), s.human ? 'cli' : 'broker', ); return rows.map((m) => ({ ...m, request: this.#store.get( "SELECT id FROM events WHERE business=? AND kind='human.input' AND subject=? ORDER BY seq LIMIT 1", s.business, m.id, )?.id ?? null, })); } case 'message.read': { keys(a, ['id'], ['id']); if (s.human) this.#human(s); else this.#agent(s); const m = this.#store.get('SELECT * FROM messages WHERE business=? AND id=?', s.business, a.id); if (!m || m.to_role !== (s.human ? 'human' : s.role)) fail('message-not-found'); const delivery = this.#store.get( "SELECT * FROM deliveries WHERE message=? AND op='delivered' ORDER BY seq DESC LIMIT 1", m.id, ); if (!delivery || delivery.holder_run !== (s.run ?? null)) fail('not-recipient'); if (!this.#store.get("SELECT seq FROM deliveries WHERE message=? AND op='read'", m.id)) this.#delivery(m.id, 'read', s.human ? null : this.#holder(s.business, s.role)); return { read: true }; } case 'launch.revoke': case 'launch.restore': { keys(a, []); this.#human(s); const kind = verb === 'launch.revoke' ? 'launch.revoked' : 'launch.restored'; const eid = this.#event(s, kind, { via: 'cli' }); this.#event(s, 'human.input', { kind: 'admin', event: eid }); return { id: eid }; } case 'event.emit': { keys(a, ['kind', 'body', 'subject'], ['kind', 'body']); this.#agent(s); // Authoritative lifecycle/task/credential events are emitted by trusted handlers, not arbitrary clients. if (a.kind !== 'action.allowed') fail('reserved-event'); keys(a.body, ['action', 'target'], ['action']); if (a.body.action !== 'routine') fail('reserved-event'); return { id: this.#event( s, 'action.allowed', { action: 'routine', target: a.body.target ?? null }, a.subject ?? null, ), }; } case 'inbox': case 'agents': case 'tasks': case 'trail': { keys(a, verb === 'trail' ? ['subject'] : [], verb === 'trail' ? ['subject'] : []); if (!s.human && !s.reader) this.#agent(s); if (verb === 'inbox') return this.#store .all( `SELECT d.* FROM decisions d WHERE business=? AND route_to=? AND NOT EXISTS(SELECT 1 FROM decision_events x WHERE x.decision=d.id AND x.op IN ${CLOSED}) ORDER BY CASE class WHEN 'gated' THEN 0 ELSE 1 END,seq`, s.business, s.human || s.reader ? 'human' : s.role, ) .map((d) => this.#decisionView(d)); if (verb === 'agents') return this.#store.all( "SELECT * FROM role_claims c WHERE business=? AND op='claim' AND seq=(SELECT max(seq) FROM role_claims WHERE business=c.business AND role=c.role) ORDER BY role", s.business, ); if (verb === 'tasks') return this.#store .all('SELECT * FROM task_current WHERE business=? ORDER BY task_ref', s.business) .map((r) => ({ ...r, fields: JSON.parse(r.fields) })); return this.#trail(s, identifier(a.subject)); } default: fail('unknown-verb'); } } #claimEnd(s, h, op, decision) { this.#store.run( 'INSERT INTO role_claims(at,business,role,op,holder_run,harness,address,by,decision) VALUES(?,?,?,?,?,?,?,?,?)', this.#now(), s.business, h.role, op, h.holder_run, h.harness, h.address, s.run ?? s.human, decision, ); } #delivery(mid, op, h, transport = null) { this.#store.run( 'INSERT INTO deliveries(message,at,op,holder_run,transport,address) VALUES(?,?,?,?,?,?)', mid, this.#now(), op, h?.holder_run ?? null, transport ?? h?.harness ?? null, h?.address ?? null, ); } #routeWaiting(business, role) { const h = this.#holder(business, role); const rows = this.#store.all( "SELECT id FROM messages m WHERE business=? AND to_role=? AND NOT EXISTS(SELECT 1 FROM deliveries WHERE message=m.id AND op IN ('delivered','read'))", business, role, ); for (const m of rows) this.#delivery(m.id, 'routed', h); } #trail(s, subject) { const rows = new Map(); const add = (table, list) => { for (const row of list) { const r = { ...row, table }; for (const k of ['body', 'options', 'fields']) if (k in r && table !== 'messages') try { r[k] = JSON.parse(r[k]); } catch {} rows.set(table + ':' + row.seq, r); } }; add( 'events', this.#store.all('SELECT * FROM events WHERE business=? AND subject=?', s.business, subject), ); add('messages', this.#store.all('SELECT * FROM messages WHERE business=? AND id=?', s.business, subject)); const decisions = this.#store.all( 'SELECT * FROM decisions WHERE business=? AND (id=? OR task_ref=?)', s.business, subject, subject, ); add('decisions', decisions); for (const d of decisions) { add('decision_events', this.#store.all('SELECT * FROM decision_events WHERE decision=?', d.id)); add( 'events', this.#store.all( "SELECT * FROM events WHERE business=? AND (subject=? OR json_extract(body,'$.decision')=?)", s.business, d.id, d.id, ), ); add( 'messages', this.#store.all('SELECT * FROM messages WHERE business=? AND decision=?', s.business, d.id), ); } for (const e of [...rows.values()]) if (e.table === 'events' && e.kind === 'task.created') { const input = this.#store.get( "SELECT * FROM events WHERE business=? AND kind='human.input' AND id=?", s.business, e.body.request, ); if (input) { add('events', [input]); const body = JSON.parse(input.body); if (body.message) add( 'messages', this.#store.all('SELECT * FROM messages WHERE business=? AND id=?', s.business, body.message), ); } } for (const m of [...rows.values()]) if (m.table === 'messages') add('deliveries', this.#store.all('SELECT * FROM deliveries WHERE message=?', m.id)); add( 'task_snapshots', this.#store.all('SELECT * FROM task_snapshots WHERE business=? AND task_ref=?', s.business, subject), ); const runs = new Set( [...rows.values()].map((r) => r.actor_run ?? r.raised_by_run ?? r.run).filter(Boolean), ); for (const run of runs) { add( 'events', this.#store.all( "SELECT * FROM events WHERE business=? AND actor_run=? AND kind IN ('session.launched','session.ended')", s.business, run, ), ); add( 'role_claims', this.#store.all('SELECT * FROM role_claims WHERE business=? AND holder_run=?', s.business, run), ); } return [...rows.values()].sort( (a, b) => a.at.localeCompare(b.at) || a.table.localeCompare(b.table) || a.seq - b.seq, ); } }