diff --git a/config/scripts/native-chat-compaction-pagination.test.mjs b/config/scripts/native-chat-compaction-pagination.test.mjs index 0dd58413812..2e3e8173add 100644 --- a/config/scripts/native-chat-compaction-pagination.test.mjs +++ b/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( diff --git a/mobile/src/session/mobile-native-chat-host-journal-frames.test-fixture.ts b/mobile/src/session/mobile-native-chat-host-journal-frames.test-fixture.ts index d57fb48f531..ca082a9f03b 100644 --- a/mobile/src/session/mobile-native-chat-host-journal-frames.test-fixture.ts +++ b/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[] = [] const aliases: ReadonlyMap[] = [] 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, diff --git a/src/main/native-chat/agent-session-journal/journal-public-read.ts b/src/main/native-chat/agent-session-journal/journal-public-read.ts new file mode 100644 index 00000000000..65a146e3dfc --- /dev/null +++ b/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]) +} diff --git a/src/main/native-chat/agent-session-journal/journal-queue-import.ts b/src/main/native-chat/agent-session-journal/journal-queue-import.ts index c32ec506413..9cd3a8b6197 100644 --- a/src/main/native-chat/agent-session-journal/journal-queue-import.ts +++ b/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 } diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index c23e966d890..4950017699d 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -75,6 +75,9 @@ export type JournalReducerState = { string, Pick > + /** Historical display corrections, rebuilt by the hand-off fold on replay. */ + normalizedHandOffs: Set + /** 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 } diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index 163a5d017d3..0f113b7cdec 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/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 = ( diff --git a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts b/src/main/native-chat/agent-session-journal/journal-submission-fold.ts index 25e3c2552ef..35566c1d84d 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts +++ b/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) diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-latest-turn.ts b/src/main/native-chat/agent-session-wire/agent-session-history-latest-turn.ts new file mode 100644 index 00000000000..63dfbc65ff6 --- /dev/null +++ b/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() + +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 +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-limit.ts b/src/main/native-chat/agent-session-wire/agent-session-history-limit.ts new file mode 100644 index 00000000000..ca013451e2e --- /dev/null +++ b/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))) +} 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 2ef2eaf038e..2de11b9ee8d 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 @@ -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 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() - -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 -} diff --git a/src/main/native-chat/agent-session-wire/agent-session-published-work.ts b/src/main/native-chat/agent-session-wire/agent-session-published-work.ts index ec951ce26c1..09cfb794f49 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-published-work.ts +++ b/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. */ diff --git a/src/main/native-chat/agent-session-wire/agent-session-subscribe-input.ts b/src/main/native-chat/agent-session-wire/agent-session-subscribe-input.ts new file mode 100644 index 00000000000..5f50e76ec09 --- /dev/null +++ b/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 +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts b/src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts index 54282137752..00c27060413 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts +++ b/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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts index e88a07ab873..faedfa91ba9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts +++ b/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 { 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) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts index 5932a12c58f..f7c3d17f8aa 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts +++ b/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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 2ba9748f40c..445b0df0064 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/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) => diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts index 42f2b7ea151..7b199246fd8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-legacy-import.test.ts +++ b/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, + answeredInTurn?: AgentJournalAnsweredTurnIdentity ): Promise { 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(iterator: MapIterator, correction = false): MapIterator { + 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 + ) { + return counted(values.call(this)) + }) + const entriesRead = vi.spyOn(Map.prototype, Symbol.iterator).mockImplementation(function ( + this: Map + ) { + return counted(entries.call(this)) + }) + const setIterator = Set.prototype[Symbol.iterator] + const sets = vi.spyOn(Set.prototype, Symbol.iterator).mockImplementation(function ( + this: Set + ) { + 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)[] = [] + 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() +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts index 235e5f29758..fc1a080a22d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts +++ b/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) } diff --git a/src/main/runtime/rpc/methods/structured-agent-session-history-capability.ts b/src/main/runtime/rpc/methods/structured-agent-session-history-capability.ts new file mode 100644 index 00000000000..76d2f589e5a --- /dev/null +++ b/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 +): boolean { + return ( + context.clientKind === undefined || + context.clientKind === 'mobile' || + (context.clientCapabilities?.includes(AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY) === + true && + clientReadsCurrentRequests(context)) + ) +} diff --git a/src/main/runtime/rpc/methods/structured-agent-session-wire-contracts.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-wire-contracts.test.ts index ad2341500b9..9a8202c6fd3 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-wire-contracts.test.ts +++ b/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 () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index 3e6c736f976..42bee88bbbb 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/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 } : {}) })