import { randomUUID, randomBytes } from 'node:crypto'; import { loadRequirements } from '../../tasks/src/requirements.mjs'; // 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')"; // How many closed decisions `decisions` with list 'recent' returns, newest close first. export const RECENT = 50; // How many replies `replies` returns, oldest first. export const REPLIES = 50; // Row 80 (#1574): how many stranded messages `agents` lists for each role, oldest first. export const STRANDED = 50; // Row 89 (#1583): the arguments `tasks` takes, and its page size by default and at most. const TASK_ARGS = ['done', 'assignee', 'label', 'requirement', 'limit', 'after', 'task_ref']; export const TASK_PAGE = 50; export const TASK_PAGE_MAX = 200; 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); // A git commit id, SHA-1 or SHA-256. const oid = (x) => typeof x === 'string' && /^[0-9a-f]{40}([0-9a-f]{24})?$/.test(x); const VERDICTS = ['approve', 'changes']; // Row 93 (#1587): the review loop's evidence, written only by the verbs below. A push is an // action.allowed for git.push.working that names its commit, written only by recordPush. const REVIEW_KINDS = ['review.requested', 'review.verdict']; export class Broker { #store; #businesses; #sessions = new Map(); #runs = new Map(); #ended = new Set(); #secretCheck; #credentialStatus; #requirements; #clock = 0; // credentialStatus is the runtime's metadata-only Credentials.status(); the broker never gets use(). constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [] }) { if (!object(businesses) || !Object.keys(businesses).length) fail('invalid-business'); this.#businesses = structuredClone(businesses); this.#store = store; this.#secretCheck = secretCheck; this.#credentialStatus = credentialStatus; // Row 90 (#1584): the approved PRD's requirement ids. A bad requirements.json refuses here. this.#requirements = loadRequirements(); 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. // A task ref as subject files the refusal on that task's trail, as request() does. refuse(cap, code, subject = null) { const s = this.#session(cap); if (!/^[a-z-]{1,64}$/.test(code)) fail('invalid-request'); return this.#refused(s, new BusError(code), { args: { task_ref: subject } }).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, ); } // Row 58 (#1547): the 'tasks' rows, each with the task's priority and requirement. Priority // is the current row's; a tombstone has none. The requirement is the first task.created for // the same Vikunja task id, which a move to another project keeps; null for a task the stack // didn't create. Both come from append-only rows at each read; nothing is stored. #taskRows(business) { const taskId = (ref) => ref.slice(ref.indexOf('/') + 1); const requirement = new Map(); for (const e of this.#store.all( "SELECT subject,json_extract(body,'$.requirement') AS requirement FROM events WHERE business=? AND kind='task.created' ORDER BY seq", business, )) if (!requirement.has(taskId(e.subject))) requirement.set(taskId(e.subject), e.requirement); return this.#store .all('SELECT * FROM task_current WHERE business=? ORDER BY task_ref', business) .map((r) => { const fields = JSON.parse(r.fields); const priority = fields.priority ?? null; return { ...r, fields, priority, requirement: requirement.get(taskId(r.task_ref)) ?? null }; }); } // Row 89 (#1583): with no arguments, every row as #taskRows gives it, which the Console and // `mosaic tasks` read. `task_ref`, alone, gives that one row. Any other argument filters and pages // the short form in task_ref order: no description, and the assignees that are a role's tracker // bot, by role. A tombstone isn't on the board, so a filtered read leaves it out. `total` counts // every task the filters match; `next` is the `after` for the next page, null on the last. #tasks(business, a) { const rows = this.#taskRows(business); if (!Object.keys(a).length) return rows; if ('task_ref' in a) { if (Object.keys(a).length !== 1 || !taskRef(a.task_ref)) fail('invalid-request'); return rows.find((r) => r.task_ref === a.task_ref) ?? fail('task-not-found'); } const roles = this.#business(business).roles ?? {}; const positive = (x, max = Number.MAX_SAFE_INTEGER) => Number.isSafeInteger(x) && x >= 1 && x <= max; if ( ('done' in a && typeof a.done !== 'boolean') || ('assignee' in a && !(typeof a.assignee === 'string' && Object.hasOwn(roles, a.assignee))) || ('label' in a && !positive(a.label)) || ('requirement' in a && !(typeof a.requirement === 'string' && /^REQ-[A-Z]+-[1-9][0-9]*$/.test(a.requirement))) || ('limit' in a && !positive(a.limit, TASK_PAGE_MAX)) || ('after' in a && !taskRef(a.after)) ) fail('invalid-request'); const bots = new Map( Object.entries(roles) .filter(([, r]) => r.tracker?.botId !== undefined) .map(([role, r]) => [r.tracker.botId, role]), ); const matched = rows .filter((r) => !r.fields.gone) .map(({ task_ref, fields: f, priority, requirement, at }) => ({ task_ref, title: f.title ?? null, done: Boolean(f.done), bucket: f.bucket ?? null, assignees: (f.assignees ?? []).filter((u) => bots.has(u)).map((u) => bots.get(u)).sort(), labels: f.labels ?? [], priority, requirement, changed_at: at, })) .filter( (t) => (!('done' in a) || t.done === a.done) && (!('assignee' in a) || t.assignees.includes(a.assignee)) && (!('label' in a) || t.labels.includes(a.label)) && (!('requirement' in a) || t.requirement === a.requirement), ); const rest = 'after' in a ? matched.filter((t) => t.task_ref > a.after) : matched; const limit = a.limit ?? TASK_PAGE; const tasks = rest.slice(0, limit); return { tasks, total: matched.length, next: rest.length > limit ? tasks.at(-1).task_ref : null }; } #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. // detail adds to the action.allowed body: a push's commit (row 93). #consumeAuthority(s, action, context = {}, detail = {}) { 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', { ...detail, 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. // Row 93 (#1587): task.close on a task with a recorded push also needs // context.verdict, checked by #closeVerdict; a refusal there rolls the action.allowed back. authorize(cap, action, context = {}) { const s = this.#session(cap); return this.#store.transaction(() => { const result = this.#consumeAuthority(s, action, context); if (action === 'task.close') this.#closeVerdict(s, context.target, context.verdict); return result; }); } // A task with a pushed commit closes only on an approve verdict for the latest push. A task with // no push closes as before, its verdict a citation the broker doesn't check. #closeVerdict(s, task, verdict) { const push = this.#latestPush(s.business, task); if (!push) return; const v = id(verdict) ? this.#store.get( "SELECT body FROM events WHERE business=? AND kind='review.verdict' AND id=? AND subject=?", s.business, verdict, task, ) : null; if (!v) fail('verdict-not-found'); const body = JSON.parse(v.body); if (body.verdict !== 'approve') fail('verdict-not-approve'); if (body.commit !== push.commit) fail('verdict-stale'); } // A push is the action.allowed recordPush wrote, the only one for git.push.working with a commit. #latestPush(business, task) { const r = this.#store.get( "SELECT body FROM events WHERE business=? AND kind='action.allowed' AND json_extract(body,'$.action')='git.push.working' AND json_type(body,'$.commit')='text' AND subject=? ORDER BY seq DESC LIMIT 1", business, task, ); return r ? JSON.parse(r.body) : null; } // Row 93: the bus host's git socket asks before it takes a push's bytes. Nothing is recorded // unless it refuses: the push is recorded, and its authority consumed, once the ref moved. checkPush(cap, task_ref) { const s = this.#session(cap); return this.#store.transaction(() => { this.#agent(s); if (!taskRef(task_ref)) fail('invalid-request'); return this.#checkAuthority(s, 'git.push.working', { target: task_ref }); }); } // Row 93: records a push the git socket made (packages/seat/src/working.mjs) as the action.allowed // for git.push.working, with the commit. A refusal here moves the ref back. Trusted host only. recordPush(cap, push) { const s = this.#session(cap); return this.#store.transaction(() => { this.#agent(s); keys(push, ['task_ref', 'project', 'branch', 'ref', 'commit', 'previous'], ['task_ref', 'project', 'branch', 'ref', 'commit', 'previous']); if ( !taskRef(push.task_ref) || !Object.hasOwn(this.#business(s.business).projects ?? {}, push.project) || !oid(push.commit) || (push.previous !== null && !oid(push.previous)) ) fail('invalid-request'); string(push.branch, 200); string(push.ref, 200); const { task_ref, project, branch, ref, commit, previous } = push; this.#consumeAuthority(s, 'git.push.working', { target: task_ref }, { project, branch, ref, commit, previous }); return { task_ref, commit }; }); } // Row 93: what the git socket's fetch for a reviewer may send, from the review.requested event the // reviewer cites. The caller must hold review.verdict and be the request's reviewer. Trusted host only. reviewFetch(cap, { task_ref, request } = {}) { const s = this.#session(cap); return this.#store.transaction(() => { this.#agent(s); if (!taskRef(task_ref) || !id(request)) fail('invalid-request'); if (this.#classification(s, 'review.verdict') === 'gated') fail('not-authorized'); const r = this.#reviewRequest(s, task_ref, request); return { project: r.project, commit: r.commit, ref: r.ref }; }); } // A review.requested event on the task, addressed to this session's role. #reviewRequest(s, task, request) { const r = this.#store.get( "SELECT actor_role, body FROM events WHERE business=? AND kind='review.requested' AND id=? AND subject=?", s.business, request, task, ); const body = r ? JSON.parse(r.body) : null; if (!body || body.to !== s.role) fail('request-not-found'); return { ...body, from: r.actor_role }; } // 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', ...REVIEW_KINDS].includes(kind)) fail('reserved-event'); if (kind === 'action.allowed' && body?.action === 'git.push.working') 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, which adds // priority and requirement (#taskRows). 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', 'decisions', 'credentials', 'replies', 'messages'].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 }; } // Row 93 (#1587): the coder asks a reviewer role to review the task's latest pushed commit. case 'review.request': { keys(a, ['task_ref', 'commit', 'to', 'note', 'decision'], ['task_ref', 'commit', 'to']); if (!taskRef(a.task_ref) || !oid(a.commit)) fail('invalid-request'); const withinRole = this.#classification(s, 'review.request') === 'within-role'; this.#consumeAuthority(s, 'review.request', { decision: withinRole ? null : (a.decision ?? null), target: a.task_ref, }); if (a.note !== undefined) string(a.note, 16384); if (!Object.hasOwn(b.roles, a.to)) fail('unknown-role'); if (a.to === s.role) fail('self-review'); const reviewer = b.roles[a.to].authority; if (![...reviewer.withinRole, ...reviewer.crossRole].includes('review.verdict')) fail('not-a-reviewer'); const push = this.#latestPush(s.business, a.task_ref); if (push?.commit !== a.commit) { const older = this.#store.get( "SELECT 1 FROM events WHERE business=? AND kind='action.allowed' AND json_extract(body,'$.action')='git.push.working' AND subject=? AND json_extract(body,'$.commit')=?", s.business, a.task_ref, a.commit, ); fail(older ? 'stale-commit' : 'unknown-commit'); } const mid = randomUUID(); // Row 105 (#1599): the run holding the reviewer role, or null when nobody holds it and the // request waits for a claim. #reviewMessage routes on the same holder. const h = this.#holder(s.business, a.to); const held = h?.op === 'claim' ? h.holder_run : null; const eid = this.#event( s, 'review.requested', { task_ref: a.task_ref, project: push.project, ref: push.ref, commit: a.commit, to: a.to, message: mid, held }, a.task_ref, ); this.#reviewMessage(s, mid, { to: a.to, cls: 'REVIEW-REQUEST', decision: withinRole ? null : (a.decision ?? null), body: `Review request ${eid} for ${a.task_ref}: commit ${a.commit} on ${push.ref} in project ${push.project}. Fetch it with fetch_review, then answer with review_verdict citing this request.${a.note ? '\n\n' + a.note : ''}`, }); return { request: eid, message: mid, commit: a.commit, ref: push.ref, held }; } // Row 93: the reviewer's answer to one review.request, sent back to the role that asked. case 'review.verdict': { keys(a, ['task_ref', 'request', 'verdict', 'body', 'decision'], ['task_ref', 'request', 'verdict', 'body']); if (!taskRef(a.task_ref) || !id(a.request) || !VERDICTS.includes(a.verdict)) fail('invalid-request'); const withinRole = this.#classification(s, 'review.verdict') === 'within-role'; this.#consumeAuthority(s, 'review.verdict', { decision: withinRole ? null : (a.decision ?? null), target: a.task_ref, }); string(a.body, 32768); const r = this.#reviewRequest(s, a.task_ref, a.request); // One verdict per request: a changed mind needs a new request, so a close can't cite an // approve that a later verdict on the same request took back. if ( this.#store.get( "SELECT 1 FROM events WHERE business=? AND kind='review.verdict' AND subject=? AND json_extract(body,'$.request')=?", s.business, a.task_ref, a.request, ) ) fail('verdict-exists'); const mid = randomUUID(); const eid = this.#event( s, 'review.verdict', { task_ref: a.task_ref, request: a.request, commit: r.commit, verdict: a.verdict, message: mid }, a.task_ref, ); this.#reviewMessage(s, mid, { to: r.from, cls: 'REVIEW-RESULT', in_reply_to: r.message, decision: withinRole ? null : (a.decision ?? null), body: `Review verdict ${eid} on ${a.task_ref}, commit ${r.commit}: ${a.verdict}. A task_close cites this verdict id.\n\n${a.body}`, }); return { verdict: eid, message: mid, commit: r.commit }; } 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, ), }; } // Row 90 (#1584): the approved PRD's requirement ids, which task.create takes. A read for // every session that can create tasks: task.create is an agent verb, so a human refuses // agent-required here as there, and a reader read-only. case 'requirements': { keys(a, []); this.#agent(s); return this.#requirements; } case 'inbox': case 'agents': case 'tasks': case 'trail': { const allowed = verb === 'trail' ? ['subject'] : verb === 'tasks' ? TASK_ARGS : []; keys(a, allowed, 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') { const claims = 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, ); // A role session gets the claims only: another role's decisions aren't its to read. return s.human || s.reader ? claims.map((c) => this.#agentState(c)) : claims; } if (verb === 'tasks') return this.#tasks(s.business, a); return this.#trail(s, identifier(a.subject)); } // Row 57 (#1546): the Inbox's "Routed to roles" and "Recent" lists, for human and // reader sessions only. Each row is inbox's view of the decision: the `decisions` row trail // gives the same session for that id, with blocking as a boolean, plus `authorization` from // the raise's `action.allowed` event, which trail also gives. A `recent` row adds `closed`, // the decision's closing `decision_events` row, as trail gives it. case 'decisions': { if (!s.human && !s.reader) fail('human-required'); keys(a, ['list'], ['list']); if (a.list === 'roles') return this.#store .all( `SELECT d.* FROM decisions d WHERE business=? AND route_to<>'human' AND NOT EXISTS(SELECT 1 FROM decision_events x WHERE x.decision=d.id AND x.op IN ${CLOSED}) ORDER BY seq`, s.business, ) .map((d) => this.#decisionView(d)); if (a.list === 'recent') return this.#store .all( `SELECT d.* FROM decisions d JOIN decision_events x ON x.decision=d.id AND x.op IN ${CLOSED} WHERE d.business=? ORDER BY x.seq DESC LIMIT ${RECENT}`, s.business, ) .map((d) => ({ ...this.#decisionView(d), closed: this.#closed(d.id) })); fail('invalid-request'); } // Row 60 (#1549): what the running host loaded, checked against its clock, for the session's // business. Service, account, role, date and state only; no token, file path or variable name. case 'credentials': { if (!s.human && !s.reader) fail('human-required'); keys(a, [], []); const prefix = s.business + '/'; return this.#credentialStatus() .filter((c) => c.instance.startsWith(prefix)) .map(({ instance, service, date, state }) => { const role = instance.slice(prefix.length), sync = role === '@sync', bot = sync ? b.tracker?.sync?.bot : service === 'vikunja' ? b.roles?.[role]?.tracker?.bot : null; return { service, account: bot ?? null, role: sync ? 'tracker sync' : role, dateKind: service === 'gitea' ? 'rotateBy' : 'expires', date, state, }; }); } // Row 94: the human's message history. Sequence-based cursors remain stable // as new messages arrive. Successful reads write no delivery or read receipt. case 'messages': { if (!s.human && !s.reader) fail('human-required'); keys(a, ['limit', 'after']); if (('limit' in a && !(Number.isSafeInteger(a.limit) && a.limit >= 1 && a.limit <= 200)) || ('after' in a && !id(a.after))) fail('invalid-request'); let cursor = null; if ('after' in a) { cursor = this.#store.get("SELECT seq FROM messages WHERE business=? AND to_role='human' AND id=?", s.business, a.after); if (!cursor) fail('message-not-found'); } const counts = this.#store.get( "SELECT count(*) AS total, coalesce(sum(NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='read')),0) AS unread FROM messages m WHERE business=? AND to_role='human'", s.business); const limit = a.limit ?? 50; const rows = this.#store.all( "SELECT m.*, EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='read') AS read FROM messages m WHERE business=? AND to_role='human' AND (? IS NULL OR m.seq ({ ...m, read: !!m.read })); return { messages, ...counts, next: rows.length > limit ? messages.at(-1).id : null }; } // Row 81 (#1575): the replies to a message, for the Console's web-send receipt. Human and // reader sessions only, like decisions. It writes no delivery or read row, so a reply it // lists still reaches the next `mosaic talk`. An id naming no message in the business // refuses message-not-found, as message.send does for in_reply_to; trail would return []. case 'replies': { if (!s.human && !s.reader) fail('human-required'); keys(a, ['id'], ['id']); identifier(a.id); if (!this.#store.get('SELECT id FROM messages WHERE id=? AND business=?', a.id, s.business)) fail('message-not-found'); return this.#store.all( `SELECT * FROM messages WHERE business=? AND in_reply_to=? ORDER BY seq LIMIT ${REPLIES}`, s.business, a.id, ); } default: fail('unknown-verb'); } } // Row 59 (#1548): a held role's state for the Agents view, derived from what the bus // already records. Working: messages the current holder received and hasn't marked read; // the runner marks each one read after its turn. Input needed: the role's open decisions // routed to the human, in Inbox order. Working wins over waiting, as on the control board. #agentState(c) { const busy = this.#store.get( "SELECT count(*) AS messages, min(d.at) AS since FROM messages m JOIN deliveries d ON d.message=m.id AND d.op='delivered' WHERE m.business=? AND m.to_role=? AND d.holder_run=? AND NOT EXISTS(SELECT 1 FROM deliveries r WHERE r.message=m.id AND r.op='read')", c.business, c.role, c.holder_run, ); const inputNeeded = this.#store .all( `SELECT d.* FROM decisions d WHERE business=? AND raised_by_role=? AND route_to='human' 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`, c.business, c.role, ) .map((d) => ({ id: d.id, question: d.question, class: d.class, blocking: !!d.blocking, task_ref: d.task_ref, at: d.at })); const working = busy.messages ? { messages: busy.messages, since: busy.since } : null; const state = working ? 'working' : inputNeeded.length ? 'waiting' : 'idle'; // Row 80 (#1574): messages to the role that an earlier holder was delivered and never read. // The run they went to no longer holds the role. A delivered row already exists, so routing // and receive skip them. Listing them writes nothing. const stranded = this.#store.all( `SELECT m.id, m.from_role AS "from", d.holder_run AS run, d.at FROM messages m JOIN deliveries d ON d.message=m.id AND d.op='delivered' WHERE m.business=? AND m.to_role=? AND d.holder_run IS NOT ? AND NOT EXISTS(SELECT 1 FROM deliveries r WHERE r.message=m.id AND r.op='read') ORDER BY m.seq LIMIT ${STRANDED}`, c.business, c.role, c.holder_run, ).map((m) => ({ id: m.id, from: m.from, run: m.run, at: m.at })); return { ...c, state, working, inputNeeded, stranded }; } #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, ); } // The message a review verb sends, routed like message.send's. #reviewMessage(s, mid, { to, cls, in_reply_to = null, decision = null, body }) { 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.role, s.run, to, cls, in_reply_to, decision, null, body, ); const h = this.#holder(s.business, to); this.#delivery(mid, 'routed', h?.op === 'claim' ? h : null); } #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, ); } }