fix(native-chat): validate restart continuation at provider dispatch

src/main/claude/claude-structured-dispatch-boundary.test.ts

@@ -0,0 +1,27 @@
+import { expect, it, vi } from 'vitest'
+import { AgentSessionPreDispatchError } from '../native-chat/agent-session-wire/structured-agent-session-operation-settlement'
+import { acquired, fakeClaude, USER_MESSAGE } from './claude-structured-session-test-support'
+
+it('does not enqueue a continuation refused at the provider dispatch boundary', async () => {
+  const claude = fakeClaude()
+  const settled = vi.fn()
+  const adapter = await acquired(claude, {}, [], settled)
+  const refusal = new AgentSessionPreDispatchError('agent_session_restart_work_superseded')
+  try {
+    await expect(
+      adapter.dispatch({
+        sessionId: 'session-1',
+        clientMessageId: 'continuation',
+        body: USER_MESSAGE,
+        fence: 7,
+        beforeDispatch: async () => {
+          throw refusal
+        }
+      })
+    ).rejects.toBe(refusal)
+    expect(claude.connections[0]?.sent).toEqual([])
+    expect(settled).not.toHaveBeenCalled()
+  } finally {
+    await adapter.closeAll()
+  }
+})

src/main/claude/claude-structured-dispatch.ts

