feat: analytics parity — tool cost, date ranges, budgets, query explorer, replay, snapshots (v2.4.0)
Analytics parity phases 2–7 (docs/plans/analytics-parity/), built on
138e706 and ported onto v2.3.1:
- Cost by AI tool (?tool= on cost/analytics/stats/workflows, by_tool,
Codex fast/priority and flex tiers, agenthub_cost_usd_total{tool,model})
- Date ranges (from/to) on cost/analytics/workflows; Analytics range select
with Today / This week (Monday) / This month (1st) / 7d / 30d / 90d / All /
Custom; workflow filters carried into workflow-stats.js (read worker)
- Budgets and cost alerts as alert rule types (alert_rules rebuilt without
its rule_type CHECK), GET /api/budgets/status, rule preview, spend pace,
status-bar budget chip; rules stay editable in place
- Query Explorer (/query, allowlisted sort, CSV/JSON export) and scheduled
reports (report_definitions/report_runs, Settings → Reports)
- Session Replay tab and immutable snapshots (session_snapshots, /snapshot/:token)
- Remote Codex sync, push ingest with a hashed token, privacy redaction and
ignored folders (Settings → Data), Cursor sessions (no cost)
- Read-only MCP server, statusline, Grafana/Prometheus files
Also: resizable sidebar, pet run/wave animations, the workspace home keeps a
half-typed prompt, and the pre-commit hook runs only the related tests for a
small commit.
This commit is contained in:
+84
-30
@@ -9,7 +9,9 @@
|
||||
* - line 1 `session_meta`: `id` = the thread, `session_id` = the ROOT session
|
||||
* (a sub-agent thread's parent; absent on old CLIs, then `id`), `cwd`,
|
||||
* `timestamp`, `forked_from_id` on a fork;
|
||||
* - `turn_context.model` is the model for the lines that follow;
|
||||
* - `turn_context.model` is the model for the lines that follow, and
|
||||
* `thread_settings_applied.thread_settings.service_tier` the tier (fast /
|
||||
* priority → `speed = fast`, flex → `service_tier = flex`, else standard);
|
||||
* - `token_count` events carry a cumulative `total_token_usage` and its delta
|
||||
* `last_token_usage` — `input_tokens` INCLUDES `cached_input_tokens`,
|
||||
* `output_tokens` includes reasoning. The same total is often sent twice.
|
||||
@@ -29,6 +31,13 @@
|
||||
* minutes) keeps the status, end and last-active time the hooks gave it; this
|
||||
* module still writes its usage, model and name. Hooks that stopped reporting
|
||||
* (removed, no longer trusted) hand the status back to the journal.
|
||||
*
|
||||
* A remote source (server/lib/remote-sync.js) mirrors another machine's
|
||||
* `sessions/` and `session_index.jsonl` into its staging dir and runs the same
|
||||
* pass over it (`home`), with its own `source` tag: a session row is only ever
|
||||
* written by passes of the source that created it, and the local settle never
|
||||
* touches a remote row. A session whose folder the user ignores
|
||||
* (`server/lib/privacy.js`) is never written.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
@@ -40,6 +49,7 @@ const { claimToolSessions } = require("./dashboard-runs");
|
||||
const { getCodexHome, codexHomes } = require("./codex-home");
|
||||
const { allRolloutFiles, FUTURE_SLACK_MS } = require("./codex-ingest");
|
||||
const pricingSync = require("./pricing-sync");
|
||||
const { isIgnoredCwd } = require("./privacy");
|
||||
|
||||
const ACTIVE_MS = 120_000; // same window codex-ingest.js calls a lane running
|
||||
// Read at most this much of one journal per pass: a sync read of a 100 MB
|
||||
@@ -48,12 +58,26 @@ const MAX_CHUNK = 16 * 1024 * 1024;
|
||||
const NEEDLES = [
|
||||
'"session_meta"',
|
||||
'"turn_context"',
|
||||
'"thread_settings_applied"',
|
||||
'"token_count"',
|
||||
'"user_message"', // a prompt, older CLIs
|
||||
'"UserMessage"', // a prompt, as an item_completed event (0.149+)
|
||||
];
|
||||
const FILE_NAME = /^rollout-\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}-(.+)\.jsonl$/;
|
||||
|
||||
// A usage bucket is keyed `model` (the default tier — and every state saved
|
||||
// before tiers were read) or `model␁speed␁service_tier`.
|
||||
const SEP = "\u0001";
|
||||
function usageKey(model, tier) {
|
||||
if (tier === "fast" || tier === "priority") return `${model}${SEP}fast${SEP}standard`;
|
||||
if (tier === "flex") return `${model}${SEP}standard${SEP}flex`;
|
||||
return model;
|
||||
}
|
||||
function splitUsageKey(key) {
|
||||
const [model, speed = "standard", serviceTier = "standard"] = key.split(SEP);
|
||||
return { model, speed, serviceTier };
|
||||
}
|
||||
|
||||
function totalKey(t) {
|
||||
return [
|
||||
t.input_tokens,
|
||||
@@ -102,6 +126,8 @@ function parseRolloutChunk(text, state = {}, skip = null) {
|
||||
s.forkedFrom = p.forked_from_id || null;
|
||||
} else if (o.type === "turn_context") {
|
||||
if (p.model) s.model = p.model;
|
||||
} else if (o.type === "event_msg" && p.type === "thread_settings_applied") {
|
||||
s.tier = p.thread_settings?.service_tier || null;
|
||||
} else if (o.type === "event_msg" && !s.firstPrompt && promptOf(p)) {
|
||||
// The session's label when Codex has no thread name for it — the same
|
||||
// first-line, 60-character rule hooks.js applies to a Claude prompt.
|
||||
@@ -117,7 +143,7 @@ function parseRolloutChunk(text, state = {}, skip = null) {
|
||||
if (skip) s.pastFork = true;
|
||||
}
|
||||
const l = p.info.last_token_usage || {};
|
||||
const u = (s.usage[s.model || "unknown"] ||= {
|
||||
const u = (s.usage[usageKey(s.model || "unknown", s.tier)] ||= {
|
||||
input: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
@@ -199,12 +225,12 @@ function threadOf(file) {
|
||||
return thread;
|
||||
}
|
||||
|
||||
/** `thread_name` per id from `$CODEX_HOME/session_index.jsonl` (last wins). */
|
||||
function threadNames() {
|
||||
/** `thread_name` per id from `<home>/session_index.jsonl` (last wins). */
|
||||
function threadNames(home = getCodexHome()) {
|
||||
const names = new Map();
|
||||
let text = "";
|
||||
try {
|
||||
text = fs.readFileSync(path.join(getCodexHome(), "session_index.jsonl"), "utf8");
|
||||
text = fs.readFileSync(path.join(home, "session_index.jsonl"), "utf8");
|
||||
} catch {
|
||||
return names;
|
||||
}
|
||||
@@ -234,8 +260,8 @@ const rolloutsOf = db.prepare(
|
||||
"SELECT thread_id, mtime_ms, state FROM codex_rollouts WHERE session_id = ?"
|
||||
);
|
||||
const insertCodexSession = db.prepare(
|
||||
`INSERT INTO sessions (id, name, status, cwd, model, started_at, ended_at, updated_at, metadata, tool)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'codex')`
|
||||
`INSERT INTO sessions (id, name, status, cwd, model, started_at, ended_at, updated_at, metadata, tool, source)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'codex', ?)`
|
||||
);
|
||||
const updateCodexSession = db.prepare(
|
||||
`UPDATE sessions SET name = ?, status = ?, model = ?, ended_at = ?, updated_at = ?,
|
||||
@@ -257,21 +283,29 @@ function hooksCurrent(sessionId, newestMtime) {
|
||||
}
|
||||
// A journal written "in the future" (a clock that runs ahead) is not proof of
|
||||
// activity — codex-ingest.js treats it the same way.
|
||||
const fresh = (mtime, now) => mtime - now < FUTURE_SLACK_MS && now - mtime < ACTIVE_MS;
|
||||
const fresh = (mtime, now, activeMs = ACTIVE_MS) =>
|
||||
mtime - now < FUTURE_SLACK_MS && now - mtime < activeMs;
|
||||
|
||||
// This machine's journals. A remote source passes its id and a longer window:
|
||||
// its mirror is only as fresh as its last sync.
|
||||
const LOCAL = { source: "local", activeMs: ACTIVE_MS };
|
||||
|
||||
/** Re-derive one root session's row and usage from all its threads. */
|
||||
function writeSession(sessionId, names, now) {
|
||||
function writeSession(sessionId, names, now, origin = LOCAL) {
|
||||
const rows = rolloutsOf.all(sessionId).map((r) => ({ ...r, state: JSON.parse(r.state) }));
|
||||
if (rows.length === 0) return null;
|
||||
const existing = stmts.getSession.get(sessionId);
|
||||
if (existing && existing.tool !== "codex") return null;
|
||||
// Another machine's row (or this machine's, seen from a remote pass).
|
||||
if (existing && (existing.source || "local") !== origin.source) return null;
|
||||
|
||||
// The root thread's own file, when it has been read: a sub-agent file read
|
||||
// first must not fix the session's cwd, start or model.
|
||||
const ownRoot = rows.find((r) => r.thread_id === sessionId);
|
||||
const root = ownRoot || rows[0];
|
||||
if (isIgnoredCwd(root.state.cwd)) return null;
|
||||
const newest = rows.reduce((a, b) => (b.mtime_ms > a.mtime_ms ? b : a));
|
||||
const active = rows.some((r) => fresh(r.mtime_ms, now));
|
||||
const active = rows.some((r) => fresh(r.mtime_ms, now, origin.activeMs));
|
||||
const name =
|
||||
names.get(sessionId) ||
|
||||
root.state.firstPrompt ||
|
||||
@@ -319,19 +353,21 @@ function writeSession(sessionId, names, now) {
|
||||
root.state.startedAt || new Date(now).toISOString(),
|
||||
endedAt,
|
||||
lastWrite,
|
||||
JSON.stringify({ source: "codex-rollout" })
|
||||
JSON.stringify({ source: "codex-rollout" }),
|
||||
origin.source
|
||||
);
|
||||
claimToolSessions(sessionId); // its workspace, before the broadcast
|
||||
// Every usage row of a Codex session is this module's, so it is rewritten
|
||||
// whole: a model gone from a re-read journal leaves no stale row behind.
|
||||
clearUsage.run(sessionId);
|
||||
for (const [m, t] of Object.entries(totals)) {
|
||||
for (const [key, t] of Object.entries(totals)) {
|
||||
const { model: m, speed, serviceTier } = splitUsageKey(key);
|
||||
stmts.setTokenUsage.run(
|
||||
sessionId,
|
||||
m,
|
||||
"standard",
|
||||
speed,
|
||||
"global",
|
||||
"standard",
|
||||
serviceTier,
|
||||
t.input,
|
||||
t.output,
|
||||
t.cacheRead,
|
||||
@@ -346,17 +382,26 @@ function writeSession(sessionId, names, now) {
|
||||
return existing ? "session_updated" : "session_created";
|
||||
}
|
||||
|
||||
let running = null;
|
||||
const running = new Map(); // home ("" = CODEX_HOME) → the pass in flight
|
||||
|
||||
/**
|
||||
* One pass over every journal. Yields to the event loop between files; a
|
||||
* second call while one runs joins it.
|
||||
* second call for the same home while one runs joins it. `home`, `source` and
|
||||
* `activeMs` are a remote source's (remote-sync.js); the defaults are this
|
||||
* machine's CODEX_HOME.
|
||||
* @returns {Promise<{files: number, sessions: number}>} files read, sessions written
|
||||
*/
|
||||
function ingestCodexUsage({ broadcast = () => {}, now = Date.now() } = {}) {
|
||||
if (running) return running;
|
||||
running = (async () => {
|
||||
const files = allRolloutFiles();
|
||||
function ingestCodexUsage({
|
||||
broadcast = () => {},
|
||||
now = Date.now(),
|
||||
home,
|
||||
source = LOCAL.source,
|
||||
activeMs = LOCAL.activeMs,
|
||||
} = {}) {
|
||||
const key = home || "";
|
||||
if (running.has(key)) return running.get(key);
|
||||
const pass = (async () => {
|
||||
const files = allRolloutFiles(home);
|
||||
const byThread = new Map();
|
||||
for (const f of files) {
|
||||
const m = FILE_NAME.exec(path.basename(f.path));
|
||||
@@ -403,10 +448,10 @@ function ingestCodexUsage({ broadcast = () => {}, now = Date.now() } = {}) {
|
||||
await new Promise((r) => setImmediate(r));
|
||||
}
|
||||
let written = 0;
|
||||
const names = touched.size ? threadNames() : new Map();
|
||||
const names = touched.size ? threadNames(home) : new Map();
|
||||
for (const sid of touched) {
|
||||
try {
|
||||
const type = writeSession(sid, names, now);
|
||||
const type = writeSession(sid, names, now, { source, activeMs });
|
||||
if (!type) continue;
|
||||
written += 1;
|
||||
broadcast(type, stmts.getSession.get(sid));
|
||||
@@ -416,9 +461,10 @@ function ingestCodexUsage({ broadcast = () => {}, now = Date.now() } = {}) {
|
||||
}
|
||||
return { files: read, sessions: written };
|
||||
})().finally(() => {
|
||||
running = null;
|
||||
running.delete(key);
|
||||
});
|
||||
return running;
|
||||
running.set(key, pass);
|
||||
return pass;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -447,20 +493,28 @@ function startCodexUsagePoll(broadcast) {
|
||||
if (timer.unref) timer.unref();
|
||||
}
|
||||
|
||||
/** Codex sessions still `active` whose every journal has gone quiet. */
|
||||
function settleActive(broadcast, now = Date.now()) {
|
||||
/**
|
||||
* Codex sessions of one source (this machine's by default) still `active`
|
||||
* whose every journal has gone quiet.
|
||||
*/
|
||||
function settleActive(
|
||||
broadcast,
|
||||
now = Date.now(),
|
||||
{ home, source = LOCAL.source, activeMs = LOCAL.activeMs } = {}
|
||||
) {
|
||||
const quiet = db
|
||||
.prepare(
|
||||
`SELECT s.id FROM sessions s WHERE s.tool = 'codex' AND s.status = 'active'
|
||||
`SELECT s.id FROM sessions s WHERE s.tool = 'codex' AND s.status = 'active' AND s.source = ?
|
||||
AND NOT EXISTS (SELECT 1 FROM codex_rollouts r WHERE r.session_id = s.id
|
||||
AND r.mtime_ms > ? AND r.mtime_ms < ?)`
|
||||
)
|
||||
.all(now - ACTIVE_MS, now + FUTURE_SLACK_MS);
|
||||
.all(source, now - activeMs, now + FUTURE_SLACK_MS);
|
||||
if (quiet.length === 0) return;
|
||||
const names = threadNames();
|
||||
const names = threadNames(home);
|
||||
for (const { id } of quiet) {
|
||||
if (hooksCurrent(id, newestRollout.get(id)?.m || 0)) continue;
|
||||
if (writeSession(id, names, now)) broadcast("session_updated", stmts.getSession.get(id));
|
||||
if (writeSession(id, names, now, { source, activeMs }))
|
||||
broadcast("session_updated", stmts.getSession.get(id));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user