perf(ai-vault): pipeline fresh transcript discovery with bounded I/O (#27009)

Overlap transcript stat/sidecar reads in ordered batches (4 workers per root, 8 global slots); WSL stays sequential.

Fixes #27008

Co-authored-by: outof-place <contact@outofplace.space>

src/main/ai-vault/session-newest-files.ts

@@ -16,11 +16,7 @@ export class SessionNewestFiles {
       this.files.push(file)
       return
     }
-    if (this.limit <= 0) {
-      return
-    }
-    const last = this.files.at(-1)
-    if (this.files.length >= this.limit && last && file.mtimeMs <= last.mtimeMs) {
+    if (!this.wouldRetain(file.mtimeMs)) {
       return
     }
     let low = 0
@@ -39,6 +35,14 @@ export class SessionNewestFiles {
     }
   }
 
+  wouldRetain(mtimeMs: number): boolean {
+    if (this.limit <= 0) {
+      return false
+    }
+    const last = this.files.at(-1)
+    return !(this.files.length >= this.limit && last && mtimeMs <= last.mtimeMs)
+  }
+
   get size(): number {
     return this.files.length
   }

src/main/ai-vault/session-scanner-discovery-pipeline.test.ts

@@ -0,0 +1,273 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type * as FsPromises from 'node:fs/promises'
+import { join } from 'node:path'
+import { tmpdir } from 'node:os'
+import type { AiVaultScanIssue } from '../../shared/ai-vault-types'
+
+const fsMocks = vi.hoisted(() => ({ readdir: vi.fn(), stat: vi.fn() }))
+vi.mock('node:fs/promises', async (importOriginal) => ({
+  ...(await importOriginal<typeof FsPromises>()),
+  readdir: fsMocks.readdir,
+  stat: fsMocks.stat
+}))
+
+import { discoverFiles } from './session-scanner-discovery'
+
+const fs = await vi.importActual<typeof FsPromises>('node:fs/promises')
+let root: string
+
+beforeEach(async () => {
+  fsMocks.readdir.mockReset().mockImplementation(fs.readdir)
+  fsMocks.stat.mockReset().mockImplementation(fs.stat)
+  root = await fs.mkdtemp(join(tmpdir(), 'orca-fresh-discovery-'))
+})
+
+afterEach(async () => {
+  vi.restoreAllMocks()
+  await fs.rm(root, { recursive: true, force: true })
+})
+
+async function createFiles(directory: string, count: number): Promise<string[]> {
+  await fs.mkdir(directory, { recursive: true })
+  await Promise.all(
+    Array.from({ length: count }, (_, index) =>
+      fs.writeFile(join(directory, `${index.toString().padStart(3, '0')}.jsonl`), '{}\n')
+    )
+  )
+  return (await fs.readdir(directory)).map((name) => join(directory, name))
+}
+
+function scan(scanRoot = root, issues: AiVaultScanIssue[] = [], limit = 100) {
+  return discoverFiles({
+    rootDir: scanRoot,
+    issues,
+    limit,
+    agent: 'claude',
+    extensions: ['.jsonl']
+  })
+}
+
+describe('fresh discovery pipeline', () => {
+  it('formats timestamps only for retained files while still reading every candidate', async () => {
+    await createFiles(root, 35)
+    fsMocks.stat.mockImplementation(async (path: string) =>
+      Object.assign(await fs.stat(path), { mtimeMs: 1_000 })
+    )
+    const format = vi.spyOn(Date.prototype, 'toISOString')
+    expect((await scan(root, [], 3)).files).toHaveLength(3)
+    expect(format).toHaveBeenCalledTimes(3)
+    expect(fsMocks.stat).toHaveBeenCalledTimes(35)
+    format.mockClear()
+    expect((await scan(root, [], 0)).files).toEqual([])
+    expect(format).not.toHaveBeenCalled()
+    expect(fsMocks.stat).toHaveBeenCalledTimes(70)
+    expect((await scan(root, [], Infinity)).files).toHaveLength(35)
+    expect(format).toHaveBeenCalledTimes(35)
+  })
+
+  it('reports invalid timestamps even for empty requests, after unavailable sidecar issues', async () => {
+    const [file] = await createFiles(root, 1)
+    fsMocks.stat.mockImplementation(async (path: string) => {
+      if (path.endsWith('.meta')) {
+        throw Object.assign(new Error('metadata refused'), { code: 'EACCES' })
+      }
+      return Object.assign(await fs.stat(path), { mtimeMs: Infinity })
+    })
+    const issues: AiVaultScanIssue[] = []
+    const discovery = await discoverFiles({
+      rootDir: root,
+      issues,
+      limit: 0,
+      agent: 'claude',
+      extensions: ['.jsonl'],
+      contentDependencyPath: (path) => `${path}.meta`
+    })
+    expect(discovery.files).toEqual([])
+    expect(issues).toEqual([
+      {
+        agent: 'claude',
+        path: `${file}.meta`,
+        message: 'Session metadata could not be read this scan.'
+      },
+      { agent: 'claude', path: file, message: expect.stringMatching(/Invalid (?:time value|Date)/) }
+    ])
+  })
+
+  it('preserves traversal ties when later metadata completes first, including the partial batch', async () => {
+    const paths = await createFiles(root, 131)
+    let releaseFirst = () => {}
+    const firstBlocked = new Promise<void>((resolve) => {
+      releaseFirst = resolve
+    })
+    let reachedLater = () => {}
+    const laterRead = new Promise<void>((resolve) => {
+      reachedLater = resolve
+    })
+    fsMocks.stat.mockImplementation(async (path: string) => {
+      if (path === paths[0]) {
+        await firstBlocked
+      }
+      if (path === paths[15]) {
+        reachedLater()
+      }
+      return Object.assign(await fs.stat(path), { mtimeMs: 1_000 })
+    })
+
+    const discovery = scan(root, [], 130)
+    try {
+      await laterRead
+    } finally {
+      releaseFirst()
+    }
+    expect((await discovery).files.map((file) => file.path)).toEqual(paths.slice(0, 130))
+    expect(fsMocks.stat).toHaveBeenCalledTimes(paths.length)
+  })
+
+  it('bounds directory and metadata reads across simultaneous roots', async () => {
+    const roots = Array.from({ length: 8 }, (_, index) => join(root, `root-${index}`))
+    await Promise.all(roots.map((directory) => createFiles(directory, 70)))
+    let active = 0
+    let peak = 0
+    async function tracked<T>(read: () => Promise<T>): Promise<T> {
+      active++
+      peak = Math.max(peak, active)
+      try {
+        await new Promise<void>((resolve) => setImmediate(resolve))
+        return await read()
+      } finally {
+        active--
+      }
+    }
+    fsMocks.readdir.mockImplementation((path: string, options: { withFileTypes: true }) =>
+      tracked(() => fs.readdir(path, options))
+    )
+    fsMocks.stat.mockImplementation((path: string) => tracked(() => fs.stat(path)))
+
+    const results = await Promise.all(roots.map((directory) => scan(directory, [], 3)))
+    expect(results.every((result) => result.files.length === 3)).toBe(true)
+    expect(peak).toBeGreaterThan(1)
+    expect(peak).toBeLessThanOrEqual(8)
+    expect(active).toBe(0)
+  })
+
+  it('sees file, sidecar, and nested membership changes on the next scan without notifications', async () => {
+    const [file] = await createFiles(root, 1)
+    if (!file) {
+      throw new Error('missing fixture')
+    }
+    const sidecar = join(root, 'metadata.json')
+    await fs.writeFile(sidecar, '{}')
+    const args = {
+      rootDir: root,
+      issues: [],
+      limit: 100,
+      agent: 'claude' as const,
+      extensions: ['.jsonl'],
+      contentDependencyPath: () => sidecar
+    }
+    const first = await discoverFiles(args)
+    await fs.writeFile(file, '{"new":true}\n')
+    await fs.utimes(file, 2_000, 2_000)
+    await fs.writeFile(sidecar, '{"new":true}')
+    await fs.utimes(sidecar, 3_000, 3_000)
+    const [added] = await createFiles(join(root, 'nested'), 1)
+    const second = await discoverFiles(args)
+    expect(second.files.map((entry) => entry.path)).toContain(added)
+    expect(second.files.find((entry) => entry.path === file)).toMatchObject({
+      sizeBytes: 13,
+      mtimeMs: 2_000_000,
+      sidecar: { path: sidecar, mtimeMs: 3_000_000, sizeBytes: 12 }
+    })
+    expect(first.files[0]?.sizeBytes).toBe(3)
+    await fs.rm(file)
+    expect((await discoverFiles(args)).files.map((entry) => entry.path)).not.toContain(file)
+  })
+
+  it('keeps issue ordering at the cap despite reversed metadata failures', async () => {
+    const paths = await createFiles(root, 65)
+    let releaseFirst = () => {}
+    const blocked = new Promise<void>((resolve) => {
+      releaseFirst = resolve
+    })
+    let reachedLater = () => {}
+    const later = new Promise<void>((resolve) => {
+      reachedLater = resolve
+    })
+    fsMocks.stat.mockImplementation(async (path: string) => {
+      if (path === paths[0]) {
+        await blocked
+      }
+      if (path === paths[15]) {
+        reachedLater()
+      }
+      throw new Error(`unreadable ${path}`)
+    })
+    const issues: AiVaultScanIssue[] = Array.from({ length: 498 }, (_, index) => ({
+      agent: 'claude',
+      path: `previous-${index}`,
+      message: 'previous issue'
+    }))
+    const discovery = scan(root, issues)
+    try {
+      await later
+    } finally {
+      releaseFirst()
+    }
+    expect((await discovery).files).toEqual([])
+    expect(issues).toHaveLength(500)
+    expect(issues[498]?.path).toBe(paths[0])
+    expect(issues[499]?.kind).toBe('notice')
+  })
+
+  it('reports one unavailable sidecar in traversal order and distinguishes missing metadata', async () => {
+    const paths = await createFiles(root, 5)
+    fsMocks.stat.mockImplementation(async (path: string) => {
+      if (path.endsWith('.meta')) {
+        throw Object.assign(new Error('metadata refused'), {
+          code: path === `${paths[0]}.meta` ? 'ENOENT' : 'EACCES'
+        })
+      }
+      return fs.stat(path)
+    })
+    const issues: AiVaultScanIssue[] = []
+    const discovery = await discoverFiles({
+      rootDir: root,
+      issues,
+      limit: 100,
+      agent: 'claude',
+      extensions: ['.jsonl'],
+      contentDependencyPath: (path) => `${path}.meta`
+    })
+    expect(discovery.files.find((entry) => entry.path === paths[0])?.sidecar).toBe('none')
+    expect(discovery.files.filter((entry) => entry.sidecar === 'unknown')).toHaveLength(4)
+    expect(issues).toEqual([
+      {
+        agent: 'claude',
+        path: `${paths[1]}.meta`,
+        message: 'Session metadata could not be read this scan.'
+      }
+    ])
+  })
+
+  it.skipIf(process.platform === 'win32')(
+    'reads retargeted and retargeted-back roots from disk',
+    async () => {
+      const [a] = await createFiles(join(root, 'a'), 1)
+      const [b] = await createFiles(join(root, 'b'), 1)
+      const alias = join(root, 'alias')
+      await fs.symlink(join(root, 'a'), alias, 'dir')
+      expect((await scan(alias)).files[0]?.sizeBytes).toBe(3)
+      if (!a || !b) {
+        throw new Error('missing fixture')
+      }
+      await fs.writeFile(b, '{"target":"b"}\n')
+      await fs.unlink(alias)
+      await fs.symlink(join(root, 'b'), alias, 'dir')
+      expect((await scan(alias)).files[0]?.sizeBytes).toBe(15)
+      await fs.writeFile(a, '{"target":"a"}\n')
+      await fs.unlink(alias)
+      await fs.symlink(join(root, 'a'), alias, 'dir')
+      expect((await scan(alias)).files[0]?.sizeBytes).toBe(15)
+    }
+  )
+})

src/main/ai-vault/session-scanner-discovery.ts

@@ -1,14 +1,34 @@
 import type { Dirent } from 'node:fs'
 import { extname, join } from 'node:path'
+import { mapWithConcurrency } from '../../shared/map-with-concurrency'
+import { PrioritySemaphore } from '../../shared/priority-semaphore'
+import { isWslUncPath } from '../../shared/wsl-paths'
 import { SessionNewestFiles } from './session-newest-files'
 import type { SessionSidecarObservation } from './session-sidecar-stat'
 import type { AiVaultAgent, AiVaultScanIssue } from '../../shared/ai-vault-types'
 import { wslGatedReaddir, wslGatedStat } from '../native-chat/wsl-transcript-fs-access'
 import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate'
 import { recordSessionScanIssue } from './session-scan-issues'
-import type { SessionFileDiscovery } from './session-scanner-types'
+import type { FileWithMtime, SessionFileDiscovery } from './session-scanner-types'
 import { errorMessage } from './session-scanner-values'
 
+const NATIVE_DISCOVERY_CONCURRENCY = 4
+const NATIVE_DISCOVERY_BATCH_SIZE = 16
+const nativeDiscoverySlots = new PrioritySemaphore(8)
+
+type SessionFileObservation =
+  | { file: Omit<FileWithMtime, 'modifiedAt'>; sidecarPath?: string }
+  | { issue: { path: string; message: string } }
+
+async function withNativeDiscoverySlot<T>(read: () => Promise<T>): Promise<T> {
+  const release = await nativeDiscoverySlots.acquire(0)
+  try {
+    return await read()
+  } finally {
+    release()
+  }
+}
+
 export async function discoverFiles(args: {
   rootDir: string
   limit: number
@@ -21,50 +41,103 @@ export async function discoverFiles(args: {
 }): Promise<SessionFileDiscovery> {
   const files = new SessionNewestFiles(args.limit)
   let refusedSidecar = false
-  try {
-    await forEachSessionFile(
-      args.rootDir,
-      args.agent,
-      args.issues,
-      {
-        extensions: new Set(args.extensions),
-        filePredicate: args.filePredicate,
-        directoryPredicate: args.directoryPredicate
-      },
-      async (path) => {
-        try {
-          const fileStat = await wslGatedStat(path, 'scan')
-          const sidecarPath = await args.contentDependencyPath?.(path)
-          const sidecar = await observeSessionSidecar(sidecarPath)
-          if (sidecar === 'unknown' && !refusedSidecar) {
-            // One issue per root: a refused sibling is a property of the tree,
-            // not of each transcript that happens to point at it.
-            refusedSidecar = true
-            recordSessionScanIssue(args.issues, {
-              agent: args.agent,
-              path: sidecarPath ?? args.rootDir,
-              message: 'Session metadata could not be read this scan.'
-            })
-          }
-          files.add({
-            path,
-            mtimeMs: fileStat.mtimeMs,
-            modifiedAt: new Date(fileStat.mtimeMs).toISOString(),
-            sizeBytes: fileStat.size,
-            sidecar,
-            dev: fileStat.dev,
-            ino: fileStat.ino,
-            nlink: fileStat.nlink
-          })
-        } catch (err) {
-          recordSessionScanIssue(args.issues, {
-            agent: args.agent,
-            path,
-            message: errorMessage(err)
-          })
-        }
+  let paths: string[] = []
+  let pending: Promise<SessionFileObservation[]> | null = null
+  const native = !isWslUncPath(args.rootDir)
+
+  function consume(observations: readonly SessionFileObservation[]): void {
+    for (const observation of observations) {
+      if ('issue' in observation) {
+        recordSessionScanIssue(args.issues, { agent: args.agent, ...observation.issue })
+        continue
       }
+      if (observation.file.sidecar === 'unknown' && !refusedSidecar) {
+        refusedSidecar = true
+        recordSessionScanIssue(args.issues, {
+          agent: args.agent,
+          path: observation.sidecarPath ?? args.rootDir,
+          message: 'Session metadata could not be read this scan.'
+        })
+      }
+      try {
+        const modifiedAt = new Date(observation.file.mtimeMs)
+        if (Number.isNaN(modifiedAt.getTime()) || files.wouldRetain(observation.file.mtimeMs)) {
+          files.add({ ...observation.file, modifiedAt: modifiedAt.toISOString() })
+        }
+      } catch (err) {
+        recordSessionScanIssue(args.issues, {
+          agent: args.agent,
+          path: observation.file.path,
+          message: errorMessage(err)
+        })
+      }
+    }
+  }
+
+  async function finishPending(): Promise<void> {
+    if (pending) {
+      consume(await pending)
+      pending = null
+    }
+  }
+
+  function queueNativeFile(path: string): void | Promise<void> {
+    paths.push(path)
+    if (paths.length < NATIVE_DISCOVERY_BATCH_SIZE) {
+      return
+    }
+    return finishPending().then(() => {
+      pending = readBatch(paths)
+      paths = []
+    })
+  }
+
+  async function readBatch(batch: readonly string[]): Promise<SessionFileObservation[]> {
+    const groups: string[][] = []
+    const groupSize = Math.ceil(batch.length / NATIVE_DISCOVERY_CONCURRENCY)
+    for (let index = 0; index < batch.length; index += groupSize) {
+      groups.push(batch.slice(index, index + groupSize))
+    }
+    const observations = await mapWithConcurrency(groups, NATIVE_DISCOVERY_CONCURRENCY, (group) =>
+      withNativeDiscoverySlot(async () => {
+        const files: SessionFileObservation[] = []
+        for (const path of group) {
+          files.push(await observeSessionFile(path, args.contentDependencyPath))
+        }
+        return files
+      })
     )
+    return observations.flat()
+  }
+
+  try {
+    try {
+      await forEachSessionFile(
+        args.rootDir,
+        args.agent,
+        args.issues,
+        {
+          extensions: new Set(args.extensions),
+          filePredicate: args.filePredicate,
+          directoryPredicate: args.directoryPredicate,
+          ...(native
+            ? {
+                readDirectory: (path: string) =>
+                  withNativeDiscoverySlot(() => wslGatedReaddir(path, 'scan'))
+              }
+            : {})
+        },
+        native
+          ? queueNativeFile
+          : async (path) => consume([await observeSessionFile(path, args.contentDependencyPath)])
+      )
+    } finally {
+      // One read batch overlaps directory traversal; commit in traversal order.
+      await finishPending()
+      if (paths.length) {
+        consume(await readBatch(paths))
+      }
+    }
   } catch (err) {
     // Why: discoverAiVaultSessionSources fans out with Promise.all, so one
     // stalled distro would otherwise reject the whole vault scan — including
@@ -82,6 +155,31 @@ export async function discoverFiles(args: {
   return { agent: args.agent, rootDir: args.rootDir, files: files.newest() }
 }
 
+async function observeSessionFile(
+  path: string,
+  contentDependencyPath?: (path: string) => string | undefined | Promise<string | undefined>
+): Promise<SessionFileObservation> {
+  try {
+    const fileStat = await wslGatedStat(path, 'scan')
+    const sidecarPath = contentDependencyPath ? await contentDependencyPath(path) : undefined
+    const sidecar = sidecarPath ? await observeSessionSidecar(sidecarPath) : 'none'
+    return {
+      file: {
+        path,
+        mtimeMs: fileStat.mtimeMs,
+        sizeBytes: fileStat.size,
+        sidecar,
+        dev: fileStat.dev,
+        ino: fileStat.ino,
+        nlink: fileStat.nlink
+      },
+      sidecarPath
+    }
+  } catch (err) {
+    return { issue: { path, message: errorMessage(err) } }
+  }
+}
+
 /**
  * A sibling that cannot be statted is not "no sibling": it must not take the
  * transcript down with it, and it must not read as absent either, or the parse
@@ -144,7 +242,7 @@ export async function forEachSessionFile(
   agent: AiVaultAgent,
   issues: AiVaultScanIssue[],
   options: SessionFileWalkOptions,
-  onFile: (path: string) => Promise<void>,
+  onFile: (path: string) => void | Promise<void>,
   depth = 0
 ): Promise<void> {
   options.signal?.throwIfAborted()
@@ -179,7 +277,10 @@ export async function forEachSessionFile(
       options.extensions.has(extname(entry.name).toLowerCase()) &&
       (options.filePredicate?.(fullPath) ?? true)
     ) {
-      await onFile(fullPath)
+      const observation = onFile(fullPath)
+      if (observation) {
+        await observation
+      }
     }
   }
 }