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

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<typeof readJournalPublicView>
       }
     | 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))
-    )
+    ))
   }
 }
 

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, {

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 })
   }

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)

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)

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: {

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 }