mirror of
https://github.com/AgentSeal/codeburn.git
synced 2026-08-26 17:03:31 +00:00
perf: fan out the discovery sweep instead of walking it one syscall at a time
Every dated command re-walks and re-stats every provider tree before it can decide what the cache already covers. That sweep was strictly serial at every level -- one readdir, one stat, one state.json read at a time, and one provider after another -- so on a 21k-file / 9-provider corpus it owned most of a warm run's wall clock with the machine idle waiting on the kernel. Measured per warm `today` pass on that corpus before this change: 859 ms in discoverAllSessions (kimicode 467, codex 143, claude 135, grok 77) and 727 ms in the Claude project walk + fingerprint pass. - fs-utils: mapWithConcurrency + FS_SCAN_CONCURRENCY, one bounded, order- preserving helper for the whole sweep. - providers/index: run provider discovery concurrently, concatenated in registry order. - claude/codex/grok/kimicode: walk each level with the level fanned out, re-concatenated in readdir order before anything reconciles. - parser: the Claude dir walk and both fingerprint passes (scanProjectDirs and parseProviderSources) fan out, then reconcile serially in discovery order, which is what changedFiles ordering and the seenMsgIds pre-seed depend on. - parser: collectJsonlFiles reads entries with their types, so a plain file no longer costs a wasted subagents/ probe. - codex-cache: share one in-flight load between concurrent readers. The memo is only populated after the read + parse resolves, so concurrent discovery had every caller re-reading and re-parsing the same (here 59 MB) file. No reconciliation logic changed: the same fingerprints reach the same cache comparison in the same order. Warm `today` 5.30s -> 2.99s, cold 34.7s -> 31.1s (medians of 5 / 2, isolated HOME + cache). today/report -p month/models/sessions JSON, warm and cold, per-provider and combined, are byte-identical apart from the run's own `generated` timestamp. Closes #1104
This commit is contained in:
iamtoruk
parent
554d32eba6
commit
d5e485f415
11 changed files with 376 additions and 125 deletions
|
|
@ -19,6 +19,7 @@
|
|||
- **DeepSeek Harness (`dsh`) is now a supported provider.** Reads DeepSeek's open-source agent harness from `~/.dsh/sessions` (`DSH_HOME` relocates the root), both the default zstd logs and the uncompressed `session.jsonl` variant. A `.zstd` log is a concatenation of independent zstd frames, one per write batch, so it is decoded frame by frame behind a structural frame scan and a torn trailing frame from a crashed writer is ignored rather than failing the file (needs Node 22.15+ for `zlib` zstd; below that dsh is skipped with a notice instead of counted as $0). One call per `(turn, step)`, with the step's final `assistant/message` usage superseding the streamed `assistant/chunk` sample of the same call rather than adding to it, the model taken from the message that served the step, and reasoning tokens billed at the output rate. DSH records tokens but no cost, so calls are priced from the shared tables. The events a forked session replays from its parent are skipped, since codeburn already counts the parent's own log. The session format is pinned at version 0 upstream with no compatibility implied, so a log stamped with any other version is skipped with a notice instead of read under today's assumptions.
|
||||
|
||||
### Changed
|
||||
- **The discovery sweep issues its metadata syscalls concurrently.** Every dated command re-walks and re-stats every provider tree before it can decide what is already cached, and that sweep was strictly serial: one `readdir`, one `stat`, one `state.json` read at a time, per provider, one provider after another. On a 21k-file / 9-provider corpus it owned most of a warm run's wall clock while the machine sat idle waiting on the kernel. Provider discovery now runs across providers at once, and the four walks that dominate it (claude, codex, kimicode, grok) plus the Claude project-dir walk and both fingerprint passes fan out through a shared bounded-concurrency helper. Order is unchanged everywhere — each level is re-concatenated in registry/`readdir` order before anything reconciles against the cache — so what is discovered, in what sequence, is byte-identical to the serial walk. Two smaller cuts ride along: the Claude walk reads directory entries with their types so a plain file no longer costs a wasted `subagents/` probe, and the Codex result cache (a single file that can reach hundreds of MB) now shares one in-flight load between concurrent readers instead of letting each one re-read and re-parse it. Warm `codeburn today` on that corpus: 5.30s to 2.99s; a cold parse 34.7s to 31.1s. (#1104)
|
||||
- **A kill mid-way through a non-Claude provider phase no longer restarts that whole phase.** `scanProjectDirs` (Claude) has long taken a throttled `saveProgress` callback so a killed cold parse resumes from a warm cache; `parseProviderSources` (every other provider — codex, cursor, gemini, and the rest) did not, and only persisted at the whole-provider boundary. On a large single-provider corpus (a multi-GB codex history is the common case) an app-timeout SIGKILL, crash, or force-quit during that phase discarded everything parsed since the last provider finished, forcing the entire phase to re-parse from zero on the next run. `parseProviderSources` now takes the same callback, invoked once per source right after that source's cache entry lands (mirroring `scanProjectDirs`' placement, outside the per-file try/catch), on the same file-count/wall-clock throttle. A file only ever gets a fingerprint once it has fully parsed, so a mid-file kill can never leave a half-parsed file's entry looking complete on resume.
|
||||
- **Routed model ids price as the model they wrap, and an unknown vendor prefix no longer prices by blind stripping.** Token-plan and gateway spellings of the same model (`omniroute:`, `cp/`, `cline-pass/`, `cline-free/`, `cmd/`, `antigravity/`) are peeled and the remaining id is priced, so a Cline Pass or OmniRoute session shows a `~` estimate instead of $0. In exchange, `provider/model` is no longer treated as authority on its own: the leading segment is stripped only when it is a namespace the bundled pricing catalog itself uses (`anthropic/`, `openai/`, `google/`, `x-ai/`, `qwen/`, `moonshotai/`, `nousresearch/`, `xiaomi/`, `z-ai/`, and every other vendor prefix in the LiteLLM snapshot), one of the routing wrappers above, or one of the client-side spellings `kimi/`, `mimo/`, `zhipu/`, `litellm_proxy/` and `openai_like/`. Anything else stays unpriced and is reported as unpriced rather than inheriting the price of a same-named cloud row, and local-runner prefixes (`ollama/`, `lmstudio/`, `hosted_vllm/`, `local/`) are excluded on purpose so an unlisted local tag can never invent cloud spend. A user price override for the bare id wins over the catalog row a routed spelling would otherwise hit.
|
||||
- **SQLite providers now survive read-only database parents.** A read-only SQLite open is not read-only on disk: on a WAL database SQLite must create `<db>-shm` and `<db>-wal` in the database's own directory, so a source on read-only media, under restrictive permissions, or inside a Flatpak/snap confinement failed with `attempt to write a readonly database` (or `unable to open database file` when a `-wal` was present without its `-shm`), and both discovery sites swallowed it — the provider read as "not installed" rather than as an error. That covers cursor, cursor-agent, opencode, goose, warp, kilo-code, zerostack and the copilot agent-traces database. The direct open stays the fast path and is byte-identical when it succeeds. When it fails for want of sidecars: a database with no WAL frames to lose is opened in place with `immutable=1`, which costs nothing and cannot go stale; a database with a non-empty `-wal` is copied with its `-wal` into the CodeBurn cache and read there, so its un-checkpointed rows are never silently dropped. The copy costs one database's worth of disk and is taken once per change — it is keyed by the main-plus-WAL fingerprint, published under a fingerprint-stamped name so a refresh never overwrites a copy another process is reading, and superseded copies are evicted once a day has passed without a read, keeping at most one predecessor. If the cache itself cannot be written, the database is skipped with a notice naming it and the reason rather than in silence. The original provider database is never opened writable or modified.
|
||||
|
|
|
|||
|
|
@ -113,11 +113,25 @@ const memCaches = new Map<string, ResultCache>()
|
|||
// flushCodexCache() in the parse's finally, so the next load re-reads disk.
|
||||
export function clearCodexMemCaches(): void {
|
||||
memCaches.clear()
|
||||
inFlightLoads.clear()
|
||||
}
|
||||
|
||||
async function loadCache(cacheDir: string): Promise<ResultCache> {
|
||||
// Concurrent callers must share one load. The memo below is only populated
|
||||
// after the read + JSON.parse resolves, so without this every in-flight caller
|
||||
// would re-read and re-parse the same (hundreds-of-MB) cache file.
|
||||
const inFlightLoads = new Map<string, Promise<ResultCache>>()
|
||||
|
||||
function loadCache(cacheDir: string): Promise<ResultCache> {
|
||||
const inMemory = memCaches.get(cacheDir)
|
||||
if (inMemory) return inMemory
|
||||
if (inMemory) return Promise.resolve(inMemory)
|
||||
const pending = inFlightLoads.get(cacheDir)
|
||||
if (pending) return pending
|
||||
const load = loadCacheFromDisk(cacheDir).finally(() => inFlightLoads.delete(cacheDir))
|
||||
inFlightLoads.set(cacheDir, load)
|
||||
return load
|
||||
}
|
||||
|
||||
async function loadCacheFromDisk(cacheDir: string): Promise<ResultCache> {
|
||||
const empty = { version: CODEX_CACHE_VERSION, files: {} }
|
||||
const versioned = await readExistingTextFile(getCachePath(cacheDir))
|
||||
if (versioned.status === 'ok') {
|
||||
|
|
|
|||
|
|
@ -225,3 +225,26 @@ export async function* readSessionLines(
|
|||
stream.destroy()
|
||||
}
|
||||
}
|
||||
|
||||
// Metadata syscalls (readdir/stat) issued one-at-a-time leave the OS idle
|
||||
// between them: on a 21k-file corpus the discovery sweep spent most of its wall
|
||||
// clock waiting rather than working. Results are returned in input order so
|
||||
// callers keep their deterministic, order-sensitive downstream processing.
|
||||
export const FS_SCAN_CONCURRENCY = 32
|
||||
|
||||
export async function mapWithConcurrency<T, R>(
|
||||
items: readonly T[],
|
||||
limit: number,
|
||||
worker: (item: T, index: number) => Promise<R>,
|
||||
): Promise<R[]> {
|
||||
const out = new Array<R>(items.length)
|
||||
let idx = 0
|
||||
const run = async (): Promise<void> => {
|
||||
while (idx < items.length) {
|
||||
const current = idx++
|
||||
out[current] = await worker(items[current]!, current)
|
||||
}
|
||||
}
|
||||
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, run))
|
||||
return out
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import { existsSync } from 'fs'
|
||||
import { lstat, readFile, readdir, stat } from 'fs/promises'
|
||||
import { basename, dirname, join, resolve, sep } from 'path'
|
||||
import { readSessionLines } from './fs-utils.js'
|
||||
import { FS_SCAN_CONCURRENCY, mapWithConcurrency, readSessionLines } from './fs-utils.js'
|
||||
import { billableOutputTokens, calculateCost, calculateLocalModelSavings, getShortModelName, isProxiedPath, getProxyPathsConfigHash, getModelAliasesConfigHash, getPriceOverridesConfigHash, getLocalModelSavingsConfigHash } from './models.js'
|
||||
import { resolveSubagentAttribution, sessionIdentity } from './sessions-report.js'
|
||||
import { normalizeContentBlocks, flatSlice, flatString } from './content-utils.js'
|
||||
|
|
@ -1922,13 +1922,17 @@ async function collectJsonlInto(dir: string, out: Set<string>): Promise<void> {
|
|||
}
|
||||
|
||||
export async function collectJsonlFiles(dirPath: string): Promise<string[]> {
|
||||
const files = await readdir(dirPath).catch(() => [])
|
||||
const jsonlFiles = new Set(files.filter(f => f.endsWith('.jsonl')).map(f => join(dirPath, f)))
|
||||
const files = await readdir(dirPath, { withFileTypes: true }).catch(() => [])
|
||||
const jsonlFiles = new Set(files.filter(f => f.name.endsWith('.jsonl')).map(f => join(dirPath, f.name)))
|
||||
|
||||
await collectJsonlInto(join(dirPath, 'subagents'), jsonlFiles)
|
||||
for (const entry of files) {
|
||||
if (entry.endsWith('.jsonl')) continue
|
||||
await collectJsonlInto(join(dirPath, entry, 'subagents'), jsonlFiles)
|
||||
if (entry.name.endsWith('.jsonl')) continue
|
||||
// A plain file can't hold a subagents/ dir, so don't spend a readdir
|
||||
// finding out. Anything else (real dir, symlink, unknown type) still gets
|
||||
// probed, matching what the untyped readdir used to do.
|
||||
if (entry.isFile()) continue
|
||||
await collectJsonlInto(join(dirPath, entry.name, 'subagents'), jsonlFiles)
|
||||
}
|
||||
|
||||
return [...jsonlFiles]
|
||||
|
|
@ -1969,36 +1973,47 @@ async function scanProjectDirs(
|
|||
|
||||
const discoverProgress = createScanProgress('scanning claude project dirs', dirs.length)
|
||||
let dirsDone = 0
|
||||
for (const { path: dirPath, name: dirName, source } of dirs) {
|
||||
// Walk and fingerprint concurrently, then reconcile serially in discovery
|
||||
// order: the reconcile loop feeds order-sensitive state (changedFiles order
|
||||
// drives the worker-result pairing, seenMsgIds pre-seeding), so only the
|
||||
// syscalls are allowed to overlap.
|
||||
const walked = await mapWithConcurrency(dirs, FS_SCAN_CONCURRENCY, async ({ path: dirPath }) => {
|
||||
const jsonlFiles = await collectJsonlFiles(dirPath)
|
||||
for (const filePath of jsonlFiles) {
|
||||
allDiscoveredFiles.add(filePath)
|
||||
const fp = await fingerprintFile(filePath)
|
||||
if (!fp) continue
|
||||
|
||||
const cached = section.files[filePath]
|
||||
const action = reconcileFile(fp, cached)
|
||||
if (cached && (readOnly || action.action === 'unchanged')) {
|
||||
if (readOnly && action.action !== 'unchanged') readOnlyServedStale = true
|
||||
unchangedFiles.push({ filePath, dirName, source, cached: section.files[filePath]! })
|
||||
} else if (!readOnly) {
|
||||
if (action.action === 'appended') {
|
||||
changedFiles.push({
|
||||
filePath,
|
||||
info: { dirName, fp, source },
|
||||
append: { cached: section.files[filePath]!, readFromOffset: action.readFromOffset },
|
||||
})
|
||||
continue
|
||||
}
|
||||
changedFiles.push({ filePath, info: { dirName, fp, source } })
|
||||
} else {
|
||||
// Read-only with no cache entry at all: this file is dropped from what
|
||||
// we serve, so the snapshot under-reports whatever days it covers.
|
||||
readOnlyServedStale = true
|
||||
}
|
||||
}
|
||||
dirsDone++
|
||||
await discoverProgress.tick(dirsDone)
|
||||
return jsonlFiles
|
||||
})
|
||||
const discovered: Array<{ filePath: string; dirName: string; source?: SessionSourceMetadata }> = []
|
||||
for (let i = 0; i < dirs.length; i++) {
|
||||
const { name: dirName, source } = dirs[i]!
|
||||
for (const filePath of walked[i]!) discovered.push({ filePath, dirName, source })
|
||||
}
|
||||
const fingerprints = await mapWithConcurrency(discovered, FS_SCAN_CONCURRENCY, e => fingerprintFile(e.filePath))
|
||||
for (const [i, { filePath, dirName, source }] of discovered.entries()) {
|
||||
allDiscoveredFiles.add(filePath)
|
||||
const fp = fingerprints[i]
|
||||
if (!fp) continue
|
||||
|
||||
const cached = section.files[filePath]
|
||||
const action = reconcileFile(fp, cached)
|
||||
if (cached && (readOnly || action.action === 'unchanged')) {
|
||||
if (readOnly && action.action !== 'unchanged') readOnlyServedStale = true
|
||||
unchangedFiles.push({ filePath, dirName, source, cached: section.files[filePath]! })
|
||||
} else if (!readOnly) {
|
||||
if (action.action === 'appended') {
|
||||
changedFiles.push({
|
||||
filePath,
|
||||
info: { dirName, fp, source },
|
||||
append: { cached: section.files[filePath]!, readFromOffset: action.readFromOffset },
|
||||
})
|
||||
continue
|
||||
}
|
||||
changedFiles.push({ filePath, info: { dirName, fp, source } })
|
||||
} else {
|
||||
// Read-only with no cache entry at all: this file is dropped from what
|
||||
// we serve, so the snapshot under-reports whatever days it covers.
|
||||
readOnlyServedStale = true
|
||||
}
|
||||
}
|
||||
discoverProgress.finish()
|
||||
|
||||
|
|
@ -3220,19 +3235,27 @@ export async function parseProviderSources(
|
|||
const unchangedSources: Array<{ source: SessionSource; cached: CachedFile }> = []
|
||||
const changedSources: SourceInfo[] = []
|
||||
|
||||
for (const source of sources) {
|
||||
// Same shape as scanProjectDirs: overlap the stat syscalls, then reconcile in
|
||||
// discovery order. Network sources on a write run never reach fingerprintFile
|
||||
// (they take the synthetic-fingerprint branch below), so they are skipped here.
|
||||
const skipFingerprint = provider.network && !readOnly
|
||||
const sourceFingerprints = skipFingerprint
|
||||
? []
|
||||
: await mapWithConcurrency(sources, FS_SCAN_CONCURRENCY, s => fingerprintFile(s.path))
|
||||
|
||||
for (const [sourceIndex, source] of sources.entries()) {
|
||||
allDiscoveredFiles.add(source.path)
|
||||
|
||||
// Network providers (e.g. Vercel AI Gateway) have no on-disk file — their data
|
||||
// comes from a live API fetch in createSessionParser. There's nothing to
|
||||
// fingerprint or incrementally cache, so re-fetch every run with a synthetic
|
||||
// fingerprint (mtime=now so the date-range filter below never excludes it).
|
||||
if (provider.network && !readOnly) {
|
||||
if (skipFingerprint) {
|
||||
changedSources.push({ source, fp: { dev: 0, ino: 0, mtimeMs: Date.now(), sizeBytes: 0 } })
|
||||
continue
|
||||
}
|
||||
|
||||
const fp = await fingerprintFile(source.path)
|
||||
const fp = sourceFingerprints[sourceIndex]
|
||||
if (!fp) {
|
||||
// A source that was discovered but cannot be fingerprinted is skipped —
|
||||
// but skipping is only safe when the file is genuinely GONE (discovery
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import { createHash } from 'crypto'
|
|||
import type { Provider, ProbeRoot, SessionSource, SessionParser } from './types.js'
|
||||
import { getShortModelName } from '../models.js'
|
||||
import { readConfig } from '../config.js'
|
||||
import { FS_SCAN_CONCURRENCY, mapWithConcurrency } from '../fs-utils.js'
|
||||
|
||||
export type ClaudeConfigSource = {
|
||||
id: string
|
||||
|
|
@ -301,14 +302,19 @@ export const claude: Provider = {
|
|||
// a real and a stale path in CLAUDE_CONFIG_DIRS without breaking.
|
||||
continue
|
||||
}
|
||||
for (const dirName of entries) {
|
||||
// stat() (not the readdir Dirent) decides directory-ness so a symlinked
|
||||
// project dir still counts; issue them concurrently, then apply the
|
||||
// order-sensitive dedup serially.
|
||||
const dirStats = await mapWithConcurrency(entries, FS_SCAN_CONCURRENCY, dirName =>
|
||||
stat(join(projectsDir, dirName)).catch(() => null))
|
||||
for (const [i, dirName] of entries.entries()) {
|
||||
const dirPath = join(projectsDir, dirName)
|
||||
// Resolve before deduping so two CLAUDE_CONFIG_DIRS entries that
|
||||
// reach the same projects/<slug> directory (via symlinks or
|
||||
// overlapping configs) emit only one SessionSource.
|
||||
const resolved = resolve(dirPath)
|
||||
if (seenProjectDirs.has(resolved)) continue
|
||||
const dirStat = await stat(dirPath).catch(() => null)
|
||||
const dirStat = dirStats[i]
|
||||
if (!dirStat?.isDirectory()) continue
|
||||
seenProjectDirs.add(resolved)
|
||||
// `project: dirName` is identical across config dirs for the same
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import { createInterface } from 'readline'
|
|||
import { basename, join } from 'path'
|
||||
import { homedir } from 'os'
|
||||
|
||||
import { readSessionLines } from '../fs-utils.js'
|
||||
import { FS_SCAN_CONCURRENCY, mapWithConcurrency, readSessionLines } from '../fs-utils.js'
|
||||
import { billableOutputTokens, calculateCost, getModelCosts } from '../models.js'
|
||||
import { readCachedCodexResults, writeCachedCodexResults, getCachedCodexProject, fingerprintFile, type CodexFileFingerprint } from '../codex-cache.js'
|
||||
import { mergeToolIntervals } from '../codex-throughput.js'
|
||||
|
|
@ -551,47 +551,46 @@ async function discoverSessionsInDir(codexDir: string): Promise<SessionSource[]>
|
|||
const seenBasenames = new Set<string>()
|
||||
const sessionsDir = join(codexDir, 'sessions')
|
||||
|
||||
const years = await readdir(sessionsDir).catch(() => [] as string[])
|
||||
const years = (await readdir(sessionsDir).catch(() => [] as string[])).filter(y => /^\d{4}$/.test(y))
|
||||
|
||||
for (const year of years) {
|
||||
if (!/^\d{4}$/.test(year)) continue
|
||||
const monthDirs = (await mapWithConcurrency(years, FS_SCAN_CONCURRENCY, async year => {
|
||||
const yearDir = join(sessionsDir, year)
|
||||
const months = await readdir(yearDir).catch(() => [] as string[])
|
||||
return (await readdir(yearDir).catch(() => [] as string[]))
|
||||
.filter(m => /^\d{2}$/.test(m))
|
||||
.map(m => join(yearDir, m))
|
||||
})).flat()
|
||||
|
||||
for (const month of months) {
|
||||
if (!/^\d{2}$/.test(month)) continue
|
||||
const monthDir = join(yearDir, month)
|
||||
const days = await readdir(monthDir).catch(() => [] as string[])
|
||||
const dayDirs = (await mapWithConcurrency(monthDirs, FS_SCAN_CONCURRENCY, async monthDir =>
|
||||
(await readdir(monthDir).catch(() => [] as string[]))
|
||||
.filter(d => /^\d{2}$/.test(d))
|
||||
.map(d => join(monthDir, d)),
|
||||
)).flat()
|
||||
|
||||
for (const day of days) {
|
||||
if (!/^\d{2}$/.test(day)) continue
|
||||
const dayDir = join(monthDir, day)
|
||||
const files = await readdir(dayDir).catch(() => [] as string[])
|
||||
|
||||
for (const file of files) {
|
||||
if (!file.startsWith('rollout-') || !file.endsWith('.jsonl')) continue
|
||||
if (seenBasenames.has(file)) continue
|
||||
seenBasenames.add(file)
|
||||
const source = await discoverSessionFile(join(dayDir, file))
|
||||
if (source) sources.push(source)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
const dated = (await mapWithConcurrency(dayDirs, FS_SCAN_CONCURRENCY, async dayDir =>
|
||||
(await readdir(dayDir).catch(() => [] as string[]))
|
||||
.filter(f => f.startsWith('rollout-') && f.endsWith('.jsonl'))
|
||||
.map(f => ({ file: f, path: join(dayDir, f) })),
|
||||
)).flat()
|
||||
|
||||
// Codex moves archived sessions into a flat directory. Keep them in usage
|
||||
// reports so archiving a conversation does not erase its historical usage.
|
||||
// Call-level deduplication (seenKeys) already collapses any remaining
|
||||
// archived copies, while basename dedup above prevents double discovery.
|
||||
// archived copies, while basename dedup below prevents double discovery.
|
||||
const archivedDir = join(codexDir, 'archived_sessions')
|
||||
const archivedFiles = await readdir(archivedDir).catch(() => [] as string[])
|
||||
for (const file of archivedFiles) {
|
||||
if (!file.startsWith('rollout-') || !file.endsWith('.jsonl')) continue
|
||||
if (seenBasenames.has(file)) continue
|
||||
seenBasenames.add(file)
|
||||
const source = await discoverSessionFile(join(archivedDir, file))
|
||||
if (source) sources.push(source)
|
||||
const archived = (await readdir(archivedDir).catch(() => [] as string[]))
|
||||
.filter(f => f.startsWith('rollout-') && f.endsWith('.jsonl'))
|
||||
.map(f => ({ file: f, path: join(archivedDir, f) }))
|
||||
|
||||
// Dedup before the reads so a basename is opened once, and keep the dated
|
||||
// pass ahead of the archived one — the same precedence the serial walk had.
|
||||
const candidates: Array<{ file: string; path: string }> = []
|
||||
for (const entry of [...dated, ...archived]) {
|
||||
if (seenBasenames.has(entry.file)) continue
|
||||
seenBasenames.add(entry.file)
|
||||
candidates.push(entry)
|
||||
}
|
||||
const discovered = await mapWithConcurrency(candidates, FS_SCAN_CONCURRENCY, c => discoverSessionFile(c.path))
|
||||
for (const source of discovered) if (source) sources.push(source)
|
||||
|
||||
return sources
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import { readdir, stat } from 'fs/promises'
|
|||
import { basename, dirname, join } from 'path'
|
||||
import { homedir } from 'os'
|
||||
|
||||
import { readSessionFile } from '../fs-utils.js'
|
||||
import { FS_SCAN_CONCURRENCY, mapWithConcurrency, readSessionFile } from '../fs-utils.js'
|
||||
import { calculateCost, getModelCosts, getShortModelName } from '../models.js'
|
||||
import { extractBashCommands } from '../bash-utils.js'
|
||||
import type { ProbeRoot, Provider, SessionSource, SessionParser, ParsedProviderCall } from './types.js'
|
||||
|
|
@ -458,30 +458,28 @@ async function discoverSessions(sessionsDir: string): Promise<SessionSource[]> {
|
|||
return sources
|
||||
}
|
||||
|
||||
for (const cwdName of cwdDirs) {
|
||||
// Fanned out per level (the tree is one summary.json read per session); the
|
||||
// per-level results are re-concatenated in readdir order so the emitted
|
||||
// source order matches the serial walk exactly.
|
||||
const cwds = (await mapWithConcurrency(cwdDirs, FS_SCAN_CONCURRENCY, async cwdName => {
|
||||
const cwdPath = join(sessionsDir, cwdName)
|
||||
const cwdStat = await stat(cwdPath).catch(() => null)
|
||||
if (!cwdStat?.isDirectory()) continue
|
||||
if (!cwdStat?.isDirectory()) return []
|
||||
const sessionDirs = await readdir(cwdPath).catch(() => null)
|
||||
if (!sessionDirs) return []
|
||||
return sessionDirs.map(sessionName => ({ cwdName, sessionPath: join(cwdPath, sessionName) }))
|
||||
})).flat()
|
||||
|
||||
let sessionDirs: string[]
|
||||
try {
|
||||
sessionDirs = await readdir(cwdPath)
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
const sessions = await mapWithConcurrency(cwds, FS_SCAN_CONCURRENCY, async ({ cwdName, sessionPath }) => {
|
||||
const sessionStat = await stat(sessionPath).catch(() => null)
|
||||
if (!sessionStat?.isDirectory()) return null
|
||||
const summary = await readJson<GrokSummary>(join(sessionPath, 'summary.json'))
|
||||
if (!summary) return null
|
||||
const cwd = summary.info?.cwd ?? safeDecode(cwdName)
|
||||
return { path: join(sessionPath, 'updates.jsonl'), project: basename(cwd), provider: 'grok' } as SessionSource
|
||||
})
|
||||
|
||||
for (const sessionName of sessionDirs) {
|
||||
const sessionPath = join(cwdPath, sessionName)
|
||||
const sessionStat = await stat(sessionPath).catch(() => null)
|
||||
if (!sessionStat?.isDirectory()) continue
|
||||
|
||||
const summary = await readJson<GrokSummary>(join(sessionPath, 'summary.json'))
|
||||
if (!summary) continue
|
||||
|
||||
const cwd = summary.info?.cwd ?? safeDecode(cwdName)
|
||||
sources.push({ path: join(sessionPath, 'updates.jsonl'), project: basename(cwd), provider: 'grok' })
|
||||
}
|
||||
}
|
||||
for (const source of sessions) if (source) sources.push(source)
|
||||
|
||||
return sources
|
||||
}
|
||||
|
|
|
|||
|
|
@ -286,12 +286,11 @@ export async function discoverAllSessions(
|
|||
const filtered = providerFilter && providerFilter !== 'all'
|
||||
? allProviders.filter(p => p.name === providerFilter)
|
||||
: allProviders
|
||||
const all: SessionSource[] = []
|
||||
for (const provider of filtered) {
|
||||
const sessions = await safeDiscoverSessions(provider)
|
||||
all.push(...sessions)
|
||||
}
|
||||
return all
|
||||
// Each provider's discovery is its own serial directory walk; run them
|
||||
// concurrently and concatenate in registry order so the result stays
|
||||
// byte-identical to the sequential version.
|
||||
const perProvider = await Promise.all(filtered.map(provider => safeDiscoverSessions(provider)))
|
||||
return perProvider.flat()
|
||||
}
|
||||
|
||||
export async function getProvider(name: string): Promise<Provider | undefined> {
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { basename, dirname, join, resolve } from 'node:path'
|
|||
|
||||
import { extractBashCommands } from '../bash-utils.js'
|
||||
import { calculateCost } from '../models.js'
|
||||
import { FS_SCAN_CONCURRENCY, mapWithConcurrency } from '../fs-utils.js'
|
||||
import type { ParsedProviderCall, ProbeRoot, Provider, SessionParser, SessionSource } from './types.js'
|
||||
|
||||
type JsonObject = Record<string, unknown>
|
||||
|
|
@ -121,39 +122,49 @@ function projectFromWorkDir(workDir: string, workDirKey: string): string {
|
|||
return match?.[1] || workDirKey.replace(/^wd_/, '') || 'kimicode'
|
||||
}
|
||||
|
||||
// Walked one level at a time with each level fanned out, rather than as nested
|
||||
// serial loops: the tree is thousands of tiny state.json reads and wire.jsonl
|
||||
// stats, and issuing them one at a time left the corpus scan waiting on the
|
||||
// kernel. The final sort makes the result order independent of completion order.
|
||||
async function discoverSources(root: string): Promise<SessionSource[]> {
|
||||
const sources: SessionSource[] = []
|
||||
const sessionsDir = join(root, 'sessions')
|
||||
|
||||
for (const workDirEntry of await directoryEntries(sessionsDir)) {
|
||||
if (!workDirEntry.isDirectory() || !workDirEntry.name.startsWith('wd_')) continue
|
||||
const workDirPath = join(sessionsDir, workDirEntry.name)
|
||||
const workDirs = (await directoryEntries(sessionsDir))
|
||||
.filter(e => e.isDirectory() && e.name.startsWith('wd_'))
|
||||
.map(e => ({ key: e.name, path: join(sessionsDir, e.name) }))
|
||||
|
||||
for (const sessionEntry of await directoryEntries(workDirPath)) {
|
||||
// Session dir naming differs by host product: the CLI uses session_*,
|
||||
// embedded runtimes (desktop app, IDE) use conv-*/ctitle-*. Any directory
|
||||
// is accepted; the agents/*/wire.jsonl probe below gates real sessions.
|
||||
if (!sessionEntry.isDirectory()) continue
|
||||
const sessionDir = join(workDirPath, sessionEntry.name)
|
||||
const state = await readState(sessionDir)
|
||||
const project = projectFromWorkDir(state.workDir ?? '', workDirEntry.name)
|
||||
// Session dir naming differs by host product: the CLI uses session_*,
|
||||
// embedded runtimes (desktop app, IDE) use conv-*/ctitle-*. Any directory is
|
||||
// accepted; the agents/*/wire.jsonl probe below gates real sessions.
|
||||
const sessionDirs = (await mapWithConcurrency(workDirs, FS_SCAN_CONCURRENCY, async wd =>
|
||||
(await directoryEntries(wd.path))
|
||||
.filter(e => e.isDirectory())
|
||||
.map(e => ({ workDirKey: wd.key, sessionDir: join(wd.path, e.name) })),
|
||||
)).flat()
|
||||
|
||||
for (const agentEntry of await directoryEntries(join(sessionDir, 'agents'))) {
|
||||
if (!agentEntry.isDirectory()) continue
|
||||
const wirePath = join(sessionDir, 'agents', agentEntry.name, 'wire.jsonl')
|
||||
if (!await isFile(wirePath)) continue
|
||||
sources.push({
|
||||
path: wirePath,
|
||||
project,
|
||||
provider: 'kimicode',
|
||||
sourceId: agentEntry.name,
|
||||
sourceLabel: agentEntry.name,
|
||||
sourcePath: state.workDir,
|
||||
})
|
||||
}
|
||||
}
|
||||
const agents = (await mapWithConcurrency(sessionDirs, FS_SCAN_CONCURRENCY, async sd => {
|
||||
const state = await readState(sd.sessionDir)
|
||||
const project = projectFromWorkDir(state.workDir ?? '', sd.workDirKey)
|
||||
const agentsDir = join(sd.sessionDir, 'agents')
|
||||
return (await directoryEntries(agentsDir))
|
||||
.filter(e => e.isDirectory())
|
||||
.map(e => ({ agentName: e.name, wirePath: join(agentsDir, e.name, 'wire.jsonl'), project, workDir: state.workDir }))
|
||||
})).flat()
|
||||
|
||||
const present = await mapWithConcurrency(agents, FS_SCAN_CONCURRENCY, a => isFile(a.wirePath))
|
||||
|
||||
const sources: SessionSource[] = []
|
||||
for (const [i, a] of agents.entries()) {
|
||||
if (!present[i]) continue
|
||||
sources.push({
|
||||
path: a.wirePath,
|
||||
project: a.project,
|
||||
provider: 'kimicode',
|
||||
sourceId: a.agentName,
|
||||
sourceLabel: a.agentName,
|
||||
sourcePath: a.workDir,
|
||||
})
|
||||
}
|
||||
|
||||
return sources.sort((a, b) => a.path.localeCompare(b.path))
|
||||
}
|
||||
|
||||
|
|
|
|||
66
tests/codex-cache-concurrent-load.test.ts
Normal file
66
tests/codex-cache-concurrent-load.test.ts
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
// The codex result cache is a single (often hundreds-of-MB) JSON file, memoized
|
||||
// in memory only once the read + parse resolves. Discovery now asks for it from
|
||||
// many concurrent callers, so without a shared in-flight promise every one of
|
||||
// them re-read and re-parsed the whole file.
|
||||
|
||||
import { mkdtemp, rm, writeFile, mkdir } from 'fs/promises'
|
||||
import { tmpdir } from 'os'
|
||||
import { join } from 'path'
|
||||
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
|
||||
|
||||
const readSpy = vi.hoisted(() => vi.fn())
|
||||
|
||||
vi.mock('../src/cache-dir.js', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import('../src/cache-dir.js')>()
|
||||
return {
|
||||
...actual,
|
||||
readExistingTextFile: (path: string) => {
|
||||
readSpy(path)
|
||||
return actual.readExistingTextFile(path)
|
||||
},
|
||||
}
|
||||
})
|
||||
|
||||
const { CODEX_CACHE_VERSION, clearCodexMemCaches, codexCacheFileName, getCachedCodexProject, withCodexCacheDirectory } =
|
||||
await import('../src/codex-cache.js')
|
||||
|
||||
let cacheDir: string
|
||||
let sessionDir: string
|
||||
|
||||
beforeEach(async () => {
|
||||
readSpy.mockClear()
|
||||
clearCodexMemCaches()
|
||||
const root = await mkdtemp(join(tmpdir(), 'codeburn-codex-cache-'))
|
||||
cacheDir = join(root, 'cache')
|
||||
sessionDir = join(root, 'sessions')
|
||||
await mkdir(cacheDir, { recursive: true })
|
||||
await mkdir(sessionDir, { recursive: true })
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
clearCodexMemCaches()
|
||||
await rm(join(cacheDir, '..'), { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('codex result cache under concurrent readers', () => {
|
||||
it('reads the cache file once and answers every caller correctly', async () => {
|
||||
const paths: string[] = []
|
||||
const files: Record<string, unknown> = {}
|
||||
for (let i = 0; i < 24; i++) {
|
||||
const p = join(sessionDir, `rollout-${i}.jsonl`)
|
||||
await writeFile(p, '{}\n')
|
||||
paths.push(p)
|
||||
const { statSync } = await import('fs')
|
||||
const s = statSync(p)
|
||||
files[p] = { dev: s.dev, ino: s.ino, mtimeMs: s.mtimeMs, sizeBytes: s.size, project: `proj-${i}`, calls: [] }
|
||||
}
|
||||
await writeFile(join(cacheDir, codexCacheFileName()), JSON.stringify({ version: CODEX_CACHE_VERSION, files }))
|
||||
|
||||
const projects = await withCodexCacheDirectory(cacheDir, () =>
|
||||
Promise.all(paths.map(p => getCachedCodexProject(p))))
|
||||
|
||||
expect(projects).toEqual(paths.map((_, i) => `proj-${i}`))
|
||||
expect(readSpy.mock.calls.filter(([p]) => String(p).includes('codex-results'))).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
111
tests/discovery-concurrency.test.ts
Normal file
111
tests/discovery-concurrency.test.ts
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
// The discovery sweep (provider walks + per-file stats) now issues its metadata
|
||||
// syscalls concurrently instead of one at a time. Nothing downstream may notice:
|
||||
// source order still follows the registry / readdir order, and nothing that was
|
||||
// discovered serially may be dropped.
|
||||
|
||||
import { mkdtemp, mkdir, rm, symlink, writeFile } from 'fs/promises'
|
||||
import { tmpdir } from 'os'
|
||||
import { basename, join } from 'path'
|
||||
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||
|
||||
import { FS_SCAN_CONCURRENCY, mapWithConcurrency } from '../src/fs-utils.js'
|
||||
import { collectJsonlFiles } from '../src/parser.js'
|
||||
import { discoverAllSessions } from '../src/providers/index.js'
|
||||
import type { Provider, SessionSource } from '../src/providers/types.js'
|
||||
|
||||
let root: string
|
||||
beforeEach(async () => { root = await mkdtemp(join(tmpdir(), 'codeburn-disc-')) })
|
||||
afterEach(async () => { await rm(root, { recursive: true, force: true }) })
|
||||
|
||||
const tick = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms))
|
||||
|
||||
describe('mapWithConcurrency', () => {
|
||||
it('returns results in input order even when they settle out of order', async () => {
|
||||
const items = [40, 0, 20, 0, 10]
|
||||
const out = await mapWithConcurrency(items, 8, async (ms, i) => {
|
||||
await tick(ms)
|
||||
return i
|
||||
})
|
||||
expect(out).toEqual([0, 1, 2, 3, 4])
|
||||
})
|
||||
|
||||
it('never exceeds the requested number of in-flight workers', async () => {
|
||||
let inFlight = 0
|
||||
let peak = 0
|
||||
await mapWithConcurrency(Array.from({ length: 50 }, (_, i) => i), 4, async () => {
|
||||
inFlight++
|
||||
peak = Math.max(peak, inFlight)
|
||||
await tick(1)
|
||||
inFlight--
|
||||
})
|
||||
expect(peak).toBe(4)
|
||||
})
|
||||
|
||||
it('handles an empty list without spawning workers', async () => {
|
||||
await expect(mapWithConcurrency([], FS_SCAN_CONCURRENCY, async () => 1)).resolves.toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
describe('collectJsonlFiles', () => {
|
||||
// The walk skips the subagents/ probe for entries readdir reports as plain
|
||||
// files. A symlink is NOT a plain file, so a symlinked session directory must
|
||||
// still be probed — dropping it would lose that session's subagent spend.
|
||||
it('still probes subagents under a symlinked session directory', async () => {
|
||||
const real = join(root, 'real-session')
|
||||
await mkdir(join(real, 'subagents'), { recursive: true })
|
||||
await writeFile(join(real, 'subagents', 'agent-linked.jsonl'), '{}\n')
|
||||
await symlink(real, join(root, 'linked-session'), 'dir')
|
||||
// A plain sibling file is the case the skip exists for: no probe, no effect.
|
||||
await writeFile(join(root, 'notes.txt'), 'x')
|
||||
|
||||
const found = (await collectJsonlFiles(root)).map(f => basename(f))
|
||||
expect(found).toContain('agent-linked.jsonl')
|
||||
expect(found).not.toContain('notes.txt')
|
||||
})
|
||||
})
|
||||
|
||||
function fakeProvider(name: string, delayMs: number, paths: string[]): Provider {
|
||||
return {
|
||||
name,
|
||||
displayName: name,
|
||||
modelDisplayName: (m: string) => m,
|
||||
toolDisplayName: (t: string) => t,
|
||||
async discoverSessions(): Promise<SessionSource[]> {
|
||||
await tick(delayMs)
|
||||
return paths.map(path => ({ path, project: name, provider: name }))
|
||||
},
|
||||
createSessionParser() { throw new Error('not used') },
|
||||
}
|
||||
}
|
||||
|
||||
describe('discoverAllSessions', () => {
|
||||
it('concatenates providers in registry order regardless of which finishes first', async () => {
|
||||
const providers = [
|
||||
fakeProvider('slow', 30, ['/s1', '/s2']),
|
||||
fakeProvider('fast', 0, ['/f1']),
|
||||
fakeProvider('mid', 10, ['/m1']),
|
||||
]
|
||||
const sources = await discoverAllSessions('all', providers)
|
||||
expect(sources.map(s => s.path)).toEqual(['/s1', '/s2', '/f1', '/m1'])
|
||||
})
|
||||
|
||||
it('keeps a throwing provider isolated without dropping the others', async () => {
|
||||
const boom = fakeProvider('boom', 0, [])
|
||||
boom.discoverSessions = async () => { throw new Error('nope') }
|
||||
const sources = await discoverAllSessions('all', [
|
||||
fakeProvider('a', 5, ['/a1']),
|
||||
boom,
|
||||
fakeProvider('b', 0, ['/b1']),
|
||||
])
|
||||
expect(sources.map(s => s.path)).toEqual(['/a1', '/b1'])
|
||||
})
|
||||
|
||||
it('honours a provider filter', async () => {
|
||||
const sources = await discoverAllSessions('b', [
|
||||
fakeProvider('a', 0, ['/a1']),
|
||||
fakeProvider('b', 0, ['/b1']),
|
||||
])
|
||||
expect(sources.map(s => s.path)).toEqual(['/b1'])
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue