fix(native-chat): close round-four recovery gaps

2026-10-24 00:04:01 +00:00 · 2026-08-28 01:09:34 -07:00
parent 4d75715bf2
commit ff77fb6215
14 changed files with 514 additions and 62 deletions
@@ -96,11 +96,14 @@ const CROSS_VERSION_WIRE_PREFIXES = [
  'src/shared/agent-session-wire',
  'src/shared/agent-session-mutation-envelope',
  'src/shared/agent-session-journal-',
+  'src/main/ai-vault/structured-session-ownership.ts',
  'src/main/native-chat/agent-session-journal/',
  'src/main/native-chat/agent-session-wire/',
  'src/main/runtime/agent-session-record-store',
  'src/main/runtime/rpc/dispatcher',
+  'src/main/runtime/rpc/methods/ai-vault.ts',
  'src/main/runtime/rpc/methods/browser-tab-create-schema',
+  'src/main/runtime/rpc/methods/session-tabs.ts',
  'src/main/runtime/rpc/methods/structured-agent-session',
  'src/main/runtime/rpc/methods/terminal',
  'src/renderer/src/runtime/remote-runtime-terminal-multiplexer'
@@ -185,13 +185,16 @@ describe('per-job path classification', () => {
      'src/shared/agent-session-mutation-envelope.ts',
      'src/shared/agent-session-journal-item-key.ts',
      'src/shared/agent-session-journal-types.ts',
+      'src/main/ai-vault/structured-session-ownership.ts',
      'src/main/native-chat/agent-session-journal/journal-cursor.ts',
      'src/main/native-chat/agent-session-journal/journal-reducer.ts',
      'src/main/native-chat/agent-session-journal/journal-row-schema.ts',
      'src/main/native-chat/agent-session-wire/structured-agent-session-host.ts',
      'src/main/runtime/agent-session-record-store.ts',
      'src/main/runtime/rpc/dispatcher.ts',
+      'src/main/runtime/rpc/methods/ai-vault.ts',
      'src/main/runtime/rpc/methods/browser-tab-create-schema.ts',
+      'src/main/runtime/rpc/methods/session-tabs.ts',
      'src/main/runtime/rpc/methods/structured-agent-session.ts',
      'src/main/runtime/rpc/methods/structured-agent-session-gate.ts',
      'src/main/runtime/rpc/methods/structured-agent-session-hold.ts',
@@ -2,10 +2,12 @@
 // keeps its intact prefix live and moves the unreadable remainder aside, so the
 // bytes stay on disk for inspection instead of being rebuilt into an empty epoch.

+import { readFile } from 'node:fs/promises'
+import { join } from 'node:path'
 import {
+  JOURNAL_SNAPSHOT_FILE,
  quarantineJournalRemainder,
  readJournalLog,
-  readJournalSnapshotFile,
  rewriteJournalLog
 } from './journal-log-file'
 import type { JournalRow } from './journal-row-schema'
@@ -23,12 +25,14 @@ export async function quarantineCorruptSuffix(
 }

 /** Copy everything aside before a read-only journal is rebuilt under a newer
- *  schema: those rows are unreadable to THIS build, not worthless. */
+ *  schema: those rows are unreadable to THIS build, not worthless. The
+ *  snapshot is preserved as raw bytes — a future-version snapshot does not
+ *  parse under this build's schema, and its bytes must survive verbatim. */
 export async function quarantineUnreadableSchema(journalDir: string): Promise<void> {
-  const snapshot = await readJournalSnapshotFile(journalDir)
+  const snapshot = await readSnapshotBytes(journalDir)
  const log = await readJournalLog(journalDir)
  const preserved = [
-    snapshot ? JSON.stringify(snapshot) : '',
+    snapshot ?? '',
    log.rows.map((row) => JSON.stringify(row)).join('\n'),
    log.remainder ?? ''
  ]
@@ -39,6 +43,17 @@ export async function quarantineUnreadableSchema(journalDir: string): Promise<vo
  }
 }

+async function readSnapshotBytes(journalDir: string): Promise<string | null> {
+  try {
+    return await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8')
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
+      return null
+    }
+    throw error
+  }
+}
+
 /** The disclosure row for lines that failed to parse. Skipped lines are lost
 *  rows; counting them silently is the drop this exists to prevent. */
 export function malformedRowsDisclosure(count: number): {
@@ -194,3 +194,78 @@ describe('oversized identity bounding on Unicode boundaries', () => {
    expect(() => encodeURIComponent(bounded)).not.toThrow()
  })
 })
