diff --git a/mobile/src/session/native-chat/mobile-native-chat-host-journal-frames.test-fixture.ts b/mobile/src/session/native-chat/mobile-native-chat-host-journal-frames.test-fixture.ts index 2f6baba4c14..ea9a0de185c 100644 --- a/mobile/src/session/native-chat/mobile-native-chat-host-journal-frames.test-fixture.ts +++ b/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 diff --git a/src/main/native-chat/agent-session-journal/journal-cursor.ts b/src/main/native-chat/agent-session-journal/journal-cursor.ts index 4463f66d26e..118ea714a2d 100644 --- a/src/main/native-chat/agent-session-journal/journal-cursor.ts +++ b/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() } } diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index 82138bda6c9..6841163a429 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/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() } } diff --git a/src/main/native-chat/agent-session-journal/journal-outcome-scaling.test.ts b/src/main/native-chat/agent-session-journal/journal-outcome-scaling.test.ts index 98f4b991fed..2b0cd06ffcf 100644 --- a/src/main/native-chat/agent-session-journal/journal-outcome-scaling.test.ts +++ b/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 diff --git a/src/main/native-chat/agent-session-journal/journal-outcome-wire.test.ts b/src/main/native-chat/agent-session-journal/journal-outcome-wire.test.ts index 200b8db2b4e..5b1fc9a367b 100644 --- a/src/main/native-chat/agent-session-journal/journal-outcome-wire.test.ts +++ b/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() } + expect(reusePromptProjection(journal, prior, journal.cursor(), 1)).toBe(prior.items) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-projection-bootstrap.test.ts b/src/main/native-chat/agent-session-journal/journal-projection-bootstrap.test.ts index 292e5c4b2a3..b464b8a9f29 100644 --- a/src/main/native-chat/agent-session-journal/journal-projection-bootstrap.test.ts +++ b/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]) diff --git a/src/main/native-chat/agent-session-journal/journal-read-row.ts b/src/main/native-chat/agent-session-journal/journal-read-row.ts index cd40045658a..b3f90aa9c8b 100644 --- a/src/main/native-chat/agent-session-journal/journal-read-row.ts +++ b/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( diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index f580428f95f..a18dc00e952 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/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 = { diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts index 6b0ba723542..b9d7e663a18 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts @@ -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 } }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-actionable-prompts.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-actionable-prompts.ts index cc0a0cc769f..1b75875e7bd 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-actionable-prompts.ts +++ b/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) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-native-responses.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-native-responses.ts index 66c23d610bc..b3984397259 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-native-responses.ts +++ b/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() } diff --git a/src/shared/agent-session/agent-session-wire.ts b/src/shared/agent-session/agent-session-wire.ts index 716067afe24..553d9d41ea5 100644 --- a/src/shared/agent-session/agent-session-wire.ts +++ b/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 diff --git a/src/shared/agent-session/structured-agent-session-reducer.ts b/src/shared/agent-session/structured-agent-session-reducer.ts index 6b93a091e20..460fff35569 100644 --- a/src/shared/agent-session/structured-agent-session-reducer.ts +++ b/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)) diff --git a/src/shared/agent-session/view/structured-agent-session-item-retention.test.ts b/src/shared/agent-session/view/structured-agent-session-item-retention.test.ts index b076ad718fd..2a1f6a0f50c 100644 --- a/src/shared/agent-session/view/structured-agent-session-item-retention.test.ts +++ b/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))