Merge branch 'perf/local-ingest' into perf/local-opt

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/<conversation id>.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/<project slug>/agent-transcripts/<id>/<id>.jsonl`
 (`AGENTHUB_CURSOR_DIR` overrides `~/.cursor`). Each becomes a `sessions` row with

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)
 

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.
 

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

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

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ĩ <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);
+  });
+});

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 }]);
+  });
+});

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");
+  });
+});

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ĩ <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);
+  });
+});

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

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 = {

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ĩ <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 };

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;

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 = {

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 `<id>-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;