diff --git a/src/main/ipc/desktop-renderer-runtime-capabilities.ts b/src/main/ipc/desktop-renderer-runtime-capabilities.ts index b15d2060c0c..a1df52b5e26 100644 --- a/src/main/ipc/desktop-renderer-runtime-capabilities.ts +++ b/src/main/ipc/desktop-renderer-runtime-capabilities.ts @@ -4,6 +4,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, @@ -42,6 +43,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 7cce60a3bdd..9a2454080f5 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 2915f480acd..0f8fde1769b 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 @@ -116,9 +116,8 @@ export function deliverToSubscriber( }) 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 f447ff90da8..19dbfacf462 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 @@ -39,6 +39,8 @@ export type AgentSessionSubscribeInput = { /** Every frame to fit a phone's bridge frame; the RPC sets it for every phone. */ boundedFrames?: boolean readsCurrentRequests?: boolean + /** False for desktops that need exact replay to keep send echoes and current requests. */ + snapshotCatchUp?: boolean } export type Subscriber = CurrentRequestsSubscriber & { @@ -48,6 +50,7 @@ export type Subscriber = CurrentRequestsSubscriber & { 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. */ @@ -99,6 +102,7 @@ export class AgentSessionSubscribers { cursor?: AgentJournalCursor boundedFrames?: boolean readsCurrentRequests?: boolean + snapshotCatchUp?: boolean }): () => void { const liveCursor = input.journal.cursor() const subscriber: Subscriber = { @@ -108,7 +112,8 @@ export class AgentSessionSubscribers { cursor: input.cursor ?? { epoch: liveCursor.epoch, sequence: 0 }, fence: input.fence, readsCurrentRequests: input.readsCurrentRequests === true, - 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 d70228762bf..e537f24db49 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-rpc.test-fixture.ts b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts index ce9b35c8f3c..e12a7d67a15 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts @@ -268,7 +268,7 @@ export async function call( client?: { clientId?: string clientKind?: 'mobile' | 'runtime' - clientCapabilities?: string[] + clientCapabilities?: readonly string[] signal?: AbortSignal }, runtimeOverrides: Record = {}, 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 20d04ee3c34..cc4702432ab 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 @@ -3,7 +3,11 @@ import { readFileSync } from 'node:fs' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { AGENT_SESSION_OPTIONAL_MODEL_CLIENT_CAPABILITY } from '../../../../shared/agent-session-optional-model-capability' import { ELECTRON_REMOTE_RUNTIME_CLIENT_CAPABILITIES } from '../../../../shared/electron-remote-runtime-client-capabilities' -import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' +import { + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, + STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY +} from '../../../../shared/protocol-version' +import { AGENT_SESSION_CURRENT_REQUESTS_CLIENT_CAPABILITY } from '../../../../shared/agent-session-current-requests-capability' import { remoteRuntimeClientCapabilities } from '../../../../shared/remote-runtime-client-capabilities' import { AcpStructuredOptions } from '../../../acp/acp-structured-options' import { NewSessionResponseSchema } from '../../../acp/generated/acp-protocol.generated' @@ -28,12 +32,34 @@ import { beforeEach(installStructuredHostStub) afterEach(clearStructuredHostStub) +const currentDesktop = { + ...STRUCTURED_CLIENT, + clientCapabilities: [ + ...STRUCTURED_CLIENT.clientCapabilities, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, + AGENT_SESSION_CURRENT_REQUESTS_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], + [ + 'a desktop that only settles skipped sends', + { + ...STRUCTURED_CLIENT, + clientCapabilities: [ + ...STRUCTURED_CLIENT.clientCapabilities, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY + ] + }, + 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 @@ -42,6 +68,28 @@ 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) }) describe('agentSession.conversationOutline', () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index 5d5ddea42bb..e0c92cc03f4 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -62,6 +62,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' import { ORCA_LAYOUT_OWNER_IS_MAIN } from '../../../../shared/layout-owner-switch' import { noteStructuredAgentSessionClose, @@ -200,6 +201,13 @@ 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)), // 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 587849965da..0008648b079 100644 --- a/src/renderer/src/web/web-runtime-client-capabilities.ts +++ b/src/renderer/src/web/web-runtime-client-capabilities.ts @@ -5,6 +5,7 @@ import { AGENT_STATUS_RESTORED_UNCONFIRMED_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, @@ -22,6 +23,7 @@ export const WEB_RUNTIME_CLIENT_CAPABILITIES = [ AGENT_SESSION_CURRENT_REQUESTS_CLIENT_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 0a5d99c7c40..aaa73a4917f 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, AGENT_STATUS_RESTORED_UNCONFIRMED_RUNTIME_CAPABILITY, REPO_SEARCH_QUALIFIED_REFS_RUNTIME_CAPABILITY, @@ -132,6 +133,7 @@ describe('WebRuntimeClient', () => { AGENT_SESSION_CURRENT_REQUESTS_CLIENT_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 122bd6ff710..0cb63d32f23 100644 --- a/src/shared/electron-remote-runtime-client-capabilities.ts +++ b/src/shared/electron-remote-runtime-client-capabilities.ts @@ -3,6 +3,7 @@ import { TERMINAL_IMAGE_VIEWER_CAPABILITY } from './terminal-image-viewer-capabi import { AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY, + AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY, AGENT_STATUS_RESTORED_UNCONFIRMED_RUNTIME_CAPABILITY, BROWSER_CLIENT_HOST_RUNTIME_CAPABILITY, BROWSER_CLIENT_PAGE_METADATA_RUNTIME_CAPABILITY, @@ -28,6 +29,7 @@ export const ELECTRON_REMOTE_RUNTIME_CLIENT_CAPABILITIES = [ AGENT_SESSION_CURRENT_REQUESTS_CLIENT_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 a2ef9fcf7af..cd0004699ef 100644 --- a/src/shared/runtime-capability-constants.ts +++ b/src/shared/runtime-capability-constants.ts @@ -190,6 +190,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 8d3dc4f4d43..d99b91336c2 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 d14aee3b599..2c3523f7027 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 0220a33e59a..26b75cd6560 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({