+
+describe('ill-formed UTF-16 identity totality', () => {
+  // JSON.parse admits lone surrogates, so any JSON string is a legal component;
+  // pre-fix these threw `URIError: URI malformed` in `encodeURIComponent`.
+  const LONE_HIGH = '\ud83d'
+  const LONE_LOW = '\ude00'
+
+  it('keys a lone-high-surrogate id without throwing, deterministically', () => {
+    const identity: AgentJournalItemIdentity = {
+      provider: 'claude',
+      sessionId: LONE_HIGH,
+      uuid: 'u-1'
+    }
+    expect(() => agentJournalItemKey(identity)).not.toThrow()
+    expect(agentJournalItemKey(identity)).toBe(agentJournalItemKey(identity))
+  })
+
+  it('keys an oversized id carrying a lone low surrogate inside the head', () => {
+    const identity: AgentJournalItemIdentity = {
+      provider: 'codex',
+      threadId: THREAD,
+      turnId: `${LONE_LOW}${'x'.repeat(1100)}`,
+      ordinal: 0
+    }
+    expect(() => agentJournalItemKey(identity)).not.toThrow()
+  })
+
+  it('keys an oversized id with a lone high surrogate away from the head cut', () => {
+    const identity: AgentJournalItemIdentity = {
+      provider: 'codex',
+      threadId: THREAD,
+      turnId: `${'a'.repeat(10)}${LONE_HIGH}${'b'.repeat(1100)}`,
+      ordinal: 0
+    }
+    expect(() => agentJournalItemKey(identity)).not.toThrow()
+  })
+
+  it('cannot collide an ill-formed id with its replacement-character spelling', () => {
+    const keyFor = (sessionId: string) =>
+      agentJournalItemKey({ provider: 'claude', sessionId, uuid: 'u-1' })
+    expect(keyFor(LONE_HIGH)).not.toBe(keyFor('�'))
+    expect(keyFor(LONE_HIGH)).not.toBe(keyFor(LONE_LOW))
+    const oversized = (head: string) => `${head}${'x'.repeat(1100)}`
+    expect(keyFor(oversized(LONE_HIGH))).not.toBe(keyFor(oversized('�')))
+  })
+
+  it('re-deriving from a parsed ill-formed key is a fixed point', () => {
+    const key = agentJournalItemKey({ provider: 'claude', sessionId: LONE_HIGH, uuid: 'u-1' })
+    const parsed = parseAgentJournalItemKey(key)
+    expect(parsed).not.toBeNull()
+    expect(agentJournalItemKey(parsed as AgentJournalItemIdentity)).toBe(key)
+  })
+
+  it('leaves well-formed short components verbatim', () => {
+    expect(boundJournalKeyComponent('turn-1')).toBe('turn-1')
+    expect(boundJournalKeyComponent('😀 café')).toBe('😀 café')
+  })
+})
+
+describe('malformed persisted keys decode to null instead of throwing', () => {
+  it('returns null for malformed percent sequences', () => {
+    for (const key of ['%', 'claude:%E0%A4%A:u-1', 'codex:a:b:1%ZZ', 'claude:%ED%A0%BD:u-1']) {
+      expect(parseAgentJournalItemKey(key)).toBeNull()
+    }
+  })
+
+  it('still round-trips well-formed keys containing the delimiter and spaces', () => {
+    const identity: AgentJournalItemIdentity = {
+      provider: 'claude',
+      sessionId: 's:1',
+      uuid: 'u 1'
+    }
+    expect(parseAgentJournalItemKey(agentJournalItemKey(identity))).toEqual(identity)
+  })
+})
@@ -234,6 +234,39 @@ describe('readJournalSnapshot validation', () => {
  })
 })

