fix(native-chat): every journal append reaches the chats that are open

A journal write and its delivery to open readers were two calls, and some
writers made only the first. A failed start whose lease could not be handed
back, a provider revision with no frame behind it, and eviction's settlement
were all journaled without reaching an open chat.

A journal handle now reports every durable change, and the host's session map
binds that report to the session's readers when the handle is set. Writers no
longer publish what they append; the per-writer publish calls are deleted.

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,188 @@
+// 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
+    }
+  }
+  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'])
+  })
+})

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

@@ -0,0 +1,126 @@
+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 { 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('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.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-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-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,