Show retained chat messages after reconnect without reopening (#32255)

* fix(native-chat): publish retained history corrections to capable clients

* test(native-chat): expose canonical reads in phone history fixture

* test(native-chat): migrate remaining history reader fixtures

* fix(native-chat): preserve expanded history on late acceptance

* fix(native-chat): share lazy correction metadata by journal source

* fix(native-chat): reuse raw legacy display selections

config/scripts/native-chat-compaction-pagination.test.mjs

@@ -102,9 +102,8 @@ function slotsFor(page) {
 it('keeps the captured skip visible when the actual 200-item history tail omits only its opener', async () => {
   const snapshot = await snapshotWithLaterHistory()
   const result = readAgentSessionHistory(
-    { snapshot: () => snapshot },
-    { sessionId: 'session-1', direction: 'tail', limit: 200 },
-    snapshot
+    { publicRead: () => ({ snapshot, normalizationKey: '' }) },
+    { sessionId: 'session-1', direction: 'tail', limit: 200 }
   )
   expect(result.ok).toBe(true)
   const page = result.page
@@ -125,14 +124,12 @@ it('keeps the captured skip visible when the actual 200-item history tail omits
 it('restores the captured skip after the real older-page read loads its opener', async () => {
   const snapshot = await snapshotWithLaterHistory()
   const tail = readAgentSessionHistory(
-    { snapshot: () => snapshot },
-    { sessionId: 'session-1', direction: 'tail', limit: 200 },
-    snapshot
+    { publicRead: () => ({ snapshot, normalizationKey: '' }) },
+    { sessionId: 'session-1', direction: 'tail', limit: 200 }
   ).page
   const older = readAgentSessionHistory(
-    { snapshot: () => snapshot },
-    { sessionId: 'session-1', direction: 'before', cursor: tail.window.nextCursor, limit: 200 },
-    snapshot
+    { publicRead: () => ({ snapshot, normalizationKey: '' }) },
+    { sessionId: 'session-1', direction: 'before', cursor: tail.window.nextCursor, limit: 200 }
   ).page
   expect(older.items.map((item) => item.itemId)).toEqual(['orca:compact-1'])
   const slots = slotsFor({
@@ -156,9 +153,8 @@ it('continues folding an ordinary warning at the same real history boundary', as
     item.body.kind === 'status' ? { ...item, itemId: 'orca:ordinary-warning' } : item
   )
   const result = readAgentSessionHistory(
-    { snapshot: () => snapshot },
-    { sessionId: 'session-1', direction: 'tail', limit: 200 },
-    snapshot
+    { publicRead: () => ({ snapshot, normalizationKey: '' }) },
+    { sessionId: 'session-1', direction: 'tail', limit: 200 }
   )
   expect(result.ok).toBe(true)
   const slot = slotsFor(result.page).find(

mobile/src/session/mobile-native-chat-host-journal-frames.test-fixture.ts

@@ -1,15 +1,12 @@
 // What the host's own fold and history readers send a phone subscribed to a journal, at any row.
 
-import type {
-  AgentJournalCursor,
-  AgentJournalSnapshot
-} from '../../../src/shared/agent-session-journal-types'
+import type { AgentJournalCursor } from '../../../src/shared/agent-session-journal-types'
 import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire'
 import {
   applyJournalRow,
-  createJournalReducerState,
-  renderJournalState
+  createJournalReducerState
 } from '../../../src/main/native-chat/agent-session-journal/journal-reducer'
+import { readJournalPublicView } from '../../../src/main/native-chat/agent-session-journal/journal-public-read'
 import type { JournalRow } from '../../../src/main/native-chat/agent-session-journal/journal-row-schema'
 import type { AgentSessionJournal } from '../../../src/main/native-chat/agent-session-journal/journal-store'
 import {
@@ -40,21 +37,23 @@ export function hostJournalFrames(
 ): HostJournalFrames {
   // One fold, read at every row: each row's snapshot and the aliases its readers resolve with.
   const state = createJournalReducerState(sessionId, epoch)
-  const snapshots: AgentJournalSnapshot[] = []
+  const views: ReturnType<typeof readJournalPublicView>[] = []
   const aliases: ReadonlyMap<string, string>[] = []
   const fences: number[] = []
   for (const row of rows) {
     applyJournalRow(state, row)
     // A copy: the fold revises its submissions in place.
-    snapshots.push(structuredClone(renderJournalState(state)))
+    views.push(structuredClone(readJournalPublicView(state)))
     aliases.push(new Map(state.aliases))
     fences.push(Math.max(row.fence, fences.at(-1) ?? 0))
   }
   const journalAt = (upTo: number): AgentSessionJournal => {
-    const snapshot = snapshots[upTo - 1]!
+    const view = views[upTo - 1]!
+    const { snapshot } = view
     const journal = {
       isReadOnly: false,
       snapshot: () => snapshot,
+      publicRead: () => view,
       cursor: () => snapshot.cursor,
       // Without a body the store's own resolution reads only the aliases.
       canonicalItemId: (itemId: string) => aliases[upTo - 1]!.get(itemId) ?? itemId,

src/main/native-chat/agent-session-journal/journal-public-read.ts

@@ -0,0 +1,62 @@
+import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types'
+import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
+import { renderJournalState, type JournalReducerState } from './journal-reducer'
+
+/** Only a committed journal owner can certify reuse across public reads. */
+export function createJournalPublicReader() {
+  let current:
+    | {
+        raw: JournalReducerState
+        epoch: string
+        sequence: number
+        revision: string
+        normalizationKey?: string
+      }
+    | undefined
+  return (raw: JournalReducerState, display: JournalReducerState, revision: string) => {
+    const { epoch, lastSequence: sequence } = raw
+    if (
+      current?.raw !== raw ||
+      current.epoch !== epoch ||
+      current.sequence !== sequence ||
+      current.revision !== revision
+    ) {
+      current = { raw, epoch, sequence, revision }
+    }
+    const source = current
+    return readJournalPublicView(
+      display,
+      () => (source.normalizationKey ??= journalNormalizationKey(display))
+    )
+  }
+}
+
+/** Row progress cannot certify a historical display that the queue fold corrected. */
+export function readJournalPublicView(
+  state: JournalReducerState,
+  normalizationKey = () => journalNormalizationKey(state)
+): {
+  snapshot: AgentJournalSnapshot
+  normalizationKey: string
+} {
+  let snapshot: AgentJournalSnapshot | undefined
+  return {
+    get snapshot() {
+      return (snapshot ??= renderJournalState(state))
+    },
+    get normalizationKey() {
+      return normalizationKey()
+    }
+  }
+}
+
+function journalNormalizationKey(state: JournalReducerState): string {
+  const corrections = [...state.normalizedHandOffs].sort().flatMap((id) => {
+    const item = state.items.get(agentJournalSubmissionKey(id))
+    const submission = state.submissions.get(id)
+    return item && submission
+      ? [[id, item.sequence, item.observedAt, item.turnScope, submission.queuedMessageId ?? null]]
+      : []
+  })
+  return corrections.length === 0 ? '' : JSON.stringify([state.epoch, corrections])
+}

src/main/native-chat/agent-session-journal/journal-queue-import.ts

@@ -91,7 +91,12 @@ function unanswered(
 export class LegacyQueuedCards {
   private revision = 0
   private stored: { revision: number; rows: readonly QueuedMessageRow[] } | null = null
-  private viewed: { key: string; cards: readonly QueuedMessageRow[] } | null = null
+  private viewed: {
+    owner: JournalReducerState
+    key: string
+    cards: readonly QueuedMessageRow[]
+    display?: JournalReducerState
+  } | null = null
 
   constructor(private readonly deps: LegacyDeps) {}
 
@@ -107,15 +112,16 @@ export class LegacyQueuedCards {
   cards(): readonly QueuedMessageRow[] {
     const key = this.key()
     if (key === '') {
+      this.viewed = null
       return []
     }
-    if (this.viewed?.key !== key) {
-      const state = this.deps.state()
+    const state = this.deps.state()
+    if (this.viewed?.key !== key || this.viewed.owner !== state) {
       const cards = this.rows().flatMap((row) => {
         const change = legacyQueuedChange(state, row)
         return change ? [queuedCardFromChange(state, row.messageId, change)] : []
       })
-      this.viewed = { key, cards }
+      this.viewed = { owner: state, key, cards }
     }
     return this.viewed.cards
   }
@@ -123,16 +129,24 @@ export class LegacyQueuedCards {
   // Failed adoption cannot hide spent sends; derive display changes only on copied state.
   displayState(): JournalReducerState {
     const state = this.deps.state()
-    const handedOff = this.cards().flatMap((card) => {
+    const cards = this.cards()
+    if (this.viewed?.display) {
+      return this.viewed.display
+    }
+    const handedOff = cards.flatMap((card) => {
       const submission =
         card.consumedAs === null ? undefined : state.submissions.get(card.consumedAs)
       return submission && handOffInDoubt(submission) ? [{ ...submission }] : []
     })
     if (handedOff.length === 0) {
+      if (this.viewed) {
+        this.viewed.display = state
+      }
       return state
     }
     const display = {
       ...state,
+      normalizedHandOffs: new Set(state.normalizedHandOffs),
       items: new Map(state.items),
       submissions: new Map(state.submissions)
     }
@@ -140,6 +154,9 @@ export class LegacyQueuedCards {
       display.submissions.set(submission.clientMessageId, submission)
       foldQueuedHandOffInDoubt(display, submission)
     }
+    if (this.viewed) {
+      this.viewed.display = display
+    }
     return display
   }
 

src/main/native-chat/agent-session-journal/journal-reducer.ts

@@ -75,6 +75,9 @@ export type JournalReducerState = {
     string,
     Pick<AgentJournalRenderItem, 'sequence' | 'observedAt' | 'turnScope'>
   >
+  /** Historical display corrections, rebuilt by the hand-off fold on replay. */
+  normalizedHandOffs: Set<string>
+  /** The queued messages, folded from their own rows and their hand-offs (`journal-queue-fold.ts`). */
   queue: JournalQueueState
   historyOrigin: JournalHistoryOrigin | null
 }
@@ -103,6 +106,7 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou
     queuePauseMarks: createJournalQueuePauseMarks(),
     stoppedHistoryCarried: new Map(),
     queuedHandOffPlaces: new Map(),
+    normalizedHandOffs: new Set(),
     queue: createJournalQueueState(),
     historyOrigin: null
   }

src/main/native-chat/agent-session-journal/journal-store.ts

@@ -2,6 +2,7 @@
 
 import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words'
 import { randomUUID } from 'node:crypto'
+import { createJournalPublicReader } from './journal-public-read'
 import type {
   AgentJournalAcceptanceReceipt,
   AgentJournalCursor,
@@ -32,7 +33,6 @@ import {
 import {
   applyJournalRow,
   createJournalReducerState,
-  renderJournalState,
   resolveJournalItemId,
   type JournalReducerState
 } from './journal-reducer'
@@ -80,6 +80,7 @@ export class AgentSessionJournal {
   private readonly mintEpoch: () => string
 
   private state: JournalReducerState
+  private readonly readPublic = createJournalPublicReader()
   private openedThrough: AgentJournalCursor = { epoch: '', sequence: 0 }
   private reopenUnmarked: AgentJournalCursor | null = null
   private onCommitted: ((rows?: readonly JournalRow[]) => void) | null = null
@@ -184,7 +185,10 @@ export class AgentSessionJournal {
     sequence: this.state.lastSequence
   })
 
-  snapshot = (): AgentJournalSnapshot => renderJournalState(this.queuedMessages.displayState())
+  snapshot = (): AgentJournalSnapshot => this.publicRead().snapshot
+
+  publicRead = () =>
+    this.readPublic(this.state, this.queuedMessages.displayState(), this.queuedMessages.revision())
 
   /** Visit without allocating/sorting a snapshot. */
   visitItems = (

src/main/native-chat/agent-session-journal/journal-submission-fold.ts

@@ -79,6 +79,7 @@ export function foldQueuedHandOffInDoubt(
   state: JournalReducerState,
   submission: AgentJournalSubmission
 ): void {
+  state.normalizedHandOffs.add(submission.clientMessageId)
   const itemId = agentJournalSubmissionKey(submission.clientMessageId)
   const item = state.items.get(itemId)
   const place = state.queuedHandOffPlaces.get(submission.clientMessageId)

src/main/native-chat/agent-session-wire/agent-session-history-latest-turn.ts

@@ -0,0 +1,15 @@
+import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types'
+import type { AgentSessionLatestTurn } from '../../../shared/agent-session-wire'
+import { latestStructuredAgentSessionTurn } from '../../../shared/structured-agent-session-live-turn'
+
+// A catch-up run pages one snapshot many times; its scan back to the turn record is paid once.
+const latestTurnBySnapshot = new WeakMap<AgentJournalSnapshot, AgentSessionLatestTurn | null>()
+
+export function snapshotLatestTurn(snapshot: AgentJournalSnapshot): AgentSessionLatestTurn | null {
+  let latest = latestTurnBySnapshot.get(snapshot)
+  if (latest === undefined) {
+    latest = latestStructuredAgentSessionTurn(snapshot.items)
+    latestTurnBySnapshot.set(snapshot, latest)
+  }
+  return latest
+}

src/main/native-chat/agent-session-wire/agent-session-history-limit.ts

@@ -0,0 +1,13 @@
+import {
+  AGENT_SESSION_HISTORY_DEFAULT_LIMIT,
+  AGENT_SESSION_HISTORY_MAX_LIMIT
+} from '../../../shared/agent-session-wire'
+
+/** Clamped, never rejected: a client asking for more than the host will serve
+ *  should get a smaller page and keep paging, not an error mid-scroll. */
+export function resolveHistoryLimit(limit: number | undefined): number {
+  if (limit === undefined || !Number.isFinite(limit)) {
+    return AGENT_SESSION_HISTORY_DEFAULT_LIMIT
+  }
+  return Math.min(AGENT_SESSION_HISTORY_MAX_LIMIT, Math.max(1, Math.floor(limit)))
+}

src/main/native-chat/agent-session-wire/agent-session-history-page.ts

@@ -1,8 +1,17 @@
-// Backward pages read reduced items; catch-up reads rows so late revisions of early items are not lost.
+import { resolveHistoryLimit } from './agent-session-history-limit'
+export { resolveHistoryLimit } from './agent-session-history-limit'
+// Paged history over one journal.
+//
+// `tail` and `before` read the REDUCED timeline, so backward paging keeps
+// working after compaction — the folded snapshot still holds every live item.
+// `after` is the catch-up direction and must read rows instead: an item created
+// early and revised late orders by its creation sequence, so an item-window
+// read would silently skip that revision. Rows carry the revision, which is why
+// `after` is the only direction that can answer `cursor_compacted`.
 
 import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
 import { isRootAgentJournalItem } from '../../../shared/agent-session-journal-producer'
-import { latestStructuredAgentSessionTurn } from '../../../shared/structured-agent-session-live-turn'
+import { snapshotLatestTurn } from './agent-session-history-latest-turn'
 import { isSubagentBlock } from '../../../shared/native-chat-types'
 import type {
   AgentJournalCursor,
@@ -10,13 +19,11 @@ import type {
   AgentJournalSnapshot
 } from '../../../shared/agent-session-journal-types'
 import {
-  AGENT_SESSION_HISTORY_DEFAULT_LIMIT,
   AGENT_SESSION_HISTORY_MAX_LIMIT,
   type AgentSessionHistoryDirection,
   type AgentSessionHistoryPage,
   type AgentSessionHistoryRequest,
   type AgentSessionHistoryResult,
-  type AgentSessionLatestTurn,
   type AgentSessionSubagentRosterEntry
 } from '../../../shared/agent-session-wire'
 import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
@@ -35,15 +42,9 @@ import { offPageSubagentRoster } from './agent-session-history-subagent-roster'
 
 export { AGENT_SESSION_HISTORY_MAX_PAGE_BYTES } from './agent-session-history-page-bounds'
 
-/** Clamp oversized requests so clients can keep paging. */
-export function resolveHistoryLimit(limit: number | undefined): number {
-  if (limit === undefined || !Number.isFinite(limit)) {
-    return AGENT_SESSION_HISTORY_DEFAULT_LIMIT
-  }
-  return Math.min(AGENT_SESSION_HISTORY_MAX_LIMIT, Math.max(1, Math.floor(limit)))
-}
-
-/** Both scopes window on session-owned rows; every-agent also includes interleaved subagent rows. */
+/** Whose rows a backward read serves. Both window over the session's own rows; a
+ *  reader of the session's own agent alone gets only those. In-process only: no wire
+ *  request carries it. */
 export type AgentSessionHistoryScope = 'every-agent' | 'own-agent'
 
 /** Window by own-agent rows, retaining contiguous subagent work and whole sibling groups within byte bounds. */
@@ -74,18 +75,31 @@ export function readAgentSessionHistory(
   journal: AgentSessionJournal,
   request: AgentSessionHistoryRequest,
   {
-    snapshot = journal.snapshot(),
+    view = journal.publicRead(),
+    snapshot = view.snapshot,
     scope = 'every-agent',
-    reader = CLIENT_HISTORY_READER
+    reader = CLIENT_HISTORY_READER,
+    rowReplay = false
   }: {
     /** Reuse a snapshot across synchronous catch-up pages to avoid repeated full reduction. */
     snapshot?: AgentJournalSnapshot
+    view?: ReturnType<AgentSessionJournal['publicRead']>
     scope?: AgentSessionHistoryScope
     reader?: HistoryReader
+    /** Internal catch-up and older item-backed clients keep their existing row replay. */
+    rowReplay?: boolean
   } = {}
 ): AgentSessionHistoryResult {
   const limit = resolveHistoryLimit(request.limit)
   if (request.direction === 'after') {
+    if (
+      !rowReplay &&
+      request.cursor?.epoch === snapshot.cursor.epoch &&
+      request.cursor.sequence <= snapshot.cursor.sequence &&
+      view.normalizationKey !== ''
+    ) {
+      return historyReset(snapshot, 'cursor_compacted', reader)
+    }
     return readForward(journal, snapshot, request.cursor, limit, reader)
   }
   const cursor = request.direction === 'before' ? request.cursor : undefined
@@ -128,24 +142,27 @@ export function readAgentSessionHistory(
 /** Reuse reduced state until the journal cursor advances. */
 export function createAgentSessionCatchUpReader(
   journal: AgentSessionJournal,
-  reader: HistoryReader = CLIENT_HISTORY_READER
+  reader: HistoryReader = CLIENT_HISTORY_READER,
+  initialView = journal.publicRead()
 ): (request: AgentSessionHistoryRequest) => AgentSessionHistoryResult {
-  let snapshot = journal.snapshot()
+  let view = initialView
   return (request) => {
     const live = journal.cursor()
-    if (live.epoch !== snapshot.cursor.epoch || live.sequence !== snapshot.cursor.sequence) {
-      snapshot = journal.snapshot()
+    const prior = view.snapshot.cursor
+    if (live.epoch !== prior.epoch || live.sequence !== prior.sequence) {
+      view = journal.publicRead()
     }
-    return readAgentSessionHistory(journal, request, { snapshot, reader })
+    return readAgentSessionHistory(journal, request, { view, reader, rowReplay: true })
   }
 }
 
 export function readAgentSessionHydrationPage(
   journal: AgentSessionJournal,
   fence?: number,
-  reader: HistoryReader = CLIENT_HISTORY_READER
+  reader: HistoryReader = CLIENT_HISTORY_READER,
+  snapshot: AgentJournalSnapshot = journal.snapshot()
 ): AgentSessionHistoryPage {
-  return buildHydrationPage(journal.snapshot(), fence, reader)
+  return buildHydrationPage(snapshot, fence, reader)
 }
 
 function buildHydrationPage(
@@ -314,15 +331,3 @@ function buildPage(input: {
     latestTurn: snapshotLatestTurn(input.snapshot)
   }
 }
-
-// Cache the turn scan across catch-up pages sharing a snapshot.
-const latestTurnBySnapshot = new WeakMap<AgentJournalSnapshot, AgentSessionLatestTurn | null>()
-
-export function snapshotLatestTurn(snapshot: AgentJournalSnapshot): AgentSessionLatestTurn | null {
-  let latest = latestTurnBySnapshot.get(snapshot)
-  if (latest === undefined) {
-    latest = latestStructuredAgentSessionTurn(snapshot.items)
-    latestTurnBySnapshot.set(snapshot, latest)
-  }
-  return latest
-}

src/main/native-chat/agent-session-wire/agent-session-published-work.ts

@@ -3,7 +3,7 @@ import type {
   AgentSessionHistoryResult
 } from '../../../shared/agent-session-wire'
 import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
-import { snapshotLatestTurn } from './agent-session-history-page'
+import { snapshotLatestTurn } from './agent-session-history-latest-turn'
 import type { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work'
 
 /** Absent projection preserves legacy raw-turn behavior without current prompt IDs. */

src/main/native-chat/agent-session-wire/agent-session-subscribe-input.ts

@@ -0,0 +1,16 @@
+import type { AgentJournalCursor } from '../../../shared/agent-session-journal-types'
+import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire'
+import type { SubscriberFieldState } from './agent-session-subscriber-frame-fields'
+
+export type AgentSessionSubscribeInput = {
+  id: string
+  sessionId: string
+  emit: (event: AgentSessionSubscribeEvent) => void
+  cursor?: AgentJournalCursor
+  /** Every frame to fit a phone's bridge frame; the RPC sets it for every phone. */
+  boundedFrames?: boolean
+  projectQueuePublication?: SubscriberFieldState['projectQueuePublication']
+  readsCurrentRequests?: boolean
+  /** False for desktops that need exact replay to keep send echoes and current requests. */
+  snapshotCatchUp?: boolean
+}

src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts

@@ -37,6 +37,7 @@ export type SubscriberDeliveryPort = {
     options?: {
       withholdQueued?: boolean
       currentWork?: AgentSessionPublishedWork
+      historyView?: string
     }
   ) => void
   isActive: (subscriber: Subscriber) => boolean
@@ -64,6 +65,41 @@ export function deliverToSubscriber(
   const emit: SubscriberDeliveryPort['emit'] = (target, event, options) =>
     port.emit(target, event, { ...options, currentWork: work })
   const delivery = { ...port, emit }
+  const view = journal.publicRead()
+  if (
+    subscriber.snapshotCatchUp !== false &&
+    subscriber.readsCurrentRequests &&
+    subscriber.cursor.epoch === journal.epoch &&
+    view.normalizationKey !== '' &&
+    subscriber.historyView !== view.normalizationKey
+  ) {
+    const page = withPublishedWork(
+      readAgentSessionHydrationPage(
+        journal,
+        subscriber.fence,
+        subscriberHistoryReader(subscriber),
+        view.snapshot
+      ),
+      work
+    )
+    emit(
+      subscriber,
+      {
+        type: 'snapshot',
+        sessionId: subscriber.sessionId,
+        page,
+        fence: subscriber.fence,
+        ...shared
+      },
+      { historyView: view.normalizationKey }
+    )
+    if (port.isActive(subscriber)) {
+      subscriber.cursor = page.liveCursor ?? page.window.nextCursor
+      refreshPublishedWork(delivery, subscriber, journal, work, shared)
+    }
+    return
+  }
+  // Caught up, so there are no rows to read: every publish behind a commit's own delivery.
   if (sameJournalCursor(subscriber.cursor, journal.cursor())) {
     emitCaughtUp(delivery, subscriber, emitCheckpoint, shared, {
       journal,
@@ -72,7 +108,7 @@ export function deliverToSubscriber(
     return
   }
   const reader = subscriberHistoryReader(subscriber)
-  const readPage = createAgentSessionCatchUpReader(journal, reader)
+  const readPage = createAgentSessionCatchUpReader(journal, reader, view)
   while (true) {
     const result = withPublishedWorkResult(
       readPage({
@@ -85,14 +121,18 @@ export function deliverToSubscriber(
     )
     if (!result.ok) {
       const page = { ...result.page, fence: subscriber.fence }
-      emit(subscriber, {
-        type: 'reset',
-        sessionId: subscriber.sessionId,
-        reset: result.reset,
-        page,
-        fence: subscriber.fence,
-        ...shared
-      })
+      emit(
+        subscriber,
+        {
+          type: 'reset',
+          sessionId: subscriber.sessionId,
+          reset: result.reset,
+          page,
+          fence: subscriber.fence,
+          ...shared
+        },
+        { historyView: view.normalizationKey }
+      )
       subscriber.cursor = page.liveCursor ?? page.window.nextCursor
       refreshPublishedWork(delivery, subscriber, journal, work, shared)
       return
@@ -112,14 +152,18 @@ export function deliverToSubscriber(
         readAgentSessionHydrationPage(journal, subscriber.fence, reader),
         work
       )
-      emit(subscriber, {
-        type: 'snapshot',
-        sessionId: subscriber.sessionId,
-        page: tail,
-        fence: subscriber.fence,
-        hostNow,
-        activity: port.activity(subscriber.sessionId)
-      })
+      emit(
+        subscriber,
+        {
+          type: 'snapshot',
+          sessionId: subscriber.sessionId,
+          page: tail,
+          fence: subscriber.fence,
+          hostNow,
+          activity: port.activity(subscriber.sessionId)
+        },
+        { historyView: view.normalizationKey }
+      )
       subscriber.cursor = tail.liveCursor ?? tail.window.nextCursor
       refreshPublishedWork(delivery, subscriber, journal, work, shared)
       return

src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts

@@ -38,7 +38,8 @@ export class StructuredAgentSessionBackgroundTaskChannel {
 
   async history(
     request: AgentSessionHistoryRequest,
-    scope?: AgentSessionHistoryScope
+    scope?: AgentSessionHistoryScope,
+    replaceNormalizedHistory = true
   ): Promise<AgentSessionHistoryResult> {
     const journal = (await this.conversation(request.sessionId)).journal
     const work = hostStructuredAgentSessionCurrentWork(
@@ -50,6 +51,7 @@ export class StructuredAgentSessionBackgroundTaskChannel {
       record: this.deps.store.getRecord(request.sessionId),
       request,
       scope,
+      replaceNormalizedHistory,
       ...(work ? { work } : {})
     })
     const backgroundTasks = this.read(request.sessionId)

src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts

@@ -27,10 +27,15 @@ export function readStructuredAgentSessionHistoryResult(input: {
   record: AgentSessionRecord | null
   request: AgentSessionHistoryRequest
   scope?: AgentSessionHistoryScope
+  replaceNormalizedHistory?: boolean
+  /** The host's current-work projection, which the page's turn and prompt fields state. */
   work?: AgentSessionPublishedWork
 }): AgentSessionHistoryResult {
   const result = withPublishedWorkResult(
-    readAgentSessionHistory(input.journal, input.request, { scope: input.scope }),
+    readAgentSessionHistory(input.journal, input.request, {
+      scope: input.scope,
+      rowReplay: input.replaceNormalizedHistory === false
+    }),
     input.work
   )
   const fence = input.record?.lease.runtimeFence

src/main/native-chat/agent-session-wire/structured-agent-session-host.ts

@@ -317,8 +317,8 @@ export class StructuredAgentSessionHost {
   /** Read the store; idle-released chats have no session-map entry. */
   handoffStatus = (sessionId: string) => structuredAgentSessionOwnerStatus(this.deps, sessionId)
 
-  history: StructuredAgentSessionBackgroundTaskChannel['history'] = (request, scope) =>
-    this.backgroundTasks.history(request, scope)
+  history: StructuredAgentSessionBackgroundTaskChannel['history'] = (...args) =>
+    this.backgroundTasks.history(...args)
 
   /** Full reduction: absence from a bounded page cannot prove a revised/tombstoned item is absent. */
   journalSnapshot = async (sessionId: string) =>

src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts

@@ -4,12 +4,21 @@ import { agentSessionFailureFact } from '../../../shared/agent-session-failure'
 import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words'
 import {
   AGENT_JOURNAL_THREAD_SCOPE,
+  type AgentJournalAnsweredTurnIdentity,
   type AgentJournalMessageItem
 } from '../../../shared/agent-session-journal-types'
 import { activeProviderContext } from '../../../shared/agent-session-provider-context'
 import { projectStructuredAgentSessionMessages } from '../../../shared/structured-agent-session-message-projection'
 import { structuredAgentSessionCommandTurnIdentity } from '../../../shared/structured-agent-session-command-turn-identity'
 import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation'
+import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire'
+import {
+  EMPTY_STRUCTURED_AGENT_SESSION,
+  reduceStructuredAgentSession
+} from '../../../shared/structured-agent-session-reducer'
+import { AgentSessionSubscribers } from './structured-agent-session-subscribers'
+import { 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'
 import {
@@ -38,6 +47,7 @@ const message = (text: string): AgentJournalMessageItem => ({
   blocks: [{ type: 'text', text }],
   ...(text === '/compact' ? { command: { name: 'compact' } } : {})
 })
+const threadWrite = { fence: 0, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
 const words = (fact: 'providerRejected' | 'cancelled') =>
   agentSessionFailureWords(
     fact === 'providerRejected'
@@ -65,7 +75,9 @@ async function handOff(
   card: string,
   text: string,
   as: 'providerRejected' | 'cancelled' | 'accepted',
-  fence = 0
+  fence = 0,
+  beforeAnswer?: () => Promise<void>,
+  answeredInTurn?: AgentJournalAnsweredTurnIdentity
 ): Promise<void> {
   const body = message(text)
   const payloadFingerprint = agentSessionSendBodyFingerprint(journal.queuedMessages.sessionId, body)
@@ -83,10 +95,11 @@ async function handOff(
     fence,
     turnScope: AGENT_JOURNAL_THREAD_SCOPE
   })
+  await beforeAnswer?.()
   await journal.resolveDispatch(
     as === 'accepted'
       ? { clientMessageId: attempt, state: 'accepted', providerIdentity: null, fence }
-      : { clientMessageId: attempt, state: 'rejected', ...words(as), fence }
+      : { clientMessageId: attempt, state: 'rejected', ...words(as), fence, answeredInTurn }
   )
 }
 
@@ -298,6 +311,267 @@ it("moves an earlier build's cards into the journal, each action its own card, t
   await reopened.close()
 })
 
+it.each([100, 1_000].flatMap((size) => [false, true].map((adopt) => ({ size, adopt }))))(
+  'publication reuse: $size historical items, adopted=$adopt',
+  async ({ size, adopt }) => {
+    const corrections = size / 10
+    const earlier = journalFor(SESSION)
+    await earlier.open()
+    await handOff(earlier, 'u-0', 'q-0', 'stopped words', 'cancelled', 0, async () => {
+      for (let ordinal = 0; ordinal < size; ordinal++) {
+        await earlier.appendItem(
+          { provider: 'codex', threadId: 'history', turnId: 't', ordinal },
+          { ...message('old answer'), role: 'assistant' },
+          { fence: 0, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+        )
+      }
+    })
+    for (let index = 1; index < corrections; index++) {
+      await handOff(earlier, `u-${index}`, `q-${index}`, 'stopped words', 'cancelled')
+    }
+    await earlier.close()
+    seedLegacy(
+      SESSION,
+      Array.from({ length: corrections }, (_, i): Legacy => [`q-${i}`, 'waiting', 'stopped words'])
+    )
+    if (!adopt) {
+      db().exec(`CREATE TEMP TRIGGER b2_perf_fail_import BEFORE DELETE ON queued_messages
+      BEGIN SELECT RAISE(ABORT, 'disk full'); END`)
+    }
+    const journal = journalFor(SESSION)
+    await journal.open()
+    expect(legacyIds()).toHaveLength(adopt ? 0 : corrections)
+    const display = journal.queuedMessages.displayState()
+    const feed = new AgentSessionSubscribers()
+    const events: AgentSessionSubscribeEvent[] = []
+    let visits = 0
+    let keyVisits = 0
+    function counted<T>(iterator: MapIterator<T>, correction = false): MapIterator<T> {
+      const next = iterator.next.bind(iterator)
+      iterator.next = (...args) => {
+        const result = next(...args)
+        if (!result.done) {
+          if (correction) {
+            keyVisits++
+          } else {
+            visits++
+          }
+        }
+        return result
+      }
+      return iterator
+    }
+    const values = Map.prototype.values
+    const entries = Map.prototype[Symbol.iterator]
+    const valuesRead = vi.spyOn(Map.prototype, 'values').mockImplementation(function (
+      this: Map<unknown, unknown>
+    ) {
+      return counted(values.call(this))
+    })
+    const entriesRead = vi.spyOn(Map.prototype, Symbol.iterator).mockImplementation(function (
+      this: Map<unknown, unknown>
+    ) {
+      return counted(entries.call(this))
+    })
+    const setIterator = Set.prototype[Symbol.iterator]
+    const sets = vi.spyOn(Set.prototype, Symbol.iterator).mockImplementation(function (
+      this: Set<unknown>
+    ) {
+      return this.has('u-0') ? counted(setIterator.call(this), true) : setIterator.call(this)
+    })
+    const gets = vi.spyOn(Map.prototype, 'get')
+    const sourceGets = () =>
+      gets.mock.contexts.filter((m) => m === display.items || m === display.submissions)
+    const sorts = vi.spyOn(Array.prototype, 'sort')
+    const json = vi.spyOn(JSON, 'stringify')
+    const work = () => [
+      keyVisits,
+      sorts.mock.contexts.filter((v) => Array.isArray(v) && v[0] === 'u-0').length,
+      json.mock.calls.filter(
+        ([v]) => Array.isArray(v) && Array.isArray(v[1]) && v[1][0]?.[0] === 'u-0'
+      ).length
+    ]
+    const reset = () => {
+      keyVisits = 0
+      gets.mockClear()
+      sorts.mockClear()
+      json.mockClear()
+    }
+    try {
+      for (const id of ['perf-a', 'perf-b', 'old']) {
+        feed.open({
+          sessionId: SESSION,
+          journal,
+          fence: 0,
+          id,
+          boundedFrames: id === 'perf-b',
+          cursor: journal.cursor(),
+          snapshotCatchUp: id !== 'old',
+          readsCurrentRequests: id !== 'old',
+          emit: (event) => events.push(event)
+        })
+      }
+      expect(work()).toEqual([corrections, 1, 1])
+      expect(sourceGets()).toHaveLength(2 * corrections)
+      expect(visits).toBeGreaterThanOrEqual(size)
+      const key = journal.publicRead().normalizationKey
+      reset()
+      visits = 0
+      for (let publication = 0; publication < 5; publication++) {
+        feed.publish(SESSION, journal)
+      }
+      // Only subscriber visits; history copies/materialization would add N visits.
+      expect(visits).toBe(15)
+      expect(work()).toEqual([0, 0, 0])
+      expect(sourceGets()).toHaveLength(0)
+      const cursor = journal.cursor()
+      await journal.open()
+      expect(journal.cursor()).toEqual(cursor)
+      expect(journal.queuedMessages.displayState() === display).toBe(false)
+      feed.publish(SESSION, journal)
+      expect(work()).toEqual([corrections, 1, 1])
+      reset()
+      await journal.appendItem(
+        { provider: 'codex', threadId: 'progress', turnId: 't', ordinal: 0 },
+        message('progress'),
+        { fence: 0, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+      )
+      expect(journal.snapshot().items.at(-1)?.body).toEqual(message('progress'))
+      expect(work()).toEqual([0, 0, 0])
+      reset()
+      feed.publish(SESSION, journal)
+      const latest = events.at(-1)
+      expect(latest?.type === 'batch' && latest.batch.items.at(-1)?.body).toEqual(
+        message('progress')
+      )
+      expect(work()).toEqual([corrections, 1, 1])
+      expect(journal.publicRead().normalizationKey).toBe(key)
+    } finally {
+      for (const spy of [sets, gets, sorts, json, entriesRead, valuesRead]) {
+        spy.mockRestore()
+      }
+      if (!adopt) {
+        db().exec('DROP TRIGGER b2_perf_fail_import')
+      }
+    }
+    expect(events.filter((event) => event.type === 'snapshot')).toHaveLength(2)
+    await journal.close()
+  }
+)
+
+it.each([100, 1_000])('raw selection reuse: %i waiting cards', async (size) => {
+  const journal = journalFor(SESSION)
+  const rows = Array.from({ length: size }, (_, i): Legacy => [`waiting-${i}`, 'waiting', 'words'])
+  seedLegacy(SESSION, rows)
+  db().exec(`CREATE TEMP TRIGGER b2_raw_fail_import BEFORE DELETE ON queued_messages
+    BEGIN SELECT RAISE(ABORT, 'disk full'); END`)
+  await journal.open()
+  expect(legacyIds()).toHaveLength(size)
+  const feed = new AgentSessionSubscribers()
+  const base = { sessionId: SESSION, journal, fence: 0, cursor: journal.cursor() }
+  const received: AgentSessionSubscribeEvent[][] = [[], [], []]
+  received.forEach((events, index) =>
+    feed.open({
+      ...base,
+      id: String(index),
+      boundedFrames: index === 1,
+      snapshotCatchUp: index !== 2,
+      readsCurrentRequests: index !== 2,
+      emit: (event) => events.push(event)
+    })
+  )
+  let visits = 0
+  const flatMap = Array.prototype.flatMap
+  const scans = vi.spyOn(Array.prototype, 'flatMap')
+  scans.mockImplementation(function (this: { messageId?: string }[], callback, thisArg) {
+    const legacy = this.length === size && this[0]?.messageId === 'waiting-0'
+    return flatMap.call(this, (...args) => {
+      visits += legacy ? 1 : 0
+      return callback.apply(thisArg, args)
+    })
+  })
+  try {
+    const turn = { provider: 'codex' as const, threadId: 'progress', turnId: 't', ordinal: 0 }
+    await journal.appendItem(turn, message('progress'), threadWrite)
+    feed.publish(SESSION, journal)
+    expect(visits).toBe(2 * size)
+    expect(journal.publicRead().normalizationKey).toBe('')
+    expect(journal.queuedMessages.list().filter((c) => c.state === 'waiting')).toHaveLength(size)
+    for (const events of received) {
+      const items = events.flatMap((e) => (e.type === 'batch' ? e.batch.items : []))
+      expect(items.map((i) => i.body)).toContainEqual(message('progress'))
+      expect(events.filter((e) => e.type === 'snapshot')).toHaveLength(0)
+    }
+    visits = 0
+    for (let p = 0; p < 5; p++) {
+      feed.publish(SESSION, journal)
+    }
+    expect(visits).toBe(0)
+    const answered = { turn: { ...turn, ordinal: -1 }, via: 'steer' as const }
+    await handOff(journal, 'u', 'waiting-0', 'words', 'cancelled', 0, undefined, answered)
+    feed.publish(SESSION, journal)
+    const view = journal.publicRead()
+    expect(view.normalizationKey).not.toBe('')
+    expect(view.snapshot.submissions.find((s) => s.clientMessageId === 'u')).toMatchObject({
+      dispatchState: 'unknown',
+      queuedMessageId: 'waiting-0'
+    })
+    const item = view.snapshot.items.find((i) => i.itemId === agentJournalSubmissionKey('u'))
+    expect(item?.body).toEqual(message('words'))
+    expect(item?.sequence).toBeLessThan(journal.cursor().sequence)
+    expect(journal.submission('u')?.dispatchState).toBe('rejected')
+    expect(visits).toBeGreaterThanOrEqual(2 * size)
+    visits = 0
+    for (let p = 0; p < 5; p++) {
+      feed.publish(SESSION, journal)
+    }
+    expect(visits).toBe(0)
+    expect(received.map((es) => es.filter((e) => e.type === 'snapshot').length)).toEqual([1, 1, 0])
+  } finally {
+    scans.mockRestore()
+    db().exec('DROP TRIGGER b2_raw_fail_import')
+    await journal.close()
+  }
+})
+
+it('public metadata follows legacy-only commits, item recreation and epoch replacement', async () => {
+  const earlier = journalFor(SESSION)
+  await earlier.open()
+  await earlier.queuedMessages.insert({ messageId: 'q', body: message('words') })
+  await handOff(earlier, 'u', 'q', 'words', 'cancelled')
+  await earlier.close()
+  seedLegacy(SESSION, [['gone', 'withdrawn', 'deleted']])
+  db().exec(`CREATE TEMP TRIGGER b2_source_fail_import BEFORE DELETE ON queued_messages
+    BEGIN SELECT RAISE(ABORT, 'disk full'); END`)
+  const journal = journalFor(SESSION)
+  await journal.open()
+  const ids = vi.spyOn(journal.queuedMessages.displayState().normalizedHandOffs, Symbol.iterator)
+  const key = journal.publicRead().normalizationKey
+  expect(key).not.toBe('')
+  const cursor = journal.cursor()
+  db().exec('DROP TRIGGER b2_source_fail_import')
+  await journal.queuedMessages.importAtOpen()
+  expect(journal.cursor()).toEqual(cursor)
+  expect(legacyIds()).toEqual([])
+  journal.snapshot()
+  expect(ids).toHaveBeenCalledTimes(1)
+  expect(journal.publicRead().normalizationKey).toBe(key)
+  expect(ids).toHaveBeenCalledTimes(2)
+  ids.mockRestore()
+  const identity = { provider: 'orca' as const, clientMessageId: 'u' }
+  await journal.appendTombstone(identity, { fence: 0 })
+  expect(journal.publicRead().normalizationKey).toBe('')
+  expect(journal.snapshot().items).toEqual([])
+  await journal.appendItem(identity, message('recreated'), threadWrite)
+  expect([key, '']).not.toContain(journal.publicRead().normalizationKey)
+  expect(journal.snapshot().items[0]?.body).toEqual(message('recreated'))
+  await journal.rollEpoch('handle_forked', 0)
+  expect(journal.epoch).not.toBe(cursor.epoch)
+  expect(journal.publicRead().normalizationKey).toBe('')
+  expect(journal.snapshot().items).toEqual([])
+  await journal.close()
+})
+
 it('a rewind while the import has failed moves the legacy cards first: a refused one stays returned', async () => {
   const earlier = journalFor(SESSION)
   const clear = { command: 'clear' as const }
@@ -354,3 +628,192 @@ it('a rewind while the import has failed moves the legacy cards first: a refused
   await nothingSentTo('edited')
   expect(legacyIds()).toEqual([])
 })
+
+const retainedHistoryCases = [
+  { adopt: false, proof: 'none' },
+  { adopt: true, proof: 'none' },
+  { adopt: false, proof: 'dispatch' },
+  { adopt: true, proof: 'item' }
+]
+it.each(retainedHistoryCases)('retained history: $adopt / $proof', async ({ adopt, proof }) => {
+  const earlier = journalFor(SESSION)
+  await earlier.open()
+  const append = async (journal: AgentSessionJournal, ordinal: number) => {
+    await journal.appendItem(
+      { provider: 'codex', threadId: 'history', turnId: 't', ordinal },
+      { ...message('x'.repeat(4_000)), role: 'assistant' },
+      { fence: 0, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+    )
+  }
+  await append(earlier, 0)
+  await handOff(
+    earlier,
+    'u',
+    'q',
+    'stopped words',
+    'cancelled',
+    0,
+    async () => {
+      for (let index = 1; index <= 205; index++) {
+        await append(earlier, index)
+      }
+    },
+    proof === 'none'
+      ? undefined
+      : {
+          turn: { provider: 'codex', threadId: 'history', turnId: 't', ordinal: -1 },
+          via: 'steer'
+        }
+  )
+  const stale = readAgentSessionHistory(earlier, { sessionId: SESSION, direction: 'tail' })
+  expect(stale.ok).toBe(true)
+  await earlier.close()
+  seedLegacy(SESSION, [['q', 'waiting', 'stopped words']])
+  if (!adopt) {
+    db().exec(`CREATE TEMP TRIGGER b2_fail_import BEFORE DELETE ON queued_messages
+    BEGIN SELECT RAISE(ABORT, 'disk full'); END`)
+  }
+  const journal = journalFor(SESSION)
+  await journal.open()
+  if (!adopt) {
+    db().exec('DROP TRIGGER b2_fail_import')
+  }
+  const queue = { queuedMessages: [], queuePause: null, nextQueuedMessageId: null }
+  const feed = new AgentSessionSubscribers({ readQueuePublication: () => queue })
+  const base = { sessionId: SESSION, journal, fence: 0 }
+  const oldEvents: AgentSessionSubscribeEvent[] = []
+  feed.open({
+    ...base,
+    id: 'old',
+    cursor: journal.cursor(),
+    snapshotCatchUp: false,
+    readsCurrentRequests: false,
+    emit: (event) => oldEvents.push(event)
+  })
+  expect(oldEvents).toEqual([
+    {
+      type: 'batch',
+      sessionId: SESSION,
+      fence: 0,
+      hostNow: expect.any(Number),
+      activity: null,
+      ...queue,
+      batch: { cursor: journal.cursor(), items: [], removedItemIds: [], submissions: [] }
+    }
+  ])
+  const corrected: AgentSessionSubscribeEvent[][] = []
+  const clients: (() => ReturnType<typeof reduceStructuredAgentSession>)[] = []
+  for (const boundedFrames of [false, true]) {
+    let state = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
+      type: 'history-page',
+      page: structuredClone(stale.page)
+    })
+    const events: AgentSessionSubscribeEvent[] = []
+    corrected.push(events)
+    clients.push(() => state)
+    const reader = subscriberHistoryReader({ boundedFrames })
+    feed.open({
+      ...base,
+      id: `new-${boundedFrames}`,
+      cursor: journal.cursor(),
+      boundedFrames,
+      snapshotCatchUp: true,
+      readsCurrentRequests: true,
+      emit: (event) => {
+        events.push(event)
+        state = reduceStructuredAgentSession(state, {
+          type: 'event',
+          event: structuredClone(event)
+        })
+      }
+    })
+    expect(events[0]?.type).toBe('snapshot')
+    while (state.hasOlder) {
+      const requestedCursor = { epoch: journal.epoch, sequence: state.items[0]!.sequence }
+      const older = readAgentSessionHistory(
+        journal,
+        { sessionId: SESSION, direction: 'before', cursor: requestedCursor },
+        { reader }
+      )
+      expect(older.ok).toBe(true)
+      state = reduceStructuredAgentSession(state, {
+        type: 'older-page',
+        requestedCursor,
+        page: structuredClone(older.page)
+      })
+    }
+    const canonical = journal.snapshot()
+    expect([state.items, state.submissions]).toEqual([canonical.items, canonical.submissions])
+    feed.publish(SESSION, journal)
+    await append(journal, 205)
+    feed.publish(SESSION, journal)
+  }
+  const cursor = journal.cursor()
+  const after = readAgentSessionHistory(journal, {
+    sessionId: SESSION,
+    direction: 'after',
+    cursor
+  })
+  expect(after).toMatchObject({ ok: false, reset: 'cursor_compacted' })
+  if (!adopt) {
+    await journal.queuedMessages.importAtOpen()
+    feed.publish(SESSION, journal)
+  }
+  if (proof !== 'none') {
+    const placed = journal
+      .snapshot()
+      .items.find((item) => item.itemId === agentJournalSubmissionKey('u'))
+    const loaded = clients.map((client) => client().items.length)
+    const normalizationKey = journal.publicRead().normalizationKey
+    const identity = {
+      provider: 'codex' as const,
+      threadId: 'history',
+      turnId: 't',
+      ordinal: 206
+    }
+    if (proof === 'dispatch') {
+      await journal.acceptLate({
+        clientMessageId: 'u',
+        state: 'accepted',
+        fence: 0,
+        providerIdentity: identity
+      })
+    }
+    await journal.appendItem(identity, message('stopped words'), threadWrite)
+    feed.publish(SESSION, journal)
+    expect(journal.submission('u')?.dispatchState).toBe('accepted')
+    expect(journal.receiptFor('u')).not.toBeNull()
+    expect(journal.publicRead().normalizationKey).toBe(normalizationKey)
+    expect(clients.map((client) => client().items.length)).toEqual(loaded)
+    for (const client of clients) {
+      expect(client().items.find((item) => item.itemId === placed?.itemId)).toMatchObject({
+        sequence: placed?.sequence,
+        observedAt: placed?.observedAt,
+        turnScope: placed?.turnScope,
+        revision: 1
+      })
+      expect(
+        client().submissions.find((submission) => submission.clientMessageId === 'u')?.dispatchState
+      ).toBe('accepted')
+    }
+  }
+  expect(
+    corrected.map((events) => events.filter((event) => event.type === 'snapshot').length)
+  ).toEqual([1, 1])
+  expect(oldEvents.every((event) => event.type === 'batch')).toBe(true)
+  await journal.close()
+  const reopened = journalFor(SESSION)
+  await reopened.open()
+  const resumed: AgentSessionSubscribeEvent[] = []
+  new AgentSessionSubscribers().open({
+    ...base,
+    journal: reopened,
+    id: 'reopen',
+    cursor: reopened.cursor(),
+    snapshotCatchUp: true,
+    readsCurrentRequests: true,
+    emit: (event) => resumed.push(event)
+  })
+  expect(resumed[0]?.type).toBe('snapshot')
+  await reopened.close()
+})

src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts

@@ -1,3 +1,5 @@
+import type { AgentSessionSubscribeInput } from './agent-session-subscribe-input'
+export type { AgentSessionSubscribeInput } from './agent-session-subscribe-input'
 import type { CurrentRequestsSubscriber } from './agent-session-current-requests'
 // Advance subscriber cursors independently, sending resets or reducer-projected batches as needed.
 
@@ -31,19 +33,6 @@ import { refreshPublishedWork } from './agent-session-work-refresh'
 import type { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work'
 
 export type AgentSessionSubscriberEmit = (event: AgentSessionSubscribeEvent) => void
-export type AgentSessionSubscribeInput = {
-  id: string
-  sessionId: string
-  emit: AgentSessionSubscriberEmit
-  cursor?: AgentJournalCursor
-  /** Bound frames to the phone bridge limit. */
-  boundedFrames?: boolean
-  projectQueuePublication?: SubscriberFieldState['projectQueuePublication']
-  readsCurrentRequests?: boolean
-  /** False for desktops that need exact replay to keep send echoes and current requests. */
-  snapshotCatchUp?: boolean
-}
-
 export type Subscriber = CurrentRequestsSubscriber & {
   id: string
   sessionId: string
@@ -59,6 +48,8 @@ export type Subscriber = CurrentRequestsSubscriber & {
   backgroundTasks?: string
   /** Last frame’s current-work key; row-free changes still require a frame. */
   workView?: string
+  /** Semantic history baseline actually emitted, independent of token/row progress. */
+  historyView?: string
 }
 
 export type AgentSessionSubscribersHooks = {
@@ -118,12 +109,14 @@ export class AgentSessionSubscribers {
         this.hooks.readCurrentWork?.(input.sessionId, input.journal)
       )
     } else {
+      const view = input.journal.publicRead()
       const work = this.hooks.readCurrentWork?.(input.sessionId, input.journal)
       const page = withPublishedWork(
         readAgentSessionHydrationPage(
           input.journal,
           input.fence,
-          subscriberHistoryReader(subscriber)
+          subscriberHistoryReader(subscriber),
+          view.snapshot
         ),
         work
       )
@@ -137,7 +130,7 @@ export class AgentSessionSubscribers {
           hostNow,
           ...this.activityField(input.sessionId)
         },
-        { currentWork: work }
+        { currentWork: work, historyView: view.normalizationKey }
       )
       subscriber.cursor = page.liveCursor ?? page.window.nextCursor
       refreshPublishedWork(this.deliveryPort(), subscriber, input.journal, work, { hostNow })
@@ -194,13 +187,19 @@ export class AgentSessionSubscribers {
   }
 
   snapshot(sessionId: string, journal: AgentSessionJournal, fence: number): void {
+    const view = journal.publicRead()
     const work = this.hooks.readCurrentWork?.(sessionId, journal)
     // Cache each stored/bounded form separately; each sizes its own page.
     let storedPage: AgentSessionHistoryPage | undefined
     let boundedPage: AgentSessionHistoryPage | undefined
     const read = (subscriber: Subscriber): AgentSessionHistoryPage =>
       withPublishedWork(
-        readAgentSessionHydrationPage(journal, fence, subscriberHistoryReader(subscriber)),
+        readAgentSessionHydrationPage(
+          journal,
+          fence,
+          subscriberHistoryReader(subscriber),
+          view.snapshot
+        ),
         work
       )
     const pageFor = (subscriber: Subscriber): AgentSessionHistoryPage =>
@@ -220,7 +219,7 @@ export class AgentSessionSubscribers {
           hostNow,
           ...this.activityField(sessionId)
         },
-        { currentWork: work }
+        { currentWork: work, historyView: view.normalizationKey }
       )
       subscriber.cursor = page.liveCursor ?? page.window.nextCursor
       subscriber.fence = fence
@@ -311,6 +310,9 @@ export class AgentSessionSubscribers {
   ): void {
     try {
       emitSubscriberFrame(this.hooks, subscriber, event, options)
+      if (options?.historyView !== undefined && this.isActive(subscriber)) {
+        subscriber.historyView = options.historyView
+      }
     } catch {
       this.drop(subscriber)
     }

src/main/runtime/rpc/methods/structured-agent-session-history-capability.ts

@@ -0,0 +1,16 @@
+import type { RpcContext } from '../core'
+import { AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY } from '../../../../shared/protocol-version'
+import { clientReadsCurrentRequests } from './structured-agent-session-current-requests'
+
+/** Desktop snapshots require settled send echoes and independent current requests. */
+export function clientSupportsSnapshotCatchUp(
+  context: Pick<RpcContext, 'clientKind' | 'clientCapabilities'>
+): boolean {
+  return (
+    context.clientKind === undefined ||
+    context.clientKind === 'mobile' ||
+    (context.clientCapabilities?.includes(AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY) ===
+      true &&
+      clientReadsCurrentRequests(context))
+  )
+}

src/main/runtime/rpc/methods/structured-agent-session-wire-contracts.test.ts

@@ -81,6 +81,10 @@ it.each([
   await call('agentSession.subscribe', { sessionId: SESSION }, client)
   expect(input?.boundedFrames === true).toBe(bounded)
   expect(input?.snapshotCatchUp).toBe(snapshot)
+  const request = { sessionId: SESSION, direction: 'after', cursor: { epoch: 'e', sequence: 1 } }
+  await call('agentSession.history', request, client)
+  const replaceHistory = snapshot && input?.readsCurrentRequests === true
+  expect(hostCalls.history).toHaveBeenCalledWith(request, undefined, replaceHistory)
 })
 
 it('allows the local desktop renderer to replace a missed gap with a snapshot', async () => {

src/main/runtime/rpc/methods/structured-agent-session.ts

@@ -2,6 +2,7 @@ import {
   clientReadsCurrentRequests,
   STRUCTURED_AGENT_SESSION_CURRENT_REQUESTS_METHODS
 } from './structured-agent-session-current-requests'
+import { clientSupportsSnapshotCatchUp } from './structured-agent-session-history-capability'
 // `agentSession.*` — the structured session RPC surface.
 //
 // Every method here is gated on the client advertising
@@ -72,7 +73,6 @@ import { sendStructuredAgentSessionForClient } from './structured-agent-session-
 import { getStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
 import { structuredSessionNeverStarted } from '../../agent-session-record-founding'
 import { LOCAL_EXECUTION_HOST_ID } from '../../../../shared/execution-host'
-import { AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY } from '../../../../shared/protocol-version'
 import { ORCA_LAYOUT_OWNER_IS_MAIN } from '../../../../shared/layout-owner-switch'
 import {
   noteStructuredAgentSessionClose,
@@ -229,7 +229,14 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [
       const host = await requireInstalledHost(ctx, params.sessionId)
       return projectTurnItemHistory(
         projectQueuedImageCardHistory(
-          projectBackgroundTaskHistory(await host.history(params), ctx),
+          projectBackgroundTaskHistory(
+            await host.history(
+              params,
+              undefined,
+              clientSupportsSnapshotCatchUp(ctx) && clientReadsCurrentRequests(ctx)
+            ),
+            ctx
+          ),
           ctx
         ),
         ctx,
@@ -267,13 +274,7 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [
             )
           ),
         ...(params.cursor ? { cursor: params.cursor } : {}),
-        // Desktop snapshots require settled send echoes and independent current requests.
-        snapshotCatchUp:
-          ctx.clientKind === undefined ||
-          ctx.clientKind === 'mobile' ||
-          (ctx.clientCapabilities?.includes(AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY) ===
-            true &&
-            clientReadsCurrentRequests(ctx)),
+        snapshotCatchUp: clientSupportsSnapshotCatchUp(ctx),
         // Every phone's app, whatever its version, ends a stream over a frame its bridge cannot carry.
         ...(ctx.clientKind === 'mobile' ? { boundedFrames: true } : {})
       })