diff --git a/src/main/claude/claude-structured-dispatch-test-support.ts b/src/main/claude/claude-structured-dispatch-test-support.ts index b6084026818..83971a38d13 100644 --- a/src/main/claude/claude-structured-dispatch-test-support.ts +++ b/src/main/claude/claude-structured-dispatch-test-support.ts @@ -20,7 +20,6 @@ export function sessionFor(send: Mock = vi.fn().mockResolvedValue(undefined)): C backgroundTasks: new ClaudeBackgroundTaskTracker(), commands: new ClaudeSlashCommandCatalog(), dispatchSequence: 0, - lastAdmittedDispatchSequence: undefined, optionMutationSequence: 0, options: new Map(), reportedOptions: {}, diff --git a/src/main/claude/claude-structured-dispatch.ts b/src/main/claude/claude-structured-dispatch.ts index f5ae9399460..a69508ce071 100644 --- a/src/main/claude/claude-structured-dispatch.ts +++ b/src/main/claude/claude-structured-dispatch.ts @@ -151,9 +151,6 @@ function settleWaiter( providerIdentity: { provider: 'claude', sessionId: session.providerSessionId, uuid } }) } - if (waiter.dispatchSequence === session.dispatchSequence) { - session.lastAdmittedDispatchSequence = waiter.dispatchSequence - } } function forgetRetiredWaiter(session: ClaudeSession, waiter: ClaudeDispatchWaiter): void { @@ -182,9 +179,6 @@ function recoverLateIdentity( providerIdentity: { provider: 'claude', sessionId: session.providerSessionId, uuid } }) } - if (waiter.dispatchSequence === session.dispatchSequence) { - session.lastAdmittedDispatchSequence = waiter.dispatchSequence - } return isUserReplay && waiter.dispatchSequence === session.dispatchSequence } @@ -316,9 +310,6 @@ export async function dispatchClaudeTurn( if (waiter.settledUuid) { const uuid = await replayed if (uuid) { - if (waiter.dispatchSequence === session.dispatchSequence) { - session.lastAdmittedDispatchSequence = waiter.dispatchSequence - } return { state: 'accepted', providerIdentity: { provider: 'claude', sessionId: session.providerSessionId, uuid } diff --git a/src/main/claude/claude-structured-options.test.ts b/src/main/claude/claude-structured-options.test.ts index b4198dbc8e2..9b962fc33d7 100644 --- a/src/main/claude/claude-structured-options.test.ts +++ b/src/main/claude/claude-structured-options.test.ts @@ -31,7 +31,6 @@ function sessionFor(setModel: ClaudeSession['connection']['setModel']): ClaudeSe backgroundTasks: new ClaudeBackgroundTaskTracker(), commands: new ClaudeSlashCommandCatalog(), dispatchSequence: 0, - lastAdmittedDispatchSequence: undefined, optionMutationSequence: 0, options: new Map(), reportedOptions: {}, diff --git a/src/main/claude/claude-structured-prompt-ownership.ts b/src/main/claude/claude-structured-prompt-ownership.ts index 051e5bf4171..07236e7c960 100644 --- a/src/main/claude/claude-structured-prompt-ownership.ts +++ b/src/main/claude/claude-structured-prompt-ownership.ts @@ -83,10 +83,20 @@ export async function cancelClaudeStructuredTurn(input: { ? session.dispatchSequence === 0 : currentTurnId === request.turnId } + // The host supplies the durable latest submission; direct adapter callers fall back to + // the current in-memory waiter so an unknown dispatch remains fenced without a latch. const dispatchAdmissionIsCurrent = (): boolean => session.dispatchSequence === 0 || - session.lastAdmittedDispatchSequence === session.dispatchSequence || - supportsClaudeQueuedInterruptCancellation(session) + (request.dispatchStatus + ? request.dispatchStatus.state === 'accepted' || + request.dispatchStatus.state === 'rejected' || + (request.dispatchStatus.state === 'unknown' && request.dispatchStatus.recovered) + : ![...session.dispatchWaiters, ...session.retiredDispatchWaiters].some( + (waiter) => waiter.dispatchSequence === session.dispatchSequence + )) + const dispatchAdmissionAllowsCancellation = (): boolean => + dispatchAdmissionIsCurrent() || + (Boolean(prompt) && supportsClaudeQueuedInterruptCancellation(session)) const isCurrent = (): boolean => sessions.get(request.sessionId) === session && session.fence === request.fence && @@ -94,9 +104,9 @@ export async function cancelClaudeStructuredTurn(input: { (claim && prompt ? ownsRequestedTurn() && session.prompts.ownsBoundClaim(claim, prompt.itemId, request.turnId) && - dispatchAdmissionIsCurrent() + dispatchAdmissionAllowsCancellation() : compactions.ownsTurn(request.sessionId, request.turnId) || - (ownsRequestedTurn() && dispatchAdmissionIsCurrent())) + (ownsRequestedTurn() && dispatchAdmissionAllowsCancellation())) let interruptConfirmed = false try { const result = await cancelClaudeTurn( diff --git a/src/main/claude/claude-structured-session-publication.ts b/src/main/claude/claude-structured-session-publication.ts index 9b15691eb9b..395335332e7 100644 --- a/src/main/claude/claude-structured-session-publication.ts +++ b/src/main/claude/claude-structured-session-publication.ts @@ -61,7 +61,6 @@ export function createClaudeSessionPublication(input: { backgroundTasks: new ClaudeBackgroundTaskTracker(), commands: new ClaudeSlashCommandCatalog(input.init.message, input.initialization), dispatchSequence: 0, - lastAdmittedDispatchSequence: undefined, optionMutationSequence: 0, options: new Map(input.options), capabilities: input.capabilities, diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index bc7e972d05c..4c4808b64d8 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -153,8 +153,6 @@ export type ClaudeSession = { commands: ClaudeSlashCommandCatalog /** Monotonic fence advanced when a dispatch starts, including unresolved dispatches. */ dispatchSequence: number - /** Dispatch sequence whose provider replay admitted the most recent send. */ - lastAdmittedDispatchSequence?: number /** Fences overlapping option writes so a late completion cannot restore stale state. */ optionMutationSequence: number /** Shared durable-close write; a failed write clears this for a retry. */ diff --git a/src/main/claude/claude-turn-ownership.test.ts b/src/main/claude/claude-turn-ownership.test.ts index 0dd90e9316a..80dd3998ea2 100644 --- a/src/main/claude/claude-turn-ownership.test.ts +++ b/src/main/claude/claude-turn-ownership.test.ts @@ -163,6 +163,32 @@ describe('Claude turn ownership', () => { expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) }) + it('keeps ordinary Stop fenced when the latest journal submission is unresolved', async () => { + const claude = fakeClaude({ + replayUuid: 'echo-turn', + capabilities: ['interrupt_cancel_queued_v1'] + }) + const { adapter, bodies, connection } = await acquiredWithJournal(claude) + + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + expect(runningTurnId(bodies)).toBe('echo-turn') + + await expect( + adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'echo-turn', + fence: 7, + dispatchStatus: { state: 'unknown', recovered: false } + }) + ).resolves.toEqual({ cancelled: false }) + expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) + }) + it('still stops an echo-opened turn', async () => { const claude = fakeClaude({ replayUuid: 'echo-turn' }) const { adapter, bodies, connection } = await acquiredWithJournal(claude) diff --git a/src/main/native-chat/agent-session-journal/journal-dispatch-observation.test.ts b/src/main/native-chat/agent-session-journal/journal-dispatch-observation.test.ts new file mode 100644 index 00000000000..4a554d82662 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-dispatch-observation.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, it } from 'vitest' +import { latestJournalDispatchObservation } from './journal-dispatch-observation' + +describe('latestJournalDispatchObservation', () => { + it('uses the newest submission in the requested fence', () => { + const journal = { + submissions: () => [ + { + fence: 7, + dispatchState: 'unknown' as const, + recovered: true as const, + submittedAt: 1 + }, + { + fence: 8, + dispatchState: 'pending' as const, + submittedAt: 2 + }, + { + fence: 7, + dispatchState: 'accepted' as const, + submittedAt: 1 + } + ] + } + + expect(latestJournalDispatchObservation(journal, 7)).toEqual({ + state: 'accepted', + recovered: false + }) + expect(latestJournalDispatchObservation(journal, 8)).toEqual({ + state: 'pending', + recovered: false + }) + }) + + it('returns no observation when the fence has no submission', () => { + expect(latestJournalDispatchObservation({ submissions: () => [] }, 7)).toBeNull() + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-dispatch-observation.ts b/src/main/native-chat/agent-session-journal/journal-dispatch-observation.ts new file mode 100644 index 00000000000..b202c2e71a2 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-dispatch-observation.ts @@ -0,0 +1,27 @@ +import type { AgentJournalDispatchState } from '../../../shared/agent-session-journal-types' + +export type AgentJournalDispatchObservation = { + state: AgentJournalDispatchState + recovered: boolean +} + +/** Returns the latest write-ahead submission for the execution fence. */ +export function latestJournalDispatchObservation( + journal: { + submissions: () => readonly { + fence: number + dispatchState: AgentJournalDispatchState + recovered?: true + submittedAt: number + }[] + }, + fence: number +): AgentJournalDispatchObservation | null { + const latest = journal.submissions().reduce<{ + fence: number + dispatchState: AgentJournalDispatchState + recovered?: true + submittedAt: number + } | null>((current, submission) => (submission.fence === fence && (current === null || submission.submittedAt >= current.submittedAt) ? submission : current), null) + return latest ? { state: latest.dispatchState, recovered: latest.recovered === true } : null +} 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 81ef7f79062..779f34cadb4 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 @@ -14,6 +14,7 @@ import type { AgentJournalItemIdentity, AgentJournalItemBody, AgentJournalMessageItem, + AgentJournalDispatchState, AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' import type { AgentSessionProviderHandleLink } from '../../../shared/agent-session-provider-handle' @@ -202,6 +203,8 @@ export type StructuredAgentSessionAdapter = { turnId: string fence: number prompt?: { itemId: string } + /** Latest journal submission for this fence, when the host has one. */ + dispatchStatus?: { state: AgentJournalDispatchState; recovered: boolean } | null }): Promise<{ cancelled: boolean }> stopBackgroundTasks?(input: { sessionId: string diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts index 9a3266541be..5c798d8a275 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts @@ -16,6 +16,7 @@ import type { AgentSessionSubscribers } from './structured-agent-session-subscri import { StructuredTuiTranscriptCatchup } from './structured-tui-transcript-catchup' import { adapterSupportsCreateIfDeclared } from './structured-agent-session-provider-support' import { retryLoadedStructuredAgentSessionSettlement } from './structured-agent-session-settlement-retry' +import { latestJournalDispatchObservation } from '../agent-session-journal/journal-dispatch-observation' type HostHandoffAccess = { session: (sessionId: string) => StructuredAgentSessionHostSession @@ -101,8 +102,17 @@ export function createStructuredAgentSessionHostHandoff( }, acknowledgeNativeRelease: (sessionId) => deps.adapter.acknowledgeSessionRelease?.(sessionId), acquireNative: (input) => acquireNativeHandoffOwner(deps, host, input), - acquireNativeStop: async (sessionId, turnId, fence) => - (await deps.adapter.cancelTurn({ sessionId, turnId, fence })).cancelled, + acquireNativeStop: async (sessionId, turnId, fence) => { + const session = host.session(sessionId) + return ( + await deps.adapter.cancelTurn({ + sessionId, + turnId, + fence, + dispatchStatus: latestJournalDispatchObservation(session.journal, fence) + }) + ).cancelled + }, importTuiHistory: (input) => importTuiHistory(deps, host, input), retryPendingSettlement: (sessionId) => retryLoadedStructuredAgentSessionSettlement({ 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 c666047e1e7..a81cfef0dca 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 @@ -17,6 +17,7 @@ import type { } from '../../../shared/agent-session-wire' import { DISPATCH_DOUBT_PERSISTENCE_FAILED } from '../agent-session-journal/journal-dispatch-doubt-reasons' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { latestJournalDispatchObservation } from '../agent-session-journal/journal-dispatch-observation' import type { AgentSessionDispatchOutcome, StructuredAgentSessionAdapter @@ -214,6 +215,7 @@ export async function performCancel( sessionId: ctx.sessionId, turnId: input.turnId, fence: ctx.fence, + dispatchStatus: latestJournalDispatchObservation(ctx.journal, ctx.fence), ...(input.prompt ? { prompt: { itemId: input.prompt.itemId } } : {}) }) ).cancelled