diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts index 8b8d6fb5ee8..c8c908aae70 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts @@ -192,6 +192,11 @@ describe('an open chat receives each row once', () => { it('when the provider frame that wrote it also publishes', async () => { const pane = liveReader() const sink = providerSink() + const journal = host['sessions'].get(SESSION)?.journal + if (!journal) { + throw new Error('the attached chat has no journal') + } + const readSince = vi.spyOn(journal, 'readSince') sink.appendItem( { provider: 'orca', clientMessageId: 'streamed' }, @@ -201,6 +206,8 @@ describe('an open chat receives each row once', () => { await host.flushStreamedEvents(SESSION) expect(pane.received().statuses).toEqual(['streamed row']) + // The frame's publish finds the reader caught up and reads no rows: streaming stays one read. + expect(readSince).toHaveBeenCalledOnce() }) it('when a writer publishes the row it appended', async () => { 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 af5e231730c..0b54ea7c430 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 @@ -15,6 +15,7 @@ import { type AgentSessionSubscribeEvent, type AgentSessionTurnActivity } from '../../../shared/agent-session-wire' +import { sameJournalCursor } from '../agent-session-journal/journal-cursor' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { emptyAgentSessionBatch } from './agent-session-empty-batch' import { @@ -213,6 +214,11 @@ export class AgentSessionSubscribers { ? this.activityField(subscriber.sessionId).activity : undefined const publishedActivity = activity !== undefined ? activity : checkpointActivity + // Caught up, so there are no rows to read: every publish behind a commit's own delivery. + if (!journal.isReadOnly && sameJournalCursor(subscriber.cursor, journal.cursor())) { + this.emitCaughtUp(subscriber, hostNow, emitCheckpoint, backgroundTasks, publishedActivity) + return + } const readPage = createAgentSessionCatchUpReader(journal) while (true) { const result = readPage({ @@ -239,20 +245,7 @@ export class AgentSessionSubscribers { const page = result.page const advanced = page.window.nextCursor.sequence > subscriber.cursor.sequence if (!advanced) { - const commandsChanged = - this.hooks.readCommands !== undefined && - (this.hooks.readCommands(subscriber.sessionId) ?? null) !== subscriber.commands - if (emitCheckpoint || publishedActivity !== undefined || commandsChanged) { - this.emit(subscriber, { - type: 'batch', - sessionId: subscriber.sessionId, - batch: emptyAgentSessionBatch(page.window.nextCursor), - fence: subscriber.fence, - hostNow, - ...(backgroundTasks !== undefined ? { backgroundTasks } : {}), - ...(publishedActivity !== undefined ? { activity: publishedActivity } : {}) - }) - } + this.emitCaughtUp(subscriber, hostNow, emitCheckpoint, backgroundTasks, publishedActivity) return } this.emit(subscriber, { @@ -276,6 +269,29 @@ export class AgentSessionSubscribers { } } + private emitCaughtUp( + subscriber: Subscriber, + hostNow: number, + emitCheckpoint: boolean, + backgroundTasks: AgentSessionBackgroundTaskState | null | undefined, + activity: AgentSessionTurnActivity | null | undefined + ): void { + const commandsChanged = + this.hooks.readCommands !== undefined && + (this.hooks.readCommands(subscriber.sessionId) ?? null) !== subscriber.commands + if (emitCheckpoint || activity !== undefined || commandsChanged) { + this.emit(subscriber, { + type: 'batch', + sessionId: subscriber.sessionId, + batch: emptyAgentSessionBatch(subscriber.cursor), + fence: subscriber.fence, + hostNow, + ...(backgroundTasks !== undefined ? { backgroundTasks } : {}), + ...(activity !== undefined ? { activity } : {}) + }) + } + } + private now = (): number => this.hooks.now?.() ?? Date.now() private isActive = (subscriber: Subscriber): boolean =>