diff --git a/config/scripts/session-search-pass-benchmark.ts b/config/scripts/session-search-pass-benchmark.ts new file mode 100644 index 00000000000..6576baa9589 --- /dev/null +++ b/config/scripts/session-search-pass-benchmark.ts @@ -0,0 +1,93 @@ +import assert from 'node:assert/strict' +import { mkdir, mkdtemp, rename, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { basename, join } from 'node:path' +import { isolatedScanRoots } from '../../src/main/ai-vault/session-scanner-test-fixtures' +import { resetSessionParseCacheForTests } from '../../src/main/ai-vault/session-scanner-parse-cache' +import { resetTranscriptConsumersForTests } from '../../src/main/ai-vault/session-transcript-consumers' +import { SessionSearchIndexer } from '../../src/main/ai-vault-search/session-search-indexer' +import { writeSyntheticTranscriptCorpus } from '../../src/main/ai-vault-search/session-search-synthetic-corpus' + +// Bundle with esbuild --bundle --platform=node, then run on the host under test. +// What a warm pass costs on a machine with a real number of transcripts: a cycle +// stats the newest N per agent, a sweep stats every file under every root, and +// neither reads anything the index already holds at its current stat. This is the +// number the reconcile interval is chosen against; it does not set one. +// Never point this at a real transcript tree. + +const SESSIONS = 5_000 +const TURNS_PER_SESSION = 4 +const PROJECTS = 40 + +const corpus = await writeSyntheticTranscriptCorpus({ + sessions: SESSIONS, + turnsPerSession: TURNS_PER_SESSION +}) +const root = await mkdtemp(join(tmpdir(), 'orca-search-pass-')) +const roots = isolatedScanRoots(root) +const databasePath = join(root, 'index', 'session-search.sqlite') + +try { + // A flat corpus is not what discovery walks: spread it over project directories + // so the readdir count is realistic rather than one enormous listing. + for (let index = 0; index < PROJECTS; index++) { + await mkdir(join(roots.claudeProjectsDir, `project-${index}`), { recursive: true }) + } + await Promise.all( + corpus.files.map((path, index) => + rename(path, join(roots.claudeProjectsDir, `project-${index % PROJECTS}`, basename(path))) + ) + ) + + resetSessionParseCacheForTests() + resetTranscriptConsumersForTests() + const errors: unknown[] = [] + const indexer = new SessionSearchIndexer({ + databasePath, + roots, + historyDays: null, + // No wall-clock ceiling: the cold build has to finish before a warm pass can + // be measured, and a deadline would leave a backlog priced into every number. + passDeadlineMs: Number.MAX_SAFE_INTEGER, + onError: (error) => errors.push(error) + }) + try { + const coldStarted = performance.now() + await indexer.start() + const coldMs = performance.now() - coldStarted + assert.deepEqual(errors, []) + assert.equal(indexer.status().filesIndexed, SESSIONS, 'indexed file count') + + const sweepStarted = performance.now() + await indexer.reconcile({ full: true }) + const sweepMs = performance.now() - sweepStarted + + const cycleStarted = performance.now() + await indexer.reconcile({ full: false }) + const cycleMs = performance.now() - cycleStarted + + assert.deepEqual(errors, []) + assert.equal(indexer.status().filesDue, 0, 'nothing owed after a warm sweep') + console.log( + JSON.stringify( + { + transcripts: SESSIONS, + projectDirectories: PROJECTS, + transcriptMb: Math.round((corpus.transcriptBytes / (1024 * 1024)) * 100) / 100, + coldBuildMs: Math.round(coldMs), + warmSweepMs: Math.round(sweepMs), + warmCycleMs: Math.round(cycleMs) + }, + null, + 2 + ) + ) + } finally { + indexer.close() + resetTranscriptConsumersForTests() + resetSessionParseCacheForTests() + } +} finally { + await rm(corpus.root, { recursive: true, force: true }) + await rm(root, { recursive: true, force: true }) +} diff --git a/src/main/ai-vault-search/session-search-child-service.test.ts b/src/main/ai-vault-search/session-search-child-service.test.ts new file mode 100644 index 00000000000..ddd3e8cb03b --- /dev/null +++ b/src/main/ai-vault-search/session-search-child-service.test.ts @@ -0,0 +1,55 @@ +import { expect, it, vi } from 'vitest' +import { unavailableSessionSearchStatus } from '../../shared/ai-vault-search-client' +import type { AiVaultSearchStatus } from '../../shared/ai-vault-search-types' +import { createChildSessionSearchService } from './session-search-child-service' + +const indexingStatus: AiVaultSearchStatus = { + ...unavailableSessionSearchStatus(), + enabled: true, + phase: 'indexing', + filesIndexed: 3, + generation: 7 +} + +function stubCalls(overrides: Partial[0]> = {}) { + return { + search: vi.fn(async () => ({ kind: 'unavailable', reason: 'disabled' }) as const), + status: vi.fn(async () => indexingStatus), + reconcile: vi.fn(async () => undefined), + ...overrides + } +} + +it('forwards every call to the child and returns what it answered', async () => { + const calls = stubCalls() + const service = createChildSessionSearchService(calls) + + expect(await service.search({ query: 'ledger' })).toEqual({ + kind: 'unavailable', + reason: 'disabled' + }) + expect(calls.search).toHaveBeenCalledWith({ query: 'ledger' }) + expect(await service.status()).toEqual(indexingStatus) + await service.reconcile() + expect(calls.reconcile).toHaveBeenCalledTimes(1) +}) + +// A child that is starting, restarting or refusing is "not yet", which is an +// answer to the caller's question; turning it into a throw would make a paired +// client show a transport error for a host that is simply booting. +it('maps a child that cannot answer to not-ready rather than an error', async () => { + const service = createChildSessionSearchService( + stubCalls({ + search: vi.fn(() => Promise.reject(new Error('AI Vault service did not become ready.'))), + status: vi.fn(() => Promise.reject(new Error('AI Vault service queue is full.'))), + reconcile: vi.fn(() => Promise.reject(new Error('AI Vault service disconnected.'))) + }) + ) + + expect(await service.search({ query: 'ledger' })).toEqual({ + kind: 'unavailable', + reason: 'not-ready' + }) + expect(await service.status()).toEqual(unavailableSessionSearchStatus()) + await expect(service.reconcile()).resolves.toBeUndefined() +}) diff --git a/src/main/ai-vault-search/session-search-child-service.ts b/src/main/ai-vault-search/session-search-child-service.ts new file mode 100644 index 00000000000..379efeff4b1 --- /dev/null +++ b/src/main/ai-vault-search/session-search-child-service.ts @@ -0,0 +1,47 @@ +import type { AiVaultSearchStatus } from '../../shared/ai-vault-search-types' +import { unavailableSessionSearchStatus } from '../../shared/ai-vault-search-client' +import { + reconcileSessionSearchInService, + searchSessionsInService, + sessionSearchStatusInService +} from '../ai-vault/session-scanner-service-spawn' +import type { SessionSearchService } from './session-search-service' + +/** + * The desktop's `SessionSearchService`: every call is forwarded to the scanner + * child that owns the database. This process never opens the index file. + * + * A transport failure is a child that is starting, restarting or refusing, which + * is `not-ready` rather than an error: the caller asked whether this host can + * answer, and "not yet" is an answer. A child that is up and has no indexer says + * `disabled` for itself. + */ +export function createChildSessionSearchService( + calls = { + search: searchSessionsInService, + status: sessionSearchStatusInService, + reconcile: reconcileSessionSearchInService + } +): SessionSearchService { + return { + search: async (request) => { + try { + return await calls.search(request) + } catch { + return { kind: 'unavailable', reason: 'not-ready' } + } + }, + status: async (): Promise => { + try { + return await calls.status() + } catch { + return unavailableSessionSearchStatus() + } + }, + reconcile: async () => { + // Swallowed for the same reason: the caller's next search reports the state + // of the index, and a freshness wait that cannot run is a stale page, not a throw. + await calls.reconcile().catch(() => undefined) + } + } +} diff --git a/src/main/ai-vault-search/session-search-database-path.ts b/src/main/ai-vault-search/session-search-database-path.ts new file mode 100644 index 00000000000..5fcaf405f57 --- /dev/null +++ b/src/main/ai-vault-search/session-search-database-path.ts @@ -0,0 +1,13 @@ +import { join } from 'node:path' + +/** + * Where one host keeps its index. + * + * Beside the scanner's parse cache (`/ai-vault/`), because the two are + * the same kind of thing: a disposable derivative of the transcripts this host + * can read, scoped to this host's data root. One file per host, never shared — + * a second process writing the same file is the rebuild race PR 2 recorded. + */ +export function sessionSearchDatabasePath(dataRoot: string): string { + return join(dataRoot, 'ai-vault', 'session-search.sqlite') +} diff --git a/src/main/ai-vault-search/session-search-enablement.ts b/src/main/ai-vault-search/session-search-enablement.ts new file mode 100644 index 00000000000..e292e4a8551 --- /dev/null +++ b/src/main/ai-vault-search/session-search-enablement.ts @@ -0,0 +1,70 @@ +import { + resolveAiVaultSearchSettings, + sameAiVaultSearchSettings +} from '../../shared/ai-vault-search-settings' +import type { GlobalSettings } from '../../shared/global-settings-types' +import { updateSessionSearchInService } from '../ai-vault/session-scanner-service-spawn' +import { createChildSessionSearchService } from './session-search-child-service' +import { installSessionSearchPolicySource } from './session-search-policy' +import { setSessionSearchService } from './session-search-service-registry' +import { + installSessionSearchDataRoot, + sessionSearchServiceInit +} from './session-search-service-init' +import { sessionSearchSqliteAvailable } from './session-search-sqlite-support' +let installed = false + +/** + * The desktop's one wiring point: search answers from the scanner child, and the + * child's consent comes from the settings store. + * + * Registered whether or not the setting is on, because "off" is an answer this + * host can give (`unavailable/disabled`) and `no-service` is not — that reason + * means nothing here owns an index, which stops being true the moment this runs. + */ +export function installChildSessionSearchService(args: { + dataRoot: string + getSettings: () => Pick +}): { dispose(): void } | null { + if (!sessionSearchSqliteAvailable()) { + return null + } + installed = true + installSessionSearchDataRoot(args.dataRoot) + installSessionSearchPolicySource(args.getSettings) + setSessionSearchService(createChildSessionSearchService()) + pushSessionSearchPolicy() + return { + dispose: () => { + installed = false + } + } +} + +/** + * Reconciles a settings write. An unchanged policy is not forwarded, so re-saving + * the same value never restarts a running index. + */ +export function applySessionSearchSettingsChange( + before: Pick, + after: Pick +): void { + if ( + sameAiVaultSearchSettings( + resolveAiVaultSearchSettings(before), + resolveAiVaultSearchSettings(after) + ) + ) { + return + } + if (installed) { + pushSessionSearchPolicy() + } +} + +function pushSessionSearchPolicy(): void { + const init = sessionSearchServiceInit() + if (init) { + updateSessionSearchInService(init) + } +} diff --git a/src/main/ai-vault-search/session-search-host-registration.test.ts b/src/main/ai-vault-search/session-search-host-registration.test.ts new file mode 100644 index 00000000000..0e42ac68cde --- /dev/null +++ b/src/main/ai-vault-search/session-search-host-registration.test.ts @@ -0,0 +1,206 @@ +import { readFileSync } from 'node:fs' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { resetSessionParseCacheForTests } from '../ai-vault/session-scanner-parse-cache' +import { resetTranscriptConsumersForTests } from '../ai-vault/session-transcript-consumers' +import { installInProcessSessionSearchService } from './session-search-in-process-service' +import { + openSessionSearchIndexerHarness, + writeClaudeTranscript, + type SessionSearchIndexerHarness +} from './session-search-indexer-test-fixture' +import { searchSessionService } from './session-search-service-registry' +import { resetSessionSearchPolicyForTests } from './session-search-policy' +import { resetSessionSearchServiceInitForTests } from './session-search-service-init' + +/** + * Every host that answers a search has to register a service, or its answer is + * `no-service` — which means "this host does not have the feature", not "it is + * off". Two halves: the installers really register, and each host's boot module + * really calls the installer that suits it. + */ + +const updateSessionSearchInService = vi.hoisted(() => vi.fn()) +vi.mock('../ai-vault/session-scanner-service-spawn', async (importOriginal) => ({ + ...(await importOriginal()), + updateSessionSearchInService +})) + +const localAiVaultScanRoots = vi.hoisted(() => vi.fn()) +vi.mock('../ai-vault/cached-session-list', async (importOriginal) => ({ + ...(await importOriginal()), + localAiVaultScanRoots +})) + +const ROOT = join(import.meta.dirname, '..', '..', '..') + +let harness: SessionSearchIndexerHarness +let installed: { dispose(): void } | null + +beforeEach(async () => { + resetSessionParseCacheForTests() + resetTranscriptConsumersForTests() + updateSessionSearchInService.mockClear() + harness = await openSessionSearchIndexerHarness('ss-registration') + installed = null + localAiVaultScanRoots.mockReset().mockResolvedValue(harness.roots) +}) + +afterEach(async () => { + installed?.dispose() + vi.useRealTimers() + const { setSessionSearchService } = await import('./session-search-service-registry') + setSessionSearchService(null) + resetSessionSearchPolicyForTests() + resetSessionSearchServiceInitForTests() + resetTranscriptConsumersForTests() + resetSessionParseCacheForTests() + await harness.cleanup() +}) + +it('answers no-service until a host registers one', async () => { + expect(await searchSessionService({ query: 'ledger' }, 'ipc')).toEqual({ + kind: 'unavailable', + reason: 'no-service' + }) +}) + +it('registers the desktop service and pushes the stored policy at boot', async () => { + const { installChildSessionSearchService } = await import('./session-search-enablement') + installed = installChildSessionSearchService({ + dataRoot: harness.root, + getSettings: () => ({ aiVaultSearch: { enabled: true, historyDays: 30 } }) + }) + + expect(await searchSessionService({ query: 'ledger' }, 'ipc')).not.toEqual({ + kind: 'unavailable', + reason: 'no-service' + }) + await vi.waitFor(() => expect(updateSessionSearchInService).toHaveBeenCalledTimes(1)) + expect(updateSessionSearchInService.mock.calls[0]?.[0]).toMatchObject({ + settings: { enabled: true, historyDays: 30 }, + databasePath: join(harness.root, 'ai-vault', 'session-search.sqlite') + }) +}) + +it('forwards only a real settings change to the child', async () => { + const { applySessionSearchSettingsChange, installChildSessionSearchService } = + await import('./session-search-enablement') + installed = installChildSessionSearchService({ + dataRoot: harness.root, + getSettings: () => ({ aiVaultSearch: { enabled: false, historyDays: null } }) + }) + await vi.waitFor(() => expect(updateSessionSearchInService).toHaveBeenCalledTimes(1)) + + applySessionSearchSettingsChange( + { aiVaultSearch: { enabled: false, historyDays: null } }, + { aiVaultSearch: { enabled: false, historyDays: null } } + ) + expect(updateSessionSearchInService).toHaveBeenCalledTimes(1) + + applySessionSearchSettingsChange( + { aiVaultSearch: { enabled: false, historyDays: null } }, + { aiVaultSearch: { enabled: true, historyDays: null } } + ) + await vi.waitFor(() => expect(updateSessionSearchInService).toHaveBeenCalledTimes(2)) +}) + +it('does not discover roots or arm a timer during registration', async () => { + vi.useFakeTimers() + const { installChildSessionSearchService } = await import('./session-search-enablement') + installed = installChildSessionSearchService({ + dataRoot: harness.root, + getSettings: () => ({ aiVaultSearch: { enabled: false, historyDays: null } }) + }) + expect(updateSessionSearchInService).toHaveBeenCalledTimes(1) + await vi.advanceTimersByTimeAsync(600_000) + expect(localAiVaultScanRoots).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) +}) + +it('registers an in-process service for a host with no scanner child', async () => { + installed = installInProcessSessionSearchService({ + dataRoot: harness.root, + roots: harness.roots, + settings: { enabled: false, historyDays: null } + }) + expect(installed).not.toBeNull() + + // Off, not absent: the caller can tell consent from a host that lacks the feature. + expect(await searchSessionService({ query: 'ledger' }, 'relay')).toEqual({ + kind: 'unavailable', + reason: 'disabled' + }) + + installed?.dispose() + installed = null + expect(await searchSessionService({ query: 'ledger' }, 'relay')).toEqual({ + kind: 'unavailable', + reason: 'no-service' + }) +}) + +// The behavioural tests above prove the installers register; these prove each +// host's boot path reaches one, which no unit of either module can show. +it.each([ + [ + 'desktop and headless serve', + 'src/main/startup/main-process-runtime-service.ts', + 'installChildSessionSearchService' + ], + ['orcad', 'src/main/orcad/orcad-session-search.ts', 'installInProcessSessionSearchService'], + [ + 'the relay daemon', + 'src/relay/relay-runtime-services.ts', + 'installInProcessSessionSearchService' + ] +])('boots %s with a registered session search service', (_host, file, installer) => { + const source = readFileSync(join(ROOT, file), 'utf8') + expect(source).toContain(installer) + expect(source).toMatch(new RegExp(`${installer}\\(\\{`)) +}) + +it('disables immediately without root discovery', async () => { + const { installChildSessionSearchService, applySessionSearchSettingsChange } = + await import('./session-search-enablement') + let settings = { aiVaultSearch: { enabled: true, historyDays: null } } + installed = installChildSessionSearchService({ + dataRoot: harness.root, + getSettings: () => settings + }) + updateSessionSearchInService.mockClear() + const before = settings + settings = { aiVaultSearch: { enabled: false, historyDays: null } } + applySessionSearchSettingsChange(before, settings) + expect(updateSessionSearchInService).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ settings: settings.aiVaultSearch }) + ) + expect(localAiVaultScanRoots).not.toHaveBeenCalled() +}) + +it('orcad resolves no roots while disabled and discovers late roots when enabled', async () => { + const { installOrcadSessionSearchService } = await import('../orcad/orcad-session-search') + installed = await installOrcadSessionSearchService({ + userDataPath: harness.root, + getSettings: () => ({ aiVaultSearch: { enabled: false, historyDays: null } }) + }) + expect(localAiVaultScanRoots).not.toHaveBeenCalled() + installed?.dispose() + installed = await installOrcadSessionSearchService({ + userDataPath: harness.root, + getSettings: () => ({ aiVaultSearch: { enabled: true, historyDays: null } }) + }) + await searchSessionService({ query: 'latehostroot', freshness: 'wait-until-current' }, 'ipc') + const late = join(harness.root, 'late-claude') + const id = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' + await writeClaudeTranscript(join(late, 'project', `${id}.jsonl`), ['latehostroot'], id) + localAiVaultScanRoots.mockResolvedValue({ ...harness.roots, claudeProjectsDir: late }) + const response = await searchSessionService( + { query: 'latehostroot', freshness: 'wait-until-current' }, + 'ipc' + ) + expect(response.kind).toBe('results') + if (response.kind === 'results') { + expect(response.hits.map((hit) => hit.sessionId)).toEqual([id]) + } +}) diff --git a/src/main/ai-vault-search/session-search-in-process-service.ts b/src/main/ai-vault-search/session-search-in-process-service.ts new file mode 100644 index 00000000000..536ad742a98 --- /dev/null +++ b/src/main/ai-vault-search/session-search-in-process-service.ts @@ -0,0 +1,53 @@ +import type { AiVaultSearchSettings } from '../../shared/ai-vault-search-settings' +import { sessionSearchDatabasePath } from './session-search-database-path' +import type { SessionSearchIndexerOptions } from './session-search-indexer-options' +import { SessionSearchInstance } from './session-search-instance' +import type { SessionSearchScanRoots } from './session-search-scan-roots' +import { setSessionSearchService } from './session-search-service-registry' +import { sessionSearchSqliteAvailable } from './session-search-sqlite-support' + +/** + * Registration for the two hosts that have no scanner-service child of their own. + * + * The desktop puts the index in that child because the child is where the + * transcript reader runs, so one read serves both the session list and the index. + * Neither of these hosts has that child: orcad ships only the watcher and daemon + * entries beside `orcad.js`, and the relay's AI Vault sidecar runs the remote + * scanner, which reads through a filesystem provider and publishes nothing to the + * transcript channel. On both, the process that would drive the index's reads is + * this one, and it is the only writer, so the two-process rebuild race the + * desktop rule avoids cannot arise here. + * + * Returns null on a runtime with no `node:sqlite`: both hosts are built for a + * Node 18 floor, and a host that cannot hold an index registers nothing rather + * than answering `disabled` for a reason that is not consent. + */ +export function installInProcessSessionSearchService(args: { + dataRoot: string + roots: SessionSearchScanRoots + resolveRoots?: SessionSearchIndexerOptions['resolveRoots'] + settings: AiVaultSearchSettings + onError?: (error: unknown) => void +}): { dispose(): void } | null { + if (!sessionSearchSqliteAvailable()) { + return null + } + const instance = new SessionSearchInstance({ + databasePath: sessionSearchDatabasePath(args.dataRoot), + roots: args.roots, + resolveRoots: args.resolveRoots, + ...(args.onError ? { onError: args.onError } : {}) + }) + instance.apply(args.settings) + setSessionSearchService({ + search: (request) => instance.search(request), + status: async () => instance.status(), + reconcile: () => instance.reconcile() + }) + return { + dispose: () => { + setSessionSearchService(null) + instance.close() + } + } +} diff --git a/src/main/ai-vault-search/session-search-indexer-options.ts b/src/main/ai-vault-search/session-search-indexer-options.ts index 08eb4aeb2af..2c32c8381da 100644 --- a/src/main/ai-vault-search/session-search-indexer-options.ts +++ b/src/main/ai-vault-search/session-search-indexer-options.ts @@ -36,6 +36,8 @@ export const DEFAULT_SESSION_SEARCH_FULL_SWEEP_EVERY_CYCLES = 15 export type SessionSearchIndexerOptions = { databasePath: string roots: SessionSearchScanRoots + /** Full sweeps refresh host roots; recent cycles reuse the last snapshot. */ + resolveRoots?: (signal: AbortSignal) => Promise /** null = all history; otherwise only transcripts modified within this many days. */ historyDays: number | null clock?: SessionSearchClock diff --git a/src/main/ai-vault-search/session-search-indexer.ts b/src/main/ai-vault-search/session-search-indexer.ts index 26213242c49..d70d3718f9c 100644 --- a/src/main/ai-vault-search/session-search-indexer.ts +++ b/src/main/ai-vault-search/session-search-indexer.ts @@ -58,6 +58,7 @@ export type SessionSearchIndexStatus = { * counter with a reset rule. * * What is left here, and why none of it can be a row: + * - `roots`, the latest full-sweep snapshot reused by recent cycles. * - `previousRootsWithFiles`, the one bit per root the retirement walk's grace * needs. Deliberately not durable: see the mountpoint trade in * `session-search-deleted-sources.ts`. @@ -81,6 +82,7 @@ export type SessionSearchIndexStatus = { * one reconcile interval. Everything else is reached by the periodic sweep. */ export class SessionSearchIndexer { + private roots: SessionSearchIndexerOptions['roots'] private readonly ownershipPath: string private readonly clock: SessionSearchClock private readonly intervalMs: number @@ -104,6 +106,7 @@ export class SessionSearchIndexer { private closed = false constructor(private readonly options: SessionSearchIndexerOptions) { + this.roots = options.roots this.ownershipPath = resolve(options.databasePath) this.clock = options.clock ?? systemSessionSearchClock this.intervalMs = options.reconcileIntervalMs ?? DEFAULT_SESSION_SEARCH_RECONCILE_INTERVAL_MS @@ -273,9 +276,16 @@ export class SessionSearchIndexer { // end would erase that request along with this pass's own. this.sweepNext = false try { + if (full && this.options.resolveRoots) { + const roots = await this.options.resolveRoots(signal) + if (signal.aborted) { + return + } + this.roots = roots + } const result = await runSessionSearchPass({ store: this.store, - roots: this.options.roots, + roots: this.roots, full, recentPerAgent: this.recentPerAgent, previousRootsWithFiles: this.previousRootsWithFiles ?? undefined, diff --git a/src/main/ai-vault-search/session-search-instance.test.ts b/src/main/ai-vault-search/session-search-instance.test.ts new file mode 100644 index 00000000000..c917bd2d8a9 --- /dev/null +++ b/src/main/ai-vault-search/session-search-instance.test.ts @@ -0,0 +1,195 @@ +import { existsSync } from 'node:fs' +import { utimes } from 'node:fs/promises' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { resetSessionParseCacheForTests } from '../ai-vault/session-scanner-parse-cache' +import { resetTranscriptConsumersForTests } from '../ai-vault/session-transcript-consumers' +import { SessionSearchInstance } from './session-search-instance' +import { + openSessionSearchIndexerHarness, + writeClaudeTranscript, + type SessionSearchIndexerHarness +} from './session-search-indexer-test-fixture' + +const RECENT_SESSION_ID = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' +const ANCIENT_SESSION_ID = 'bbbbbbbb-cccc-4ddd-8eee-ffffffffffff' + +let harness: SessionSearchIndexerHarness +let instance: SessionSearchInstance | null +let errors: unknown[] + +beforeEach(async () => { + resetSessionParseCacheForTests() + resetTranscriptConsumersForTests() + errors = [] + harness = await openSessionSearchIndexerHarness('ss-instance') + instance = null +}) + +afterEach(async () => { + vi.restoreAllMocks() + instance?.close() + resetTranscriptConsumersForTests() + resetSessionParseCacheForTests() + await harness.cleanup() +}) + +function newInstance(): SessionSearchInstance { + instance = new SessionSearchInstance({ + databasePath: harness.databasePath, + roots: harness.roots, + onError: (error) => errors.push(error) + }) + return instance +} + +function transcriptPath(sessionId: string): string { + return join(harness.claudeProjectDir, `${sessionId}.jsonl`) +} + +async function searchFor(query: string): Promise { + const response = await instance!.search({ query }) + if (response.kind !== 'results') { + throw new Error(`expected results, got ${response.kind}`) + } + return response.hits.map((hit) => hit.sessionId).sort() +} + +it('constructs nothing and touches no disk while the setting is off', async () => { + await writeClaudeTranscript( + transcriptPath(RECENT_SESSION_ID), + ['a conversation'], + RECENT_SESSION_ID + ) + const subject = newInstance() + subject.apply({ enabled: false, historyDays: null }) + await subject.settled() + + expect(subject.running).toBe(false) + expect(existsSync(harness.databasePath)).toBe(false) + expect(await subject.search({ query: 'conversation' })).toEqual({ + kind: 'unavailable', + reason: 'disabled' + }) + expect(subject.status()).toMatchObject({ enabled: false, phase: 'idle', generation: 0 }) + expect(errors).toEqual([]) +}) + +it('indexes and answers once the setting is on', async () => { + await writeClaudeTranscript( + transcriptPath(RECENT_SESSION_ID), + ['a distinctive conversation'], + RECENT_SESSION_ID + ) + const subject = newInstance() + subject.apply({ enabled: true, historyDays: null }) + await subject.settled() + + expect(await searchFor('distinctive')).toEqual([RECENT_SESSION_ID]) + const status = subject.status() + expect(status.enabled).toBe(true) + expect(status.filesIndexed).toBeGreaterThan(0) + expect(status.generation).toBeGreaterThan(0) + expect(errors).toEqual([]) +}) + +// The whole reason the indexer is immutable: a change is a new instance, and the +// old one is closed before it exists, so there is never a second writer. +it('closes the live pair and starts a new one on a settings change', async () => { + const recent = transcriptPath(RECENT_SESSION_ID) + const ancient = transcriptPath(ANCIENT_SESSION_ID) + await writeClaudeTranscript(recent, ['a recent conversation'], RECENT_SESSION_ID) + await writeClaudeTranscript(ancient, ['an ancient conversation'], ANCIENT_SESSION_ID) + const longAgo = new Date(Date.now() - 120 * 86_400_000) + await utimes(ancient, longAgo, longAgo) + + const subject = newInstance() + subject.apply({ enabled: true, historyDays: null }) + await subject.settled() + expect(await searchFor('ancient')).toEqual([ANCIENT_SESSION_ID]) + + // Narrowing: the new instance's opening sweep purges what the window no longer covers. + subject.apply({ enabled: true, historyDays: 30 }) + await subject.settled() + expect(await searchFor('ancient')).toEqual([]) + expect(await searchFor('recent')).toEqual([RECENT_SESSION_ID]) + + // Widening: the same recipe the other way, admitting files no read ever saw. + subject.apply({ enabled: true, historyDays: null }) + await subject.settled() + expect(await searchFor('ancient')).toEqual([ANCIENT_SESSION_ID]) + expect(errors).toEqual([]) +}) + +it('leaves nothing running and no live claim when the setting goes off', async () => { + await writeClaudeTranscript( + transcriptPath(RECENT_SESSION_ID), + ['a conversation'], + RECENT_SESSION_ID + ) + const subject = newInstance() + subject.apply({ enabled: true, historyDays: null }) + await subject.settled() + expect(subject.running).toBe(true) + + subject.apply({ enabled: false, historyDays: null }) + expect(subject.running).toBe(false) + // The index is left on disk: disabling is not a deletion, and the claim the + // closed indexer staked on the path has to be released or nothing can reopen it. + expect(existsSync(harness.databasePath)).toBe(true) + subject.apply({ enabled: true, historyDays: null }) + await subject.settled() + expect(subject.running).toBe(true) + expect(errors).toEqual([]) +}) + +it('removes the database on clear and rebuilds only while consent stands', async () => { + await writeClaudeTranscript( + transcriptPath(RECENT_SESSION_ID), + ['a distinctive conversation'], + RECENT_SESSION_ID + ) + const subject = newInstance() + subject.apply({ enabled: true, historyDays: null }) + await subject.settled() + expect(await searchFor('distinctive')).toEqual([RECENT_SESSION_ID]) + + subject.clear() + expect(subject.running).toBe(true) + expect(existsSync(harness.databasePath)).toBe(true) + await subject.settled() + expect(await searchFor('distinctive')).toEqual([RECENT_SESSION_ID]) + + subject.apply({ enabled: false, historyDays: null }) + subject.clear() + expect(subject.running).toBe(false) + expect(existsSync(harness.databasePath)).toBe(false) + expect(errors).toEqual([]) +}) + +it('keeps pagination stable when the clock crosses retention before a purge', async () => { + for (const id of [RECENT_SESSION_ID, ANCIENT_SESSION_ID]) { + await writeClaudeTranscript(transcriptPath(id), [`distinctive conversation ${id}`], id) + } + const subject = newInstance() + subject.apply({ enabled: true, historyDays: 30 }) + await subject.settled() + const first = await subject.search({ query: 'distinctive', limit: 1 }) + if (first.kind !== 'results') { + throw new Error('expected results') + } + expect(first.page.cursor).toBeTruthy() + vi.spyOn(Date, 'now').mockReturnValue(Date.now() + 31 * 86_400_000) + const second = await subject.search({ + query: 'distinctive', + limit: 1, + cursor: first.page.cursor! + }) + if (second.kind !== 'results') { + throw new Error('expected results') + } + expect(second.generation).toBe(first.generation) + expect(second.hits).toHaveLength(1) + expect(second.hits[0].sessionId).not.toBe(first.hits[0].sessionId) + expect(errors).toEqual([]) +}) diff --git a/src/main/ai-vault-search/session-search-instance.ts b/src/main/ai-vault-search/session-search-instance.ts new file mode 100644 index 00000000000..b17103b004b --- /dev/null +++ b/src/main/ai-vault-search/session-search-instance.ts @@ -0,0 +1,162 @@ +import type SyncDatabase from '../sqlite/sync-database' +import type { + AiVaultSearchRequest, + AiVaultSearchResponse, + AiVaultSearchStatus +} from '../../shared/ai-vault-search-types' +import { unavailableSessionSearchStatus } from '../../shared/ai-vault-search-client' +import type { AiVaultSearchSettings } from '../../shared/ai-vault-search-settings' +import { SessionSearchEngine } from './session-search-engine' +import { SessionSearchIndexer } from './session-search-indexer' +import { sessionSearchHistoryCutoffMs } from './session-search-retention-policy' +import { openSessionSearchDatabase, removeSessionSearchDatabase } from './session-search-schema' +import type { SessionSearchScanRoots } from './session-search-scan-roots' +import type { SessionSearchIndexerOptions } from './session-search-indexer-options' +import { createSessionSearchService, type SessionSearchService } from './session-search-service' + +export type SessionSearchInstanceOptions = { + databasePath: string + roots: SessionSearchScanRoots + resolveRoots?: SessionSearchIndexerOptions['resolveRoots'] + onError?: (error: unknown) => void + /** Tests only: shortens the loop so a settings change is observable in one tick. */ + reconcileIntervalMs?: number +} + +type LiveIndex = { + indexer: SessionSearchIndexer + engine: SessionSearchEngine + /** The engine's own handle; the indexer's store keeps a second, private one. */ + db: SyncDatabase + service: SessionSearchService +} + +/** + * The one object that holds a host's live indexer and engine, and the three + * recipes that change them. + * + * The indexer is immutable after construction, so there is nothing here that + * reconfigures one: a settings change is `close()` and a new instance, disabling + * is `close()` with no replacement, and clearing is `close()`, remove the + * database, construct again. The new instance's first sweep purges a narrowed + * window and admits a widened one, so neither of those needs a path of its own. + * + * Lives in whichever process runs the transcript reader for this host. Nothing + * here knows about IPC, Electron or a settings store; the caller supplies the + * resolved settings and scan roots. + */ +export class SessionSearchInstance { + private live: LiveIndex | null = null + private settings: AiVaultSearchSettings = { enabled: false, historyDays: null } + private readonly onError: (error: unknown) => void + + constructor(private readonly options: SessionSearchInstanceOptions) { + this.onError = options.onError ?? ((error) => console.warn('[ai-vault-search]', error)) + } + + /** True once an indexer exists; false while disabled or while a construction is failing. */ + get running(): boolean { + return this.live !== null + } + + /** Close whatever is live and construct from `next`. A no-op change still restarts. */ + apply(next: AiVaultSearchSettings): void { + this.settings = next + this.closeLive() + this.construct() + } + + /** Throw the index away, then rebuild it if consent still stands. */ + clear(): void { + this.closeLive() + removeSessionSearchDatabase(this.options.databasePath) + this.construct() + } + + close(): void { + this.closeLive() + } + + async search(request: AiVaultSearchRequest): Promise { + const live = this.live + if (!live) { + return { kind: 'unavailable', reason: this.settings.enabled ? 'not-ready' : 'disabled' } + } + return live.service.search(request) + } + + status(): AiVaultSearchStatus { + const live = this.live + if (!live) { + return { ...unavailableSessionSearchStatus(), enabled: this.settings.enabled } + } + return { + enabled: true, + ...live.indexer.status(), + generation: live.engine.generation() + } + } + + async reconcile(): Promise { + await this.live?.service.reconcile() + } + + /** Tests only: resolves once the work loop has no pass in flight. */ + settled(): Promise { + return this.live?.indexer.settled() ?? Promise.resolve() + } + + private construct(): void { + if (!this.settings.enabled) { + return + } + const { historyDays } = this.settings + let indexer: SessionSearchIndexer | null = null + let db: SyncDatabase | null = null + try { + indexer = new SessionSearchIndexer({ + databasePath: this.options.databasePath, + roots: this.options.roots, + resolveRoots: this.options.resolveRoots, + historyDays, + onError: this.onError, + ...(this.options.reconcileIntervalMs === undefined + ? {} + : { reconcileIntervalMs: this.options.reconcileIntervalMs }) + }) + db = openSessionSearchDatabase(this.options.databasePath) + // Later expiry comes from the indexer purge, which also invalidates page cursors. + const engineOptions = { + retentionCutoffMs: sessionSearchHistoryCutoffMs(historyDays, Date.now()) + } + const engine = new SessionSearchEngine(db, engineOptions) + this.live = { + indexer, + engine, + db, + service: createSessionSearchService({ engine, indexer }) + } + void indexer.start().catch(this.onError) + } catch (error) { + // A failed open must leave nothing half-built: the indexer stakes the + // database path when its store opens, and only close() releases it. + db?.close() + indexer?.close() + this.live = null + this.onError(error) + } + } + + private closeLive(): void { + const live = this.live + this.live = null + if (!live) { + return + } + try { + live.indexer.close() + } finally { + live.db.close() + } + } +} diff --git a/src/main/ai-vault-search/session-search-policy.ts b/src/main/ai-vault-search/session-search-policy.ts new file mode 100644 index 00000000000..9f706f35fea --- /dev/null +++ b/src/main/ai-vault-search/session-search-policy.ts @@ -0,0 +1,25 @@ +import { + DEFAULT_AI_VAULT_SEARCH_SETTINGS, + resolveAiVaultSearchSettings, + type AiVaultSearchSettings +} from '../../shared/ai-vault-search-settings' +import type { GlobalSettings } from '../../shared/global-settings-types' + +// Why a source and not a captured value: the scanner child is spawned lazily and +// respawned after a fault, so its init frame has to read consent at spawn time. +// Before a composition root installs one, every read is the safe default (off). +let readSettings: (() => AiVaultSearchSettings) | null = null + +export function installSessionSearchPolicySource( + source: (() => Pick) | null +): void { + readSettings = source ? () => resolveAiVaultSearchSettings(source()) : null +} + +export function sessionSearchPolicy(): AiVaultSearchSettings { + return readSettings?.() ?? DEFAULT_AI_VAULT_SEARCH_SETTINGS +} + +export function resetSessionSearchPolicyForTests(): void { + readSettings = null +} diff --git a/src/main/ai-vault-search/session-search-retention-policy.ts b/src/main/ai-vault-search/session-search-retention-policy.ts index c7fa8a0b6d1..f6b7c483d48 100644 --- a/src/main/ai-vault-search/session-search-retention-policy.ts +++ b/src/main/ai-vault-search/session-search-retention-policy.ts @@ -1,25 +1,12 @@ -const DAY_MS = 86_400_000 -const HISTORY_DAYS_MAX = 3_650 +import { normalizeAiVaultSearchHistoryDays } from '../../shared/ai-vault-search-settings' -/** - * The retention window, as the indexer's callers state it and as the store - * consumes it. Settings storage is PR 3b's problem; this is the arithmetic. - */ -function normalizeSessionSearchHistoryDays(value: number | null): number | null { - if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) { - return null - } - // Why floor then re-check: a fractional day floors to 0, which reads as "all - // history" on one side and "now" on the other; make the two agree. - const days = Math.floor(value) - return days <= 0 ? null : Math.min(HISTORY_DAYS_MAX, days) -} +const DAY_MS = 86_400_000 /** The oldest transcript mtime worth indexing; null means no bound. */ export function sessionSearchHistoryCutoffMs( historyDays: number | null, nowMs: number ): number | null { - const days = normalizeSessionSearchHistoryDays(historyDays) + const days = normalizeAiVaultSearchHistoryDays(historyDays) return days === null ? null : nowMs - days * DAY_MS } diff --git a/src/main/ai-vault-search/session-search-root-refresh.test.ts b/src/main/ai-vault-search/session-search-root-refresh.test.ts new file mode 100644 index 00000000000..43d57e19536 --- /dev/null +++ b/src/main/ai-vault-search/session-search-root-refresh.test.ts @@ -0,0 +1,99 @@ +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { SessionSearchIndexer } from './session-search-indexer' +import { + FakeSessionSearchClock, + openSessionSearchIndexerHarness, + writeClaudeTranscript, + type SessionSearchIndexerHarness +} from './session-search-indexer-test-fixture' +import { resetSessionParseCacheForTests } from '../ai-vault/session-scanner-parse-cache' +import { resetTranscriptConsumersForTests } from '../ai-vault/session-transcript-consumers' + +let harness: SessionSearchIndexerHarness +let indexer: SessionSearchIndexer | undefined +beforeEach(async () => { + resetSessionParseCacheForTests() + resetTranscriptConsumersForTests() + harness = await openSessionSearchIndexerHarness('search-root-refresh') +}) +afterEach(async () => { + indexer?.close() + indexer = undefined + resetSessionParseCacheForTests() + resetTranscriptConsumersForTests() + await harness.cleanup() +}) + +it('refreshes roots on scheduled full sweeps and reuses them on recent cycles', async () => { + const clock = new FakeSessionSearchClock() + let roots = harness.roots + const resolveRoots = vi.fn(async () => roots) + indexer = new SessionSearchIndexer({ + databasePath: harness.databasePath, + roots, + resolveRoots, + historyDays: null, + clock, + fullSweepEveryCycles: 1 + }) + await indexer.start() + expect(resolveRoots).toHaveBeenCalledTimes(1) + const newRoot = join(harness.root, 'late') + const id = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' + await writeClaudeTranscript(join(newRoot, 'project', `${id}.jsonl`), ['a late conversation'], id) + roots = { ...roots, claudeProjectsDir: newRoot } + clock.advance(20_000) + await indexer.settled() + expect(resolveRoots).toHaveBeenCalledTimes(1) + expect(indexer.status().filesIndexed).toBe(0) + clock.advance(20_000) + await indexer.settled() + expect(resolveRoots).toHaveBeenCalledTimes(2) + expect(indexer.status().filesIndexed).toBe(1) + clock.advance(20_000) + await indexer.settled() + expect(resolveRoots).toHaveBeenCalledTimes(2) + expect(indexer.status().filesIndexed).toBe(1) +}) + +it('does not access a closed store when pending discovery completes', async () => { + const pending = Promise.withResolvers() + const errors = vi.fn() + const resolver = vi.fn(() => pending.promise) + indexer = new SessionSearchIndexer({ + databasePath: harness.databasePath, + roots: harness.roots, + resolveRoots: resolver, + historyDays: null, + onError: errors + }) + const start = indexer.start() + await vi.waitFor(() => expect(resolver).toHaveBeenCalledTimes(1)) + indexer.close() + pending.resolve(harness.roots) + await start + expect(errors).not.toHaveBeenCalled() + expect(indexer.status().filesIndexed).toBe(0) +}) + +it('retries discovery after failure without silently sweeping stale roots', async () => { + const errors = vi.fn() + const resolveRoots = vi + .fn() + .mockRejectedValueOnce(new Error('unavailable')) + .mockResolvedValue(harness.roots) + indexer = new SessionSearchIndexer({ + databasePath: harness.databasePath, + roots: harness.roots, + resolveRoots, + historyDays: null, + onError: errors + }) + await indexer.start() + expect(errors).toHaveBeenCalledTimes(1) + expect(indexer.status().lastSweepCompletedAt).toBeNull() + await indexer.reconcile() + expect(resolveRoots).toHaveBeenCalledTimes(2) + expect(indexer.status().lastSweepCompletedAt).not.toBeNull() +}) diff --git a/src/main/ai-vault-search/session-search-scan-roots.test.ts b/src/main/ai-vault-search/session-search-scan-roots.test.ts index 51b7381c78b..39324ca5da8 100644 --- a/src/main/ai-vault-search/session-search-scan-roots.test.ts +++ b/src/main/ai-vault-search/session-search-scan-roots.test.ts @@ -1,7 +1,7 @@ import { expect, it } from 'vitest' import { delimiter, join } from 'node:path' import type { SessionFileDiscovery } from '../ai-vault/session-scanner-types' -import { sessionSearchRootListings } from './session-search-scan-roots' +import { sameSessionSearchRoots, sessionSearchRootListings } from './session-search-scan-roots' const STATE = '/tmp/ss-roots/openclaw-state' const LEGACY = '/tmp/ss-roots/openclaw-legacy' @@ -56,3 +56,23 @@ it('attributes a file by path segment, not by string prefix', () => { expect(byRoot[agents]).toBe(0) expect(byRoot[legacy]).toBe(0) }) + +it('reads a re-resolved root set as the same trees when only spelling order differs', () => { + expect( + sameSessionSearchRoots( + { openclawStateDir: STATE, wslHomeDirs: ['/home/a', '/home/b'] }, + { wslHomeDirs: ['/home/b', '/home/a'], openclawStateDir: STATE } + ) + ).toBe(true) + // An absent key and an explicitly undefined one are the same absence. + expect(sameSessionSearchRoots({ openclawStateDir: STATE }, { openclawStateDir: STATE })).toBe( + true + ) +}) + +it('reads an added, dropped or changed root as a different set', () => { + const base = { openclawStateDir: STATE, wslHomeDirs: ['/home/a'] } + expect(sameSessionSearchRoots(base, { ...base, openclawLegacyStateDir: LEGACY })).toBe(false) + expect(sameSessionSearchRoots(base, { openclawStateDir: STATE })).toBe(false) + expect(sameSessionSearchRoots(base, { ...base, wslHomeDirs: ['/home/b'] })).toBe(false) +}) diff --git a/src/main/ai-vault-search/session-search-scan-roots.ts b/src/main/ai-vault-search/session-search-scan-roots.ts index 8df3510199e..5c1041eac25 100644 --- a/src/main/ai-vault-search/session-search-scan-roots.ts +++ b/src/main/ai-vault-search/session-search-scan-roots.ts @@ -133,3 +133,30 @@ export function sessionSearchEmptiedRoots( ): Set { return new Set([...previous].filter((root) => !current.has(root))) } + +/** + * Whether two root sets name the same trees. + * + * Structural, not by reference: the caller re-resolves roots on every policy + * push, so a live index that already walks these trees must not be rebuilt just + * because the object is new. Key-sorted rather than a plain JSON compare because + * nothing fixes the key order two producers write, and list-sorted because the + * indexer walks every root, so a re-enumeration that reorders is not a change. + */ +export function sameSessionSearchRoots( + a: SessionSearchScanRoots, + b: SessionSearchScanRoots +): boolean { + const left = comparableRootFields(a) + const right = comparableRootFields(b) + return left.length === right.length && left.every((field, index) => field === right[index]) +} + +function comparableRootFields(roots: SessionSearchScanRoots): string[] { + return Object.entries(roots) + .filter(([, value]) => value !== undefined) + .map( + ([key, value]) => `${key}=${JSON.stringify(Array.isArray(value) ? [...value].sort() : value)}` + ) + .sort() +} diff --git a/src/main/ai-vault-search/session-search-service-init.ts b/src/main/ai-vault-search/session-search-service-init.ts new file mode 100644 index 00000000000..1ad4e2a2a3a --- /dev/null +++ b/src/main/ai-vault-search/session-search-service-init.ts @@ -0,0 +1,27 @@ +import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' +import type { AiVaultSessionSearchInit } from '../ai-vault/session-scanner-service-protocol' +import { sessionSearchDatabasePath } from './session-search-database-path' +import { sessionSearchPolicy } from './session-search-policy' + +// Captured once from the composition root's data path, like the parse cache: +// every export is inert until then, so no test or early import can index. +let databasePath: string | null = null + +export function installSessionSearchDataRoot(dataRoot: string): void { + databasePath = sessionSearchDatabasePath(dataRoot) +} + +/** Read at every spawn and every settings change; null before the data root is installed. */ +export function sessionSearchServiceInit(): AiVaultSessionSearchInit | null { + return databasePath + ? { + databasePath, + settings: sessionSearchPolicy(), + roots: { executionHostId: LOCAL_EXECUTION_HOST_ID } + } + : null +} + +export function resetSessionSearchServiceInitForTests(): void { + databasePath = null +} diff --git a/src/main/ai-vault-search/session-search-service.test.ts b/src/main/ai-vault-search/session-search-service.test.ts index 235b00684cb..8d990bcda5c 100644 --- a/src/main/ai-vault-search/session-search-service.test.ts +++ b/src/main/ai-vault-search/session-search-service.test.ts @@ -86,7 +86,7 @@ describe('real index to public service adapter', () => { hits: [expect.objectContaining({ evidence: null })] }) await service.reconcile() - expect(indexer.reconcile).toHaveBeenCalledExactlyOnceWith({ full: false }) + expect(indexer.reconcile).toHaveBeenCalledExactlyOnceWith({ full: true }) const status = await service.status() expect(AiVaultSearchStatusSchema.parse(status)).toEqual(status) expect(status.generation).toBeGreaterThan(0) diff --git a/src/main/ai-vault-search/session-search-service.ts b/src/main/ai-vault-search/session-search-service.ts index 0f9cf6c615e..237d59eebf7 100644 --- a/src/main/ai-vault-search/session-search-service.ts +++ b/src/main/ai-vault-search/session-search-service.ts @@ -21,7 +21,7 @@ export function createSessionSearchService({ indexer: Pick }): SessionSearchService { return { - reconcile: () => indexer.reconcile({ full: false }), + reconcile: () => indexer.reconcile({ full: true }), status: async () => ({ enabled: true, ...indexer.status(), generation: engine.generation() }), search: async (request) => { if (request.cursor === '') { diff --git a/src/main/ai-vault-search/session-search-sqlite-support.ts b/src/main/ai-vault-search/session-search-sqlite-support.ts new file mode 100644 index 00000000000..c59fe8d3db3 --- /dev/null +++ b/src/main/ai-vault-search/session-search-sqlite-support.ts @@ -0,0 +1,26 @@ +/** + * Whether this Node can hold an index at all. + * + * The store is `node:sqlite`, reached through `process.getBuiltinModule`, which + * neither exists on Node 18. That is not a hypothetical floor: orcad and the SSH + * relay are both built for Node 18 and run on whatever the host has, and + * build-orcad.mjs keeps that floor deliberately by excluding the only clusters + * that import `node:sqlite` statically. A host without it registers no search + * service at all rather than one that fails at every call. + */ +export function sessionSearchSqliteAvailable(): boolean { + if (typeof process.getBuiltinModule !== 'function') { + return false + } + try { + const sqlite: unknown = process.getBuiltinModule('node:sqlite') + return ( + typeof sqlite === 'object' && + sqlite !== null && + 'DatabaseSync' in sqlite && + typeof sqlite.DatabaseSync === 'function' + ) + } catch { + return false + } +} diff --git a/src/main/ai-vault/cached-session-list.ts b/src/main/ai-vault/cached-session-list.ts index c9feb5b9daf..673de66e666 100644 --- a/src/main/ai-vault/cached-session-list.ts +++ b/src/main/ai-vault/cached-session-list.ts @@ -7,6 +7,7 @@ import { import { getCachedWslDistros, hasCachedWslDistros, listRunningWslHomeDirsAsync } from '../wsl' import { filterPathsToRunningWslDistrosAsync } from '../wsl-running-path-filter' import type { AiVaultListArgs, AiVaultListResult } from '../../shared/ai-vault-types' +import type { AiVaultScanOptions } from './session-scanner-types' import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' import { AiVaultScanCoordinator } from './ai-vault-scan-coordinator' import { @@ -49,6 +50,28 @@ export function configureAiVaultSessionSources(next: AiVaultSessionSources): voi sources = next } +/** + * The trees a local scan enumerates, resolved fresh because a WSL distro can start + * or stop between scans. The search index reads the same function, so it walks + * exactly what the session list walks. + */ +export async function localAiVaultScanRoots(): Promise< + Required> & + Pick +> { + const [additionalCodexHomes, wslHomeDirs] = await Promise.all([ + filterPathsToRunningWslDistrosAsync(configuredAdditionalCodexHomePaths()), + getAiVaultWslHomeDirs() + ]) + return { + additionalCodexSessionsDirs: additionalCodexHomes.map((homePath) => join(homePath, 'sessions')), + wslHomeDirs, + // Why: this scan is always host-local; callers addressing this host by a + // runtime id get the result restamped at the RPC edge, never rescanned. + executionHostId: LOCAL_EXECUTION_HOST_ID + } +} + /** The extra Codex homes session discovery scans. Anything that decides what a listed row may be * resumed from must read the same set, or a row can be listed and then refuse to resume. */ export function configuredAdditionalCodexHomePaths(): readonly string[] { @@ -86,24 +109,12 @@ export async function listAiVaultSessions( force: args?.force, signal: options.signal, start: async (scanSignal) => { - const configuredCodexHomes = sources.getAdditionalCodexHomePaths?.() ?? [] - const [additionalCodexHomes, wslHomeDirs] = await Promise.all([ - filterPathsToRunningWslDistrosAsync(configuredCodexHomes), - getAiVaultWslHomeDirs() - ]) - const additionalCodexSessionsDirs = additionalCodexHomes.map((homePath) => - join(homePath, 'sessions') - ) const result = await scanAiVaultSessionsInBackground( { limit: args?.limit, unlimited: args?.unlimited, scopePaths: args?.scopePaths, - additionalCodexSessionsDirs, - wslHomeDirs, - // Why: this scan is always host-local; callers addressing this host by a - // runtime id get the result restamped at the RPC edge, never rescanned. - executionHostId: LOCAL_EXECUTION_HOST_ID + ...(await localAiVaultScanRoots()) }, scanSignal ) diff --git a/src/main/ai-vault/session-scanner-service-client-state.ts b/src/main/ai-vault/session-scanner-service-client-state.ts index 9b64431e219..cc5b514988b 100644 --- a/src/main/ai-vault/session-scanner-service-client-state.ts +++ b/src/main/ai-vault/session-scanner-service-client-state.ts @@ -1,9 +1,12 @@ +import type { SessionSearchScanRoots } from '../ai-vault-search/session-search-scan-roots' import type { ChildProcess } from 'node:child_process' +import { createAiVaultScanCancelledError } from './ai-vault-scan-cancellation' import { AI_VAULT_SERVICE_PROTOCOL_VERSION, type AiVaultServiceInit, type AiVaultServiceLane, - type AiVaultServiceRequest + type AiVaultServiceRequest, + type AiVaultSessionSearchInit } from './session-scanner-service-protocol' export const AI_VAULT_SERVICE_READY_TIMEOUT_MS = 5_000 @@ -16,7 +19,9 @@ export const AI_VAULT_SERVICE_SHUTDOWN_TIMEOUT_MS = 2_000 export type AiVaultServiceProcessFactory = () => ChildProcess export type AiVaultServiceClientOptions = { processFactory: AiVaultServiceProcessFactory - init: Omit + /** Resolved per spawn: a respawned child must see current consent, not the first frame's. */ + init: () => Omit + resolveSessionSearchRoots?: () => Promise idleTimeoutMs?: number onStderr?: (text: string) => void } @@ -49,6 +54,29 @@ export class AiVaultServiceInvalidations { }) } + /** + * Sends one invalidation and resolves on the child's acknowledgement. + * + * The deadline is a startup-sized budget, but a child mid-scan can be slow to + * turn the channel around. Fork IPC ordering already guarantees the child + * applies the invalidation before any request sent after it, so a busy child + * owes nothing here -- only an idle one that misses the deadline is wedged. + */ + send( + child: ChildProcess, + paths: string[], + lanes: { busy: () => boolean; onFault: (error: Error) => void } + ): Promise { + return this.open( + AI_VAULT_SERVICE_READY_TIMEOUT_MS, + (generation) => + lanes.busy() + ? void this.settle(generation) + : lanes.onFault(new Error('AI Vault service cache invalidation timed out.')), + (generation) => child.send({ type: 'invalidate', generation, paths }) + ) + } + settle(generation: number): boolean { const entry = this.pending.get(generation) if (!entry) { @@ -96,7 +124,7 @@ export function retireAiVaultServiceChild(child: ChildProcess): void { child.unref() } -export function armAiVaultServiceCancellationTimeout( +function armAiVaultServiceCancellationTimeout( call: AiVaultServicePendingCall, onExpired: () => void ): void { @@ -107,14 +135,50 @@ export function armAiVaultServiceCancellationTimeout( call.timer.unref?.() } -/** - * A cold start that faults before the request reached the child self-heals on - * the scheduled respawn. Requeue once; the caller rejects when this returns false. - */ +/** Abandons one call, and waits for the child's acknowledgement only when it owes one. */ +export function cancelAiVaultServiceCall( + call: AiVaultServicePendingCall, + lanes: { + queue: AiVaultServicePendingCall[] + active: Map + child: ChildProcess | null + pump: () => void + onFault: (error: Error) => void + } +): void { + if (call.cancelled) { + return + } + call.cancelled = true + call.reject(createAiVaultScanCancelledError()) + const queuedIndex = lanes.queue.indexOf(call) + if (queuedIndex !== -1) { + lanes.queue.splice(queuedIndex, 1) + clearAiVaultServiceCall(call) + lanes.pump() + return + } + if (lanes.active.get(call.lane) !== call) { + return + } + // Why: a call cancelled before it reached the child gets no acknowledgement, + // so waiting on one would kill a healthy service and stall the lane. + if (!call.sent) { + lanes.active.delete(call.lane) + clearAiVaultServiceCall(call) + lanes.pump() + return + } + lanes.child?.send({ type: 'cancel', id: call.request.id }) + armAiVaultServiceCancellationTimeout(call, () => + lanes.onFault(new Error('AI Vault service did not cancel within 2000ms.')) + ) +} + /** Wires a freshly forked child to the client's callbacks and hands it the init frame. */ export function attachAiVaultServiceChild( child: ChildProcess, - init: AiVaultServiceClientOptions['init'], + init: ReturnType, handlers: { onMessage: (message: unknown) => void onFault: (error: Error) => void @@ -133,16 +197,26 @@ export function attachAiVaultServiceChild( } satisfies AiVaultServiceInit) } -export function requeueAiVaultServiceStart( +/** + * A cold start that faults before the request reached the child self-heals on + * the scheduled respawn. Requeue once; anything else is the caller's error. + */ +export function requeueOrRejectAiVaultServiceStart( call: AiVaultServicePendingCall, - queue: AiVaultServicePendingCall[] -): boolean { - if (call.sent || call.cancelled || call.startRetried) { - return false + queue: AiVaultServicePendingCall[], + error: Error, + respawning: boolean +): void { + if (!respawning || call.sent || call.cancelled || call.startRetried) { + rejectAiVaultServiceCall(call, error) + return } call.startRetried = true queue.unshift(call) - return true +} + +export function aiVaultServiceErrorText(error: unknown): string { + return error instanceof Error ? error.message : String(error) } export function clearAiVaultServiceCall(call: AiVaultServicePendingCall): void { @@ -205,3 +279,52 @@ export class AiVaultServiceIdleRetirement { this.timer.unref?.() } } + +/** + * The parent's half of the index setting. + * + * A child running the index is never idle from out here -- its reconcile loop is + * invisible to the parent -- so this is what stops idle retirement ending the + * indexing until some later scan happens to respawn a child. + */ +export class AiVaultServiceSessionSearchHold { + private enabled = false + + /** True while a running index needs a child to exist. */ + get holdsChild(): boolean { + return this.enabled + } + + /** + * Records the policy and tells a live child. A missing one reads the same + * policy out of its init frame, which is why `init` is a factory, not a value. + * @returns whether a child now has to exist. + */ + record(init: AiVaultSessionSearchInit, child: ChildProcess | null): boolean { + this.enabled = init.settings.enabled + child?.send({ type: 'sessionSearch', init }) + return this.enabled + } +} + +/** Starts the request deadline only once the child is ready to receive it. */ +export function sendAiVaultServiceCall( + child: ChildProcess, + call: AiVaultServicePendingCall, + isActive: () => boolean, + onFault: (error: Error) => void +): void { + if (call.cancelled || !isActive()) { + return + } + const timeoutMs = + call.request.operation === 'scan' + ? AI_VAULT_SERVICE_SCAN_TIMEOUT_MS + : AI_VAULT_SERVICE_INTERACTIVE_TIMEOUT_MS + call.timer = setTimeout(() => { + onFault(new Error(`AI Vault service timed out after ${timeoutMs}ms.`)) + }, timeoutMs) + call.timer.unref?.() + call.sent = true + child.send(call.request) +} diff --git a/src/main/ai-vault/session-scanner-service-client.test.ts b/src/main/ai-vault/session-scanner-service-client.test.ts index 97cb0d46afd..1e8a67602f3 100644 --- a/src/main/ai-vault/session-scanner-service-client.test.ts +++ b/src/main/ai-vault/session-scanner-service-client.test.ts @@ -1,12 +1,36 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { AiVaultScannerServiceClient } from './session-scanner-service-client' import { AI_VAULT_SERVICE_READY_TIMEOUT_MS } from './session-scanner-service-client-state' +import type { AiVaultSessionSearchInit } from './session-scanner-service-protocol' import { AiVaultServiceTestChild, aiVaultServiceRequestId, readyAiVaultServiceChild } from './session-scanner-service-test-child' +const SESSION_SEARCH_ON: AiVaultSessionSearchInit = { + databasePath: '/data/ai-vault/session-search.sqlite', + settings: { enabled: true, historyDays: null }, + roots: {} +} + +/** Every fork the client makes, so a respawn can be told from the first start. */ +function setupChildren(policy: () => AiVaultSessionSearchInit | null): { + children: AiVaultServiceTestChild[] + client: AiVaultScannerServiceClient +} { + const children: AiVaultServiceTestChild[] = [] + const client = new AiVaultScannerServiceClient({ + processFactory: () => { + const child = new AiVaultServiceTestChild(12_345 + children.length) + children.push(child) + return child.asChildProcess() + }, + init: () => ({ sessionParseCache: null, sessionSearch: policy() }) + }) + return { children, client } +} + function setup(idleTimeoutMs?: number): { child: AiVaultServiceTestChild client: AiVaultScannerServiceClient @@ -14,7 +38,7 @@ function setup(idleTimeoutMs?: number): { const child = new AiVaultServiceTestChild() const client = new AiVaultScannerServiceClient({ processFactory: () => child.asChildProcess(), - init: { sessionParseCache: null }, + init: () => ({ sessionParseCache: null, sessionSearch: null }), idleTimeoutMs }) return { child, client } @@ -135,7 +159,7 @@ describe('AiVaultScannerServiceClient', () => { children.push(child) return child.asChildProcess() }, - init: { sessionParseCache: null } + init: () => ({ sessionParseCache: null, sessionSearch: null }) }) const titles = client.request({ type: 'request', operation: 'titles', requests: [] }) expect(children).toHaveLength(1) @@ -167,7 +191,7 @@ describe('AiVaultScannerServiceClient', () => { children.push(child) return child.asChildProcess() }, - init: { sessionParseCache: null } + init: () => ({ sessionParseCache: null, sessionSearch: null }) }) const titles = client.request({ type: 'request', operation: 'titles', requests: [] }) @@ -190,7 +214,7 @@ describe('AiVaultScannerServiceClient', () => { children.push(child) return child.asChildProcess() }, - init: { sessionParseCache: null } + init: () => ({ sessionParseCache: null, sessionSearch: null }) }) const titles = client.request({ type: 'request', operation: 'titles', requests: [] }) @@ -224,7 +248,7 @@ describe('AiVaultScannerServiceClient', () => { children.push(child) return child.asChildProcess() }, - init: { sessionParseCache: null } + init: () => ({ sessionParseCache: null, sessionSearch: null }) }) // Each request retries its cold start once, so two requests spend the three // faults the circuit breaker needs. @@ -242,11 +266,9 @@ describe('AiVaultScannerServiceClient', () => { vi.advanceTimersByTime(AI_VAULT_SERVICE_READY_TIMEOUT_MS) await Promise.resolve() vi.advanceTimersByTime(5_000) - await expect(blocked).rejects.toThrow('circuit is open') expect(children).toHaveLength(3) client.clearRestartCircuit() - const retried = client.request({ type: 'request', operation: 'titles', requests: [] }) await vi.waitFor(() => expect(children).toHaveLength(4)) readyAiVaultServiceChild(children[3]!) await vi.waitFor(() => @@ -258,7 +280,7 @@ describe('AiVaultScannerServiceClient', () => { operation: 'titles', value: { titles: [] } }) - await expect(retried).resolves.toEqual({ titles: [] }) + await expect(blocked).resolves.toEqual({ titles: [] }) client.dispose() }) @@ -314,7 +336,7 @@ describe('AiVaultScannerServiceClient', () => { children.push(child) return child.asChildProcess() }, - init: { sessionParseCache: null } + init: () => ({ sessionParseCache: null, sessionSearch: null }) }) const invalidation = client.invalidate(['/tmp/deleted.jsonl']) readyAiVaultServiceChild(children[0]!) @@ -350,7 +372,7 @@ describe('AiVaultScannerServiceClient', () => { children.push(child) return child.asChildProcess() }, - init: { sessionParseCache: null }, + init: () => ({ sessionParseCache: null, sessionSearch: null }), idleTimeoutMs: 100 }) @@ -392,6 +414,152 @@ describe('AiVaultScannerServiceClient', () => { client.dispose() }) + // The child holds the index while the setting is on, and its reconcile loop is + // invisible from here: retiring it would stop indexing until the next scan + // happened to respawn one, which is not a guarantee anyone stated. + it('spawns a child for the index and never retires it while the index is on', async () => { + vi.useFakeTimers() + const { child, client } = setup(100) + const on = { + databasePath: '/data/ai-vault/session-search.sqlite', + settings: { enabled: true, historyDays: null }, + roots: {} + } + + // No request outstanding: turning the index on is itself what spawns a child. + client.updateSessionSearch(on) + readyAiVaultServiceChild(child) + await Promise.resolve() + expect(child.sent).toContainEqual(expect.objectContaining({ type: 'init' })) + + vi.advanceTimersByTime(10_000) + expect(child.sent).not.toContainEqual({ type: 'shutdown' }) + + // A live child hears the change directly rather than waiting for a respawn. + const narrowed = { ...on, settings: { enabled: true, historyDays: 30 } } + client.updateSessionSearch(narrowed) + expect(child.sent).toContainEqual({ type: 'sessionSearch', init: narrowed }) + vi.advanceTimersByTime(10_000) + expect(child.sent).not.toContainEqual({ type: 'shutdown' }) + + client.updateSessionSearch({ ...on, settings: { enabled: false, historyDays: null } }) + vi.advanceTimersByTime(100) + expect(child.sent).toContainEqual({ type: 'shutdown' }) + client.dispose() + }) + + it('re-reads the init frame on every spawn so a respawn sees current consent', async () => { + const children: AiVaultServiceTestChild[] = [] + let enabled = false + const client = new AiVaultScannerServiceClient({ + processFactory: () => { + const child = new AiVaultServiceTestChild(12_345 + children.length) + children.push(child) + return child.asChildProcess() + }, + init: () => ({ + sessionParseCache: null, + sessionSearch: { + databasePath: '/data/ai-vault/session-search.sqlite', + settings: { enabled, historyDays: null }, + roots: {} + } + }) + }) + + const first = client.request({ type: 'request', operation: 'titles', requests: [] }) + readyAiVaultServiceChild(children[0]!) + await Promise.resolve() + expect(children[0]!.sent[0]).toMatchObject({ sessionSearch: { settings: { enabled: false } } }) + + enabled = true + children[0]!.emit('error', new Error('crashed')) + await expect(first).rejects.toThrow('crashed') + void client.request({ type: 'request', operation: 'titles', requests: [] }).catch(() => {}) + await vi.waitFor(() => expect(children.length).toBeGreaterThan(1)) + for (const respawned of children.slice(1)) { + expect(respawned.sent[0]).toMatchObject({ sessionSearch: { settings: { enabled: true } } }) + } + client.dispose() + }) + + // The hold is the only thing keeping this child alive, so nothing else will + // restart it: without its own restart, an idle indexing child that crashes + // leaves the index stopped until some unrelated request happens to arrive. + it('restarts a child that faulted while the index was holding it', async () => { + vi.useFakeTimers() + const { children, client } = setupChildren(() => SESSION_SEARCH_ON) + client.updateSessionSearch(SESSION_SEARCH_ON) + readyAiVaultServiceChild(children[0]!) + await Promise.resolve() + + // No queued call and no outstanding invalidation: an idle child simply dies. + children[0]!.emit('error', new Error('crashed')) + expect(children).toHaveLength(1) + vi.advanceTimersByTime(250) + + expect(children).toHaveLength(2) + expect(children[1]!.sent[0]).toMatchObject({ + type: 'init', + sessionSearch: { settings: { enabled: true } } + }) + client.dispose() + }) + + it.each([false, true])( + 'waits for circuit expiry before restarting a held child (dispose=%s)', + async (dispose) => { + vi.useFakeTimers() + const { children, client } = setupChildren(() => SESSION_SEARCH_ON) + try { + client.updateSessionSearch(SESSION_SEARCH_ON) + for (const delay of [250, 1_000]) { + readyAiVaultServiceChild(children.at(-1)!) + await Promise.resolve() + children.at(-1)!.emit('error', new Error('temporary fault')) + await vi.advanceTimersByTimeAsync(delay) + } + expect(children).toHaveLength(3) + readyAiVaultServiceChild(children[2]!) + await Promise.resolve() + children[2]!.emit('error', new Error('temporary fault')) + await vi.advanceTimersByTimeAsync(59_999) + expect(children).toHaveLength(3) + if (dispose) { + client.dispose() + } + await vi.advanceTimersByTimeAsync(1) + expect(children).toHaveLength(dispose ? 3 : 4) + if (!dispose) { + readyAiVaultServiceChild(children[3]!) + } + } finally { + client.dispose() + } + } + ) + + it('leaves a faulted idle child dead while the index is off', async () => { + vi.useFakeTimers() + const { children, client } = setupChildren(() => null) + const titles = client.request({ type: 'request', operation: 'titles', requests: [] }) + readyAiVaultServiceChild(children[0]!) + await Promise.resolve() + children[0]!.emit('message', { + type: 'result', + id: aiVaultServiceRequestId(children[0]!, 'titles'), + operation: 'titles', + value: { titles: [] } + }) + await titles + + children[0]!.emit('error', new Error('crashed')) + vi.advanceTimersByTime(5_000) + + expect(children).toHaveLength(1) + client.dispose() + }) + it('retires an idle child gracefully, then kills it after the shutdown bound', async () => { vi.useFakeTimers() const { child, client } = setup(100) diff --git a/src/main/ai-vault/session-scanner-service-client.ts b/src/main/ai-vault/session-scanner-service-client.ts index e068f2e2e0a..e547ee1dee7 100644 --- a/src/main/ai-vault/session-scanner-service-client.ts +++ b/src/main/ai-vault/session-scanner-service-client.ts @@ -2,19 +2,20 @@ import type { ChildProcess } from 'node:child_process' import { createAiVaultScanCancelledError } from './ai-vault-scan-cancellation' import { AI_VAULT_SERVICE_IDLE_TIMEOUT_MS, - AI_VAULT_SERVICE_INTERACTIVE_TIMEOUT_MS, AI_VAULT_SERVICE_MAX_CALLS, AI_VAULT_SERVICE_READY_TIMEOUT_MS, - AI_VAULT_SERVICE_SCAN_TIMEOUT_MS, AiVaultServiceIdleRetirement, AiVaultServiceInvalidations, - armAiVaultServiceCancellationTimeout, + AiVaultServiceSessionSearchHold, + aiVaultServiceErrorText, attachAiVaultServiceChild, + cancelAiVaultServiceCall, clearAiVaultServiceCall, createAiVaultServiceReadyWaiter, rejectAiVaultServiceCall, - requeueAiVaultServiceStart, + requeueOrRejectAiVaultServiceStart, retireAiVaultServiceChild, + sendAiVaultServiceCall, type AiVaultServiceClientOptions, type AiVaultServicePendingCall, type AiVaultServiceReadyWaiter @@ -23,7 +24,7 @@ import { AiVaultServiceRestartPolicy } from './session-scanner-service-restart-p import { aiVaultServiceLane, isAiVaultServiceChildMessage, - type AiVaultServiceChildMessage, + type AiVaultSessionSearchInit, type AiVaultServiceRequest, type AiVaultServiceRequestBody, type AiVaultServiceResultValue @@ -38,6 +39,7 @@ export class AiVaultScannerServiceClient { private nextId = 1 private readonly idleRetirement = new AiVaultServiceIdleRetirement() private readonly restartPolicy = new AiVaultServiceRestartPolicy() + private readonly sessionSearch = new AiVaultServiceSessionSearchHold() private disposed = false constructor(private readonly options: AiVaultServiceClientOptions) {} @@ -76,6 +78,19 @@ export class AiVaultScannerServiceClient { }) } + /** Push a consent or retention change, and while the index is on keep a child. */ + updateSessionSearch(init: AiVaultSessionSearchInit): void { + if (this.disposed) { + return + } + if (!this.sessionSearch.record(init, this.child)) { + this.scheduleIdleIfNeeded() + return + } + this.idleRetirement.clear() + this.startSessionSearchChild() + } + clearRestartCircuit(): void { this.restartPolicy.clearCircuit() this.pump() @@ -87,25 +102,10 @@ export class AiVaultScannerServiceClient { } this.idleRetirement.clear() const child = await this.ensureChild() - return this.invalidations.open( - AI_VAULT_SERVICE_READY_TIMEOUT_MS, - (generation) => this.onInvalidationDeadline(generation), - (generation) => child.send({ type: 'invalidate', generation, paths }) - ) - } - - /** - * The deadline is a startup-sized budget, but a child mid-scan can be slow to - * turn the channel around. Fork IPC ordering already guarantees the child - * applies the invalidation before any request sent after it, so a busy child - * owes nothing here — only an idle one that misses the deadline is wedged. - */ - private onInvalidationDeadline(generation: number): void { - if (this.active.size > 0) { - this.invalidations.settle(generation) - return - } - this.onFault(new Error('AI Vault service cache invalidation timed out.')) + return this.invalidations.send(child, paths, { + busy: () => this.active.size > 0, + onFault: (error) => this.onFault(error) + }) } dispose(): void { @@ -140,7 +140,13 @@ export class AiVaultScannerServiceClient { const call = this.queue.splice(index, 1)[0]! this.active.set(lane, call) void this.ensureChild().then( - (child) => this.sendCall(child, call), + (child) => + sendAiVaultServiceCall( + child, + call, + () => this.active.get(call.lane) === call, + (error) => this.onFault(error) + ), (error: Error) => { if (this.active.get(lane) !== call) { return @@ -151,33 +157,32 @@ export class AiVaultScannerServiceClient { } ) } + this.startSessionSearchChild() this.scheduleIdleIfNeeded() } - private sendCall(child: ChildProcess, call: AiVaultServicePendingCall): void { - if (call.cancelled || this.active.get(call.lane) !== call) { + /** + * The index's own restart. A child indexing for the hold has no queued call to + * bring it back, so without this a fault stops the indexing until an unrelated + * request happens to arrive. The restart delay and circuit bound it, exactly as + * they bound a queued call's start. + */ + private startSessionSearchChild(): void { + if (this.disposed || !this.sessionSearch.holdsChild || this.child || this.readyWaiter) { return } - const timeoutMs = - call.request.operation === 'scan' - ? AI_VAULT_SERVICE_SCAN_TIMEOUT_MS - : AI_VAULT_SERVICE_INTERACTIVE_TIMEOUT_MS - call.timer = setTimeout(() => { - this.onFault(new Error(`AI Vault service timed out after ${timeoutMs}ms.`)) - }, timeoutMs) - call.timer.unref?.() - call.sent = true - child.send(call.request) + void this.ensureChild().catch((error: unknown) => { + this.options.onStderr?.(`session search child unavailable: ${aiVaultServiceErrorText(error)}`) + }) } private retryStartOrReject(call: AiVaultServicePendingCall, error: Error): void { - if ( - this.disposed || - !this.restartPolicy.restartScheduled || - !requeueAiVaultServiceStart(call, this.queue) - ) { - rejectAiVaultServiceCall(call, error) - } + requeueOrRejectAiVaultServiceStart( + call, + this.queue, + error, + !this.disposed && this.restartPolicy.restartScheduled + ) } private ensureChild(): Promise { @@ -203,7 +208,7 @@ export class AiVaultScannerServiceClient { this.onFault(new Error('AI Vault service did not become ready.')) ) this.readyWaiter = waiter - attachAiVaultServiceChild(child, this.options.init, { + attachAiVaultServiceChild(child, this.options.init(), { onMessage: (message) => this.onMessage(message), onFault: (error) => this.onFault(error), onStderr: this.options.onStderr @@ -211,12 +216,24 @@ export class AiVaultScannerServiceClient { return waiter.promise } - private onMessage(raw: unknown): void { - if (!isAiVaultServiceChildMessage(raw)) { + private onMessage(message: unknown): void { + if (!isAiVaultServiceChildMessage(message)) { this.onFault(new Error('AI Vault service sent a malformed message.')) return } - const message = raw as AiVaultServiceChildMessage + if (message.type === 'sessionSearchRoots') { + const child = this.child + const resolve = this.options.resolveSessionSearchRoots + void Promise.resolve() + .then(() => (resolve ? resolve() : (this.options.init().sessionSearch?.roots ?? null))) + .catch(() => null) + .then((roots) => { + if (child && this.child === child && child.connected) { + child.send({ type: 'sessionSearchRoots', id: message.id, roots }, () => undefined) + } + }) + return + } if (message.type === 'ready') { const waiter = this.readyWaiter if (!waiter || !this.child) { @@ -250,32 +267,13 @@ export class AiVaultScannerServiceClient { } private cancel(call: AiVaultServicePendingCall): void { - if (call.cancelled) { - return - } - call.cancelled = true - call.reject(createAiVaultScanCancelledError()) - const queuedIndex = this.queue.indexOf(call) - if (queuedIndex !== -1) { - this.queue.splice(queuedIndex, 1) - clearAiVaultServiceCall(call) - this.pump() - return - } - if (this.active.get(call.lane) === call) { - // Why: a call cancelled before it reached the child gets no acknowledgement, - // so waiting on one would kill a healthy service and stall the lane. - if (!call.sent) { - this.active.delete(call.lane) - clearAiVaultServiceCall(call) - this.pump() - return - } - this.child?.send({ type: 'cancel', id: call.request.id }) - armAiVaultServiceCancellationTimeout(call, () => - this.onFault(new Error('AI Vault service did not cancel within 2000ms.')) - ) - } + cancelAiVaultServiceCall(call, { + queue: this.queue, + active: this.active, + child: this.child, + pump: () => this.pump(), + onFault: (error) => this.onFault(error) + }) } private onFault(error: Error): void { @@ -304,7 +302,11 @@ export class AiVaultScannerServiceClient { private scheduleIdleIfNeeded(): void { this.idleRetirement.schedule( - this.active.size > 0 || this.queue.length > 0 || this.invalidations.size > 0 || !this.child, + this.sessionSearch.holdsChild || + this.active.size > 0 || + this.queue.length > 0 || + this.invalidations.size > 0 || + !this.child, this.options.idleTimeoutMs ?? AI_VAULT_SERVICE_IDLE_TIMEOUT_MS, () => this.retireChild() ) diff --git a/src/main/ai-vault/session-scanner-service-entry.ts b/src/main/ai-vault/session-scanner-service-entry.ts index 74a5ea9c355..7458db74e3a 100644 --- a/src/main/ai-vault/session-scanner-service-entry.ts +++ b/src/main/ai-vault/session-scanner-service-entry.ts @@ -1,3 +1,4 @@ +import { requestSessionSearchRoots } from './session-scanner-service-root-request' import type { AiVaultSessionTitle } from '../../shared/ai-vault-session-title' import { readAiVaultFirstUserPrompt } from './session-first-user-prompt-read' import { @@ -6,6 +7,7 @@ import { } from './session-parse-cache-persistence' import { scanAiVaultSessions } from './session-scanner' import { invalidateSessionParseCacheEntry } from './session-scanner-parse-cache' +import { SessionScannerServiceSearch } from './session-scanner-service-search' import { AI_VAULT_SERVICE_PROTOCOL_VERSION, aiVaultServiceLane, @@ -29,6 +31,7 @@ const cancelled = new Set() const pending = new Set() const titleIndex = new Map() const invalidatedPaths = new Set() +const sessionSearch = new SessionScannerServiceSearch(requestSessionSearchRoots) let initialized = false let shuttingDown = false let cacheLane = Promise.resolve() @@ -43,6 +46,15 @@ function titleKey(request: { agent: string; sessionId: string }): string { } async function executeRequest(request: AiVaultServiceRequest): Promise { + if (sessionSearch.handles(request)) { + try { + return await sessionSearch.execute(request) + } finally { + // A search registers no controller, so nothing else consumes a cancel sent + // for one; without this the id sits in the set for the process's life. + cancelled.delete(request.id) + } + } const controller = new AbortController() controllers.set(request.id, controller) try { @@ -149,6 +161,7 @@ async function shutdown(): Promise { for (const controller of controllers.values()) { controller.abort() } + sessionSearch.close() await Promise.allSettled([cacheLane, interactiveLane]) await flushSessionParseCachePersist() process.disconnect?.() @@ -164,6 +177,9 @@ process.on('message', (raw: AiVaultServiceParentMessage) => { if (raw.sessionParseCache) { initSessionParseCachePersistence(raw.sessionParseCache) } + if (raw.sessionSearch) { + sessionSearch.apply(raw.sessionSearch) + } send({ type: 'ready', protocol: AI_VAULT_SERVICE_PROTOCOL_VERSION, pid: process.pid }) return } @@ -192,6 +208,10 @@ process.on('message', (raw: AiVaultServiceParentMessage) => { send({ type: 'invalidated', generation: raw.generation }) return } + if (raw?.type === 'sessionSearch') { + sessionSearch.apply(raw.init) + return + } if (raw?.type === 'shutdown') { void shutdown() return diff --git a/src/main/ai-vault/session-scanner-service-protocol.ts b/src/main/ai-vault/session-scanner-service-protocol.ts index f2842751eb1..df89600934c 100644 --- a/src/main/ai-vault/session-scanner-service-protocol.ts +++ b/src/main/ai-vault/session-scanner-service-protocol.ts @@ -4,6 +4,13 @@ import type { AiVaultSessionTitleRequest, AiVaultSessionTitlesResult } from '../../shared/ai-vault-session-title' +import type { + AiVaultSearchRequest, + AiVaultSearchResponse, + AiVaultSearchStatus +} from '../../shared/ai-vault-search-types' +import type { AiVaultSearchSettings } from '../../shared/ai-vault-search-settings' +import type { SessionSearchScanRoots } from '../ai-vault-search/session-search-scan-roots' import type { ReadAiVaultFirstUserPromptArgs } from './session-first-user-prompt-read' import type { SessionParseCachePersistenceOptions } from './session-parse-cache-persistence' import type { AiVaultWorkerScanOptions } from './session-scanner-worker-protocol' @@ -11,17 +18,49 @@ import type { AiVaultWorkerScanOptions } from './session-scanner-worker-protocol export const AI_VAULT_SERVICE_PROTOCOL_VERSION = 1 export type AiVaultServiceLane = 'cache' | 'interactive' -export type AiVaultServiceOperation = 'scan' | 'titles' | 'subagents' | 'firstPrompt' +export type AiVaultServiceOperation = + | 'scan' + | 'titles' + | 'subagents' + | 'firstPrompt' + | 'searchSessions' + | 'searchStatus' + | 'searchReconcile' + +// Typed from the union so a new operation cannot be added without landing here, +// and held as strings so recognising one costs no assertion. +const AI_VAULT_SERVICE_OPERATIONS: ReadonlySet = new Set([ + 'scan', + 'titles', + 'subagents', + 'firstPrompt', + 'searchSessions', + 'searchStatus', + 'searchReconcile' +]) export type AiVaultServiceSubagentRequest = { agent: 'claude' | 'omp' parentFilePath: string } +/** + * Everything the child needs to own this host's index. + * + * Initial roots also support standalone tests. Production asks the parent for + * a fresh snapshot on each full sweep; the parent owns managed account homes. + */ +export type AiVaultSessionSearchInit = { + databasePath: string + settings: AiVaultSearchSettings + roots: SessionSearchScanRoots +} + export type AiVaultServiceInit = { type: 'init' protocol: typeof AI_VAULT_SERVICE_PROTOCOL_VERSION sessionParseCache: SessionParseCachePersistenceOptions | null + sessionSearch: AiVaultSessionSearchInit | null } export type AiVaultServiceRequestBody = @@ -41,6 +80,9 @@ export type AiVaultServiceRequestBody = operation: 'firstPrompt' request: ReadAiVaultFirstUserPromptArgs } + | { type: 'request'; operation: 'searchSessions'; request: AiVaultSearchRequest } + | { type: 'request'; operation: 'searchStatus' } + | { type: 'request'; operation: 'searchReconcile' } export type AiVaultServiceRequest = AiVaultServiceRequestBody & { id: number } @@ -49,6 +91,9 @@ export type AiVaultServiceParentMessage = | AiVaultServiceRequest | { type: 'cancel'; id: number } | { type: 'invalidate'; generation: number; paths: string[] } + // Fire-and-forget: the child closes the live pair and constructs from this. + | { type: 'sessionSearch'; init: AiVaultSessionSearchInit } + | { type: 'sessionSearchRoots'; id: number; roots: SessionSearchScanRoots | null } | { type: 'shutdown' } export type AiVaultServiceResultValue = @@ -56,8 +101,12 @@ export type AiVaultServiceResultValue = | { operation: 'titles'; value: AiVaultSessionTitlesResult } | { operation: 'subagents'; value: AiVaultSubagentListResult } | { operation: 'firstPrompt'; value: { prompt: string | null } } + | { operation: 'searchSessions'; value: AiVaultSearchResponse } + | { operation: 'searchStatus'; value: AiVaultSearchStatus } + | { operation: 'searchReconcile'; value: null } export type AiVaultServiceChildMessage = + | { type: 'sessionSearchRoots'; id: number } | { type: 'ready' protocol: typeof AI_VAULT_SERVICE_PROTOCOL_VERSION @@ -67,22 +116,23 @@ export type AiVaultServiceChildMessage = | { type: 'error'; id: number; message: string; retryable: boolean } | { type: 'invalidated'; generation: number } +/** Everything but the two bulk reads is interactive: a search must not queue behind a scan. */ export function aiVaultServiceLane(operation: AiVaultServiceOperation): AiVaultServiceLane { - return operation === 'subagents' || operation === 'firstPrompt' ? 'interactive' : 'cache' + return operation === 'scan' || operation === 'titles' ? 'cache' : 'interactive' } export function isAiVaultServiceRequest(value: unknown): value is AiVaultServiceRequest { if (!value || typeof value !== 'object') { return false } - const message = value as Record return ( - message.type === 'request' && - Number.isSafeInteger(message.id) && - (message.operation === 'scan' || - message.operation === 'titles' || - message.operation === 'subagents' || - message.operation === 'firstPrompt') + 'type' in value && + value.type === 'request' && + 'id' in value && + Number.isSafeInteger(value.id) && + 'operation' in value && + typeof value.operation === 'string' && + AI_VAULT_SERVICE_OPERATIONS.has(value.operation) ) } @@ -94,6 +144,9 @@ export function isAiVaultServiceChildMessage(value: unknown): value is AiVaultSe if (message.type === 'ready') { return message.protocol === AI_VAULT_SERVICE_PROTOCOL_VERSION && Number.isInteger(message.pid) } + if (message.type === 'sessionSearchRoots') { + return Number.isSafeInteger(message.id) + } if (message.type === 'invalidated') { return Number.isSafeInteger(message.generation) } diff --git a/src/main/ai-vault/session-scanner-service-restart-policy.ts b/src/main/ai-vault/session-scanner-service-restart-policy.ts index ae9f437ab3e..02f972691ed 100644 --- a/src/main/ai-vault/session-scanner-service-restart-policy.ts +++ b/src/main/ai-vault/session-scanner-service-restart-policy.ts @@ -43,10 +43,13 @@ export class AiVaultServiceRestartPolicy { if (this.timer) { clearTimeout(this.timer) } - this.timer = setTimeout(() => { - this.timer = null - restart() - }, delay) + this.timer = setTimeout( + () => { + this.timer = null + restart() + }, + Math.max(delay, this.circuitUntil - now) + ) this.timer.unref?.() } diff --git a/src/main/ai-vault/session-scanner-service-root-request.test.ts b/src/main/ai-vault/session-scanner-service-root-request.test.ts new file mode 100644 index 00000000000..0f07b415b4f --- /dev/null +++ b/src/main/ai-vault/session-scanner-service-root-request.test.ts @@ -0,0 +1,57 @@ +import { afterEach, beforeEach, expect, it } from 'vitest' +import { requestSessionSearchRoots } from './session-scanner-service-root-request' +import { + isAiVaultServiceChildMessage, + type AiVaultServiceChildMessage +} from './session-scanner-service-protocol' + +let originalSend: typeof process.send +let lastRequest: Extract +let listeners: number +beforeEach(() => { + originalSend = process.send + listeners = process.listenerCount('message') + process.send = (message) => { + if (!isAiVaultServiceChildMessage(message) || message.type !== 'sessionSearchRoots') { + throw new Error('Unexpected child message') + } + lastRequest = message + return true + } +}) +afterEach(() => { + process.send = originalSend + expect(process.listenerCount('message')).toBe(listeners) +}) +it('matches the requested snapshot and removes its listener', async () => { + const pending = requestSessionSearchRoots(new AbortController().signal) + process.emit( + 'message', + { type: 'sessionSearchRoots', id: lastRequest.id + 1, roots: {} }, + undefined + ) + const roots = { additionalCodexSessionsDirs: ['/late'] } + process.emit('message', { type: 'sessionSearchRoots', id: lastRequest.id, roots }, undefined) + await expect(pending).resolves.toEqual(roots) +}) +it('releases a pending request when indexing is disabled', async () => { + const controller = new AbortController() + const pending = requestSessionSearchRoots(controller.signal) + controller.abort(new Error('disabled')) + await expect(pending).rejects.toThrow('disabled') +}) +it('reports discovery and send failures instead of using stale roots', async () => { + const pending = requestSessionSearchRoots(new AbortController().signal) + process.emit( + 'message', + { type: 'sessionSearchRoots', id: lastRequest.id, roots: null }, + undefined + ) + await expect(pending).rejects.toThrow('discovery failed') + process.send = () => { + throw new Error('channel closed') + } + await expect(requestSessionSearchRoots(new AbortController().signal)).rejects.toThrow( + 'channel closed' + ) +}) diff --git a/src/main/ai-vault/session-scanner-service-root-request.ts b/src/main/ai-vault/session-scanner-service-root-request.ts new file mode 100644 index 00000000000..4d510dd2b73 --- /dev/null +++ b/src/main/ai-vault/session-scanner-service-root-request.ts @@ -0,0 +1,40 @@ +import type { SessionSearchScanRoots } from '../ai-vault-search/session-search-scan-roots' +import type { AiVaultServiceParentMessage } from './session-scanner-service-protocol' + +let nextId = 1 + +/** The parent owns managed-account discovery; the child owns the sweep's lifetime. */ +export async function requestSessionSearchRoots( + signal: AbortSignal +): Promise { + signal.throwIfAborted() + const id = nextId++ + const pending = Promise.withResolvers() + const onAbort = (): void => pending.reject(signal.reason) + const onMessage = (message: AiVaultServiceParentMessage): void => { + if (message?.type !== 'sessionSearchRoots' || message.id !== id) { + return + } + if (message.roots) { + pending.resolve(message.roots) + } else { + pending.reject(new Error('Session search root discovery failed.')) + } + } + process.on('message', onMessage) + signal.addEventListener('abort', onAbort, { once: true }) + try { + if (!process.send) { + throw new Error('Session search root discovery requires parent IPC.') + } + process.send({ type: 'sessionSearchRoots', id }, (error) => { + if (error) { + pending.reject(error) + } + }) + return await pending.promise + } finally { + process.removeListener('message', onMessage) + signal.removeEventListener('abort', onAbort) + } +} diff --git a/src/main/ai-vault/session-scanner-service-root-response.test.ts b/src/main/ai-vault/session-scanner-service-root-response.test.ts new file mode 100644 index 00000000000..6d1d81fd351 --- /dev/null +++ b/src/main/ai-vault/session-scanner-service-root-response.test.ts @@ -0,0 +1,63 @@ +import { expect, it, vi } from 'vitest' +import { AiVaultScannerServiceClient } from './session-scanner-service-client' +import { + AiVaultServiceTestChild, + readyAiVaultServiceChild +} from './session-scanner-service-test-child' + +it('answers root requests freshly without forwarding another settings change', async () => { + const child = new AiVaultServiceTestChild() + Object.assign(child, { connected: true }) + const roots = { additionalCodexSessionsDirs: ['/late'] } + const resolveSessionSearchRoots = vi + .fn() + .mockResolvedValueOnce(roots) + .mockRejectedValueOnce(new Error('offline')) + const client = new AiVaultScannerServiceClient({ + processFactory: () => child.asChildProcess(), + init: () => ({ sessionSearch: null, sessionParseCache: null }), + resolveSessionSearchRoots + }) + const status = client.request({ type: 'request', operation: 'searchStatus' }) + try { + readyAiVaultServiceChild(child) + await Promise.resolve() + child.emit('message', { type: 'result', operation: 'searchStatus', id: 1, value: {} }) + await status + child.emit('message', { type: 'sessionSearchRoots', id: 5 }) + await vi.waitFor(() => + expect(child.sent).toContainEqual({ type: 'sessionSearchRoots', id: 5, roots }) + ) + child.emit('message', { type: 'sessionSearchRoots', id: 6 }) + await vi.waitFor(() => + expect(child.sent).toContainEqual({ type: 'sessionSearchRoots', id: 6, roots: null }) + ) + expect(resolveSessionSearchRoots).toHaveBeenCalledTimes(2) + expect(child.sent).not.toContainEqual(expect.objectContaining({ type: 'sessionSearch' })) + } finally { + client.dispose() + } +}) + +it('does not deliver a delayed snapshot after the child is disposed', async () => { + const child = new AiVaultServiceTestChild() + Object.assign(child, { connected: true }) + const pending = Promise.withResolvers<{}>() + const resolveSessionSearchRoots = vi.fn(() => pending.promise) + const client = new AiVaultScannerServiceClient({ + processFactory: () => child.asChildProcess(), + init: () => ({ sessionSearch: null, sessionParseCache: null }), + resolveSessionSearchRoots + }) + const status = client.request({ type: 'request', operation: 'searchStatus' }) + readyAiVaultServiceChild(child) + await Promise.resolve() + child.emit('message', { type: 'result', operation: 'searchStatus', id: 1, value: {} }) + await status + child.emit('message', { type: 'sessionSearchRoots', id: 5 }) + await vi.waitFor(() => expect(resolveSessionSearchRoots).toHaveBeenCalledTimes(1)) + client.dispose() + pending.resolve({}) + await new Promise((resolve) => setImmediate(resolve)) + expect(child.sent).not.toContainEqual(expect.objectContaining({ type: 'sessionSearchRoots' })) +}) diff --git a/src/main/ai-vault/session-scanner-service-search-roots.test.ts b/src/main/ai-vault/session-scanner-service-search-roots.test.ts new file mode 100644 index 00000000000..5a206646ea0 --- /dev/null +++ b/src/main/ai-vault/session-scanner-service-search-roots.test.ts @@ -0,0 +1,106 @@ +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { + openSessionSearchIndexerHarness, + writeMessageGraphTranscript, + type SessionSearchIndexerHarness +} from '../ai-vault-search/session-search-indexer-test-fixture' +import { SessionSearchIndexer } from '../ai-vault-search/session-search-indexer' +import type { SessionSearchScanRoots } from '../ai-vault-search/session-search-scan-roots' +import { resetSessionParseCacheForTests } from './session-scanner-parse-cache' +import type { AiVaultSessionSearchInit } from './session-scanner-service-protocol' +import { SessionScannerServiceSearch } from './session-scanner-service-search' +import { resetTranscriptConsumersForTests } from './session-transcript-consumers' + +let harness: SessionSearchIndexerHarness +let subject: SessionScannerServiceSearch +let spawnRoot: string +let lateRoot: string +let currentRoots: SessionSearchScanRoots +let spawnRoots: SessionSearchScanRoots + +beforeEach(async () => { + resetSessionParseCacheForTests() + resetTranscriptConsumersForTests() + harness = await openSessionSearchIndexerHarness('ss-service-roots') + subject = new SessionScannerServiceSearch(async () => currentRoots) + const { openclawLegacyStateDir, ...rest } = harness.roots + spawnRoot = harness.roots.openclawStateDir ?? '' + lateRoot = openclawLegacyStateDir ?? '' + spawnRoots = rest + currentRoots = rest +}) + +afterEach(async () => { + subject.close() + vi.restoreAllMocks() + resetTranscriptConsumersForTests() + resetSessionParseCacheForTests() + await harness.cleanup() +}) + +function init(roots: SessionSearchScanRoots): AiVaultSessionSearchInit { + return { + databasePath: harness.databasePath, + settings: { enabled: true, historyDays: null }, + roots + } +} + +/** OpenClaw reads `/agents/**` and keeps only paths through `sessions`. */ +function openclawTranscript(stateDir: string, name: string): string { + return join(stateDir, 'agents', 'main', 'sessions', `${name}.jsonl`) +} + +async function sessionsMatching(term: string): Promise { + const reply = await subject.execute({ + type: 'request', + id: 1, + operation: 'searchSessions', + request: { query: term } + }) + if (reply.operation !== 'searchSessions' || reply.value.kind !== 'results') { + throw new Error(`expected results, got ${JSON.stringify(reply)}`) + } + return reply.value.hits.map((hit) => hit.sessionId).sort() +} + +async function indexedSessions(term: string, expected: string[]): Promise { + await vi.waitFor( + async () => { + await subject.execute({ type: 'request', id: 2, operation: 'searchReconcile' }) + expect(await sessionsMatching(term)).toEqual(expected) + }, + { timeout: 20_000 } + ) +} + +it('refreshes a late root without rebuilding the index', async () => { + await writeMessageGraphTranscript(openclawTranscript(spawnRoot, 'early-session'), [ + 'a conversation in a root the spawn already knew' + ]) + await writeMessageGraphTranscript(openclawTranscript(lateRoot, 'late-session'), [ + 'a conversation in a distro that started later' + ]) + + subject.apply(init(spawnRoots)) + await indexedSessions('conversation', ['early-session']) + + const close = vi.spyOn(SessionSearchIndexer.prototype, 'close') + currentRoots = harness.roots + await indexedSessions('conversation', ['early-session', 'late-session']) + expect(close).not.toHaveBeenCalled() +}) + +it('keeps the live indexer when an unchanged root snapshot is refreshed', async () => { + await writeMessageGraphTranscript(openclawTranscript(spawnRoot, 'early-session'), [ + 'a conversation in a root the spawn already knew' + ]) + subject.apply(init(harness.roots)) + await indexedSessions('conversation', ['early-session']) + + const close = vi.spyOn(SessionSearchIndexer.prototype, 'close') + currentRoots = { ...spawnRoots } + await indexedSessions('conversation', ['early-session']) + expect(close).not.toHaveBeenCalled() +}) diff --git a/src/main/ai-vault/session-scanner-service-search.test.ts b/src/main/ai-vault/session-scanner-service-search.test.ts new file mode 100644 index 00000000000..d0c3394883d --- /dev/null +++ b/src/main/ai-vault/session-scanner-service-search.test.ts @@ -0,0 +1,166 @@ +import { existsSync } from 'node:fs' +import { join } from 'node:path' +import { afterAll, beforeAll, expect, it, vi } from 'vitest' +import type { AiVaultSearchResponse, AiVaultSearchStatus } from '../../shared/ai-vault-search-types' +import { + openSessionSearchIndexerHarness, + writeClaudeTranscript, + type SessionSearchIndexerHarness +} from '../ai-vault-search/session-search-indexer-test-fixture' +import { + AI_VAULT_SERVICE_PROTOCOL_VERSION, + type AiVaultServiceChildMessage, + type AiVaultServiceParentMessage, + type AiVaultServiceRequestBody, + type AiVaultServiceResultValue, + type AiVaultSessionSearchInit +} from './session-scanner-service-protocol' + +/** + * The child, booted the way a spawn boots it: an init frame and messages, with + * no renderer, no Electron and no scan request. What this proves is that consent + * alone constructs the indexer and that every search answer crosses the protocol. + */ + +const SESSION_ID = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' + +let harness: SessionSearchIndexerHarness +let currentRoots: SessionSearchIndexerHarness['roots'] +let originalSend: typeof process.send +const sent: AiVaultServiceChildMessage[] = [] +let nextId = 1 + +function emit(message: AiVaultServiceParentMessage): void { + process.emit('message', message, undefined) +} + +/** One request, and the reply the child sent for it, still discriminated by operation. */ +async function call(body: AiVaultServiceRequestBody): Promise { + const id = nextId++ + emit({ ...body, id }) + const reply = await vi.waitFor(() => { + const found = sent.find( + (message) => (message.type === 'result' || message.type === 'error') && message.id === id + ) + expect(found).toBeDefined() + return found! + }) + if (reply.type === 'error') { + throw new Error(reply.message) + } + if (reply.type !== 'result') { + throw new Error(`expected a result, got ${reply.type}`) + } + return reply +} + +async function searchStatus(): Promise { + const reply = await call({ type: 'request', operation: 'searchStatus' }) + if (reply.operation !== 'searchStatus') { + throw new Error(`expected searchStatus, got ${reply.operation}`) + } + return reply.value +} + +async function searchSessions(query: string): Promise { + const reply = await call({ type: 'request', operation: 'searchSessions', request: { query } }) + if (reply.operation !== 'searchSessions') { + throw new Error(`expected searchSessions, got ${reply.operation}`) + } + return reply.value +} + +function searchInit(enabled: boolean): AiVaultSessionSearchInit { + return { + databasePath: harness.databasePath, + settings: { enabled, historyDays: null }, + roots: harness.roots + } +} + +beforeAll(async () => { + harness = await openSessionSearchIndexerHarness('ss-child') + currentRoots = harness.roots + await writeClaudeTranscript( + join(harness.claudeProjectDir, `${SESSION_ID}.jsonl`), + ['a distinctive conversation'], + SESSION_ID + ) + originalSend = process.send + const record: NonNullable = (message) => { + sent.push(message) + if (message.type === 'sessionSearchRoots') { + queueMicrotask(() => + emit({ type: 'sessionSearchRoots', id: message.id, roots: currentRoots }) + ) + } + return true + } + process.send = record + await import('./session-scanner-service-entry') + emit({ + type: 'init', + protocol: AI_VAULT_SERVICE_PROTOCOL_VERSION, + sessionParseCache: null, + sessionSearch: searchInit(true) + }) + await vi.waitFor(() => expect(sent.some((message) => message.type === 'ready')).toBe(true)) +}) + +afterAll(async () => { + emit({ type: 'sessionSearch', init: searchInit(false) }) + process.send = originalSend + await harness.cleanup() +}) + +it('reports the indexer phase and a live generation over the protocol', async () => { + const status = await vi.waitFor(async () => { + const value = await searchStatus() + expect(value.filesIndexed).toBeGreaterThan(0) + return value + }) + expect(status.enabled).toBe(true) + expect(status.phase).toBe('current') + expect(status.generation).toBeGreaterThan(0) + expect(existsSync(harness.databasePath)).toBe(true) +}) + +it('answers a search and a reconcile over the protocol', async () => { + expect(await call({ type: 'request', operation: 'searchReconcile' })).toEqual({ + operation: 'searchReconcile', + value: null, + type: 'result', + id: expect.any(Number) + }) + const response = await searchSessions('distinctive') + expect(response.kind).toBe('results') + if (response.kind === 'results') { + expect(response.hits.map((hit) => hit.sessionId)).toEqual([SESSION_ID]) + } +}) + +it('discovers a new root through the parent exchange on manual reconciliation', async () => { + const lateHome = join(harness.root, 'late-home') + const id = 'bbbbbbbb-cccc-4ddd-8eee-ffffffffffff' + await writeClaudeTranscript( + join(lateHome, '.claude', 'projects', 'late', `${id}.jsonl`), + ['freshroots'], + id + ) + currentRoots = { ...harness.roots, wslHomeDirs: [lateHome] } + await call({ type: 'request', operation: 'searchReconcile' }) + const response = await searchSessions('freshroots') + expect(response.kind).toBe('results') + if (response.kind === 'results') { + expect(response.hits.map((hit) => hit.sessionId)).toEqual([id]) + } +}) + +it('answers disabled once consent is withdrawn, without a respawn', async () => { + emit({ type: 'sessionSearch', init: searchInit(false) }) + expect(await searchSessions('distinctive')).toEqual({ kind: 'unavailable', reason: 'disabled' }) + expect(await searchStatus()).toMatchObject({ enabled: false, phase: 'idle' }) + // Re-consenting reuses the index that was left on disk rather than rebuilding it. + emit({ type: 'sessionSearch', init: searchInit(true) }) + expect((await searchSessions('distinctive')).kind).toBe('results') +}) diff --git a/src/main/ai-vault/session-scanner-service-search.ts b/src/main/ai-vault/session-scanner-service-search.ts new file mode 100644 index 00000000000..deab1c26beb --- /dev/null +++ b/src/main/ai-vault/session-scanner-service-search.ts @@ -0,0 +1,94 @@ +import type { SessionSearchIndexerOptions } from '../ai-vault-search/session-search-indexer-options' +import { unavailableSessionSearchStatus } from '../../shared/ai-vault-search-client' +import { AiVaultSearchRequestSchema } from '../../shared/ai-vault-search-contract' +import { SessionSearchInstance } from '../ai-vault-search/session-search-instance' +import { + sameSessionSearchRoots, + type SessionSearchScanRoots +} from '../ai-vault-search/session-search-scan-roots' +import { sessionSearchSqliteAvailable } from '../ai-vault-search/session-search-sqlite-support' +import type { + AiVaultServiceRequest, + AiVaultServiceResultValue, + AiVaultSessionSearchInit +} from './session-scanner-service-protocol' + +type SearchOperation = Extract< + AiVaultServiceRequest, + { operation: 'searchSessions' | 'searchStatus' | 'searchReconcile' } +> + +/** + * The scanner-service child's half of session search. + * + * Why the child and not the parent: the transcript reader runs here, so the + * index consumer has to as well — one process reads a transcript once and both + * the session list and the index see that read. Main, the CLI and a remote + * server never open the database; they ask over this protocol. + */ +export class SessionScannerServiceSearch { + private instance: SessionSearchInstance | null = null + private databasePath: string | null = null + private roots: SessionSearchScanRoots | null = null + + constructor(private readonly resolveRoots?: SessionSearchIndexerOptions['resolveRoots']) {} + + /** Applied at init and again on every settings change; both are close-and-construct. */ + apply(init: AiVaultSessionSearchInit): void { + if (!sessionSearchSqliteAvailable()) { + return + } + if (this.instance && this.databasePath !== init.databasePath) { + // A data root cannot move under a running process, so this is a caller bug + // rather than a case to support: close the old one before it writes there. + this.close() + } + if (this.instance && this.roots && !sameSessionSearchRoots(this.roots, init.roots)) { + // Explicit init-root changes replace the fallback used by callers without a resolver. + this.close() + } + this.databasePath = init.databasePath + this.roots = init.roots + this.instance ??= new SessionSearchInstance({ + databasePath: init.databasePath, + roots: init.roots, + resolveRoots: this.resolveRoots + }) + this.instance.apply(init.settings) + } + + handles(request: AiVaultServiceRequest): request is SearchOperation { + return ( + request.operation === 'searchSessions' || + request.operation === 'searchStatus' || + request.operation === 'searchReconcile' + ) + } + + async execute(request: SearchOperation): Promise { + const instance = this.instance + if (request.operation === 'searchStatus') { + return { + operation: 'searchStatus', + value: instance?.status() ?? unavailableSessionSearchStatus() + } + } + if (request.operation === 'searchReconcile') { + await instance?.reconcile() + return { operation: 'searchReconcile', value: null } + } + return { + operation: 'searchSessions', + value: instance + ? await instance.search(AiVaultSearchRequestSchema.parse(request.request)) + : { kind: 'unavailable', reason: 'disabled' } + } + } + + close(): void { + this.instance?.close() + this.instance = null + this.databasePath = null + this.roots = null + } +} diff --git a/src/main/ai-vault/session-scanner-service-spawn.ts b/src/main/ai-vault/session-scanner-service-spawn.ts index 3ba12322734..d841fc62593 100644 --- a/src/main/ai-vault/session-scanner-service-spawn.ts +++ b/src/main/ai-vault/session-scanner-service-spawn.ts @@ -1,5 +1,11 @@ +import { localAiVaultScanRoots } from './cached-session-list' import { fork, type ChildProcess } from 'node:child_process' import { existsSync } from 'node:fs' +import type { + AiVaultSearchRequest, + AiVaultSearchResponse, + AiVaultSearchStatus +} from '../../shared/ai-vault-search-types' import type { AiVaultListResult, AiVaultSubagentListResult } from '../../shared/ai-vault-types' import type { AiVaultSessionTitleRequest, @@ -10,12 +16,16 @@ import type { ReadAiVaultFirstUserPromptArgs, ReadAiVaultFirstUserPromptResult } from './session-first-user-prompt-read' +import { sessionSearchServiceInit } from '../ai-vault-search/session-search-service-init' import { getSessionParseCachePersistenceOptions } from './session-parse-cache-persistence' import { buildAiVaultServiceEnv } from './session-scanner-service-env' import { AiVaultScannerServiceClient } from './session-scanner-service-client' import { getAiVaultServiceEntryPath } from './session-scanner-service-entry-path' import { lowerAiVaultServicePriority } from './session-scanner-service-priority' -import type { AiVaultServiceSubagentRequest } from './session-scanner-service-protocol' +import type { + AiVaultServiceSubagentRequest, + AiVaultSessionSearchInit +} from './session-scanner-service-protocol' import type { AiVaultWorkerScanOptions } from './session-scanner-worker-protocol' export function spawnAiVaultServiceProcess(): ChildProcess { @@ -39,7 +49,11 @@ let sharedClient: AiVaultScannerServiceClient | null = null function getSharedClient(): AiVaultScannerServiceClient { sharedClient ??= new AiVaultScannerServiceClient({ processFactory: spawnAiVaultServiceProcess, - init: { sessionParseCache: getSessionParseCachePersistenceOptions() }, + resolveSessionSearchRoots: localAiVaultScanRoots, + init: () => ({ + sessionParseCache: getSessionParseCachePersistenceOptions(), + sessionSearch: sessionSearchServiceInit() + }), onStderr: (text) => console.error('[ai-vault-service]', text.trimEnd()) }) return sharedClient @@ -81,6 +95,25 @@ export function readAiVaultFirstUserPromptInService( return getSharedClient().request({ type: 'request', operation: 'firstPrompt', request }, signal) } +export function searchSessionsInService( + request: AiVaultSearchRequest +): Promise { + return getSharedClient().request({ type: 'request', operation: 'searchSessions', request }) +} + +export function sessionSearchStatusInService(): Promise { + return getSharedClient().request({ type: 'request', operation: 'searchStatus' }) +} + +export function reconcileSessionSearchInService(): Promise { + return getSharedClient().request({ type: 'request', operation: 'searchReconcile' }) +} + +/** Boot and every settings change: push the policy and keep a child while the index runs. */ +export function updateSessionSearchInService(init: AiVaultSessionSearchInit): void { + getSharedClient().updateSessionSearch(init) +} + export function invalidateAiVaultServiceCache(paths: string[]): Promise { return sharedClient?.invalidate(paths) ?? Promise.resolve() } diff --git a/src/main/ipc/register-core-handlers/register-core-handlers.test.ts b/src/main/ipc/register-core-handlers/register-core-handlers.test.ts index ac9e6a9bb61..870eac98d46 100644 --- a/src/main/ipc/register-core-handlers/register-core-handlers.test.ts +++ b/src/main/ipc/register-core-handlers/register-core-handlers.test.ts @@ -138,7 +138,8 @@ const { vi.mock('electron', () => ({ app: { - getPath: getPathMock + getPath: getPathMock, + once: vi.fn() } })) diff --git a/src/main/ipc/settings.test.ts b/src/main/ipc/settings.test.ts index a6a90f2b9bf..31d587bd056 100644 --- a/src/main/ipc/settings.test.ts +++ b/src/main/ipc/settings.test.ts @@ -13,6 +13,7 @@ const { resolveEnvironmentMock, rebuildAppMenuMock, applyBrowserSessionProxiesMock, + applySessionSearchSettingsChangeMock, listProfilesMock } = vi.hoisted(() => ({ applyAppIconMock: vi.fn(), @@ -27,6 +28,7 @@ const { resolveEnvironmentMock: vi.fn(), rebuildAppMenuMock: vi.fn(), applyBrowserSessionProxiesMock: vi.fn(), + applySessionSearchSettingsChangeMock: vi.fn(), listProfilesMock: vi.fn(() => []) })) @@ -61,6 +63,10 @@ vi.mock('../app-icon', () => ({ applyAppIcon: applyAppIconMock })) +vi.mock('../ai-vault-search/session-search-enablement', () => ({ + applySessionSearchSettingsChange: applySessionSearchSettingsChangeMock +})) + vi.mock('../agent-hooks/managed-agent-hook-controls', () => ({ applyAgentStatusHooksEnabled: applyAgentStatusHooksEnabledMock })) @@ -113,6 +119,7 @@ describe('registerSettingsHandlers', () => { }) rebuildAppMenuMock.mockClear() applyBrowserSessionProxiesMock.mockReset().mockResolvedValue(undefined) + applySessionSearchSettingsChangeMock.mockClear() listProfilesMock.mockReset().mockReturnValue([]) browserWindowGetAllWindowsMock.mockReset() store.getSettings.mockReset() @@ -827,4 +834,44 @@ describe('registerSettingsHandlers', () => { expect(rebuildAppMenuMock).toHaveBeenCalledTimes(1) }) + + // 3b stores the two booleans and nothing else; the consent copy and the + // history picker are PR 8's. A profile that has never opted in has no key. + it('normalizes an agent-session-search write and hands the change to the index', async () => { + const before = { aiVaultSearch: { enabled: false, historyDays: null } } + store.getSettings.mockReturnValue(before) + store.updateSettings.mockImplementation((args: object) => ({ ...before, ...args })) + registerSettingsHandlers(store as never) + const handler = handleMock.mock.calls.find((call) => call[0] === 'settings:set')?.[1] as ( + event: typeof settingsInvokeEvent, + args: unknown + ) => Promise + + await handler(settingsInvokeEvent, { + aiVaultSearch: { enabled: true, historyDays: 30.7, paused: true } + }) + + expect(store.updateSettings).toHaveBeenCalledWith( + expect.objectContaining({ aiVaultSearch: { enabled: true, historyDays: 30 } }), + expect.anything() + ) + expect(applySessionSearchSettingsChangeMock).toHaveBeenCalledWith( + before, + expect.objectContaining({ aiVaultSearch: { enabled: true, historyDays: 30 } }) + ) + }) + + it('leaves the index alone for a settings write that does not mention it', async () => { + store.getSettings.mockReturnValue({ appIcon: 'default' }) + store.updateSettings.mockReturnValue({ appIcon: 'default' }) + registerSettingsHandlers(store as never) + const handler = handleMock.mock.calls.find((call) => call[0] === 'settings:set')?.[1] as ( + event: typeof settingsInvokeEvent, + args: unknown + ) => Promise + + await handler(settingsInvokeEvent, { appIcon: 'default' }) + + expect(applySessionSearchSettingsChangeMock).not.toHaveBeenCalled() + }) }) diff --git a/src/main/ipc/settings.ts b/src/main/ipc/settings.ts index 1d4194825d9..f3c1b8aeaf8 100644 --- a/src/main/ipc/settings.ts +++ b/src/main/ipc/settings.ts @@ -36,6 +36,8 @@ import { computerAwakeSettingsForMode, normalizeComputerAwakeMode } from '../../shared/computer-awake-mode' +import { resolveAiVaultSearchSettings } from '../../shared/ai-vault-search-settings' +import { applySessionSearchSettingsChange } from '../ai-vault-search/session-search-enablement' // Why: the whitelist is the source-of-truth for which keys we emit on. Casting // to a Set once at module load lets the IPC handler's per-key membership @@ -160,6 +162,9 @@ export function registerSettingsHandlers( if ('appIcon' in args) { sanitizedArgs.appIcon = normalizeAppIconId(args.appIcon) } + if ('aiVaultSearch' in args) { + sanitizedArgs.aiVaultSearch = resolveAiVaultSearchSettings(args) + } if ('terminalCustomThemes' in args) { sanitizedArgs.terminalCustomThemes = normalizeTerminalCustomThemes(args.terminalCustomThemes) } @@ -266,6 +271,9 @@ export function registerSettingsHandlers( if ('appIcon' in sanitizedArgs && before.appIcon !== result.appIcon) { applyAppIcon(result.appIcon) } + if ('aiVaultSearch' in sanitizedArgs) { + applySessionSearchSettingsChange(before, result) + } // Why: telemetry-plan.md§Settings — fire `settings_changed` only for // whitelisted keys, with `value_kind` distinguishing booleans from diff --git a/src/main/orcad/orcad-command-arguments.ts b/src/main/orcad/orcad-command-arguments.ts new file mode 100644 index 00000000000..f4fd748de61 --- /dev/null +++ b/src/main/orcad/orcad-command-arguments.ts @@ -0,0 +1,43 @@ +import type { OrcadOptions } from './orcad-entry' + +/** + * orcad's flags. A value-taking flag consumes the next token whatever it looks + * like, so `--bind --json` binds to the literal `--json`; only a missing token + * is an error. Pinned by orcad-launch-contract.test.ts. + */ +export function parseArgs(argv: string[]): OrcadOptions { + const options: OrcadOptions = {} + for (let i = 0; i < argv.length; i += 1) { + const arg = argv[i] + if (arg === '--port') { + const raw = argv[i + 1] + const port = Number(raw) + if (!Number.isInteger(port) || port < 0 || port > 65535) { + throw new Error(`--port expects an integer 0-65535, got ${raw ?? "''"}`) + } + options.port = port + i += 1 + } else if (arg === '--json') { + options.json = true + } else if (arg === '--no-pairing') { + options.noPairing = true + } else if (arg === '--bind') { + const value = argv[i + 1] + if (value === undefined) { + throw new Error('--bind expects a value') + } + options.bind = value + i += 1 + } else if (arg === '--pairing-address') { + const value = argv[i + 1] + if (!value) { + throw new Error('--pairing-address expects a value') + } + options.pairingAddress = value + i += 1 + } else { + throw new Error(`Unknown argument: ${arg}`) + } + } + return options +} diff --git a/src/main/orcad/orcad-entry.ts b/src/main/orcad/orcad-entry.ts index 137894f87b8..3dc84906c99 100644 --- a/src/main/orcad/orcad-entry.ts +++ b/src/main/orcad/orcad-entry.ts @@ -25,6 +25,9 @@ import { } from './orcad-bind-address' import { acquireOrcadInstanceLock, OrcadInstanceLockError } from './orcad-instance-lock' import { startOrcadWithLifecycle } from './orcad-lifecycle' +import { parseArgs } from './orcad-command-arguments' + +export { parseArgs } let runOrcadQuitHandlers = (): void => {} @@ -242,6 +245,13 @@ async function startOrcadRuntime( isAgentStatusHooksEnabled(store.getSettings()) ? agentHookServer.buildPtyEnv() : {} }) + const { installOrcadSessionSearchService } = await import('./orcad-session-search') + const sessionSearch = await installOrcadSessionSearchService({ + userDataPath: runtimeUserDataPath, + getSettings: () => store.getSettings() + }) + getAppEnvironment().onWillQuit(() => sessionSearch?.dispose()) + // Why here too and not only on the desktop: nothing else republishes `session.tabs` when a // pane's status row changes, and orcad's whole job is serving paired clients. uninstallHookStatusRepublish = installHookStatusSessionTabsRepublish( @@ -338,43 +348,6 @@ async function startOrcadRuntime( return { readiness } } -export function parseArgs(argv: string[]): OrcadOptions { - const options: OrcadOptions = {} - for (let i = 0; i < argv.length; i += 1) { - const arg = argv[i] - if (arg === '--port') { - const raw = argv[i + 1] - const port = Number(raw) - if (!Number.isInteger(port) || port < 0 || port > 65535) { - throw new Error(`--port expects an integer 0-65535, got ${raw ?? "''"}`) - } - options.port = port - i += 1 - } else if (arg === '--json') { - options.json = true - } else if (arg === '--no-pairing') { - options.noPairing = true - } else if (arg === '--bind') { - const value = argv[i + 1] - if (value === undefined) { - throw new Error('--bind expects a value') - } - options.bind = value - i += 1 - } else if (arg === '--pairing-address') { - const value = argv[i + 1] - if (!value) { - throw new Error('--pairing-address expects a value') - } - options.pairingAddress = value - i += 1 - } else { - throw new Error(`Unknown argument: ${arg}`) - } - } - return options -} - /** * Exit codes a supervisor can act on. Closed set — see docs/reference/orcad-operations.md. * diff --git a/src/main/orcad/orcad-session-search.ts b/src/main/orcad/orcad-session-search.ts new file mode 100644 index 00000000000..7f5d67010a8 --- /dev/null +++ b/src/main/orcad/orcad-session-search.ts @@ -0,0 +1,26 @@ +import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' +import { resolveAiVaultSearchSettings } from '../../shared/ai-vault-search-settings' +import type { GlobalSettings } from '../../shared/global-settings-types' +import { localAiVaultScanRoots } from '../ai-vault/cached-session-list' +import { installInProcessSessionSearchService } from '../ai-vault-search/session-search-in-process-service' + +/** + * orcad's session search registration. + * + * In this process and not a scanner child: orcad ships only the watcher and the + * daemon entries beside `orcad.js`, so there is no scanner-service child here to + * own the index — and this process is the sole writer, so nothing can race it. + * Null on a host whose Node has no `node:sqlite`, which is orcad's stated floor. + */ +export async function installOrcadSessionSearchService(args: { + userDataPath: string + getSettings: () => Pick +}): Promise<{ dispose(): void } | null> { + return installInProcessSessionSearchService({ + dataRoot: args.userDataPath, + roots: { executionHostId: LOCAL_EXECUTION_HOST_ID }, + resolveRoots: localAiVaultScanRoots, + settings: resolveAiVaultSearchSettings(args.getSettings()), + onError: (error) => console.error('[orcad] session search:', error) + }) +} diff --git a/src/main/startup/main-process-runtime-service.ts b/src/main/startup/main-process-runtime-service.ts index 3aac4a03b3c..a0a65d54e7a 100644 --- a/src/main/startup/main-process-runtime-service.ts +++ b/src/main/startup/main-process-runtime-service.ts @@ -1,3 +1,5 @@ +import { installChildSessionSearchService } from '../ai-vault-search/session-search-enablement' +import { getCanonicalUserDataPath } from '../persistence/loading-store/user-data-path' import { app } from 'electron' import { OrcaRuntimeService } from '../runtime/orca-runtime' import { getLocalPtyProvider, getSshPtyProvider, clearProviderPtyState } from '../ipc/pty' @@ -131,6 +133,12 @@ export function initializeMainProcessRuntime(): OrcaRuntimeService { orchestrationEnvironmentTransport, skillTransactionRecovery: state.skillTransactionRecovery }) + // Both desktop and headless serve own a host-local search service. + const sessionSearch = installChildSessionSearchService({ + dataRoot: getCanonicalUserDataPath(), + getSettings: () => store.getSettings() + }) + app.once('will-quit', () => sessionSearch?.dispose()) state.runtime = runtime agentHookServer.subscribeEnrichedStatus((enriched) => recordObservedAgentStatusPaneIdentity(observedPaneIdentities, enriched.paneKey, runtime) diff --git a/src/relay/relay-runtime-services.ts b/src/relay/relay-runtime-services.ts index 73e03242af6..4295014782f 100644 --- a/src/relay/relay-runtime-services.ts +++ b/src/relay/relay-runtime-services.ts @@ -1,6 +1,10 @@ import { homedir } from 'node:os' +import { join } from 'node:path' import { getRemoteHostPlatform } from '../main/ssh/ssh-remote-platform' -import { parseUnameToRelayPlatform } from '../main/ssh/relay-protocol' +import { parseUnameToRelayPlatform, RELAY_REMOTE_DIR } from '../main/ssh/relay-protocol' +import { DEFAULT_AI_VAULT_SEARCH_SETTINGS } from '../shared/ai-vault-search-settings' +import { LOCAL_EXECUTION_HOST_ID } from '../shared/execution-host' +import { installInProcessSessionSearchService } from '../main/ai-vault-search/session-search-in-process-service' import type { RelayDispatcher } from './dispatcher' import { RelayContext, expandTilde } from './context' import { PtyHandler } from './pty-handler' @@ -29,6 +33,7 @@ export class RelayRuntimeServices { readonly gitHandler: GitHandler readonly skillInstallHandler: SkillInstallHandler private readonly aiVaultService: ReturnType | null + private readonly sessionSearch: { dispose(): void } | null private readonly registeredHandlers: readonly unknown[] constructor( @@ -77,6 +82,22 @@ export class RelayRuntimeServices { const relayPlatform = parseUnameToRelayPlatform(process.platform, process.arch) const hostPlatform = relayPlatform ? getRemoteHostPlatform(relayPlatform) : undefined this.aiVaultService = hostPlatform ? createRelayAiVaultService(homedir(), hostPlatform) : null + // Why beside the AI Vault sidecar and not inside it: that sidecar runs the + // remote scanner, which reads through a filesystem provider and publishes + // nothing to the transcript channel the index consumes. This process is the + // one that would drive the index's own reads, and the only writer on the file. + // Off until something can carry consent to a remote host (see the PR body); + // registering it anyway is what makes this host answer `disabled` and not + // `no-service`, which is the difference between off and too old. + this.sessionSearch = installInProcessSessionSearchService({ + dataRoot: join(homedir(), RELAY_REMOTE_DIR), + roots: { executionHostId: LOCAL_EXECUTION_HOST_ID }, + settings: DEFAULT_AI_VAULT_SEARCH_SETTINGS, + onError: (error) => + relayLogLine( + `[relay] session search: ${error instanceof Error ? error.message : String(error)}` + ) + }) this.registeredHandlers = [ preflightHandler, this.skillInstallHandler, @@ -112,6 +133,7 @@ export class RelayRuntimeServices { } disposeHandlers(): void { + this.sessionSearch?.dispose() this.fsHandler.dispose() this.gitHandler.dispose() void this.registeredHandlers diff --git a/src/shared/ai-vault-search-settings.test.ts b/src/shared/ai-vault-search-settings.test.ts new file mode 100644 index 00000000000..78ff95fa2f3 --- /dev/null +++ b/src/shared/ai-vault-search-settings.test.ts @@ -0,0 +1,81 @@ +import { expect, it } from 'vitest' +import { + AiVaultSearchSettingsSchema, + DEFAULT_AI_VAULT_SEARCH_SETTINGS, + resolveAiVaultSearchSettings, + sameAiVaultSearchSettings +} from './ai-vault-search-settings' + +// Off is the only safe default: building the index reads every transcript on the +// machine, so a profile that has never answered must read as "no". +it('reads anything that is not an explicit opt-in as off', () => { + expect(resolveAiVaultSearchSettings(undefined)).toEqual(DEFAULT_AI_VAULT_SEARCH_SETTINGS) + expect(resolveAiVaultSearchSettings({})).toEqual(DEFAULT_AI_VAULT_SEARCH_SETTINGS) + expect(resolveAiVaultSearchSettings({ aiVaultSearch: null })).toEqual( + DEFAULT_AI_VAULT_SEARCH_SETTINGS + ) + expect(resolveAiVaultSearchSettings({ aiVaultSearch: { enabled: 'yes' } })).toEqual( + DEFAULT_AI_VAULT_SEARCH_SETTINGS + ) + expect(resolveAiVaultSearchSettings({ aiVaultSearch: 'on' })).toEqual( + DEFAULT_AI_VAULT_SEARCH_SETTINGS + ) +}) + +it('normalizes a history bound and drops anything that is not one', () => { + expect( + resolveAiVaultSearchSettings({ aiVaultSearch: { enabled: true, historyDays: 30.7 } }) + ).toEqual({ enabled: true, historyDays: 30 }) + // A fractional day floors to zero, which would read as "all history" on one + // side and "cutoff is now" on the other. + for (const historyDays of [0.4, 0, -30, Number.NaN] as const) { + expect(resolveAiVaultSearchSettings({ aiVaultSearch: { enabled: true, historyDays } })).toEqual( + { enabled: true, historyDays: null } + ) + } + expect( + resolveAiVaultSearchSettings({ aiVaultSearch: { enabled: true, historyDays: 999_999 } }) + ).toEqual({ enabled: true, historyDays: 3_650 }) +}) + +// There is no `paused`: the indexer is immutable, so a pause would be a second +// lifetime for one object's store, queue and sweep flag. +it('keeps only the two fields the indexer is constructed from', () => { + expect( + resolveAiVaultSearchSettings({ + aiVaultSearch: { enabled: true, historyDays: 90, paused: true } + }) + ).toEqual({ enabled: true, historyDays: 90 }) +}) + +it('accepts what it produces and refuses what it does not', () => { + expect(AiVaultSearchSettingsSchema.parse({ enabled: true, historyDays: 90 })).toEqual({ + enabled: true, + historyDays: 90 + }) + expect(AiVaultSearchSettingsSchema.safeParse({ enabled: true, historyDays: 0 }).success).toBe( + false + ) + expect(AiVaultSearchSettingsSchema.safeParse({ historyDays: null }).success).toBe(false) +}) + +it('treats an unchanged policy as unchanged so a re-save never restarts the index', () => { + expect( + sameAiVaultSearchSettings( + { enabled: true, historyDays: 30 }, + { enabled: true, historyDays: 30 } + ) + ).toBe(true) + expect( + sameAiVaultSearchSettings( + { enabled: true, historyDays: 30 }, + { enabled: true, historyDays: 90 } + ) + ).toBe(false) + expect( + sameAiVaultSearchSettings( + { enabled: true, historyDays: null }, + { enabled: false, historyDays: null } + ) + ).toBe(false) +}) diff --git a/src/shared/ai-vault-search-settings.ts b/src/shared/ai-vault-search-settings.ts new file mode 100644 index 00000000000..8b6b8e1b07c --- /dev/null +++ b/src/shared/ai-vault-search-settings.ts @@ -0,0 +1,65 @@ +import { z } from 'zod' + +/** + * Consent and retention for the agent-session transcript index. + * + * Off until the user turns it on: building the index reads every transcript on + * the machine, so nothing constructs an indexer, opens the database or reads a + * transcript for it before that choice is recorded. + * + * There is no `paused`. The indexer is immutable after construction, so every + * change here is close-and-construct (see session-search-instance.ts). + */ +export type AiVaultSearchSettings = { + enabled: boolean + /** null = all history; otherwise only transcripts modified within this many days. */ + historyDays: number | null +} + +export const DEFAULT_AI_VAULT_SEARCH_SETTINGS: AiVaultSearchSettings = { + enabled: false, + historyDays: null +} + +const HISTORY_DAYS_MAX = 3_650 + +export const AiVaultSearchSettingsSchema: z.ZodType = z.object({ + enabled: z.boolean(), + historyDays: z.number().int().positive().max(HISTORY_DAYS_MAX).nullable() +}) + +export function normalizeAiVaultSearchHistoryDays(value: unknown): number | null { + if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) { + return null + } + // A fractional day floors to 0, which reads as "all history" on one side and + // "now" on the other; make the two agree. + const days = Math.floor(value) + return days <= 0 ? null : Math.min(HISTORY_DAYS_MAX, days) +} + +/** + * The persisted shape, from whatever a settings write or an old profile left behind. + * + * The input is `unknown` on purpose: this is the sanitizer, and what it reads is a + * JSON profile that may predate either field or hold a value no version wrote. + */ +export function resolveAiVaultSearchSettings( + settings: { aiVaultSearch?: unknown } | null | undefined +): AiVaultSearchSettings { + const raw = settings?.aiVaultSearch + if (typeof raw !== 'object' || raw === null) { + return { ...DEFAULT_AI_VAULT_SEARCH_SETTINGS } + } + return { + enabled: 'enabled' in raw && raw.enabled === true, + historyDays: normalizeAiVaultSearchHistoryDays('historyDays' in raw ? raw.historyDays : null) + } +} + +export function sameAiVaultSearchSettings( + a: AiVaultSearchSettings, + b: AiVaultSearchSettings +): boolean { + return a.enabled === b.enabled && a.historyDays === b.historyDays +} diff --git a/src/shared/global-settings-types.ts b/src/shared/global-settings-types.ts index e65fec28c39..6369033c892 100644 --- a/src/shared/global-settings-types.ts +++ b/src/shared/global-settings-types.ts @@ -1,6 +1,7 @@ import type { ExecutionHostId } from './execution-host' import type { GitHubProjectSettings } from './github/project-types' import type { VoiceSettings } from './speech-types' +import type { AiVaultSearchSettings } from './ai-vault-search-settings' import type { GitLabProjectSettings } from './gitlab-types' import type { TaskProvider } from './task-providers' import type { KeybindingOverrides, TerminalShortcutPolicy } from './keybindings' @@ -488,6 +489,8 @@ export type GlobalSettings = { tabSwitchKeybindingSeed?: 'pending' | 'done' /** Local voice/dictation config. Optional for pre-voice profiles; getDefaultSettings() hydrates defaults via the persistence merge. */ voice?: VoiceSettings + /** Transcript full-text search consent + retention. Absent means off; nothing indexes until the user opts in. */ + aiVaultSearch?: AiVaultSearchSettings } export type OrcaWorkspaceLayout = {