From 3c31f0b4997bcd42919239018a0315b9541c53dc Mon Sep 17 00:00:00 2001 From: mesalogo Date: Sat, 3 Oct 2026 05:53:10 +0800 Subject: [PATCH] perf: read conversation lists in the readonly worker and diff activity history saves conversations:list-summaries and conversations:get run on the readonly query worker. Activity history replace validates and serializes only changed records; a new activity-history:update IPC writes one record. Adds npm run perf:main. --- build/run-main-perf.cjs | 251 ++++++++++++++++++ package.json | 1 + .../assistant/activity-history-plan.test.ts | 197 ++++++++++++++ src/main/assistant/activity-history-plan.ts | 203 ++++++++++++++ src/main/assistant/assistant-database.ts | 177 ++++++++---- src/main/ipc.test.ts | 7 +- src/main/ipc.ts | 24 +- src/main/readonly-query-reader.test.ts | 53 ++++ src/main/readonly-query-worker.ts | 7 +- src/preload/index.ts | 4 + src/shared/assistant-contracts.ts | 32 +++ src/shared/contracts.ts | 7 + src/shared/ipc-channels.ts | 1 + 13 files changed, 908 insertions(+), 56 deletions(-) create mode 100644 build/run-main-perf.cjs create mode 100644 src/main/assistant/activity-history-plan.test.ts create mode 100644 src/main/assistant/activity-history-plan.ts diff --git a/build/run-main-perf.cjs b/build/run-main-perf.cjs new file mode 100644 index 00000000..ac2934ce --- /dev/null +++ b/build/run-main-perf.cjs @@ -0,0 +1,251 @@ +// Main-process database benchmark (PERF-15). Bundles the production +// AssistantDatabase and readonly worker with esbuild, seeds a temporary +// assistant database (default 1,000 conversations / 20,000 messages and 5,000 +// activity records) and measures, for the IPC paths the renderer hits often: +// - per-call time +// - Main event-loop delay while the calls run back to back +// for the previous synchronous path ("sync") and the current path ("after"). +// conversations:list-summaries sync listConversationSummaries vs worker +// conversations:get sync getConversation vs worker +// activity-history:replace legacy replace (zod parse + BEGIN IMMEDIATE + +// JSON.stringify of every record) vs current +// activity-history:update one changed record via the incremental API +// Env: GB_MPERF_CONVERSATIONS, GB_MPERF_MESSAGES, GB_MPERF_ACTIVITY, +// GB_MPERF_CALLS, GB_MPERF_OUTPUT (JSON), GB_MPERF_NODE=1 (skip Electron). +const { buildSync } = require('esbuild') +const { mkdtempSync, rmSync, writeFileSync, mkdirSync } = require('node:fs') +const { tmpdir } = require('node:os') +const { join, resolve } = require('node:path') +const { randomUUID } = require('node:crypto') +const { monitorEventLoopDelay, performance } = require('node:perf_hooks') + +const root = resolve(__dirname, '..') + +if (process.env.GB_MPERF_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 conversationCount = Number(process.env.GB_MPERF_CONVERSATIONS || 1000) +const messageCount = Number(process.env.GB_MPERF_MESSAGES || 20000) +const activityCount = Number(process.env.GB_MPERF_ACTIVITY || 5000) +const calls = Number(process.env.GB_MPERF_CALLS || 15) + +const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-main-perf-')) +const bundle = join(directory, 'assistant-database.cjs') +const workerBundle = join(directory, 'readonly-query-worker.cjs') +const schemaBundle = join(directory, 'assistant-contracts.cjs') +for (const [entry, outfile] of [ + ['src/main/assistant/assistant-database.ts', bundle], + ['src/main/readonly-query-worker.ts', workerBundle], + ['src/shared/assistant-contracts.ts', schemaBundle] +]) { + buildSync({ + entryPoints: [join(root, entry)], bundle: true, platform: 'node', format: 'cjs', + target: 'node22', outfile, logLevel: 'error', external: ['electron'] + }) +} +const { AssistantDatabase } = require(bundle) +const { activityHistorySnapshotSchema } = require(schemaBundle) + +let seed = 42 +const random = () => ((seed = (seed * 1664525 + 1013904223) >>> 0) / 4294967296) +const words = 'lighthouse harbor policy engine renderer stream vector index budget latency schedule archive network storage contract invoice release window memory review 发布 审批 流程'.split(' ') +const sentence = n => Array.from({ length: n }, () => words[Math.floor(random() * words.length)]).join(' ') +const round = value => Math.round(value * 10) / 10 +const percentile = (values, p) => { + const sorted = [...values].sort((a, b) => a - b) + return sorted[Math.min(sorted.length - 1, Math.max(0, Math.ceil((p / 100) * sorted.length) - 1))] +} + +/** + * Runs `body` `calls` times (after one warm-up) and reports the per-call wall + * time and the Main event-loop delay histogram (p99/max) over the run. + */ +async function measure(body) { + await body(-1) + await new Promise(resolve => setTimeout(resolve, 50)) + // monitorEventLoopDelay is bound to the ~15 ms Windows timer granularity, so + // the longest block is measured as the largest gap between setImmediate + // ticks (each tick is one event-loop turn). + const histogram = monitorEventLoopDelay({ resolution: 1 }) + histogram.enable() + let probing = true + let last = performance.now() + let maxGap = 0 + const tick = () => { + const now = performance.now() + maxGap = Math.max(maxGap, now - last) + last = now + if (probing) setImmediate(tick) + } + setImmediate(tick) + const samples = [] + for (let index = 0; index < calls; index += 1) { + const started = performance.now() + await body(index) + samples.push(performance.now() - started) + // Yield like the IPC loop does between messages. + await new Promise(resolve => setImmediate(resolve)) + } + await new Promise(resolve => setTimeout(resolve, 20)) + probing = false + histogram.disable() + return { + callP50: round(percentile(samples, 50)), callP95: round(percentile(samples, 95)), + loopP99: round(histogram.percentile(99) / 1e6), loopMax: round(maxGap) + } +} + +/** The pre-PERF-15 replace implementation, kept as the "before" reference. */ +function legacyReplace(raw, cache, input) { + const snapshot = activityHistorySnapshotSchema.parse(input) + raw.exec('BEGIN IMMEDIATE') + try { + const { data_version: dataVersion } = raw.prepare('PRAGMA data_version').get() + let previous = cache.value?.dataVersion === dataVersion ? cache.value.records : undefined + if (!previous) { + previous = new Map(raw.prepare('SELECT record_key, record_json FROM activity_history_records').all() + .map(row => [row.record_key, row.record_json])) + } + const records = new Map() + const occurrences = new Map() + const order = [] + const upsert = raw.prepare(`INSERT INTO activity_history_records (record_key, record_json) VALUES (?, ?) + ON CONFLICT(record_key) DO UPDATE SET record_json = excluded.record_json`) + for (const record of snapshot.records) { + const occurrence = occurrences.get(record.id) ?? 0 + occurrences.set(record.id, occurrence + 1) + const key = JSON.stringify([record.id, occurrence]) + const json = JSON.stringify(record) + order.push(key) + if (previous.get(key) !== json) upsert.run(key, json) + records.set(key, json) + } + const remove = raw.prepare('DELETE FROM activity_history_records WHERE record_key = ?') + for (const key of previous.keys()) if (!records.has(key)) remove.run(key) + const orderJson = JSON.stringify(order) + const incomplete = Number(snapshot.legacyHistoryMayBeIncomplete) + raw.prepare(`UPDATE activity_history SET record_order_json = ?, legacy_history_may_be_incomplete = ? + WHERE singleton = 1 AND (record_order_json != ? OR legacy_history_may_be_incomplete != ?)`) + .run(orderJson, incomplete, orderJson, incomplete) + raw.exec('COMMIT') + cache.value = { records, dataVersion } + } catch (error) { + raw.exec('ROLLBACK') + throw error + } +} + +async function main() { + console.log(`[mperf] runtime: node ${process.version}${process.versions.electron ? ` (electron ${process.versions.electron})` : ''}`) + const workspace = join(directory, 'workspace') + mkdirSync(workspace) + const path = join(directory, 'assistant.sqlite') + const seeding = new AssistantDatabase(path) + seeding.initialize(workspace) + const seedStarted = performance.now() + const projectId = seeding.listProjects()[0].id + // Message counts follow a skewed distribution; the first conversation is the largest. + const weights = Array.from({ length: conversationCount }, (_, index) => 1 / (1 + index * 0.01)) + const weightSum = weights.reduce((sum, value) => sum + value, 0) + const counts = weights.map(weight => Math.max(1, Math.round(messageCount * weight / weightSum))) + const ids = counts.map(() => randomUUID()) + for (let start = 0; start < conversationCount; start += 50) { + seeding.saveLocalConversations(ids.slice(start, start + 50).map((id, offset) => { + const index = start + offset + return { + header: { id, projectId, title: `Conversation ${index} ${sentence(3)}`, updatedAt: 1_700_000_000_000 + index }, + messages: Array.from({ length: counts[index] }, (_, at) => ({ + id: randomUUID(), role: at % 2 === 0 ? 'user' : 'assistant', state: 'complete', + content: sentence(at % 2 === 0 ? 30 : 120), createdAt: 1_700_000_000_000 + index * 1000 + at + })) + } + })) + } + const records = Array.from({ length: activityCount }, (_, index) => ({ + id: randomUUID(), conversationId: ids[index % conversationCount], requestId: randomUUID(), + ...(index % 2 === 0 ? { callId: `call-${index}` } : {}), + scope: index % 3 === 0 ? { kind: 'project', projectId, projectName: 'Perf' } : { kind: 'global' }, + kind: index % 5 === 0 ? 'request' : 'tool', title: `Tool ${index}`, detail: sentence(40), + status: 'completed', createdAt: 1_700_000_000_000 + index + })) + seeding.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: false }) + const seedSeconds = round((performance.now() - seedStarted) / 1000) + seeding.close() + + const database = new AssistantDatabase(path) + database.initialize(workspace) + const raw = database.database + const totalMessages = counts.reduce((sum, value) => sum + value, 0) + const largest = ids[0] + const typical = ids[Math.floor(conversationCount / 2)] + // The renderer retains the current conversation's history. + const detailIds = [ids[1]] + const results = { conversations: conversationCount, messages: totalMessages, activityRecords: activityCount, + largestConversationMessages: counts[0], seedSeconds } + + results.listSummariesSync = await measure(async () => database.listConversationSummaries(detailIds)) + results.getLargestSync = await measure(async () => database.getConversation(largest)) + results.getTypicalSync = await measure(async () => database.getConversation(typical)) + database.enableReadonlyWorker(workerBundle) + results.listSummariesWorker = await measure(() => database.listConversationSummariesAsync(detailIds)) + results.getLargestWorker = await measure(() => database.getConversationAsync(largest)) + results.getTypicalWorker = await measure(() => database.getConversationAsync(typical)) + + // The renderer sends a fresh structured clone of the whole list every save. + const unchanged = () => structuredClone({ records, legacyHistoryMayBeIncomplete: false }) + const oneChanged = index => { + const snapshot = unchanged() + const record = snapshot.records[(index + 2) % 50] + record.detail = `${record.detail} ${index}` + return snapshot + } + // Inputs are cloned up front: cloning happens in the renderer / IPC layer. + const run = async (makeInput, call) => { + const inputs = Array.from({ length: calls + 1 }, (_, index) => makeInput(index - 1)) + return measure(async index => call(inputs[index + 1])) + } + const legacyCache = {} + results.replaceUnchangedLegacy = await run(unchanged, input => legacyReplace(raw, legacyCache, input)) + results.replaceOneChangedLegacy = await run(oneChanged, input => legacyReplace(raw, legacyCache, input)) + // Restore a consistent state for the current implementation's cache. + database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: false }) + results.replaceUnchangedAfter = await run(unchanged, input => database.replaceActivityHistory(input)) + results.replaceOneChangedAfter = await run(oneChanged, input => database.replaceActivityHistory(input)) + results.updateOneRecord = await run(index => { + const record = { ...records[(index + 2) % 50], detail: `incremental ${index}`, status: 'running' } + return { changes: [{ type: 'upsert', position: 'front', record }] } + }, input => database.updateActivityHistory(input)) + database.close() + + console.log(`[mperf] seeded ${results.conversations} conversations / ${results.messages} messages (largest ${results.largestConversationMessages}), ${results.activityRecords} activity records in ${seedSeconds} s`) + const rows = [ + ['conversations:list-summaries', results.listSummariesSync, results.listSummariesWorker], + [`conversations:get (${counts[0]} msgs)`, results.getLargestSync, results.getLargestWorker], + [`conversations:get (${counts[Math.floor(conversationCount / 2)]} msgs)`, results.getTypicalSync, results.getTypicalWorker], + ['activity replace, unchanged', results.replaceUnchangedLegacy, results.replaceUnchangedAfter], + ['activity replace, 1 changed', results.replaceOneChangedLegacy, results.replaceOneChangedAfter], + ['activity update, 1 record', results.replaceOneChangedLegacy, results.updateOneRecord] + ] + console.table(rows.map(([name, before, after]) => ({ + path: name, + 'before call p50/p95 ms': `${before.callP50} / ${before.callP95}`, + 'before loop max ms': before.loopMax, + 'after call p50/p95 ms': `${after.callP50} / ${after.callP95}`, + 'after loop max ms': after.loopMax + }))) + const output = process.env.GB_MPERF_OUTPUT + if (output) writeFileSync(output, JSON.stringify({ kind: 'goodbuddy-main-database-benchmark', node: process.version, results }, null, 2)) +} + +main().catch(error => { process.exitCode = 1; console.error(error) }).finally(() => { + rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }) +}) diff --git a/package.json b/package.json index df6221ee..bb8b91f3 100644 --- a/package.json +++ b/package.json @@ -28,6 +28,7 @@ "perf:app": "node build/run-app-perf.cjs", "perf:check": "node build/check-perf-thresholds.cjs", "perf:knowledge": "node build/run-knowledge-perf.cjs", + "perf:main": "node build/run-main-perf.cjs", "eval:retrieval": "vitest run --config tests/support/knowledge-retrieval-evaluation.ts tests/knowledge-retrieval-metrics.test.ts tests/knowledge-retrieval-evaluation.test.ts", "build": "npm run typecheck && npm run build:bundle", "build:bundle": "electron-vite build && node build/check-control-plane-package-installer.cjs", diff --git a/src/main/assistant/activity-history-plan.test.ts b/src/main/assistant/activity-history-plan.test.ts new file mode 100644 index 00000000..b432c2ff --- /dev/null +++ b/src/main/assistant/activity-history-plan.test.ts @@ -0,0 +1,197 @@ +// @vitest-environment node +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import type { DatabaseSync } from 'node:sqlite' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { z } from 'zod' +import { activityHistorySnapshotSchema, type ActivityHistoryChange, type ActivityRecord } from '../../shared/assistant-contracts' +import { AssistantDatabase } from './assistant-database' +import { parseActivityHistorySnapshot, planActivityHistoryReplace, planActivityHistoryUpdate } from './activity-history-plan' + +const directories: string[] = [] +const opened: AssistantDatabase[] = [] +afterEach(() => { + for (const database of opened.splice(0)) database.close() + for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true }) +}) + +function open(): { database: AssistantDatabase; raw: DatabaseSync; directory: string } { + const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-activity-plan-')) + directories.push(directory) + const database = new AssistantDatabase(join(directory, 'assistant.sqlite')) + database.initialize(directory) + opened.push(database) + return { database, raw: (database as unknown as { database: DatabaseSync }).database, directory } +} + +/** The complete persisted state, independent of the read API. */ +function storedState(raw: DatabaseSync): unknown { + return { + header: raw.prepare('SELECT record_order_json, legacy_history_may_be_incomplete FROM activity_history').all(), + rows: raw.prepare('SELECT record_key, record_json FROM activity_history_records ORDER BY record_key').all() + } +} + +let counter = 0 +function record(id: string, overrides: Partial = {}): ActivityRecord { + counter += 1 + return { + id, conversationId: 'conversation', requestId: `request-${id}`, + ...(counter % 2 === 0 ? { callId: `call-${id}` } : {}), + scope: counter % 3 === 0 ? { kind: 'project', projectId: 'p', projectName: 'Project' } : { kind: 'global' }, + kind: 'tool', title: `Tool ${id}`, detail: `detail ${counter}`, status: 'running', createdAt: counter, + ...overrides + } +} + +/** Reference semantics of a change, applied to a plain newest-first list. */ +function applyReference(list: ActivityRecord[], change: ActivityHistoryChange): ActivityRecord[] { + if (change.type === 'remove') return list.filter(item => item.id !== change.id) + const index = list.findIndex(item => item.id === change.record.id) + if (index < 0) return [change.record, ...list] + if (change.position === 'front') return [change.record, ...list.filter((_, at) => at !== index)] + return list.map((item, at) => (at === index ? change.record : item)) +} + +describe('activity history incremental updates', () => { + it('stores exactly the state a full replace stores, for random change sequences', () => { + let seed = 11 + const random = (): number => ((seed = (seed * 1664525 + 1013904223) >>> 0) / 4294967296) + const incremental = open() + const replaced = open() + // Legacy history may contain duplicate IDs. + let list = [record('a'), record('b'), record('a'), record('c'), record('d')] + let flag = true + incremental.database.replaceActivityHistory({ records: list, legacyHistoryMayBeIncomplete: true }) + for (let round = 0; round < 60; round++) { + const changes: ActivityHistoryChange[] = Array.from({ length: 1 + Math.floor(random() * 4) }, () => { + const id = 'abcdefgh'[Math.floor(random() * 8)]! + const roll = random() + if (roll < 0.15) return { type: 'remove', id } + return { type: 'upsert', position: roll < 0.6 ? 'front' : 'in-place', + record: record(id, { status: random() < 0.5 ? 'completed' : 'running' }) } + }) + const legacyHistoryMayBeIncomplete = round % 7 === 0 ? round % 2 === 0 : undefined + for (const change of changes) list = applyReference(list, change) + flag = legacyHistoryMayBeIncomplete ?? flag + incremental.database.updateActivityHistory({ changes, ...(legacyHistoryMayBeIncomplete === undefined ? {} : { legacyHistoryMayBeIncomplete }) }) + const expected = { records: list, legacyHistoryMayBeIncomplete: flag } + replaced.database.replaceActivityHistory(expected) + expect(incremental.database.getActivityHistory()).toEqual(expected) + expect(storedState(incremental.raw)).toEqual(storedState(replaced.raw)) + } + // A fresh connection (no cache) reads the same state. + incremental.database.close() + incremental.database.initialize(incremental.directory) + expect(incremental.database.getActivityHistory().records).toEqual(list) + }) + + it('validates incremental changes and rolls back a failed batch', () => { + const { database, raw } = open() + database.replaceActivityHistory({ records: [record('a')], legacyHistoryMayBeIncomplete: false }) + const before = storedState(raw) + expect(() => database.updateActivityHistory({ changes: [{ type: 'upsert', position: 'front', record: { ...record('b'), title: '' } }] })) + .toThrow(z.ZodError) + raw.exec(`CREATE TRIGGER reject_activity BEFORE INSERT ON activity_history_records + WHEN json_extract(NEW.record_json, '$.id') = 'z' BEGIN SELECT RAISE(ABORT, 'test failure'); END`) + expect(() => database.updateActivityHistory({ changes: [ + { type: 'upsert', position: 'front', record: record('y') }, + { type: 'upsert', position: 'front', record: record('z') } + ] })).toThrow('test failure') + expect(storedState(raw)).toEqual(before) + raw.exec('DROP TRIGGER reject_activity') + const y = record('y') + database.updateActivityHistory({ changes: [{ type: 'upsert', position: 'front', record: y }] }) + expect(database.getActivityHistory().records.map(item => item.id)).toEqual(['y', 'a']) + }) + + it('writes only the touched rows', () => { + const { database, raw } = open() + const records = Array.from({ length: 2_000 }, (_, index) => record(String(index))) + database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: false }) + const changes = (): number => (raw.prepare('SELECT total_changes() AS n').get() as { n: number }).n + const start = changes() + const stringify = vi.spyOn(JSON, 'stringify') + database.updateActivityHistory({ changes: [{ type: 'upsert', position: 'front', record: { ...records[1_500]!, status: 'completed' } }] }) + const calls = stringify.mock.calls.filter(([value]) => !Array.isArray(value)).length + stringify.mockRestore() + // One record row plus the order row. + expect(changes() - start).toBe(2) + expect(calls).toBeLessThan(10) + }) +}) + +describe('activity history replace', () => { + it('does no writes, transactions or per-record serialization for 5,000 unchanged records', () => { + const { database, raw } = open() + const records = Array.from({ length: 5_000 }, (_, index) => record(String(index % 4_000))) + database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: true }) + // The renderer sends a fresh structured clone every time. + const clone = (): unknown => structuredClone({ records, legacyHistoryMayBeIncomplete: true }) + const total = (): number => (raw.prepare('SELECT total_changes() AS n').get() as { n: number }).n + const start = total() + const exec = vi.spyOn(raw, 'exec') + const stringify = vi.spyOn(JSON, 'stringify') + const parse = vi.spyOn(activityHistorySnapshotSchema, 'parse') + database.replaceActivityHistory(clone()) + // Record serialization only; [id, occurrence] keys are tiny. + const stringifyCalls = stringify.mock.calls.filter(([value]) => !Array.isArray(value)).length + stringify.mockRestore() + expect(exec).not.toHaveBeenCalled() + expect(parse).not.toHaveBeenCalled() + expect(stringifyCalls).toBeLessThan(10) + exec.mockRestore() + parse.mockRestore() + expect(total()).toBe(start) + + // One changed record: one row write plus nothing else. + const changed = structuredClone(records) + changed[42]!.detail += ' changed' + database.replaceActivityHistory({ records: changed, legacyHistoryMayBeIncomplete: true }) + expect(total() - start).toBe(1) + expect(database.getActivityHistory().records).toEqual(changed) + }) + + it('keeps the schema behavior: same errors, strictness and normalized output', () => { + const { database } = open() + const valid = [record('a'), record('b')] + database.replaceActivityHistory({ records: valid, legacyHistoryMayBeIncomplete: false }) + const cases: unknown[] = [ + { records: [valid[0], { ...valid[1], extra: 1 }], legacyHistoryMayBeIncomplete: false }, + { records: [valid[0], { ...valid[1], createdAt: -1 }], legacyHistoryMayBeIncomplete: false }, + { records: [valid[0], { ...valid[1], scope: { kind: 'global', extra: true } }], legacyHistoryMayBeIncomplete: false }, + { records: valid, legacyHistoryMayBeIncomplete: false, extra: 1 }, + { records: valid }, + { records: 'x', legacyHistoryMayBeIncomplete: false }, + [valid], + null + ] + for (const input of cases) { + const expected = activityHistorySnapshotSchema.safeParse(input) + expect(expected.success).toBe(false) + let actual: unknown + try { database.replaceActivityHistory(input) } catch (error) { actual = error } + expect(actual).toBeInstanceOf(z.ZodError) + expect((actual as z.ZodError).issues).toEqual(expected.error!.issues) + } + expect(database.getActivityHistory().records).toEqual(valid) + // Key order and an explicit undefined optional key follow the schema. + const reordered = Object.fromEntries(Object.entries(valid[0]!).reverse()) + database.replaceActivityHistory({ records: [reordered, { ...valid[1], callId: undefined }], legacyHistoryMayBeIncomplete: false }) + expect(database.getActivityHistory()).toEqual(activityHistorySnapshotSchema.parse({ records: [reordered, { ...valid[1], callId: undefined }], legacyHistoryMayBeIncomplete: false })) + }) + + it('plans identical results with and without cached records', () => { + const records = [record('a'), record('b'), record('a')] + const empty = { entries: new Map() } + const first = planActivityHistoryReplace(parseActivityHistorySnapshot({ records, legacyHistoryMayBeIncomplete: false }).items, false, empty) + const reparsed = parseActivityHistorySnapshot(structuredClone({ records, legacyHistoryMayBeIncomplete: false }), first.next) + expect(reparsed.items.every((item, index) => item.record === first.next.entries.get([...first.next.order][index]!)!.record)).toBe(true) + const second = planActivityHistoryReplace(reparsed.items, false, first.next) + expect(second.upserts).toEqual([]) + expect(second.removes).toEqual([]) + expect(second.headerChanged).toBe(false) + expect(() => planActivityHistoryUpdate({ changes: [] }, { entries: new Map() })).toThrow('unreadable') + }) +}) diff --git a/src/main/assistant/activity-history-plan.ts b/src/main/assistant/activity-history-plan.ts new file mode 100644 index 00000000..aaaed0ee --- /dev/null +++ b/src/main/assistant/activity-history-plan.ts @@ -0,0 +1,203 @@ +import { + activityHistorySnapshotSchema, + activityHistoryUpdateSchema, + activityRecordSchema, + type ActivityRecord +} from '../../shared/assistant-contracts' + +// PERF-15: activity history persistence planning. The renderer sends the whole +// newest-first list every change; most records are unchanged. These pure +// helpers validate and diff against the last committed state without +// re-validating or re-serializing unchanged records, and turn incremental +// changes into the same stored rows that `replace` would produce. + +/** A stored row; `record` is a validated, privately owned copy when known. */ +export type ActivityHistoryEntry = { json: string; record?: ActivityRecord } + +export type ActivityHistoryState = { + /** Every row in activity_history_records, keyed by record_key. */ + entries: ReadonlyMap + /** Stored order; undefined when unknown (always rewritten). */ + order?: readonly string[] + orderJson?: string + incomplete?: boolean +} + +export type ActivityHistoryPlan = { + next: ActivityHistoryState & { order: readonly string[]; orderJson: string; incomplete: boolean } + upserts: Array<[key: string, json: string]> + removes: string[] + /** Order or legacy flag changed: the singleton row must be updated. */ + headerChanged: boolean +} + +type ParsedItem = { record: ActivityRecord; json?: string; key?: string } + +/** IDs were never unique in the snapshot schema; every occurrence is kept. */ +export const activityRecordKey = (id: string, occurrence: number): string => JSON.stringify([id, occurrence]) + +export const isActivityHistoryPlanEmpty = (plan: ActivityHistoryPlan): boolean => + plan.upserts.length === 0 && plan.removes.length === 0 && !plan.headerChanged + +function isPlainObject(value: unknown): value is Record { + if (typeof value !== 'object' || value === null || Array.isArray(value)) return false + const prototype = Object.getPrototypeOf(value) as unknown + return prototype === Object.prototype || prototype === null +} + +/** Exact structural equality against a validated record (same own keys, same values). */ +function sameValue(raw: unknown, validated: unknown): boolean { + if (raw === validated) return true + if (!isPlainObject(validated) || !isPlainObject(raw)) return false + const keys = Object.keys(validated) + if (Object.keys(raw).length !== keys.length) return false + for (const key of keys) { + if (!Object.hasOwn(raw, key) || !sameValue(raw[key], validated[key])) return false + } + return true +} + +function sameOrder(left: readonly string[], right: readonly string[] | undefined): boolean { + if (!right || left.length !== right.length) return false + for (let index = 0; index < left.length; index++) if (left[index] !== right[index]) return false + return true +} + +/** + * Validates a replace snapshot with the same accept/reject behavior and errors + * as `activityHistorySnapshotSchema.parse`, reusing previously validated + * records (and their serialized form) that are structurally identical. + */ +export function parseActivityHistorySnapshot( + input: unknown, + previous?: ActivityHistoryState +): { items: ParsedItem[]; incomplete: boolean } { + let raw: { records: readonly unknown[]; legacyHistoryMayBeIncomplete: boolean } + if ( + isPlainObject(input) && Object.keys(input).length === 2 && + Array.isArray(input.records) && typeof input.legacyHistoryMayBeIncomplete === 'boolean' + ) { + raw = input as typeof raw + } else { + raw = activityHistorySnapshotSchema.parse(input) + } + const occurrences = new Map() + const items: ParsedItem[] = new Array(raw.records.length) + for (let index = 0; index < raw.records.length; index++) { + const candidate = raw.records[index] + const id = isPlainObject(candidate) ? candidate.id : undefined + if (typeof id === 'string') { + const occurrence = occurrences.get(id) ?? 0 + const key = activityRecordKey(id, occurrence) + const entry = previous?.entries.get(key) + if (entry?.record && sameValue(candidate, entry.record)) { + occurrences.set(id, occurrence + 1) + items[index] = { record: entry.record, json: entry.json, key } + continue + } + } + const parsed = activityRecordSchema.safeParse(candidate) + // Throw exactly the error (paths included) the whole-snapshot schema throws. + if (!parsed.success) activityHistorySnapshotSchema.parse(input) + const record = parsed.success ? parsed.data : activityRecordSchema.parse(candidate) + occurrences.set(record.id, (occurrences.get(record.id) ?? 0) + 1) + items[index] = { record } + } + return { items, incomplete: raw.legacyHistoryMayBeIncomplete } +} + +/** Diffs a full newest-first list against the committed state. */ +export function planActivityHistoryReplace( + items: readonly ParsedItem[], + incomplete: boolean, + previous: ActivityHistoryState, + onProgress?: (processed: number) => void +): ActivityHistoryPlan { + const entries = new Map() + const occurrences = new Map() + const order: string[] = new Array(items.length) + const upserts: Array<[string, string]> = [] + for (let index = 0; index < items.length; index++) { + if (index % 128 === 0) onProgress?.(index) + const { record } = items[index]! + const occurrence = occurrences.get(record.id) ?? 0 + occurrences.set(record.id, occurrence + 1) + const key = items[index]!.key ?? activityRecordKey(record.id, occurrence) + const json = items[index]!.json ?? JSON.stringify(record) + order[index] = key + if (previous.entries.get(key)?.json !== json) upserts.push([key, json]) + entries.set(key, { json, record }) + } + const removes: string[] = [] + for (const key of previous.entries.keys()) if (!entries.has(key)) removes.push(key) + return finish(entries, order, incomplete, previous, upserts, removes) +} + +function finish( + entries: Map, + order: string[], + incomplete: boolean, + previous: ActivityHistoryState, + upserts: Array<[string, string]>, + removes: string[] +): ActivityHistoryPlan { + const orderUnchanged = sameOrder(order, previous.order) && previous.orderJson !== undefined + const orderJson = orderUnchanged ? previous.orderJson! : JSON.stringify(order) + return { + next: { entries, order, orderJson, incomplete }, + upserts, + removes, + headerChanged: !orderUnchanged || previous.incomplete !== incomplete + } +} + +/** + * Applies incremental changes (see `activityHistoryChangeSchema`). Only the + * changed records are validated and serialized; the result is identical to + * `planActivityHistoryReplace` with the resulting list. + */ +export function planActivityHistoryUpdate(input: unknown, previous: ActivityHistoryState): ActivityHistoryPlan { + const update = activityHistoryUpdateSchema.parse(input) + if (!previous.order) throw new Error('Activity history order is unreadable; save the full history instead') + const entries = new Map(previous.entries) + let order = [...previous.order] + const touched = new Set() + for (const change of update.changes) { + if (change.type === 'remove') { + // Keys of one ID share the JSON prefix `["id",`. + const prefix = `${JSON.stringify([change.id]).slice(0, -1)},` + const removed = order.filter(key => key.startsWith(prefix)) + if (removed.length > 0) order = order.filter(key => !key.startsWith(prefix)) + for (const key of removed) { entries.delete(key); touched.add(key) } + for (let occurrence = 0; ; occurrence++) { + const key = activityRecordKey(change.id, occurrence) + if (!entries.delete(key)) break + touched.add(key) + } + continue + } + const { record } = change + // The first occurrence stays first when moved to the front, so its key is stable. + const key = activityRecordKey(record.id, 0) + const index = order.indexOf(key) + if (index < 0) order.unshift(key) + else if (change.position === 'front' && index > 0) { + order.splice(index, 1) + order.unshift(key) + } + entries.set(key, { json: JSON.stringify(record), record }) + touched.add(key) + } + const upserts: Array<[string, string]> = [] + const removes: string[] = [] + for (const key of touched) { + const entry = entries.get(key) + if (!entry) { + if (previous.entries.has(key)) removes.push(key) + } else if (previous.entries.get(key)?.json !== entry.json) { + upserts.push([key, entry.json]) + } + } + return finish(entries, order, update.legacyHistoryMayBeIncomplete ?? previous.incomplete ?? false, + previous, upserts, removes) +} diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index a5c5f367..fe5b054e 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -6,6 +6,14 @@ import { appendConversationQuestionBlock } from '../../shared/conversation-quest import { DatabaseSync } from 'node:sqlite' import { ExecutionStatsReader } from './execution-stats-reader' import { ReadonlyQueryReader, readWithFallback } from '../readonly-query-reader' +import { + isActivityHistoryPlanEmpty, + parseActivityHistorySnapshot, + planActivityHistoryReplace, + planActivityHistoryUpdate, + type ActivityHistoryPlan, + type ActivityHistoryState +} from './activity-history-plan' import { statSync } from 'node:fs' import { MagicNoteStorage } from '../magic-notes/magic-note-storage' import type { AssistantStorageProgress } from '../../shared/assistant-storage-contracts' @@ -1963,7 +1971,7 @@ export class AssistantDatabase { private static readonly executionStatsCacheTtlMs = 30_000 private static readonly executionStatsCacheLimit = 8 private database?: DatabaseSync - private activityHistoryCache?: { records: Map; dataVersion: number } + private activityHistoryCache?: { state: ActivityHistoryState; dataVersion: number } private executionStatsReader?: ExecutionStatsReader private readonlyReader?: ReadonlyQueryReader private readonlyWorkerPath?: string @@ -3008,6 +3016,25 @@ export class AssistantDatabase { return this.readConversationList(new Set(detailIds)) } + /** + * Conversation summaries on the readonly worker (PERF-15). Every write runs + * synchronously on Main, so once no transaction is open, everything written + * before this call is committed and visible to the worker. Inside an open + * transaction the synchronous path keeps read-your-writes semantics. + */ + listConversationSummariesAsync(detailIds: string[] = [], signal?: AbortSignal): Promise { + const database = this.requireDatabase() + return readWithFallback(database.isTransaction ? undefined : this.readonlyQueryReader(), + 'listConversationSummaries', [detailIds], signal, () => this.listConversationSummaries(detailIds)) + } + + /** Full conversation history on the readonly worker; same rules as listConversationSummariesAsync. */ + getConversationAsync(conversationId: string, signal?: AbortSignal): Promise { + const database = this.requireDatabase() + return readWithFallback(database.isTransaction ? undefined : this.readonlyQueryReader(), + 'getConversation', [conversationId], signal, () => this.getConversation(conversationId)) + } + private listConversationRows(): ConversationRow[] { const database = this.requireDatabase() return database @@ -4969,6 +4996,25 @@ export class AssistantDatabase { this.database = new DatabaseSync(this.databasePath, { readOnly: true, timeout: 5_000 }) } + /** + * Runs multi-statement reads against one WAL snapshot. Used by the readonly + * worker, where Main may commit between statements; on Main itself the + * event loop already serializes reads and writes. + */ + readSnapshot(read: () => T): T { + const database = this.requireDatabase() + if (database.isTransaction) return read() + database.exec('BEGIN') + try { + const result = read() + database.exec('COMMIT') + return result + } catch (error) { + database.exec('ROLLBACK') + throw error + } + } + /** * Routes heavy read-only queries (story graph, conversation search) through a * read-only worker thread. In-memory databases keep the synchronous path. @@ -5255,67 +5301,108 @@ export class AssistantDatabase { }) } - replaceActivityHistory(input: ActivityHistorySnapshot): void { - const snapshot = activityHistorySnapshotSchema.parse(input) + /** + * Persists the whole newest-first list. Unchanged records are neither + * re-validated, re-serialized nor written; a fully unchanged snapshot does + * not open a write transaction (PERF-15). + */ + replaceActivityHistory(input: unknown): void { const database = this.requireDatabase() + // Reuse is safe even from a stale cache: cached records are validated copies. + const { items, incomplete } = parseActivityHistorySnapshot(input, this.activityHistoryCache?.state) + const committed = this.committedActivityHistoryState(database) + const planned = committed && planActivityHistoryReplace(items, incomplete, committed) + if (planned && isActivityHistoryPlanEmpty(planned)) return + this.writeActivityHistoryPlan(database, (previous) => previous === committed && planned + ? planned : planActivityHistoryReplace(items, incomplete, previous)) + } + + /** Applies incremental changes in one transaction (PERF-15). */ + updateActivityHistory(input: unknown): void { + const database = this.requireDatabase() + this.writeActivityHistoryPlan(database, (previous) => planActivityHistoryUpdate(input, previous)) + } + + /** The cached committed state, when no other connection committed since. */ + private committedActivityHistoryState(database: DatabaseSync): ActivityHistoryState | undefined { + const cache = this.activityHistoryCache + if (!cache || database.isTransaction) return undefined + const { data_version: dataVersion } = database.prepare('PRAGMA data_version').get() as { data_version: number } + return cache.dataVersion === dataVersion ? cache.state : undefined + } + + private writeActivityHistoryPlan( + database: DatabaseSync, + plan: (previous: ActivityHistoryState) => ActivityHistoryPlan + ): void { database.exec('BEGIN IMMEDIATE') try { const { data_version: dataVersion } = database.prepare('PRAGMA data_version').get() as { data_version: number } const previous = this.activityHistoryCache?.dataVersion === dataVersion - ? this.activityHistoryCache.records : undefined - const records = this.writeActivityHistory(database, snapshot, undefined, previous) + ? this.activityHistoryCache.state : this.readActivityHistoryState(database) + const result = plan(previous) + this.applyActivityHistoryPlan(database, result) database.exec('COMMIT') // Publish only committed records; a failed save must remain retryable. - this.activityHistoryCache = { records, dataVersion } + this.activityHistoryCache = { state: result.next, dataVersion } } catch (error) { database.exec('ROLLBACK') throw error } } + private readActivityHistoryState(database: DatabaseSync): ActivityHistoryState { + const rows = database.prepare( + 'SELECT record_key, record_json FROM activity_history_records' + ).all() as Array<{ record_key: string; record_json: string }> + const header = database.prepare( + 'SELECT record_order_json, legacy_history_may_be_incomplete FROM activity_history WHERE singleton = 1' + ).get() as { record_order_json: string; legacy_history_may_be_incomplete: number } | undefined + const order = header ? JSON.parse(header.record_order_json) as unknown : undefined + return { + entries: new Map(rows.map((row) => [row.record_key, { json: row.record_json }])), + ...(Array.isArray(order) && order.every((key) => typeof key === 'string') + ? { order, orderJson: header!.record_order_json } : {}), + ...(header ? { incomplete: header.legacy_history_may_be_incomplete === 1 } : {}) + } + } + + private applyActivityHistoryPlan(database: DatabaseSync, plan: ActivityHistoryPlan): void { + if (plan.upserts.length > 0) { + const upsert = database.prepare( + `INSERT INTO activity_history_records (record_key, record_json) VALUES (?, ?) + ON CONFLICT(record_key) DO UPDATE SET record_json = excluded.record_json` + ) + for (const [key, json] of plan.upserts) upsert.run(key, json) + } + if (plan.removes.length > 0) { + const remove = database.prepare('DELETE FROM activity_history_records WHERE record_key = ?') + for (const key of plan.removes) remove.run(key) + } + if (plan.headerChanged) { + database.prepare( + `UPDATE activity_history + SET record_order_json = ?, legacy_history_may_be_incomplete = ? + WHERE singleton = 1` + ).run(plan.next.orderJson, Number(plan.next.incomplete)) + } + } + + /** Schema-44 migration: writes a parsed legacy snapshot into the new tables. */ private writeActivityHistory( database: DatabaseSync, snapshot: ActivityHistorySnapshot, - onProgress?: (processed: number) => void, - previous?: ReadonlyMap - ): Map { - if (!previous) { - const rows = database.prepare( - 'SELECT record_key, record_json FROM activity_history_records' - ).all() as Array<{ record_key: string; record_json: string }> - previous = new Map(rows.map((row) => [row.record_key, row.record_json])) + onProgress?: (processed: number) => void + ): void { + const rows = database.prepare( + 'SELECT record_key, record_json FROM activity_history_records' + ).all() as Array<{ record_key: string; record_json: string }> + // The order column still holds the legacy records; always rewrite it. + const previous: ActivityHistoryState = { + entries: new Map(rows.map((row) => [row.record_key, { json: row.record_json }])) } - const records = new Map() - const occurrences = new Map() - const order: string[] = [] - const upsert = database.prepare( - `INSERT INTO activity_history_records (record_key, record_json) VALUES (?, ?) - ON CONFLICT(record_key) DO UPDATE SET record_json = excluded.record_json` - ) - for (const record of snapshot.records) { - if (order.length % 128 === 0) onProgress?.(order.length) - // IDs were never unique in the snapshot schema; retain every occurrence. - const occurrence = occurrences.get(record.id) ?? 0 - occurrences.set(record.id, occurrence + 1) - const key = JSON.stringify([record.id, occurrence]) - const json = JSON.stringify(record) - order.push(key) - if (previous.get(key) !== json) upsert.run(key, json) - records.set(key, json) - } - const remove = database.prepare('DELETE FROM activity_history_records WHERE record_key = ?') - for (const key of previous.keys()) { - if (!records.has(key)) remove.run(key) - } - const orderJson = JSON.stringify(order) - const incomplete = Number(snapshot.legacyHistoryMayBeIncomplete) - database.prepare( - `UPDATE activity_history - SET record_order_json = ?, legacy_history_may_be_incomplete = ? - WHERE singleton = 1 - AND (record_order_json != ? OR legacy_history_may_be_incomplete != ?)` - ).run(orderJson, incomplete, orderJson, incomplete) - return records + this.applyActivityHistoryPlan(database, planActivityHistoryReplace( + snapshot.records.map((record) => ({ record })), snapshot.legacyHistoryMayBeIncomplete, previous, onProgress)) } createTask(input: { diff --git a/src/main/ipc.test.ts b/src/main/ipc.test.ts index 50557c75..c517ec1e 100644 --- a/src/main/ipc.test.ts +++ b/src/main/ipc.test.ts @@ -5404,6 +5404,9 @@ describe('registerIpcHandlers agent terminal state', () => { ), listConversations: vi.fn<() => ConversationSnapshot[]>(() => []), listConversationSummaries: vi.fn<(ids?: string[]) => ConversationSnapshot[]>(() => []), + // The readonly-worker variants delegate to the synchronous mocks. + listConversationSummariesAsync: vi.fn(async (ids?: string[]) => assistantDatabase.listConversationSummaries(ids)), + getConversationAsync: vi.fn(async (conversationId: string) => assistantDatabase.getConversation(conversationId)), listRecoverableRemoteTasks: vi.fn< () => Array<{ taskId: string @@ -8221,10 +8224,10 @@ describe('registerIpcHandlers agent terminal state', () => { harness.recoveryGetHandler?.(event) await vi.waitFor(() => expect(ready.size).toBe(2)) await vi.waitFor(() => expect(list().every(c => c.activeRequest?.questions.length === 1)).toBe(true)) - const summaries = electronMocks.handlers.get(ipcChannels.conversationsListSummaries)!(event, { detailIds: [] }) as ConversationSnapshot[] + const summaries = await electronMocks.handlers.get(ipcChannels.conversationsListSummaries)!(event, { detailIds: [] }) as ConversationSnapshot[] expect(summaries).toEqual(list()) for (const snapshot of summaries) { - expect(electronMocks.handlers.get(ipcChannels.conversationsGet)!(event, snapshot.id)).toEqual(snapshot) + expect(await electronMocks.handlers.get(ipcChannels.conversationsGet)!(event, snapshot.id)).toEqual(snapshot) } expect(harness.assistantDatabase.appendRemoteConversationTaskEventOnce).not.toHaveBeenCalled() const update = settingsKind === 'runtime' diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 264b4f08..39e21c7f 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -132,7 +132,6 @@ import { registerMagicNotesIpcHandlers, registerMagicTodosIpcHandlers } from './ import { assistantIdSchema, executionStatsInputSchema, - activityHistorySnapshotSchema, conversationBranchInputSchema, conversationSnapshotsSchema, conversationListRequestSchema, @@ -6329,16 +6328,19 @@ export function registerIpcHandlers( assertTrustedSender(event, window) return projectConversationRequests(assistantDatabase.listConversations()) }) - registerHandler(ipcChannels.conversationsListSummaries, (event, input: unknown) => { + // PERF-15: both reads run on the readonly worker. All writes are synchronous + // on Main, so a read issued after a write observes it; request projection + // uses the in-memory state current when the result arrives. + registerHandler(ipcChannels.conversationsListSummaries, async (event, input: unknown) => { assertTrustedSender(event, window) const { detailIds } = conversationListRequestSchema.parse(input) - return projectConversationRequests(assistantDatabase.listConversationSummaries([ + return projectConversationRequests(await assistantDatabase.listConversationSummariesAsync([ ...detailIds, ...[...activeRequests.values()].map(request => request.conversationId) ])) }) - registerHandler(ipcChannels.conversationsGet, (event, input: unknown) => { + registerHandler(ipcChannels.conversationsGet, async (event, input: unknown) => { assertTrustedSender(event, window) - return projectConversationRequests([assistantDatabase.getConversation(assistantIdSchema.parse(input))])[0] + return projectConversationRequests([await assistantDatabase.getConversationAsync(assistantIdSchema.parse(input))])[0] }) registerHandler(ipcChannels.conversationsSearch, (event, input: unknown) => { assertTrustedSender(event, window) @@ -6830,9 +6832,15 @@ export function registerIpcHandlers( ipcChannels.activityHistoryReplace, (event, input: unknown) => { assertTrustedSender(event, window) - assistantDatabase.replaceActivityHistory( - activityHistorySnapshotSchema.parse(input) - ) + // Validation happens inside, reusing already validated unchanged records. + assistantDatabase.replaceActivityHistory(input) + } + ) + registerHandler( + ipcChannels.activityHistoryUpdate, + (event, input: unknown) => { + assertTrustedSender(event, window) + assistantDatabase.updateActivityHistory(input) } ) diff --git a/src/main/readonly-query-reader.test.ts b/src/main/readonly-query-reader.test.ts index 1c8ef7c3..d0719b05 100644 --- a/src/main/readonly-query-reader.test.ts +++ b/src/main/readonly-query-reader.test.ts @@ -194,6 +194,59 @@ describe('assistant readonly worker', () => { } finally { database.close() } }, 60_000) + it('serves conversation summaries and full history from the worker with the synchronous results', 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 message = (content: string, createdAt: number, state: 'complete' | 'streaming' = 'complete', role: 'user' | 'assistant' = 'assistant') => + ({ id: randomUUID(), role, state, content, createdAt }) + const conversations = Array.from({ length: 12 }, (_, index) => ({ + id: randomUUID(), projectId: index % 2 === 0 ? project.id : undefined, title: `Conversation ${index}`, + updatedAt: 100 - index, pinned: index === 7, + messages: index === 0 ? [] : Array.from({ length: index }, (_, at) => + message(`m${index}-${at} 跨平台`, index * 10 + at, index === 5 && at === index - 1 ? 'streaming' : 'complete', at === 0 ? 'user' : 'assistant')) + })) + database.replaceConversations(conversations) + const detailSets = [[], [conversations[3]!.id], conversations.map(item => item.id), [randomUUID()]] + const expected = detailSets.map(ids => database.listConversationSummaries(ids)) + const expectedFull = conversations.map(item => database.getConversation(item.id)) + // Summaries without details carry messageSummary; retained/streaming ones carry messages. + expect(expected[0]!.find(item => item.id === conversations[4]!.id)?.messageSummary).toEqual({ count: 4, firstRole: 'user', latestMessageAt: 43 }) + expect(expected[0]!.find(item => item.id === conversations[5]!.id)?.messages).toHaveLength(5) + expect(expected[1]!.find(item => item.id === conversations[3]!.id)).toEqual(expectedFull[3]) + + database.enableReadonlyWorker(workerPath) + expect(await Promise.all(detailSets.map(ids => database.listConversationSummariesAsync(ids)))).toEqual(expected) + expect(await Promise.all(conversations.map(item => database.getConversationAsync(item.id)))).toEqual(expectedFull) + expect(database.readonlyWorkerForTest?.pendingCount).toBe(0) + // Missing conversations reject with the synchronous error. + await expect(database.getConversationAsync(randomUUID())).rejects.toThrow('对话不存在') + + // Read after write: a synchronous write on Main is visible to the next worker read. + database.saveLocalConversations([{ header: { id: conversations[1]!.id, title: 'Renamed', updatedAt: 500 }, + messages: [message('fresh', 999)] }]) + const afterWrite = await database.listConversationSummariesAsync([]) + expect(afterWrite).toEqual(database.listConversationSummaries([])) + expect(afterWrite[0]!.title).toBe('Renamed') + expect(await database.getConversationAsync(conversations[1]!.id)).toEqual(database.getConversation(conversations[1]!.id)) + + // Inside an open transaction the synchronous path keeps read-your-writes. + const raw = (database as unknown as { database: import('node:sqlite').DatabaseSync }).database + raw.exec('BEGIN IMMEDIATE') + try { + raw.prepare('UPDATE conversations SET title = ? WHERE id = ?').run('Uncommitted', conversations[2]!.id) + expect((await database.getConversationAsync(conversations[2]!.id)).title).toBe('Uncommitted') + } finally { raw.exec('ROLLBACK') } + + // Worker crash: served synchronously, identical. + const inFlight = database.listConversationSummariesAsync([conversations[3]!.id]) + await database.readonlyWorkerForTest!.terminateWorkerForTest() + expect(await inFlight).toEqual(database.listConversationSummaries([conversations[3]!.id])) + } 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')) diff --git a/src/main/readonly-query-worker.ts b/src/main/readonly-query-worker.ts index ef9fbfe2..1a7fdcf8 100644 --- a/src/main/readonly-query-worker.ts +++ b/src/main/readonly-query-worker.ts @@ -31,7 +31,12 @@ if (kind === 'knowledge') { 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) + searchConversations: ([query]) => database.searchConversations(query as string), + // Several statements each: read them from one snapshot. + listConversationSummaries: ([detailIds]) => database.readSnapshot(() => + database.listConversationSummaries(detailIds as string[])), + getConversation: ([conversationId]) => database.readSnapshot(() => + database.getConversation(conversationId as string)) } } diff --git a/src/preload/index.ts b/src/preload/index.ts index 49646d5b..3346ba44 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -68,6 +68,7 @@ import type { AssistantTask, ExecutionStats, ActivityHistorySnapshot, + ActivityHistoryUpdate, ActivityRecord, TokenUsageSummary, ConversationBranchInput, @@ -1286,6 +1287,9 @@ const desktopApi: DesktopApi = { records, legacyHistoryMayBeIncomplete }) + }, + update: async (update: ActivityHistoryUpdate) => { + await ipcRenderer.invoke(ipcChannels.activityHistoryUpdate, update) } }, usage: { diff --git a/src/shared/assistant-contracts.ts b/src/shared/assistant-contracts.ts index ea2acc0b..fff5e0a7 100644 --- a/src/shared/assistant-contracts.ts +++ b/src/shared/assistant-contracts.ts @@ -100,10 +100,42 @@ export const activityHistorySnapshotSchema = z }) .strict() +/** + * Incremental activity history change (PERF-15). Changes apply in order to the + * stored, newest-first list; the result is identical to `replace` with the + * same final list. A record is identified by its `id` (first occurrence when + * legacy history contains duplicate IDs). + * - `upsert` + `position: 'front'` inserts the record, or moves the existing + * record to the front with the new content (renderer `upsertActivityRecord`). + * - `upsert` + `position: 'in-place'` replaces the existing record where it + * is; a record that does not exist yet is inserted at the front. + * - `remove` deletes every record with the ID. + */ +export const activityHistoryChangeSchema = z.discriminatedUnion('type', [ + z.object({ + type: z.literal('upsert'), + record: activityRecordSchema, + position: z.enum(['front', 'in-place']) + }).strict(), + z.object({ + type: z.literal('remove'), + id: z.string().min(1).max(256) + }).strict() +]) + +export const activityHistoryUpdateSchema = z + .object({ + changes: z.array(activityHistoryChangeSchema).max(20_000), + legacyHistoryMayBeIncomplete: z.boolean().optional() + }) + .strict() + export type ActivityRecord = z.infer export type ActivityHistorySnapshot = z.infer< typeof activityHistorySnapshotSchema > +export type ActivityHistoryChange = z.infer +export type ActivityHistoryUpdate = z.infer export const projectExecutionSpaceSchema = z.discriminatedUnion( 'kind', diff --git a/src/shared/contracts.ts b/src/shared/contracts.ts index efe89926..e06201ee 100644 --- a/src/shared/contracts.ts +++ b/src/shared/contracts.ts @@ -45,6 +45,7 @@ import { type ExecutionStats, type ExecutionStatsInput, type ActivityHistorySnapshot, + type ActivityHistoryUpdate, type ActivityRecord, type TokenUsageSummary, type ConversationSnapshot, @@ -2158,6 +2159,12 @@ export type DesktopApi = { records: ActivityRecord[], legacyHistoryMayBeIncomplete: boolean ) => Promise + /** + * Incremental save (PERF-15): only the changed records cross IPC and are + * written. Optional until renderer mocks adopt it; the preload always + * provides it. + */ + update?: (update: ActivityHistoryUpdate) => Promise } usage: { getTokenSummary: () => Promise diff --git a/src/shared/ipc-channels.ts b/src/shared/ipc-channels.ts index c06314df..ffdd5201 100644 --- a/src/shared/ipc-channels.ts +++ b/src/shared/ipc-channels.ts @@ -254,6 +254,7 @@ export const ipcChannels = { tasksSetStatus: 'tasks:set-status', activityHistoryGet: 'activity-history:get', activityHistoryReplace: 'activity-history:replace', + activityHistoryUpdate: 'activity-history:update', tokenUsageSummary: 'usage:token-summary', tokenUsageChanged: 'usage:token-changed', artifactsList: 'artifacts:list',