feat(ledger): Gate F, the ledger's T3 thread source (#1506)

packages/ledger/src/t3.mjs reads ~/.t3/userdata/state.sqlite read-only,
in one transaction. It maps each thread to a seat by title and checks
self-addressed headers. Unmatched threads go in a t3:unmapped row. A
missing or locked database exits 1 and names --no-t3. Gate F is on by
default (lead decision 12). The 6a uppercase-class fix rides here.

Separate item: the Pi session reader splits lines only on \n, so a raw
U+2028 or U+2029 in a string no longer splits a record. Node 26.8.1's
readline split there, and the live ledger refused on HEAD.

Darkwing built to brief R3 (f3c05c1b); manifest ba73a163. Filbert
approved the build (e47ec6da) and the U+2028 fix as its own item; brief
review be1aa414. On an index export: the eight suites
24/90/43/17/14/15/63/18, ledger 47/47. Four nonblocking notes go to a
small follow-up.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
2026-09-26 16:39:56 -05:00
co-authored by Claude Opus 5.5
parent e58783d278
commit 136958c98b
11 changed files with 1764 additions and 31 deletions
+13 -5
View File
@@ -1,11 +1,12 @@
#!/usr/bin/env node
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { dateRange, readCommits, readIssues, readSessions, summarize, formatTable, SourceError } from './ledger.mjs';
import { dateRange, readCommits, readIssues, readSessions, mergeSources, summarize, formatTable, SourceError } from './ledger.mjs';
import { readT3, defaultT3Path } from './t3.mjs';
const usage = 'Usage: node packages/ledger/src/cli.mjs --since YYYY-MM-DD [--until YYYY-MM-DD] [--json] [--no-issues]';
const usage = 'Usage: node packages/ledger/src/cli.mjs --since YYYY-MM-DD [--until YYYY-MM-DD] [--json] [--no-issues] [--no-t3 | --t3-db PATH]';
export async function main(args, root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '../../..')) {
let since, until, json = false, noIssues = false;
let since, until, t3Db, json = false, noIssues = false, noT3 = false;
const seen = new Set();
for (let i = 0; i < args.length; i++) {
const flag = args[i];
@@ -14,13 +15,18 @@ export async function main(args, root = path.resolve(path.dirname(fileURLToPath(
if (flag === '--help') { console.log(usage); return; }
if (flag === '--json') json = true;
else if (flag === '--no-issues') noIssues = true;
else if (flag === '--since' || flag === '--until') {
else if (flag === '--no-t3') noT3 = true;
else if (flag === '--t3-db') {
t3Db = args[++i];
if (!t3Db || t3Db.startsWith('--')) throw new SourceError('--t3-db requires a path');
} else if (flag === '--since' || flag === '--until') {
const value = args[++i];
if (!value || value.startsWith('--')) throw new SourceError(`${flag} requires a date`);
if (flag === '--since') since = value; else until = value;
} else throw new SourceError('Unknown option; ' + usage);
}
if (!since) throw new SourceError(usage);
if (noT3 && t3Db !== undefined) throw new SourceError('--no-t3 and --t3-db cannot be combined');
const range = dateRange(since, until);
const commits = readCommits(root, range);
// Fixture tools may be placed first on PATH. The repository client is the
@@ -30,7 +36,9 @@ export async function main(args, root = path.resolve(path.dirname(fileURLToPath(
let issues;
try { issues = noIssues ? null : readIssues(root, range); }
finally { if (priorPath === undefined) delete process.env.PATH; else process.env.PATH = priorPath; }
const sessions = await readSessions(root, range);
// T3 is on by default. A missing or unreadable database refuses the report.
const t3 = noT3 ? null : await readT3(root, range, t3Db === undefined ? { dbPath: defaultT3Path(), isDefault: true } : { dbPath: t3Db, isDefault: false });
const sessions = mergeSources(await readSessions(root, range), t3);
const report = summarize(range, commits, issues, sessions);
console.log(json ? JSON.stringify(report, null, 2) : formatTable(report));
return report;
+63 -15
View File
@@ -1,7 +1,6 @@
import { execFileSync } from 'node:child_process';
import { createReadStream } from 'node:fs';
import { readdir, lstat } from 'node:fs/promises';
import { createInterface } from 'node:readline';
import path from 'node:path';
const DAY = 86400000;
@@ -23,7 +22,7 @@ export function dateRange(since, until = new Date().toISOString().slice(0, 10))
if (end <= start) throw new SourceError('--until must not precede --since');
return { since, until, start, end };
}
const inRange = (value, range) => {
export const inRange = (value, range) => {
const ms = typeof value === 'number' ? value : Date.parse(value);
return Number.isFinite(ms) && ms >= range.start && ms < range.end;
};
@@ -75,15 +74,32 @@ export function messageText(content) {
if (Array.isArray(content)) return content.filter(c => c?.type === 'text' && typeof c.text === 'string').map(c => c.text).join('\n');
return '';
}
// Classes are matched in either case: seats send DECISION and REVIEW-REQUEST.
// tmux preamble from agent-send.sh: [host:session -> host:session class=x]
const TMUX = /^\[([^\s:\[\]]+):([^\s\[\]]+) -> ([^\s:\[\]]+):([^\s\[\]]+)(?: class=[A-Za-z-]+)?\](?:\s|$)/;
// T3 header (docs/guides/T3-AGENT-COMMS.md): [from: role (id) -> to: role (id) class=x]
const T3 = /^\[from: ([^\s()\[\]]+) \(([^()\[\]]+)\) -> to: ([^\s()\[\]]+) \(([^()\[\]]+)\)(?: class=[A-Za-z-]+)?\](?:\s|$)/;
const firstLine = text => text.split(/\r?\n/, 1)[0];
export function t3Header(text) {
const m = firstLine(text).match(T3);
return m ? { from: m[1], fromId: m[2], to: m[3], toId: m[4] } : null;
}
export function messageKind(text) {
const firstLine = text.split(/\r?\n/, 1)[0];
// tmux preamble from agent-send.sh: [host:session -> host:session class=x]
const tmux = firstLine.match(/^\[([^\s:\[\]]+):([^\s\[\]]+) -> ([^\s:\[\]]+):([^\s\[\]]+)(?: class=[a-z-]+)?\](?:\s|$)/);
// T3 header (docs/guides/T3-AGENT-COMMS.md): [from: role (id) -> to: role (id) class=x]
const t3 = firstLine.match(/^\[from: ([^\s()\[\]]+) \(([^()\[\]]+)\) -> to: ([^\s()\[\]]+) \(([^()\[\]]+)\)(?: class=[a-z-]+)?\](?:\s|$)/);
const sender = tmux ? tmux[2] : t3 ? t3[1] : null;
const tmux = firstLine(text).match(TMUX), t3 = t3Header(text);
const sender = tmux ? tmux[2] : t3 ? t3.from : null;
return sender === null ? 'human' : sender === 'control-board' ? 'board' : 'agent';
}
// JSONL lines end at \n only. readline also ends a line at U+2028, which JSON
// allows raw inside a string, so it split valid records (Node 26.8.1).
async function* jsonLines(input) {
let rest = '';
for await (const chunk of input) {
const parts = (rest + chunk).split('\n');
rest = parts.pop();
yield* parts;
}
if (rest) yield rest;
}
async function directories(dir, optional = false) {
try {
if (!(await lstat(dir)).isDirectory()) throw new SourceError('Session source must be a real directory');
@@ -93,11 +109,15 @@ async function directories(dir, optional = false) {
throw new SourceError(`Cannot read ledger directory: ${dir}`);
}
}
// Seats are the real directories in agents/, sorted.
export async function readSeats(root) {
return (await directories(path.join(root, 'agents'))).filter(e => e.isDirectory()).map(e => e.name).sort((a, b) => a.localeCompare(b));
}
export async function readSessions(root, range) {
const rows = [];
const mentions = new Map();
// No symlink traversal, no fleet paths, no transcript content in the report.
const agents = (await directories(path.join(root, 'agents'))).filter(e => e.isDirectory()).sort((a, b) => a.name.localeCompare(b.name));
const agents = (await readSeats(root)).map(name => ({ name }));
const state = path.join(root, '.pi', 'state');
// Check every source ancestor, not only the leaf directory.
if (!(await directories(path.join(root, '.pi'), true)).length) return { rows, mentions };
@@ -109,11 +129,10 @@ export async function readSessions(root, range) {
const files = (await directories(dir, true)).filter(e => e.isFile() && e.name.endsWith('.jsonl'));
const row = { seat: agent.name, board: 0, agent: 0, human: 0 };
for (const file of files) {
const input = createReadStream(path.join(dir, file.name));
const lines = createInterface({ input, crlfDelay: Infinity });
const input = createReadStream(path.join(dir, file.name), { encoding: 'utf8' });
let lineNumber = 0;
try {
for await (const line of lines) {
for await (const line of jsonLines(input)) {
lineNumber++;
if (!line.trim()) continue;
let entry;
@@ -132,12 +151,33 @@ export async function readSessions(root, range) {
mentions.get(number).add(agent.name);
}
}
} finally { lines.close(); input.destroy(); }
} finally { input.destroy(); }
}
if (row.board + row.agent + row.human) rows.push(row);
}
return { rows, mentions };
}
// Adds T3 counts to the Pi rows per seat. Unmapped T3 threads get one row,
// last. The report keeps the Pi rows and the T3 section so the split shows.
export function mergeSources(pi, t3, unmapped = 't3:unmapped') {
if (!t3) return { rows: pi.rows, mentions: pi.mentions, pi: pi.rows, t3: { read: false } };
const bySeat = new Map(pi.rows.map(r => [r.seat, { ...r }]));
for (const [seat, counts] of t3.rows) {
if (seat === unmapped || !(counts.board + counts.agent + counts.human)) continue;
const row = bySeat.get(seat) ?? { seat, board: 0, agent: 0, human: 0 };
for (const kind of ['board', 'agent', 'human']) row[kind] += counts[kind];
bySeat.set(seat, row);
}
const rows = [...bySeat.values()].sort((a, b) => a.seat.localeCompare(b.seat));
const extra = t3.rows.get(unmapped);
if (extra.board + extra.agent + extra.human) rows.push({ seat: unmapped, ...extra });
const mentions = new Map([...pi.mentions].map(([n, seats]) => [n, new Set(seats)]));
for (const [n, seats] of t3.mentions) {
if (!mentions.has(n)) mentions.set(n, new Set());
for (const seat of seats) mentions.get(n).add(seat);
}
return { rows, mentions, pi: pi.rows, t3: t3.section };
}
const round = value => Math.round(value * 10) / 10;
function duration(issue) {
if (!issue) return UNKNOWN;
@@ -163,13 +203,14 @@ export function summarize(range, commits, issues, sessions) {
const median = hours.includes(UNKNOWN) ? UNKNOWN : hours.length ?
round(hours.length % 2 ? hours[middle] : (hours[middle - 1] + hours[middle]) / 2) : 0;
const human = sessions.rows.reduce((sum, r) => sum + r.human, 0);
return { since: range.since, until: range.until, timezone: 'UTC', issues: rows, seats: sessions.rows,
const sources = sessions.t3 ? { pi: sessions.pi, t3: sessions.t3 } : {};
return { since: range.since, until: range.until, timezone: 'UTC', issues: rows, seats: sessions.rows, ...sources,
totals: { issuesClosed: issues === null ? UNKNOWN : closed.length,
medianHoursOpen: issues === null ? UNKNOWN : median, commits: commits.length,
followUpsPerIssue: rows.length ? round(rows.reduce((sum, r) => sum + r.followUps, 0) / rows.length) : 0,
humanMessagesPerClosedIssue: issues === null ? UNKNOWN : closed.length ? round(human / closed.length) : human ? UNKNOWN : 0 } };
}
const clean = value => String(value).replace(/[\x00-\x1f\x7f-\x9f]/g, ' ');
export const clean = value => String(value).replace(/[\x00-\x1f\x7f-\x9f]/g, ' ');
const decimal = value => typeof value === 'number' ? value.toFixed(1) : value;
export function totalsLine(t) {
return `Totals: issues closed ${t.issuesClosed} | median hours open ${decimal(t.medianHoursOpen)} | commits ${t.commits} | follow-ups per issue ${decimal(t.followUpsPerIssue)} | human messages per closed issue ${decimal(t.humanMessagesPerClosedIssue)}`;
@@ -181,5 +222,12 @@ export function formatTable(report) {
decimal(r.hoursOpen), r.commits, r.followUps, r.seats.join(', ')].join(' | ')),
'', 'Seat | Board | Agent | Human',
...report.seats.map(r => [clean(r.seat), r.board, r.agent, r.human].join(' | ')),
...t3Line(report.t3),
'', totalsLine(report.totals)].join('\n');
}
// One line when T3 was skipped or read from somewhere other than the default.
function t3Line(t3) {
if (!t3) return [];
if (!t3.read) return ['T3: not read (--no-t3)'];
return t3.database.default ? [] : [`T3: read from ${clean(t3.database.path)}, not the default`];
}
+151
View File
@@ -0,0 +1,151 @@
import { lstat } from 'node:fs/promises';
import { DatabaseSync } from 'node:sqlite';
import { pathToFileURL } from 'node:url';
import os from 'node:os';
import path from 'node:path';
import { SourceError, UNKNOWN, clean, inRange, issueNumbers, messageKind, readSeats, t3Header } from './ledger.mjs';
// T3 keeps every thread message in one SQLite database. This reader opens that
// file read-only and nothing else in ~/.t3. See
// docs/plans/2026-09-26_ledger-t3-source.md for the rules below.
export const UNMAPPED = 't3:unmapped';
const SKIP = 'use --no-t3 to skip T3';
const REQUIRED = {
projection_projects: ['project_id', 'workspace_root', 'deleted_at'],
projection_threads: ['thread_id', 'project_id', 'title', 'archived_at', 'deleted_at'],
projection_thread_messages: ['message_id', 'thread_id', 'role', 'text', 'created_at'],
};
const DIAGNOSTIC = { orchestration_events: ['stream_id', 'event_type', 'payload_json', 'metadata_json'] };
export const defaultT3Path = () => path.join(os.homedir(), '.t3', 'userdata', 'state.sqlite');
// Every named path must exist and must not be a symlink. Skipping one would be
// a silent zero, so each problem refuses the report.
async function checkPaths(dbPath, isDefault) {
const dirs = isDefault ? [path.dirname(path.dirname(dbPath)), path.dirname(dbPath)] : [path.dirname(dbPath)];
for (const [target, wantDir] of [...dirs.map(d => [d, true]), [dbPath, false]]) {
let stat;
try { stat = await lstat(target); }
catch { throw new SourceError(`T3 database unavailable: ${target} is missing or unreadable; ${SKIP}`); }
if (stat.isSymbolicLink()) throw new SourceError(`T3 database refused: ${target} is a symlink; ${SKIP}`);
if (wantDir ? !stat.isDirectory() : !stat.isFile()) {
throw new SourceError(`T3 database refused: ${target} is not a ${wantDir ? 'directory' : 'regular file'}; ${SKIP}`);
}
}
}
function missingColumns(db, tables) {
const missing = [];
for (const [table, columns] of Object.entries(tables)) {
const have = new Set(db.prepare('select name from pragma_table_info(?)').all(table).map(r => r.name));
if (!have.size) missing.push(table);
else for (const column of columns) if (!have.has(column)) missing.push(`${table}.${column}`);
}
return missing;
}
// Seat for a thread title: the lower-cased title equals the seat or starts
// with the seat and a space. Longest seat first, so the most specific wins.
export function seatForTitle(title, seats) {
const lower = title.toLowerCase();
return [...seats].sort((a, b) => b.length - a.length).find(s => lower === s || lower.startsWith(`${s} `)) ?? null;
}
// Origin per message id from thread.message-sent events. Any missing table,
// column or unparseable event makes the diagnostic unknown; it decides nothing.
function origins(db, projectId) {
if (missingColumns(db, DIAGNOSTIC).length) return null;
const byMessage = new Map();
const events = db.prepare(`select e.payload_json, e.metadata_json from orchestration_events e
join projection_threads t on t.thread_id = e.stream_id
where e.event_type = 'thread.message-sent' and t.project_id = ?`).all(projectId);
for (const event of events) {
let payload, metadata;
try { payload = JSON.parse(event.payload_json); metadata = JSON.parse(event.metadata_json); }
catch { return null; }
if (typeof payload?.messageId !== 'string') return null;
byMessage.set(payload.messageId, typeof metadata?.origin?.appVersion === 'string');
}
return byMessage;
}
function query(db, root, range, seats) {
const missing = missingColumns(db, REQUIRED);
if (missing.length) throw new SourceError(`T3 schema changed: missing ${missing.join(', ')}`);
// Compared in JavaScript so a declared collation can't loosen the match.
const projects = db.prepare('select project_id, workspace_root from projection_projects where deleted_at is null').all()
.filter(p => p.workspace_root === root);
if (projects.length !== 1) {
throw new SourceError(`T3 has ${projects.length ? 'more than one project' : 'no project'} for ${root}; a project opened through a symlink does not match; ${SKIP}`);
}
const projectId = projects[0].project_id;
const threads = new Map(), excluded = { importedThreads: 0, deletedThreads: 0 };
for (const t of db.prepare('select thread_id, title, archived_at, deleted_at from projection_threads where project_id = ?').all(projectId)) {
if (typeof t.thread_id !== 'string' || typeof t.title !== 'string') throw new SourceError('T3 thread with a non-text id or title');
if (t.thread_id.startsWith('import:')) { excluded.importedThreads++; continue; }
if (t.deleted_at !== null) { excluded.deletedThreads++; continue; }
threads.set(t.thread_id, { id: t.thread_id, title: t.title, archived: t.archived_at !== null, seat: seatForTitle(t.title, seats) });
}
const rows = new Map([...seats, UNMAPPED].map(s => [s, { board: 0, agent: 0, human: 0 }]));
const mentions = new Map(), human = [];
const messages = db.prepare(`select m.message_id, m.thread_id, m.role, m.text, m.created_at from projection_thread_messages m
join projection_threads t on t.thread_id = m.thread_id where t.project_id = ?`).all(projectId);
for (const m of messages) {
const thread = threads.get(m.thread_id);
if (!thread) continue;
const where = `T3 message ${clean(m.message_id)} in thread ${clean(m.thread_id)}`;
if (m.role !== 'user' && m.role !== 'assistant') throw new SourceError(`${where} has an unknown role`);
if (typeof m.text !== 'string') throw new SourceError(`${where} has non-text content`);
if (typeof m.created_at !== 'string' || !Number.isFinite(Date.parse(m.created_at))) throw new SourceError(`${where} has an invalid created_at`);
if (m.role !== 'user') continue;
// A header addressed to its own thread must agree with the title mapping.
const header = t3Header(m.text);
if (header && header.toId === thread.id) {
const to = header.to.toLowerCase();
if (thread.seat ? to !== thread.seat : seats.includes(to)) {
throw new SourceError(`T3 header conflict: thread ${clean(thread.id)} "${clean(thread.title)}" maps to ${thread.seat ?? 'no seat'}, but a header addresses ${clean(header.to)}`);
}
}
if (!inRange(m.created_at, range)) continue;
const kind = messageKind(m.text), seat = thread.seat ?? UNMAPPED;
rows.get(seat)[kind]++;
if (kind === 'human') human.push(m.message_id);
for (const number of issueNumbers(m.text)) {
if (!mentions.has(number)) mentions.set(number, new Set());
mentions.get(number).add(seat);
}
}
const byMessage = origins(db, projectId);
const sentThroughApi = byMessage === null ? UNKNOWN : human.filter(id => byMessage.get(id) === false).length;
const noEvent = byMessage === null ? UNKNOWN : human.filter(id => !byMessage.has(id)).length;
const listed = seat => [...threads.values()].filter(t => (t.seat ?? UNMAPPED) === seat)
.sort((a, b) => a.id.localeCompare(b.id)).map(({ id, title, archived }) => ({ id, title, archived }));
const seatRows = seats.map(seat => ({ seat, ...rows.get(seat), threads: listed(seat) })).filter(r => r.threads.length);
return { rows, mentions, excluded, seats: seatRows, unmapped: { ...rows.get(UNMAPPED), threads: listed(UNMAPPED) },
diagnostic: { humanSentThroughApi: sentThroughApi, humanWithoutEvent: noEvent } };
}
// Reads one snapshot of T3's database. Returns the per-seat rows and issue
// mentions the ledger merges with Pi, and the report's `t3` section.
export async function readT3(root, range, { dbPath = defaultT3Path(), isDefault = true } = {}) {
dbPath = path.resolve(dbPath);
await checkPaths(dbPath, isDefault);
const seats = await readSeats(root);
const url = pathToFileURL(dbPath);
url.searchParams.set('mode', 'ro');
let db, result;
try {
db = new DatabaseSync(url, { readOnly: true, timeout: 5000 });
db.exec('BEGIN');
result = query(db, root, range, seats);
db.exec('COMMIT');
} catch (error) {
if (error instanceof SourceError) throw error;
throw new SourceError(`T3 database cannot be read: ${dbPath} (SQLite ${error.errcode ?? 'error'}); ${SKIP}`);
} finally {
try { if (db?.isTransaction) db.exec('ROLLBACK'); } catch { /* the close below still runs */ }
try { db?.close(); } catch { /* nothing was written */ }
}
const { rows, mentions, ...section } = result;
return { rows, mentions, section: { read: true, database: { path: dbPath, default: isDefault }, ...section } };
}