From 352da66dbd98ebbf3aafb1a3675c9b51e4ea7c47 Mon Sep 17 00:00:00 2001 From: mesalogo Date: Sat, 3 Oct 2026 21:42:52 +0800 Subject: [PATCH] perf(activity): page, summarize and reconcile activity history in Main by index Schema 57 rebuilds activity_history_records with order_seq (newer = larger; a move to the front assigns a new maximum) and stored generated columns (record_id, occurrence, status, status_group, kind, request_id, conversation_id, project_id, created_at) with indexes. The order is migrated from record_order_json exactly; legacy duplicates keep their occurrence and only the first is shown. record_order_json is no longer written. ActivityHistoryRepository owns the SQL: page({before, limit <= 500, filter, projectId, conversationId}) newest first by index, summary() with the counts and conversation titles/status the Activity page derived from the full list, reconcile() joined with tasks with a renderer-localized detail line, update() applied directly in SQL without reading the history, replace() for the legacy migration, clear(). New activity-history:page/summary/reconcile IPC with zod validation; page and summary run on the readonly worker. perf:main measures 24,000 records: startup load 106 ms on Main (14 MB) -> first page + summary 13.5 ms on the worker (0.6 ms loop block, 114 KB). --- build/run-main-perf.cjs | 31 +- .../assistant/activity-history-plan.test.ts | 15 +- src/main/assistant/activity-history-plan.ts | 134 ++---- .../activity-history-repository.test.ts | 203 +++++++++ .../assistant/activity-history-repository.ts | 418 ++++++++++++++++++ src/main/assistant/assistant-database.ts | 201 +++------ src/main/ipc.ts | 20 + src/main/readonly-query-worker.ts | 4 +- src/preload/index.ts | 13 +- src/renderer/src/App.test.tsx | 10 +- src/shared/activity-history-reference.ts | 110 ++++- src/shared/assistant-contracts.ts | 50 +++ src/shared/contracts.ts | 14 + src/shared/ipc-channels.ts | 3 + 14 files changed, 963 insertions(+), 263 deletions(-) create mode 100644 src/main/assistant/activity-history-repository.test.ts create mode 100644 src/main/assistant/activity-history-repository.ts diff --git a/build/run-main-perf.cjs b/build/run-main-perf.cjs index ac2934ce..25a005e2 100644 --- a/build/run-main-perf.cjs +++ b/build/run-main-perf.cjs @@ -1,6 +1,6 @@ // 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 +// assistant database (default 1,000 conversations / 20,000 messages and 24,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 @@ -10,6 +10,9 @@ // 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 +// activity startup load whole list (pre-paging get) vs first page (200) +// + summary on the readonly worker +// activity page / summary sync paths, for reference // 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') @@ -35,7 +38,7 @@ if (process.env.GB_MPERF_NODE !== '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 activityCount = Number(process.env.GB_MPERF_ACTIVITY || 24000) const calls = Number(process.env.GB_MPERF_CALLS || 15) const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-main-perf-')) @@ -118,7 +121,8 @@ function legacyReplace(raw, cache, input) { const records = new Map() const occurrences = new Map() const order = [] - const upsert = raw.prepare(`INSERT INTO activity_history_records (record_key, record_json) VALUES (?, ?) + // order_seq only satisfies the current schema; the old code kept order in the JSON list below. + const upsert = raw.prepare(`INSERT INTO activity_history_records (record_key, record_json, order_seq) 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 @@ -126,7 +130,7 @@ function legacyReplace(raw, cache, input) { const key = JSON.stringify([record.id, occurrence]) const json = JSON.stringify(record) order.push(key) - if (previous.get(key) !== json) upsert.run(key, json) + if (previous.get(key) !== json) upsert.run(key, json, snapshot.records.length - order.length + 1) records.set(key, json) } const remove = raw.prepare('DELETE FROM activity_history_records WHERE record_key = ?') @@ -224,6 +228,19 @@ async function main() { const record = { ...records[(index + 2) % 50], detail: `incremental ${index}`, status: 'running' } return { changes: [{ type: 'upsert', position: 'front', record }] } }, input => database.updateActivityHistory(input)) + + // Startup: the renderer used to load the whole list; now the first page and the summary. + const firstPageRequest = { limit: 200 } + const summaryRequest = () => ({ conversationIds: [...new Set(database.getActivityHistoryPage(firstPageRequest).records.map(record => record.conversationId))] }) + results.activityFullLoad = await measure(async () => database.getActivityHistory()) + results.activityFirstPageSync = await measure(async () => database.getActivityHistoryPage(firstPageRequest)) + results.activitySummarySync = await measure(async () => database.getActivityHistorySummary(summaryRequest())) + results.activityStartupWorker = await measure(async () => { + const page = await database.getActivityHistoryPageAsync(firstPageRequest) + await database.getActivityHistorySummaryAsync({ conversationIds: [...new Set(page.records.map(record => record.conversationId))] }) + }) + results.activityFullLoadBytes = JSON.stringify(database.getActivityHistory()).length + results.activityFirstPageBytes = JSON.stringify(database.getActivityHistoryPage(firstPageRequest)).length database.close() console.log(`[mperf] seeded ${results.conversations} conversations / ${results.messages} messages (largest ${results.largestConversationMessages}), ${results.activityRecords} activity records in ${seedSeconds} s`) @@ -233,7 +250,10 @@ async function main() { [`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] + ['activity update, 1 record', results.replaceOneChangedLegacy, results.updateOneRecord], + ['activity startup: full list vs page+summary (worker)', results.activityFullLoad, results.activityStartupWorker], + ['activity first page (200), sync', results.activityFullLoad, results.activityFirstPageSync], + ['activity summary, sync', results.activityFullLoad, results.activitySummarySync] ] console.table(rows.map(([name, before, after]) => ({ path: name, @@ -242,6 +262,7 @@ async function main() { 'after call p50/p95 ms': `${after.callP50} / ${after.callP95}`, 'after loop max ms': after.loopMax }))) + console.log(`[mperf] activity payload: full list ${results.activityFullLoadBytes} bytes, first page ${results.activityFirstPageBytes} bytes`) const output = process.env.GB_MPERF_OUTPUT if (output) writeFileSync(output, JSON.stringify({ kind: 'goodbuddy-main-database-benchmark', node: process.version, results }, null, 2)) } diff --git a/src/main/assistant/activity-history-plan.test.ts b/src/main/assistant/activity-history-plan.test.ts index 48e125ed..a7c85747 100644 --- a/src/main/assistant/activity-history-plan.test.ts +++ b/src/main/assistant/activity-history-plan.test.ts @@ -8,7 +8,7 @@ import { z } from 'zod' import { activityHistorySnapshotSchema, type ActivityHistoryChange, type ActivityRecord } from '../../shared/assistant-contracts' import { applyActivityChanges } from '../../shared/activity-history-reference' import { AssistantDatabase } from './assistant-database' -import { parseActivityHistorySnapshot, planActivityHistoryReplace, planActivityHistoryUpdate } from './activity-history-plan' +import { parseActivityHistorySnapshot, planActivityHistoryReplace } from './activity-history-plan' const directories: string[] = [] const opened: AssistantDatabase[] = [] @@ -29,8 +29,9 @@ function open(): { database: AssistantDatabase; raw: DatabaseSync; directory: st /** 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() + header: raw.prepare('SELECT legacy_history_may_be_incomplete FROM activity_history').all(), + // Sequence values may differ; their order may not. + rows: raw.prepare('SELECT record_key, record_json FROM activity_history_records ORDER BY order_seq DESC, record_key').all() } } @@ -112,8 +113,8 @@ describe('activity history incremental updates', () => { 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) + // One record row; no order list is rewritten. + expect(changes() - start).toBe(1) expect(calls).toBeLessThan(10) }) }) @@ -183,11 +184,11 @@ describe('activity history replace', () => { 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 keys = ['["a",0]', '["b",0]', '["a",1]'] + expect(reparsed.items.every((item, index) => item.record === first.next.entries.get(keys[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 index 3d76411d..c15753e5 100644 --- a/src/main/assistant/activity-history-plan.ts +++ b/src/main/assistant/activity-history-plan.ts @@ -1,33 +1,28 @@ 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. +// PERF-15: planning of a full activity history `replace` (legacy migration). +// These pure helpers validate and diff against the last committed rows +// without re-validating or re-serializing unchanged records. Incremental +// changes are applied by ActivityHistoryRepository directly in SQL. /** A stored row; `record` is a validated, privately owned copy when known. */ -export type ActivityHistoryEntry = { json: string; record?: ActivityRecord } +export type ActivityHistoryEntry = { json: string; seq: number; 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]> + next: { entries: Map; incomplete: boolean } + upserts: Array<[key: string, json: string, seq: number]> removes: string[] - /** Order or legacy flag changed: the singleton row must be updated. */ + /** The legacy flag changed: the singleton row must be updated. */ headerChanged: boolean } @@ -57,12 +52,6 @@ function sameValue(raw: unknown, validated: unknown): boolean { 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 @@ -106,106 +95,47 @@ export function parseActivityHistorySnapshot( return { items, incomplete: raw.legacyHistoryMayBeIncomplete } } -/** Diffs a full newest-first list against the committed state. */ +/** + * Diffs a full newest-first list against the committed rows. Order is kept + * as `order_seq` (newer = larger): walking from the oldest record, a row + * keeps its sequence when it is still above the previous one, otherwise it + * gets the next one. Prepending or moving one record to the front therefore + * writes only that record. + */ 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]> = [] + const keys: string[] = new Array(items.length) 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) + keys[index] = items[index]!.key ?? activityRecordKey(record.id, occurrence) + } + const entries = new Map() + const upserts: Array<[string, string, number]> = [] + let floor = 0 + for (let index = items.length - 1, processed = 0; index >= 0; index--, processed++) { + if (processed % 128 === 0) onProgress?.(processed) + const { record } = items[index]! + const key = keys[index]! 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 before = previous.entries.get(key) + const seq = before && before.seq > floor ? before.seq : floor + 1 + floor = seq + if (before?.json !== json || before.seq !== seq) upserts.push([key, json, seq]) + entries.set(key, { json, seq, 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 }, + next: { entries, incomplete }, upserts, removes, - headerChanged: !orderUnchanged || previous.incomplete !== incomplete + headerChanged: 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-duplicates') { - const isDuplicate = (key: string): boolean => (JSON.parse(key) as [string, number])[1] !== 0 - order = order.filter(key => !isDuplicate(key)) - for (const key of [...entries.keys()]) { - if (isDuplicate(key)) { entries.delete(key); touched.add(key) } - } - continue - } - 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/activity-history-repository.test.ts b/src/main/assistant/activity-history-repository.test.ts new file mode 100644 index 00000000..9bd65e51 --- /dev/null +++ b/src/main/assistant/activity-history-repository.test.ts @@ -0,0 +1,203 @@ +// @vitest-environment node +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join, resolve } from 'node:path' +import { DatabaseSync } from 'node:sqlite' +import { build } from 'esbuild' +import { afterEach, describe, expect, it } from 'vitest' +import type { ActivityRecord, AssistantTask } from '../../shared/assistant-contracts' +import { + firstOccurrences, + referenceActivityPage, + referenceActivityReconcile, + referenceActivitySummary +} from '../../shared/activity-history-reference' +import { AssistantDatabase } from './assistant-database' + +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; path: string; directory: string } { + const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-activity-repository-')) + directories.push(directory) + const path = join(directory, 'assistant.sqlite') + const database = new AssistantDatabase(path) + database.initialize(directory) + opened.push(database) + return { database, raw: (database as unknown as { database: DatabaseSync }).database, path, directory } +} + +function makeRandom(seed: number): () => number { + let state = seed + return () => (state = (state * 1664525 + 1013904223) >>> 0) / 4294967296 +} + +const statuses = ['pending', 'running', 'completed', 'failed', 'denied', 'cancelled', 'interrupted'] as const +const kinds = ['request', 'tool', 'approval', 'subagent', 'result'] as const + +/** A random newest-first list with duplicate IDs and createdAt ties. */ +function randomList(random: () => number, length: number): ActivityRecord[] { + const pick = (items: readonly T[]): T => items[Math.floor(random() * items.length)]! + const list: ActivityRecord[] = [] + for (let index = 0; index < length; index++) { + const duplicate = list.length > 0 && random() < 0.08 + const id = duplicate ? pick(list).id : `record-${index}` + const project = random() < 0.5 + list.push({ + id, + conversationId: `conversation-${Math.floor(random() * 6)}`, + requestId: `request-${Math.floor(random() * 10)}`, + scope: project ? { kind: 'project', projectId: `project-${Math.floor(random() * 2)}`, projectName: 'P' } : { kind: 'global' }, + kind: pick(kinds), + title: `Title ${index}`, + detail: `detail ${index}`, + status: pick(statuses), + createdAt: Math.floor(random() * 40) + }) + } + return list +} + +describe('ActivityHistoryRepository', () => { + it('pages, counts and summarizes exactly like the full-list computation', () => { + for (const seed of [1, 7, 99]) { + const random = makeRandom(seed) + const { database } = open() + const list = randomList(random, 300) + database.replaceActivityHistory({ records: list, legacyHistoryMayBeIncomplete: seed === 7 }) + const conversationIds = Array.from({ length: 7 }, (_, index) => `conversation-${index}`) + expect(database.getActivityHistorySummary({ conversationIds })) + .toEqual(referenceActivitySummary(list, conversationIds, seed === 7)) + for (const filter of ['all', 'active', 'failed'] as const) { + for (const scope of [{}, { projectId: 'project-1' }, { conversationId: 'conversation-2' }]) { + const pages: ActivityRecord[] = [] + let before: number | undefined + do { + const page = database.getActivityHistoryPage({ limit: 37, filter, ...scope, ...(before === undefined ? {} : { before }) }) + pages.push(...page.records) + before = page.nextBefore + } while (before !== undefined) + const expected = referenceActivityPage(list, { limit: 500, filter, ...scope }) + expect(pages).toEqual(expected.records) + } + } + expect(database.getActivityHistoryPage({ limit: 500 }).records).toEqual(firstOccurrences(list)) + } + }) + + it('reconciles with the stored tasks like the former renderer code', () => { + const { database } = open() + const random = makeRandom(5) + const list = randomList(random, 200) + database.replaceActivityHistory({ records: list, legacyHistoryMayBeIncomplete: false }) + const statusesByRequest: Array = + ['completed', 'failed', 'cancelled', 'interrupted', 'running', 'waiting_approval', 'queued', undefined, 'completed', undefined] + const tasks: Array> = [] + statusesByRequest.forEach((status, index) => { + if (!status) return + const task = database.createTask({ id: `request-${index}`, title: 't', instructions: 'i', workMode: 'ask' }) + database.updateTaskStatus(task.id, status) + tasks.push({ id: task.id, status }) + }) + const active = new Set(['request-8', 'request-9']) + const changed = database.reconcileActivityHistory({ activeRequestIds: [...active], interruptedDetail: '已在重启时中断' }) + const expected = referenceActivityReconcile(list, tasks, active, '已在重启时中断') + expect(database.getActivityHistory().records).toEqual(expected) + expect(changed).toBe(expected.filter((record, index) => record !== list[index]).length) + expect(database.reconcileActivityHistory({ activeRequestIds: [...active], interruptedDetail: 'x' })).toBe(0) + }) + + it('moves a record to the front without rewriting other rows', () => { + const { database, raw } = open() + const list = randomList(makeRandom(3), 1_000) + database.replaceActivityHistory({ records: list, legacyHistoryMayBeIncomplete: false }) + const total = (): number => (raw.prepare('SELECT total_changes() AS n').get() as { n: number }).n + const start = total() + const target = firstOccurrences(list)[600]! + database.updateActivityHistory({ changes: [{ type: 'upsert', position: 'front', record: { ...target, status: 'completed' } }] }) + expect(total() - start).toBe(1) + expect(database.getActivityHistoryPage({ limit: 1 }).records[0]).toEqual({ ...target, status: 'completed' }) + // A replace that prepends one record writes one row. + const current = database.getActivityHistory().records + const before = total() + database.replaceActivityHistory({ records: [{ ...list[0]!, id: 'fresh' }, ...current], legacyHistoryMayBeIncomplete: false }) + expect(total() - before).toBe(1) + }) + + it('reads pages by index, without scanning arrays', () => { + const { raw } = open() + for (const sql of [ + 'SELECT record_json FROM activity_history_records WHERE occurrence = 0 AND order_seq < 5 ORDER BY order_seq DESC LIMIT 10', + "SELECT record_json FROM activity_history_records WHERE occurrence = 0 AND status_group = 'failed' ORDER BY order_seq DESC LIMIT 10", + "SELECT record_json FROM activity_history_records WHERE occurrence = 0 AND conversation_id = 'c' ORDER BY order_seq DESC LIMIT 10" + ]) { + const plan = (raw.prepare(`EXPLAIN QUERY PLAN ${sql}`).all() as Array<{ detail: string }>).map((row) => row.detail).join('; ') + expect(plan).toMatch(/USING INDEX/) + expect(plan).not.toMatch(/TEMP B-TREE|json_each/) + } + }) + + it('migrates the stored order exactly, keeping the first occurrence of duplicates shown', () => { + const { database, raw, path, directory } = open() + const list = randomList(makeRandom(11), 120) + database.replaceActivityHistory({ records: list, legacyHistoryMayBeIncomplete: true }) + database.close() + // Rebuild the schema-56 layout: order list in record_order_json, no order columns. + const legacy = new DatabaseSync(path) + const rows = legacy.prepare( + 'SELECT record_key, record_json FROM activity_history_records ORDER BY order_seq DESC, record_key' + ).all() as Array<{ record_key: string; record_json: string }> + legacy.exec(` + DROP TABLE activity_history_records; + CREATE TABLE activity_history_records (record_key TEXT PRIMARY KEY NOT NULL, record_json TEXT NOT NULL); + `) + const insert = legacy.prepare('INSERT INTO activity_history_records VALUES (?, ?)') + for (const row of rows) insert.run(row.record_key, row.record_json) + insert.run(JSON.stringify(['orphan', 0]), JSON.stringify({ ...list[0], id: 'orphan' })) + legacy.prepare("UPDATE activity_history SET record_order_json = ? WHERE singleton = 1") + .run(JSON.stringify(rows.map((row) => row.record_key))) + legacy.exec('PRAGMA user_version = 56') + legacy.close() + + const reopened = new AssistantDatabase(path) + reopened.initialize(directory) + opened.push(reopened) + expect(reopened.getActivityHistory()).toEqual({ records: list, legacyHistoryMayBeIncomplete: true }) + expect(reopened.getActivityHistoryPage({ limit: 500 }).records).toEqual(firstOccurrences(list)) + const check = (reopened as unknown as { database: DatabaseSync }).database + expect(check.prepare("SELECT record_order_json FROM activity_history").get()).toEqual({ record_order_json: '[]' }) + void raw + }) + + it('serves pages and the summary from the readonly worker with the same results', async () => { + const { database, directory } = open() + const 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' }) + const list = randomList(makeRandom(21), 150) + database.replaceActivityHistory({ records: list, legacyHistoryMayBeIncomplete: false }) + const pageRequest = { limit: 40, filter: 'failed' as const } + const summaryRequest = { conversationIds: ['conversation-1', 'conversation-4'] } + const expectedPage = database.getActivityHistoryPage(pageRequest) + const expectedSummary = database.getActivityHistorySummary(summaryRequest) + database.enableReadonlyWorker(workerPath) + expect(await database.getActivityHistoryPageAsync(pageRequest)).toEqual(expectedPage) + expect(await database.getActivityHistorySummaryAsync(summaryRequest)).toEqual(expectedSummary) + // A committed write is visible to the next worker read. + database.updateActivityHistory({ changes: [{ type: 'upsert', position: 'front', record: { ...list[0]!, id: 'fresh', status: 'failed' } }] }) + expect((await database.getActivityHistoryPageAsync(pageRequest)).records[0]!.id).toBe('fresh') + await expect(database.getActivityHistoryPageAsync({ limit: 501 })).rejects.toThrow() + }, 60_000) + + it('clears everything in one call', () => { + const { database } = open() + database.replaceActivityHistory({ records: randomList(makeRandom(2), 50), legacyHistoryMayBeIncomplete: true }) + database.clearActivityHistory() + expect(database.getActivityHistory()).toEqual({ records: [], legacyHistoryMayBeIncomplete: false }) + expect(database.getActivityHistorySummary({ conversationIds: [] }).counts).toEqual({ all: 0, active: 0, failed: 0 }) + }) +}) diff --git a/src/main/assistant/activity-history-repository.ts b/src/main/assistant/activity-history-repository.ts new file mode 100644 index 00000000..5523ad37 --- /dev/null +++ b/src/main/assistant/activity-history-repository.ts @@ -0,0 +1,418 @@ +import { + activityHistoryPageRequestSchema, + activityHistoryReconcileRequestSchema, + activityHistorySnapshotSchema, + activityHistorySummaryRequestSchema, + activityHistoryUpdateSchema, + activityRecordSchema, + type ActivityHistoryPage, + type ActivityHistorySnapshot, + type ActivityHistorySummary, + type ActivityRecord +} from '../../shared/assistant-contracts' +import { + activityRecordKey, + isActivityHistoryPlanEmpty, + parseActivityHistorySnapshot, + planActivityHistoryReplace, + type ActivityHistoryPlan, + type ActivityHistoryState +} from './activity-history-plan' + +/** + * The connection the repository runs on; AssistantDatabase owns and passes + * it, so this module does not import node:sqlite itself (PERF-16 allowlist). + */ +type SqlValue = string | number | bigint | null | Uint8Array +type Connection = { + readonly isTransaction: boolean + exec: (sql: string) => void + prepare: (sql: string) => { + all: (...args: SqlValue[]) => unknown[] + get: (...args: SqlValue[]) => unknown + run: (...args: SqlValue[]) => { changes: number | bigint } + } +} + +/** + * Activity history storage (PERF-15). Every record is one row of + * activity_history_records: + * - `record_key` = JSON `[id, occurrence]`: legacy history may hold an ID more + * than once; only occurrence 0 (the first in list order) is shown. + * - `order_seq`: list position, newer = larger. A move to the front assigns a + * new maximum, so no order list is rewritten. + * - `record_id`, `occurrence`, `status`, `status_group`, `kind`, + * `request_id`, `conversation_id`, `project_id`, `created_at`: stored + * columns generated from the key and JSON, indexed for pages, counts and + * updates; they cannot disagree with the record. + * The singleton activity_history row keeps only the legacy-incomplete flag. + */ +const recordsTableSql = (name: string): string => ` + CREATE TABLE ${name} ( + record_key TEXT PRIMARY KEY NOT NULL, + record_json TEXT NOT NULL, + order_seq INTEGER NOT NULL, + record_id TEXT GENERATED ALWAYS AS (json_extract(record_key, '$[0]')) STORED, + occurrence INTEGER GENERATED ALWAYS AS (json_extract(record_key, '$[1]')) STORED, + status TEXT GENERATED ALWAYS AS (json_extract(record_json, '$.status')) STORED, + status_group TEXT GENERATED ALWAYS AS (CASE json_extract(record_json, '$.status') + WHEN 'pending' THEN 'active' WHEN 'running' THEN 'active' + WHEN 'completed' THEN 'other' ELSE 'failed' END) STORED, + kind TEXT GENERATED ALWAYS AS (json_extract(record_json, '$.kind')) STORED, + request_id TEXT GENERATED ALWAYS AS (json_extract(record_json, '$.requestId')) STORED, + conversation_id TEXT GENERATED ALWAYS AS (json_extract(record_json, '$.conversationId')) STORED, + project_id TEXT GENERATED ALWAYS AS (json_extract(record_json, '$.scope.projectId')) STORED, + created_at REAL GENERATED ALWAYS AS (json_extract(record_json, '$.createdAt')) STORED + )` + +/** Schema 57: order and indexed columns; replaces the record_order_json list. */ +export function migrateActivityHistoryOrder(database: Connection): void { + const columns = database.prepare( + "SELECT name FROM pragma_table_xinfo('activity_history_records')" + ).all() as Array<{ name: string }> + // Re-running the migration (a lowered user_version) changes nothing. + if (columns.some((column) => column.name === 'status_group')) return + // Tables created by the schema-44 migration of this version already carry the order. + const hasOrder = columns.some((column) => column.name === 'order_seq') + const header = database.prepare( + 'SELECT record_order_json FROM activity_history WHERE singleton = 1' + ).get() as { record_order_json: string } | undefined + database.exec(recordsTableSql('activity_history_records_v57')) + if (hasOrder) { + database.exec(`INSERT INTO activity_history_records_v57 (record_key, record_json, order_seq) + SELECT record_key, record_json, order_seq FROM activity_history_records`) + } else { + // The stored order lists every row once; rows outside it were never shown. + database.prepare(` + INSERT INTO activity_history_records_v57 (record_key, record_json, order_seq) + SELECT records.record_key, records.record_json, json_array_length(?1) - CAST(ordering.key AS INTEGER) + FROM json_each(?1) AS ordering + JOIN activity_history_records AS records ON records.record_key = ordering.value + `).run(header?.record_order_json ?? '[]') + } + database.exec(` + DROP TABLE activity_history_records; + ALTER TABLE activity_history_records_v57 RENAME TO activity_history_records; + UPDATE activity_history SET record_order_json = '[]' WHERE singleton = 1; + CREATE INDEX activity_history_records_order ON activity_history_records(occurrence, order_seq); + CREATE INDEX activity_history_records_group ON activity_history_records(occurrence, status_group, order_seq); + CREATE INDEX activity_history_records_conversation ON activity_history_records(conversation_id, occurrence, + order_seq, kind, status, request_id, created_at, record_key); + CREATE INDEX activity_history_records_project ON activity_history_records(project_id, occurrence, order_seq); + CREATE INDEX activity_history_records_request ON activity_history_records(request_id, kind); + CREATE INDEX activity_history_records_id ON activity_history_records(record_id); + `) +} +type SummaryRow = { + conversation_id: string + kind: ActivityRecord['kind'] + status: ActivityRecord['status'] + request_id: string + created_at: number + record_key: string +} + +/** + * Title (first request in list order) and status (the latest request's + * latest result, else the latest result, else the latest record) of one + * conversation; `rows` are newest first, so createdAt ties resolve to the + * record nearer the front, as the former full-list computation did. + */ +function summarizeConversation( + rows: readonly SummaryRow[], + titles: ReadonlyMap +): { title?: string; status: ActivityRecord['status'] } { + const latest = (match: (row: SummaryRow) => boolean): SummaryRow | undefined => { + let best: SummaryRow | undefined + for (const row of rows) if (match(row) && (!best || row.created_at > best.created_at)) best = row + return best + } + const firstRequest = rows.find((row) => row.kind === 'request') + const title = firstRequest ? titles.get(firstRequest.record_key) : undefined + const request = latest((row) => row.kind === 'request') + const status = request + ? (latest((row) => row.kind === 'result' && row.request_id === request.request_id)?.status ?? request.status) + : (latest((row) => row.kind === 'result')?.status ?? latest(() => true)?.status ?? 'completed') + return { ...(title === undefined ? {} : { title }), status } +} + +function parseRow(json: string): ActivityRecord { + return activityRecordSchema.parse(JSON.parse(json)) +} + +export class ActivityHistoryRepository { + /** Validated rows of the last committed replace, reused by the next one. */ + private replaceCache?: { state: ActivityHistoryState; dataVersion: number } + + constructor(private readonly database: Connection) {} + + /** Drops cached rows after writes that bypass the repository. */ + invalidate(): void { + this.replaceCache = undefined + } + + /** The whole stored list, every occurrence, newest first. */ + snapshot(): ActivityHistorySnapshot { + const rows = this.database.prepare( + 'SELECT record_json FROM activity_history_records ORDER BY order_seq DESC, record_key' + ).all() as Array<{ record_json: string }> + return activityHistorySnapshotSchema.parse({ + records: rows.map((row) => JSON.parse(row.record_json) as unknown), + legacyHistoryMayBeIncomplete: this.readIncomplete() + }) + } + + /** One page of shown records, newest first, by index. */ + page(input: unknown): ActivityHistoryPage { + const request = activityHistoryPageRequestSchema.parse(input) + const where = ['occurrence = 0'] + const args: Array = [] + if (request.before !== undefined) { + where.push('order_seq < ?') + args.push(request.before) + } + if (request.filter && request.filter !== 'all') { + where.push('status_group = ?') + args.push(request.filter) + } + if (request.projectId) { + where.push('project_id = ?') + args.push(request.projectId) + } + if (request.conversationId) { + where.push('conversation_id = ?') + args.push(request.conversationId) + } + const rows = this.database.prepare( + `SELECT order_seq, record_json FROM activity_history_records + WHERE ${where.join(' AND ')} ORDER BY order_seq DESC LIMIT ?` + ).all(...args, request.limit + 1) as Array<{ order_seq: number; record_json: string }> + const more = rows.length > request.limit + const shown = more ? rows.slice(0, request.limit) : rows + return { + records: shown.map((row) => parseRow(row.record_json)), + ...(more ? { nextBefore: shown.at(-1)!.order_seq } : {}) + } + } + + /** What the Activity page derives from the whole history. */ + summary(input: unknown): ActivityHistorySummary { + const request = activityHistorySummaryRequestSchema.parse(input) + const groupRows = this.database.prepare( + `SELECT status_group AS status_group, COUNT(*) AS count FROM activity_history_records + WHERE occurrence = 0 GROUP BY status_group` + ).all() as Array<{ status_group: 'active' | 'failed' | 'other'; count: number }> + const counts = { all: 0, active: 0, failed: 0 } + for (const { status_group: group, count } of groupRows) { + counts.all += count + if (group !== 'other') counts[group] += count + } + // One indexed read of the requested conversations' rows, newest first. + const rows = this.database.prepare( + `SELECT conversation_id, kind, status, request_id, created_at, record_key + FROM activity_history_records + WHERE conversation_id IN (SELECT value FROM json_each(?)) AND occurrence = 0 + ORDER BY conversation_id, order_seq DESC` + ).all(JSON.stringify([...new Set(request.conversationIds)])) as SummaryRow[] + const byConversation = new Map() + for (const row of rows) { + const own = byConversation.get(row.conversation_id) + if (own) own.push(row) + else byConversation.set(row.conversation_id, [row]) + } + // Titles come from the first request row of each conversation only. + const titleKeys = [...byConversation.values()].flatMap((own) => { + const first = own.find((row) => row.kind === 'request') + return first ? [first.record_key] : [] + }) + const titles = new Map((this.database.prepare( + `SELECT record_key, json_extract(record_json, '$.title') AS title FROM activity_history_records + WHERE record_key IN (SELECT value FROM json_each(?))` + ).all(JSON.stringify(titleKeys)) as Array<{ record_key: string; title: string }>) + .map((row) => [row.record_key, row.title])) + const conversations: ActivityHistorySummary['conversations'] = {} + for (const conversationId of new Set(request.conversationIds)) { + conversations[conversationId] = summarizeConversation(byConversation.get(conversationId) ?? [], titles) + } + return { counts, conversations, legacyHistoryMayBeIncomplete: this.readIncomplete() } + } + + /** + * Ends pending/running records whose request is not running in this + * window: without a task they become interrupted (with the given localized + * line appended); with a finished task they take its status, except that + * tool, approval and subagent records of a completed task become + * interrupted. Returns the number of changed records. + */ + reconcile(input: unknown): number { + const request = activityHistoryReconcileRequestSchema.parse(input) + return this.write(() => { + const result = this.database.prepare(` + WITH candidates AS ( + SELECT records.record_key, + CASE + WHEN tasks.id IS NULL THEN 'interrupted' + WHEN tasks.status IN ('completed', 'failed', 'cancelled', 'interrupted') THEN tasks.status + END AS terminal + FROM activity_history_records AS records + LEFT JOIN tasks ON tasks.id = records.request_id AND tasks.visible = 1 + WHERE records.occurrence = 0 AND records.status_group = 'active' + AND records.request_id NOT IN (SELECT value FROM json_each(?1)) + ) + UPDATE activity_history_records SET record_json = json_set(record_json, + '$.status', CASE + WHEN candidates.terminal = 'completed' AND activity_history_records.kind IN ('tool', 'approval', 'subagent') + THEN 'interrupted' + ELSE candidates.terminal END, + '$.detail', CASE + WHEN candidates.terminal = 'interrupted' + THEN json_extract(record_json, '$.detail') || char(10) || ?2 + ELSE json_extract(record_json, '$.detail') END) + FROM candidates + WHERE activity_history_records.record_key = candidates.record_key AND candidates.terminal IS NOT NULL + `).run(JSON.stringify(request.activeRequestIds), request.interruptedDetail) + return Number(result.changes) + }) + } + + /** Applies incremental changes in one transaction, without reading the history. */ + update(input: unknown): void { + const update = activityHistoryUpdateSchema.parse(input) + this.write(() => { + const database = this.database + const nextSeq = database.prepare( + 'SELECT COALESCE(MAX(order_seq), 0) + 1 AS seq FROM activity_history_records' + ) + const exists = database.prepare('SELECT 1 FROM activity_history_records WHERE record_key = ?') + const insert = database.prepare( + 'INSERT INTO activity_history_records (record_key, record_json, order_seq) VALUES (?, ?, ?)' + ) + const moveToFront = database.prepare( + 'UPDATE activity_history_records SET record_json = ?, order_seq = ? WHERE record_key = ?' + ) + const replaceInPlace = database.prepare( + 'UPDATE activity_history_records SET record_json = ? WHERE record_key = ? AND record_json != ?' + ) + const remove = database.prepare('DELETE FROM activity_history_records WHERE record_id = ?') + for (const change of update.changes) { + if (change.type === 'remove-duplicates') { + database.exec('DELETE FROM activity_history_records WHERE occurrence != 0') + } else if (change.type === 'remove') { + remove.run(change.id) + } else { + // The first occurrence stays first when moved to the front, so its key is stable. + const key = activityRecordKey(change.record.id, 0) + const json = JSON.stringify(change.record) + const seq = (): number => (nextSeq.get() as { seq: number }).seq + if (!exists.get(key)) insert.run(key, json, seq()) + else if (change.position === 'front') moveToFront.run(json, seq(), key) + else replaceInPlace.run(json, key, json) + } + } + if (update.legacyHistoryMayBeIncomplete !== undefined) { + this.writeIncomplete(update.legacyHistoryMayBeIncomplete) + } + }) + } + + /** + * Persists a whole newest-first list (legacy migration). Unchanged records + * are neither re-validated, re-serialized nor written; a fully unchanged + * snapshot does not open a write transaction. + */ + replace(input: unknown): void { + // Reuse is safe even from a stale cache: cached records are validated copies. + const { items, incomplete } = parseActivityHistorySnapshot(input, this.replaceCache?.state) + const committed = this.committedState() + const planned = committed && planActivityHistoryReplace(items, incomplete, committed) + if (planned && isActivityHistoryPlanEmpty(planned)) return + let result: ActivityHistoryPlan | undefined + this.write(() => { + const previous = committed && planned ? committed : this.readState() + result = previous === committed && planned ? planned : planActivityHistoryReplace(items, incomplete, previous) + this.applyPlan(result) + }) + // Publish only committed records; a failed save must remain retryable. + this.replaceCache = { state: result!.next, dataVersion: this.dataVersion() } + } + + /** Writes a parsed legacy snapshot into empty tables (schema migrations). */ + writeSnapshot(snapshot: ActivityHistorySnapshot, onProgress?: (processed: number) => void): void { + this.applyPlan(planActivityHistoryReplace( + snapshot.records.map((record) => ({ record })), snapshot.legacyHistoryMayBeIncomplete, + { entries: new Map(), incomplete: !snapshot.legacyHistoryMayBeIncomplete }, onProgress)) + } + + /** Deletes the whole history in one transaction. */ + clear(): void { + this.write(() => { + this.database.exec('DELETE FROM activity_history_records') + this.writeIncomplete(false) + }) + } + + private write(body: () => T): T { + const database = this.database + this.replaceCache = undefined + if (database.isTransaction) return body() + database.exec('BEGIN IMMEDIATE') + try { + const result = body() + database.exec('COMMIT') + return result + } catch (error) { + database.exec('ROLLBACK') + throw error + } + } + + private dataVersion(): number { + return (this.database.prepare('PRAGMA data_version').get() as { data_version: number }).data_version + } + + /** The cached committed state, when no other connection committed since. */ + private committedState(): ActivityHistoryState | undefined { + const cache = this.replaceCache + if (!cache || this.database.isTransaction) return undefined + return cache.dataVersion === this.dataVersion() ? cache.state : undefined + } + + private readState(): ActivityHistoryState { + const rows = this.database.prepare( + 'SELECT record_key, record_json, order_seq FROM activity_history_records' + ).all() as Array<{ record_key: string; record_json: string; order_seq: number }> + return { + entries: new Map(rows.map((row) => [row.record_key, { json: row.record_json, seq: row.order_seq }])), + incomplete: this.readIncomplete() + } + } + + private applyPlan(plan: ActivityHistoryPlan): void { + const database = this.database + 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.upserts.length > 0) { + const upsert = database.prepare( + `INSERT INTO activity_history_records (record_key, record_json, order_seq) VALUES (?, ?, ?) + ON CONFLICT(record_key) DO UPDATE SET record_json = excluded.record_json, order_seq = excluded.order_seq` + ) + for (const [key, json, seq] of plan.upserts) upsert.run(key, json, seq) + } + if (plan.headerChanged) this.writeIncomplete(plan.next.incomplete) + } + + private readIncomplete(): boolean { + const row = this.database.prepare( + 'SELECT legacy_history_may_be_incomplete AS value FROM activity_history WHERE singleton = 1' + ).get() as { value: number } | undefined + return row?.value === 1 + } + + private writeIncomplete(value: boolean): void { + this.database.prepare( + 'UPDATE activity_history SET legacy_history_may_be_incomplete = ? WHERE singleton = 1' + ).run(Number(value)) + } +} diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index 42b2d158..db5d3d7c 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -6,14 +6,7 @@ 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 { ActivityHistoryRepository, migrateActivityHistoryOrder } from './activity-history-repository' import { statSync } from 'node:fs' import { MagicNoteStorage } from '../magic-notes/magic-note-storage' import type { AssistantStorageProgress } from '../../shared/assistant-storage-contracts' @@ -45,7 +38,9 @@ import type { AssistantTask, ExecutionStats, ExecutionStatsInput, + ActivityHistoryPage, ActivityHistorySnapshot, + ActivityHistorySummary, ConversationQueueItem, ConversationAnsweredQuestion, ConversationBranchInput, @@ -142,7 +137,7 @@ import { SUBAGENT_PROGRESS_STORAGE_SCHEMA_VERSION } from './subagent-progress-storage' -export const ASSISTANT_DATABASE_SCHEMA_VERSION = 56 +export const ASSISTANT_DATABASE_SCHEMA_VERSION = 57 export type RemoteTaskEventInput = { taskId: string @@ -1974,7 +1969,7 @@ export class AssistantDatabase { private static readonly executionStatsCacheTtlMs = 30_000 private static readonly executionStatsCacheLimit = 8 private database?: DatabaseSync - private activityHistoryCache?: { state: ActivityHistoryState; dataVersion: number } + private activityHistoryRepository?: { database: DatabaseSync; repository: ActivityHistoryRepository } private executionStatsReader?: ExecutionStatsReader private readonlyReader?: ReadonlyQueryReader private readonlyWorkerPath?: string @@ -2412,7 +2407,7 @@ export class AssistantDatabase { this.readonlyReader = undefined this.foldedSearchConnection = undefined this.executionStatsCache.clear() - this.activityHistoryCache = undefined + this.activityHistoryRepository = undefined this.database?.close() this.database = undefined this.subagentProgress = undefined @@ -2472,7 +2467,7 @@ export class AssistantDatabase { throw error } this.dirtyMagicNotes.clear() - this.activityHistoryCache = undefined + this.activityHistoryRepository?.repository.invalidate() this.pendingMagicNoteCleanup.clear() this.reconcileMagicNoteFiles() this.options.onMagicNotesChanged?.() @@ -5295,154 +5290,59 @@ export class AssistantDatabase { return result } - getActivityHistory(): ActivityHistorySnapshot { + /** Activity history storage and queries (PERF-15); the SQL lives in the repository. */ + activityHistory(): ActivityHistoryRepository { const database = this.requireDatabase() - const row = 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 records = database.prepare( - `SELECT records.record_json - FROM json_each(?) AS ordering - JOIN activity_history_records AS records ON records.record_key = ordering.value - ORDER BY CAST(ordering.key AS INTEGER)` - ).all(row?.record_order_json ?? '[]') as Array<{ record_json: string }> - return activityHistorySnapshotSchema.parse({ - records: records.map((record) => JSON.parse(record.record_json)), - legacyHistoryMayBeIncomplete: - row?.legacy_history_may_be_incomplete === 1 - }) + if (this.activityHistoryRepository?.database !== database) { + this.activityHistoryRepository = { database, repository: new ActivityHistoryRepository(database) } + } + return this.activityHistoryRepository.repository } - /** - * 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). - */ + /** The whole stored list (tests, diagnostics); the renderer reads pages. */ + getActivityHistory(): ActivityHistorySnapshot { + return this.activityHistory().snapshot() + } + + /** Persists a whole newest-first list; used for the legacy localStorage migration. */ 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)) + this.activityHistory().replace(input) } /** Applies incremental changes in one transaction (PERF-15). */ updateActivityHistory(input: unknown): void { - const database = this.requireDatabase() - this.writeActivityHistoryPlan(database, (previous) => planActivityHistoryUpdate(input, previous)) + this.activityHistory().update(input) } /** Deletes every activity record in one transaction. */ clearActivityHistory(): void { + this.activityHistory().clear() + } + + getActivityHistoryPage(input: unknown): ActivityHistoryPage { + return this.activityHistory().page(input) + } + + getActivityHistorySummary(input: unknown): ActivityHistorySummary { + return this.readSnapshot(() => this.activityHistory().summary(input)) + } + + /** Pages and the summary on the readonly worker; same rules as listConversationSummariesAsync. */ + getActivityHistoryPageAsync(input: unknown): Promise { const database = this.requireDatabase() - database.exec('BEGIN IMMEDIATE') - try { - database.exec('DELETE FROM activity_history_records') - database.exec( - `UPDATE activity_history - SET record_order_json = '[]', legacy_history_may_be_incomplete = 0 - WHERE singleton = 1` - ) - database.exec('COMMIT') - } catch (error) { - database.exec('ROLLBACK') - throw error - } - this.activityHistoryCache = undefined + return readWithFallback(database.isTransaction ? undefined : this.readonlyQueryReader(), + 'activityHistoryPage', [input], undefined, () => this.getActivityHistoryPage(input)) } - /** 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 + getActivityHistorySummaryAsync(input: unknown): Promise { + const database = this.requireDatabase() + return readWithFallback(database.isTransaction ? undefined : this.readonlyQueryReader(), + 'activityHistorySummary', [input], undefined, () => this.getActivityHistorySummary(input)) } - 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.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 = { 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 - ): 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 }])) - } - this.applyActivityHistoryPlan(database, planActivityHistoryReplace( - snapshot.records.map((record) => ({ record })), snapshot.legacyHistoryMayBeIncomplete, previous, onProgress)) + /** Ends records left running by requests that are no longer active (see the repository). */ + reconcileActivityHistory(input: unknown): number { + return this.activityHistory().reconcile(input) } createTask(input: { @@ -12088,11 +11988,13 @@ export class AssistantDatabase { database.exec(` CREATE TABLE activity_history_records ( record_key TEXT PRIMARY KEY NOT NULL, - record_json TEXT NOT NULL + record_json TEXT NOT NULL, + order_seq INTEGER NOT NULL DEFAULT 0 ); ALTER TABLE activity_history RENAME COLUMN records_json TO record_order_json; `) - this.writeActivityHistory(database, snapshot, onProgress || isCancelled ? (processed) => { + // Writes order_seq directly; schema 57 then only adds the indexed columns. + new ActivityHistoryRepository(database).writeSnapshot(snapshot, onProgress || isCancelled ? (processed) => { onProgress?.({ stage: 'converting', processed, total: snapshot.records.length, bytesBefore }) if (isCancelled?.()) throw new DOMException('Upgrade cancelled', 'AbortError') } : undefined) @@ -12345,6 +12247,15 @@ export class AssistantDatabase { database.exec('PRAGMA user_version = 56; COMMIT;') } catch (error) { database.exec('ROLLBACK'); throw error } } + if (version.user_version < 57) { + database.exec('BEGIN IMMEDIATE') + try { + // Activity history order and indexed columns (PERF-15): pages read by index, + // and a move to the front no longer rewrites the order list. + migrateActivityHistoryOrder(database) + database.exec('PRAGMA user_version = 57; COMMIT;') + } catch (error) { database.exec('ROLLBACK'); throw error } + } } supervisionExperiences(): SupervisionExperienceStore { diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 20cc4e3b..7e318163 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -130,6 +130,9 @@ import { import { registerMagicNotesAnalysisIpcHandlers, registerMagicTodosAnalysisIpcHandlers } from './magic-notes/magic-notes-analysis-ipc' import { registerMagicNotesIpcHandlers, registerMagicTodosIpcHandlers } from './magic-notes/magic-notes-ipc' import { + activityHistoryPageRequestSchema, + activityHistoryReconcileRequestSchema, + activityHistorySummaryRequestSchema, assistantIdSchema, executionStatsInputSchema, conversationBranchInputSchema, @@ -6873,6 +6876,23 @@ export function registerIpcHandlers( assertTrustedSender(event, window) assistantDatabase.clearActivityHistory() }) + registerHandler(ipcChannels.activityHistoryPage, (event, input: unknown) => { + assertTrustedSender(event, window) + return assistantDatabase.getActivityHistoryPageAsync(activityHistoryPageRequestSchema.parse(input)) + }) + registerHandler(ipcChannels.activityHistorySummary, (event, input: unknown) => { + assertTrustedSender(event, window) + return assistantDatabase.getActivityHistorySummaryAsync(activityHistorySummaryRequestSchema.parse(input)) + }) + registerHandler(ipcChannels.activityHistoryReconcile, (event, input: unknown) => { + assertTrustedSender(event, window) + const request = activityHistoryReconcileRequestSchema.parse(input) + // Requests still running in Main are active too, whichever window started them. + return assistantDatabase.reconcileActivityHistory({ + ...request, + activeRequestIds: [...new Set([...request.activeRequestIds, ...activeRequests.keys()])] + }) + }) registerHandler(ipcChannels.tokenUsageSummary, (event) => { assertTrustedSender(event, window) diff --git a/src/main/readonly-query-worker.ts b/src/main/readonly-query-worker.ts index 1a7fdcf8..0197135d 100644 --- a/src/main/readonly-query-worker.ts +++ b/src/main/readonly-query-worker.ts @@ -36,7 +36,9 @@ if (kind === 'knowledge') { listConversationSummaries: ([detailIds]) => database.readSnapshot(() => database.listConversationSummaries(detailIds as string[])), getConversation: ([conversationId]) => database.readSnapshot(() => - database.getConversation(conversationId as string)) + database.getConversation(conversationId as string)), + activityHistoryPage: ([input]) => database.getActivityHistoryPage(input), + activityHistorySummary: ([input]) => database.getActivityHistorySummary(input) } } diff --git a/src/preload/index.ts b/src/preload/index.ts index c961dc24..18d2bd43 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -67,7 +67,12 @@ import type { AssistantExpert, AssistantTask, ExecutionStats, + ActivityHistoryPage, + ActivityHistoryPageRequest, + ActivityHistoryReconcileRequest, ActivityHistorySnapshot, + ActivityHistorySummary, + ActivityHistorySummaryRequest, ActivityHistoryUpdate, ActivityRecord, TokenUsageSummary, @@ -1296,7 +1301,13 @@ const desktopApi: DesktopApi = { }, clear: async () => { await ipcRenderer.invoke(ipcChannels.activityHistoryClear) - } + }, + page: (request: ActivityHistoryPageRequest) => + ipcRenderer.invoke(ipcChannels.activityHistoryPage, request) as Promise, + summary: (request: ActivityHistorySummaryRequest) => + ipcRenderer.invoke(ipcChannels.activityHistorySummary, request) as Promise, + reconcile: (request: ActivityHistoryReconcileRequest) => + ipcRenderer.invoke(ipcChannels.activityHistoryReconcile, request) as Promise }, usage: { getTokenSummary: () => diff --git a/src/renderer/src/App.test.tsx b/src/renderer/src/App.test.tsx index 7a5db312..0f6d7089 100644 --- a/src/renderer/src/App.test.tsx +++ b/src/renderer/src/App.test.tsx @@ -193,7 +193,11 @@ vi.mock("./TerminalPanel", () => ({ import App from "./App"; import { ACTIVITY_STORAGE_KEY, type ActivityRecord } from "./activity-store"; -import { applyActivityChanges } from "../../shared/activity-history-reference"; +import { + applyActivityChanges, + referenceActivityPage, + referenceActivitySummary, +} from "../../shared/activity-history-reference"; import type { ActivityHistoryUpdate } from "../../shared/assistant-contracts"; import { changeUiLocale } from "./i18n"; import { UiLocaleProvider } from "./i18n/UiLocaleProvider"; @@ -689,6 +693,10 @@ const api: DesktopApi & RuntimeNativeClientApi = { clear: vi.fn(async () => { savedActivity = []; }), + page: vi.fn(async (request) => referenceActivityPage(savedActivity, request)), + summary: vi.fn(async (request) => + referenceActivitySummary(savedActivity, request.conversationIds)), + reconcile: vi.fn(async () => 0), }, usage: { getTokenSummary: vi.fn(async () => ({ diff --git a/src/shared/activity-history-reference.ts b/src/shared/activity-history-reference.ts index bb4ada67..7f9c1972 100644 --- a/src/shared/activity-history-reference.ts +++ b/src/shared/activity-history-reference.ts @@ -1,8 +1,116 @@ import type { ActivityHistoryChange, - ActivityRecord + ActivityHistoryPage, + ActivityHistoryPageRequest, + ActivityHistorySummary, + ActivityRecord, + AssistantTask } from './assistant-contracts' +const isActive = (record: ActivityRecord): boolean => + record.status === 'pending' || record.status === 'running' +const isFailed = (record: ActivityRecord): boolean => + record.status === 'failed' || record.status === 'denied' || + record.status === 'cancelled' || record.status === 'interrupted' + +/** The records a list shows: the first occurrence of each ID. */ +export function firstOccurrences(list: readonly ActivityRecord[]): ActivityRecord[] { + const seen = new Set() + return list.filter((item) => !seen.has(item.id) && Boolean(seen.add(item.id))) +} + +/** + * Reference page over a plain newest-first list; a record's sequence is its + * distance from the end of the list (newer = larger), like `order_seq`. + */ +export function referenceActivityPage( + list: readonly ActivityRecord[], + request: ActivityHistoryPageRequest +): ActivityHistoryPage { + const seen = new Set() + const matching: Array<{ record: ActivityRecord; seq: number }> = [] + list.forEach((record, index) => { + if (seen.has(record.id)) return + seen.add(record.id) + const seq = list.length - index + if (request.before !== undefined && seq >= request.before) return + if (request.filter === 'active' && !isActive(record)) return + if (request.filter === 'failed' && !isFailed(record)) return + if (request.projectId && (record.scope.kind !== 'project' || record.scope.projectId !== request.projectId)) return + if (request.conversationId && record.conversationId !== request.conversationId) return + matching.push({ record, seq }) + }) + const shown = matching.slice(0, request.limit) + return { + records: shown.map((item) => item.record), + ...(matching.length > request.limit ? { nextBefore: shown.at(-1)!.seq } : {}) + } +} + +/** + * The former ActivityPanel computation over the whole list: counts, the + * first request title and the conversation status. + */ +export function referenceActivitySummary( + list: readonly ActivityRecord[], + conversationIds: readonly string[], + legacyHistoryMayBeIncomplete = false +): ActivityHistorySummary { + const records = firstOccurrences(list) + const latest = (candidates: readonly ActivityRecord[]): ActivityRecord | undefined => + candidates.reduce( + (best, record) => (!best || record.createdAt > best.createdAt ? record : best), undefined) + const conversations: ActivityHistorySummary['conversations'] = {} + for (const conversationId of conversationIds) { + const own = records.filter((record) => record.conversationId === conversationId) + const title = own.find((record) => record.kind === 'request')?.title + const latestRequest = latest(own.filter((record) => record.kind === 'request')) + const status = latestRequest + ? (latest(own.filter((record) => record.kind === 'result' && record.requestId === latestRequest.requestId)) + ?.status ?? latestRequest.status) + : (latest(own.filter((record) => record.kind === 'result'))?.status ?? latest(own)?.status ?? 'completed') + conversations[conversationId] = { ...(title === undefined ? {} : { title }), status } + } + return { + counts: { + all: records.length, + active: records.filter(isActive).length, + failed: records.filter(isFailed).length + }, + conversations, + legacyHistoryMayBeIncomplete + } +} + +/** Reference of `reconcile` (the former renderer reconcileActivityRecords) on shown records. */ +export function referenceActivityReconcile( + list: readonly ActivityRecord[], + tasks: readonly Pick[], + activeRequestIds: ReadonlySet, + interruptedDetail: string +): ActivityRecord[] { + const tasksById = new Map(tasks.map((task) => [task.id, task])) + const seen = new Set() + return list.map((record) => { + const first = !seen.has(record.id) + seen.add(record.id) + if (!first || activeRequestIds.has(record.requestId) || !isActive(record)) return record + const task = tasksById.get(record.requestId) + const terminal = !task + ? 'interrupted' + : task.status === 'completed' || task.status === 'failed' || + task.status === 'cancelled' || task.status === 'interrupted' + ? task.status : undefined + if (!terminal) return record + return { + ...record, + status: terminal === 'completed' && (record.kind === 'tool' || record.kind === 'approval' || + record.kind === 'subagent') ? 'interrupted' : terminal, + detail: terminal === 'interrupted' ? `${record.detail}\n${interruptedDetail}` : record.detail + } + }) +} + /** * Reference semantics of `activityHistory.update` on a plain newest-first * list (see `activityHistoryChangeSchema`), for tests and test doubles. diff --git a/src/shared/assistant-contracts.ts b/src/shared/assistant-contracts.ts index e8c7f530..b793b34e 100644 --- a/src/shared/assistant-contracts.ts +++ b/src/shared/assistant-contracts.ts @@ -133,6 +133,56 @@ export const activityHistoryUpdateSchema = z }) .strict() +/** + * Activity history pages (PERF-15), newest first. A record ID shows once: + * legacy duplicates after the first occurrence are left out. `before` is the + * `nextBefore` of the previous page. + */ +export const activityHistoryFilterSchema = z.enum(['all', 'active', 'failed']) +export const activityHistoryPageRequestSchema = z + .object({ + before: z.number().int().optional(), + limit: z.number().int().min(1).max(500), + filter: activityHistoryFilterSchema.optional(), + projectId: z.string().min(1).max(256).optional(), + conversationId: z.string().min(1).max(256).optional() + }) + .strict() +export const activityHistorySummaryRequestSchema = z + .object({ + /** Conversations whose title and status are needed (the loaded ones). */ + conversationIds: z.array(z.string().min(1).max(256)).max(5_000) + }) + .strict() +export const activityHistoryReconcileRequestSchema = z + .object({ + activeRequestIds: z.array(z.string().min(1).max(256)).max(10_000), + /** The localized line appended to records marked interrupted. */ + interruptedDetail: z.string().max(1_000) + }) + .strict() + +export type ActivityHistoryFilter = z.infer +export type ActivityHistoryPageRequest = z.infer +export type ActivityHistoryPage = { + records: ActivityRecord[] + /** Set when older records may follow. */ + nextBefore?: number +} +export type ActivityHistorySummaryRequest = z.infer +export type ActivityHistorySummary = { + /** Record counts per filter, over the whole history. */ + counts: Record + /** + * Per requested conversation: the title of its newest request record and + * the conversation status (latest request's result, else latest result, + * else latest record), over the whole history. + */ + conversations: Record + legacyHistoryMayBeIncomplete: boolean +} +export type ActivityHistoryReconcileRequest = z.infer + export type ActivityRecord = z.infer export type ActivityHistorySnapshot = z.infer< typeof activityHistorySnapshotSchema diff --git a/src/shared/contracts.ts b/src/shared/contracts.ts index 6005c0d2..4ebe8f34 100644 --- a/src/shared/contracts.ts +++ b/src/shared/contracts.ts @@ -44,7 +44,12 @@ import { type AssistantTask, type ExecutionStats, type ExecutionStatsInput, + type ActivityHistoryPage, + type ActivityHistoryPageRequest, + type ActivityHistoryReconcileRequest, type ActivityHistorySnapshot, + type ActivityHistorySummary, + type ActivityHistorySummaryRequest, type ActivityHistoryUpdate, type ActivityRecord, type TokenUsageSummary, @@ -2167,6 +2172,15 @@ export type DesktopApi = { update: (update: ActivityHistoryUpdate) => Promise /** Deletes the whole history in one transaction. */ clear: () => Promise + /** One page of shown records, newest first (see activityHistoryPageRequestSchema). */ + page: (request: ActivityHistoryPageRequest) => Promise + /** Counts and conversation titles/status over the whole history. */ + summary: (request: ActivityHistorySummaryRequest) => Promise + /** + * Ends records left pending/running by requests that are no longer + * active, against the stored tasks; returns how many changed. + */ + reconcile: (request: ActivityHistoryReconcileRequest) => Promise } usage: { getTokenSummary: () => Promise diff --git a/src/shared/ipc-channels.ts b/src/shared/ipc-channels.ts index 72966906..7789cddd 100644 --- a/src/shared/ipc-channels.ts +++ b/src/shared/ipc-channels.ts @@ -257,6 +257,9 @@ export const ipcChannels = { activityHistoryReplace: 'activity-history:replace', activityHistoryUpdate: 'activity-history:update', activityHistoryClear: 'activity-history:clear', + activityHistoryPage: 'activity-history:page', + activityHistorySummary: 'activity-history:summary', + activityHistoryReconcile: 'activity-history:reconcile', tokenUsageSummary: 'usage:token-summary', tokenUsageChanged: 'usage:token-changed', artifactsList: 'artifacts:list',