Merge #22811 into the PR 1 base

src/main/native-chat/agent-session-journal/journal-store.ts

@@ -76,6 +76,7 @@ export class AgentSessionJournal {
   private readOnly = false
   private malformedRows = 0
   private database: OpenJournalDatabase | null = null
+  private onCommitted: (() => void) | null = null
   private readonly queue: JournalWriteQueue
   private readonly closer: JournalConnectionCloser
   private readonly rowWriter: JournalRowWriter
@@ -111,8 +112,14 @@ export class AgentSessionJournal {
         this.readOnly = readOnly
       },
       cursor: this.cursor,
-      adopt: (loaded) => this.adoptLoadedJournal(loaded),
-      commit: (row) => applyJournalRow(this.state, row),
+      adopt: (loaded) => {
+        this.adoptLoadedJournal(loaded)
+        this.onCommitted?.()
+      },
+      commit: (row) => {
+        applyJournalRow(this.state, row)
+        this.onCommitted?.()
+      },
       loaded: () => this.loaded,
       malformedRows: () => this.malformedRows,
       setMalformedRows: (count) => {
@@ -167,6 +174,12 @@ export class AgentSessionJournal {
     return this.closer.close()
   }
 
+  /** Told of every durable change, epoch replacements included, so a reader learns of a write
+   *  without its writer saying so. One listener: a later call replaces it. It must not throw. */
+  observeCommits(listener: () => void): void {
+    this.onCommitted = listener
+  }
+
   cursor = (): AgentJournalCursor => ({
     epoch: this.state.epoch,
     sequence: this.state.lastSequence

src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts

@@ -0,0 +1,231 @@
+// What an open chat receives, asserted at its subscriber rather than in the journal: a fresh
+// subscribe re-reads the journal and hides a write that never reached the readers already open.
+
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest'
+import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
+import type {
+  AgentJournalRenderItem,
+  AgentJournalSubmission
+} from '../../../shared/agent-session-journal-types'
+import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire'
+import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
+import { StructuredAgentSessionHost } from './structured-agent-session-host'
+import {
+  HOST_TEST_NOW as NOW,
+  HOST_TEST_SESSION as SESSION,
+  HOST_TEST_THREAD as THREAD,
+  hostTestAttachParams,
+  hostTestMessage,
+  hostTestOperationId,
+  resetHostTestOperationIds
+} from './structured-agent-session-host-test-data'
+
+const CALLER = { callerKey: 'client-1' }
+const EXIT_REASON = 'Claude Code is not signed in. Sign in with the Claude CLI'
+
+let root: string
+let store: AgentSessionRecordStore
+let host: StructuredAgentSessionHost
+let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
+let generation = 0
+
+/** Everything a live subscriber was sent after it opened. */
+function liveReader() {
+  const events: AgentSessionSubscribeEvent[] = []
+  host.subscribe({ id: 'pane', sessionId: SESSION, emit: (event) => events.push(event) })
+  const opened = events.length
+  const received = () => {
+    const items: AgentJournalRenderItem[] = []
+    const submissions: AgentJournalSubmission[] = []
+    for (const event of events.slice(opened)) {
+      if (event.type === 'batch') {
+        items.push(...event.batch.items)
+        submissions.push(...event.batch.submissions)
+      } else if (event.type === 'snapshot' || event.type === 'reset') {
+        items.push(...event.page.items)
+        submissions.push(...event.page.submissions)
+      }
+    }
+    return {
+      statuses: items.flatMap((item) => (item.body.kind === 'status' ? [item.body.text] : [])),
+      submissions,
+      batches: events.slice(opened).filter((event) => event.type === 'batch').length
+    }
+  }
+  return { received }
+}
+
+async function send(text: string): Promise<string> {
+  const body = hostTestMessage(text)
+  const sent = await host.send(CALLER, {
+    envelope: {
+      sessionId: SESSION,
+      clientOperationId: hostTestOperationId(),
+      expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 0,
+      payloadFingerprint: computeAgentSessionPayloadFingerprint({
+        method: 'agentSession.send',
+        sessionId: SESSION,
+        fields: { body }
+      })
+    },
+    body
+  })
+  expect(sent, JSON.stringify(sent)).toMatchObject({ ok: true })
+  return sent.ok ? sent.value.clientMessageId : ''
+}
+
+function exitBeforeProof(): Promise<void> {
+  return host.handleAdapterEvent({
+    type: 'ended',
+    sessionId: SESSION,
+    fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 0,
+    acquisitionGeneration: `generation-${generation}`,
+    reason: EXIT_REASON,
+    cause: 'unexpected-exit',
+    startupUnproven: true
+  })
+}
+
+function providerSink() {
+  const events = acquire.mock.calls.at(-1)?.[0].events
+  if (!events) {
+    throw new Error('no acquired provider sink')
+  }
+  return events
+}
+
+beforeEach(async () => {
+  root = await mkdtemp(join(tmpdir(), 'orca-append-delivery-'))
+  resetHostTestOperationIds()
+  generation = 0
+  acquire = vi.fn(async ({ fence, spawnToken }) => ({
+    process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken },
+    link: {
+      linkId: `link-${fence}`,
+      handle: { provider: 'codex', threadId: THREAD },
+      origin: generation === 0 ? ('created' as const) : ('resumed' as const),
+      mintedAtFence: fence,
+      observedAt: NOW
+    },
+    acquisitionGeneration: `generation-${++generation}`,
+    providerChildPhase: 'starting' as const
+  }))
+  store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
+  host = new StructuredAgentSessionHost({
+    store,
+    adapter: {
+      acquire,
+      releaseAcquisition: vi.fn(async () => true),
+      closeSession: vi.fn(async () => true),
+      dispatch: vi.fn(async () => ({ state: 'admitted' as const })),
+      cancelTurn: vi.fn(async () => ({ cancelled: true })),
+      answerPrompt: vi.fn(async () => undefined),
+      setOption: vi.fn(async () => undefined)
+    },
+    journalRoot: root,
+    claimKeyId: 'key-1',
+    mintSpawnToken: () => `spawn-${generation + 1}`,
+    now: () => NOW
+  })
+  await expect(host.attach(CALLER, hostTestAttachParams(null))).resolves.toMatchObject({
+    ok: true
+  })
+})
+
+afterEach(async () => {
+  await host.flushAllStreamedEvents()
+  await rm(root, { recursive: true, force: true })
+})
+
+describe('an open chat receives every row its journal commits', () => {
+  it('shows a failed start whose lease could not be handed back', async () => {
+    const held = await send('hello')
+    const pane = liveReader()
+    // The exit settles the journal, then fails to release the lease: nothing moves the fence.
+    vi.spyOn(store, 'transitionHandoff').mockRejectedValueOnce(new Error('record store busy'))
+
+    await exitBeforeProof()
+
+    expect(pane.received().statuses).toEqual([
+      expect.stringMatching(/stopped before it finished starting: .*not signed in/)
+    ])
+    expect(pane.received().submissions).toContainEqual(
+      expect.objectContaining({ clientMessageId: held, dispatchState: 'rejected' })
+    )
+  })
+
+  it('shows a revision the provider queued with no publish behind it', async () => {
+    const pane = liveReader()
+    const identity = { provider: 'orca' as const, clientMessageId: 'context-usage' }
+    const body = { kind: 'status' as const, text: 'context usage answered after the turn' }
+
+    expect(providerSink().tryReviseResolvedItem?.(4_096, () => ({ identity, body }))).toEqual({
+      accepted: true
+    })
+    await host.flushStreamedEvents(SESSION)
+
+    expect(pane.received().statuses).toEqual(['context usage answered after the turn'])
+  })
+
+  it('shows a row appended straight to the journal', async () => {
+    const pane = liveReader()
+    const journal = host['sessions'].get(SESSION)?.journal
+    if (!journal) {
+      throw new Error('the attached chat has no journal')
+    }
+
+    await journal.appendItem(
+      { provider: 'orca', clientMessageId: 'host-note' },
+      { kind: 'status', text: 'written by a writer that publishes nothing' },
+      { fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 0 }
+    )
+
+    expect(pane.received().statuses).toEqual(['written by a writer that publishes nothing'])
+  })
+})
+
+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' },
+      { kind: 'status', text: 'streamed row' }
+    )
+    sink.publish()
+    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 () => {
+    const pane = liveReader()
+    const journal = host['sessions'].get(SESSION)?.journal
+    if (!journal) {
+      throw new Error('the attached chat has no journal')
+    }
+
+    await journal.appendItem(
+      { provider: 'orca', clientMessageId: 'host-row' },
+      { kind: 'status', text: 'host row' },
+      { fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 0 }
+    )
+    host['subscribers'].publish(SESSION, journal)
+
+    expect(pane.received().statuses).toEqual(['host row'])
+    // The second publish found nothing past the reader's cursor, so it sent nothing at all.
+    expect(pane.received().batches).toBe(1)
+  })
+})

