diff --git a/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts b/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts index 21bbe468adf..0303bd1727a 100644 --- a/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts +++ b/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts @@ -264,6 +264,28 @@ describe('local Cursor sidecar discovery bounds', () => { ).rejects.toThrow('cursor_sidecar_scan_cancelled') }) + it('honors cancellation that lands while resolving a missing chats root', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-cursor-cancel-missing-')) + tempRoots.push(root) + let checks = 0 + const signal = { + get aborted() { + checks += 1 + return checks > 1 + } + } as AbortSignal + + await expect( + discoverLocalCursorSidecarsBounded({ + chatsDir: join(root, 'missing-chats'), + scopePaths: [], + issues: [], + signal + }) + ).rejects.toThrow('cursor_sidecar_scan_cancelled') + expect(checks).toBe(2) + }) + it('rejects on second-phase cancellation instead of resolving with an issue', async () => { const root = await mkdtemp(join(tmpdir(), 'orca-cursor-cancel-2nd-')) tempRoots.push(root) @@ -276,7 +298,7 @@ describe('local Cursor sidecar discovery bounds', () => { const signal = { get aborted() { checks += 1 - // First check (post-realpath) passes; later enumeration-phase checks abort. + // The pre-realpath check passes; the post-realpath check observes cancellation. return checks > 1 } } as AbortSignal diff --git a/src/relay/cursor-sidecar-scan.test.ts b/src/relay/cursor-sidecar-scan.test.ts index 7641e5d9d7a..6de8ebeef8c 100644 --- a/src/relay/cursor-sidecar-scan.test.ts +++ b/src/relay/cursor-sidecar-scan.test.ts @@ -12,7 +12,7 @@ import { cursorBucketForCwd } from '../main/ai-vault/session-scanner-cursor-path const roots: string[] = [] const context = { clientId: 1, isStale: () => false } -const CHECKS_BEFORE_FIRST_VERIFIED_READ = 8 +const CHECKS_BEFORE_FIRST_VERIFIED_READ = 9 async function createRoot(): Promise { const root = await mkdtemp(join(tmpdir(), 'orca-cursor-scan-')) @@ -89,6 +89,48 @@ describe('scanCursorSidecars', () => { }) }) + it('shares bounded session capacity across exact scope buckets', async () => { + const root = await createRoot() + const chatsRoot = join(root, 'chats') + const scopes = [join(root, 'a-workspace'), join(root, 'z-workspace')] + await Promise.all(scopes.map((scope) => mkdir(scope))) + const firstBucket = cursorBucketForCwd(scopes[0], process.platform) + await Promise.all([ + addSession(chatsRoot, firstBucket, 'first-a'), + addSession(chatsRoot, firstBucket, 'first-b'), + addSession(chatsRoot, cursorBucketForCwd(scopes[1], process.platform), 'second') + ]) + const request = defaultCursorSidecarScanRequest(chatsRoot, scopes, process.platform) + request.maxSessionDirs = 2 + + const result = await scanCursorSidecars(request, context) + + expect(result.sidecars.map((sidecar) => sidecar.sessionId).sort()).toEqual([ + 'first-a', + 'second' + ]) + expect(result.truncated.sessionDirs).toBe(true) + }) + + it('keeps exact scope sessions within the requested cap', async () => { + const root = await createRoot() + const chatsRoot = join(root, 'chats') + const scope = join(root, 'workspace') + await mkdir(scope) + const bucket = cursorBucketForCwd(scope, process.platform) + await Promise.all([ + addSession(chatsRoot, bucket, 'first-scoped'), + addSession(chatsRoot, bucket, 'second-scoped') + ]) + const request = defaultCursorSidecarScanRequest(chatsRoot, [scope], process.platform) + request.maxSessionDirs = 1 + + const result = await scanCursorSidecars(request, context) + + expect(result.sidecars).toHaveLength(1) + expect(result.truncated.sessionDirs).toBe(true) + }) + it.skipIf(process.platform === 'win32')( 'keeps missing files silent and rejects symlinked sidecars', async () => { diff --git a/src/shared/cursor-sidecar-scan-cancellation.ts b/src/shared/cursor-sidecar-scan-cancellation.ts new file mode 100644 index 00000000000..906cef1675a --- /dev/null +++ b/src/shared/cursor-sidecar-scan-cancellation.ts @@ -0,0 +1,19 @@ +export type CursorSidecarScanCancellation = { + throwIfCancelled: () => void +} + +export function isCursorSidecarScanCancelledError(error: unknown): boolean { + return error instanceof Error && error.message === 'cursor_sidecar_scan_cancelled' +} + +export function cursorSidecarScanCancellationFromSignal( + signal?: AbortSignal +): CursorSidecarScanCancellation { + return { + throwIfCancelled: () => { + if (signal?.aborted) { + throw new Error('cursor_sidecar_scan_cancelled') + } + } + } +} diff --git a/src/shared/cursor-sidecar-scan-directory.test.ts b/src/shared/cursor-sidecar-scan-directory.test.ts index 02fad24c122..6d332ec4eb3 100644 --- a/src/shared/cursor-sidecar-scan-directory.test.ts +++ b/src/shared/cursor-sidecar-scan-directory.test.ts @@ -1,5 +1,5 @@ import type { Dirent } from 'node:fs' -import { mkdir, mkdtemp, readdir, rm, writeFile } from 'node:fs/promises' +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' @@ -85,22 +85,14 @@ describe('cursor sidecar directory examination bounds', () => { expect(onDirentCount).toBe(budget) }) - it('hard-bounds the readdir fallback without retaining an unbounded listing', async () => { + it('fails bounded when directory streaming is unsupported', async () => { const root = await tempDir('orca-cursor-dir-fallback-') const budget = 16 - const names = await fillDirectory(root, budget + 8) + await fillDirectory(root, budget + 8) - let readdirCalls = 0 - let observedListingLength = 0 setStreamDirectoryIoForTests({ opendir: async () => { throw unsupportedDirectoryStreamError() - }, - readdirWithFileTypes: async (dirPath) => { - readdirCalls += 1 - const entries = await readdir(dirPath, { withFileTypes: true }) - observedListingLength = entries.length - return entries } }) @@ -113,13 +105,8 @@ describe('cursor sidecar directory examination bounds', () => { { maxEntriesExamined: budget } ) - expect(readdirCalls).toBe(1) - expect(result.entriesExamined).toBe(budget) - expect(result.examinationTruncated).toBe(true) - expect(visited).toHaveLength(budget) - expect(observedListingLength).toBe(names.length) - // After return, the eager listing must not remain as a retained examination set. - expect(visited.every((name) => names.includes(name))).toBe(true) + expect(result).toEqual({ entriesExamined: 0, examinationTruncated: true }) + expect(visited).toEqual([]) }) it('propagates cancellation from onDirent during examination', async () => { @@ -140,29 +127,6 @@ describe('cursor sidecar directory examination bounds', () => { expect(seen).toBe(10) }) - it('propagates cancellation on the readdir fallback path', async () => { - const root = await tempDir('orca-cursor-dir-cancel-fallback-') - await fillDirectory(root, 40) - setStreamDirectoryIoForTests({ - opendir: async () => { - throw unsupportedDirectoryStreamError() - } - }) - let seen = 0 - await expect( - streamDirectoryNames(root, () => undefined, { - maxEntriesExamined: 40, - onDirent: () => { - seen += 1 - if (seen >= 5) { - throw new Error('cursor_sidecar_scan_cancelled') - } - } - }) - ).rejects.toThrow('cursor_sidecar_scan_cancelled') - expect(seen).toBe(5) - }) - it('keeps listLexicographic examination-truncated when the walk hits the budget with more left', async () => { const root = await tempDir('orca-cursor-dir-list-trunc-') await fillDirectory(root, 20, 'z') @@ -226,7 +190,7 @@ describe('cursor sidecar directory examination bounds', () => { expect(listed.names).toHaveLength(64) }, 60_000) - it('fallback path never visits more than the examination budget on a 10k store', async () => { + it('unsupported streaming does not traverse a 10k store', async () => { const root = await tempDir('orca-cursor-dir-10k-fallback-') await fillDirectory(root, 10_000) setStreamDirectoryIoForTests({ @@ -243,9 +207,8 @@ describe('cursor sidecar directory examination bounds', () => { }, { maxEntriesExamined: budget } ) - expect(visits).toBe(budget) - expect(result.entriesExamined).toBe(budget) - expect(result.examinationTruncated).toBe(true) + expect(visits).toBe(0) + expect(result).toEqual({ entriesExamined: 0, examinationTruncated: true }) }, 60_000) it('retainLexicographic keeps only the first `limit` names in order', () => { @@ -257,6 +220,32 @@ describe('cursor sidecar directory examination bounds', () => { expect(selected).toEqual(['a', 'b', 'm']) }) + it('bounds comparisons for reverse-ordered retained names', () => { + const originalLocaleCompare = String.prototype.localeCompare + let comparisons = 0 + const localeCompare = vi.spyOn(String.prototype, 'localeCompare').mockImplementation(function ( + this: string, + value: string + ) { + comparisons += 1 + return originalLocaleCompare.call(this, value) + }) + const selected: string[] = [] + + try { + for (let index = 255; index >= 0; index -= 1) { + retainLexicographic(selected, String(index).padStart(3, '0'), 64) + } + } finally { + localeCompare.mockRestore() + } + + expect(comparisons).toBeLessThan(2_500) + expect(selected).toEqual( + Array.from({ length: 64 }, (_, index) => String(index).padStart(3, '0')) + ) + }) + it('treats a zero examination budget as already truncated', async () => { const root = await tempDir('orca-cursor-dir-zero-') await fillDirectory(root, 3) diff --git a/src/shared/cursor-sidecar-scan-directory.ts b/src/shared/cursor-sidecar-scan-directory.ts index 8b91c4bed14..09dcb733262 100644 --- a/src/shared/cursor-sidecar-scan-directory.ts +++ b/src/shared/cursor-sidecar-scan-directory.ts @@ -1,5 +1,5 @@ import type { Dir, Dirent } from 'node:fs' -import { opendir, readdir, realpath } from 'node:fs/promises' +import { opendir, realpath } from 'node:fs/promises' import { posix, win32 } from 'node:path' /** Per-directory dirent examination budget; keeps cold scans hard-bounded. */ @@ -18,12 +18,18 @@ export function retainLexicographic(selected: string[], name: string, limit: num if (name.localeCompare(last) >= 0) { return true } - let index = limit - 1 - while (index > 0 && name.localeCompare(selected[index - 1]) < 0) { - selected[index] = selected[index - 1] - index -= 1 + let low = 0 + let high = limit + while (low < high) { + const middle = Math.floor((low + high) / 2) + if (name.localeCompare(selected[middle]) < 0) { + high = middle + } else { + low = middle + 1 + } } - selected[index] = name + selected.splice(low, 0, name) + selected.pop() return true } @@ -36,22 +42,18 @@ export type StreamDirectoryNamesOptions = { type StreamDirectoryIo = { opendir: (path: string) => ReturnType - readdirWithFileTypes: (path: string) => Promise } const defaultStreamDirectoryIo: StreamDirectoryIo = { - opendir, - readdirWithFileTypes: (path) => readdir(path, { withFileTypes: true }) + opendir } let streamDirectoryIo: StreamDirectoryIo = defaultStreamDirectoryIo -/** Test isolation for opendir vs readdir-fallback paths. */ +/** Test isolation for streamed and unsupported directory paths. */ export function setStreamDirectoryIoForTests(next?: Partial): void { streamDirectoryIo = next ? { - opendir: next.opendir ?? defaultStreamDirectoryIo.opendir, - readdirWithFileTypes: - next.readdirWithFileTypes ?? defaultStreamDirectoryIo.readdirWithFileTypes + opendir: next.opendir ?? defaultStreamDirectoryIo.opendir } : defaultStreamDirectoryIo } @@ -82,12 +84,12 @@ export async function streamDirectoryNames( await directory.close().catch(() => undefined) } } catch (error) { - // Fallback keeps remote-wire hosts that only expose readdir working. if (!isUnsupportedDirectoryStream(error)) { throw error } + // readdir is eager and cannot preserve this function's memory/IO bound. + return { entriesExamined: 0, examinationTruncated: true } } - return examineReaddirFallback(dirPath, visit, options, maxEntries) } /** @@ -198,33 +200,6 @@ async function examineDirectoryStream( return { entriesExamined, examinationTruncated } } -async function examineReaddirFallback( - dirPath: string, - visit: (name: string, entry: Dirent) => void, - options: StreamDirectoryNamesOptions, - maxEntries: number -): Promise<{ entriesExamined: number; examinationTruncated: boolean }> { - // Why: readdir eagerly allocates the full listing. Shrink immediately to the - // examination window so adversarial directories cannot keep an unbounded - // Dirent array alive past the budget (peak alloc is platform-limited). - const entries = await streamDirectoryIo.readdirWithFileTypes(dirPath) - let entriesExamined = 0 - try { - const examinationTruncated = entries.length > maxEntries - if (examinationTruncated) { - entries.length = maxEntries - } - for (const entry of entries) { - entriesExamined += 1 - options.onDirent?.() - visit(entry.name, entry) - } - return { entriesExamined, examinationTruncated } - } finally { - entries.length = 0 - } -} - function isUnsupportedDirectoryStream(error: unknown): boolean { const code = (error as NodeJS.ErrnoException | null)?.code return code === 'ERR_FEATURE_UNAVAILABLE_ON_PLATFORM' || code === 'ENOTSUP' diff --git a/src/shared/cursor-sidecar-scan-discovery.ts b/src/shared/cursor-sidecar-scan-discovery.ts index 25e98caeaff..73bec4e896a 100644 --- a/src/shared/cursor-sidecar-scan-discovery.ts +++ b/src/shared/cursor-sidecar-scan-discovery.ts @@ -2,16 +2,34 @@ import { createHash } from 'node:crypto' import type { Stats } from 'node:fs' import { lstat, realpath } from 'node:fs/promises' import { join } from 'node:path' -import type { CursorSidecarScanRequest, CursorSidecarScanResponse } from './cursor-sidecar-scan' +import { + CURSOR_REMOTE_MAX_SESSION_DIRS, + type CursorSidecarScanRequest, + type CursorSidecarScanResponse +} from './cursor-sidecar-scan' +import { + isCursorSidecarScanCancelledError, + type CursorSidecarScanCancellation +} from './cursor-sidecar-scan-cancellation' import { CURSOR_DIR_MAX_ENTRIES_EXAMINED, listLexicographicDirectoryNames, - resolveTargetScopePathVariants, - safeBasename + resolveTargetScopePathVariants } from './cursor-sidecar-scan-directory' +import { + CURSOR_SIDECAR_SCAN_CONCURRENCY, + retainCursorSidecarSessions, + type CursorSidecarScanBucket as Bucket, + type CursorSidecarScanSession +} from './cursor-sidecar-scan-session-retention' + +export { + cursorSidecarScanCancellationFromSignal, + isCursorSidecarScanCancelledError, + type CursorSidecarScanCancellation +} from './cursor-sidecar-scan-cancellation' const BUCKET_PATTERN = /^[0-9a-f]{32}$/u -const BUCKET_READ_CONCURRENCY = 8 /** Check cancellation every N dirents during cold directory walks. */ const CANCEL_CHECK_EVERY_DIRENTS = 64 @@ -23,11 +41,6 @@ export type CursorSidecarScanCaps = { aggregateBytes: number } -export type CursorSidecarScanCancellation = { - throwIfCancelled: () => void -} - -type Bucket = { name: string; path: string; scopeCwd: string | null } type ScanArgs = { caps: CursorSidecarScanCaps response: CursorSidecarScanResponse @@ -42,22 +55,6 @@ export type CursorSidecarScanCandidate = Bucket & { store: Stats } -export function isCursorSidecarScanCancelledError(error: unknown): boolean { - return error instanceof Error && error.message === 'cursor_sidecar_scan_cancelled' -} - -export function cursorSidecarScanCancellationFromSignal( - signal?: AbortSignal -): CursorSidecarScanCancellation { - return { - throwIfCancelled: () => { - if (signal?.aborted) { - throw new Error('cursor_sidecar_scan_cancelled') - } - } - } -} - export async function discoverCursorSidecarCandidates(args: { request: CursorSidecarScanRequest caps: CursorSidecarScanCaps @@ -71,10 +68,12 @@ export async function discoverCursorSidecarCandidates(args: { const pathPlatform = args.pathPlatform ?? process.platform const chatsRoot = args.request.chatsRoot let rootRealPath: string + args.cancellation.throwIfCancelled() try { rootRealPath = await realpath(chatsRoot) } catch (error) { rethrowCancel(error) + args.cancellation.throwIfCancelled() if (!isMissing(error)) { addIssue(args.response, chatsRoot, error) } @@ -85,7 +84,12 @@ export async function discoverCursorSidecarCandidates(args: { const direct = await scopeBuckets(args.request, chatsRoot, pathPlatform, args) const enumerated = await enumeratedBuckets(chatsRoot, direct, args) - const sessions = await retainSessions([...direct.values(), ...enumerated], args) + const sessions = await retainCursorSidecarSessions({ + buckets: [...direct.values(), ...enumerated], + sessionLimit: args.caps.sessions, + response: args.response, + cancellation: args.cancellation + }) return { rootRealPath, candidates: await eligibleCandidates(sessions, args) } } @@ -155,86 +159,16 @@ async function enumeratedBuckets( } } -async function retainSessions( - buckets: readonly Bucket[], - args: ScanArgs -): Promise<(Bucket & { sessionId: string })[]> { - const retained: (Bucket & { sessionId: string })[] = [] - for ( - let index = 0; - index < buckets.length && retained.length < args.caps.sessions; - index += BUCKET_READ_CONCURRENCY - ) { - args.cancellation.throwIfCancelled() - const batch = buckets.slice(index, index + BUCKET_READ_CONCURRENCY) - const listings = await Promise.all( - batch.map((bucket) => listBucketSessions(bucket, args, retained.length)) - ) - args.cancellation.throwIfCancelled() - for (let listingIndex = 0; listingIndex < listings.length; listingIndex += 1) { - const { bucket, names, truncated } = listings[listingIndex] - if (retained.length >= args.caps.sessions) { - if ( - listings.slice(listingIndex).some((listing) => listing.truncated || listing.names.length) - ) { - args.response.truncated.sessionDirs = true - } - break - } - const capacity = args.caps.sessions - retained.length - retained.push(...names.slice(0, capacity).map((sessionId) => ({ ...bucket, sessionId }))) - if (truncated || names.length > capacity) { - args.response.truncated.sessionDirs = true - } - } - if (retained.length >= args.caps.sessions && index + batch.length < buckets.length) { - args.response.truncated.sessionDirs = true - } - } - return retained -} - -async function listBucketSessions( - bucket: Bucket, - args: ScanArgs, - retainedCount: number -): Promise<{ bucket: Bucket; names: string[]; truncated: boolean }> { - try { - if (bucket.scopeCwd) { - args.response.counters.fileLstat++ - const stats = await lstat(bucket.path) - if (!stats.isDirectory() || stats.isSymbolicLink()) { - return { bucket, names: [], truncated: false } - } - } - args.response.counters.bucketReaddir++ - const listed = await listLexicographicDirectoryNames({ - dirPath: bucket.path, - limit: Math.max(0, args.caps.sessions - retainedCount), - maxEntriesExamined: CURSOR_DIR_MAX_ENTRIES_EXAMINED, - accept: (name, entry) => entry.isDirectory() && !entry.isSymbolicLink() && safeBasename(name), - onDirent: createDirentCancelChecker(args.cancellation) - }) - return { bucket, names: listed.names, truncated: listed.truncated } - } catch (error) { - rethrowCancel(error) - if (!isMissing(error)) { - addIssue(args.response, bucket.path, error) - } - return { bucket, names: [], truncated: false } - } -} - async function eligibleCandidates( - sessions: readonly (Bucket & { sessionId: string })[], + sessions: readonly CursorSidecarScanSession[], args: ScanArgs ): Promise { // Collect every eligible candidate first, then apply newest-first retention // before the aggregate byte budget (stable lexical ties). const eligible: CursorSidecarScanCandidate[] = [] - for (let index = 0; index < sessions.length; index += BUCKET_READ_CONCURRENCY) { + for (let index = 0; index < sessions.length; index += CURSOR_SIDECAR_SCAN_CONCURRENCY) { args.cancellation.throwIfCancelled() - const batch = sessions.slice(index, index + BUCKET_READ_CONCURRENCY) + const batch = sessions.slice(index, index + CURSOR_SIDECAR_SCAN_CONCURRENCY) const results = await Promise.all(batch.map((session) => inspectSession(session, args))) args.cancellation.throwIfCancelled() for (const result of results) { @@ -247,7 +181,7 @@ async function eligibleCandidates( } async function inspectSession( - session: Bucket & { sessionId: string }, + session: CursorSidecarScanSession, args: ScanArgs ): Promise { const sessionDir = join(session.path, session.sessionId) @@ -287,9 +221,25 @@ function retainByNewestThenAggregate( Math.max(left.meta.mtimeMs, left.store.mtimeMs) || `${left.name}\0${left.sessionId}`.localeCompare(`${right.name}\0${right.sessionId}`) ) + const representatives = new Set() + const representedBuckets = new Set() + for (const candidate of ranked) { + if (candidate.scopeCwd && !representedBuckets.has(candidate.name)) { + representedBuckets.add(candidate.name) + representatives.add(candidate) + } + } + const prioritized = [ + ...representatives, + ...ranked.filter((candidate) => !representatives.has(candidate)) + ] const retained: CursorSidecarScanCandidate[] = [] let aggregateBytes = 0 - for (const candidate of ranked) { + for (const candidate of prioritized) { + if (retained.length >= CURSOR_REMOTE_MAX_SESSION_DIRS) { + args.response.truncated.sessionDirs = true + break + } if (aggregateBytes + candidate.meta.size > args.caps.aggregateBytes) { args.response.truncated.sidecarBytes = true break diff --git a/src/shared/cursor-sidecar-scan-session-retention.ts b/src/shared/cursor-sidecar-scan-session-retention.ts new file mode 100644 index 00000000000..ad0832cac48 --- /dev/null +++ b/src/shared/cursor-sidecar-scan-session-retention.ts @@ -0,0 +1,203 @@ +import { lstat } from 'node:fs/promises' +import type { CursorSidecarScanResponse } from './cursor-sidecar-scan' +import { + isCursorSidecarScanCancelledError, + type CursorSidecarScanCancellation +} from './cursor-sidecar-scan-cancellation' +import { + CURSOR_DIR_MAX_ENTRIES_EXAMINED, + listLexicographicDirectoryNames, + safeBasename +} from './cursor-sidecar-scan-directory' + +export const CURSOR_SIDECAR_SCAN_CONCURRENCY = 8 +const CANCEL_CHECK_EVERY_DIRENTS = 64 + +export type CursorSidecarScanBucket = { name: string; path: string; scopeCwd: string | null } +export type CursorSidecarScanSession = CursorSidecarScanBucket & { sessionId: string } + +type RetentionArgs = { + buckets: readonly CursorSidecarScanBucket[] + sessionLimit: number + response: CursorSidecarScanResponse + cancellation: CursorSidecarScanCancellation +} + +export async function retainCursorSidecarSessions( + args: RetentionArgs +): Promise { + const retained: CursorSidecarScanSession[] = [] + const scopedBuckets = args.buckets.filter((bucket) => bucket.scopeCwd !== null) + const unscopedBuckets = args.buckets.filter((bucket) => bucket.scopeCwd === null) + const scopedBucketLimit = scopedBuckets.length + ? Math.max(1, Math.floor(CURSOR_DIR_MAX_ENTRIES_EXAMINED / scopedBuckets.length)) + : 0 + const scopedListings = await listBucketSessionsBatched(scopedBuckets, args, scopedBucketLimit) + retainScopedSessionListings(scopedListings, retained, args) + if (retained.length >= args.sessionLimit) { + if (unscopedBuckets.length) { + args.response.truncated.sessionDirs = true + } + return retained + } + await retainUnscopedSessions(unscopedBuckets, retained, args) + return retained +} + +async function listBucketSessionsBatched( + buckets: readonly CursorSidecarScanBucket[], + args: RetentionArgs, + limit: number +): Promise { + const listings: BucketSessionListing[] = [] + for (let index = 0; index < buckets.length; index += CURSOR_SIDECAR_SCAN_CONCURRENCY) { + args.cancellation.throwIfCancelled() + const batch = buckets.slice(index, index + CURSOR_SIDECAR_SCAN_CONCURRENCY) + listings.push( + ...(await Promise.all(batch.map((bucket) => listBucketSessions(bucket, args, limit, limit)))) + ) + args.cancellation.throwIfCancelled() + } + return listings +} + +function retainScopedSessionListings( + listings: readonly BucketSessionListing[], + retained: CursorSidecarScanSession[], + args: RetentionArgs +): void { + let depth = 0 + while (retained.length < args.sessionLimit) { + let found = false + for (const { bucket, names } of listings) { + const sessionId = names[depth] + if (sessionId === undefined) { + continue + } + found = true + retained.push({ ...bucket, sessionId }) + if (retained.length >= args.sessionLimit) { + break + } + } + if (!found) { + break + } + depth += 1 + } + const listedCount = listings.reduce((total, listing) => total + listing.names.length, 0) + if (listings.some((listing) => listing.truncated) || listedCount > retained.length) { + args.response.truncated.sessionDirs = true + } +} + +async function retainUnscopedSessions( + buckets: readonly CursorSidecarScanBucket[], + retained: CursorSidecarScanSession[], + args: RetentionArgs +): Promise { + for (let index = 0; index < buckets.length; index += CURSOR_SIDECAR_SCAN_CONCURRENCY) { + args.cancellation.throwIfCancelled() + const batch = buckets.slice(index, index + CURSOR_SIDECAR_SCAN_CONCURRENCY) + const listings = await Promise.all( + batch.map((bucket) => + listBucketSessions( + bucket, + args, + args.sessionLimit - retained.length, + CURSOR_DIR_MAX_ENTRIES_EXAMINED + ) + ) + ) + args.cancellation.throwIfCancelled() + retainUnscopedSessionListings(listings, retained, args) + if (retained.length >= args.sessionLimit && index + batch.length < buckets.length) { + args.response.truncated.sessionDirs = true + break + } + } +} + +function retainUnscopedSessionListings( + listings: readonly BucketSessionListing[], + retained: CursorSidecarScanSession[], + args: RetentionArgs +): void { + for (let index = 0; index < listings.length; index += 1) { + const { bucket, names, truncated } = listings[index] + if (retained.length >= args.sessionLimit) { + if (listings.slice(index).some((listing) => listing.truncated || listing.names.length)) { + args.response.truncated.sessionDirs = true + } + break + } + const capacity = args.sessionLimit - retained.length + const selected = names.slice(0, capacity) + retained.push(...selected.map((sessionId) => ({ ...bucket, sessionId }))) + if (truncated || names.length > capacity) { + args.response.truncated.sessionDirs = true + } + } +} + +type BucketSessionListing = { + bucket: CursorSidecarScanBucket + names: string[] + truncated: boolean +} + +async function listBucketSessions( + bucket: CursorSidecarScanBucket, + args: RetentionArgs, + limit: number, + maxEntriesExamined: number +): Promise { + try { + if (bucket.scopeCwd) { + args.response.counters.fileLstat++ + const stats = await lstat(bucket.path) + if (!stats.isDirectory() || stats.isSymbolicLink()) { + return { bucket, names: [], truncated: false } + } + } + args.response.counters.bucketReaddir++ + const listed = await listLexicographicDirectoryNames({ + dirPath: bucket.path, + limit: Math.max(0, limit), + maxEntriesExamined, + accept: (name, entry) => entry.isDirectory() && !entry.isSymbolicLink() && safeBasename(name), + onDirent: createDirentCancelChecker(args.cancellation) + }) + return { bucket, names: listed.names, truncated: listed.truncated } + } catch (error) { + if (isCursorSidecarScanCancelledError(error)) { + throw error + } + if (!isMissing(error)) { + addIssue(args.response, bucket.path, error) + } + return { bucket, names: [], truncated: false } + } +} + +function createDirentCancelChecker(cancellation: CursorSidecarScanCancellation): () => void { + let seen = 0 + return () => { + seen += 1 + if (seen % CANCEL_CHECK_EVERY_DIRENTS === 0) { + cancellation.throwIfCancelled() + } + } +} + +function addIssue(response: CursorSidecarScanResponse, path: string, error: unknown): void { + response.issues.push({ + path, + message: error instanceof Error ? error.message.slice(0, 1_024) : 'Cursor scan failed.' + }) +} + +function isMissing(error: unknown): boolean { + const code = (error as NodeJS.ErrnoException | null)?.code + return code === 'ENOENT' || code === 'ENOTDIR' +}