From 2e181b1b7853e861c462dbdf7e2128f5be82a7f9 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 26 Oct 2026 22:19:15 -0700 Subject: [PATCH] Reuse chat history snapshots across readers (#34215) * Reuse rendered chat history across readers * test: cover current journal reads across source changes * test: align journal assertions with current snapshot reuse --- .../journal-public-read.ts | 5 +- .../agent-session-history-page.test.ts | 95 ++++++++++++++++++- .../agent-session-history-page.ts | 9 +- ...agent-session-stop-note-projection.test.ts | 2 +- ...agent-session-queued-legacy-import.test.ts | 16 +++- ...ructured-agent-session-subscribers.test.ts | 81 +++++++++++++++- .../agent-session-publication-scale.test.ts | 8 +- 7 files changed, 200 insertions(+), 16 deletions(-) diff --git a/src/main/native-chat/agent-session-journal/journal-public-read.ts b/src/main/native-chat/agent-session-journal/journal-public-read.ts index 65a146e3dfc..f01fe97c566 100644 --- a/src/main/native-chat/agent-session-journal/journal-public-read.ts +++ b/src/main/native-chat/agent-session-journal/journal-public-read.ts @@ -11,6 +11,7 @@ export function createJournalPublicReader() { sequence: number revision: string normalizationKey?: string + view?: ReturnType } | undefined return (raw: JournalReducerState, display: JournalReducerState, revision: string) => { @@ -24,10 +25,10 @@ export function createJournalPublicReader() { current = { raw, epoch, sequence, revision } } const source = current - return readJournalPublicView( + return (source.view ??= readJournalPublicView( display, () => (source.normalizationKey ??= journalNormalizationKey(display)) - ) + )) } } diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts index 6a2b47fe7fe..6e335b8420d 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts @@ -34,7 +34,16 @@ import type { } from '../agent-session-journal/journal-row-schema' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { projectJournalBatch } from './agent-session-journal-batch' -import { readAgentSessionHistory, resolveHistoryLimit } from './agent-session-history-page' +import { hostTestMessage } from './structured-agent-session-host-test-data' +import { + createAgentSessionCatchUpReader, + readAgentSessionHistory, + resolveHistoryLimit +} from './agent-session-history-page' +import { readAgentSessionConversationOutline } from './agent-session-conversation-outline' +import { shortenedAgentJournalItem } from './agent-session-item-wire-form' +import { firstStructuredChatNamingPrompt } from '../../../shared/structured-chat-naming-eligibility' +import { structuredSessionGateFacts } from '../../runtime/orchestration/structured-session-pointer-delivery' import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' const IDENTITY: AgentSessionJournalIdentity = { @@ -104,6 +113,90 @@ describe('resolveHistoryLimit', () => { }) describe('readAgentSessionHistory', () => { + it('refreshes a retained catch-up reader after a same-cursor disk reload', async () => { + const options = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + await journal.appendItem(item(0), { kind: 'turn', turnId: 'turn-1', state: 'running' }, options) + const view = journal.publicRead() + expect(view.snapshot.items).toHaveLength(1) + const cursor = journal.cursor() + const read = createAgentSessionCatchUpReader(journal, undefined, view) + const request = { + sessionId: IDENTITY.sessionId, + direction: 'after' as const, + cursor: { epoch: cursor.epoch, sequence: 0 } + } + expect(read(request)).toMatchObject({ + ok: true, + page: { items: [{ body: { state: 'running' } }], latestTurn: { turn: { state: 'running' } } } + }) + openTestJournalHostDatabase(root) + .db.prepare( + "UPDATE journal_rows SET row_json = json_set(row_json, '$.body.state', 'completed') WHERE session_id = ? AND epoch = ? AND seq = ?" + ) + .run(IDENTITY.sessionId, cursor.epoch, cursor.sequence) + await journal.open() + expect(journal.cursor()).toEqual(cursor) + expect(read(request)).toMatchObject({ + ok: true, + page: { + items: [{ body: { state: 'completed' } }], + latestTurn: { turn: { state: 'completed' } } + } + }) + }) + + it('serves current detail, outline, naming and mail facts without changing later history', async () => { + const options = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + await journal.appendSubmission({ + clientMessageId: 'u', + payloadFingerprint: 'u', + body: hostTestMessage('Name this'), + fence: 1 + }) + const turn = await journal.appendItem( + item(0), + { kind: 'turn', turnId: 'turn-1', state: 'running' }, + options + ) + const replies = { ...options, turnScope: { kind: 'turn' as const, turnItemId: turn.itemId } } + await journal.appendItem(item(1), body('obsolete reply'), replies) + await journal.appendItem(item(2), body('initial reply'), replies) + const warm = journal.snapshot() + await journal.resolveDispatch({ + clientMessageId: 'u', + state: 'accepted', + providerIdentity: null, + fence: 1 + }) + await journal.appendItem(item(2), body('Revised reply'), replies) + await journal.appendTombstone(item(1), { fence: 1 }) + await journal.appendItem( + item(0), + { kind: 'turn', turnId: 'turn-1', state: 'completed' }, + options + ) + const current = journal.snapshot() + const expected = structuredClone(current) + expect(current).not.toBe(warm) + const detail = current.items.at(-1)! + expect(shortenedAgentJournalItem(detail, 1024)?.body).toMatchObject({ + blocks: [{ text: 'Revised reply' }] + }) + expect(readAgentSessionConversationOutline(current).entries).toMatchObject([ + { preview: 'Name this', reply: 'Revised reply' } + ]) + expect(firstStructuredChatNamingPrompt(current, 0)).toBe('Name this') + expect(structuredSessionGateFacts(current.items).turnRunning).toBe(false) + expect(current.submissions).toMatchObject([{ clientMessageId: 'u', dispatchState: 'accepted' }]) + expect( + readAgentSessionHistory(journal, { sessionId: IDENTITY.sessionId, direction: 'tail' }) + ).toMatchObject({ + ok: true, + page: { items: expected.items, submissions: expected.submissions } + }) + expect(journal.snapshot()).toEqual(expected) + }) + it('serves the newest page on tail and pages backward from it', async () => { await appendItems(5) const tail = readAgentSessionHistory(journal, { diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts index 0d04e33eea5..7c2263effc3 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts @@ -132,7 +132,7 @@ export function readAgentSessionHistory( } } -/** Reuse reduced state until the journal cursor advances. */ +/** Reuse reduced state until the complete journal source changes. */ export function createAgentSessionCatchUpReader( journal: AgentSessionJournal, reader: HistoryReader = CLIENT_HISTORY_READER, @@ -140,10 +140,9 @@ export function createAgentSessionCatchUpReader( ): (request: AgentSessionHistoryRequest) => AgentSessionHistoryResult { let view = initialView return (request) => { - const live = journal.cursor() - const prior = view.snapshot.cursor - if (live.epoch !== prior.epoch || live.sequence !== prior.sequence) { - view = journal.publicRead() + const live = journal.publicRead() + if (view !== live) { + view = live } return readAgentSessionHistory(journal, request, { view, reader, rowReplay: true }) } diff --git a/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts b/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts index 8cf0d4d3883..844b1a625b2 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts @@ -98,7 +98,7 @@ it.each([false, true])( expect(projected.body).toEqual(ordinary) expect(projected).toEqual({ ...prior, body: ordinary }) expect(projected).not.toBe(prior) - expect(readNote()).not.toBe(projected) + expect(readNote()).toBe(projected) expect(journal.itemBody(noteId)).toBe(raw) expect(rows()).toEqual(stored) expect(stored.filter((row) => row.kind === 'item' && row.itemId === noteId)).toHaveLength(1) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts index 7b199246fd8..d23c33a6ba1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts @@ -17,7 +17,10 @@ import { reduceStructuredAgentSession } from '../../../shared/structured-agent-session-reducer' import { AgentSessionSubscribers } from './structured-agent-session-subscribers' -import { readAgentSessionHistory } from './agent-session-history-page' +import { + createAgentSessionCatchUpReader, + readAgentSessionHistory +} from './agent-session-history-page' import { subscriberHistoryReader } from './agent-session-bounded-frames' import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' import { AgentSessionJournal } from '../agent-session-journal/journal-store' @@ -425,9 +428,11 @@ it.each([100, 1_000].flatMap((size) => [false, true].map((adopt) => ({ size, ado expect(work()).toEqual([0, 0, 0]) expect(sourceGets()).toHaveLength(0) const cursor = journal.cursor() + const snapshot = journal.snapshot() await journal.open() expect(journal.cursor()).toEqual(cursor) expect(journal.queuedMessages.displayState() === display).toBe(false) + expect(journal.snapshot()).not.toBe(snapshot) feed.publish(SESSION, journal) expect(work()).toEqual([corrections, 1, 1]) reset() @@ -547,13 +552,20 @@ it('public metadata follows legacy-only commits, item recreation and epoch repla await journal.open() const ids = vi.spyOn(journal.queuedMessages.displayState().normalizedHandOffs, Symbol.iterator) const key = journal.publicRead().normalizationKey + const snapshot = journal.snapshot() + expect(journal.snapshot()).toBe(snapshot) expect(key).not.toBe('') const cursor = journal.cursor() + const catchUp = createAgentSessionCatchUpReader(journal, undefined, journal.publicRead()) + const request = { sessionId: SESSION, direction: 'tail' as const } + const beforeImport = catchUp(request) + expect(beforeImport).toMatchObject({ ok: true, page: { items: snapshot.items } }) db().exec('DROP TRIGGER b2_source_fail_import') await journal.queuedMessages.importAtOpen() expect(journal.cursor()).toEqual(cursor) expect(legacyIds()).toEqual([]) - journal.snapshot() + expect(journal.snapshot()).not.toBe(snapshot) + expect(catchUp(request)).toEqual(beforeImport) expect(ids).toHaveBeenCalledTimes(1) expect(journal.publicRead().normalizationKey).toBe(key) expect(ids).toHaveBeenCalledTimes(2) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts index 8d3c1fddc77..0e1c74f7f3f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts @@ -1,7 +1,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { AGENT_JOURNAL_THREAD_SCOPE, AGENT_SESSION_JOURNAL_SCHEMA_VERSION @@ -21,6 +21,9 @@ import { openTestJournalHostDatabase, insertTestJournalRow } from '../agent-session-journal/journal-host-database-test-support' +import * as journalReducer from '../agent-session-journal/journal-reducer' +import { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work' +import { StructuredAgentSessionJournalProjections } from './structured-agent-session-status-journal-projection' import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' import { MAX_RETAINED_SESSION_ACTIVITIES } from './structured-agent-session-activity-retention' import { AgentSessionSubscribers } from './structured-agent-session-subscribers' @@ -45,6 +48,82 @@ afterEach(async () => { }) describe('AgentSessionSubscribers', () => { + it('renders history once per source for N/10N items and one/three readers', async () => { + const render = vi.spyOn(journalReducer, 'renderJournalState') + try { + for (const size of [100, 1000]) { + for (const readers of [1, 3]) { + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: codexProviderHandle('thread-1') + }, + stateDirectory: join(root, `${size}-${readers}`) + }) + const options = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + const identity = (ordinal: number) => ({ + provider: 'codex' as const, + threadId: 'thread-1', + turnId: 'turn', + ordinal + }) + const body = (text: string) => ({ + kind: 'message' as const, + role: 'assistant' as const, + blocks: [{ type: 'text' as const, text }] + }) + for (let index = 0; index < size; index++) { + await journal.appendItem(identity(index), body('initial'), options) + } + const projections = new StructuredAgentSessionJournalProjections() + const events: AgentSessionSubscribeEvent[] = [] + const subscribers = new AgentSessionSubscribers({ + readCurrentWork: () => new StructuredAgentSessionCurrentWork(journal, 1, 0), + onJournalPublished: () => { + projections.read(journal, null) + } + }) + for (let index = 0; index < readers; index++) { + subscribers.open({ + id: String(index), + sessionId: SESSION, + journal, + fence: 1, + boundedFrames: index === 1, + emit: (event) => events.push(event) + }) + } + subscribers.publish(SESSION, journal) + for (const text of ['streamed', 'newer']) { + const prior = journal.snapshot() + await journal.appendItem(identity(size - 1), body(text), options) + render.mockClear() + events.length = 0 + subscribers.publish(SESSION, journal) + expect(render).toHaveBeenCalledTimes(1) + const current = journal.snapshot() + expect(current).not.toBe(prior) + expect(current.items).toHaveLength(size) + expect(current.items.at(-1)?.body).toEqual(body(text)) + expect(journal.publicRead().snapshot).toBe(current) + expect(events.filter((event) => event.type === 'batch')).toHaveLength(2 * readers) + expect( + events + .flatMap((event) => (event.type === 'batch' ? event.batch.items : [])) + .map((item) => item.body) + ).toEqual(Array.from({ length: readers }, () => body(text))) + expect(render).toHaveBeenCalledTimes(1) + } + } + } + } finally { + render.mockRestore() + } + }) + it('bounds retained turn activity across session churn', async () => { const journal = await journals.open({ identity: { diff --git a/src/main/runtime/agent-session-publication-scale.test.ts b/src/main/runtime/agent-session-publication-scale.test.ts index d9bb285c76e..7c9b87da283 100644 --- a/src/main/runtime/agent-session-publication-scale.test.ts +++ b/src/main/runtime/agent-session-publication-scale.test.ts @@ -201,10 +201,10 @@ describe('real-store status and queue publications', () => { .filter((event) => event.type === 'snapshot') .every((event) => 'queuedMessages' in event && event.queuedMessages?.length === cards) ).toBe(true) - // Status projection, hydration and initial current-work restatement each read once. - expect(counts.snapshots).toBe(3 * n) - expect(counts.snapshotItems).toBe(3 * n * (history + sends)) - expect(counts.snapshotSubmissions).toBe(3 * n * sends) + // Status projection, hydration and current-work restatement reuse one current-source render. + expect(counts.snapshots).toBe(n) + expect(counts.snapshotItems).toBe(n * (history + sends)) + expect(counts.snapshotSubmissions).toBe(n * sends) expect(counts.serializations).toBe(n) expect(counts.serializedRows).toBe(n * cards) const warm = { ...counts }