src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts

@@ -0,0 +1,154 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types'
+import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
+import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open'
+import { StructuredAgentSessionConversations } from './structured-agent-session-conversations'
+import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types'
+import { hostTestAttachParams } from './structured-agent-session-host-test-data'
+
+const IDENTITY: AgentSessionJournalIdentity = {
+  sessionId: 'session-1',
+  workspaceId: 'workspace-1',
+  hostId: 'host-1',
+  agent: 'codex',
+  providerHandle: { kind: 'codex', threadId: 'thread-1' }
+}
+
+const journals = createTrackedJournalOpener()
+let root: string | null = null
+
+afterEach(async () => {
+  await journals.closeAll()
+  if (root) {
+    await rm(root, { recursive: true, force: true })
+    root = null
+  }
+})
+
+async function openJournal(name: string): Promise<AgentSessionJournal> {
+  root ??= await mkdtemp(join(tmpdir(), 'orca-conversations-'))
+  return journals.open({ identity: IDENTITY, journalDir: join(root, name) })
+}
+
+function session(journal: AgentSessionJournal) {
+  return {
+    journal,
+    params: hostTestAttachParams(null),
+    fence: 0,
+    hasProviderChild: false,
+    providerChildPhase: 'ready' as const,
+    acquisitionGeneration: null
+  }
+}
+
+function appendStatus(journal: AgentSessionJournal, text: string) {
+  return journal.appendItem(
+    { provider: 'orca', clientMessageId: text },
+    { kind: 'status', text },
+    { fence: 0 }
+  )
+}
+
+describe('a conversation delivers what its journal commits', () => {
+  it('delivers an append that no writer published', async () => {
+    const deliver = vi.fn()
+    const conversations = new StructuredAgentSessionConversations({
+      deliver,
+      onDeliveryError: vi.fn()
+    })
+    const journal = await openJournal('a')
+    conversations.set('session-1', session(journal))
+
+    await appendStatus(journal, 'written')
+
+    // Already delivered when the writer's await returns, as an explicit publish would have been.
+    expect(deliver).toHaveBeenCalledOnce()
+    expect(deliver).toHaveBeenCalledWith('session-1', journal)
+  })
+
+  it('binds a handle set through a plain map reference', async () => {
+    const deliver = vi.fn()
+    // Collaborators hold the host's map as a plain `Map`; `set` still reaches the binding.
+    const sessions: Map<string, StructuredAgentSessionHostSession> =
+      new StructuredAgentSessionConversations({ deliver, onDeliveryError: vi.fn() })
+    const journal = await openJournal('a')
+    sessions.set('session-1', session(journal))
+
+    await appendStatus(journal, 'through the plain map')
+
+    expect(deliver).toHaveBeenCalledExactlyOnceWith('session-1', journal)
+  })
+
+  it('delivers an epoch replacement, which readers must reload from', async () => {
+    const deliver = vi.fn()
+    const conversations = new StructuredAgentSessionConversations({
+      deliver,
+      onDeliveryError: vi.fn()
+    })
+    const journal = await openJournal('a')
+    conversations.set('session-1', session(journal))
+
+    await journal.replaceEpochItems('handle_forked', 0, [])
+
+    expect(deliver).toHaveBeenCalledExactlyOnceWith('session-1', journal)
+  })
+
+  it('delivers nothing for a handle the conversation has replaced', async () => {
+    const deliver = vi.fn()
+    const conversations = new StructuredAgentSessionConversations({
+      deliver,
+      onDeliveryError: vi.fn()
+    })
+    const replaced = await openJournal('a')
+    const current = await openJournal('b')
+    conversations.set('session-1', session(replaced))
+    conversations.set('session-1', session(current))
+
+    await appendStatus(replaced, 'stale')
+    expect(deliver).not.toHaveBeenCalled()
+
+    await appendStatus(current, 'live')
+    expect(deliver).toHaveBeenCalledExactlyOnceWith('session-1', current)
+  })
+
+  it('delivers nothing once the conversation is dropped', async () => {
+    const deliver = vi.fn()
+    const conversations = new StructuredAgentSessionConversations({
+      deliver,
+      onDeliveryError: vi.fn()
+    })
+    const journal = await openJournal('a')
+    conversations.set('session-1', session(journal))
+    conversations.delete('session-1')
+
+    await appendStatus(journal, 'after close')
+
+    expect(deliver).not.toHaveBeenCalled()
+  })
+
+  it('reports a reader failure without failing the durable write', async () => {
+    const failure = new Error('reader failed')
+    const onDeliveryError = vi.fn()
+    const conversations = new StructuredAgentSessionConversations({
+      deliver: () => {
+        throw failure
+      },
+      onDeliveryError
+    })
+    const journal = await openJournal('a')
+    conversations.set('session-1', session(journal))
+
+    await expect(appendStatus(journal, 'durable')).resolves.toMatchObject({
+      itemId: expect.any(String)
+    })
+
+    expect(onDeliveryError).toHaveBeenCalledExactlyOnceWith('session-1', failure)
+    expect(journal.snapshot().items.map((item) => item.body)).toContainEqual({
+      kind: 'status',
+      text: 'durable'
+    })
+  })
+})

