From be2470d0e59a47c51308ec2e000dc6177798ab52 Mon Sep 17 00:00:00 2001 From: AgentSeal Date: Fri, 3 Jul 2026 12:32:46 +0200 Subject: [PATCH] fix(guard): dedupe streaming message copies in the incremental cost fold Claude Code rewrites each assistant message several times as it streams, every copy carrying the full final usage; the shipped parser dedupes these last-wins (dedupeStreamingMessageIds) but the guard fold summed every line, measuring real sessions at 2.5-2.8x their true cost and false-blocking the hard cap at roughly 40% of the configured spend. - The session cache now maps message id -> that id's cost contribution and each id-carrying line replaces its previous contribution; id-less lines keep plain adds. Validated against two real transcripts (90MB and 116MB): guard totals now equal the shipped deduped totals exactly. - Replace semantics also self-heal the trailing-line case: a complete final line without its newline is folded but byteOffset stops before it, so the next invocation re-reads it as a replace, not a double add. - editCount becomes a set-once sawEdit boolean so duplicate copies of an edit tool_use cannot inflate it; cache schema bumped to v2 (old caches cold-reparse once). - Per-session state moves to guard/sessions/ so a session id can never collide with the shared flags.json, dropping the doAllow special case. - The git-commit detector now requires commit as the git subcommand at a command boundary (start of string or line, or ; & |), with intra-command gaps that never cross newlines: 'git log --grep commit' and 'git diff && echo commit' no longer match, while newline-separated 'git add ...\ngit commit' in multi-line Bash calls now does (verified as a real false negative on a live transcript). - Corrected the statusline protocol note: each stdout line renders as its own row; we emit exactly one. - New tests: streaming-duplicate fixtures (3x identical, growing last-wins, incremental replace) asserted equal to a cold shipped-parser computation, the trailing-partial-line scenario, the commit-detector matrix, and a stale-plan test proving guard-install plans carry expectedHash (a concurrent settings edit aborts the apply and survives). The act list CLI spawn test now anchors to the repo root from the test file location. --- src/guard/cli.ts | 6 +-- src/guard/hooks.ts | 5 +- src/guard/store.ts | 10 +++- src/guard/usage.ts | 56 ++++++++++++++++----- src/parser.ts | 2 +- tests/act-journal.test.ts | 8 ++- tests/guard-hooks.test.ts | 97 ++++++++++++++++++++++++++++++++++++- tests/guard-install.test.ts | 14 ++++++ 8 files changed, 174 insertions(+), 24 deletions(-) diff --git a/src/guard/cli.ts b/src/guard/cli.ts index f62941d..0cc3ca7 100644 --- a/src/guard/cli.ts +++ b/src/guard/cli.ts @@ -116,12 +116,12 @@ async function doStatus(): Promise { } async function doAllow(sessionId: string | undefined): Promise { - const { guardDir } = await import('./store.js') + const { sessionsDir } = await import('./store.js') const { writeAllow } = await import('./usage.js') let id = sessionId if (!id) { - const dir = guardDir() - const names = (await readdir(dir).catch(() => [])).filter(f => f.endsWith('.json') && f !== 'flags.json') + const dir = sessionsDir() + const names = (await readdir(dir).catch(() => [])).filter(f => f.endsWith('.json')) let newest = { at: -1, id: '' } for (const name of names) { const st = await stat(join(dir, name)).catch(() => null) diff --git a/src/guard/hooks.ts b/src/guard/hooks.ts index 236b3e9..767b666 100644 --- a/src/guard/hooks.ts +++ b/src/guard/hooks.ts @@ -9,7 +9,8 @@ // cwd, hook_event_name, permission_mode. PreToolUse adds tool_name + // tool_input; SessionStart adds source (startup|resume|clear|compact). // statusLine stdin differs: session_id, transcript_path, workspace.current_dir, -// cost.total_cost_usd (plain text out, first stdout line rendered). +// cost.total_cost_usd (plain text out; each stdout line renders as its own +// status row, so we emit exactly one). // Exit/stdout contract: exit 0 -> stdout parsed as JSON output; exit 2 -> // blocking, stderr fed to the model. We ALWAYS exit 0 and encode any decision // as JSON, so an internal error is indistinguishable from "no opinion" @@ -106,7 +107,7 @@ async function handleStop(input: unknown, opts: HookOpts): Promise { if ( config.checkpointUSD !== null && cache.costUSD > config.checkpointUSD - && cache.editCount === 0 + && !cache.sawEdit && !cache.sawGitCommit && !cache.stopNotified ) { diff --git a/src/guard/store.ts b/src/guard/store.ts index bfe1ffb..c9e30e6 100644 --- a/src/guard/store.ts +++ b/src/guard/store.ts @@ -23,12 +23,18 @@ export function flagsPath(base?: string): string { return join(guardDir(base), 'flags.json') } +// Per-session state sits one level below the shared flags.json so a session id +// can never collide with it (e.g. a session literally named "flags"). +export function sessionsDir(base?: string): string { + return join(guardDir(base), 'sessions') +} + export function sessionCachePath(sessionId: string, base?: string): string { - return join(guardDir(base), `${sanitizeId(sessionId)}.json`) + return join(sessionsDir(base), `${sanitizeId(sessionId)}.json`) } export function allowPath(sessionId: string, base?: string): string { - return join(guardDir(base), `${sanitizeId(sessionId)}.allow`) + return join(sessionsDir(base), `${sanitizeId(sessionId)}.allow`) } // Session ids come from the hook payload; keep them to a filesystem-safe set so diff --git a/src/guard/usage.ts b/src/guard/usage.ts index 80edc72..eaafdc7 100644 --- a/src/guard/usage.ts +++ b/src/guard/usage.ts @@ -2,18 +2,28 @@ import { mkdir, readFile, stat, writeFile } from 'fs/promises' import { readSessionLines } from '../fs-utils.js' import { parseApiCall, parseJsonlLine } from '../parser.js' import { EDIT_TOOLS } from '../classifier.js' -import { allowPath, guardDir, sessionCachePath } from './store.js' +import { allowPath, sessionCachePath, sessionsDir } from './store.js' // Per-session running totals. The transcript is append-only, so each invocation // streams only the bytes after `byteOffset` (the offset of the last complete // line parsed) and folds them into the totals; a cold parse of a multi-hundred- // MB transcript on every tool call is what this avoids. +// +// Claude Code rewrites each assistant message several times as it streams, and +// every copy carries the full final usage. The shipped parser dedupes those +// copies last-wins (dedupeStreamingMessageIds); a plain sum here measured real +// sessions at ~3x their true cost. `perMessage` maps message id -> that id's +// current cost contribution, and each id-carrying line REPLACES its previous +// contribution instead of adding. This also self-heals the trailing-line case: +// a complete final line without its newline is folded but byteOffset stops +// before it, so the next invocation re-reads it as a replace, not a double add. export type GuardCache = { version: number sessionId: string byteOffset: number costUSD: number - editCount: number + perMessage: Record + sawEdit: boolean sawGitCommit: boolean lastTurnAt: string | null updatedAt: string @@ -21,11 +31,18 @@ export type GuardCache = { stopNotified: boolean } -export const GUARD_CACHE_VERSION = 1 +// v2: per-message-id replace fold (perMessage map, sawEdit boolean) and the +// guard/sessions/ cache location. v1 caches are ignored and cold-reparse once. +export const GUARD_CACHE_VERSION = 2 -// Raw command text is needed (not the base-binary names parseApiCall's -// bashCommands reduces to), so the Stop check reads call.toolSequence. -const GIT_COMMIT = /\bgit\b[\s\S]*?\bcommit\b/ +// `commit` must be the git subcommand: `git`, optionally flag tokens (long +// flags, or a short flag with an optional separate value like `-c k=v`), then +// `commit` as the next word, anchored at a command boundary: string/line start +// (multi-line Bash calls separate commands with newlines) or ; & |. The gaps +// inside the command never cross a newline, so "git diff\ncommit msg" is two +// commands, not a commit. "git log --grep commit" and "git diff && echo +// commit" don't match either. +const GIT_COMMIT = /(?:^|[;&|])[^\S\n]*git(?:[^\S\n]+(?:--\S+|-\w+(?:[^\S\n]+\S+)?))*[^\S\n]+commit(?![-\w])/m export function emptyCache(sessionId: string): GuardCache { return { @@ -33,7 +50,8 @@ export function emptyCache(sessionId: string): GuardCache { sessionId, byteOffset: 0, costUSD: 0, - editCount: 0, + perMessage: {}, + sawEdit: false, sawGitCommit: false, lastTurnAt: null, updatedAt: '', @@ -51,7 +69,11 @@ export async function readCache(sessionId: string, base?: string): Promise - if (parsed.version !== GUARD_CACHE_VERSION || typeof parsed.byteOffset !== 'number') { + if ( + parsed.version !== GUARD_CACHE_VERSION + || typeof parsed.byteOffset !== 'number' + || !parsed.perMessage || typeof parsed.perMessage !== 'object' + ) { return emptyCache(sessionId) } return { ...emptyCache(sessionId), ...parsed, sessionId } @@ -61,7 +83,7 @@ export async function readCache(sessionId: string, base?: string): Promise { - await mkdir(guardDir(base), { recursive: true }) + await mkdir(sessionsDir(base), { recursive: true }) await writeFile(sessionCachePath(cache.sessionId, base), JSON.stringify(cache), 'utf-8') } @@ -86,7 +108,7 @@ export async function computeSessionUsage( // offset into different bytes. const cache = size < prev.byteOffset ? { ...emptyCache(prev.sessionId), softWarned: prev.softWarned, stopNotified: prev.stopNotified } - : { ...prev } + : { ...prev, perMessage: { ...prev.perMessage } } const resumedFrom = cache.byteOffset const tracker = { lastCompleteLineOffset: resumedFrom } @@ -99,9 +121,17 @@ export async function computeSessionUsage( if (!entry) continue const call = parseApiCall(entry) if (!call) continue - cache.costUSD += call.costUSD + // Last-wins per message id, matching the shipped dedupeStreamingMessageIds. + // Lines without an id (rare, and never streamed in copies) just add. + const msgId = (entry.message as { id?: string } | undefined)?.id + if (msgId) { + cache.costUSD += call.costUSD - (cache.perMessage[msgId] ?? 0) + cache.perMessage[msgId] = call.costUSD + } else { + cache.costUSD += call.costUSD + } for (const tc of call.toolSequence?.flat() ?? []) { - if (EDIT_TOOLS.has(tc.tool)) cache.editCount++ + if (!cache.sawEdit && EDIT_TOOLS.has(tc.tool)) cache.sawEdit = true if (!cache.sawGitCommit && tc.command && GIT_COMMIT.test(tc.command)) cache.sawGitCommit = true } if (call.timestamp) cache.lastTurnAt = call.timestamp @@ -122,6 +152,6 @@ export async function isAllowed(sessionId: string, base?: string): Promise { - await mkdir(guardDir(base), { recursive: true }) + await mkdir(sessionsDir(base), { recursive: true }) await writeFile(allowPath(sessionId, base), '', 'utf-8') } diff --git a/src/parser.ts b/src/parser.ts index a6e2f80..4d407b9 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -1148,7 +1148,7 @@ export function parseApiCall(entry: JournalEntry): ParsedApiCall | null { }) } -function dedupeStreamingMessageIds(entries: JournalEntry[]): JournalEntry[] { +export function dedupeStreamingMessageIds(entries: JournalEntry[]): JournalEntry[] { const firstIdxById = new Map() const lastIdxById = new Map() for (let i = 0; i < entries.length; i++) { diff --git a/tests/act-journal.test.ts b/tests/act-journal.test.ts index 659cc5a..5732899 100644 --- a/tests/act-journal.test.ts +++ b/tests/act-journal.test.ts @@ -3,7 +3,8 @@ import { mkdtemp, mkdir, readFile, rm, utimes, writeFile } from 'node:fs/promise import { existsSync } from 'node:fs' import { spawnSync } from 'node:child_process' import { tmpdir } from 'node:os' -import { join } from 'node:path' +import { dirname, join } from 'node:path' +import { fileURLToPath } from 'node:url' import { runAction } from '../src/act/apply.js' import { journalPath, readRecords } from '../src/act/journal.js' import { sha256 } from '../src/act/backup.js' @@ -182,8 +183,11 @@ describe('act list --json (CLI)', () => { kind: 'mcp-remove', description: 'newer', changes: [{ op: 'edit', path: p2, content: 'b2' }], }, actionsDir) + // Anchor the spawn to this checkout so running vitest from another cwd + // cannot execute a different checkout's cli.ts. + const repoRoot = join(dirname(fileURLToPath(import.meta.url)), '..') const res = spawnSync(process.execPath, ['--import', 'tsx', 'src/cli.ts', 'act', 'list', '--json'], { - cwd: process.cwd(), + cwd: repoRoot, env: { ...process.env, HOME: home, USERPROFILE: home, HOMEPATH: home, HOMEDRIVE: '' }, encoding: 'utf-8', }) diff --git a/tests/guard-hooks.test.ts b/tests/guard-hooks.test.ts index 04f8b16..ef5d6b2 100644 --- a/tests/guard-hooks.test.ts +++ b/tests/guard-hooks.test.ts @@ -1,7 +1,8 @@ import { afterAll, describe, expect, it } from 'vitest' -import { appendFile, mkdtemp, rm, stat, writeFile } from 'node:fs/promises' +import { appendFile, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { dedupeStreamingMessageIds, parseApiCall, parseJsonlLine } from '../src/parser.js' import { computeSessionUsage, emptyCache, readCache, writeAllow, writeCache } from '../src/guard/usage.js' import { runGuardHook, runGuardStatusline } from '../src/guard/hooks.js' import { writeGuardConfig, DEFAULT_GUARD_CONFIG } from '../src/guard/store.js' @@ -46,6 +47,18 @@ function hookInput(path: string): string { return JSON.stringify({ session_id: SID, transcript_path: path, hook_event_name: 'PreToolUse', tool_name: 'Bash' }) } +// The shipped pipeline's cost for a transcript: parse every line, dedupe +// streaming copies last-wins exactly like parseSessionFile does, sum call +// costs. The guard's incremental fold must match this to the cent. +async function shippedDedupedCost(path: string): Promise { + const entries = (await readFile(path, 'utf-8')) + .split('\n') + .filter(l => l.trim()) + .map(l => parseJsonlLine(l)) + .filter((e): e is NonNullable> => e !== null) + return dedupeStreamingMessageIds(entries).reduce((s, e) => s + (parseApiCall(e)?.costUSD ?? 0), 0) +} + describe('incremental session cache', () => { it('parses only the appended tail and totals match a cold parse', async () => { const base = await tmp() @@ -89,6 +102,88 @@ describe('incremental session cache', () => { }) }) +describe('streaming duplicates (per-message-id replace)', () => { + it('counts a message written 3x with identical usage once, matching the shipped dedup', async () => { + const dup = assistantLine('msg-dup', { inTok: 500_000, outTok: 100_000 }) + const path = await transcript([dup, dup, dup, assistantLine('msg-other')]) + const singlePath = await transcript([dup, assistantLine('msg-other')]) + + const { cache } = await computeSessionUsage(emptyCache(SID), path) + expect(cache.costUSD).toBeCloseTo(await shippedDedupedCost(path), 10) + const single = await computeSessionUsage(emptyCache(SID), singlePath) + expect(cache.costUSD).toBeCloseTo(single.cache.costUSD, 10) + }) + + it('keeps the last copy when streamed usage grows', async () => { + const copies = [10_000, 50_000, 120_000].map(outTok => assistantLine('msg-grow', { outTok })) + const path = await transcript(copies) + const lastOnly = await transcript([copies[2]!]) + + const { cache } = await computeSessionUsage(emptyCache(SID), path) + expect(cache.costUSD).toBeCloseTo(await shippedDedupedCost(path), 10) + const last = await computeSessionUsage(emptyCache(SID), lastOnly) + expect(cache.costUSD).toBeCloseTo(last.cache.costUSD, 10) + }) + + it('replaces across incremental invocations when a later copy arrives', async () => { + const base = await tmp() + const path = await transcript([assistantLine('msg-x', { outTok: 20_000 })]) + const r1 = await computeSessionUsage(emptyCache(SID), path) + await writeCache(r1.cache, base) + + await appendFile(path, assistantLine('msg-x', { outTok: 90_000 }) + assistantLine('msg-y'), 'utf-8') + const r2 = await computeSessionUsage(await readCache(SID, base), path) + expect(r2.resumedFrom).toBe(r1.cache.byteOffset) + expect(r2.cache.costUSD).toBeCloseTo(await shippedDedupedCost(path), 10) + }) +}) + +describe('trailing complete line without newline', () => { + it('does not double count the line once it completes (A,B,C then D)', async () => { + const base = await tmp() + const a = assistantLine('msg-a') + const b = assistantLine('msg-b') + const c = assistantLine('msg-c') + const d = assistantLine('msg-d') + const dir = await tmp() + const path = join(dir, 'session.jsonl') + await writeFile(path, a + b + c.slice(0, -1), 'utf-8') // C complete but unterminated + + const r1 = await computeSessionUsage(emptyCache(SID), path) + await writeCache(r1.cache, base) + // C was folded, but the offset stops after B's newline. + expect(r1.cache.costUSD).toBeCloseTo(await shippedDedupedCost(path), 10) + expect(r1.cache.byteOffset).toBe((a + b).length) + + await appendFile(path, '\n' + d, 'utf-8') + const r2 = await computeSessionUsage(await readCache(SID, base), path) + // C is re-read as a replace, not a second add: totals equal a cold parse. + const cold = await computeSessionUsage(emptyCache(SID), path) + expect(r2.cache.costUSD).toBeCloseTo(cold.cache.costUSD, 10) + expect(r2.cache.costUSD).toBeCloseTo(await shippedDedupedCost(path), 10) + }) +}) + +describe('git commit detection', () => { + it('requires commit to be the git subcommand', async () => { + const cases: [string, boolean][] = [ + ['git commit -m x', true], + ['git -c user.name=x commit', true], + ['npm test && git commit -am done', true], + ['git add src\ngit commit -m "step"', true], // newline-separated multi-command Bash call + ['git log --grep commit', false], + ['git diff && echo commit', false], + ['echo git then commit', false], + ['git diff\ncommit notes to self', false], // commit on the next line is not a git subcommand + ] + for (const [command, expected] of cases) { + const path = await transcript([assistantLine('msg-1', { tools: [{ name: 'Bash', input: { command } }] })]) + const { cache } = await computeSessionUsage(emptyCache(SID), path) + expect(cache.sawGitCommit, command).toBe(expected) + } + }) +}) + describe('budget hook (PreToolUse)', () => { async function costOf(path: string): Promise { return (await computeSessionUsage(emptyCache(SID), path)).cache.costUSD diff --git a/tests/guard-install.test.ts b/tests/guard-install.test.ts index 65b67a5..8c1e877 100644 --- a/tests/guard-install.test.ts +++ b/tests/guard-install.test.ts @@ -90,6 +90,20 @@ describe('guard install', () => { await runAction(built.plan!, a2) expect((await readJson(s2)).statusLine.command).toBe('my-statusline.sh') }) + + it('refuses to apply a plan when the settings file changed after it was built', async () => { + const { settings, actionsDir } = await makeRoot() + await writeFile(settings, canonical({ permissions: { allow: [] } })) + const built = buildInstall(settings) + expect(built.plan).not.toBeNull() + + const concurrent = canonical({ permissions: { allow: ['Bash(ls:*)'] } }) + await writeFile(settings, concurrent) + + await expect(runAction(built.plan!, actionsDir)).rejects.toThrow(/changed since the plan was built/) + expect(await readFile(settings, 'utf-8')).toBe(concurrent) // the concurrent edit survives + expect(await readRecords(actionsDir)).toEqual([]) // nothing journaled + }) }) describe('guard uninstall', () => {