jason.woltjeandClaude Opus 5.5
3a1e11b83e
Row 105. Built by a Haiku subagent under Researcher (decision 129),
packet agents/researcher/work/queue-105/r1/, candidate manifest
cd6d5eb3e4093ac157f63d05cb639d243d76c6bde1a03e78efb8f29e191b6b44.
Filbert approved on #1599, comment 28247.
The review.request reply and the review.requested event carry held,
the run id holding the reviewer role or null. The request_review tool
description says what null means. The broker change is live only at
a bus restart.
Sage's gate rerun on 79e126be0 with rows 101 and 105 applied, Docker
up: node suites bus 128, business 67, cli 146, control-board 124,
conversation 182, discord 178, harness 70, ledger 78, mosaic 69,
queue 155, runs 43, seat 109, tasks 92, webui 108; scripts auth 15,
conductor 17, config 24, discord 66, extension-package 18,
foundation 44, queue 28, release 14, task 98. All pass, 0 fail.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
1373 lines
57 KiB
JavaScript
1373 lines
57 KiB
JavaScript
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<?) ORDER BY m.seq DESC LIMIT ?",
|
|
s.business, cursor?.seq ?? null, cursor?.seq ?? null, limit + 1);
|
|
const messages = rows.slice(0, limit).map(m => ({ ...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,
|
|
);
|
|
}
|
|
}
|