Catch up desktop chats from the newest page (#29125)

* fix(native-chat): retire acknowledged send echoes after catch-up

* test: follow phone lint rules in reconnect regression

* test: format reconnect message expectation

* Keep current chat requests independent of loaded history

* Read current chat requests in bounded cursor pieces

* Bound desktop catch-up after independent request support

* Fix current request subscription static checks

* Match generated request schema import ordering

* fix(native-chat): correct current request type boundaries

* Accept readonly client capability fixtures

* Exercise current request reads in wire compatibility inventory

* Align current request fixtures with provider commit and RPC surface

* test(native-chat): include current request browser capability

* Keep the mobile chat session hook within its line budget

* Format mobile current request readers

* fix(native-chat): reopen once for a current-request version whose read keeps failing

A deterministic read failure reopened the subscription, whose first frame re-read the same version and reset the backoff, so the pane resubscribed every 750 ms forever.

* test(native-chat): type the current-request snapshot fixture

* test(native-chat): wait for the hydrated open before driving the subscription
This commit is contained in:
Brennan Benson
2026-10-16 17:46:54 -07:00
committed by GitHub
parent a2933244e3
commit dd3bbf680e
15 changed files with 124 additions and 57 deletions
@@ -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,
@@ -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<boolean>(9).fill(true),
false
])
expect(returnedRows).toEqual([...Array<number>(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 () => {
@@ -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
@@ -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<string, Subscriber>()
session.set(input.id, subscriber)
@@ -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
})
})
@@ -268,7 +268,7 @@ export async function call(
client?: {
clientId?: string
clientKind?: 'mobile' | 'runtime'
clientCapabilities?: string[]
clientCapabilities?: readonly string[]
signal?: AbortSignal
},
runtimeOverrides: Record<string, unknown> = {},
@@ -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', () => {
@@ -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 } : {})
})
@@ -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,
@@ -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,
@@ -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,
@@ -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
@@ -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',
@@ -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<RpcReply[]> {
return call('agentSession.subscribe', { sessionId: SESSION, cursor })
async function subscribe(cursor: AgentJournalCursor | null = journal.cursor()): Promise<RpcReply[]> {
return call('agentSession.subscribe', { sessionId: SESSION, ...(cursor ? { cursor } : {}) })
}
function events(replies: RpcReply[]): AgentSessionSubscribeEvent[] {
return replies.map((reply) => {
@@ -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({