diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 0917ae55..3518293c 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -1066,7 +1066,7 @@ flowchart TD - Uses Node.js `http` module directly — no dependencies. - Resolution order is **env override → discovery file → default**: - `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`. - 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 path, and broadcasts `session_created` for newly imported sessions / `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. ### 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 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 -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 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 @@ -1654,7 +1663,9 @@ the binary's own proto descriptors. **Sessions and tokens** (`server/lib/antigravity-usage.js`, `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/.db`. Each is opened read-only whenever its mtime (or its non-empty `-wal`'s) changes; `antigravity_conversations` holds the mtime read, so a session the user deleted @@ -1743,7 +1754,8 @@ is ever inferred. ## Cursor Sessions `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 — at `~/.cursor/projects//agent-transcripts//.jsonl` (`AGENTHUB_CURSOR_DIR` overrides `~/.cursor`). Each becomes a `sessions` row with diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index bd06d876..13e0977c 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -256,9 +256,9 @@ DASHBOARD_DB_PATH=/var/lib/agenthub/dashboard.db DASHBOARD_WORKFLOW_POLL_MS=12000 # Workflow-tool run journals DASHBOARD_CODEX_POLL_MS=15000 # Codex rollout journals → codex lane running/idle + Codex sessions/tokens 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) -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_AGY_USAGE= # 0 never runs `agy --print /usage` for the Usage panel's Antigravity quota (default on; running agy may update it) diff --git a/docs/HOOKS.md b/docs/HOOKS.md index fe5e24a4..c560629d 100644 --- a/docs/HOOKS.md +++ b/docs/HOOKS.md @@ -491,7 +491,8 @@ graph TB > 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 > 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. > 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. - **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`). - **Heavy reads off the event loop** — aggregates run in a worker (`server/lib/read-pool.js`), so a slow page never delays ingest. diff --git a/scripts/import-history.js b/scripts/import-history.js index b0cedc11..41a2b5f5 100644 --- a/scripts/import-history.js +++ b/scripts/import-history.js @@ -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. * @@ -2024,6 +2065,25 @@ async function syncDefaultProjects(dbModule, options = {}) { 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 session = await parseSessionForImport(projPath, sourcePath); // Record the mtime even when the file yields no session, so we don't diff --git a/server/README.md b/server/README.md index 7aa72d63..ea567128 100644 --- a/server/README.md +++ b/server/README.md @@ -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 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) -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 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_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 diff --git a/server/__tests__/backoff-poll.test.js b/server/__tests__/backoff-poll.test.js new file mode 100644 index 00000000..ade35d60 --- /dev/null +++ b/server/__tests__/backoff-poll.test.js @@ -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ĩ + */ + +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); + }); +}); diff --git a/server/__tests__/codex-usage.test.js b/server/__tests__/codex-usage.test.js index 554336b1..fb92a986 100644 --- a/server/__tests__/codex-usage.test.js +++ b/server/__tests__/codex-usage.test.js @@ -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 }]); + }); +}); diff --git a/server/__tests__/session-sync.test.js b/server/__tests__/session-sync.test.js index bc35afb3..afd7df7f 100644 --- a/server/__tests__/session-sync.test.js +++ b/server/__tests__/session-sync.test.js @@ -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"); + }); +}); diff --git a/server/__tests__/transcript-marks.test.js b/server/__tests__/transcript-marks.test.js new file mode 100644 index 00000000..82c4d244 --- /dev/null +++ b/server/__tests__/transcript-marks.test.js @@ -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ĩ + */ + +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); + }); +}); diff --git a/server/index.js b/server/index.js index 7e503bbf..28668678 100644 --- a/server/index.js +++ b/server/index.js @@ -825,18 +825,35 @@ function startBackgroundServices() { } // 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 - // 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); if (Number.isFinite(LANE_SYNC_MS) && LANE_SYNC_MS > 0) { 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 { broadcastForeignLaneChanges(); } catch (err) { 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 // legacy backfill above runs only once (marker-gated), so a project added diff --git a/server/lib/antigravity-usage.js b/server/lib/antigravity-usage.js index a415d53f..ab4453a2 100644 --- a/server/lib/antigravity-usage.js +++ b/server/lib/antigravity-usage.js @@ -425,6 +425,9 @@ function writeSession(f, conv, now) { } 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 @@ -440,12 +443,16 @@ function ingestAntigravityUsage({ running = (async () => { let read = 0; let written = 0; + let pending = 0; // still-fresh sessions a later pass may have to settle for (const f of conversationFiles()) { try { const seen = getSeen.get(f.path); if (seen && seen.mtime_ms === 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); if (s && s.tool === "antigravity" && s.status === "active") { const at = new Date(a.last).toISOString(); @@ -486,6 +493,7 @@ function ingestAntigravityUsage({ } await new Promise((r) => setImmediate(r)); } + lastPending = pending; return { files: read, sessions: written }; })().finally(() => { running = null; @@ -504,16 +512,16 @@ function startAntigravityUsagePoll(broadcast, broadcastLane) { if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return; // Files an older build could not read get another try. db.prepare("DELETE FROM antigravity_conversations WHERE session_id = ''").run(); - const tick = () => { - ingestAntigravityUsage({ broadcast, broadcastLane }) - .then((r) => { + // Slowed to once a minute after a pass that read nothing, wrote nothing and + // left no fresh session to settle; back to POLL_MS once one finds work. + require("./backoff-poll").startBackoffPoll( + () => + ingestAntigravityUsage({ broadcast, broadcastLane }).then((r) => { if (r.sessions > 0 && pricingSync.syncEnabled()) void pricingSync.syncPricing(db); - }) - .catch(() => {}); - }; - tick(); - const timer = setInterval(tick, POLL_MS); - if (timer.unref) timer.unref(); + return r.files + r.sessions + lastPending > 0; + }), + { baseMs: POLL_MS } + ); } module.exports = { diff --git a/server/lib/backoff-poll.js b/server/lib/backoff-poll.js new file mode 100644 index 00000000..3e4e64cb --- /dev/null +++ b/server/lib/backoff-poll.js @@ -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ĩ + */ + +/** 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 }; diff --git a/server/lib/codex-usage.js b/server/lib/codex-usage.js index c4b136bc..723ebb69 100644 --- a/server/lib/codex-usage.js +++ b/server/lib/codex-usage.js @@ -235,7 +235,9 @@ function threadNames(home = getCodexHome()) { const names = new Map(); let text = ""; 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 { return names; } @@ -251,6 +253,17 @@ function threadNames(home = getCodexHome()) { } 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( `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')) @@ -400,6 +413,59 @@ function writeSession(sessionId, names, now, origin = LOCAL) { 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 * second call for the same home while one runs joins it. `home`, `source` and @@ -413,21 +479,42 @@ function ingestCodexUsage({ home, source = LOCAL.source, activeMs = LOCAL.activeMs, + recent = false, } = {}) { const key = home || ""; if (running.has(key)) return running.get(key); const pass = (async () => { - const files = allRolloutFiles(home); + const files = recent ? recentRolloutFiles(home, now) : allRolloutFiles(home); const byThread = new Map(); for (const f of files) { const m = FILE_NAME.exec(path.basename(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(); let read = 0; for (const f of files) { try { 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); let offset = row ? row.offset : 0; let state = row ? JSON.parse(row.state) : {}; @@ -439,7 +526,8 @@ function ingestCodexUsage({ let skip = null; const parent = offset === 0 ? firstLineMeta(f.path).forked_from_id : state.forkedFrom; 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) console.warn(`[codex-usage] fork parent ${parent} not found; ${f.path} may overcount`); } @@ -508,10 +596,15 @@ function startCodexUsagePoll(broadcast) { .memberHomes() .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 total = { sessions: 0 }; + const recent = passNo++ % FULL_EVERY !== 0; for (const home of homes()) { - const r = await ingestCodexUsage({ broadcast, home }); + const r = await ingestCodexUsage({ broadcast, home, recent }); total.sessions += r?.sessions || 0; } return total; diff --git a/server/lib/cursor-usage.js b/server/lib/cursor-usage.js index deb6412e..053d3a2b 100644 --- a/server/lib/cursor-usage.js +++ b/server/lib/cursor-usage.js @@ -182,14 +182,7 @@ function writeSession(f, now) { const endedAt = active ? null : lastWrite; const name = firstPrompt(f.path) || existing?.name || `Cursor ${f.id.slice(0, 8)}`; if (existing) { - updateSession.run( - name, - status, - endedAt, - lastWrite, - cwd, - f.id - ); + updateSession.run(name, status, endedAt, lastWrite, cwd, f.id); return "session_updated"; } insertSession.run( @@ -206,6 +199,9 @@ function writeSession(f, now) { } 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 @@ -217,11 +213,15 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) { running = (async () => { let read = 0; let written = 0; + let pending = 0; // still-fresh sessions a later pass may have to settle for (const f of transcriptFiles()) { try { const seen = getSeen.get(f.path); 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); if (s && s.tool === "cursor" && s.status === "active") { 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)); } + lastPending = pending; return { files: read, sessions: written }; })().finally(() => { running = null; @@ -249,18 +250,21 @@ function ingestCursorUsage({ broadcast = () => {}, now = Date.now() } = {}) { 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) { const POLL_MS = process.env.DASHBOARD_CURSOR_POLL_MS ? Number(process.env.DASHBOARD_CURSOR_POLL_MS) : 15_000; if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return; - const tick = () => { - ingestCursorUsage({ broadcast }).catch(() => {}); - }; - tick(); - const timer = setInterval(tick, POLL_MS); - if (timer.unref) timer.unref(); + require("./backoff-poll").startBackoffPoll( + () => ingestCursorUsage({ broadcast }).then((r) => r.files + r.sessions + lastPending > 0), + { baseMs: POLL_MS } + ); } module.exports = { diff --git a/server/routes/hooks.js b/server/routes/hooks.js index 107ba529..ecc105fc 100644 --- a/server/routes/hooks.js +++ b/server/routes/hooks.js @@ -294,6 +294,46 @@ const turnDurationStampsStmt = db.prepare( "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 `-main` agent, created and broadcast. */ function createMainAgent(session) { const mainAgentId = `${session.id}-main`; @@ -944,6 +984,11 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => { const result = transcriptCache.extract(transcriptPath); if (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 // 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. // Each isCompactSummary entry in the JSONL = one compaction that occurred. // Deduplicate by uuid so we only create once per compaction. - if (compaction) { - for (const entry of compaction.entries) { + if (compaction && !unchanged) { + const fresh = tailAfter( + compaction.entries, + mark?.compactionUuid, + (e, uuid) => e.uuid === uuid + ); + for (const entry of fresh) { const compactId = `${sessionId}-compact-${entry.uuid}`; 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 // the 1h cache-write split and server-tool request counts. 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.) - if (result.errors) { + const freshErrors = unchanged + ? null + : tailAfter(result.errors, mark?.lastError, (a, b) => a === b); + if (freshErrors && freshErrors.length) { let newErrorRecorded = false; // Deduplicate against the errors already recorded, read ONCE per hook: // `result.errors` is the transcript's whole history, not just new ones. 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}`; if (recorded.has(errSummary)) continue; recorded.add(errSummary); @@ -1136,13 +1189,16 @@ const processEventTx = db.transaction((hookType, data, tool = "claude") => { } // 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 // 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 // a 763-turn session before. 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(); if (recorded.has(tdTs)) continue; 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) if (result.usageExtras || result.thinkingBlockCount > 0) { 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. */ 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 @@ -2176,6 +2252,7 @@ const watchdogTimer = setInterval(watchdogCheck, WATCHDOG_INTERVAL_MS); if (watchdogTimer.unref) watchdogTimer.unref(); router.transcriptCache = transcriptCache; +router.transcriptMarks = transcriptMarks; router.watchdogCheck = watchdogCheck; router.livenessReap = livenessReap; module.exports = router;