feat(db): trim stored hook payloads; price the session list like a session
Analytics parity, phase 1 (data hygiene), from the analytics-parity session. events.data is now stored through the SQL function trim_event_data(?) (server/lib/event-payload.js, registered in server/db.js) on every event INSERT and in the duplicate guard: Edit/MultiEdit/NotebookEdit/Write drop the whole-file mirrors (originalFile, oldString/newString, Write content), Read drops file.content/base64, and any string in tool_input/tool_response over 8 KB keeps a 6 KB head + 2 KB tail. structuredPatch and every field outside those two stay whole; _trimmed names what was dropped. The full text stays in the transcript and its snapshot. Rows stored before this are trimmed once in the background (sweepEventPayloads, 200 rows per write transaction, user_version 4). No VACUUM, no row pruned. The session list now prices every token column /api/pricing/cost/:id does (speed, geo, tier, the 1h cache write, server-tool fees) through one tokensForSessions helper, so a row's cost equals the session's own.
This commit is contained in:
+1
-1
@@ -796,7 +796,7 @@ db.pragma("journal_size_limit = 67108864"); // a checkpoint truncates the WAL b
|
|||||||
|
|
||||||
**Upkeep.** `maintainDatabase()` (exported by `server/db.js`) runs hourly from `server/index.js` (`DASHBOARD_DB_MAINTAIN_MS`, `0` = off): a `wal_checkpoint(PASSIVE)` — it never waits on a reader, such as the other dashboard — after which the next write shrinks the WAL to `journal_size_limit`, and `PRAGMA optimize`. Nothing runs `VACUUM`: it rewrites the whole file under an exclusive lock, so space freed by a cleanup is reused for new rows instead of returned to the disk.
|
**Upkeep.** `maintainDatabase()` (exported by `server/db.js`) runs hourly from `server/index.js` (`DASHBOARD_DB_MAINTAIN_MS`, `0` = off): a `wal_checkpoint(PASSIVE)` — it never waits on a reader, such as the other dashboard — after which the next write shrinks the WAL to `journal_size_limit`, and `PRAGMA optimize`. Nothing runs `VACUUM`: it rewrites the whole file under an exclusive lock, so space freed by a cleanup is reused for new rows instead of returned to the disk.
|
||||||
|
|
||||||
**Growth.** `events.data` keeps each hook's full payload (a `PostToolUse` carries the tool's whole output), and no data is pruned automatically — `events` is ~95% of a long-lived database (on the order of 300–500 MB a month of heavy use). The only deletions are the user's: Settings → clear data / cleanup, a workspace's purge, `scripts/clear-data.js`. Clear data runs in ONE transaction (foreign keys off around it, so workflows are deleted explicitly), cleanup deletes in 500-id chunks inside one transaction. `GET /api/settings/export` streams `events` in keyset pages of 500 instead of loading the table.
|
**Growth.** `events.data` is the hook payload TRIMMED by `trim_event_data(?)` (`server/lib/event-payload.js`, a SQL function registered on the connection in `server/db.js`, used by every event `INSERT` and by the duplicate guard): Edit/MultiEdit/NotebookEdit/Write drop `tool_response.originalFile`/`oldString`/`newString` (Write also `content`), Read drops `tool_response.file.content`/`base64`, and any string inside `tool_input`/`tool_response` over 8 KB keeps its first 6 KB and last 2 KB; `structuredPatch` and every field outside those two are kept, and `_trimmed` lists what was dropped. The full text stays in the session transcript and AgentHub's snapshot of it under `<data dir>/transcripts/`, which the Conversation tab reads. Rows stored before this are trimmed once in the background (`sweepEventPayloads`, 200 rows per transaction, `user_version` 4 after marker 3); measured on a 1.2 GB install it took `events.data` from 913 MB to ~400 MB. Freed pages are reused, not returned to the disk (no `VACUUM`). No ROW is pruned automatically. The only deletions are the user's: Settings → clear data / cleanup, a workspace's purge, `scripts/clear-data.js`. Clear data runs in ONE transaction (foreign keys off around it, so workflows are deleted explicitly), cleanup deletes in 500-id chunks inside one transaction. `GET /api/settings/export` streams `events` in keyset pages of 500 instead of loading the table.
|
||||||
|
|
||||||
### Prepared Statements
|
### Prepared Statements
|
||||||
|
|
||||||
|
|||||||
@@ -562,7 +562,7 @@ function importCompactions(dbModule, sessionId, mainAgentId, compactions) {
|
|||||||
if (!compactions || compactions.length === 0) return 0;
|
if (!compactions || compactions.length === 0) return 0;
|
||||||
const { db, stmts } = dbModule;
|
const { db, stmts } = dbModule;
|
||||||
const insertEvent = db.prepare(
|
const insertEvent = db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), ?)"
|
||||||
);
|
);
|
||||||
let created = 0;
|
let created = 0;
|
||||||
for (let i = 0; i < compactions.length; i++) {
|
for (let i = 0; i < compactions.length; i++) {
|
||||||
@@ -620,7 +620,7 @@ function importSubagents(dbModule, sessionId, mainAgentId, toolUses) {
|
|||||||
if (!toolUses || toolUses.length === 0) return 0;
|
if (!toolUses || toolUses.length === 0) return 0;
|
||||||
const { stmts } = dbModule;
|
const { stmts } = dbModule;
|
||||||
const insertEvent = dbModule.db.prepare(
|
const insertEvent = dbModule.db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), ?)"
|
||||||
);
|
);
|
||||||
|
|
||||||
let created = 0;
|
let created = 0;
|
||||||
@@ -679,7 +679,7 @@ function importApiErrors(dbModule, sessionId, mainAgentId, apiErrors) {
|
|||||||
if (!apiErrors || apiErrors.length === 0) return 0;
|
if (!apiErrors || apiErrors.length === 0) return 0;
|
||||||
const { db } = dbModule;
|
const { db } = dbModule;
|
||||||
const insertEvent = db.prepare(
|
const insertEvent = db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), ?)"
|
||||||
);
|
);
|
||||||
let created = 0;
|
let created = 0;
|
||||||
for (const err of apiErrors) {
|
for (const err of apiErrors) {
|
||||||
@@ -996,7 +996,7 @@ function importSubagentFromJsonl(dbModule, sessionId, mainAgentId, subData) {
|
|||||||
// importSession level (subagents have their own JSONL with separate usage).
|
// importSession level (subagents have their own JSONL with separate usage).
|
||||||
|
|
||||||
const insertEvent = db.prepare(
|
const insertEvent = db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), ?)"
|
||||||
);
|
);
|
||||||
|
|
||||||
// Spawn marker under the parent (main) agent — only emit once per subagent,
|
// Spawn marker under the parent (main) agent — only emit once per subagent,
|
||||||
@@ -1192,7 +1192,7 @@ function importSession(dbModule, session) {
|
|||||||
|
|
||||||
const mainAgentId = `${session.sessionId}-main`;
|
const mainAgentId = `${session.sessionId}-main`;
|
||||||
const insertEvent = db.prepare(
|
const insertEvent = db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), ?)"
|
||||||
);
|
);
|
||||||
const importedData = JSON.stringify({ imported: true });
|
const importedData = JSON.stringify({ imported: true });
|
||||||
let backfilled = false;
|
let backfilled = false;
|
||||||
@@ -1591,7 +1591,7 @@ function importSession(dbModule, session) {
|
|||||||
// Create synthetic events at actual message timestamps so the activity heatmap
|
// Create synthetic events at actual message timestamps so the activity heatmap
|
||||||
// reflects when work actually happened, not just session start/end.
|
// reflects when work actually happened, not just session start/end.
|
||||||
const insertEvent = db.prepare(
|
const insertEvent = db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), ?)"
|
||||||
);
|
);
|
||||||
const importedData = JSON.stringify({ imported: true });
|
const importedData = JSON.stringify({ imported: true });
|
||||||
|
|
||||||
|
|||||||
@@ -226,3 +226,32 @@ describe("TranscriptCache tokensByHour", () => {
|
|||||||
);
|
);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("session list cost", () => {
|
||||||
|
it("equals the session's own cost, speed and server-tool fees included", async () => {
|
||||||
|
session("list-fast", "2026-10-01T00:00:00.000Z");
|
||||||
|
// A fast-mode bucket with web searches: the list once priced only model + 4 token columns.
|
||||||
|
stmts.setTokenUsage.run(
|
||||||
|
"list-fast",
|
||||||
|
"daily-model",
|
||||||
|
"fast",
|
||||||
|
"global",
|
||||||
|
"standard",
|
||||||
|
M,
|
||||||
|
0,
|
||||||
|
0,
|
||||||
|
0,
|
||||||
|
0,
|
||||||
|
100,
|
||||||
|
0,
|
||||||
|
0
|
||||||
|
);
|
||||||
|
const own = await call("GET", "/api/pricing/cost/list-fast");
|
||||||
|
const list = await call("GET", "/api/sessions?q=list-fast");
|
||||||
|
const row = list.body.sessions.find((s) => s.id === "list-fast");
|
||||||
|
assert.ok(own.body.total_cost > 1, "fast + web search cost more than the plain input rate");
|
||||||
|
assert.equal(row.cost, own.body.total_cost);
|
||||||
|
const byPrice = await call("GET", "/api/sessions?q=list-fast&sort_by=price");
|
||||||
|
assert.equal(byPrice.body.sessions[0].cost, own.body.total_cost);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
@@ -0,0 +1,128 @@
|
|||||||
|
/**
|
||||||
|
* @file Tests for hook-payload trimming (`server/lib/event-payload.js`): what is
|
||||||
|
* dropped and kept, that every event insert goes through `trim_event_data`, that
|
||||||
|
* the duplicate guard still matches a trimmed payload, and the one-time sweep.
|
||||||
|
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||||
|
*/
|
||||||
|
|
||||||
|
const { describe, it } = require("node:test");
|
||||||
|
const assert = require("node:assert/strict");
|
||||||
|
const path = require("path");
|
||||||
|
const os = require("os");
|
||||||
|
|
||||||
|
const TMP = path.join(os.tmpdir(), `event-payload-${Date.now()}-${process.pid}`);
|
||||||
|
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 {
|
||||||
|
trimHookPayload,
|
||||||
|
trimEventData,
|
||||||
|
sweepEventPayloads,
|
||||||
|
STR_CAP,
|
||||||
|
} = require("../lib/event-payload");
|
||||||
|
const { db, stmts } = require("../db");
|
||||||
|
|
||||||
|
const big = (n, ch = "x") => ch.repeat(n);
|
||||||
|
|
||||||
|
describe("trimHookPayload", () => {
|
||||||
|
it("drops an Edit's whole-file mirror and keeps the patch", () => {
|
||||||
|
const patch = [{ oldStart: 1, lines: ["-a", "+b"] }];
|
||||||
|
const out = trimHookPayload({
|
||||||
|
tool_name: "Edit",
|
||||||
|
tool_input: { file_path: "/a.js", old_string: "a", new_string: "b" },
|
||||||
|
tool_response: {
|
||||||
|
originalFile: big(50),
|
||||||
|
oldString: "a",
|
||||||
|
newString: "b",
|
||||||
|
structuredPatch: patch,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
assert.equal(out.tool_response.originalFile, undefined);
|
||||||
|
assert.equal(out.tool_response.oldString, undefined);
|
||||||
|
assert.deepEqual(out.tool_response.structuredPatch, patch);
|
||||||
|
assert.equal(out.tool_input.old_string, "a");
|
||||||
|
assert.deepEqual(out._trimmed.sort(), [
|
||||||
|
"tool_response.newString",
|
||||||
|
"tool_response.oldString",
|
||||||
|
"tool_response.originalFile",
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("drops a Read's file body but keeps its counts", () => {
|
||||||
|
const out = trimHookPayload({
|
||||||
|
tool_name: "Read",
|
||||||
|
tool_response: {
|
||||||
|
type: "text",
|
||||||
|
file: { filePath: "/a", content: "hi", numLines: 1, totalLines: 9 },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
assert.deepEqual(out.tool_response.file, { filePath: "/a", numLines: 1, totalLines: 9 });
|
||||||
|
assert.deepEqual(out._trimmed, ["tool_response.file.content"]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps another tool's `content` (Grep matches), only capping it", () => {
|
||||||
|
const out = trimHookPayload({ tool_name: "Grep", tool_response: { content: "a:1:x" } });
|
||||||
|
assert.equal(out.tool_response.content, "a:1:x");
|
||||||
|
assert.equal(out._trimmed, undefined);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("caps long strings to head + tail, idempotently, and leaves other fields", () => {
|
||||||
|
const stdout = `HEAD${big(STR_CAP * 3)}TAIL`;
|
||||||
|
const data = {
|
||||||
|
tool_name: "Bash",
|
||||||
|
prompt_id: "p",
|
||||||
|
tool_response: { stdout, stderr: "", interrupted: false },
|
||||||
|
tool_input: { command: "ls" },
|
||||||
|
};
|
||||||
|
const once = trimHookPayload(data);
|
||||||
|
assert.ok(once.tool_response.stdout.length <= STR_CAP);
|
||||||
|
assert.ok(once.tool_response.stdout.startsWith("HEAD"));
|
||||||
|
assert.ok(once.tool_response.stdout.endsWith("TAIL"));
|
||||||
|
assert.equal(once.tool_response.interrupted, false);
|
||||||
|
assert.equal(once.prompt_id, "p");
|
||||||
|
assert.deepEqual(trimHookPayload(once), once);
|
||||||
|
assert.equal(data.tool_response.stdout, stdout, "input not mutated");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("caps a string-typed tool_response", () => {
|
||||||
|
const out = trimHookPayload({ tool_name: "Read", tool_response: big(STR_CAP * 2) });
|
||||||
|
assert.ok(out.tool_response.length <= STR_CAP);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe("trimEventData", () => {
|
||||||
|
it("passes short, null and non-JSON input through", () => {
|
||||||
|
assert.equal(trimEventData(null), null);
|
||||||
|
assert.equal(trimEventData('{"a":1}'), '{"a":1}');
|
||||||
|
const junk = big(STR_CAP + 10, "{");
|
||||||
|
assert.equal(trimEventData(junk), junk);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe("event inserts", () => {
|
||||||
|
const payload = JSON.stringify({
|
||||||
|
session_id: "s-trim",
|
||||||
|
tool_name: "Edit",
|
||||||
|
tool_response: { originalFile: big(STR_CAP * 4) },
|
||||||
|
});
|
||||||
|
|
||||||
|
it("store the trimmed payload and the duplicate guard still matches it", () => {
|
||||||
|
db.prepare("INSERT OR IGNORE INTO sessions (id) VALUES ('s-trim')").run();
|
||||||
|
stmts.insertEvent.run("s-trim", null, "PostToolUse", "Edit", "x", payload);
|
||||||
|
const row = db.prepare("SELECT data FROM events WHERE session_id = 's-trim'").get();
|
||||||
|
assert.ok(row.data.length < 500);
|
||||||
|
assert.ok(stmts.findRecentDuplicateEvent.get("s-trim", "PostToolUse", payload));
|
||||||
|
});
|
||||||
|
|
||||||
|
it("the sweep trims rows stored untrimmed", async () => {
|
||||||
|
db.prepare(
|
||||||
|
"INSERT INTO events (session_id, event_type, data, created_at) VALUES ('s-trim', 'PostToolUse', ?, '2026-01-01T00:00:00Z')"
|
||||||
|
).run(payload);
|
||||||
|
const r = await sweepEventPayloads(db, { chunk: 1 });
|
||||||
|
assert.equal(r.rows, 1);
|
||||||
|
assert.ok(r.bytesSaved > STR_CAP * 3);
|
||||||
|
const again = await sweepEventPayloads(db);
|
||||||
|
assert.equal(again.rows, 0);
|
||||||
|
});
|
||||||
|
});
|
||||||
+11
-3
@@ -151,6 +151,14 @@ db.pragma("synchronous = NORMAL");
|
|||||||
// (an import, a cleanup) left a WAL of hundreds of MB that never shrank while
|
// (an import, a cleanup) left a WAL of hundreds of MB that never shrank while
|
||||||
// any dashboard kept the database open. See also `maintainDatabase`.
|
// any dashboard kept the database open. See also `maintainDatabase`.
|
||||||
db.pragma("journal_size_limit = 67108864");
|
db.pragma("journal_size_limit = 67108864");
|
||||||
|
// 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`.
|
||||||
|
db.function(
|
||||||
|
"trim_event_data",
|
||||||
|
{ deterministic: true },
|
||||||
|
require("./lib/event-payload").trimEventData
|
||||||
|
);
|
||||||
|
|
||||||
db.exec(`
|
db.exec(`
|
||||||
CREATE TABLE IF NOT EXISTS sessions (
|
CREATE TABLE IF NOT EXISTS sessions (
|
||||||
@@ -1684,11 +1692,11 @@ const stmts = {
|
|||||||
findRecentDuplicateEvent: db.prepare(
|
findRecentDuplicateEvent: db.prepare(
|
||||||
`SELECT 1 FROM (SELECT event_type, data, created_at FROM events
|
`SELECT 1 FROM (SELECT event_type, data, created_at FROM events
|
||||||
WHERE session_id = ? ORDER BY id DESC LIMIT 1)
|
WHERE session_id = ? ORDER BY id DESC LIMIT 1)
|
||||||
WHERE event_type = ? AND data = ?
|
WHERE event_type = ? AND data = trim_event_data(?)
|
||||||
AND created_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-10 seconds')`
|
AND created_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-10 seconds')`
|
||||||
),
|
),
|
||||||
insertEvent: db.prepare(
|
insertEvent: db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))"
|
||||||
),
|
),
|
||||||
// Same insert with an EXPLICIT created_at, for events recovered from a
|
// Same insert with an EXPLICIT created_at, for events recovered from a
|
||||||
// transcript: their real time is in the file, not the moment we happened to
|
// transcript: their real time is in the file, not the moment we happened to
|
||||||
@@ -1697,7 +1705,7 @@ const stmts = {
|
|||||||
// re-parse re-insert the whole turn history (587k rows across this DB, 75%
|
// re-parse re-insert the whole turn history (587k rows across this DB, 75%
|
||||||
// of the events table, one session holding 68,747 copies of the same turns).
|
// of the events table, one session holding 68,747 copies of the same turns).
|
||||||
insertEventAt: db.prepare(
|
insertEventAt: db.prepare(
|
||||||
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
"INSERT INTO events (session_id, agent_id, event_type, tool_name, summary, data, created_at) VALUES (?, ?, ?, ?, ?, trim_event_data(?), ?)"
|
||||||
),
|
),
|
||||||
listEvents: db.prepare("SELECT * FROM events ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?"),
|
listEvents: db.prepare("SELECT * FROM events ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?"),
|
||||||
listEventsBySession: db.prepare(
|
listEventsBySession: db.prepare(
|
||||||
|
|||||||
@@ -479,6 +479,29 @@ function startBackgroundServices() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// One-time (user_version 4) trim of the hook payloads stored before
|
||||||
|
// `trim_event_data` existed — see `server/lib/event-payload.js`. Waits for
|
||||||
|
// marker 3, so it never claims a DB whose token repair has not run yet.
|
||||||
|
{
|
||||||
|
const dbModule = require("./db");
|
||||||
|
if (dbModule.db.pragma("user_version", { simple: true }) === 3) {
|
||||||
|
const t = setTimeout(() => {
|
||||||
|
require("./lib/event-payload")
|
||||||
|
.sweepEventPayloads(dbModule.db)
|
||||||
|
.then((r) => {
|
||||||
|
if (r.rows > 0) {
|
||||||
|
console.log(
|
||||||
|
`Trimmed ${r.rows} stored event payload(s), ${(r.bytesSaved / 1048576).toFixed(0)} MB`
|
||||||
|
);
|
||||||
|
}
|
||||||
|
dbModule.db.pragma("user_version = 4");
|
||||||
|
})
|
||||||
|
.catch((err) => console.warn("event payload trim failed:", err?.message || err));
|
||||||
|
}, 5_000);
|
||||||
|
if (t.unref) t.unref();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Per-hour tokens for sessions recorded before `token_usage_hourly`, so the
|
// Per-hour tokens for sessions recorded before `token_usage_hourly`, so the
|
||||||
// daily cost split covers them too. Picks only sessions with no hourly rows,
|
// daily cost split covers them too. Picks only sessions with no hourly rows,
|
||||||
// so a later start re-reads nothing already done; non-blocking.
|
// so a later start re-reads nothing already done; non-blocking.
|
||||||
|
|||||||
@@ -0,0 +1,143 @@
|
|||||||
|
/**
|
||||||
|
* @file event-payload.js
|
||||||
|
* @description Trims a hook payload before it is stored as `events.data`. A
|
||||||
|
* PostToolUse payload carries whole files — Edit's `originalFile`, Read's file
|
||||||
|
* `content`/`base64`, Write's `content` — and long command output, none of which
|
||||||
|
* any analysis reads: the server reads only `tool_input.skill`, the event views
|
||||||
|
* only preview. The full text stays in the session's transcript (and AgentHub's
|
||||||
|
* snapshot of it under `<data dir>/transcripts/`), which the Conversation tab
|
||||||
|
* reads. Kept: every field outside `tool_input`/`tool_response`, every number and
|
||||||
|
* flag, `structuredPatch` (the +/- line counts), and each long string's head and
|
||||||
|
* tail. Dropped fields are named in `_trimmed`, so a view can say so.
|
||||||
|
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||||
|
*/
|
||||||
|
|
||||||
|
// Strings inside tool_input / tool_response longer than this keep head + tail.
|
||||||
|
const STR_CAP = 8192;
|
||||||
|
const HEAD = 6000;
|
||||||
|
const TAIL = 2000;
|
||||||
|
// A payload this short cannot hold anything worth trimming, so it is not parsed.
|
||||||
|
const SKIP_UNDER = STR_CAP;
|
||||||
|
|
||||||
|
// Per tool, the tool_response fields that mirror a whole file or repeat
|
||||||
|
// tool_input. Other tools' fields (Grep's `content` is its matches) only get capped.
|
||||||
|
const EDIT_DROP = ["originalFile", "oldString", "newString"];
|
||||||
|
const DROP_RESPONSE = {
|
||||||
|
Edit: EDIT_DROP,
|
||||||
|
MultiEdit: EDIT_DROP,
|
||||||
|
NotebookEdit: EDIT_DROP,
|
||||||
|
Write: [...EDIT_DROP, "content"],
|
||||||
|
};
|
||||||
|
const DROP_RESPONSE_FILE = { Read: ["content", "base64"] };
|
||||||
|
|
||||||
|
function capString(s) {
|
||||||
|
if (s.length <= STR_CAP) return s;
|
||||||
|
return `${s.slice(0, HEAD)}\n…[${s.length - HEAD - TAIL} chars trimmed]…\n${s.slice(-TAIL)}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Caps every string at any depth; `structuredPatch` is left whole.
|
||||||
|
function capDeep(v, key) {
|
||||||
|
if (typeof v === "string") return capString(v);
|
||||||
|
if (key === "structuredPatch" || v === null || typeof v !== "object") return v;
|
||||||
|
if (Array.isArray(v)) return v.map((x) => capDeep(x));
|
||||||
|
const out = {};
|
||||||
|
for (const [k, x] of Object.entries(v)) out[k] = capDeep(x, k);
|
||||||
|
return out;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param {object} data A parsed hook payload.
|
||||||
|
* @returns {object} A trimmed copy (the input is not mutated).
|
||||||
|
*/
|
||||||
|
function trimHookPayload(data) {
|
||||||
|
if (!data || typeof data !== "object" || Array.isArray(data)) return data;
|
||||||
|
const out = { ...data };
|
||||||
|
const dropped = new Set(Array.isArray(data._trimmed) ? data._trimmed : []);
|
||||||
|
const res = out.tool_response;
|
||||||
|
if (res && typeof res === "object" && !Array.isArray(res)) {
|
||||||
|
const r = { ...res };
|
||||||
|
for (const k of DROP_RESPONSE[data.tool_name] || []) {
|
||||||
|
if (typeof r[k] === "string" && r[k].length > 0) {
|
||||||
|
delete r[k];
|
||||||
|
dropped.add(`tool_response.${k}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const fileDrop = DROP_RESPONSE_FILE[data.tool_name];
|
||||||
|
if (fileDrop && r.file && typeof r.file === "object") {
|
||||||
|
const f = { ...r.file };
|
||||||
|
for (const k of fileDrop) {
|
||||||
|
if (typeof f[k] === "string" && f[k].length > 0) {
|
||||||
|
delete f[k];
|
||||||
|
dropped.add(`tool_response.file.${k}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
r.file = f;
|
||||||
|
}
|
||||||
|
out.tool_response = r;
|
||||||
|
}
|
||||||
|
if ("tool_response" in out) out.tool_response = capDeep(out.tool_response);
|
||||||
|
if ("tool_input" in out) out.tool_input = capDeep(out.tool_input);
|
||||||
|
if (dropped.size > 0) out._trimmed = [...dropped];
|
||||||
|
return out;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The SQL-side entry point (`trim_event_data(?)`, registered in `server/db.js`):
|
||||||
|
* a JSON string in, a JSON string out. Short, null or unparseable input is
|
||||||
|
* returned unchanged, and trimming is idempotent.
|
||||||
|
* @param {string|null} json
|
||||||
|
* @returns {string|null}
|
||||||
|
*/
|
||||||
|
function trimEventData(json) {
|
||||||
|
if (typeof json !== "string" || json.length <= SKIP_UNDER) return json;
|
||||||
|
let parsed;
|
||||||
|
try {
|
||||||
|
parsed = JSON.parse(json);
|
||||||
|
} catch {
|
||||||
|
return json;
|
||||||
|
}
|
||||||
|
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return json;
|
||||||
|
return JSON.stringify(trimHookPayload(parsed));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Trims the rows stored before trimming existed, a chunk per write transaction
|
||||||
|
* so a hook never waits long for the lock. Idempotent: a crash just resumes from
|
||||||
|
* the start on the next run.
|
||||||
|
* @param {import("better-sqlite3").Database} db
|
||||||
|
* @param {{chunk?: number}} [opts]
|
||||||
|
* @returns {Promise<{rows: number, bytesSaved: number}>}
|
||||||
|
*/
|
||||||
|
function sweepEventPayloads(db, { chunk = 200 } = {}) {
|
||||||
|
const select = db.prepare(
|
||||||
|
"SELECT id, data FROM events WHERE id > ? AND length(data) > ? ORDER BY id LIMIT ?"
|
||||||
|
);
|
||||||
|
const update = db.prepare("UPDATE events SET data = ? WHERE id = ?");
|
||||||
|
const out = { rows: 0, bytesSaved: 0 };
|
||||||
|
const trimChunk = db.transaction((rows) => {
|
||||||
|
for (const r of rows) {
|
||||||
|
const t = trimEventData(r.data);
|
||||||
|
if (t === r.data) continue;
|
||||||
|
update.run(t, r.id);
|
||||||
|
out.rows++;
|
||||||
|
out.bytesSaved += r.data.length - t.length;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
let lastId = 0;
|
||||||
|
return new Promise((resolve, reject) => {
|
||||||
|
const step = () => {
|
||||||
|
try {
|
||||||
|
const rows = select.all(lastId, SKIP_UNDER, chunk);
|
||||||
|
if (rows.length === 0) return resolve(out);
|
||||||
|
trimChunk(rows);
|
||||||
|
lastId = rows[rows.length - 1].id;
|
||||||
|
setImmediate(step);
|
||||||
|
} catch (err) {
|
||||||
|
reject(err);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
step();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
module.exports = { trimHookPayload, trimEventData, sweepEventPayloads, STR_CAP };
|
||||||
+26
-35
@@ -121,6 +121,30 @@ async function readFirstLine(filePath) {
|
|||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Every priced column, as `stmts.getTokensBySession` reads one session — so a
|
||||||
|
// list row's cost equals the session's own `/api/pricing/cost/:id` (speed, geo,
|
||||||
|
// tier, the 1h cache write and server-tool fees included). ≤ 900 ids per call.
|
||||||
|
function tokensForSessions(ids) {
|
||||||
|
const out = {};
|
||||||
|
if (ids.length === 0) return out;
|
||||||
|
const rows = db
|
||||||
|
.prepare(
|
||||||
|
`SELECT session_id, model, speed, inference_geo, service_tier,
|
||||||
|
input_tokens + baseline_input as input_tokens,
|
||||||
|
output_tokens + baseline_output as output_tokens,
|
||||||
|
cache_read_tokens + baseline_cache_read as cache_read_tokens,
|
||||||
|
cache_write_tokens + baseline_cache_write as cache_write_tokens,
|
||||||
|
cache_write_1h_tokens + baseline_cache_write_1h as cache_write_1h_tokens,
|
||||||
|
web_search_requests + baseline_web_search as web_search_requests,
|
||||||
|
web_fetch_requests + baseline_web_fetch as web_fetch_requests,
|
||||||
|
code_execution_requests + baseline_code_execution as code_execution_requests
|
||||||
|
FROM token_usage WHERE session_id IN (${ids.map(() => "?").join(",")})`
|
||||||
|
)
|
||||||
|
.all(...ids);
|
||||||
|
for (const t of rows) (out[t.session_id] ||= []).push(t);
|
||||||
|
return out;
|
||||||
|
}
|
||||||
|
|
||||||
router.get("/", (req, res) => {
|
router.get("/", (req, res) => {
|
||||||
const limit = Math.min(parseInt(req.query.limit) || 50, 10000);
|
const limit = Math.min(parseInt(req.query.limit) || 50, 10000);
|
||||||
const offset = parseInt(req.query.offset) || 0;
|
const offset = parseInt(req.query.offset) || 0;
|
||||||
@@ -201,23 +225,7 @@ router.get("/", (req, res) => {
|
|||||||
for (let i = 0; i < allRows.length; i += 900) {
|
for (let i = 0; i < allRows.length; i += 900) {
|
||||||
const chunk = allRows.slice(i, i + 900);
|
const chunk = allRows.slice(i, i + 900);
|
||||||
const ids = chunk.map((r) => r.id);
|
const ids = chunk.map((r) => r.id);
|
||||||
const placeholders = ids.map(() => "?").join(",");
|
const tokensBySession = tokensForSessions(ids);
|
||||||
const chunkTokens = db
|
|
||||||
.prepare(
|
|
||||||
`SELECT session_id, model,
|
|
||||||
input_tokens + baseline_input as input_tokens,
|
|
||||||
output_tokens + baseline_output as output_tokens,
|
|
||||||
cache_read_tokens + baseline_cache_read as cache_read_tokens,
|
|
||||||
cache_write_tokens + baseline_cache_write as cache_write_tokens
|
|
||||||
FROM token_usage WHERE session_id IN (${placeholders})`
|
|
||||||
)
|
|
||||||
.all(...ids);
|
|
||||||
|
|
||||||
const tokensBySession = {};
|
|
||||||
for (const t of chunkTokens) {
|
|
||||||
if (!tokensBySession[t.session_id]) tokensBySession[t.session_id] = [];
|
|
||||||
tokensBySession[t.session_id].push(t);
|
|
||||||
}
|
|
||||||
|
|
||||||
for (const row of chunk) {
|
for (const row of chunk) {
|
||||||
const sessionTokens = tokensBySession[row.id];
|
const sessionTokens = tokensBySession[row.id];
|
||||||
@@ -250,25 +258,8 @@ router.get("/", (req, res) => {
|
|||||||
.all(...params, limit, offset);
|
.all(...params, limit, offset);
|
||||||
|
|
||||||
if (rows.length > 0) {
|
if (rows.length > 0) {
|
||||||
const ids = rows.map((r) => r.id);
|
|
||||||
const placeholders = ids.map(() => "?").join(",");
|
|
||||||
const allTokens = db
|
|
||||||
.prepare(
|
|
||||||
`SELECT session_id, model,
|
|
||||||
input_tokens + baseline_input as input_tokens,
|
|
||||||
output_tokens + baseline_output as output_tokens,
|
|
||||||
cache_read_tokens + baseline_cache_read as cache_read_tokens,
|
|
||||||
cache_write_tokens + baseline_cache_write as cache_write_tokens
|
|
||||||
FROM token_usage WHERE session_id IN (${placeholders})`
|
|
||||||
)
|
|
||||||
.all(...ids);
|
|
||||||
|
|
||||||
const rules = stmts.listPricing.all();
|
const rules = stmts.listPricing.all();
|
||||||
const tokensBySession = {};
|
const tokensBySession = tokensForSessions(rows.map((r) => r.id));
|
||||||
for (const t of allTokens) {
|
|
||||||
if (!tokensBySession[t.session_id]) tokensBySession[t.session_id] = [];
|
|
||||||
tokensBySession[t.session_id].push(t);
|
|
||||||
}
|
|
||||||
|
|
||||||
for (const row of rows) {
|
for (const row of rows) {
|
||||||
const sessionTokens = tokensBySession[row.id];
|
const sessionTokens = tokensBySession[row.id];
|
||||||
|
|||||||
Reference in New Issue
Block a user