perf: v2.5.10 — ready for 10+ concurrent users, leaner locally

Measured on a copy of a real 1.2 GB / 253K-event database, 10 simulated
browsers + 10 hooks/s for 90 s: event-loop delay p99 1239 ms → 32 ms (max
2.0 s → 94 ms), 266 failed requests → 0, 74 → 111 requests/s served, server
RSS 633 → 346 MB. Entry bundle 975 KB raw → 394 KB (106 KB brotli).

Server
- /api/analytics: the events-table scans run in the read worker
  (scoped-stats eventAggregates, read-pool job analyticsEvents); runRead takes
  maxAgeMs — analytics 10 s, /api/workflows 30 s — keyed by the args
- /api/workflows/skill-usage: +event_type and a tool_name range instead of
  LIKE, so it uses idx_events_tool (784 ms → a few ms)
- idx_events_session_type_tool (session_id, event_type, tool_name,
  created_at) replaces idx_events_session_type: covering reads for session
  stats, the hook dedupes and the import resume point (1.2 s → 20 ms); 64 MB
  cache_size and 256 MB mmap_size on both connections
- WAL: the server sets wal_autocheckpoint = 0 and checkpoints PASSIVE every
  DASHBOARD_WAL_CHECKPOINT_MS (5 s) from the read worker's own writable
  connection; a failed checkpoint restores the automatic one (hook COMMIT p95
  109 ms → 15 ms at 100 hooks/s)
- tmux: one list-sessions answers a whole GET /api/lanes and the heal beat
  (liveSessions), shared for 1 s and cleared by this process's new/kill —
  GET /api/run's sync spawns had blocked 60 s of a 90 s run; the heal beat
  skips tmux when nothing runs; attach options chained into 2 spawns
- hooks: TurnDuration/APIError deduped against one read per hook instead of a
  query per transcript entry (23 ms/hook on a 763-turn session); one lane
  scan per hook; hourly upserts skip rows that did not grow; frames are sent
  after the transaction commits (and after the pane binding); new_event
  carries its row id; a split pane's (slot 1–9) hooks no longer rebind the
  lane's session_id
- watchdog liveness probe runs ps asynchronously; transcript newline scan via
  indexOf; startup subagent backfill fingerprints the files it already checked
- pty attach: backpressure (pause past 1 MB unsent, resume under 256 KB)
- static: hashed assets served from their pre-compressed .br/.gz sibling

Client
- hidden tabs neither refetch on WS events nor poll (useLiveRefresh, panes,
  lane cards, git header, changes panel, settings); no per-tab hourly
  /updates/check (the server already checks)
- refetch storms throttled (workspace tree/session list 3 s, /api/run only on
  a run-key change, notification pane lookup only when shown); WS reconnect
  jitter; terminal retry backoff 1.5 → 30 s, stops on 1008
- heavier throttles for aggregates (15 s) and detail views; Monitor keepIfSame;
  GET in-flight dedupe in request()
- lazy Dashboard/History, xterm (with the first terminal), lane modals, Sagi
  (only while enabled); only the active locale loads; Workspace chunk
  prefetched on a /workspace deep link
- Sagi: no work while disabled, pupil via CSS variables, memo avatar
- build: .br/.gz siblings, sourcemap hidden; service worker no longer caches
  /assets/ (cache v3 drops the old one)

Tests: scripts/test-server.js drops the GIT_DIR-family variables a hook
exports, so a suite run from a linked worktree's pre-commit can no longer
commit into the real repository or set core.bare.

