bus, cli, webui: token accounting per run, schema 3h usage_records (row 124, #1618)

Rocko's round 2 candidate e09313aba4d155a1c5af7bfb227ec397e96c0d57444782cd0b48593102bcc684 (38 paths, base 23533625d), approved by Dewey in #1618 comment 28923 (queue rev 1757). Packet: agents/rocko/work/queue-124/r2/. - Schema 3g to 3h: packages/bus/src/migrations/3g-3h.sql adds an append-only usage_records table; nothing earlier changes. usage_records is in timedTables. - The transcript usage-source module reads token counts from Pi and Claude Code session transcripts and follows no symlink. - `mosaic usage` in the CLI and a read-only usage section in the Console's Settings (one GET route; 400 on a query string). Decision 123 go: Mos, mos-0737. Activation (3g to 3h) is a migrating run with Jason present, in Mos's combined re-pin; landing changes nothing live. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-18 11:08:17 -05:00
1 parent d523272292
commit 70504d23a4
@@ -653,3 +653,15 @@ record-repository path; the default is
 `<dataRoot>/projects/<business>/<project>/`. `root` remains the code checkout.
 The package README describes schemas and the shared agent record writer; agent
 writes never enter `spec/`, never absorb pending edits and refuse remotes.
+
+### Token usage (row 124)
+
+`mosaic usage [--business ID] [--json]` reads token accounting from the bus.
+Each total, run, instance, model, provider and account group shows input,
+output, cache write, cache read and unknown turns. A malformed or unavailable
+transcript contributes an unknown turn, never a zero-token reading. Accounts
+are unassigned until account tracking ships. Exact totals larger than the
+safe integer range are decimal strings in JSON. Settings has the same
+read-only Token usage section. Counts are a best reading, not proof: the
+session can edit its own transcript. This command changes no setting and
+performs no token-limit enforcement.
@@ -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.
@@ -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;
@@ -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);
@@ -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;
@@ -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: {
@@ -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;
@@ -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',
@@ -61,6 +65,7 @@ const timedTables = [
  'thread_claims',
  'thread_opens',
  'instance_names',
+  'usage_records',
 ];
 const hex = (value) => createHash('sha256').update(value).digest('hex');
 if (hex(ddl) !== SCHEMA_SHA256) throw Error('schema-source-mismatch');
@@ -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 };
+}
@@ -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'),
@@ -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 });
@@ -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);
@@ -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;
@@ -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) });
@@ -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);
 });
@@ -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'), []);
@@ -157,3 +157,17 @@ test('async transactions refuse before invoking their function', async (t) => {
  await Promise.resolve();
  assert.equal(ran, false);
 });
+
+// Row 124 review 28920(a): usage evidence participates in the Store's timestamp invariant.
+test('usage_records refuses non-canonical timestamps atomically', t => {
+  const root = scratch(t), store = new Store(root);
+  t.after(() => store.close());
+  const insert = at => store.run(
+    'INSERT INTO usage_records(at,business,run,instance,turn,source,phase,status,code) VALUES(?,?,?,?,?,?,?,?,?)',
+    at, 'demo', 'run1', 'coder', 'turn1', '@none', 'end', 'unknown', 'no-source',
+  );
+  assert.throws(() => insert('2026-10-04T00:00:00Z'), /invalid-timestamp/);
+  assert.equal(store.get('SELECT count(*) n FROM usage_records').n, 0);
+  insert('2026-10-04T00:00:00.000Z');
+  assert.equal(store.get('SELECT count(*) n FROM usage_records').n, 1);
+});
@@ -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'));
 });
