perf(activity): keep only loaded activity pages in the renderer, driven by Main's summary

The renderer no longer holds the whole history. Startup sends queued
changes, lets Main reconcile leftover pending/running records against the
stored tasks (same localized interruptedOnRestart line), then loads the
first page (200). The Activity page loads a full visible batch (500, as
before paging); "load more" shows the next batch and loads the next page
behind it; a filter switch refetches that filter's first page; leaving the
page trims to one batch. Counts, project/conversation groups, titles,
conversation status and the clear count come from activityHistory.summary
(refreshed on entry, by the page timer, the refresh button and after load
more / filter changes).

Records that are not loaded are still updated incrementally: the update
contract gains content-matched changes (upsert-call, update-request,
resolve-approval, settle-request, remove-request, remove-call) that Main
applies in SQL to shown rows; the store applies the same semantics to its
pages through the shared reference. update returns Main's stored call
records so a call whose earlier record was not loaded keeps one id. Task
refresh reconciles via Main SQL. Legacy localStorage records migrate as
incremental changes; the renderer no longer calls replace.

Tests: randomized sequences of every action, page loads, trims, flushes and
injected failures against the former full-list code on a real database;
first page <= 200 after startup; flush payload depends only on changed
records (including unloaded ones); ActivityPanel output with pages + summary
equals the full-list rendering for randomized datasets with duplicates;
restart interrupts leftover records with the same detail; clear empties the
database, pages and summary; SQL change semantics equal the reference.
This commit is contained in:
mesalogo
2026-10-04 00:03:33 +08:00
parent 0e9c9d4927
commit bf9850a6a9
16 changed files with 1294 additions and 493 deletions
@@ -5,8 +5,9 @@ import { join, resolve } from 'node:path'
import { DatabaseSync } from 'node:sqlite' import { DatabaseSync } from 'node:sqlite'
import { build } from 'esbuild' import { build } from 'esbuild'
import { afterEach, describe, expect, it } from 'vitest' import { afterEach, describe, expect, it } from 'vitest'
import type { ActivityRecord, AssistantTask } from '../../shared/assistant-contracts' import type { ActivityHistoryChange, ActivityRecord, AssistantTask } from '../../shared/assistant-contracts'
import { import {
applyActivityChanges,
firstOccurrences, firstOccurrences,
referenceActivityPage, referenceActivityPage,
referenceActivityReconcile, referenceActivityReconcile,
@@ -89,6 +90,34 @@ describe('ActivityHistoryRepository', () => {
} }
}) })
it('applies content-matched changes in SQL exactly like the reference, including legacy duplicates', () => {
for (const seed of [4, 8, 15]) {
const random = makeRandom(seed)
const pick = <T,>(items: readonly T[]): T => items[Math.floor(random() * items.length)]!
const { database } = open()
let list = randomList(random, 150).map((item, index) =>
item.kind === 'tool' || item.kind === 'subagent' ? { ...item, callId: `call-${index % 4}` } : item)
database.replaceActivityHistory({ records: list, legacyHistoryMayBeIncomplete: false })
list = firstOccurrences(list)
for (let round = 0; round < 80; round++) {
const requestId = `request-${Math.floor(random() * 10)}`
const status = pick(statuses)
const changes: ActivityHistoryChange[] = [pick([
(): ActivityHistoryChange => ({ type: 'upsert-call', record: { ...randomList(random, 1)[0]!, id: `new-${round}`,
kind: pick(['tool', 'subagent'] as const), requestId, callId: `call-${Math.floor(random() * 4)}` } }),
(): ActivityHistoryChange => ({ type: 'update-request', requestId, status, ...(random() < 0.5 ? { detail: 'd' } : {}) }),
(): ActivityHistoryChange => ({ type: 'resolve-approval', conversationId: `conversation-${Math.floor(random() * 6)}`, status, detailLine: 'x' }),
(): ActivityHistoryChange => ({ type: 'settle-request', requestId, status, detailLine: 'y' }),
(): ActivityHistoryChange => ({ type: 'remove-request', requestId }),
(): ActivityHistoryChange => ({ type: 'remove-call', requestId, callId: `call-${Math.floor(random() * 4)}` })
])()]
database.updateActivityHistory({ changes })
list = applyActivityChanges(list, changes)
expect(firstOccurrences(database.getActivityHistory().records)).toEqual(list)
}
}
})
it('reconciles with the stored tasks like the former renderer code', () => { it('reconciles with the stored tasks like the former renderer code', () => {
const { database } = open() const { database } = open()
const random = makeRandom(5) const random = makeRandom(5)
@@ -8,6 +8,7 @@ import {
type ActivityHistoryPage, type ActivityHistoryPage,
type ActivityHistorySnapshot, type ActivityHistorySnapshot,
type ActivityHistorySummary, type ActivityHistorySummary,
type ActivityHistoryUpdateResult,
type ActivityRecord type ActivityRecord
} from '../../shared/assistant-contracts' } from '../../shared/assistant-contracts'
import { import {
@@ -275,9 +276,15 @@ export class ActivityHistoryRepository {
}) })
} }
/** Applies incremental changes in one transaction, without reading the history. */ /**
update(input: unknown): void { * Applies incremental changes in one transaction, without reading the
* history. Returns the stored record of every upsert-call, which keeps
* the id and createdAt of an earlier record of the call the renderer may
* not have loaded.
*/
update(input: unknown): ActivityHistoryUpdateResult {
const update = activityHistoryUpdateSchema.parse(input) const update = activityHistoryUpdateSchema.parse(input)
const calls: ActivityRecord[] = []
this.write(() => { this.write(() => {
const database = this.database const database = this.database
const nextSeq = database.prepare( const nextSeq = database.prepare(
@@ -294,25 +301,95 @@ export class ActivityHistoryRepository {
'UPDATE activity_history_records SET record_json = ? WHERE record_key = ? AND record_json != ?' 'UPDATE activity_history_records SET record_json = ? WHERE record_key = ? AND record_json != ?'
) )
const remove = database.prepare('DELETE FROM activity_history_records WHERE record_id = ?') const remove = database.prepare('DELETE FROM activity_history_records WHERE record_id = ?')
const seq = (): number => (nextSeq.get() as { seq: number }).seq
// Shown rows (first occurrences) a content-matched change applies to, newest first.
type Row = { record_key: string; order_seq: number; record: ActivityRecord }
const select = (where: string, limit: number, ...args: SqlValue[]): Row[] =>
(database.prepare(
`SELECT record_key, order_seq, record_json FROM activity_history_records
WHERE occurrence = 0 AND ${where} ORDER BY order_seq DESC LIMIT ${limit}`
).all(...args) as Array<{ record_key: string; order_seq: number; record_json: string }>)
.map((row) => ({ record_key: row.record_key, order_seq: row.order_seq, record: parseRow(row.record_json) }))
const frontSeq = database.prepare(
'SELECT MAX(order_seq) AS seq FROM activity_history_records WHERE occurrence = 0'
)
const rewrite = (key: string, record: ActivityRecord): void => {
const json = JSON.stringify(activityRecordSchema.parse(record))
replaceInPlace.run(json, key, json)
}
for (const change of update.changes) { for (const change of update.changes) {
if (change.type === 'remove-duplicates') { switch (change.type) {
database.exec('DELETE FROM activity_history_records WHERE occurrence != 0') case 'remove-duplicates':
} else if (change.type === 'remove') { database.exec('DELETE FROM activity_history_records WHERE occurrence != 0')
remove.run(change.id) break
} else { case 'remove':
// The first occurrence stays first when moved to the front, so its key is stable. remove.run(change.id)
const key = activityRecordKey(change.record.id, 0) break
const json = JSON.stringify(change.record) case 'upsert': {
const seq = (): number => (nextSeq.get() as { seq: number }).seq // The first occurrence stays first when moved to the front, so its key is stable.
if (!exists.get(key)) insert.run(key, json, seq()) const key = activityRecordKey(change.record.id, 0)
else if (change.position === 'front') moveToFront.run(json, seq(), key) const json = JSON.stringify(change.record)
else replaceInPlace.run(json, key, json) if (!exists.get(key)) insert.run(key, json, seq())
else if (change.position === 'front') moveToFront.run(json, seq(), key)
else replaceInPlace.run(json, key, json)
break
}
case 'upsert-call': {
const incoming = change.record
const [existing] = select(
"request_id = ? AND kind = ? AND json_extract(record_json, '$.callId') = ?", 1,
incoming.requestId, incoming.kind, incoming.callId!
)
if (!existing) {
insert.run(activityRecordKey(incoming.id, 0), JSON.stringify(incoming), seq())
calls.push(incoming)
break
}
const before = existing.record
const unchanged = before.conversationId === incoming.conversationId && before.title === incoming.title &&
before.detail === incoming.detail && before.status === incoming.status
const front = existing.order_seq === (frontSeq.get() as { seq: number }).seq
const next = unchanged ? before : { ...incoming, id: before.id, createdAt: before.createdAt, scope: before.scope }
calls.push(next)
if (unchanged && front) break
moveToFront.run(JSON.stringify(activityRecordSchema.parse(next)), seq(), existing.record_key)
break
}
case 'update-request':
for (const row of select("request_id = ? AND kind = 'request'", -1, change.requestId)) {
rewrite(row.record_key, { ...row.record, status: change.status, detail: change.detail ?? row.record.detail })
}
break
case 'resolve-approval': {
const [row] = select("conversation_id = ? AND kind = 'approval' AND status = 'pending'", 1,
change.conversationId)
if (row) {
rewrite(row.record_key, { ...row.record, status: change.status, detail: `${row.record.detail}\n${change.detailLine}` })
}
break
}
case 'settle-request':
for (const row of select("request_id = ? AND kind != 'request' AND status_group = 'active'", -1,
change.requestId)) {
rewrite(row.record_key, { ...row.record, status: change.status, detail: `${row.record.detail}\n${change.detailLine}` })
}
break
case 'remove-request':
for (const row of select('request_id = ?', -1, change.requestId)) remove.run(row.record.id)
break
case 'remove-call':
for (const row of select(
"request_id = ? AND kind = 'tool' AND json_extract(record_json, '$.callId') = ?", -1,
change.requestId, change.callId
)) remove.run(row.record.id)
break
} }
} }
if (update.legacyHistoryMayBeIncomplete !== undefined) { if (update.legacyHistoryMayBeIncomplete !== undefined) {
this.writeIncomplete(update.legacyHistoryMayBeIncomplete) this.writeIncomplete(update.legacyHistoryMayBeIncomplete)
} }
}) })
return { calls }
} }
/** /**
+3 -2
View File
@@ -41,6 +41,7 @@ import type {
ActivityHistoryPage, ActivityHistoryPage,
ActivityHistorySnapshot, ActivityHistorySnapshot,
ActivityHistorySummary, ActivityHistorySummary,
ActivityHistoryUpdateResult,
ConversationQueueItem, ConversationQueueItem,
ConversationAnsweredQuestion, ConversationAnsweredQuestion,
ConversationBranchInput, ConversationBranchInput,
@@ -5310,8 +5311,8 @@ export class AssistantDatabase {
} }
/** Applies incremental changes in one transaction (PERF-15). */ /** Applies incremental changes in one transaction (PERF-15). */
updateActivityHistory(input: unknown): void { updateActivityHistory(input: unknown): ActivityHistoryUpdateResult {
this.activityHistory().update(input) return this.activityHistory().update(input)
} }
/** Deletes every activity record in one transaction. */ /** Deletes every activity record in one transaction. */
+2 -7
View File
@@ -6869,7 +6869,7 @@ export function registerIpcHandlers(
ipcChannels.activityHistoryUpdate, ipcChannels.activityHistoryUpdate,
(event, input: unknown) => { (event, input: unknown) => {
assertTrustedSender(event, window) assertTrustedSender(event, window)
assistantDatabase.updateActivityHistory(input) return assistantDatabase.updateActivityHistory(input)
} }
) )
registerHandler(ipcChannels.activityHistoryClear, (event) => { registerHandler(ipcChannels.activityHistoryClear, (event) => {
@@ -6886,12 +6886,7 @@ export function registerIpcHandlers(
}) })
registerHandler(ipcChannels.activityHistoryReconcile, (event, input: unknown) => { registerHandler(ipcChannels.activityHistoryReconcile, (event, input: unknown) => {
assertTrustedSender(event, window) assertTrustedSender(event, window)
const request = activityHistoryReconcileRequestSchema.parse(input) return assistantDatabase.reconcileActivityHistory(activityHistoryReconcileRequestSchema.parse(input))
// Requests still running in Main are active too, whichever window started them.
return assistantDatabase.reconcileActivityHistory({
...request,
activeRequestIds: [...new Set([...request.activeRequestIds, ...activeRequests.keys()])]
})
}) })
registerHandler(ipcChannels.tokenUsageSummary, (event) => { registerHandler(ipcChannels.tokenUsageSummary, (event) => {
+3 -3
View File
@@ -74,6 +74,7 @@ import type {
ActivityHistorySummary, ActivityHistorySummary,
ActivityHistorySummaryRequest, ActivityHistorySummaryRequest,
ActivityHistoryUpdate, ActivityHistoryUpdate,
ActivityHistoryUpdateResult,
ActivityRecord, ActivityRecord,
TokenUsageSummary, TokenUsageSummary,
ConversationBranchInput, ConversationBranchInput,
@@ -1296,9 +1297,8 @@ const desktopApi: DesktopApi = {
legacyHistoryMayBeIncomplete legacyHistoryMayBeIncomplete
}) })
}, },
update: async (update: ActivityHistoryUpdate) => { update: (update: ActivityHistoryUpdate) =>
await ipcRenderer.invoke(ipcChannels.activityHistoryUpdate, update) ipcRenderer.invoke(ipcChannels.activityHistoryUpdate, update) as Promise<ActivityHistoryUpdateResult>,
},
clear: async () => { clear: async () => {
await ipcRenderer.invoke(ipcChannels.activityHistoryClear) await ipcRenderer.invoke(ipcChannels.activityHistoryClear)
}, },
+93
View File
@@ -991,3 +991,96 @@ describe('ActivityPanel', () => {
expect(screen.queryByRole('button', { name: '刷新' })).not.toBeInTheDocument() expect(screen.queryByRole('button', { name: '刷新' })).not.toBeInTheDocument()
}) })
}) })
describe('ActivityPanel paged by Main', () => {
function makeRandom(seed: number): () => number {
let state = seed
return () => (state = (state * 1664525 + 1013904223) >>> 0) / 4294967296
}
function randomList(random: () => number, length: number): ActivityRecord[] {
const pick = <T,>(items: readonly T[]): T => items[Math.floor(random() * items.length)]!
const list: ActivityRecord[] = []
for (let index = 0; index < length; index++) {
const duplicate = list.length > 0 && random() < 0.08
list.push({
id: duplicate ? pick(list).id : `random-${index}`,
conversationId: `conversation-${Math.floor(random() * 5)}`,
requestId: `request-${Math.floor(random() * 8)}`,
scope: random() < 0.5 ? { kind: 'project', projectId: 'p1', projectName: '项目一' } : { kind: 'global' },
kind: pick(['request', 'tool', 'approval', 'subagent', 'result'] as const),
title: `标题 ${index}`,
detail: `详情 ${index}`,
status: pick(['pending', 'running', 'completed', 'failed', 'denied', 'cancelled', 'interrupted'] as const),
createdAt: Date.UTC(2026, 0, 1) + Math.floor(random() * 30) * 1000
})
}
return list
}
const filterButtons = (container: HTMLElement): HTMLButtonElement[] =>
[...container.querySelectorAll<HTMLButtonElement>('.activity-panel__toolbar [role="group"] button')]
/** The visible text and status classes of the page; what the user can compare. */
function snapshot(container: HTMLElement): string {
return [...container.querySelectorAll('.activity-panel__toolbar, .activity-project, .activity-panel__load-more')]
.map((element) => `${element.textContent ?? ''}|${[...element.querySelectorAll('.status-badge')].map((badge) => badge.className).join(',')}`)
.join('\n')
}
it('shows the same counts, groups, titles, status and clear count as the full-list computation', async () => {
const { firstOccurrences, referenceActivitySummary } = await import('../../shared/activity-history-reference')
for (const seed of [2, 13, 77, 404]) {
const random = makeRandom(seed)
const stored = randomList(random, 120 + Math.floor(random() * 500))
// The former page got the whole shown list (first occurrences).
const full = firstOccurrences(stored)
const props = { tokenUsage: makeTokenUsage(), onClear: vi.fn(), onOpenConversation: vi.fn() }
for (const filter of ['all', 'active', 'failed'] as const) {
const before = render(<ActivityPanel {...props} records={full} />)
if (filter !== 'all') {
fireEvent.click(filterButtons(before.container)[filter === 'active' ? 1 : 2]!)
}
const expected = snapshot(before.container)
before.unmount()
// Now: only a loaded prefix (as Main pages it) plus Main's summary.
const matching = full.filter((record) =>
filter === 'all' ? true : filter === 'active'
? record.status === 'pending' || record.status === 'running'
: ['failed', 'denied', 'cancelled', 'interrupted'].includes(record.status))
const loaded = matching.slice(0, 500)
const summary = referenceActivitySummary(stored, [...new Set(loaded.map((record) => record.conversationId))])
const after = render(
<ActivityPanel
{...props}
paging={{ filter, summary, hasMore: loaded.length < matching.length, onFilterChange: vi.fn(), onLoadMore: vi.fn() }}
records={loaded}
/>
)
expect(snapshot(after.container)).toBe(expected)
after.unmount()
}
}
})
it('loads the next page behind "load more" and switches filters through the callbacks', () => {
const records = Array.from({ length: 500 }, (_, index) => makeRecord(index))
const onLoadMore = vi.fn()
const onFilterChange = vi.fn()
const summary = { counts: { all: 1_200, active: 0, failed: 0 }, conversations: {}, legacyHistoryMayBeIncomplete: false }
const { container } = render(
<ActivityPanel
onClear={vi.fn()}
onOpenConversation={vi.fn()}
paging={{ filter: 'all', summary, hasMore: true, onFilterChange, onLoadMore }}
records={records}
tokenUsage={makeTokenUsage()}
/>
)
expect(container.querySelectorAll('.activity-list__item')).toHaveLength(500)
const loadMore = container.querySelector<HTMLButtonElement>('.activity-panel__load-more')!
expect(loadMore.textContent).toContain('700')
fireEvent.click(loadMore)
expect(onLoadMore).toHaveBeenCalledOnce()
fireEvent.click(filterButtons(container)[2]!)
expect(onFilterChange).toHaveBeenCalledWith('failed')
expect(container.textContent).toContain('1,200')
})
})
+116 -33
View File
@@ -1,8 +1,9 @@
import { Activity, ChevronRight, RefreshCw, Trash2 } from 'lucide-react' import { Activity, ChevronRight, RefreshCw, Trash2 } from 'lucide-react'
import { memo, useMemo, useState } from 'react' import { memo, useCallback, useMemo, useState } from 'react'
import { useTranslation } from 'react-i18next' import { useTranslation } from 'react-i18next'
import { InlineHelp } from './InlineHelp' import { InlineHelp } from './InlineHelp'
import type { import type {
ActivityHistorySummary,
AssistantProject, AssistantProject,
TokenUsageSummary TokenUsageSummary
} from '../../shared/assistant-contracts' } from '../../shared/assistant-contracts'
@@ -41,7 +42,23 @@ const activityRenderBatchSize = 500
export type ActivityPanelProps = { export type ActivityPanelProps = {
projects?: readonly AssistantProject[] projects?: readonly AssistantProject[]
/**
* The records to show, newest first. With `paging` they are the loaded
* pages of `paging.filter`; without, the whole history.
*/
records: readonly ActivityRecord[] records: readonly ActivityRecord[]
/**
* Paged history (PERF-15): counts, titles and conversation status come
* from `summary`, computed by Main over the whole history; "load more"
* and filter changes go through the callbacks.
*/
paging?: {
filter: ActivityFilter
summary?: ActivityHistorySummary
hasMore: boolean
onFilterChange: (filter: ActivityFilter) => void
onLoadMore: () => void
}
tokenUsage: TokenUsageSummary tokenUsage: TokenUsageSummary
onClear: () => void onClear: () => void
onOpenConversation: (conversationId: string) => void onOpenConversation: (conversationId: string) => void
@@ -187,11 +204,15 @@ function getConversationTitles(
return conversationTitles return conversationTitles
} }
function groupActivityRecordsByProject( type ConversationFacts = {
records: readonly ActivityRecord[], titles: ReadonlyMap<string, string>
status: (conversationId: string, shown: readonly ActivityRecord[]) => ActivityRecord['status']
}
/** Titles and status from the whole list (unpaged). */
function conversationFactsFromRecords(
allRecords: readonly ActivityRecord[] allRecords: readonly ActivityRecord[]
): ProjectActivityGroup[] { ): ConversationFacts {
const conversationTitles = getConversationTitles(allRecords)
const allConversationRecords = new Map<string, ActivityRecord[]>() const allConversationRecords = new Map<string, ActivityRecord[]>()
for (const record of allRecords) { for (const record of allRecords) {
const conversationRecords = const conversationRecords =
@@ -199,6 +220,35 @@ function groupActivityRecordsByProject(
conversationRecords.push(record) conversationRecords.push(record)
allConversationRecords.set(record.conversationId, conversationRecords) allConversationRecords.set(record.conversationId, conversationRecords)
} }
return {
titles: getConversationTitles(allRecords),
status: (conversationId, shown) =>
conversationStatus(allConversationRecords.get(conversationId) ?? shown)
}
}
/** Titles and status Main computed over the whole history; loaded records fill gaps. */
function conversationFactsFromSummary(
summary: ActivityHistorySummary,
loaded: readonly ActivityRecord[]
): ConversationFacts {
const fallback = conversationFactsFromRecords(loaded)
const titles = new Map(fallback.titles)
for (const [conversationId, facts] of Object.entries(summary.conversations)) {
if (facts.title !== undefined) titles.set(conversationId, facts.title)
}
return {
titles,
status: (conversationId, shown) =>
summary.conversations[conversationId]?.status ?? fallback.status(conversationId, shown)
}
}
function groupActivityRecordsByProject(
records: readonly ActivityRecord[],
facts: ConversationFacts
): ProjectActivityGroup[] {
const conversationTitles = facts.titles
const projectGroups = new Map< const projectGroups = new Map<
string, string,
{ {
@@ -231,9 +281,7 @@ function groupActivityRecordsByProject(
conversationTitles.get(conversationId) ?? items[0]!.title, conversationTitles.get(conversationId) ?? items[0]!.title,
records: items, records: items,
latestAt: Math.max(...items.map((record) => record.createdAt)), latestAt: Math.max(...items.map((record) => record.createdAt)),
status: conversationStatus( status: facts.status(conversationId, items)
allConversationRecords.get(conversationId) ?? items
)
}) })
) )
return { return {
@@ -254,13 +302,23 @@ export const ActivityPanel = memo(function ActivityPanel({
tokenUsage, tokenUsage,
onClear, onClear,
onOpenConversation, onOpenConversation,
onRefresh onRefresh,
paging
}: ActivityPanelProps): React.JSX.Element { }: ActivityPanelProps): React.JSX.Element {
const { t, i18n } = useTranslation('activity') const { t, i18n } = useTranslation('activity')
const { t: tWorkspace } = useTranslation('workspace') const { t: tWorkspace } = useTranslation('workspace')
const [activeView, setActiveView] = const [activeView, setActiveView] =
useState<ActivityView>('tasks') useState<ActivityView>('tasks')
const [filter, setFilter] = useState<ActivityFilter>('all') const [localFilter, setLocalFilter] = useState<ActivityFilter>('all')
const filter = paging?.filter ?? localFilter
const onFilterChange = paging?.onFilterChange
const setFilter = useCallback(
(next: ActivityFilter): void => {
if (onFilterChange) onFilterChange(next)
else setLocalFilter(next)
},
[onFilterChange]
)
const [tokenGroup, setTokenGroup] = const [tokenGroup, setTokenGroup] =
useState<Exclude<TokenUsageGroup, 'source'>>('project') useState<Exclude<TokenUsageGroup, 'source'>>('project')
const [systemTokenGroup, setSystemTokenGroup] = const [systemTokenGroup, setSystemTokenGroup] =
@@ -414,6 +472,7 @@ export const ActivityPanel = memo(function ActivityPanel({
}), }),
[projectDisplayNames, tokenUsage] [projectDisplayNames, tokenUsage]
) )
const summary = paging?.summary
const matchingRecords = useMemo( const matchingRecords = useMemo(
() => displayRecords.filter((record) => matchesFilter(record, filter)), () => displayRecords.filter((record) => matchesFilter(record, filter)),
[displayRecords, filter] [displayRecords, filter]
@@ -422,9 +481,16 @@ export const ActivityPanel = memo(function ActivityPanel({
() => matchingRecords.slice(0, visibleRecordCount), () => matchingRecords.slice(0, visibleRecordCount),
[matchingRecords, visibleRecordCount] [matchingRecords, visibleRecordCount]
) )
const conversationFacts = useMemo(
() =>
summary
? conversationFactsFromSummary(summary, displayRecords)
: conversationFactsFromRecords(displayRecords),
[displayRecords, summary]
)
const projectGroups = useMemo( const projectGroups = useMemo(
() => groupActivityRecordsByProject(filteredRecords, displayRecords), () => groupActivityRecordsByProject(filteredRecords, conversationFacts),
[displayRecords, filteredRecords] [conversationFacts, filteredRecords]
) )
const timelineBounds = useMemo(() => { const timelineBounds = useMemo(() => {
const timestamps = filteredRecords.map((record) => record.createdAt) const timestamps = filteredRecords.map((record) => record.createdAt)
@@ -450,12 +516,23 @@ export const ActivityPanel = memo(function ActivityPanel({
const selectedTimelineRecord = filteredRecords.find( const selectedTimelineRecord = filteredRecords.find(
(record) => record.id === selectedTimelineRecordId (record) => record.id === selectedTimelineRecordId
) )
const conversationTitles = useMemo( const conversationTitles = conversationFacts.titles
() => getConversationTitles(displayRecords), // Before the first summary arrives, the loaded records are the best estimate.
[displayRecords] const totalCount = summary?.counts.all ?? displayRecords.length
) const activeCount =
const activeCount = displayRecords.filter(isActive).length summary?.counts.active ?? displayRecords.filter(isActive).length
const failedCount = displayRecords.filter(isFailed).length const failedCount =
summary?.counts.failed ?? displayRecords.filter(isFailed).length
const matchingTotal =
filter === 'all' ? totalCount : filter === 'active' ? activeCount : failedCount
// Records of this filter that exist but are not shown yet.
const remainingCount = paging
? Math.max(
matchingTotal - filteredRecords.length,
matchingRecords.length - filteredRecords.length,
paging.hasMore ? 1 : 0
)
: matchingRecords.length - filteredRecords.length
const filters: ReadonlyArray<{ const filters: ReadonlyArray<{
value: ActivityFilter value: ActivityFilter
label: string label: string
@@ -463,8 +540,8 @@ export const ActivityPanel = memo(function ActivityPanel({
{ {
value: 'all', value: 'all',
label: t('filters.all', { label: t('filters.all', {
count: displayRecords.length, count: totalCount,
formattedCount: formatCount(displayRecords.length) formattedCount: formatCount(totalCount)
}) })
}, },
{ {
@@ -691,19 +768,19 @@ export const ActivityPanel = memo(function ActivityPanel({
/> />
<DestructiveConfirmActions <DestructiveConfirmActions
confirmAriaLabel={t('clear.confirmAriaLabel', { confirmAriaLabel={t('clear.confirmAriaLabel', {
count: displayRecords.length, count: totalCount,
formattedCount: formatCount(displayRecords.length) formattedCount: formatCount(totalCount)
})} })}
confirmLabel={t('clear.confirmLabel', { confirmLabel={t('clear.confirmLabel', {
count: displayRecords.length, count: totalCount,
formattedCount: formatCount(displayRecords.length) formattedCount: formatCount(totalCount)
})} })}
confirming={confirmingClear} confirming={confirmingClear}
disabled={!confirmingClear && displayRecords.length === 0} disabled={!confirmingClear && totalCount === 0 && displayRecords.length === 0}
icon={<Trash2 aria-hidden="true" size={15} />} icon={<Trash2 aria-hidden="true" size={15} />}
message={t('clear.message', { message={t('clear.message', {
count: displayRecords.length, count: totalCount,
formattedCount: formatCount(displayRecords.length) formattedCount: formatCount(totalCount)
})} })}
onCancel={() => setConfirmingClear(false)} onCancel={() => setConfirmingClear(false)}
onConfirm={() => { onConfirm={() => {
@@ -739,20 +816,26 @@ export const ActivityPanel = memo(function ActivityPanel({
/> />
) )
const loadMoreRecords = const loadMoreRecords =
filteredRecords.length < matchingRecords.length ? ( remainingCount > 0 ? (
<button <button
className="secondary-button activity-panel__load-more" className="secondary-button activity-panel__load-more"
onClick={() => onClick={() => {
// Show the next loaded batch at once; load another page behind it.
setVisibleRecordCount( setVisibleRecordCount(
(current) => current + activityRenderBatchSize (current) => current + activityRenderBatchSize
) )
} if (
paging?.hasMore &&
matchingRecords.length <
visibleRecordCount + 2 * activityRenderBatchSize
) {
paging.onLoadMore()
}
}}
type="button" type="button"
> >
{t('records.loadMore', { {t('records.loadMore', {
remaining: formatCount( remaining: formatCount(remainingCount)
matchingRecords.length - filteredRecords.length
)
})} })}
</button> </button>
) : null ) : null
+58 -10
View File
@@ -14222,20 +14222,68 @@ describe("App", () => {
expect(await screen.findByLabelText("项目:默认项目")).toHaveClass("scope-badge"); expect(await screen.findByLabelText("项目:默认项目")).toHaveClass("scope-badge");
await waitFor(() => { await waitFor(() => {
expect(api.activityHistory.replace).toHaveBeenCalledWith( expect(savedActivity).toEqual([
[ expect.objectContaining({
expect.objectContaining({ id: "legacy-activity",
id: "legacy-activity", detail: "x".repeat(4_000),
detail: "x".repeat(4_000), }),
}), ]);
], expect(api.activityHistory.update).toHaveBeenCalledWith({
true, changes: [],
); legacyHistoryMayBeIncomplete: true,
});
expect(localStorage.getItem(ACTIVITY_STORAGE_KEY)).toBeNull(); expect(localStorage.getItem(ACTIVITY_STORAGE_KEY)).toBeNull();
}); });
// The legacy migration no longer writes the whole list.
expect(api.activityHistory.replace).not.toHaveBeenCalled();
}); });
it("keeps at most one page of activity after startup and trims when the page is left", async () => {
savedActivity = Array.from({ length: 1_200 }, (_, index) => ({
id: `stored-${index}`,
conversationId: `stored-conversation-${index % 7}`,
requestId: `stored-request-${index}`,
scope: { kind: "global" as const },
kind: index % 3 === 0 ? ("request" as const) : ("tool" as const),
title: `存量记录 ${index}`,
detail: "detail",
status: index % 5 === 0 ? ("failed" as const) : ("completed" as const),
createdAt: 1_200 - index,
}));
render(<App />);
// Startup: Main reconciles, then one page of at most 200 records.
await waitFor(() => expect(api.activityHistory.page).toHaveBeenCalledWith({ limit: 200 }));
expect(api.activityHistory.get).not.toHaveBeenCalled();
expect(api.activityHistory.reconcile).toHaveBeenCalledOnce();
expect(vi.mocked(api.activityHistory.reconcile).mock.invocationCallOrder[0])
.toBeLessThan(vi.mocked(api.activityHistory.page).mock.invocationCallOrder[0]!);
expect(api.activityHistory.page).toHaveBeenCalledTimes(1);
fireEvent.click(screen.getByRole("button", { name: "运行记录" }));
// The page loads a full visible batch (500, as before paging) and counts the whole history.
expect(await screen.findByRole("button", { name: "全部 1,200" })).toBeInTheDocument();
expect(api.activityHistory.page).toHaveBeenLastCalledWith({ limit: 500 });
const records = (): number =>
[...document.querySelectorAll(".activity-group small")]
.reduce((total, element) => total + Number(/\d[\d,]*/u.exec(element.textContent ?? "")![0]!.replace(",", "")), 0);
await waitFor(() => expect(records()).toBe(500));
expect(screen.getByRole("button", { name: "继续显示(剩余 700 条)" })).toBeInTheDocument();
fireEvent.click(screen.getByRole("button", { name: "清空记录" }));
expect(screen.getByText("永久清空 1,200 条活动记录?此操作不可撤销。")).toBeInTheDocument();
expect(screen.getByRole("button", { name: "确认清空 1,200 条活动记录" })).toBeInTheDocument();
fireEvent.click(screen.getByRole("button", { name: "取消" }));
fireEvent.click(screen.getByRole("button", { name: "继续显示(剩余 700 条)" }));
await waitFor(() => expect(records()).toBe(1_000));
await waitFor(() => expect(api.activityHistory.page).toHaveBeenLastCalledWith({ before: expect.any(Number), limit: 500 }));
// Leaving the page keeps one visible batch; coming back shows the same first rows.
fireEvent.click(screen.getByRole("button", { name: "对话" }));
fireEvent.click(screen.getByRole("button", { name: "运行记录" }));
expect(await screen.findByRole("button", { name: "全部 1,200" })).toBeInTheDocument();
expect(records()).toBe(500);
}, 30_000);
it("marks the current primary navigation page and hides decorative icons", async () => { it("marks the current primary navigation page and hides decorative icons", async () => {
render(<App />); render(<App />);
+27 -13
View File
@@ -98,6 +98,7 @@ import {
runtimeProviderLabel, runtimeProviderLabel,
} from "./runtime-selection"; } from "./runtime-selection";
import type { import type {
ActivityHistoryFilter,
AssistantProject, AssistantProject,
AssistantArtifact, AssistantArtifact,
AssistantMemory, AssistantMemory,
@@ -202,7 +203,7 @@ import {
loadLegacyActivityHistory, loadLegacyActivityHistory,
type ActivityRecord, type ActivityRecord,
} from "./activity-store"; } from "./activity-store";
import { useActivityPanelRecords } from "./activity-selectors"; import { useActivityPanel } from "./activity-selectors";
import { useActivityHistorySync } from "./activity-sync"; import { useActivityHistorySync } from "./activity-sync";
import { import {
KnowledgeCitationDialog, KnowledgeCitationDialog,
@@ -1769,9 +1770,8 @@ function App(): React.JSX.Element {
const citationRequestRef = useRef(0); const citationRequestRef = useRef(0);
const failedKnowledgeLibraryIdRef = useRef<string | undefined>(undefined); const failedKnowledgeLibraryIdRef = useRef<string | undefined>(undefined);
const [legacyActivityHistory] = useState(loadLegacyActivityHistory); const [legacyActivityHistory] = useState(loadLegacyActivityHistory);
const [activityStore] = useState(() => createActivityStore( const [activityStore] = useState(() => createActivityStore(legacyActivityHistory.records));
legacyActivityHistory.records, legacyActivityHistory.historyMayBeIncomplete)); const activityPanel = useActivityPanel(activityStore);
const activityPanelRecords = useActivityPanelRecords(activityStore);
const activeRuns = useRef(new Map<string, ActiveRun>()); const activeRuns = useRef(new Map<string, ActiveRun>());
const preparingConversations = useRef(new Set<string>()); const preparingConversations = useRef(new Set<string>());
const [activeConversationIds, setActiveConversationIds] = useState< const [activeConversationIds, setActiveConversationIds] = useState<
@@ -3267,7 +3267,8 @@ function App(): React.JSX.Element {
return stop; return stop;
}, [conversationStoreReady, conversationPersistence]); }, [conversationStoreReady, conversationPersistence]);
const persistActivityHistory = useActivityHistorySync(activityStore, legacyActivityHistory, { const activitySync = useActivityHistorySync(activityStore, legacyActivityHistory, {
activeRequestIds: () => new Set(activeRuns.current.keys()),
onReadFailed: () => notify({ onReadFailed: () => notify({
tone: "error", message: tRef.current("notices.activityHistoryReadFailed"), dedupeKey: "activity-history-read", tone: "error", message: tRef.current("notices.activityHistoryReadFailed"), dedupeKey: "activity-history-read",
}), }),
@@ -3278,9 +3279,10 @@ function App(): React.JSX.Element {
}), }),
// The loaded history appears at once when the Activity page is open. // The loaded history appears at once when the Activity page is open.
onLoaded: () => { onLoaded: () => {
if (viewRef.current === "activity") activityStore.refreshPanel(); if (viewRef.current === "activity") void activitySync.show();
}, },
}); });
const persistActivityHistory = activitySync.flush;
useEffect( useEffect(
() => () =>
@@ -3457,9 +3459,11 @@ function App(): React.JSX.Element {
}; };
}, []); }, []);
const refreshActivityPanelRecords = activityStore.refreshPanel; const { show: showActivity, hide: hideActivity } = activitySync;
const refreshActivityPanelRecords = useCallback((): void => void showActivity(), [showActivity]);
// Layout effect so entering the page never paints a stale snapshot. // Layout effect so entering the page never paints a stale snapshot; leaving
// it keeps only the first page loaded.
useLayoutEffect(() => { useLayoutEffect(() => {
if (view !== "activity") { if (view !== "activity") {
return; return;
@@ -3469,8 +3473,11 @@ function App(): React.JSX.Element {
refreshActivityPanelRecords, refreshActivityPanelRecords,
activityPanelRefreshIntervalMs, activityPanelRefreshIntervalMs,
); );
return () => window.clearInterval(interval); return () => {
}, [refreshActivityPanelRecords, view]); window.clearInterval(interval);
hideActivity();
};
}, [hideActivity, refreshActivityPanelRecords, view]);
const resumeProjectConversationQueues = useCallback( const resumeProjectConversationQueues = useCallback(
(projectId: string): void => { (projectId: string): void => {
@@ -4042,7 +4049,7 @@ function App(): React.JSX.Element {
}, [refreshTokenUsage, view]); }, [refreshTokenUsage, view]);
const manualRefreshActivity = useCallback(async (): Promise<void> => { const manualRefreshActivity = useCallback(async (): Promise<void> => {
refreshActivityPanelRecords(); await showActivity();
try { try {
await refreshTokenUsage(); await refreshTokenUsage();
} catch { } catch {
@@ -4051,7 +4058,13 @@ function App(): React.JSX.Element {
message: tRef.current("notices.tokenUsageReadFailed"), message: tRef.current("notices.tokenUsageReadFailed"),
}); });
} }
}, [notify, refreshActivityPanelRecords, refreshTokenUsage]); }, [notify, showActivity, refreshTokenUsage]);
const { loadMore: loadMoreActivity, setFilter: setActivityFilter } = activitySync;
const activityPaging = useMemo(() => ({
filter: activityPanel.page.filter, summary: activityPanel.summary, hasMore: activityPanel.page.hasMore,
onFilterChange: (filter: ActivityHistoryFilter) => void setActivityFilter(filter),
onLoadMore: () => void loadMoreActivity(),
}), [activityPanel, loadMoreActivity, setActivityFilter]);
useInitialArtifactSync(taskStore, () => useInitialArtifactSync(taskStore, () =>
notify({ notify({
@@ -7564,7 +7577,8 @@ function App(): React.JSX.Element {
onOpenConversation={openActivityConversation} onOpenConversation={openActivityConversation}
onRefresh={manualRefreshActivity} onRefresh={manualRefreshActivity}
projects={projects} projects={projects}
records={activityPanelRecords} paging={activityPaging}
records={activityPanel.page.records}
tokenUsage={tokenUsage} tokenUsage={tokenUsage}
/> />
</Suspense> </Suspense>
+22 -8
View File
@@ -1,12 +1,26 @@
import { useSyncExternalStore } from 'react' import { useCallback, useRef, useSyncExternalStore } from 'react'
import type { ActivityRecord, ActivityStore } from './activity-store' import type { ActivityHistorySummary } from '../../shared/assistant-contracts'
import type { ActivityPageState, ActivityStore } from './activity-store'
export type ActivityPanelView = {
page: ActivityPageState
summary?: ActivityHistorySummary
}
/** /**
* The records the Activity page shows: a snapshot refreshed on entry, by the * What the Activity page shows: the snapshot of the loaded pages (refreshed
* page timer and the refresh button, so streaming runs do not re-render it. * on entry, by the page timer and the refresh button, so streaming runs do
* not re-render it) and Main's summary. The same object while both are
* unchanged.
*/ */
export function useActivityPanelRecords( export function useActivityPanel(store: ActivityStore): ActivityPanelView {
store: ActivityStore const cache = useRef<ActivityPanelView | undefined>(undefined)
): readonly ActivityRecord[] { const getSnapshot = useCallback((): ActivityPanelView => {
return useSyncExternalStore(store.subscribePanel, store.getPanelRecords) const page = store.getPanel()
const summary = store.getSummary()
const cached = cache.current
if (cached && cached.page === page && cached.summary === summary) return cached
return (cache.current = { page, ...(summary ? { summary } : {}) })
}, [store])
return useSyncExternalStore(store.subscribePanel, getSnapshot)
} }
+275 -171
View File
@@ -3,9 +3,14 @@ import i18n from './i18n'
import { import {
activityRecordSchema, activityRecordSchema,
type ActivityHistoryChange, type ActivityHistoryChange,
type ActivityHistoryFilter,
type ActivityHistoryPage,
type ActivityHistoryReconcileRequest,
type ActivityHistorySummary,
type ActivityRecord as SharedActivityRecord, type ActivityRecord as SharedActivityRecord,
type AssistantTask type AssistantTask
} from '../../shared/assistant-contracts' } from '../../shared/assistant-contracts'
import { applyActivityChanges } from '../../shared/activity-history-reference'
export const ACTIVITY_STORAGE_KEY = 'goodbuddy.activity-records.v1' export const ACTIVITY_STORAGE_KEY = 'goodbuddy.activity-records.v1'
@@ -225,36 +230,71 @@ export function clearLegacyActivityHistory(
} }
/** /**
* What still has to reach Main, in order: incremental changes, or a marker * What still has to reach Main, in order: incremental changes, a reconcile
* that the whole history must be cleared or written in full first. * against the stored tasks, the legacy-incomplete flag, or clearing it all.
*/ */
export type ActivityOutboxEntry = export type ActivityOutboxEntry =
| ActivityHistoryChange | ActivityHistoryChange
| {
type: 'reconcile'
request: ActivityHistoryReconcileRequest
/** Kept to apply the same reconcile to pages read before Main ran it. */
tasks: readonly AssistantTask[]
}
| { type: 'legacy-flag'; value: boolean }
| { type: 'clear' } | { type: 'clear' }
| { type: 'replace' }
type Listener = () => void type Listener = () => void
export type ActivityStore = ReturnType<typeof createActivityStore> export type ActivityStore = ReturnType<typeof createActivityStore>
/** The loaded records of one filter. */
export type ActivityPageState = {
filter: ActivityHistoryFilter
/** Newest first: a prefix of Main's list for `filter`, plus this session's unsaved records. */
records: readonly ActivityRecord[]
/** `before` of the next page; undefined when everything is loaded or not known (trimmed). */
nextBefore?: number
/** Whether more records may exist than are loaded. */
hasMore: boolean
}
export const ACTIVITY_FIRST_PAGE_SIZE = 200
export const ACTIVITY_PAGE_SIZE = 500
export function matchesActivityFilter(record: ActivityRecord, filter: ActivityHistoryFilter): boolean {
if (filter === 'active') return record.status === 'pending' || record.status === 'running'
if (filter === 'failed') {
return record.status === 'failed' || record.status === 'denied' ||
record.status === 'cancelled' || record.status === 'interrupted'
}
return true
}
export const isActivityChange = (entry: ActivityOutboxEntry): entry is ActivityHistoryChange =>
entry.type !== 'reconcile' && entry.type !== 'clear' && entry.type !== 'legacy-flag'
/** /**
* Activity history, kept outside App state (P2/P4). Every domain action * Activity history, kept outside App state (P2/P4). Main owns the history;
* updates the newest-first list and appends the matching contract changes to * the renderer keeps only the pages it loaded. Every domain action applies
* the outbox, which activity-sync.ts sends to Main; the list is never written * its contract change to the loaded records (`applyActivityChanges`, the
* back as a whole except for the legacy migration fallback. * same semantics Main applies) and appends it to the outbox, which
* activity-sync.ts sends to Main. Main matches changes by content, so
* records that are not loaded are updated too.
* *
* The Activity page shows a snapshot (`getPanelRecords`) instead of the live * The Activity page shows a snapshot (`getPanel`) instead of the live list so
* list so streaming runs do not re-render it; `refreshPanel` updates it. * streaming runs do not re-render it; `refreshPanel` updates it.
*/ */
export function createActivityStore( export function createActivityStore(initial: readonly ActivityRecord[] = []) {
initial: readonly ActivityRecord[] = [], let page: ActivityPageState = { filter: 'all', records: mergeActivityRecords(initial, []), hasMore: true }
initialLegacyIncomplete = false let panel = page
) { let summary: ActivityHistorySummary | undefined
let records: readonly ActivityRecord[] = mergeActivityRecords(initial, [])
let panelRecords = records
let ready = false let ready = false
let legacyIncomplete = initialLegacyIncomplete /** Bumped whenever the loaded pages are replaced; stale page responses are dropped. */
let generation = 0
let outbox: ActivityOutboxEntry[] = [] let outbox: ActivityOutboxEntry[] = []
/** Taken by the sync and not acknowledged yet. */
let sending: ActivityOutboxEntry[] = []
const listeners = new Set<Listener>() const listeners = new Set<Listener>()
const panelListeners = new Set<Listener>() const panelListeners = new Set<Listener>()
const outboxListeners = new Set<Listener>() const outboxListeners = new Set<Listener>()
@@ -267,187 +307,202 @@ export function createActivityStore(
set.delete(listener) set.delete(listener)
} }
} }
const setPage = (next: ActivityPageState): void => {
page = next
notify(listeners)
}
const enqueue = (entry: ActivityOutboxEntry): void => { 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 const wasEmpty = outbox.length === 0
records = next // Only the latest reconcile matters: it reads the stored tasks when it runs.
for (const change of changes) enqueue(change) if (entry.type === 'reconcile' && outbox.at(-1)?.type === 'reconcile') outbox[outbox.length - 1] = entry
notify(listeners) else outbox.push(entry)
if (wasEmpty && outbox.length > 0) notify(outboxListeners) if (wasEmpty) notify(outboxListeners)
} }
/** Maps records in place; each changed one becomes an in-place upsert. */ const apply = (change: ActivityHistoryChange): void => {
const mapInPlace = ( const records = applyActivityChanges(page.records, [change])
update: (record: ActivityRecord) => ActivityRecord enqueue(change)
): void => { if (!sameRecords(records, page.records)) setPage({ ...page, records })
const changes: ActivityOutboxEntry[] = [] }
const next = records.map((record) => {
const updated = update(record) /** What Main does not have yet: the changes after the last queued clear (and whether there is one). */
if (updated !== record) { const queuedChanges = (): { cleared: boolean; entries: ActivityOutboxEntry[] } => {
changes.push({ type: 'upsert', record: updated, position: 'in-place' }) const queued = [...sending, ...outbox]
const cleared = queued.findLastIndex((entry) => entry.type === 'clear')
return { cleared: cleared >= 0, entries: queued.slice(cleared + 1) }
}
const replay = (
records: readonly ActivityRecord[],
entries: readonly ActivityOutboxEntry[],
inserts: boolean
): ActivityRecord[] => {
let result = [...records]
for (const entry of entries) {
if (entry.type === 'reconcile') {
result = reconcileActivityRecords(result, entry.tasks, new Set(entry.request.activeRequestIds))
} else if (isActivityChange(entry) && (inserts || (entry.type !== 'upsert' && entry.type !== 'upsert-call'))) {
result = applyActivityChanges(result, [entry])
} }
return updated }
}) return result
if (changes.length > 0) commit(next, changes)
} }
/** A first page from Main with the queued changes applied, as Main will store it. */
const removeWhere = (match: (record: ActivityRecord) => boolean): void => { const firstPageRecords = (loaded: readonly ActivityRecord[], filter: ActivityHistoryFilter): ActivityRecord[] => {
const removed = records.filter(match) const { cleared, entries } = queuedChanges()
if (removed.length === 0) return return replay(cleared ? [] : loaded, entries, true).filter((record) => matchesActivityFilter(record, filter))
commit(
records.filter((record) => !match(record)),
[...new Set(removed.map((record) => record.id))].map((id) => ({
type: 'remove' as const,
id
}))
)
} }
/** An older page: queued upserts already put their records in front of the loaded ones. */
const olderPageRecords = (loaded: readonly ActivityRecord[]): ActivityRecord[] => {
const { cleared, entries } = queuedChanges()
if (cleared) return []
const moved = new Set<string>()
for (const entry of entries) {
if (entry.type === 'upsert') moved.add(entry.record.id)
if (entry.type !== 'upsert-call') continue
// Inserted here because the call's record was not loaded: Main moves
// the first record of the call to the front instead.
const key = callKey(entry.record)
const local = page.records.find((record) => record.callId !== undefined && callKey(record) === key)
if (local?.id !== entry.record.id) continue
const match = loaded.find((record) => !moved.has(record.id) && record.callId !== undefined && callKey(record) === key)
if (match) moved.add(match.id)
}
return replay(loaded.filter((record) => !moved.has(record.id)), entries, false)
.filter((record) => matchesActivityFilter(record, page.filter))
}
const clearQueued = (): boolean => queuedChanges().cleared
const pageEnd = (loaded: ActivityHistoryPage): Pick<ActivityPageState, 'hasMore' | 'nextBefore'> =>
loaded.nextBefore === undefined || clearQueued()
? { hasMore: false }
: { hasMore: true, nextBefore: loaded.nextBefore }
const store = { const emptySummary = (): ActivityHistorySummary => ({
counts: { all: 0, active: 0, failed: 0 }, conversations: {}, legacyHistoryMayBeIncomplete: false
})
return {
/** The loaded records, newest first. */
getRecords(): readonly ActivityRecord[] { getRecords(): readonly ActivityRecord[] {
return records return page.records
},
getPage(): ActivityPageState {
return page
},
getGeneration(): number {
return generation
}, },
subscribe: subscribe(listeners), subscribe: subscribe(listeners),
getPanelRecords(): readonly ActivityRecord[] { /** What the Activity page shows (a snapshot). */
return panelRecords getPanel(): ActivityPageState {
return panel
},
/** Counts and conversation titles/status over the whole history (Main). */
getSummary(): ActivityHistorySummary | undefined {
return summary
}, },
subscribePanel: subscribe(panelListeners), subscribePanel: subscribe(panelListeners),
/** Shows the current list on the Activity page. */ /** Shows the current records on the Activity page. */
refreshPanel(): void { refreshPanel(): void {
if (panelRecords === records) return if (panel === page) return
panelRecords = records panel = page
notify(panelListeners)
},
setSummary(next: ActivityHistorySummary): void {
summary = next
notify(panelListeners) notify(panelListeners)
}, },
/** Whether the stored history was loaded; changes are sent only after. */ /** Whether the stored history was loaded; changes are sent only after. */
isReady(): boolean { isReady(): boolean {
return ready return ready
}, },
/** Whether the migrated legacy history may have lost records. */ markReady(): void {
isLegacyIncomplete(): boolean { ready = true
return legacyIncomplete if (outbox.length > 0) notify(outboxListeners)
}, },
/** Adds a record, or moves a tool/subagent record of the same call to the front. */ /** Adds a record, or moves a tool/subagent record of the same call to the front. */
record(incoming: ActivityRecord): void { record(incoming: ActivityRecord): void {
const next = upsertActivityRecord(records, incoming) apply((incoming.kind === 'tool' || incoming.kind === 'subagent') && incoming.callId
if (next === records) return ? { type: 'upsert-call', record: incoming }
commit(next, [{ type: 'upsert', record: next[0]!, position: 'front' }]) : { type: 'upsert', record: incoming, position: 'front' })
}, },
/** Updates the request record(s) of a run where they are. */ /** Updates the request record(s) of a run where they are. */
updateRequest( updateRequest(requestId: string, status: ActivityRecord['status'], detail?: string): void {
requestId: string, apply({ type: 'update-request', requestId, status, ...(detail === undefined ? {} : { detail }) })
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. */ /** Resolves the newest pending approval of a conversation. */
updateApproval( updateApproval(conversationId: string, status: ActivityRecord['status'], detailLine: string): void {
conversationId: string, apply({ type: 'resolve-approval', conversationId, status, detailLine })
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. */ /** Drops every record of a request that never started. */
removeByRequest(requestId: string): void { removeByRequest(requestId: string): void {
removeWhere((record) => record.requestId === requestId) apply({ type: 'remove-request', requestId })
}, },
/** Drops the provisional tool record a subagent replaces. */ /** Drops the provisional tool record a subagent replaces. */
removeToolByCallId(requestId: string, callId: string): void { removeToolByCallId(requestId: string, callId: string): void {
removeWhere( apply({ type: 'remove-call', requestId, callId })
(record) =>
record.requestId === requestId &&
record.kind === 'tool' &&
record.callId === callId
)
}, },
/** Ends the unfinished non-request records of a finished run. */ /** Ends the unfinished non-request records of a finished run. */
settleRequest( settleRequest(requestId: string, status: ActivityRecord['status'], detailLine: string): void {
requestId: string, apply({ type: 'settle-request', requestId, status, detailLine })
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 * Ends records left running by requests that are no longer active: the
* duplicates of an ID are dropped, as the list always showed only the * loaded ones here, all of them in Main (joined with the stored tasks).
* first one.
*/ */
mergeLoaded( reconcile(tasks: readonly AssistantTask[], activeRequestIds: ReadonlySet<string>): void {
loaded: readonly ActivityRecord[], const interruptedDetail = i18n.t('records.interruptedOnRestart', { ns: 'activity' })
loadedLegacyIncomplete: boolean const records = reconcileActivityRecords(page.records, tasks, activeRequestIds)
): void { enqueue({ type: 'reconcile', request: { activeRequestIds: [...activeRequestIds], interruptedDetail }, tasks })
const next = mergeActivityRecords(records, loaded) if (!sameRecords(records, page.records)) setPage({ ...page, records })
const hasDuplicates = },
new Set(loaded.map((record) => record.id)).size < loaded.length
ready = true /**
legacyIncomplete = legacyIncomplete || loadedLegacyIncomplete * Replaces the loaded records with a first page from Main; this
commit(next, hasDuplicates ? [{ type: 'remove-duplicates' }] : []) * session's unsaved records stay in front. `generation` is the value
if (outbox.length > 0) notify(outboxListeners) * when the request started: a response superseded meanwhile is dropped.
*/
setFirstPage(filter: ActivityHistoryFilter, loaded: ActivityHistoryPage, requestGeneration: number): boolean {
if (requestGeneration !== generation) return false
generation++
setPage({
filter,
records: firstPageRecords(loaded.records, filter),
...pageEnd(loaded)
})
return true
},
/** Appends the page that starts at the current `nextBefore`. */
appendPage(loaded: ActivityHistoryPage, requestGeneration: number, before: number): boolean {
if (requestGeneration !== generation || page.nextBefore !== before) return false
generation++
setPage({
filter: page.filter,
records: mergeActivityRecords(page.records, olderPageRecords(loaded.records)),
...pageEnd(loaded)
})
return true
},
/**
* Shows the loaded records of another filter at once; the sync then
* loads its first page. They are a prefix of that filter's list.
*/
showFilter(filter: ActivityHistoryFilter): void {
if (filter === page.filter) return
generation++
setPage({ filter, records: page.records.filter((record) => matchesActivityFilter(record, filter)), hasMore: true })
panel = page
notify(panelListeners)
},
/** Keeps at most the first page (leaving the Activity page); the next entry reloads. */
trim(size = ACTIVITY_PAGE_SIZE): void {
if (page.records.length <= size) return
generation++
setPage({ filter: page.filter, records: page.records.slice(0, size), hasMore: true })
// The hidden page lets go of the rest too; entering shows the same first rows.
panel = page
notify(panelListeners)
}, },
/** /**
* Clears the history here and, through the sync, in Main. Queued changes * Clears the history here and, through the sync, in Main. Queued changes
@@ -455,27 +510,40 @@ export function createActivityStore(
* loaded only this session's records go, as before. * loaded only this session's records go, as before.
*/ */
clear(): void { clear(): void {
records = [] generation++
panelRecords = [] page = { filter: page.filter, records: [], hasMore: false }
legacyIncomplete = false panel = page
summary = summary && emptySummary()
outbox = ready ? [{ type: 'clear' }] : [] outbox = ready ? [{ type: 'clear' }] : []
sending = []
notify(listeners) notify(listeners)
notify(panelListeners) notify(panelListeners)
notify(outboxListeners) notify(outboxListeners)
}, },
/** Main already deleted everything (clear local data): nothing to send. */ /** Main already deleted everything (clear local data): nothing to send. */
reset(): void { reset(): void {
records = [] generation++
panelRecords = [] page = { filter: 'all', records: [], hasMore: false }
legacyIncomplete = false panel = page
summary = summary && emptySummary()
outbox = [] outbox = []
notify(listeners) notify(listeners)
notify(panelListeners) notify(panelListeners)
}, },
/** The next save writes the whole list (Main's copy is not a known base). */ /**
requireReplace(): void { * Queues the migration of the legacy localStorage history in front of
outbox = [{ type: 'replace' }] * this session's changes: the stored list becomes
notify(outboxListeners) * mergeActivityRecords(legacy, stored), as the former full replace did.
*/
queueLegacyMigration(legacy: readonly ActivityRecord[], mayBeIncomplete: boolean): void {
const changes: ActivityOutboxEntry[] = [...legacy].reverse().map((record) =>
({ type: 'upsert', record, position: 'front' }))
changes.push({ type: 'remove-duplicates' })
// The stored flag stays set when it was; the legacy history can only add to it.
if (mayBeIncomplete) changes.push({ type: 'legacy-flag', value: true })
const wasEmpty = outbox.length === 0
outbox = [...changes, ...outbox]
if (wasEmpty) notify(outboxListeners)
}, },
getOutbox(): readonly ActivityOutboxEntry[] { getOutbox(): readonly ActivityOutboxEntry[] {
@@ -484,13 +552,49 @@ export function createActivityStore(
subscribeOutbox: subscribe(outboxListeners), subscribeOutbox: subscribe(outboxListeners),
/** Removes and returns the first `count` outbox entries. */ /** Removes and returns the first `count` outbox entries. */
takeOutbox(count: number): ActivityOutboxEntry[] { takeOutbox(count: number): ActivityOutboxEntry[] {
return outbox.splice(0, count) sending = outbox.splice(0, count)
return sending
}, },
/** Puts entries that failed to save back in front, in order. */ /**
* The taken entries are saved. calls are Main's stored records of the
* sent upsert-call changes: a call whose earlier record was not loaded
* keeps that record's id and createdAt, so the loaded copy takes them.
*/
acknowledge(calls: readonly ActivityRecord[] = []): void {
const sent = sending.flatMap((entry) => (entry.type === 'upsert-call' ? [entry.record.id] : []))
sending = []
// The renderer's id of each sent call record -> the id Main keeps.
const renamed = new Map<string, ActivityRecord>()
sent.forEach((id, index) => {
const stored = calls[index]
if (stored && stored.id !== id) renamed.set(id, stored)
})
if (renamed.size === 0) return
const records = page.records.map((record) => {
const stored = renamed.get(record.id)
return stored ? { ...record, id: stored.id, createdAt: stored.createdAt, scope: stored.scope } : record
})
// Later queued changes refer to the renderer's id only through content, never by id.
setPage({ ...page, records: mergeActivityRecords(records, []) })
}, /** Puts entries that failed to save back in front, in order. */
requeue(entries: readonly ActivityOutboxEntry[]): void { requeue(entries: readonly ActivityOutboxEntry[]): void {
if (outbox[0]?.type === 'clear' || outbox[0]?.type === 'replace') return sending = []
if (outbox[0]?.type === 'clear') return
outbox = [...entries, ...outbox] outbox = [...entries, ...outbox]
} }
} }
return store
} }
const callKey = (record: ActivityRecord): string =>
JSON.stringify([record.kind, record.requestId, record.callId])
const recordKeys: readonly (keyof ActivityRecord)[] = [
'id', 'conversationId', 'requestId', 'callId', 'kind', 'title', 'detail', 'status', 'createdAt'
]
function sameRecords(left: readonly ActivityRecord[], right: readonly ActivityRecord[]): boolean {
return left.length === right.length && left.every((record, index) => {
const other = right[index]!
return record === other || (recordKeys.every((key) => record[key] === other[key]) &&
JSON.stringify(record.scope) === JSON.stringify(other.scope))
})
}
+259 -135
View File
@@ -4,14 +4,28 @@ import { tmpdir } from 'node:os'
import { join } from 'node:path' import { join } from 'node:path'
import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest' import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest'
import type { import type {
ActivityHistoryPage,
ActivityHistorySnapshot, ActivityHistorySnapshot,
ActivityHistorySummary,
ActivityHistoryUpdate, ActivityHistoryUpdate,
ActivityRecord, ActivityRecord,
AssistantTask AssistantTask
} from '../../shared/assistant-contracts' } from '../../shared/assistant-contracts'
import { createActivityStore, type ActivityStore } from './activity-store' import { firstOccurrences } from '../../shared/activity-history-reference'
import i18n from './i18n'
import {
ACTIVITY_FIRST_PAGE_SIZE,
createActivityStore,
mergeActivityRecords,
reconcileActivityRecords,
upsertActivityRecord,
type ActivityStore
} from './activity-store'
import { import {
loadActivityHistory, loadActivityHistory,
loadFirstActivityPage,
loadNextActivityPage,
refreshActivitySummary,
startActivitySync, startActivitySync,
type ActivityHistoryApi type ActivityHistoryApi
} from './activity-sync' } from './activity-sync'
@@ -22,8 +36,13 @@ type AssistantDatabase = {
close: () => void close: () => void
getActivityHistory: () => ActivityHistorySnapshot getActivityHistory: () => ActivityHistorySnapshot
replaceActivityHistory: (input: unknown) => void replaceActivityHistory: (input: unknown) => void
updateActivityHistory: (input: unknown) => void updateActivityHistory: (input: unknown) => { calls: ActivityRecord[] }
clearActivityHistory: () => void clearActivityHistory: () => void
getActivityHistoryPage: (input: unknown) => ActivityHistoryPage
getActivityHistorySummary: (input: unknown) => ActivityHistorySummary
reconcileActivityHistory: (input: unknown) => number
createTask: (input: { id: string; title: string; instructions: string; workMode: 'ask' }) => AssistantTask
updateTaskStatus: (taskId: string, status: AssistantTask['status']) => void
} }
let AssistantDatabase: new (path: string) => AssistantDatabase let AssistantDatabase: new (path: string) => AssistantDatabase
beforeAll(async () => { beforeAll(async () => {
@@ -52,40 +71,44 @@ function openDatabase(): AssistantDatabase {
return database return database
} }
/** IPC double over a real database; `fail()` decides whether a call throws first. */ type TestApi = ActivityHistoryApi & { updates: ActivityHistoryUpdate[]; calls: string[] }
function databaseApi(
database: AssistantDatabase, /** IPC double over a real database (structured clones, like IPC); `fail()` makes a save throw. */
fail: () => boolean = () => false function databaseApi(database: AssistantDatabase, fail: () => boolean = () => false): TestApi {
): ActivityHistoryApi & { updates: ActivityHistoryUpdate[] } {
const updates: ActivityHistoryUpdate[] = [] const updates: ActivityHistoryUpdate[] = []
const guard = (): void => { const calls: string[] = []
const guard = (name: string): void => {
calls.push(name)
if (fail()) throw new Error('injected save failure') if (fail()) throw new Error('injected save failure')
} }
return { return {
updates, updates,
get: async () => database.getActivityHistory(), calls,
replace: async (records, legacyHistoryMayBeIncomplete) => {
guard()
database.replaceActivityHistory(
structuredClone({ records, legacyHistoryMayBeIncomplete })
)
},
update: async (update) => { update: async (update) => {
guard() guard('update')
updates.push(update) updates.push(update)
database.updateActivityHistory(structuredClone(update)) return structuredClone(database.updateActivityHistory(structuredClone(update)))
}, },
clear: async () => { clear: async () => {
guard() guard('clear')
database.clearActivityHistory() database.clearActivityHistory()
},
reconcile: async (request) => {
guard('reconcile')
return database.reconcileActivityHistory(structuredClone(request))
},
page: async (request) => {
calls.push('page')
return database.getActivityHistoryPage(structuredClone(request))
},
summary: async (request) => {
calls.push('summary')
return database.getActivityHistorySummary(structuredClone(request))
} }
} }
} }
const manualTimers = { const manualTimers = { setTimeout: () => 1, clearTimeout: () => undefined }
setTimeout: () => 1,
clearTimeout: () => undefined
}
function makeRandom(seed: number): () => number { function makeRandom(seed: number): () => number {
let state = seed let state = seed
@@ -109,140 +132,240 @@ function record(overrides: Partial<ActivityRecord> = {}): ActivityRecord {
} }
} }
function task(id: string, status: AssistantTask['status']): AssistantTask { async function startup(
store: ActivityStore,
api: TestApi,
sync: { flush: () => Promise<void> },
activeRequestIds: ReadonlySet<string> = new Set(),
legacy: ActivityRecord[] = [],
legacyIncomplete = false
): Promise<void> {
await loadActivityHistory(store, api, { records: legacy, historyMayBeIncomplete: legacyIncomplete }, {
flush: sync.flush,
activeRequestIds: () => activeRequestIds,
onReadFailed: () => undefined,
isActive: () => true
})
}
/**
* The former App behavior over the whole list (activity state = every
* record), used as the oracle the paged store plus Main must match.
*/
function oracle(initial: readonly ActivityRecord[]) {
let list = firstOccurrences(initial)
return { return {
id, get: () => list,
title: id, record: (incoming: ActivityRecord) => { list = upsertActivityRecord(list, incoming) },
instructions: id, updateRequest: (requestId: string, status: ActivityRecord['status'], detail?: string) => {
workMode: 'ask', list = list.map((item) => item.requestId === requestId && item.kind === 'request'
status, ? { ...item, status, detail: detail ?? item.detail } : item)
createdAt: new Date(0).toISOString() },
} as AssistantTask updateApproval: (conversationId: string, status: ActivityRecord['status'], line: string) => {
let updated = false
list = list.map((item) => {
if (updated || item.conversationId !== conversationId || item.kind !== 'approval' || item.status !== 'pending') return item
updated = true
return { ...item, status, detail: `${item.detail}\n${line}` }
})
},
removeByRequest: (requestId: string) => { list = list.filter((item) => item.requestId !== requestId) },
removeToolByCallId: (requestId: string, callId: string) => {
list = list.filter((item) => !(item.requestId === requestId && item.kind === 'tool' && item.callId === callId))
},
settleRequest: (requestId: string, status: ActivityRecord['status'], line: string) => {
list = list.map((item) => item.requestId === requestId && item.kind !== 'request' &&
(item.status === 'pending' || item.status === 'running')
? { ...item, status, detail: `${item.detail}\n${line}` } : item)
},
reconcile: (tasks: readonly AssistantTask[], active: ReadonlySet<string>) => {
list = reconcileActivityRecords(list, tasks, active)
},
clear: () => { list = [] },
prepend: (records: readonly ActivityRecord[]) => { list = mergeActivityRecords(records, list) }
}
} }
async function loadedStore(api: ActivityHistoryApi): Promise<ActivityStore> { describe('activity history sync (paged)', () => {
const store = createActivityStore() it('ends in the state the former full-list code saved, for random actions, pages, trims, flushes and failures', async () => {
await loadActivityHistory( for (const seed of [3, 17, 2026, 41, 7777, 90210]) {
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 random = makeRandom(seed)
const pick = <T,>(items: readonly T[]): T => const pick = <T,>(items: readonly T[]): T => items[Math.floor(random() * items.length)]!
items[Math.floor(random() * items.length)]! const database = openDatabase()
const incremental = openDatabase() // Stored history with legacy duplicate IDs; the first occurrence is the shown one.
// Legacy history with duplicate IDs, stored by the former full replace. const stored = Array.from({ length: 260 }, (_, index) => record({
const first = record({ id: 'dup' }) kind: pick(['tool', 'subagent', 'request', 'result', 'approval'] as const),
const legacy = [ status: pick(['pending', 'running', 'completed', 'failed'] as const),
first, requestId: `request-${index % 7}`,
record({ status: 'completed' }), conversationId: `conversation-${index % 4}`,
{ ...first, detail: 'older duplicate' }, ...(index % 3 === 0 ? { callId: pick(['a', 'b', 'c']) } : {})
record({ kind: 'request', status: 'running', requestId: 'request-1' }), }))
record({ kind: 'approval', status: 'pending', conversationId: 'conversation-1' }) stored.splice(40, 0, { ...stored[3]!, detail: 'hidden duplicate' })
] stored.splice(250, 0, { ...stored[200]!, detail: 'hidden duplicate' })
incremental.replaceActivityHistory({ database.replaceActivityHistory({ records: stored, legacyHistoryMayBeIncomplete: false })
records: legacy, const tasks: AssistantTask[] = []
legacyHistoryMayBeIncomplete: seed % 2 === 0 for (const [index, status] of (['completed', 'failed', 'running', 'cancelled'] as const).entries()) {
}) const created = database.createTask({ id: `request-${index}`, title: 't', instructions: 'i', workMode: 'ask' })
database.updateTaskStatus(created.id, status)
tasks.push({ ...created, status })
}
const expected = oracle(stored)
let failing = false let failing = false
let expectedFlag = seed % 2 === 0 const api = databaseApi(database, () => failing && random() < 0.5)
const api = databaseApi(incremental, () => failing && random() < 0.5) const store = createActivityStore()
const store = await loadedStore(api) const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers })
const sync = startActivitySync(store, api, { const active = new Set(['request-5'])
onSaveFailed: () => undefined, await startup(store, api, sync, active)
timers: manualTimers expected.reconcile(tasks, active)
}) expect(store.getRecords().length).toBeLessThanOrEqual(ACTIVITY_FIRST_PAGE_SIZE)
const callIds = ['a', 'b', 'c', 'd'] const callIds = ['a', 'b', 'c', 'd']
for (let step = 0; step < 400; step++) { for (let step = 0; step < 500; step++) {
const roll = random() const roll = random()
const requestId = `request-${Math.floor(random() * 5)}` const requestId = `request-${Math.floor(random() * 8)}`
if (roll < 0.3) { const conversationId = `conversation-${Math.floor(random() * 4)}`
if (roll < 0.25) {
const kind = pick(['tool', 'subagent', 'request', 'result', 'approval'] as const) const kind = pick(['tool', 'subagent', 'request', 'result', 'approval'] as const)
store.record(record({ const incoming = record({
kind, kind, requestId, conversationId,
requestId,
...(kind === 'tool' || kind === 'subagent' ? { callId: pick(callIds) } : {}), ...(kind === 'tool' || kind === 'subagent' ? { callId: pick(callIds) } : {}),
status: pick(['pending', 'running', 'completed', 'failed'] as const) status: pick(['pending', 'running', 'completed', 'failed'] as const)
})) })
} else if (roll < 0.4) { store.record(incoming)
store.updateRequest(requestId, pick(['completed', 'failed'] as const), random() < 0.5 ? 'done' : undefined) expected.record(incoming)
} else if (roll < 0.47) { } else if (roll < 0.35) {
store.updateApproval(`conversation-${Math.floor(random() * 3)}`, 'denied', 'decision') const status = pick(['completed', 'failed'] as const)
} else if (roll < 0.52) { const detail = random() < 0.5 ? 'done' : undefined
store.updateRequest(requestId, status, detail)
expected.updateRequest(requestId, status, detail)
} else if (roll < 0.42) {
store.updateApproval(conversationId, 'denied', 'decision')
expected.updateApproval(conversationId, 'denied', 'decision')
} else if (roll < 0.46) {
store.removeByRequest(requestId) store.removeByRequest(requestId)
} else if (roll < 0.58) { expected.removeByRequest(requestId)
store.removeToolByCallId(requestId, pick(callIds)) } else if (roll < 0.51) {
} else if (roll < 0.65) { const callId = pick(callIds)
store.removeToolByCallId(requestId, callId)
expected.removeToolByCallId(requestId, callId)
} else if (roll < 0.57) {
store.settleRequest(requestId, 'interrupted', 'incomplete') store.settleRequest(requestId, 'interrupted', 'incomplete')
} else if (roll < 0.72) { expected.settleRequest(requestId, 'interrupted', 'incomplete')
store.reconcile( } else if (roll < 0.62) {
[task('request-0', 'completed'), task('request-2', 'failed'), task('request-3', 'running')], const running = new Set(random() < 0.5 ? ['request-1'] : [])
new Set(random() < 0.5 ? ['request-1'] : []) store.reconcile(tasks, running)
) expected.reconcile(tasks, running)
} else if (roll < 0.74) { } else if (roll < 0.635) {
store.clear() store.clear()
expectedFlag = false expected.clear()
} else if (roll < 0.8) { } else if (roll < 0.68) {
failing = random() < 0.4 failing = random() < 0.4
} else if (roll < 0.73) {
await loadNextActivityPage(store, api)
} else if (roll < 0.76) {
store.trim(Math.floor(random() * 60))
} else if (roll < 0.79) {
await loadFirstActivityPage(store, api, 1 + Math.floor(random() * 80))
} else { } else {
await sync.flush() await sync.flush()
} }
// Loaded records are always the newest records of the full list.
if (store.getOutbox().length === 0) {
const loaded = store.getRecords()
expect(loaded).toEqual(expected.get().slice(0, loaded.length))
}
} }
failing = false failing = false
await sync.flush() await sync.flush()
expect(store.getOutbox()).toEqual([]) expect(store.getOutbox()).toEqual([])
const shown = firstOccurrences(database.getActivityHistory().records)
expect(shown).toEqual(expected.get())
const replaced = openDatabase() const replaced = openDatabase()
replaced.replaceActivityHistory({ replaced.replaceActivityHistory({ records: expected.get(), legacyHistoryMayBeIncomplete: false })
records: [...store.getRecords()], expect(shown).toEqual(replaced.getActivityHistory().records)
legacyHistoryMayBeIncomplete: expectedFlag
})
expect(incremental.getActivityHistory()).toEqual(replaced.getActivityHistory())
expect(incremental.getActivityHistory().records).toEqual(store.getRecords())
sync.stop() sync.stop()
} }
}) }, 60_000)
it('sends only the changed records', async () => { it('keeps at most one page after startup and sends only changed records, including unloaded ones', async () => {
const database = openDatabase() const database = openDatabase()
const records = Array.from({ length: 5_000 }, () => record({ status: 'completed' })) const records = Array.from({ length: 5_000 }, (_, index) => record({ status: 'completed', kind: index % 5 === 0 ? 'request' : 'tool', requestId: `request-${index}` }))
database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: false }) database.replaceActivityHistory({ records, legacyHistoryMayBeIncomplete: false })
const api = databaseApi(database) const api = databaseApi(database)
const store = await loadedStore(api) const store = createActivityStore()
const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers }) const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers })
await startup(store, api, sync)
expect(store.getRecords()).toHaveLength(ACTIVITY_FIRST_PAGE_SIZE)
expect(api.calls).toEqual(['reconcile', 'page'])
const tool = record({ callId: 'call-1', requestId: 'request-x' }) const tool = record({ callId: 'call-1', requestId: 'request-x' })
store.record(tool) store.record(tool)
store.record({ ...tool, id: 'ignored', detail: 'progress', status: 'completed' }) store.record({ ...tool, id: 'ignored', detail: 'progress', status: 'completed' })
store.updateRequest(records[4_000]!.requestId, 'failed') // records[4_000] is far outside the loaded page.
store.updateRequest(records[4_000]!.requestId, 'failed', 'late failure')
await sync.flush() await sync.flush()
// Two consecutive upserts of the tool collapse into one.
expect(api.updates).toHaveLength(1) expect(api.updates).toHaveLength(1)
const changes = api.updates[0]!.changes expect(api.updates[0]!.changes).toHaveLength(3)
const requestRecords = records.filter( expect(JSON.stringify(api.updates[0]).length).toBeLessThan(1_500)
(item) => item.requestId === records[4_000]!.requestId && item.kind === 'request' const after = database.getActivityHistory().records
) expect(after[0]).toMatchObject({ id: tool.id, detail: 'progress', status: 'completed' })
expect(changes).toHaveLength(1 + requestRecords.length) expect(after.find((item) => item.id === records[4_000]!.id)).toMatchObject({ status: 'failed', detail: 'late failure' })
expect(JSON.stringify(api.updates[0]).length).toBeLessThan(1_000) expect(store.getRecords().length).toBeLessThanOrEqual(ACTIVITY_FIRST_PAGE_SIZE + 1)
expect(database.getActivityHistory().records[0]).toMatchObject({ sync.stop()
id: tool.id, })
detail: 'progress'
}) it('shows a tool call once when its earlier record arrives on a later page while the update is queued', async () => {
const database = openDatabase()
const old = record({ kind: 'tool', callId: 'call-z', requestId: 'request-z', status: 'running' })
const newer = Array.from({ length: 4 }, () => record({ status: 'completed' }))
database.replaceActivityHistory({ records: [...newer, old], legacyHistoryMayBeIncomplete: false })
const api = databaseApi(database)
const store = createActivityStore()
const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers })
await startup(store, api, sync)
await loadFirstActivityPage(store, api, 2)
const incoming = record({ kind: 'tool', callId: 'call-z', requestId: 'request-z', status: 'completed', detail: 'done' })
store.record(incoming)
await loadNextActivityPage(store, api)
const expected = upsertActivityRecord([...newer, old], incoming)
// Until saved, the call carries this session's id; content and order already match.
const withoutIds = (list: readonly ActivityRecord[]): unknown[] =>
list.map((item) => ({ ...item, id: undefined, createdAt: undefined }))
expect(store.getRecords()).toHaveLength(5)
expect(withoutIds(store.getRecords())).toEqual(withoutIds(expected))
await sync.flush()
expect(store.getRecords()).toEqual(expected.slice(0, store.getRecords().length))
expect(database.getActivityHistory().records).toEqual(expected)
sync.stop()
})
it('marks records left running by the previous session interrupted, with the same localized line', async () => {
const database = openDatabase()
const leftover = record({ status: 'running', requestId: 'gone', detail: 'reading' })
const done = record({ status: 'running', requestId: 'finished', kind: 'request', detail: 'started' })
const live = record({ status: 'running', requestId: 'live' })
database.replaceActivityHistory({ records: [leftover, done, live], legacyHistoryMayBeIncomplete: false })
const task = database.createTask({ id: 'finished', title: 't', instructions: 'i', workMode: 'ask' })
database.updateTaskStatus(task.id, 'completed')
const api = databaseApi(database)
const store = createActivityStore()
const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers })
await startup(store, api, sync, new Set(['live']))
const line = i18n.t('records.interruptedOnRestart', { ns: 'activity' })
expect(line.length).toBeGreaterThan(0)
const expected = reconcileActivityRecords([leftover, done, live], [{ ...task, status: 'completed' }], new Set(['live']))
expect(expected[0]!.detail).toBe(`reading\n${line}`)
expect(database.getActivityHistory().records).toEqual(expected)
expect(store.getRecords()).toEqual(expected)
sync.stop() sync.stop()
}) })
it('keeps failed changes queued and retries them in order with backoff', async () => { it('keeps failed changes queued and retries them in order with backoff', async () => {
const database = openDatabase() const database = openDatabase()
let fail = true let fail = false
const api = databaseApi(database, () => fail) const api = databaseApi(database, () => fail)
const store = await loadedStore(api) const store = createActivityStore()
const delays: number[] = [] const delays: number[] = []
const callbacks: Array<() => void> = [] const callbacks: Array<() => void> = []
const onSaveFailed = vi.fn() const onSaveFailed = vi.fn()
@@ -257,6 +380,8 @@ describe('activity history sync', () => {
clearTimeout: () => undefined clearTimeout: () => undefined
} }
}) })
await startup(store, api, sync)
fail = true
const a = record() const a = record()
const b = record() const b = record()
store.record(a) store.record(a)
@@ -276,45 +401,44 @@ describe('activity history sync', () => {
sync.stop() sync.stop()
}) })
it('clears the database in one call and drops queued changes', async () => { it('clear empties the database, the loaded pages and the summary', async () => {
const database = openDatabase() const database = openDatabase()
database.replaceActivityHistory({ records: [record(), record()], legacyHistoryMayBeIncomplete: true }) database.replaceActivityHistory({ records: [record(), record()], legacyHistoryMayBeIncomplete: true })
const api = databaseApi(database) const api = databaseApi(database)
const update = vi.spyOn(api, 'update') const store = createActivityStore()
const store = await loadedStore(api)
const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers }) const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers })
await startup(store, api, sync)
store.refreshPanel()
await refreshActivitySummary(store, api)
expect(store.getSummary()?.counts.all).toBe(2)
store.record(record()) store.record(record())
store.clear() store.clear()
await sync.flush() await sync.flush()
expect(update).not.toHaveBeenCalled() expect(api.updates).toEqual([])
expect(database.getActivityHistory()).toEqual({ records: [], legacyHistoryMayBeIncomplete: false }) expect(database.getActivityHistory()).toEqual({ records: [], legacyHistoryMayBeIncomplete: false })
expect(store.getRecords()).toEqual([]) expect(store.getRecords()).toEqual([])
expect(store.getPanel().records).toEqual([])
expect(store.getSummary()?.counts).toEqual({ all: 0, active: 0, failed: 0 })
sync.stop() sync.stop()
}) })
it('migrates legacy records with one replace and retries it in full after a failure', async () => { it('migrates legacy localStorage records in front of the stored ones, retrying after a failure', async () => {
const database = openDatabase() const database = openDatabase()
const stored = record({ status: 'completed' }) const stored = record({ status: 'completed' })
database.replaceActivityHistory({ records: [stored], legacyHistoryMayBeIncomplete: false }) const both = record({ status: 'completed' })
database.replaceActivityHistory({ records: [stored, both], legacyHistoryMayBeIncomplete: false })
let fail = true let fail = true
const api = databaseApi(database, () => fail) const api = databaseApi(database, () => fail)
const legacyRecord = record({ status: 'completed' }) const legacyRecord = record({ status: 'completed' })
const store = createActivityStore([legacyRecord], true) const store = createActivityStore([legacyRecord, both])
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 }) const sync = startActivitySync(store, api, { onSaveFailed: () => undefined, timers: manualTimers })
await startup(store, api, sync, new Set(), [legacyRecord, both], true)
fail = false
await sync.flush() await sync.flush()
expect(database.getActivityHistory()).toEqual({ await loadFirstActivityPage(store, api)
records: [legacyRecord, stored], const expected = mergeActivityRecords([legacyRecord, both], [stored, both])
legacyHistoryMayBeIncomplete: true expect(database.getActivityHistory()).toEqual({ records: expected, legacyHistoryMayBeIncomplete: true })
}) expect(store.getRecords()).toEqual(expected)
sync.stop() sync.stop()
}) })
}) })
+186 -77
View File
@@ -1,34 +1,47 @@
import { useCallback, useEffect, useRef } from 'react' import { useCallback, useEffect, useRef } from 'react'
import type { import type {
ActivityHistoryChange, ActivityHistoryChange,
ActivityHistorySnapshot, ActivityHistoryFilter,
ActivityHistoryPage,
ActivityHistoryPageRequest,
ActivityHistoryReconcileRequest,
ActivityHistorySummary,
ActivityHistorySummaryRequest,
ActivityHistoryUpdate, ActivityHistoryUpdate,
ActivityRecord ActivityHistoryUpdateResult
} from '../../shared/assistant-contracts' } from '../../shared/assistant-contracts'
import i18n from './i18n'
import { import {
ACTIVITY_FIRST_PAGE_SIZE,
ACTIVITY_PAGE_SIZE,
clearLegacyActivityHistory, clearLegacyActivityHistory,
mergeActivityRecords, isActivityChange,
type ActivityOutboxEntry, type ActivityOutboxEntry,
type ActivityStore, type ActivityStore,
type LegacyActivityHistory type LegacyActivityHistory
} from './activity-store' } from './activity-store'
/** /**
* Activity history IPC with Main (P4): the first load with the legacy * Activity history IPC with Main (P4): startup (legacy migration, reconcile,
* localStorage migration, and the incremental save queue. Changes collect in * first page and summary), pages on demand, and the incremental save queue.
* the store's outbox and leave 250 ms after the first one, at most 20,000 per * Changes collect in the store's outbox and leave 250 ms after the first
* `update` call. A failed save keeps its changes queued, in order, and is * one, at most 20,000 per `update` call. A failed save keeps its changes
* retried with backoff; quitting and unmounting flush what is left. * queued, in order, and is retried with backoff; quitting and unmounting
* flush what is left.
*/ */
export type ActivityHistoryApi = { export type ActivityHistoryApi = {
get: () => Promise<ActivityHistorySnapshot> update: (update: ActivityHistoryUpdate) => Promise<ActivityHistoryUpdateResult | void>
replace: (records: ActivityRecord[], legacyHistoryMayBeIncomplete: boolean) => Promise<void>
update: (update: ActivityHistoryUpdate) => Promise<void>
clear: () => Promise<void> clear: () => Promise<void>
page: (request: ActivityHistoryPageRequest) => Promise<ActivityHistoryPage>
summary: (request: ActivityHistorySummaryRequest) => Promise<ActivityHistorySummary>
reconcile: (request: ActivityHistoryReconcileRequest) => Promise<number>
} }
export const ACTIVITY_SAVE_DELAY_MS = 250 export const ACTIVITY_SAVE_DELAY_MS = 250
export const ACTIVITY_MAX_CHANGES_PER_UPDATE = 20_000 export const ACTIVITY_MAX_CHANGES_PER_UPDATE = 20_000
/** The Activity page shows this many records before "load more", as before paging. */
export const ACTIVITY_VISIBLE_BATCH = 500
const MAX_SUMMARY_CONVERSATIONS = 5_000
const RETRY_DELAYS_MS = [1_000, 2_000, 5_000, 10_000, 30_000] const RETRY_DELAYS_MS = [1_000, 2_000, 5_000, 10_000, 30_000]
export type ActivitySync = { export type ActivitySync = {
@@ -51,8 +64,8 @@ const saving = new WeakMap<ActivityStore, Promise<void>>()
export function startActivitySync( export function startActivitySync(
store: ActivityStore, store: ActivityStore,
api: ActivityHistoryApi, api: Pick<ActivityHistoryApi, 'update' | 'clear' | 'reconcile'>,
options: { onSaveFailed: () => void; timers?: Timers; delayMs?: number } options: { onSaveFailed: () => void; onDrained?: () => void; timers?: Timers; delayMs?: number }
): ActivitySync { ): ActivitySync {
const timers = options.timers ?? defaultTimers const timers = options.timers ?? defaultTimers
const delayMs = options.delayMs ?? ACTIVITY_SAVE_DELAY_MS const delayMs = options.delayMs ?? ACTIVITY_SAVE_DELAY_MS
@@ -72,33 +85,33 @@ export function startActivitySync(
const sendNext = async (): Promise<boolean> => { const sendNext = async (): Promise<boolean> => {
const first = store.getOutbox()[0] const first = store.getOutbox()[0]
if (!first) return true if (!first) return true
let batch: ActivityOutboxEntry[] let batch: ActivityOutboxEntry[] = []
try { try {
if (first.type === 'clear') { if (first.type === 'clear') {
batch = store.takeOutbox(1) batch = store.takeOutbox(1)
await api.clear() await api.clear()
} else if (first.type === 'replace') { } else if (first.type === 'reconcile') {
// The full list already contains everything queued after the marker. batch = store.takeOutbox(1)
batch = store.takeOutbox(store.getOutbox().length) await api.reconcile(first.request)
await api.replace([...store.getRecords()], store.isLegacyIncomplete()) } else if (first.type === 'legacy-flag') {
batch = store.takeOutbox(1)
await api.update({ changes: [], legacyHistoryMayBeIncomplete: first.value })
} else { } else {
const outbox = store.getOutbox() const outbox = store.getOutbox()
let count = 0 let count = 0
while ( while (count < outbox.length && count < ACTIVITY_MAX_CHANGES_PER_UPDATE && isActivityChange(outbox[count]!)) {
count < outbox.length &&
count < ACTIVITY_MAX_CHANGES_PER_UPDATE &&
outbox[count]!.type !== 'clear' &&
outbox[count]!.type !== 'replace'
) {
count++ count++
} }
batch = store.takeOutbox(count) batch = store.takeOutbox(count)
await api.update({ changes: batch as ActivityHistoryChange[] }) const result = await api.update({ changes: batch as ActivityHistoryChange[] })
store.acknowledge(result ? result.calls : [])
return true
} }
} catch { } catch {
store.requeue(batch!) store.requeue(batch)
return false return false
} }
store.acknowledge()
return true return true
} }
@@ -112,6 +125,7 @@ export function startActivitySync(
} }
failures = 0 failures = 0
} }
options.onDrained?.()
} }
const flush = async (): Promise<void> => { const flush = async (): Promise<void> => {
@@ -128,9 +142,7 @@ export function startActivitySync(
} }
const unsubscribe = store.subscribeOutbox(() => { const unsubscribe = store.subscribeOutbox(() => {
if (store.isReady() && store.getOutbox().length > 0 && failures === 0) { if (store.isReady() && store.getOutbox().length > 0 && failures === 0) schedule(delayMs)
schedule(delayMs)
}
}) })
if (store.isReady() && store.getOutbox().length > 0) schedule(delayMs) if (store.isReady() && store.getOutbox().length > 0) schedule(delayMs)
@@ -145,33 +157,132 @@ export function startActivitySync(
} }
} }
/** Loads the first page of the store's filter (replacing the loaded pages). */
export async function loadFirstActivityPage(
store: ActivityStore,
api: Pick<ActivityHistoryApi, 'page'>,
limit: number = ACTIVITY_FIRST_PAGE_SIZE,
filter: ActivityHistoryFilter = store.getPage().filter
): Promise<void> {
const generation = store.getGeneration()
const page = await api.page({ limit, ...(filter === 'all' ? {} : { filter }) })
store.setFirstPage(filter, page, generation)
}
/** Loads the page after the loaded ones, when there is one. */
export async function loadNextActivityPage(
store: ActivityStore,
api: Pick<ActivityHistoryApi, 'page'>
): Promise<void> {
const { filter, nextBefore, hasMore } = store.getPage()
if (!hasMore) return
if (nextBefore === undefined) {
// Trimmed: reload the head, then continue from its end as if never trimmed.
await loadFirstActivityPage(store, api, ACTIVITY_PAGE_SIZE)
if (store.getPage().nextBefore !== undefined) await loadNextActivityPage(store, api)
return
}
const generation = store.getGeneration()
const page = await api.page({ before: nextBefore, limit: ACTIVITY_PAGE_SIZE, ...(filter === 'all' ? {} : { filter }) })
store.appendPage(page, generation, nextBefore)
}
/** Reloads the summary for the conversations the page shows. */
export async function refreshActivitySummary(
store: ActivityStore,
api: Pick<ActivityHistoryApi, 'summary'>
): Promise<void> {
const conversationIds = [...new Set(store.getPanel().records.map((record) => record.conversationId))]
.slice(0, MAX_SUMMARY_CONVERSATIONS)
store.setSummary(await api.summary({ conversationIds }))
}
/**
* Startup: queue the legacy migration, let saves start and send what is
* queued, let Main end records left running by the previous session, then
* load the first page.
*/
export async function loadActivityHistory(
store: ActivityStore,
api: Pick<ActivityHistoryApi, 'page' | 'reconcile'>,
legacy: LegacyActivityHistory,
handlers: {
flush: () => Promise<void>
activeRequestIds: () => ReadonlySet<string>
onReadFailed: () => void
isActive: () => boolean
}
): Promise<void> {
if (legacy.records.length > 0 || legacy.historyMayBeIncomplete) {
store.queueLegacyMigration(legacy.records, legacy.historyMayBeIncomplete)
}
store.markReady()
await handlers.flush()
if (!handlers.isActive()) return
try {
await api.reconcile({
activeRequestIds: [...handlers.activeRequestIds()],
interruptedDetail: i18n.t('records.interruptedOnRestart', { ns: 'activity' })
})
} catch {
// The next task refresh reconciles again.
}
if (!handlers.isActive()) return
try {
await loadFirstActivityPage(store, api)
} catch {
if (handlers.isActive()) handlers.onReadFailed()
}
}
/** /**
* Loads the stored history on mount, then saves changes incrementally. The * Loads the stored history on mount, then saves changes incrementally. The
* returned `flush` (stable) sends what is queued, for the quit handshake; the * returned actions are stable: `flush` (quit handshake), `show` (the
* queue is also flushed on unmount. * Activity page opens or its timer/refresh runs: flush, load at least one
* visible batch, refresh the snapshot and summary), `hide`, `loadMore` and
* `setFilter`.
*/ */
export function useActivityHistorySync( export function useActivityHistorySync(
store: ActivityStore, store: ActivityStore,
legacy: LegacyActivityHistory, legacy: LegacyActivityHistory,
notices: { onReadFailed: () => void; onSaveFailed: () => void; onLoaded?: () => void } options: {
): () => Promise<void> { activeRequestIds: () => ReadonlySet<string>
const noticesRef = useRef(notices) onReadFailed: () => void
onSaveFailed: () => void
onLoaded?: () => void
}
): {
flush: () => Promise<void>
show: () => Promise<void>
hide: () => void
loadMore: () => Promise<void>
setFilter: (filter: ActivityHistoryFilter) => Promise<void>
} {
const optionsRef = useRef(options)
useEffect(() => { useEffect(() => {
noticesRef.current = notices optionsRef.current = options
}) })
const syncRef = useRef<ActivitySync | undefined>(undefined) const syncRef = useRef<ActivitySync | undefined>(undefined)
useEffect(() => { useEffect(() => {
let active = true let active = true
let migrating = legacy.records.length > 0 || legacy.historyMayBeIncomplete
const api = window.goodbuddy.activityHistory const api = window.goodbuddy.activityHistory
const onSaveFailed = (): void => noticesRef.current.onSaveFailed() const sync = startActivitySync(store, api, {
const sync = startActivitySync(store, api, { onSaveFailed }) onSaveFailed: () => optionsRef.current.onSaveFailed(),
onDrained: () => {
// The legacy records are in Main once the queue in front of them drained.
if (migrating) clearLegacyActivityHistory()
migrating = false
}
})
syncRef.current = sync syncRef.current = sync
void loadActivityHistory(store, api, legacy, { void loadActivityHistory(store, api, legacy, {
onReadFailed: () => noticesRef.current.onReadFailed(), flush: sync.flush,
onSaveFailed, activeRequestIds: () => optionsRef.current.activeRequestIds(),
onReadFailed: () => optionsRef.current.onReadFailed(),
isActive: () => active isActive: () => active
}).then(() => { }).then(() => {
if (active && store.isReady()) noticesRef.current.onLoaded?.() if (active) optionsRef.current.onLoaded?.()
}) })
return () => { return () => {
active = false active = false
@@ -180,46 +291,44 @@ export function useActivityHistorySync(
void sync.flush() void sync.flush()
} }
}, [legacy, store]) }, [legacy, store])
return useCallback(async () => { const flush = useCallback(async () => {
await syncRef.current?.flush() await syncRef.current?.flush()
}, []) }, [])
} const api = (): ActivityHistoryApi => window.goodbuddy.activityHistory
const show = useCallback(async () => {
/** store.refreshPanel()
* Loads the stored history once. A legacy localStorage history is written await flush()
* 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 { try {
await api.replace( if (store.getRecords().length < ACTIVITY_VISIBLE_BATCH && store.getPage().hasMore) {
mergeActivityRecords(store.getRecords(), snapshot.records), await loadFirstActivityPage(store, api(), ACTIVITY_VISIBLE_BATCH)
legacy.historyMayBeIncomplete || snapshot.legacyHistoryMayBeIncomplete store.refreshPanel()
) }
clearLegacyActivityHistory() await refreshActivitySummary(store, api())
} catch { } catch {
replaceFailed = true optionsRef.current.onReadFailed()
handlers.onSaveFailed()
} }
if (!handlers.isActive()) return }, [flush, store])
} const hide = useCallback(() => store.trim(), [store])
store.mergeLoaded(snapshot.records, snapshot.legacyHistoryMayBeIncomplete) const loadMore = useCallback(async () => {
if (replaceFailed) store.requireReplace() try {
await flush()
await loadNextActivityPage(store, api())
store.refreshPanel()
await refreshActivitySummary(store, api())
} catch {
optionsRef.current.onReadFailed()
}
}, [flush, store])
const setFilter = useCallback(async (filter: ActivityHistoryFilter) => {
store.showFilter(filter)
try {
await flush()
await loadFirstActivityPage(store, api(), ACTIVITY_VISIBLE_BATCH, filter)
store.refreshPanel()
await refreshActivitySummary(store, api())
} catch {
optionsRef.current.onReadFailed()
}
}, [flush, store])
return { flush, show, hide, loadMore, setFilter }
} }
+71 -14
View File
@@ -120,21 +120,78 @@ export function applyActivityChanges(
changes: readonly ActivityHistoryChange[] changes: readonly ActivityHistoryChange[]
): ActivityRecord[] { ): ActivityRecord[] {
let next = [...list] let next = [...list]
/** Indexes of shown records (first occurrences) that match. */
const shown = (match: (record: ActivityRecord) => boolean): number[] => {
const seen = new Set<string>()
const indexes: number[] = []
next.forEach((record, index) => {
if (!seen.has(record.id) && match(record)) indexes.push(index)
seen.add(record.id)
})
return indexes
}
const updateAt = (indexes: readonly number[], update: (record: ActivityRecord) => ActivityRecord): void => {
const set = new Set(indexes)
next = next.map((record, index) => (set.has(index) ? update(record) : record))
}
const removeIds = (ids: ReadonlySet<string>): void => {
next = next.filter((record) => !ids.has(record.id))
}
for (const change of changes) { for (const change of changes) {
if (change.type === 'remove') { switch (change.type) {
next = next.filter((item) => item.id !== change.id) case 'remove':
continue removeIds(new Set([change.id]))
break
case 'remove-duplicates':
next = firstOccurrences(next)
break
case 'upsert': {
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))
break
}
case 'upsert-call': {
const incoming = change.record
const [index] = shown((record) => record.kind === incoming.kind &&
record.requestId === incoming.requestId && record.callId === incoming.callId)
if (index === undefined) {
next = [incoming, ...next]
break
}
const existing = next[index]!
const unchanged = existing.conversationId === incoming.conversationId && existing.title === incoming.title &&
existing.detail === incoming.detail && existing.status === incoming.status
if (unchanged && index === 0) break
next = [
unchanged ? existing : { ...incoming, id: existing.id, createdAt: existing.createdAt, scope: existing.scope },
...next.filter((_, at) => at !== index)
]
break
}
case 'update-request':
updateAt(shown((record) => record.requestId === change.requestId && record.kind === 'request'),
(record) => ({ ...record, status: change.status, detail: change.detail ?? record.detail }))
break
case 'resolve-approval':
updateAt(shown((record) => record.conversationId === change.conversationId &&
record.kind === 'approval' && record.status === 'pending').slice(0, 1),
(record) => ({ ...record, status: change.status, detail: `${record.detail}\n${change.detailLine}` }))
break
case 'settle-request':
updateAt(shown((record) => record.requestId === change.requestId && record.kind !== 'request' &&
isActive(record)),
(record) => ({ ...record, status: change.status, detail: `${record.detail}\n${change.detailLine}` }))
break
case 'remove-request':
removeIds(new Set(shown((record) => record.requestId === change.requestId).map((index) => next[index]!.id)))
break
case 'remove-call':
removeIds(new Set(shown((record) => record.requestId === change.requestId && record.kind === 'tool' &&
record.callId === change.callId).map((index) => next[index]!.id)))
break
} }
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 return next
} }
+56 -4
View File
@@ -110,9 +110,25 @@ export const activityHistorySnapshotSchema = z
* - `upsert` + `position: 'in-place'` replaces the existing record where it * - `upsert` + `position: 'in-place'` replaces the existing record where it
* is; a record that does not exist yet is inserted at the front. * is; a record that does not exist yet is inserted at the front.
* - `remove` deletes every record with the ID. * - `remove` deletes every record with the ID.
* - `remove-duplicates` deletes every record but the first of each ID; the * The other changes match shown records (first occurrences) by content, so
* renderer shows only the first (see `mergeActivityRecords`). * the renderer does not need to have them loaded:
* - `upsert-call`: a tool/subagent record with a callId. The first record
* of the same kind, request and call is moved to the front with the new
* content but its own id, createdAt and scope (nothing happens when it is
* unchanged and already first); without one the record is inserted at the
* front (renderer `upsertActivityRecord`).
* - `update-request`: every request record of the request gets the status
* (and the detail, when given), in place.
* - `resolve-approval`: the first pending approval of the conversation gets
* the status and `detailLine` appended on a new line, in place.
* - `settle-request`: every pending/running non-request record of the
* request gets the status and `detailLine` appended, in place.
* - `remove-request`: deletes every record of the request.
* - `remove-call`: deletes the tool records of the request with the callId.
*/ */
const activityIdSchema = z.string().min(1).max(256)
const activityStatusSchema = activityRecordSchema.shape.status
const activityDetailLineSchema = z.string().max(1_000_000)
export const activityHistoryChangeSchema = z.discriminatedUnion('type', [ export const activityHistoryChangeSchema = z.discriminatedUnion('type', [
z.object({ z.object({
type: z.literal('upsert'), type: z.literal('upsert'),
@@ -121,9 +137,43 @@ export const activityHistoryChangeSchema = z.discriminatedUnion('type', [
}).strict(), }).strict(),
z.object({ z.object({
type: z.literal('remove'), type: z.literal('remove'),
id: z.string().min(1).max(256) id: activityIdSchema
}).strict(), }).strict(),
z.object({ type: z.literal('remove-duplicates') }).strict() z.object({ type: z.literal('remove-duplicates') }).strict(),
z.object({
type: z.literal('upsert-call'),
record: activityRecordSchema.refine(
(record) => (record.kind === 'tool' || record.kind === 'subagent') && record.callId !== undefined,
'upsert-call needs a tool or subagent record with a callId'
)
}).strict(),
z.object({
type: z.literal('update-request'),
requestId: activityIdSchema,
status: activityStatusSchema,
detail: z.string().optional()
}).strict(),
z.object({
type: z.literal('resolve-approval'),
conversationId: activityIdSchema,
status: activityStatusSchema,
detailLine: activityDetailLineSchema
}).strict(),
z.object({
type: z.literal('settle-request'),
requestId: activityIdSchema,
status: activityStatusSchema,
detailLine: activityDetailLineSchema
}).strict(),
z.object({
type: z.literal('remove-request'),
requestId: activityIdSchema
}).strict(),
z.object({
type: z.literal('remove-call'),
requestId: activityIdSchema,
callId: activityIdSchema
}).strict()
]) ])
export const activityHistoryUpdateSchema = z export const activityHistoryUpdateSchema = z
@@ -189,6 +239,8 @@ export type ActivityHistorySnapshot = z.infer<
> >
export type ActivityHistoryChange = z.infer<typeof activityHistoryChangeSchema> export type ActivityHistoryChange = z.infer<typeof activityHistoryChangeSchema>
export type ActivityHistoryUpdate = z.infer<typeof activityHistoryUpdateSchema> export type ActivityHistoryUpdate = z.infer<typeof activityHistoryUpdateSchema>
/** The stored record of each upsert-call change, in order. */
export type ActivityHistoryUpdateResult = { calls: ActivityRecord[] }
export const projectExecutionSpaceSchema = z.discriminatedUnion( export const projectExecutionSpaceSchema = z.discriminatedUnion(
'kind', 'kind',
+2 -1
View File
@@ -51,6 +51,7 @@ import {
type ActivityHistorySummary, type ActivityHistorySummary,
type ActivityHistorySummaryRequest, type ActivityHistorySummaryRequest,
type ActivityHistoryUpdate, type ActivityHistoryUpdate,
type ActivityHistoryUpdateResult,
type ActivityRecord, type ActivityRecord,
type TokenUsageSummary, type TokenUsageSummary,
type ConversationSnapshot, type ConversationSnapshot,
@@ -2169,7 +2170,7 @@ export type DesktopApi = {
* Incremental save (PERF-15): only the changed records cross IPC and are * Incremental save (PERF-15): only the changed records cross IPC and are
* written. `replace` remains for the legacy localStorage migration. * written. `replace` remains for the legacy localStorage migration.
*/ */
update: (update: ActivityHistoryUpdate) => Promise<void> update: (update: ActivityHistoryUpdate) => Promise<ActivityHistoryUpdateResult | void>
/** Deletes the whole history in one transaction. */ /** Deletes the whole history in one transaction. */
clear: () => Promise<void> clear: () => Promise<void>
/** One page of shown records, newest first (see activityHistoryPageRequestSchema). */ /** One page of shown records, newest first (see activityHistoryPageRequestSchema). */