diff --git a/src/main/ai-vault/session-newest-files.ts b/src/main/ai-vault/session-newest-files.ts index c315c26dfad..c5fc252f338 100644 --- a/src/main/ai-vault/session-newest-files.ts +++ b/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 } diff --git a/src/main/ai-vault/session-scanner-discovery-pipeline.test.ts b/src/main/ai-vault/session-scanner-discovery-pipeline.test.ts new file mode 100644 index 00000000000..4ec44226820 --- /dev/null +++ b/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()), + readdir: fsMocks.readdir, + stat: fsMocks.stat +})) + +import { discoverFiles } from './session-scanner-discovery' + +const fs = await vi.importActual('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 { + 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((resolve) => { + releaseFirst = resolve + }) + let reachedLater = () => {} + const laterRead = new Promise((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(read: () => Promise): Promise { + active++ + peak = Math.max(peak, active) + try { + await new Promise((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((resolve) => { + releaseFirst = resolve + }) + let reachedLater = () => {} + const later = new Promise((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) + } + ) +}) diff --git a/src/main/ai-vault/session-scanner-discovery.ts b/src/main/ai-vault/session-scanner-discovery.ts index f7835dbdcc5..84c2ffdd1e4 100644 --- a/src/main/ai-vault/session-scanner-discovery.ts +++ b/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; sidecarPath?: string } + | { issue: { path: string; message: string } } + +async function withNativeDiscoverySlot(read: () => Promise): Promise { + 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 { 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 | 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 { + if (pending) { + consume(await pending) + pending = null + } + } + + function queueNativeFile(path: string): void | Promise { + 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 { + 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 +): Promise { + 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, + onFile: (path: string) => void | Promise, depth = 0 ): Promise { 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 + } } } }