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 `/projects///`. `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 '

Token usage

Usage unavailable. No count was read.

'; + const columns = row => ['input', 'output', 'cacheWrite', 'cacheRead', 'unknownTurns'].map(key => `${txt(row[key])}`).join(''); + const rows = [`Total${columns(value.total)}`]; + 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(`${label}: ${txt(name)}${columns(row)}`); + } + } + return `

Token usage

Best reading of session-editable transcripts, not proof. Unknown turns may have used more tokens. Account is unassigned until account tracking is configured.

${rows.join('')}
SubjectInputOutputCache writeCache readUnknown turns
`; + } 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' ? `
Binding
${txt(n.binding)}

The binding name only. Its channels, server and user ids stay in the binding file.

` : empty('No notifier binding to show.', `${n.refused ? `${txt(n.refused)}: ` : ''}${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)}

Notifications

${notify}

About

Address
${txt(s.address) || 'unknown'}
Release
${s.release ? `${txt(s.release)}` : 'unknown: no RELEASE file was read'}
Business
${txt(s.business) || none}
Board
${txt(s.board) || 'unknown'}
`; } 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 = ''; + 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" "$@"