mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 08:03:28 +00:00
fix(ai-vault): close scan phase boundary
This commit is contained in:
@@ -97,6 +97,14 @@ function assistantRecord(index: number, text: string): string {
|
||||
})
|
||||
}
|
||||
|
||||
function nestedUnknownSession(depth: number): PersistedSessionParseCacheEntry['session'] {
|
||||
let value: unknown = null
|
||||
for (let index = 0; index < depth; index += 1) {
|
||||
value = { next: value }
|
||||
}
|
||||
return value as PersistedSessionParseCacheEntry['session']
|
||||
}
|
||||
|
||||
async function writeTranscript(root: string): Promise<string> {
|
||||
const path = join(root, 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee.jsonl')
|
||||
await writeFile(path, `${userRecord(0, 'first question')}\n${assistantRecord(1, 'answer')}\n`)
|
||||
@@ -184,6 +192,39 @@ describe('session parse cache persistence', () => {
|
||||
expect(stats.reused).toBe(0)
|
||||
})
|
||||
|
||||
it('accepts an exact-byte-cap cache file', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
const transcript = await writeTranscript(root)
|
||||
const candidate = await claudeCandidate(transcript)
|
||||
const file = {
|
||||
schemaVersion: 1,
|
||||
appVersion: APP_VERSION,
|
||||
entries: [
|
||||
[
|
||||
transcript,
|
||||
{
|
||||
mtimeMs: candidate.file.mtimeMs,
|
||||
sizeBytes: candidate.file.sizeBytes,
|
||||
platform: process.platform,
|
||||
session: null
|
||||
}
|
||||
]
|
||||
],
|
||||
padding: ''
|
||||
}
|
||||
const base = JSON.stringify(file)
|
||||
file.padding = 'x'.repeat(SESSION_PARSE_CACHE_MAX_BYTES - Buffer.byteLength(base))
|
||||
const payload = JSON.stringify(file)
|
||||
expect(Buffer.byteLength(payload)).toBe(SESSION_PARSE_CACHE_MAX_BYTES)
|
||||
await writeFile(cacheFile, payload)
|
||||
initSessionParseCachePersistence({ filePath: cacheFile, appVersion: APP_VERSION })
|
||||
|
||||
const stats = await coldParseStats(transcript)
|
||||
expect(stats.reused).toBe(1)
|
||||
expect(stats.fullParses).toBe(0)
|
||||
})
|
||||
|
||||
it('rejects structurally excessive JSON before parsing the cache graph', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
@@ -215,6 +256,43 @@ describe('session parse cache persistence', () => {
|
||||
expect(stats.reused).toBe(0)
|
||||
})
|
||||
|
||||
it('accepts nesting depth 32 and rejects depth 33 through the real loader', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
const transcript = await writeTranscript(root)
|
||||
const candidate = await claudeCandidate(transcript)
|
||||
const writeNestedCache = async (sessionDepth: number): Promise<void> => {
|
||||
await writeFile(
|
||||
cacheFile,
|
||||
JSON.stringify({
|
||||
schemaVersion: 1,
|
||||
appVersion: APP_VERSION,
|
||||
entries: [
|
||||
[
|
||||
transcript,
|
||||
{
|
||||
mtimeMs: candidate.file.mtimeMs,
|
||||
sizeBytes: candidate.file.sizeBytes,
|
||||
platform: process.platform,
|
||||
session: nestedUnknownSession(sessionDepth)
|
||||
}
|
||||
]
|
||||
]
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
await writeNestedCache(28)
|
||||
initSessionParseCachePersistence({ filePath: cacheFile, appVersion: APP_VERSION })
|
||||
expect((await coldParseStats(transcript)).reused).toBe(1)
|
||||
|
||||
simulateRestart(cacheFile)
|
||||
await writeNestedCache(29)
|
||||
const rejectedStats = await coldParseStats(transcript)
|
||||
expect(rejectedStats.reused).toBe(0)
|
||||
expect(rejectedStats.fullParses).toBe(1)
|
||||
})
|
||||
|
||||
it('ignores a cache file with a mismatched schemaVersion', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
@@ -407,6 +485,41 @@ describe('session parse cache persistence', () => {
|
||||
expect(stats.fullParses).toBe(0)
|
||||
})
|
||||
|
||||
it('round-trips a full 4,096-entry writer snapshot', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
const transcript = await writeTranscript(root)
|
||||
const candidate = await claudeCandidate(transcript)
|
||||
const entry: PersistedSessionParseCacheEntry = {
|
||||
mtimeMs: candidate.file.mtimeMs,
|
||||
sizeBytes: candidate.file.sizeBytes ?? null,
|
||||
platform: process.platform,
|
||||
session: null
|
||||
}
|
||||
seedSessionParseCache([
|
||||
...Array.from({ length: 4095 }, (_, index): [string, PersistedSessionParseCacheEntry] => [
|
||||
`/fake-${index}`,
|
||||
entry
|
||||
]),
|
||||
[transcript, entry]
|
||||
])
|
||||
initSessionParseCachePersistence({ filePath: cacheFile, appVersion: APP_VERSION })
|
||||
scheduleSessionParseCachePersist({
|
||||
reused: 0,
|
||||
incremental: 0,
|
||||
fullParses: 1,
|
||||
bytesRead: 1
|
||||
})
|
||||
await flushSessionParseCachePersistForTests()
|
||||
|
||||
simulateRestart(cacheFile)
|
||||
await ensureSessionParseCacheLoaded()
|
||||
const stats = createSessionParseStats()
|
||||
await parseAgentSessionFileCached(candidate, process.platform, stats)
|
||||
expect(stats.reused).toBe(1)
|
||||
expect(JSON.parse(await readFile(cacheFile, 'utf8')).entries).toHaveLength(4096)
|
||||
})
|
||||
|
||||
it('rejects malformed entries even when they fall outside the retained tail', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
@@ -501,6 +614,39 @@ describe('session parse cache persistence', () => {
|
||||
debugSpy.mockRestore()
|
||||
})
|
||||
|
||||
it('leaves the previous snapshot intact when writer byte admission fails', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
initSessionParseCachePersistence({ filePath: cacheFile, appVersion: APP_VERSION })
|
||||
const transcript = await writeTranscript(root)
|
||||
await parseAndPersist(transcript)
|
||||
const previousSnapshot = await readFile(cacheFile, 'utf8')
|
||||
|
||||
seedSessionParseCache([
|
||||
[
|
||||
'x'.repeat(SESSION_PARSE_CACHE_MAX_BYTES),
|
||||
{
|
||||
mtimeMs: 1,
|
||||
sizeBytes: 1,
|
||||
platform: process.platform,
|
||||
session: null
|
||||
}
|
||||
]
|
||||
])
|
||||
const debugSpy = vi.spyOn(console, 'debug').mockImplementation(() => {})
|
||||
scheduleSessionParseCachePersist({
|
||||
reused: 0,
|
||||
incremental: 0,
|
||||
fullParses: 1,
|
||||
bytesRead: 1
|
||||
})
|
||||
await flushSessionParseCachePersistForTests()
|
||||
|
||||
expect(await readFile(cacheFile, 'utf8')).toBe(previousSnapshot)
|
||||
expect(debugSpy).toHaveBeenCalled()
|
||||
debugSpy.mockRestore()
|
||||
})
|
||||
|
||||
it('a failing rename cleans up its temp file and keeps the previous snapshot usable', async () => {
|
||||
const root = await makeTempDir()
|
||||
const cacheFile = join(root, 'session-parse-cache.json')
|
||||
|
||||
@@ -40,15 +40,18 @@ describe('OpenCodeSqliteCandidatePhase', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('disarms after the last SQLite batch or an early parser stop', async () => {
|
||||
it('disarms after the last SQLite promise without awaiting an unrelated parser', async () => {
|
||||
vi.useFakeTimers()
|
||||
const sqlite = candidate('/data/opencode.db#session')
|
||||
const legacy = candidate('/legacy/session.json')
|
||||
const completedContext = new OpenCodeSqliteScanContext(1)
|
||||
const stoppedContext = new OpenCodeSqliteScanContext(1)
|
||||
try {
|
||||
const completed = new OpenCodeSqliteCandidatePhase([sqlite], completedContext)
|
||||
expect(completed.prepareBatch([sqlite])).toEqual([sqlite])
|
||||
completed.completeBatch()
|
||||
const completed = new OpenCodeSqliteCandidatePhase([sqlite, legacy], completedContext)
|
||||
const batch = completed.prepareBatch([sqlite, legacy])
|
||||
const neverSettles = new Promise<void>(() => {})
|
||||
completed.trackBatch(batch, [Promise.resolve(), neverSettles])
|
||||
await Promise.resolve()
|
||||
|
||||
const stopped = new OpenCodeSqliteCandidatePhase([sqlite], stoppedContext)
|
||||
stopped.finish()
|
||||
|
||||
@@ -32,10 +32,20 @@ export class OpenCodeSqliteCandidatePhase {
|
||||
})
|
||||
}
|
||||
|
||||
completeBatch(): void {
|
||||
if (this.remainingCandidates === 0) {
|
||||
this.context?.disarmDeadline()
|
||||
trackBatch(
|
||||
candidates: readonly SessionFileCandidate[],
|
||||
promises: readonly Promise<unknown>[]
|
||||
): void {
|
||||
if (this.remainingCandidates !== 0 || !this.context) {
|
||||
return
|
||||
}
|
||||
const sqlitePromises = promises.filter((_, index) =>
|
||||
looksLikeOpenCodeSqliteCandidate(candidates[index]?.file.path ?? '')
|
||||
)
|
||||
if (sqlitePromises.length === 0) {
|
||||
return
|
||||
}
|
||||
void Promise.allSettled(sqlitePromises).then(() => this.context?.disarmDeadline())
|
||||
}
|
||||
|
||||
finish(): void {
|
||||
|
||||
@@ -5,6 +5,31 @@ import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { scanAiVaultSessions } from './session-scanner'
|
||||
import Database from '../sqlite/sync-database'
|
||||
import type { AiVaultScanIssue, AiVaultSession } from '../../shared/ai-vault-types'
|
||||
import type { SessionFileCandidate } from './session-scanner-types'
|
||||
import type { OpenCodeSqliteScanContext } from './session-scanner-opencode-sqlite-scan-context'
|
||||
import { buildOpenCodeSqliteCandidatePath } from './session-scanner-opencode-sqlite-paths'
|
||||
|
||||
type WorkerListArgs = {
|
||||
context: OpenCodeSqliteScanContext
|
||||
dbPaths: readonly string[]
|
||||
limit: number
|
||||
issues: AiVaultScanIssue[]
|
||||
}
|
||||
|
||||
type WorkerParseArgs = {
|
||||
context: OpenCodeSqliteScanContext
|
||||
dbPath: string
|
||||
sessionId: string
|
||||
platform: NodeJS.Platform
|
||||
}
|
||||
|
||||
const workerMock = vi.hoisted(() => ({
|
||||
listCalls: 0,
|
||||
parseCalls: 0,
|
||||
listOverride: null as null | ((args: WorkerListArgs) => Promise<SessionFileCandidate[]>),
|
||||
parseOverride: null as null | ((args: WorkerParseArgs) => Promise<AiVaultSession | null>)
|
||||
}))
|
||||
|
||||
// Why: this source-level integration suite has no built worker entry. Keep its
|
||||
// SQLite fixtures inline explicitly; production fails closed if the bundle is absent.
|
||||
@@ -14,8 +39,14 @@ vi.mock('./session-scanner-opencode-sqlite-worker-spawn', async () => {
|
||||
import('./session-scanner-opencode-sqlite')
|
||||
])
|
||||
return {
|
||||
listOpenCodeSqliteSessionsViaWorker: listOpenCodeSqliteSessions,
|
||||
parseOpenCodeSqliteSessionViaWorker: parseOpenCodeSqliteSession
|
||||
listOpenCodeSqliteSessionsViaWorker: (args: WorkerListArgs) => {
|
||||
workerMock.listCalls += 1
|
||||
return workerMock.listOverride?.(args) ?? listOpenCodeSqliteSessions(args)
|
||||
},
|
||||
parseOpenCodeSqliteSessionViaWorker: (args: WorkerParseArgs) => {
|
||||
workerMock.parseCalls += 1
|
||||
return workerMock.parseOverride?.(args) ?? parseOpenCodeSqliteSession(args)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
@@ -23,6 +54,10 @@ let tempRoots: string[] = []
|
||||
let tempDbDirs: string[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
workerMock.listCalls = 0
|
||||
workerMock.parseCalls = 0
|
||||
workerMock.listOverride = null
|
||||
workerMock.parseOverride = null
|
||||
await Promise.all(tempRoots.map((root) => rm(root, { recursive: true, force: true })))
|
||||
for (const dir of tempDbDirs) {
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
@@ -104,6 +139,69 @@ function applyOpenCodeSchema(db: Database.Database): void {
|
||||
}
|
||||
|
||||
describe('scanAiVaultSessions — OpenCode SQLite + legacy file coexistence', () => {
|
||||
it('stops native and WSL SQLite parsing at one scan deadline', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-deadline-'))
|
||||
tempRoots.push(root)
|
||||
const roots = isolatedScanRoots(root)
|
||||
const nativeDataDir = join(root, 'native-opencode')
|
||||
const wslHome = join(root, 'wsl-home')
|
||||
await mkdir(join(nativeDataDir, 'storage'), { recursive: true })
|
||||
await mkdir(join(wslHome, '.local', 'share', 'opencode'), { recursive: true })
|
||||
await writeFile(join(nativeDataDir, 'opencode.db'), '')
|
||||
await writeFile(join(wslHome, '.local', 'share', 'opencode', 'opencode.db'), '')
|
||||
|
||||
workerMock.listOverride = async (args) =>
|
||||
args.dbPaths.flatMap((dbPath, sourceIndex) =>
|
||||
Array.from({ length: 120 }, (_, candidateIndex) => ({
|
||||
agent: 'opencode' as const,
|
||||
codexHome: null,
|
||||
file: {
|
||||
path: buildOpenCodeSqliteCandidatePath(
|
||||
dbPath,
|
||||
`session-${sourceIndex}-${candidateIndex}`
|
||||
),
|
||||
mtimeMs: 10_000 - candidateIndex,
|
||||
modifiedAt: new Date(10_000 - candidateIndex).toISOString()
|
||||
}
|
||||
}))
|
||||
)
|
||||
workerMock.parseOverride = (args) =>
|
||||
new Promise((_, reject) => {
|
||||
const rejectTerminated = (): void => {
|
||||
args.context.markWorkOmitted()
|
||||
reject(args.context.terminationError())
|
||||
}
|
||||
if (args.context.isTerminated) {
|
||||
rejectTerminated()
|
||||
} else {
|
||||
args.context.signal.addEventListener('abort', rejectTerminated, { once: true })
|
||||
}
|
||||
})
|
||||
|
||||
const scan = scanAiVaultSessions({
|
||||
...roots,
|
||||
opencodeStorageDir: join(nativeDataDir, 'storage'),
|
||||
opencodeDbPaths: undefined,
|
||||
wslHomeDirs: [wslHome],
|
||||
limit: 500
|
||||
})
|
||||
await vi.waitFor(() => expect(workerMock.parseCalls).toBe(8))
|
||||
await vi.advanceTimersByTimeAsync(45_000)
|
||||
const result = await scan
|
||||
|
||||
expect(workerMock.listCalls).toBe(2)
|
||||
expect(workerMock.parseCalls).toBe(8)
|
||||
expect(result.sessions).toEqual([])
|
||||
expect(
|
||||
result.issues.filter((issue) => /OpenCode history was skipped/.test(issue.message))
|
||||
).toHaveLength(1)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('discovers SQLite sessions next to a custom OpenCode storage directory', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-custom-opencode-'))
|
||||
tempRoots.push(root)
|
||||
|
||||
@@ -189,6 +189,9 @@ describe('OpenCodeSqliteWorkerClient', () => {
|
||||
// Exactly one respawn; the queued call drains on the new worker.
|
||||
expect(workers).toHaveLength(2)
|
||||
const respawned = workers[1]!
|
||||
workers[0]!.emit('error', new Error('late old-worker fault'))
|
||||
workers[0]!.emit('message', { id: respawned.lastId(), ok: true, value: 'stale' })
|
||||
expect(workers).toHaveLength(2)
|
||||
respawned.emit('message', { id: respawned.lastId(), ok: true, value: 'B' })
|
||||
await expect(queued).resolves.toBe('B')
|
||||
})
|
||||
@@ -351,6 +354,24 @@ describe('OpenCodeSqliteWorkerClient', () => {
|
||||
).rejects.toThrow(/background scanner could not start/)
|
||||
})
|
||||
|
||||
it('rejects already-aborted work without spawning a worker', async () => {
|
||||
const workers: FakeWorker[] = []
|
||||
const client = new OpenCodeSqliteWorkerClient({ workerFactory: makeFactory(workers), log() {} })
|
||||
const abortedContext = new OpenCodeSqliteScanContext()
|
||||
abortedContext.dispose()
|
||||
|
||||
await expect(
|
||||
client.parse({
|
||||
context: abortedContext,
|
||||
dbPath: '/db#a',
|
||||
sessionId: 'a',
|
||||
platform: 'darwin'
|
||||
})
|
||||
).rejects.toThrow(/scan ended/)
|
||||
expect(workers).toEqual([])
|
||||
expect(abortedContext.metrics().workOmitted).toBe(true)
|
||||
})
|
||||
|
||||
it('stops respawning after the consecutive-death cap and fails the rest to issues', async () => {
|
||||
const workers: FakeWorker[] = []
|
||||
const client = new OpenCodeSqliteWorkerClient({ workerFactory: makeFactory(workers), log() {} })
|
||||
@@ -505,6 +526,19 @@ describe('OpenCodeSqliteWorkerClient', () => {
|
||||
await rejection
|
||||
expect(addListener).toHaveBeenCalledTimes(index + 1)
|
||||
expect(removeListener).toHaveBeenCalledTimes(index + 1)
|
||||
|
||||
const interleaved = client.parse({
|
||||
context: otherContext,
|
||||
dbPath: `/db#b${index}`,
|
||||
sessionId: `b${index}`,
|
||||
platform: 'darwin'
|
||||
})
|
||||
workers.at(-1)!.emit('message', {
|
||||
id: workers.at(-1)!.lastId(),
|
||||
ok: true,
|
||||
value: `B${index}`
|
||||
})
|
||||
await expect(interleaved).resolves.toBe(`B${index}`)
|
||||
}
|
||||
|
||||
const third = client.parse({
|
||||
|
||||
@@ -244,19 +244,18 @@ async function parseSessionCandidates(args: {
|
||||
const batchSize = Math.min(SESSION_PARSE_CONCURRENCY, needed, remaining)
|
||||
const batch = args.candidates.slice(index, index + batchSize)
|
||||
const parseableBatch = opencodePhase.prepareBatch(batch)
|
||||
const results = await Promise.all(
|
||||
parseableBatch.map((candidate) =>
|
||||
parseSessionCandidate(
|
||||
candidate,
|
||||
args.platform,
|
||||
args.executionHostId,
|
||||
args.parseStats,
|
||||
args.antigravityWorkspaceResolver,
|
||||
args.opencodeSqliteScanContext
|
||||
)
|
||||
const parsePromises = parseableBatch.map((candidate) =>
|
||||
parseSessionCandidate(
|
||||
candidate,
|
||||
args.platform,
|
||||
args.executionHostId,
|
||||
args.parseStats,
|
||||
args.antigravityWorkspaceResolver,
|
||||
args.opencodeSqliteScanContext
|
||||
)
|
||||
)
|
||||
opencodePhase.completeBatch()
|
||||
opencodePhase.trackBatch(parseableBatch, parsePromises)
|
||||
const results = await Promise.all(parsePromises)
|
||||
|
||||
for (const result of results) {
|
||||
if (result.issue) {
|
||||
|
||||
Reference in New Issue
Block a user