+describe('future-version snapshot classification', () => {
+  it('classifies a future version before shape validation so unknown bodies stay unreadable', async () => {
+    // The version can only advance because bodies changed, so a future snapshot
+    // legitimately carries kinds this build cannot parse. That is the
+    // schema-unreadable contract, not corruption.
+    const future = validSnapshot() as unknown as Record<string, unknown>
+    future.v = 99
+    future.items = [
+      {
+        itemId: 'codex:thread-1:turn-1:1',
+        revision: 1,
+        body: { kind: 'future-render-kind', payload: { anything: true } },
+        sequence: 2,
+        observedAt: 1_000
+      }
+    ]
+    await writeSnapshot(future)
+    expect((await readJournalSnapshot(root)).status).toBe('unreadable')
+  })
+
+  it('classifies a future version as unreadable even when its shapes still parse today', async () => {
+    await writeSnapshot({ ...validSnapshot(), v: 99 })
+    expect((await readJournalSnapshot(root)).status).toBe('unreadable')
+  })
+
+  it('treats a non-integer or sub-1 version as invalid, matching row admission', async () => {
+    for (const v of [0, 1.5]) {
+      await writeSnapshot({ ...validSnapshot(), v })
+      expect((await readJournalSnapshot(root)).status).toBe('invalid')
+    }
+  })
+})
+
 describe('journal startup isolation from a malformed snapshot', () => {
  it('quarantines a JSON-valid malformed snapshot instead of throwing through open', async () => {
    await writeSnapshot({ ...validSnapshot(), tombstones: {} })
@@ -12,9 +12,10 @@ import { appendFile, mkdir, open, readFile, type FileHandle } from 'node:fs/prom
 import { randomUUID } from 'node:crypto'
 import { join } from 'node:path'
 import { durableWriteTempPath, renameDurable, writeFileDurable } from '../../durable-file-write'
-import type {
-  AgentJournalRenderItem,
-  AgentJournalSubmission
+import {
+  AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
+  type AgentJournalRenderItem,
+  type AgentJournalSubmission
 } from '../../../shared/agent-session-journal-types'
 import {
  isAdmissibleAgentJournalRenderItem,
@@ -68,6 +69,9 @@ export type JournalSnapshotReadResult =
  | { status: 'missing' }
  | { status: 'valid'; snapshot: JournalSnapshotFile }
  | { status: 'invalid' }
+  /** A future schema version: unreadable by this build, not corrupt. The file
+   *  stays authoritative in place and the caller degrades to read-only. */
+  | { status: 'unreadable' }

 const NEWLINE_BYTE = 0x0a

@@ -75,17 +79,20 @@ export async function ensureJournalDir(journalDir: string): Promise<void> {
  await mkdir(journalDir, { recursive: true })
 }

-export async function readJournalSnapshotFile(
-  journalDir: string
-): Promise<JournalSnapshotFile | null> {
-  const result = await readJournalSnapshot(journalDir)
-  return result.status === 'valid' ? result.snapshot : null
-}
-
 export async function readJournalSnapshot(journalDir: string): Promise<JournalSnapshotReadResult> {
  try {
    const raw = await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8')
    const parsed: unknown = JSON.parse(raw)
+    const version = snapshotSchemaVersion(parsed)
+    if (version === null) {
+      return { status: 'invalid' }
+    }
+    // Version is classified BEFORE shape validation, matching row admission: a
+    // version only advances because bodies changed, so a valid newer snapshot
+    // carries kinds this build cannot parse — unreadable, never corruption.
+    if (version > AGENT_SESSION_JOURNAL_SCHEMA_VERSION) {
+      return { status: 'unreadable' }
+    }
    return isJournalSnapshotFile(parsed)
      ? { status: 'valid', snapshot: parsed }
      : { status: 'invalid' }
@@ -151,6 +158,14 @@ export async function readJournalLog(journalDir: string): Promise<JournalReadRes
  return { rows, unreadable, malformed, hasBytes: raw.length > 0 }
 }

+/** Row admission requires an integer version of at least 1; a snapshot whose
+ *  version cannot even be read is malformed, not a schema statement. */
+function snapshotSchemaVersion(value: unknown): number | null {
+  const snapshot = recordOf(value)
+  const version = snapshot?.v
+  return typeof version === 'number' && Number.isInteger(version) && version >= 1 ? version : null
+}
+
 function isJournalSnapshotFile(value: unknown): value is JournalSnapshotFile {
  if (!value || typeof value !== 'object' || Array.isArray(value)) {
    return false
@@ -14,7 +14,6 @@ import {
  readJournalSnapshot,
  type JournalSnapshotFile
 } from './journal-log-file'
-import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types'
 import {
  applyJournalRow,
  createJournalReducerState,
@@ -46,6 +45,13 @@ export async function loadJournal(
  sessionId: string
 ): Promise<JournalLoad | null> {
  const snapshotRead = await readJournalSnapshot(journalDir)
+  if (snapshotRead.status === 'unreadable') {
+    // Written by a newer schema: not corrupt, so never quarantined. The file
+    // stays authoritative in place, and this build must not write, compact,
+    // delete, or render a partial timeline from rows it cannot anchor to the
+    // snapshot it cannot read.
+    return emptyReadOnlyLoad(sessionId)
+  }
  if (snapshotRead.status === 'invalid') {
    await quarantineInvalidJournalSnapshot(journalDir)
  }
@@ -94,7 +100,9 @@ export async function loadJournal(
    state,
    tailRows,
    compactedThrough,
-    readOnly: log.unreadable || (snapshot?.v ?? 0) > AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
+    // A future-version snapshot never reaches here: it is classified
+    // unreadable above, so `valid` implies a version this build can write.
+    readOnly: log.unreadable,
    corrupt,
    malformedRows: log.malformed,
    sizeBytes: tailRows.reduce((total, row) => total + journalRowByteLength(row), 0),
@@ -407,3 +407,14 @@ describe('blob retention', () => {
    expect(referencedBlobDigests(state).size).toBe(0)
  })
 })
+
+describe('malformed persisted item keys', () => {
+  it('degrades a malformed-percent item id to an opaque key instead of throwing', () => {
+    // A user-message body drives identity resolution through the key parser;
+    // pre-fix `parseAgentJournalItemKey('%')` threw `URIError: URI malformed`.
+    const state = fold([
+      { kind: 'item', itemId: '%', revision: 1, body: userText('hi'), ...base(1) }
+    ])
+    expect(renderJournalState(state).items[0]?.itemId).toBe('%')
+  })
+})
@@ -568,6 +568,87 @@ describe('schema', () => {
    })
  })

+  it('preserves a future-version snapshot with an unknown body kind in place instead of quarantining it', async () => {
+    const journal = await open()
+    await journal.appendItem(item(0), body('a'), { fence: 1 })
+    const snapshotPath = join(root, JOURNAL_SNAPSHOT_FILE)
+    const snapshot = JSON.parse(await readFile(snapshotPath, 'utf-8')) as Record<string, unknown>
+    snapshot.v = 99
+    // The version advances because bodies changed: a valid newer snapshot
+    // carries kinds this build cannot parse and must stay unreadable in place.
+    snapshot.items = [
+      {
+        itemId: 'codex:thread-1:turn-1:1',
+        revision: 1,
+        body: { kind: 'future-render-kind', payload: { anything: true } },
+        sequence: 1,
+        observedAt: 1_000
+      }
+    ]
+    await writeFile(snapshotPath, JSON.stringify(snapshot), 'utf-8')
+
+    const reopened = await open()
+    const entries = await readdir(root)
+    expect(entries.some((name) => name.startsWith('quarantine-'))).toBe(false)
+    expect(entries.includes(JOURNAL_SNAPSHOT_FILE)).toBe(true)
+    expect(reopened.isReadOnly).toBe(true)
+    expect(reopened.snapshot().items).toHaveLength(0)
+    await expect(reopened.appendItem(item(1), body('b'), { fence: 1 })).rejects.toMatchObject({
+      code: 'journal_read_only'
+    })
+  })
+
+  it('keeps the future-version snapshot bytes when the schema escape hatch rolls the epoch', async () => {
+    const journal = await open()
+    await journal.appendItem(item(0), body('a'), { fence: 1 })
+    const snapshotPath = join(root, JOURNAL_SNAPSHOT_FILE)
+    const snapshot = JSON.parse(await readFile(snapshotPath, 'utf-8')) as Record<string, unknown>
+    snapshot.v = 99
+    snapshot.items = [
+      {
+        itemId: 'codex:thread-1:turn-1:1',
+        revision: 1,
+        body: { kind: 'future-render-kind', payload: { anything: true } },
+        sequence: 1,
+        observedAt: 1_000
+      }
+    ]
+    await writeFile(snapshotPath, JSON.stringify(snapshot), 'utf-8')
+    const reopened = await open()
+    // Still live in place before the explicit escape hatch runs.
+    expect((await readdir(root)).some((name) => name.startsWith('quarantine-'))).toBe(false)
+
+    await reopened.rollEpoch('schema_unreadable', 2)
+    expect(reopened.isReadOnly).toBe(false)
+    const quarantine = (await readdir(root)).find((name) => name.startsWith('quarantine-'))
+    expect(quarantine).toBeDefined()
+    expect(await readFile(join(root, quarantine!), 'utf-8')).toContain('future-render-kind')
+  })
+
+  it('reopens a log holding an admitted malformed-percent item id without throwing', async () => {
+    const journal = await open()
+    await journal.appendItem(item(0), body('a'), { fence: 1 })
+    const logPath = join(root, JOURNAL_LOG_FILE)
+    // `parseJournalRow` admits any string itemId, so replay must degrade a
+    // malformed percent key to an opaque id instead of throwing URIError.
+    const malformedKeyRow = JSON.stringify({
+      v: 1,
+      epoch: journal.epoch,
+      seq: journal.cursor().sequence + 1,
+      fence: 1,
+      ts: 1,
+      kind: 'item',
+      itemId: '%',
+      revision: 1,
+      body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }
+    })
+    await writeFile(logPath, `${await readFile(logPath, 'utf-8')}${malformedKeyRow}\n`, 'utf-8')
+
+    const reopened = await open()
+    expect(reopened.isReadOnly).toBe(false)
+    expect(reopened.snapshot().items.some((entry) => entry.itemId === '%')).toBe(true)
+  })
+
  it('allows the explicit schema-unreadable epoch escape hatch while preserving the old files', async () => {
    const journal = await open()
    await journal.appendItem(item(0), body('a'), { fence: 1 })
@@ -54,6 +54,8 @@ function createReadOwner(
  }
  let stopActiveRun: (() => void) | null = null
  let refreshActiveRun = (): void => {}
+  const retiredHistoryRead = (): boolean => true
+  let captureActiveHistoryReadGuard = (): (() => boolean) => retiredHistoryRead
  const activations = new Set<symbol>()
  const listeners = new Set<() => void>()

@@ -80,6 +82,11 @@ function createReadOwner(
      setSnapshot({ ...snapshot, providerSession })
    }
  }
+  const clearLoadingOlder = (): void => {
+    if (snapshot.loadingOlder) {
+      setSnapshot({ ...snapshot, loadingOlder: false })
+    }
+  }
  const refreshTail = async (shouldStop: () => boolean): Promise<void> => {
    const result = await callStructuredAgentSession<AgentSessionHistoryResult>(
      target,
@@ -91,6 +98,9 @@ function createReadOwner(
    }
    setProviderSession(result.providerSession)
    if (!result.ok) {
+      if (shouldStop()) {
+        return
+      }
      apply({
        type: 'event',
        event: {
@@ -103,7 +113,13 @@ function createReadOwner(
      })
      return
    }
+    if (shouldStop()) {
+      return
+    }
    apply({ type: 'tail-page', page: result.page })
+    if (shouldStop()) {
+      return
+    }
    let restored = snapshot.state.items.filter(countsTowardInitialHistory).length
    while (snapshot.state.hasOlder && restored < NATIVE_CHAT_INITIAL_LIMIT) {
      const oldest = oldestStructuredAgentSessionCursor(snapshot.state)
@@ -127,7 +143,13 @@ function createReadOwner(
      if (!older.ok || older.page.window.oldest?.sequence === oldest.sequence) {
        break
      }
+      if (shouldStop()) {
+        return
+      }
      apply({ type: 'older-page', requestedEpoch: oldest.epoch, page: older.page })
+      if (shouldStop()) {
+        return
+      }
      restored = snapshot.state.items.filter(countsTowardInitialHistory).length
    }
  }
@@ -140,12 +162,15 @@ function createReadOwner(
      applyEvent: (event) => apply({ type: 'event', event }),
      applyError: (message) => apply({ type: 'error', message }),
      getCursor: () => snapshot.state.cursor,
+      onHistoryReadInvalidated: clearLoadingOlder,
      refreshTail,
      sessionId,
      target
    })
+    captureActiveHistoryReadGuard = transport.captureHistoryReadGuard
    refreshActiveRun = transport.refresh
    stopActiveRun = () => {
+      captureActiveHistoryReadGuard = () => retiredHistoryRead
      refreshActiveRun = (): void => {}
      transport.dispose()
      stopActiveRun = null
@@ -180,10 +205,17 @@ function createReadOwner(
    },
    getSnapshot: () => snapshot,
    loadOlder: async () => {
+      const shouldStop = captureActiveHistoryReadGuard()
+      if (shouldStop()) {
+        return
+      }
      const cursor = oldestStructuredAgentSessionCursor(snapshot.state)
      if (!cursor || !snapshot.state.hasOlder || snapshot.loadingOlder) {
        return
      }
+      if (shouldStop()) {
+        return
+      }
      setSnapshot({ ...snapshot, loadingOlder: true })
      try {
        const result = await callStructuredAgentSession<AgentSessionHistoryResult>(
@@ -191,13 +223,20 @@ function createReadOwner(
          'agentSession.history',
          { sessionId, direction: 'before', cursor, limit: AGENT_SESSION_HISTORY_MAX_LIMIT }
        )
-        if (result.ok) {
+        if (shouldStop()) {
+          return
+        }
+        if (result.ok && !shouldStop()) {
          apply({ type: 'older-page', requestedEpoch: cursor.epoch, page: result.page })
        }
      } catch (error) {
-        apply({ type: 'error', message: String(error) })
+        if (!shouldStop()) {
+          apply({ type: 'error', message: String(error) })
+        }
      } finally {
-        setSnapshot({ ...snapshot, loadingOlder: false })
+        if (!shouldStop()) {
+          clearLoadingOlder()
+        }
      }
    },
    refresh: () => refreshActiveRun(),
@@ -69,6 +69,7 @@ describe('structured agent-session read transport generations', () => {
      applyEvent,
      applyError,
      getCursor: () => null,
+      onHistoryReadInvalidated: () => undefined,
      refreshTail: async () => undefined,
      sessionId: 'session-a',
      target
@@ -32,27 +32,54 @@ export function startStructuredAgentSessionReadTransport(args: {
  applyEvent: (event: AgentSessionSubscribeEvent) => void
  applyError: (message: string) => void
  getCursor: () => AgentJournalCursor | null
+  onHistoryReadInvalidated: () => void
  refreshTail: (shouldStop: () => boolean) => Promise<void>
  sessionId: string
  target: RuntimeClientTarget
-}): { dispose: () => void; refresh: () => void } {
+}): {
+  captureHistoryReadGuard: () => () => boolean
+  dispose: () => void
+  refresh: () => void
+} {
  let stopped = false
  let connected = false
  let opening = false
-  let generation = 0
+  let openGeneration = 0
+  let stateGeneration = 0
  let unsubscribe = (): void => {}
  let resumeCursor = args.getCursor()
-  const coalescer = createStructuredAgentSessionEventCoalescer(args.applyEvent)
+  let shouldStopCoalescedEvent = (): boolean => true
+  const coalescer = createStructuredAgentSessionEventCoalescer((event) => {
+    if (!shouldStopCoalescedEvent()) {
+      args.applyEvent(event)
+    }
+  })
  const reconnectScheduler = createReconnectScheduler({
    shouldStop: () => stopped || connected,
    reconnect: () => void open()
  })
-  const isCurrentGeneration = (candidate: number): boolean => !stopped && candidate === generation
-  const handleEvent = (event: AgentSessionSubscribeEvent, eventGeneration: number): void => {
-    if (!isCurrentGeneration(eventGeneration)) {
+  const isCurrentOpenGeneration = (candidate: number): boolean =>
+    !stopped && candidate === openGeneration
+  const captureHistoryReadGuard = (): (() => boolean) => {
+    const readOpenGeneration = openGeneration
+    const readStateGeneration = stateGeneration
+    return () =>
+      !isCurrentOpenGeneration(readOpenGeneration) || readStateGeneration !== stateGeneration
+  }
+  const handleEvent = (event: AgentSessionSubscribeEvent, eventOpenGeneration: number): void => {
+    if (!isCurrentOpenGeneration(eventOpenGeneration)) {
      return
    }
    if (event.type === 'snapshot' || event.type === 'reset') {
+      coalescer.flush()
+      if (!isCurrentOpenGeneration(eventOpenGeneration)) {
+        return
+      }
+      stateGeneration += 1
+      args.onHistoryReadInvalidated()
+      if (!isCurrentOpenGeneration(eventOpenGeneration)) {
+        return
+      }
      resumeCursor = event.page.liveCursor ?? event.page.window.nextCursor
    } else if (
      event.type === 'batch' &&
@@ -63,6 +90,7 @@ export function startStructuredAgentSessionReadTransport(args: {
      connected = false
      reconnectScheduler.schedule()
    }
+    shouldStopCoalescedEvent = captureHistoryReadGuard()
    coalescer.push(event)
  }
  async function open(): Promise<void> {
@@ -75,17 +103,25 @@ export function startStructuredAgentSessionReadTransport(args: {
    }
    opening = true
    coalescer.flush()
-    const openGeneration = ++generation
+    if (stopped) {
+      opening = false
+      return
+    }
+    const currentOpenGeneration = ++openGeneration
+    args.onHistoryReadInvalidated()
    unsubscribe()
    unsubscribe = (): void => {}
    try {
+      if (!isCurrentOpenGeneration(currentOpenGeneration)) {
+        return
+      }
      let closedDuringOpen = false
      const handle = await subscribeStructuredAgentSession(
        args.target,
        { sessionId: args.sessionId, ...(resumeCursor ? { cursor: resumeCursor } : {}) },
-        (event) => handleEvent(event, openGeneration),
+        (event) => handleEvent(event, currentOpenGeneration),
        (error) => {
-          if (!isCurrentGeneration(openGeneration)) {
+          if (!isCurrentOpenGeneration(currentOpenGeneration)) {
            return
          }
          closedDuringOpen = true
@@ -94,7 +130,7 @@ export function startStructuredAgentSessionReadTransport(args: {
          reconnectScheduler.schedule()
        },
        () => {
-          if (!isCurrentGeneration(openGeneration)) {
+          if (!isCurrentOpenGeneration(currentOpenGeneration)) {
            return
          }
          closedDuringOpen = true
@@ -102,9 +138,9 @@ export function startStructuredAgentSessionReadTransport(args: {
          reconnectScheduler.schedule()
        }
      )
-      if (!isCurrentGeneration(openGeneration) || closedDuringOpen) {
+      if (!isCurrentOpenGeneration(currentOpenGeneration) || closedDuringOpen) {
        handle.unsubscribe()
-        if (isCurrentGeneration(openGeneration)) {
+        if (isCurrentOpenGeneration(currentOpenGeneration)) {
          reconnectScheduler.schedule()
        }
      } else {
@@ -112,24 +148,24 @@ export function startStructuredAgentSessionReadTransport(args: {
        unsubscribe = handle.unsubscribe
      }
    } catch (error) {
-      if (!isCurrentGeneration(openGeneration)) {
+      if (!isCurrentOpenGeneration(currentOpenGeneration)) {
        return
      }
      connected = false
      args.applyError(String(error))
      reconnectScheduler.schedule()
    } finally {
-      if (openGeneration === generation) {
+      if (currentOpenGeneration === openGeneration) {
        opening = false
      }
    }
  }
  const refresh = (): void => {
-    const refreshGeneration = generation
+    const shouldStop = captureHistoryReadGuard()
    void args
-      .refreshTail(() => !isCurrentGeneration(refreshGeneration))
+      .refreshTail(shouldStop)
      .then(() => {
-        if (!isCurrentGeneration(refreshGeneration)) {
+        if (shouldStop()) {
          return
        }
        resumeCursor = args.getCursor()
@@ -138,31 +174,33 @@ export function startStructuredAgentSessionReadTransport(args: {
        }
      })
      .catch((error) => {
-        if (isCurrentGeneration(refreshGeneration)) {
+        if (!shouldStop()) {
          args.applyError(String(error))
        }
      })
  }
-  const initialGeneration = generation
+  const shouldStopInitialRead = captureHistoryReadGuard()
  void args
-    .refreshTail(() => !isCurrentGeneration(initialGeneration))
+    .refreshTail(shouldStopInitialRead)
    .then(() => {
-      if (!isCurrentGeneration(initialGeneration)) {
+      if (shouldStopInitialRead()) {
        return
      }
      resumeCursor = args.getCursor()
      return open()
    })
    .catch((error) => {
-      if (isCurrentGeneration(initialGeneration)) {
+      if (!shouldStopInitialRead()) {
        args.applyError(String(error))
        reconnectScheduler.schedule()
      }
    })
  return {
+    captureHistoryReadGuard,
    dispose: () => {
      stopped = true
-      generation += 1
+      openGeneration += 1
+      args.onHistoryReadInvalidated()
      reconnectScheduler.dispose()
      coalescer.dispose()
      unsubscribe()
@@ -8,7 +8,8 @@ import type {
 } from '../../../../shared/agent-session-journal-types'
 import {
  AGENT_SESSION_HISTORY_MAX_LIMIT,
-  type AgentSessionHistoryPage
+  type AgentSessionHistoryPage,
+  type AgentSessionSubscribeEvent
 } from '../../../../shared/agent-session-wire'

 const mocks = vi.hoisted(() => ({ call: vi.fn(), subscribe: vi.fn() }))
@@ -57,14 +58,15 @@ function providerFrame(id: string, sequence: number): AgentJournalRenderItem {
 function page(
  direction: 'tail' | 'before',
  items: AgentJournalRenderItem[],
-  hasOlder: boolean
+  hasOlder: boolean,
+  epoch = 'epoch-a'
 ): AgentSessionHistoryPage {
-  const cursor = (sequence: number): AgentJournalCursor => ({ epoch: 'epoch-a', sequence })
+  const cursor = (sequence: number): AgentJournalCursor => ({ epoch, sequence })
  const oldest = items[0]?.sequence ?? 0
  const newest = items.at(-1)?.sequence ?? oldest
  return {
    sessionId: 'session-a',
-    epoch: 'epoch-a',
+    epoch,
    direction,
    items,
    removedItemIds: [],
@@ -255,6 +257,82 @@ describe('useStructuredAgentSessionRead history window', () => {
    }
  })

+  it.each(['snapshot', 'reset'] as const)(
+    'drops a delayed refresh after a same-stream %s advances the epoch',
+    async (eventType) => {
+      const hasFocus = vi.spyOn(document, 'hasFocus').mockReturnValue(true)
+      const delayedRefresh = Promise.withResolvers<{
+        ok: true
+        page: AgentSessionHistoryPage
+        providerSession: { key: 'session_id'; id: string }
+      }>()
+      const onEvents: ((event: AgentSessionSubscribeEvent) => void)[] = []
+      const initialProviderSession = { key: 'session_id', id: 'provider-initial' } as const
+      mocks.call
+        .mockResolvedValueOnce({
+          ok: true,
+          page: page('tail', [message('initial', 1, 'assistant')], false),
+          providerSession: initialProviderSession
+        })
+        .mockReturnValueOnce(delayedRefresh.promise)
+      mocks.subscribe.mockImplementation((_target, _params, onEvent) => {
+        onEvents.push(onEvent)
+        return Promise.resolve({ unsubscribe: vi.fn() })
+      })
+
+      const view = renderHook(() =>
+        useStructuredAgentSessionRead({ sessionId: 'session-a', target: LOCAL_TARGET })
+      )
+
+      try {
+        await waitFor(() => expect(onEvents).toHaveLength(1))
+        act(() => window.dispatchEvent(new Event('focus')))
+        await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2))
+
+        const replacementPage = page(
+          'tail',
+          [message('new-epoch', 2, 'assistant')],
+          false,
+          'epoch-b'
+        )
+        const replacementEvent: AgentSessionSubscribeEvent =
+          eventType === 'reset'
+            ? {
+                type: 'reset',
+                sessionId: 'session-a',
+                reset: 'epoch_changed',
+                page: replacementPage,
+                fence: 2
+              }
+            : { type: 'snapshot', sessionId: 'session-a', page: replacementPage, fence: 2 }
+        act(() => onEvents[0]?.(replacementEvent))
+
+        expect(view.result.current.state.epoch).toBe('epoch-b')
+        expect(view.result.current.state.items[0]?.itemId).toBe('new-epoch')
+        expect(view.result.current.providerSession).toBe(initialProviderSession)
+        const stateAfterReplacement = view.result.current.state
+
+        await act(async () => {
+          delayedRefresh.resolve({
+            ok: true,
+            page: page('tail', [message('stale-refresh', 3, 'assistant')], false),
+            providerSession: { key: 'session_id', id: 'provider-stale' }
+          })
+          await delayedRefresh.promise
+          await Promise.resolve()
+        })
+
+        expect(view.result.current.state).toBe(stateAfterReplacement)
+        expect(view.result.current.state.epoch).toBe('epoch-b')
+        expect(view.result.current.state.items[0]?.itemId).toBe('new-epoch')
+        expect(view.result.current.providerSession).toBe(initialProviderSession)
+      } finally {
+        view.unmount()
+        hasFocus.mockRestore()
+      }
+    }
+  )
+
  it('does no host work for retained inactive sessions', async () => {
    const first = renderHook(() =>
      useStructuredAgentSessionRead({
@@ -14,29 +14,72 @@ const KEY_DELIMITER = ':'
 export const MAX_JOURNAL_KEY_COMPONENT_CHARS = 1024

 /**
- * Deterministic stand-in for an oversized key component: same input, same
- * output, so revisions and tombstones of one identity still share a key, and
- * re-deriving from a parsed key is a fixed point (the bounded form is far
- * below the cap). The head keeps keys debuggable; length plus two independent
- * hashes makes an accidental collision practically impossible. Pure JS because
- * clients derive keys too and cannot reach node:crypto.
+ * Deterministic stand-in for an oversized or ill-formed key component: same
+ * input, same output, so revisions and tombstones of one identity still share
+ * a key, and re-deriving from a parsed key is a fixed point (the bounded form
+ * is well-formed and far below the cap). The head keeps keys debuggable;
+ * length plus two independent hashes makes an accidental collision practically
+ * impossible. Pure JS because clients derive keys too and cannot reach
+ * node:crypto.
+ *
+ * JSON strings are arbitrary UTF-16 code units, so a component can carry a
+ * lone surrogate that `encodeURIComponent` throws on. Those values take the
+ * digest form too: the hashes run over the raw code units, so a value and its
+ * replacement-character spelling keep distinct keys.
 */
 export function boundJournalKeyComponent(value: string): string {
-  if (value.length <= MAX_JOURNAL_KEY_COMPONENT_CHARS) {
+  if (value.length <= MAX_JOURNAL_KEY_COMPONENT_CHARS && !hasLoneSurrogate(value)) {
    return value
  }
  const h1 = fnv1a32(value, 0x811c9dc5).toString(16).padStart(8, '0')
  const h2 = fnv1a32(value, 0x0100_0193).toString(16).padStart(8, '0')
-  return `${codePointBoundedHead(value, 40)}~orca-oversized~${value.length}~${h1}${h2}`
+  return `${wellFormedBoundedHead(value, 40)}~orca-oversized~${value.length}~${h1}${h2}`
 }

-/** `slice` cuts UTF-16 code units, so a head ending mid-astral-character would
- *  carry a lone high surrogate and make `encodeURIComponent` throw on a valid
- *  id. Dropping that unit keeps the head valid Unicode and deterministic. */
-function codePointBoundedHead(value: string, maxUnits: number): string {
-  const head = value.slice(0, maxUnits)
-  const last = head.charCodeAt(head.length - 1)
-  return last >= 0xd800 && last <= 0xdbff ? head.slice(0, -1) : head
+/** The diagnostic head must be valid Unicode for `encodeURIComponent`: a pair
+ *  split by the cut is dropped and a lone surrogate becomes U+FFFD — the
+ *  hashes over the raw units keep the full key collision-safe regardless.
+ *  Digest forms are persisted, so for well-formed input the head must stay
+ *  byte-stable across builds or one identity would stop sharing a key. */
+function wellFormedBoundedHead(value: string, maxUnits: number): string {
+  let head = ''
+  let index = 0
+  while (index < value.length && index < maxUnits) {
+    const unit = value.charCodeAt(index)
+    if (unit >= 0xd800 && unit <= 0xdbff) {
+      const next = index + 1 < value.length ? value.charCodeAt(index + 1) : 0
+      if (next >= 0xdc00 && next <= 0xdfff) {
+        if (index + 1 >= maxUnits) {
+          break
+        }
+        head += value.charAt(index) + value.charAt(index + 1)
+        index += 2
+        continue
+      }
+      head += '�'
+      index += 1
+      continue
+    }
+    head += unit >= 0xdc00 && unit <= 0xdfff ? '�' : value.charAt(index)
+    index += 1
+  }
+  return head
+}
+
+function hasLoneSurrogate(value: string): boolean {
+  for (let index = 0; index < value.length; index += 1) {
+    const unit = value.charCodeAt(index)
+    if (unit >= 0xd800 && unit <= 0xdbff) {
+      const next = index + 1 < value.length ? value.charCodeAt(index + 1) : 0
+      if (next < 0xdc00 || next > 0xdfff) {
+        return true
+      }
+      index += 1
+    } else if (unit >= 0xdc00 && unit <= 0xdfff) {
+      return true
+    }
+  }
+  return false
 }

 function fnv1a32(value: string, seed: number): number {
@@ -97,7 +140,16 @@ export function agentJournalSubmissionKey(clientMessageId: string): string {
 * `encodeURIComponent` escapes the delimiter, so the split is unambiguous.
 */
 export function parseAgentJournalItemKey(key: string): AgentJournalItemIdentity | null {
-  const parts = key.split(KEY_DELIMITER).map((part) => decodeURIComponent(part))
+  // Persisted keys can be corrupted: a malformed percent sequence must fail
+  // the parse, never throw through journal replay or open.
+  const parts: string[] = []
+  for (const part of key.split(KEY_DELIMITER)) {
+    try {
+      parts.push(decodeURIComponent(part))
+    } catch {
+      return null
+    }
+  }
  const [provider, ...rest] = parts
  if (provider === 'codex' && rest.length === 3) {
    const ordinal = Number(rest[2])