fix(cli): dedup claude usage and reset backfill state schema

Author: Jannchie

packages/cli/src/adapters/claude-code.ts

@@ -53,6 +53,11 @@ async function parseClaudeCodeSessionFile(
   const lines = text.split('\n').filter(Boolean)
   const projectContext = await claudeProjectContextFromLines(filePath, lines, options)
   const pendingTools = new Map<string, ClaudePendingTool>()
+  // Claude Code occasionally writes the same assistant message to the
+  // jsonl more than once (streaming flushes, retries, replays). ccusage
+  // dedups by `${messageId}:${requestId}` — without it codetime double-
+  // counts tokens 2-6×. Track which usage rows we've already emitted.
+  const seenUsageKeys = new Set<string>()
   let sessionId = sessionIdFromFilePath(filePath, 'claude')
   let cwd: string | undefined
   let project: string | undefined = projectContext.project
@@ -221,16 +226,37 @@ async function parseClaudeCodeSessionFile(
     }
 
     model = stringField(message, 'model') || model
+    // Skip the entire assistant entry when (messageId, requestId) was
+    // already processed — applies to both the usage metrics and any
+    // tool_use items so we don't double-emit tool.started either.
+    // When messageId is absent we cannot dedup safely — emit and accept
+    // the risk (matches ccusage's createUniqueHash returning null).
+    const messageId = stringField(message, 'id')
+    const requestId = stringField(raw, 'requestId')
+    const usageKey = messageId ? `${messageId}:${requestId}` : null
+    if (usageKey != null && seenUsageKeys.has(usageKey)) {
+      continue
+    }
+    if (usageKey != null) {
+      seenUsageKeys.add(usageKey)
+    }
     const usage = claudeUsageFromMessage(message)
     if (usage) {
+      // Anthropic surfaces fast inference via `usage.speed === 'fast'`.
+      // Append `-fast` so the model name lines up with OpenRouter's
+      // separate `anthropic/claude-opus-4.7-fast` pricing entry (~6×
+      // standard). Tag only the model.usage event to keep downstream
+      // tool events on the base model name.
+      const speed = stringField(objectField(message, 'usage'), 'speed')
+      const usageModel = speed === 'fast' && model ? `${model}-fast` : model
       push(baseClaudeEvent({
         ts,
         type: 'model.usage',
         sessionId,
         turnId: state.currentTurnId,
         cwd,
         project,
-        model,
+        model: usageModel,
         confidence: 'partial',
         metrics: usage,
       }), lineNumber, topType, 'usage')

packages/cli/src/cli.ts

@@ -7,6 +7,7 @@ import type {
 } from '@codetime/shared'
 import type { BackfillSourceDefinition } from './lib/backfill.js'
 import type { BackfillImportCounts, BackfillIncrementalState, BackfillSourceFile, ParsedArgs, RunContext, SyncLocalLock, SyncLocalTriggerState, WritableLike } from './lib/types.js'
+import { BACKFILL_STATE_SCHEMA_VERSION } from './lib/types.js'
 import { spawn } from 'node:child_process'
 import { mkdir, rm, stat, writeFile } from 'node:fs/promises'
 import os from 'node:os'
@@ -973,26 +974,33 @@ function syncLocalTriggerLockPath(home: string): string {
 
 async function readBackfillIncrementalState(home: string, ctx?: RunContext): Promise<BackfillIncrementalState> {
   // Corrupt JSON now surfaces from readJsonIfExists; a missing file
-  // resolves to null. Anything else (wrong shape, future schema
+  // resolves to null. Anything else (wrong shape, mismatched schema
   // version, manual edits that dropped `sources`) lands here and would
   // previously vanish silently — log via debug so the user can see
   // when watermarks were dropped.
+  //
+  // When the on-disk schema version doesn't match the CLI's current
+  // BACKFILL_STATE_SCHEMA_VERSION we deliberately drop every watermark.
+  // The next sync-local-runner then re-parses every jsonl from scratch
+  // and the server upserts via `replace: true`, so a CLI upgrade that
+  // changed parser semantics (e.g. v2's dedup fix) silently rewrites
+  // historical rollups without the user knowing.
   const statePath = backfillIncrementalStatePath(home)
   const state = await readJsonIfExists(statePath)
   if (state === null) {
-    return { version: 1, sources: {} }
+    return { version: BACKFILL_STATE_SCHEMA_VERSION, sources: {} }
   }
   if (!isPlainObject(state) || !isPlainObject(state.sources)) {
     if (ctx) {
       debug(ctx, `backfill-state malformed at ${statePath}; ignoring watermarks\n`)
     }
-    return { version: 1, sources: {} }
+    return { version: BACKFILL_STATE_SCHEMA_VERSION, sources: {} }
   }
-  if (state.version !== undefined && state.version !== 1) {
+  if (state.version !== BACKFILL_STATE_SCHEMA_VERSION) {
     if (ctx) {
-      debug(ctx, `backfill-state version ${String(state.version)} at ${statePath} is not supported; ignoring watermarks\n`)
+      debug(ctx, `backfill-state version ${String(state.version)} at ${statePath} differs from current v${BACKFILL_STATE_SCHEMA_VERSION}; dropping watermarks so the next sync re-imports under the new parser\n`)
     }
-    return { version: 1, sources: {} }
+    return { version: BACKFILL_STATE_SCHEMA_VERSION, sources: {} }
   }
 
   const sources: BackfillIncrementalState['sources'] = {}
@@ -1003,7 +1011,7 @@ async function readBackfillIncrementalState(home: string, ctx?: RunContext): Pro
     }
   }
 
-  return { version: 1, sources }
+  return { version: BACKFILL_STATE_SCHEMA_VERSION, sources }
 }
 
 async function updateBackfillIncrementalState(

packages/cli/src/lib/types.ts

@@ -27,8 +27,17 @@ export interface BackfillSourceFile {
   modifiedAt: string
 }
 
+// Bump `BACKFILL_STATE_SCHEMA_VERSION` whenever the offline parsers
+// change in a way that invalidates already-uploaded rollups (e.g. the
+// Claude assistant-message dedup added in v2). The CLI compares the
+// constant against the on-disk schema; on a mismatch it drops every
+// watermark so the next sync silently re-parses all jsonl from scratch
+// and upserts the deduped rollups (`replace: true` is already set).
+// Users get the fix transparently the next time their agent runs.
+export const BACKFILL_STATE_SCHEMA_VERSION = 2
+
 export interface BackfillIncrementalState {
-  version: 1
+  version: typeof BACKFILL_STATE_SCHEMA_VERSION
   sources: Partial<Record<
     import('@codetime/shared').BackfillSourceId,
     { watermarkTs: string }