Merge pull request #1008 from getagentseal/perf/parallel-cold-parse

perf(parser): parallelize the cold Claude parse across worker threads, hardware-adaptively

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 ### Changed
+- **A large cold Claude parse now runs across worker threads.** Reading, decoding and line-parsing a session JSONL is per-file work that never touches anything shared, so it moves onto `worker_threads`; each worker ships its parsed turns back as a JSON string and the parent installs them in the exact order the serial loop would. Everything with cross-file state — the streaming-message dedup, canonical project paths, spawn links, PR correlation, progress saves — stays on the main thread, and a file whose message ids were already claimed by an earlier file (or whose worker failed) is simply re-parsed in-process, so the session cache and every payload are identical either way. On a 6 GB corpus a cold `status` drops from 27.5s to 14.8s with peak RSS up 2.27 GB → 2.52 GB. Threads only engage for a genuinely large cold parse: never with fewer than 200 pending whole-file re-parses, under 200 MB behind them, 2 or fewer cores, or under 4 GB of available memory — so warm and incremental runs are untouched and spawn nothing. Otherwise the count is `min(cores - 1, min(0.25 × available, 2 GB) / 256 MB, pendingFiles / 50)`, where available is `process.availableMemory()` (cgroup-aware in containers) rather than free memory, which on macOS reports free pages and would switch the feature on and off between runs. `CODEBURN_PARSE_WORKERS=0` forces the serial parse and `CODEBURN_PARSE_WORKERS=N` forces N (capped at the core count), both bypassing every gate; `CODEBURN_VERBOSE=1` prints the resolved count and why. Only Claude sessions are parallelized so far.
 - **A warm launch rewrites only the month that changed, and a ranged query reads only the months it can report on.** Per-provider shards still meant one appended session republished that provider's entire history — 95 MB for Claude on a 6 GB corpus. Each provider's shard is now split again by the UTC month of the cached session's FIRST turn, a bucket that never moves as a session grows, so an append rewrites one month. Every shard records the newest month it holds, which lets `--period today/week` skip the shards that cannot contribute a turn to the range; the skipped months stay on disk untouched across the save, and providers whose cache is the only surviving record (durable) or whose parse fingerprint moved are always read in full. Remaining shards are read concurrently. Existing v8 and v7 caches are re-laid-out losslessly on first load and the old layout removed once the new one is published: nothing re-parses.
 - **A warm launch rewrites only the provider that changed.** The session cache was a single blob, so any provider appending a few KB republished the whole thing — 147 MB of stringify + fsync on a 6 GB corpus, ~18% of a warm run. It is now a version-suffixed directory holding one shard per provider plus a small envelope, written per provider and published by a single envelope rename. An existing v7 cache is re-laid-out losslessly on first load and the old file removed once the new layout is on disk: nothing re-parses. One unreadable shard now costs that provider a re-parse instead of discarding every provider's history, and partial saves during a cold parse are triggered every 2000 files rather than every 5 seconds, so a slow cold parse no longer rewrites the growing cache on a wall clock.
 - **An appended Codex rollout parses only its tail.** Rollout files are append-only and the active ones run to hundreds of MB, but the Codex result cache keyed on mtime + size alone, so any growth re-read the file from byte 0. Each entry now records a restart point at the last task boundary — byte offset plus the state the single-pass decode carries across it — and a grown file with the same inode resumes there, producing output identical to a full re-parse. An entry without a usable restart point simply re-parses in full once and gains one.

app/scripts/stage-cli.mjs

@@ -11,6 +11,7 @@
 //   build/cli/package.json          (root package.json: {version}, type:module)
 //   build/cli/dist/cli.js           (Node-version-guard launcher → ./main.js)
 //   build/cli/dist/main.js          (the bundle)
+//   build/cli/dist/parse-worker.js  (the parse worker thread's own entry)
 //   build/cli/node_modules/         (production dependency closure)
 //
 // The production closure is copied out of the already-installed root
@@ -29,7 +30,7 @@ const dist = join(root, 'dist')
 const rootModules = join(root, 'node_modules')
 const stage = join(appDir, 'build', 'cli')
 
