perf(activity): save activity history incrementally through an activity store and sync queue

The renderer no longer sends the whole history on every change. Domain
actions on an external activity store update the list and queue the matching
contract changes; activity-sync.ts flushes them every 250 ms via
activityHistory.update (<= 20,000 per call), keeps failed changes queued in
order with backoff retry, and flushes on quit and unmount. replace remains
only for the legacy localStorage migration; clearing uses the new
activityHistory.clear IPC. App's activity state is gone; ActivityPanel reads
a selector. update is now required in the desktop API.
This commit is contained in:
mesalogo
2026-10-03 20:45:51 +08:00
parent 2adee13638
commit 6aab78b9fa
16 changed files with 987 additions and 263 deletions
@@ -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<ActivityRecord> = {}): 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', () => {
@@ -163,6 +163,14 @@ export function planActivityHistoryUpdate(input: unknown, previous: ActivityHist
let order = [...previous.order]
const touched = new Set<string>()
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)},`
+19
View File
@@ -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
+4
View File
@@ -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)
+3
View File
@@ -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: {
+21 -8
View File
@@ -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 === "记录项目范围",
+47 -210
View File
@@ -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<string | undefined>(undefined);
const [legacyActivityHistory] = useState(loadLegacyActivityHistory);
const [activityRecords, setActivityRecords] = useState<ActivityRecord[]>(
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<ActivityRecord[]>(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<string, ActiveRun>());
const preparingConversations = useRef(new Set<string>());
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<void> => {
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<void> => {
@@ -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([]);
+12
View File
@@ -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)
}
+272
View File
@@ -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<typeof createActivityStore>
/**
* 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<Listener>()
const panelListeners = new Set<Listener>()
const outboxListeners = new Set<Listener>()
const notify = (set: Set<Listener>): void => {
for (const listener of [...set]) listener()
}
const subscribe = (set: Set<Listener>) => (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<string>
): 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
}
+320
View File
@@ -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> = {}): 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<ActivityStore> {
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 = <T,>(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()
})
})
+225
View File
@@ -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<ActivityHistorySnapshot>
replace: (records: ActivityRecord[], legacyHistoryMayBeIncomplete: boolean) => Promise<void>
update: (update: ActivityHistoryUpdate) => Promise<void>
clear: () => Promise<void>
}
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<void>
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<typeof setTimeout>)
}
const saving = new WeakMap<ActivityStore, Promise<void>>()
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<boolean> => {
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<void> => {
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<void> => {
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<void> {
const noticesRef = useRef(notices)
useEffect(() => {
noticesRef.current = notices
})
const syncRef = useRef<ActivitySync | undefined>(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<void> {
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()
}
+12 -34
View File
@@ -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<SetStateAction<ActivityRecord[]>>;
/** Activity history; its actions also queue the incremental save. */
activityStore: ActivityStore;
setUnreadConversationIds: Dispatch<SetStateAction<Set<string>>>;
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<ActivityRecord, "id" | "createdAt" | "scope">) => void;
updateRequestActivity: (requestId: string, status: ActivityRecord["status"], detail?: string) => void;
loadWorkspaceChanges: (projectId: string) => Promise<void>;
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,
+32
View File
@@ -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<string>()
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
}
+4 -1
View File
@@ -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
+4 -3
View File
@@ -2162,10 +2162,11 @@ export type DesktopApi = {
) => Promise<void>
/**
* 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<void>
update: (update: ActivityHistoryUpdate) => Promise<void>
/** Deletes the whole history in one transaction. */
clear: () => Promise<void>
}
usage: {
getTokenSummary: () => Promise<TokenUsageSummary>
+1
View File
@@ -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',