src/main/native-chat/agent-session-wire/structured-agent-session-conversations.ts

@@ -0,0 +1,47 @@
+import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
+import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types'
+
+/**
+ * The host's open conversations. A journal handle becomes a conversation's when it is set here,
+ * and from then on every change it commits reaches that conversation's readers: no writer
+ * publishes what it appended, so none can forget to.
+ *
+ * Delivery runs a microtask after the commit, so a reader that throws cannot fail a write that is
+ * already durable. A handle this map has since replaced or dropped delivers nothing.
+ */
+export class StructuredAgentSessionConversations extends Map<
+  string,
+  StructuredAgentSessionHostSession
+> {
+  constructor(
+    private readonly delivery: {
+      deliver: (sessionId: string, journal: AgentSessionJournal) => void
+      onDeliveryError: (sessionId: string, error: unknown) => void
+    }
+  ) {
+    super()
+  }
+
+  override set(sessionId: string, session: StructuredAgentSessionHostSession): this {
+    const { journal } = session
+    let queued = false
+    journal.observeCommits(() => {
+      if (queued) {
+        return
+      }
+      queued = true
+      queueMicrotask(() => {
+        queued = false
+        if (this.get(sessionId)?.journal !== journal) {
+          return
+        }
+        try {
+          this.delivery.deliver(sessionId, journal)
+        } catch (error) {
+          this.delivery.onDeliveryError(sessionId, error)
+        }
+      })
+    })
+    return super.set(sessionId, session)
+  }
+}

