From b2f72902e8ada58bdf5816f168dc5c7dedfa40bb Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 17 Sep 2026 00:07:28 -0700 Subject: [PATCH] fix(native-chat): validate restart continuation at provider dispatch --- ...laude-structured-dispatch-boundary.test.ts | 27 +++++++++++ src/main/claude/claude-structured-dispatch.ts | 4 +- .../claude-structured-session-adapter.ts | 2 +- ...codex-structured-dispatch-boundary.test.ts | 46 +++++++++++++++++++ .../codex-structured-dispatch-test-support.ts | 4 +- .../codex/codex-structured-session-adapter.ts | 2 + .../structured-agent-session-adapter.ts | 2 + ...uctured-agent-session-host-test-harness.ts | 5 +- .../structured-agent-session-host.ts | 2 +- ...agent-session-operation-settlement.test.ts | 6 +-- ...ed-agent-session-restart-ownership.test.ts | 25 ++++++++++ ...red-agent-session-restart-resume-wiring.ts | 12 ++--- .../structured-agent-session-turns.ts | 33 ++++++++----- 13 files changed, 141 insertions(+), 29 deletions(-) create mode 100644 src/main/claude/claude-structured-dispatch-boundary.test.ts create mode 100644 src/main/codex/codex-structured-dispatch-boundary.test.ts diff --git a/src/main/claude/claude-structured-dispatch-boundary.test.ts b/src/main/claude/claude-structured-dispatch-boundary.test.ts new file mode 100644 index 00000000000..a8fdb572e18 --- /dev/null +++ b/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() + } +}) diff --git a/src/main/claude/claude-structured-dispatch.ts b/src/main/claude/claude-structured-dispatch.ts index b5e676056b4..c04c3c82ef2 100644 --- a/src/main/claude/claude-structured-dispatch.ts +++ b/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 ): Promise { 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 } } diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 9357e9635f6..6daff234ea3 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/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 = (input) => compactClaudeSession(this.session(input.sessionId), this.compactions, input) diff --git a/src/main/codex/codex-structured-dispatch-boundary.test.ts b/src/main/codex/codex-structured-dispatch-boundary.test.ts new file mode 100644 index 00000000000..5d4f0912bcc --- /dev/null +++ b/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() + const captured = Promise.withResolvers() + 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() + } +}) diff --git a/src/main/codex/codex-structured-dispatch-test-support.ts b/src/main/codex/codex-structured-dispatch-test-support.ts index 5519ffdfdb8..42efd57bab4 100644 --- a/src/main/codex/codex-structured-dispatch-test-support.ts +++ b/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 settlements: LateSettlement[] sink?: StructuredAgentSessionEventSink + captureTurnProcesses?: CodexStructuredSessionAdapterDeps['captureTurnProcesses'] }): Promise { 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) }) diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index fff23b05eb1..993678b82f2 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/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 }): Promise { 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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 5a816d78252..ac73526fd1c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/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 }): Promise rewindSupport?(sessionId: string): AgentSessionRewindSupport recoverRewind?(input: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts index bed2a2d747e..e0ac9c00d88 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts +++ b/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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index a2c52f37b4c..781745ec26d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/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() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts index e2d5799a07d..bc4f360e79b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts +++ b/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) } ) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-ownership.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-ownership.test.ts index be5cc5210f8..e3108e556ea 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-ownership.test.ts +++ b/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) => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts index 954018e5bd7..4494392fb48 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts +++ b/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 { 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 } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts index 5fb91f199ee..4c9cc2213d5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts +++ b/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 { - 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) } } }