@@ -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/);
+});
@@ -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();
+});
@@ -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/);
+});
@@ -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);
+});
@@ -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.
@@ -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);
@@ -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,
@@ -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');
+}
@@ -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/);
+});
@@ -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 {
@@ -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 },
+  ]);
+});
@@ -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.
@@ -0,0 +1,151 @@
+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 harness of ['pi', 'claude-code']) 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[f.entry.type === 'assistant' ? 'input_tokens' : 'input']; f.put(f.entry); }, 'transcript-usage'],
+  ['negative count', f => { f.entry.message.usage[f.entry.type === 'assistant' ? 'output_tokens' : 'output'] = -1; f.put(f.entry); }, 'transcript-usage'],
+  ['fractional count', f => { f.entry.message.usage[f.entry.type === 'assistant' ? 'output_tokens' : 'output'] = 0.5; f.put(f.entry); }, 'transcript-usage'],
+  ['unsafe integer', f => { f.entry.message.usage[f.entry.type === 'assistant' ? 'output_tokens' : '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 ${harness} ${name} is unknown, never zero or content`, async t => {
+  const f = fixture(t, harness); change(f);
+  assert.deepEqual(await f.read(), { status: 'unknown', code });
+});
+for (const harness of ['pi', 'claude-code']) test(`REQ-LIMIT-4 ${harness} a symlink directory anywhere in the root refuses`, async t => {
+  const f = fixture(t, harness), alias = join(f.root, 'alias');
+  symlinkSync(f.root, alias);
+  assert.deepEqual(await readTurn({ harness, 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' });
+});
+
+// Claude's project-directory descent is separate from root and session-file opens.
+test('REQ-LIMIT-4 claude-code refuses a symlinked project directory', async t => {
+  const f = fixture(t, 'claude-code');
+  const project = join(f.root, 'project');
+  const target = join(f.root, '..', 'project-target-' + f.root.split('/').at(-1));
+  mkdirSync(target);
+  t.after(() => rmSync(target, { recursive: true, force: true }));
+  writeFileSync(join(target, 'session.jsonl'), JSON.stringify(f.entry) + '\n');
+  rmSync(project, { recursive: true });
+  symlinkSync(target, project);
+  assert.deepEqual(await f.read(), { status: 'unknown', code: 'transcript-path' });
+});
@@ -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.
@@ -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); }
+}
@@ -585,12 +585,26 @@
    try{const response=await fetch('/api/settings',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify(payload),cache:'no-store',redirect:'manual'});if(response.type==='opaqueredirect'){settingsReceipt='Sign in again. Draft kept.';signOut();}else{const result=await response.json();if(response.ok&&/^[a-f0-9]{64}$/.test(result?.revision??'')){settingsInputs=null;settingsSnapshot=result;const prior=last.get('api:settings');if(prior)last.set('api:settings',{...prior,launch:result});settingsReceipt='Launch settings saved.';}else settingsReceipt=result?.sent==='unknown'?'Outcome unknown. Read current settings before trying again. Draft kept.':`${result?.message||'Settings refused.'} Draft kept.`;}}
    catch{settingsReceipt='Outcome unknown. Read current settings before trying again. Draft kept.';}finally{settingsSaving=false;if(!out())render(false);}
  });
