From f57130176480b57fc43f72556d933b3f3e17661c Mon Sep 17 00:00:00 2001 From: mesalogo Date: Sat, 3 Oct 2026 02:53:51 +0800 Subject: [PATCH] perf: run knowledge, story graph and conversation search reads in a read-only worker --- build/run-knowledge-perf.cjs | 82 ++++-- electron.vite.config.ts | 1 + src/main/agent/knowledge-mcp-gateway.ts | 4 +- src/main/assistant/assistant-database.ts | 153 ++++++++--- src/main/assistant/story-graph-reader.test.ts | 22 ++ src/main/assistant/story-graph-reader.ts | 17 +- src/main/index.ts | 7 +- src/main/ipc.ts | 2 +- src/main/knowledge/knowledge-database.ts | 52 ++++ src/main/knowledge/knowledge-service.ts | 11 +- src/main/readonly-query-reader.test.ts | 241 ++++++++++++++++++ src/main/readonly-query-reader.ts | 204 +++++++++++++++ src/main/readonly-query-worker.ts | 68 +++++ 13 files changed, 799 insertions(+), 65 deletions(-) create mode 100644 src/main/readonly-query-reader.test.ts create mode 100644 src/main/readonly-query-reader.ts create mode 100644 src/main/readonly-query-worker.ts diff --git a/build/run-knowledge-perf.cjs b/build/run-knowledge-perf.cjs index 42b66758..e906458a 100644 --- a/build/run-knowledge-perf.cjs +++ b/build/run-knowledge-perf.cjs @@ -8,9 +8,24 @@ const { createHash } = require('node:crypto') const { mkdtempSync, rmSync, writeFileSync } = require('node:fs') const { tmpdir } = require('node:os') const { join, resolve } = require('node:path') -const { performance } = require('node:perf_hooks') +const { monitorEventLoopDelay, performance } = require('node:perf_hooks') const root = resolve(__dirname, '..') + +// Measure on the project's Electron (the production runtime). Older runtimes +// (Node 24.18 / Electron 43) stall worker threads for hundreds of ms on many +// small ArrayBuffer allocations (SQLite BLOB rows); GB_KPERF_NODE=1 opts out. +if (process.env.GB_KPERF_NODE !== '1') { + let electron + let version + try { electron = require('electron'); version = require('electron/package.json').version } catch { electron = undefined } + if (typeof electron === 'string' && process.versions.electron !== version) { + const { status } = require('node:child_process').spawnSync(electron, [__filename], { + stdio: 'inherit', env: { ...process.env, ELECTRON_RUN_AS_NODE: '1' } + }) + process.exit(status ?? 1) + } +} const sizes = (process.env.GB_KPERF_SIZES || '5000,20000,50000').split(',').map(Number) const dimensions = Number(process.env.GB_KPERF_DIMENSIONS || 384) const chunksPerDocument = 100 @@ -18,15 +33,23 @@ const queries = Number(process.env.GB_KPERF_QUERIES || 15) const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-knowledge-perf-')) const bundle = join(directory, 'knowledge-database.cjs') -buildSync({ - entryPoints: [join(root, 'src/main/knowledge/knowledge-database.ts')], - bundle: true, - platform: 'node', - format: 'cjs', - target: 'node22', - outfile: bundle, - logLevel: 'error' -}) +const workerBundle = join(directory, 'readonly-query-worker.cjs') +for (const [entry, outfile] of [ + ['src/main/knowledge/knowledge-database.ts', bundle], + ['src/main/readonly-query-worker.ts', workerBundle] +]) { + buildSync({ + entryPoints: [join(root, entry)], + bundle: true, + platform: 'node', + format: 'cjs', + target: 'node22', + outfile, + logLevel: 'error', + // The worker reuses AssistantDatabase, which imports Electron-only modules lazily. + external: ['electron'] + }) +} const { KnowledgeDatabase } = require(bundle) // Deterministic pseudo-random generator so runs are comparable. @@ -53,8 +76,30 @@ function time(body) { } const round = value => Math.round(value * 10) / 10 +// Main event-loop delay while `queries` searches run back to back. A sync +// search blocks the loop for its whole duration; a worker search should not. +async function loopDelay(body) { + const histogram = monitorEventLoopDelay({ resolution: 5 }) + await body() // warm worker / caches outside the measurement + // Let the warm-up's blocked interval drain before sampling. + await new Promise(resolve => setTimeout(resolve, 50)) + histogram.enable() + const started = performance.now() + for (let index = 0; index < queries; index += 1) { + await body() + // Yield like a real IPC handler, so the histogram samples between queries. + await new Promise(resolve => setImmediate(resolve)) + } + const wallMs = performance.now() - started + await new Promise(resolve => setTimeout(resolve, 20)) + histogram.disable() + const ms = value => round(value / 1e6) + return { p50: ms(histogram.percentile(50)), p99: ms(histogram.percentile(99)), max: ms(histogram.max), perQueryMs: round(wallMs / queries) } +} + const results = [] -try { +async function main() { + console.log(`[kperf] runtime: node ${process.version}${process.versions.electron ? ` (electron ${process.versions.electron})` : ''}`) for (const size of sizes) { const path = join(directory, `knowledge-${size}.sqlite`) const database = new KnowledgeDatabase(path) @@ -82,15 +127,22 @@ try { seedSeconds: round(seedMs / 1000), ftsMs: time(() => reopened.search({ knowledgeBaseId: library.id, query: 'lighthouse budget', limit: 40 })), vectorMs: time(() => reopened.vectorSearch({ knowledgeBaseId: library.id, provider: 'perf', model: 'perf-model', vector: query, limit: 40 })), - hybridMs: time(() => reopened.hybridSearchWithDiagnostics({ knowledgeBaseId: library.id, query: 'lighthouse budget', limit: 40, provider: 'perf', model: 'perf-model', vector: query, graphEnabled: false, candidateMultiplier: 1 })) + hybridMs: time(() => reopened.hybridSearchWithDiagnostics(hybridOptions(library.id, query))) } + // Main event-loop delay: synchronous search on Main vs the readonly worker. + entry.mainLoopDelaySyncMs = await loopDelay(async () => reopened.hybridSearchWithDiagnostics(hybridOptions(library.id, query))) + reopened.enableReadonlyWorker(workerBundle) + entry.mainLoopDelayWorkerMs = await loopDelay(() => reopened.hybridSearchWithDiagnosticsAsync(hybridOptions(library.id, query))) reopened.close() results.push(entry) console.log(`[kperf] ${size} chunks: fts p95 ${entry.ftsMs.p95} ms, vector p95 ${entry.vectorMs.p95} ms, hybrid p95 ${entry.hybridMs.p95} ms (seed ${entry.seedSeconds} s)`) + console.log(`[kperf] ${size} chunks: Main loop delay max sync ${entry.mainLoopDelaySyncMs.max} ms vs worker ${entry.mainLoopDelayWorkerMs.max} ms (p99 ${entry.mainLoopDelaySyncMs.p99} / ${entry.mainLoopDelayWorkerMs.p99} ms; per query ${entry.mainLoopDelaySyncMs.perQueryMs} / ${entry.mainLoopDelayWorkerMs.perQueryMs} ms)`) } const output = process.env.GB_KPERF_OUTPUT if (output) writeFileSync(output, JSON.stringify({ kind: 'goodbuddy-knowledge-retrieval-benchmark', node: process.version, results }, null, 2)) - console.table(results.map(r => ({ chunks: r.chunks, 'fts p50/p95': `${r.ftsMs.p50} / ${r.ftsMs.p95}`, 'vector p50/p95': `${r.vectorMs.p50} / ${r.vectorMs.p95}`, 'hybrid p50/p95': `${r.hybridMs.p50} / ${r.hybridMs.p95}` }))) -} finally { - rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }) + console.table(results.map(r => ({ chunks: r.chunks, 'fts p50/p95': `${r.ftsMs.p50} / ${r.ftsMs.p95}`, 'vector p50/p95': `${r.vectorMs.p50} / ${r.vectorMs.p95}`, 'hybrid p50/p95': `${r.hybridMs.p50} / ${r.hybridMs.p95}`, 'loop max sync/worker': `${r.mainLoopDelaySyncMs.max} / ${r.mainLoopDelayWorkerMs.max}` }))) } +const hybridOptions = (knowledgeBaseId, vector) => ({ knowledgeBaseId, query: 'lighthouse budget', limit: 40, provider: 'perf', model: 'perf-model', vector, graphEnabled: false, candidateMultiplier: 1 }) +main().catch(error => { process.exitCode = 1; console.error(error) }).finally(() => { + rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }) +}) diff --git a/electron.vite.config.ts b/electron.vite.config.ts index f9c77955..4f62eb7f 100644 --- a/electron.vite.config.ts +++ b/electron.vite.config.ts @@ -133,6 +133,7 @@ export default defineConfig({ index: resolve('src/main/index.ts'), 'assistant-storage-worker': resolve('src/main/assistant-storage-worker.ts'), 'execution-stats-worker': resolve('src/main/execution-stats-worker.ts'), + 'readonly-query-worker': resolve('src/main/readonly-query-worker.ts'), 'wechat-sidecar': resolve( 'src/main/channels/wechat-sidecar.ts' ), diff --git a/src/main/agent/knowledge-mcp-gateway.ts b/src/main/agent/knowledge-mcp-gateway.ts index d2419386..8a3bd6a5 100644 --- a/src/main/agent/knowledge-mcp-gateway.ts +++ b/src/main/agent/knowledge-mcp-gateway.ts @@ -10,7 +10,7 @@ export type StoryGraphRemoteBinding = { } export type StoryGraphService = { available(binding: StoryGraphBinding): Promise - read(name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record + read(name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record | Promise> } import { knowledgeReferenceKey, toKnowledgeReference } from '../../shared/knowledge-reference' import { @@ -1611,7 +1611,7 @@ export class KnowledgeMcpGateway { effectiveSignal.throwIfAborted() if (!await this.isStoryGraphAvailable(token)) throw new Error('story_graph_unavailable: Supervisor or runtime capability is disabled') effectiveSignal.throwIfAborted() - const result = this.storyGraphService!.read(name, input, capability.storyGraph!.projectId, effectiveSignal) + const result = await this.storyGraphService!.read(name, input, capability.storyGraph!.projectId, effectiveSignal) effectiveSignal.throwIfAborted() if (!await this.isStoryGraphAvailable(token)) throw new Error('story_graph_unavailable: Supervisor or runtime capability is disabled') effectiveSignal.throwIfAborted() diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index ee4b4e92..a5c5f367 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -5,6 +5,7 @@ import { knowledgeReferenceKey } from '../../shared/knowledge-reference' import { appendConversationQuestionBlock } from '../../shared/conversation-question-blocks' import { DatabaseSync } from 'node:sqlite' import { ExecutionStatsReader } from './execution-stats-reader' +import { ReadonlyQueryReader, readWithFallback } from '../readonly-query-reader' import { statSync } from 'node:fs' import { MagicNoteStorage } from '../magic-notes/magic-note-storage' import type { AssistantStorageProgress } from '../../shared/assistant-storage-contracts' @@ -1964,6 +1965,9 @@ export class AssistantDatabase { private database?: DatabaseSync private activityHistoryCache?: { records: Map; dataVersion: number } private executionStatsReader?: ExecutionStatsReader + private readonlyReader?: ReadonlyQueryReader + private readonlyWorkerPath?: string + private foldedSearchConnection?: DatabaseSync private readonly noteStorage: MagicNoteStorage private readonly dirtyMagicNotes = new Set() private readonly pendingMagicNoteCleanup = new Set() @@ -2393,6 +2397,9 @@ export class AssistantDatabase { close(): void { this.executionStatsReader?.close() this.executionStatsReader = undefined + this.readonlyReader?.close() + this.readonlyReader = undefined + this.foldedSearchConnection = undefined this.executionStatsCache.clear() this.activityHistoryCache = undefined this.database?.close() @@ -3021,61 +3028,99 @@ export class AssistantDatabase { private readConversationList(detailIds?: ReadonlySet): import('../../shared/assistant-contracts').ConversationListSnapshot[] { const database = this.requireDatabase() const conversations = this.listConversationRows() - const messageStatement = database.prepare( - `SELECT id, conversation_id, role, content, state, metadata_json, - created_at - FROM ( - SELECT id, conversation_id, role, content, state, metadata_json, - created_at, sequence - FROM messages - WHERE conversation_id = ? - ORDER BY sequence DESC - ) - ORDER BY sequence ASC` - ) - const summaryStatement = database.prepare( - `SELECT count(*) AS count, - max(state = 'streaming') AS streaming, - max(created_at) AS latestMessageAt, - (SELECT role FROM messages WHERE conversation_id = ? ORDER BY sequence LIMIT 1) AS firstRole - FROM messages WHERE conversation_id = ?` - ) + type MessageSummaryRow = { + conversation_id: string; count: number; streaming: number | null + firstRole: ConversationMessage['role'] | null; latestMessageAt: string | null + } const activeIds = new Set( (database.prepare(`SELECT conversation_id FROM (${activeVisibleTaskSelect})`).all() as { conversation_id: string }[]).map(row => row.conversation_id) ) + // Batched reads (PERF-15): one aggregate for every summary and one ordered + // scan for every detailed conversation, instead of 1-2 statements each. + const summaries = new Map() + if (detailIds) { + for (const row of database.prepare( + `SELECT m.conversation_id AS conversation_id, count(*) AS count, + max(m.state = 'streaming') AS streaming, + max(m.created_at) AS latestMessageAt, + (SELECT f.role FROM messages f WHERE f.conversation_id = m.conversation_id + ORDER BY f.sequence LIMIT 1) AS firstRole + FROM messages m JOIN conversations c ON c.id = m.conversation_id + WHERE c.status = 'active' + GROUP BY m.conversation_id` + ).all() as MessageSummaryRow[]) summaries.set(row.conversation_id, row) + } + const summaryOf = (id: string): MessageSummaryRow => summaries.get(id) ?? + { conversation_id: id, count: 0, streaming: null, firstRole: null, latestMessageAt: null } + const detailed = conversations.filter(conversation => + !detailIds || detailIds.has(conversation.id) || activeIds.has(conversation.id) || + Boolean(summaryOf(conversation.id).streaming)) + const messagesByConversation = new Map() + if (detailed.length > 0) { + for (const row of database.prepare( + `SELECT id, conversation_id, role, content, state, metadata_json, created_at + FROM messages + WHERE conversation_id IN (SELECT value FROM json_each(?)) + ORDER BY conversation_id, sequence ASC` + ).all(JSON.stringify(detailed.map(conversation => conversation.id))) as MessageRow[]) { + const list = messagesByConversation.get(row.conversation_id) + if (list) list.push(row) + else messagesByConversation.set(row.conversation_id, [row]) + } + } + const detailedIds = new Set(detailed.map(conversation => conversation.id)) return conversations.map((conversation) => { - if (detailIds && !detailIds.has(conversation.id) && !activeIds.has(conversation.id)) { - const summary = summaryStatement.get(conversation.id, conversation.id) as { - count: number; streaming: number | null; firstRole: ConversationMessage['role'] | null; latestMessageAt: string | null - } - if (!summary.streaming) { - return { - ...toConversationSnapshot(conversation, []), - messageSummary: { - count: summary.count, - firstRole: summary.firstRole ?? undefined, - latestMessageAt: summary.latestMessageAt === null ? undefined : Date.parse(summary.latestMessageAt) - } + if (!detailedIds.has(conversation.id)) { + const summary = summaryOf(conversation.id) + return { + ...toConversationSnapshot(conversation, []), + messageSummary: { + count: summary.count, + firstRole: summary.firstRole ?? undefined, + latestMessageAt: summary.latestMessageAt === null ? undefined : Date.parse(summary.latestMessageAt) } } } - return toConversationSnapshot(conversation, messageStatement.all(conversation.id) as MessageRow[]) + return toConversationSnapshot(conversation, messagesByConversation.get(conversation.id) ?? []) }) } + /** Registers the Unicode case-folding matcher used by conversation search. */ + private foldedSearchDatabase(): DatabaseSync { + const database = this.requireDatabase() + if (this.foldedSearchConnection !== database) { + // Same folding as the renderer's search (String#toLocaleLowerCase), which + // SQLite's ASCII-only lower()/LIKE cannot reproduce for non-ASCII text. + database.function('goodbuddy_folded_includes', { deterministic: true }, (text, needle) => + typeof text === 'string' && typeof needle === 'string' && + text.toLocaleLowerCase().includes(needle) ? 1 : 0) + this.foldedSearchConnection = database + } + return database + } + searchConversations(query: string): string[] { const normalized = query.trim().toLocaleLowerCase() if (!normalized) return [] - const database = this.requireDatabase() - // Read only searchable text, never tool/subagent/image metadata. Use the - // same Unicode case folding as the renderer's existing search. - const messages = database.prepare('SELECT content FROM messages WHERE conversation_id = ?') - return this.listConversationRows().filter(conversation => - conversation.title.toLocaleLowerCase().includes(normalized) || - (messages.all(conversation.id) as { content: string }[]).some(message => - message.content.toLocaleLowerCase().includes(normalized)) - ).map(conversation => conversation.id) + // Read only searchable text, never tool/subagent/image metadata. One query + // with the same Unicode case folding as the renderer's existing search. + return (this.foldedSearchDatabase().prepare( + `SELECT c.id FROM conversations c + WHERE c.status = 'active' AND ( + goodbuddy_folded_includes(c.title, ?1) + OR EXISTS (SELECT 1 FROM messages m WHERE m.conversation_id = c.id + AND goodbuddy_folded_includes(m.content, ?1))) + ORDER BY c.pinned DESC, c.updated_at DESC` + ).all(normalized) as { id: string }[]).map(row => row.id) + } + + /** Conversation search on the readonly worker; falls back to the synchronous query. */ + searchConversationsAsync(query: string, signal?: AbortSignal): Promise { + this.requireDatabase() + if (!query.trim()) return Promise.resolve([]) + return readWithFallback(this.readonlyQueryReader(), 'searchConversations', [query], signal, + () => this.searchConversations(query)) } getConversation(conversationId: string): ConversationSnapshot { @@ -4924,6 +4969,25 @@ export class AssistantDatabase { this.database = new DatabaseSync(this.databasePath, { readOnly: true, timeout: 5_000 }) } + /** + * Routes heavy read-only queries (story graph, conversation search) through a + * read-only worker thread. In-memory databases keep the synchronous path. + */ + enableReadonlyWorker(workerPath: string): void { + this.readonlyWorkerPath = workerPath + } + + private readonlyQueryReader(): ReadonlyQueryReader | undefined { + if (!this.readonlyWorkerPath || this.databasePath === ':memory:') return undefined + this.readonlyReader ??= new ReadonlyQueryReader('assistant', this.databasePath, this.readonlyWorkerPath) + return this.readonlyReader + } + + /** Test hook for the worker crash and cancellation paths. */ + get readonlyWorkerForTest(): ReadonlyQueryReader | undefined { + return this.readonlyQueryReader() + } + private readExecutionStatsSnapshot( scope: ExecutionStatsInput, activeRequestIds: ReadonlySet @@ -9236,6 +9300,13 @@ export class AssistantDatabase { return readStoryGraph(this.requireDatabase(), name, input, projectId, signal) } + /** Story graph read on the readonly worker; falls back to the synchronous read. */ + readStoryGraphAsync(name: import('../../shared/story-graph-tools').StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Promise> { + this.requireDatabase() + return readWithFallback(this.readonlyQueryReader(), 'readStoryGraph', [name, input, projectId], signal, + () => this.readStoryGraph(name, input, projectId, signal)) + } + getSupervisionResult(id: string): { id: string; summary: string } | undefined { return this.requireDatabase().prepare('SELECT id, summary FROM supervision_results WHERE id = ?').get(id) as { id: string; summary: string } | undefined } diff --git a/src/main/assistant/story-graph-reader.test.ts b/src/main/assistant/story-graph-reader.test.ts index 7750d050..b55f6b9c 100644 --- a/src/main/assistant/story-graph-reader.test.ts +++ b/src/main/assistant/story-graph-reader.test.ts @@ -138,3 +138,25 @@ it('pages Unicode snapshots, reads explicit current versions, and reports deleti f.sql.prepare('UPDATE supervision_sources SET locator_json = ? WHERE id = ?').run(JSON.stringify({ detail: 'x'.repeat(100_001) }), source.id!) expect(() => f.read('story_graph_read_source', { source_reference_id: source.id })).toThrow('response_too_large') }) + +it('prefilters story lines by scope in SQL with the exact normalized match, and caps the scanned objects', async () => { + const f = await fixture() + f.save('Decision A', '2026-09-01T00:00:00Z') + const insert = f.sql.prepare('INSERT INTO story_lines (id, scope_json, title, created_at, updated_at) VALUES (?, ?, ?, ?, ?)') + const now = '2026-09-01T00:00:00.000Z' + // Unsorted / duplicate ids still normalize to the requested scope; supersets and other kinds do not. + insert.run('dup', JSON.stringify({ kind: 'projects', projectIds: [f.project.id, f.project.id] }), 'Decision dup', now, now) + insert.run('superset', JSON.stringify({ kind: 'projects', projectIds: [f.other.id, f.project.id] }), 'Decision superset', now, now) + insert.run('global', JSON.stringify({ kind: 'global' }), 'Decision global', now, now) + const lines = (scope?: unknown) => ((f.read('story_graph_search', { query: 'Decision', page_size: 50, object_types: ['story_line'], ...(scope ? { scope } : {}) }) as Page) + .items.map(item => item.object_ref.id)).sort() + const all = f.sql.prepare('SELECT id FROM story_lines').all().map(row => String(row.id)) + expect(lines()).toEqual(expect.arrayContaining(['dup'])) + expect(lines()).not.toContain('superset') + expect(lines()).not.toContain('global') + expect(lines({ kind: 'global' })).toEqual(['global']) + expect(lines({ kind: 'projects', projectIds: [f.project.id, f.other.id] })).toEqual(['superset']) + expect(all.length).toBeGreaterThanOrEqual(4) + expect(() => readStoryGraph(f.sql, 'story_graph_search', { query: 'Decision' }, f.project.id, undefined, 2)).toThrow('scan_limit_exceeded') + expect(readStoryGraph(f.sql, 'story_graph_search', { query: 'Decision' }, f.project.id)).toMatchObject({ page: { complete: true } }) +}) diff --git a/src/main/assistant/story-graph-reader.ts b/src/main/assistant/story-graph-reader.ts index 061d3c53..a8cc8f80 100644 --- a/src/main/assistant/story-graph-reader.ts +++ b/src/main/assistant/story-graph-reader.ts @@ -21,6 +21,8 @@ type Fact = { source_reference_ids: string[] data: Row } +/** Upper bound on projected objects per read (PERF-15). Normal scopes are far below it. */ +export const STORY_GRAPH_MAX_FACTS = 50_000 const digest = (value: unknown): string => createHash('sha256').update(JSON.stringify(value)).digest('hex') const cursorSchema = z.object({ key: z.string(), revision: z.string(), index: z.number().int().nonnegative(), offset: z.number().int().nonnegative() }).strict() const normalizeScope = (scope: HeartbeatScope): HeartbeatScope => scope.kind === 'global' ? scope : { kind: 'projects', projectIds: [...new Set(scope.projectIds)].sort() } @@ -48,7 +50,8 @@ export function queryTerms(query: string): string[] { } // This is a projection of existing storage, not a second graph or a historical state store. -export function readStoryGraph(db: DatabaseSync, name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record { +export function readStoryGraph(db: DatabaseSync, name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal, + maxFacts = STORY_GRAPH_MAX_FACTS): Record { signal?.throwIfAborted() if (input && typeof input === 'object' && ('as_of' in input || ('mode' in input && input.mode === 'as_of'))) throw new Error('unsupported_mode: as_of is not supported') const args = name === 'story_graph_search' ? storyGraphSearchSchema.parse(input) @@ -139,8 +142,18 @@ export function readStoryGraph(db: DatabaseSync, name: StoryGraphToolName, input } const facts: Fact[] = [] const results = new Map() - const stories = db.prepare('SELECT * FROM story_lines ORDER BY id').all().filter(row => sameScope(row.scope_json)) + // SQL prefilter on the scope kind and project membership (a superset of exact + // matches); the exact normalized comparison still runs in JS. + const stories = db.prepare(`SELECT * FROM story_lines + WHERE json_extract(scope_json, '$.kind') = ? + AND (? IS NULL OR NOT EXISTS (SELECT 1 FROM json_each(?) requested + WHERE requested.value NOT IN (SELECT value FROM json_each(story_lines.scope_json, '$.projectIds')))) + ORDER BY id`).all(scope.kind, scope.kind === 'projects' ? 1 : null, scope.kind === 'projects' ? JSON.stringify(scope.projectIds) : '[]') + .filter(row => sameScope(row.scope_json)) const add = (type: string, data: Row, result?: Row, sources?: string[]): void => { + // Bound the in-memory projection; a scope this large must be narrowed rather than truncated, + // since a partial projection would yield misleading revisions and cursors. + if (facts.length >= maxFacts) throw new Error(`scan_limit_exceeded: more than ${maxFacts} stored objects in scope; narrow the scope`) facts.push({ object_ref: { type, id: String(data.id) }, object_revision: digest(data), result_id: result ? String(result.id) : null, scope, event_time: type === 'event' ? timestamp(data.occurred_at) : null, review_generated_at: result ? timestamp(result.created_at) : null, diff --git a/src/main/index.ts b/src/main/index.ts index 1717e63a..0fa95913 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -933,11 +933,13 @@ if (hasSingleInstanceLock) { onExtensionStartupFailures: (extensionIds) => runtimeExtensionStore.markStartupFailed(extensionIds) }) + const readonlyQueryWorkerPath = join(app.getAppPath(), 'out/main/readonly-query-worker.js') const startupKnowledgeService = new KnowledgeService({ documentResults: documentParsingService.results, credentialCipher: secureCipher, databasePath: join(app.getPath('userData'), 'knowledge.sqlite'), managedRoot: join(app.getPath('userData'), 'knowledge'), + readonlyQueryWorkerPath, extractStructured: createModelGraphExtractor( settingsStore, fetch, @@ -1037,7 +1039,7 @@ if (hasSingleInstanceLock) { storyGraphService: { available: async ({ runtimeTarget }) => (await applicationSettingsStore.get()).heartbeatEnabled === true && (await capabilityService.getEnabledBuiltinMcpServerIds(runtimeTarget)).includes('story-graph'), - read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraph(name, input, projectId, signal) + read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraphAsync(name, input, projectId, signal) }, magicNotesDatabase: startupAssistantDatabase, configService: goodbuddyConfigService, @@ -1246,6 +1248,7 @@ if (hasSingleInstanceLock) { createConfiguredRuntime(initialResolvedSettings), initializeAssistant: () => { startupAssistantDatabase.initialize(defaultWorkspace) + startupAssistantDatabase.enableReadonlyWorker(readonlyQueryWorkerPath) imageGenerationService!.initialize() startupAssistantDatabase.ensureChannelProjects( defaultWorkspace, @@ -1504,7 +1507,7 @@ if (hasSingleInstanceLock) { storyGraphService: { available: async ({ runtimeTarget }) => (await applicationSettingsStore.get()).heartbeatEnabled === true && (await capabilityService.getEnabledBuiltinMcpServerIds(runtimeTarget)).includes('story-graph'), - read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraph(name, input, projectId, signal) + read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraphAsync(name, input, projectId, signal) }, magicNotesDatabase: startupAssistantDatabase, configService: goodbuddyConfigService, obsidianService, launchEnvironmentProvider: startupLocalToolEnvironmentService.launchEnvironmentProvider diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 30764108..264b4f08 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -6343,7 +6343,7 @@ export function registerIpcHandlers( registerHandler(ipcChannels.conversationsSearch, (event, input: unknown) => { assertTrustedSender(event, window) const { query } = conversationSearchRequestSchema.parse(input) - return assistantDatabase.searchConversations(query) + return assistantDatabase.searchConversationsAsync(query) }) registerHandler( diff --git a/src/main/knowledge/knowledge-database.ts b/src/main/knowledge/knowledge-database.ts index 64ae11a0..da8887aa 100644 --- a/src/main/knowledge/knowledge-database.ts +++ b/src/main/knowledge/knowledge-database.ts @@ -47,6 +47,7 @@ import { knowledgeRetrievalTerms } from './retrieval-text' import { normalizeEntityAlias } from './graph-extractor' +import { ReadonlyQueryReader, readWithFallback } from '../readonly-query-reader' import type { Chunk, ChunkEmbeddingInput, @@ -742,6 +743,7 @@ function mapKnowledgeTask(row: Row): KnowledgeTaskItem { export class KnowledgeDatabase { externalStore!: ExternalKnowledgeStore private database?: DatabaseSync + private readonlyReader?: ReadonlyQueryReader private readonly taskPrunedLibraries = new Set() private readonly taskPrunePending = new Set() @@ -779,7 +781,57 @@ export class KnowledgeDatabase { } } + /** + * Opens a read-only connection for the readonly query worker. No migration, + * no task recovery: the Main connection owns the schema and all writes. + */ + openReadOnly(): void { + if (this.database) { + throw new Error('Knowledge database is already open') + } + const database = new DatabaseSync(this.databasePath, { + readOnly: true, + timeout: 5_000 + }) + this.database = database + this.externalStore = new ExternalKnowledgeStore(database) + } + + /** + * Routes the async search methods through a read-only worker thread. In-memory + * databases cannot be shared with a worker and keep the synchronous path. + */ + enableReadonlyWorker(workerPath: string): void { + if (this.databasePath === ':memory:' || this.readonlyReader) return + this.readonlyReader = new ReadonlyQueryReader('knowledge', this.databasePath, workerPath) + } + + /** Test hook for the worker crash and cancellation paths. */ + get readonlyWorkerForTest(): ReadonlyQueryReader | undefined { + return this.readonlyReader + } + + searchAsync(options: SearchOptions, signal?: AbortSignal): Promise { + this.requireDatabase() + return readWithFallback(this.readonlyReader, 'search', [options], signal, + () => this.search(options)) + } + + hybridSearchAsync(options: HybridSearchOptions): Promise { + return this.hybridSearchWithDiagnosticsAsync(options).then(page => page.results) + } + + hybridSearchWithDiagnosticsAsync(options: HybridSearchOptions): Promise { + this.requireDatabase() + // AbortSignal is not cloneable; the reader forwards it as a shared cancel flag. + const { signal, ...cloneable } = options + return readWithFallback(this.readonlyReader, 'hybridSearchWithDiagnostics', [cloneable], signal, + () => this.hybridSearchWithDiagnostics(options)) + } + close(): void { + this.readonlyReader?.close() + this.readonlyReader = undefined if (!this.database) { return } diff --git a/src/main/knowledge/knowledge-service.ts b/src/main/knowledge/knowledge-service.ts index c453c597..c3975162 100644 --- a/src/main/knowledge/knowledge-service.ts +++ b/src/main/knowledge/knowledge-service.ts @@ -159,6 +159,8 @@ export type KnowledgeServiceOptions = { embeddingProvider?: EmbeddingProvider rerankProvider?: RerankProvider embeddingBatchSize?: number + /** Read-only query worker bundle; searches run off the Main thread when set. */ + readonlyQueryWorkerPath?: string parseDocument?: ( name: string, buffer: Buffer, @@ -263,9 +265,11 @@ export class KnowledgeService { EmbeddingIndexCoordinator >() private readonly lifecycleController = new AbortController() + private readonly readonlyQueryWorkerPath?: string constructor(options: KnowledgeServiceOptions) { this.database = new KnowledgeDatabase(options.databasePath) + this.readonlyQueryWorkerPath = options.readonlyQueryWorkerPath this.external = new ExternalKnowledgeService(this.database, options.credentialCipher, options.externalFetcher) this.managedRoot = resolve(options.managedRoot) this.documentResults = options.documentResults @@ -297,6 +301,9 @@ export class KnowledgeService { async initialize(): Promise { await mkdir(this.managedRoot, { recursive: true }) this.database.initialize() + if (this.readonlyQueryWorkerPath) { + this.database.enableReadonlyWorker(this.readonlyQueryWorkerPath) + } await this.reconcileDocumentResults() for (const library of this.database.listKnowledgeBases()) { for (const source of this.database.listSourcesForSnapshot(library.id)) { @@ -1352,7 +1359,7 @@ export class KnowledgeService { 100, settings.topK * settings.candidateMultiplier ) - const searchPage = this.database.hybridSearchWithDiagnostics({ + const searchPage = await this.database.hybridSearchWithDiagnosticsAsync({ knowledgeBaseId: library.id, query: input.query, limit: candidateLimit, @@ -1892,7 +1899,7 @@ export class KnowledgeService { const library = this.requireLibrary(knowledgeBaseId) const settings = library.retrievalSettings const vector = await this.embedQuery(query, signal) - return this.database.hybridSearch({ + return this.database.hybridSearchAsync({ knowledgeBaseId, query, limit: Math.min(limit, settings.topK), diff --git a/src/main/readonly-query-reader.test.ts b/src/main/readonly-query-reader.test.ts new file mode 100644 index 00000000..1c8ef7c3 --- /dev/null +++ b/src/main/readonly-query-reader.test.ts @@ -0,0 +1,241 @@ +// @vitest-environment node +import { createHash, randomUUID } from 'node:crypto' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join, resolve } from 'node:path' +import { build } from 'esbuild' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { AssistantDatabase } from './assistant/assistant-database' +import { KnowledgeDatabase } from './knowledge/knowledge-database' +import { ReadonlyQueryReader, ReadonlyWorkerUnavailableError, readWithFallback } from './readonly-query-reader' + +let directory: string +let workerPath: string +beforeAll(async () => { + directory = await mkdtemp(join(tmpdir(), 'goodbuddy-readonly-reader-')) + workerPath = join(directory, 'readonly-query-worker.cjs') + await build({ entryPoints: [resolve('src/main/readonly-query-worker.ts')], outfile: workerPath, + bundle: true, platform: 'node', format: 'cjs', logLevel: 'silent' }) +}, 60_000) +afterAll(async () => { await rm(directory, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }) }) + +const checksum = (value: string): string => createHash('sha256').update(value).digest('hex') + +function seedKnowledge(path: string): { database: KnowledgeDatabase; libraryId: string; query: number[] } { + const database = new KnowledgeDatabase(path) + database.initialize() + const library = database.createKnowledgeBase({ name: 'Worker parity', storageMode: 'reference' }) + let seed = 7 + const random = (): number => ((seed = (seed * 1664525 + 1013904223) >>> 0) / 4294967296) + const words = 'lighthouse harbor budget latency 发布 审批 流程 跨平台 桌面 vector'.split(' ') + const sqlite = (database as unknown as { database: import('node:sqlite').DatabaseSync }).database + for (let documentIndex = 0; documentIndex < 6; documentIndex += 1) { + const source = database.upsertSource({ knowledgeBaseId: library.id, type: 'file', + location: `C:\\parity\\doc-${documentIndex}.md`, displayName: `doc-${documentIndex}.md`, status: 'ready' }) + const chunks = Array.from({ length: 30 }, (_, ordinal) => ({ id: `d${documentIndex}-c${ordinal}`, ordinal, + content: Array.from({ length: 24 }, () => words[Math.floor(random() * words.length)]).join(' '), location: `line ${ordinal}` })) + const document = database.upsertDocument({ knowledgeBaseId: library.id, sourceId: source.id, + externalId: `doc-${documentIndex}`, title: `doc-${documentIndex}`, sourceLocation: source.location }, chunks) + const stored = sqlite.prepare("SELECT id, index_content FROM chunks WHERE document_id = ? AND enabled = 1 AND role <> 'parent'") + .all(document.id) as Array<{ id: string; index_content: string }> + database.replaceDocumentEmbeddings(document.id, 'p', 'm', stored.map(row => ({ chunkId: row.id, + // Duplicate vectors exercise the tie-break on chunk id. + contentChecksum: checksum(row.index_content), vector: row.id.endsWith('-c3') ? [1, 0, 0, 0] : [random() - 0.5, random() - 0.5, random() - 0.5, random() - 0.5] }))) + } + return { database, libraryId: library.id, query: [0.9, 0.1, -0.2, 0.05] } +} + +describe('knowledge search worker', () => { + it('returns exactly the synchronous results, order and scores', async () => { + const { database, libraryId, query } = seedKnowledge(join(directory, `knowledge-${randomUUID()}.sqlite`)) + try { + const cases = [ + { knowledgeBaseId: libraryId, query: 'lighthouse budget', limit: 40, provider: 'p', model: 'm', vector: query, graphEnabled: false, candidateMultiplier: 1 }, + { knowledgeBaseId: libraryId, query: '发布审批流程', limit: 20, provider: 'p', model: 'm', vector: query, minimumVectorSimilarity: 0.1 }, + { knowledgeBaseId: libraryId, query: 'harbor', limit: 5, ftsWeight: 0 , provider: 'p', model: 'm', vector: query }, + { knowledgeBaseId: libraryId, query: 'latency', limit: 10 } + ] + const expected = cases.map(options => database.hybridSearchWithDiagnostics(options)) + const expectedFts = database.search({ knowledgeBaseId: libraryId, query: '跨平台 vector', limit: 30 }) + database.enableReadonlyWorker(workerPath) + const actual = await Promise.all(cases.map(options => database.hybridSearchWithDiagnosticsAsync(options))) + expect(database.readonlyWorkerForTest?.pendingCount).toBe(0) + expect(actual).toEqual(expected) + expect(actual[0]!.results.length).toBeGreaterThan(10) + expect(actual[0]!.vectorScannedCount).toBe(180) + expect(await database.searchAsync({ knowledgeBaseId: libraryId, query: '跨平台 vector', limit: 30 })).toEqual(expectedFts) + // Validation errors keep their type and message across the thread boundary. + await expect(database.hybridSearchWithDiagnosticsAsync({ ...cases[0]!, limit: 500 })).rejects.toThrow(RangeError) + // A write committed on Main is visible to the next worker read. + database.updateKnowledgeBase(libraryId, { name: 'Renamed' }) + const source = database.upsertSource({ knowledgeBaseId: libraryId, type: 'file', location: 'C:\\parity\\new.md', displayName: 'new.md', status: 'ready' }) + database.upsertDocument({ knowledgeBaseId: libraryId, sourceId: source.id, externalId: 'new', title: 'new', sourceLocation: source.location }, + [{ id: 'fresh-chunk', ordinal: 0, content: 'zebracorn unique', location: 'l' }]) + expect((await database.searchAsync({ knowledgeBaseId: libraryId, query: 'zebracorn' })).map(item => item.chunk.id)).toEqual(['fresh-chunk']) + } finally { database.close() } + }, 60_000) + + it('cancels through the AbortSignal and keeps serving later requests', async () => { + const { database, libraryId, query } = seedKnowledge(join(directory, `knowledge-${randomUUID()}.sqlite`)) + try { + database.enableReadonlyWorker(workerPath) + const options = { knowledgeBaseId: libraryId, query: 'harbor', provider: 'p', model: 'm', vector: query } + const preAborted = new AbortController() + preAborted.abort(new Error('cancelled before')) + await expect(database.hybridSearchWithDiagnosticsAsync({ ...options, signal: preAborted.signal })).rejects.toThrow('cancelled before') + const controller = new AbortController() + const pending = database.hybridSearchWithDiagnosticsAsync({ ...options, signal: controller.signal }) + controller.abort(new Error('cancelled during')) + await expect(pending).rejects.toThrow('cancelled during') + expect(database.readonlyWorkerForTest?.pendingCount).toBe(0) + expect(await database.hybridSearchWithDiagnosticsAsync(options)).toEqual(database.hybridSearchWithDiagnostics(options)) + } finally { database.close() } + }, 60_000) + + it('falls back to the synchronous path when the worker crashes and restarts later', async () => { + const { database, libraryId, query } = seedKnowledge(join(directory, `knowledge-${randomUUID()}.sqlite`)) + try { + database.enableReadonlyWorker(workerPath) + const reader = database.readonlyWorkerForTest! + const options = { knowledgeBaseId: libraryId, query: 'budget', provider: 'p', model: 'm', vector: query } + const expected = database.hybridSearchWithDiagnostics(options) + expect(await database.hybridSearchWithDiagnosticsAsync(options)).toEqual(expected) + // In-flight request when the worker dies: rejected internally, served synchronously. + const inFlight = database.hybridSearchWithDiagnosticsAsync(options) + await reader.terminateWorkerForTest() + expect(await inFlight).toEqual(expected) + expect(reader.available).toBe(false) + // During the backoff the synchronous path serves requests. + expect(await database.hybridSearchWithDiagnosticsAsync(options)).toEqual(expected) + reader.resetBackoffForTest() + expect(reader.available).toBe(true) + expect(await reader.call('hybridSearchWithDiagnostics', [options])).toEqual(expected) + } finally { database.close() } + }, 60_000) + + it('keeps the synchronous path for in-memory databases and startup failures', async () => { + const memory = new KnowledgeDatabase(':memory:') + memory.initialize() + try { + memory.enableReadonlyWorker(workerPath) + expect(memory.readonlyWorkerForTest).toBeUndefined() + const library = memory.createKnowledgeBase({ name: 'Memory', storageMode: 'reference' }) + expect(await memory.searchAsync({ knowledgeBaseId: library.id, query: 'x' })).toEqual([]) + } finally { memory.close() } + const missing = new ReadonlyQueryReader('knowledge', join(directory, 'missing.sqlite'), workerPath) + try { + await expect(missing.call('search', [{}])).rejects.toBeInstanceOf(ReadonlyWorkerUnavailableError) + expect(await readWithFallback(missing, 'search', [{}], undefined, () => 'sync')).toBe('sync') + } finally { missing.close() } + }, 60_000) +}) + +/** The previous per-conversation JS implementation, kept as the semantic reference. */ +function referenceSearch(database: AssistantDatabase, query: string): string[] { + const normalized = query.trim().toLocaleLowerCase() + if (!normalized) return [] + const sqlite = (database as unknown as { database: import('node:sqlite').DatabaseSync }).database + const rows = sqlite.prepare(`SELECT id, title FROM conversations WHERE status = 'active' ORDER BY pinned DESC, updated_at DESC`).all() as Array<{ id: string; title: string }> + const messages = sqlite.prepare('SELECT content FROM messages WHERE conversation_id = ?') + return rows.filter(row => row.title.toLocaleLowerCase().includes(normalized) || + (messages.all(row.id) as Array<{ content: string }>).some(message => message.content.toLocaleLowerCase().includes(normalized))) + .map(row => row.id) +} + +describe('assistant readonly worker', () => { + it('matches the reference conversation search, including Unicode case folding and CJK, on Main and in the worker', async () => { + const root = await mkdtemp(join(directory, 'assistant-')) + const database = new AssistantDatabase(join(root, 'assistant.sqlite')) + database.initialize(root) + try { + const texts = ['ÄPFEL und Birnen', 'straße STRASSE', 'İstanbul ısı', 'ΣΊΣΥΦΟΣ', '跨平台桌面应用发布审批', '100%_literal\\x', 'DŽemal Dž', 'plain body', ''] + const conversations = texts.map((text, index) => ({ + id: randomUUID(), title: index % 3 === 0 ? `Title ${text}` : `Conversation ${index}`, updatedAt: index + 1, pinned: index === 4, + messages: [{ id: randomUUID(), role: 'user' as const, state: 'complete' as const, content: text, createdAt: index * 10 }, + { id: randomUUID(), role: 'assistant' as const, state: 'complete' as const, content: `reply ${index}`, createdAt: index * 10 + 1 }] + })) + database.replaceConversations(conversations) + const queries = ['äpfel', 'ÄPFEL', 'Äp', 'STRASSE', 'straße', 'İstanbul', 'i̇stanbul', 'ISI', 'σίσυφος', 'ΣΊΣΥΦΟΣ', '平台桌面', '审批', + '%_', '100%', '_', '\\x', 'dž', 'DŽemal', 'reply 1', 'REPLY', 'title', 'conversation 7', ' plain ', 'missing', ' '] + const expected = queries.map(query => referenceSearch(database, query)) + expect(expected.some(ids => ids.length > 1)).toBe(true) + expect(queries.map(query => database.searchConversations(query))).toEqual(expected) + database.enableReadonlyWorker(workerPath) + expect(await Promise.all(queries.map(query => database.searchConversationsAsync(query)))).toEqual(expected) + expect(database.readonlyWorkerForTest?.pendingCount).toBe(0) + } finally { database.close() } + }, 60_000) + + it('lists conversations with batched queries matching per-conversation reads', async () => { + const root = await mkdtemp(join(directory, 'assistant-')) + const database = new AssistantDatabase(join(root, 'assistant.sqlite')) + database.initialize(root) + try { + const message = (content: string, createdAt: number, state: 'complete' | 'streaming' = 'complete', role: 'user' | 'assistant' = 'assistant') => + ({ id: randomUUID(), role, state, content, createdAt }) + const conversations = [ + { id: randomUUID(), title: 'Empty', updatedAt: 1, messages: [] }, + { id: randomUUID(), title: 'Streaming', updatedAt: 2, messages: [message('q', 1, 'complete', 'user'), message('partial', 2, 'streaming')] }, + { id: randomUUID(), title: 'Done', updatedAt: 3, messages: [message('q', 5, 'complete', 'user'), message('a', 9), message('b', 7)] }, + { id: randomUUID(), title: 'Single', updatedAt: 0, messages: [message('x', 3)] } + ] + database.replaceConversations(conversations) + const full = database.listConversations() + expect(full).toHaveLength(4) + for (const item of full) expect(item).toEqual(database.getConversation(item.id)) + const summaries = database.listConversationSummaries([conversations[2]!.id]) + const byId = new Map(summaries.map(item => [item.id, item])) + expect(byId.get(conversations[2]!.id)).toEqual(database.getConversation(conversations[2]!.id)) + // A streaming conversation always carries its messages. + expect(byId.get(conversations[1]!.id)).toEqual(database.getConversation(conversations[1]!.id)) + expect(byId.get(conversations[0]!.id)?.messageSummary).toEqual({ count: 0, firstRole: undefined, latestMessageAt: undefined }) + expect(byId.get(conversations[3]!.id)?.messageSummary).toEqual({ count: 1, firstRole: 'assistant', latestMessageAt: 3 }) + expect(byId.get(conversations[3]!.id)?.messages).toEqual([]) + } finally { database.close() } + }, 60_000) + + it('serves story graph reads from the worker with the synchronous result and errors', async () => { + const root = await mkdtemp(join(directory, 'assistant-')) + const database = new AssistantDatabase(join(root, 'assistant.sqlite')) + database.initialize(root) + try { + const project = database.listProjects()[0]! + const other = database.createProject({ name: 'Other', description: '', rootPath: root, defaultWorkMode: 'ask' }) + const conversationId = randomUUID(), messageId = randomUUID() + const body = 'Choose A, then revise A to B because of measured latency.' + database.saveLocalConversations([{ header: { id: conversationId, projectId: project.id, title: 'Decision', updatedAt: 1 }, + messages: [{ id: messageId, role: 'user', content: body, createdAt: Date.parse('2026-09-01T00:00:00Z'), state: 'complete' }] }]) + const request = { trigger: 'manual' as const, scope: { kind: 'projects' as const, projectIds: [project.id] }, + timeRange: { from: '2026-09-01T00:00:00Z', to: '2026-09-30T00:00:00Z' } } + database.saveSupervisionResult({ request, candidates: database.listSupervisionCandidates(request), + evidence: [{ id: 'quote', sourceType: 'conversation', sourceId: conversationId, title: 'Decision evidence', content: body, + occurredAt: '2026-09-01T00:00:00Z', locator: { source: `message:${messageId}`, projectId: project.id, conversationId, messageId, start: 0, end: body.length, length: body.length } }], + output: { summary: 'Decision A', changeDigest: '', openItems: [], relations: [], + entities: [{ id: 'choice', label: 'Decision', description: 'Decision A', sourceReferenceIds: ['quote'] }], + events: [{ title: 'Decision A', description: 'Decision A', occurredAt: '2026-09-01T00:00:00Z', eventType: 'decision', entityIds: ['choice'], sourceReferenceIds: ['quote'] }], + entityChanges: [{ entityId: 'choice', changeType: 'proposed', description: 'Decision A', sourceReferenceIds: ['quote'] }] } }) + const withoutReadAt = (value: Record): Record => ({ ...value, read_at: undefined }) + const inputs = [ + { query: 'Decision', page_size: 50 }, + { query: 'decision latency', page_size: 2 }, + { query: 'Decision', page_size: 50, scope: { kind: 'global' } } + ] + const expected = inputs.map(input => withoutReadAt(database.readStoryGraph('story_graph_search', input, project.id))) + expect((expected[0]!.items as unknown[]).length).toBeGreaterThan(2) + database.enableReadonlyWorker(workerPath) + const actual = await Promise.all(inputs.map(input => database.readStoryGraphAsync('story_graph_search', input, project.id))) + expect(actual.map(withoutReadAt)).toEqual(expected) + const entity = (expected[0]!.items as Array<{ object_ref: { type: string; id: string } }>).find(item => item.object_ref.type === 'entity')! + const contextInput = { object_ref: entity.object_ref, mode: 'timeline', page_size: 5 } + expect(withoutReadAt(await database.readStoryGraphAsync('story_graph_get_context', contextInput, project.id))) + .toEqual(withoutReadAt(database.readStoryGraph('story_graph_get_context', contextInput, project.id))) + await expect(database.readStoryGraphAsync('story_graph_get_context', contextInput, other.id)).rejects.toThrow('scope_mismatch') + await expect(database.readStoryGraphAsync('story_graph_search', { query: 'x' })).rejects.toThrow('scope_required') + // Schema validation errors stay ZodErrors so MCP callers classify them unchanged. + await expect(database.readStoryGraphAsync('story_graph_search', { query: 1 }, project.id)).rejects.toMatchObject({ name: 'ZodError' }) + const aborted = new AbortController(); aborted.abort(new Error('stop')) + await expect(database.readStoryGraphAsync('story_graph_search', inputs[0], project.id, aborted.signal)).rejects.toThrow('stop') + expect(database.readonlyWorkerForTest?.pendingCount).toBe(0) + } finally { database.close() } + }, 60_000) +}) diff --git a/src/main/readonly-query-reader.ts b/src/main/readonly-query-reader.ts new file mode 100644 index 00000000..8887a5dc --- /dev/null +++ b/src/main/readonly-query-reader.ts @@ -0,0 +1,204 @@ +import { Worker } from 'node:worker_threads' +import { z } from 'zod' + +// PERF-15/16 first slice: hot read paths run in a read-only worker per database +// file so a long scan does not block the Main event loop. The worker opens its +// own read-only connection (WAL), so it observes committed data only. + +export type ReadonlyQueryKind = 'assistant' | 'knowledge' + +/** Worker infrastructure failure; callers fall back to the synchronous path. */ +export class ReadonlyWorkerUnavailableError extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options) + this.name = 'ReadonlyWorkerUnavailableError' + } +} + +export type ReadonlyQueryRequest = { + type: 'query' + id: number + op: string + args: unknown[] + /** Int32Array(1) cancel flag, polled by the worker through `signal.throwIfAborted()`. */ + cancel: SharedArrayBuffer +} + +export type ReadonlyQueryResponse = + | { id: number; result: unknown } + | { id: number; error: { name: string; message: string; issues?: unknown } } + | { id: number; ready: true } + +type Pending = { + resolve: (value: unknown) => void + reject: (error: Error) => void + cleanup: () => void +} + +const MAX_PENDING = 32 +const RESTART_BACKOFF_MS = 30_000 + +export function deserializeWorkerError(error: { name: string; message: string; issues?: unknown }): Error { + if (error.name === 'ZodError' && Array.isArray(error.issues)) { + return new z.ZodError(error.issues as z.core.$ZodIssue[]) + } + const Constructor = error.name === 'RangeError' ? RangeError + : error.name === 'TypeError' ? TypeError : Error + const result = new Constructor(error.message) + if (result.name !== error.name) result.name = error.name + return result +} + +export class ReadonlyQueryReader { + private worker?: Worker + private nextId = 0 + private disabledUntil = 0 + private closed = false + private readonly pending = new Map() + + constructor( + private readonly kind: ReadonlyQueryKind, + private readonly databasePath: string, + private readonly workerPath: string, + private readonly now: () => number = Date.now + ) {} + + /** False while backing off after a crash or startup failure. */ + get available(): boolean { + return !this.closed && this.now() >= this.disabledUntil && this.pending.size < MAX_PENDING + } + + get pendingCount(): number { + return this.pending.size + } + + call(op: string, args: unknown[], signal?: AbortSignal): Promise { + if (this.closed) return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query reader closed')) + if (signal?.aborted) return Promise.reject(signal.reason as Error) + if (this.now() < this.disabledUntil) { + return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query worker is backing off')) + } + // Bound outstanding work; the caller runs the query synchronously instead. + if (this.pending.size >= MAX_PENDING) { + return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query worker is busy')) + } + let worker: Worker + try { + worker = this.ensureWorker() + } catch (error) { + this.disabledUntil = this.now() + RESTART_BACKOFF_MS + return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query worker failed to start', { cause: error })) + } + const id = ++this.nextId + const cancel = new SharedArrayBuffer(4) + const flag = new Int32Array(cancel) + return new Promise((resolve, reject) => { + const onAbort = (): void => { + Atomics.store(flag, 0, 1) + const request = this.pending.get(id) + if (!request) return + this.pending.delete(id) + request.cleanup() + if (this.pending.size === 0) worker.unref() + reject(signal!.reason as Error) + } + signal?.addEventListener('abort', onAbort, { once: true }) + this.pending.set(id, { + resolve: resolve as (value: unknown) => void, + reject, + cleanup: () => signal?.removeEventListener('abort', onAbort) + }) + // Keep the process alive only while a request is outstanding. + worker.ref() + try { + worker.postMessage({ type: 'query', id, op, args, cancel } satisfies ReadonlyQueryRequest) + } catch (error) { + this.pending.delete(id) + signal?.removeEventListener('abort', onAbort) + if (this.pending.size === 0) worker.unref() + // DataCloneError etc. are caller errors, but stay safe and let the caller fall back. + reject(new ReadonlyWorkerUnavailableError('Readonly query could not be posted', { cause: error })) + } + }) + } + + private ensureWorker(): Worker { + if (this.worker) return this.worker + const worker = new Worker(this.workerPath, { + workerData: { kind: this.kind, databasePath: this.databasePath } + }) + this.worker = worker + worker.on('message', (message: ReadonlyQueryResponse) => { + if ('ready' in message) return + const request = this.pending.get(message.id) + if (!request) return + this.pending.delete(message.id) + request.cleanup() + if (this.pending.size === 0) worker.unref() + if ('error' in message) request.reject(deserializeWorkerError(message.error)) + else request.resolve(message.result) + }) + const failed = (error: Error): void => { + if (this.worker !== worker) return + this.worker = undefined + // Restart lazily on a later request, after a short backoff. + this.disabledUntil = this.now() + RESTART_BACKOFF_MS + const pending = [...this.pending.values()] + this.pending.clear() + for (const request of pending) { + request.cleanup() + request.reject(new ReadonlyWorkerUnavailableError(error.message, { cause: error })) + } + void worker.terminate() + } + worker.once('error', failed) + worker.once('exit', (code) => failed(new Error(`Readonly query worker exited (${code})`))) + worker.unref() + return worker + } + + /** Test hook: simulate a crash by terminating the worker. */ + async terminateWorkerForTest(): Promise { + await this.worker?.terminate() + } + + /** Test hook: end the post-crash backoff. */ + resetBackoffForTest(): void { + this.disabledUntil = 0 + } + + close(): void { + this.closed = true + const worker = this.worker + this.worker = undefined + const pending = [...this.pending.values()] + this.pending.clear() + for (const request of pending) { + request.cleanup() + request.reject(new ReadonlyWorkerUnavailableError('Readonly query reader closed')) + } + if (worker) void worker.terminate() + } +} + +/** + * Runs `op` in the worker when available, otherwise synchronously. Worker + * infrastructure failures fall back to the synchronous path; query errors and + * aborts propagate unchanged. + */ +export async function readWithFallback( + reader: ReadonlyQueryReader | undefined, + op: string, + args: unknown[], + signal: AbortSignal | undefined, + sync: () => T +): Promise { + signal?.throwIfAborted() + if (!reader?.available) return sync() + try { + return await reader.call(op, args, signal) + } catch (error) { + if (error instanceof ReadonlyWorkerUnavailableError && !signal?.aborted) return sync() + throw error + } +} diff --git a/src/main/readonly-query-worker.ts b/src/main/readonly-query-worker.ts new file mode 100644 index 00000000..ef9fbfe2 --- /dev/null +++ b/src/main/readonly-query-worker.ts @@ -0,0 +1,68 @@ +import { parentPort, workerData } from 'node:worker_threads' +import { z } from 'zod' +import { AssistantDatabase } from './assistant/assistant-database' +import { KnowledgeDatabase } from './knowledge/knowledge-database' +import type { ReadonlyQueryKind, ReadonlyQueryRequest } from './readonly-query-reader' + +// Read-only query worker (PERF-15/16 first slice). One worker per database file. +// It reuses the production classes on a read-only connection, so results are +// produced by exactly the same code as the synchronous path. + +const { kind, databasePath } = workerData as { kind: ReadonlyQueryKind; databasePath: string } + +type Handler = (args: unknown[], signal: AbortSignal) => unknown +let handlers: Record +let close: () => void + +if (kind === 'knowledge') { + const database = new KnowledgeDatabase(databasePath) + database.openReadOnly() + close = () => database.close() + handlers = { + search: ([options]) => database.search(options as Parameters[0]), + hybridSearchWithDiagnostics: ([options], signal) => database.hybridSearchWithDiagnostics({ + ...(options as Parameters[0]), signal + }) + } +} else { + const database = new AssistantDatabase(databasePath) + database.openReadOnly() + close = () => database.close() + handlers = { + readStoryGraph: ([name, input, projectId], signal) => database.readStoryGraph( + name as Parameters[0], input, projectId as string | undefined, signal), + searchConversations: ([query]) => database.searchConversations(query as string) + } +} + +/** A signal backed by the shared cancel flag; queries poll it via throwIfAborted(). */ +function sharedSignal(cancel: SharedArrayBuffer): AbortSignal { + const flag = new Int32Array(cancel) + const reason = (): Error => Object.assign(new Error('This operation was aborted'), { name: 'AbortError' }) + const signal = new AbortController().signal + Object.defineProperties(signal, { + aborted: { get: () => Atomics.load(flag, 0) === 1 }, + reason: { get: () => (Atomics.load(flag, 0) === 1 ? reason() : undefined) }, + throwIfAborted: { value: () => { if (Atomics.load(flag, 0) === 1) throw reason() } } + }) + return signal +} + +function serializeError(error: unknown): { name: string; message: string; issues?: unknown } { + if (error instanceof z.ZodError) return { name: 'ZodError', message: error.message, issues: error.issues } + if (error instanceof Error) return { name: error.name, message: error.message } + return { name: 'Error', message: String(error) } +} + +parentPort!.on('message', (message: ReadonlyQueryRequest) => { + try { + const handler = Object.hasOwn(handlers, message.op) ? handlers[message.op] : undefined + if (!handler) throw new Error(`Unknown readonly query: ${message.op}`) + const signal = sharedSignal(message.cancel) + signal.throwIfAborted() + parentPort!.postMessage({ id: message.id, result: handler(message.args, signal) }) + } catch (error) { + parentPort!.postMessage({ id: message.id, error: serializeError(error) }) + } +}) +parentPort!.on('close', () => close())