diff --git a/src/main/ai-vault-search/session-search-dependency-work.test.ts b/src/main/ai-vault-search/session-search-dependency-work.test.ts index d00b8340665..f57995424f6 100644 --- a/src/main/ai-vault-search/session-search-dependency-work.test.ts +++ b/src/main/ai-vault-search/session-search-dependency-work.test.ts @@ -1,8 +1,5 @@ import './worker/session-search-worker.test-fixture' -import { - notifySessionSearchFileChange, - notifySessionSearchOverflow -} from './session-search-watcher.test-fixture' +import { notifySessionSearchFileChange } from './session-search-watcher.test-fixture' import { appendFile, mkdir, writeFile } from 'node:fs/promises' import { zstdCompressSync } from 'node:zlib' import { dirname, join } from 'node:path' @@ -19,128 +16,74 @@ import { openSessionSearchIndexerHarness } from './indexing/session-search-indexer-test-fixture' -it.each([40, 400])( - 'bounds one title append independently of %s index records and transcript owners', - async (population) => { - const harness = await openSessionSearchIndexerHarness('ss-title-input-work') - const clock = new FakeSessionSearchClock() - const root = harness.roots.codexSessionsDir! - const index = join(dirname(root), 'session_index.jsonl') - const fixture = codexFixture() - const ids = Array.from( - { length: population }, - (_, id) => `019f0000-1111-7222-8333-${String(id).padStart(12, '0')}` +it('updates an archived Codex title while retaining its neighboring conversation', async () => { + const harness = await openSessionSearchIndexerHarness('ss-archive-title') + const clock = new FakeSessionSearchClock() + const root = harness.roots.codexSessionsDir! + const index = join(dirname(root), 'session_index.jsonl') + const fixture = codexFixture() + const neighbor = '019f0000-1111-7222-8333-000000000001' + const subject = new SessionSearchIndexer({ + discoveryPolicy: 'event-led', + databasePath: harness.databasePath, + roots: harness.roots, + historyDays: null, + clock + }) + const titles = () => + harness.read((db) => + db.prepare('SELECT title,count(*) AS n FROM sessions GROUP BY title ORDER BY title').all() ) - const records = ids.map((id) => `${JSON.stringify({ id, thread_name: 'Original title' })}\n`) - const subject = new SessionSearchIndexer({ - discoveryPolicy: 'event-led', - databasePath: harness.databasePath, - roots: { ...harness.roots, kimiSessionsDir: join(harness.root, 'kimi-home', 'sessions') }, - historyDays: null, - clock - }) - const streams = vi.spyOn(filesystem, 'openTranscriptReadStream') - const slices = vi.spyOn(filesystem, 'readTranscriptSlice') - const traffic = vi.spyOn(SessionSearchWorkerClient.prototype, 'request') - const transcripts = new Set() - try { - await mkdir(root, { recursive: true }) - for (const id of ids) { - const archived = id === ids[0] - const path = join( - root, - fixture.fileName.replace(CODEX_FIXTURE_SESSION_ID, id) + (archived ? '.zst' : '') - ) - transcripts.add(path) - const text = `${fixture.seedLines.join('\n').replaceAll(CODEX_FIXTURE_SESSION_ID, id)}\n` - await writeFile(path, archived ? zstdCompressSync(Buffer.from(text)) : text) - } - await writeFile(index, records.join('')) - await subject.start() - streams.mockClear() - slices.mockClear() - // Lost delivery must re-observe the input, not reread it or force every transcript. - notifySessionSearchOverflow(dirname(root)) - clock.advance(20000) - await subject.settled() - expect(streams.mock.calls.filter(([path]) => path === index)).toHaveLength(0) - expect(streams.mock.calls.filter(([path]) => transcripts.has(path))).toHaveLength(0) - expect(slices.mock.calls.filter(([path]) => transcripts.has(path))).toHaveLength(0) - const append = `${JSON.stringify({ id: ids[0], thread_name: 'Changed title' })}\n` - const size = Buffer.byteLength(records.join('') + append) - await appendFile(index, append) - notifySessionSearchFileChange(index) - clock.advance(20000) - await subject.settled() - const indexBytes = - streams.mock.calls.reduce( - (bytes, [path, options]) => - path !== index - ? bytes - : bytes + Math.max(0, (options?.end ?? size - 1) - (options?.start ?? 0) + 1), - 0 - ) + - slices.mock.calls.reduce( - (bytes, [path, , length]) => (path === index ? bytes + length : bytes), - 0 - ) - // Two independent consumers need the appended record plus one fixed continuity record each. - expect(indexBytes).toBeLessThanOrEqual( - 2 * (Buffer.byteLength(append) + Buffer.byteLength(records[0]) + 1) - ) - expect(streams.mock.calls.filter(([path]) => transcripts.has(path))).toHaveLength(0) - expect(slices.mock.calls.filter(([path]) => transcripts.has(path))).toHaveLength(0) - expect( - harness.read((db) => - db.prepare("SELECT count(*) AS n FROM sessions WHERE title='Changed title'").get() - ) - ).toEqual({ n: 1 }) - expect( - harness.read((db) => - db.prepare("SELECT count(*) AS n FROM sessions WHERE title='Original title'").get() - ) - ).toEqual({ n: population - 1 }) - traffic.mockClear() - await writeFile( - index, - ids.map((id) => `${JSON.stringify({ id, thread_name: 'Replaced title' })}\n`).join('') - ) - notifySessionSearchFileChange(index) - clock.advance(20000) - await subject.settled() - expect( - harness.read((db) => - db.prepare("SELECT count(*) AS n FROM sessions WHERE title='Replaced title'").get() - ) - ).toEqual({ n: population }) - const firstRead = traffic.mock.calls.findIndex( - ([command]) => command.op === 'work' && command.command.kind === 'next' - ) - const fanout = traffic.mock.calls - .slice(0, firstRead === -1 ? undefined : firstRead) - .filter( - ([command]) => - command.op === 'catalog' && - (command.command.scope.startsWith(`dependency:codex:${root}:`) || - command.command.scope.startsWith(`physical:codex:${root}`)) && - (command.command.kind === 'dependency-fanout' || - command.command.kind === 'dependency-deliver' || - command.command.kind === 'dirty-page' || - command.command.kind === 'frontier-set' || - command.command.kind === 'get' || - command.command.kind === 'put') - ) - // Accepted changed keys, their reverse owners and physical observations each have one bounded page handoff, plus fixed source phases. - expect(fanout.length).toBeLessThanOrEqual(40 + 12 * Math.ceil(population / 256)) - } finally { - streams.mockRestore() - slices.mockRestore() - traffic.mockRestore() - await subject.close() - await harness.cleanup() - } + try { + await mkdir(root, { recursive: true }) + await writeFile( + join(root, `${fixture.fileName}.zst`), + zstdCompressSync(Buffer.from(`${fixture.seedLines.join('\n')}\n`)) + ) + await writeFile( + join(root, fixture.fileName.replace(CODEX_FIXTURE_SESSION_ID, neighbor)), + `${fixture.seedLines.join('\n').replaceAll(CODEX_FIXTURE_SESSION_ID, neighbor)}\n` + ) + await writeFile( + index, + [ + { id: CODEX_FIXTURE_SESSION_ID, thread_name: 'Original title' }, + { id: neighbor, thread_name: 'Original title' } + ] + .map((record) => `${JSON.stringify(record)}\n`) + .join('') + ) + await subject.start() + expect(titles()).toEqual([{ title: 'Original title', n: 2 }]) + await appendFile( + index, + `${JSON.stringify({ id: CODEX_FIXTURE_SESSION_ID, thread_name: 'Changed archive title' })}\n` + ) + notifySessionSearchFileChange(index) + clock.advance(20000) + await subject.settled() + expect(titles()).toEqual([ + { title: 'Changed archive title', n: 1 }, + { title: 'Original title', n: 1 } + ]) + await writeFile( + index, + [ + { id: CODEX_FIXTURE_SESSION_ID, thread_name: 'Replacement title' }, + { id: neighbor, thread_name: 'Replacement title' } + ] + .map((record) => `${JSON.stringify(record)}\n`) + .join('') + ) + notifySessionSearchFileChange(index) + clock.advance(20000) + await subject.settled() + expect(titles()).toEqual([{ title: 'Replacement title', n: 2 }]) + } finally { + await subject.close() + await harness.cleanup() } -) +}) it('settles a refused JSONL observer without losing its accepted title and wakes it on delivery', async () => { const harness = await openSessionSearchIndexerHarness('ss-refused-title-index')