@@ -279,7 +279,8 @@ export function retireClaudeDispatchWaiters(session: ClaudeSession): void {
 
 export async function dispatchClaudeTurn(
   session: ClaudeSession,
-  input: { clientMessageId?: string; body: AgentJournalMessageItem; requestedAt?: number }
+  input: { clientMessageId?: string; body: AgentJournalMessageItem; requestedAt?: number },
+  beforeDispatch?: () => Promise<void>
 ): Promise<AgentSessionDispatchOutcome> {
   let content: unknown[]
   try {
@@ -287,6 +288,7 @@ export async function dispatchClaudeTurn(
   } catch (error) {
     return { state: 'rejected', reason: (error as Error).message }
   }
+  await beforeDispatch?.()
   if (session.dispatchWaiters.length >= MAX_ACTIVE_DISPATCH_WAITERS) {
     return { state: 'rejected', reason: DISPATCH_REJECTED_QUEUE_FULL }
   }

src/main/claude/claude-structured-session-adapter.ts

@@ -237,7 +237,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
   }
 
   dispatch: StructuredAgentSessionAdapter['dispatch'] = (input) =>
-    dispatchClaudeTurn(this.session(input.sessionId), input)
+    dispatchClaudeTurn(this.session(input.sessionId), input, input.beforeDispatch)
 
   compact: NonNullable<StructuredAgentSessionAdapter['compact']> = (input) =>
     compactClaudeSession(this.session(input.sessionId), this.compactions, input)

src/main/codex/codex-structured-dispatch-boundary.test.ts

@@ -0,0 +1,46 @@
+import { expect, it, vi } from 'vitest'
+import { AgentSessionPreDispatchError } from '../native-chat/agent-session-wire/structured-agent-session-operation-settlement'
+import {
+  acquiredCodexAdapter,
+  CODEX_TEST_USER_MESSAGE,
+  fakeCodexAppServer
+} from './codex-structured-dispatch-test-support'
+
+it('checks continuation authority after process capture and before writing turn/start', async () => {
+  const capturing = Promise.withResolvers<void>()
+  const captured = Promise.withResolvers<null>()
+  const codex = fakeCodexAppServer()
+  const adapter = await acquiredCodexAdapter({
+    codex,
+    settlements: [],
+    captureTurnProcesses: () => {
+      capturing.resolve()
+      return captured.promise
+    }
+  })
+  let superseded = false
+  const beforeDispatch = vi.fn(async () => {
+    if (superseded) {
+      throw new AgentSessionPreDispatchError('agent_session_restart_work_superseded')
+    }
+  })
+  try {
+    const result = adapter.dispatch({
+      sessionId: 'session-1',
+      clientMessageId: 'continuation',
+      body: CODEX_TEST_USER_MESSAGE,
+      fence: 7,
+      beforeDispatch
+    })
+    const verdict = result.catch((error: unknown) => error)
+    await capturing.promise
+    superseded = true
+    captured.resolve(null)
+    expect(await verdict).toBeInstanceOf(AgentSessionPreDispatchError)
+    expect(beforeDispatch).toHaveBeenCalledOnce()
+    expect(codex.connections[0]?.calls.some((call) => call.method === 'turn/start')).toBe(false)
+  } finally {
+    captured.resolve(null)
+    await adapter.closeAll()
+  }
+})

src/main/codex/codex-structured-dispatch-test-support.ts

@@ -11,6 +11,7 @@ import type {
 } from './codex-app-server-connection'
 import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
 import { CodexStructuredSessionAdapter } from './codex-structured-session-adapter'
+import type { CodexStructuredSessionAdapterDeps } from './codex-structured-session-state'
 
 export const CODEX_TEST_THREAD_ID = 'thread-abc'
 
@@ -84,6 +85,7 @@ export async function acquiredCodexAdapter(input: {
   codex: ReturnType<typeof fakeCodexAppServer>
   settlements: LateSettlement[]
   sink?: StructuredAgentSessionEventSink
+  captureTurnProcesses?: CodexStructuredSessionAdapterDeps['captureTurnProcesses']
 }): Promise<CodexStructuredSessionAdapter> {
   const adapter = new CodexStructuredSessionAdapter({
     resolveLaunch: async () => ({
@@ -95,7 +97,7 @@ export async function acquiredCodexAdapter(input: {
     }),
     openConnection: input.codex.openConnection,
     readProcessStartTime: async () => 1_700_000_000_000,
-    captureTurnProcesses: async () => null,
+    captureTurnProcesses: input.captureTurnProcesses ?? (async () => null),
     now: () => 1_700_000_000_500,
     onDispatchSettledLate: (settlement) => input.settlements.push(settlement)
   })

src/main/codex/codex-structured-session-adapter.ts

@@ -211,11 +211,13 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
     body: AgentJournalMessageItem
     fence: number
     requestedAt?: number
+    beforeDispatch?: () => Promise<void>
   }): Promise<AgentSessionDispatchOutcome> {
     const session = this.session(input.sessionId)
     session.dispatchPending = true
     try {
       await this.turnCancellation.captureBaseline(session)
+      await input.beforeDispatch?.()
       return await dispatchCodexTurn(session, input, this.deps.requestTimeoutMs)
     } finally {
       session.dispatchPending = false

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

@@ -173,6 +173,8 @@ export type StructuredAgentSessionAdapter = {
     /** Host clock on the submission row this send came from; the origin the turn
      *  it opens records as `requestedAt`. */
     requestedAt?: number
+    /** Revalidate after preparation, immediately before writing to the provider. */
+    beforeDispatch?: () => Promise<void>
   }): Promise<AgentSessionDispatchOutcome>
   rewindSupport?(sessionId: string): AgentSessionRewindSupport
   recoverRewind?(input: {

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

@@ -75,7 +75,10 @@ function adapter(): StructuredAgentSessionAdapter {
   return {
     acquire,
     releaseAcquisition,
-    dispatch,
+    dispatch: async (input) => {
+      await input.beforeDispatch?.()
+      return dispatch(input)
+    },
     cancelTurn,
     answerPrompt,
     setOption

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

@@ -149,7 +149,7 @@ export class StructuredAgentSessionHost {
       onBarrierError: (sessionId, error) => deps.onEventSinkError?.({ sessionId, error })
     })
     this.restartResume = createStructuredAgentSessionRestartResume(deps, this.sessions, {
-      ...structuredAgentSessionRestartResumeSurfaces(this, this.now, deps.onEventSinkError),
+      ...structuredAgentSessionRestartResumeSurfaces(this, this.now),
       publish: this.subscribers.publish.bind(this.subscribers)
     })
     this.runtimeState.startLeaseRenewal()

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

@@ -83,7 +83,7 @@ it('returns a pre-dispatch refusal without waiting on redundant uncertainty pers
     await vi.advanceTimersByTimeAsync(AGENT_SESSION_ADMISSION_BARRIER_TIMEOUT_MS)
     expect(returned).toBe(true)
     expect(writes).toHaveBeenCalledOnce()
-    expect(ctx.adapter.dispatch).not.toHaveBeenCalled()
+    expect(hostTestState().dispatch).not.toHaveBeenCalled()
     expect(vi.getTimerCount()).toBe(0)
   } finally {
     stalled.resolve()
@@ -168,13 +168,13 @@ it.each(['stalled', 'failed', 'stalled-with-refusal-write'] as const)(
     }
     expect(await result).toBeInstanceOf(AgentSessionPreDispatchError)
     expect(beforeRun).not.toHaveBeenCalled()
-    expect(ctx.adapter.dispatch).not.toHaveBeenCalled()
+    expect(hostTestState().dispatch).not.toHaveBeenCalled()
     expect(ctx.journal.submissions()[0]?.dispatchState).toBe(
       barrier === 'stalled-with-refusal-write' ? 'pending' : 'rejected'
     )
     pending.resolve()
     await vi.advanceTimersByTimeAsync(0)
-    expect(ctx.adapter.dispatch).not.toHaveBeenCalled()
+    expect(hostTestState().dispatch).not.toHaveBeenCalled()
     expect(vi.getTimerCount()).toBe(0)
   }
 )

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

@@ -108,6 +108,31 @@ it('publishes continuation attribution to the subscribed chat without another pr
   }
 })
 
+it('reports a failed attribution note without an installed error sink or private details', async () => {
+  const { host } = await interruptedRestart()
+  const append = AgentSessionJournal.prototype.appendItem
+  const warning = vi.spyOn(console, 'warn').mockImplementation(() => {})
+  const write = vi.spyOn(AgentSessionJournal.prototype, 'appendItem').mockImplementation(function (
+    this: AgentSessionJournal,
+    ...args
+  ) {
+    if (args[1].kind === 'status' && args[1].text === AGENT_SESSION_RESTART_CONTINUATION_NOTE) {
+      return Promise.reject(new Error('private recovery payload at /private/account/session.json'))
+    }
+    return append.apply(this, args)
+  })
+  try {
+    const result = await host.restartResume.continueAfterRestart([SESSION], 'modal')
+    expect(result.continued).toMatchObject([{ outcome: 'continued' }])
+    expect(warning).toHaveBeenCalledExactlyOnceWith(
+      '[structured-agent-session] restart continuation attribution failed'
+    )
+  } finally {
+    write.mockRestore()
+    warning.mockRestore()
+  }
+})
+
 it.each(['turn', 'submission'] as const)(
   'does not continue a marked %s after the user submits new work without a provider echo',
   async (work) => {

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

@@ -40,10 +40,7 @@ type RestartResumeHostBindings = {
 
 export function structuredAgentSessionRestartResumeSurfaces(
   host: RestartResumeHostBindings,
-  now: () => number,
-  /** The host's error sink. Absent on a host built without one, which only means a failed note goes
-   *  unreported — never that the continuation fails. */
-  reportError?: (input: { sessionId: string; error: Error }) => void
+  now: () => number
 ): Omit<StructuredAgentSessionRestartResumeSurfaces, 'publish'> {
   return {
     revealSession: host.revealSession,
@@ -52,11 +49,8 @@ export function structuredAgentSessionRestartResumeSurfaces(
     send: (params) =>
       host.send({ callerKey: STRUCTURED_AGENT_SESSION_RESTART_CONTINUATION_CALLER }, params),
     awaitSendSettlement: host.waitForSendSettlement,
-    onNoteFailed: (sessionId, error) =>
-      reportError?.({
-        sessionId,
-        error: error instanceof Error ? error : new Error(String(error))
-      }),
+    onNoteFailed: () =>
+      console.warn('[structured-agent-session] restart continuation attribution failed'),
     now
   }
 }

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

@@ -65,27 +65,36 @@ async function dispatchSafely(
   body: AgentJournalMessageItem,
   requestedAt: number | undefined
 ): Promise<AgentSessionDispatchOutcome> {
-  if (ctx.beforeDispatch) {
-    const ready = await withTimeout(
-      ctx.flushStreamedEvents().then(() => true),
-      AGENT_SESSION_ADMISSION_BARRIER_TIMEOUT_MS,
-      false
-    )
-    // A fixed barrier may finish while newer events are still queued; never dispatch past them.
-    if (!ready || ctx.hasPendingStreamedEvents?.()) {
-      throw new AgentSessionPreDispatchError('agent_session_admission_evidence_unavailable')
-    }
-    ctx.beforeDispatch()
-  }
   try {
     return await ctx.adapter.dispatch({
       sessionId: ctx.sessionId,
       clientMessageId,
       body,
       fence: ctx.fence,
+      ...(ctx.beforeDispatch
+        ? {
+            beforeDispatch: async () => {
+              const ready = await withTimeout(
+                ctx.flushStreamedEvents().then(() => true),
+                AGENT_SESSION_ADMISSION_BARRIER_TIMEOUT_MS,
+                false
+              )
+              // A drained barrier may be followed by newer accepted events.
+              if (!ready || ctx.hasPendingStreamedEvents?.()) {
+                throw new AgentSessionPreDispatchError(
+                  'agent_session_admission_evidence_unavailable'
+                )
+              }
+              ctx.beforeDispatch?.()
+            }
+          }
+        : {}),
       ...(requestedAt === undefined ? {} : { requestedAt })
     })
   } catch (error) {
+    if (error instanceof AgentSessionPreDispatchError) {
+      throw error
+    }
     return { state: 'unknown', reason: error instanceof Error ? error.message : String(error) }
   }
 }