+  function tokenUsage(value) {
+    if (!value?.total) return '<section class="panel" aria-labelledby="token-usage-title"><h2 id="token-usage-title">Token usage</h2><p>Usage unavailable. No count was read.</p></section>';
+    const columns = row => ['input', 'output', 'cacheWrite', 'cacheRead', 'unknownTurns'].map(key => `<td>${txt(row[key])}</td>`).join('');
+    const rows = [`<tr><th scope="row">Total</th>${columns(value.total)}</tr>`];
+    for (const [group, label] of [['byRun', 'Run'], ['byInstance', 'Instance'], ['byModel', 'Model'], ['byProvider', 'Provider'], ['byAccount', 'Account']]) {
+      for (const row of value[group] || []) {
+        const name = ['run', 'instance', 'provider', 'model', 'account'].filter(key => key in row).map(key => row[key] ?? 'unassigned').join(' / ');
+        rows.push(`<tr><th scope="row">${label}: ${txt(name)}</th>${columns(row)}</tr>`);
+      }
+    }
+    return `<section class="panel" aria-labelledby="token-usage-title"><h2 id="token-usage-title">Token usage</h2><p>Best reading of session-editable transcripts, not proof. Unknown turns may have used more tokens. Account is unassigned until account tracking is configured.</p><div class="table-wrap"><table id="token-usage-table"><thead><tr><th scope="col">Subject</th><th scope="col">Input</th><th scope="col">Output</th><th scope="col">Cache write</th><th scope="col">Cache read</th><th scope="col">Unknown turns</th></tr></thead><tbody>${rows.join('')}</tbody></table></div></section>`;
+  }
  async function settingsView(get) {
    const s = await get.api('settings'), n = s.notifier || {};
+    let usage;
+    try { [usage] = await get('usage'); } catch (err) { if (err.status === 403 || err.code === 'signed-out') throw err; }
    const notify = typeof n.binding === 'string'
      ? `<dl class="kv"><dt>Binding</dt><dd><code>${txt(n.binding)}</code></dd></dl><p class="small muted">The binding name only. Its channels, server and user ids stay in the binding file.</p>`
      : empty('No notifier binding to show.', `${n.refused ? `<code>${txt(n.refused)}</code>: ` : ''}${txt(n.message || 'the notifier settings could not be read')}`);
-    return `${head('Settings', `${readLine()} Agent limits and appearance.`)}${appearance()}${launchForm(s.launch)}
+    return `${head('Settings', `${readLine()} Agent limits and appearance.`)}${appearance()}${launchForm(s.launch)}${tokenUsage(usage)}
      <section aria-labelledby="st-notify"><h2 id="st-notify">Notifications</h2>${notify}</section>
      <section aria-labelledby="st-about" class="panel"><h2 id="st-about">About</h2><dl class="kv"><dt>Address</dt><dd><code>${txt(s.address) || 'unknown'}</code></dd><dt>Release</dt><dd>${s.release ? `<code>${txt(s.release)}</code>` : '<span class="muted">unknown: no RELEASE file was read</span>'}</dd><dt>Business</dt><dd>${txt(s.business) || none}</dd><dt>Board</dt><dd><code>${txt(s.board) || 'unknown'}</code></dd></dl></section>`;
  }
@@ -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' });
@@ -0,0 +1,26 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { browser } from './browser.mjs';
+import { startServer } from '../src/serve.mjs';
+import { close } from './fixture.mjs';
+test('REQ-LIMIT-4 Settings shows all token kinds and unknown turns beside every total, as a read-only separate section', { timeout: 60000 }, async () => {
+  const count = { input: 2, output: 3, cacheWrite: 4, cacheRead: 5, unknownTurns: 7 };
+  const hostile = '<img src=x onerror="window.injected=true">';
+  const usage = { total: count, byRun: [{ run: 'run1', ...count }], byInstance: [{ instance: hostile, ...count }],
+    byModel: [{ provider: 'provider', model: 'model', ...count }], byProvider: [{ provider: 'provider', ...count }], byAccount: [{ account: null, ...count }] };
+  const server = await startServer({ port: 0, board: 'http://127.0.0.1:9',
+    bus: { messages: async () => ({ messages: [], unread: 0 }), usage: async () => usage },
+    reads: { settings: async () => ({ business: 'demo', notifier: {}, launch: { refused: 'not-configured' } }) },
+  });
+  const b = await browser();
+  try {
+    await b.navigate(`http://127.0.0.1:${server.address().port}/#/settings`);
+    await b.evaluate(`(async()=>{for(let n=0;n<200;n++){if(document.querySelector('#token-usage-table'))return;await new Promise(r=>setTimeout(r,30));}throw Error('usage table missing');})()`);
+    const rows = await b.evaluate("[...document.querySelectorAll('#token-usage-table tbody tr')].map(row=>[...row.querySelectorAll('td')].map(c=>c.textContent))");
+    assert.equal(rows.length, 6);
+    for (const row of rows) assert.deepEqual(row, ['2', '3', '4', '5', '7']);
+    assert.equal(await b.evaluate("document.querySelectorAll('#token-usage-table input, #token-usage-table button, #token-usage-table img').length"), 0);
+    assert.equal(await b.evaluate('window.injected===true'), false);
+    assert.ok((await b.evaluate("document.querySelector('#token-usage-title').parentElement.textContent")).includes('not proof'));
+  } finally { await b.close(); await close(server); }
+});
@@ -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);
+});
@@ -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" "$@"