From 0cd56e415fdc859b9aa5ece3c971792d79d49117 Mon Sep 17 00:00:00 2001 From: e2720pjk Date: Thu, 5 Nov 2026 17:05:24 +0800 Subject: [PATCH 1/7] feat(insights): add Antigravity CLI usage support Read local generation metadata from Antigravity conversation databases, with canonical compact transcripts as a fallback. Keep input, cached input, output and thinking accounting separate and deduplicate copies. Register Antigravity in collector and UI defaults. Migrate only untouched previous default source lists; preserve customized configurations. Keep DB and transcript replacements atomic on errors and interrupted reads. Reuse the existing indexer for budget-limited scans without changing other adapters' read behavior. Cover decoding, read-only access, privacy, WAL invalidation, fallback replacement, interrupted scans and source migration with synthetic tests. --- herdr-insights/README.md | 4 +- herdr-insights/RECIPE.md | 14 +- herdr-insights/mods/insights/analytics.js | 7 + .../mods/insights/core/adapters.mjs | 31 ++ .../mods/insights/core/antigravity.mjs | 286 ++++++++++++++++++ herdr-insights/mods/insights/core/config.mjs | 36 ++- herdr-insights/mods/insights/core/scan.mjs | 53 +++- tests/herdr-insights/analytics.test.mjs | 2 +- .../collector-adapters.test.mjs | 54 ++++ .../collector-antigravity.test.mjs | 268 ++++++++++++++++ .../herdr-insights/collector-config.test.mjs | 54 +++- 11 files changed, 780 insertions(+), 29 deletions(-) create mode 100644 herdr-insights/mods/insights/core/antigravity.mjs create mode 100644 tests/herdr-insights/collector-antigravity.test.mjs diff --git a/herdr-insights/README.md b/herdr-insights/README.md index ac1dfcc..6353f10 100644 --- a/herdr-insights/README.md +++ b/herdr-insights/README.md @@ -2,7 +2,7 @@ A native [Herdr](https://herdr.dev) dashboard for coding-agent token usage. See which models you use, how usage changes over time, and your daily activity. -Reads local sessions from **Pi, Codex, Claude Code, and OpenCode**. +Reads local sessions from **Pi, Codex, Claude Code, OpenCode, and Antigravity**. Tokens history @@ -40,7 +40,7 @@ click-state rules. Set `HERDR_APP_HOME` to use a different app home. ## Privacy No network requests or uploads. The mod runs a local Node helper, reads session -files and OpenCode's SQLite database, and writes usage metadata to +files and OpenCode/Antigravity SQLite databases, and writes usage metadata to `~/.config/herdr-app/data/insights/insights.json`. It does not store prompt or response text or modify source transcripts. diff --git a/herdr-insights/RECIPE.md b/herdr-insights/RECIPE.md index a3b8f0f..ad71c3a 100644 --- a/herdr-insights/RECIPE.md +++ b/herdr-insights/RECIPE.md @@ -1,7 +1,7 @@ # Insights -A native Herdr dashboard for token usage across Pi, Codex, Claude Code, and -OpenCode. It reads existing local session data, shows model rankings and activity, +A native Herdr dashboard for token usage across Pi, Codex, Claude Code, OpenCode, +and Antigravity. It reads existing local session data, shows model rankings and activity, and opens in its own tab from an Insights button beside the native tabs. One mod provides both the button and the dashboard; no built-in Herdr code, terminals, or keys are changed. @@ -122,10 +122,14 @@ Keep raw rows and analytics work in Node, not QuickJS rendering. Return a versio and error. Coalesce filter requests and reject stale results. Send diagnostics to stderr and surface nonzero exits. Settings controls are initialized lazily. -`collector.mjs` and `core/{config,adapters,scan,store,util}.mjs` implement collection. +`collector.mjs` and `core/{config,adapters,antigravity,scan,store,util}.mjs` implement collection. Default roots are `~/.pi/agent/sessions`, `~/.codex/sessions`, -`~/.codex/archived_sessions`, `~/.claude/projects`, and `~/.local/share/opencode`. -Read JSONL/zstd transcripts and OpenCode SQLite in read-only mode. The metadata-only +`~/.codex/archived_sessions`, `~/.claude/projects`, `~/.local/share/opencode`, +`~/.gemini/antigravity-cli/conversations`, and `~/.gemini/antigravity-cli/brain`. +Read JSONL/zstd transcripts and OpenCode/Antigravity SQLite in read-only mode. +For Antigravity, prefer generation metadata over compact transcript exports for +one conversation; decoding and accounting notes live in `core/antigravity.mjs`. +The metadata-only store is `/data/insights/insights.json`; never save prompt/response text. Use source-tagged status/warnings for missing, disabled, empty, and failed sources. Preserve historical rows from removed files until explicit rebuild. diff --git a/herdr-insights/mods/insights/analytics.js b/herdr-insights/mods/insights/analytics.js index afd80b0..0e8dd3c 100644 --- a/herdr-insights/mods/insights/analytics.js +++ b/herdr-insights/mods/insights/analytics.js @@ -33,6 +33,13 @@ export const DEFAULT_SOURCES = [ enabled: true, roots: ["~/.claude/projects"], }, + { + id: "antigravity", + label: "Antigravity", + adapter: "antigravity", + enabled: true, + roots: ["~/.gemini/antigravity-cli/conversations", "~/.gemini/antigravity-cli/brain"], + }, ]; export const PALETTES = ["blue", "green", "purple"]; export const DEFAULT_REFRESH_SECONDS = 60; diff --git a/herdr-insights/mods/insights/core/adapters.mjs b/herdr-insights/mods/insights/core/adapters.mjs index ab9e5ef..2503d70 100644 --- a/herdr-insights/mods/insights/core/adapters.mjs +++ b/herdr-insights/mods/insights/core/adapters.mjs @@ -1,3 +1,4 @@ +import { antigravityConversation } from "./antigravity.mjs"; import { MAX_TURN_MS, asString, @@ -932,6 +933,35 @@ function parseCodex(record, ctx) { } } +// --------------------------------------------------------------------------- +// Antigravity compact transcripts (CLI and current IDE/desktop surfaces). +// Counts are per model step; input_tokens excludes cache_read_tokens. Missing +// model/provider/workspace fields remain unknown: artifacts are not usage evidence. +function parseAntigravity(record, ctx) { + if (!isPlainObject(record) || record.source !== "MODEL") return; + if (!Number.isSafeInteger(record.step_index) || record.step_index < 0) return; + const conversation = antigravityConversation(ctx.file.path); + const timestamp = parseTimestamp(record.created_at); + if (timestamp === null) { + ctx.warn("Antigravity model step has no usable timestamp; skipped"); + return; + } + const tokens = normalizeTokens({ + input: record.input_tokens, + output: record.output_tokens, + cacheRead: record.cache_read_tokens, + }); + if (!tokens.nonzero) return; + // Path-independent identity deduplicates copied sessions across product roots + // and repeated revisions of one step; separate conversations stay separate. + const session = `antigravity:${conversation}`; + ctx.addUsage( + metaFor("antigravity", session, timestamp, record.workspace, record.provider, record.model), + tokens, + makeDedupKey("antigravity", conversation, record.step_index), + ); +} + // --------------------------------------------------------------------------- // Declarative custom JSONL @@ -1050,6 +1080,7 @@ export const ADAPTER_PARSERS = { pi: parsePi, claude: parseClaude, codex: parseCodex, + antigravity: parseAntigravity, custom: parseCustom, }; diff --git a/herdr-insights/mods/insights/core/antigravity.mjs b/herdr-insights/mods/insights/core/antigravity.mjs new file mode 100644 index 0000000..0c51da0 --- /dev/null +++ b/herdr-insights/mods/insights/core/antigravity.mjs @@ -0,0 +1,286 @@ +/** + * Antigravity CLI usage from conversations/.db (read-only SQLite). + * Canonical brain//.system_generated/logs/transcript.jsonl exports are a + * fallback in adapters.mjs. Discovery selects one file per conversation: SQLite + * first, then configured root order. Cached alternatives are removed only after + * a replacement is accepted. Unreadable databases retain accepted history. + * + * Reverse-engineered protobuf mapping (not an official stable API): + * gen_metadata.data #1.#19: response model; #1.#4: usage. + * Usage #1 + #2: system + uncached input; #5: cache read; #3: total output; + * #9: text output; #10: thinking (a subset of output); #11: response ID. + * Envelope #2: packed step indexes. steps.metadata #1: creation timestamp, + * used only when #9.#11 matches the generation's response ID. Otherwise use + * trajectory_metadata_blob creation time, then database mtime, with a note. + * Trajectory metadata also supplies the workspace URI. + * Reject inconsistent output totals; skip prefilled system-only records without + * measured usage. Response IDs deduplicate copies; missing IDs use conversation + * and generation index. Provider denotes model publisher, not billing gateway. + * + * DB/WAL fingerprints invalidate cached usage. Blobs are read one row at a time, + * capped at 16 MiB; malformed/oversized records warn. Only usage metadata is + * persisted, never prompt/output/thinking text. No RPC, credentials or uploads. + * Retired Gemini CLI history and Antigravity IDE .pb stores are not decoded. + * See also: https://github.com/junhoyeo/tokscale/pull/713 + * https://github.com/robinebers/openusage/pull/1139 + */ +import path from "node:path"; +import { stat } from "node:fs/promises"; +import { fileURLToPath } from "node:url"; +import { + addUsage, + beginStoreTransaction, + commitStoreTransaction, + rollbackStoreTransaction, + fileKey, + newFileEntry, + resetFileEntry, + dedupKeyFor as makeDedupKey, +} from "./store.mjs"; +import { localDay } from "./util.mjs"; + +const VERSION = 1; +const MAX_BLOB = 16 * 1024 * 1024; +const MAX_FIELDS = 100_000; + +export function antigravityConversation(file) { + return file.endsWith(".db") + ? path.basename(file, ".db") + : path.basename(path.dirname(path.dirname(path.dirname(file)))); +} + +// Read only selected protobuf fields. Unknown fields (including prompt payloads) +// are bounded byte slices, never decoded to text or persisted. uint64 uses BigInt +// so unrelated signed/sentinel fields cannot silently overflow JS numbers. +function varint(bytes, cursor) { + let value = 0n; + for (let shift = 0; shift < 70; shift += 7) { + if (cursor.pos >= bytes.length) throw new Error("truncated protobuf varint"); + const byte = bytes[cursor.pos++]; + if (shift === 63 && byte > 1) throw new Error("protobuf uint64 overflow"); + value |= BigInt(byte & 127) << BigInt(shift); + if (byte < 128) return value; + } + throw new Error("protobuf varint overflow"); +} + +export function protobufFields(bytes) { + const cursor = { pos: 0 }; + const out = []; + while (cursor.pos < bytes.length) { + if (out.length >= MAX_FIELDS) throw new Error("too many protobuf fields"); + const tag = varint(bytes, cursor); + const number = Number(tag >> 3n), wire = Number(tag & 7n); + if (number < 1 || number > 536870911) throw new Error("invalid protobuf tag"); + let value; + if (wire === 0) value = varint(bytes, cursor); + else if (wire === 1 || wire === 2 || wire === 5) { + const size = wire === 2 ? Number(varint(bytes, cursor)) : wire === 1 ? 8 : 4; + if (!Number.isSafeInteger(size) || size > bytes.length - cursor.pos) { + throw new Error("invalid protobuf length"); + } + value = bytes.subarray(cursor.pos, cursor.pos + size); + cursor.pos += size; + } else throw new Error("unsupported protobuf wire type"); + out.push({ number, wire, value }); + } + return out; +} + +function blob(fields, number) { + return fields.find((f) => f.number === number && f.wire === 2)?.value; +} +function integer(fields, number) { + const value = fields.find((f) => f.number === number && f.wire === 0)?.value ?? 0n; + if (value > BigInt(Number.MAX_SAFE_INTEGER)) throw new Error("unsafe usage integer"); + return Number(value); +} +function text(bytes, max = 256) { + if (!bytes || bytes.length === 0 || bytes.length > max) return ""; + const value = Buffer.from(bytes).toString("utf8"); + return /[\x00-\x1f\x7f\ufffd]/.test(value) ? "" : value; +} +function timestamp(bytes) { + if (!bytes) return null; + const fields = protobufFields(bytes); + const seconds = integer(fields, 1), nanos = integer(fields, 2); + const ms = seconds * 1000 + Math.floor(nanos / 1e6); + return seconds > 0 && nanos < 1e9 && Number.isSafeInteger(ms) && + Number.isFinite(new Date(ms).getTime()) ? ms : null; +} +function responseId(usage) { return text(blob(usage, 11)); } +function provider(model) { + // Model publisher, not the account/billing gateway used by Antigravity. + if (model.startsWith("gemini-")) return "google"; + if (model.startsWith("claude-")) return "anthropic"; + return "unknown"; +} + +export function decodeAntigravityGeneration(bytes) { + const outer = protobufFields(bytes); + const packedSteps = blob(outer, 2); + let step = null; + if (packedSteps?.length) { + const value = varint(packedSteps, { pos: 0 }); + if (value <= BigInt(Number.MAX_SAFE_INTEGER)) step = Number(value); + } + const generations = []; + for (const field of outer) { + if (field.number !== 1 || field.wire !== 2) continue; + const inner = protobufFields(field.value); + const usageBytes = blob(inner, 4); + if (!usageBytes) continue; + const usage = protobufFields(usageBytes); + const system = integer(usage, 1), input = integer(usage, 2); + const output = integer(usage, 3), cacheRead = integer(usage, 5); + const reasoning = integer(usage, 10), outputText = integer(usage, 9); + if (!Number.isSafeInteger(system + input + output + cacheRead)) throw new Error("unsafe token total"); + if (outputText + reasoning !== output) throw new Error("inconsistent output accounting"); + // A fixed system-prompt count is prefilled even on abandoned/failed + // requests with no measured usage. It is not a completed generation. + if (input + output + cacheRead === 0) continue; + generations.push({ + step, model: text(blob(inner, 19)) || "unknown", responseId: responseId(usage), + // Output includes thinking; reasoning is a subset, never added to total. + tokens: { input: system + input, output, cacheRead, cacheWrite: 0, reasoning }, + }); + } + return generations; +} + +function sessionMetadata(bytes) { + if (!bytes) return {}; + const fields = protobufFields(bytes); + let project = "unknown"; + const workspace = blob(fields, 1); + if (workspace) { + const uri = text(blob(protobufFields(workspace), 1), 8192); + try { if (uri.startsWith("file://")) project = fileURLToPath(uri); } catch { /* unknown */ } + } + return { timestamp: timestamp(blob(fields, 2)), project }; +} + +function stepTimestamp(bytes, id) { + if (!bytes || !id) return null; + const fields = protobufFields(bytes); + const usage = blob(fields, 9); + // Packed step references were validated against response IDs on real stores. + // Do not attribute a timestamp if a future schema points at a different step. + if (!usage || responseId(protobufFields(usage)) !== id) return null; + return timestamp(blob(fields, 1)); +} + +export async function scanAntigravityDb(store, source, file, sink, stats, options = {}) { + let info; + try { + info = await stat(file); + } catch (error) { + if (error.code !== "ENOENT") sink.warn(`Antigravity: cannot stat database ${path.basename(file)}; previous accepted data retained.`); + return { dirty: false, aborted: false }; + } + if (!info.isFile()) return { dirty: false, aborted: false }; + let wal = null; + try { + wal = await stat(`${file}-wal`); + } catch (error) { + if (error.code !== "ENOENT") { + sink.warn(`Antigravity: cannot stat database WAL ${path.basename(file)}; previous accepted data retained.`); + return { dirty: false, aborted: false }; + } + } + const fingerprint = `${info.dev}:${info.ino}:${info.size}:${info.mtimeMs}:${wal?.size ?? 0}:${wal?.mtimeMs ?? 0}`; + let entry = store.files[fileKey(source.id, file)]; + if (entry?.state.agyDbVersion === VERSION && entry.state.agyDbFingerprint === fingerprint) { + for (const warning of entry.state.agyWarnings ?? []) sink.warn(warning); + for (const note of entry.state.agyNotes ?? []) sink.note(note); + stats.unchangedFiles += 1; + return { dirty: false, aborted: false, accepted: true }; + } + let sqlite; + try { sqlite = await import("node:sqlite"); } catch { + sink.warn("Antigravity SQLite requires a Node runtime with node:sqlite; skipped."); + return { dirty: false, aborted: false }; + } + const tx = beginStoreTransaction(store, { fileId: entry?.id }); + let db; + const counts = { oversized: 0, malformed: 0, approximateTime: 0, missingId: 0 }; + let rows = 0; + try { + db = new sqlite.DatabaseSync(file, { readOnly: true }); + db.exec("BEGIN"); + // SQLite limits the returned blob before it crosses into JS memory. + const limited = `CASE WHEN length(data) <= ${MAX_BLOB} THEN data ELSE NULL END`; + let session = {}; + const meta = db.prepare(`SELECT ${limited} AS data FROM trajectory_metadata_blob WHERE id = 'main'`).get(); + if (meta?.data) { + try { session = sessionMetadata(meta.data); } catch { counts.malformed += 1; } + } else if (meta) counts.oversized += 1; + const stepQuery = db.prepare(`SELECT CASE WHEN length(metadata) <= ${MAX_BLOB} THEN metadata ELSE NULL END AS data FROM steps WHERE idx = ?`); + const records = db.prepare(`SELECT idx, ${limited} AS data FROM gen_metadata ORDER BY idx`).iterate(); + if (!entry) entry = newFileEntry(store, { + sourceId: source.id, adapter: source.adapter, path: file, + dev: String(info.dev), ino: String(info.ino), size: 0, mtimeMs: 0, compressed: false, + }); + else resetFileEntry(store, entry); + const conversation = antigravityConversation(file); + for (const record of records) { + const stopped = options.shouldStop?.(); + const expired = rows > 0 && Number.isFinite(options.deadline) && + (options.now ?? Date.now)() >= options.deadline; + if (stopped || expired) { + rollbackStoreTransaction(tx); + return { dirty: false, aborted: true, reason: stopped ? "stop" : "budget" }; + } + rows += 1; + if (!record.data) { counts.oversized += 1; continue; } + let generations; + try { generations = decodeAntigravityGeneration(record.data); } catch { counts.malformed += 1; continue; } + for (let index = 0; index < generations.length; index += 1) { + const generation = generations[index]; + let created = null; + if (generation.step !== null) { + const step = stepQuery.get(generation.step); + try { created = stepTimestamp(step?.data, generation.responseId); } catch { /* use documented fallback */ } + } + if (created === null) { + counts.approximateTime += 1; + created = session.timestamp ?? info.mtimeMs; + } + if (!generation.responseId) counts.missingId += 1; + addUsage(store, entry, { + day: localDay(created), session: `antigravity:${conversation}`, harness: "antigravity", + model: generation.model, provider: provider(generation.model), project: session.project ?? "unknown", + }, generation.tokens, generation.responseId + ? makeDedupKey("antigravity", "response", generation.responseId) + : makeDedupKey("antigravity", "generation", conversation, record.idx, index)); + } + } + db.exec("COMMIT"); + const warnings = []; + const notes = []; + const prefix = `Antigravity ${path.basename(file)}:`; + if (counts.oversized) warnings.push(`${prefix} ${counts.oversized} blobs exceed the 16 MiB limit; skipped.`); + if (counts.malformed) warnings.push(`${prefix} ${counts.malformed} malformed metadata records were skipped.`); + if (counts.approximateTime) notes.push(`${prefix} ${counts.approximateTime} generations use session time or database mtime (no verified step timestamp).`); + if (counts.missingId) notes.push(`${prefix} ${counts.missingId} generations have no response ID; deduplicated by conversation and generation index.`); + entry.dev = String(info.dev); entry.ino = String(info.ino); + entry.size = info.size; entry.mtimeMs = info.mtimeMs; + entry.state.agyDbVersion = VERSION; + entry.state.agyDbFingerprint = fingerprint; + entry.state.agyWarnings = warnings; + entry.state.agyNotes = notes; + commitStoreTransaction(tx); + for (const warning of warnings) sink.warn(warning); + for (const note of notes) sink.note(note); + stats.filesRead += 1; + stats.bytesRead += info.size + (wal?.size ?? 0); + return { dirty: true, aborted: false, accepted: true }; + } catch { + rollbackStoreTransaction(tx); + // Do not leak strings from SQLite or protobuf source payloads in diagnostics. + sink.warn(`Antigravity: cannot read generation database ${path.basename(file)}; previous accepted data retained.`); + return { dirty: false, aborted: false }; + } finally { + try { db?.close(); } catch { /* best effort */ } + } +} diff --git a/herdr-insights/mods/insights/core/config.mjs b/herdr-insights/mods/insights/core/config.mjs index 33b1e6d..56d0fce 100644 --- a/herdr-insights/mods/insights/core/config.mjs +++ b/herdr-insights/mods/insights/core/config.mjs @@ -45,12 +45,21 @@ const CLAUDE_SOURCE = Object.freeze({ roots: Object.freeze(["~/.claude/projects"]), }); -// Stores written before Claude Code became a default source get the same -// upgrade, but only when they still contain the untouched previous defaults. -const LEGACY_DEFAULT_SOURCES = Object.freeze([ - PI_SOURCE, - CODEX_SOURCE, - OPENCODE_SOURCE, +const ANTIGRAVITY_SOURCE = Object.freeze({ + id: "antigravity", + label: "Antigravity", + adapter: "antigravity", + enabled: true, + roots: Object.freeze([ + "~/.gemini/antigravity-cli/conversations", + "~/.gemini/antigravity-cli/brain", + ]), +}); + +// Upgrade only untouched previous defaults; customized sources stay as saved. +const LEGACY_DEFAULT_SOURCE_SETS = Object.freeze([ + [PI_SOURCE, CODEX_SOURCE, OPENCODE_SOURCE], + [PI_SOURCE, CODEX_SOURCE, OPENCODE_SOURCE, CLAUDE_SOURCE], ]); export const DEFAULT_CONFIG = Object.freeze({ @@ -59,10 +68,10 @@ export const DEFAULT_CONFIG = Object.freeze({ includeCache: true, weekStartsOn: 0, refreshSeconds: 60, - sources: Object.freeze([PI_SOURCE, CODEX_SOURCE, OPENCODE_SOURCE, CLAUDE_SOURCE]), + sources: Object.freeze([PI_SOURCE, CODEX_SOURCE, OPENCODE_SOURCE, CLAUDE_SOURCE, ANTIGRAVITY_SOURCE]), }); -export const BUILTIN_ADAPTERS = new Set(["pi", "claude", "codex", "opencode"]); +export const BUILTIN_ADAPTERS = new Set(["pi", "claude", "codex", "opencode", "antigravity"]); export const ADAPTERS = new Set([...BUILTIN_ADAPTERS, "custom"]); const TOP_KEYS = new Set([ @@ -368,16 +377,17 @@ function sameSource(left, right) { } export function migrateLoadedConfig(config) { - if ( - config.sources.length !== LEGACY_DEFAULT_SOURCES.length || - !config.sources.every((source, index) => sameSource(source, LEGACY_DEFAULT_SOURCES[index])) - ) { + const untouchedDefaults = LEGACY_DEFAULT_SOURCE_SETS.some((sources) => + config.sources.length === sources.length && + config.sources.every((source, index) => sameSource(source, sources[index])) + ); + if (!untouchedDefaults) { return { config, migrated: false }; } return { config: { ...config, - sources: [...config.sources, { ...CLAUDE_SOURCE, roots: [...CLAUDE_SOURCE.roots] }], + sources: DEFAULT_CONFIG.sources.map((source) => ({ ...source, roots: [...source.roots] })), }, migrated: true, }; diff --git a/herdr-insights/mods/insights/core/scan.mjs b/herdr-insights/mods/insights/core/scan.mjs index 80641ff..90d5c8d 100644 --- a/herdr-insights/mods/insights/core/scan.mjs +++ b/herdr-insights/mods/insights/core/scan.mjs @@ -7,6 +7,7 @@ import { PassThrough } from "node:stream"; import { pipeline } from "node:stream/promises"; import * as zlib from "node:zlib"; import { parseAdapterRecord } from "./adapters.mjs"; +import { antigravityConversation, scanAntigravityDb } from "./antigravity.mjs"; import { MAX_LINE_BYTES, expandHome, @@ -19,6 +20,7 @@ import { beginStoreTransaction, commitStoreTransaction, dedupKeyFor, + deleteFileEntry, fileKey, newFileEntry, removeLatestFallbackTaskForSession, @@ -30,7 +32,7 @@ import { const CHUNK_BYTES = 256 * 1024; const MAX_WARNINGS = 50; const MAX_TRAVERSAL_ENTRIES = 250_000; -const PARSER_VERSIONS = { pi: 3, codex: 5, claude: 2 }; +const PARSER_VERSIONS = { pi: 3, codex: 5, claude: 2, antigravity: 1 }; const PROGRESS_CHUNK_INTERVAL = 64; export class ScanStopError extends Error { @@ -83,6 +85,14 @@ function warningSink(warnings, stats, sourceState) { } function isEligible(adapter, filePath) { + // Compact transcripts carry the same usage as full logs. Never scan both, + // chunks, workflow artifacts, or arbitrary JSONL files in a brain directory. + if (adapter === "antigravity") { + if (filePath.endsWith(".db")) return path.basename(path.dirname(filePath)) === "conversations"; + return path.basename(filePath) === "transcript.jsonl" && + path.basename(path.dirname(filePath)) === "logs" && + path.basename(path.dirname(path.dirname(filePath))) === ".system_generated"; + } if (adapter === "opencode") return path.basename(filePath) === "opencode.db"; if (adapter === "codex") { return filePath.endsWith(".jsonl") || filePath.endsWith(".jsonl.zst"); @@ -128,7 +138,7 @@ async function walkRoot(root, adapter) { const full = path.join(dir, entry.name); if (entry.isDirectory()) { stack.push(full); - } else if (entry.isFile() && isEligible(adapter, entry.name)) { + } else if (entry.isFile() && isEligible(adapter, full)) { files.push(full); } else if (entry.isSymbolicLink()) { // Do not follow directory symlinks (loops); a symlinked file is harmless. @@ -184,7 +194,17 @@ export async function discoverFiles(config, home, sink, sourceState) { a.rootIndex - b.rootIndex || a.file.localeCompare(b.file), ); - return discovered; + // One authoritative file per Antigravity conversation. SQLite wins over + // exports; configured root order resolves copied/divergent databases. + const preferred = new Map(); + for (const found of discovered) { + if (found.source.adapter !== "antigravity") continue; + const id = antigravityConversation(found.file); + const previous = preferred.get(id); + if (!previous || (!previous.file.endsWith(".db") && found.file.endsWith(".db"))) preferred.set(id, found); + } + return discovered.filter((found) => found.source.adapter !== "antigravity" || + preferred.get(antigravityConversation(found.file)) === found); } function hashBytes(buffer) { @@ -638,6 +658,7 @@ async function scanPlainFile( if (!stopped) consumer.finish(start + bytes); if (!stopped) delete entry.state.readFailed; } catch (error) { + if (adapter === "antigravity") throw error; // Keep the accepted complete prefix and its parser state together. The // unfinished suffix is retried even if the file metadata is unchanged. entry.state.readFailed = true; @@ -987,6 +1008,9 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { if (source.adapter === "opencode") { return scanOpenCodeDb(store, source, filePath, sourceSink, stats, options); } + if (source.adapter === "antigravity" && filePath.endsWith(".db")) { + return scanAntigravityDb(store, source, filePath, sourceSink, stats, options); + } let fileStat; try { fileStat = await stat(filePath); @@ -1018,7 +1042,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { && existing.mtimeMs === fileStat.mtimeMs ) { stats.unchangedFiles += 1; - return { dirty: false, aborted: false }; + return { dirty: false, aborted: false, accepted: true }; } // Every changed file gets a precise transaction. The previous committed @@ -1164,6 +1188,12 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { stats, options, ); + // Match SQLite's atomic reads: an unfinished fallback must not add rows + // alongside the conversation's previously accepted usage. + if (source.adapter === "antigravity" && plain.pending) { + rollbackStoreTransaction(tx); + return { dirty: false, aborted: true, reason: stopRequested(options) ? "stop" : "budget" }; + } dirty = plain.dirty || dirty; commitStoreTransaction(tx); entry.path = filePath; @@ -1173,7 +1203,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { entry.mtimeMs = fileStat.mtimeMs; delete entry.state.scanPending; } - return { dirty, aborted: false, pending: plain.pending }; + return { dirty, aborted: false, pending: plain.pending, accepted: !entry.state.readFailed && !plain.pending }; } catch (error) { rollbackStoreTransaction(tx); sourceSink.warn(`${source.adapter} ${filePath}: scan failed: ${error.message}`); @@ -1295,6 +1325,19 @@ export async function scanSources(store, { sourceState[found.source.id].bytesRead += stats.bytesRead - beforeBytes; dirty = Boolean(result?.dirty) || dirty; + if (found.source.adapter === "antigravity" && result?.accepted) { + // Superseded cached exports must disappear too, not just be excluded + // from discovery. Keep prior accepted data until its replacement succeeds. + const conversation = antigravityConversation(found.file); + const selectedKey = fileKey(found.source.id, found.file); + for (const [key, entry] of Object.entries(store.files)) { + if (key !== selectedKey && entry.adapter === "antigravity" && + antigravityConversation(entry.path) === conversation) { + deleteFileEntry(store, key); + dirty = true; + } + } + } if (result?.aborted) { stopReason = result.reason ?? "budget"; abortedFile = result.abortedFile ?? { diff --git a/tests/herdr-insights/analytics.test.mjs b/tests/herdr-insights/analytics.test.mjs index c5c9abf..2c3c036 100644 --- a/tests/herdr-insights/analytics.test.mjs +++ b/tests/herdr-insights/analytics.test.mjs @@ -400,7 +400,7 @@ test("config and year helpers are defensive", () => { assert.deepEqual(availableYears([row({ day: "2023-01-01" }), row({ day: "2024-01-01" })], "2025-01-01"), [2025, 2024, 2023]); const defaults = normalizeConfig({}); - assert.deepEqual(defaults.sources.map((source) => source.adapter), ["pi", "codex", "opencode", "claude"]); + assert.deepEqual(defaults.sources.map((source) => source.adapter), ["pi", "codex", "opencode", "claude", "antigravity"]); assert.equal(defaults.sources.some((source) => source.adapter === "claude"), true); assert.deepEqual(normalizeConfig({ sources: [] }).sources, [], "an explicit empty source list is preserved"); const claude = normalizeConfig({ sources: [{ id: "claude", label: "Claude Code", adapter: "claude", enabled: true, roots: ["~/.claude/projects"] }] }); diff --git a/tests/herdr-insights/collector-adapters.test.mjs b/tests/herdr-insights/collector-adapters.test.mjs index 04d1ecd..f6260c2 100644 --- a/tests/herdr-insights/collector-adapters.test.mjs +++ b/tests/herdr-insights/collector-adapters.test.mjs @@ -585,6 +585,60 @@ test("Custom declarative mapping: role filter, max revisions, harness/provider/m assert.equal(snapshot.stats.errors, 0); }); +test("Antigravity: compact-only, per-step accounting, copied sessions, unknown metadata and privacy", async () => { + const context = await env(); + const logs = ".system_generated/logs"; + const step = { + source: "MODEL", type: "PLANNER_RESPONSE", status: "DONE", step_index: 1, + created_at: "2026-03-04T04:00:00Z", input_tokens: 10, + output_tokens: 5, cache_read_tokens: 20, + content: "private agy output", thinking: "private agy reasoning", + }; + async function put(relative, records, tail = "") { + const target = path.join(context.home, relative); + await mkdir(path.dirname(target), { recursive: true }); + await writeFile(target, records.map((r) => JSON.stringify(r)).join("\n") + "\n" + tail); + return target; + } + const compact = [ + { ...step, source: "USER_EXPLICIT", step_index: 0 }, + step, + { ...step, output_tokens: 8 }, // revision, not another billable step + { ...step, step_index: 3, input_tokens: 2, output_tokens: 4, cache_read_tokens: 6 }, + { ...step, step_index: 5, input_tokens: 0, output_tokens: 0, cache_read_tokens: 0 }, + { ...step, step_index: 7, created_at: "bad-time" }, + { ...step, step_index: -1 }, + ]; + const file = await put(`cli/conv-a/${logs}/transcript.jsonl`, compact, '{"source":'); + await put(`desktop/conv-a/${logs}/transcript.jsonl`, compact); + await put(`cli/conv-a/${logs}/transcript_full.jsonl`, [{ ...step, step_index: 99, input_tokens: 9999 }]); + await put(`cli/conv-a/${logs}/chunks/chunk.jsonl`, [{ ...step, input_tokens: 9999 }]); + await put(`cli/conv-a/transcript.jsonl`, [step]); // not a canonical logs directory + await put(`cli/conv-b/${logs}/transcript.jsonl`, [step]); + const snapshot = await configure(context, [{ + id: "agy", label: "Antigravity", adapter: "antigravity", roots: ["~/cli", "~/desktop"], + }]); + const a = snapshot.rows.find((r) => r.session === "antigravity:conv-a"); + const b = snapshot.rows.find((r) => r.session === "antigravity:conv-b"); + assert.equal(snapshot.rows.length, 2); + assert.deepEqual(a.tokens, { input: 12, output: 12, cacheRead: 26, cacheWrite: 0, reasoning: 0, total: 50 }); + assert.equal(a.messages, 2); + assert.equal(b.tokens.total, 35); + assert.equal(a.harness, "antigravity"); + assert.equal(a.model, "unknown"); + assert.equal(a.provider, "unknown"); + assert.equal(a.project, "unknown"); + assert.ok(snapshot.warnings.some((w) => w.includes("timestamp"))); + assert.equal((await readFile(context.store, "utf8")).includes("private agy"), false); + // The incomplete tail is completed on a later scan and counted exactly once. + await writeFile(file, compact.map((r) => JSON.stringify(r)).join("\n") + "\n" + + JSON.stringify({ ...step, step_index: 9 }) + "\n"); + const next = await run(["snapshot", "--store", context.store, "--home", context.home]); + assert.equal(next.code, 0, next.stderr); + const updated = JSON.parse(next.stdout); + assert.equal(updated.rows.find((r) => r.session === "antigravity:conv-a").tokens.total, 85); +}); + test("Snapshot rows never include transcript bodies", async () => { const context = await env(); await placeFixture( diff --git a/tests/herdr-insights/collector-antigravity.test.mjs b/tests/herdr-insights/collector-antigravity.test.mjs new file mode 100644 index 0000000..fc01e34 --- /dev/null +++ b/tests/herdr-insights/collector-antigravity.test.mjs @@ -0,0 +1,268 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { createHash } from "node:crypto"; +import { chmod, mkdir, mkdtemp, open, readFile, realpath, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import test from "node:test"; +import { decodeAntigravityGeneration, protobufFields, scanAntigravityDb } from "../../herdr-insights/mods/insights/core/antigravity.mjs"; +import { emptyStore, snapshotRows } from "../../herdr-insights/mods/insights/core/store.mjs"; +import { scanSources } from "../../herdr-insights/mods/insights/core/scan.mjs"; + +const collector = new URL("../../herdr-insights/mods/insights/collector.mjs", import.meta.url); +function vi(value) { + let n = BigInt(value); const out = []; + do { const b = Number(n & 127n); n >>= 7n; out.push(b | (n ? 128 : 0)); } while (n); + return Buffer.from(out); +} +function num(n, value) { return Buffer.concat([vi(n * 8), vi(value)]); } +function msg(n, value) { const b = typeof value === "string" ? Buffer.from(value) : value; return Buffer.concat([vi(n * 8 + 2), vi(b.length), b]); } +const join = (...values) => Buffer.concat(values); +const ts = (seconds) => join(num(1, seconds), num(2, 123000000)); +function generation({ id = "response-one", model = "gemini-test", step = 1, input = 10, system = 3, output = 7, thinking = 2, cache = 20 } = {}) { + const usage = join(num(1, system), num(2, input), num(3, output), num(5, cache), num(9, output - thinking), num(10, thinking), msg(11, id)); + return { usage, data: join(msg(2, vi(step)), msg(1, join(msg(4, usage), msg(19, model))), msg(8, "PRIVATE_PROMPT_AND_THINKING")) }; +} +async function env(t) { + const root = await mkdtemp(path.join(tmpdir(), "insights-agy-db-")); + t.after(() => rm(root, { recursive: true, force: true })); + const home = path.join(root, "home"); + await mkdir(path.join(home, "conversations"), { recursive: true }); + await mkdir(path.join(home, "brain"), { recursive: true }); + return { root, home, store: path.join(root, "data/insights.json"), sources: [{ id: "agy", label: "Antigravity", adapter: "antigravity", roots: ["~/conversations", "~/brain"] }] }; +} +function database(context, name = "conv", records = [{ idx: 0, ...generation() }]) { + const file = path.join(context.home, "conversations", `${name}.db`); + const db = new DatabaseSync(file); + db.exec("CREATE TABLE gen_metadata(idx INTEGER PRIMARY KEY, data BLOB); CREATE TABLE steps(idx INTEGER PRIMARY KEY, metadata BLOB); CREATE TABLE trajectory_metadata_blob(id TEXT PRIMARY KEY, data BLOB)"); + db.prepare("INSERT INTO trajectory_metadata_blob VALUES ('main', ?)").run(join(msg(1, msg(1, "file:///workspace/test")), msg(2, ts(1790400000)))); + for (const r of records) { + db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(r.idx, r.data); + if (r.usage) db.prepare("INSERT OR IGNORE INTO steps VALUES (?, ?)").run(r.step ?? 1, join(msg(1, ts(r.seconds ?? 1790486400)), msg(9, r.usage))); + } + return { db, file }; +} +async function transcript(context, name = "conv") { + const file = path.join(context.home, "brain", name, ".system_generated/logs/transcript.jsonl"); + await mkdir(path.dirname(file), { recursive: true }); + await writeFile(file, JSON.stringify({ source: "MODEL", step_index: 1, created_at: "2026-09-27T00:00:00Z", input_tokens: 999, output_tokens: 2, cache_read_tokens: 3, content: "PRIVATE_TRANSCRIPT" }) + "\n"); + return file; +} +async function run(context, configure = false) { + const args = [collector.pathname, configure ? "configure" : "snapshot", "--store", context.store, "--home", context.home]; + if (configure) args.push("--config", JSON.stringify({ sources: context.sources })); + const child = spawn(process.execPath, args, { env: { ...process.env, TZ: "UTC" }, stdio: ["ignore", "pipe", "pipe"] }); + let stdout = "", stderr = ""; + child.stdout.on("data", (x) => stdout += x); child.stderr.on("data", (x) => stderr += x); + const code = await new Promise((resolve, reject) => { child.on("close", resolve); child.on("error", reject); }); + assert.equal(code, 0, stderr); + return JSON.parse(stdout); +} +const total = (snapshot) => snapshot.rows.reduce((n, r) => n + r.tokens.total, 0); + +test("Antigravity protobuf decoder validates accounting, bounds and uint64", () => { + const [g] = decodeAntigravityGeneration(generation().data); + assert.equal(g.model, "gemini-test"); assert.equal(g.responseId, "response-one"); + assert.deepEqual(decodeAntigravityGeneration(generation({ input: 0, output: 0, thinking: 0, cache: 0 }).data), [], "prefilled system-only metadata is not billable usage"); + assert.deepEqual(g.tokens, { input: 13, output: 7, cacheRead: 20, cacheWrite: 0, reasoning: 2 }); + assert.throws(() => protobufFields(Buffer.from([0])), /tag/); + assert.throws(() => protobufFields(Buffer.from([10, 255])), /varint/); + assert.throws(() => protobufFields(Buffer.from([10, 5, 0])), /length/); + assert.throws(() => protobufFields(Buffer.from([15])), /wire/); + assert.throws(() => protobufFields(Buffer.from([8, ...Array(10).fill(255)])), /overflow/); + assert.equal(protobufFields(join(num(7, 18446744073709551615n)))[0].value, 18446744073709551615n); + assert.throws(() => decodeAntigravityGeneration(msg(1, msg(4, join(num(3, 10), num(9, 11))))), /output accounting/); + assert.throws(() => decodeAntigravityGeneration(msg(1, msg(4, num(1, 9007199254740992n)))), /unsafe/); +}); + +test("Antigravity SQLite: actual model, system input, reasoning subset, verified step day, read-only and privacy", async (t) => { + const context = await env(t); const { db, file } = database(context); db.close(); + const digest = async () => createHash("sha256").update(await readFile(file)).digest("hex"); + const before = await digest(); + const first = await run(context, true); + assert.equal(first.rows.length, 1); + const row = first.rows[0]; + assert.equal(row.model, "gemini-test"); assert.equal(row.provider, "google"); + assert.equal(row.project, "/workspace/test"); assert.equal(row.day, "2026-09-27"); + assert.deepEqual(row.tokens, { input: 13, output: 7, cacheRead: 20, cacheWrite: 0, reasoning: 2, total: 40 }); + assert.equal(row.messages, 1); assert.equal(row.session, "antigravity:conv"); + assert.equal(first.warnings.length, 0); assert.equal(await digest(), before); + assert.equal((await readFile(context.store, "utf8")).includes("PRIVATE_"), false); + const warm = await run(context); + assert.equal(warm.stats.filesRead, 0); assert.equal(warm.stats.unchangedFiles, 1); assert.equal(total(warm), 40); +}); + +test("Antigravity SQLite supersedes cached transcript without duplicate totals, with or without source reconfiguration", async (t) => { + const context = await env(t); await transcript(context); + assert.equal(total(await run(context, true)), 1004); + const { db } = database(context); db.close(); + const next = await run(context); + assert.equal(total(next), 40); assert.equal(next.rows.length, 1); + const persisted = JSON.parse(await readFile(context.store, "utf8")); + assert.equal(Object.values(persisted.files).filter((f) => f.adapter === "antigravity").length, 1); + assert.equal(Object.values(persisted.files)[0].path.endsWith(".db"), true); + assert.equal(total(await run(context)), 40); +}); + +test("Antigravity SQLite handles WAL append and response-ID dedup across sessions and copied roots", async (t) => { + const context = await env(t); const { db, file } = database(context); + db.exec("PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=0"); + assert.equal(total(await run(context, true)), 40); + const g = generation({ id: "two", model: "claude-test", step: 3 }); + db.prepare("INSERT INTO gen_metadata VALUES (1, ?)").run(g.data); + db.prepare("INSERT INTO steps VALUES (3, ?)").run(join(msg(1, ts(1790486401)), msg(9, g.usage))); + const next = await run(context); + assert.equal(total(next), 80); assert.equal(next.rows.find((r) => r.model === "claude-test").provider, "anthropic"); + db.close(); + const other = database(context, "copy"); other.db.close(); // same response ID + const copiedRoot = path.join(context.home, "second/conversations"); + await mkdir(copiedRoot, { recursive: true }); + await writeFile(path.join(copiedRoot, "conv.db"), await readFile(file)); + context.sources[0].roots.push("~/second/conversations"); + const copy = await run(context, true); + assert.equal(total(copy), 80); +}); + +test("Antigravity warns for malformed/oversized blobs and missing IDs/times, replays cached warnings", async (t) => { + const context = await env(t); + const unknown = generation({ id: "", model: "", step: 5 }); + const { db } = database(context, "conv", [ + { idx: 0, ...unknown }, + { idx: 1, data: Buffer.from([10,255]) }, + { idx: 2, data: Buffer.alloc(16 * 1024 * 1024 + 1) }, + ]); db.close(); + const first = await run(context, true); + assert.equal(total(first), 40); assert.equal(first.rows[0].model, "unknown"); + assert.equal(first.rows[0].day, "2026-09-26"); // session-created, not mismatched step + assert.match(first.warnings.join(" "), /malformed/); + assert.match(first.warnings.join(" "), /16 MiB/); + assert.match(first.warnings.join(" "), /session time/); + assert.match(first.warnings.join(" "), /no response ID/); + const warm = await run(context); assert.equal(warm.stats.filesRead, 0); assert.deepEqual(warm.warnings, first.warnings); +}); + +test("Antigravity preserves accepted history on corrupt DB, excludes duplicate transcript, restores fallback after DB removal", async (t) => { + const context = await env(t); await transcript(context); + const { db, file } = database(context); db.close(); + assert.equal(total(await run(context, true)), 40); + await writeFile(file, "not sqlite PRIVATE_DATABASE_CONTENT"); + const failed = await run(context); assert.equal(total(failed), 40); + assert.match(failed.warnings.join(" "), /previous accepted data retained/); + assert.equal(JSON.stringify(failed).includes("PRIVATE_DATABASE_CONTENT"), false); + await rm(file); + const fallback = await run(context); + assert.equal(total(fallback), 1004); assert.equal(fallback.rows[0].model, "unknown"); +}); + +test("Antigravity retains accepted SQLite history until its transcript fallback is readable", async (t) => { + if (process.platform === "win32" || process.getuid?.() === 0) { + t.skip("requires POSIX file permissions enforced for a non-root user"); + return; + } + const context = await env(t); + const fallback = await transcript(context); + const { db, file } = database(context); + db.close(); + assert.equal(total(await run(context, true)), 40); + await rm(file); + await chmod(fallback, 0o000); + try { + const failed = await run(context); + assert.equal(total(failed), 40); + assert.match(failed.warnings.join(" "), /scan failed/); + } finally { + await chmod(fallback, 0o600); + } + assert.equal(total(await run(context)), 1004); +}); + +for (const interruption of ["deadline", "read error"]) { + test(`Antigravity rolls back transcript replacement on ${interruption} and retains SQLite history`, async (t) => { + const context = await env(t); + const fallback = await realpath(await transcript(context)); + const step = JSON.parse(await readFile(fallback, "utf8")); + await writeFile(fallback, Array.from({ length: 4000 }, (_, index) => + JSON.stringify({ ...step, step_index: index }) + ).join("\n") + "\n"); + const { db, file } = database(context); + db.close(); + const store = emptyStore({ sources: context.sources.map((source) => ({ ...source, enabled: true })) }); + await scanSources(store, { home: context.home }); + const before = JSON.stringify(store); + await rm(file); + + const options = { home: context.home }; + let mocked; + if (interruption === "deadline") { + options.deadline = 1; + options.now = () => 2; + } else { + const handle = await open(fallback, "r"); + const prototype = Object.getPrototypeOf(handle); + const read = prototype.read; + await handle.close(); + let calls = 0; + mocked = t.mock.method(prototype, "read", function (...args) { + if (++calls === 2) throw new Error("synthetic read failure"); + return read.apply(this, args); + }); + } + try { + const interrupted = await scanSources(store, options); + assert.equal(total({ rows: snapshotRows(store) }), 40, "an unfinished fallback must not add a second copy of usage"); + assert.equal(JSON.stringify(store), before, "rows, dedup ownership and file state must roll back together"); + if (interruption === "deadline") { + assert.equal(interrupted.complete, false); + assert.equal(interrupted.abortedFile.path, fallback, "use the existing indexer handoff"); + } else { + assert.equal(interrupted.stats.errors, 1); + assert.match(interrupted.warnings.join(" "), /synthetic read failure/); + assert.doesNotMatch(interrupted.warnings.join(" "), /accepted prefix retained/); + } + } finally { + mocked?.mock.restore(); + } + + const completed = await scanSources(store, { home: context.home, mode: "worker" }); + assert.equal(completed.complete, true); + assert.equal(total({ rows: snapshotRows(store) }), 4000 * 1004); + assert.deepEqual(Object.values(store.files).map((entry) => entry.path), [fallback]); + assert.equal(Object.keys(store.dedup).length, 4000); + }); +} + +test("Antigravity handles databases disappearing during discovery and unreadable WAL metadata", async (t) => { + const context = await env(t); + const { db, file } = database(context); + db.close(); + const store = emptyStore({ sources: context.sources.map((source) => ({ ...source, enabled: true })) }); + await scanSources(store, { home: context.home }); + const before = JSON.stringify(store); + const warnings = []; + const sink = { warn: (message) => warnings.push(message), note: (message) => warnings.push(message) }; + const stats = { filesRead: 0, bytesRead: 0, unchangedFiles: 0 }; + const missing = await scanAntigravityDb(store, store.config.sources[0], `${file}.removed`, sink, stats); + assert.equal(missing.accepted, undefined); + assert.equal(JSON.stringify(store), before); + assert.equal(warnings.length, 0); + if (process.platform === "win32") return; + await symlink(`${file}-wal`, `${file}-wal`); // stat reports ELOOP, not a missing WAL + const unreadable = await scanAntigravityDb(store, store.config.sources[0], file, sink, stats); + assert.equal(unreadable.accepted, undefined); + assert.equal(JSON.stringify(store), before); + assert.match(warnings.join(" "), /cannot stat database WAL/); +}); + +test("Antigravity SQLite abort rolls back all file contributions and resumes on next scan", async (t) => { + const context = await env(t); + const second = generation({ id: "two", step: 3 }); + const { db } = database(context, "conv", [{ idx: 0, ...generation() }, { idx: 1, ...second, step: 3 }]); db.close(); + const store = emptyStore({ sources: context.sources.map((s) => ({ ...s, enabled: true })) }); + let checks = 0; + const stopped = await scanSources(store, { home: context.home, shouldStop: () => ++checks >= 3 }); + assert.equal(stopped.complete, false); assert.equal(Object.keys(store.files).length, 0); assert.equal(Object.keys(store.dedup).length, 0); + const complete = await scanSources(store, { home: context.home }); + assert.equal(complete.complete, true); + assert.equal(Object.keys(store.dedup).length, 2); +}); diff --git a/tests/herdr-insights/collector-config.test.mjs b/tests/herdr-insights/collector-config.test.mjs index 4284ba4..8f66bb7 100644 --- a/tests/herdr-insights/collector-config.test.mjs +++ b/tests/herdr-insights/collector-config.test.mjs @@ -18,6 +18,7 @@ import test from "node:test"; import { DEFAULT_CONFIG, defaultConfig, + migrateLoadedConfig, normalizeConfig, } from "../../herdr-insights/mods/insights/core/config.mjs"; import { emptyStore, newFileEntry, saveStore } from "../../herdr-insights/mods/insights/core/store.mjs"; @@ -392,7 +393,7 @@ test("sourceStats distinguishes disabled, missing, empty, ok, and error sources" assert.ok(result.warnings.some((warning) => warning.startsWith("[bad]"))); }); -test("default config and empty input include Pi, Codex, OpenCode and Claude Code sources", () => { +test("default config and empty input include Pi, Codex, OpenCode, Claude Code and Antigravity sources", () => { const expectedSources = [ { id: "pi", @@ -422,6 +423,13 @@ test("default config and empty input include Pi, Codex, OpenCode and Claude Code enabled: true, roots: ["~/.claude/projects"], }, + { + id: "antigravity", + label: "Antigravity", + adapter: "antigravity", + enabled: true, + roots: ["~/.gemini/antigravity-cli/conversations", "~/.gemini/antigravity-cli/brain"], + }, ]; assert.deepEqual(DEFAULT_CONFIG.sources, expectedSources); assert.deepEqual(defaultConfig().sources, expectedSources); @@ -466,7 +474,7 @@ test("untouched pre-Claude default stores gain Claude Code and scan local sessio const result = await snapshot(context); assert.deepEqual( result.config.sources.map((source) => source.id), - ["pi", "codex", "opencode", "claude"], + ["pi", "codex", "opencode", "claude", "antigravity"], ); assert.equal(result.sourceStats.claude.status, "ok"); assert.equal(result.rows.some((row) => row.harness === "claude"), true); @@ -515,7 +523,7 @@ test("configure with untouched legacy defaults keeps Claude and cached history", const result = await configure(context, { sources: legacyConfig.sources }); assert.deepEqual( result.config.sources.map((source) => source.id), - ["pi", "codex", "opencode", "claude"], + ["pi", "codex", "opencode", "claude", "antigravity"], ); assert.equal( result.rows.some((row) => row.session === "pi:removed" && row.model === "gpt-old"), @@ -524,6 +532,46 @@ test("configure with untouched legacy defaults keeps Claude and cached history", ); }); +test("pre-Antigravity default stores discover compact transcripts at the new default root", async () => { + const context = await env(); + const logs = path.join(context.home, ".gemini/antigravity-cli/brain/conversation/.system_generated/logs"); + await mkdir(logs, { recursive: true }); + await writeFile(path.join(logs, "transcript.jsonl"), JSON.stringify({ + source: "MODEL", step_index: 1, created_at: "2026-03-04T04:00:00Z", + input_tokens: 10, output_tokens: 5, cache_read_tokens: 20, + }) + "\n"); + const previous = normalizeConfig({ sources: DEFAULT_CONFIG.sources.slice(0, 4) }); + await saveStore(context.store, emptyStore(previous)); + const result = await snapshot(context); + assert.deepEqual(result.config.sources, defaultConfig().sources); + assert.equal(result.sourceStats.antigravity.status, "ok"); + assert.equal(result.rows.find((row) => row.harness === "antigravity").tokens.total, 35); + const persisted = JSON.parse(await readFile(context.store, "utf8")); + assert.deepEqual(persisted.config.sources, defaultConfig().sources); + assert.equal((await snapshot(context)).stats.unchangedFiles, 1); +}); + +test("untouched pre-Antigravity defaults migrate once and customized sources stay unchanged", () => { + const previous = normalizeConfig({ sources: DEFAULT_CONFIG.sources.slice(0, 4) }); + const upgraded = migrateLoadedConfig(previous); + assert.equal(upgraded.migrated, true); + assert.deepEqual(upgraded.config.sources, defaultConfig().sources); + assert.equal(migrateLoadedConfig(upgraded.config).migrated, false); + assert.equal(previous.sources.length, 4, "migration must not mutate its input"); + + for (const change of [ + (sources) => { sources[0].enabled = false; }, + (sources) => { sources[0].roots = ["~/custom-pi"]; }, + (sources) => { sources[0].label = "My Pi"; }, + (sources) => { sources.reverse(); }, + (sources) => { sources.length = 0; }, + ]) { + const custom = normalizeConfig({ sources: DEFAULT_CONFIG.sources.slice(0, 4) }); + change(custom.sources); + assert.deepEqual(migrateLoadedConfig(custom), { config: custom, migrated: false }); + } +}); + test("customized pre-Claude stores are not rewritten by the default-source migration", async () => { const context = await env(); const customDefaults = normalizeConfig({ From c4bb8b037d127b0daadd6880f6ef16a0a8c621a5 Mon Sep 17 00:00:00 2001 From: e2720pjk Date: Thu, 5 Nov 2026 17:12:02 +0800 Subject: [PATCH 2/7] refactor(insights): remove redundant Antigravity decoder checks Remove the unreachable varint fallback and duplicate length and metadata checks. The uint64 guard, remaining-buffer bound and checked metadata caller already cover these cases. --- herdr-insights/mods/insights/core/antigravity.mjs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/herdr-insights/mods/insights/core/antigravity.mjs b/herdr-insights/mods/insights/core/antigravity.mjs index 0c51da0..f9aeee7 100644 --- a/herdr-insights/mods/insights/core/antigravity.mjs +++ b/herdr-insights/mods/insights/core/antigravity.mjs @@ -61,7 +61,6 @@ function varint(bytes, cursor) { value |= BigInt(byte & 127) << BigInt(shift); if (byte < 128) return value; } - throw new Error("protobuf varint overflow"); } export function protobufFields(bytes) { @@ -76,7 +75,7 @@ export function protobufFields(bytes) { if (wire === 0) value = varint(bytes, cursor); else if (wire === 1 || wire === 2 || wire === 5) { const size = wire === 2 ? Number(varint(bytes, cursor)) : wire === 1 ? 8 : 4; - if (!Number.isSafeInteger(size) || size > bytes.length - cursor.pos) { + if (size > bytes.length - cursor.pos) { throw new Error("invalid protobuf length"); } value = bytes.subarray(cursor.pos, cursor.pos + size); @@ -149,7 +148,6 @@ export function decodeAntigravityGeneration(bytes) { } function sessionMetadata(bytes) { - if (!bytes) return {}; const fields = protobufFields(bytes); let project = "unknown"; const workspace = blob(fields, 1); From 5a2cd2407845f0259219f1fc3fb451fa83b4e954 Mon Sep 17 00:00:00 2001 From: GaganSD Date: Thu, 5 Nov 2026 18:33:47 +0530 Subject: [PATCH 3/7] refactor(insights): simplify Antigravity config and tests --- herdr-insights/mods/insights/core/config.mjs | 13 +++---------- tests/herdr-insights/collector-antigravity.test.mjs | 9 +++------ 2 files changed, 6 insertions(+), 16 deletions(-) diff --git a/herdr-insights/mods/insights/core/config.mjs b/herdr-insights/mods/insights/core/config.mjs index 56d0fce..5557444 100644 --- a/herdr-insights/mods/insights/core/config.mjs +++ b/herdr-insights/mods/insights/core/config.mjs @@ -56,12 +56,6 @@ const ANTIGRAVITY_SOURCE = Object.freeze({ ]), }); -// Upgrade only untouched previous defaults; customized sources stay as saved. -const LEGACY_DEFAULT_SOURCE_SETS = Object.freeze([ - [PI_SOURCE, CODEX_SOURCE, OPENCODE_SOURCE], - [PI_SOURCE, CODEX_SOURCE, OPENCODE_SOURCE, CLAUDE_SOURCE], -]); - export const DEFAULT_CONFIG = Object.freeze({ profile: { name: "Developer", handle: "", avatar: "" }, palette: "blue", @@ -377,10 +371,9 @@ function sameSource(left, right) { } export function migrateLoadedConfig(config) { - const untouchedDefaults = LEGACY_DEFAULT_SOURCE_SETS.some((sources) => - config.sources.length === sources.length && - config.sources.every((source, index) => sameSource(source, sources[index])) - ); + // Upgrade only untouched previous defaults; customized sources stay as saved. + const untouchedDefaults = [3, 4].includes(config.sources.length) && + config.sources.every((source, index) => sameSource(source, DEFAULT_CONFIG.sources[index])); if (!untouchedDefaults) { return { config, migrated: false }; } diff --git a/tests/herdr-insights/collector-antigravity.test.mjs b/tests/herdr-insights/collector-antigravity.test.mjs index fc01e34..ef6d88f 100644 --- a/tests/herdr-insights/collector-antigravity.test.mjs +++ b/tests/herdr-insights/collector-antigravity.test.mjs @@ -1,11 +1,12 @@ import assert from "node:assert/strict"; -import { spawn } from "node:child_process"; +import { execFile } from "node:child_process"; import { createHash } from "node:crypto"; import { chmod, mkdir, mkdtemp, open, readFile, realpath, rm, symlink, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { DatabaseSync } from "node:sqlite"; import test from "node:test"; +import { promisify } from "node:util"; import { decodeAntigravityGeneration, protobufFields, scanAntigravityDb } from "../../herdr-insights/mods/insights/core/antigravity.mjs"; import { emptyStore, snapshotRows } from "../../herdr-insights/mods/insights/core/store.mjs"; import { scanSources } from "../../herdr-insights/mods/insights/core/scan.mjs"; @@ -52,11 +53,7 @@ async function transcript(context, name = "conv") { async function run(context, configure = false) { const args = [collector.pathname, configure ? "configure" : "snapshot", "--store", context.store, "--home", context.home]; if (configure) args.push("--config", JSON.stringify({ sources: context.sources })); - const child = spawn(process.execPath, args, { env: { ...process.env, TZ: "UTC" }, stdio: ["ignore", "pipe", "pipe"] }); - let stdout = "", stderr = ""; - child.stdout.on("data", (x) => stdout += x); child.stderr.on("data", (x) => stderr += x); - const code = await new Promise((resolve, reject) => { child.on("close", resolve); child.on("error", reject); }); - assert.equal(code, 0, stderr); + const { stdout } = await promisify(execFile)(process.execPath, args, { env: { ...process.env, TZ: "UTC" } }); return JSON.parse(stdout); } const total = (snapshot) => snapshot.rows.reduce((n, r) => n + r.tokens.total, 0); From 6d4c06bbb90101e00ec59beaa7c2f217ae9b66d3 Mon Sep 17 00:00:00 2001 From: GaganSD Date: Thu, 5 Nov 2026 18:56:34 +0530 Subject: [PATCH 4/7] fix(insights): replace Antigravity history atomically --- .../mods/insights/core/antigravity.mjs | 21 +++- herdr-insights/mods/insights/core/scan.mjs | 42 ++++--- .../collector-antigravity.test.mjs | 113 ++++++++++++++++++ 3 files changed, 154 insertions(+), 22 deletions(-) diff --git a/herdr-insights/mods/insights/core/antigravity.mjs b/herdr-insights/mods/insights/core/antigravity.mjs index f9aeee7..617ab13 100644 --- a/herdr-insights/mods/insights/core/antigravity.mjs +++ b/herdr-insights/mods/insights/core/antigravity.mjs @@ -32,6 +32,7 @@ import { beginStoreTransaction, commitStoreTransaction, rollbackStoreTransaction, + deleteFileEntry, fileKey, newFileEntry, resetFileEntry, @@ -39,7 +40,7 @@ import { } from "./store.mjs"; import { localDay } from "./util.mjs"; -const VERSION = 1; +const VERSION = 2; const MAX_BLOB = 16 * 1024 * 1024; const MAX_FIELDS = 100_000; @@ -124,12 +125,14 @@ export function decodeAntigravityGeneration(bytes) { if (value <= BigInt(Number.MAX_SAFE_INTEGER)) step = Number(value); } const generations = []; + let recognizedUsage = false; for (const field of outer) { if (field.number !== 1 || field.wire !== 2) continue; const inner = protobufFields(field.value); const usageBytes = blob(inner, 4); if (!usageBytes) continue; const usage = protobufFields(usageBytes); + recognizedUsage = true; const system = integer(usage, 1), input = integer(usage, 2); const output = integer(usage, 3), cacheRead = integer(usage, 5); const reasoning = integer(usage, 10), outputText = integer(usage, 9); @@ -144,6 +147,7 @@ export function decodeAntigravityGeneration(bytes) { tokens: { input: system + input, output, cacheRead, cacheWrite: 0, reasoning }, }); } + if (!recognizedUsage) throw new Error("unrecognized generation metadata"); return generations; } @@ -188,11 +192,12 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option } const fingerprint = `${info.dev}:${info.ino}:${info.size}:${info.mtimeMs}:${wal?.size ?? 0}:${wal?.mtimeMs ?? 0}`; let entry = store.files[fileKey(source.id, file)]; - if (entry?.state.agyDbVersion === VERSION && entry.state.agyDbFingerprint === fingerprint) { + const supersededKeys = options.supersededKeys ?? []; + if (supersededKeys.length === 0 && entry?.state.agyDbVersion === VERSION && entry.state.agyDbFingerprint === fingerprint) { for (const warning of entry.state.agyWarnings ?? []) sink.warn(warning); for (const note of entry.state.agyNotes ?? []) sink.note(note); stats.unchangedFiles += 1; - return { dirty: false, aborted: false, accepted: true }; + return { dirty: false, aborted: false }; } let sqlite; try { sqlite = await import("node:sqlite"); } catch { @@ -203,7 +208,9 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option let db; const counts = { oversized: 0, malformed: 0, approximateTime: 0, missingId: 0 }; let rows = 0; + let decodedRows = 0; try { + for (const key of supersededKeys) deleteFileEntry(store, key); db = new sqlite.DatabaseSync(file, { readOnly: true }); db.exec("BEGIN"); // SQLite limits the returned blob before it crosses into JS memory. @@ -233,6 +240,7 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option if (!record.data) { counts.oversized += 1; continue; } let generations; try { generations = decodeAntigravityGeneration(record.data); } catch { counts.malformed += 1; continue; } + decodedRows += 1; for (let index = 0; index < generations.length; index += 1) { const generation = generations[index]; let created = null; @@ -253,6 +261,11 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option : makeDedupKey("antigravity", "generation", conversation, record.idx, index)); } } + if (rows > 0 && decodedRows === 0) { + rollbackStoreTransaction(tx); + sink.warn(`Antigravity ${path.basename(file)}: no decodable generation records; previous accepted data retained.`); + return { dirty: false, aborted: false }; + } db.exec("COMMIT"); const warnings = []; const notes = []; @@ -272,7 +285,7 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option for (const note of notes) sink.note(note); stats.filesRead += 1; stats.bytesRead += info.size + (wal?.size ?? 0); - return { dirty: true, aborted: false, accepted: true }; + return { dirty: true, aborted: false }; } catch { rollbackStoreTransaction(tx); // Do not leak strings from SQLite or protobuf source payloads in diagnostics. diff --git a/herdr-insights/mods/insights/core/scan.mjs b/herdr-insights/mods/insights/core/scan.mjs index 90d5c8d..5bb4503 100644 --- a/herdr-insights/mods/insights/core/scan.mjs +++ b/herdr-insights/mods/insights/core/scan.mjs @@ -32,7 +32,7 @@ import { const CHUNK_BYTES = 256 * 1024; const MAX_WARNINGS = 50; const MAX_TRAVERSAL_ENTRIES = 250_000; -const PARSER_VERSIONS = { pi: 3, codex: 5, claude: 2, antigravity: 1 }; +const PARSER_VERSIONS = { pi: 3, codex: 5, claude: 2, antigravity: 2 }; const PROGRESS_CHUNK_INTERVAL = 64; export class ScanStopError extends Error { @@ -1026,6 +1026,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { const ino = String(fileStat.ino); const key = fileKey(source.id, filePath); const existing = store.files[key]; + const supersededKeys = options.supersededKeys ?? []; const isNew = !existing; const parserVersion = PARSER_VERSIONS[existing?.adapter ?? source.adapter]; const parserKey = `${existing?.adapter ?? source.adapter}ParserVersion`; @@ -1033,6 +1034,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { && !(existing?.state.unsupported === "zstd" && hasZstdSupport()); if ( existing + && supersededKeys.length === 0 && parserCurrent && !existing.state.readFailed && !existing.state.scanPending @@ -1042,7 +1044,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { && existing.mtimeMs === fileStat.mtimeMs ) { stats.unchangedFiles += 1; - return { dirty: false, aborted: false, accepted: true }; + return { dirty: false, aborted: false }; } // Every changed file gets a precise transaction. The previous committed @@ -1052,6 +1054,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { let entry = existing; let dirty = false; try { + for (const superseded of supersededKeys) deleteFileEntry(store, superseded); if (!entry) { entry = newFileEntry(store, { sourceId: source.id, @@ -1066,7 +1069,11 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { dirty = true; } - let forceRead = false; + let forceRead = supersededKeys.length > 0; + if (forceRead) { + resetFileEntry(store, entry); + dirty = true; + } if (!isNew && parserVersion && entry.state[parserKey] !== parserVersion) { // Migrate existing history once for accounting fixes. New files own no // metadata and must never trigger a global ownership cleanup scan. @@ -1203,7 +1210,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { entry.mtimeMs = fileStat.mtimeMs; delete entry.state.scanPending; } - return { dirty, aborted: false, pending: plain.pending, accepted: !entry.state.readFailed && !plain.pending }; + return { dirty, aborted: false, pending: plain.pending }; } catch (error) { rollbackStoreTransaction(tx); sourceSink.warn(`${source.adapter} ${filePath}: scan failed: ${error.message}`); @@ -1245,6 +1252,13 @@ export async function scanSources(store, { } const sink = warningSink(warnings, stats, sourceState); const discovered = await discoverFiles(store.config, home, sink, sourceState); + const antigravityFiles = new Map(); + for (const [key, entry] of Object.entries(store.files)) { + if (entry.adapter !== "antigravity") continue; + const conversation = antigravityConversation(entry.path); + if (!antigravityFiles.has(conversation)) antigravityFiles.set(conversation, []); + antigravityFiles.get(conversation).push(key); + } const seenInodes = new Set(); const seenPaths = new Set(); const options = { @@ -1306,13 +1320,18 @@ export async function scanSources(store, { sourceState[found.source.id].files += 1; const beforeRead = stats.filesRead; const beforeBytes = stats.bytesRead; + // Remove old owners inside the file transaction, before max-upsert can inherit their values. + const supersededKeys = found.source.adapter === "antigravity" + ? (antigravityFiles.get(antigravityConversation(found.file)) ?? []) + .filter((key) => key !== fileKey(found.source.id, found.file)) + : []; const result = await scanOneFile( store, found.source, found.file, sink, stats, - options, + { ...options, supersededKeys }, ).catch((error) => { sink.warn( `[${found.source.id}] ${found.source.adapter} ${found.file}: scan failed: ${error.message}`, @@ -1325,19 +1344,6 @@ export async function scanSources(store, { sourceState[found.source.id].bytesRead += stats.bytesRead - beforeBytes; dirty = Boolean(result?.dirty) || dirty; - if (found.source.adapter === "antigravity" && result?.accepted) { - // Superseded cached exports must disappear too, not just be excluded - // from discovery. Keep prior accepted data until its replacement succeeds. - const conversation = antigravityConversation(found.file); - const selectedKey = fileKey(found.source.id, found.file); - for (const [key, entry] of Object.entries(store.files)) { - if (key !== selectedKey && entry.adapter === "antigravity" && - antigravityConversation(entry.path) === conversation) { - deleteFileEntry(store, key); - dirty = true; - } - } - } if (result?.aborted) { stopReason = result.reason ?? "budget"; abortedFile = result.abortedFile ?? { diff --git a/tests/herdr-insights/collector-antigravity.test.mjs b/tests/herdr-insights/collector-antigravity.test.mjs index ef6d88f..66af121 100644 --- a/tests/herdr-insights/collector-antigravity.test.mjs +++ b/tests/herdr-insights/collector-antigravity.test.mjs @@ -102,6 +102,119 @@ test("Antigravity SQLite supersedes cached transcript without duplicate totals, assert.equal(total(await run(context)), 40); }); +test("Antigravity database handoff replaces token maxima and attribution, but rolls back an unusable replacement", async (t) => { + const context = await env(t); + context.sources[0].roots.push("~/second/conversations"); + const original = database(context); + original.db.close(); + await mkdir(path.join(context.home, "second/conversations"), { recursive: true }); + const replacement = database({ home: path.join(context.home, "second") }, "conv", [ + { idx: 0, data: Buffer.from([10, 255]) }, + ]); + replacement.db.close(); + const first = await run(context, true); + assert.equal(total(first), 40); + const before = JSON.parse(await readFile(context.store, "utf8")); + await rm(original.file); + + const failed = await run(context); + assert.equal(total(failed), 40); + const retained = JSON.parse(await readFile(context.store, "utf8")); + assert.deepEqual(retained.files, before.files); + assert.deepEqual(retained.dedup, before.dedup); + assert.equal(retained.nextFileId, before.nextFileId); + + const g = generation({ model: "claude-new", system: 0, input: 10, output: 0, thinking: 0, cache: 0 }); + const db = new DatabaseSync(replacement.file); + db.prepare("UPDATE gen_metadata SET data = ?").run(g.data); + db.prepare("INSERT INTO steps VALUES (1, ?)").run(join(msg(1, ts(1790572800)), msg(9, g.usage))); + db.close(); + const next = await run(context); + assert.equal(next.rows.length, 1); + assert.equal(total(next), 10); + assert.equal(next.rows[0].model, "claude-new"); + assert.equal(next.rows[0].provider, "anthropic"); + assert.equal(next.rows[0].day, "2026-09-28"); + const warm = await run(context); + assert.deepEqual(warm.rows, next.rows); + assert.equal(warm.stats.filesRead, 0); + const saved = JSON.parse(await readFile(context.store, "utf8")); + assert.deepEqual(Object.values(saved.files).map((entry) => entry.path), [await realpath(replacement.file)]); +}); + +test("Antigravity transcript handoff replaces revisions from the superseded copy", async (t) => { + const context = await env(t); + const original = await transcript(context); + context.sources[0].roots.push("~/second/brain"); + const replacement = await transcript({ home: path.join(context.home, "second") }); + await writeFile(replacement, JSON.stringify({ + source: "MODEL", step_index: 1, created_at: "2026-09-28T00:00:00Z", + input_tokens: 10, model: "claude-new", provider: "anthropic", + }) + "\n"); + assert.equal(total(await run(context, true)), 1004); + await rm(original); + const next = await run(context); + assert.equal(total(next), 10); + assert.equal(next.rows[0].model, "claude-new"); + assert.equal(next.rows[0].day, "2026-09-28"); + assert.deepEqual((await run(context)).rows, next.rows); +}); + +for (const invalid of ["malformed", "oversized", "unrecognized"]) { + test(`Antigravity retains accepted database history when all generation records are ${invalid}`, async (t) => { + const context = await env(t); + const { db, file } = database(context); + db.close(); + const first = await run(context, true); + const before = JSON.parse(await readFile(context.store, "utf8")); + const bad = invalid === "malformed" ? Buffer.from([10, 255]) + : invalid === "oversized" ? Buffer.alloc(16 * 1024 * 1024 + 1) + : num(7, 1); + const changed = new DatabaseSync(file); + changed.prepare("UPDATE gen_metadata SET data = ?").run(bad); + changed.close(); + const failed = await run(context); + assert.deepEqual(failed.rows, first.rows); + assert.match(failed.warnings.join(" "), /previous accepted data retained/); + const retained = JSON.parse(await readFile(context.store, "utf8")); + assert.deepEqual(retained.files, before.files); + assert.deepEqual(retained.dedup, before.dedup); + const repaired = new DatabaseSync(file); + repaired.prepare("UPDATE gen_metadata SET data = ?").run(generation().data); + repaired.close(); + assert.deepEqual((await run(context)).rows, first.rows); + }); +} + +test("Antigravity keeps a transcript until the preferred database has decodable metadata", async (t) => { + const context = await env(t); + await transcript(context); + const first = await run(context, true); + const { db } = database(context, "conv", [{ idx: 0, data: Buffer.from([10, 255]) }]); + db.close(); + const failed = await run(context); + assert.deepEqual(failed.rows, first.rows); + assert.match(failed.warnings.join(" "), /previous accepted data retained/); + const saved = JSON.parse(await readFile(context.store, "utf8")); + assert.equal(Object.values(saved.files)[0].path.endsWith("transcript.jsonl"), true); +}); + +test("Antigravity accepts genuinely empty and system-only databases", async (t) => { + const context = await env(t); + await transcript(context); + assert.equal(total(await run(context, true)), 1004); + const { db } = database(context, "conv", []); + db.close(); + assert.equal(total(await run(context)), 0); + const systemOnly = database(context, "system-only", [ + { idx: 0, ...generation({ input: 0, output: 0, thinking: 0, cache: 0 }) }, + ]); + systemOnly.db.close(); + const next = await run(context); + assert.equal(next.rows.length, 0); + assert.equal(next.stats.errors, 0); +}); + test("Antigravity SQLite handles WAL append and response-ID dedup across sessions and copied roots", async (t) => { const context = await env(t); const { db, file } = database(context); db.exec("PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=0"); From 671b4b4fbf2501a1204d989b18a7791c2da13168 Mon Sep 17 00:00:00 2001 From: GaganSD Date: Thu, 5 Nov 2026 19:01:09 +0530 Subject: [PATCH 5/7] fix(insights): yield during Antigravity database scans --- .../mods/insights/core/antigravity.mjs | 10 ++- .../collector-antigravity.test.mjs | 84 +++++++++++++++---- 2 files changed, 77 insertions(+), 17 deletions(-) diff --git a/herdr-insights/mods/insights/core/antigravity.mjs b/herdr-insights/mods/insights/core/antigravity.mjs index 617ab13..0757b90 100644 --- a/herdr-insights/mods/insights/core/antigravity.mjs +++ b/herdr-insights/mods/insights/core/antigravity.mjs @@ -27,6 +27,7 @@ import path from "node:path"; import { stat } from "node:fs/promises"; import { fileURLToPath } from "node:url"; +import { setImmediate } from "node:timers/promises"; import { addUsage, beginStoreTransaction, @@ -209,6 +210,7 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option const counts = { oversized: 0, malformed: 0, approximateTime: 0, missingId: 0 }; let rows = 0; let decodedRows = 0; + let bytes = 0; try { for (const key of supersededKeys) deleteFileEntry(store, key); db = new sqlite.DatabaseSync(file, { readOnly: true }); @@ -229,6 +231,11 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option else resetFileEntry(store, entry); const conversation = antigravityConversation(file); for (const record of records) { + if (rows > 0 && rows % 128 === 0) { + options.onStreamProgress?.({ file, bytes, phase: "scan" }); + // Let signal handlers and generation-fence polling run between batches. + await setImmediate(); + } const stopped = options.shouldStop?.(); const expired = rows > 0 && Number.isFinite(options.deadline) && (options.now ?? Date.now)() >= options.deadline; @@ -237,6 +244,7 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option return { dirty: false, aborted: true, reason: stopped ? "stop" : "budget" }; } rows += 1; + bytes += record.data?.length ?? 0; if (!record.data) { counts.oversized += 1; continue; } let generations; try { generations = decodeAntigravityGeneration(record.data); } catch { counts.malformed += 1; continue; } @@ -244,7 +252,7 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option for (let index = 0; index < generations.length; index += 1) { const generation = generations[index]; let created = null; - if (generation.step !== null) { + if (generation.step !== null && generation.responseId) { const step = stepQuery.get(generation.step); try { created = stepTimestamp(step?.data, generation.responseId); } catch { /* use documented fallback */ } } diff --git a/tests/herdr-insights/collector-antigravity.test.mjs b/tests/herdr-insights/collector-antigravity.test.mjs index 66af121..6c8d912 100644 --- a/tests/herdr-insights/collector-antigravity.test.mjs +++ b/tests/herdr-insights/collector-antigravity.test.mjs @@ -10,6 +10,10 @@ import { promisify } from "node:util"; import { decodeAntigravityGeneration, protobufFields, scanAntigravityDb } from "../../herdr-insights/mods/insights/core/antigravity.mjs"; import { emptyStore, snapshotRows } from "../../herdr-insights/mods/insights/core/store.mjs"; import { scanSources } from "../../herdr-insights/mods/insights/core/scan.mjs"; +import { readWorkerClaim, writeFence } from "../../herdr-insights/mods/insights/core/fence.mjs"; +import { ensureIndexer } from "../../herdr-insights/mods/insights/core/indexer-control.mjs"; +import { readLiveProgress } from "../../herdr-insights/mods/insights/core/progress.mjs"; +import { isProcessAlive, sleep } from "../../herdr-insights/mods/insights/core/util.mjs"; const collector = new URL("../../herdr-insights/mods/insights/collector.mjs", import.meta.url); function vi(value) { @@ -352,27 +356,75 @@ test("Antigravity handles databases disappearing during discovery and unreadable const warnings = []; const sink = { warn: (message) => warnings.push(message), note: (message) => warnings.push(message) }; const stats = { filesRead: 0, bytesRead: 0, unchangedFiles: 0 }; - const missing = await scanAntigravityDb(store, store.config.sources[0], `${file}.removed`, sink, stats); - assert.equal(missing.accepted, undefined); + await scanAntigravityDb(store, store.config.sources[0], `${file}.removed`, sink, stats); assert.equal(JSON.stringify(store), before); assert.equal(warnings.length, 0); if (process.platform === "win32") return; await symlink(`${file}-wal`, `${file}-wal`); // stat reports ELOOP, not a missing WAL - const unreadable = await scanAntigravityDb(store, store.config.sources[0], file, sink, stats); - assert.equal(unreadable.accepted, undefined); + await scanAntigravityDb(store, store.config.sources[0], file, sink, stats); assert.equal(JSON.stringify(store), before); assert.match(warnings.join(" "), /cannot stat database WAL/); }); -test("Antigravity SQLite abort rolls back all file contributions and resumes on next scan", async (t) => { - const context = await env(t); - const second = generation({ id: "two", step: 3 }); - const { db } = database(context, "conv", [{ idx: 0, ...generation() }, { idx: 1, ...second, step: 3 }]); db.close(); - const store = emptyStore({ sources: context.sources.map((s) => ({ ...s, enabled: true })) }); - let checks = 0; - const stopped = await scanSources(store, { home: context.home, shouldStop: () => ++checks >= 3 }); - assert.equal(stopped.complete, false); assert.equal(Object.keys(store.files).length, 0); assert.equal(Object.keys(store.dedup).length, 0); - const complete = await scanSources(store, { home: context.home }); - assert.equal(complete.complete, true); - assert.equal(Object.keys(store.dedup).length, 2); -}); +for (const interruption of ["SIGTERM", "generation fence"]) { + test(`Antigravity worker rolls back a database scan on ${interruption} and resumes`, async (t) => { + const context = await env(t); + const original = database(context); + original.db.close(); + const first = await run(context, true); + const before = JSON.parse(await readFile(context.store, "utf8")); + const db = new DatabaseSync(original.file); + db.prepare(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 100000) + INSERT INTO gen_metadata SELECT i, ? FROM n`).run(generation({ id: "appended", input: 11, step: 2 }).data); + db.close(); + + async function waitFor(predicate) { + const deadline = Date.now() + 30_000; + while (Date.now() < deadline) { + const result = await predicate(); + if (result) return result; + await sleep(10); + } + throw new Error("Antigravity worker did not settle"); + } + + let worker; + try { + worker = await ensureIndexer(context.store, context.home, { generation: before.generation }); + assert.equal(worker.spawned, true); + const progress = await waitFor(async () => { + const live = await readLiveProgress(context.store); + if (live?.pid === worker.pid && live.decodedBytes > 0) return live; + return (await readWorkerClaim(context.store)) === null ? { finished: true } : null; + }); + assert.ok(progress.decodedBytes > 0, "interruption must occur inside the database scan"); + if (interruption === "SIGTERM") process.kill(worker.pid, "SIGTERM"); + else await writeFence(context.store, before.generation + 1); + await waitFor(async () => (await readWorkerClaim(context.store)) === null); + const stopped = JSON.parse(await readFile(context.store, "utf8")); + assert.deepEqual(snapshotRows(stopped), first.rows); + assert.deepEqual(stopped.files, before.files); + assert.deepEqual(stopped.dedup, before.dedup); + assert.notEqual(stopped.scan.status, "complete"); + if (interruption === "SIGTERM") { + assert.equal(stopped.scan.status, "partial"); + assert.equal(stopped.scan.pendingFile.path, await realpath(original.file)); + } else { + await writeFence(context.store, before.generation); + } + + worker = await ensureIndexer(context.store, context.home, { generation: before.generation }); + assert.equal(worker.spawned, true); + await waitFor(async () => (await readWorkerClaim(context.store)) === null); + const completed = JSON.parse(await readFile(context.store, "utf8")); + assert.equal(completed.scan.status, "complete"); + assert.equal(total({ rows: snapshotRows(completed) }), 81); + assert.equal(snapshotRows(completed).reduce((n, row) => n + row.messages, 0), 2); + } finally { + if (worker?.pid && isProcessAlive(worker.pid)) { + process.kill(worker.pid, "SIGTERM"); + await waitFor(() => !isProcessAlive(worker.pid)); + } + } + }); +} From 69b6988a561b57f612a429739a1dc5586c9397b4 Mon Sep 17 00:00:00 2001 From: GaganSD Date: Thu, 5 Nov 2026 19:05:02 +0530 Subject: [PATCH 6/7] test(insights): document and independently check Antigravity wire decoding --- .../mods/insights/core/adapters.mjs | 6 +-- .../mods/insights/core/antigravity.mjs | 41 ++++--------------- herdr-insights/mods/insights/core/scan.mjs | 9 ++-- .../collector-antigravity.test.mjs | 16 +++++++- .../fixtures/antigravity-generation.hex | 1 + .../fixtures/antigravity-generation.md | 18 ++++++++ 6 files changed, 46 insertions(+), 45 deletions(-) create mode 100644 tests/herdr-insights/fixtures/antigravity-generation.hex create mode 100644 tests/herdr-insights/fixtures/antigravity-generation.md diff --git a/herdr-insights/mods/insights/core/adapters.mjs b/herdr-insights/mods/insights/core/adapters.mjs index 2503d70..8645405 100644 --- a/herdr-insights/mods/insights/core/adapters.mjs +++ b/herdr-insights/mods/insights/core/adapters.mjs @@ -934,9 +934,7 @@ function parseCodex(record, ctx) { } // --------------------------------------------------------------------------- -// Antigravity compact transcripts (CLI and current IDE/desktop surfaces). -// Counts are per model step; input_tokens excludes cache_read_tokens. Missing -// model/provider/workspace fields remain unknown: artifacts are not usage evidence. +// Antigravity compact transcript fallback; counts are per model step. function parseAntigravity(record, ctx) { if (!isPlainObject(record) || record.source !== "MODEL") return; if (!Number.isSafeInteger(record.step_index) || record.step_index < 0) return; @@ -952,8 +950,6 @@ function parseAntigravity(record, ctx) { cacheRead: record.cache_read_tokens, }); if (!tokens.nonzero) return; - // Path-independent identity deduplicates copied sessions across product roots - // and repeated revisions of one step; separate conversations stay separate. const session = `antigravity:${conversation}`; ctx.addUsage( metaFor("antigravity", session, timestamp, record.workspace, record.provider, record.model), diff --git a/herdr-insights/mods/insights/core/antigravity.mjs b/herdr-insights/mods/insights/core/antigravity.mjs index 0757b90..d116e58 100644 --- a/herdr-insights/mods/insights/core/antigravity.mjs +++ b/herdr-insights/mods/insights/core/antigravity.mjs @@ -1,28 +1,9 @@ /** - * Antigravity CLI usage from conversations/.db (read-only SQLite). - * Canonical brain//.system_generated/logs/transcript.jsonl exports are a - * fallback in adapters.mjs. Discovery selects one file per conversation: SQLite - * first, then configured root order. Cached alternatives are removed only after - * a replacement is accepted. Unreadable databases retain accepted history. - * - * Reverse-engineered protobuf mapping (not an official stable API): - * gen_metadata.data #1.#19: response model; #1.#4: usage. - * Usage #1 + #2: system + uncached input; #5: cache read; #3: total output; - * #9: text output; #10: thinking (a subset of output); #11: response ID. - * Envelope #2: packed step indexes. steps.metadata #1: creation timestamp, - * used only when #9.#11 matches the generation's response ID. Otherwise use - * trajectory_metadata_blob creation time, then database mtime, with a note. - * Trajectory metadata also supplies the workspace URI. - * Reject inconsistent output totals; skip prefilled system-only records without - * measured usage. Response IDs deduplicate copies; missing IDs use conversation - * and generation index. Provider denotes model publisher, not billing gateway. - * - * DB/WAL fingerprints invalidate cached usage. Blobs are read one row at a time, - * capped at 16 MiB; malformed/oversized records warn. Only usage metadata is - * persisted, never prompt/output/thinking text. No RPC, credentials or uploads. - * Retired Gemini CLI history and Antigravity IDE .pb stores are not decoded. - * See also: https://github.com/junhoyeo/tokscale/pull/713 - * https://github.com/robinebers/openusage/pull/1139 + * Reverse-engineered CLI schema, not a vendor API: gen_metadata #1.#19 is the + * model and #1.#4 is usage. Usage #1+#2 is uncached input, #3 is total output + * (including #10 reasoning), #5 is cache read, #9 is text output, #11 is response ID. + * References: https://github.com/junhoyeo/tokscale/pull/713 + * https://github.com/robinebers/openusage/pull/1139 */ import path from "node:path"; import { stat } from "node:fs/promises"; @@ -51,9 +32,7 @@ export function antigravityConversation(file) { : path.basename(path.dirname(path.dirname(path.dirname(file)))); } -// Read only selected protobuf fields. Unknown fields (including prompt payloads) -// are bounded byte slices, never decoded to text or persisted. uint64 uses BigInt -// so unrelated signed/sentinel fields cannot silently overflow JS numbers. +// Keep uint64 fields exact, including unrelated sentinel values outside JS's safe range. function varint(bytes, cursor) { let value = 0n; for (let shift = 0; shift < 70; shift += 7) { @@ -139,12 +118,10 @@ export function decodeAntigravityGeneration(bytes) { const reasoning = integer(usage, 10), outputText = integer(usage, 9); if (!Number.isSafeInteger(system + input + output + cacheRead)) throw new Error("unsafe token total"); if (outputText + reasoning !== output) throw new Error("inconsistent output accounting"); - // A fixed system-prompt count is prefilled even on abandoned/failed - // requests with no measured usage. It is not a completed generation. + // A system-prompt estimate alone is not measured request usage. if (input + output + cacheRead === 0) continue; generations.push({ step, model: text(blob(inner, 19)) || "unknown", responseId: responseId(usage), - // Output includes thinking; reasoning is a subset, never added to total. tokens: { input: system + input, output, cacheRead, cacheWrite: 0, reasoning }, }); } @@ -167,8 +144,7 @@ function stepTimestamp(bytes, id) { if (!bytes || !id) return null; const fields = protobufFields(bytes); const usage = blob(fields, 9); - // Packed step references were validated against response IDs on real stores. - // Do not attribute a timestamp if a future schema points at a different step. + // Trust a step timestamp only when it identifies the same response. if (!usage || responseId(protobufFields(usage)) !== id) return null; return timestamp(blob(fields, 1)); } @@ -269,6 +245,7 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option : makeDedupKey("antigravity", "generation", conversation, record.idx, index)); } } + // An undecodable snapshot is not evidence that previously measured usage vanished. if (rows > 0 && decodedRows === 0) { rollbackStoreTransaction(tx); sink.warn(`Antigravity ${path.basename(file)}: no decodable generation records; previous accepted data retained.`); diff --git a/herdr-insights/mods/insights/core/scan.mjs b/herdr-insights/mods/insights/core/scan.mjs index 5bb4503..3db2855 100644 --- a/herdr-insights/mods/insights/core/scan.mjs +++ b/herdr-insights/mods/insights/core/scan.mjs @@ -85,8 +85,7 @@ function warningSink(warnings, stats, sourceState) { } function isEligible(adapter, filePath) { - // Compact transcripts carry the same usage as full logs. Never scan both, - // chunks, workflow artifacts, or arbitrary JSONL files in a brain directory. + // Exclude full logs and artifacts; only the canonical compact export is a fallback. if (adapter === "antigravity") { if (filePath.endsWith(".db")) return path.basename(path.dirname(filePath)) === "conversations"; return path.basename(filePath) === "transcript.jsonl" && @@ -194,8 +193,7 @@ export async function discoverFiles(config, home, sink, sourceState) { a.rootIndex - b.rootIndex || a.file.localeCompare(b.file), ); - // One authoritative file per Antigravity conversation. SQLite wins over - // exports; configured root order resolves copied/divergent databases. + // SQLite wins over exports; configured root order breaks ties between copies. const preferred = new Map(); for (const found of discovered) { if (found.source.adapter !== "antigravity") continue; @@ -1195,8 +1193,7 @@ async function scanOneFile(store, source, filePath, sink, stats, options = {}) { stats, options, ); - // Match SQLite's atomic reads: an unfinished fallback must not add rows - // alongside the conversation's previously accepted usage. + // A partial fallback must not replace the conversation's last complete snapshot. if (source.adapter === "antigravity" && plain.pending) { rollbackStoreTransaction(tx); return { dirty: false, aborted: true, reason: stopRequested(options) ? "stop" : "budget" }; diff --git a/tests/herdr-insights/collector-antigravity.test.mjs b/tests/herdr-insights/collector-antigravity.test.mjs index 6c8d912..223f47e 100644 --- a/tests/herdr-insights/collector-antigravity.test.mjs +++ b/tests/herdr-insights/collector-antigravity.test.mjs @@ -62,6 +62,14 @@ async function run(context, configure = false) { } const total = (snapshot) => snapshot.rows.reduce((n, r) => n + r.tokens.total, 0); +test("Antigravity decodes the independently encoded protobuf wire vector", async () => { + const hex = await readFile(new URL("./fixtures/antigravity-generation.hex", import.meta.url), "utf8"); + assert.deepEqual(decodeAntigravityGeneration(Buffer.from(hex.trim(), "hex")), [{ + step: null, model: "gemini-3-flash-a", responseId: "resp-1", + tokens: { input: 1632, output: 340, cacheRead: 16000, cacheWrite: 0, reasoning: 40 }, + }]); +}); + test("Antigravity protobuf decoder validates accounting, bounds and uint64", () => { const [g] = decodeAntigravityGeneration(generation().data); assert.equal(g.model, "gemini-test"); assert.equal(g.responseId, "response-one"); @@ -77,7 +85,7 @@ test("Antigravity protobuf decoder validates accounting, bounds and uint64", () assert.throws(() => decodeAntigravityGeneration(msg(1, msg(4, num(1, 9007199254740992n)))), /unsafe/); }); -test("Antigravity SQLite: actual model, system input, reasoning subset, verified step day, read-only and privacy", async (t) => { +test("Antigravity SQLite: model attribution, token accounting, step day, read-only and privacy", async (t) => { const context = await env(t); const { db, file } = database(context); db.close(); const digest = async () => createHash("sha256").update(await readFile(file)).digest("hex"); const before = await digest(); @@ -94,7 +102,7 @@ test("Antigravity SQLite: actual model, system input, reasoning subset, verified assert.equal(warm.stats.filesRead, 0); assert.equal(warm.stats.unchangedFiles, 1); assert.equal(total(warm), 40); }); -test("Antigravity SQLite supersedes cached transcript without duplicate totals, with or without source reconfiguration", async (t) => { +test("Antigravity SQLite supersedes a cached transcript without duplicate totals", async (t) => { const context = await env(t); await transcript(context); assert.equal(total(await run(context, true)), 1004); const { db } = database(context); db.close(); @@ -368,6 +376,10 @@ test("Antigravity handles databases disappearing during discovery and unreadable for (const interruption of ["SIGTERM", "generation fence"]) { test(`Antigravity worker rolls back a database scan on ${interruption} and resumes`, async (t) => { + if (interruption === "SIGTERM" && process.platform === "win32") { + t.skip("requires POSIX signal delivery"); + return; + } const context = await env(t); const original = database(context); original.db.close(); diff --git a/tests/herdr-insights/fixtures/antigravity-generation.hex b/tests/herdr-insights/fixtures/antigravity-generation.hex new file mode 100644 index 0000000..e2f55e0 --- /dev/null +++ b/tests/herdr-insights/fixtures/antigravity-generation.hex @@ -0,0 +1 @@ +0a34221908ec0810f40318d40228807d48ac0250285a06726573702d319a011067656d696e692d332d666c6173682d61a50198badcfe38ffffffffffffffffff01411032547698badcfe diff --git a/tests/herdr-insights/fixtures/antigravity-generation.md b/tests/herdr-insights/fixtures/antigravity-generation.md new file mode 100644 index 0000000..fc78de4 --- /dev/null +++ b/tests/herdr-insights/fixtures/antigravity-generation.md @@ -0,0 +1,18 @@ +# Antigravity wire-format vector + +`antigravity-generation.hex` is synthetic, not a captured conversation. It was +encoded and round-tripped with Google's Python `protobuf==6.33.0`, independently +of this repository's JavaScript fixture builder. + +The model is `gemini-3-flash-a`, response ID `resp-1`; usage fields are +`1=1132, 2=500, 3=340, 5=16000, 9=300, 10=40`. Expected uncached input is +1632; output is 340, including 40 reasoning tokens. Unknown fields exercise +fixed32, fixed64 and the maximum uint64 without interpreting them as usage. + +Model, input, total-output and cache fields agree across these public implementations; +text-output, reasoning and response-ID fields follow tokscale's documented mapping: +- [tokscale, e9af34f](https://github.com/junhoyeo/tokscale/blob/e9af34fbd805674df0181dfaa2c6820da0b2b71d/crates/tokscale-core/src/sessions/antigravity_cli.rs) +- [OpenUsage, 2fd4fb0](https://github.com/robinebers/openusage/blob/2fd4fb071f7720e2f9cee7ac52c2c9684dd96e31/Sources/OpenUsage/Providers/Antigravity/AntigravityProtoDecoder.swift) + +This checks wire decoding and accounting, not compatibility with every Antigravity +version. Neither reference is an official schema; no real-session fixture is supplied. From fda65dcb0bf04552bb7a6afa07586d2c465046f5 Mon Sep 17 00:00:00 2001 From: GaganSD Date: Thu, 5 Nov 2026 19:18:37 +0530 Subject: [PATCH 7/7] fix(insights): retain SQLite statements on Node 22.15 --- .../mods/insights/core/antigravity.mjs | 4 ++- .../collector-antigravity.test.mjs | 28 ++++++++++++++++++- 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/herdr-insights/mods/insights/core/antigravity.mjs b/herdr-insights/mods/insights/core/antigravity.mjs index d116e58..e0855c3 100644 --- a/herdr-insights/mods/insights/core/antigravity.mjs +++ b/herdr-insights/mods/insights/core/antigravity.mjs @@ -199,7 +199,9 @@ export async function scanAntigravityDb(store, source, file, sink, stats, option try { session = sessionMetadata(meta.data); } catch { counts.malformed += 1; } } else if (meta) counts.oversized += 1; const stepQuery = db.prepare(`SELECT CASE WHEN length(metadata) <= ${MAX_BLOB} THEN metadata ELSE NULL END AS data FROM steps WHERE idx = ?`); - const records = db.prepare(`SELECT idx, ${limited} AS data FROM gen_metadata ORDER BY idx`).iterate(); + // Node 22.15 iterators do not retain their prepared statement across garbage collection. + const query = db.prepare(`SELECT idx, ${limited} AS data FROM gen_metadata ORDER BY idx`); + const records = Object.assign(query.iterate(), { statementOwner: query }); if (!entry) entry = newFileEntry(store, { sourceId: source.id, adapter: source.adapter, path: file, dev: String(info.dev), ino: String(info.ino), size: 0, mtimeMs: 0, compressed: false, diff --git a/tests/herdr-insights/collector-antigravity.test.mjs b/tests/herdr-insights/collector-antigravity.test.mjs index 223f47e..2b16b2c 100644 --- a/tests/herdr-insights/collector-antigravity.test.mjs +++ b/tests/herdr-insights/collector-antigravity.test.mjs @@ -374,6 +374,29 @@ test("Antigravity handles databases disappearing during discovery and unreadable assert.match(warnings.join(" "), /cannot stat database WAL/); }); +test("Antigravity keeps the SQLite statement alive when GC runs during iteration", async (t) => { + const context = await env(t); + const { db, file } = database(context); + db.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 256) + INSERT INTO gen_metadata SELECT i, (SELECT data FROM gen_metadata WHERE idx = 0) FROM n`); + db.close(); + const core = new URL("../../herdr-insights/mods/insights/core/", import.meta.url); + const script = ` + import assert from "node:assert/strict"; + import { scanAntigravityDb } from ${JSON.stringify(new URL("antigravity.mjs", core).href)}; + import { emptyStore, snapshotRows } from ${JSON.stringify(new URL("store.mjs", core).href)}; + const store = emptyStore(); + const warnings = []; + const result = await scanAntigravityDb(store, { id: "agy", adapter: "antigravity" }, + process.argv[1], { warn: (w) => warnings.push(w), note() {} }, + { filesRead: 0, bytesRead: 0, unchangedFiles: 0 }, { onStreamProgress: () => globalThis.gc() }); + assert.equal(result.dirty, true, warnings.join(" ")); + assert.equal(snapshotRows(store)[0].tokens.total, 40); + assert.equal(snapshotRows(store)[0].messages, 1); + `; + await promisify(execFile)(process.execPath, ["--expose-gc", "--input-type=module", "-e", script, file]); +}); + for (const interruption of ["SIGTERM", "generation fence"]) { test(`Antigravity worker rolls back a database scan on ${interruption} and resumes`, async (t) => { if (interruption === "SIGTERM" && process.platform === "win32") { @@ -409,7 +432,10 @@ for (const interruption of ["SIGTERM", "generation fence"]) { if (live?.pid === worker.pid && live.decodedBytes > 0) return live; return (await readWorkerClaim(context.store)) === null ? { finished: true } : null; }); - assert.ok(progress.decodedBytes > 0, "interruption must occur inside the database scan"); + if (!(progress.decodedBytes > 0)) { + const failed = JSON.parse(await readFile(context.store, "utf8")); + assert.fail(`worker finished before interruption: ${JSON.stringify(failed.scan)}`); + } if (interruption === "SIGTERM") process.kill(worker.pid, "SIGTERM"); else await writeFence(context.store, before.generation + 1); await waitFor(async () => (await readWorkerClaim(context.store)) === null);