diff --git a/src/main/assistant/activity-history-plan.test.ts b/src/main/assistant/activity-history-plan.test.ts index b432c2ff..48e125ed 100644 --- a/src/main/assistant/activity-history-plan.test.ts +++ b/src/main/assistant/activity-history-plan.test.ts @@ -6,6 +6,7 @@ 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 { applyActivityChanges } from '../../shared/activity-history-reference' import { AssistantDatabase } from './assistant-database' import { parseActivityHistorySnapshot, planActivityHistoryReplace, planActivityHistoryUpdate } from './activity-history-plan' @@ -46,13 +47,8 @@ function record(id: string, overrides: Partial = {}): ActivityRe } /** 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)) -} +const applyReference = (list: ActivityRecord[], change: ActivityHistoryChange): ActivityRecord[] => + applyActivityChanges(list, [change]) describe('activity history incremental updates', () => { it('stores exactly the state a full replace stores, for random change sequences', () => { diff --git a/src/main/assistant/activity-history-plan.ts b/src/main/assistant/activity-history-plan.ts index aaaed0ee..3d76411d 100644 --- a/src/main/assistant/activity-history-plan.ts +++ b/src/main/assistant/activity-history-plan.ts @@ -163,6 +163,14 @@ export function planActivityHistoryUpdate(input: unknown, previous: ActivityHist 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)},` diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index 0d9a4d4c..42b2d158 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -5344,6 +5344,25 @@ export class AssistantDatabase { this.writeActivityHistoryPlan(database, (previous) => planActivityHistoryUpdate(input, previous)) } + /** Deletes every activity record in one transaction. */ + clearActivityHistory(): void { + 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 + } + /** The cached committed state, when no other connection committed since. */ private committedActivityHistoryState(database: DatabaseSync): ActivityHistoryState | undefined { const cache = this.activityHistoryCache diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 7e7d2f1c..20cc4e3b 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -6869,6 +6869,10 @@ export function registerIpcHandlers( assistantDatabase.updateActivityHistory(input) } ) + registerHandler(ipcChannels.activityHistoryClear, (event) => { + assertTrustedSender(event, window) + assistantDatabase.clearActivityHistory() + }) registerHandler(ipcChannels.tokenUsageSummary, (event) => { assertTrustedSender(event, window) diff --git a/src/preload/index.ts b/src/preload/index.ts index 4202a3da..c961dc24 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -1293,6 +1293,9 @@ const desktopApi: DesktopApi = { }, update: async (update: ActivityHistoryUpdate) => { await ipcRenderer.invoke(ipcChannels.activityHistoryUpdate, update) + }, + clear: async () => { + await ipcRenderer.invoke(ipcChannels.activityHistoryClear) } }, usage: { diff --git a/src/renderer/src/App.test.tsx b/src/renderer/src/App.test.tsx index 8477e363..7a5db312 100644 --- a/src/renderer/src/App.test.tsx +++ b/src/renderer/src/App.test.tsx @@ -192,10 +192,15 @@ vi.mock("./TerminalPanel", () => ({ })); import App from "./App"; -import { ACTIVITY_STORAGE_KEY } from "./activity-store"; +import { ACTIVITY_STORAGE_KEY, type ActivityRecord } from "./activity-store"; +import { applyActivityChanges } from "../../shared/activity-history-reference"; +import type { ActivityHistoryUpdate } from "../../shared/assistant-contracts"; import { changeUiLocale } from "./i18n"; import { UiLocaleProvider } from "./i18n/UiLocaleProvider"; +/** What the activity history mock holds after replace/update/clear calls. */ +let savedActivity: ActivityRecord[] = []; + let agentListener: ((event: AgentEvent) => void) | undefined; let browserListener: ((state: BrowserLiveState) => void) | undefined; const browserListeners = new Set<(state: BrowserLiveState) => void>(); @@ -675,7 +680,15 @@ const api: DesktopApi & RuntimeNativeClientApi = { records: [], legacyHistoryMayBeIncomplete: false, })), - replace: vi.fn(async () => {}), + replace: vi.fn(async (records: ActivityRecord[]) => { + savedActivity = [...records]; + }), + update: vi.fn(async (update: ActivityHistoryUpdate) => { + savedActivity = applyActivityChanges(savedActivity, update.changes); + }), + clear: vi.fn(async () => { + savedActivity = []; + }), }, usage: { getTokenSummary: vi.fn(async () => ({ @@ -1420,7 +1433,10 @@ describe("App", () => { records: [], legacyHistoryMayBeIncomplete: false, }); - vi.mocked(api.activityHistory.replace).mockReset().mockResolvedValue(); + vi.mocked(api.activityHistory.replace).mockClear(); + vi.mocked(api.activityHistory.update).mockClear(); + vi.mocked(api.activityHistory.clear).mockClear(); + savedActivity = []; vi.mocked(api.schedules.list).mockReset().mockResolvedValue([]); api.channels = undefined; installRemoteProjectsSetting(false); @@ -5506,8 +5522,7 @@ describe("App", () => { ).not.toBeInTheDocument(); await waitFor( () => { - const saved = - vi.mocked(api.activityHistory.replace).mock.calls.at(-1)?.[0] ?? []; + const saved = savedActivity; expect(saved.filter((record) => record.kind === "tool")).toHaveLength( 501, ); @@ -14164,9 +14179,7 @@ describe("App", () => { fireEvent.click(screen.getByLabelText("发送")); await waitFor(() => expect( - vi - .mocked(api.activityHistory.replace) - .mock.calls.at(-1)?.[0] + savedActivity .find( (record) => record.kind === "request" && record.title === "记录项目范围", diff --git a/src/renderer/src/App.tsx b/src/renderer/src/App.tsx index e2fdc4d9..50f2c7dc 100644 --- a/src/renderer/src/App.tsx +++ b/src/renderer/src/App.tsx @@ -98,7 +98,6 @@ import { runtimeProviderLabel, } from "./runtime-selection"; import type { - ActivityHistorySnapshot, AssistantProject, AssistantArtifact, AssistantMemory, @@ -199,13 +198,12 @@ import { type AgentEventDependencies, } from "./agent-event-handler"; import { - clearLegacyActivityHistory, + createActivityStore, loadLegacyActivityHistory, - mergeActivityRecords, - reconcileActivityRecords, - upsertActivityRecord, type ActivityRecord, } from "./activity-store"; +import { useActivityPanelRecords } from "./activity-selectors"; +import { useActivityHistorySync } from "./activity-sync"; import { KnowledgeCitationDialog, type KnowledgeCitationContextView, @@ -1771,22 +1769,9 @@ function App(): React.JSX.Element { const citationRequestRef = useRef(0); const failedKnowledgeLibraryIdRef = useRef(undefined); const [legacyActivityHistory] = useState(loadLegacyActivityHistory); - const [activityRecords, setActivityRecords] = useState( - legacyActivityHistory.records, - ); - const [ - legacyActivityHistoryMayBeIncomplete, - setLegacyActivityHistoryMayBeIncomplete, - ] = useState(legacyActivityHistory.historyMayBeIncomplete); - const [activityHistoryReady, setActivityHistoryReady] = useState(false); - // The Activity page shows a snapshot instead of live records so streaming - // runs do not re-render it; it refreshes on entry, periodically, or manually. - const [activityPanelRecords, setActivityPanelRecords] = - useState(activityRecords); - const activityRecordsRef = useRef(activityRecords); - const legacyActivityHistoryMayBeIncompleteRef = useRef( - legacyActivityHistoryMayBeIncomplete, - ); + const [activityStore] = useState(() => createActivityStore( + legacyActivityHistory.records, legacyActivityHistory.historyMayBeIncomplete)); + const activityPanelRecords = useActivityPanelRecords(activityStore); const activeRuns = useRef(new Map()); const preparingConversations = useRef(new Set()); const [activeConversationIds, setActiveConversationIds] = useState< @@ -2944,39 +2929,18 @@ function App(): React.JSX.Element { projectName: project.name.slice(0, 120), } : { kind: "unavailable" }); - setActivityRecords((current) => - upsertActivityRecord(current, { - ...record, - title: record.title.slice(0, 240), - scope, - id: crypto.randomUUID(), - createdAt: Date.now(), - }), - ); + activityStore.record({ + ...record, + title: record.title.slice(0, 240), + scope, + id: crypto.randomUUID(), + createdAt: Date.now(), + }); }, - [conversationStore], + [activityStore, conversationStore], ); - const updateRequestActivity = useCallback( - ( - requestId: string, - status: ActivityRecord["status"], - detail?: string, - ): void => { - setActivityRecords((current) => - current.map((record) => - record.requestId === requestId && record.kind === "request" - ? { - ...record, - status, - detail: detail ?? record.detail, - } - : record, - ), - ); - }, - [], - ); + const updateRequestActivity = activityStore.updateRequest; useEffect(() => { const api = window.goodbuddy.channels; @@ -3235,12 +3199,12 @@ function App(): React.JSX.Element { activeRuns, activeConversationIdRef, activeProjectIdRef, taskStore, hydratingArtifactIds, requestPersistenceFlush: conversationPersistence.requestFlushAfterCommit, tRef, conversationStore, liveMessages, setConversations, - setActivityRecords, setUnreadConversationIds, notify, updateMessage, recordActivity, - updateRequestActivity, loadWorkspaceChanges, markConversationCompleted, setConversationActivity, + activityStore, setUnreadConversationIds, notify, updateMessage, recordActivity, + loadWorkspaceChanges, markConversationCompleted, setConversationActivity, releaseConversationQueueAfterRun, }; - }, [conversationPersistence, conversationStore, liveMessages, setConversations, taskStore, updateMessage, - recordActivity, updateRequestActivity, loadWorkspaceChanges, markConversationCompleted, + }, [activityStore, conversationPersistence, conversationStore, liveMessages, setConversations, taskStore, + updateMessage, recordActivity, loadWorkspaceChanges, markConversationCompleted, setConversationActivity, releaseConversationQueueAfterRun]); const handleAgentEvent = useCallback((event: AgentEvent): void => { const dependencies = agentEventDependenciesRef.current; @@ -3303,23 +3267,20 @@ function App(): React.JSX.Element { return stop; }, [conversationStoreReady, conversationPersistence]); - const persistActivityHistory = useCallback(async (): Promise => { - if (!activityHistoryReady) { - return; - } - try { - await window.goodbuddy.activityHistory.replace( - activityRecordsRef.current, - legacyActivityHistoryMayBeIncompleteRef.current, - ); - } catch { - notify({ - tone: "error", - message: tRef.current("notices.activityHistoryPersistenceFailed"), - dedupeKey: "activity-history-persistence", - }); - } - }, [activityHistoryReady]); + const persistActivityHistory = useActivityHistorySync(activityStore, legacyActivityHistory, { + onReadFailed: () => notify({ + tone: "error", message: tRef.current("notices.activityHistoryReadFailed"), dedupeKey: "activity-history-read", + }), + onSaveFailed: () => notify({ + tone: "error", + message: tRef.current("notices.activityHistoryPersistenceFailed"), + dedupeKey: "activity-history-persistence", + }), + // The loaded history appears at once when the Activity page is open. + onLoaded: () => { + if (viewRef.current === "activity") activityStore.refreshPanel(); + }, + }); useEffect( () => @@ -3352,9 +3313,7 @@ function App(): React.JSX.Element { }), onTasks: (tasks) => { setAssistantTasks(tasks); - setActivityRecords((current) => - reconcileActivityRecords(current, tasks, new Set(activeRuns.current.keys())), - ); + activityStore.reconcile(tasks, new Set(activeRuns.current.keys())); }, onSchedules: setAssistantSchedules, onUnread: (unread) => { @@ -3385,6 +3344,7 @@ function App(): React.JSX.Element { }, }); }, [ + activityStore, conversationStore, conversationPersistence, conversationStoreReady, @@ -3497,84 +3457,7 @@ function App(): React.JSX.Element { }; }, []); - useEffect(() => { - let active = true; - void (async () => { - let snapshot: ActivityHistorySnapshot; - try { - snapshot = await window.goodbuddy.activityHistory.get(); - } catch { - if (active) { - notify({ - tone: "error", - message: tRef.current("notices.activityHistoryReadFailed"), - dedupeKey: "activity-history-read", - }); - } - return; - } - if (!active) { - return; - } - const legacyHistoryMayBeIncomplete = - legacyActivityHistory.historyMayBeIncomplete || - snapshot.legacyHistoryMayBeIncomplete; - if ( - legacyActivityHistory.records.length > 0 || - legacyActivityHistory.historyMayBeIncomplete - ) { - try { - await window.goodbuddy.activityHistory.replace( - mergeActivityRecords( - legacyActivityHistory.records, - snapshot.records, - ), - legacyHistoryMayBeIncomplete, - ); - clearLegacyActivityHistory(); - } catch { - notify({ - tone: "error", - message: tRef.current("notices.activityHistoryPersistenceFailed"), - dedupeKey: "activity-history-persistence", - }); - } - if (!active) { - return; - } - } - setActivityRecords((current) => - mergeActivityRecords(current, snapshot.records), - ); - setLegacyActivityHistoryMayBeIncomplete(legacyHistoryMayBeIncomplete); - setActivityHistoryReady(true); - })(); - return () => { - active = false; - }; - }, [legacyActivityHistory]); - - useEffect(() => { - activityRecordsRef.current = activityRecords; - legacyActivityHistoryMayBeIncompleteRef.current = - legacyActivityHistoryMayBeIncomplete; - if (!activityHistoryReady) { - return; - } - const timeout = window.setTimeout(() => { - void persistActivityHistory(); - }, 250); - return () => window.clearTimeout(timeout); - }, [ - activityHistoryReady, - activityRecords, - legacyActivityHistoryMayBeIncomplete, - persistActivityHistory, - ]); - - const refreshActivityPanelRecords = useCallback((): void => { - setActivityPanelRecords(activityRecordsRef.current); - }, []); + const refreshActivityPanelRecords = activityStore.refreshPanel; // Layout effect so entering the page never paints a stale snapshot. useLayoutEffect(() => { @@ -3589,20 +3472,6 @@ function App(): React.JSX.Element { return () => window.clearInterval(interval); }, [refreshActivityPanelRecords, view]); - // Runs after the ref sync effect above, so the loaded history is visible. - useEffect(() => { - if (activityHistoryReady && viewRef.current === "activity") { - refreshActivityPanelRecords(); - } - }, [activityHistoryReady, refreshActivityPanelRecords]); - - useEffect( - () => () => { - void persistActivityHistory(); - }, - [persistActivityHistory], - ); - const resumeProjectConversationQueues = useCallback( (projectId: string): void => { if ( @@ -4139,13 +4008,7 @@ function App(): React.JSX.Element { .list() .then((tasks) => { setAssistantTasks(tasks); - setActivityRecords((current) => - reconcileActivityRecords( - current, - tasks, - new Set(activeRuns.current.keys()), - ), - ); + activityStore.reconcile(tasks, new Set(activeRuns.current.keys())); }) .catch(() => notify({ @@ -4153,7 +4016,7 @@ function App(): React.JSX.Element { message: tRef.current("notices.taskHistoryReadFailed"), }), ); - }, [setAssistantTasks]); + }, [activityStore, setAssistantTasks]); useEffect(() => { if (view !== "activity") { @@ -5540,9 +5403,7 @@ function App(): React.JSX.Element { setAssistantTasks((current) => current.filter((task) => task.id !== requestId), ); - setActivityRecords((current) => - current.filter((record) => record.requestId !== requestId), - ); + activityStore.removeByRequest(requestId); notify({ tone: "error", message: displayErrorMessage(error, t("notices.sendFailed")), @@ -5687,28 +5548,11 @@ function App(): React.JSX.Element { session: tRef.current("chat.approval.decisionSession"), permanent: tRef.current("chat.approval.decisionPermanent"), }[decision]; - setActivityRecords((current) => { - let updated = false; - return current.map((record) => { - if ( - !updated && - record.conversationId === conversationId && - record.kind === "approval" && - record.status === "pending" - ) { - updated = true; - return { - ...record, - status: approved ? ("completed" as const) : ("denied" as const), - detail: `${record.detail}\n${tRef.current( - "notices.userDecision", - { decision: decisionLabel }, - )}`, - }; - } - return record; - }); - }); + activityStore.updateApproval( + conversationId, + approved ? "completed" : "denied", + tRef.current("notices.userDecision", { decision: decisionLabel }), + ); updateMessage(conversationId, messageId, (message) => { if (message.approval?.id !== approvalId) return message; if (!message.pendingQuestions?.length) { @@ -5737,7 +5581,7 @@ function App(): React.JSX.Element { })); } }, - [updateMessage, conversationStore, setAssistantTasks], + [activityStore, updateMessage, conversationStore, setAssistantTasks], ); const respondToQuestion = useCallback( @@ -6212,12 +6056,7 @@ function App(): React.JSX.Element { return 'opened'; }; - const clearActivity = useCallback((): void => { - legacyActivityHistoryMayBeIncompleteRef.current = false; - setLegacyActivityHistoryMayBeIncomplete(false); - setActivityRecords([]); - setActivityPanelRecords([]); - }, []); + const clearActivity = activityStore.clear; const openActivityConversation = useCallback((conversationId: string): void => { const open = async (): Promise => { @@ -6541,9 +6380,7 @@ function App(): React.JSX.Element { conversationPersistence.acknowledged().clear(); setConversations([conversation]); setActiveId(conversation.id); - legacyActivityHistoryMayBeIncompleteRef.current = false; - setLegacyActivityHistoryMayBeIncomplete(false); - setActivityRecords([]); + activityStore.reset(); setAssistantTasks([]); setTokenUsage(emptyTokenUsage); setAssistantArtifacts([]); diff --git a/src/renderer/src/activity-selectors.ts b/src/renderer/src/activity-selectors.ts new file mode 100644 index 00000000..8f5a99db --- /dev/null +++ b/src/renderer/src/activity-selectors.ts @@ -0,0 +1,12 @@ +import { useSyncExternalStore } from 'react' +import type { ActivityRecord, ActivityStore } from './activity-store' + +/** + * The records the Activity page shows: a snapshot refreshed on entry, by the + * page timer and the refresh button, so streaming runs do not re-render it. + */ +export function useActivityPanelRecords( + store: ActivityStore +): readonly ActivityRecord[] { + return useSyncExternalStore(store.subscribePanel, store.getPanelRecords) +} diff --git a/src/renderer/src/activity-store.ts b/src/renderer/src/activity-store.ts index 6253d58a..91a6558c 100644 --- a/src/renderer/src/activity-store.ts +++ b/src/renderer/src/activity-store.ts @@ -2,6 +2,7 @@ import i18n from './i18n' import { activityRecordSchema, + type ActivityHistoryChange, type ActivityRecord as SharedActivityRecord, type AssistantTask } from '../../shared/assistant-contracts' @@ -222,3 +223,274 @@ export function clearLegacyActivityHistory( // A failed cleanup leaves the migration source intact for the next start. } } + +/** + * What still has to reach Main, in order: incremental changes, or a marker + * that the whole history must be cleared or written in full first. + */ +export type ActivityOutboxEntry = + | ActivityHistoryChange + | { type: 'clear' } + | { type: 'replace' } + +type Listener = () => void + +export type ActivityStore = ReturnType + +/** + * Activity history, kept outside App state (P2/P4). Every domain action + * updates the newest-first list and appends the matching contract changes to + * the outbox, which activity-sync.ts sends to Main; the list is never written + * back as a whole except for the legacy migration fallback. + * + * The Activity page shows a snapshot (`getPanelRecords`) instead of the live + * list so streaming runs do not re-render it; `refreshPanel` updates it. + */ +export function createActivityStore( + initial: readonly ActivityRecord[] = [], + initialLegacyIncomplete = false +) { + let records: readonly ActivityRecord[] = mergeActivityRecords(initial, []) + let panelRecords = records + let ready = false + let legacyIncomplete = initialLegacyIncomplete + let outbox: ActivityOutboxEntry[] = [] + const listeners = new Set() + const panelListeners = new Set() + const outboxListeners = new Set() + const notify = (set: Set): void => { + for (const listener of [...set]) listener() + } + const subscribe = (set: Set) => (listener: Listener): (() => void) => { + set.add(listener) + return () => { + set.delete(listener) + } + } + + const enqueue = (entry: ActivityOutboxEntry): void => { + const last = outbox.at(-1) + // Consecutive upserts of one record collapse: front then in-place is a + // front upsert with the newer content, and so on. + if ( + entry.type === 'upsert' && + last?.type === 'upsert' && + last.record.id === entry.record.id + ) { + outbox[outbox.length - 1] = { + type: 'upsert', + record: entry.record, + position: + last.position === 'front' || entry.position === 'front' + ? 'front' + : 'in-place' + } + return + } + outbox.push(entry) + } + + const commit = ( + next: readonly ActivityRecord[], + changes: readonly ActivityOutboxEntry[] + ): void => { + if (next === records && changes.length === 0) return + const wasEmpty = outbox.length === 0 + records = next + for (const change of changes) enqueue(change) + notify(listeners) + if (wasEmpty && outbox.length > 0) notify(outboxListeners) + } + + /** Maps records in place; each changed one becomes an in-place upsert. */ + const mapInPlace = ( + update: (record: ActivityRecord) => ActivityRecord + ): void => { + const changes: ActivityOutboxEntry[] = [] + const next = records.map((record) => { + const updated = update(record) + if (updated !== record) { + changes.push({ type: 'upsert', record: updated, position: 'in-place' }) + } + return updated + }) + if (changes.length > 0) commit(next, changes) + } + + const removeWhere = (match: (record: ActivityRecord) => boolean): void => { + const removed = records.filter(match) + if (removed.length === 0) return + commit( + records.filter((record) => !match(record)), + [...new Set(removed.map((record) => record.id))].map((id) => ({ + type: 'remove' as const, + id + })) + ) + } + + const store = { + getRecords(): readonly ActivityRecord[] { + return records + }, + subscribe: subscribe(listeners), + getPanelRecords(): readonly ActivityRecord[] { + return panelRecords + }, + subscribePanel: subscribe(panelListeners), + /** Shows the current list on the Activity page. */ + refreshPanel(): void { + if (panelRecords === records) return + panelRecords = records + notify(panelListeners) + }, + /** Whether the stored history was loaded; changes are sent only after. */ + isReady(): boolean { + return ready + }, + /** Whether the migrated legacy history may have lost records. */ + isLegacyIncomplete(): boolean { + return legacyIncomplete + }, + + /** Adds a record, or moves a tool/subagent record of the same call to the front. */ + record(incoming: ActivityRecord): void { + const next = upsertActivityRecord(records, incoming) + if (next === records) return + commit(next, [{ type: 'upsert', record: next[0]!, position: 'front' }]) + }, + /** Updates the request record(s) of a run where they are. */ + updateRequest( + requestId: string, + status: ActivityRecord['status'], + detail?: string + ): void { + mapInPlace((record) => + record.requestId === requestId && record.kind === 'request' + ? { ...record, status, detail: detail ?? record.detail } + : record + ) + }, + /** Resolves the newest pending approval of a conversation. */ + updateApproval( + conversationId: string, + status: ActivityRecord['status'], + detailLine: string + ): void { + let updated = false + mapInPlace((record) => { + if ( + updated || + record.conversationId !== conversationId || + record.kind !== 'approval' || + record.status !== 'pending' + ) { + return record + } + updated = true + return { ...record, status, detail: `${record.detail}\n${detailLine}` } + }) + }, + /** Drops every record of a request that never started. */ + removeByRequest(requestId: string): void { + removeWhere((record) => record.requestId === requestId) + }, + /** Drops the provisional tool record a subagent replaces. */ + removeToolByCallId(requestId: string, callId: string): void { + removeWhere( + (record) => + record.requestId === requestId && + record.kind === 'tool' && + record.callId === callId + ) + }, + /** Ends the unfinished non-request records of a finished run. */ + settleRequest( + requestId: string, + status: ActivityRecord['status'], + detailLine: string + ): void { + mapInPlace((record) => + record.requestId === requestId && + record.kind !== 'request' && + (record.status === 'pending' || record.status === 'running') + ? { ...record, status, detail: `${record.detail}\n${detailLine}` } + : record + ) + }, + /** Ends records left running by a run that is no longer active. */ + reconcile( + tasks: readonly AssistantTask[], + activeRequestIds: ReadonlySet + ): void { + const next = reconcileActivityRecords(records, tasks, activeRequestIds) + const changes: ActivityOutboxEntry[] = [] + next.forEach((record, index) => { + if (record !== records[index]) { + changes.push({ type: 'upsert', record, position: 'in-place' }) + } + }) + if (changes.length > 0) commit(next, changes) + }, + /** + * Adds the stored history behind the records of this session. Stored + * duplicates of an ID are dropped, as the list always showed only the + * first one. + */ + mergeLoaded( + loaded: readonly ActivityRecord[], + loadedLegacyIncomplete: boolean + ): void { + const next = mergeActivityRecords(records, loaded) + const hasDuplicates = + new Set(loaded.map((record) => record.id)).size < loaded.length + ready = true + legacyIncomplete = legacyIncomplete || loadedLegacyIncomplete + commit(next, hasDuplicates ? [{ type: 'remove-duplicates' }] : []) + if (outbox.length > 0) notify(outboxListeners) + }, + /** + * Clears the history here and, through the sync, in Main. Queued changes + * are dropped: the clear supersedes them. Before the stored history + * loaded only this session's records go, as before. + */ + clear(): void { + records = [] + panelRecords = [] + legacyIncomplete = false + outbox = ready ? [{ type: 'clear' }] : [] + notify(listeners) + notify(panelListeners) + notify(outboxListeners) + }, + /** Main already deleted everything (clear local data): nothing to send. */ + reset(): void { + records = [] + panelRecords = [] + legacyIncomplete = false + outbox = [] + notify(listeners) + notify(panelListeners) + }, + /** The next save writes the whole list (Main's copy is not a known base). */ + requireReplace(): void { + outbox = [{ type: 'replace' }] + notify(outboxListeners) + }, + + getOutbox(): readonly ActivityOutboxEntry[] { + return outbox + }, + subscribeOutbox: subscribe(outboxListeners), + /** Removes and returns the first `count` outbox entries. */ + takeOutbox(count: number): ActivityOutboxEntry[] { + return outbox.splice(0, count) + }, + /** Puts entries that failed to save back in front, in order. */ + requeue(entries: readonly ActivityOutboxEntry[]): void { + if (outbox[0]?.type === 'clear' || outbox[0]?.type === 'replace') return + outbox = [...entries, ...outbox] + } + } + return store +} diff --git a/src/renderer/src/activity-sync.test.ts b/src/renderer/src/activity-sync.test.ts new file mode 100644 index 00000000..3b27ca57 --- /dev/null +++ b/src/renderer/src/activity-sync.test.ts @@ -0,0 +1,320 @@ +// @vitest-environment node +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest' +import type { + ActivityHistorySnapshot, + ActivityHistoryUpdate, + ActivityRecord, + AssistantTask +} from '../../shared/assistant-contracts' +import { createActivityStore, type ActivityStore } from './activity-store' +import { + loadActivityHistory, + startActivitySync, + type ActivityHistoryApi +} from './activity-sync' + +/** The part of Main's AssistantDatabase these tests use (a real database on a temp file). */ +type AssistantDatabase = { + initialize: (directory: string) => void + close: () => void + getActivityHistory: () => ActivityHistorySnapshot + replaceActivityHistory: (input: unknown) => void + updateActivityHistory: (input: unknown) => void + clearActivityHistory: () => void +} +let AssistantDatabase: new (path: string) => AssistantDatabase +beforeAll(async () => { + // Main code is outside the renderer TypeScript project; load it untyped. + const mainModule = '../../main/assistant/assistant-database' + ;({ AssistantDatabase } = (await import(mainModule)) as { + AssistantDatabase: typeof AssistantDatabase + }) +}) + +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 openDatabase(): AssistantDatabase { + const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-activity-sync-')) + directories.push(directory) + const database = new AssistantDatabase(join(directory, 'assistant.sqlite')) + database.initialize(directory) + opened.push(database) + return database +} + +/** IPC double over a real database; `fail()` decides whether a call throws first. */ +function databaseApi( + database: AssistantDatabase, + fail: () => boolean = () => false +): ActivityHistoryApi & { updates: ActivityHistoryUpdate[] } { + const updates: ActivityHistoryUpdate[] = [] + const guard = (): void => { + if (fail()) throw new Error('injected save failure') + } + return { + updates, + get: async () => database.getActivityHistory(), + replace: async (records, legacyHistoryMayBeIncomplete) => { + guard() + database.replaceActivityHistory( + structuredClone({ records, legacyHistoryMayBeIncomplete }) + ) + }, + update: async (update) => { + guard() + updates.push(update) + database.updateActivityHistory(structuredClone(update)) + }, + clear: async () => { + guard() + database.clearActivityHistory() + } + } +} + +const manualTimers = { + setTimeout: () => 1, + clearTimeout: () => undefined +} + +function makeRandom(seed: number): () => number { + let state = seed + return () => (state = (state * 1664525 + 1013904223) >>> 0) / 4294967296 +} + +let counter = 0 +function record(overrides: Partial = {}): ActivityRecord { + counter += 1 + return { + id: `record-${counter}`, + conversationId: `conversation-${counter % 3}`, + requestId: `request-${counter % 5}`, + scope: { kind: 'global' }, + kind: 'tool', + title: `Tool ${counter}`, + detail: `detail ${counter}`, + status: 'running', + createdAt: counter, + ...overrides + } +} + +function task(id: string, status: AssistantTask['status']): AssistantTask { + return { + id, + title: id, + instructions: id, + workMode: 'ask', + status, + createdAt: new Date(0).toISOString() + } as AssistantTask +} + +async function loadedStore(api: ActivityHistoryApi): Promise { + const store = createActivityStore() + await loadActivityHistory( + store, + api, + { records: [], historyMayBeIncomplete: false }, + { onReadFailed: () => undefined, onSaveFailed: () => undefined, isActive: () => true } + ) + return store +} + +describe('activity history sync', () => { + it('stores what a full replace of the final list stores, for random actions, flushes and failures', async () => { + for (const seed of [3, 17, 2026]) { + const random = makeRandom(seed) + const pick = (items: readonly T[]): T => + items[Math.floor(random() * items.length)]! + const incremental = openDatabase() + // Legacy history with duplicate IDs, stored by the former full replace. + const first = record({ id: 'dup' }) + const legacy = [ + first, + record({ status: 'completed' }), + { ...first, detail: 'older duplicate' }, + record({ kind: 'request', status: 'running', requestId: 'request-1' }), + record({ kind: 'approval', status: 'pending', conversationId: 'conversation-1' }) + ] + incremental.replaceActivityHistory({ + records: legacy, + legacyHistoryMayBeIncomplete: seed % 2 === 0 + }) + let failing = false + let expectedFlag = seed % 2 === 0 + const api = databaseApi(incremental, () => failing && random() < 0.5) + const store = await loadedStore(api) + const sync = startActivitySync(store, api, { + onSaveFailed: () => undefined, + timers: manualTimers + }) + const callIds = ['a', 'b', 'c', 'd'] + for (let step = 0; step < 400; step++) { + const roll = random() + const requestId = `request-${Math.floor(random() * 5)}` + if (roll < 0.3) { + const kind = pick(['tool', 'subagent', 'request', 'result', 'approval'] as const) + store.record(record({ + kind, + requestId, + ...(kind === 'tool' || kind === 'subagent' ? { callId: pick(callIds) } : {}), + status: pick(['pending', 'running', 'completed', 'failed'] as const) + })) + } else if (roll < 0.4) { + store.updateRequest(requestId, pick(['completed', 'failed'] as const), random() < 0.5 ? 'done' : undefined) + } else if (roll < 0.47) { + store.updateApproval(`conversation-${Math.floor(random() * 3)}`, 'denied', 'decision') + } else if (roll < 0.52) { + store.removeByRequest(requestId) + } else if (roll < 0.58) { + store.removeToolByCallId(requestId, pick(callIds)) + } else if (roll < 0.65) { + store.settleRequest(requestId, 'interrupted', 'incomplete') + } else if (roll < 0.72) { + store.reconcile( + [task('request-0', 'completed'), task('request-2', 'failed'), task('request-3', 'running')], + new Set(random() < 0.5 ? ['request-1'] : []) + ) + } else if (roll < 0.74) { + store.clear() + expectedFlag = false + } else if (roll < 0.8) { + failing = random() < 0.4 + } else { + await sync.flush() + } + } + failing = false + await sync.flush() + expect(store.getOutbox()).toEqual([]) + + const replaced = openDatabase() + replaced.replaceActivityHistory({ + records: [...store.getRecords()], + legacyHistoryMayBeIncomplete: expectedFlag + }) + expect(incremental.getActivityHistory()).toEqual(replaced.getActivityHistory()) + expect(incremental.getActivityHistory().records).toEqual(store.getRecords()) + sync.stop() + } + }) + + it('sends only the changed records', async () => { + const database = openDatabase() + const records = Array.from({ length: 5_000 }, () => record({ status: 'completed' })) + database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: false }) + const api = databaseApi(database) + const store = await loadedStore(api) + const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers }) + + const tool = record({ callId: 'call-1', requestId: 'request-x' }) + store.record(tool) + store.record({ ...tool, id: 'ignored', detail: 'progress', status: 'completed' }) + store.updateRequest(records[4_000]!.requestId, 'failed') + await sync.flush() + // Two consecutive upserts of the tool collapse into one. + expect(api.updates).toHaveLength(1) + const changes = api.updates[0]!.changes + const requestRecords = records.filter( + (item) => item.requestId === records[4_000]!.requestId && item.kind === 'request' + ) + expect(changes).toHaveLength(1 + requestRecords.length) + expect(JSON.stringify(api.updates[0]).length).toBeLessThan(1_000) + expect(database.getActivityHistory().records[0]).toMatchObject({ + id: tool.id, + detail: 'progress' + }) + sync.stop() + }) + + it('keeps failed changes queued and retries them in order with backoff', async () => { + const database = openDatabase() + let fail = true + const api = databaseApi(database, () => fail) + const store = await loadedStore(api) + const delays: number[] = [] + const callbacks: Array<() => void> = [] + const onSaveFailed = vi.fn() + const sync = startActivitySync(store, api, { + onSaveFailed, + timers: { + setTimeout: (callback, delay) => { + delays.push(delay) + callbacks.push(callback) + return callbacks.length + }, + clearTimeout: () => undefined + } + }) + const a = record() + const b = record() + store.record(a) + expect(delays).toEqual([250]) + callbacks.shift()!() + await vi.waitFor(() => expect(delays).toEqual([250, 1_000])) + store.record(b) + callbacks.shift()!() + await vi.waitFor(() => expect(delays).toEqual([250, 1_000, 2_000])) + expect(onSaveFailed).toHaveBeenCalledTimes(1) + fail = false + callbacks.shift()!() + await vi.waitFor(() => + expect(database.getActivityHistory().records.map((item) => item.id)).toEqual([b.id, a.id]) + ) + expect(store.getOutbox()).toEqual([]) + sync.stop() + }) + + it('clears the database in one call and drops queued changes', async () => { + const database = openDatabase() + database.replaceActivityHistory({ records: [record(), record()], legacyHistoryMayBeIncomplete: true }) + const api = databaseApi(database) + const update = vi.spyOn(api, 'update') + const store = await loadedStore(api) + const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers }) + store.record(record()) + store.clear() + await sync.flush() + expect(update).not.toHaveBeenCalled() + expect(database.getActivityHistory()).toEqual({ records: [], legacyHistoryMayBeIncomplete: false }) + expect(store.getRecords()).toEqual([]) + sync.stop() + }) + + it('migrates legacy records with one replace and retries it in full after a failure', async () => { + const database = openDatabase() + const stored = record({ status: 'completed' }) + database.replaceActivityHistory({ records: [stored], legacyHistoryMayBeIncomplete: false }) + let fail = true + const api = databaseApi(database, () => fail) + const legacyRecord = record({ status: 'completed' }) + const store = createActivityStore([legacyRecord], true) + await loadActivityHistory( + store, + api, + { records: [legacyRecord], historyMayBeIncomplete: true }, + { onReadFailed: () => undefined, onSaveFailed: () => undefined, isActive: () => true } + ) + expect(store.getRecords()).toEqual([legacyRecord, stored]) + expect(store.getOutbox()).toEqual([{ type: 'replace' }]) + fail = false + const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers }) + await sync.flush() + expect(database.getActivityHistory()).toEqual({ + records: [legacyRecord, stored], + legacyHistoryMayBeIncomplete: true + }) + sync.stop() + }) +}) diff --git a/src/renderer/src/activity-sync.ts b/src/renderer/src/activity-sync.ts new file mode 100644 index 00000000..f986b6eb --- /dev/null +++ b/src/renderer/src/activity-sync.ts @@ -0,0 +1,225 @@ +import { useCallback, useEffect, useRef } from 'react' +import type { + ActivityHistoryChange, + ActivityHistorySnapshot, + ActivityHistoryUpdate, + ActivityRecord +} from '../../shared/assistant-contracts' +import { + clearLegacyActivityHistory, + mergeActivityRecords, + type ActivityOutboxEntry, + type ActivityStore, + type LegacyActivityHistory +} from './activity-store' + +/** + * Activity history IPC with Main (P4): the first load with the legacy + * localStorage migration, and the incremental save queue. Changes collect in + * the store's outbox and leave 250 ms after the first one, at most 20,000 per + * `update` call. A failed save keeps its changes queued, in order, and is + * retried with backoff; quitting and unmounting flush what is left. + */ +export type ActivityHistoryApi = { + get: () => Promise + replace: (records: ActivityRecord[], legacyHistoryMayBeIncomplete: boolean) => Promise + update: (update: ActivityHistoryUpdate) => Promise + clear: () => Promise +} + +export const ACTIVITY_SAVE_DELAY_MS = 250 +export const ACTIVITY_MAX_CHANGES_PER_UPDATE = 20_000 +const RETRY_DELAYS_MS = [1_000, 2_000, 5_000, 10_000, 30_000] + +export type ActivitySync = { + /** Sends every queued change; resolves when done or after a failure. */ + flush: () => Promise + stop: () => void +} + +type Timers = { + setTimeout: (callback: () => void, delay: number) => unknown + clearTimeout: (handle: unknown) => void +} + +const defaultTimers: Timers = { + setTimeout: (callback, delay) => globalThis.setTimeout(callback, delay), + clearTimeout: (handle) => globalThis.clearTimeout(handle as ReturnType) +} + +const saving = new WeakMap>() + +export function startActivitySync( + store: ActivityStore, + api: ActivityHistoryApi, + options: { onSaveFailed: () => void; timers?: Timers; delayMs?: number } +): ActivitySync { + const timers = options.timers ?? defaultTimers + const delayMs = options.delayMs ?? ACTIVITY_SAVE_DELAY_MS + let timer: unknown + let failures = 0 + let stopped = false + + const schedule = (delay: number): void => { + if (stopped || timer !== undefined) return + timer = timers.setTimeout(() => { + timer = undefined + void flush() + }, delay) + } + + /** Sends the next batch; false when it failed. */ + const sendNext = async (): Promise => { + const first = store.getOutbox()[0] + if (!first) return true + let batch: ActivityOutboxEntry[] + try { + if (first.type === 'clear') { + batch = store.takeOutbox(1) + await api.clear() + } else if (first.type === 'replace') { + // The full list already contains everything queued after the marker. + batch = store.takeOutbox(store.getOutbox().length) + await api.replace([...store.getRecords()], store.isLegacyIncomplete()) + } else { + const outbox = store.getOutbox() + let count = 0 + while ( + count < outbox.length && + count < ACTIVITY_MAX_CHANGES_PER_UPDATE && + outbox[count]!.type !== 'clear' && + outbox[count]!.type !== 'replace' + ) { + count++ + } + batch = store.takeOutbox(count) + await api.update({ changes: batch as ActivityHistoryChange[] }) + } + } catch { + store.requeue(batch!) + return false + } + return true + } + + const run = async (): Promise => { + while (store.getOutbox().length > 0) { + if (!(await sendNext())) { + failures++ + if (failures === 1) options.onSaveFailed() + schedule(RETRY_DELAYS_MS[Math.min(failures, RETRY_DELAYS_MS.length) - 1]!) + return + } + failures = 0 + } + } + + const flush = async (): Promise => { + if (!store.isReady()) return + if (timer !== undefined) { + timers.clearTimeout(timer) + timer = undefined + } + // One save at a time per store, also across a remounted sync. + const previous = saving.get(store) ?? Promise.resolve() + const current = previous.then(run) + saving.set(store, current) + await current + } + + const unsubscribe = store.subscribeOutbox(() => { + if (store.isReady() && store.getOutbox().length > 0 && failures === 0) { + schedule(delayMs) + } + }) + if (store.isReady() && store.getOutbox().length > 0) schedule(delayMs) + + return { + flush, + stop: () => { + unsubscribe() + stopped = true + if (timer !== undefined) timers.clearTimeout(timer) + timer = undefined + } + } +} + +/** + * Loads the stored history on mount, then saves changes incrementally. The + * returned `flush` (stable) sends what is queued, for the quit handshake; the + * queue is also flushed on unmount. + */ +export function useActivityHistorySync( + store: ActivityStore, + legacy: LegacyActivityHistory, + notices: { onReadFailed: () => void; onSaveFailed: () => void; onLoaded?: () => void } +): () => Promise { + const noticesRef = useRef(notices) + useEffect(() => { + noticesRef.current = notices + }) + const syncRef = useRef(undefined) + useEffect(() => { + let active = true + const api = window.goodbuddy.activityHistory + const onSaveFailed = (): void => noticesRef.current.onSaveFailed() + const sync = startActivitySync(store, api, { onSaveFailed }) + syncRef.current = sync + void loadActivityHistory(store, api, legacy, { + onReadFailed: () => noticesRef.current.onReadFailed(), + onSaveFailed, + isActive: () => active + }).then(() => { + if (active && store.isReady()) noticesRef.current.onLoaded?.() + }) + return () => { + active = false + sync.stop() + if (syncRef.current === sync) syncRef.current = undefined + void sync.flush() + } + }, [legacy, store]) + return useCallback(async () => { + await syncRef.current?.flush() + }, []) +} + +/** + * Loads the stored history once. A legacy localStorage history is written + * together with it as a full `replace` and then removed; the store is ready + * (saves start) only after the load succeeded. + */ +export async function loadActivityHistory( + store: ActivityStore, + api: ActivityHistoryApi, + legacy: LegacyActivityHistory, + handlers: { onReadFailed: () => void; onSaveFailed: () => void; isActive: () => boolean } +): Promise { + let snapshot: ActivityHistorySnapshot + try { + snapshot = await api.get() + } catch { + if (handlers.isActive()) handlers.onReadFailed() + return + } + if (!handlers.isActive()) return + let replaceFailed = false + if (legacy.records.length > 0 || legacy.historyMayBeIncomplete) { + // Everything queued so far is part of the list sent here. + store.takeOutbox(store.getOutbox().length) + try { + await api.replace( + mergeActivityRecords(store.getRecords(), snapshot.records), + legacy.historyMayBeIncomplete || snapshot.legacyHistoryMayBeIncomplete + ) + clearLegacyActivityHistory() + } catch { + replaceFailed = true + handlers.onSaveFailed() + } + if (!handlers.isActive()) return + } + store.mergeLoaded(snapshot.records, snapshot.legacyHistoryMayBeIncomplete) + if (replaceFailed) store.requireReplace() +} diff --git a/src/renderer/src/agent-event-handler.ts b/src/renderer/src/agent-event-handler.ts index 64a7bb77..40d324e2 100644 --- a/src/renderer/src/agent-event-handler.ts +++ b/src/renderer/src/agent-event-handler.ts @@ -8,7 +8,7 @@ import type { } from "../../shared/assistant-contracts"; import { appendConversationQuestionBlock } from "../../shared/conversation-question-blocks"; import { knowledgeReferenceKey } from "../../shared/knowledge-reference"; -import type { ActivityRecord } from "./activity-store"; +import type { ActivityRecord, ActivityStore } from "./activity-store"; import type { Message, SubagentActivity, ToolActivity } from "./ChatTimeline"; import type { ConversationStore } from "./conversation-store"; import type { LiveMessageStore } from "./live-message-store"; @@ -48,12 +48,13 @@ export type AgentEventDependencies = { setConversations: ConversationStore["set"]; /** Tasks and artifacts; updates re-render only the views that select them. */ taskStore: TaskStore; - setActivityRecords: Dispatch>; + /** Activity history; its actions also queue the incremental save. */ + activityStore: ActivityStore; setUnreadConversationIds: Dispatch>>; notify: (notification: AppNotificationInput) => void; updateMessage: (conversationId: string, messageId: string, update: (message: Message) => Message) => void; + /** Adds a record with the conversation's project scope. */ recordActivity: (record: Omit) => void; - updateRequestActivity: (requestId: string, status: ActivityRecord["status"], detail?: string) => void; loadWorkspaceChanges: (projectId: string) => Promise; markConversationCompleted: (conversationId: string) => void; setConversationActivity: (conversationId: string, active: boolean) => void; @@ -169,8 +170,8 @@ export function handleAgentEvent(event: AgentEvent, deps: AgentEventDependencies const { activeRuns, activeConversationIdRef, activeProjectIdRef, taskStore, hydratingArtifactIds, requestPersistenceFlush, tRef, conversationStore, liveMessages, setConversations, - setActivityRecords, setUnreadConversationIds, notify, - updateMessage, recordActivity, updateRequestActivity, loadWorkspaceChanges, markConversationCompleted, + activityStore, setUnreadConversationIds, notify, + updateMessage, recordActivity, loadWorkspaceChanges, markConversationCompleted, setConversationActivity, releaseConversationQueueAfterRun, } = deps; const setAssistantTasks = taskStore.setTasks; @@ -565,17 +566,7 @@ export function handleAgentEvent(event: AgentEvent, deps: AgentEventDependencies }); } if (event.runtimeCallId) { - setActivityRecords((current) => { - const remaining = current.filter( - (record) => - !( - record.requestId === event.requestId && - record.kind === "tool" && - record.callId === event.runtimeCallId - ), - ); - return remaining.length === current.length ? current : remaining; - }); + activityStore.removeToolByCallId(event.requestId, event.runtimeCallId); } recordActivity({ conversationId: run.conversationId, @@ -744,30 +735,17 @@ export function handleAgentEvent(event: AgentEvent, deps: AgentEventDependencies const incompleteActivityDetail = tRef.current( "chat.status.activityIncomplete", ); - updateRequestActivity( + activityStore.updateRequest( event.requestId, terminalStatus, event.type === "error" ? event.message : tRef.current("chat.status.taskCompleted"), ); - setActivityRecords((current) => - current.map((record) => - record.requestId === event.requestId && - record.kind !== "request" && - (record.status === "pending" || record.status === "running") - ? { - ...record, - status: - event.type === "done" ? "interrupted" : terminalStatus, - detail: `${record.detail}\n${ - event.type === "done" - ? incompleteActivityDetail - : event.message - }`, - } - : record, - ), + activityStore.settleRequest( + event.requestId, + event.type === "done" ? "interrupted" : terminalStatus, + event.type === "done" ? incompleteActivityDetail : event.message, ); recordActivity({ conversationId: run.conversationId, diff --git a/src/shared/activity-history-reference.ts b/src/shared/activity-history-reference.ts new file mode 100644 index 00000000..bb4ada67 --- /dev/null +++ b/src/shared/activity-history-reference.ts @@ -0,0 +1,32 @@ +import type { + ActivityHistoryChange, + ActivityRecord +} from './assistant-contracts' + +/** + * Reference semantics of `activityHistory.update` on a plain newest-first + * list (see `activityHistoryChangeSchema`), for tests and test doubles. + */ +export function applyActivityChanges( + list: readonly ActivityRecord[], + changes: readonly ActivityHistoryChange[] +): ActivityRecord[] { + let next = [...list] + for (const change of changes) { + if (change.type === 'remove') { + next = next.filter((item) => item.id !== change.id) + continue + } + if (change.type === 'remove-duplicates') { + const seen = new Set() + next = next.filter((item) => !seen.has(item.id) && Boolean(seen.add(item.id))) + continue + } + const index = next.findIndex((item) => item.id === change.record.id) + if (index < 0) next = [change.record, ...next] + else if (change.position === 'front') { + next = [change.record, ...next.filter((_, at) => at !== index)] + } else next = next.map((item, at) => (at === index ? change.record : item)) + } + return next +} diff --git a/src/shared/assistant-contracts.ts b/src/shared/assistant-contracts.ts index 829b9147..e8c7f530 100644 --- a/src/shared/assistant-contracts.ts +++ b/src/shared/assistant-contracts.ts @@ -110,6 +110,8 @@ export const activityHistorySnapshotSchema = z * - `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. + * - `remove-duplicates` deletes every record but the first of each ID; the + * renderer shows only the first (see `mergeActivityRecords`). */ export const activityHistoryChangeSchema = z.discriminatedUnion('type', [ z.object({ @@ -120,7 +122,8 @@ export const activityHistoryChangeSchema = z.discriminatedUnion('type', [ z.object({ type: z.literal('remove'), id: z.string().min(1).max(256) - }).strict() + }).strict(), + z.object({ type: z.literal('remove-duplicates') }).strict() ]) export const activityHistoryUpdateSchema = z diff --git a/src/shared/contracts.ts b/src/shared/contracts.ts index 6efff2a9..6005c0d2 100644 --- a/src/shared/contracts.ts +++ b/src/shared/contracts.ts @@ -2162,10 +2162,11 @@ export type DesktopApi = { ) => 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. + * written. `replace` remains for the legacy localStorage migration. */ - update?: (update: ActivityHistoryUpdate) => Promise + update: (update: ActivityHistoryUpdate) => Promise + /** Deletes the whole history in one transaction. */ + clear: () => Promise } usage: { getTokenSummary: () => Promise diff --git a/src/shared/ipc-channels.ts b/src/shared/ipc-channels.ts index a51b64b5..72966906 100644 --- a/src/shared/ipc-channels.ts +++ b/src/shared/ipc-channels.ts @@ -256,6 +256,7 @@ export const ipcChannels = { activityHistoryGet: 'activity-history:get', activityHistoryReplace: 'activity-history:replace', activityHistoryUpdate: 'activity-history:update', + activityHistoryClear: 'activity-history:clear', tokenUsageSummary: 'usage:token-summary', tokenUsageChanged: 'usage:token-changed', artifactsList: 'artifacts:list',