mosaicstack/stack
25
0 Fork 1
Code Issues 404 Pull Requests 13 Packages Projects Releases 15 Wiki Activity
Files
org-cred-bind-design
stack/agents/rocko/work/queue-124/build.patch
T
jason.woltje 068d868a99 lead: commit rocko's untracked evidence (4471 files: queue-124 queue-129 queue-171 queue-178 queue-280 queue-31 queue-84 queue-85 queue-94) 
Unclaimed files left in the canonical checkout. Inventory: canonical-cleanup-inventory.md (commit-as-evidence groups). detect-secrets scan clean on 2026-11-02: hits are sha256 pins, git object ids, test sentinels already in tracked tests and unit-file lines.
2026-11-02 15:05:15 -06:00

1853 lines
114 KiB
Diff
Raw Permalink Blame History

diff --git a/docs/TOOLS.md b/docs/TOOLS.md
index aff8a9c95..9a2c23ec2 100644
--- a/docs/TOOLS.md
+++ b/docs/TOOLS.md
@@ -653,3 +653,15 @@ record-repository path; the default is
`<dataRoot>/projects/<business>/<project>/`. `root` remains the code checkout.
The package README describes schemas and the shared agent record writer; agent
writes never enter `spec/`, never absorb pending edits and refuse remotes.
+
+### Token usage (row 124)
+
+`mosaic usage [--business ID] [--json]` reads token accounting from the bus.
+Each total, run, instance, model, provider and account group shows input,
+output, cache write, cache read and unknown turns. A malformed or unavailable
+transcript contributes an unknown turn, never a zero-token reading. Accounts
+are unassigned until account tracking ships. Exact totals larger than the
+safe integer range are decimal strings in JSON. Settings has the same
+read-only Token usage section. Counts are a best reading, not proof: the
+session can edit its own transcript. This command changes no setting and
+performs no token-limit enforcement.
diff --git a/packages/bus/README.md b/packages/bus/README.md
index f18c527a0..c95f39278 100644
--- a/packages/bus/README.md
+++ b/packages/bus/README.md
@@ -797,3 +797,29 @@ its triggers, and a nullable `to_name` column on messages with nulls on existing
changes no other row. The 3f fixture, `tests/fixtures/schema-3f.sql`, is pinned from
4ebef2284's tree; no live store is used. The step runs only in its own stopped-bus/re-pin
window on Mos's go.
+
+## Token accounting per run (row 124, REQ-LIMIT-4)
+
+Schema 3h adds only the append-only usage_records table and its three protection
+triggers. The 3g→3h step preserves all prior evidence; migration is explicit and
+offline in Mos's re-pin window. The normal bus never auto-migrates an old store.
+
+The managed runner calls usage.begin before an actual harness turn and usage.end
+after it, including a failed/stopped turn. Each takes only {turn: deliveredMessageId}.
+The broker requires the bound role holder and its delivery, binds run identity,
+and calls the runtime-injected accountant. These are runner protocol verbs,
+not harness tools. The accountant calls configured usage sources only through
+the plug-in loader. It durably baselines history, then appends one outcome per
+source/run/turn. Replays return the recorded result without charging twice.
+An absent/failed source, parse failure, missing baseline, transcript rewrite or
+incomplete turn remains unknown. Unknown outcomes are logged with fixed codes,
+never transcript text. A pending turn is unknown until its end arrives.
+
+The business-scoped usage read is available to its human, reader and held role.
+It returns total plus byRun, byInstance, byProvider, byModel and byAccount arrays.
+Each total has input, output, cacheWrite, cacheRead and unknownTurns. Known
+readings sum; unknown turns count distinctly by run and turn next to every total.
+An exact aggregate beyond JavaScript's safe integer range is a decimal string,
+not a rounded count or a policy ceiling. Account is null until account tracking
+ships. A session can edit its own transcript: this is a best reading, not proof.
+No enforcement or resets are added. Settings reads through the same bus view.
diff --git a/packages/bus/schema.sql b/packages/bus/schema.sql
index 6e306431f..53059db47 100644
--- a/packages/bus/schema.sql
+++ b/packages/bus/schema.sql
@@ -335,3 +335,31 @@ CREATE TRIGGER instance_names_unique BEFORE INSERT ON instance_names
CREATE TRIGGER instance_names_no_update BEFORE UPDATE ON instance_names BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END;
CREATE TRIGGER instance_names_no_delete BEFORE DELETE ON instance_names BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END;
CREATE TRIGGER instance_names_no_replace BEFORE INSERT ON instance_names WHEN EXISTS (SELECT 1 FROM instance_names WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END;
+-- Row 124 (#1618): immutable source snapshots and turn outcomes, never transcript text.
+CREATE TABLE usage_records (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ run TEXT NOT NULL,
+ instance TEXT NOT NULL,
+ provider TEXT,
+ model TEXT,
+ account TEXT CHECK (account IS NULL),
+ turn TEXT NOT NULL,
+ source TEXT NOT NULL,
+ phase TEXT NOT NULL CHECK (phase IN ('begin','end')),
+ status TEXT NOT NULL CHECK (status IN ('known','unknown')),
+ code TEXT,
+ cursor TEXT CHECK (cursor IS NULL OR json_valid(cursor)),
+ input INTEGER,
+ output INTEGER,
+ cache_write INTEGER,
+ cache_read INTEGER,
+ UNIQUE (business,run,turn,source,phase),
+ CHECK ((status='unknown' AND input IS NULL AND output IS NULL AND cache_write IS NULL AND cache_read IS NULL)
+ OR (status='known' AND input >= 0 AND output >= 0 AND cache_write >= 0 AND cache_read >= 0
+ AND input IS NOT NULL AND output IS NOT NULL AND cache_write IS NOT NULL AND cache_read IS NOT NULL))
+) STRICT;
+CREATE TRIGGER usage_records_no_update BEFORE UPDATE ON usage_records BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END;
+CREATE TRIGGER usage_records_no_delete BEFORE DELETE ON usage_records BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END;
+CREATE TRIGGER usage_records_no_replace BEFORE INSERT ON usage_records WHEN EXISTS (SELECT 1 FROM usage_records WHERE seq=NEW.seq OR (business=NEW.business AND run=NEW.run AND turn=NEW.turn AND source=NEW.source AND phase=NEW.phase)) BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END;
diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs
index 3f760abe3..cf362407c 100644
--- a/packages/bus/src/broker.mjs
+++ b/packages/bus/src/broker.mjs
@@ -182,9 +182,10 @@ export class Broker {
#requirements;
#organization;
#project;
+ #usage;
#clock = 0;
// credentialStatus is the runtime's metadata-only Credentials.status(); the broker never gets use().
- constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null }) {
+ constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null, usage = null }) {
if (!object(businesses) || !Object.keys(businesses).length) fail('invalid-business');
this.#businesses = structuredClone(businesses);
this.#store = store;
@@ -192,6 +193,7 @@ export class Broker {
this.#credentialStatus = credentialStatus;
this.#organization = organization;
this.#project = project;
+ this.#usage = usage;
// Row 90 (#1584): the approved PRD's requirement ids. A bad requirements.json refuses here.
this.#requirements = loadRequirements();
for (const table of [
@@ -390,7 +392,7 @@ export class Broker {
bindLaunch(record) {
keys(
record,
- ['business', 'role', 'run', 'harness', 'address', 'pid', 'startTime', 'thread'],
+ ['business', 'role', 'run', 'harness', 'address', 'pid', 'startTime', 'thread', 'provider', 'model'],
['business', 'role', 'run', 'harness'],
);
const b = this.#business(identifier(record.business));
@@ -398,6 +400,8 @@ export class Broker {
identifier(record.run);
if (!['pi', 'claude-code'].includes(record.harness)) fail('invalid-harness');
if (record.address !== undefined) string(record.address, 1024);
+ if (record.provider !== undefined) string(record.provider, 1024);
+ if (record.model !== undefined) string(record.model, 1024);
// Row 121: a thread instance's run is for one thread, and no other run names one.
if (this.#threaded(record.business, record.role)) {
if (record.thread === undefined) fail('thread-required');
@@ -1119,6 +1123,8 @@ export class Broker {
const args = request.args ?? {};
if (!object(args)) fail('invalid-request');
this.#secretCheck(args);
+ if (request.verb === 'usage.begin' || request.verb === 'usage.end')
+ return this.#requestUsage(s, request.verb, structuredClone(args));
if (request.verb === 'spec.read' || request.verb === 'record.write' ||
(request.verb === 'decision.raise' && ('spec_path' in args || 'spec_commit' in args)))
return this.#requestProject(s, { verb: request.verb, args: structuredClone(args) });
@@ -1159,7 +1165,7 @@ export class Broker {
return result;
}
return this.#store.transaction(() => {
- if (s.reader && !['inbox', 'agents', 'tasks', 'trail', 'decisions', 'credentials', 'replies', 'messages', 'state'].includes(request.verb))
+ if (s.reader && !['inbox', 'agents', 'tasks', 'trail', 'decisions', 'credentials', 'replies', 'messages', 'state', 'usage'].includes(request.verb))
fail('read-only');
if (s.tester && !TESTER_VERBS.includes(request.verb)) fail('tester-refused');
// A tester reads the human's views, as a reader does, and its message verbs are its own.
@@ -1172,6 +1178,20 @@ export class Broker {
throw this.#refused(s, e, request);
}
}
+ // Only the runner's bound session and delivered message identify a turn; no host path crosses the wire.
+ async #requestUsage(s, verb, args) {
+ try {
+ keys(args, ['turn'], ['turn']);
+ identifier(args.turn);
+ this.#store.transaction(() => {
+ this.#agent(s);
+ if (!this.#store.get("SELECT 1 FROM messages m JOIN deliveries d ON d.message=m.id WHERE m.id=? AND m.business=? AND m.to_role=? AND d.op='delivered' AND d.holder_run=?", args.turn, s.business, s.role, s.run))
+ fail('not-recipient');
+ });
+ if (!this.#usage) fail('usage-not-configured');
+ return await this.#usage.collect(s, verb === 'usage.begin' ? 'begin' : 'end', args.turn);
+ } catch (e) { throw this.#refused(s, e, { verb, args }); }
+ }
// Git and the record writer are async effects: never hold a SQL transaction across an await.
async #requestProject(s, request) {
const { verb, args } = request;
@@ -1896,6 +1916,12 @@ export class Broker {
// the time and id of the event that set it (`allowed`, no time and no id when none has), and
// the open tasks as the tracker snapshot shows them: refs and titles, at most `cap` of them,
// `total` counting every open task. A read for role, reader and human sessions; nothing is stored.
+ case 'usage': {
+ keys(a, [], []);
+ if (!s.human && !s.reader) this.#agent(s);
+ if (!this.#usage) fail('usage-not-configured');
+ return this.#usage.view(s.business);
+ }
case 'state': {
keys(a, []);
if (!s.human && !s.reader) this.#agent(s);
diff --git a/packages/bus/src/migrations/3g-3h.sql b/packages/bus/src/migrations/3g-3h.sql
new file mode 100644
index 000000000..32880e24d
--- /dev/null
+++ b/packages/bus/src/migrations/3g-3h.sql
@@ -0,0 +1,28 @@
+-- Row 124 (#1618): immutable source snapshots and turn outcomes, never transcript text.
+CREATE TABLE usage_records (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ run TEXT NOT NULL,
+ instance TEXT NOT NULL,
+ provider TEXT,
+ model TEXT,
+ account TEXT CHECK (account IS NULL),
+ turn TEXT NOT NULL,
+ source TEXT NOT NULL,
+ phase TEXT NOT NULL CHECK (phase IN ('begin','end')),
+ status TEXT NOT NULL CHECK (status IN ('known','unknown')),
+ code TEXT,
+ cursor TEXT CHECK (cursor IS NULL OR json_valid(cursor)),
+ input INTEGER,
+ output INTEGER,
+ cache_write INTEGER,
+ cache_read INTEGER,
+ UNIQUE (business,run,turn,source,phase),
+ CHECK ((status='unknown' AND input IS NULL AND output IS NULL AND cache_write IS NULL AND cache_read IS NULL)
+ OR (status='known' AND input >= 0 AND output >= 0 AND cache_write >= 0 AND cache_read >= 0
+ AND input IS NOT NULL AND output IS NOT NULL AND cache_write IS NOT NULL AND cache_read IS NOT NULL))
+) STRICT;
+CREATE TRIGGER usage_records_no_update BEFORE UPDATE ON usage_records BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END;
+CREATE TRIGGER usage_records_no_delete BEFORE DELETE ON usage_records BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END;
+CREATE TRIGGER usage_records_no_replace BEFORE INSERT ON usage_records WHEN EXISTS (SELECT 1 FROM usage_records WHERE seq=NEW.seq OR (business=NEW.business AND run=NEW.run AND turn=NEW.turn AND source=NEW.source AND phase=NEW.phase)) BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END;
diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs
index 6ae5098e3..985ac7953 100644
--- a/packages/bus/src/runtime.mjs
+++ b/packages/bus/src/runtime.mjs
@@ -11,6 +11,7 @@ import { createLoader, hasModule } from '../../plugins/src/loader.mjs';
import { pluginSettings } from '../../business/src/plugin-settings.mjs';
import { webService } from './web.mjs';
import { verifyHuman } from './human.mjs';
+import { createUsage } from './usage.mjs';
import { verifyTester } from './tester.mjs';
// Trusted host API. S1 supplies resolved definitions, S6 supplies launch records and ends them.
// Human limits writes use the shared organization writer behind the verified transport.
@@ -64,7 +65,10 @@ export async function startBroker({
// plugins and plugins.set, logged in changes.jsonl with the actor, a missing module refused.
const plugins=organization?pluginSettings({dir:organization.dir,rolesDir:organization.rolesDir,exists:hasModule}):null;
const settings=organization?launchSettings({...organization,sessions:()=>readSessions(dataRoot)}):null;
+ const usage = await createUsage({ store, dataRoot, businesses,
+ log: entry => process.stderr.write(`usage: ${JSON.stringify(entry)}\n`) });
const broker = new Broker({
+ usage,
store,
businesses,
project: {
diff --git a/packages/bus/src/server.mjs b/packages/bus/src/server.mjs
index 32b91e877..7ea3f581d 100644
--- a/packages/bus/src/server.mjs
+++ b/packages/bus/src/server.mjs
@@ -103,7 +103,8 @@ export async function serve({ broker, path, authenticateHuman = null, authentica
return;
}
// Row138: only project Git effects need an await; preserve every other sync path below.
- if (r.verb === 'spec.read' || r.verb === 'record.write' || PROJECT_FILE_VERBS.includes(r.verb) ||
+ if (r.verb === 'usage.begin' || r.verb === 'usage.end' ||
+ r.verb === 'spec.read' || r.verb === 'record.write' || PROJECT_FILE_VERBS.includes(r.verb) ||
(r.verb === 'decision.raise' && (Object.hasOwn(r.args ?? {}, 'spec_path') || Object.hasOwn(r.args ?? {}, 'spec_commit')))) {
const own = transient;
transient = null;
diff --git a/packages/bus/src/store.mjs b/packages/bus/src/store.mjs
index 8f29dd37c..d995d7f4d 100644
--- a/packages/bus/src/store.mjs
+++ b/packages/bus/src/store.mjs
@@ -13,8 +13,8 @@ import { join, isAbsolute } from 'node:path';
import { createHash } from 'node:crypto';
const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
-export const SCHEMA_SHA256 = 'a79b3dbe6f4dfba10871c01ad8513ea290071eb8d8af2b87be6db52dbcf23428';
-export const SCHEMA_VERSION = '3g';
+export const SCHEMA_SHA256 = 'bb2ba58175a111bb99681778911507846bef714a6d573f8a8af67580a3636950';
+export const SCHEMA_VERSION = '3h';
// Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned.
const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8');
const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65';
@@ -47,6 +47,10 @@ const migrations = [{
from: '3f',
to: '3g',
sql: readFileSync(new URL('./migrations/3f-3g.sql', import.meta.url), 'utf8'),
+}, {
+ from: '3g',
+ to: '3h',
+ sql: readFileSync(new URL('./migrations/3g-3h.sql', import.meta.url), 'utf8'),
}];
const timedTables = [
'events',
diff --git a/packages/bus/src/usage.mjs b/packages/bus/src/usage.mjs
new file mode 100644
index 000000000..d1118dd20
--- /dev/null
+++ b/packages/bus/src/usage.mjs
@@ -0,0 +1,123 @@
+// REQ-LIMIT-4: the host calls sources only through the registry loader.
+import { join } from 'node:path';
+import { createLoader } from '../../plugins/src/loader.mjs';
+const KINDS = ['input', 'output', 'cacheWrite', 'cacheRead'];
+const zero = () => ({ input: 0, output: 0, cacheWrite: 0, cacheRead: 0 });
+const unknown = code => ({ status: 'unknown', code });
+function snapshot(value) {
+ if (value?.status === 'unknown') return unknown(/^transcript-(path|size|line|entries|records|changed|parse|usage|harness)$/.test(value.code) ? value.code : 'source-unknown');
+ if (value?.status !== 'known' || !value.entries || typeof value.entries !== 'object' || Array.isArray(value.entries))
+ return unknown('source-invalid');
+ const entries = Object.entries(value.entries);
+ if (entries.length > 20000) return unknown('source-invalid');
+ for (const [id, v] of entries) {
+ if (id.length > 512 || !v || Object.keys(v).length !== 4 ||
+ KINDS.some(k => !Number.isSafeInteger(v[k]) || v[k] < 0)) return unknown('source-invalid');
+ }
+ return { status: 'known', entries: value.entries };
+}
+function delta(before, after) {
+ if (!before || before.status !== 'known') return unknown('baseline-unknown');
+ if (after.status !== 'known') return after;
+ for (const [id, v] of Object.entries(before.entries)) {
+ if (!Object.hasOwn(after.entries, id) || KINDS.some(k => after.entries[id][k] !== v[k])) return unknown('transcript-rewritten');
+ }
+ const sum = zero(); let added = false;
+ for (const [id, v] of Object.entries(after.entries)) {
+ if (Object.hasOwn(before.entries, id)) continue;
+ added = true;
+ for (const k of KINDS) {
+ sum[k] += v[k];
+ if (!Number.isSafeInteger(sum[k])) return unknown('usage-overflow');
+ }
+ }
+ return added ? { status: 'known', ...sum } : unknown('no-new-usage');
+}
+export async function createUsage({ store, dataRoot, businesses, log }) {
+ const loaders = new Map();
+ for (const [id, b] of Object.entries(businesses)) {
+ const loader = createLoader({ log });
+ await loader.load(b.plugins ?? {});
+ loaders.set(id, loader);
+ }
+ // This map serializes retries per run, without keeping a SQL transaction across source awaits.
+ const pending = new Map();
+ async function perform(session, phase, turn) {
+ const { business, run, role: instance, harness, provider = null, model = null } = session;
+ const existing = store.all('SELECT * FROM usage_records WHERE business=? AND run=? AND turn=? AND phase=?', business, run, turn, phase);
+ if (existing.length) return { recorded: true };
+ const loader = loaders.get(business);
+ const modules = loader.modules('usage-source'), names = loader.loaded()['usage-source'];
+ const bases = store.all("SELECT * FROM usage_records WHERE business=? AND run=? AND turn=? AND phase='begin'", business, run, turn);
+ const sources = modules.length ? modules.map((module, i) => ({ module, name: names[i] })) : [{ name: '@none' }];
+ // Preserve a failed/disabled baseline even if configuration changes between calls/restarts.
+ for (const base of bases) if (!sources.some(s => s.name === base.source)) sources.push({ name: base.source });
+ const rows = [];
+ const runDir = join(dataRoot, 'launches', business, run);
+ const root = session.thread
+ ? join(dataRoot, 'threads', business, session.thread, harness === 'pi' ? 'session' : 'claude-projects')
+ : harness === 'pi' ? join(runDir, 'session') : join(runDir, 'home/.claude/projects');
+ for (const { module, name } of sources) {
+ let sample = unknown('no-source');
+ if (module) {
+ try { sample = snapshot(await module.readTurn({ harness, root })); }
+ catch { sample = unknown('source-failed'); }
+ }
+ const base = bases.find(row => row.source === name);
+ const result = phase === 'begin' ? sample : delta(base?.cursor ? JSON.parse(base.cursor) : null, sample);
+ rows.push({ name, result });
+ }
+ store.transaction(() => {
+ for (const { name, result } of rows) {
+ const known = result.status === 'known';
+ const counts = phase === 'begin' ? zero() : result;
+ store.run(`INSERT INTO usage_records(at,business,run,instance,provider,model,account,turn,source,phase,status,code,cursor,input,output,cache_write,cache_read)
+ VALUES (?,?,?,?,?,?,NULL,?,?,?,?,?,?,?,?,?,?)`, new Date().toISOString(), business, run, instance,
+ provider, model, turn, name, phase, result.status, result.code ?? null,
+ phase === 'begin' ? JSON.stringify(result) : null,
+ known ? counts.input : null, known ? counts.output : null, known ? counts.cacheWrite : null, known ? counts.cacheRead : null);
+ }
+ });
+ for (const { name, result } of rows) if (phase === 'end' && result.status === 'unknown') {
+ log({ event: 'usage-unknown', business, run, instance, turn, source: name, code: result.code });
+ }
+ return { recorded: true };
+ }
+ function collect(session, phase, turn) {
+ const key = session.business + ':' + session.run;
+ const op = (pending.get(key) ?? Promise.resolve()).then(() => perform(session, phase, turn));
+ const tail = op.catch(() => {});
+ pending.set(key, tail);
+ return op.finally(() => { if (pending.get(key) === tail) pending.delete(key); });
+ }
+ function view(business) {
+ const rows = store.all(`SELECT * FROM usage_records WHERE business=? ORDER BY seq`, business);
+ const turns = new Map();
+ for (const row of rows) {
+ const key = JSON.stringify([row.run, row.turn, row.source]);
+ if (row.phase === 'end' || !turns.has(key)) turns.set(key, row);
+ }
+ const groups = { byRun: new Map(), byInstance: new Map(), byProvider: new Map(), byModel: new Map(), byAccount: new Map() };
+ const total = { ...zero(), unknown: new Set() };
+ const add = (target, row) => {
+ if (row.phase !== 'end' || row.status !== 'known') target.unknown.add(row.run + ':' + row.turn);
+ else for (const [key, column] of [['input', 'input'], ['output', 'output'], ['cacheWrite', 'cache_write'], ['cacheRead', 'cache_read']]) target[key] = BigInt(target[key]) + BigInt(row[column]);
+ };
+ for (const row of turns.values()) {
+ add(total, row);
+ for (const [name, identity] of Object.entries({ byRun: { run: row.run }, byInstance: { instance: row.instance },
+ byProvider: { provider: row.provider }, byModel: { provider: row.provider, model: row.model }, byAccount: { account: row.account } })) {
+ const key = JSON.stringify(identity);
+ if (!groups[name].has(key)) groups[name].set(key, { ...identity, ...zero(), unknown: new Set() });
+ add(groups[name].get(key), row);
+ }
+ }
+ const finish = value => {
+ const { unknown, ...rest } = value;
+ for (const kind of KINDS) rest[kind] = BigInt(rest[kind]) <= BigInt(Number.MAX_SAFE_INTEGER) ? Number(rest[kind]) : String(rest[kind]);
+ return { ...rest, unknownTurns: unknown.size };
+ };
+ return { total: finish(total), ...Object.fromEntries(Object.entries(groups).map(([key, map]) => [key, [...map.values()].map(finish)])) };
+ }
+ return { collect, view };
+}
diff --git a/packages/bus/src/views.mjs b/packages/bus/src/views.mjs
index d5dc5a2c2..f8d1386c6 100644
--- a/packages/bus/src/views.mjs
+++ b/packages/bus/src/views.mjs
@@ -1,6 +1,7 @@
// Shared by the S4 CLI and S5 WebUI. Client-only: never opens bus.sqlite.
export function views(client) {
return Object.freeze({
+ usage: () => client.call('usage'),
inbox: () => client.call('inbox'),
tasks: () => client.call('tasks'),
agents: () => client.call('agents'),
diff --git a/packages/bus/tests/chief-of-staff.test.mjs b/packages/bus/tests/chief-of-staff.test.mjs
index 46505ef73..c7c650f1a 100644
--- a/packages/bus/tests/chief-of-staff.test.mjs
+++ b/packages/bus/tests/chief-of-staff.test.mjs
@@ -369,7 +369,7 @@ async function database3c(t) {
let source = readFileSync(here('../src/store.mjs'), 'utf8');
const pinned = source
.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3c).digest('hex')}';`)
- .replace("export const SCHEMA_VERSION = '3g';", "export const SCHEMA_VERSION = '3c';")
+ .replace("export const SCHEMA_VERSION = '3h';", "export const SCHEMA_VERSION = '3c';")
.replace(/\}, \{\n {2}from: '3c',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
assert.notEqual(pinned, source);
assert.ok(!pinned.includes('3c-3d'), 'the 3c engine has no 3c→3d step');
@@ -393,7 +393,7 @@ test('REQ-ROLE-2, REQ-COS-1: the 3c→3d migration lets the chief of staff end a
assert.throws(() => new Store(root), /schema-mismatch/);
const store = new Store(root, { migrateSchema: true });
t.after(() => store.close());
- assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3g');
+ assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3h');
assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}');
assert.equal(store.get("SELECT holder_run FROM role_claims WHERE role='coder'").holder_run, 'coder-run');
const b = new Broker({ store, businesses: demo(), organization: null });
diff --git a/packages/bus/tests/comment-migration.test.mjs b/packages/bus/tests/comment-migration.test.mjs
index d7675e297..9db61ce43 100644
--- a/packages/bus/tests/comment-migration.test.mjs
+++ b/packages/bus/tests/comment-migration.test.mjs
@@ -33,7 +33,7 @@ test('explicit 3b copy migration preserves evidence and opens as 3g; normal star
const original = readFileSync(source);
assert.throws(() => new Store(root), /schema-mismatch/);
const store = new Store(root, { migrateSchema: true });
- assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3g');
+ assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3h');
assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}');
assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
const epoch = store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value;
@@ -75,15 +75,15 @@ test('named migration registry chains 3b through 3g to a fixture-only future ver
const sha = createHash('sha256').update(futureDDL).digest('hex');
let source = readFileSync(new URL('../src/store.mjs', import.meta.url), 'utf8');
source = source.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${sha}';`)
- .replace("export const SCHEMA_VERSION = '3g';", "export const SCHEMA_VERSION = '3h';")
- .replace('}];\nconst timedTables', `}, { from: '3g', to: '3h', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
+ .replace("export const SCHEMA_VERSION = '3h';", "export const SCHEMA_VERSION = '3i';")
+ .replace('}];\nconst timedTables', `}, { from: '3h', to: '3i', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
writeFileSync(join(moduleRoot, 'src/store.mjs'), source);
writeFileSync(join(moduleRoot, 'schema.sql'), futureDDL);
- for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql'])
+ for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql'])
copyFileSync(new URL('../src/migrations/' + name, import.meta.url), join(moduleRoot, 'src/migrations', name));
const { Store: FutureStore } = await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')));
const future = new FutureStore(root, { migrateSchema: true });
- assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3h');
+ assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3i');
assert.equal(future.get("SELECT name FROM sqlite_master WHERE name='future_fixture'").name, 'future_fixture');
assert.equal(future.get("SELECT count(*) n FROM task_comment_cursors").n, 0);
assert.equal(future.get("SELECT id FROM events").id, 'preserved');
@@ -109,7 +109,7 @@ test('offline command migrates only the supplied fixture root and reports usage
const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname;
const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' });
assert.equal(migrated.status, 0, migrated.stderr);
- assert.match(migrated.stdout, /bus schema 3g ready/);
+ assert.match(migrated.stdout, /bus schema 3h ready/);
new Store(root).close();
const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' });
assert.equal(usage.status, 2);
diff --git a/packages/bus/tests/fixtures/schema-3g.sql b/packages/bus/tests/fixtures/schema-3g.sql
new file mode 100644
index 000000000..6e306431f
--- /dev/null
+++ b/packages/bus/tests/fixtures/schema-3g.sql
@@ -0,0 +1,337 @@
+PRAGMA journal_mode = WAL;
+PRAGMA foreign_keys = ON;
+CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT;
+CREATE TABLE events (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ kind TEXT NOT NULL CHECK (kind IN (
+ 'session.launched',
+ 'session.ended',
+ 'action.allowed',
+ 'action.refused',
+ 'task.created',
+ 'task.assigned',
+ 'task.state',
+ 'task.closed',
+ 'task.changed.external',
+ 'task.conflict',
+ 'task.missing',
+ 'review.requested',
+ 'review.verdict',
+ 'human.input',
+ 'config.refused',
+ 'credential.expiring',
+ 'credential.expired',
+ 'credential.changed',
+ 'launch.revoked',
+ 'launch.restored',
+ 'digest.sent')),
+ actor_role TEXT, actor_run TEXT,
+ subject TEXT,
+ corrects TEXT REFERENCES events(id),
+ body TEXT NOT NULL CHECK (json_valid(body))
+) STRICT;
+CREATE TABLE role_claims (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL, role TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')),
+ holder_run TEXT NOT NULL,
+ harness TEXT NOT NULL, address TEXT,
+ by TEXT NOT NULL, reason TEXT,
+ decision TEXT
+) STRICT;
+CREATE TABLE decisions (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL, project TEXT,
+ raised_by_role TEXT NOT NULL, raised_by_run TEXT NOT NULL,
+ class TEXT NOT NULL CHECK (class IN ('routine','within-role','cross-role','gated')),
+ action TEXT NOT NULL,
+ route_to TEXT NOT NULL,
+ question TEXT NOT NULL,
+ options TEXT NOT NULL CHECK (json_valid(options) AND json_array_length(options) BETWEEN 2 AND 9),
+ recommendation TEXT NOT NULL,
+ task_ref TEXT CHECK (task_ref IS NULL OR (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*')), requirement_ref TEXT,
+ blocking INTEGER NOT NULL CHECK (blocking IN (0,1)),
+ supersedes TEXT REFERENCES decisions(id)
+, spec_path TEXT, spec_commit TEXT) STRICT;
+CREATE TABLE decision_events (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ decision TEXT NOT NULL REFERENCES decisions(id),
+ at TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('seen','resolved','withdrawn','expired')),
+ by TEXT NOT NULL,
+ choice TEXT, note TEXT, via TEXT
+) STRICT;
+CREATE TABLE messages (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ from_role TEXT NOT NULL, from_run TEXT NOT NULL,
+ to_role TEXT NOT NULL,
+ class TEXT NOT NULL,
+ in_reply_to TEXT REFERENCES messages(id),
+ decision TEXT REFERENCES decisions(id),
+ corrects TEXT REFERENCES messages(id),
+ body TEXT NOT NULL
+, to_name TEXT) STRICT;
+CREATE TABLE deliveries (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ message TEXT NOT NULL REFERENCES messages(id),
+ at TEXT NOT NULL,
+ op TEXT NOT NULL CHECK (op IN ('routed','delivered','failed','read')),
+ holder_run TEXT, transport TEXT, address TEXT, detail TEXT
+) STRICT;
+CREATE TABLE task_snapshots (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ task_ref TEXT NOT NULL CHECK (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*'),
+ updated TEXT NOT NULL,
+ etag TEXT,
+ digest TEXT NOT NULL CHECK (length(digest) = 64 AND NOT digest GLOB '*[^0-9a-f]*'),
+ fields TEXT NOT NULL CHECK (json_valid(fields)),
+ source TEXT NOT NULL CHECK (source IN ('self','poll')),
+ via TEXT CHECK (via IN ('board','cursor','task','reconcile')),
+ read_at TEXT,
+ role TEXT, run TEXT,
+ CHECK ((source = 'self') = (role IS NOT NULL AND run IS NOT NULL)),
+ CHECK ((source = 'poll') = (via IS NOT NULL AND read_at IS NOT NULL)),
+ CHECK (json_type(fields, '$.bucket') IS 'integer'
+ OR (source IS 'poll' AND via IS 'task' AND json_type(fields, '$.gone') IS 'text'))
+) STRICT;
+CREATE INDEX task_snapshots_ref ON task_snapshots (business, task_ref, seq);
+CREATE TRIGGER decisions_resolve_once BEFORE INSERT ON decision_events
+ WHEN NEW.op IN ('resolved','withdrawn','expired') AND EXISTS (
+ SELECT 1 FROM decision_events WHERE decision = NEW.decision AND op IN ('resolved','withdrawn','expired'))
+ BEGIN SELECT RAISE(ABORT, 'decision already closed'); END;
+CREATE TRIGGER decisions_resolved_choice BEFORE INSERT ON decision_events
+ WHEN NEW.op = 'resolved' AND (NEW.choice IS NULL OR NOT EXISTS (
+ SELECT 1 FROM decisions d, json_each(d.options) o WHERE d.id = NEW.decision AND json_extract(o.value,'$.key') = NEW.choice))
+ BEGIN SELECT RAISE(ABORT, 'resolution must name one of the options'); END;
+CREATE TRIGGER role_one_holder BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'claim' AND (SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) = 'claim'
+ BEGIN SELECT RAISE(ABORT, 'role already held'); END;
+CREATE TRIGGER role_release_by_holder BEFORE INSERT ON role_claims
+ WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1),'') <> 'claim'
+ BEGIN SELECT RAISE(ABORT, 'role is not held'); END;
+CREATE TRIGGER role_release_same_run BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'release' AND (SELECT holder_run FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) <> NEW.holder_run
+ BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END;
+CREATE TRIGGER role_revoke_needs_decision BEFORE INSERT ON role_claims
+ WHEN NEW.op = 'revoke' AND NEW.decision IS NULL AND NEW.reason IS NULL
+ BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision or a reason'); END;
+CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER meta_no_delete BEFORE DELETE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER events_no_update BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER events_no_delete BEFORE DELETE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER role_claims_no_update BEFORE UPDATE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER role_claims_no_delete BEFORE DELETE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER decisions_no_update BEFORE UPDATE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decisions_no_delete BEFORE DELETE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decision_events_no_update BEFORE UPDATE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER decision_events_no_delete BEFORE DELETE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER messages_no_update BEFORE UPDATE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER messages_no_delete BEFORE DELETE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER deliveries_no_update BEFORE UPDATE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER deliveries_no_delete BEFORE DELETE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER meta_no_replace BEFORE INSERT ON meta WHEN EXISTS (SELECT 1 FROM meta WHERE key = NEW.key) BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
+CREATE TRIGGER events_no_replace BEFORE INSERT ON events WHEN EXISTS (SELECT 1 FROM events WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
+CREATE TRIGGER role_claims_no_replace BEFORE INSERT ON role_claims WHEN EXISTS (SELECT 1 FROM role_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
+CREATE TRIGGER decisions_no_replace BEFORE INSERT ON decisions WHEN EXISTS (SELECT 1 FROM decisions WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
+CREATE TRIGGER decision_events_no_replace BEFORE INSERT ON decision_events WHEN EXISTS (SELECT 1 FROM decision_events WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
+CREATE TRIGGER messages_no_replace BEFORE INSERT ON messages WHEN EXISTS (SELECT 1 FROM messages WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
+CREATE TRIGGER deliveries_no_replace BEFORE INSERT ON deliveries WHEN EXISTS (SELECT 1 FROM deliveries WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
+CREATE TRIGGER task_snapshots_no_update BEFORE UPDATE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER task_snapshots_no_delete BEFORE DELETE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER task_snapshots_no_replace BEFORE INSERT ON task_snapshots WHEN EXISTS (SELECT 1 FROM task_snapshots WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
+CREATE TRIGGER decisions_blocking_needs_task BEFORE INSERT ON decisions
+ WHEN NEW.blocking = 1 AND NEW.task_ref IS NULL
+ BEGIN SELECT RAISE(ABORT, 'a blocking decision cites the task it blocks'); END;
+CREATE TRIGGER events_launch_by_human BEFORE INSERT ON events
+ WHEN NEW.kind IN ('launch.revoked','launch.restored') AND (NEW.actor_role IS NOT NULL OR NEW.actor_run IS NOT NULL)
+ BEGIN SELECT RAISE(ABORT, 'only the human revokes or restores launching'); END;
+CREATE TRIGGER events_credential_body BEFORE INSERT ON events
+ WHEN NEW.kind GLOB 'credential.*' AND (
+ json_extract(NEW.body, '$.service') IS NULL OR json_extract(NEW.body, '$.service') NOT IN ('gitea','vikunja')
+ OR json_extract(NEW.body, '$.instance') IS NULL)
+ BEGIN SELECT RAISE(ABORT, 'credential events name a service and a role instance'); END;
+CREATE TRIGGER events_task_missing_body BEFORE INSERT ON events
+ WHEN NEW.kind = 'task.missing' AND (
+ json_extract(NEW.body, '$.reason') IS NULL OR json_extract(NEW.body, '$.reason') NOT IN ('moved','not-found','no-access')
+ OR (json_extract(NEW.body, '$.reason') = 'moved' AND json_type(NEW.body, '$.project') IS NOT 'integer'))
+ BEGIN SELECT RAISE(ABORT, 'task.missing carries a reason, and the new project when moved'); END;
+CREATE TRIGGER events_task_subject BEFORE INSERT ON events
+ WHEN NEW.kind GLOB 'task.*' AND (NEW.subject IS NULL OR NOT (NEW.subject GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(NEW.subject, 9) GLOB '*[^0-9/]*' AND NOT substr(NEW.subject, 9) GLOB '*/*/*'))
+ BEGIN SELECT RAISE(ABORT, 'a task event names its task in subject'); END;
+CREATE TRIGGER events_task_created_body BEFORE INSERT ON events
+ WHEN NEW.kind = 'task.created' AND (
+ json_type(NEW.body, '$.request') IS NOT 'text'
+ OR NOT EXISTS (SELECT 1 FROM events WHERE id = json_extract(NEW.body, '$.request')
+ AND kind = 'human.input' AND business = NEW.business)
+ OR json_type(NEW.body, '$.requirement') IS NOT 'text'
+ OR NOT json_extract(NEW.body, '$.requirement') GLOB 'REQ-[A-Z]*-[1-9]*'
+ OR json_extract(NEW.body, '$.requirement') GLOB 'REQ-*[^A-Z0-9-]*')
+ BEGIN SELECT RAISE(ABORT, 'task.created cites the human.input that asked for it and a requirement id'); END;
+CREATE VIEW launch_state AS
+ SELECT business, CASE kind WHEN 'launch.revoked' THEN 'revoked' ELSE 'allowed' END AS state, at
+ FROM events e WHERE kind IN ('launch.revoked','launch.restored')
+ AND seq = (SELECT max(seq) FROM events WHERE business = e.business AND kind IN ('launch.revoked','launch.restored'));
+CREATE VIEW task_external_changes AS
+ SELECT p.business, p.task_ref, p.seq, p.via, p.updated, p.digest, ls.digest AS self_digest
+ FROM task_snapshots p
+ LEFT JOIN task_snapshots ls ON ls.seq = (SELECT max(seq) FROM task_snapshots
+ WHERE business = p.business AND task_ref = p.task_ref AND source = 'self')
+ WHERE p.source = 'poll'
+ AND p.seq = (SELECT max(seq) FROM task_snapshots WHERE business = p.business AND task_ref = p.task_ref)
+ AND (ls.seq IS NULL OR (p.updated >= ls.updated AND p.read_at > ls.at AND p.digest <> ls.digest));
+CREATE VIEW task_current AS
+ SELECT s.seq, s.at, s.business, s.task_ref, s.updated, s.etag, s.digest, s.fields, s.source, s.via, s.read_at, s.role, s.run
+ FROM task_snapshots s
+ WHERE s.seq = (SELECT max(c.seq) FROM task_snapshots c
+ WHERE c.business = s.business AND c.task_ref = s.task_ref
+ AND (c.source = 'self'
+ OR c.read_at > coalesce((SELECT ls.at FROM task_snapshots ls
+ WHERE ls.business = c.business AND ls.task_ref = c.task_ref AND ls.source = 'self'
+ ORDER BY ls.seq DESC LIMIT 1), '')));
+CREATE VIEW tasks_open AS
+ SELECT s.business, s.task_ref, json_extract(s.fields, '$.bucket') AS bucket, s.seq
+ FROM task_current s
+ WHERE json_type(s.fields, '$.gone') IS NULL
+ AND json_extract(s.fields, '$.done') = 0;
+CREATE VIEW urgent_inbox AS
+ SELECT d.id, d.business, d.task_ref, d.question, d.at
+ FROM decisions d
+ WHERE d.class = 'gated' AND d.blocking = 1
+ AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired'));
+CREATE TABLE task_comment_cursors (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ business TEXT NOT NULL,
+ task_id INTEGER NOT NULL CHECK (task_id > 0),
+ last_comment INTEGER NOT NULL CHECK (last_comment > 0)
+) STRICT;
+CREATE INDEX task_comment_cursors_latest ON task_comment_cursors (business, task_id, seq);
+CREATE TRIGGER task_comment_cursors_no_update BEFORE UPDATE ON task_comment_cursors
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_no_delete BEFORE DELETE ON task_comment_cursors
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_no_replace BEFORE INSERT ON task_comment_cursors
+ WHEN EXISTS (SELECT 1 FROM task_comment_cursors WHERE seq = NEW.seq)
+ BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
+CREATE TRIGGER task_comment_cursors_forward BEFORE INSERT ON task_comment_cursors
+ WHEN NEW.last_comment <= COALESCE((SELECT last_comment FROM task_comment_cursors
+ WHERE business = NEW.business AND task_id = NEW.task_id ORDER BY seq DESC LIMIT 1), 0)
+ BEGIN SELECT RAISE(ABORT, 'comment cursor must advance'); END;
+CREATE TABLE projects (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ name TEXT NOT NULL,
+ repo TEXT,
+ created_by TEXT NOT NULL,
+ UNIQUE (business, name)
+) STRICT;
+CREATE TABLE threads (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ id TEXT NOT NULL UNIQUE,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ project TEXT NOT NULL REFERENCES projects(id),
+ instance TEXT NOT NULL,
+ owner TEXT NOT NULL,
+ name TEXT NOT NULL,
+ UNIQUE (project, name)
+) STRICT;
+CREATE TABLE thread_claims (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ thread TEXT NOT NULL REFERENCES threads(id),
+ op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')),
+ holder_run TEXT NOT NULL,
+ harness TEXT NOT NULL, address TEXT,
+ by TEXT NOT NULL, reason TEXT,
+ decision TEXT
+) STRICT;
+CREATE TABLE message_threads (
+ message TEXT PRIMARY KEY REFERENCES messages(id),
+ thread TEXT NOT NULL REFERENCES threads(id)
+) STRICT;
+CREATE TABLE thread_opens (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ thread TEXT NOT NULL REFERENCES threads(id),
+ by TEXT NOT NULL
+) STRICT;
+CREATE INDEX thread_claims_thread ON thread_claims (thread, seq);
+CREATE INDEX message_threads_thread ON message_threads (thread);
+CREATE TRIGGER threads_project_business BEFORE INSERT ON threads
+ WHEN (SELECT business FROM projects WHERE id = NEW.project) IS NOT NEW.business
+ BEGIN SELECT RAISE(ABORT, 'a thread belongs to its project''s business'); END;
+CREATE TRIGGER thread_one_holder BEFORE INSERT ON thread_claims
+ WHEN NEW.op = 'claim' AND (SELECT op FROM thread_claims WHERE thread = NEW.thread ORDER BY seq DESC LIMIT 1) = 'claim'
+ BEGIN SELECT RAISE(ABORT, 'thread already held'); END;
+CREATE TRIGGER thread_one_per_run BEFORE INSERT ON thread_claims
+ WHEN NEW.op = 'claim' AND EXISTS (SELECT 1 FROM thread_claims c WHERE c.holder_run = NEW.holder_run AND c.business = NEW.business
+ AND c.seq = (SELECT max(seq) FROM thread_claims WHERE thread = c.thread) AND c.op = 'claim')
+ BEGIN SELECT RAISE(ABORT, 'run already holds a thread'); END;
+CREATE TRIGGER thread_release_by_holder BEFORE INSERT ON thread_claims
+ WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM thread_claims WHERE thread = NEW.thread ORDER BY seq DESC LIMIT 1),'') <> 'claim'
+ BEGIN SELECT RAISE(ABORT, 'thread is not held'); END;
+CREATE TRIGGER thread_release_same_run BEFORE INSERT ON thread_claims
+ WHEN NEW.op = 'release' AND (SELECT holder_run FROM thread_claims WHERE thread = NEW.thread ORDER BY seq DESC LIMIT 1) <> NEW.holder_run
+ BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END;
+CREATE TRIGGER thread_revoke_needs_decision BEFORE INSERT ON thread_claims
+ WHEN NEW.op = 'revoke' AND NEW.decision IS NULL AND NEW.reason IS NULL
+ BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision or a reason'); END;
+CREATE TRIGGER thread_claims_business BEFORE INSERT ON thread_claims
+ WHEN (SELECT business FROM threads WHERE id = NEW.thread) IS NOT NEW.business
+ BEGIN SELECT RAISE(ABORT, 'a thread claim names the thread''s business'); END;
+CREATE TRIGGER message_threads_match BEFORE INSERT ON message_threads
+ WHEN NOT EXISTS (SELECT 1 FROM messages m JOIN threads t ON t.id = NEW.thread
+ WHERE m.id = NEW.message AND m.business = t.business AND t.instance IN (m.to_role, m.from_role))
+ BEGIN SELECT RAISE(ABORT, 'a threaded message is to or from the thread''s instance'); END;
+CREATE TRIGGER delivery_same_thread BEFORE INSERT ON deliveries
+ WHEN NEW.op IN ('routed','delivered') AND (NEW.op = 'delivered' OR NEW.holder_run IS NOT NULL)
+ AND EXISTS (SELECT 1 FROM message_threads mt JOIN messages m ON m.id = mt.message JOIN threads t ON t.id = mt.thread
+ WHERE mt.message = NEW.message AND m.to_role = t.instance
+ AND NOT EXISTS (SELECT 1 FROM thread_claims c WHERE c.seq = (SELECT max(seq) FROM thread_claims WHERE thread = t.id)
+ AND c.op = 'claim' AND c.holder_run IS NEW.holder_run))
+ BEGIN SELECT RAISE(ABORT, 'a threaded message goes only to its thread''s holder'); END;
+CREATE TRIGGER projects_no_update BEFORE UPDATE ON projects BEGIN SELECT RAISE(ABORT, 'projects is append-only'); END;
+CREATE TRIGGER projects_no_delete BEFORE DELETE ON projects BEGIN SELECT RAISE(ABORT, 'projects is append-only'); END;
+CREATE TRIGGER projects_no_replace BEFORE INSERT ON projects WHEN EXISTS (SELECT 1 FROM projects WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'projects is append-only'); END;
+CREATE TRIGGER threads_no_update BEFORE UPDATE ON threads BEGIN SELECT RAISE(ABORT, 'threads is append-only'); END;
+CREATE TRIGGER threads_no_delete BEFORE DELETE ON threads BEGIN SELECT RAISE(ABORT, 'threads is append-only'); END;
+CREATE TRIGGER threads_no_replace BEFORE INSERT ON threads WHEN EXISTS (SELECT 1 FROM threads WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'threads is append-only'); END;
+CREATE TRIGGER thread_claims_no_update BEFORE UPDATE ON thread_claims BEGIN SELECT RAISE(ABORT, 'thread_claims is append-only'); END;
+CREATE TRIGGER thread_claims_no_delete BEFORE DELETE ON thread_claims BEGIN SELECT RAISE(ABORT, 'thread_claims is append-only'); END;
+CREATE TRIGGER thread_claims_no_replace BEFORE INSERT ON thread_claims WHEN EXISTS (SELECT 1 FROM thread_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'thread_claims is append-only'); END;
+CREATE TRIGGER message_threads_no_update BEFORE UPDATE ON message_threads BEGIN SELECT RAISE(ABORT, 'message_threads is append-only'); END;
+CREATE TRIGGER message_threads_no_delete BEFORE DELETE ON message_threads BEGIN SELECT RAISE(ABORT, 'message_threads is append-only'); END;
+CREATE TRIGGER message_threads_no_replace BEFORE INSERT ON message_threads WHEN EXISTS (SELECT 1 FROM message_threads WHERE message = NEW.message) BEGIN SELECT RAISE(ABORT, 'message_threads is append-only'); END;
+CREATE TRIGGER thread_opens_no_update BEFORE UPDATE ON thread_opens BEGIN SELECT RAISE(ABORT, 'thread_opens is append-only'); END;
+CREATE TRIGGER thread_opens_no_delete BEFORE DELETE ON thread_opens BEGIN SELECT RAISE(ABORT, 'thread_opens is append-only'); END;
+CREATE TRIGGER thread_opens_no_replace BEFORE INSERT ON thread_opens WHEN EXISTS (SELECT 1 FROM thread_opens WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'thread_opens is append-only'); END;
+CREATE TABLE instance_names (
+ seq INTEGER PRIMARY KEY AUTOINCREMENT,
+ at TEXT NOT NULL,
+ business TEXT NOT NULL,
+ instance TEXT NOT NULL,
+ name TEXT NOT NULL,
+ by TEXT NOT NULL
+) STRICT;
+CREATE TRIGGER instance_names_unique BEFORE INSERT ON instance_names
+ WHEN EXISTS (SELECT 1 FROM instance_names n WHERE n.business = NEW.business AND n.name = NEW.name
+ AND n.instance <> NEW.instance
+ AND n.seq = (SELECT max(seq) FROM instance_names WHERE business = n.business AND instance = n.instance))
+ BEGIN SELECT RAISE(ABORT, 'an instance name is unique in its business'); END;
+CREATE TRIGGER instance_names_no_update BEFORE UPDATE ON instance_names BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END;
+CREATE TRIGGER instance_names_no_delete BEFORE DELETE ON instance_names BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END;
+CREATE TRIGGER instance_names_no_replace BEFORE INSERT ON instance_names WHEN EXISTS (SELECT 1 FROM instance_names WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END;
diff --git a/packages/bus/tests/name-migration.test.mjs b/packages/bus/tests/name-migration.test.mjs
index 3c3e361f9..d63a23c67 100644
--- a/packages/bus/tests/name-migration.test.mjs
+++ b/packages/bus/tests/name-migration.test.mjs
@@ -55,7 +55,7 @@ test('3f copy migrates to 3g preserving every old column/row, with an empty inst
assert.throws(() => new Store(root), /schema-mismatch/);
const store = new Store(root, { migrateSchema: true });
try {
- assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3g');
+ assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3h');
assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null });
assert.equal(store.get('SELECT count(*) AS n FROM instance_names').n, 0);
assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: 'docs/spec.md', spec_commit: 'c'.repeat(40) });
diff --git a/packages/bus/tests/schema.test.mjs b/packages/bus/tests/schema.test.mjs
index a975d8fbf..153ee1776 100644
--- a/packages/bus/tests/schema.test.mjs
+++ b/packages/bus/tests/schema.test.mjs
@@ -518,7 +518,8 @@ test('v3b prototype refusals, views and append-only mutations', (t) => {
for (const label of refused) assert.ok(tested.has(label), label);
// Row 121 (#1615) adds 5 tables and 24 triggers for projects and threads; threads.test.mjs tests them.
// Row 141 (#1635) adds instance_names and its 4 triggers; names.test.mjs tests them.
- assert.equal(count('table') - 1, 15);
- assert.equal(count('trigger'), 67);
+ // Row 124 adds usage_records and its three append-only triggers.
+ assert.equal(count('table') - 1, 16);
+ assert.equal(count('trigger'), 70);
assert.equal(count('view'), 5);
});
diff --git a/packages/bus/tests/spec-migration.test.mjs b/packages/bus/tests/spec-migration.test.mjs
index 6034c4598..d108bf0a2 100644
--- a/packages/bus/tests/spec-migration.test.mjs
+++ b/packages/bus/tests/spec-migration.test.mjs
@@ -53,7 +53,7 @@ test('3e copy migrates through 3f to 3g preserving every old column/row and addi
assert.throws(() => new Store(root), /schema-mismatch/);
const store = new Store(root, { migrateSchema: true });
try {
- assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3g');
+ assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3h');
assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: null, spec_commit: null });
assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok');
assert.deepEqual(store.all('PRAGMA foreign_key_check'), []);
diff --git a/packages/bus/tests/thread-migration.test.mjs b/packages/bus/tests/thread-migration.test.mjs
index c0c80db67..6b2e914dc 100644
--- a/packages/bus/tests/thread-migration.test.mjs
+++ b/packages/bus/tests/thread-migration.test.mjs
@@ -49,7 +49,7 @@ async function database3d(t) {
const source = readFileSync(here('../src/store.mjs'), 'utf8');
const pinned = source
.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3d).digest('hex')}';`)
- .replace("export const SCHEMA_VERSION = '3g';", "export const SCHEMA_VERSION = '3d';")
+ .replace("export const SCHEMA_VERSION = '3h';", "export const SCHEMA_VERSION = '3d';")
.replace(/\}, \{\n {2}from: '3d',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables');
assert.ok(!pinned.includes('3d-3e'), 'the 3d engine has no 3d→3e step');
writeFileSync(join(moduleRoot, 'src/store.mjs'), pinned);
@@ -101,7 +101,7 @@ test('REQ-THREAD-1 REQ-ORG-2: the 3d→3e migration adds the thread tables and l
assert.throws(() => new Store(root), /schema-mismatch/);
const store = new Store(root, { migrateSchema: true });
t.after(() => store.close());
- assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3g');
+ assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3h');
for (const table of ['projects', 'threads', 'thread_claims', 'message_threads', 'thread_opens'])
assert.equal(store.get(`SELECT count(*) n FROM ${table}`).n, 0, table);
// Parked: the undelivered ones are counted per business and role, and none is changed or given a thread.
@@ -161,5 +161,5 @@ test('REQ-THREAD-1: the offline command prints the kept-message counts before th
const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname;
const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' });
assert.equal(migrated.status, 0, migrated.stderr);
- assert.equal(migrated.stdout, [...REPORT, 'bus schema 3g ready', ''].join('\n'));
+ assert.equal(migrated.stdout, [...REPORT, 'bus schema 3h ready', ''].join('\n'));
});
diff --git a/packages/bus/tests/usage-broker.test.mjs b/packages/bus/tests/usage-broker.test.mjs
new file mode 100644
index 000000000..c1556b7c5
--- /dev/null
+++ b/packages/bus/tests/usage-broker.test.mjs
@@ -0,0 +1,30 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { Store } from '../src/store.mjs';
+import { Broker } from '../src/broker.mjs';
+test('REQ-LIMIT-4 usage protocol binds run identity, requires a delivered turn and isolates businesses', async t => {
+ const root = mkdtempSync(join(tmpdir(), 'usage-broker-'));
+ const store = new Store(root); t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
+ const businesses = { demo: { id: 'demo', human: 'jason', roles: { coder: { definition: 'coder', authority: { withinRole: [], crossRole: [] } } }, arbiters: { technical: 'coder', delivery: 'coder' } } };
+ const recorded = [];
+ const broker = new Broker({ store, businesses, usage: {
+ collect: async (...args) => { recorded.push(args); return { recorded: true }; },
+ view: business => ({ business }),
+ } });
+ const coder = broker.bindLaunch({ business: 'demo', role: 'coder', run: 'run1', harness: 'pi', provider: 'fixture', model: 'custom-model' });
+ broker.request(coder, { verb: 'role.claim' });
+ const human = broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
+ const { id } = broker.request(human, { verb: 'message.send', args: { to: 'coder', body: 'Work' } });
+ await assert.rejects(broker.request(coder, { verb: 'usage.begin', args: { turn: id } }), { code: 'not-recipient' });
+ broker.request(coder, { verb: 'message.receive' });
+ for (const phase of ['begin', 'end']) assert.deepEqual(await broker.request(coder, { verb: `usage.${phase}`, args: { turn: id } }), { recorded: true });
+ assert.equal(recorded.length, 2); assert.equal(recorded[0][0].run, 'run1');
+ assert.equal(recorded[0][0].provider, 'fixture'); assert.equal(recorded[0][0].model, 'custom-model');
+ await assert.rejects(broker.request(coder, { verb: 'usage.end', args: { turn: id, root: '/etc' } }), { code: 'invalid-request' });
+ await assert.rejects(broker.request(human, { verb: 'usage.begin', args: { turn: id } }), { code: 'agent-required' });
+ assert.deepEqual(broker.request(broker.bindReader({ business: 'demo' }), { verb: 'usage' }), { business: 'demo' });
+ assert.throws(() => broker.request(human, { verb: 'usage', args: { business: 'other' } }), /invalid-request/);
+});
diff --git a/packages/bus/tests/usage-migration.test.mjs b/packages/bus/tests/usage-migration.test.mjs
new file mode 100644
index 000000000..421b1d417
--- /dev/null
+++ b/packages/bus/tests/usage-migration.test.mjs
@@ -0,0 +1,71 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { chmodSync, copyFileSync, mkdirSync, mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { createHash } from 'node:crypto';
+import { DatabaseSync } from 'node:sqlite';
+import { Store, schemaDigest } from '../src/store.mjs';
+// Row 124 (#1618, REQ-LIMIT-4): the 3g→3h step on a copy of a 3g store built here from the pinned 3g
+// schema (4ebef2284's schema.sql) and fixture rows. No live store is read or copied.
+const AT = '2026-10-16T00:00:00.000Z';
+function source3g(t) {
+ const root = mkdtempSync(join(tmpdir(), 'usage-migrate-'));
+ t.after(() => rmSync(root, { recursive: true, force: true }));
+ mkdirSync(join(root, 'bus'), { mode: 0o700 });
+ const path = join(root, 'bus/bus.sqlite'), db = new DatabaseSync(path);
+ const ddl = readFileSync(new URL('./fixtures/schema-3g.sql', import.meta.url), 'utf8');
+ assert.equal(createHash('sha256').update(ddl).digest('hex'), 'a79b3dbe6f4dfba10871c01ad8513ea290071eb8d8af2b87be6db52dbcf23428');
+ db.exec(ddl);
+ const run = (sql, ...args) => db.prepare(sql).run(...args);
+ run('INSERT INTO meta VALUES (?,?)', 'schema_version', '3g');
+ run('INSERT INTO meta VALUES (?,?)', 'schema_digest', schemaDigest(db));
+ run('INSERT INTO meta VALUES (?,?)', 'comment_cursor_epoch', AT);
+ run('INSERT INTO events(id,at,business,kind,body) VALUES(?,?,?,?,?)', 'event', AT, 'demo', 'action.allowed', '{}');
+ run('INSERT INTO role_claims(at,business,role,op,holder_run,harness,by) VALUES(?,?,?,?,?,?,?)', AT, 'demo', 'coder', 'claim', 'run', 'pi', 'run');
+ run('INSERT INTO decisions(id,at,business,raised_by_role,raised_by_run,class,action,route_to,question,options,recommendation,blocking,spec_path,spec_commit) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)',
+ 'decision', AT, 'demo', 'coder', 'run', 'gated', 'deploy', 'human', '?', '[{"key":"yes","text":"Yes"},{"key":"no","text":"No"}]', 'yes', 0, 'docs/spec.md', 'c'.repeat(40));
+ run('INSERT INTO decision_events(decision,at,op,by,choice) VALUES(?,?,?,?,?)', 'decision', AT, 'resolved', 'jason', 'yes');
+ run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)', 'message', AT, 'demo', 'human', 'human', 'assistant', 'INFO', 'kept');
+ run('INSERT INTO deliveries(message,at,op) VALUES(?,?,?)', 'message', AT, 'routed');
+ run('INSERT INTO task_snapshots(at,business,task_ref,updated,digest,fields,source,role,run) VALUES(?,?,?,?,?,?,?,?,?)', AT, 'demo', 'vikunja:1/1', AT, 'a'.repeat(64), '{"bucket":1,"done":false}', 'self', 'coder', 'run');
+ run('INSERT INTO task_comment_cursors(business,task_id,last_comment) VALUES(?,?,?)', 'demo', 1, 8);
+ run('INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)', 'project', AT, 'demo', 'Work', 'stack', 'jason');
+ run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', 'thread', AT, 'demo', 'project', 'assistant', 'jason', 'One');
+ run('INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,by) VALUES(?,?,?,?,?,?,?)', AT, 'demo', 'thread', 'claim', 'assistant-run', 'pi', 'assistant-run');
+ run('INSERT INTO message_threads VALUES(?,?)', 'message', 'thread');
+ run('INSERT INTO thread_opens(at,thread,by) VALUES(?,?,?)', AT, 'thread', 'jason');
+ run('INSERT INTO instance_names(at,business,instance,name,by) VALUES(?,?,?,?,?)', AT, 'demo', 'coder', 'Builder', 'jason');
+ db.close(); chmodSync(path, 0o600);
+ const copy = join(root, 'untouched-3g.sqlite'); copyFileSync(path, copy);
+ return { root, path, copy };
+}
+function snapshot(path, columns) {
+ const db = new DatabaseSync(path, { readOnly: true });
+ try {
+ columns ??= Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all()
+ .map(({ name }) => [name, db.prepare(`PRAGMA table_info(${name})`).all().map(c => c.name)]));
+ const rows = Object.fromEntries(Object.entries(columns).map(([table, keys]) => [table,
+ db.prepare(`SELECT ${keys.join(',')} FROM ${table} ORDER BY rowid`).all()
+ .filter(r => table !== 'meta' || !['schema_version', 'schema_digest'].includes(r.key))]));
+ return { columns, rows };
+ } finally { db.close(); }
+}
+test('REQ-LIMIT-4 3g copy migrates to 3h preserving every old column, row and sequence; new usage table is empty', t => {
+ const { root, path, copy } = source3g(t), before = snapshot(path), untouched = readFileSync(copy);
+ assert.throws(() => new Store(root), /schema-mismatch/);
+ const store = new Store(root, { migrateSchema: true });
+ try {
+ assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3h');
+ assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null });
+ assert.equal(store.get('SELECT count(*) AS n FROM usage_records').n, 0);
+ assert.equal(store.get('SELECT name FROM instance_names').name, 'Builder');
+ assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: 'docs/spec.md', spec_commit: 'c'.repeat(40) });
+ assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok');
+ assert.deepEqual(store.all('PRAGMA foreign_key_check'), []);
+ } finally { store.close(); }
+ assert.deepEqual(snapshot(path, before.columns).rows, before.rows);
+ assert.deepEqual(readFileSync(copy), untouched);
+ assert.deepEqual(snapshot(copy).rows, before.rows);
+ new Store(root).close();
+});
diff --git a/packages/bus/tests/usage-runtime.test.mjs b/packages/bus/tests/usage-runtime.test.mjs
new file mode 100644
index 000000000..1b3ccc122
--- /dev/null
+++ b/packages/bus/tests/usage-runtime.test.mjs
@@ -0,0 +1,36 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, mkdirSync, writeFileSync, rmSync, readFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { startBroker } from '../src/runtime.mjs';
+import { Client } from '../src/client.mjs';
+import { startTimeOf } from '../../seat/src/proc.mjs';
+test('REQ-PLUG-1 runtime reaches transcript only through loader, over the real bus socket', async t => {
+ const root = mkdtempSync(join(tmpdir(), 'usage-runtime-'));
+ let runtime;
+ t.after(async () => { await runtime?.close(); rmSync(root, { recursive: true, force: true }); });
+ const dir = join(root, 'launches/demo/run1/session'); mkdirSync(dir, { recursive: true });
+ const business = { id: 'demo', human: 'jason', plugins: { 'usage-source': ['transcript'] },
+ roles: { coder: { authority: { withinRole: [], crossRole: [] } } }, arbiters: { technical: 'coder', delivery: 'coder' } };
+ runtime = await startBroker({ dataRoot: root, businesses: { demo: business }, launches: [{
+ business: 'demo', role: 'coder', run: 'run1', harness: 'pi', provider: 'other-provider', model: 'other-model',
+ pid: process.pid, startTime: startTimeOf(process.pid),
+ }] });
+ const client = new Client({ path: runtime.path, cap: runtime.launches[0].cap });
+ await client.call('role.claim');
+ const human = runtime.broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
+ const { id } = runtime.broker.request(human, { verb: 'message.send', args: { to: 'coder', body: 'Work' } });
+ await client.call('message.receive');
+ assert.deepEqual(await client.call('usage.begin', { turn: id }), { recorded: true });
+ writeFileSync(join(dir, 'session.jsonl'), JSON.stringify({ type: 'message', id: 'new', message: {
+ role: 'assistant', usage: { input: 2, output: 3, cacheWrite: 4, cacheRead: 5 },
+ } }) + '\n');
+ assert.deepEqual(await client.call('usage.end', { turn: id }), { recorded: true });
+ const view = await client.call('usage');
+ assert.deepEqual(view.total, { input: 2, output: 3, cacheWrite: 4, cacheRead: 5, unknownTurns: 0 });
+ assert.equal(view.byProvider[0].provider, 'other-provider');
+ const source = readFileSync(new URL('../src/usage.mjs', import.meta.url), 'utf8');
+ assert.match(source, /createLoader/); assert.match(source, /loader.modules\('usage-source'\)/);
+ assert.doesNotMatch(source, /import[^\n]*usage-source\/transcript/);
+});
diff --git a/packages/bus/tests/usage.test.mjs b/packages/bus/tests/usage.test.mjs
new file mode 100644
index 000000000..a9179a6d5
--- /dev/null
+++ b/packages/bus/tests/usage.test.mjs
@@ -0,0 +1,140 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { Store } from '../src/store.mjs';
+import { createUsage } from '../src/usage.mjs';
+async function fixture(t, plugins = { 'usage-source': ['transcript'] }) {
+ const root = mkdtempSync(join(tmpdir(), 'bus-usage-'));
+ t.after(() => rmSync(root, { recursive: true, force: true }));
+ const store = new Store(root); t.after(() => store.close());
+ const logs = [], session = { business: 'demo', role: 'coder', run: 'run1', harness: 'pi', provider: 'fixture', model: 'model' };
+ const dir = join(root, 'launches/demo/run1/session'); mkdirSync(dir, { recursive: true });
+ const usage = await createUsage({ store, dataRoot: root, businesses: { demo: { plugins } }, log: e => logs.push(e) });
+ const put = (id, input = 2) => ({ type: 'message', id, message: { role: 'assistant', usage: { input, output: 3, cacheWrite: 4, cacheRead: 5 } } });
+ const file = join(dir, 'session.jsonl');
+ const save = records => writeFileSync(file, records.map(JSON.stringify).join('\n') + '\n');
+ return { root, store, usage, session, file, save, put, logs };
+}
+test('REQ-LIMIT-4 loader source records new turn kinds, excludes history and retries exactly once', async t => {
+ const f = await fixture(t);
+ f.save([f.put('old', 99)]);
+ await f.usage.collect(f.session, 'begin', 'turn1');
+ f.save([f.put('old', 99), f.put('new')]);
+ await f.usage.collect(f.session, 'end', 'turn1');
+ await f.usage.collect(f.session, 'end', 'turn1');
+ const view = f.usage.view('demo');
+ assert.deepEqual(view.total, { input: 2, output: 3, cacheWrite: 4, cacheRead: 5, unknownTurns: 0 });
+ assert.equal(view.byInstance[0].instance, 'coder'); assert.equal(view.byRun[0].run, 'run1');
+ assert.equal(view.byModel[0].model, 'model'); assert.equal(view.byProvider[0].provider, 'fixture');
+ assert.equal(view.byAccount[0].account, null);
+ assert.equal(f.usage.view('other').byRun.length, 0);
+ assert.equal(f.store.get('SELECT count(*) n FROM usage_records').n, 2);
+});
+for (const kind of ['cut', 'disabled', 'missing-begin', 'no-new-usage', 'rewrite']) test(`REQ-LIMIT-4 ${kind}: unknown is durable and logged next to every total`, async t => {
+ const f = await fixture(t, kind === 'disabled' ? {} : undefined);
+ f.save([f.put('old')]);
+ if (kind !== 'missing-begin') await f.usage.collect(f.session, 'begin', 'turn1');
+ if (kind === 'cut') writeFileSync(f.file, '{');
+ else if (kind === 'rewrite') f.save([f.put('old', 90), f.put('new')]);
+ else if (kind !== 'no-new-usage') f.save([f.put('old'), f.put('new')]);
+ await f.usage.collect(f.session, 'end', 'turn1');
+ const view = f.usage.view('demo');
+ for (const row of [view.total, ...view.byRun, ...view.byInstance, ...view.byModel, ...view.byProvider, ...view.byAccount]) {
+ assert.equal(row.unknownTurns, 1); assert.equal(row.input, 0);
+ }
+ assert.equal(f.logs.filter(e => e.event === 'usage-unknown').length, 1);
+ assert.equal(f.store.get("SELECT input FROM usage_records WHERE phase='end'").input, null);
+});
+test('REQ-LIMIT-4 unfinished turn remains visibly unknown after accounting restarts', async t => {
+ const f = await fixture(t);
+ await f.usage.collect(f.session, 'begin', 'turn1');
+ const again = await createUsage({ store: f.store, dataRoot: f.root, businesses: { demo: { plugins: {} } }, log: () => {} });
+ assert.equal(again.view('demo').total.unknownTurns, 1);
+});
+test('usage records are append-only', async t => {
+ const f = await fixture(t); await f.usage.collect(f.session, 'begin', 'turn1');
+ for (const sql of ['DELETE FROM usage_records', "UPDATE usage_records SET code='changed'"])
+ assert.throws(() => f.store.run(sql), /append-only/);
+});
+
+test('REQ-LIMIT-4 parse failure logs its safe code, never transcript content', async t => {
+ const f = await fixture(t);
+ await f.usage.collect(f.session, 'begin', 'turn1');
+ writeFileSync(f.file, 'PRIVATE-CUT-LINE');
+ await f.usage.collect(f.session, 'end', 'turn1');
+ assert.equal(f.logs.at(-1).code, 'transcript-parse');
+ assert.ok(!JSON.stringify(f.logs).includes('PRIVATE-CUT-LINE'));
+});
+test('REQ-LIMIT-4 resumed run baselines history and exact totals do not overflow', async t => {
+ const f = await fixture(t);
+ for (const [n, run] of [['1', 'run1'], ['2', 'run2']]) {
+ const session = { ...f.session, run };
+ const dir = join(f.root, 'launches/demo', run, 'session'); mkdirSync(dir, { recursive: true });
+ const file = join(dir, 'session.jsonl');
+ await f.usage.collect(session, 'begin', 'turn' + n);
+ writeFileSync(file, JSON.stringify(f.put('new', n === '1' ? Number.MAX_SAFE_INTEGER : 2)) + '\n');
+ await f.usage.collect(session, 'end', 'turn' + n);
+ }
+ assert.equal(f.usage.view('demo').total.input, '9007199254740993');
+});
+
+
+test('REQ-PLUG-1 core has no transcript import and disabled sources do not read any transcript', async t => {
+ const f = await fixture(t, {});
+ await f.usage.collect(f.session, 'begin', 'turn1');
+ f.save([f.put('new', 500)]);
+ await f.usage.collect(f.session, 'end', 'turn1');
+ assert.equal(f.usage.view('demo').total.input, 0);
+ assert.equal(f.usage.view('demo').total.unknownTurns, 1);
+});
+
+test('REQ-LIMIT-4 a resumed thread bills only new assistant messages to its next run', async t => {
+ const f = await fixture(t);
+ const root = join(f.root, 'threads/demo/thread1/session'); mkdirSync(root, { recursive: true });
+ const file = join(root, 'session.jsonl');
+ writeFileSync(file, JSON.stringify(f.put('historical', 900)) + '\n');
+ const session = { ...f.session, run: 'next-run', thread: 'thread1' };
+ await f.usage.collect(session, 'begin', 'turn2');
+ writeFileSync(file, [f.put('historical', 900), f.put('new', 6)].map(JSON.stringify).join('\n') + '\n');
+ await f.usage.collect(session, 'end', 'turn2');
+ const view = f.usage.view('demo');
+ assert.equal(view.total.input, 6); assert.equal(view.byRun[0].run, 'next-run');
+});
+
+test('REQ-LIMIT-4 every enabled source is called; failure is unknown and a partial total stays labelled', async t => {
+ const f = await fixture(t);
+ const { tree } = await import('../../plugins/tests/helpers.mjs');
+ const { copyFileSync } = await import('node:fs');
+ const { pathToFileURL } = await import('node:url');
+ const plugins = tree(t);
+ plugins.write('usage-source', 'meter', `export const interfaceVersion=1;
+ let n=0; export function readTurn(){ return {status:'known', entries:++n===1?{}:{a:{input:2,output:3,cacheWrite:4,cacheRead:5}}}; }`);
+ plugins.write('usage-source', 'broken', `export const interfaceVersion=1;
+ export function readTurn(){throw Error('PRIVATE-ERROR-MUST-NOT-LOG');}`);
+ plugins.write('usage-source', 'invalid', `export const interfaceVersion=1;
+ let n=0; export function readTurn(){return {status:'known', entries:++n===1?{}:{bad:{input:-1,output:0,cacheWrite:0,cacheRead:0}}};}`);
+ const bus = join(plugins.root, 'packages/bus/src'); mkdirSync(bus, { recursive: true });
+ copyFileSync(new URL('../src/usage.mjs', import.meta.url), join(bus, 'usage.mjs'));
+ const { createUsage: create } = await import(pathToFileURL(join(bus, 'usage.mjs')));
+ const usage = await create({ store: f.store, dataRoot: f.root,
+ businesses: { demo: { plugins: { 'usage-source': ['meter', 'broken', 'invalid'] } } }, log: e => f.logs.push(e) });
+ await usage.collect(f.session, 'begin', 'turn1');
+ await usage.collect(f.session, 'end', 'turn1');
+ assert.equal(usage.view('demo').total.input, 2);
+ assert.equal(usage.view('demo').total.unknownTurns, 1);
+ assert.equal(f.store.get("SELECT count(*) n FROM usage_records WHERE phase='end'").n, 3);
+ assert.ok(!JSON.stringify(f.logs).includes('PRIVATE-ERROR-MUST-NOT-LOG'));
+ assert.equal(f.store.get("SELECT code FROM usage_records WHERE source='invalid' AND phase='end'").code, 'source-invalid');
+});
+
+
+test('REQ-LIMIT-4 a turn whose reported sum exceeds safe precision is unknown, not rounded', async t => {
+ const f = await fixture(t);
+ await f.usage.collect(f.session, 'begin', 'turn1');
+ f.save([f.put('one', Number.MAX_SAFE_INTEGER), f.put('two', Number.MAX_SAFE_INTEGER)]);
+ await f.usage.collect(f.session, 'end', 'turn1');
+ assert.equal(f.store.get("SELECT code FROM usage_records WHERE phase='end'").code, 'usage-overflow');
+ assert.equal(f.usage.view('demo').total.unknownTurns, 1);
+});
diff --git a/packages/cli/README.md b/packages/cli/README.md
index 5c40d1e0d..b02cee8e2 100644
--- a/packages/cli/README.md
+++ b/packages/cli/README.md
@@ -810,3 +810,12 @@ provider/model pairs), `families` (old family names to keys), and optional
mappings refuse. Every held role must map too. A migrated file prints unchanged.
Jason installs the output himself in Mos's window with the PM run idle, and
the re-pin follows in the same window (decision 145).
+
+## Token usage
+
+`mosaic usage [--business ID] [--json]` reads the bus's accounting, without
+opening its database or transcripts. Text lists the total and run, instance,
+provider/model, provider and account groups, with input/output/cacheWrite/
+cacheRead and unknown turns beside every total. JSON retains the full view.
+Account is unassigned today. Unknown turns can have used tokens: known sums
+are a best reading of session-editable transcripts, not proof or a limit check.
diff --git a/packages/cli/src/cli.mjs b/packages/cli/src/cli.mjs
index 3163f0fae..7b903e506 100644
--- a/packages/cli/src/cli.mjs
+++ b/packages/cli/src/cli.mjs
@@ -29,7 +29,10 @@ import { quitLine, runApp } from "./app.mjs";
import { readSessions, stopSession } from "../../seat/src/session.mjs";
import { SeatError } from "../../seat/src/seat.mjs";
+import { usageCommand, USAGE_USAGE } from "./usage.mjs";
+
export const USAGE = `usage:
+ ${USAGE_USAGE}
${LIMITS_USAGE}
${PLUGINS_USAGE}
${PROJECT_USAGE}
@@ -539,6 +542,7 @@ export async function main(argv, io = { env: process.env, stdin: process.stdin,
if (verb === "threads") return threadsCommand(rest, io, (p) => callFor(p, io, deps));
if (verb === "thread") return threadCommand(rest, io, (p) => callFor(p, io, deps), formatMessage);
if (verb === "limits") return limitsCommand(rest,io,p=>callFor(p,io,deps));
+ if (verb === "usage") return usageCommand(rest, io, p => callFor(p, io, deps));
if (verb === "plugins") return pluginsCommand(rest, io, (p) => callFor(p, io, deps));
if (verb === "talk") return talk(rest, io, deps);
if (verb === "follow") return talk(rest, io, deps, true);
diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs
index 68b3e7f0b..3a19900e6 100644
--- a/packages/cli/src/launcher.mjs
+++ b/packages/cli/src/launcher.mjs
@@ -320,7 +320,7 @@ export function createLauncher({
const outerStart = startTimeOf(child.pid);
const runner = await runnerOf(child);
if (!outerStart) throw new Error("the session process exited at once");
- const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, pid: child.pid, startTime: outerStart, ...(thread ? { thread } : {}) });
+ const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, provider: model.provider, model: model.model, pid: child.pid, startTime: outerStart, ...(thread ? { thread } : {}) });
bound = true;
entry = {
business: b.id, instance, ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey,
diff --git a/packages/cli/src/usage.mjs b/packages/cli/src/usage.mjs
new file mode 100644
index 000000000..ae212c2cd
--- /dev/null
+++ b/packages/cli/src/usage.mjs
@@ -0,0 +1,23 @@
+import { CliError } from './errors.mjs';
+export const USAGE_USAGE = 'mosaic usage [--business ID] [--json]';
+export async function usageCommand(args, io, callFor) {
+ const parsed = { flags: new Set(), values: {} };
+ for (let i = 0; i < args.length; i++) {
+ const arg = args[i];
+ if (arg === '--json' && !parsed.flags.has(arg)) parsed.flags.add(arg);
+ else if (arg === '--business' && args[i + 1] && !args[i + 1].startsWith('--') && !parsed.values[arg]) parsed.values[arg] = args[++i];
+ else throw new CliError(USAGE_USAGE, 4);
+ }
+ const result = await callFor(parsed)('usage', {});
+ if (parsed.flags.has('--json')) { io.stdout.write(JSON.stringify(result, null, 2) + '\n'); return; }
+ const counts = row => `input=${row.input} output=${row.output} cacheWrite=${row.cacheWrite} cacheRead=${row.cacheRead} unknown=${row.unknownTurns}`;
+ const lines = [`Total: ${counts(result.total)}`];
+ for (const group of ['byRun', 'byInstance', 'byModel', 'byProvider', 'byAccount']) {
+ for (const row of result[group]) {
+ const name = ['run', 'instance', 'provider', 'model', 'account'].filter(key => key in row)
+ .map(key => row[key] ?? '(unassigned)').join(' / ');
+ lines.push(`${group} ${name}: ${counts(row)}`);
+ }
+ }
+ io.stdout.write(lines.join('\n') + '\n');
+}
diff --git a/packages/cli/tests/usage.test.mjs b/packages/cli/tests/usage.test.mjs
new file mode 100644
index 000000000..0ec48abf9
--- /dev/null
+++ b/packages/cli/tests/usage.test.mjs
@@ -0,0 +1,30 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { main } from '../src/cli.mjs';
+import { fixture, io, tmp } from './helpers.mjs';
+const count = { input: 11, output: 3, cacheWrite: 5, cacheRead: 7, unknownTurns: 2 };
+const view = { total: count, byRun: [{ run: 'run1', ...count }], byInstance: [{ instance: 'coder', ...count }],
+ byModel: [{ provider: 'anthropic', model: 'fixture-model', ...count }], byProvider: [{ provider: 'anthropic', ...count }],
+ byAccount: [{ account: null, ...count }] };
+test('REQ-LIMIT-4 mosaic usage shows each token kind and unknown turns beside every total', async t => {
+ const f = fixture(tmp(t), 'acme'), calls = [];
+ const deps = { transport: business => async (verb, args) => { calls.push({ business, verb, args }); return view; } };
+ const text = io(f.env); await main(['usage', '--business', 'acme'], text, deps);
+ for (const expected of ['input=11', 'output=3', 'cacheWrite=5', 'cacheRead=7', 'unknown=2', 'run1', 'coder', 'fixture-model', 'anthropic', '(unassigned)']) assert.ok(text.out.text.includes(expected), expected);
+ assert.equal((text.out.text.match(/unknown=2/g) ?? []).length, 6);
+ const json = io(f.env); await main(['usage', '--business', 'acme', '--json'], json, deps);
+ assert.deepEqual(JSON.parse(json.out.text), view);
+ assert.deepEqual(calls[0], { business: 'acme', verb: 'usage', args: {} });
+ await assert.rejects(main(['usage', '--bad'], io(f.env), deps), { exitCode: 4 });
+});
+
+test('usage is dispatched by scripts/mosaic and the shared read adapter', async () => {
+ const { spawnSync } = await import('node:child_process');
+ const { views } = await import('../../bus/src/views.mjs');
+ const calls = [];
+ const result = await views({ call: (...args) => { calls.push(args); return view; } }).usage();
+ assert.deepEqual(result, view); assert.deepEqual(calls, [['usage']]);
+ const script = new URL('../../../scripts/mosaic', import.meta.url).pathname;
+ const command = spawnSync(script, ['usage', '--bad'], { encoding: 'utf8', env: { PATH: process.env.PATH } });
+ assert.equal(command.status, 4); assert.match(command.stderr, /mosaic usage/);
+});
diff --git a/packages/harness/src/runner.mjs b/packages/harness/src/runner.mjs
index 503ef89cf..820d03c89 100644
--- a/packages/harness/src/runner.mjs
+++ b/packages/harness/src/runner.mjs
@@ -378,9 +378,17 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
unread = `could not read the current state: ${e.code ?? e.message}`;
}
++n;
+ // REQ-LIMIT-4: baseline the transcript before invoking the harness, including resumed history.
+ if (!unread) {
+ try { await client.call("usage.begin", { turn: m.id }); }
+ catch (e) { log(`runner: usage baseline unavailable: ${e.code ?? "failed"}`); }
+ }
const turn = unread
? { ok: false, reason: unread }
: await runTurn({ session, request: `${block}\n${turnRequest(m)}`, n, env: turnEnv, onChild: (c) => (child = c) });
+ // A failed/stopped harness can still have used tokens. Missing data remains unknown.
+ try { await client.call("usage.end", { turn: m.id }); }
+ catch (e) { log(`runner: usage outcome unavailable: ${e.code ?? "failed"}`); }
if (stopping) break loop;
if (!turn.ok) log(`runner: turn ${n} for message ${m.id} failed: ${turn.reason}`);
try {
diff --git a/packages/harness/tests/runner.test.mjs b/packages/harness/tests/runner.test.mjs
index e409f43b0..85fd3ba9c 100644
--- a/packages/harness/tests/runner.test.mjs
+++ b/packages/harness/tests/runner.test.mjs
@@ -90,11 +90,11 @@ test("stateBlock names the launch state with its time and event, and the open ta
assert.match(capped, /Open tasks: 9, the first 1 of cap 1\./);
});
-async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy } = {}) {
+async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy, usage = null } = {}) {
const dir = scratch(t);
const store = new Store(dir);
const demo = launchBy ? { ...businesses.demo, launch: { ...businesses.demo.launch, by: launchBy } } : businesses.demo;
- const broker = new Broker({ store, businesses: { demo }, organization });
+ const broker = new Broker({ store, businesses: { demo }, organization, usage });
const brokerSocket = join(dir, "bus.sock");
const server = await serve({ broker, path: brokerSocket });
t.after(async () => {
@@ -560,3 +560,25 @@ test("a missing or malformed policy exits 2 before the claim", async (t) => {
assert.match(missing.stderr, /runner: ENOENT/);
assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});
+
+
+test("REQ-LIMIT-4 runner brackets its actual harness turn before the reply and read receipt", async t => {
+ const calls = [];
+ let ctx;
+ ctx = await setup(t, { usage: { collect: async (session, phase, turn) => {
+ calls.push({ phase, turn, run: session.run, started: existsSync(join(ctx.runDir, "turns/0001.stderr")) });
+ return { recorded: true };
+ } } });
+ ctx.inbox = [];
+ const runner = startRunner(ctx.runDir, ctx.cap);
+ t.after(() => runner.child.kill("SIGKILL"));
+ await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
+ const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "account this turn", class: "REQUEST" });
+ await until(replies(ctx));
+ runner.child.kill("SIGTERM");
+ assert.equal((await runner.exited).code, 0);
+ assert.deepEqual(calls, [
+ { phase: 'begin', turn: sent.id, run: 'coder-run', started: false },
+ { phase: 'end', turn: sent.id, run: 'coder-run', started: true },
+ ]);
+});
diff --git a/packages/plugins/README.md b/packages/plugins/README.md
index 8bde4df0b..97703951f 100644
--- a/packages/plugins/README.md
+++ b/packages/plugins/README.md
@@ -26,7 +26,7 @@ do: a usage source reads what a session used after each turn, and a limit
check answers whether a launch may start (`mayLaunch`) and whether a running
session must stop at the end of its turn (`mustStop`). The arguments and
return values of these functions are set by the rows that call them (row 124
-converts the first usage source). No usage-source or limit-check module ships yet.
+converts the first usage source). The transcript usage source ships with row 124; no limit-check module ships yet.
A `search-source` answers the broker's `web.search` (REQ-HARN-2, REQ-PLUG-2, decision 0b81b0c5): `search(settings, query, deps, { signal })` resolves `{ query, total, results: [{ title, url, snippet }] }` or throws a `WebRefusal`; `signal` is an `AbortSignal` the broker fires at its deadline, and a module must stop its request when it does. The kind switches on one module, and the module's name is the source: `searxng-url` queries the operator's SearXNG at the business var `web.searxng`; `searxng-bundled` queries the SearXNG the Stack installs, at a fixed loopback address (the install is REQ-INST-1's first component); `harness-builtin` is served by the harness itself: a granted claude-code role gets WebSearch and WebFetch, the broker serves no web verbs, and `search` only refuses. The choice and every change to it are the `plugins` setting, logged in `changes.jsonl` by the plug-in setter.
@@ -122,3 +122,12 @@ Limits of the checks:
until the process restarts.
- The loader is not wired into any host process yet. The rows that call
the first two kinds (124, 125, 126) wire it.
+
+## Transcript usage source (row 124)
+
+`usage-source/transcript` now ships interface 1's `readTurn({harness, root})`.
+The bus brackets each actual harness turn, reads all sources through the loader,
+and appends baselines and outcomes. See that module's README for JSONL formats,
+security/resource bounds and the best-reading limitation. No limit check ships
+with this row. Enable transcript through the human plug-in settings writer;
+a restart/re-pin remains a separate authorized window.
diff --git a/packages/plugins/tests/transcript.test.mjs b/packages/plugins/tests/transcript.test.mjs
new file mode 100644
index 000000000..adc08e39a
--- /dev/null
+++ b/packages/plugins/tests/transcript.test.mjs
@@ -0,0 +1,138 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, mkdirSync, writeFileSync, rmSync, symlinkSync, linkSync } from 'node:fs';
+import fs, { appendFileSync } from 'node:fs';
+import { syncBuiltinESMExports } from 'node:module';
+import { join } from 'node:path';
+import { tmpdir } from 'node:os';
+import { readTurn } from '../usage-source/transcript/index.mjs';
+function fixture(t, harness = 'pi') {
+ const root = mkdtempSync(join(tmpdir(), 'usage-source-'));
+ t.after(() => rmSync(root, { recursive: true, force: true }));
+ const dir = harness === 'pi' ? root : join(root, 'project');
+ mkdirSync(dir, { recursive: true });
+ const file = join(dir, 'session.jsonl');
+ const pi = { type: 'message', id: 'assistant1', message: { role: 'assistant', usage: { input: 12, output: 3, cacheWrite: 4, cacheRead: 5 } } };
+ const claude = { type: 'assistant', uuid: 'record1', message: { id: 'assistant1', role: 'assistant', usage: { input_tokens: 12, output_tokens: 3, cache_creation_input_tokens: 4, cache_read_input_tokens: 5 } } };
+ const entry = harness === 'pi' ? pi : claude;
+ const put = value => writeFileSync(file, JSON.stringify(value) + '\n');
+ put(entry);
+ return { root, file, entry, put, read: () => readTurn({ harness, root }) };
+}
+for (const harness of ['pi', 'claude-code']) test(`REQ-LIMIT-4 ${harness}: reports all token kinds and deduplicates assistant ids`, async t => {
+ const f = fixture(t, harness);
+ writeFileSync(f.file, [f.entry, f.entry].map(JSON.stringify).join('\n') + '\n');
+ const result = await f.read();
+ assert.equal(result.status, 'known');
+ assert.deepEqual(Object.values(result.entries), [{ input: 12, output: 3, cacheWrite: 4, cacheRead: 5 }]);
+});
+for (const [name, change, code] of [
+ ['cut line', f => writeFileSync(f.file, '{"type":'), 'transcript-parse'],
+ ['invalid UTF-8', f => writeFileSync(f.file, Buffer.concat([Buffer.from('{"type":"user","text":"'), Buffer.from([0xff]), Buffer.from('"}\n')])), 'transcript-parse'],
+ ['missing input', f => { delete f.entry.message.usage.input; f.put(f.entry); }, 'transcript-usage'],
+ ['negative count', f => { f.entry.message.usage.output = -1; f.put(f.entry); }, 'transcript-usage'],
+ ['fractional count', f => { f.entry.message.usage.output = 0.5; f.put(f.entry); }, 'transcript-usage'],
+ ['unsafe integer', f => { f.entry.message.usage.output = Number.MAX_SAFE_INTEGER + 1; f.put(f.entry); }, 'transcript-usage'],
+ ['file size', f => writeFileSync(f.file, ' '.repeat(8 * 1024 * 1024 + 1)), 'transcript-size'],
+ ['line size', f => writeFileSync(f.file, JSON.stringify({ type: 'user', text: 'x'.repeat(1024 * 1024) }) + '\n'), 'transcript-line'],
+ ['symlink file', f => { const other = join(f.root, 'other'); writeFileSync(other, JSON.stringify(f.entry) + '\n'); rmSync(f.file); symlinkSync(other, f.file); }, 'transcript-path'],
+ ['hardlink file', f => linkSync(f.file, join(f.root, 'linked')), 'transcript-path'],
+]) test(`REQ-LIMIT-4 ${name} is unknown, never zero or content`, async t => {
+ const f = fixture(t); change(f);
+ assert.deepEqual(await f.read(), { status: 'unknown', code });
+});
+test('REQ-LIMIT-4 a symlink directory anywhere in the root refuses', async t => {
+ const f = fixture(t), alias = join(f.root, 'alias');
+ symlinkSync(f.root, alias);
+ assert.deepEqual(await readTurn({ harness: 'pi', root: alias }), { status: 'unknown', code: 'transcript-path' });
+});
+test('REQ-LIMIT-4 bounded directory entries refuse before scanning an arbitrary tree', async t => {
+ const f = fixture(t);
+ for (let n = 0; n < 513; n++) writeFileSync(join(f.root, `ignored-${n}`), '');
+ assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-entries' });
+});
+test('REQ-LIMIT-4 conflicting repeated usage refuses rather than double counting', async t => {
+ const f = fixture(t), second = structuredClone(f.entry); second.message.usage.input++;
+ writeFileSync(f.file, [f.entry, second].map(JSON.stringify).join('\n') + '\n');
+ assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-usage' });
+});
+
+
+test('REQ-LIMIT-4 oversized metadata refuses before reading any transcript bytes', async t => {
+ const f = fixture(t); writeFileSync(f.file, 'x'.repeat(8 * 1024 * 1024 + 1));
+ let reads = 0;
+ const original = fs.readSync;
+ fs.readSync = (...args) => { reads++; return original(...args); };
+ syncBuiltinESMExports();
+ try { assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-size' }); assert.equal(reads, 0); }
+ finally { fs.readSync = original; syncBuiltinESMExports(); }
+});
+test('REQ-LIMIT-4 a file growing after fstat cannot exceed the streaming byte budget', async t => {
+ const f = fixture(t);
+ let grew = false;
+ const original = fs.readSync;
+ fs.readSync = (...args) => {
+ if (!grew) { grew = true; appendFileSync(f.file, 'x'.repeat(8 * 1024 * 1024)); }
+ return original(...args);
+ };
+ syncBuiltinESMExports();
+ try { assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-size' }); }
+ finally { fs.readSync = original; syncBuiltinESMExports(); }
+});
+test('REQ-LIMIT-4 a same-size edit during a read is unknown rather than a stable snapshot', async t => {
+ const f = fixture(t);
+ const original = fs.readSync;
+ let edited = false;
+ fs.readSync = (...args) => {
+ if (!edited) { edited = true; const bytes = fs.readFileSync(f.file); fs.writeFileSync(f.file, bytes); }
+ return original(...args);
+ };
+ syncBuiltinESMExports();
+ try { assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-changed' }); }
+ finally { fs.readSync = original; syncBuiltinESMExports(); }
+});
+test('REQ-LIMIT-4 record count is bounded even when each record is tiny', async t => {
+ const f = fixture(t); writeFileSync(f.file, '{"type":"user"}\n'.repeat(20001));
+ assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-records' });
+});
+test('REQ-LIMIT-4 missing assistant usage, malformed JSON and conflicting identities are unknown', async t => {
+ const f = fixture(t);
+ for (const value of [{ type: 'message', id: 'a', message: { role: 'assistant' } }, { ...f.entry, id: '../bad' }]) {
+ f.put(value); assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-usage' });
+ }
+ writeFileSync(f.file, 'not-json\n'); assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-parse' });
+});
+
+test('REQ-LIMIT-4 caller roots, unsupported harnesses and non-object JSON never become zero usage', async t => {
+ assert.deepEqual(await readTurn({ harness: 'pi', root: 'relative' }), { status: 'unknown', code: 'transcript-path' });
+ const f = fixture(t);
+ assert.deepEqual(await readTurn({ harness: 'other', root: f.root }), { status: 'unknown', code: 'transcript-harness' });
+ for (const value of [null, [], 7]) { f.put(value); assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-parse' }); }
+});
+
+
+test('REQ-LIMIT-4 even a complete JSON record with no final newline is an unknown cut line', async t => {
+ const f = fixture(t); writeFileSync(f.file, JSON.stringify(f.entry));
+ assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-parse' });
+});
+
+
+test('REQ-LIMIT-4 final file opens are nonblocking before hostile file type is known', async t => {
+ const f = fixture(t), original = fs.openSync;
+ const flags = [];
+ fs.openSync = (path, mode, ...args) => {
+ if (String(path).endsWith('/session.jsonl')) flags.push(mode);
+ return original(path, mode, ...args);
+ };
+ syncBuiltinESMExports();
+ try {
+ assert.equal((await f.read()).status, 'known');
+ assert.equal(flags.length, 1);
+ assert.ok(flags[0] & fs.constants.O_NONBLOCK);
+ } finally { fs.openSync = original; syncBuiltinESMExports(); }
+});
+test('REQ-LIMIT-4 invalid transcript entry names refuse before file lookup', async t => {
+ const f = fixture(t);
+ writeFileSync(join(f.root, 'bad name.jsonl'), JSON.stringify(f.entry) + '\n');
+ assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-path' });
+});
diff --git a/packages/plugins/usage-source/transcript/README.md b/packages/plugins/usage-source/transcript/README.md
new file mode 100644
index 000000000..e87c15a12
--- /dev/null
+++ b/packages/plugins/usage-source/transcript/README.md
@@ -0,0 +1,38 @@
+# Transcript usage source
+
+REQ-LIMIT-4, interface version 1. Enable with
+`"plugins": { "usage-source": ["transcript"] }` in the organization file,
+keeping its other plug-in kinds. Human changes use `mosaic plugins set` and
+the current revision. No agent edits the live organization file.
+
+`readTurn({ harness, root })` returns `{status:'known', entries}` or
+`{status:'unknown', code}`. Entries map assistant message identities to
+`{input, output, cacheWrite, cacheRead}`. This is a snapshot; the bus stores a
+baseline and computes the turn's new entries. No known new entry means unknown,
+not zero. The bus attributes records to the bound run/instance/provider/model;
+the account is null. No token limit or enforcement ships here.
+
+For Pi, root is the run's session directory, or a resumed thread's persistent
+session directory. JSONL `message` entries with `message.role === 'assistant'`
+have `id` and `message.usage` input/output/cacheWrite/cacheRead counters.
+For Claude Code, root is `home/.claude/projects`, or the thread's persistent
+`claude-projects` directory: one project-directory level, then session JSONL
+files. Assistant entries have message.id and message.usage input_tokens,
+output_tokens, cache_creation_input_tokens and cache_read_input_tokens. Claude's
+absent optional cache fields mean no reported cache usage; missing required
+input/output is unknown. Duplicate identical identities count once; conflicting
+counters refuse the snapshot.
+
+The source follows no user symlink: root components and project directories
+are opened with directory descriptors and O_NOFOLLOW; final opens add
+O_NONBLOCK. Files must be regular and single-link. Both fstat size and bytes
+actually read are bounded, and edits during a read refuse the snapshot. Limits:
+8 MiB total bytes, 1 MiB per line, 512 directory entries and 20,000 JSON records
+per snapshot. These are parser resource bounds, not policy token ceilings.
+Invalid UTF-8, malformed/cut JSONL or invalid counters return unknown, never
+zero. Reads/logs return no transcript text or parser exception text.
+
+A session can edit its own transcript. Counts are the stack's best reading,
+not proof. Resumed history present before a run's next turn is excluded by its
+baseline. Rewriting that baseline makes the turn unknown. Missing data or a
+source disabled/failed at a restart leaves its pending turn visibly unknown.
diff --git a/packages/plugins/usage-source/transcript/index.mjs b/packages/plugins/usage-source/transcript/index.mjs
new file mode 100644
index 000000000..ce8fc652a
--- /dev/null
+++ b/packages/plugins/usage-source/transcript/index.mjs
@@ -0,0 +1,113 @@
+// REQ-LIMIT-4: hostile transcripts are a best reading, never proof of usage.
+// Linux dirfds anchor every lookup; no caller-provided symlink is followed.
+import { constants, openSync, closeSync, fstatSync, readSync, opendirSync } from 'node:fs';
+import { isAbsolute } from 'node:path';
+
+export const interfaceVersion = 1;
+const MAX_BYTES = 8 * 1024 * 1024;
+const MAX_LINE = 1024 * 1024;
+const MAX_ENTRIES = 512;
+const MAX_RECORDS = 20000;
+const ID = /^[A-Za-z0-9_-]{1,128}$/;
+const NAME = /^[A-Za-z0-9_.-]{1,240}$/;
+const DIR = constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW;
+const FILE = constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK;
+const refuse = code => { throw Object.assign(new Error(code), { code }); };
+const at = (fd, name) => `/proc/self/fd/${fd}/${name}`;
+
+function rootFd(root) {
+ if (typeof root !== 'string' || !isAbsolute(root) || root.split('/').includes('..')) refuse('transcript-path');
+ let fd = openSync('/', DIR);
+ try {
+ for (const part of root.split('/').filter(Boolean)) {
+ const next = openSync(at(fd, part), DIR);
+ closeSync(fd); fd = next;
+ }
+ return fd;
+ } catch (e) { closeSync(fd); throw e; }
+}
+function readFile(parent, name, budget) {
+ const fd = openSync(at(parent, name), FILE);
+ try {
+ const before = fstatSync(fd);
+ if (!before.isFile() || before.nlink !== 1) refuse('transcript-path');
+ if (before.size > MAX_BYTES - budget.bytes) refuse('transcript-size');
+ const parts = [];
+ let total = 0;
+ const buffer = Buffer.alloc(65536);
+ for (;;) {
+ const n = readSync(fd, buffer, 0, Math.min(buffer.length, MAX_BYTES - budget.bytes + 1), null);
+ if (!n) break;
+ total += n; budget.bytes += n;
+ if (budget.bytes > MAX_BYTES) refuse('transcript-size');
+ parts.push(Buffer.from(buffer.subarray(0, n)));
+ }
+ const after = fstatSync(fd);
+ if (before.size !== after.size || total !== after.size || before.mtimeMs !== after.mtimeMs ||
+ before.ctimeMs !== after.ctimeMs) refuse('transcript-changed');
+ return Buffer.concat(parts);
+ } finally { closeSync(fd); }
+}
+function counters(usage, harness) {
+ if (!usage || typeof usage !== 'object' || Array.isArray(usage)) refuse('transcript-usage');
+ const fields = harness === 'pi' ? ['input', 'output', 'cacheWrite', 'cacheRead'] :
+ ['input_tokens', 'output_tokens', 'cache_creation_input_tokens', 'cache_read_input_tokens'];
+ const values = fields.map((key, i) => harness === 'claude-code' && i > 1 ? (usage[key] ?? 0) : usage[key]);
+ if (values.some(n => !Number.isSafeInteger(n) || n < 0)) refuse('transcript-usage');
+ return Object.fromEntries(['input', 'output', 'cacheWrite', 'cacheRead'].map((key, i) => [key, values[i]]));
+}
+function parse(bytes, harness, file, entries, budget) {
+ let text;
+ try { text = new TextDecoder('utf-8', { fatal: true }).decode(bytes); }
+ catch { refuse('transcript-parse'); }
+ if (text && !text.endsWith('\n')) refuse('transcript-parse');
+ for (const line of text.split('\n')) {
+ if (!line.trim()) continue;
+ if (Buffer.byteLength(line) > MAX_LINE) refuse('transcript-line');
+ if (++budget.records > MAX_RECORDS) refuse('transcript-records');
+ let record;
+ try { record = JSON.parse(line); } catch { refuse('transcript-parse'); }
+ if (!record || typeof record !== 'object' || Array.isArray(record)) refuse('transcript-parse');
+ if (harness === 'pi' ? record.type !== 'message' || record.message?.role !== 'assistant' : record.type !== 'assistant') continue;
+ const id = harness === 'pi' ? record.id : record.message?.id;
+ if (typeof id !== 'string' || !ID.test(id)) refuse('transcript-usage');
+ const value = counters(record.message?.usage, harness);
+ const key = `${file}:${id}`;
+ if (Object.hasOwn(entries, key) && JSON.stringify(entries[key]) !== JSON.stringify(value)) refuse('transcript-usage');
+ entries[key] = value;
+ }
+}
+function walk(fd, harness, depth, prefix, entries, budget) {
+ const directory = opendirSync(`/proc/self/fd/${fd}`);
+ try {
+ let entry;
+ while ((entry = directory.readSync())) {
+ if (++budget.entries > MAX_ENTRIES) refuse('transcript-entries');
+ if (!NAME.test(entry.name) || entry.name === '.' || entry.name === '..') refuse('transcript-path');
+ if (harness === 'claude-code' && depth === 0) {
+ // Only the standard project directory / session.jsonl shape is admitted.
+ if (entry.isFile()) continue;
+ const child = openSync(at(fd, entry.name), DIR);
+ try { walk(child, harness, 1, entry.name + '/', entries, budget); }
+ finally { closeSync(child); }
+ } else if (entry.name.endsWith('.jsonl')) {
+ parse(readFile(fd, entry.name, budget), harness, prefix + entry.name, entries, budget);
+ }
+ }
+ } finally { directory.closeSync(); }
+}
+export async function readTurn({ harness, root }) {
+ let fd;
+ try {
+ if (!['pi', 'claude-code'].includes(harness)) refuse('transcript-harness');
+ fd = rootFd(root);
+ const entries = Object.create(null);
+ walk(fd, harness, 0, '', entries, { bytes: 0, entries: 0, records: 0 });
+ return { status: 'known', entries };
+ } catch (e) {
+ // A new run can have no transcript directory before its first turn.
+ if (e.code === 'ENOENT' && fd === undefined) return { status: 'known', entries: {} };
+ const code = typeof e.code === 'string' && e.code.startsWith('transcript-') ? e.code : 'transcript-path';
+ return { status: 'unknown', code };
+ } finally { if (fd !== undefined) closeSync(fd); }
+}
diff --git a/packages/webui/src/public/bus.js b/packages/webui/src/public/bus.js
index abf786a8c..4a2d0c5fe 100644
--- a/packages/webui/src/public/bus.js
+++ b/packages/webui/src/public/bus.js
@@ -585,12 +585,26 @@
try{const response=await fetch('/api/settings',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify(payload),cache:'no-store',redirect:'manual'});if(response.type==='opaqueredirect'){settingsReceipt='Sign in again. Draft kept.';signOut();}else{const result=await response.json();if(response.ok&&/^[a-f0-9]{64}$/.test(result?.revision??'')){settingsInputs=null;settingsSnapshot=result;const prior=last.get('api:settings');if(prior)last.set('api:settings',{...prior,launch:result});settingsReceipt='Launch settings saved.';}else settingsReceipt=result?.sent==='unknown'?'Outcome unknown. Read current settings before trying again. Draft kept.':`${result?.message||'Settings refused.'} Draft kept.`;}}
catch{settingsReceipt='Outcome unknown. Read current settings before trying again. Draft kept.';}finally{settingsSaving=false;if(!out())render(false);}
});
+ function tokenUsage(value) {
+ if (!value?.total) return '<section class="panel" aria-labelledby="token-usage-title"><h2 id="token-usage-title">Token usage</h2><p>Usage unavailable. No count was read.</p></section>';
+ const columns = row => ['input', 'output', 'cacheWrite', 'cacheRead', 'unknownTurns'].map(key => `<td>${txt(row[key])}</td>`).join('');
+ const rows = [`<tr><th scope="row">Total</th>${columns(value.total)}</tr>`];
+ for (const [group, label] of [['byRun', 'Run'], ['byInstance', 'Instance'], ['byModel', 'Model'], ['byProvider', 'Provider'], ['byAccount', 'Account']]) {
+ for (const row of value[group] || []) {
+ const name = ['run', 'instance', 'provider', 'model', 'account'].filter(key => key in row).map(key => row[key] ?? 'unassigned').join(' / ');
+ rows.push(`<tr><th scope="row">${label}: ${txt(name)}</th>${columns(row)}</tr>`);
+ }
+ }
+ return `<section class="panel" aria-labelledby="token-usage-title"><h2 id="token-usage-title">Token usage</h2><p>Best reading of session-editable transcripts, not proof. Unknown turns may have used more tokens. Account is unassigned until account tracking is configured.</p><div class="table-wrap"><table id="token-usage-table"><thead><tr><th scope="col">Subject</th><th scope="col">Input</th><th scope="col">Output</th><th scope="col">Cache write</th><th scope="col">Cache read</th><th scope="col">Unknown turns</th></tr></thead><tbody>${rows.join('')}</tbody></table></div></section>`;
+ }
async function settingsView(get) {
const s = await get.api('settings'), n = s.notifier || {};
+ let usage;
+ try { [usage] = await get('usage'); } catch (err) { if (err.status === 403 || err.code === 'signed-out') throw err; }
const notify = typeof n.binding === 'string'
? `<dl class="kv"><dt>Binding</dt><dd><code>${txt(n.binding)}</code></dd></dl><p class="small muted">The binding name only. Its channels, server and user ids stay in the binding file.</p>`
: empty('No notifier binding to show.', `${n.refused ? `<code>${txt(n.refused)}</code>: ` : ''}${txt(n.message || 'the notifier settings could not be read')}`);
- return `${head('Settings', `${readLine()} Agent limits and appearance.`)}${appearance()}${launchForm(s.launch)}
+ return `${head('Settings', `${readLine()} Agent limits and appearance.`)}${appearance()}${launchForm(s.launch)}${tokenUsage(usage)}
<section aria-labelledby="st-notify"><h2 id="st-notify">Notifications</h2>${notify}</section>
<section aria-labelledby="st-about" class="panel"><h2 id="st-about">About</h2><dl class="kv"><dt>Address</dt><dd><code>${txt(s.address) || 'unknown'}</code></dd><dt>Release</dt><dd>${s.release ? `<code>${txt(s.release)}</code>` : '<span class="muted">unknown: no RELEASE file was read</span>'}</dd><dt>Business</dt><dd>${txt(s.business) || none}</dd><dt>Board</dt><dd><code>${txt(s.board) || 'unknown'}</code></dd></dl></section>`;
}
diff --git a/packages/webui/src/serve.mjs b/packages/webui/src/serve.mjs
index 1249349f8..7c5e74a3c 100644
--- a/packages/webui/src/serve.mjs
+++ b/packages/webui/src/serve.mjs
@@ -49,10 +49,22 @@ async function body(req, max = 4096) {
// Row 57 (#1546) adds `decisions`, also a read, for the Inbox tabs. Row 60
// (#1549) adds `credentials`: what the bus host loaded, metadata only. Row 81
// (#1575) adds `replies`: the messages answering one message, for the send receipt.
-const BUS_VERBS = ['inbox', 'tasks', 'agents', 'trail', 'decisions', 'credentials', 'replies', 'messages'];
+const BUS_VERBS = ['inbox', 'tasks', 'agents', 'trail', 'decisions', 'credentials', 'replies', 'messages', 'usage'];
const DECISION_LISTS = ['roles', 'recent'];
async function busRead(res, bus, verb, search) {
let arg;
+ if (verb === 'usage') {
+ if (search) return json(res, 400, { error: 'invalid-request', message: 'usage takes no arguments' });
+ if (!bus) return json(res, 503, { error: 'not-configured', message: 'the Console has no bus configured' });
+ try {
+ const value = await bus.usage();
+ if (!value || typeof value.total !== 'object' || !Array.isArray(value.byInstance)) throw new BusReadError('invalid-response');
+ return json(res, 200, { rows: [value], at: new Date().toISOString() });
+ } catch (err) {
+ const code = err instanceof BusReadError ? err.code : 'invalid-response';
+ return json(res, busStatus(code), { error: code, message: code });
+ }
+ }
if (verb === 'trail') {
arg = new URLSearchParams(search).get('subject');
if (!subjectOk(arg)) return json(res, 400, { error: 'invalid-request', message: 'subject must be a decision id, message id or task ref' });
diff --git a/packages/webui/tests/usage-browser.test.mjs b/packages/webui/tests/usage-browser.test.mjs
new file mode 100644
index 000000000..fac72a6f3
--- /dev/null
+++ b/packages/webui/tests/usage-browser.test.mjs
@@ -0,0 +1,26 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { browser } from './browser.mjs';
+import { startServer } from '../src/serve.mjs';
+import { close } from './fixture.mjs';
+test('REQ-LIMIT-4 Settings shows all token kinds and unknown turns beside every total, as a read-only separate section', { timeout: 60000 }, async () => {
+ const count = { input: 2, output: 3, cacheWrite: 4, cacheRead: 5, unknownTurns: 7 };
+ const hostile = '<img src=x onerror="window.injected=true">';
+ const usage = { total: count, byRun: [{ run: 'run1', ...count }], byInstance: [{ instance: hostile, ...count }],
+ byModel: [{ provider: 'provider', model: 'model', ...count }], byProvider: [{ provider: 'provider', ...count }], byAccount: [{ account: null, ...count }] };
+ const server = await startServer({ port: 0, board: 'http://127.0.0.1:9',
+ bus: { messages: async () => ({ messages: [], unread: 0 }), usage: async () => usage },
+ reads: { settings: async () => ({ business: 'demo', notifier: {}, launch: { refused: 'not-configured' } }) },
+ });
+ const b = await browser();
+ try {
+ await b.navigate(`http://127.0.0.1:${server.address().port}/#/settings`);
+ await b.evaluate(`(async()=>{for(let n=0;n<200;n++){if(document.querySelector('#token-usage-table'))return;await new Promise(r=>setTimeout(r,30));}throw Error('usage table missing');})()`);
+ const rows = await b.evaluate("[...document.querySelectorAll('#token-usage-table tbody tr')].map(row=>[...row.querySelectorAll('td')].map(c=>c.textContent))");
+ assert.equal(rows.length, 6);
+ for (const row of rows) assert.deepEqual(row, ['2', '3', '4', '5', '7']);
+ assert.equal(await b.evaluate("document.querySelectorAll('#token-usage-table input, #token-usage-table button, #token-usage-table img').length"), 0);
+ assert.equal(await b.evaluate('window.injected===true'), false);
+ assert.ok((await b.evaluate("document.querySelector('#token-usage-title').parentElement.textContent")).includes('not proof'));
+ } finally { await b.close(); await close(server); }
+});
diff --git a/packages/webui/tests/usage.test.mjs b/packages/webui/tests/usage.test.mjs
new file mode 100644
index 000000000..49127d4fd
--- /dev/null
+++ b/packages/webui/tests/usage.test.mjs
@@ -0,0 +1,25 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { startServer } from '../src/serve.mjs';
+import { close } from './fixture.mjs';
+const count = { input: 2, output: 3, cacheWrite: 4, cacheRead: 5, unknownTurns: 1 };
+export const usage = { total: count, byRun: [{ run: 'run1', ...count }], byInstance: [{ instance: 'coder', ...count }],
+ byModel: [{ provider: 'provider', model: 'model', ...count }], byProvider: [{ provider: 'provider', ...count }], byAccount: [{ account: null, ...count }] };
+test('REQ-WEB-1 usage read keeps unknown counts and each token kind; no query or write is accepted', async t => {
+ let calls = 0;
+ const server = await startServer({ port: 0, bus: { usage: async () => { calls++; return usage; } } });
+ t.after(() => close(server));
+ const url = `http://127.0.0.1:${server.address().port}/api/bus/usage`;
+ const response = await fetch(url), body = await response.json();
+ assert.equal(response.status, 200); assert.deepEqual(body.rows, [usage]);
+ assert.equal((await fetch(url + '?business=other')).status, 400);
+ assert.equal((await fetch(url, { method: 'POST' })).status, 405);
+ assert.equal(calls, 1);
+});
+
+test('REQ-WEB-1 malformed usage is refused instead of shown as zero', async t => {
+ const server = await startServer({ port: 0, bus: { usage: async () => ({}) } });
+ t.after(() => close(server));
+ const response = await fetch(`http://127.0.0.1:${server.address().port}/api/bus/usage`);
+ assert.equal(response.status, 502);
+});
diff --git a/scripts/mosaic b/scripts/mosaic
index a85679e2c..22087e2fd 100755
--- a/scripts/mosaic
+++ b/scripts/mosaic
@@ -30,6 +30,6 @@ if [ "${1:-}" = business ]; then
exec node "$REPO/packages/business/src/cli.mjs" "$@"
fi
case "${1:-}" in
- project|limits|inbox|decide|tasks|agents|trail|talk|stop|launches|bus) exec node "$REPO/packages/cli/src/cli.mjs" "$@" ;;
+ usage|project|limits|inbox|decide|tasks|agents|trail|talk|stop|launches|bus) exec node "$REPO/packages/cli/src/cli.mjs" "$@" ;;
esac
exec node "$REPO/packages/seat/src/cli.mjs" "$@"
Reference in new issue View Git Blame Copy Permalink