Docs: ARCHITECTURE, DATABASE, API, DEPLOYMENT, client README,
docs/plans/perf/README.md (audit, checklist, results).
This commit is contained in:
2026-10-07 13:24:15 +07:00
parent 8fdda66893
commit ca821008e2
75 changed files with 1711 additions and 699 deletions
+3 -1
View File
@@ -1813,11 +1813,13 @@ describe("Database Integrity", () => {
assert.ok(indexes.includes("idx_agents_session"));
assert.ok(indexes.includes("idx_agents_status"));
// (session_id, id) is NOT a prefix of idx_events_session_type: the hook
// (session_id, id) is NOT a prefix of idx_events_session_type_tool: the hook
// duplicate guard reads a session's latest event by id through it.
assert.ok(indexes.includes("idx_events_session_created"));
assert.ok(indexes.includes("idx_events_session"));
assert.ok(indexes.includes("idx_events_type"));
assert.ok(indexes.includes("idx_events_session_type_tool"));
assert.ok(!indexes.includes("idx_events_session_type"));
assert.ok(indexes.includes("idx_events_created"));
assert.ok(indexes.includes("idx_sessions_status"));
assert.ok(indexes.includes("idx_sessions_started"));
+47 -6
View File
@@ -2,7 +2,8 @@
* @file Tests the DASHBOARD_CLIENT_DIST override in server/index.js. A plugin
* install runs the server from a read-only plugin cache directory, so the
* client bundle is served from the writable runtime dir instead of the
* checkout's client/dist. Also asserts the API still answers when the
* checkout's client/dist, with a hashed asset's pre-compressed `.br`/`.gz`
* sibling sent to a browser that accepts it. Also asserts the API still answers when the
* configured bundle directory does not exist yet (the normal state before
* /agenthub-open builds it).
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
@@ -19,6 +20,16 @@ const TMP = fs.mkdtempSync(path.join(os.tmpdir(), "agenthub-clientdist-"));
const DIST = path.join(TMP, "client-dist");
fs.mkdirSync(DIST);
fs.writeFileSync(path.join(DIST, "index.html"), "<!doctype html><title>from-runtime</title>");
// A hashed bundle with the pre-compressed siblings the client build writes.
fs.mkdirSync(path.join(DIST, "assets"));
const BUNDLE = "console.log('plain bundle');".repeat(50);
fs.writeFileSync(path.join(DIST, "assets", "app-abc123.js"), BUNDLE);
fs.writeFileSync(
path.join(DIST, "assets", "app-abc123.js.br"),
require("zlib").brotliCompressSync(BUNDLE)
);
fs.writeFileSync(path.join(DIST, "assets", "app-abc123.js.gz"), require("zlib").gzipSync(BUNDLE));
fs.writeFileSync(path.join(DIST, "assets", "only-plain.css"), "body{}");
// Must be set BEFORE requiring the server: the data dir, discovery file and
// the static mount are all resolved at startup. NODE_ENV=production is the
@@ -38,12 +49,15 @@ const { db } = require("../db");
let server;
let BASE;
function get(urlPath) {
function get(urlPath, headers = {}) {
return new Promise((resolve, reject) => {
const req = http.get(new URL(urlPath, BASE), (res) => {
let body = "";
res.on("data", (c) => (body += c));
res.on("end", () => resolve({ status: res.statusCode, body }));
const req = http.get(new URL(urlPath, BASE), { headers }, (res) => {
const chunks = [];
res.on("data", (c) => chunks.push(c));
res.on("end", () => {
const raw = Buffer.concat(chunks);
resolve({ status: res.statusCode, body: raw.toString(), raw, headers: res.headers });
});
});
req.on("error", reject);
});
@@ -68,6 +82,33 @@ describe("DASHBOARD_CLIENT_DIST override", () => {
assert.match(res.body, /from-runtime/);
});
it("serves a hashed asset's pre-compressed sibling to a browser that takes it", async () => {
const zlib = require("zlib");
const br = await get("/assets/app-abc123.js", { "accept-encoding": "gzip, deflate, br" });
assert.equal(br.status, 200);
assert.equal(br.headers["content-encoding"], "br");
assert.match(br.headers["content-type"], /javascript/);
assert.match(br.headers.vary, /Accept-Encoding/i);
assert.match(br.headers["cache-control"], /immutable/);
assert.equal(zlib.brotliDecompressSync(br.raw).toString(), BUNDLE);
const gz = await get("/assets/app-abc123.js", { "accept-encoding": "gzip" });
assert.equal(gz.headers["content-encoding"], "gzip");
assert.equal(zlib.gunzipSync(gz.raw).toString(), BUNDLE);
const plain = await get("/assets/app-abc123.js", { "accept-encoding": "identity" });
assert.equal(plain.headers["content-encoding"], undefined);
assert.equal(plain.body, BUNDLE);
// No sibling on disk: the plain file, as before.
const css = await get("/assets/only-plain.css", { "accept-encoding": "br" });
assert.equal(css.headers["content-encoding"], undefined);
assert.equal(css.body, "body{}");
// A name that is not a flat file name never reaches the disk lookup.
const walk = await get("/assets/..%2findex.html", { "accept-encoding": "br" });
assert.equal(walk.headers["content-encoding"], undefined);
});
it("keeps the API working alongside the override", async () => {
const res = await get("/api/health");
assert.equal(res.status, 200);
+24
View File
@@ -541,6 +541,30 @@ describe("hook → lane binding", () => {
assert.equal((await request("GET", "/api/lanes")).body.lanes.length, before);
});
it("keeps the lane's session when a split pane (slot 1–9) of the workspace sends hooks", async () => {
const created = await request("POST", "/api/lanes", { cwd: "/tmp/lane-split-pane" });
const id = created.body.lane.id;
const hook = (session_id, run_id) =>
request("POST", "/api/hooks/event", {
hook_type: "PostToolUse",
run_id,
data: {
session_id,
cwd: "/tmp/lane-split-pane",
tool_name: "Read",
tool_use_id: session_id,
},
});
await hook("sess-main", `agenthub-lane-${id}`);
assert.equal((await request("GET", `/api/lanes/${id}`)).body.lane.session_id, "sess-main");
// A split pane's session never takes the lane over…
await hook("sess-split", `agenthub-lane-${id}-2`);
assert.equal((await request("GET", `/api/lanes/${id}`)).body.lane.session_id, "sess-main");
// …while a session outside the dashboard still does, as before.
await hook("sess-outside", undefined);
assert.equal((await request("GET", `/api/lanes/${id}`)).body.lane.session_id, "sess-outside");
});
it("clears needs_action only from the session that raised it, not from cross-session rebinds", async () => {
const created = await request("POST", "/api/lanes", {
cwd: "/tmp/lane-cross-session",
+7 -1
View File
@@ -68,7 +68,13 @@ describe("pty-attach", () => {
return "";
});
ptyAttach.attach(makeFakeWs(), "agenthub-lane-1", { cols: 80, rows: 24 });
assert.deepEqual(calls[0], ["set-option", "-t", "=agenthub-lane-1:", "mouse", "on"]);
assert.deepEqual(calls[0].slice(0, 5), [
"set-option",
"-t",
"=agenthub-lane-1:",
"mouse",
"on",
]);
});
it("attaches with the RUNNING tmux server's own binary, not the one on PATH", () => {
+39 -2
View File
@@ -1,8 +1,10 @@
/**
* @file Tests server/lib/read-pool.js: a read job runs in the worker on its own
* connection, identical calls in flight share one run, a job over the timeout
* is refused with EREADTIMEOUT and the next one gets a fresh worker, and
* AGENTHUB_READ_WORKER=0 runs the job inline.
* is refused with EREADTIMEOUT and the next one gets a fresh worker,
* AGENTHUB_READ_WORKER=0 runs the job inline, `maxAgeMs` reuses a young result,
* the analytics events job matches the main connection, and the WAL checkpoint
* runs on the worker's writable connection.
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
*/
@@ -63,6 +65,41 @@ describe("read-pool", () => {
assert.equal(data.stats.totalSessions, 1);
});
it("reuses a result younger than maxAgeMs, and only for the same args", async () => {
const first = await runRead(
"workflows",
{ sections: ["stats"], tag: "memo" },
{ maxAgeMs: 60_000 }
);
db.prepare("INSERT INTO sessions (id, status) VALUES ('rp-2', 'completed')").run();
const again = await runRead(
"workflows",
{ sections: ["stats"], tag: "memo" },
{ maxAgeMs: 60_000 }
);
assert.equal(again, first); // the cached object, not a new run
const fresh = await runRead("workflows", { sections: ["stats"], tag: "memo" });
assert.equal(fresh.stats.totalSessions, first.stats.totalSessions + 1);
db.prepare("DELETE FROM sessions WHERE id = 'rp-2'").run();
});
it("computes the analytics events numbers in the worker like the main connection", async () => {
const { eventAggregates } = require("../lib/scoped-stats");
const args = { scope: null, range: null, tzModifier: "+0 minutes" };
const inWorker = await runRead("analyticsEvents", args);
assert.deepEqual(inWorker, eventAggregates(db, null, null, "+0 minutes"));
assert.deepEqual(inWorker.toolUsage.map((r) => [r.tool_name, r.count]).sort(), [
["Edit", 2],
["Read", 2],
]);
});
it("checkpoints the WAL from the worker", async () => {
const [row] = await runRead("walCheckpoint");
assert.equal(row.busy, 0);
assert.ok(row.log >= row.checkpointed);
});
it("runs inline with AGENTHUB_READ_WORKER=0, and rejects an unknown job", async () => {
process.env.AGENTHUB_READ_WORKER = "0";
const data = await runRead("workflows", { sections: ["stats"], tag: "inline" });
+8 -2
View File
@@ -401,8 +401,14 @@ describe("watchdog liveness reap", () => {
const cwd = "/tmp/liveness-watchdog";
await seedSession(sid, cwd);
liveness.probeLiveCwds = () => ({ available: true, cwds: new Set() });
hooksRouter.watchdogCheck();
// The watchdog's beat probes asynchronously (`ps` off the event loop).
const realAsync = liveness.probeLiveCwdsAsync;
liveness.probeLiveCwdsAsync = async () => ({ available: true, cwds: new Set() });
try {
await hooksRouter.watchdogCheck();
} finally {
liveness.probeLiveCwdsAsync = realAsync;
}
assert.equal(stmts.getSession.get(sid).status, "completed");
});
@@ -143,4 +143,26 @@ describe("subagent token backfill", () => {
// a tokens key), so no session is re-scanned.
assert.equal(res.sessions, 0);
});
it("parses a row it can never stamp once, until its files change", async () => {
// A subagent row whose transcript holds no usage stays tokens-less forever,
// which used to re-parse its whole session tree on every boot.
stmts.insertAgent.run(
`${SESSION}-jsonl-ghost`,
SESSION,
"ghost",
"subagent",
"general-purpose",
"completed",
null,
MAIN,
JSON.stringify({ imported: true, source: "jsonl" })
);
fs.writeFileSync(path.join(tmpDir, SESSION, "subagents", "agent-ghost.jsonl"), "");
assert.equal((await importHistory.backfillSubagentTokenMetadata(dbModule)).sessions, 1);
assert.equal((await importHistory.backfillSubagentTokenMetadata(dbModule)).sessions, 0);
// A new subagent file appears: the session is looked at again.
fs.writeFileSync(path.join(tmpDir, SESSION, "subagents", "agent-new.jsonl"), "");
assert.equal((await importHistory.backfillSubagentTokenMetadata(dbModule)).sessions, 1);
});
});
+71 -8
View File
@@ -28,6 +28,35 @@ describe("tmux wrapper", () => {
assert.equal(tmux.hasSession("agenthub-lane-1"), false);
});
it("liveSessions answers every live name with ONE list-sessions call", () => {
const calls = [];
tmux.__setExecImpl((args) => {
calls.push(args[0]);
return "agenthub-lane-1\nagenthub-lane-12\nmain\n";
});
assert.deepEqual([...tmux.liveSessions()], ["agenthub-lane-1", "agenthub-lane-12", "main"]);
assert.deepEqual(calls, ["list-sessions"]);
});
it("liveSessions is empty when tmux answers that no server runs (or is not installed)", () => {
tmux.__setExecImpl(() => {
throw Object.assign(new Error("no server running"), { status: 1 });
});
assert.equal(tmux.liveSessions().size, 0);
tmux.__setExecImpl(() => {
throw Object.assign(new Error("spawn tmux ENOENT"), { code: "ENOENT" });
});
assert.equal(tmux.liveSessions().size, 0);
});
it("liveSessions is null when tmux could not be asked — not 'every pane is gone'", () => {
// Out of memory/processes: the panes are fine, the question failed.
tmux.__setExecImpl(() => {
throw Object.assign(new Error("spawnSync tmux EAGAIN"), { code: "EAGAIN" });
});
assert.equal(tmux.liveSessions(), null);
});
it("newSession names a command that exited at once instead of tmux's set-option error", () => {
tmux.__setExecImpl((args) => {
if (args[0] === "new-session") return "";
@@ -79,6 +108,22 @@ describe("tmux wrapper", () => {
calls = fake(true);
tmux.enableMouse("agenthub-lane-1");
assert.ok(calls.some((c) => c[0] === "set-option" && c[2] === "=agenthub-lane-1:"));
// tmux chains the four options into two spawns; psmux above got four.
const sets = calls.filter((c) => c[0] === "set-option");
assert.equal(sets.length, 2);
assert.deepEqual(sets[0], [
"set-option",
"-t",
"=agenthub-lane-1:",
"mouse",
"on",
";",
"set-option",
"-t",
"=agenthub-lane-1:",
"status",
"off",
]);
});
it("newSession builds the correct argv", () => {
@@ -104,16 +149,18 @@ describe("tmux wrapper", () => {
"--model",
"opus",
]);
assert.deepEqual(calls[1], ["set-option", "-t", "=agenthub-lane-1:", "mouse", "on"]);
assert.deepEqual(calls[2], ["set-option", "-t", "=agenthub-lane-1:", "status", "off"]);
assert.deepEqual(calls[3], ["set-option", "-t", "=agenthub-lane-1:", "pane-scrollbars", "on"]);
assert.deepEqual(calls[4], [
"set-option",
"-t",
"=agenthub-lane-1:",
"pane-scrollbars-style",
// One tmux call per `;`-chained group: the options every tmux knows, then
// the 3.7+ scrollbar pair (allowed to fail on its own).
assert.deepEqual(calls[1], [
...["set-option", "-t", "=agenthub-lane-1:", "mouse", "on", ";"],
...["set-option", "-t", "=agenthub-lane-1:", "status", "off"],
]);
assert.deepEqual(calls[2], [
...["set-option", "-t", "=agenthub-lane-1:", "pane-scrollbars", "on", ";"],
...["set-option", "-t", "=agenthub-lane-1:", "pane-scrollbars-style"],
"fg=brightwhite,bg=black,width=1,pad=0",
]);
assert.equal(calls.length, 3);
});
it("newSession still succeeds on a tmux older than 3.7 (no pane-scrollbars)", () => {
@@ -171,6 +218,22 @@ describe("tmux wrapper", () => {
assert.deepEqual(tmux.listSessions("agenthub-lane-"), ["agenthub-lane-1", "agenthub-lane-2"]);
});
it("shares one list-sessions answer for a second, until this process creates or kills one", () => {
const calls = [];
let names = "agenthub-lane-1\n";
tmux.__setExecImpl((args) => {
calls.push(args[0]);
return args[0] === "list-sessions" ? names : "";
});
assert.deepEqual(tmux.listSessions("agenthub-lane-"), ["agenthub-lane-1"]);
assert.deepEqual([...tmux.liveSessions()], ["agenthub-lane-1"]);
assert.equal(calls.filter((c) => c === "list-sessions").length, 1);
tmux.killSession("agenthub-lane-1");
names = "";
assert.deepEqual(tmux.listSessions("agenthub-lane-"), []);
assert.equal(calls.filter((c) => c === "list-sessions").length, 2);
});
it("listSessions returns [] when tmux has no sessions at all (exit 1)", () => {
tmux.__setExecImpl(() => {
const e = new Error("no server running");
+26 -28
View File
@@ -151,6 +151,11 @@ db.pragma("synchronous = NORMAL");
// (an import, a cleanup) left a WAL of hundreds of MB that never shrank while
// any dashboard kept the database open. See also `maintainDatabase`.
db.pragma("journal_size_limit = 67108864");
// A 64 MB page cache (default 2 MB) and memory-mapped reads: the aggregates
// re-read the same index pages, 5–45 % faster on a large DB. Both are per
// connection and cost address space, not a fixed allocation.
db.pragma("cache_size = -65536");
db.pragma("mmap_size = 268435456");
// Every event insert stores its payload through `trim_event_data(?)`
// (`server/lib/event-payload.js`): whole-file mirrors and long output stay in
// the transcript, not in `events.data`.
@@ -322,7 +327,7 @@ db.exec(`
-- read walked the rows' overflow chains (created_at sits after data).
CREATE INDEX IF NOT EXISTS idx_events_session_created ON events(session_id, created_at);
-- (session_id, id): a session's events in insert order, NOT a prefix of
-- idx_events_session_type (that one is not id-ordered). Every hook's
-- idx_events_session_type_tool (that one is not id-ordered). Every hook's
-- duplicate guard reads a session's latest event through it; dropped, that
-- read sorted the whole session per hook.
CREATE INDEX IF NOT EXISTS idx_events_session ON events(session_id, id);
@@ -335,7 +340,14 @@ db.exec(`
CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC);
-- Composite indexes for frequent query patterns (columns that exist at table creation time)
CREATE INDEX IF NOT EXISTS idx_events_session_type ON events(session_id, event_type);
-- (session_id, event_type, tool_name, created_at) covers a session's per-type,
-- per-tool and newest-per-type reads (session stats, the import's resume
-- point) without touching the rows: 1.2 s → 20 ms on a 29K-event session. It
-- replaces the (session_id, event_type) index it extends, so a hook still
-- writes the same number of index entries.
CREATE INDEX IF NOT EXISTS idx_events_session_type_tool
ON events(session_id, event_type, tool_name, created_at);
DROP INDEX IF EXISTS idx_events_session_type;
-- Subagent JSONL import dedups each tool event with
-- "WHERE agent_id = ? AND event_type = ? AND data LIKE '%tool_use_id%'".
-- Without an agent_id index that is a full events-table scan per tool event;
@@ -1963,6 +1975,16 @@ const stmts = {
web_search_requests = MAX(web_search_requests, excluded.web_search_requests),
web_fetch_requests = MAX(web_fetch_requests, excluded.web_fetch_requests),
code_execution_requests = MAX(code_execution_requests, excluded.code_execution_requests)
-- Every hook re-sends every hour of the transcript; most rows did not grow.
-- Skipping those leaves the values identical (every SET is a MAX) and
-- writes no page, which keeps the WAL — and its checkpoints — small.
WHERE excluded.input_tokens > input_tokens OR excluded.output_tokens > output_tokens
OR excluded.cache_read_tokens > cache_read_tokens
OR excluded.cache_write_tokens > cache_write_tokens
OR excluded.cache_write_1h_tokens > cache_write_1h_tokens
OR excluded.web_search_requests > web_search_requests
OR excluded.web_fetch_requests > web_fetch_requests
OR excluded.code_execution_requests > code_execution_requests
`),
getTokenTotals: db.prepare(`
SELECT
@@ -2035,22 +2057,8 @@ const stmts = {
matchPricing: db.prepare(
"SELECT * FROM model_pricing WHERE ? LIKE model_pattern ORDER BY LENGTH(model_pattern) DESC LIMIT 1"
),
toolUsageCounts: db.prepare(`
SELECT tool_name, COUNT(*) as count
FROM events
WHERE tool_name IS NOT NULL
GROUP BY tool_name
ORDER BY count DESC
LIMIT 20
`),
// Accept a timezone modifier (e.g. '-420 minutes') so GROUP BY uses local dates
dailyEventCounts: db.prepare(`
SELECT DATE(created_at, ?) as date, COUNT(*) as count
FROM events
WHERE created_at >= DATE('now', '-365 days')
GROUP BY 1
ORDER BY date ASC
`),
// Accept a timezone modifier (e.g. '-420 minutes') so GROUP BY uses local dates.
// The events-table analytics live in scoped-stats.js's eventAggregates (read worker).
dailySessionCounts: db.prepare(`
SELECT DATE(started_at, ?) as date, COUNT(*) as count
FROM sessions
@@ -2066,16 +2074,6 @@ const stmts = {
ORDER BY count DESC
`),
totalSubagentCount: db.prepare("SELECT COUNT(*) as count FROM agents WHERE type = 'subagent'"),
eventTypeCounts: db.prepare(`
SELECT event_type, COUNT(*) as count
FROM events
GROUP BY event_type
ORDER BY count DESC
`),
avgEventsPerSession: db.prepare(`
SELECT ROUND(CAST(COUNT(*) AS REAL) / MAX(1, (SELECT COUNT(*) FROM sessions)), 1) as avg
FROM events
`),
// Per-session aggregations powering the SessionOverview panel.
sessionEventCount: db.prepare("SELECT COUNT(*) as count FROM events WHERE session_id = ?"),
+61 -5
View File
@@ -51,6 +51,7 @@ if (require.main === module) require("./lib/claude-home").migrateLegacyDataDir()
const express = require("express");
const cors = require("cors");
const fs = require("fs");
const path = require("path");
const http = require("http");
const swaggerUi = require("swagger-ui-express");
@@ -216,6 +217,30 @@ function startServer(app, port) {
// so they must revalidate every load — without this the browser's
// heuristic cache happily serves a stale index.html that references
// asset hashes that no longer exist on disk.
//
// A hashed asset's `.br`/`.gz` sibling (written by the client build) goes
// to a browser that accepts it: ~75 % fewer bytes over a network, and no
// compression CPU here. Only a flat file name is looked up.
const assetsDir = path.join(clientDist, "assets");
const ENCODINGS = [
["br", ".br", /\bbr\b/],
["gzip", ".gz", /\bgzip\b/],
];
app.use("/assets", (req, res, next) => {
res.setHeader("Vary", "Accept-Encoding");
if (req.method !== "GET" && req.method !== "HEAD") return next();
const name = req.path.slice(1);
if (!/^[\w.-]+$/.test(name) || name.startsWith(".")) return next();
const accept = String(req.headers["accept-encoding"] || "");
for (const [encoding, ext, re] of ENCODINGS) {
if (!re.test(accept) || !fs.existsSync(path.join(assetsDir, name + ext))) continue;
res.setHeader("Content-Encoding", encoding);
res.setHeader("Cache-Control", "public, max-age=31536000, immutable");
res.type(path.extname(name)); // the original's type, not the sibling's
return res.sendFile(name + ext, { root: assetsDir }, (err) => err && next(err));
}
next();
});
app.use(
express.static(clientDist, {
etag: true,
@@ -377,6 +402,29 @@ function startBackgroundServices() {
}
}
// WAL checkpoints off the event loop. SQLite's automatic checkpoint runs
// inside whichever COMMIT crosses 1000 pages — a hook's, on this thread —
// and measured up to 2.7 s per commit under load. This process stops doing
// that and checkpoints every few seconds from the read worker instead; a
// failed checkpoint puts the automatic one back. Other processes on the same
// file (a second dashboard, the CLI) keep their own setting. Off with
// DASHBOARD_WAL_CHECKPOINT_MS=0.
{
const every = Number(process.env.DASHBOARD_WAL_CHECKPOINT_MS ?? 5000);
if (every > 0) {
const { db } = require("./db");
const { runRead } = require("./lib/read-pool");
db.pragma("wal_autocheckpoint = 0");
const t = setInterval(() => {
runRead("walCheckpoint").catch(() => {
db.pragma("wal_autocheckpoint = 1000");
clearInterval(t);
});
}, every);
if (t.unref) t.unref();
}
}
// Scheduled reports (Settings → Reports). Off with DASHBOARD_REPORTS_MS=0.
require("./lib/reports").startReportScheduler();
@@ -619,18 +667,26 @@ function startBackgroundServices() {
}
}
// A codex/antigravity pane that exits fires no hook, so release its lane on
// a beat instead (one `tmux has-session` per lane with a run), and end the
// run record of any pane that is gone. Off with DASHBOARD_RUN_HEAL_MS=0.
// a beat instead (ONE `tmux list-sessions` per beat — each tmux call is
// synchronous and stalls every terminal's keystrokes), and end the run
// record of any pane that is gone. Off with DASHBOARD_RUN_HEAL_MS=0.
const HEAL_MS = Number(process.env.DASHBOARD_RUN_HEAL_MS ?? 10_000);
if (Number.isFinite(HEAL_MS) && HEAL_MS > 0) {
const { healAllRunIds } = require("./routes/lanes");
const { reconcileOrphans } = require("./lib/dashboard-runs");
const { hasSession } = require("./lib/tmux");
const { liveSessions } = require("./lib/tmux");
const anyRun = require("./db").db.prepare(
`SELECT EXISTS (SELECT 1 FROM lanes WHERE run_id IS NOT NULL)
OR EXISTS (SELECT 1 FROM dashboard_runs WHERE status IN ('running', 'spawning')) AS any`
);
const healTimer = setInterval(() => {
try {
healAllRunIds();
if (!anyRun.get().any) return; // nothing to heal: no tmux spawn at all
const live = liveSessions();
if (!live) return; // tmux could not be asked: skip, never end every run
healAllRunIds(live);
// A pane that quit (Ctrl+C, `exit`) ends its run record too.
reconcileOrphans(hasSession);
reconcileOrphans((id) => live.has(id));
} catch (err) {
console.warn("run heal failed:", err.message);
}
+2 -2
View File
@@ -12,8 +12,8 @@ const TranscriptCache = require("./transcript-cache");
/** Write every hour of one transcript parse (MAX per hour, see the statement). */
function writeHourly(sessionId, tokensByHour) {
// ponytail: rewrites every hour of the session on each hook — ~1 row per
// active hour per model, cheap; write only the newest hours if it ever shows.
// Runs every hour of the session on each hook — ~1 row per active hour per
// model; the statement leaves a row that did not grow untouched (no write).
for (const t of Object.values(tokensByHour || {})) {
stmts.upsertHourlyUsage.run(
sessionId,
+8 -6
View File
@@ -601,14 +601,15 @@ function recoverInterruptedProvisioning() {
* and a `codex` lane (see `idx_lanes_cwd_tool`). Omitted, it matches any tool,
* which stays correct for callers that already know they hold the lane's own
* cwd (there's only ever one match for that exact path per tool, and this
* caller shape doesn't care which).
* caller shape doesn't care which). `lanes` lets a caller that resolves several
* paths (a hook) read the lane table once.
*/
function resolveLaneByCwd(cwd, tool) {
function resolveLaneByCwd(cwd, tool, lanes = null) {
if (!cwd || typeof cwd !== "string") return null;
// A Windows cwd ends in and nests under `\`.
const target = cwd.replace(/[\\/]+$/, "");
let best = null;
for (const lane of listLanes()) {
for (const lane of lanes || listLanes()) {
if (tool && lane.tool !== tool) continue;
const base = lane.cwd.replace(/[\\/]+$/, "");
if (target === base || target.startsWith(`${base}/`) || target.startsWith(base + path.sep)) {
@@ -654,13 +655,14 @@ function absolutePathsIn(toolInput) {
* has no equivalent hook mechanism here) — so both lookups are pinned to the
* `claude` lane at each path, never the `codex` one that might share it.
*/
function resolveLaneForWork(sessionCwd, toolInput) {
function resolveLaneForWork(sessionCwd, toolInput, lanes = null) {
const all = lanes || listLanes();
let best = null;
for (const candidate of absolutePathsIn(toolInput)) {
const lane = resolveLaneByCwd(candidate, "claude");
const lane = resolveLaneByCwd(candidate, "claude", all);
if (lane && (!best || lane.cwd.length > best.cwd.length)) best = lane;
}
return best || resolveLaneByCwd(sessionCwd, "claude");
return best || resolveLaneByCwd(sessionCwd, "claude", all);
}
function classifyLiveness({ status, stage, ageSec }, deadSec = DEAD_SEC) {
+16 -1
View File
@@ -81,9 +81,24 @@ function attach(ws, runId, { cols, rows }) {
rows: rows || 24,
});
// Backpressure: a slow viewer (a remote browser on a thin link) must not
// queue a chatty pane's output in this process without bound. Past 1 MB
// unsent the pty is paused — tmux holds the pane and redraws — and resumed
// once the socket drains below 256 KB.
let paused = false;
const drained = () => {
if (paused && ws.bufferedAmount < 256 * 1024) {
paused = false;
pty.resume?.();
}
};
pty.onData((data) => {
try {
ws.send(data, { binary: true });
ws.send(data, { binary: true }, drained);
if (!paused && ws.bufferedAmount > 1024 * 1024) {
paused = true;
pty.pause?.();
}
} catch {
/* client gone between data event and send — safe to ignore */
}
+45 -16
View File
@@ -2,10 +2,11 @@
* @file Runs the dashboard's heavy read-only aggregations in a worker thread
* on its own SQLite connection, so a slow one never blocks the event loop that
* serves every other page, hook and websocket. Identical requests in flight are
* shared, a job over `AGENTHUB_READ_TIMEOUT_MS` (default 60 s) is answered with
* `EREADTIMEOUT` and its worker replaced, and `AGENTHUB_READ_WORKER=0` — or a
* worker that cannot start — runs the job inline as before. This file is also
* the worker's entry point.
* shared (and, when the caller asks, reused for a short window), a job over
* `AGENTHUB_READ_TIMEOUT_MS` (default 60 s) is answered with `EREADTIMEOUT` and
* its worker replaced, and `AGENTHUB_READ_WORKER=0` — or a worker that cannot
* start — runs the job inline as before. The server's periodic WAL checkpoint
* also runs here (`walCheckpoint`). This file is also the worker's entry point.
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
*/
@@ -19,6 +20,11 @@ const JOBS = {
{ status: a.status || null, tool: a.tool || null, range: a.range || null },
a.sections || null
),
// Not a read: the worker runs it on its own writable connection (below), so
// the checkpoint's copy into the database file never blocks the event loop.
walCheckpoint: (conn) => conn.pragma("wal_checkpoint(PASSIVE)", { simple: false }),
analyticsEvents: (conn, a) =>
require("./scoped-stats").eventAggregates(conn, a.scope, a.range, a.tzModifier),
};
function timeoutMs() {
@@ -35,6 +41,7 @@ let worker = null;
let seq = 0;
const pending = new Map(); // id → { resolve, reject, timer, w }
const inflight = new Map(); // job+args → Promise
const recent = new Map(); // job+args → { at, result } for callers passing maxAgeMs
function spawn(dbPath) {
const w = new Worker(__filename, { workerData: { agenthubReadWorker: true, dbPath } });
@@ -93,37 +100,59 @@ function dispatch(job, args) {
/**
* Run a named read job off the event loop. Concurrent calls with the same job
* and args share one run.
* and args share one run; with `maxAgeMs`, a result that young is reused too,
* so N open dashboards cost one query per window instead of N. Args are part
* of the key, so a result is only reused for the very same query.
*
* @param {keyof JOBS} job
* @param {object} [args] structured-cloneable
* @param {{maxAgeMs?: number}} [opts]
* @returns {Promise<any>}
*/
function runRead(job, args = {}) {
function runRead(job, args = {}, { maxAgeMs = 0 } = {}) {
if (!JOBS[job]) return Promise.reject(codedError(`unknown read job ${job}`, "EBADJOB"));
const key = `${job}:${JSON.stringify(args)}`;
const now = Date.now();
const fresh = recent.get(key);
if (fresh && now - fresh.at < maxAgeMs) return Promise.resolve(fresh.result);
const hit = inflight.get(key);
if (hit) return hit;
const p = dispatch(job, args).finally(() => inflight.delete(key));
const p = dispatch(job, args)
.then((result) => {
if (maxAgeMs > 0) {
// ponytail: drop-expired-on-write keeps this bounded by the keys used
// within one window; an LRU if a caller ever varies args per request.
for (const [k, v] of recent) if (now - v.at > 60_000) recent.delete(k);
recent.set(key, { at: Date.now(), result });
}
return result;
})
.finally(() => inflight.delete(key));
inflight.set(key, p);
return p;
}
if (!isMainThread && workerData?.agenthubReadWorker) {
// Same driver order as server/db.js: better-sqlite3, else node:sqlite.
let conn;
try {
const Database = require("better-sqlite3");
conn = new Database(workerData.dbPath);
} catch {
const Database = require("../compat-sqlite");
conn = new Database(workerData.dbPath);
}
const open = () => {
try {
const Database = require("better-sqlite3");
return new Database(workerData.dbPath);
} catch {
const Database = require("../compat-sqlite");
return new Database(workerData.dbPath);
}
};
const conn = open();
let writer = null; // walCheckpoint's connection, opened on first use
conn.pragma("busy_timeout = 5000");
conn.pragma("query_only = ON"); // this connection can never write
conn.pragma("cache_size = -65536"); // same read tuning as server/db.js
conn.pragma("mmap_size = 268435456");
parentPort.on("message", ({ id, job, args }) => {
try {
parentPort.postMessage({ id, result: JOBS[job](conn, args) });
const c = job === "walCheckpoint" ? (writer ??= open()) : conn;
parentPort.postMessage({ id, result: JOBS[job](c, args) });
} catch (err) {
parentPort.postMessage({ id, error: { message: err.message, code: err.code } });
}
+42
View File
@@ -197,7 +197,49 @@ function avgEventsPerSession(db, scope, range) {
.get(...scope.params, ...s.params, ...scope.params, ...e.params);
}
/**
* The analytics numbers that scan `events` (tool usage, daily events, event
* types, events per session) for a scope or none. Takes the connection as an
* argument so the read worker can run it (`analyticsEvents` in read-pool.js).
*/
function eventAggregates(db, scope, range, tzModifier) {
if (scope) {
return {
toolUsage: toolUsageCounts(db, scope, range),
dailyEvents: dailyEventCounts(db, scope, tzModifier),
eventTypes: eventTypeCounts(db, scope, range),
avgEvents: avgEventsPerSession(db, scope, range),
};
}
return {
toolUsage: db
.prepare(
`SELECT tool_name, COUNT(*) as count FROM events WHERE tool_name IS NOT NULL
GROUP BY tool_name ORDER BY count DESC LIMIT 20`
)
.all(),
dailyEvents: db
.prepare(
`SELECT DATE(created_at, ?) as date, COUNT(*) as count FROM events
WHERE created_at >= DATE('now', '-365 days') GROUP BY 1 ORDER BY date ASC`
)
.all(tzModifier),
eventTypes: db
.prepare(
"SELECT event_type, COUNT(*) as count FROM events GROUP BY event_type ORDER BY count DESC"
)
.all(),
avgEvents: db
.prepare(
`SELECT ROUND(CAST(COUNT(*) AS REAL) / MAX(1, (SELECT COUNT(*) FROM sessions)), 1) as avg
FROM events`
)
.get(),
};
}
module.exports = {
eventAggregates,
statsOverview,
agentStatusCounts,
sessionStatusCounts,
+35 -11
View File
@@ -16,7 +16,7 @@
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
*/
const { execFileSync } = require("node:child_process");
const { execFile, execFileSync } = require("node:child_process");
const fs = require("node:fs");
const path = require("node:path");
const { isInsideContainer } = require("../../scripts/install-hooks");
@@ -63,21 +63,45 @@ function probeDisabledByEnv() {
* with an empty set genuinely means no claude process is running.
*/
function probeLiveCwds() {
if (probeDisabledByEnv()) return UNAVAILABLE();
if (process.platform === "win32") return UNAVAILABLE();
if (isInsideContainer()) return UNAVAILABLE();
if (probeBlocked()) return UNAVAILABLE();
let psOut;
try {
psOut = execImpl("ps", ["-Ao", "pid=,args="], {
encoding: "utf8",
timeout: 5_000,
maxBuffer: 16 * 1024 * 1024,
});
psOut = execImpl("ps", PS_ARGS, PS_OPTS);
} catch {
return UNAVAILABLE();
}
return cwdsFromPs(psOut);
}
/**
* `probeLiveCwds` with `ps` run asynchronously — what the watchdog's 15 s
* beat uses, so the spawn (24–300 ms inside a large server process) never
* blocks the event loop. Same answer shape; resolves, never rejects.
*
* @returns {Promise<{ available: boolean, cwds: Set<string> }>}
*/
function probeLiveCwdsAsync() {
if (probeBlocked()) return Promise.resolve(UNAVAILABLE());
return new Promise((resolve) => {
execFile("ps", PS_ARGS, PS_OPTS, (err, out) => {
try {
resolve(err ? UNAVAILABLE() : cwdsFromPs(out));
} catch {
resolve(UNAVAILABLE());
}
});
});
}
const PS_ARGS = ["-Ao", "pid=,args="];
const PS_OPTS = { encoding: "utf8", timeout: 5_000, maxBuffer: 16 * 1024 * 1024 };
function probeBlocked() {
return probeDisabledByEnv() || process.platform === "win32" || isInsideContainer();
}
/** The cwds of the `claude` processes in a `ps -Ao pid=,args=` listing. */
function cwdsFromPs(psOut) {
const pids = [];
for (const line of psOut.split("\n")) {
const m = line.match(/^\s*(\d+)\s+(.*)$/);
@@ -120,4 +144,4 @@ function probeLiveCwds() {
return { available: true, cwds };
}
module.exports = { probeLiveCwds, isClaudeCommand, __setExecImpl };
module.exports = { probeLiveCwds, probeLiveCwdsAsync, isClaudeCommand, __setExecImpl };
+77 -16
View File
@@ -34,10 +34,35 @@ let exactTargets = null;
function __setExecImpl(fn, { exact = true } = {}) {
execImpl = fn;
exactTargets = exact;
listed = null;
}
function __reset() {
execImpl = (args) => execFileSync("tmux", args, EXEC_OPTS);
exactTargets = null;
listed = null;
}
// The `list-sessions` answer, shared for LIST_TTL_MS. Every open dashboard
// polls GET /api/run, and each synchronous spawn forks this whole process —
// measured 2,247 calls, 60 s of a 90 s run blocked, at 10 browsers. A session
// this process creates or kills clears it, so only a pane that exits on its
// own can read as live for up to a second. "No server" (exit 1) is an answer
// too and is shared; any other failure is not.
const LIST_TTL_MS = 1000;
let listed = null; // { at, out } | { at, err }
function listOutput() {
if (listed && Date.now() - listed.at < LIST_TTL_MS) {
if (listed.err) throw listed.err;
return listed.out;
}
try {
const out = String(execImpl(["list-sessions", "-F", "#{session_name}"]));
listed = { at: Date.now(), out };
return out;
} catch (err) {
listed = err.status === 1 || err.code === "ENOENT" ? { at: Date.now(), err } : null;
throw err;
}
}
/**
@@ -117,6 +142,7 @@ function envFlags(env) {
function newSession({ name, cwd, argv, env }) {
// No argv = tmux's own default shell (a Windows Terminal pane).
const cmd = argv.length ? ["--", ...argv] : [];
listed = null;
execImpl(["new-session", "-d", "-s", name, "-c", cwd, ...envFlags(env), ...cmd]);
try {
enableMouse(name);
@@ -144,25 +170,37 @@ function newSession({ name, cwd, argv, env }) {
// The status bar goes off here too: the dashboard's tab bar already names the
// pane, and tmux's green line only eats a row of the terminal.
function enableMouse(name) {
execImpl(["set-option", "-t", pane(name), "mouse", "on"]);
execImpl(["set-option", "-t", pane(name), "status", "off"]);
// tmux runs `;`-separated commands from ONE argv (each tmux call is a
// synchronous spawn on the event loop), stopping at the first that fails —
// so the options every tmux knows go first, in their own call. psmux (no
// `=name` targets) is not known to chain, so it gets one call per option.
const run = (...cmds) => {
if (exact()) return execImpl(cmds.flatMap((c, i) => (i ? [";", ...c] : c)));
for (const c of cmds) execImpl(c);
};
run(
["set-option", "-t", pane(name), "mouse", "on"],
["set-option", "-t", pane(name), "status", "off"]
);
// A draggable scrollbar beside the pane (mouse mode makes it drag). Only a
// tmux SERVER of 3.7+ knows the option; an older one refuses it, which is
// fine — the pane just has none (`/agenthub-doctor` says how to upgrade).
try {
execImpl(["set-option", "-t", pane(name), "pane-scrollbars", "on"]);
// tmux's default (`fg=white,bg=black,width=1`) is a one-column track in
// ANSI black — the terminal theme's border colour, almost the background —
// so it reads as no scrollbar at all. Bright slider. ONE column: tmux 3.7c
// reserves `width` columns but paints only the first, so a second one is
// never cleared and shows stale cells — a ghost second scrollbar.
execImpl([
"set-option",
"-t",
pane(name),
"pane-scrollbars-style",
"fg=brightwhite,bg=black,width=1,pad=0",
]);
run(
["set-option", "-t", pane(name), "pane-scrollbars", "on"],
// tmux's default (`fg=white,bg=black,width=1`) is a one-column track in
// ANSI black — the terminal theme's border colour, almost the background —
// so it reads as no scrollbar at all. Bright slider. ONE column: tmux 3.7c
// reserves `width` columns but paints only the first, so a second one is
// never cleared and shows stale cells — a ghost second scrollbar.
[
"set-option",
"-t",
pane(name),
"pane-scrollbars-style",
"fg=brightwhite,bg=black,width=1,pad=0",
]
);
} catch {
/* tmux < 3.7 */
}
@@ -239,6 +277,7 @@ function pasteText(name, text) {
/** Idempotent — a session that's already gone is not an error. */
function killSession(name) {
listed = null;
try {
execImpl(["kill-session", "-t", session(name)]);
} catch {
@@ -250,7 +289,7 @@ function killSession(name) {
function listSessions(prefix) {
let out;
try {
out = execImpl(["list-sessions", "-F", "#{session_name}"]);
out = listOutput();
} catch {
return [];
}
@@ -260,6 +299,27 @@ function listSessions(prefix) {
.filter((s) => s && s.startsWith(prefix));
}
/**
* Every live session name from ONE `list-sessions` — what a poll asks instead
* of a synchronous has-session per name, each of which blocks the event loop
* (and so every terminal keystroke) for a whole tmux spawn. Empty when no
* server runs (tmux exits 1) or tmux is not installed; null when the call
* itself failed (EAGAIN/ENOMEM under memory pressure), which proves nothing
* about the panes — a caller must not read that as "every pane is gone".
*/
function liveSessions() {
try {
return new Set(
listOutput()
.split("\n")
.map((s) => s.trim())
.filter(Boolean)
);
} catch (err) {
return err.status === 1 || err.code === "ENOENT" ? new Set() : null;
}
}
function isTmuxAvailable() {
return clientVersion() !== null;
}
@@ -297,6 +357,7 @@ module.exports = {
pasteText,
killSession,
listSessions,
liveSessions,
isTmuxAvailable,
clientVersion,
serverVersion,
+4 -3
View File
@@ -312,9 +312,10 @@ class TranscriptCache {
pos += got;
let lineStart = 0;
for (let i = 0; i < got; i++) {
if (buf[i] !== 0x0a) continue;
// indexOf finds each newline natively — a JS byte loop was ~100 ms of
// a 76 MB transcript's cold read on the event loop.
const view = buf.subarray(0, got);
for (let i = view.indexOf(0x0a); i !== -1; i = view.indexOf(0x0a, i + 1)) {
let line;
if (pendingLen) {
const need = pendingLen + (i - lineStart);
+20 -10
View File
@@ -14,6 +14,7 @@ const {
BAD_RANGE,
} = require("../lib/source-filter");
const scoped = require("../lib/scoped-stats");
const { runRead } = require("../lib/read-pool");
const { calculateCost, dailyTokenRows, costWhere } = require("./pricing");
@@ -34,12 +35,12 @@ const USAGE_TOKEN_COLS = {
* With a range, counts follow it and tokens/cost come from the cost rows of the
* days in it (`token_usage` has no date); the daily charts keep 365 days.
*/
function computeAnalytics({ scope = null, range = null, tzModifier = "+0 minutes" } = {}) {
function computeAnalytics(
{ scope = null, range = null, tzModifier = "+0 minutes" } = {},
events = scoped.eventAggregates(db, scope || (range ? ALL : null), range, tzModifier)
) {
const sc = scope || (range ? ALL : null);
const toolUsage = sc ? scoped.toolUsageCounts(db, sc, range) : stmts.toolUsageCounts.all();
const dailyEvents = sc
? scoped.dailyEventCounts(db, sc, tzModifier)
: stmts.dailyEventCounts.all(tzModifier);
const { toolUsage, dailyEvents, eventTypes, avgEvents } = events;
const dailySessions = sc
? scoped.dailySessionCounts(db, sc, tzModifier)
: stmts.dailySessionCounts.all(tzModifier);
@@ -56,10 +57,6 @@ function computeAnalytics({ scope = null, range = null, tzModifier = "+0 minutes
const totalSubagents = sc
? scoped.totalSubagentCount(db, sc, range)
: stmts.totalSubagentCount.get();
const eventTypes = sc ? scoped.eventTypeCounts(db, sc, range) : stmts.eventTypeCounts.all();
const avgEvents = sc
? scoped.avgEventsPerSession(db, sc, range)
: stmts.avgEventsPerSession.get();
// Total cost exactly as GET /api/pricing/cost computes it (baselines, the
// per-day rate, the same scope and range), so the two numbers always agree.
@@ -108,7 +105,20 @@ router.get("/", (req, res) => {
if (parseTool(req) === false) return res.status(400).json(BAD_TOOL);
const range = parseRange(req);
if (range === false) return res.status(400).json(BAD_RANGE);
res.json(computeAnalytics({ scope: parseScope(req), range, tzModifier }));
const scope = parseScope(req);
// The events scans run in the read worker (a 365-day daily count is ~0.5 s
// on a large DB), and a 10 s memo lets every open dashboard share one run.
runRead(
"analyticsEvents",
{ scope: scope || (range ? ALL : null), range, tzModifier },
{ maxAgeMs: 10_000 }
)
.then((events) => res.json(computeAnalytics({ scope, range, tzModifier }, events)))
.catch((err) =>
res
.status(err.code === "EREADTIMEOUT" ? 503 : 500)
.json({ error: { code: err.code, message: err.message } })
);
});
module.exports = router;
+110 -40
View File
@@ -7,7 +7,15 @@ const { Router } = require("express");
const { v4: uuidv4 } = require("uuid");
const dbModule = require("../db");
const { stmts, db } = dbModule;
const { broadcast } = require("../websocket");
const { broadcast: sendNow } = require("../websocket");
// Frames emitted while a hook's transaction is open wait in `held` for its
// commit (see processEvent); everywhere else they go out at once.
let held = null;
function broadcast(type, data) {
if (held) held.push([type, data]);
else sendNow(type, data);
}
const TranscriptCache = require("../lib/transcript-cache");
const { writeHourly } = require("../lib/hourly-usage");
const { scanAndImportSubagents } = require("../../scripts/import-history");
@@ -173,13 +181,16 @@ const LANE_DONE_HOOKS = new Set(["Stop", "SessionEnd"]);
* running/idle, so without this an adopted lane read "idle" for the entire
* time Claude was working in it.
*/
function touchLaneFromHook(hookType, data, tool = "claude") {
function touchLaneFromHook(hookType, data, tool = "claude", runId = null) {
try {
if (!data || !data.cwd) return;
// One read of the lane table for every path this hook resolves.
// ponytail: a scan per hook — fine at dozens of lanes; index by cwd prefix if lane counts grow.
const allLanes = lanesLib.listLanes();
// Pin to the lane of the CLI that sent the hook — a Claude hook never
// touches the codex lane sharing its directory, nor a Codex hook the
// claude one.
const lane = lanesLib.resolveLaneByCwd(data.cwd, tool);
const lane = lanesLib.resolveLaneByCwd(data.cwd, tool, allLanes);
if (!lane) return;
// Infer the lane's stage from this one tool event, best-effort. A real
@@ -203,7 +214,7 @@ function touchLaneFromHook(hookType, data, tool = "claude") {
// Credit the lane whose files/directory the tool actually touched, which
// is not always the session's lane - see resolveLaneForWork. Bookkeeping
// below still uses `lane`, the session's own.
const workLane = lanesLib.resolveLaneForWork(data.cwd, data.tool_input) || lane;
const workLane = lanesLib.resolveLaneForWork(data.cwd, data.tool_input, allLanes) || lane;
const detection = detect(getPipeline(workLane.pipeline), {
tool_name: data.tool_name,
tool_input: data.tool_input,
@@ -222,7 +233,14 @@ function touchLaneFromHook(hookType, data, tool = "claude") {
// resuming, so it neither clears needs_action nor moves the lane.
const childEvent = tool === "codex" && data._codexChild === true;
const idleNudge = hookType === "Notification" && IDLE_NUDGE_RE.test(data.message || "");
if (data.session_id && lane.session_id !== data.session_id) patch.session_id = data.session_id;
// The lane's session follows its slot-0 pane (or a session outside the
// dashboard), never a split pane's: only slot 0 is the lane's run, and two
// panes of one workspace would otherwise flip session_id — and broadcast
// the lane to every browser — on every hook.
const splitPane = typeof runId === "string" && /^agenthub-lane-\d+-[1-9]$/.test(runId);
if (data.session_id && lane.session_id !== data.session_id && !splitPane) {
patch.session_id = data.session_id;
}
if (hookType === "Notification" && !idleNudge) {
patch.needs_action = data.message || "needs you";
patch.needs_action_session = data.session_id || null;
@@ -260,6 +278,14 @@ function touchLaneFromHook(hookType, data, tool = "claude") {
}
const setSessionTool = db.prepare("UPDATE sessions SET tool = ? WHERE id = ?");
// What processEvent dedupes the transcript's API errors and turn durations
// against — one read per hook, not one per transcript entry.
const apiErrorSummariesStmt = db.prepare(
"SELECT summary FROM events WHERE session_id = ? AND event_type = 'APIError'"
);
const turnDurationStampsStmt = db.prepare(
"SELECT created_at FROM events WHERE session_id = ? AND event_type = 'TurnDuration'"
);
/** The session's `<id>-main` agent, created and broadcast. */
function createMainAgent(session) {
@@ -448,7 +474,7 @@ function applyFirstUserDescriptor(sessionId, result) {
if (refreshedAgent) broadcast("agent_updated", refreshedAgent);
}
const processEvent = db.transaction((hookType, data, tool = "claude") => {
const processEventTx = db.transaction((hookType, data, tool = "claude") => {
const sessionId = data.session_id;
if (!sessionId) return null;
@@ -1017,16 +1043,13 @@ const processEvent = db.transaction((hookType, data, tool = "claude") => {
// Register API errors from transcript (quota limits, rate limits, overloaded, etc.)
if (result.errors) {
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) {
// Deduplicate: check if we already recorded this error (same type+message+timestamp)
const errKey = `${apiErr.type}:${apiErr.timestamp || ""}`;
const existing = db
.prepare(
`SELECT 1 FROM events WHERE session_id = ? AND event_type = 'APIError'
AND summary = ? LIMIT 1`
)
.get(sessionId, `${apiErr.type}: ${apiErr.message}`);
if (existing) continue;
const errSummary = `${apiErr.type}: ${apiErr.message}`;
if (recorded.has(errSummary)) continue;
recorded.add(errSummary);
stmts.insertEvent.run(
sessionId,
@@ -1096,15 +1119,15 @@ const processEvent = db.transaction((hookType, data, tool = "claude") => {
// Register turn duration events from transcript
if (result.turnDurations) {
// `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) {
const tdTs = td.timestamp || new Date().toISOString();
// Deduplicate by checking if we already have this turn duration event
const existing = db
.prepare(
"SELECT 1 FROM events WHERE session_id = ? AND event_type = 'TurnDuration' AND created_at = ? LIMIT 1"
)
.get(sessionId, tdTs);
if (existing) continue;
if (recorded.has(tdTs)) continue;
recorded.add(tdTs);
const tdSummary = `Turn completed in ${(td.durationMs / 1000).toFixed(1)}s`;
// Stamped with the TRANSCRIPT's timestamp, not NOW — that is what the
@@ -1168,7 +1191,7 @@ const processEvent = db.transaction((hookType, data, tool = "claude") => {
// Bump session updated_at on every event
stmts.touchSession.run(sessionId);
stmts.insertEvent.run(
const inserted = stmts.insertEvent.run(
sessionId,
agentId,
eventType,
@@ -1179,6 +1202,8 @@ const processEvent = db.transaction((hookType, data, tool = "claude") => {
);
const event = {
// The row id (additive): lets a client drop a frame it already holds.
id: Number(inserted.lastInsertRowid),
session_id: sessionId,
agent_id: agentId,
event_type: eventType,
@@ -1190,6 +1215,31 @@ const processEvent = db.transaction((hookType, data, tool = "claude") => {
return event;
});
/**
* One hook's writes in one transaction, its WebSocket frames sent only once
* that transaction COMMITTED — in order, in the same tick. A frame sent from
* inside it described a row a later throw could still roll back.
*/
function processEvent(hookType, data, tool) {
return holdFrames(() => processEventTx(hookType, data, tool));
}
/** Run `fn` with frames held, then send them if it returned (nested: the
* outermost call sends). A throw drops them. */
function holdFrames(fn) {
if (held) return fn();
held = [];
let frames = null;
try {
const out = fn();
frames = held;
return out;
} finally {
held = null;
if (frames) for (const [type, payload] of frames) sendNow(type, payload);
}
}
// A Codex sub-agent thread's own lifecycle is not the user's session's: its
// Stop is not the end of the user's turn, nor its SessionStart a new session.
const CODEX_CHILD_DROPPED = new Set([
@@ -1414,26 +1464,38 @@ function ingestHook(body, pushSource = null) {
transcriptCache.extract(data.transcript_path);
}
const result = processEvent(hook_type, data, tool);
// The hook's frames also wait for bindRunSession: in server mode a member
// is sent only frames of sessions they own (broadcastAudience), and a new
// Codex/agy session has no workspace — no owner — until that stamp. They are
// dropped only when processEvent threw (nothing was committed).
held = [];
let result = null;
try {
result = processEvent(hook_type, data, tool);
if (result) {
if (pushSource) {
// Tagged in the same tick processEvent created it, so no local liveness
// sweep ever sees it as this machine's.
if (pushedNew) {
stmts.setSessionSource.run(pushSource, data.session_id);
broadcast("session_updated", stmts.getSession.get(data.session_id));
}
} else {
touchLaneFromHook(hook_type, data, tool, body.run_id);
bindRunSession(body.run_id, data.session_id, hook_type, data.source, tool);
}
}
} finally {
const frames = held;
held = null;
if (result) for (const [type, payload] of frames) sendNow(type, payload);
}
if (!result) {
return {
status: 400,
body: { error: { code: "MISSING_SESSION", message: "session_id is required in data" } },
};
}
if (pushSource) {
// Tagged in the same tick processEvent created it, so no local liveness
// sweep ever sees it as this machine's.
if (pushedNew) {
stmts.setSessionSource.run(pushSource, data.session_id);
broadcast("session_updated", stmts.getSession.get(data.session_id));
}
} else {
// ponytail: full lane scan per hook — fine at dozens of lanes; index by cwd prefix if lane counts grow.
touchLaneFromHook(hook_type, data, tool);
bindRunSession(body.run_id, data.session_id, hook_type, data.source, tool);
}
return { status: 200, body: { ok: true, event: result }, result, hook_type, data, tool };
}
@@ -1880,7 +1942,8 @@ function watchdogCheck() {
// sessions never flicker out on a transient mismatch. A false
// completion self-heals anyway: the next hook event reactivates the
// session via the existing needsReactivation path.
livenessReap();
// The probe's `ps` runs asynchronously here (not on the event loop).
return livenessReap({ asyncProbe: true });
} catch (err) {
// Watchdog is best-effort — log but never crash the server
console.warn("[WATCHDOG] Error during check:", err?.message || err);
@@ -1901,7 +1964,7 @@ function watchdogCheck() {
* existing reactivation path. Watchdog ticks keep the gate, where it still
* guards long-running steady-state work against transient probe misses.
*/
function livenessReap({ ignoreIdleGate = false } = {}) {
function livenessReap({ ignoreIdleGate = false, asyncProbe = false, probe = null } = {}) {
const fs = require("fs");
const path = require("path");
@@ -1924,7 +1987,14 @@ function livenessReap({ ignoreIdleGate = false } = {}) {
.all();
if (activeSessions.length === 0) return; // nothing to check — skip the ps/lsof cost
const probe = liveness.probeLiveCwds();
if (asyncProbe) {
// Re-enters with the answer; the session list is read again then, fresh.
return liveness
.probeLiveCwdsAsync()
.then((p) => livenessReap({ ignoreIdleGate, probe: p }))
.catch((err) => console.warn("[WATCHDOG] Liveness reap failed:", err?.message || err));
}
if (!probe) probe = liveness.probeLiveCwds();
if (!probe.available) return;
const now = Date.now();
for (const sess of activeSessions) {
+19 -9
View File
@@ -28,6 +28,7 @@ const laneFeatures = require("../lib/lane-features");
const proofLib = require("../lib/proof");
const { broadcast } = require("../websocket");
const runs = require("../lib/pty-run");
const tmux = require("../lib/tmux");
const { resumeProxy } = require("../lib/proxy-apply");
const { sameOriginGuard } = require("./run");
const { writeLaneFile } = require("../lib/lane-file-write");
@@ -115,10 +116,14 @@ function lastEventAge(lane) {
* it, exactly like the old push-based handler did, just pulled instead of
* pushed.
*/
function healRunId(lane) {
function healRunId(lane, live = null) {
if (!lane.run_id) return lane;
const run = runs.getRun(lane.run_id);
if (run && run.status === "running") return lane;
if (live) {
if (live.has(lane.run_id)) return lane;
} else {
const run = runs.getRun(lane.run_id);
if (run && run.status === "running") return lane;
}
lanesLib.updateLane(lane.id, { run_id: null, status: "idle" });
broadcastLane(lane.id);
return lanesLib.getLane(lane.id);
@@ -128,10 +133,11 @@ function healRunId(lane) {
* Heal every lane at once. A claude pane that exits fires a hook, which
* broadcasts the lane and so re-reads it; a codex or antigravity pane fires
* nothing, so without this sweep its lane read "running" until something
* else happened to fetch it.
* else happened to fetch it. `live` (a `tmux.liveSessions()` set) answers
* every lane from one tmux call instead of a has-session per lane.
*/
function healAllRunIds() {
for (const lane of lanesLib.listLanes()) if (lane.run_id) healRunId(lane);
function healAllRunIds(live = null) {
for (const lane of lanesLib.listLanes()) if (lane.run_id) healRunId(lane, live);
}
/**
@@ -171,8 +177,8 @@ function broadcastForeignLaneChanges() {
return out;
}
function payload(lane) {
const healed = healRunId(lane);
function payload(lane, live = null) {
const healed = healRunId(lane, live);
return lanesLib.lanePayload(healed, lastEventAge(healed));
}
@@ -195,7 +201,11 @@ function broadcastLane(id) {
}
router.get("/", (_req, res) => {
const lanes = lanesLib.listLanes().map(payload);
const rows = lanesLib.listLanes();
// ONE list-sessions answers every lane's run, instead of a synchronous
// has-session per lane (null when tmux could not be asked: per-lane then).
const live = rows.some((l) => l.run_id) ? tmux.liveSessions() : null;
const lanes = rows.map((l) => payload(l, live));
res.json({
lanes,
counts: {
+14 -4
View File
@@ -37,7 +37,14 @@ router.get("/", async (req, res) => {
}
}
try {
res.json(await runRead("workflows", { status: statusFilter, tool, range, sections }));
res.json(
await runRead(
"workflows",
{ status: statusFilter, tool, range, sections },
// Every open Workflows page shares one run (~1.5 s on a large DB).
{ maxAgeMs: 30_000 }
)
);
} catch (err) {
if (err.code === "EREADTIMEOUT") {
return res.status(503).json({ error: { code: err.code, message: err.message } });
@@ -73,7 +80,7 @@ router.get("/skill-usage", (req, res) => {
COUNT(*) AS count,
COUNT(DISTINCT session_id) AS sessions
FROM events
WHERE tool_name = 'Skill' AND event_type = 'PreToolUse'
WHERE tool_name = 'Skill' AND +event_type = 'PreToolUse'
AND json_extract(data, '$.tool_input.skill') IS NOT NULL${sf.clause}
GROUP BY name ORDER BY count DESC LIMIT 30`
)
@@ -83,11 +90,14 @@ router.get("/skill-usage", (req, res) => {
return { ...r, plugin: i > 0 ? r.name.slice(0, i) : null };
});
// A range on tool_name ("`" sorts right after "_") instead of LIKE, and
// `+event_type` to keep the planner off idx_events_type: both make the
// query use idx_events_tool (784 ms → a few ms on 253K events).
const mcpCalls = db
.prepare(
`SELECT tool_name, COUNT(*) AS count FROM events
WHERE tool_name LIKE 'mcp__plugin\\_%' ESCAPE '\\'
AND event_type = 'PreToolUse'${sf.clause}
WHERE tool_name >= 'mcp__plugin_' AND tool_name < 'mcp__plugin\`'
AND +event_type = 'PreToolUse'${sf.clause}
GROUP BY tool_name`
)
.all(...sf.params);