src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts

@@ -224,7 +224,6 @@ export async function settleStructuredAgentSessionLateDispatch(
           fence: session.fence
         }
   )
-  context.publish(input.sessionId, session.journal)
 }
 
 /**
@@ -240,7 +239,7 @@ export async function settleStructuredAgentSessionLateDispatch(
  * it never makes a send re-deliverable, because the provider may well have run it.
  */
 export async function releaseStructuredAgentSessionUnansweredDispatches(
-  context: Pick<StructuredAgentSessionMutationContext, 'sessions' | 'publish'>,
+  context: Pick<StructuredAgentSessionMutationContext, 'sessions'>,
   input: { sessionId: string; reason: string }
 ): Promise<void> {
   const session = context.sessions.get(input.sessionId)
@@ -263,7 +262,6 @@ export async function releaseStructuredAgentSessionUnansweredDispatches(
       recovered: true
     })
   }
-  context.publish(input.sessionId, session.journal)
 }
 
 /** The host's thin mutation surface. Each call re-reads the context, so a session

src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts

@@ -27,7 +27,8 @@ export type StructuredAgentSessionReveal = {
 }
 
 export type StructuredAgentSessionHostSession = {
-  journal: AgentSessionJournal
+  /** Readonly: a new handle enters only through the session map's `set`, which binds its delivery. */
+  readonly journal: AgentSessionJournal
   params: AgentSessionAttachParams
   fence: number
   /** Whether THIS host generation is running the provider process behind the session. A journal

src/main/native-chat/agent-session-wire/structured-agent-session-host.ts

@@ -46,6 +46,7 @@ import type { StructuredAgentSessionTurnCompletionSubscriber } from './structure
 import { StructuredAgentSessionEventRecovery } from './structured-agent-session-event-recovery'
 import { StructuredAgentSessionBackgroundTaskChannel } from './structured-agent-session-background-task-channel'
 import { StructuredAgentSessionClientDelivery } from './structured-agent-session-client-delivery'
+import { StructuredAgentSessionConversations } from './structured-agent-session-conversations'
 import {
   createStructuredAgentSessionRestartResume,
   type StructuredAgentSessionRestartResume
@@ -58,7 +59,10 @@ export class StructuredAgentSessionHost {
     () => this.mutationContext(),
     this
   )
-  private readonly sessions = new Map<string, StructuredAgentSessionHostSession>()
+  private readonly sessions = new StructuredAgentSessionConversations({
+    deliver: (sessionId, journal) => this.subscribers.publish(sessionId, journal),
+    onDeliveryError: (sessionId, error) => this.deps.onEventSinkError?.({ sessionId, error })
+  })
   private readonly clientDelivery = new StructuredAgentSessionClientDelivery(
     this.sessions,
     () => this.now(),
@@ -127,10 +131,11 @@ export class StructuredAgentSessionHost {
       ensureProviderChild: (id, options) => this.holds.ensureProviderChild(id, options),
       onBarrierError: (sessionId, error) => deps.onEventSinkError?.({ sessionId, error })
     })
-    this.restartResume = createStructuredAgentSessionRestartResume(deps, this.sessions, {
-      ...structuredAgentSessionRestartResumeSurfaces(this, this.now),
-      publish: this.subscribers.publish.bind(this.subscribers)
-    })
+    this.restartResume = createStructuredAgentSessionRestartResume(
+      deps,
+      this.sessions,
+      structuredAgentSessionRestartResumeSurfaces(this, this.now)
+    )
     this.runtimeState.startLeaseRenewal()
   }
 

src/main/native-chat/agent-session-wire/structured-agent-session-journal-handles.test.ts

@@ -242,9 +242,13 @@ describe('site 11: host teardown is failure-complete', () => {
     const failing = sessions.get(SESSION)
     const closeError = new Error('close rejected')
     if (failing) {
-      failing.journal = {
-        close: () => Promise.reject(closeError)
-      } as unknown as AgentSessionJournal
+      sessions.set(SESSION, {
+        ...failing,
+        // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: teardown calls only `close`, and this map is a plain `Map` that binds no delivery.
+        journal: {
+          close: () => Promise.reject(closeError)
+        } as unknown as AgentSessionJournal
+      })
     }
 
     await expect(

src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts

@@ -55,7 +55,6 @@ export type StructuredAgentSessionContinuationHost = {
     clientMessageId: string
   ) => Promise<{ value: AgentSessionSendResult } | undefined>
   onNoteFailed: (sessionId: string, error: unknown) => void
-  publish: (sessionId: string, journal: AgentSessionJournal) => void
   now: () => number
   /** Whether the marker is still an offer, with the continuation's own submission set aside.
    *  Re-asked right before dispatch, so a newer user message refuses the send; a provider turn
@@ -106,9 +105,9 @@ export function noteRestartReattachFailed(
   )
 }
 
-/** Writes a host-authored status note into the chat and publishes it to open panes. */
+/** Writes a host-authored status note into the chat. */
 function restartNoteWriter(
-  host: Pick<StructuredAgentSessionContinuationHost, 'sessions' | 'publish' | 'now'>
+  host: Pick<StructuredAgentSessionContinuationHost, 'sessions' | 'now'>
 ): StructuredAgentSessionContinuationDeps['note'] {
   return async (sessionId, text, tone) => {
     const session = host.sessions.get(sessionId)
@@ -120,7 +119,6 @@ function restartNoteWriter(
       { kind: 'status', text, ...(tone ? { tone } : {}) },
       { fence: session.fence }
     )
-    host.publish(sessionId, session.journal)
   }
 }
 