-for (const f of ['cli.js', 'main.js']) {
+for (const f of ['cli.js', 'main.js', 'parse-worker.js']) {
   if (!existsSync(join(dist, f))) {
     throw new Error(`stage-cli: ${join(dist, f)} is missing — build the root CLI first`)
   }
@@ -41,6 +42,10 @@ mkdirSync(join(stage, 'dist'), { recursive: true })
 copyFileSync(join(root, 'package.json'), join(stage, 'package.json'))
 copyFileSync(join(dist, 'cli.js'), join(stage, 'dist', 'cli.js'))
 copyFileSync(join(dist, 'main.js'), join(stage, 'dist', 'main.js'))
+// The cold-parse worker pool resolves this as a sibling of the bundle it runs
+// from, so it has to be staged alongside main.js or a packaged app silently
+// loses every parse thread.
+copyFileSync(join(dist, 'parse-worker.js'), join(stage, 'dist', 'parse-worker.js'))
 
 // Desktop-app launch shim (the app spawns this, not cli.js). The packaged app
 // runs the CLI with Electron's own binary as Node (ELECTRON_RUN_AS_NODE=1).

docs/architecture.md

@@ -69,6 +69,45 @@ output formatter (Ink TUI, JSON, or menubar-json)
 
 `src/parser.ts` is the central aggregator. Public exports: `parseAllSessions`, `filterProjectsByName`, `extractMcpInventory`. It owns the dedup `Set` (`seenKeys`) that is passed into every provider parser so a turn that surfaces in two providers (Claude logs vs. Cursor mirror, for instance) is counted once.
 
+### Parallel Cold Parse
+
+A cold Claude parse spends most of its time on work that is per-file and pure:
+reading a session JSONL, decoding it, and turning each line into a journal entry.
+`src/parse-workers.ts` moves that onto `worker_threads` when the pending workload
+is big enough to pay for them. Each worker runs `parseClaudeFileFull` against an
+empty dedup set and ships the result back as a JSON string; the parent installs
+results in the same order the serial loop would, and everything with cross-file
+state (the `seenMsgIds` dedup, canonical project paths, spawn links, PR
+correlation) stays on the main thread. A file whose message ids were already
+claimed by an earlier file, or whose worker failed, is re-parsed in-process — so
+the output is identical to the serial path either way. Only whole-file re-parses
+go off-thread; the append/incremental path is untouched. Workers are created at
+the start of a qualifying parse and terminated when it ends, so the resident
+`serve` child never accumulates threads.
+
+The pool is off by default for anything that is not a large cold parse:
+
+| Gate | Serial when |
+|---|---|
+| Pending files | fewer than 200 whole-file re-parses |
+| Pending bytes | under 200 MB behind those files |
+| Cores | `availableParallelism() <= 2` |
+| Memory | under 4 GB available |
+
+Otherwise the worker count is
+`min(cores - 1, min(0.25 * available, 2 GB) / 256 MB, pendingFiles / 50)`.
+
+"Available" is `process.availableMemory()`, falling back to `os.totalmem()`. It is
+deliberately not `os.freemem()`: on macOS that counts free pages rather than
+available memory and reads as a few hundred MB on an idle 128 GB machine, so a
+gate built on it switches the feature on and off between runs. On Linux outside a
+memory-limited cgroup, `availableMemory()` reports free memory and can still
+under-report on a busy host — which fails safe, to fewer threads or none.
+
+`CODEBURN_PARSE_WORKERS` overrides the decision and skips every gate above:
+`0` forces the serial parse, `N` forces N workers (capped at the core count).
+`CODEBURN_VERBOSE=1` prints the resolved worker count and the reason for it.
+
 ### Cache Layers
 
 Three caches under `~/.cache/codeburn/` (override with `CODEBURN_CACHE_DIR`):

package.json

@@ -8,7 +8,8 @@
     "codeburn": "dist/cli.js"
   },
   "files": [
-    "dist"
+    "dist",
+    "!dist/parse-worker.js.map"
   ],
   "scripts": {
     "bundle-litellm": "node scripts/bundle-litellm.mjs",

src/models.ts

@@ -2,8 +2,8 @@ import { readFile, writeFile, mkdir } from 'fs/promises'
 import { join } from 'path'
 
 import { getCodeburnCacheDir } from './cache-dir.js'
-import snapshotData from './data/litellm-snapshot.json'
-import fallbackData from './data/pricing-fallback.json'
+import snapshotData from './data/litellm-snapshot.json' with { type: 'json' }
+import fallbackData from './data/pricing-fallback.json' with { type: 'json' }
 import { fetchWithTimeout } from './fetch-utils.js'
 
 export type ModelCosts = {
@@ -995,3 +995,33 @@ export function getShortModelName(model: string): string {
   }
   return canonical
 }
+
+// Pricing is process-global state assembled at CLI startup from the cached
+// LiteLLM snapshot plus user config. A parse worker thread starts with none of
+// it, and re-running loadPricing() there would mean N more disk reads (or, on a
+// cold pricing cache, N network fetches). Ship the resolved state across
+// instead, so every thread prices a call exactly as the main thread would.
+export type PricingSnapshot = {
+  pricing: Map<string, ModelCosts>
+  aliases: Record<string, string>
+  priceOverrides: Record<string, PriceOverrideRates>
+  localModelSavings: Record<string, string>
+}
+
+export function snapshotPricingState(): PricingSnapshot {
+  return {
+    pricing: pricingCache,
+    aliases: userAliases,
+    priceOverrides: userPriceOverridesConfig,
+    localModelSavings: userLocalModelSavings,
+  }
+}
+
+export function restorePricingState(snapshot: PricingSnapshot): void {
+  pricingCache = snapshot.pricing
+  sortedPricingKeys = null
+  lowercasePricingIndex = null
+  setModelAliases(snapshot.aliases)
+  setPriceOverrides(snapshot.priceOverrides)
+  setLocalModelSavings(snapshot.localModelSavings)
+}

src/parse-worker.ts

@@ -0,0 +1,25 @@
+import { parentPort, workerData } from 'worker_threads'
+import { restorePricingState, type PricingSnapshot } from './models.js'
+import { parseClaudeFileFull } from './parser.js'
+
+const port = parentPort
+if (!port) throw new Error('parse-worker must be started as a worker thread')
+
+restorePricingState((workerData as { pricing: PricingSnapshot }).pricing)
+
+// The parsed turns go back as a JSON string rather than as a live object graph:
+// structured-cloning a whole corpus of turns costs more than the parallel parse
+// saves, while a string is a single copy the parent re-parses at memcpy speed.
+// `msgIds` is every streaming message id this file claimed; the parent uses it to
+// prove no earlier file already owned one before installing the result.
+port.on('message', (msg: { filePath: string }) => {
+  void (async () => {
+    try {
+      const seenMsgIds = new Set<string>()
+      const parsed = await parseClaudeFileFull(msg.filePath, seenMsgIds)
+      port.postMessage({ json: parsed === null ? null : JSON.stringify({ ...parsed, msgIds: [...seenMsgIds] }) })
+    } catch (err) {
+      port.postMessage({ error: err instanceof Error ? err.message : String(err) })
+    }
+  })()
+})

src/parse-workers.ts

@@ -0,0 +1,220 @@
+import { availableParallelism, totalmem } from 'os'
+import { Worker } from 'worker_threads'
+import { snapshotPricingState } from './models.js'
+import type { ClaudeFileParse } from './parser.js'
+
+// Each worker holds one file's entry list plus its serialized result. 256 MB is
+// the observed high-water mark for the largest real session files; going over the
+// budget is what turns a parallel parse into a swapping one.
+const PER_WORKER_RSS_BYTES = 256 * 1024 * 1024
+const MEMORY_BUDGET_CAP_BYTES = 2 * 1024 * 1024 * 1024
+const MIN_AVAILABLE_BYTES = 4 * 1024 * 1024 * 1024
+const MIN_FILES_PER_WORKER = 50
+// Below these, a parse is warm/incremental and the thread startup + result
+// transfer costs more than the parallelism buys.
+const MIN_PENDING_FILES = 200
+const MIN_PENDING_BYTES = 200 * 1024 * 1024
+
+export type ParseWorkerDecision = { workers: number; reason: string }
+
+export type SystemCapacity = { cores: number; availableBytes: number }
+
+// `process.availableMemory()` respects a container's cgroup / rlimit, which is the
+// case this gate exists for; where it is absent it falls back to total RAM. Free
+// memory is deliberately NOT used: on macOS `os.freemem()` counts free pages, not
+// available memory, and reads as a few hundred MB on an idle 128 GB machine — a
+// gate built on it turns the feature on and off at random.
+function currentSystemCapacity(): SystemCapacity {
+  return {
+    cores: availableParallelism(),
+    availableBytes: typeof process.availableMemory === 'function' ? process.availableMemory() : totalmem(),
+  }
+}
+
+/// Decide how many parse worker threads a pending workload earns. Returning 0
+/// means "parse serially" — the only behaviour before this existed, and still
+/// the behaviour for every warm run, every small corpus, and every low-spec box.
+export function decideParseWorkers(
+  pending: { files: number; bytes: number },
+  sys: SystemCapacity = currentSystemCapacity(),
+  env: NodeJS.ProcessEnv = process.env,
+): ParseWorkerDecision {
+  // Every reason carries the full decision input, so a support log line explains
+  // itself without a second run.
+  const inputs = `${sys.cores} cores, ${Math.round(sys.availableBytes / 1e9 * 10) / 10} GB available, ${pending.files} pending files / ${Math.round(pending.bytes / 1e6)} MB`
+
+  const override = env['CODEBURN_PARSE_WORKERS']
+  if (override !== undefined && override !== '') {
+    const n = Number(override)
+    if (!Number.isFinite(n) || n < 0) return { workers: 0, reason: `invalid CODEBURN_PARSE_WORKERS=${override}` }
+    const capped = Math.min(Math.floor(n), sys.cores)
+    return { workers: capped, reason: `${capped === 0 ? 'forced serial' : 'forced'} by CODEBURN_PARSE_WORKERS=${override}; ${inputs}` }
+  }
+
+  // Workload gates first, so a warm run's log line says "warm", not whatever the
+  // machine happened to look like at that moment.
+  if (pending.files < MIN_PENDING_FILES) return { workers: 0, reason: `below ${MIN_PENDING_FILES} pending files; ${inputs}` }
+  if (pending.bytes < MIN_PENDING_BYTES) return { workers: 0, reason: `below ${Math.round(MIN_PENDING_BYTES / 1e6)} MB pending; ${inputs}` }
+  if (sys.cores <= 2) return { workers: 0, reason: `too few cores; ${inputs}` }
+  if (sys.availableBytes < MIN_AVAILABLE_BYTES) return { workers: 0, reason: `below ${Math.round(MIN_AVAILABLE_BYTES / 1e9)} GB available memory; ${inputs}` }
+
+  const memoryBudget = Math.min(0.25 * sys.availableBytes, MEMORY_BUDGET_CAP_BYTES)
+  const workers = Math.min(
+    sys.cores - 1,
+    Math.floor(memoryBudget / PER_WORKER_RSS_BYTES),
+    Math.floor(pending.files / MIN_FILES_PER_WORKER),
+  )
+  return { workers, reason: inputs }
+}
+
+// In dist the entry is the bundled sibling of this module and a worker can load
+// it directly. Running from source (tsx, vitest) the entry is TypeScript, and a
+// worker thread inherits none of the parent's loader hooks — so register tsx's
+// inside the thread before importing. tsx is a devDependency, which is exactly
+// the only situation where the entry can be a .ts file at all.
+function workerBootstrap(entryUrl: string): { source: string | URL; eval: boolean } {
+  if (!entryUrl.endsWith('.ts')) return { source: new URL(entryUrl), eval: false }
+  return {
+    eval: true,
+    // Chained, not awaited: the eval scope is CommonJS, and a top-level await of
+    // the entry re-enters it as a require(esm) cycle.
+    source: `
+process.noDeprecation = true
+import('tsx/esm/api').then(tsx => { tsx.register(); return import(${JSON.stringify(entryUrl)}) })
+`,
+  }
+}
+
+function workerEntryUrl(): string {
+  const ext = import.meta.url.endsWith('.ts') ? '.ts' : '.js'
+  return new URL(`./parse-worker${ext}`, import.meta.url).href
+}
+
+export type ClaudeWorkerResult =
+  | { ok: true; parsed: (ClaudeFileParse & { msgIds: string[] }) | null }
+  | { ok: false; error: string }
+
+type Task = { filePath: string; resolve: (r: ClaudeWorkerResult) => void }
+
+type WorkerMessage = { json?: string | null; error?: string }
+
+export class ParseWorkerPool {
+  private readonly workers: Worker[] = []
+  private readonly idle: Worker[] = []
+  private readonly inflight = new Map<Worker, Task>()
+  private readonly queue: Task[] = []
+  private closed = false
+
+  constructor(size: number) {
+    const boot = workerBootstrap(workerEntryUrl())
+    const workerData = { pricing: snapshotPricingState() }
+    try {
+      for (let i = 0; i < size; i++) {
+        const worker = new Worker(boot.source, { eval: boot.eval, workerData })
+        worker.on('message', (msg: WorkerMessage) => this.settle(worker, msg))
+        worker.on('error', (err: Error) => this.settle(worker, { error: err.message }, true))
+        worker.on('exit', () => this.drop(worker))
+        this.workers.push(worker)
+        this.idle.push(worker)
+      }
+    } catch (err) {
+      for (const w of this.workers) void w.terminate()
+      this.workers.length = 0
+      this.idle.length = 0
+      throw err
+    }
+  }
+
+  get size(): number {
+    return this.workers.length
+  }
+
+  /// Parse one file off-thread. Never rejects: a worker-side failure (or a dead
+  /// pool) comes back as `ok: false` so the caller can fall back to an in-process
+  /// parse and never lose a file to a crashed thread.
+  submit(filePath: string): Promise<ClaudeWorkerResult> {
+    return new Promise<ClaudeWorkerResult>((resolve) => {
+      if (this.closed || this.workers.length === 0) {
+        resolve({ ok: false, error: 'parse worker pool unavailable' })
+        return
+      }
+      this.queue.push({ filePath, resolve })
+      this.pump()
+    })
+  }
+
+  async close(): Promise<void> {
+    this.closed = true
+    const pending = [...this.queue]
+    this.queue.length = 0
+    for (const task of pending) task.resolve({ ok: false, error: 'parse worker pool closed' })
+    await Promise.all(this.workers.map(w => w.terminate()))
+    this.workers.length = 0
+    this.idle.length = 0
+    this.inflight.clear()
+  }
+
+  private pump(): void {
+    while (this.queue.length > 0 && this.idle.length > 0) {
+      const worker = this.idle.pop()!
+      const task = this.queue.shift()!
+      this.inflight.set(worker, task)
+      worker.postMessage({ filePath: task.filePath })
+    }
+  }
+
+  private settle(worker: Worker, msg: WorkerMessage, fatal = false): void {
+    const task = this.inflight.get(worker)
+    this.inflight.delete(worker)
+    if (task) {
+      if (msg.error !== undefined) task.resolve({ ok: false, error: msg.error })
+      else task.resolve({ ok: true, parsed: msg.json == null ? null : JSON.parse(msg.json) })
+    }
+    if (fatal) return
+    if (!this.closed) {
+      this.idle.push(worker)
+      this.pump()
+    }
+  }
+
+  // A thread that died takes its queue slot with it; the remaining files are
+  // handed back for a serial parse rather than being lost.
+  private drop(worker: Worker): void {
+    const i = this.workers.indexOf(worker)
+    if (i >= 0) this.workers.splice(i, 1)
+    const j = this.idle.indexOf(worker)
+    if (j >= 0) this.idle.splice(j, 1)
+    const task = this.inflight.get(worker)
+    if (task) {
+      this.inflight.delete(worker)
+      task.resolve({ ok: false, error: 'parse worker exited' })
+    }
+    if (this.workers.length === 0) {
+      const pending = [...this.queue]
+      this.queue.length = 0
+      for (const t of pending) t.resolve({ ok: false, error: 'all parse workers exited' })
+    }
+  }
+}
+
+/// Yield results for `filePaths` in the SAME order they were given, no matter
+/// which worker finishes first. Keeps exactly `pool.size` files in flight, so at
+/// most that many parsed results are buffered while the caller installs one.
+export async function* parseFilesInOrder(
+  pool: ParseWorkerPool,
+  filePaths: readonly string[],
+): AsyncGenerator<ClaudeWorkerResult, void, void> {
+  const inflight: Array<Promise<ClaudeWorkerResult>> = []
+  let next = 0
+  const fill = (): void => {
+    while (inflight.length < Math.max(1, pool.size) && next < filePaths.length) {
+      inflight.push(pool.submit(filePaths[next++]!))
+    }
+  }
+  fill()
+  for (let i = 0; i < filePaths.length; i++) {
+    const result = await inflight.shift()!
+    fill()
+    yield result
+  }
+}

src/parser.ts

@@ -31,6 +31,7 @@ import {
   saveCache,
 } from './session-cache.js'
 import { acquireCacheRefreshLock, type RefreshLockHandle } from './cache-refresh-lock.js'
+import { decideParseWorkers, parseFilesInOrder, ParseWorkerPool } from './parse-workers.js'
 import { dateKey } from './day-aggregator.js'
 import type { ParsedProviderCall, SessionSource } from './providers/types.js'
 import type {
@@ -2012,171 +2013,228 @@ async function scanProjectDirs(
   const progressTotal = changedFiles.length
   let filesDone = 0
   emitScanProgress({ kind: 'tick', provider: 'claude', done: 0, total: progressTotal })
-  for (const { filePath, info, append } of changedFiles) {
-    // Marked here, not after the re-parse: an unreadable file `continue`s out
-    // below, and the deletion would otherwise live only in memory.
-    delete section.files[filePath]
-    markCacheDirty(diskCache, 'claude', filePath)
 
+  // Only whole-file re-parses go off-thread. Appends are the warm path: they read
+  // a few KB past the cached offset, so a thread hop would cost more than it saves.
+  const fullReparsePaths = changedFiles.filter(f => !f.append).map(f => f.filePath)
+  const pendingBytes = changedFiles.reduce((n, f) => f.append ? n : n + f.info.fp.sizeBytes, 0)
+  const decision = decideParseWorkers({ files: fullReparsePaths.length, bytes: pendingBytes })
+  if (process.env['CODEBURN_VERBOSE'] === '1') {
+    process.stderr.write(`codeburn: claude parse workers=${decision.workers} (${decision.reason})\n`)
+  }
+  // A pool that cannot even start (worker entry missing from an odd packaging,
+  // thread limit reached) must degrade to the serial parse, not fail the run.
+  let pool: ParseWorkerPool | null = null
+  if (decision.workers > 0) {
     try {
-      if (append) {
-        // Append-only growth: parse ONLY the bytes past the cached resume offset
-        // and merge with the cached turns, rather than re-reading the file from 0.
-        // On a studio machine where live agents constantly append to session
-        // JSONL, this is the dominant warm-run cost. The merged result is
-        // byte-for-byte identical to a full re-parse (see mergeBoundaryCalls).
-        const tracker = { lastCompleteLineOffset: append.readFromOffset }
-        const toolResultMeta = new Map<string, ToolResultMeta>()
-        const sessionMeta = emptySessionMeta()
-        const newEntries = await parseClaudeEntries(filePath, tracker, append.readFromOffset, { toolResultMeta, sessionMeta })
-        const cached = append.cached
-
-        // Straddle guard: a streamed assistant message id that first appeared in
-        // the committed prefix can be restated inside the appended region
-        // (image-heavy turns stream one id across several records over seconds).
-        // The appended region is grouped before this file's cached keys join
-        // seenMsgIds, so the restated id would count twice; suppressing it
-        // instead would freeze the stale first emission. Neither matches a full
-        // re-parse, so on any id overlap the shortcut is abandoned and the file
-        // re-parses from byte 0 (rare: ~0.3% of real files).
-        const cachedIds = new Set(cached.turns.flatMap(t => t.calls.map(c => c.deduplicationKey)))
-        const straddles = newEntries !== null && newEntries.some(e => {
-          const id = getMessageId(e)
-          return id !== null && cachedIds.has(id)
-        })
-        if (!straddles) {
-          const newTurns = newEntries
-            ? parsedTurnsToCachedTurns(groupIntoTurns(dedupeStreamingMessageIds(newEntries), seenMsgIds, toolResultMeta))
-            : []
-
-          const mergedTurns: CachedTurn[] = cached.turns.map(t => ({ ...t, calls: [...t.calls] }))
-          if (newTurns.length > 0) {
-            let startIdx = 0
-            // A first new turn with no leading user message is a continuation of
-            // the last cached turn — merge its calls in (a full re-parse would put
-            // them in that same turn), then append the remaining new turns.
-            if (!newTurns[0]!.userMessage.trim() && mergedTurns.length > 0) {
-              const last = mergedTurns[mergedTurns.length - 1]!
-              last.calls = mergeBoundaryCalls(last.calls, newTurns[0]!.calls)
-              // A PR referenced in the appended continuation belongs to this same
-              // turn: union its refs in so the shortcut matches a full re-parse.
-              const refs = Array.from(new Set([...(last.prRefs ?? []), ...(newTurns[0]!.prRefs ?? [])])).sort()
-              if (refs.length > 0) last.prRefs = refs
-              // A subagent spawned in the appended continuation belongs to this
-              // same turn: union its spawn ids in for the same reason.
-              const spawnIds = Array.from(new Set([...(last.spawnToolUseIds ?? []), ...(newTurns[0]!.spawnToolUseIds ?? [])]))
-              if (spawnIds.length > 0) last.spawnToolUseIds = spawnIds
-              startIdx = 1
-            }
-            for (let i = startIdx; i < newTurns.length; i++) mergedTurns.push(newTurns[i]!)
-          }
-
-          // The cached region's dedup keys were not added to seenMsgIds (only
-          // unchanged files pre-seed it), so add them now — a full re-parse would
-          // have, and later files dedup cross-file against them.
-          for (const t of cached.turns) for (const c of t.calls) seenMsgIds.add(c.deduplicationKey)
-
-          // First-cwd wins, and the first cwd lives in the cached region whenever
-          // one was resolved there; only re-derive if the cached region had none.
-          let canonicalCwd = cached.canonicalCwd
-          let canonicalProjectName = cached.canonicalProjectName
-          let workingDirectory = cached.workingDirectory
-          if (canonicalCwd === undefined && newEntries) {
-            const cwd = extractCanonicalCwd(newEntries)
-            workingDirectory = workingDirectory ?? cwd
-            const canonical = (cwd && !isCoworkSession(cwd, filePath)) ? await resolveCanonicalProjectPath(cwd) : undefined
-            canonicalCwd = canonical?.path
-            canonicalProjectName = canonical?.isWorktree ? projectNameFromPath(canonical.path, info.dirName) : undefined
-          }
-
-          // Inventory is a sorted set union; cached (older entries) ∪ new = full.
-          const mcpInventory = newEntries
-            ? Array.from(new Set([...cached.mcpInventory, ...extractMcpInventory(newEntries)])).sort()
-            : cached.mcpInventory
-
-          // Session meta merges across the append boundary: title is last-wins
-          // (prefer the newly-parsed tail), PR links union, isSidechain is sticky.
-          // parentSessionId is sticky (cached-first, it is the earliest region);
-          // agentSpawnLinks union (cached-first, first-seen spawn id per agent wins).
-          const mergedTitle = sessionMeta.title ?? cached.title
-          const mergedPrLinks = Array.from(new Set([...(cached.prLinks ?? []), ...sessionMeta.prLinks]))
-          const mergedSidechain = cached.isSidechain === true || sessionMeta.isSidechain
-          const mergedParentSessionId = cached.parentSessionId ?? sessionMeta.parentSessionId
-          const mergedSpawnLinks = { ...sessionMeta.agentSpawnLinks, ...cached.agentSpawnLinks }
-          const mergedAmbiguousIds = Array.from(new Set([...(cached.ambiguousSpawnAgentIds ?? []), ...sessionMeta.ambiguousSpawnAgentIds]))
-
-          section.files[filePath] = {
-            fingerprint: info.fp,
-            lastCompleteLineOffset: tracker.lastCompleteLineOffset,
-            canonicalCwd,
-            ...(workingDirectory ? { workingDirectory } : {}),
-            canonicalProjectName,
-            mcpInventory,
-            turns: mergedTurns,
-            agentType: cached.agentType,
-            ...(mergedTitle ? { title: mergedTitle } : {}),
-            ...(mergedPrLinks.length > 0 ? { prLinks: mergedPrLinks } : {}),
-            ...(mergedSidechain ? { isSidechain: true } : {}),
-            ...(mergedParentSessionId ? { parentSessionId: mergedParentSessionId } : {}),
-            ...(Object.keys(mergedSpawnLinks).length > 0 ? { agentSpawnLinks: mergedSpawnLinks } : {}),
-            ...(mergedAmbiguousIds.length > 0 ? { ambiguousSpawnAgentIds: mergedAmbiguousIds } : {}),
-          }
-          markCacheDirty(diskCache, 'claude', filePath)
-          filesDone++
-          await parseProgress.tick(filesDone)
-          if (filesDone % 50 === 0 || filesDone === progressTotal) {
-            emitScanProgress({ kind: 'tick', provider: 'claude', done: filesDone, total: progressTotal })
-          }
-          if (onFileParsed) await onFileParsed()
-          continue
-        }
-        // Straddled: fall through to the full re-parse below.
-      }
-
-      const tracker = { lastCompleteLineOffset: 0 }
-      const toolResultMeta = new Map<string, ToolResultMeta>()
-      const sessionMeta = emptySessionMeta()
-      const entries = await parseClaudeEntries(filePath, tracker, undefined, { toolResultMeta, sessionMeta })
-      if (!entries) { filesDone++; await parseProgress.tick(filesDone); continue }
-
-      const turns = groupIntoTurns(dedupeStreamingMessageIds(entries), seenMsgIds, toolResultMeta)
-      const cwd = extractCanonicalCwd(entries)
-      const canonical = (cwd && !isCoworkSession(cwd, filePath)) ? await resolveCanonicalProjectPath(cwd) : undefined
-      section.files[filePath] = {
-        fingerprint: info.fp,
-        lastCompleteLineOffset: tracker.lastCompleteLineOffset,
-        canonicalCwd: canonical?.path,
-        ...(cwd ? { workingDirectory: cwd } : {}),
-        canonicalProjectName: canonical?.isWorktree ? projectNameFromPath(canonical.path, info.dirName) : undefined,
-        mcpInventory: extractMcpInventory(entries),
-        turns: parsedTurnsToCachedTurns(turns),
-        agentType: await readAgentType(filePath),
-        ...(sessionMeta.title ? { title: sessionMeta.title } : {}),
-        ...(sessionMeta.prLinks.length > 0 ? { prLinks: sessionMeta.prLinks } : {}),
-        ...(sessionMeta.isSidechain ? { isSidechain: true } : {}),
-        ...(sessionMeta.parentSessionId ? { parentSessionId: sessionMeta.parentSessionId } : {}),
-        ...(Object.keys(sessionMeta.agentSpawnLinks).length > 0 ? { agentSpawnLinks: sessionMeta.agentSpawnLinks } : {}),
-        ...(sessionMeta.ambiguousSpawnAgentIds.length > 0 ? { ambiguousSpawnAgentIds: sessionMeta.ambiguousSpawnAgentIds } : {}),
-      }
-      markCacheDirty(diskCache, 'claude', filePath)
+      pool = new ParseWorkerPool(decision.workers)
     } catch (err) {
-      // A single malformed Claude session file must not abort the whole run — that
-      // would empty the daily-cache backfill and wipe the trend/history (issue #441,
-      // same isolation the provider path already has). Record a failure marker keyed
-      // by the current fingerprint so it isn't re-read and re-thrown every run; it
-      // re-parses only if the file changes.
-      section.files[filePath] = { fingerprint: info.fp, mcpInventory: [], turns: [], failed: true }
+      process.stderr.write(`codeburn: parse workers unavailable, parsing serially (${err instanceof Error ? err.message : String(err)})\n`)
+    }
+  }
+  const offThread = pool ? parseFilesInOrder(pool, fullReparsePaths) : null
+  // Files whose worker result had to be thrown away because an earlier file had
+  // already claimed one of its message ids. Expected to stay near zero; a large
+  // count means the corpus is full of resumed sessions and the pool is doing
+  // double work.
+  let workerDiscards = 0
+
+  const installClaudeFile = async (filePath: string, info: FileInfo, parsed: ClaudeFileParse): Promise<void> => {
+    const cwd = parsed.workingDirectory
+    const canonical = (cwd && !isCoworkSession(cwd, filePath)) ? await resolveCanonicalProjectPath(cwd) : undefined
+    section.files[filePath] = {
+      fingerprint: info.fp,
+      lastCompleteLineOffset: parsed.lastCompleteLineOffset,
+      canonicalCwd: canonical?.path,
+      ...(cwd ? { workingDirectory: cwd } : {}),
+      canonicalProjectName: canonical?.isWorktree ? projectNameFromPath(canonical.path, info.dirName) : undefined,
+      mcpInventory: parsed.mcpInventory,
+      turns: parsed.turns,
+      agentType: parsed.agentType,
+      ...(parsed.title ? { title: parsed.title } : {}),
+      ...(parsed.prLinks?.length ? { prLinks: parsed.prLinks } : {}),
+      ...(parsed.isSidechain ? { isSidechain: true } : {}),
+      ...(parsed.parentSessionId ? { parentSessionId: parsed.parentSessionId } : {}),
+      ...(Object.keys(parsed.agentSpawnLinks ?? {}).length > 0 ? { agentSpawnLinks: parsed.agentSpawnLinks } : {}),
+      ...(parsed.ambiguousSpawnAgentIds?.length ? { ambiguousSpawnAgentIds: parsed.ambiguousSpawnAgentIds } : {}),
+    }
+    markCacheDirty(diskCache, 'claude', filePath)
+  }
+
+  try {
+    for (const { filePath, info, append } of changedFiles) {
+      // Marked here, not after the re-parse: an unreadable file `continue`s out
+      // below, and the deletion would otherwise live only in memory.
+      delete section.files[filePath]
       markCacheDirty(diskCache, 'claude', filePath)
-      warnProviderParseFailure('claude', filePath, err)
+
+      try {
+        if (append) {
+          // Append-only growth: parse ONLY the bytes past the cached resume offset
+          // and merge with the cached turns, rather than re-reading the file from 0.
+          // On a studio machine where live agents constantly append to session
+          // JSONL, this is the dominant warm-run cost. The merged result is
+          // byte-for-byte identical to a full re-parse (see mergeBoundaryCalls).
+          const tracker = { lastCompleteLineOffset: append.readFromOffset }
+          const toolResultMeta = new Map<string, ToolResultMeta>()
+          const sessionMeta = emptySessionMeta()
+          const newEntries = await parseClaudeEntries(filePath, tracker, append.readFromOffset, { toolResultMeta, sessionMeta })
+          const cached = append.cached
+
+          // Straddle guard: a streamed assistant message id that first appeared in
+          // the committed prefix can be restated inside the appended region
+          // (image-heavy turns stream one id across several records over seconds).
+          // The appended region is grouped before this file's cached keys join
+          // seenMsgIds, so the restated id would count twice; suppressing it
+          // instead would freeze the stale first emission. Neither matches a full
+          // re-parse, so on any id overlap the shortcut is abandoned and the file
+          // re-parses from byte 0 (rare: ~0.3% of real files).
+          const cachedIds = new Set(cached.turns.flatMap(t => t.calls.map(c => c.deduplicationKey)))
+          const straddles = newEntries !== null && newEntries.some(e => {
+            const id = getMessageId(e)
+            return id !== null && cachedIds.has(id)
+          })
+          if (!straddles) {
+            const newTurns = newEntries
+              ? parsedTurnsToCachedTurns(groupIntoTurns(dedupeStreamingMessageIds(newEntries), seenMsgIds, toolResultMeta))
+              : []
+
+            const mergedTurns: CachedTurn[] = cached.turns.map(t => ({ ...t, calls: [...t.calls] }))
+            if (newTurns.length > 0) {
+              let startIdx = 0
+              // A first new turn with no leading user message is a continuation of
+              // the last cached turn — merge its calls in (a full re-parse would put
+              // them in that same turn), then append the remaining new turns.
+              if (!newTurns[0]!.userMessage.trim() && mergedTurns.length > 0) {
+                const last = mergedTurns[mergedTurns.length - 1]!
+                last.calls = mergeBoundaryCalls(last.calls, newTurns[0]!.calls)
+                // A PR referenced in the appended continuation belongs to this same
+                // turn: union its refs in so the shortcut matches a full re-parse.
+                const refs = Array.from(new Set([...(last.prRefs ?? []), ...(newTurns[0]!.prRefs ?? [])])).sort()
+                if (refs.length > 0) last.prRefs = refs
+                // A subagent spawned in the appended continuation belongs to this
+                // same turn: union its spawn ids in for the same reason.
+                const spawnIds = Array.from(new Set([...(last.spawnToolUseIds ?? []), ...(newTurns[0]!.spawnToolUseIds ?? [])]))
+                if (spawnIds.length > 0) last.spawnToolUseIds = spawnIds
+                startIdx = 1
+              }
+              for (let i = startIdx; i < newTurns.length; i++) mergedTurns.push(newTurns[i]!)
+            }
+
+            // The cached region's dedup keys were not added to seenMsgIds (only
+            // unchanged files pre-seed it), so add them now — a full re-parse would
+            // have, and later files dedup cross-file against them.
+            for (const t of cached.turns) for (const c of t.calls) seenMsgIds.add(c.deduplicationKey)
+
+            // First-cwd wins, and the first cwd lives in the cached region whenever
+            // one was resolved there; only re-derive if the cached region had none.
+            let canonicalCwd = cached.canonicalCwd
+            let canonicalProjectName = cached.canonicalProjectName
+            let workingDirectory = cached.workingDirectory
+            if (canonicalCwd === undefined && newEntries) {
+              const cwd = extractCanonicalCwd(newEntries)
+              workingDirectory = workingDirectory ?? cwd
+              const canonical = (cwd && !isCoworkSession(cwd, filePath)) ? await resolveCanonicalProjectPath(cwd) : undefined
+              canonicalCwd = canonical?.path
+              canonicalProjectName = canonical?.isWorktree ? projectNameFromPath(canonical.path, info.dirName) : undefined
+            }
+
+            // Inventory is a sorted set union; cached (older entries) ∪ new = full.
+            const mcpInventory = newEntries
+              ? Array.from(new Set([...cached.mcpInventory, ...extractMcpInventory(newEntries)])).sort()
+              : cached.mcpInventory
+
+            // Session meta merges across the append boundary: title is last-wins
+            // (prefer the newly-parsed tail), PR links union, isSidechain is sticky.
+            // parentSessionId is sticky (cached-first, it is the earliest region);
+            // agentSpawnLinks union (cached-first, first-seen spawn id per agent wins).
+            const mergedTitle = sessionMeta.title ?? cached.title
+            const mergedPrLinks = Array.from(new Set([...(cached.prLinks ?? []), ...sessionMeta.prLinks]))
+            const mergedSidechain = cached.isSidechain === true || sessionMeta.isSidechain
+            const mergedParentSessionId = cached.parentSessionId ?? sessionMeta.parentSessionId
+            const mergedSpawnLinks = { ...sessionMeta.agentSpawnLinks, ...cached.agentSpawnLinks }
+            const mergedAmbiguousIds = Array.from(new Set([...(cached.ambiguousSpawnAgentIds ?? []), ...sessionMeta.ambiguousSpawnAgentIds]))
+
+            section.files[filePath] = {
+              fingerprint: info.fp,
+              lastCompleteLineOffset: tracker.lastCompleteLineOffset,
+              canonicalCwd,
+              ...(workingDirectory ? { workingDirectory } : {}),
+              canonicalProjectName,
+              mcpInventory,
+              turns: mergedTurns,
+              agentType: cached.agentType,
+              ...(mergedTitle ? { title: mergedTitle } : {}),
+              ...(mergedPrLinks.length > 0 ? { prLinks: mergedPrLinks } : {}),
+              ...(mergedSidechain ? { isSidechain: true } : {}),
+              ...(mergedParentSessionId ? { parentSessionId: mergedParentSessionId } : {}),
+              ...(Object.keys(mergedSpawnLinks).length > 0 ? { agentSpawnLinks: mergedSpawnLinks } : {}),
+              ...(mergedAmbiguousIds.length > 0 ? { ambiguousSpawnAgentIds: mergedAmbiguousIds } : {}),
+            }
+            markCacheDirty(diskCache, 'claude', filePath)
+            filesDone++
+            await parseProgress.tick(filesDone)
+            if (filesDone % 50 === 0 || filesDone === progressTotal) {
+              emitScanProgress({ kind: 'tick', provider: 'claude', done: filesDone, total: progressTotal })
+            }
+            if (onFileParsed) await onFileParsed()
+            continue
+          }
+          // Straddled: fall through to the full re-parse below.
+        }
+
+        // Off-thread results arrive in this order (parseFilesInOrder), so the Nth
+        // full re-parse here is the Nth yielded result. A worker parses against an
+        // EMPTY dedup set, so an EMPTY id intersection is the proof that a serial
+        // parse would have dropped nothing either — that, and only that, makes the
+        // result installable. On any overlap the WHOLE file is discarded and
+        // re-parsed in-process. Never patch the overlapping turns out of a worker
+        // result instead: a drop is not local to its own turn, because
+        // parsedTurnsToCachedTurns delta-encodes gitBranch across turns, so removing
+        // one turn changes whether a LATER turn carries a gitBranch key.
+        let parsed: ClaudeFileParse | null | undefined
+        if (offThread && !append) {
+          const result = (await offThread.next()).value
+          if (result?.ok && result.parsed) {
+            if (result.parsed.msgIds.some(id => seenMsgIds.has(id))) {
+              workerDiscards++
+              parsed = undefined
+            } else {
+              for (const id of result.parsed.msgIds) seenMsgIds.add(id)
+              parsed = result.parsed
+            }
+          } else if (result?.ok) {
+            parsed = null
+          }
+        }
+        if (parsed === undefined) parsed = await parseClaudeFileFull(filePath, seenMsgIds)
+        if (!parsed) { filesDone++; await parseProgress.tick(filesDone); continue }
+
+        await installClaudeFile(filePath, info, parsed)
+      } catch (err) {
+        // A single malformed Claude session file must not abort the whole run — that
+        // would empty the daily-cache backfill and wipe the trend/history (issue #441,
+        // same isolation the provider path already has). Record a failure marker keyed
+        // by the current fingerprint so it isn't re-read and re-thrown every run; it
+        // re-parses only if the file changes.
+        section.files[filePath] = { fingerprint: info.fp, mcpInventory: [], turns: [], failed: true }
+        markCacheDirty(diskCache, 'claude', filePath)
+        warnProviderParseFailure('claude', filePath, err)
+      }
+      filesDone++
+      await parseProgress.tick(filesDone)
+      // Machine-readable tick for the app splash (throttled to ~every 50 files so
+      // a large cold run doesn't flood stderr), plus a partial-progress save.
+      if (filesDone % 50 === 0 || filesDone === progressTotal) {
+        emitScanProgress({ kind: 'tick', provider: 'claude', done: filesDone, total: progressTotal })
+      }
+      if (onFileParsed) await onFileParsed()
     }
-    filesDone++
-    await parseProgress.tick(filesDone)
-    // Machine-readable tick for the app splash (throttled to ~every 50 files so
-    // a large cold run doesn't flood stderr), plus a partial-progress save.
-    if (filesDone % 50 === 0 || filesDone === progressTotal) {
-      emitScanProgress({ kind: 'tick', provider: 'claude', done: filesDone, total: progressTotal })
-    }
-    if (onFileParsed) await onFileParsed()
+  } finally {
+    await pool?.close()
+  }
+  if (pool && process.env['CODEBURN_VERBOSE'] === '1') {
+    process.stderr.write(`codeburn: claude parse workers done, ${workerDiscards}/${fullReparsePaths.length} results re-parsed in-process on id overlap\n`)
   }
   parseProgress.finish()
 
@@ -2676,6 +2734,54 @@ async function parseClaudeEntries(
   return entries
 }
 
+// Everything a cold Claude re-parse does for ONE file: read + decode + line-parse
+// the JSONL, group it into turns, shape it for the cache. Depends on nothing
+// process-wide except `seenMsgIds`, so a worker thread can run it against a fresh
+// empty set and the parent can install the result verbatim once it has confirmed
+// none of those ids were already claimed by an earlier file. Canonical-path
+// resolution deliberately stays with the caller: it walks the filesystem behind a
+// process-global memo.
+export type ClaudeFileParse = {
+  lastCompleteLineOffset: number
+  workingDirectory?: string
+  mcpInventory: string[]
+  turns: CachedTurn[]
+  agentType?: string
+  title?: string
+  prLinks?: string[]
+  isSidechain?: boolean
+  parentSessionId?: string
+  agentSpawnLinks?: Record<string, string>
+  ambiguousSpawnAgentIds?: string[]
+}
+
+export async function parseClaudeFileFull(
+  filePath: string,
+  seenMsgIds: Set<string>,
+): Promise<ClaudeFileParse | null> {
+  const tracker = { lastCompleteLineOffset: 0 }
+  const toolResultMeta = new Map<string, ToolResultMeta>()
+  const sessionMeta = emptySessionMeta()
+  const entries = await parseClaudeEntries(filePath, tracker, undefined, { toolResultMeta, sessionMeta })
+  if (!entries) return null
+
+  const turns = groupIntoTurns(dedupeStreamingMessageIds(entries), seenMsgIds, toolResultMeta)
+  const cwd = extractCanonicalCwd(entries)
+  return {
+    lastCompleteLineOffset: tracker.lastCompleteLineOffset,
+    ...(cwd ? { workingDirectory: cwd } : {}),
+    mcpInventory: extractMcpInventory(entries),
+    turns: parsedTurnsToCachedTurns(turns),
+    agentType: await readAgentType(filePath),
+    ...(sessionMeta.title ? { title: sessionMeta.title } : {}),
+    ...(sessionMeta.prLinks.length > 0 ? { prLinks: sessionMeta.prLinks } : {}),
+    ...(sessionMeta.isSidechain ? { isSidechain: true } : {}),
+    ...(sessionMeta.parentSessionId ? { parentSessionId: sessionMeta.parentSessionId } : {}),
+    ...(Object.keys(sessionMeta.agentSpawnLinks).length > 0 ? { agentSpawnLinks: sessionMeta.agentSpawnLinks } : {}),
+    ...(sessionMeta.ambiguousSpawnAgentIds.length > 0 ? { ambiguousSpawnAgentIds: sessionMeta.ambiguousSpawnAgentIds } : {}),
+  }
+}
+
 function getOrCreateProviderSection(cache: SessionCache, provider: string): ProviderSection {
   const envFp = computeEnvFingerprint(provider)
   const existing = cache.providers[provider]

tests/parse-workers.test.ts

@@ -0,0 +1,297 @@
+import { spawnSync } from 'node:child_process'
+import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { createHash } from 'node:crypto'
+
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { decideParseWorkers, ParseWorkerPool, parseFilesInOrder } from '../src/parse-workers.js'
+import { clearSessionCache, parseAllSessions, parseClaudeFileFull } from '../src/parser.js'
+
+// Two full cold CLI parses of a multi-hundred-file corpus, plus in-process parses
+// that spawn real threads.
+vi.setConfig({ testTimeout: 60_000 })
+
+const BIG_SYSTEM = { cores: 16, availableBytes: 32 * 1024 ** 3 }
+const BIG_PENDING = { files: 5000, bytes: 6 * 1024 ** 3 }
+const NO_ENV = {} as NodeJS.ProcessEnv
+
+describe('decideParseWorkers', () => {
+  it('scales with cores, memory budget and pending file count', () => {
+    // 15 (cores-1) vs 8 (2 GB budget / 256 MB) vs 100 (5000/50) -> memory cap wins
+    expect(decideParseWorkers(BIG_PENDING, BIG_SYSTEM, NO_ENV).workers).toBe(8)
+    // Fewer cores than the memory budget allows -> cores-1 wins
+    expect(decideParseWorkers(BIG_PENDING, { cores: 6, availableBytes: 32 * 1024 ** 3 }, NO_ENV).workers).toBe(5)
+    // 8 GB reaches the same cap as 32 GB: a quarter of it is the 2 GB budget
+    expect(decideParseWorkers(BIG_PENDING, { cores: 16, availableBytes: 8 * 1024 ** 3 }, NO_ENV).workers).toBe(8)
+    // Under that, the quarter-of-available budget is the binding constraint
+    expect(decideParseWorkers(BIG_PENDING, { cores: 16, availableBytes: 6 * 1024 ** 3 }, NO_ENV).workers).toBe(6)
+    // The smallest machine that clears every gate still only earns 2 threads
+    expect(decideParseWorkers({ files: 200, bytes: 300 * 1024 ** 2 }, { cores: 3, availableBytes: 4 * 1024 ** 3 }, NO_ENV).workers).toBe(2)
+    // Few enough files that MIN_FILES_PER_WORKER is the binding constraint
+    expect(decideParseWorkers({ files: 300, bytes: 6 * 1024 ** 3 }, BIG_SYSTEM, NO_ENV).workers).toBe(6)
+  })
+
+  it('stays serial on low-spec machines and on warm/small parses', () => {
+    expect(decideParseWorkers(BIG_PENDING, { cores: 2, availableBytes: 32 * 1024 ** 3 }, NO_ENV).workers).toBe(0)
+    // A 4 GB box: availableMemory() always reads a little under the nominal size
+    expect(decideParseWorkers(BIG_PENDING, { cores: 16, availableBytes: 3.9 * 1024 ** 3 }, NO_ENV).workers).toBe(0)
+    // Warm/incremental: a handful of appended files
+    expect(decideParseWorkers({ files: 12, bytes: 6 * 1024 ** 3 }, BIG_SYSTEM, NO_ENV).workers).toBe(0)
+    // Many files but almost no bytes behind them
+    expect(decideParseWorkers({ files: 5000, bytes: 10 * 1024 ** 2 }, BIG_SYSTEM, NO_ENV).workers).toBe(0)
+  })
+
+  it('honours CODEBURN_PARSE_WORKERS, which also bypasses the auto gates', () => {
+    expect(decideParseWorkers(BIG_PENDING, BIG_SYSTEM, { CODEBURN_PARSE_WORKERS: '0' }).workers).toBe(0)
+    expect(decideParseWorkers(BIG_PENDING, BIG_SYSTEM, { CODEBURN_PARSE_WORKERS: '4' }).workers).toBe(4)
+    // Capped by the core count
+    expect(decideParseWorkers(BIG_PENDING, { cores: 4, availableBytes: 32 * 1024 ** 3 }, { CODEBURN_PARSE_WORKERS: '32' }).workers).toBe(4)
+    // A tiny fixture corpus still gets threads when forced — that is what makes
+    // the determinism test below able to exercise them at all.
+    expect(decideParseWorkers({ files: 3, bytes: 1000 }, BIG_SYSTEM, { CODEBURN_PARSE_WORKERS: '3' }).workers).toBe(3)
+    expect(decideParseWorkers(BIG_PENDING, BIG_SYSTEM, { CODEBURN_PARSE_WORKERS: 'nonsense' }).workers).toBe(0)
+  })
+
+  it('reports the decision inputs in every reason, gate or not', () => {
+    for (const d of [
+      decideParseWorkers(BIG_PENDING, BIG_SYSTEM, NO_ENV),
+      decideParseWorkers({ files: 12, bytes: 1000 }, BIG_SYSTEM, NO_ENV),
+      decideParseWorkers(BIG_PENDING, BIG_SYSTEM, { CODEBURN_PARSE_WORKERS: '2' }),
+    ]) {
+      expect(d.reason).toContain('16 cores')
+      expect(d.reason).toContain('GB available')
+      expect(d.reason).toContain('pending files')
+    }
+  })
+})
+
+type Turn = { id: string; t: number }
+
+function sessionLines(project: string, session: string, turns: Turn[]): string {
+  const lines: string[] = []
+  for (const { id, t } of turns) {
+    const ts = new Date(Date.UTC(2026, 4, 4 + (t % 5), 9, t % 60, 0)).toISOString()
+    const gitBranch = t % 3 === 0 ? 'main' : 'feature'
+    lines.push(JSON.stringify({
+      type: 'user', sessionId: session, timestamp: ts, cwd: `/tmp/proj${project}`, gitBranch,
+      message: { role: 'user', content: `task ${t} in ${project}` },
+    }))
+    lines.push(JSON.stringify({
+      type: 'assistant', sessionId: session, timestamp: ts, cwd: `/tmp/proj${project}`, gitBranch,
+      message: {
+        id, type: 'message', role: 'assistant', model: 'claude-sonnet-4-5',
+        content: [
+          { type: 'text', text: 'x'.repeat(200) },
+          { type: 'tool_use', id: `tu-${id}`, name: 'Edit', input: { file_path: '/tmp/x', old_string: 'a', new_string: 'b' } },
+        ],
+        usage: { input_tokens: 400 + t, output_tokens: 40 + t, cache_read_input_tokens: 9 },
+      },
+    }))
+  }
+  return lines.join('\n') + '\n'
+}
+
+const range = (n: number, from = 0): number[] => Array.from({ length: n }, (_, i) => i + from)
+
+async function writeCorpus(claudeDir: string, projects: number, filesPerProject: number): Promise<string[]> {
+  const written: string[] = []
+  for (let p = 0; p < projects; p++) {
+    const dir = join(claudeDir, 'projects', `-tmp-proj${p}`)
+    await mkdir(dir, { recursive: true })
+    for (let f = 0; f < filesPerProject; f++) {
+      const session = `${p}${f}`.padStart(8, '0') + '-aaaa-bbbb-cccc-000000000000'
+      const path = join(dir, `${session}.jsonl`)
+      await writeFile(path, sessionLines(String(p), session, range(12).map(t => ({ id: `msg-${p}-${session}-${t}`, t }))))
+      written.push(path)
+    }
+  }
+  return written
+}
+
+/// A resumed Claude session: the new transcript restates the original's assistant
+/// messages verbatim (same message ids) before adding its own. Cross-file dedup
+/// means whichever file is installed FIRST keeps those turns and the other loses
+/// them, so this fixture is only stable if worker results are installed in the
+/// serial order — and it is the only fixture that drives the discard/re-parse path,
+/// since a worker parses against an empty dedup set and cannot see the overlap.
+async function writeResumedPair(claudeDir: string, tag: string, originalName: string, resumedName: string): Promise<void> {
+  const dir = join(claudeDir, 'projects', `-tmp-${tag}`)
+  await mkdir(dir, { recursive: true })
+  const shared = range(6).map(t => ({ id: `${tag}-m${t}`, t }))
+  await writeFile(join(dir, `${originalName}.jsonl`), sessionLines(tag, originalName, shared))
+  await writeFile(
+    join(dir, `${resumedName}.jsonl`),
+    sessionLines(tag, resumedName, [...shared, ...range(4, 6).map(t => ({ id: `${tag}-n${t}`, t }))]),
+  )
+}
+
+/// Cache shard file names carry a random nonce, so compare bodies keyed by
+/// `<provider>.<month>` instead of by file name.
+async function shardBodies(cacheDir: string): Promise<Record<string, string>> {
+  const dir = join(cacheDir, 'session-cache.v9')
+  const out: Record<string, string> = {}
+  for (const name of (await readdir(dir).catch(() => []))) {
+    if (name === 'envelope.json' || !name.endsWith('.json')) continue
+    const key = name.split('.').slice(0, 2).join('.')
+    out[key] = createHash('sha256').update(await readFile(join(dir, name))).digest('hex')
+  }
+  return out
+}
+
+function runCli(args: string[], home: string, extraEnv: Record<string, string>) {
+  return spawnSync(process.execPath, ['--import', 'tsx', 'src/cli.ts', ...args], {
+    cwd: process.cwd(),
+    env: {
+      ...process.env,
+      CLAUDE_CONFIG_DIR: join(home, '.claude'),
+      CODEBURN_CACHE_DIR: join(home, '.cache', 'codeburn'),
+      HOME: home,
+      TZ: 'UTC',
+      ...extraEnv,
+    },
+    encoding: 'utf-8',
+    timeout: 60_000,
+  })
+}
+
+function stripVolatile(payload: unknown): unknown {
+  if (Array.isArray(payload)) return payload.map(stripVolatile)
+  if (payload && typeof payload === 'object') {
+    return Object.fromEntries(
+      Object.entries(payload as Record<string, unknown>)
+        .filter(([k]) => !k.toLowerCase().startsWith('generated'))
+        .map(([k, v]) => [k, stripVolatile(v)]),
+    )
+  }
+  if (typeof payload === 'number') return Math.round(payload * 1e9) / 1e9
+  return payload
+}
+
+describe('parallel cold parse', () => {
+  let home: string
+
+  beforeEach(async () => {
+    home = await mkdtemp(join(tmpdir(), 'cb-cold-'))
+  })
+
+  afterEach(async () => {
+    await rm(home, { recursive: true, force: true })
+  })
+
+  /// Both runs read the SAME corpus, so the absolute paths embedded in the cache
+  /// shards match and the bodies can be compared byte for byte.
+  async function bothWays(extraParallelEnv: Record<string, string> = {}) {
+    const serialCache = join(home, 'cache-serial')
+    const parallelCache = join(home, 'cache-parallel')
+    const args = ['status', '--format', 'menubar-json']
+    const serial = runCli(args, home, { CODEBURN_PARSE_WORKERS: '0', CODEBURN_CACHE_DIR: serialCache })
+    const parallel = runCli(args, home, { CODEBURN_PARSE_WORKERS: '3', CODEBURN_CACHE_DIR: parallelCache, ...extraParallelEnv })
+
+    expect(serial.status, serial.stderr).toBe(0)
+    expect(parallel.status, parallel.stderr).toBe(0)
+    expect(stripVolatile(JSON.parse(parallel.stdout))).toEqual(stripVolatile(JSON.parse(serial.stdout)))
+
+    const serialShards = await shardBodies(serialCache)
+    expect(Object.keys(serialShards).length).toBeGreaterThan(0)
+    expect(await shardBodies(parallelCache)).toEqual(serialShards)
+    return parallel
+  }
+
+  // The whole point of the feature: threads may only ever be a speed change.
+  it('produces an identical payload and byte-identical cache shards with and without workers', async () => {
+    await writeCorpus(join(home, '.claude'), 4, 12)
+    await bothWays()
+  })
+
+  // Resumed sessions in both filename orders: the restating file sorts after the
+  // original in one project and before it in the other, so install order decides
+  // which file keeps the shared turns either way. Out-of-order installation, or
+  // any attempt to patch overlapping turns out of a worker result instead of
+  // discarding the whole file, changes the answer.
+  it("matches the serial parse when files restate each other's message ids", async () => {
+    const claude = join(home, '.claude')
+    await writeResumedPair(claude, 'fwd', '00000000-aaaa-bbbb-cccc-000000000000', '99999999-aaaa-bbbb-cccc-000000000000')
+    await writeResumedPair(claude, 'rev', '99999999-dddd-bbbb-cccc-000000000000', '00000000-dddd-bbbb-cccc-000000000000')
+
+    const parallel = await bothWays({ CODEBURN_VERBOSE: '1' })
+
+    // Pin that the discard path actually ran rather than passing by luck.
+    const overlaps = [...parallel.stderr.matchAll(/(\d+)\/\d+ results re-parsed in-process on id overlap/g)]
+      .reduce((n, m) => n + Number(m[1]), 0)
+    expect(overlaps).toBeGreaterThan(0)
+  })
+})
+
+describe('ParseWorkerPool', () => {
+  let home: string
+  let files: string[]
+
+  beforeEach(async () => {
+    clearSessionCache()
+    home = await mkdtemp(join(tmpdir(), 'cb-pool-'))
+    files = await writeCorpus(join(home, '.claude'), 2, 4)
+    process.env['CLAUDE_CONFIG_DIR'] = join(home, '.claude')
+    process.env['CODEBURN_CACHE_DIR'] = join(home, '.cache', 'codeburn')
+  })
+
+  afterEach(async () => {
+    clearSessionCache()
+    delete process.env['CODEBURN_PARSE_WORKERS']
+    await rm(home, { recursive: true, force: true })
+  })
+
+  function liveWorkers(): number {
+    return process.getActiveResourcesInfo().filter(r => r === 'Worker').length
+  }
+
+  it('returns results in submission order and terminates every thread on close', async () => {
+    const before = liveWorkers()
+    const pool = new ParseWorkerPool(3)
+    const results = []
+    for await (const r of parseFilesInOrder(pool, files)) results.push(r)
+    await pool.close()
+
+    expect(results).toHaveLength(files.length)
+    for (const r of results) expect(r.ok).toBe(true)
+    // Each fixture session's first turn names its own project, which pins the
+    // yielded order to the submitted order rather than to completion order.
+    const projects = results.map(r => (r.ok && r.parsed ? r.parsed.turns[0]?.userMessage : undefined))
+    expect(projects).toEqual(files.map((_, i) => `task 0 in ${Math.floor(i / 4)}`))
+    expect(liveWorkers()).toBe(before)
+  })
+
+  // A worker that cannot answer must hand the file back, never drop it: the
+  // caller's fallback is an in-process parse, and it has to land on the same
+  // result the worker would have produced.
+  it('reports failures instead of throwing, and the serial fallback matches', async () => {
+    const pool = new ParseWorkerPool(1)
+    const fromWorker = await pool.submit(files[0]!)
+    await pool.close()
+
+    const afterClose = await pool.submit(files[1]!)
+    expect(afterClose.ok).toBe(false)
+
+    const serial = await parseClaudeFileFull(files[0]!, new Set<string>())
+    expect(fromWorker.ok).toBe(true)
+    if (!fromWorker.ok || !fromWorker.parsed) throw new Error('expected a parsed result')
+    const { msgIds, ...worker } = fromWorker.parsed
+    expect(msgIds.length).toBeGreaterThan(0)
+    expect(worker).toEqual(JSON.parse(JSON.stringify(serial)))
+  })
+
+  // The resident `serve` child parses over and over in one process; a thread
+  // that outlives its parse would accumulate across requests.
+  it('leaves no live worker behind after back-to-back parses', async () => {
+    const before = liveWorkers()
+    process.env['CODEBURN_PARSE_WORKERS'] = '2'
+
+    await parseAllSessions()
+    expect(liveWorkers()).toBe(before)
+
+    clearSessionCache()
+    await parseAllSessions()
+    expect(liveWorkers()).toBe(before)
+  })
+})

tsup.config.ts

@@ -1,7 +1,7 @@
 import { defineConfig } from 'tsup'
 
 export default defineConfig({
-  entry: ['src/main.ts'],
+  entry: ['src/main.ts', 'src/parse-worker.ts'],
   format: ['esm'],
   target: 'node20',
   outDir: 'dist',