diff --git a/src/main/ipc/desktop-renderer-runtime-capabilities.ts b/src/main/ipc/desktop-renderer-runtime-capabilities.ts index d44fda04208a..233597a4e003 100644 --- a/src/main/ipc/desktop-renderer-runtime-capabilities.ts +++ b/src/main/ipc/desktop-renderer-runtime-capabilities.ts @@ -3,6 +3,7 @@ import { AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY, AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY, AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, AGENT_SESSION_TURN_ITEM_CAPABILITY, CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY, PI_STRUCTURED_DIALOGS_RUNTIME_CAPABILITY, @@ -39,6 +40,8 @@ export const DESKTOP_RENDERER_RUNTIME_CLIENT_CAPABILITIES: readonly RuntimeCapab AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY, AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY, AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, + // The renderer settles acknowledged echoes from the read cursor even when a snapshot skips them. + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, AGENT_SESSION_TURN_ITEM_CAPABILITY, AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY, AGENT_SESSION_BACKGROUND_TASK_CHILD_VIEWS_CAPABILITY, diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts b/src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts index 7cce60a3bddc..9a2454080f5e 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts @@ -8,7 +8,6 @@ import { type AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' -import * as journalReducer from '../agent-session-journal/journal-reducer' import * as rowSchema from '../agent-session-journal/journal-row-schema' import { createTrackedJournalOpener, @@ -101,8 +100,7 @@ function observeForwardReads() { } describe('forward history SQL read budget', () => { - it('reconnects through every page with one lookahead row per page', async () => { - const count = 2_000 + it.each([200, 2_000, 20_000])('bounds reconnect reads after %i missed rows', async (count) => { const journal = await seedJournal(count) const { returnedRows, parse } = observeForwardReads() const events: AgentSessionSubscribeEvent[] = [] @@ -114,35 +112,16 @@ describe('forward history SQL read budget', () => { cursor: { epoch: journal.epoch, sequence: 1 }, emit: (event) => events.push(event) }) - const batches = events.filter((event) => event.type === 'batch') - expect(batches.flatMap((event) => event.batch.items.map((item) => item.itemId))).toEqual( - Array.from({ length: count }, (_, index) => `item-${index}`) + expect(events).toHaveLength(1) + const event = events[0]! + expect(event.type).toBe(count > 200 ? 'snapshot' : 'batch') + const items = + event.type === 'snapshot' ? event.page.items : event.type === 'batch' ? event.batch.items : [] + expect(items.map((item) => item.itemId)).toEqual( + Array.from({ length: 200 }, (_, index) => `item-${count - 200 + index}`) ) - expect(batches.at(-1)?.batch.cursor).toEqual(journal.cursor()) - // Only the last page says the reader is caught up. - expect(batches.map((event) => event.hasNewer === true)).toEqual([ - ...Array(9).fill(true), - false - ]) - expect(returnedRows).toEqual([...Array(9).fill(201), 200]) - expect(parse).toHaveBeenCalledTimes(2_009) - }) - - it('reduces the timeline once for the whole catch-up, not once per page', async () => { - const journal = await seedJournal(2_000) - const render = vi.spyOn(journalReducer, 'renderJournalState') - const events: AgentSessionSubscribeEvent[] = [] - new AgentSessionSubscribers().open({ - id: 'reader', - sessionId: identity.sessionId, - journal, - fence: 1, - cursor: { epoch: journal.epoch, sequence: 1 }, - emit: (event) => events.push(event) - }) - // Catch-up is synchronous, so the reduced timeline cannot change between pages. - expect(events.filter((event) => event.type === 'batch')).toHaveLength(10) - expect(render).toHaveBeenCalledTimes(1) + expect(returnedRows).toEqual([Math.min(count, 201)]) + expect(parse).toHaveBeenCalledTimes(Math.min(count, 201)) }) it('keeps an exact final page final and preserves unlimited journal readers', async () => { 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 7c546de29273..282ccc6d10ba 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 @@ -103,9 +103,8 @@ export function deliverToSubscriber( emitCaughtUp(port, subscriber, emitCheckpoint, shared, { journal, work }) return } - // A phone's app ends a stream that outruns its unacked window, so a gap of more than one page - // arrives as the newest page instead; older rows load on scroll. - if (page.hasNewer && subscriber.boundedFrames) { + // Replaying an unbounded gap blocks the host; older rows remain available on scroll. + if (page.hasNewer && subscriber.snapshotCatchUp !== false) { const tail = withPublishedWork( readAgentSessionHydrationPage(journal, subscriber.fence, reader), work 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 559d53640dcd..39080bbf6dab 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 @@ -36,6 +36,8 @@ export type AgentSessionSubscribeInput = { cursor?: AgentJournalCursor /** Every frame to fit a phone's bridge frame; the RPC sets it for every phone. */ boundedFrames?: boolean + /** False for older desktops whose send echoes require exact row replay. */ + snapshotCatchUp?: boolean } export type Subscriber = { @@ -45,6 +47,7 @@ export type Subscriber = { cursor: AgentJournalCursor fence: number boundedFrames: boolean + snapshotCatchUp?: boolean commands?: AgentSessionSlashCommand[] | null /** The last draft list actually SENT — never advanced on a page that withheld * it, or the final replacement would be suppressed by the identity dedup. */ @@ -95,6 +98,7 @@ export class AgentSessionSubscribers { emit: AgentSessionSubscriberEmit cursor?: AgentJournalCursor boundedFrames?: boolean + snapshotCatchUp?: boolean }): () => void { const liveCursor = input.journal.cursor() const subscriber: Subscriber = { @@ -103,7 +107,8 @@ export class AgentSessionSubscribers { emit: input.emit, cursor: input.cursor ?? { epoch: liveCursor.epoch, sequence: 0 }, fence: input.fence, - boundedFrames: input.boundedFrames === true + boundedFrames: input.boundedFrames === true, + snapshotCatchUp: input.snapshotCatchUp !== false } const session = this.bySession.get(input.sessionId) ?? new Map() session.set(input.id, subscriber) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-wire-admission.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-wire-admission.test.ts index d70228762bf4..e537f24db49a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-wire-admission.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-wire-admission.test.ts @@ -121,7 +121,7 @@ describe('structured agent-session outbound admission', () => { dispose() }) - it('splits valid-cursor catch-up into admitted batch frames', () => { + it('replaces an over-budget catch-up with one admitted newest page', () => { const subscribers = new AgentSessionSubscribers() const catchup: AgentSessionSubscribeEvent[] = [] const firstItemSequence = journal.snapshot().items[0]!.sequence @@ -135,14 +135,11 @@ describe('structured agent-session outbound admission', () => { emit: (event) => catchup.push(event) }) - expect(catchup.length).toBeGreaterThan(1) + expect(catchup).toHaveLength(1) catchup.forEach(expectAdmitted) - expect( - catchup.flatMap((event) => (event.type === 'batch' ? event.batch.items : [])) - ).toHaveLength(19) - expect(catchup.at(-1)).toMatchObject({ - type: 'batch', - batch: { cursor: journal.cursor() }, + expect(catchup[0]).toMatchObject({ + type: 'snapshot', + page: { hasOlder: true, hasNewer: false, liveCursor: journal.cursor() }, fence: 2 }) }) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts index 0d96a3861acd..2e0ab3b05915 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts @@ -3,6 +3,8 @@ import { afterEach, beforeEach, expect, it } from 'vitest' import type { AgentSessionSubscribeInput } from '../../../native-chat/agent-session-wire/structured-agent-session-subscribers' +import { AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY } from '../../../../shared/protocol-version' +import { DESKTOP_RENDERER_RUNTIME_CLIENT_CAPABILITIES } from '../../../ipc/desktop-renderer-runtime-capabilities' import { call, clearStructuredHostStub, @@ -16,12 +18,21 @@ import { beforeEach(installStructuredHostStub) afterEach(clearStructuredHostStub) +const currentDesktop = { + ...STRUCTURED_CLIENT, + clientCapabilities: [ + ...STRUCTURED_CLIENT.clientCapabilities, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY + ] +} + it.each([ // Being a phone is enough: no capability or param is asked of it. - ['a phone of any version', STRUCTURED_MOBILE_CLIENT, true], - ['a paired desktop', STRUCTURED_CLIENT, false], - ['an in-process reader', undefined, false] -] as const)('subscribes %s with bounded frames: %s', async (_label, client, bounded) => { + ['a phone of any version', STRUCTURED_MOBILE_CLIENT, true, true], + ['an older paired desktop', STRUCTURED_CLIENT, false, false], + ['an updated paired desktop', currentDesktop, false, true], + ['an in-process reader', undefined, false, true] +] as const)('subscribes %s with bounded frames: %s', async (_label, client, bounded, snapshot) => { let input: AgentSessionSubscribeInput | undefined hostCalls.subscribe.mockImplementation((opened: AgentSessionSubscribeInput) => { input = opened @@ -30,4 +41,22 @@ it.each([ }) await call('agentSession.subscribe', { sessionId: SESSION }, client) expect(input?.boundedFrames === true).toBe(bounded) + expect(input?.snapshotCatchUp).toBe(snapshot) +}) + +it('allows the local desktop renderer to replace a missed gap with a snapshot', async () => { + let input: AgentSessionSubscribeInput | undefined + hostCalls.subscribe.mockImplementation((opened: AgentSessionSubscribeInput) => { + input = opened + opened.emit({ type: 'end' }) + return () => {} + }) + const cursor = { epoch: 'epoch-a', sequence: 10 } + await call('agentSession.subscribe', { sessionId: SESSION, cursor }, { + clientId: 'desktop-renderer', + clientKind: 'runtime', + clientCapabilities: [...DESKTOP_RENDERER_RUNTIME_CLIENT_CAPABILITIES] + }) + expect(input?.cursor).toEqual(cursor) + expect(input?.snapshotCatchUp).toBe(true) }) diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index a25d40b52b28..3af0d34d9262 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -58,6 +58,7 @@ import { UnsubscribeParams } from './structured-agent-session-schemas' import { sendStructuredAgentSessionForClient } from './structured-agent-session-send-compatibility' +import { AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY } from '../../../../shared/protocol-version' export const STRUCTURED_AGENT_SESSION_METHODS = [ defineMethod({ @@ -182,6 +183,11 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [ ) ), ...(params.cursor ? { cursor: params.cursor } : {}), + // Older desktops still need each missed submission to settle their local echoes. + snapshotCatchUp: + ctx.clientKind === undefined || + ctx.clientKind === 'mobile' || + ctx.clientCapabilities?.includes(AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY) === true, // Every phone's app, whatever its version, ends a stream over a frame its bridge cannot carry. ...(ctx.clientKind === 'mobile' ? { boundedFrames: true } : {}) }) diff --git a/src/renderer/src/web/web-runtime-client-capabilities.ts b/src/renderer/src/web/web-runtime-client-capabilities.ts index f7f8b9f53da4..378ca00ce8b4 100644 --- a/src/renderer/src/web/web-runtime-client-capabilities.ts +++ b/src/renderer/src/web/web-runtime-client-capabilities.ts @@ -3,6 +3,7 @@ import { AGENT_SESSION_BOUNDARY_RUNTIME_CAPABILITY, REPO_SEARCH_QUALIFIED_REFS_RUNTIME_CAPABILITY, AGENT_SESSION_TURN_ITEM_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY, SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY, SESSION_TABS_SPLIT_GROUP_PLACEMENT_RUNTIME_CAPABILITY, @@ -18,6 +19,7 @@ export const WEB_RUNTIME_CLIENT_CAPABILITIES = [ AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY, AGENT_SESSION_BACKGROUND_TASK_CHILD_VIEWS_CAPABILITY, AGENT_SESSION_TURN_ITEM_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY, SESSION_TABS_SPLIT_GROUP_PLACEMENT_RUNTIME_CAPABILITY, SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY, diff --git a/src/renderer/src/web/web-runtime-client.test.ts b/src/renderer/src/web/web-runtime-client.test.ts index fa1eb7301bf3..b4d1c0164978 100644 --- a/src/renderer/src/web/web-runtime-client.test.ts +++ b/src/renderer/src/web/web-runtime-client.test.ts @@ -17,6 +17,7 @@ import type { RuntimeRpcResponse } from '../../../shared/runtime-rpc-envelope' import { AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY, AGENT_SESSION_TURN_ITEM_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, AGENT_SESSION_BOUNDARY_RUNTIME_CAPABILITY, REPO_SEARCH_QUALIFIED_REFS_RUNTIME_CAPABILITY, SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY, @@ -128,6 +129,7 @@ describe('WebRuntimeClient', () => { AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY, AGENT_SESSION_BACKGROUND_TASK_CHILD_VIEWS_CAPABILITY, AGENT_SESSION_TURN_ITEM_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY, SESSION_TABS_SPLIT_GROUP_PLACEMENT_RUNTIME_CAPABILITY, SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY, diff --git a/src/shared/electron-remote-runtime-client-capabilities.ts b/src/shared/electron-remote-runtime-client-capabilities.ts index c6f3882e4eda..379c286509a3 100644 --- a/src/shared/electron-remote-runtime-client-capabilities.ts +++ b/src/shared/electron-remote-runtime-client-capabilities.ts @@ -1,6 +1,7 @@ import { AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, BROWSER_CLIENT_HOST_RUNTIME_CAPABILITY, BROWSER_CLIENT_PAGE_METADATA_RUNTIME_CAPABILITY, CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY, @@ -22,6 +23,7 @@ export const ELECTRON_REMOTE_RUNTIME_CLIENT_CAPABILITIES = [ AGENT_SESSION_CREATE_MESSAGE_RUNTIME_CAPABILITY, AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY, AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, // The renderer that reads a paired host's structured chats is the one that reads its own. STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY, CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY, diff --git a/src/shared/runtime-capability-constants.ts b/src/shared/runtime-capability-constants.ts index 3f540b33f1d9..2a3dfbbb560f 100644 --- a/src/shared/runtime-capability-constants.ts +++ b/src/shared/runtime-capability-constants.ts @@ -183,6 +183,9 @@ export const AGENT_SESSION_REWIND_RECOVERY_CAPABILITY = 'agent-session.rewind-re // status form to clients that do not advertise this. Transitional: drop the downgrade once no // supported release lacks the capability. export const AGENT_SESSION_TURN_ITEM_CAPABILITY = 'agent-session.turn-item.v1' as const +/** The client retires acknowledged echoes even when a catch-up snapshot skips their rows. */ +export const AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY = + 'agent-session.snapshot-catch-up.v1' as const export const AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY = 'agent-session.background-task-stop.v1' as const // Why: agentSession.respondToQuestion has a strict schema, so clients must not send structured diff --git a/tests/e2e/cross-version-wire/cross-version-stop-note-scenarios.ts b/tests/e2e/cross-version-wire/cross-version-stop-note-scenarios.ts index 8d3dc4f4d436..d99b91336c24 100644 --- a/tests/e2e/cross-version-wire/cross-version-stop-note-scenarios.ts +++ b/tests/e2e/cross-version-wire/cross-version-stop-note-scenarios.ts @@ -111,7 +111,7 @@ export function describeReleasedStopNoteProjection(port: ScenarioPort): void { expect(render(state)).toEqual({ type: 'text', text: ordinary.text }) expect(rig.journal.cursor()).toEqual(held) }) - it('catches up through byte-bounded pages without losing the dependent note or cursor', async () => { + it('replays to a released desktop and pages a fresh snapshot back to the dependent note', async () => { const before = await seed() const held = rig.journal.cursor() for (const clientMessageId of ['large-a', 'large-b']) { @@ -134,10 +134,22 @@ export function describeReleasedStopNoteProjection(port: ScenarioPort): void { const first = await page({ direction: 'after', cursor: held, limit: 3 }) expect(first.hasNewer).toBe(true) const caughtUp = await subscribe(held) - const after = reduce(before, caughtUp) + expect(events(caughtUp).every((event) => event.type === 'batch')).toBe(true) + expect(reduce(before, caughtUp).cursor).toEqual(rig.journal.cursor()) + const fresh = await subscribe(null) + let after = reduce(before, fresh) + expect(events(fresh)[0]).toMatchObject({ type: 'snapshot', page: { hasOlder: true } }) + while (after.hasOlder && !after.items.some((item) => item.itemId === noteId)) { + const requestedCursor = { epoch: after.epoch!, sequence: after.items[0]!.sequence } + const older = await page({ direction: 'before', cursor: requestedCursor }) + after = rig.client.reduceStructuredAgentSession(after, { + type: 'older-page', + requestedCursor, + page: older + }) + } expect(render(after)).toEqual({ type: 'text', text: ordinary.text }) expect(after.cursor).toEqual(rig.journal.cursor()) - expect(events(caughtUp).filter((event) => event.type === 'batch').length).toBeGreaterThan(1) }) it.each(['running', 'completed', 'unverifiable', 'missing', 'other-failure'] as const)( 'keeps %s wording through the released page reader', diff --git a/tests/e2e/cross-version-wire/cross-version-stop-note-test-rig.ts b/tests/e2e/cross-version-wire/cross-version-stop-note-test-rig.ts index d14aee3b5996..2c3523f70272 100644 --- a/tests/e2e/cross-version-wire/cross-version-stop-note-test-rig.ts +++ b/tests/e2e/cross-version-wire/cross-version-stop-note-test-rig.ts @@ -14,6 +14,7 @@ import { agentSessionFailureFact } from '../../../src/shared/agent-session-failu import { agentSessionFailureWords } from '../../../src/shared/agent-session-failure-words' import { AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalCursor, type AgentJournalItemBody, type AgentJournalTurnLifecycle } from '../../../src/shared/agent-session-journal-types' @@ -138,8 +139,8 @@ export function createReleasedStopNoteRig(port: ScenarioPort, ref: string) { items: result.page.items.map((item) => client.AgentJournalRenderItemSchema.parse(item)) } } - async function subscribe(cursor = journal.cursor()): Promise { - return call('agentSession.subscribe', { sessionId: SESSION, cursor }) + async function subscribe(cursor: AgentJournalCursor | null = journal.cursor()): Promise { + return call('agentSession.subscribe', { sessionId: SESSION, ...(cursor ? { cursor } : {}) }) } function events(replies: RpcReply[]): AgentSessionSubscribeEvent[] { return replies.map((reply) => { diff --git a/tests/e2e/structured-agent-session-read-owner.unit.test.ts b/tests/e2e/structured-agent-session-read-owner.unit.test.ts index 0220a33e59a7..26b75cd65604 100644 --- a/tests/e2e/structured-agent-session-read-owner.unit.test.ts +++ b/tests/e2e/structured-agent-session-read-owner.unit.test.ts @@ -122,7 +122,7 @@ describe('structured session cursor/body regression', () => { }) it.each([40, 401])( - 'replays an off-page dispatch after %i missed rows without stranding pending state', + 'catches up an off-page dispatch after %i missed rows without stranding pending state', async (missedRows) => { const { journal, appendOutput, accept } = await fixture() let hostSummary: AgentSessionStatusSummary | undefined @@ -234,12 +234,18 @@ describe('structured session cursor/body regression', () => { expect( projectStructuredAgentSessionStatus(settled.items, settled.submissions, settled.fence) ).toBe(hostSummary?.status) - expect(settled.submissions).toEqual(journal.snapshot().submissions) + expect(settled.submissions).toEqual( + missedRows > 200 ? tail.submissions : journal.snapshot().submissions + ) expect({ cursor: caughtUp.cursor?.sequence, dispatch: caughtUp.submissions[0]?.dispatchState, unansweredDispatch: hasUnansweredStructuredAgentSessionDispatch(caughtUp.submissions, 1) - }).toEqual({ cursor: 100 + missedRows, dispatch: 'accepted', unansweredDispatch: false }) + }).toEqual({ + cursor: 100 + missedRows, + dispatch: missedRows > 200 ? undefined : 'accepted', + unansweredDispatch: false + }) expect(mocks.call).toHaveBeenCalledTimes(1) expect(mocks.subscribe).toHaveBeenCalledTimes(2) expect(mocks.subscribe.mock.calls[1]?.[1]).toEqual({