src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts

@@ -45,7 +45,6 @@ import { createStructuredAgentSessionRestartWitnesses } from './structured-agent
 type LiveSession = { journal: AgentSessionJournal; hasProviderChild: boolean; fence: number }
 
 export type StructuredAgentSessionRestartResumeSurfaces = {
-  publish: (sessionId: string, journal: AgentSessionJournal) => void
   revealSession: (sessionId: string) => Promise<{ readable: boolean }>
   hold: (sessionId: string, holderId: string) => Promise<void>
   release: (sessionId: string, holderId: string) => void

src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts

@@ -41,7 +41,7 @@ type RestartResumeHostBindings = {
 export function structuredAgentSessionRestartResumeSurfaces(
   host: RestartResumeHostBindings,
   now: () => number
-): Omit<StructuredAgentSessionRestartResumeSurfaces, 'publish'> {
+): StructuredAgentSessionRestartResumeSurfaces {
   return {
     revealSession: host.revealSession,
     hold: host.hold,

src/main/native-chat/agent-session-wire/structured-agent-session-send-preparation.ts

@@ -114,7 +114,7 @@ export function structuredAgentSessionSendNeedsOwner(
 
 type SendPreparationContext = Pick<
   StructuredAgentSessionMutationContext,
-  'deps' | 'sessions' | 'holds' | 'restoreReadable' | 'publish'
+  'deps' | 'sessions' | 'holds' | 'restoreReadable'
 >
 
 export async function prepareStructuredAgentSessionSend(
@@ -224,7 +224,6 @@ async function recordFailedRestart(
         }
       ]
     })
-    context.publish(sessionId, session.journal)
   } catch (error) {
     context.deps.onEventSinkError?.({ sessionId, error })
   }

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 =>

src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts

@@ -246,7 +246,7 @@ describe('a chat that closes', () => {
     expect(host.hasSession(SESSION)).toBe(true)
   })
 
-  it('releases a compatibility wait when the session is evicted', async () => {
+  it('answers a compatibility wait with what eviction recorded', async () => {
     await attach()
     dispatch.mockResolvedValueOnce({ state: 'admitted' })
     const body = hostTestMessage('pending until close')
@@ -265,7 +265,12 @@ describe('a chat that closes', () => {
 
     await host.close(SESSION)
 
-    await expect(settlement).resolves.toBeUndefined()
+    // Eviction's settlement is a journal write, so the wait sees it rather than timing out.
+    await expect(settlement).resolves.toMatchObject({
+      value: {
+        submission: { dispatchState: 'unknown', reason: 'provider_closed_before_acknowledgement' }
+      }
+    })
   })
 
   it('retries teardown after journal close loses its result', async () => {

src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.ts

@@ -68,13 +68,11 @@ export async function performThreadGoalChange(
       },
       { fence: ctx.fence }
     )
-    ctx.publish()
   }
   const withdrawObjective = async (): Promise<void> => {
     if (change.kind === 'set') {
       // Nothing was sent as a goal.
       await ctx.journal.appendTombstone(identity, { fence: ctx.fence })
-      ctx.publish()
     }
   }
   let result: Awaited<ReturnType<typeof ctx.adapter.changeThreadGoal>>

src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts

@@ -73,7 +73,6 @@ export async function performPrompt(
             fence: ctx.fence
           }
         )
-        ctx.publish()
       }
     })
   } catch (error) {
@@ -93,7 +92,6 @@ export async function performPrompt(
       },
       { fence: ctx.fence }
     )
-    ctx.publish()
   }
   const appended = committed.item
   if (!appended) {

src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts

@@ -42,6 +42,8 @@ export type AgentSessionTurnContext = {
   persistOptions: (options: Readonly<Record<string, string>>) => Promise<void>
   /** Opaque client identity recorded as the resolver of a prompt. */
   resolvedBy: string
+  /** Republishes state kept outside the journal, such as the record's options or rewind phase.
+   *  Journal appends reach readers on their own. */
   publish: () => void
   /** Drains provider lifecycle already accepted by the execution host. */
   flushStreamedEvents: () => Promise<void>
@@ -100,7 +102,6 @@ async function appendStatus(
     { kind: 'status', text },
     { fence: ctx.fence }
   )
-  ctx.publish()
 }
 
 /**
@@ -136,7 +137,6 @@ export async function performSend(
   } catch {
     return invalid('The message could not be recorded and was not sent.')
   }
-  ctx.publish()
 
   // The row just written is the send's instant on the host clock; the turn this
   // dispatch opens records it so the live counter never re-anchors at turn-open.
@@ -161,14 +161,12 @@ export async function performSend(
         if (!recorded) {
           console.warn('[structured-agent-session] pre-dispatch refusal persistence failed')
         }
-        ctx.publish()
       }
       throw error
     }
   )
   // An admission needs no dispatch row: the submission is already pending.
   if (outcome.state === 'admitted') {
-    ctx.publish()
     return {
       ok: true,
       value: {
@@ -206,10 +204,8 @@ export async function performSend(
     } catch {
       // Nothing further to record; the pending row is settled on the next attach.
     }
-    ctx.publish()
     throw error
   }
-  ctx.publish()
   return {
     ok: true,
     value: {

src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts

@@ -35,9 +35,8 @@ function contextWith(submissions: AgentJournalSubmission[]) {
   // The mutation reads only `journal.submissions`, `journal.resolveDispatch` and `fence`.
   // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: all three are supplied here; the rest of the host session is unreachable from this mutation.
   const session = { journal, fence: FENCE } as unknown as ReleaseSession
-  const publish = vi.fn()
-  const context: ReleaseContext = { sessions: new Map([['s-1', session]]), publish }
-  return { context, resolved, publish, journal }
+  const context: ReleaseContext = { sessions: new Map([['s-1', session]]) }
+  return { context, resolved, journal }
 }
 
 describe('releasing dispatches the provider can no longer answer', () => {
@@ -45,7 +44,7 @@ describe('releasing dispatches the provider can no longer answer', () => {
     const live = [submission({})]
     // POSITIVE CONTROL: this is the latch being dissolved.
     expect(projectStructuredAgentSessionStatus([], live, FENCE)).toBe('working')
-    const { context, resolved, publish } = contextWith(live)
+    const { context, resolved } = contextWith(live)
 
     await releaseStructuredAgentSessionUnansweredDispatches(context, {
       sessionId: 's-1',
@@ -62,14 +61,13 @@ describe('releasing dispatches the provider can no longer answer', () => {
         recovered: true
       }
     ])
-    expect(publish).toHaveBeenCalledOnce()
     expect(projectStructuredAgentSessionStatus([], [submission({ recovered: true })], FENCE)).toBe(
       'idle'
     )
   })
 
   it('never touches a pending send, whose dispatch may still be in flight', async () => {
-    const { context, resolved, publish } = contextWith([
+    const { context, resolved } = contextWith([
       submission({ clientMessageId: 'm-2', dispatchState: 'pending', reason: null })
     ])
 
@@ -79,7 +77,6 @@ describe('releasing dispatches the provider can no longer answer', () => {
     })
 
     expect(resolved).toEqual([])
-    expect(publish).not.toHaveBeenCalled()
   })
 
   it('leaves an already recovered unknown alone', async () => {

src/main/native-chat/agent-session-wire/structured-conversation-command.ts

@@ -176,7 +176,6 @@ export function runStructuredConversationCommand(
               },
               { fence: ctx.fence }
             )
-            ctx.publish()
             try {
               error = (
                 await ctx.adapter.compact({
@@ -212,7 +211,6 @@ export function runStructuredConversationCommand(
                           conversationCommand: matching()!
                         }
                       })
-                      ctx.publish()
                     })
                 })
               ).error
@@ -223,7 +221,6 @@ export function runStructuredConversationCommand(
                 { kind: 'status', text: 'Compaction completion is unconfirmed.' },
                 { fence: ctx.fence }
               )
-              ctx.publish()
               throw cause
             }
             await ctx.journal.appendItem(
@@ -231,7 +228,6 @@ export function runStructuredConversationCommand(
               { kind: 'status', text: error ?? 'Conversation compacted.' },
               { fence: ctx.fence }
             )
-            ctx.publish()
           }
           const completed = {
             ...prepared,