Move chat readers to the newest saved step after a no-op save (#37118)

* fix(chat): move forward readers to the covered tip after a no-op save

A write whose fold changes nothing stores no outcome record, so its
sequence never appears in a forward read. Readers placed themselves at
the last returned row and stayed behind the live cursor, so open chats
re-read an empty page and recomputed the prompt projection on every
later delivery until the next real save.

Forward journal reads now report the highest sequence they covered.
History pages move there when the read was complete, prompt reuse and
native responses compare against it, and a read cut short by its limit,
byte budget or damage still stops at the last returned row.

* fix(chat): append unlimited outcome reads without spreading, and report coverage from the mobile journal fake

* fix(chat): skip the retention scan for batches that change no journal items

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

@@ -48,11 +48,15 @@ export function hostJournalFrames(
       publicRead: () => view,
       cursor: () => snapshot.cursor,
       canonicalItemId: (itemId: string) => aliases[upTo - 1]!.get(itemId) ?? itemId,
-      readSince: (cursor: AgentJournalCursor, limit?: number) => ({
-        ok: true as const,
-        rows: rows.slice(cursor.sequence, upTo).slice(0, limit),
-        cursor: snapshot.cursor
-      })
+      readSince: (cursor: AgentJournalCursor, limit?: number) => {
+        const read = rows.slice(cursor.sequence, upTo).slice(0, limit)
+        // Like the store, a read cut by the limit covers only its last row; a complete one covers the tip.
+        const through =
+          limit !== undefined && read.length >= limit
+            ? (read.at(-1)?.seq ?? cursor.sequence)
+            : snapshot.cursor.sequence
+        return { ok: true as const, rows: read, through, cursor: snapshot.cursor }
+      }
     }
     // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the history readers call only the members above.
     return journal as unknown as AgentSessionJournal

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

@@ -3,7 +3,7 @@
 import type { AgentJournalCursor } from '../../../shared/agent-session-journal-identity'
 import type { AgentJournalResetReason } from '../../../shared/agent-session/journal/agent-session-journal-types'
 import type { JournalReadSince } from './journal-store-contracts'
-import type { JournalReadRow } from './journal-read-row'
+import type { JournalRowsRead } from './journal-read-row'
 
 export type JournalCursorRange = {
   epoch: string
@@ -68,7 +68,7 @@ export function findSequenceGap(
 export function readJournalSince(
   source: {
     state: { epoch: string; lastSequence: number; oldestSequence: number }
-    rowsAfter: (afterSequence: number) => JournalReadRow[]
+    rowsAfter: (afterSequence: number) => JournalRowsRead
   },
   cursor: AgentJournalCursor,
   currentCursor: () => AgentJournalCursor
@@ -77,9 +77,5 @@ export function readJournalSince(
   if (!resume.ok) {
     return { ok: false, reset: resume.reset }
   }
-  return {
-    ok: true,
-    rows: source.rowsAfter(resume.afterSequence),
-    cursor: currentCursor()
-  }
+  return { ok: true, ...source.rowsAfter(resume.afterSequence), cursor: currentCursor() }
 }

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

@@ -5,7 +5,11 @@ import { journalInputOrder } from './journal-input-order'
 import { journalOutcomeSource } from './journal-outcome-format'
 import { readJournalOutcomes } from './journal-outcome-table'
 import { JournalOutcomeDecodeError } from './journal-outcome-codec'
-import { readJournalOutcomeGroups, type JournalReadRow } from './journal-read-row'
+import {
+  readJournalOutcomeGroups,
+  type JournalReadRow,
+  type JournalRowsRead
+} from './journal-read-row'
 import { AgentSessionJournalError } from './journal-write-guards'
 import type Database from '../../sqlite/sync-database'
 import { journalInputAcceptance } from './journal-input-provenance'
@@ -192,7 +196,7 @@ export function journalRowsAfterReader(
   sessionId: string,
   epoch: string,
   limit?: number
-): (afterSequence: number) => JournalReadRow[] {
+): (afterSequence: number) => JournalRowsRead {
   return (afterSequence) => readJournalRowsAfterCursor(db, sessionId, epoch, afterSequence, limit)
 }
 
@@ -204,34 +208,41 @@ export function readJournalRowsAfterCursor(
   epoch: string,
   afterSequence: number,
   limit?: number
-): JournalReadRow[] {
+): JournalRowsRead {
   const source = journalOutcomeSource(db, sessionId, epoch)
   const prefix = source ? source.prefix : Number.POSITIVE_INFINITY
   const rows: JournalReadRow[] = []
+  // A read that stops early covers only the rows it returned.
+  const stopped = (): JournalRowsRead => ({ rows, through: rows.at(-1)?.seq ?? afterSequence })
   if (afterSequence < prefix) {
     for (const stored of readJournalRowsAfter(db, sessionId, epoch, afterSequence, limit)) {
       const parsed = stored.seq <= prefix ? parseJournalRow(stored.rowJson) : null
       if (!parsed?.ok) {
         // An unreadable prefix row ends the page; outcome groups never skip past it.
-        return parsed ? rows : [...rows, ...groupsAfter(prefix)]
+        return parsed ? stopped() : groupsAfter(prefix)
       }
       rows.push(parsed.row)
     }
     if (!source || (limit !== undefined && rows.length >= limit)) {
-      return rows
+      return stopped()
     }
   }
-  return [...rows, ...groupsAfter(Math.max(afterSequence, prefix))]
+  return groupsAfter(Math.max(afterSequence, prefix))
 
-  function groupsAfter(after: number): JournalReadRow[] {
-    return source
-      ? readJournalOutcomeGroups(
-          db,
-          epoch,
-          source,
-          after,
-          limit === undefined ? undefined : limit - rows.length
-        )
-      : []
+  function groupsAfter(after: number): JournalRowsRead {
+    if (!source) {
+      return stopped()
+    }
+    const remaining = limit === undefined ? undefined : limit - rows.length
+    const groups = readJournalOutcomeGroups(db, epoch, source, after, remaining)
+    // Append one by one: spreading an unlimited read into push's arguments overflows the stack.
+    for (const group of groups) {
+      rows.push(group)
+    }
+    // A write whose fold changed nothing stores no record, so no group names its sequence; a read
+    // that returned every group still covers the epoch through its newest write.
+    return remaining === undefined || groups.length < remaining
+      ? { rows, through: source.through }
+      : stopped()
   }
 }

src/main/native-chat/agent-session-journal/journal-outcome-scaling.test.ts

@@ -681,7 +681,7 @@ it('admits each row and record an existing chat reads or writes once, at N and 1
     parses.mockRestore()
     const open = { ...admission, parsed, entries: entries.records, read: watched.take().read }
     admission.rows = 0
-    const page = readJournalRowsAfterCursor(database.db, sessionId, epoch, 0, 100)
+    const { rows: page } = readJournalRowsAfterCursor(database.db, sessionId, epoch, 0, 100)
     const paged = { walks: admission.rows, rows: page.filter((row) => row.kind !== 'outcome') }
     admission.rows = 0
     admission.records = 0

src/main/native-chat/agent-session-journal/journal-outcome-wire.test.ts

@@ -1,7 +1,7 @@
 import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
-import { afterEach, expect, it } from 'vitest'
+import { afterEach, expect, it, vi } from 'vitest'
 import {
   createTrackedJournalOpener,
   openTestJournalHostDatabase
@@ -20,6 +20,10 @@ import {
   seedOutcomeRows
 } from './journal-projection-test-support'
 import type { JournalRow } from './journal-row-schema'
+import type { AgentSessionJournal } from './journal-store'
+import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session/agent-session-wire'
+import { AgentSessionSubscribers } from '../agent-session-wire/structured-agent-session-subscribers'
+import { reusePromptProjection } from '../agent-session-wire/structured-agent-session-actionable-prompts'
 
 const journals = createTrackedJournalOpener()
 let root: string
@@ -31,7 +35,17 @@ const identity = {
   providerHandle: codexProviderHandle('thread-1')
 } as const
 const item = (id: string) => ({ provider: 'orca', clientMessageId: id }) as const
+// A queue change naming a missing card folds to nothing, so its sequence stores no record.
+const writeNoOp = (journal: AgentSessionJournal) =>
+  journal['rowWriter'].enqueue((seq, ts) => ({
+    ...base(seq),
+    ts,
+    kind: 'queue',
+    messageId: 'missing',
+    change: { op: 'withdrawn' }
+  }))
 afterEach(async () => {
+  vi.restoreAllMocks()
   await journals.closeAll()
   if (root) {
     await rm(root, { recursive: true, force: true })
@@ -90,6 +104,7 @@ it('keeps sparse sequence groups whole across catch-up, lookahead, history limit
     { kind: 'status', text: 'later' },
     { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
   )
+  await writeNoOp(journal)
   const first = readAgentSessionHistory(journal, {
     sessionId,
     direction: 'after',
@@ -113,9 +128,14 @@ it('keeps sparse sequence groups whole across catch-up, lookahead, history limit
     cursor: first.page.window.nextCursor,
     limit: 1
   })
+  // Only a page that read everything moves past the trailing write that saved nothing.
   expect(second).toMatchObject({
     ok: true,
-    page: { hasNewer: false, items: [{ body: { text: 'later' } }] }
+    page: {
+      hasNewer: false,
+      items: [{ body: { text: 'later' } }],
+      window: { nextCursor: journal.cursor() }
+    }
   })
   const bounded = readAgentSessionHistory(
     journal,
@@ -124,3 +144,33 @@ it('keeps sparse sequence groups whole across catch-up, lookahead, history limit
   )
   expect(bounded).toMatchObject({ ok: false, reset: 'cursor_compacted' })
 })
+
+it('moves subscribers and prompt reuse to the newest write when that write saved nothing', async () => {
+  root = await mkdtemp(join(tmpdir(), 'orca-journal-outcomes-wire-'))
+  const database = openTestJournalHostDatabase(root)
+  seedOutcomeRows(database, [projectionEpochRow])
+  const journal = await journals.open({ identity, stateDirectory: root })
+  const subscribers = new AgentSessionSubscribers()
+  const frames: AgentSessionSubscribeEvent[] = []
+  subscribers.open({ id: 'phone', sessionId, journal, fence: 1, emit: (e) => frames.push(e) })
+  await journal.appendItem(
+    item('material'),
+    { kind: 'status', text: 'material' },
+    { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+  )
+  const material = journal.cursor()
+  await writeNoOp(journal)
+  expect(journal.cursor().sequence).toBe(material.sequence + 1)
+
+  subscribers.publish(sessionId, journal)
+  expect(frames.findLast((frame) => frame.type === 'batch')).toMatchObject({
+    batch: { cursor: journal.cursor(), items: [{ body: { text: 'material' } }] }
+  })
+  const reads = vi.spyOn(journal, 'readSince')
+  subscribers.publish(sessionId, journal)
+  expect(reads).not.toHaveBeenCalled()
+  reads.mockRestore()
+
+  const prior = { ...material, fence: 1, items: [], dependencies: new Set<string>() }
+  expect(reusePromptProjection(journal, prior, journal.cursor(), 1)).toBe(prior.items)
+})

src/main/native-chat/agent-session-journal/journal-projection-bootstrap.test.ts

@@ -95,7 +95,9 @@ it("keeps an existing chat's rows byte for byte, starts its records after one ap
 
   // Catch-up pages the prefix rows, then the record groups, each sequence once.
   const read = (after: number, limit?: number) =>
-    readJournalRowsAfterCursor(database.db, sessionId, epoch, after, limit).map((row) => row.seq)
+    readJournalRowsAfterCursor(database.db, sessionId, epoch, after, limit).rows.map(
+      (row) => row.seq
+    )
   expect(read(0)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
   expect(read(3, 4)).toEqual([4, 5, 6, 7])
   expect(read(5, 2)).toEqual([6, 7])

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

@@ -13,6 +13,8 @@ export type JournalOutcomeReadRow = JournalOutcomeKeys & {
   deleted: JournalOutcomeKeys
 }
 export type JournalReadRow = JournalRow | JournalOutcomeReadRow
+/** `through` is the highest sequence the read covered, which may store no record of its own. */
+export type JournalRowsRead = { rows: JournalReadRow[]; through: number }
 
 /** LIMIT counts logical sequences, never the independently compactable physical contributions. */
 export function readJournalOutcomeGroups(

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

@@ -18,7 +18,7 @@ import type { AgentSessionMessageSource } from '../../../shared/agent-session/ag
 import type { AgentSessionExecutionOptions } from '../../../shared/agent-session/options/agent-session-execution-options'
 import type { JournalHostDatabase } from './journal-host-database'
 import type { JournalLifecycleMutationInput } from './journal-row-builders'
-import type { JournalReadRow } from './journal-read-row'
+import type { JournalRowsRead } from './journal-read-row'
 import type { JournalReducerState } from './journal-reducer'
 import type { JournalRowWriter } from './journal-row-writer'
 import type { JournalAttachmentClaim } from './journal-submission-hook'
@@ -32,7 +32,7 @@ export type AgentSessionJournalOptions = {
 }
 
 export type JournalReadSince =
-  | { ok: true; rows: JournalReadRow[]; cursor: AgentJournalCursor }
+  | ({ ok: true; cursor: AgentJournalCursor } & JournalRowsRead)
   | { ok: false; reset: AgentJournalResetReason }
 
 export type ResolveDispatchInput = {

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

@@ -268,7 +268,10 @@ function readForward(
     // Unsplittable removal IDs require whole-state reset; resume at the live cursor past the oversized row.
     return historyReset(snapshot, 'cursor_compacted', reader)
   }
-  const lastSequence = rows.at(-1)?.seq ?? cursor.sequence
+  const hasNewer = since.rows.length > rows.length
+  // A complete read moves past trailing writes that saved nothing. `through` cannot run ahead of
+  // `snapshot`: each write commits and adopts its state in one synchronous step on the write lane.
+  const lastSequence = hasNewer ? (rows.at(-1)?.seq ?? cursor.sequence) : since.through
   return {
     ok: true,
     page: buildPage({
@@ -277,7 +280,7 @@ function readForward(
       items: items.map(({ item }) => item),
       removedItemIds: projected.batch.removedItemIds,
       hasOlder: cursor.sequence > 0,
-      hasNewer: since.rows.length > rows.length,
+      hasNewer,
       fallbackCursor: cursor,
       nextCursor: { epoch: cursor.epoch, sequence: lastSequence }
     })

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

@@ -45,7 +45,7 @@ export function reusePromptProjection(
     if (
       delta.ok &&
       delta.cursor.sequence === cursor.sequence &&
-      delta.rows.at(-1)?.seq === cursor.sequence &&
+      delta.through === cursor.sequence &&
       delta.rows.every((row, index) =>
         row.kind === 'outcome'
           ? row.submissionIds.length === 0 && !row.itemIds.some(outcomeTouches)

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

@@ -99,7 +99,7 @@ export class StructuredNativeResponses {
     }
     this.executionFence = executionFence
     this.recovery.ready()
-    this.cursor = next
+    this.cursor = changes?.ok ? { epoch: next.epoch, sequence: changes.through } : next
     return this.result()
   }
 

src/shared/agent-session/agent-session-wire.ts

@@ -83,7 +83,8 @@ export type AgentSessionHistoryPage = {
   removedItemIds: string[]
   /** Include overlapping submissions so uncertain bubbles retain their dispatch state. */
   submissions: AgentJournalSubmission[]
-  /** Return the next same-direction request cursor; keep the request cursor on empty pages. */
+  /** Return the next same-direction request cursor. An empty page keeps the request cursor, except
+   *  that an `after` page moves past newer writes that saved no history. */
   window: {
     oldest: AgentJournalCursor | null
     newest: AgentJournalCursor | null

src/shared/agent-session/structured-agent-session-reducer.ts

@@ -267,7 +267,10 @@ export function reduceStructuredAgentSession(
   const merged = journalUnchanged
     ? state.items
     : mergeItems(state.items, liveItems, event.batch.removedItemIds)
-  const items = trimRetainedItems(merged, state.retainedOwnItemLimit, state.retainedItemCap)
+  // Retained items already fit the limits, which only grow, so an unchanged journal skips the scan.
+  const items = journalUnchanged
+    ? merged
+    : trimRetainedItems(merged, state.retainedOwnItemLimit, state.retainedItemCap)
   const outsideWindow = [
     ...(liveItems.length < event.batch.items.length
       ? event.batch.items.filter((item) => !liveItems.includes(item))

src/shared/agent-session/view/structured-agent-session-item-retention.test.ts

@@ -1,4 +1,5 @@
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } from 'vitest'
+import * as producers from '../journal/agent-session-journal-producer'
 import type { AgentJournalRenderItem } from '../journal/agent-session-journal-types'
 import type { AgentSessionHistoryPage } from '../agent-session-wire'
 import {
@@ -351,6 +352,43 @@ describe('structured agent session item retention', () => {
     expect(unchanged.items).toBe(hydrated.items)
   })
 
+  it('advances the cursor of a batch with no journal change without rescanning the window', () => {
+    for (const length of [1_000, 10_000]) {
+      let paged = hydrate(range(length - 200, 200).map(item), true)
+      for (let end = length - 200; end > 0; end -= 200) {
+        paged = reduceStructuredAgentSession(paged, {
+          type: 'older-page',
+          requestedCursor: { epoch: 'epoch-a', sequence: end },
+          page: page(range(end - 200, 200).map(item), end > 200)
+        })
+      }
+      expect(paged.items).toHaveLength(length)
+      const checks = vi.spyOn(producers, 'isRootAgentJournalItem')
+      try {
+        const advanced = reduceStructuredAgentSession(paged, {
+          type: 'event',
+          event: {
+            type: 'batch',
+            sessionId: 'session-a',
+            batch: {
+              cursor: { epoch: 'epoch-a', sequence: length + 1 },
+              items: [],
+              removedItemIds: [],
+              submissions: []
+            }
+          }
+        })
+
+        expect(advanced.cursor).toEqual({ epoch: 'epoch-a', sequence: length + 1 })
+        expect(advanced.items).toBe(paged.items)
+        expect(advanced.hasOlder).toBe(paged.hasOlder)
+        expect(checks).not.toHaveBeenCalled()
+      } finally {
+        checks.mockRestore()
+      }
+    }
+  })
+
   it("counts the session's own rows, so a subagent's burst keeps the rows before it", () => {
     const streamed = streamBatches(hydrate([item(0), item(1)]), range(2, CAP + 500).map(childItem))