Keep Session History checking totals fixed throughout a scan (#30949)

* fix(session-history): freeze scan workload before checking candidates

* test(session-history): handle native fixture path separators

* test(session-history): control bucket fixture enumeration order

* test(session-history): preserve scoped fixture timestamps

* fix(session-history): repair integrated test and locale catalog

* ci: refresh history checks against current main

* ci: refresh history checks after catalog repair
This commit is contained in:
Brennan Benson
2026-10-22 01:03:09 -07:00
committed by GitHub
parent 33ec2ba9e7
commit f6585a3916
24 changed files with 897 additions and 198 deletions
@@ -1,6 +1,6 @@
import { link, mkdir, mkdtemp, rm, utimes } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { basename, join } from 'node:path'
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { AiVaultScanProgress, AiVaultScanWork } from '../../shared/ai-vault-scan-progress'
import {
@@ -10,8 +10,11 @@ import {
resetLocalAiVaultScanProgressForTests,
startLocalAiVaultScanProgress
} from './ai-vault-local-scan-progress'
import * as parseCache from './session-scanner-parse-cache'
import * as transcriptFs from '../native-chat/wsl-transcript-fs-access'
import { AiVaultScanTally } from './ai-vault-scan-tally'
import { aiVaultScanWork, trackAiVaultScanProgress } from './ai-vault-scan-progress'
import { writeOpenCodeSqliteDatabase } from './session-scanner-opencode-sqlite-fixture'
import { scanAiVaultSessions } from './session-scanner'
import { resetSessionParseCacheForTests } from './session-scanner-parse-cache'
import { AiVaultScannerServiceClient } from './session-scanner-service-client'
@@ -22,10 +25,25 @@ import {
} from './session-scanner-service-test-child'
import { isolatedScanRoots, writeJsonlFile } from './session-scanner-test-fixtures'
// Source fixtures exercise the real SQLite readers without a packaged worker bundle.
vi.mock('./session-scanner-opencode-sqlite-worker-spawn', async () => {
const { listOpenCodeSqliteSessions } = await import('./session-scanner-opencode-sqlite-list')
const { listOpenCode2SqliteSessions } = await import('./session-scanner-opencode2-sqlite-list')
const { parseOpenCodeSqliteSession } = await import('./session-scanner-opencode-sqlite')
return {
listOpenCode2SqliteSessionsViaWorker: listOpenCode2SqliteSessions,
listOpenCodeSqliteSessionsViaWorker: listOpenCodeSqliteSessions,
listZcodeSqliteSessionsViaWorker: (args: Parameters<typeof listOpenCodeSqliteSessions>[0]) =>
listOpenCodeSqliteSessions({ ...args, agent: 'zcode' }),
parseOpenCodeSqliteSessionViaWorker: parseOpenCodeSqliteSession
}
})
let tempRoot: string | null = null
afterEach(async () => {
vi.useRealTimers()
vi.restoreAllMocks()
resetSessionParseCacheForTests()
resetLocalAiVaultScanProgressForTests()
if (tempRoot) {
@@ -102,8 +120,16 @@ describe('Session History scan progress, scanner child to window', () => {
expect(reports.some((work) => work.stage === 'finding')).toBe(true)
expect(final).toEqual({
stage: 'reading',
agents: [{ agent: 'claude', files: 3, databases: 0, walking: false }],
read: { done: 4, total: 4 }
agents: [
{
agent: 'claude',
files: 3,
databases: 0,
walking: false,
read: { done: 4, total: 4, skipped: 0 }
}
],
read: { done: 4, total: 4, skipped: 0 }
})
const { child, scan, published } = startScan()
@@ -135,8 +161,11 @@ describe('Session History scan progress, scanner child to window', () => {
it('reaches its total when the first pass stops before reading every file it planned', async () => {
// The newest file fills a limit of 1, so the pass never reads the older ones it announced.
const { final } = await scanInChild(1)
expect(final.read).toEqual({ done: 1, total: 1 })
const { reports, final } = await scanInChild(1)
expect(final.read).toEqual({ done: 1, total: 2, skipped: 1 })
expect(
reports.filter((work) => work.stage === 'reading').every((work) => work.read.total === 2)
).toBe(true)
})
it('keeps a replacement entry when the scan it replaced fails late', () => {
@@ -202,3 +231,204 @@ describe('Session History scan progress, scanner child to window', () => {
expect(tally.snapshot().agents).toMatchObject([{ agent: 'codex', files: 3 }])
})
})
it.each(['cutoff', 'bucket', 'filtered'] as const)(
'freezes %s work without changing cold/warm execution',
async (flow) => {
tempRoot = await mkdtemp(join(tmpdir(), 'orca-scan-frozen-'))
const roots = isolatedScanRoots(tempRoot)
for (let index = 0; index < (flow === 'filtered' ? 27 : 6); index++) {
const timestamp = new Date(Date.UTC(2026, 9, 19) - index * 60_000).toISOString()
const cwd =
index >= 3 && (flow === 'bucket' || (index - 3) % 8 === 0) ? '/repo' : '/elsewhere'
const path =
flow === 'filtered'
? join(roots.codexSessionsDir, `rollout-${index}.jsonl`)
: join(
roots.claudeProjectsDir,
cwd === '/repo' ? '-repo' : '-elsewhere',
`${index}.jsonl`
)
await writeJsonlFile(
path,
flow === 'filtered'
? [
{ timestamp, type: 'session_meta', payload: { id: `fixture-${index}`, cwd } },
{
timestamp,
type: 'event_msg',
payload: { type: 'user_message', message: `Generated ${index}` }
}
]
: [
{
type: 'user',
sessionId: `fixture-${index}`,
timestamp,
cwd,
message: { role: 'user', content: `Generated ${index}` }
}
]
)
await utimes(path, Date.parse(timestamp) / 1000, Date.parse(timestamp) / 1000)
}
const listed = vi.spyOn(parseCache, 'parseListedSessionFile')
const walked = vi.spyOn(parseCache, 'parseAgentSessionFileCached')
const stats = vi.spyOn(parseCache, 'createSessionParseStats')
const opened = vi.spyOn(transcriptFs, 'openTranscriptReadStream')
const ids =
flow === 'cutoff'
? [0, 1, 2]
: flow === 'filtered'
? [0, 1, 2, 3, 11, 19]
: [0, 1, 2, 3, 4, 5]
for (const warm of [false, true]) {
const reports: AiVaultScanWork[] = []
const signal = new AbortController().signal
let now = 0
const stop = trackAiVaultScanProgress(
signal,
(work) => reports.push(work),
() => (now += 1000)
)
try {
const result = await scanAiVaultSessions({
...roots,
platform: 'darwin',
limit: 3,
limitPerAgent: 50,
signal,
...(flow === 'cutoff' ? {} : { scopePaths: ['/repo'] })
})
expect(result.issues).toEqual([])
expect(result.sessions.map((row) => row.sessionId)).toEqual(
ids.map((id) => `fixture-${id}`)
)
const attempts = [...listed.mock.calls, ...walked.mock.calls].map(
([candidate]) => candidate.file.path
)
expect(
attempts.map((path) =>
Number(basename(path).replace('rollout-', '').replace('.jsonl', ''))
)
).toEqual(ids)
// Captured from the same generated corpus before the change; representatives move earlier.
expect(opened.mock.calls).toHaveLength(warm ? 0 : flow === 'bucket' ? 7 : ids.length)
expect(stats.mock.results.at(-1)?.value).toEqual({
reused: warm ? ids.length : 0,
incremental: 0,
fullParses: warm ? 0 : ids.length,
earlyStopped: 0,
bytesRead: warm ? 0 : flow === 'filtered' ? 1363 : flow === 'bucket' ? 873 : 444
})
const reading = reports.filter((work) => work.stage === 'reading')
const total = flow === 'filtered' ? 27 : 6
let settled = 0
for (const work of reading) {
expect(work.read.total).toBe(total)
expect(work.read.done + (work.read.skipped ?? 0)).toBeGreaterThanOrEqual(settled)
settled = work.read.done + (work.read.skipped ?? 0)
expect(work.agents.reduce((sum, agent) => sum + (agent.read?.total ?? 0), 0)).toBe(total)
expect(work.agents.reduce((sum, agent) => sum + (agent.read?.done ?? 0), 0)).toBe(
work.read.done
)
expect(work.agents.reduce((sum, agent) => sum + (agent.read?.skipped ?? 0), 0)).toBe(
work.read.skipped
)
}
expect(aiVaultScanWork(signal)?.read).toEqual({
done: ids.length,
total,
skipped: total - ids.length
})
} finally {
stop()
}
listed.mockClear()
walked.mockClear()
opened.mockClear()
}
}
)
it.each([false, true])(
'counts repeated global visits and database sessions once (scoped=%s)',
async (scoped) => {
tempRoot = await mkdtemp(join(tmpdir(), 'orca-scan-repeat-'))
const roots = isolatedScanRoots(tempRoot)
await writeJsonlFile(join(roots.droidSessionsDir, 'same.jsonl'), [
{ type: 'session_start', id: 'same', cwd: '/repo', timestamp: '2026-10-19T00:00:00Z' },
{ type: 'message', message: { role: 'user', content: 'Generated' } }
])
const database = join(tempRoot, 'opencode.db')
writeOpenCodeSqliteDatabase(database, [
{ id: 'db-session', directory: '/repo', turns: [{ role: 'user', parts: ['Generated'] }] }
])
const listed = vi.spyOn(parseCache, 'parseListedSessionFile')
const signal = new AbortController().signal
const reports: AiVaultScanWork[] = []
let pendingAfterParses = false
let now = 0
const stop = trackAiVaultScanProgress(
signal,
(work) => {
reports.push(work)
if (listed.mock.calls.length === 3 && work.stage === 'reading' && work.read.done === 0) {
pendingAfterParses = true
}
},
() => (now += 1000)
)
try {
const result = await scanAiVaultSessions({
...roots,
droidProjectsDir: roots.droidSessionsDir,
opencodeDbPaths: [database],
scopePaths: scoped ? ['/repo'] : [],
signal,
limit: 20,
platform: 'darwin'
})
expect(result.sessions.map((row) => row.sessionId).sort()).toEqual(
scoped ? ['db-session', 'same'] : ['db-session', 'same', 'same']
)
expect(listed.mock.calls.filter(([candidate]) => candidate.agent === 'droid')).toHaveLength(2)
expect(aiVaultScanWork(signal)?.read).toEqual({ done: 2, total: 2, skipped: 0 })
if (scoped) {
expect(pendingAfterParses).toBe(true)
}
expect(
reports.filter((work) => work.stage === 'reading').every((work) => work.read.total === 2)
).toBe(true)
expect(
aiVaultScanWork(signal)?.agents.find((agent) => agent.agent === 'opencode')?.read
).toEqual({ done: 1, total: 1, skipped: 0 })
} finally {
stop()
}
}
)
it('does not settle a batch interrupted during its real parser invocation', async () => {
tempRoot = await mkdtemp(join(tmpdir(), 'orca-scan-abort-'))
const roots = isolatedScanRoots(tempRoot)
await writeJsonlFile(join(roots.claudeProjectsDir, 'project', 'a.jsonl'), [
{ type: 'user', sessionId: 'a', message: { role: 'user', content: 'Generated' } }
])
const controller = new AbortController()
const original = parseCache.parseListedSessionFile
vi.spyOn(parseCache, 'parseListedSessionFile').mockImplementation(async (...args) => {
const result = await original(...args)
controller.abort()
return result
})
const stop = trackAiVaultScanProgress(controller.signal, () => undefined)
try {
await expect(
scanAiVaultSessions({ ...roots, signal: controller.signal })
).rejects.toMatchObject({ name: 'AbortError' })
expect(aiVaultScanWork(controller.signal)?.read).toEqual({ done: 0, total: 1, skipped: 0 })
} finally {
stop()
}
})
@@ -6,6 +6,8 @@ import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress'
import {
AI_VAULT_SCAN_PROGRESS_INTERVAL_MS,
noteAiVaultScanFileFound,
aiVaultScanWork,
openAiVaultScanSource,
noteAiVaultScanWorkChanged,
trackAiVaultScanProgress
} from './ai-vault-scan-progress'
@@ -103,3 +105,19 @@ describe('scan progress reporting', () => {
expect(reports.at(-1)).toMatchObject({ stage: 'reading', read: { done: 3, total: 3 } })
})
})
it('keeps source ownership and final facts independent of immediate and trailing sink faults', () => {
vi.useFakeTimers()
const signal = new AbortController().signal
const sink = vi.fn(() => {
throw new Error('Synthetic reporting sink fault')
})
const stop = trackAiVaultScanProgress(signal, sink, () => Date.now())
const close = openAiVaultScanSource(signal, 'claude')
expect(aiVaultScanWork(signal)?.agents[0]?.walking).toBe(true)
close()
expect(() => vi.advanceTimersByTime(AI_VAULT_SCAN_PROGRESS_INTERVAL_MS)).not.toThrow()
expect(sink).toHaveBeenCalledTimes(2)
expect(aiVaultScanWork(signal)?.agents).toEqual([])
stop()
})
+11 -15
View File
@@ -20,7 +20,12 @@ export function trackAiVaultScanProgress(
const tally = new AiVaultScanTally()
const publish = (): void => {
lastReportedAt = now()
report(tally.snapshot())
const snapshot = tally.snapshot()
try {
report(snapshot)
} catch {
// Reporting cannot interrupt scanning or strand source ownership.
}
}
const workChanged = (): void => {
if (trailing) {
@@ -90,24 +95,15 @@ export function noteAiVaultScanDatabases(
reporter?.workChanged()
}
/** Discovery is over: the scan now reads `total` more transcripts. */
export function noteAiVaultScanReading(signal: AbortSignal | undefined, total: number): void {
/** The scan has registered all candidate opportunities. */
export function noteAiVaultScanReading(signal: AbortSignal | undefined): void {
const reporter = signal ? reporters.get(signal) : undefined
reporter?.tally.startReading(total)
reporter?.tally.startReading()
reporter?.workChanged()
}
/** A pass stopped early: `count` transcripts it announced will not be read. */
export function noteAiVaultScanUnread(signal: AbortSignal | undefined, count: number): void {
const reporter = signal ? reporters.get(signal) : undefined
reporter?.tally.dropUnread(count)
reporter?.workChanged()
}
export function noteAiVaultScanFilesRead(signal: AbortSignal | undefined, count: number): void {
const reporter = signal ? reporters.get(signal) : undefined
reporter?.tally.addRead(count)
reporter?.workChanged()
export function aiVaultScanWorkload(signal: AbortSignal | undefined) {
return tallyFor(signal)?.workload
}
/** Sends the tally after a batch of files was counted. */
+17 -18
View File
@@ -1,5 +1,6 @@
import { AI_VAULT_AGENTS, type AiVaultAgent } from '../../shared/ai-vault-types'
import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress'
import { AiVaultScanWorkload } from './ai-vault-scan-workload'
import { sessionFileExecutionNamespace } from './codex-session-root-dedup'
type AgentTally = {
@@ -12,8 +13,7 @@ type AgentTally = {
export class AiVaultScanTally {
private readonly agents = new Map<AiVaultAgent, AgentTally>()
private stage: AiVaultScanWork['stage'] = 'finding'
private readDone = 0
private readTotal = 0
readonly workload = new AiVaultScanWorkload()
openSource(agent: AiVaultAgent): () => void {
const tally = this.agent(agent)
@@ -45,28 +45,30 @@ export class AiVaultScanTally {
this.agent(agent).databases += count
}
startReading(total: number): void {
startReading(): void {
this.stage = 'reading'
this.readTotal += total
}
addRead(count: number): void {
this.readDone += count
}
dropUnread(count: number): void {
this.readTotal -= count
}
snapshot(): AiVaultScanWork {
const agents: AiVaultScanWork['agents'] = []
for (const agent of AI_VAULT_AGENTS) {
if (this.workload.has(agent)) {
this.agent(agent)
}
}
for (const [agent, tally] of this.agents) {
if (tally.files.size > 0 || tally.databases > 0 || tally.openSources > 0) {
if (
tally.files.size > 0 ||
tally.databases > 0 ||
tally.openSources > 0 ||
this.workload.has(agent)
) {
agents.push({
agent,
files: tally.files.size,
databases: tally.databases,
walking: tally.openSources > 0
walking: tally.openSources > 0,
...(this.stage === 'reading' ? { read: this.workload.snapshot(agent) } : {})
})
}
}
@@ -75,10 +77,7 @@ export class AiVaultScanTally {
return {
stage: this.stage,
agents,
read: {
done: Math.min(this.readDone, this.readTotal),
total: this.readTotal
}
read: this.stage === 'reading' ? this.workload.total() : { done: 0, total: 0 }
}
}
+105
View File
@@ -0,0 +1,105 @@
import type { AiVaultAgent } from '../../shared/ai-vault-types'
import type { AiVaultScanReadingProgress } from '../../shared/ai-vault-scan-progress'
import type { SessionFileCandidate } from './session-scanner-types'
type Ownership = { globalRemaining: number; owners: number; parsed: boolean }
export type ScanReadPass = 'global' | 'walk' | 'bucket'
const OWNER = { walk: 1, bucket: 2 } as const
/** Logical membership is fixed before execution; repeated execution visits stay independent. */
export class AiVaultScanWorkload {
private readonly keys = new Map<AiVaultAgent, Map<string, Ownership>>()
private readonly agents = new Map<AiVaultAgent, AiVaultScanReadingProgress>()
register(candidate: SessionFileCandidate, pass: ScanReadPass): void {
this.registerPath(candidate.agent, candidate.file.path, pass)
}
registerPath(agent: AiVaultAgent, path: string, pass: ScanReadPass): void {
let keys = this.keys.get(agent)
let key = keys?.get(path)
if (!key) {
this.counts(agent).total++
// Walk-only paths are unique and cannot occur in bucket agents.
if (pass === 'walk') {
return
}
keys ??= new Map<string, Ownership>()
key = { globalRemaining: 0, owners: 0, parsed: false }
keys.set(path, key)
this.keys.set(agent, keys)
}
if (pass === 'global') {
key.globalRemaining++
} else {
key.owners |= OWNER[pass]
}
}
settle(candidate: SessionFileCandidate, pass: ScanReadPass, parsed: boolean): void {
this.settlePath(candidate.agent, candidate.file.path, pass, parsed)
}
settlePath(agent: AiVaultAgent, path: string, pass: ScanReadPass, parsed: boolean): void {
const keys = this.keys.get(agent)
const key = keys?.get(path)
if (!key) {
this.complete(agent, parsed)
return
}
key.parsed ||= parsed
if (pass === 'global') {
key.globalRemaining--
} else {
key.owners &= ~OWNER[pass]
}
if (key.globalRemaining === 0 && key.owners === 0) {
this.complete(agent, key.parsed)
keys?.delete(path)
}
}
*bucketPaths(agent: AiVaultAgent): Iterable<string> {
for (const [path, key] of this.keys.get(agent) ?? []) {
if ((key.owners & OWNER.bucket) !== 0) {
yield path
}
}
}
snapshot(agent: AiVaultAgent): AiVaultScanReadingProgress {
return { ...this.counts(agent) }
}
has(agent: AiVaultAgent): boolean {
return this.agents.has(agent)
}
total(): AiVaultScanReadingProgress {
const total = { done: 0, total: 0, skipped: 0 }
for (const count of this.agents.values()) {
total.done += count.done
total.total += count.total
total.skipped += count.skipped ?? 0
}
return total
}
private complete(agent: AiVaultAgent, parsed: boolean): void {
const count = this.counts(agent)
if (parsed) {
count.done++
} else {
count.skipped = (count.skipped ?? 0) + 1
}
}
private counts(agent: AiVaultAgent): AiVaultScanReadingProgress {
let counts = this.agents.get(agent)
if (!counts) {
counts = { done: 0, total: 0, skipped: 0 }
this.agents.set(agent, counts)
}
return counts
}
}
+1 -1
View File
@@ -4,7 +4,7 @@ import type { AiVaultScanIssue } from '../../shared/ai-vault-types'
// discovered path, so an uncapped list ships one row per transcript file in the
// tree over IPC. Local and remote scans share the bound — the failure shape is
// the same and both feed the same panel.
const SCAN_ISSUE_LIMIT = 500
export const SCAN_ISSUE_LIMIT = 500
/** Append a scan issue, collapsing everything past the cap into one notice. */
export function recordSessionScanIssue(issues: AiVaultScanIssue[], issue: AiVaultScanIssue): void {
@@ -21,11 +21,7 @@ import { canStopParsingSessions } from './session-scan-cutoff'
import type { SessionFileCandidate, SessionParseResult } from './session-scanner-types'
import { errorMessage } from './session-scanner-values'
import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation'
import {
noteAiVaultScanFilesRead,
noteAiVaultScanReading,
noteAiVaultScanUnread
} from './ai-vault-scan-progress'
import { noteAiVaultScanWorkChanged } from './ai-vault-scan-progress'
const SESSION_PARSE_CONCURRENCY = 8
@@ -42,16 +38,11 @@ export async function parseSessionCandidates(args: {
selectCandidate?: (candidate: SessionFileCandidate) => Promise<boolean>
admitSession?: (session: AiVaultSession) => boolean
listed?: boolean
settled?: (candidate: SessionFileCandidate, parsed: boolean) => void
}): Promise<AiVaultSession[]> {
const sessions = new ScannedSessionCollection()
let nextYieldIndex = 256
let nextCutoffIndex = 0
// Why: the loop alone knows what it will read, so it owns its share of the read total;
// a filtered walk announces each selected batch, an unfiltered pass its whole list.
if (!args.selectCandidate) {
noteAiVaultScanReading(args.signal, args.candidates.length)
}
let index = 0
while (index < args.candidates.length) {
if (index >= nextYieldIndex) {
@@ -89,9 +80,6 @@ export async function parseSessionCandidates(args: {
: batch.map(() => true)
throwIfAiVaultScanCancelled(args.signal)
const selected = batch.filter((_, batchIndex) => verdicts[batchIndex])
if (args.selectCandidate) {
noteAiVaultScanReading(args.signal, selected.length)
}
const results = await Promise.all(
selected.map((candidate) =>
parseSessionCandidate(candidate, {
@@ -107,7 +95,8 @@ export async function parseSessionCandidates(args: {
// An abort can land while a batch settles: a read it cut short is not that
// file's issue, and a partial parse must never be cached or returned.
throwIfAiVaultScanCancelled(args.signal)
noteAiVaultScanFilesRead(args.signal, selected.length)
batch.forEach((candidate, index) => args.settled?.(candidate, verdicts[index]))
noteAiVaultScanWorkChanged(args.signal)
results.forEach((result, resultIndex) => {
if (result.issue) {
@@ -127,9 +116,14 @@ export async function parseSessionCandidates(args: {
})
}
if (!args.selectCandidate) {
noteAiVaultScanUnread(args.signal, args.candidates.length - index)
for (; index < args.candidates.length; index++) {
if (index % 256 === 0) {
await yieldToEventLoop()
}
throwIfAiVaultScanCancelled(args.signal)
args.settled?.(args.candidates[index], false)
}
noteAiVaultScanWorkChanged(args.signal)
return [...sessions.values()]
}
@@ -63,10 +63,9 @@ it.each([false, true])('keeps cold authority with read failure=%s', async (failR
)
if (index === 0) {
const context = jsonLines([{ type: 'turn_context', payload: { cwd: '/repo' } }])
const { atime, mtime } = await stat(path)
await appendFile(path, `\n${context}\n`)
// Why: the append stamps wall-clock mtime, which overtakes the busy files once real time passes their date.
const day1 = new Date(Date.UTC(2026, 9, 1, 10))
await utimes(path, day1, day1)
await utimes(path, atime, mtime)
}
const { mtimeMs, size: sizeBytes, mtime } = await stat(path)
const parsed = await parseAgentSessionFileCached(
@@ -6,11 +6,12 @@ import {
wslGatedStat
} from '../native-chat/wsl-transcript-fs-access'
import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate'
import { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match'
import type { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match'
import type { CwdBucketLayout } from './session-cwd-bucket-layouts'
import type { AiVaultAgent, AiVaultScanIssue } from '../../shared/ai-vault-types'
import { parseWslUncPath } from '../../shared/wsl-paths'
import { recordSessionScanIssue } from './session-scan-issues'
import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation'
import { yieldToEventLoop } from '../../shared/event-loop-yield'
import type { FileWithMtime } from './session-scanner-types'
import { errorMessage, extractString, parseJsonObject } from './session-scanner-values'
@@ -23,6 +24,7 @@ const TRANSCRIPT_EXTENSIONS = new Set(['.jsonl'])
// Bucket encodings are lossy; retain Claude’s cache and verify other transcripts individually.
const PROJECT_DIR_CWD_CACHE_MAX = 2048
const projectDirCwdCache = new Map<string, string>()
export type IssueSink = (issue: AiVaultScanIssue) => void
export function resetProjectDirCwdCacheForTests(): void {
projectDirCwdCache.clear()
@@ -30,21 +32,22 @@ export function resetProjectDirCwdCacheForTests(): void {
// A gate refusal is a stalled WSL root, not an empty one: keep the existing
// degrade-to-empty containment but make the gap visible in the scan issues.
function recordRefusal(
issues: AiVaultScanIssue[],
export function recordRefusal(
issues: IssueSink,
path: string,
error: unknown,
agent: AiVaultAgent
): void {
if (error instanceof WslTranscriptFsError) {
recordSessionScanIssue(issues, { agent, path, message: error.message })
issues({ agent, path, message: error.message })
}
}
async function cachedProjectDirCwd(
export async function cachedProjectDirCwd(
projectDir: string,
issues: AiVaultScanIssue[],
agent: AiVaultAgent
issues: IssueSink,
agent: AiVaultAgent,
signal?: AbortSignal
): Promise<string | null> {
const cached = projectDirCwdCache.get(projectDir)
if (cached !== undefined) {
@@ -53,7 +56,7 @@ async function cachedProjectDirCwd(
projectDirCwdCache.set(projectDir, cached)
return cached
}
const cwd = await readProjectDirCwd(projectDir, issues, agent)
const cwd = await readProjectDirCwd(projectDir, issues, agent, signal)
if (cwd) {
if (projectDirCwdCache.size >= PROJECT_DIR_CWD_CACHE_MAX) {
const oldest = projectDirCwdCache.keys().next()
@@ -66,62 +69,7 @@ async function cachedProjectDirCwd(
return cwd
}
type InScopeDiscoveryArgs = {
rootDirs: readonly string[]
scopePaths: readonly string[]
limit: number
excludedFilePaths: ReadonlySet<string>
issues: AiVaultScanIssue[]
}
/**
* Fully include the transcripts of cwd-bucket directories (Claude, Pi, CodeBuddy) whose cwd
* falls inside the active workspace/project paths.
*
* Why: these agents keep one directory per cwd, e.g. `~/.claude/projects/<cwd-encoded>/`.
* The global scan is recency-capped, so a project the user hasn't touched recently
* can drop off the list entirely even though the agent's own resume still finds it.
* For scoped panel views we resolve each bucket's cwd cheaply and bypass the cap for
* the ones that belong to the active scope.
*/
export async function discoverInScopeCwdBucketFiles(
layout: CwdBucketLayout,
args: InScopeDiscoveryArgs
): Promise<FileWithMtime[]> {
if (args.scopePaths.length === 0 || args.limit <= 0) {
return []
}
const scopeProjectPrefixes = cwdBucketScopePrefixes(layout, args.scopePaths)
const inScope = createAiVaultScopeCwdMatcher(args.scopePaths)
const collected = new Map<string, FileWithMtime>()
for (const rootDir of args.rootDirs) {
for (const projectDir of await listProjectDirs(
layout,
rootDir,
scopeProjectPrefixes,
args.issues
)) {
if (layout.agent === 'claude') {
const cwd = await cachedProjectDirCwd(projectDir, args.issues, layout.agent)
if (!inScope(cwd, projectDir)) {
continue
}
}
await collectBucketFiles({
agent: layout.agent,
inScope: layout.agent !== 'claude' ? inScope : undefined,
projectDir,
issues: args.issues,
collected,
limit: args.limit,
excludedFilePaths: args.excludedFilePaths
})
}
}
return [...collected.values()].sort((left, right) => right.mtimeMs - left.mtimeMs)
}
function cwdBucketScopePrefixes(
export function cwdBucketScopePrefixes(
layout: CwdBucketLayout,
scopePaths: readonly string[]
): Set<string> {
@@ -141,16 +89,21 @@ function scopePathCandidates(scopePath: string): string[] {
return wslScopePath ? [scopePath, wslScopePath.linuxPath] : [scopePath]
}
async function listProjectDirs(
export async function listProjectDirs(
layout: CwdBucketLayout,
rootDir: string,
scopeProjectPrefixes: ReadonlySet<string>,
issues: AiVaultScanIssue[]
issues: IssueSink,
signal?: AbortSignal,
listed?: () => void
): Promise<string[]> {
let entries
try {
entries = await wslGatedReaddir(rootDir, 'scan')
entries = await wslGatedReaddir(rootDir, 'scan', signal)
throwIfAiVaultScanCancelled(signal)
listed?.()
} catch (err) {
throwIfAiVaultScanCancelled(signal)
recordRefusal(issues, rootDir, err, layout.agent)
return []
}
@@ -161,12 +114,13 @@ async function listProjectDirs(
async function readProjectDirCwd(
projectDir: string,
issues: AiVaultScanIssue[],
agent: AiVaultAgent
issues: IssueSink,
agent: AiVaultAgent,
signal?: AbortSignal
): Promise<string | null> {
const files = await newestTranscriptsInDir(projectDir, issues, agent)
const files = await newestTranscriptsInDir(projectDir, issues, agent, signal)
for (const file of files.slice(0, REPRESENTATIVE_FILE_LIMIT)) {
const cwd = await readFirstCwd(file, issues, agent)
const cwd = await readFirstCwd(file, issues, agent, signal)
if (cwd) {
return cwd
}
@@ -176,18 +130,25 @@ async function readProjectDirCwd(
async function newestTranscriptsInDir(
projectDir: string,
issues: AiVaultScanIssue[],
agent: AiVaultAgent
issues: IssueSink,
agent: AiVaultAgent,
signal?: AbortSignal
): Promise<string[]> {
let entries
try {
entries = await wslGatedReaddir(projectDir, 'scan')
entries = await wslGatedReaddir(projectDir, 'scan', signal)
throwIfAiVaultScanCancelled(signal)
} catch (err) {
throwIfAiVaultScanCancelled(signal)
recordRefusal(issues, projectDir, err, agent)
return []
}
const newest: { path: string; mtimeMs: number }[] = []
for (const entry of entries) {
for (const [index, entry] of entries.entries()) {
if (index % 256 === 0) {
await yieldToEventLoop()
}
throwIfAiVaultScanCancelled(signal)
if (!entry.isFile() || !TRANSCRIPT_EXTENSIONS.has(extname(entry.name).toLowerCase())) {
continue
}
@@ -195,9 +156,10 @@ async function newestTranscriptsInDir(
try {
addBoundedPath(newest, REPRESENTATIVE_FILE_LIMIT, {
path,
mtimeMs: (await wslGatedStat(path, 'scan')).mtimeMs
mtimeMs: (await wslGatedStat(path, 'scan', signal)).mtimeMs
})
} catch (err) {
throwIfAiVaultScanCancelled(signal)
// Best effort: unreadable candidates are ignored here and reported during
// full collection if the project directory proves in-scope.
recordRefusal(issues, path, err, agent)
@@ -208,14 +170,17 @@ async function newestTranscriptsInDir(
async function readFirstCwd(
filePath: string,
issues: AiVaultScanIssue[],
agent: AiVaultAgent
issues: IssueSink,
agent: AiVaultAgent,
signal?: AbortSignal
): Promise<string | null> {
const input = openTranscriptReadStream(filePath, { encoding: 'utf-8' }, 'scan')
const input = openTranscriptReadStream(filePath, { encoding: 'utf-8' }, 'scan', signal)
const lines = createInterface({ input, crlfDelay: Infinity })
let read = 0
try {
throwIfAiVaultScanCancelled(signal)
for await (const line of lines) {
throwIfAiVaultScanCancelled(signal)
if (read++ >= REPRESENTATIVE_CWD_LINE_LIMIT) {
break
}
@@ -225,6 +190,7 @@ async function readFirstCwd(
}
}
} catch (err) {
throwIfAiVaultScanCancelled(signal)
recordRefusal(issues, filePath, err, agent)
return null
} finally {
@@ -236,38 +202,33 @@ async function readFirstCwd(
return null
}
async function collectBucketFiles(args: {
export async function collectBucketFiles(args: {
paths: readonly string[]
agent: AiVaultAgent
inScope?: ReturnType<typeof createAiVaultScopeCwdMatcher>
projectDir: string
issues: AiVaultScanIssue[]
issues: IssueSink
collected: Map<string, FileWithMtime>
limit: number
excludedFilePaths: ReadonlySet<string>
signal?: AbortSignal
}): Promise<void> {
let entries
try {
entries = await wslGatedReaddir(args.projectDir, 'scan')
} catch (err) {
recordRefusal(args.issues, args.projectDir, err, args.agent)
return
}
for (const entry of entries) {
if (!entry.isFile() || !TRANSCRIPT_EXTENSIONS.has(extname(entry.name).toLowerCase())) {
continue
for (const [index, path] of args.paths.entries()) {
if (index % 256 === 0) {
await yieldToEventLoop()
}
const path = join(args.projectDir, entry.name)
throwIfAiVaultScanCancelled(args.signal)
if (args.collected.has(path) || args.excludedFilePaths.has(path)) {
continue
}
try {
if (args.inScope) {
const cwd = await readFirstCwd(path, args.issues, args.agent)
const cwd = await readFirstCwd(path, args.issues, args.agent, args.signal)
if (!args.inScope(cwd, path)) {
continue
}
}
const fileStat = await wslGatedStat(path, 'scan')
const fileStat = await wslGatedStat(path, 'scan', args.signal)
throwIfAiVaultScanCancelled(args.signal)
addBoundedFile(args.collected, args.limit, {
path,
mtimeMs: fileStat.mtimeMs,
@@ -275,7 +236,8 @@ async function collectBucketFiles(args: {
sizeBytes: fileStat.size
})
} catch (err) {
recordSessionScanIssue(args.issues, { agent: args.agent, path, message: errorMessage(err) })
throwIfAiVaultScanCancelled(args.signal)
args.issues({ agent: args.agent, path, message: errorMessage(err) })
}
}
}
@@ -0,0 +1,129 @@
import { mkdtemp, rm, utimes } from 'node:fs/promises'
import { PassThrough } from 'node:stream'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, expect, it, vi } from 'vitest'
import type { AiVaultScanIssue } from '../../shared/ai-vault-types'
import * as transcriptFs from '../native-chat/wsl-transcript-fs-access'
import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-error'
import { aiVaultScanWork, trackAiVaultScanProgress } from './ai-vault-scan-progress'
import { recordSessionScanIssue } from './session-scan-issues'
import { CLAUDE_CWD_BUCKET_LAYOUT, PI_CWD_BUCKET_LAYOUT } from './session-cwd-bucket-layouts'
import { inventoryCwdBucketFiles, selectCwdBucketFiles } from './session-scanner-scope-inventory'
import { writeJsonlFile } from './session-scanner-test-fixtures'
let root: string | null = null
afterEach(async () => {
vi.restoreAllMocks()
if (root) {
await rm(root, { recursive: true, force: true })
}
})
it('replays tied selection, winner exclusion and evicted paths in original root order', async () => {
root = await mkdtemp(join(tmpdir(), 'orca-bucket-replay-'))
const paths = ['b', 'a', 'c', 'winner'].map((name) => join(root!, '--repo--', `${name}.jsonl`))
for (const [index, path] of paths.entries()) {
await writeJsonlFile(path, [{ type: 'session', cwd: '/repo' }])
await utimes(path, index < 2 ? 1 : 2, index < 2 ? 1 : 2)
}
const read = transcriptFs.wslGatedReaddir
vi.spyOn(transcriptFs, 'wslGatedReaddir').mockImplementation(async (path, ...args) =>
(await read(path, ...args)).sort(
(left, right) => paths.indexOf(join(path, left.name)) - paths.indexOf(join(path, right.name))
)
)
const opened = vi.spyOn(transcriptFs, 'openTranscriptReadStream')
const inventory = await inventoryCwdBucketFiles(PI_CWD_BUCKET_LAYOUT, {
rootDirs: [root, root],
scopePaths: ['/repo'],
limit: 1
})
expect(opened).not.toHaveBeenCalled()
const selected = await selectCwdBucketFiles(inventory, {
scopePaths: ['/repo'],
limit: 1,
excludedFilePaths: new Set([paths[3]]),
issues: []
})
expect(selected.map((file) => file.path)).toEqual([paths[2]])
expect(opened.mock.calls.map(([path]) => path)).toEqual([
paths[0],
paths[1],
paths[2],
paths[0],
paths[1]
])
})
it('replays staged errors after global errors, interleaved with bucket errors under the issue cap', async () => {
root = await mkdtemp(join(tmpdir(), 'orca-bucket-issues-'))
const path = join(root, '--repo--', 'a.jsonl')
await writeJsonlFile(path, [{ type: 'session', cwd: '/repo' }])
const read = transcriptFs.wslGatedReaddir
const refused = join(root, 'refused')
vi.spyOn(transcriptFs, 'wslGatedReaddir').mockImplementation((path, ...args) => {
if (path === refused) {
throw new WslTranscriptFsError('unavailable', 'Generated root failure')
}
return read(path, ...args)
})
const inventory = await inventoryCwdBucketFiles(PI_CWD_BUCKET_LAYOUT, {
rootDirs: [root, refused],
scopePaths: ['/repo'],
limit: 1
})
vi.spyOn(transcriptFs, 'wslGatedStat').mockRejectedValue(new Error('Generated stat failure'))
const issues: AiVaultScanIssue[] = []
for (let index = 0; index < 498; index++) {
recordSessionScanIssue(issues, {
agent: 'codex',
path: String(index),
message: 'Generated global failure'
})
}
const expected = [...issues]
recordSessionScanIssue(expected, { agent: 'pi', path, message: 'Generated stat failure' })
recordSessionScanIssue(expected, {
agent: 'pi',
path: refused,
message: 'Generated root failure'
})
await selectCwdBucketFiles(inventory, {
scopePaths: ['/repo'],
limit: 1,
excludedFilePaths: new Set(),
issues
})
expect(issues).toEqual(expected)
expect(issues.at(-1)).toMatchObject({ kind: 'notice' })
})
it('closes representative streams and inventory sources on cancellation before reading', async () => {
root = await mkdtemp(join(tmpdir(), 'orca-bucket-abort-'))
await writeJsonlFile(join(root, '-repo', 'a.jsonl'), [{ cwd: '/repo' }])
const controller = new AbortController()
const input = new PassThrough()
vi.spyOn(transcriptFs, 'openTranscriptReadStream').mockImplementation(() => {
controller.abort()
return input
})
const stop = trackAiVaultScanProgress(controller.signal, () => undefined)
try {
await expect(
inventoryCwdBucketFiles(CLAUDE_CWD_BUCKET_LAYOUT, {
rootDirs: [root],
scopePaths: ['/repo'],
limit: 1,
signal: controller.signal
})
).rejects.toMatchObject({ name: 'AbortError' })
expect(input.destroyed).toBe(true)
expect(aiVaultScanWork(controller.signal)).toMatchObject({
stage: 'finding',
agents: [],
read: { done: 0, total: 0 }
})
} finally {
stop()
}
})
@@ -0,0 +1,150 @@
import { extname, join } from 'node:path'
import { yieldToEventLoop } from '../../shared/event-loop-yield'
import type { AiVaultScanIssue } from '../../shared/ai-vault-types'
import { wslGatedReaddir } from '../native-chat/wsl-transcript-fs-access'
import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation'
import { openAiVaultScanSource } from './ai-vault-scan-progress'
import { SCAN_ISSUE_LIMIT, recordSessionScanIssue } from './session-scan-issues'
import { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match'
import type { CwdBucketLayout } from './session-cwd-bucket-layouts'
import type { FileWithMtime } from './session-scanner-types'
import {
cachedProjectDirCwd,
cwdBucketScopePrefixes,
listProjectDirs,
collectBucketFiles,
recordRefusal,
type IssueSink
} from './session-scanner-scope-discovery'
const TRANSCRIPT_EXTENSIONS = new Set(['.jsonl'])
export type CwdBucketInventory = {
layout: CwdBucketLayout
directories: string[][]
issues: { before: number; issue: AiVaultScanIssue }[]
}
/** Bounded CPU chunks must let cancellation and other scanner requests run. */
export async function yieldScanInventory(index: number, signal?: AbortSignal): Promise<void> {
if (index % 256 === 0) {
await yieldToEventLoop()
}
throwIfAiVaultScanCancelled(signal)
}
type InScopeDiscoveryArgs = {
rootDirs: readonly string[]
scopePaths: readonly string[]
limit: number
excludedFilePaths: ReadonlySet<string>
issues: AiVaultScanIssue[]
signal?: AbortSignal
}
export async function discoverInScopeCwdBucketFiles(
layout: CwdBucketLayout,
args: InScopeDiscoveryArgs
): Promise<FileWithMtime[]> {
const inventory = await inventoryCwdBucketFiles(layout, args)
return selectCwdBucketFiles(inventory, args)
}
/** Freeze names only; winner-dependent header/stat decisions remain in selection. */
export async function inventoryCwdBucketFiles(
layout: CwdBucketLayout,
args: Omit<InScopeDiscoveryArgs, 'excludedFilePaths' | 'issues'>,
issueLimit = SCAN_ISSUE_LIMIT
): Promise<CwdBucketInventory> {
const inventory: CwdBucketInventory = { layout, directories: [], issues: [] }
if (args.scopePaths.length === 0 || args.limit <= 0) {
return inventory
}
const record: IssueSink = (issue) => {
if (inventory.issues.length < issueLimit) {
inventory.issues.push({ before: inventory.directories.length, issue })
}
}
const prefixes = cwdBucketScopePrefixes(layout, args.scopePaths)
const inScope = createAiVaultScopeCwdMatcher(args.scopePaths)
let index = 0
for (const rootDir of args.rootDirs) {
throwIfAiVaultScanCancelled(args.signal)
let close: () => void = () => undefined
try {
const directories = await listProjectDirs(
layout,
rootDir,
prefixes,
record,
args.signal,
() => {
close = openAiVaultScanSource(args.signal, layout.agent)
}
)
for (const directory of directories) {
await yieldScanInventory(index++, args.signal)
if (layout.agent === 'claude') {
const cwd = await cachedProjectDirCwd(directory, record, layout.agent, args.signal)
if (!inScope(cwd, directory)) {
continue
}
}
const paths: string[] = []
try {
const entries = await wslGatedReaddir(directory, 'scan', args.signal)
for (const entry of entries) {
await yieldScanInventory(index++, args.signal)
if (entry.isFile() && TRANSCRIPT_EXTENSIONS.has(extname(entry.name).toLowerCase())) {
paths.push(join(directory, entry.name))
}
}
} catch (err) {
throwIfAiVaultScanCancelled(args.signal)
recordRefusal(record, directory, err, layout.agent)
}
inventory.directories.push(paths)
}
} finally {
close()
}
}
return inventory
}
/** Replay the original selector, interleaving earlier diagnostics at their original positions. */
export async function selectCwdBucketFiles(
inventory: CwdBucketInventory,
args: Pick<
InScopeDiscoveryArgs,
'scopePaths' | 'limit' | 'excludedFilePaths' | 'issues' | 'signal'
>
): Promise<FileWithMtime[]> {
const { layout } = inventory
const collected = new Map<string, FileWithMtime>()
const record: IssueSink = (issue) => recordSessionScanIssue(args.issues, issue)
const inScope = createAiVaultScopeCwdMatcher(args.scopePaths)
let issueIndex = 0
for (let index = 0; index <= inventory.directories.length; index++) {
await yieldScanInventory(index, args.signal)
while (inventory.issues[issueIndex]?.before === index) {
record(inventory.issues[issueIndex++].issue)
}
const paths = inventory.directories[index]
if (!paths) {
break
}
await collectBucketFiles({
paths,
agent: layout.agent,
inScope: layout.agent !== 'claude' ? inScope : undefined,
issues: record,
collected,
limit: args.limit,
excludedFilePaths: args.excludedFilePaths,
signal: args.signal
})
}
inventory.directories.length = 0
inventory.issues.length = 0
return [...collected.values()].sort((left, right) => right.mtimeMs - left.mtimeMs)
}
@@ -352,6 +352,7 @@ it('walks a history larger than the spread-argument limit', async () => {
modifiedAt: new Date(1).toISOString()
}))
const sessions = await scanInScopeSessions({
inventory: [],
discoveries: [
{
agent: 'codex',
+109 -27
View File
@@ -1,3 +1,4 @@
import { SCAN_ISSUE_LIMIT } from './session-scan-issues'
import { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match'
import type { AiVaultScanIssue, AiVaultSession } from '../../shared/ai-vault-types'
import type { ExecutionHostId } from '../../shared/execution-host'
@@ -16,7 +17,13 @@ import { parseSessionCandidates } from './session-scanner-candidate-parse'
import { sessionSortTime } from './session-scanner-accumulator'
import { discoveryFileCandidates, dropCodexHardlinkAliases } from './session-scanner-candidates'
import type { SessionParseStats } from './session-scanner-parse-cache'
import { discoverInScopeCwdBucketFiles } from './session-scanner-scope-discovery'
import {
inventoryCwdBucketFiles,
selectCwdBucketFiles,
yieldScanInventory,
type CwdBucketInventory
} from './session-scanner-scope-inventory'
import { aiVaultScanWorkload, noteAiVaultScanWorkChanged } from './ai-vault-scan-progress'
import type {
AiVaultScanOptions,
FileWithMtime,
@@ -28,6 +35,7 @@ import { sidecarUnchanged } from './session-sidecar-stat'
import { prioritizeAntigravityTranscriptCandidates } from './antigravity-transcript-candidates'
type ScopeScanArgs = {
inventory: CwdBucketInventory[]
discoveries: readonly SessionFileDiscovery[]
options: AiVaultScanOptions
limit: number
@@ -50,24 +58,29 @@ export async function scanInScopeSessions(args: ScopeScanArgs): Promise<AiVaultS
return []
}
const candidates: SessionFileCandidate[] = []
for (const source of Object.values(AI_VAULT_SCOPE_CWD_SOURCES)) {
if (source.kind !== 'bucket') {
continue
}
const { layout } = source
const files = await discoverInScopeCwdBucketFiles(layout, {
rootDirs: args.discoveries
.filter((discovery) => discovery.agent === layout.agent)
.map((discovery) => discovery.rootDir),
const workload = aiVaultScanWorkload(args.options.signal)
for (const inventory of args.inventory) {
const files = await selectCwdBucketFiles(inventory, {
scopePaths,
limit: args.limit,
excludedFilePaths: args.alreadyParsedFilePaths,
issues: args.issues
issues: args.issues,
signal: args.options.signal
})
const selected = new Set(files.map((file) => file.path))
let index = 0
for (const path of workload?.bucketPaths(inventory.layout.agent) ?? []) {
await yieldScanInventory(index++, args.options.signal)
if (!selected.has(path)) {
workload?.settlePath(inventory.layout.agent, path, 'bucket', false)
}
}
for (const file of files) {
candidates.push({ agent: layout.agent, file, codexHome: null })
candidates.push({ agent: inventory.layout.agent, file, codexHome: null })
}
}
args.inventory.length = 0
noteAiVaultScanWorkChanged(args.options.signal)
const walked = await walkInScopeSessions(args, scopePaths).finally(clearSessionScopeCwdKnowledge)
if (candidates.length === 0) {
return walked
@@ -80,7 +93,8 @@ export async function scanInScopeSessions(args: ScopeScanArgs): Promise<AiVaultS
executionHostId: args.executionHostId,
issues: args.issues,
parseStats: args.parseStats,
signal: args.options.signal
signal: args.options.signal,
settled: (candidate, parsed) => workload?.settle(candidate, 'bucket', parsed)
})
return [...bucketed, ...walked]
}
@@ -96,7 +110,7 @@ async function walkInScopeSessions(
): Promise<AiVaultSession[]> {
const signal = args.options.signal
const inScope = createAiVaultScopeCwdMatcher(scopePaths)
const candidates = walkCandidates(args, scopePaths)
const candidates = await walkCandidates(args, scopePaths)
await lookupScopeCwds(candidates, signal)
const sessions = await parseSessionCandidates({
candidates,
@@ -108,6 +122,7 @@ async function walkInScopeSessions(
antigravityWorkspaceResolver: args.antigravityWorkspaceResolver,
signal,
listed: false,
settled: (candidate, parsed) => aiVaultScanWorkload(signal)?.settle(candidate, 'walk', parsed),
selectCandidate: (candidate) => mayBeInScope(candidate, inScope, args.platform, signal),
admitSession: (session) => inScope(session.cwd, session.filePath)
})
@@ -116,39 +131,106 @@ async function walkInScopeSessions(
.slice(0, args.limit)
}
function walkCandidates(
args: ScopeScanArgs,
/** Pure membership enumeration; pruning and ranking remain execution decisions. */
async function walkInventory(
args: Pick<ScopeScanArgs, 'discoveries' | 'options'>,
scopePaths: readonly string[]
): SessionFileCandidate[] {
): Promise<{ candidates: SessionFileCandidate[]; seenPaths: Set<string> }> {
const all: SessionFileCandidate[] = []
const seenPaths = new Set<string>()
let index = 0
for (const discovery of args.discoveries) {
if (!scopeWalksAgentFiles(discovery.agent, scopePaths)) {
continue
}
const files: FileWithMtime[] = []
for (const file of [...discovery.files, ...(discovery.scopeFiles ?? [])]) {
if (!seenPaths.has(file.path)) {
seenPaths.add(file.path)
files.push(file)
for (const collection of [discovery.files, discovery.scopeFiles ?? []]) {
for (const file of collection) {
await yieldScanInventory(index++, args.options.signal)
if (!seenPaths.has(file.path)) {
seenPaths.add(file.path)
files.push(file)
}
}
}
// Why: one item per push; a history-sized list spread into push() exceeds V8's argument limit.
for (const candidate of discoveryFileCandidates(discovery, files, args.options)) {
all.push(candidate)
}
}
return { candidates: dropCodexHardlinkAliases(all), seenPaths }
}
async function walkCandidates(
args: ScopeScanArgs,
scopePaths: readonly string[]
): Promise<SessionFileCandidate[]> {
const { candidates, seenPaths } = await walkInventory(args, scopePaths)
pruneSessionScopeCwds(seenPaths)
// Aliases collapse with the listed files still present, so a listed rollout's
// hardlink in another Codex home is not parsed again.
const workload = aiVaultScanWorkload(args.options.signal)
const remaining: SessionFileCandidate[] = []
for (const [index, candidate] of candidates.entries()) {
await yieldScanInventory(index, args.options.signal)
if (args.alreadyParsedFilePaths.has(candidate.file.path)) {
workload?.settle(candidate, 'walk', false)
} else {
remaining.push(candidate)
}
}
return prioritizeAntigravityTranscriptCandidates(
dropCodexHardlinkAliases(all).filter(
(candidate) => !args.alreadyParsedFilePaths.has(candidate.file.path)
),
remaining,
(candidate) => candidate.agent === 'antigravity'
)
}
export async function prepareScopeScan(
args: Pick<ScopeScanArgs, 'discoveries' | 'options' | 'limit'>
): Promise<CwdBucketInventory[]> {
const scopePaths = args.options.scopePaths ?? []
if (scopePaths.length === 0) {
return []
}
const workload = aiVaultScanWorkload(args.options.signal)
// H is temporary: walk-only keys settle directly, without a second retained registry.
if (workload) {
const { candidates } = await walkInventory(args, scopePaths)
for (const [index, candidate] of candidates.entries()) {
await yieldScanInventory(index, args.options.signal)
workload.register(candidate, 'walk')
}
}
const inventories: CwdBucketInventory[] = []
let issueLimit = SCAN_ISSUE_LIMIT
let index = 0
for (const source of Object.values(AI_VAULT_SCOPE_CWD_SOURCES)) {
if (source.kind !== 'bucket') {
continue
}
const inventory = await inventoryCwdBucketFiles(
source.layout,
{
rootDirs: args.discoveries
.filter((discovery) => discovery.agent === source.layout.agent)
.map((discovery) => discovery.rootDir),
scopePaths,
limit: args.limit,
signal: args.options.signal
},
issueLimit
)
issueLimit -= inventory.issues.length
inventories.push(inventory)
if (workload) {
for (const paths of inventory.directories) {
for (const path of paths) {
await yieldScanInventory(index++, args.options.signal)
workload.registerPath(source.layout.agent, path, 'bucket')
}
}
}
}
return inventories
}
// Per lookup batch: one database round trip, then a turn for the event loop.
const SCOPE_LOOKUP_BATCH = 500
+17 -2
View File
@@ -15,7 +15,9 @@ import { sessionCandidatesFromDiscoveries } from './session-scanner-candidates'
import { parseSessionCandidates } from './session-scanner-candidate-parse'
import { createSessionParseStats } from './session-scanner-parse-cache'
import { recordSessionScanIssue } from './session-scan-issues'
import { scanInScopeSessions } from './session-scanner-scope-walk'
import { aiVaultScanWorkload, noteAiVaultScanReading } from './ai-vault-scan-progress'
import { yieldScanInventory } from './session-scanner-scope-inventory'
import { prepareScopeScan, scanInScopeSessions } from './session-scanner-scope-walk'
import { discoverAiVaultSessionSources } from './session-scanner-source-discovery'
import { cursorChatMetaRefusals, withCursorChatMetaScan } from './session-scanner-cursor-chat-meta'
import type { AiVaultScanOptions } from './session-scanner-types'
@@ -75,6 +77,17 @@ async function scanAiVaultSessionStores(options: AiVaultScanOptions): Promise<Ai
const candidates = await sessionCandidatesFromDiscoveries(discoveries, options)
const parseCandidates = candidates.slice(0, limit * SESSION_PARSE_CANDIDATE_MULTIPLIER)
const workload = aiVaultScanWorkload(options.signal)
if (workload) {
for (const [index, candidate] of parseCandidates.entries()) {
await yieldScanInventory(index, options.signal)
workload.register(candidate, 'global')
}
}
const inventory = await prepareScopeScan({ discoveries, options, limit })
throwIfAiVaultScanCancelled(options.signal)
noteAiVaultScanReading(options.signal)
const parsedSessions = await parseSessionCandidates({
candidates: parseCandidates,
limit,
@@ -83,7 +96,8 @@ async function scanAiVaultSessionStores(options: AiVaultScanOptions): Promise<Ai
issues,
parseStats,
signal: options.signal,
antigravityWorkspaceResolver
antigravityWorkspaceResolver,
settled: (candidate, parsed) => workload?.settle(candidate, 'global', parsed)
})
const cappedSessions = dedupeScannedSessions(parsedSessions)
@@ -91,6 +105,7 @@ async function scanAiVaultSessionStores(options: AiVaultScanOptions): Promise<Ai
.slice(0, limit)
const scopeSessions = await scanInScopeSessions({
inventory,
discoveries,
options,
limit,
@@ -126,7 +126,7 @@ it('keeps real rows through partial rescans, detached keys and failed retries',
}
expect(view.getByText('12 files')).toBeTruthy()
expect(panel.container.querySelectorAll('.animate-spin').length).toBe(1)
expect(view.getByText('Reading', { exact: false })).toBeTruthy()
expect(view.getByText('Checking', { exact: false })).toBeTruthy()
// Finished before the list it produced is applied: still one row, now settled.
act(() => push({ scanKey: 'worktree-a', progress: next }))
expect(view.getByText('Scanned', { exact: false })).toBeTruthy()
@@ -161,3 +161,18 @@ it('keeps real rows through partial rescans, detached keys and failed retries',
act(() => push({ scanKey: 'worktree-c', progress: { ...next, agents: [FINISHED.agents[0]!] } }))
expect(rowCount()).toBe(1)
})
it('shows settled candidate work and keeps older snapshots without skipped compatible', () => {
const reading: AiVaultScanProgress = {
...FINISHED,
stage: 'reading',
settledAt: null,
read: { done: 3, total: 10, skipped: 4 }
}
const panel = render(<AiVaultScanProgressStrip progress={reading} />)
expect(screen.getByRole('button').textContent).toContain('Checking 7 / 10')
panel.rerender(
<AiVaultScanProgressStrip progress={{ ...reading, read: { done: 3, total: 10 } }} />
)
expect(screen.getByRole('button').textContent).toContain('Checking 3 / 10')
})
@@ -61,8 +61,8 @@ function stripSentence(progress: AiVaultScanProgress): string {
)
case 'reading':
return translate(
'auto.components.right.sidebar.AiVaultScanProgressStrip.reading',
'Reading {{value0}} files',
'auto.components.right.sidebar.AiVaultScanProgressStrip.checking',
'Checking {{value0}}',
{ value0 }
)
case 'finished':
@@ -81,7 +81,7 @@ function foundCount(agent: AiVaultScanAgentProgress): number {
function StripText({ progress, active }: StripActivity): React.JSX.Element {
const emphasis =
active && progress.stage === 'reading'
? `${formatCount(progress.read.done)} / ${formatCount(progress.read.total)}`
? `${formatCount(progress.read.done + (progress.read.skipped ?? 0))} / ${formatCount(progress.read.total)}`
: fileCountLabel(progress.agents.reduce((sum, agent) => sum + foundCount(agent), 0))
const sentence =
active || progress.stage === 'finished' ? stripSentence(progress) : EMPHASIS_TOKEN
+1 -1
View File
@@ -14824,7 +14824,7 @@
"databaseCount_one": "{{value0}} database",
"databaseCount_other": "{{value0}} databases",
"scanning": "Scanning {{value0}}",
"reading": "Reading {{value0}} files",
"checking": "Checking {{value0}}",
"scanned": "Scanned {{value0}}"
}
}
+1 -1
View File
@@ -14487,7 +14487,7 @@
"databaseCount_one": "{{value0}} base de datos",
"databaseCount_other": "{{value0}} bases de datos",
"scanning": "Analizando {{value0}}",
"reading": "Leyendo {{value0}} archivos",
"checking": "Comprobando {{value0}}",
"scanned": "Analizados {{value0}}"
}
}
+1 -1
View File
@@ -14600,7 +14600,7 @@
"databaseCount_one": "{{value0}} base de données",
"databaseCount_other": "{{value0}} bases de données",
"scanning": "Analyse de {{value0}}",
"reading": "Lecture de {{value0}} fichiers",
"checking": "Vérification {{value0}}",
"scanned": "{{value0}} analysés"
}
}
+1 -1
View File
@@ -14492,7 +14492,7 @@
"databaseCount_one": "{{value0}} 個のデータベース",
"databaseCount_other": "{{value0}} 個のデータベース",
"scanning": "{{value0}}をスキャン中",
"reading": "{{value0}} ファイルを読み込み中",
"checking": "確認中 {{value0}}",
"scanned": "{{value0}}をスキャン済み"
}
}
+1 -1
View File
@@ -14492,7 +14492,7 @@
"databaseCount_one": "데이터베이스 {{value0}}개",
"databaseCount_other": "데이터베이스 {{value0}}개",
"scanning": "{{value0}} 스캔 중",
"reading": "파일 {{value0}}개 읽는 중",
"checking": "확인 중 {{value0}}",
"scanned": "{{value0}} 스캔함"
}
}
-1
View File
@@ -10369,7 +10369,6 @@
"databaseCount_one": "{{value0}} banco de dados",
"databaseCount_other": "{{value0}} bancos de dados",
"scanning": "Verificando {{value0}}",
"reading": "Lendo {{value0}} arquivos",
"scanned": "Verificação concluída: {{value0}}"
}
}
+1 -1
View File
@@ -14492,7 +14492,7 @@
"databaseCount_one": "{{value0}} 个数据库",
"databaseCount_other": "{{value0}} 个数据库",
"scanning": "正在扫描 {{value0}}",
"reading": "正在读取 {{value0}} 个文件",
"checking": "正在检查 {{value0}}",
"scanned": "已扫描 {{value0}}"
}
}
+7 -2
View File
@@ -9,14 +9,19 @@ export type AiVaultScanAgentProgress = {
databases: number
/** A folder or database of this agent is still being listed. */
walking: boolean
/** Absent until the scan freezes its candidate workload. */
read?: AiVaultScanReadingProgress
}
/** Terminal candidate outcomes; skipped candidates needed no parser invocation. */
export type AiVaultScanReadingProgress = { done: number; total: number; skipped?: number }
/** The scanner process's view of the scan it is running. */
export type AiVaultScanWork = {
stage: 'finding' | 'reading'
agents: AiVaultScanAgentProgress[]
/** Transcripts read so far, out of the ones this scan will read. */
read: { done: number; total: number }
/** Fixed candidate workload; settled work is done + skipped. */
read: AiVaultScanReadingProgress
}
/** One local Session History scan entry, as main pushes it to windows. */