perf(native-chat): a publish behind a delivered commit reads nothing

Each commit now delivers itself, so the publish a provider frame still sends
afterwards found every reader caught up but still read rows and rebuilt the
timeline for each one. A caught-up reader now skips the read.
This commit is contained in:
Brennan Benson
2026-09-25 02:43:30 -07:00
parent 427dc8de83
commit 50d2e33099
2 changed files with 37 additions and 14 deletions
@@ -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 () => {
@@ -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 =>