perf: cheaper ingest — no re-parse of live hooked transcripts, delta hook writes, lighter pollers

- Session sync leaves a transcript alone while hooks feed its session live
  (active, newest event a hook's, touched within DASHBOARD_SYNC_LIVE_HOOK_MS,
  default 5 min) and catches it up once quiet; a hook-created session (which
  importSession skips anyway) is only snapshotted, never parsed.
- Hook writer keeps a per-transcript mark (applied after commit): an
  unchanged TranscriptCache result skips token/compaction/API-error/turn
  writes, a grown one walks only the entries after the last one handled.
- Codex usage: one stamp query per pass instead of one row per journal;
  between full passes (first + every 5 min) only the newest day folders and
  journals written in the last hour are statted; fork parents resolved from
  the table or a full walk.
- Cursor/Antigravity polls back off to 60 s after idle passes
  (server/lib/backoff-poll.js); lane sync runs only while another dashboard
  shares the data dir (checked every 30 s).
This commit is contained in:
2026-10-16 10:20:09 +07:00
parent ecbbb4dd8f
commit 5cb70d702b
15 changed files with 734 additions and 51 deletions
+17 -5
View File
@@ -1066,7 +1066,7 @@ flowchart TD
- Uses Node.js `http` module directly — no dependencies. - Uses Node.js `http` module directly — no dependencies.
- Resolution order is **env override → discovery file → default**: - Resolution order is **env override → discovery file → default**:
- `CLAUDE_DASHBOARD_PORT` forces a single target (no fan-out, no discovery). - `CLAUDE_DASHBOARD_PORT` forces a single target (no fan-out, no discovery).
- Otherwise `server/lib/server-info.js` reads `~/.claude/.agenthub.json`, prunes dead-PID entries, and returns one live `port` per unique `dataDir` (lowest port wins when two instances share `~/.claude/agenthub`). The handler POSTs to each returned port in parallel. The other instance learns of a lane the hooks moved through `broadcastForeignLaneChanges` (`server/routes/lanes.js`): every `DASHBOARD_LANE_SYNC_MS` (2 s) it reads `PRAGMA data_version`, which moves only on another connection's commit, and re-broadcasts each lane whose `updated_at` changed (and `removed` for a vanished one) to its own browsers. Sessions and events are not re-broadcast this way — a page on that instance shows them on its next fetch. - Otherwise `server/lib/server-info.js` reads `~/.claude/.agenthub.json`, prunes dead-PID entries, and returns one live `port` per unique `dataDir` (lowest port wins when two instances share `~/.claude/agenthub`). The handler POSTs to each returned port in parallel. The other instance learns of a lane the hooks moved through `broadcastForeignLaneChanges` (`server/routes/lanes.js`): every `DASHBOARD_LANE_SYNC_MS` (2 s) — while `peersSharingDataDir()` finds another live dashboard on its data dir, which it checks every 30 s — it reads `PRAGMA data_version`, which moves only on another connection's commit, and re-broadcasts each lane whose `updated_at` changed (and `removed` for a vanished one) to its own browsers. Sessions and events are not re-broadcast this way — a page on that instance shows them on its next fetch.
- If neither yields anything, the handler falls back to `4820`. - If neither yields anything, the handler falls back to `4820`.
- Per-target promises never reject — a dead listener can't starve the others, and the handler can wait on `Promise.all` for clean exit timing. - Per-target promises never reject — a dead listener can't starve the others, and the handler can wait on `Promise.all` for clean exit timing.
@@ -1223,7 +1223,12 @@ common "nothing changed" tick is just a handful of `stat` calls), funnels them
through the same `parseSessionFile` + `importSession` pipeline as every other through the same `parseSessionFile` + `importSession` pipeline as every other
path, and broadcasts `session_created` for newly imported sessions / path, and broadcasts `session_created` for newly imported sessions /
`session_updated` for grown ones — the same events hooks emit, so the UI `session_updated` for grown ones — the same events hooks emit, so the UI
refreshes live. All timers and watchers are `unref`'d and best-effort; nothing refreshes live. Two kinds of transcript are not re-parsed: one whose session
hooks are feeding **live** (status `active`, newest event a hook's, a hook
within `DASHBOARD_SYNC_LIVE_HOOK_MS`, default 5 min) is left until it goes
quiet or ends — then caught up once — and one whose session a hook created
(no `imported` flag, which `importSession` skips anyway) only has its durable
snapshot refreshed, never a parse. All timers and watchers are `unref`'d and best-effort; nothing
here blocks shutdown or can take down the server. here blocks shutdown or can take down the server.
### Upload request sequence ### Upload request sequence
@@ -1579,7 +1584,11 @@ re-point if that changes.
`DASHBOARD_CODEX_POLL_MS` beat, a first pass over every journal in the `DASHBOARD_CODEX_POLL_MS` beat, a first pass over every journal in the
background at startup). Each file's read offset and parse state live in the background at startup). Each file's read offset and parse state live in the
`codex_rollouts` table, so an unchanged file is never re-read and a grown one is `codex_rollouts` table, so an unchanged file is never re-read and a grown one is
read from where it stopped. `token_count` events are summed per `turn_context` read from where it stopped (a pass loads every row's size/mtime in one query).
The first pass and one every five minutes walk every journal; the poll's other
passes stat only the two newest day folders plus the journals the table saw
written in the last hour (or not yet read to the end), so a thread resumed
after an hour's quiet in an older folder is picked up by the next full pass. `token_count` events are summed per `turn_context`
model — `input_tokens` includes `cached_input_tokens` (stored as input − cached model — `input_tokens` includes `cached_input_tokens` (stored as input − cached
plus cache_read) and `output_tokens` includes reasoning. A **fork** copies its plus cache_read) and `output_tokens` includes reasoning. A **fork** copies its
parent's whole history into its own file, token events included, so every total parent's whole history into its own file, token events included, so every total
@@ -1654,7 +1663,9 @@ the binary's own proto descriptors.
**Sessions and tokens** (`server/lib/antigravity-usage.js`, **Sessions and tokens** (`server/lib/antigravity-usage.js`,
`startAntigravityUsagePoll`, every `DASHBOARD_ANTIGRAVITY_POLL_MS`, default `startAntigravityUsagePoll`, every `DASHBOARD_ANTIGRAVITY_POLL_MS`, default
15000, `0` disables). agy writes one SQLite file per conversation, 15000, `0` disables; after a pass that read nothing, wrote nothing and left no
fresh session to settle it slows to once a minute until one finds work —
`server/lib/backoff-poll.js`). agy writes one SQLite file per conversation,
`~/.gemini/antigravity-cli/conversations/<conversation id>.db`. Each is opened `~/.gemini/antigravity-cli/conversations/<conversation id>.db`. Each is opened
read-only whenever its mtime (or its non-empty `-wal`'s) changes; read-only whenever its mtime (or its non-empty `-wal`'s) changes;
`antigravity_conversations` holds the mtime read, so a session the user deleted `antigravity_conversations` holds the mtime read, so a session the user deleted
@@ -1743,7 +1754,8 @@ is ever inferred.
## Cursor Sessions ## Cursor Sessions
`server/lib/cursor-usage.js` (`startCursorUsagePoll`, every `server/lib/cursor-usage.js` (`startCursorUsagePoll`, every
`DASHBOARD_CURSOR_POLL_MS`, default 15000, `0` disables) reads the agent `DASHBOARD_CURSOR_POLL_MS`, default 15000, `0` disables; idle passes slow it to
once a minute like the Antigravity poll) reads the agent
transcripts Cursor writes — the CLI (`cursor-agent`) and the editor's agent both — transcripts Cursor writes — the CLI (`cursor-agent`) and the editor's agent both —
at `~/.cursor/projects/<project slug>/agent-transcripts/<id>/<id>.jsonl` at `~/.cursor/projects/<project slug>/agent-transcripts/<id>/<id>.jsonl`
(`AGENTHUB_CURSOR_DIR` overrides `~/.cursor`). Each becomes a `sessions` row with (`AGENTHUB_CURSOR_DIR` overrides `~/.cursor`). Each becomes a `sessions` row with
+2 -2
View File
@@ -256,9 +256,9 @@ DASHBOARD_DB_PATH=/var/lib/agenthub/dashboard.db
DASHBOARD_WORKFLOW_POLL_MS=12000 # Workflow-tool run journals DASHBOARD_WORKFLOW_POLL_MS=12000 # Workflow-tool run journals
DASHBOARD_CODEX_POLL_MS=15000 # Codex rollout journals → codex lane running/idle + Codex sessions/tokens DASHBOARD_CODEX_POLL_MS=15000 # Codex rollout journals → codex lane running/idle + Codex sessions/tokens
CODEX_HOME= # where Codex keeps sessions/ (default ~/.codex) CODEX_HOME= # where Codex keeps sessions/ (default ~/.codex)
DASHBOARD_ANTIGRAVITY_POLL_MS=15000 # agy conversation files → Antigravity sessions/tokens DASHBOARD_ANTIGRAVITY_POLL_MS=15000 # agy conversation files → Antigravity sessions/tokens (60 s while idle)
AGENTHUB_GEMINI_DIR= # where agy keeps its data (default ~/.gemini) AGENTHUB_GEMINI_DIR= # where agy keeps its data (default ~/.gemini)
DASHBOARD_CURSOR_POLL_MS=15000 # Cursor agent transcripts → Cursor sessions (no tokens) DASHBOARD_CURSOR_POLL_MS=15000 # Cursor agent transcripts → Cursor sessions (no tokens; 60 s while idle)
AGENTHUB_CURSOR_DIR= # where Cursor keeps its data (default ~/.cursor) AGENTHUB_CURSOR_DIR= # where Cursor keeps its data (default ~/.cursor)
AGENTHUB_AGY_USAGE= # 0 never runs `agy --print /usage` for the Usage panel's Antigravity quota (default on; running agy may update it) AGENTHUB_AGY_USAGE= # 0 never runs `agy --print /usage` for the Usage panel's Antigravity quota (default on; running agy may update it)
+3 -1
View File
@@ -491,7 +491,8 @@ graph TB
> entry whose PID is no longer alive** and POSTs the hook payload to one > entry whose PID is no longer alive** and POSTs the hook payload to one
> port per **unique SQLite data directory** (lowest port wins when two > port per **unique SQLite data directory** (lowest port wins when two
> instances share `~/.claude/agenthub`). The other instance re-broadcasts > instances share `~/.claude/agenthub`). The other instance re-broadcasts
> the workspaces those hooks moved within `DASHBOARD_LANE_SYNC_MS` (2 s), so its > the workspaces those hooks moved within `DASHBOARD_LANE_SYNC_MS` (2 s; a server
> alone on its data dir only looks for such a peer every 30 s), so its
> browsers see a workspace's status without a refresh. > browsers see a workspace's status without a refresh.
> 3. If neither yields a target, the handler falls back to `4820`. > 3. If neither yields a target, the handler falls back to `4820`.
> >
@@ -589,6 +590,7 @@ Claude Code waits only for the handler process: Node start-up, reading stdin, se
- **Fire-and-forget POST** — no wait for the dashboard's reply. - **Fire-and-forget POST** — no wait for the dashboard's reply.
- **Incremental transcript reads** — `TranscriptCache` reads only the bytes appended since the last hook. - **Incremental transcript reads** — `TranscriptCache` reads only the bytes appended since the last hook.
- **Only new transcript entries are written** — a per-transcript mark (`transcriptMarks` in `server/routes/hooks.js`, set once the hook's transaction commits) skips the token, compaction, API-error and turn-duration writes when the transcript did not change, and walks only the entries after the last one handled when it grew.
- **One POST per database, not per server** — servers sharing a database get each event once (`resolveHookIngestPorts`). - **One POST per database, not per server** — servers sharing a database get each event once (`resolveHookIngestPorts`).
- **Heavy reads off the event loop** — aggregates run in a worker (`server/lib/read-pool.js`), so a slow page never delays ingest. - **Heavy reads off the event loop** — aggregates run in a worker (`server/lib/read-pool.js`), so a slow page never delays ingest.
+60
View File
@@ -1937,6 +1937,47 @@ async function ingestWorkflowsForBatch(dbModule, sessions) {
} }
} }
// Session sync: a session counts as fed LIVE by hooks while it is active, its
// newest event came from a hook (not this importer), and a hook touched it
// within this window. Such a transcript is left to the hooks until it goes
// quiet. DASHBOARD_SYNC_LIVE_HOOK_MS=0 turns the deferral off.
const LIVE_HOOK_MS = (() => {
const n = Number(process.env.DASHBOARD_SYNC_LIVE_HOOK_MS ?? 5 * 60_000);
return Number.isFinite(n) && n >= 0 ? n : 5 * 60_000;
})();
// mtimeCache value for a deferred file: present (so the fs watcher ignores its
// appends) yet never equal to a real mtime (so every sweep re-checks it).
const DEFERRED_MTIME = -1;
const IMPORTED_DATA_RE = /"imported"\s*:\s*true/;
const latestEventStmts = new WeakMap();
function rowIsImported(row) {
try {
return !!(row.metadata && JSON.parse(row.metadata).imported);
} catch {
return false;
}
}
function isHookLive(dbModule, row) {
if (!(LIVE_HOOK_MS > 0) || row.status !== "active") return false;
const touched = Date.parse(row.updated_at);
if (!Number.isFinite(touched) || Date.now() - touched > LIVE_HOOK_MS) return false;
try {
let stmt = latestEventStmts.get(dbModule.db);
if (!stmt) {
stmt = dbModule.db.prepare(
"SELECT data FROM events WHERE session_id = ? ORDER BY id DESC LIMIT 1"
);
latestEventStmts.set(dbModule.db, stmt);
}
const ev = stmt.get(row.id);
return !!ev && !IMPORTED_DATA_RE.test(ev.data || "");
} catch {
return false; // when in doubt, sync as before
}
}
/** /**
* Incremental, change-driven sync of the default `~/.claude/projects` directory. * Incremental, change-driven sync of the default `~/.claude/projects` directory.
* *
@@ -2024,6 +2065,25 @@ async function syncDefaultProjects(dbModule, options = {}) {
continue; continue;
} }
} }
// A session hooks are feeding RIGHT NOW: its transcript keeps growing
// between hooks, so `mtime > updated_at` on nearly every sweep, and a
// full re-parse + snapshot copy here bought nothing the hooks do not
// already record. Leave it until it goes quiet (or ends): the sentinel
// keeps the watcher ignoring its appends, and never equals a real
// mtime, so the first sweep after the hooks stop catches it up.
if (existingRow && isHookLive(dbModule, existingRow)) {
mtimeCache.set(sourcePath, DEFERRED_MTIME);
continue;
}
// A session a hook created (no `imported` flag): importSession() skips
// it outright, so parsing the whole transcript + subagents was pure
// waste — only the durable snapshot copy still matters.
if (existingRow && !rowIsImported(existingRow)) {
mtimeCache.set(sourcePath, mtime);
if (require("../server/lib/run-as").sameOwner(sourcePath, existingRow.cwd))
await snapshotTranscript(sourcePath, sessionId);
continue;
}
const existed = !!existingRow; const existed = !!existingRow;
const session = await parseSessionForImport(projPath, sourcePath); const session = await parseSessionForImport(projPath, sourcePath);
// Record the mtime even when the file yields no session, so we don't // Record the mtime even when the file yields no session, so we don't
+2 -1
View File
@@ -1600,10 +1600,11 @@ DASHBOARD_DB_PATH=./data/dashboard.db # SQLite database path
DASHBOARD_AI_TOOLS_DIR= # Extra AI Tools catalog JSON files; an id here replaces the built-in one DASHBOARD_AI_TOOLS_DIR= # Extra AI Tools catalog JSON files; an id here replaces the built-in one
AGENTHUB_SHELL_PATH=1 # 0 = never run the login shell to read its PATH (startup + Refresh) AGENTHUB_SHELL_PATH=1 # 0 = never run the login shell to read its PATH (startup + Refresh)
AGENTHUB_AI_TOOLS_SYNC=1 # 0 = never copy the newest release's AI Tools catalog (server/lib/ai-tools-sync.js, daily) AGENTHUB_AI_TOOLS_SYNC=1 # 0 = never copy the newest release's AI Tools catalog (server/lib/ai-tools-sync.js, daily)
DASHBOARD_LANE_SYNC_MS=2000 # re-broadcast lanes another dashboard on the same DB changed (PRAGMA data_version); 0 = off DASHBOARD_LANE_SYNC_MS=2000 # re-broadcast lanes another dashboard on the same DB changed (PRAGMA data_version), only while one runs (checked every 30 s); 0 = off
# Background services # Background services
DASHBOARD_SESSION_SYNC_MS=30000 # Continuous project-sync poll interval (ms); 0 disables the poll (watcher stays) DASHBOARD_SESSION_SYNC_MS=30000 # Continuous project-sync poll interval (ms); 0 disables the poll (watcher stays)
DASHBOARD_SYNC_LIVE_HOOK_MS=300000 # project sync leaves a transcript alone while hooks fed its session within this window; 0 = never defer
DASHBOARD_LIVENESS_PROBE=1 # 0 disables the dead-session liveness reap (use when hooks arrive from another machine) DASHBOARD_LIVENESS_PROBE=1 # 0 disables the dead-session liveness reap (use when hooks arrive from another machine)
DASHBOARD_LIVENESS_IDLE_SECONDS=60 # Idle gate before the liveness reap may complete a process-less session DASHBOARD_LIVENESS_IDLE_SECONDS=60 # Idle gate before the liveness reap may complete a process-less session
DASHBOARD_REPORTS_MS=60000 # Scheduled-reports beat (ms); 0 disables the scheduler DASHBOARD_REPORTS_MS=60000 # Scheduled-reports beat (ms); 0 disables the scheduler
+42
View File
@@ -0,0 +1,42 @@
/**
* @file Tests server/lib/backoff-poll.js — the poll that runs at its base pace
* while passes find work (or fail) and slows to its idle pace after a pass
* that resolved `false`, returning to the base pace once work shows up again.
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
*/
const { describe, it } = require("node:test");
const assert = require("node:assert/strict");
const { startBackoffPoll, idleDefault } = require("../lib/backoff-poll");
const wait = (ms) => new Promise((r) => setTimeout(r, ms));
describe("startBackoffPoll", () => {
it("slows down after idle passes and speeds up again on work", async () => {
const answers = [true, false, false, true, "x"];
const at = [];
const t0 = Date.now();
const poll = startBackoffPoll(
() => {
at.push(Date.now() - t0);
const a = answers.shift();
if (a === "x") throw new Error("boom");
return a;
},
{ baseMs: 10, idleMs: 80 }
);
await wait(260);
poll.stop();
assert.ok(at.length >= 5, `ran ${at.length} times`);
const gaps = at.slice(1).map((t, i) => t - at[i]);
assert.ok(gaps[0] < 60, "work → base pace");
assert.ok(gaps[1] >= 75, "idle → idle pace");
assert.ok(gaps[2] >= 75, "still idle");
assert.ok(gaps[3] < 60, "work again → base pace");
});
it("idles at a minute by default, never faster than the base pace", () => {
assert.equal(idleDefault(15_000), 60_000);
assert.equal(idleDefault(120_000), 120_000);
});
});
+75
View File
@@ -536,3 +536,78 @@ describe("service tier", () => {
]); ]);
}); });
}); });
describe("ingestCodexUsage — recent pass", () => {
const NOW = Date.parse("2026-09-29T03:00:00Z");
function writeAt(day, id, lines, mtime) {
const dir = path.join(process.env.CODEX_HOME, "sessions", ...day.split("/"));
fs.mkdirSync(dir, { recursive: true });
const file = path.join(dir, `rollout-${day.replace(/\//g, "-")}T01-00-00-${id}.jsonl`);
fs.writeFileSync(file, jsonl(lines));
fs.utimesSync(file, mtime, mtime);
return file;
}
it("stats only the newest day folders and the journals still growing", async () => {
const quiet = writeAt(
"2026/09/01",
"old1",
[meta("old1"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])],
new Date(NOW - 20 * 86_400_000)
);
const live = writeAt(
"2026/09/02",
"old2",
[meta("old2"), ctx("gpt-a"), tc([10, 0, 1], [10, 0, 1])],
new Date(NOW - 60_000)
);
writeAt("2026/09/27", "a", [meta("a"), ctx("gpt-a")], new Date(NOW - 86_400_000));
writeAt("2026/09/28", "b", [meta("b"), ctx("gpt-a")], new Date(NOW - 86_400_000));
await ingestCodexUsage({ now: NOW }); // the full first pass
// Both old journals grow; only the one written within the hour is looked at.
fs.appendFileSync(quiet, jsonl([tc([30, 0, 3], [20, 0, 2])]));
fs.appendFileSync(live, jsonl([tc([30, 0, 3], [20, 0, 2])]));
fs.utimesSync(quiet, new Date(NOW - 1000), new Date(NOW - 1000));
fs.utimesSync(live, new Date(NOW - 1000), new Date(NOW - 1000));
// A new journal in today's folder is found too.
writeAt(
"2026/09/29",
"new1",
[meta("new1"), ctx("gpt-a"), tc([5, 0, 1], [5, 0, 1])],
new Date(NOW - 1000)
);
await ingestCodexUsage({ now: NOW, recent: true });
assert.deepEqual(usage("old2"), [{ model: "gpt-a", i: 30, r: 0, o: 3 }]);
assert.deepEqual(usage("old1"), [{ model: "gpt-a", i: 10, r: 0, o: 1 }], "left to a full pass");
assert.deepEqual(usage("new1"), [{ model: "gpt-a", i: 5, r: 0, o: 1 }]);
await ingestCodexUsage({ now: NOW }); // the next full pass catches it up
assert.deepEqual(usage("old1"), [{ model: "gpt-a", i: 30, r: 0, o: 3 }]);
});
it("finds a fork's parent in an older folder on a recent pass", async () => {
const parent = [
meta("p9"),
ctx("gpt-a"),
tc([100, 0, 10], [100, 0, 10]),
tc([300, 0, 30], [200, 0, 20]),
];
writeAt("2026/08/01", "p9", parent, new Date(NOW - 50 * 86_400_000));
writeAt("2026/09/28", "x", [meta("x")], new Date(NOW - 86_400_000));
await ingestCodexUsage({ now: NOW });
writeAt(
"2026/09/29",
"f9",
[
meta("f9", { forked_from_id: "p9" }),
meta("p9"),
...parent.slice(1),
tc([350, 0, 35], [50, 0, 5]),
],
new Date(NOW - 1000)
);
await ingestCodexUsage({ now: NOW, recent: true });
assert.deepEqual(usage("f9"), [{ model: "gpt-a", i: 50, r: 0, o: 5 }]);
});
});
+82
View File
@@ -150,3 +150,85 @@ describe("syncDefaultProjects", () => {
); );
}); });
}); });
describe("syncDefaultProjects — sessions hooks are feeding", () => {
const SNAP_DIR = path.join(process.env.DASHBOARD_DATA_DIR, "transcripts");
const more = (sessionId, cwd, ts) => ({
type: "assistant",
cwd,
sessionId,
timestamp: ts,
message: {
model: "claude-opus-4-8",
content: [{ type: "text", text: "later" }],
usage: { input_tokens: 3, output_tokens: 2 },
},
});
const bumpMtime = (file) => {
const future = Date.now() / 1000 + 120;
fs.utimesSync(file, future, future);
};
const ageSession = (id) =>
dbModule.db
.prepare("UPDATE sessions SET updated_at = ? WHERE id = ?")
.run(new Date(Date.now() - 10 * 60_000).toISOString(), id);
it("defers a live hook-fed imported session, then catches it up once quiet", async () => {
const ID = "33333333-cccc-4ccc-8ccc-333333333333";
const cache = new Map();
const file = writeSession("-live", ID, fixtureLines(ID, "/live"));
await syncDefaultProjects(dbModule, { mtimeCache: cache });
assert.ok(dbModule.stmts.getSession.get(ID), "imported first");
// A hook arrives: its event is the newest, and it touches the session.
dbModule.stmts.insertEvent.run(ID, `${ID}-main`, "PreToolUse", "Bash", "hook", "{}");
dbModule.stmts.touchSession.run(ID);
writeSession(
"-live",
ID,
fixtureLines(ID, "/live", [more(ID, "/live", "2026-04-18T12:09:00.000Z")])
);
bumpMtime(file);
const live = await syncDefaultProjects(dbModule, { mtimeCache: cache });
assert.equal(
live.changed.find((c) => c.sessionId === ID),
undefined,
"deferred while live"
);
assert.equal(cache.get(file), -1, "kept in the cache so the watcher ignores its appends");
ageSession(ID); // the hooks went quiet
const quiet = await syncDefaultProjects(dbModule, { mtimeCache: cache });
assert.ok(
quiet.changed.find((c) => c.sessionId === ID && !c.isNew),
"caught up once quiet"
);
assert.notEqual(cache.get(file), -1);
});
it("never re-parses a hook-created session, only refreshes its snapshot once quiet", async () => {
const ID = "44444444-dddd-4ddd-8ddd-444444444444";
dbModule.stmts.insertSession.run(ID, "hooked", "active", "/hooked", null, null);
dbModule.stmts.insertEvent.run(ID, null, "SessionStart", null, "hook", "{}");
const file = writeSession("-hooked", ID, fixtureLines(ID, "/hooked"));
bumpMtime(file);
const snap = path.join(SNAP_DIR, `${ID}.jsonl`);
const cache = new Map();
await syncDefaultProjects(dbModule, { mtimeCache: cache });
assert.equal(fs.existsSync(snap), false, "no snapshot copy while hooks are live");
ageSession(ID);
const before = dbModule.db.prepare("SELECT COUNT(*) AS n FROM events WHERE session_id = ?");
const n0 = before.get(ID).n;
const r = await syncDefaultProjects(dbModule, { mtimeCache: cache });
assert.equal(
r.changed.find((c) => c.sessionId === ID),
undefined,
"nothing imported"
);
assert.equal(before.get(ID).n, n0, "no events written for a hook-created session");
assert.equal(fs.existsSync(snap), true, "snapshot taken once quiet");
});
});
+167
View File
@@ -0,0 +1,167 @@
/**
* @file Tests for the hook writer's per-transcript "already handled" marks
* (`transcriptMarks` in server/routes/hooks.js): a hook whose transcript is
* unchanged skips the token / compaction / API-error / turn-duration writes,
* a grown transcript is walked only from the last entry handled, and the rows
* left behind are exactly the ones a full re-walk would have written.
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
*/
const { describe, it, before, after } = require("node:test");
const assert = require("node:assert/strict");
const path = require("path");
const fs = require("fs");
const os = require("os");
const http = require("http");
const TMP = fs.mkdtempSync(path.join(os.tmpdir(), "agenthub-marks-"));
process.env.DASHBOARD_DB_PATH = path.join(TMP, "dashboard.db");
process.env.CLAUDE_HOME = path.join(TMP, "home");
process.env.DASHBOARD_DATA_DIR = path.join(TMP, "data");
const { createApp, startServer } = require("../index");
const { db } = require("../db");
const hooksRouter = require("../routes/hooks");
let server;
let BASE;
function post(body) {
return new Promise((resolve, reject) => {
const url = new URL("/api/hooks/event", BASE);
const payload = JSON.stringify(body);
const r = http.request(
{
hostname: url.hostname,
port: url.port,
path: url.pathname,
method: "POST",
headers: {
"Content-Type": "application/json",
"Content-Length": Buffer.byteLength(payload),
},
},
(res) => {
res.resume();
res.on("end", () => resolve(res.statusCode));
}
);
r.on("error", reject);
r.write(payload);
r.end();
});
}
const line = (o) => JSON.stringify(o) + "\n";
const assistant = (id, inTok) => ({
type: "assistant",
timestamp: "2026-09-18T01:00:00.000Z",
message: {
id,
model: "claude-sonnet-5",
content: [{ type: "text", text: "ok" }],
usage: { input_tokens: inTok, output_tokens: 1 },
},
});
const turn = (ms, ts) => ({
type: "system",
subtype: "turn_duration",
durationMs: ms,
timestamp: ts,
});
const apiError = (text, ts) => ({
type: "assistant",
isApiErrorMessage: true,
error: "rate_limit",
timestamp: ts,
message: { content: [{ type: "text", text }] },
});
const compact = (uuid, ts) => ({ type: "user", isCompactSummary: true, uuid, timestamp: ts });
function rows(sid) {
const ev = db
.prepare(
"SELECT event_type, summary, created_at FROM events WHERE session_id = ? AND event_type IN ('TurnDuration','APIError','Compaction') ORDER BY event_type, created_at, summary"
)
.all(sid)
.map((r) =>
r.event_type === "APIError"
? `${r.event_type}|${r.summary}`
: `${r.event_type}|${r.created_at}|${r.summary}`
);
const tokens = db
.prepare("SELECT model, input_tokens, baseline_input FROM token_usage WHERE session_id = ?")
.all(sid);
return { ev, tokens };
}
before(async () => {
server = await startServer(createApp(), 0);
BASE = `http://127.0.0.1:${server.address().port}`;
});
after(() => {
server?.close();
fs.rmSync(TMP, { recursive: true, force: true });
});
describe("transcript marks", () => {
it("skips an unchanged transcript and walks only the new tail of a grown one", async () => {
const sid = "abababab-1111-4111-8111-abababababab";
const cwd = path.join(TMP, "work");
fs.mkdirSync(cwd, { recursive: true });
const tpath = path.join(TMP, `${sid}.jsonl`);
fs.writeFileSync(
tpath,
line(assistant("m1", 5)) +
line(turn(1000, "2026-09-18T01:00:01.000Z")) +
line(apiError("slow down", "2026-09-18T01:00:02.000Z")) +
line(compact("c-1", "2026-09-18T01:00:03.000Z")) +
line(assistant("m2", 7)) +
line(turn(2000, "2026-09-18T01:00:04.000Z"))
);
const data = { session_id: sid, cwd, transcript_path: tpath };
assert.equal(await post({ hook_type: "PostToolUse", data }), 200);
const first = rows(sid);
assert.equal(first.ev.filter((e) => e.startsWith("TurnDuration")).length, 2);
assert.equal(first.ev.filter((e) => e.startsWith("APIError")).length, 1);
assert.equal(first.ev.filter((e) => e.startsWith("Compaction")).length, 1);
assert.ok(hooksRouter.transcriptMarks.size >= 1, "a mark is kept after commit");
// Proof of the skip: a row removed behind the writer's back is NOT put
// back by a hook whose transcript did not change (before, every hook
// re-walked the whole history and re-inserted it).
db.prepare(
"DELETE FROM events WHERE session_id = ? AND event_type = 'TurnDuration' AND created_at = '2026-09-18T01:00:01.000Z'"
).run(sid);
assert.equal(await post({ hook_type: "PostToolUse", data: { ...data, tool_name: "x" } }), 200);
assert.equal(rows(sid).ev.filter((e) => e.startsWith("TurnDuration")).length, 1);
db.prepare("DELETE FROM events WHERE session_id = ? AND event_type = 'TurnDuration'").run(sid);
hooksRouter.transcriptMarks.clear();
assert.equal(await post({ hook_type: "PostToolUse", data: { ...data, tool_name: "y" } }), 200);
assert.deepEqual(rows(sid), first, "a cleared mark re-walks the whole history, as before");
// Grow it: one turn, one new error, a repeat of the old error, a compaction.
fs.appendFileSync(
tpath,
line(assistant("m3", 11)) +
line(turn(3000, "2026-09-18T01:00:05.000Z")) +
line(apiError("slow down", "2026-09-18T01:00:06.000Z")) +
line(apiError("overloaded", "2026-09-18T01:00:07.000Z")) +
line(compact("c-2", "2026-09-18T01:00:08.000Z"))
);
assert.equal(await post({ hook_type: "Stop", data }), 200);
const grown = rows(sid);
// The same transcript walked from scratch must produce the same rows.
hooksRouter.transcriptMarks.clear();
assert.equal(await post({ hook_type: "Stop", data: { ...data, stop_hook_active: true } }), 200);
assert.deepEqual(rows(sid), grown, "the delta wrote exactly what a full walk writes");
assert.equal(grown.ev.filter((e) => e.startsWith("TurnDuration")).length, 3);
assert.equal(grown.ev.filter((e) => e.startsWith("APIError")).length, 2, "repeat deduped");
assert.equal(grown.ev.filter((e) => e.startsWith("Compaction")).length, 2);
assert.equal(grown.tokens.length, 1);
assert.equal(grown.tokens[0].input_tokens + grown.tokens[0].baseline_input, 23);
});
});
+21 -4
View File
@@ -825,18 +825,35 @@ function startBackgroundServices() {
} }
// Another dashboard on the same database (hooks reach only one of them) // Another dashboard on the same database (hooks reach only one of them)
// moving a lane: re-broadcast it here. One pragma per beat when nothing // moving a lane: re-broadcast it here. One pragma per beat when nothing
// changed. Off with DASHBOARD_LANE_SYNC_MS=0. // changed — and only while such a dashboard runs at all: alone on its data
// dir, this server only looks for one every LANE_PEER_CHECK_MS (the first
// call primes what it has seen, so a peer that starts later is compared
// against it). Off with DASHBOARD_LANE_SYNC_MS=0.
const LANE_SYNC_MS = Number(process.env.DASHBOARD_LANE_SYNC_MS ?? 2_000); const LANE_SYNC_MS = Number(process.env.DASHBOARD_LANE_SYNC_MS ?? 2_000);
if (Number.isFinite(LANE_SYNC_MS) && LANE_SYNC_MS > 0) { if (Number.isFinite(LANE_SYNC_MS) && LANE_SYNC_MS > 0) {
const { broadcastForeignLaneChanges } = require("./routes/lanes"); const { broadcastForeignLaneChanges } = require("./routes/lanes");
const laneSyncTimer = setInterval(() => { const LANE_PEER_CHECK_MS = Math.max(LANE_SYNC_MS, 30_000);
let peersSeenAt = 0;
let hasPeers = false;
const syncLanes = () => {
try { try {
broadcastForeignLaneChanges(); broadcastForeignLaneChanges();
} catch (err) { } catch (err) {
console.warn("lane sync failed:", err.message); console.warn("lane sync failed:", err.message);
} }
}, LANE_SYNC_MS); };
if (laneSyncTimer.unref) laneSyncTimer.unref(); syncLanes();
const laneSyncTick = () => {
if (Date.now() - peersSeenAt >= LANE_PEER_CHECK_MS) {
hasPeers = peersSharingDataDir().length > 0;
peersSeenAt = Date.now();
}
if (hasPeers) syncLanes();
const t = setTimeout(laneSyncTick, hasPeers ? LANE_SYNC_MS : LANE_PEER_CHECK_MS);
if (t.unref) t.unref();
};
const first = setTimeout(laneSyncTick, LANE_SYNC_MS);
if (first.unref) first.unref();
} }
// Continuous discovery of sessions under ~/.claude/projects. The one-time // Continuous discovery of sessions under ~/.claude/projects. The one-time
// legacy backfill above runs only once (marker-gated), so a project added // legacy backfill above runs only once (marker-gated), so a project added
+18 -10
View File
@@ -425,6 +425,9 @@ function writeSession(f, conv, now) {
} }
let running = null; let running = null;
// Fresh sessions the last pass left for a later one to settle (the poll keeps
// its base pace while there are any).
let lastPending = 0;
/** /**
* One pass over every conversation file: a changed one is re-read, an * One pass over every conversation file: a changed one is re-read, an
@@ -440,12 +443,16 @@ function ingestAntigravityUsage({
running = (async () => { running = (async () => {
let read = 0; let read = 0;
let written = 0; let written = 0;
let pending = 0; // still-fresh sessions a later pass may have to settle
for (const f of conversationFiles()) { for (const f of conversationFiles()) {
try { try {
const seen = getSeen.get(f.path); const seen = getSeen.get(f.path);
if (seen && seen.mtime_ms === f.mtime) { if (seen && seen.mtime_ms === f.mtime) {
const a = activity(f.id, f.mtime); const a = activity(f.id, f.mtime);
if (fresh(a, now)) continue; if (fresh(a, now)) {
pending += 1;
continue;
}
const s = stmts.getSession.get(f.id); const s = stmts.getSession.get(f.id);
if (s && s.tool === "antigravity" && s.status === "active") { if (s && s.tool === "antigravity" && s.status === "active") {
const at = new Date(a.last).toISOString(); const at = new Date(a.last).toISOString();
@@ -486,6 +493,7 @@ function ingestAntigravityUsage({
} }
await new Promise((r) => setImmediate(r)); await new Promise((r) => setImmediate(r));
} }
lastPending = pending;
return { files: read, sessions: written }; return { files: read, sessions: written };
})().finally(() => { })().finally(() => {
running = null; running = null;
@@ -504,16 +512,16 @@ function startAntigravityUsagePoll(broadcast, broadcastLane) {
if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return; if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return;
// Files an older build could not read get another try. // Files an older build could not read get another try.
db.prepare("DELETE FROM antigravity_conversations WHERE session_id = ''").run(); db.prepare("DELETE FROM antigravity_conversations WHERE session_id = ''").run();
const tick = () => { // Slowed to once a minute after a pass that read nothing, wrote nothing and
ingestAntigravityUsage({ broadcast, broadcastLane }) // left no fresh session to settle; back to POLL_MS once one finds work.
.then((r) => { require("./backoff-poll").startBackoffPoll(
() =>
ingestAntigravityUsage({ broadcast, broadcastLane }).then((r) => {
if (r.sessions > 0 && pricingSync.syncEnabled()) void pricingSync.syncPricing(db); if (r.sessions > 0 && pricingSync.syncEnabled()) void pricingSync.syncPricing(db);
}) return r.files + r.sessions + lastPending > 0;
.catch(() => {}); }),
}; { baseMs: POLL_MS }
tick(); );
const timer = setInterval(tick, POLL_MS);
if (timer.unref) timer.unref();
} }
module.exports = { module.exports = {
+43
View File
@@ -0,0 +1,43 @@
/**
* @file A background poll that slows down while it finds nothing to do: the
* next pass runs `baseMs` after a pass that did work (or failed), `idleMs`
* after one whose `tick` resolved `false` (nothing changed, nothing pending).
* One pass at a time — the next is scheduled only once the last settled — and
* every timer is `unref`'d. Used by the Cursor and Antigravity usage polls.
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
*/
/** Default slow interval: 60 s, never faster than the base one. */
const idleDefault = (baseMs) => Math.max(baseMs, 60_000);
/**
* Run `tick` now, then again and again. `tick` resolves `false` for an idle
* pass; anything else (or a throw) keeps the base pace.
* @returns {{stop: () => void}}
*/
function startBackoffPoll(tick, { baseMs, idleMs = idleDefault(baseMs) }) {
let timer = null;
let stopped = false;
const schedule = (ms) => {
if (stopped) return;
timer = setTimeout(run, ms);
if (timer.unref) timer.unref();
};
function run() {
Promise.resolve()
.then(tick)
.then(
(busy) => schedule(busy === false ? idleMs : baseMs),
() => schedule(baseMs)
);
}
run();
return {
stop() {
stopped = true;
clearTimeout(timer);
},
};
}
module.exports = { startBackoffPoll, idleDefault };
+97 -4
View File
@@ -235,7 +235,9 @@ function threadNames(home = getCodexHome()) {
const names = new Map(); const names = new Map();
let text = ""; let text = "";
try { try {
text = require("./run-as").readFileSafe(path.join(home, "session_index.jsonl"), "utf8", { max: 16 << 20 }); text = require("./run-as").readFileSafe(path.join(home, "session_index.jsonl"), "utf8", {
max: 16 << 20,
});
} catch { } catch {
return names; return names;
} }
@@ -251,6 +253,17 @@ function threadNames(home = getCodexHome()) {
} }
const getRollout = db.prepare("SELECT * FROM codex_rollouts WHERE path = ?"); const getRollout = db.prepare("SELECT * FROM codex_rollouts WHERE path = ?");
// What a pass needs to skip an unchanged journal, for every journal at once:
// one query per pass instead of one (with its whole parse state) per file.
const rolloutStamps = db.prepare(
`SELECT path, size, mtime_ms, instr(state, '"more":true') > 0 AS more FROM codex_rollouts`
);
// Journals that may still be growing outside the newest day folders: written
// within RECENT_ROW_MS, or not yet read to the end.
const liveRolloutPaths = db.prepare(
`SELECT path FROM codex_rollouts WHERE mtime_ms > ? OR instr(state, '"more":true') > 0`
);
const threadPath = db.prepare("SELECT path FROM codex_rollouts WHERE thread_id = ? LIMIT 1");
const upsertRollout = db.prepare( const upsertRollout = db.prepare(
`INSERT INTO codex_rollouts (path, session_id, thread_id, size, mtime_ms, offset, state, updated_at) `INSERT INTO codex_rollouts (path, session_id, thread_id, size, mtime_ms, offset, state, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) VALUES (?, ?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
@@ -400,6 +413,59 @@ function writeSession(sessionId, names, now, origin = LOCAL) {
const running = new Map(); // home ("" = CODEX_HOME) → the pass in flight const running = new Map(); // home ("" = CODEX_HOME) → the pass in flight
// A RECENT pass (the poll's, between full ones) stats only the newest
// RECENT_DAY_DIRS day folders plus the journals the table says may still grow
// (RECENT_ROW_MS) — not every rollout Codex ever wrote. A journal in an older
// folder that wakes up again (a resumed thread quiet for over RECENT_ROW_MS)
// is picked up by the next full pass.
const RECENT_DAY_DIRS = 2;
const RECENT_ROW_MS = 60 * 60_000;
function numericDirs(dir) {
try {
return fs
.readdirSync(dir, { withFileTypes: true })
.filter((e) => e.isDirectory() && /^\d+$/.test(e.name))
.map((e) => e.name)
.sort((a, b) => Number(b) - Number(a));
} catch {
return [];
}
}
function recentRolloutFiles(home, now) {
const root = path.join(home || getCodexHome(), "sessions");
const days = [];
outer: for (const y of numericDirs(root))
for (const m of numericDirs(path.join(root, y)))
for (const d of numericDirs(path.join(root, y, m))) {
days.push(path.join(root, y, m, d));
if (days.length >= RECENT_DAY_DIRS) break outer;
}
const paths = new Set();
for (const dir of days) {
try {
for (const e of fs.readdirSync(dir, { withFileTypes: true }))
if (e.isFile() && /^rollout-.*\.jsonl$/.test(e.name)) paths.add(path.join(dir, e.name));
} catch {
/* vanished mid-scan */
}
}
const prefix = root + path.sep;
for (const { path: p } of liveRolloutPaths.all(now - RECENT_ROW_MS))
if (p.startsWith(prefix)) paths.add(p);
const out = [];
for (const p of paths) {
try {
const stat = fs.statSync(p);
if (stat.isFile()) out.push({ path: p, mtime: stat.mtimeMs, size: stat.size });
} catch {
/* gone */
}
}
return out;
}
/** /**
* One pass over every journal. Yields to the event loop between files; a * One pass over every journal. Yields to the event loop between files; a
* second call for the same home while one runs joins it. `home`, `source` and * second call for the same home while one runs joins it. `home`, `source` and
@@ -413,21 +479,42 @@ function ingestCodexUsage({
home, home,
source = LOCAL.source, source = LOCAL.source,
activeMs = LOCAL.activeMs, activeMs = LOCAL.activeMs,
recent = false,
} = {}) { } = {}) {
const key = home || ""; const key = home || "";
if (running.has(key)) return running.get(key); if (running.has(key)) return running.get(key);
const pass = (async () => { const pass = (async () => {
const files = allRolloutFiles(home); const files = recent ? recentRolloutFiles(home, now) : allRolloutFiles(home);
const byThread = new Map(); const byThread = new Map();
for (const f of files) { for (const f of files) {
const m = FILE_NAME.exec(path.basename(f.path)); const m = FILE_NAME.exec(path.basename(f.path));
if (m) byThread.set(m[1], f.path); if (m) byThread.set(m[1], f.path);
} }
// A recent pass lists only part of the tree: a fork's parent may live in
// an older folder, known to the table (or, failing that, to a full walk).
let walkedAll = !recent;
const parentPath = (id) => {
if (byThread.has(id)) return byThread.get(id);
if (walkedAll) return null;
const known = threadPath.get(id)?.path;
const root = path.join(home || getCodexHome(), "sessions") + path.sep;
if (known && known.startsWith(root) && fs.existsSync(known)) return known;
walkedAll = true;
for (const f of allRolloutFiles(home)) {
const m = FILE_NAME.exec(path.basename(f.path));
if (m && !byThread.has(m[1])) byThread.set(m[1], f.path);
}
return byThread.get(id) || null;
};
const stamps = new Map();
for (const r of rolloutStamps.all()) stamps.set(r.path, r);
const touched = new Set(); const touched = new Set();
let read = 0; let read = 0;
for (const f of files) { for (const f of files) {
try { try {
const mtime = Math.floor(f.mtime); const mtime = Math.floor(f.mtime);
const seen = stamps.get(f.path);
if (seen && seen.size === f.size && seen.mtime_ms === mtime && !seen.more) continue;
const row = getRollout.get(f.path); const row = getRollout.get(f.path);
let offset = row ? row.offset : 0; let offset = row ? row.offset : 0;
let state = row ? JSON.parse(row.state) : {}; let state = row ? JSON.parse(row.state) : {};
@@ -439,7 +526,8 @@ function ingestCodexUsage({
let skip = null; let skip = null;
const parent = offset === 0 ? firstLineMeta(f.path).forked_from_id : state.forkedFrom; const parent = offset === 0 ? firstLineMeta(f.path).forked_from_id : state.forkedFrom;
if (parent && !state.pastFork) { if (parent && !state.pastFork) {
if (byThread.has(parent)) skip = await parentTotals(byThread.get(parent)); const pp = parentPath(parent);
if (pp) skip = await parentTotals(pp);
else if (offset === 0) else if (offset === 0)
console.warn(`[codex-usage] fork parent ${parent} not found; ${f.path} may overcount`); console.warn(`[codex-usage] fork parent ${parent} not found; ${f.path} may overcount`);
} }
@@ -508,10 +596,15 @@ function startCodexUsagePoll(broadcast) {
.memberHomes() .memberHomes()
.map((h) => path.join(h, ".codex")), .map((h) => path.join(h, ".codex")),
]; ];
// Every FULL_EVERY-th pass (and the first) walks every journal; the others
// only the recent ones (recentRolloutFiles).
const FULL_EVERY = Math.max(1, Math.round((5 * 60_000) / POLL_MS));
let passNo = 0;
const ingestAll = async () => { const ingestAll = async () => {
const total = { sessions: 0 }; const total = { sessions: 0 };
const recent = passNo++ % FULL_EVERY !== 0;
for (const home of homes()) { for (const home of homes()) {
const r = await ingestCodexUsage({ broadcast, home }); const r = await ingestCodexUsage({ broadcast, home, recent });
total.sessions += r?.sessions || 0; total.sessions += r?.sessions || 0;
} }
return total; return total;
+20 -16
View File
@@ -182,14 +182,7 @@ function writeSession(f, now) {
const endedAt = active ? null : lastWrite; const endedAt = active ? null : lastWrite;
const name = firstPrompt(f.path) || existing?.name || `Cursor ${f.id.slice(0, 8)}`; const name = firstPrompt(f.path) || existing?.name || `Cursor ${f.id.slice(0, 8)}`;
if (existing) { if (existing) {
updateSession.run( updateSession.run(name, status, endedAt, lastWrite, cwd, f.id);
name,
status,
endedAt,
lastWrite,
cwd,
f.id
);
return "session_updated"; return "session_updated";
} }
insertSession.run( insertSession.run(
@@ -206,6 +199,9 @@ function writeSession(f, now) {
} }
let running = null; let running = null;
// Fresh sessions the last pass left for a later one to settle (the poll keeps
// its base pace while there are any).
let lastPending = 0;
/** /**
* One pass over every transcript: a changed one is re-read, an unchanged one * One pass over every transcript: a changed one is re-read, an unchanged one
@@ -217,11 +213,15 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) {
running = (async () => { running = (async () => {
let read = 0; let read = 0;
let written = 0; let written = 0;
let pending = 0; // still-fresh sessions a later pass may have to settle
for (const f of transcriptFiles()) { for (const f of transcriptFiles()) {
try { try {
const seen = getSeen.get(f.path); const seen = getSeen.get(f.path);
if (seen && seen.mtime_ms === f.mtime) { if (seen && seen.mtime_ms === f.mtime) {
if (fresh(f.mtime, now)) continue; if (fresh(f.mtime, now)) {
pending += 1;
continue;
}
const s = stmts.getSession.get(f.id); const s = stmts.getSession.get(f.id);
if (s && s.tool === "cursor" && s.status === "active") { if (s && s.tool === "cursor" && s.status === "active") {
settle.run(new Date(f.mtime).toISOString(), s.id); settle.run(new Date(f.mtime).toISOString(), s.id);
@@ -242,6 +242,7 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) {
} }
await new Promise((r) => setImmediate(r)); await new Promise((r) => setImmediate(r));
} }
lastPending = pending;
return { files: read, sessions: written }; return { files: read, sessions: written };
})().finally(() => { })().finally(() => {
running = null; running = null;
@@ -249,18 +250,21 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) {
return running; return running;
} }
/** First pass at once (in the background), then every DASHBOARD_CURSOR_POLL_MS (default 15 s; 0 disables). */ /**
* First pass at once (in the background), then every DASHBOARD_CURSOR_POLL_MS
* (default 15 s; 0 disables) — slowed to once a minute after a pass that read
* nothing, wrote nothing and left no fresh session to settle, and back to the
* base pace as soon as one finds work.
*/
function startCursorUsagePoll(broadcast) { function startCursorUsagePoll(broadcast) {
const POLL_MS = process.env.DASHBOARD_CURSOR_POLL_MS const POLL_MS = process.env.DASHBOARD_CURSOR_POLL_MS
? Number(process.env.DASHBOARD_CURSOR_POLL_MS) ? Number(process.env.DASHBOARD_CURSOR_POLL_MS)
: 15_000; : 15_000;
if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return; if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return;
const tick = () => { require("./backoff-poll").startBackoffPoll(
ingestCursorUsage({ broadcast }).catch(() => {}); () => ingestCursorUsage({ broadcast }).then((r) => r.files + r.sessions + lastPending > 0),
}; { baseMs: POLL_MS }
tick(); );
const timer = setInterval(tick, POLL_MS);
if (timer.unref) timer.unref();
} }
module.exports = { module.exports = {
+85 -8
View File
@@ -294,6 +294,46 @@ const turnDurationStampsStmt = db.prepare(
"SELECT created_at FROM events WHERE session_id = ? AND event_type = 'TurnDuration'" "SELECT created_at FROM events WHERE session_id = ? AND event_type = 'TurnDuration'"
); );
// What this process already handled of each session's transcript parse.
// `transcriptCache.extract` returns the SAME result object while the file is
// unchanged, and a grown file's result is the old arrays plus the new tail
// (errors/turnDurations keep their element objects, compaction entries their
// uuid). So an identical result skips the token, compaction, API-error and
// turn-duration writes outright — each is idempotent against the DB, so they
// would have written nothing new — and a grown one walks only the entries
// after the last one handled. Anything that does not line up (a full re-read,
// the session re-created, the head trimmed off a capped array) falls back to
// the whole array, exactly as before. Marks are applied only once the hook's
// transaction committed; bounded, oldest dropped first.
const TRANSCRIPT_MARKS_MAX = 500;
const transcriptMarks = new Map();
let pendingMarks = null;
function transcriptMarkFor(sessionId, transcriptPath, startedAt) {
const m = transcriptMarks.get(`${sessionId}\0${transcriptPath}`);
return m && m.startedAt === startedAt ? m : null;
}
function queueTranscriptMark(sessionId, transcriptPath, mark) {
if (pendingMarks) pendingMarks.push([`${sessionId}\0${transcriptPath}`, mark]);
}
function applyTranscriptMarks(marks) {
for (const [key, mark] of marks) {
transcriptMarks.delete(key);
transcriptMarks.set(key, mark);
if (transcriptMarks.size > TRANSCRIPT_MARKS_MAX)
transcriptMarks.delete(transcriptMarks.keys().next().value);
}
}
/** The entries of `arr` after `last` (found by `same`), or all of `arr`. */
function tailAfter(arr, last, same) {
if (!arr || last == null) return arr;
for (let i = arr.length - 1; i >= 0; i--) if (same(arr[i], last)) return arr.slice(i + 1);
return arr;
}
/** The session's `<id>-main` agent, created and broadcast. */ /** The session's `<id>-main` agent, created and broadcast. */
function createMainAgent(session) { function createMainAgent(session) {
const mainAgentId = `${session.id}-main`; const mainAgentId = `${session.id}-main`;
@@ -944,6 +984,11 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => {
const result = transcriptCache.extract(transcriptPath); const result = transcriptCache.extract(transcriptPath);
if (result) { if (result) {
const { tokensByModel, compaction, latestModel } = result; const { tokensByModel, compaction, latestModel } = result;
// Already handled (see transcriptMarks): the same result skips the
// idempotent writes below, a grown one handles only its new entries.
const mark = transcriptMarkFor(sessionId, transcriptPath, session.started_at);
const unchanged = !!mark && mark.result === result;
const lastOf = (arr) => (arr && arr.length ? arr[arr.length - 1] : null);
// Keep session.model in sync with the user's *current* model — the // Keep session.model in sync with the user's *current* model — the
// transcript's most recent assistant entry is the source of truth, since // transcript's most recent assistant entry is the source of truth, since
@@ -972,8 +1017,13 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => {
// Register compaction agents and events. // Register compaction agents and events.
// Each isCompactSummary entry in the JSONL = one compaction that occurred. // Each isCompactSummary entry in the JSONL = one compaction that occurred.
// Deduplicate by uuid so we only create once per compaction. // Deduplicate by uuid so we only create once per compaction.
if (compaction) { if (compaction && !unchanged) {
for (const entry of compaction.entries) { const fresh = tailAfter(
compaction.entries,
mark?.compactionUuid,
(e, uuid) => e.uuid === uuid
);
for (const entry of fresh) {
const compactId = `${sessionId}-compact-${entry.uuid}`; const compactId = `${sessionId}-compact-${entry.uuid}`;
if (stmts.getAgent.get(compactId)) continue; if (stmts.getAgent.get(compactId)) continue;
@@ -1024,7 +1074,7 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => {
} }
} }
if (tokensByModel) { if (tokensByModel && !unchanged) {
// Each bucket carries its pricing dimensions (model/speed/geo/tier) plus // Each bucket carries its pricing dimensions (model/speed/geo/tier) plus
// the 1h cache-write split and server-tool request counts. // the 1h cache-write split and server-tool request counts.
for (const tokens of Object.values(tokensByModel)) { for (const tokens of Object.values(tokensByModel)) {
@@ -1059,12 +1109,15 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => {
} }
// Register API errors from transcript (quota limits, rate limits, overloaded, etc.) // Register API errors from transcript (quota limits, rate limits, overloaded, etc.)
if (result.errors) { const freshErrors = unchanged
? null
: tailAfter(result.errors, mark?.lastError, (a, b) => a === b);
if (freshErrors && freshErrors.length) {
let newErrorRecorded = false; let newErrorRecorded = false;
// Deduplicate against the errors already recorded, read ONCE per hook: // Deduplicate against the errors already recorded, read ONCE per hook:
// `result.errors` is the transcript's whole history, not just new ones. // `result.errors` is the transcript's whole history, not just new ones.
const recorded = new Set(apiErrorSummariesStmt.all(sessionId).map((r) => r.summary)); const recorded = new Set(apiErrorSummariesStmt.all(sessionId).map((r) => r.summary));
for (const apiErr of result.errors) { for (const apiErr of freshErrors) {
const errSummary = `${apiErr.type}: ${apiErr.message}`; const errSummary = `${apiErr.type}: ${apiErr.message}`;
if (recorded.has(errSummary)) continue; if (recorded.has(errSummary)) continue;
recorded.add(errSummary); recorded.add(errSummary);
@@ -1136,13 +1189,16 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => {
} }
// Register turn duration events from transcript // Register turn duration events from transcript
if (result.turnDurations) { const freshTurns = unchanged
? null
: tailAfter(result.turnDurations, mark?.lastTurn, (a, b) => a === b);
if (freshTurns && freshTurns.length) {
// `turnDurations` is every turn in the transcript (up to 1000), on // `turnDurations` is every turn in the transcript (up to 1000), on
// every hook: dedupe against ONE read of the recorded timestamps (a // every hook: dedupe against ONE read of the recorded timestamps (a
// covering-index scan) instead of a query per turn — 23 ms per hook on // covering-index scan) instead of a query per turn — 23 ms per hook on
// a 763-turn session before. // a 763-turn session before.
const recorded = new Set(turnDurationStampsStmt.all(sessionId).map((r) => r.created_at)); const recorded = new Set(turnDurationStampsStmt.all(sessionId).map((r) => r.created_at));
for (const td of result.turnDurations) { for (const td of freshTurns) {
const tdTs = td.timestamp || new Date().toISOString(); const tdTs = td.timestamp || new Date().toISOString();
if (recorded.has(tdTs)) continue; if (recorded.has(tdTs)) continue;
recorded.add(tdTs); recorded.add(tdTs);
@@ -1171,6 +1227,14 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => {
} }
} }
queueTranscriptMark(sessionId, transcriptPath, {
result,
startedAt: session.started_at,
compactionUuid: lastOf(compaction?.entries)?.uuid ?? null,
lastError: lastOf(result.errors),
lastTurn: lastOf(result.turnDurations),
});
// Update session metadata with enriched data (thinking blocks, usage extras) // Update session metadata with enriched data (thinking blocks, usage extras)
if (result.usageExtras || result.thinkingBlockCount > 0) { if (result.usageExtras || result.thinkingBlockCount > 0) {
const session = stmts.getSession.get(sessionId); const session = stmts.getSession.get(sessionId);
@@ -1250,7 +1314,19 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => {
* inside it described a row a later throw could still roll back. * inside it described a row a later throw could still roll back.
*/ */
function processEvent(hookType, data, tool) { function processEvent(hookType, data, tool) {
return holdFrames(() => processEventTx(hookType, data, tool)); const outer = pendingMarks;
pendingMarks = [];
let marks = null;
try {
const out = holdFrames(() => processEventTx(hookType, data, tool));
marks = pendingMarks;
return out;
} finally {
pendingMarks = outer;
// Only what COMMITTED is marked handled: a rolled-back hook's entries are
// walked again by the next one.
if (marks && !db.inTransaction) applyTranscriptMarks(marks);
}
} }
/** Run `fn` with frames held, then send them if it returned (nested: the /** Run `fn` with frames held, then send them if it returned (nested: the
@@ -2176,6 +2252,7 @@ const watchdogTimer = setInterval(watchdogCheck, WATCHDOG_INTERVAL_MS);
if (watchdogTimer.unref) watchdogTimer.unref(); if (watchdogTimer.unref) watchdogTimer.unref();
router.transcriptCache = transcriptCache; router.transcriptCache = transcriptCache;
router.transcriptMarks = transcriptMarks;
router.watchdogCheck = watchdogCheck; router.watchdogCheck = watchdogCheck;
router.livenessReap = livenessReap; router.livenessReap = livenessReap;
module.exports = router; module.exports = router;