From e8b6f424d14c5a37d8bc15f9fed6ad19137eb359 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 03:28:55 -0700 Subject: [PATCH 01/10] fix(claude): stop a named running turn at once when a follow-up is queued A Stop that named the running Claude turn waited up to 3 s whenever a later send's handover was still unresolved, for example a follow-up Claude had queued. The phone app and older desktop clients always name the turn. Claude's interrupt is session-scoped, so a Stop naming the live turn now takes the same path as a Stop naming no turn: it interrupts at once, and the queue sweep settles the follow-up as withdrawn. A Stop naming a turn that is no longer live is still refused. With that case gone, the admission wait could only delay a refusal, so it is deleted rather than left as a second gate. --- .../claude-structured-prompt-ownership.ts | 96 ++---- .../claude-structured-queued-stop.test.ts | 114 +++++-- ...e-structured-session-adapter-turns.test.ts | 68 ++-- src/main/claude/claude-turn-ownership.test.ts | 302 ++++++------------ .../structured-agent-session-adapter.ts | 4 +- 5 files changed, 261 insertions(+), 323 deletions(-) diff --git a/src/main/claude/claude-structured-prompt-ownership.ts b/src/main/claude/claude-structured-prompt-ownership.ts index 01af8ec9fc0b..e53d8156c5c6 100644 --- a/src/main/claude/claude-structured-prompt-ownership.ts +++ b/src/main/claude/claude-structured-prompt-ownership.ts @@ -16,11 +16,6 @@ import type { ClaudeSession } from './claude-structured-session-state' import type { ClaudePendingPrompt } from './claude-prompt-registry' import type { PermissionResult } from '@anthropic-ai/claude-agent-sdk' -/** Conservative user-facing window: below the 30s control deadline, trading - * residual slow-pump risk for ensuring delivery bookkeeping cannot block Stop indefinitely. */ -export const CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS = 3_000 -const CLAUDE_DISPATCH_ADMISSION_POLL_MS = 50 - type CancelInput = Parameters[0] type AnswerInput = Parameters[0] @@ -56,45 +51,16 @@ function requireSession(sessions: Map, sessionId: string) return session } -function waitForClaudeDispatchAdmission( - admitted: () => boolean, - timeoutMs = CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS -): Promise { - return new Promise((resolve) => { - let settled = false - let deadline: ReturnType | null = null - let poll: ReturnType | null = null - const finish = (value: boolean): void => { - if (settled) { - return - } - settled = true - if (deadline) { - clearTimeout(deadline) - } - if (poll) { - clearInterval(poll) - } - resolve(value) - } - const check = (): void => { - if (admitted()) { - finish(true) - } - } - deadline = setTimeout(() => finish(false), timeoutMs) - poll = setInterval(check, CLAUDE_DISPATCH_ADMISSION_POLL_MS) - check() - deadline.unref?.() - poll.unref?.() - }) +/** The published journal's turn while it has one, else the in-memory turn (see the owner check). */ +function claudeLiveTurnId(session: ClaudeSession, request: CancelInput): string | null { + return request.resolveLiveTurnId?.() ?? session.translator?.currentTurnId ?? null } /** - * A Stop that names no turn: the conversation asked to stop whatever this child has in flight. - * Claude's interrupt is session-scoped, so there is no turn identity to check — only that this is - * still the child the host judged, and that it has a turn open or a written message whose turn has - * not opened yet (the gap before its echo, which no client can name). + * A Stop that names no turn, or names the live one: the conversation asked to stop whatever this + * child has in flight. Claude's interrupt is session-scoped, so there is no turn identity to check — + * only that this is still the child the host judged, and that it has a turn open or a written + * message whose turn has not opened yet (the gap before its echo, which no client can name). */ function cancelClaudeConversation( session: ClaudeSession, @@ -108,8 +74,7 @@ function cancelClaudeConversation( sessions.get(request.sessionId) === session && session.fence === request.fence && session.acquisitionGeneration === acquisitionGeneration && - ((request.resolveLiveTurnId?.() ?? session.translator?.currentTurnId ?? null) !== null || - session.dispatchWaiters.length > 0) + (claudeLiveTurnId(session, request) !== null || session.dispatchWaiters.length > 0) return cancelClaudeTurn(session, timeoutMs, isCurrent, onDispatchSettledLate) } @@ -130,10 +95,24 @@ export async function cancelClaudeStructuredTurn(input: { return { cancelled: false } } const requestedTurnId = request.turnId + // Naming the live turn asks for exactly what the conversation Stop interrupts, so a follow-up's + // pending handover is no reason to hold it; the queue sweep settles that follow-up. + if ( + !prompt && + (requestedTurnId === undefined || + (!compactions.ownsTurn(request.sessionId, requestedTurnId) && + claudeLiveTurnId(session, request) === requestedTurnId)) + ) { + return cancelClaudeConversation( + session, + sessions, + request, + timeoutMs, + input.onDispatchSettledLate + ) + } if (requestedTurnId === undefined) { - return prompt - ? { cancelled: false } - : cancelClaudeConversation(session, sessions, request, timeoutMs, input.onDispatchSettledLate) + return { cancelled: false } } if (prompt && session.fence !== request.fence) { return { cancelled: false } @@ -153,7 +132,7 @@ export async function cancelClaudeStructuredTurn(input: { // the in-memory turn there keeps Stop from being gated on bookkeeping. No live turn either way // means nothing has published an identity this request can contradict. const ownsRequestedTurn = (): boolean => { - const liveTurnId = request.resolveLiveTurnId?.() ?? session.translator?.currentTurnId ?? null + const liveTurnId = claudeLiveTurnId(session, request) return liveTurnId === null ? session.dispatchSequence === 0 : liveTurnId === requestedTurnId } // The host supplies the durable latest submission; direct adapter callers fall back to @@ -168,26 +147,11 @@ export async function cancelClaudeStructuredTurn(input: { (waiter) => waiter.dispatchSequence === session.dispatchSequence ) // Prompt cancellation has a separate callback-settlement contract, so only a provider with - // cancelQueued can release its uncertain queued send. Ordinary Stop gets a bounded escape below. + // cancelQueued can release its uncertain queued send. const dispatchAdmissionAllowsCancellation = (): boolean => dispatchAdmissionIsCurrent() || (Boolean(prompt) && supportsClaudeQueuedInterruptCancellation(session)) const compactionOwnsTurn = (): boolean => compactions.ownsTurn(request.sessionId, requestedTurnId) - const currentDispatchHasRetiredWaiter = (): boolean => - session.retiredDispatchWaiters.some( - (waiter) => waiter.dispatchSequence === session.dispatchSequence - ) - let dispatchAdmissionExpired = false - if ( - !prompt && - !compactionOwnsTurn() && - !dispatchAdmissionAllowsCancellation() && - (request.dispatchStatus !== undefined || currentDispatchHasRetiredWaiter()) - ) { - dispatchAdmissionExpired = !(await waitForClaudeDispatchAdmission( - dispatchAdmissionAllowsCancellation - )) - } const isCurrent = (): boolean => sessions.get(request.sessionId) === session && session.fence === request.fence && @@ -195,10 +159,8 @@ export async function cancelClaudeStructuredTurn(input: { (claim && prompt ? ownsRequestedTurn() && session.prompts.ownsBoundClaim(claim, prompt.itemId, requestedTurnId) && - (dispatchAdmissionAllowsCancellation() || dispatchAdmissionExpired) - : compactionOwnsTurn() || - (ownsRequestedTurn() && - (dispatchAdmissionAllowsCancellation() || dispatchAdmissionExpired))) + dispatchAdmissionAllowsCancellation() + : compactionOwnsTurn() || (ownsRequestedTurn() && dispatchAdmissionAllowsCancellation())) let interruptConfirmed = false try { const result = await cancelClaudeTurn( diff --git a/src/main/claude/claude-structured-queued-stop.test.ts b/src/main/claude/claude-structured-queued-stop.test.ts index 2dbc99bb82be..f04fb298e4af 100644 --- a/src/main/claude/claude-structured-queued-stop.test.ts +++ b/src/main/claude/claude-structured-queued-stop.test.ts @@ -8,7 +8,6 @@ import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' import { DISPATCH_REJECTED_CANCELLED } from '../../shared/structured-agent-session-dispatch-rejection' import type { ClaudeStructuredSessionAdapterDeps } from './claude-structured-session-state' -import { CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS } from './claude-structured-prompt-ownership' import { PROVIDER_SESSION_ID, USER_MESSAGE, @@ -92,40 +91,60 @@ function runningTurnId(bodies: Map): string { throw new Error('expected a running turn') } -/** Sends A and opens its turn, queues B behind it, then Stops A as the host does. */ -async function stopWithQueuedFollowUp( - cli: ReturnType, - openTurn: (connection: FakeConnection) => void -): Promise<{ interrupt: unknown; settlements: Settlement[] }> { - const settlements: Settlement[] = [] - const { adapter, bodies, connection } = await acquired(cli.claude, settlements) - const send = (clientMessageId: string) => - adapter.dispatch({ sessionId: 'session-1', clientMessageId, body: USER_MESSAGE, fence: 7 }) - await send('client-a') +/** Writes a send and opens its turn with the echo, which adopts the client uuid. */ +async function sendAndOpen( + adapter: ReturnType, + connection: FakeConnection, + clientMessageId: string, + openTurn: (connection: FakeConnection) => void = () => {} +): Promise { + await send(adapter, clientMessageId) openTurn(connection) - // Claude adopts the client uuid for the echo that opens the turn. connection.handlers.onMessage?.({ ...connection.sent.at(-1)!, uuid: connection.sent.at(-1)!.uuid }) - await send('client-b') - cli.queue(String(connection.sent.at(-1)!.uuid)) +} +function send(adapter: ReturnType, clientMessageId: string) { + return adapter.dispatch({ sessionId: 'session-1', clientMessageId, body: USER_MESSAGE, fence: 7 }) +} + +/** Stops `turnId` as the phone does, naming it, while the journal still reads the queued send as + * pending, and reads the outcome without moving the clock: that handover must hold no Stop. */ +async function namedStop(adapter: ReturnType, turnId: string): Promise { vi.useFakeTimers() try { - const stopped = adapter.cancelTurn({ - sessionId: 'session-1', - turnId: runningTurnId(bodies), - fence: 7, - // The host reads B's handover from the journal, and it is still pending. - dispatchStatus: { state: 'pending', recovered: false } - }) - await vi.advanceTimersByTimeAsync(CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS) - await expect(stopped).resolves.toEqual({ cancelled: true }) + let outcome: unknown = 'still waiting' + void adapter + .cancelTurn({ + sessionId: 'session-1', + turnId, + fence: 7, + dispatchStatus: { state: 'pending', recovered: false } + }) + .then((result) => (outcome = result)) + await vi.advanceTimersByTimeAsync(0) + return outcome } finally { vi.useRealTimers() } +} + +/** Sends A and opens its turn, queues B behind it, then Stops A by name as the phone does. */ +async function stopWithQueuedFollowUp( + cli: ReturnType, + openTurn: (connection: FakeConnection) => void +): Promise<{ stopped: unknown; interrupt: unknown; settlements: Settlement[] }> { + const settlements: Settlement[] = [] + const { adapter, bodies, connection } = await acquired(cli.claude, settlements) + await sendAndOpen(adapter, connection, 'client-a', openTurn) + await send(adapter, 'client-b') + cli.queue(String(connection.sent.at(-1)!.uuid)) + + const stopped = await namedStop(adapter, runningTurnId(bodies)) return { + stopped, interrupt: connection.calls.find((call) => call.subtype === 'interrupt')?.params, settlements: settlements.filter((settlement) => settlement.clientMessageId === 'client-b') } @@ -152,7 +171,11 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => { (connection) => connection.handlers.onMessage?.(turnInit(CAPABILITIES)) ) - expect(byInit).toEqual({ interrupt: { cancelQueued: true }, settlements: WITHDRAWN }) + expect(byInit).toEqual({ + stopped: { cancelled: true }, + interrupt: { cancelQueued: true }, + settlements: WITHDRAWN + }) expect(bySessionStart).toEqual(byInit) }) @@ -171,7 +194,11 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => { } ) - expect(result).toEqual({ interrupt: { cancelQueued: true }, settlements: WITHDRAWN }) + expect(result).toEqual({ + stopped: { cancelled: true }, + interrupt: { cancelQueued: true }, + settlements: WITHDRAWN + }) }) it('keeps what a turn advertised when the start is read after it', async () => { @@ -191,7 +218,11 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => { const result = await stopWithQueuedFollowUp(cli, () => {}) - expect(result).toEqual({ interrupt: { cancelQueued: true }, settlements: WITHDRAWN }) + expect(result).toEqual({ + stopped: { cancelled: true }, + interrupt: { cancelQueued: true }, + settlements: WITHDRAWN + }) }) it('settles a follow-up withdrawn one at a time on a CLI without cancel_queued', async () => { @@ -200,6 +231,35 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => { () => {} ) - expect(result).toEqual({ interrupt: {}, settlements: WITHDRAWN }) + expect(result).toEqual({ stopped: { cancelled: true }, interrupt: {}, settlements: WITHDRAWN }) + }) + + it('still refuses a Stop that names an older turn than the one running', async () => { + const settlements: Settlement[] = [] + const cli = claudeCli({ initProof: 'init', capabilities: CAPABILITIES }) + const { adapter, bodies, connection } = await acquired(cli.claude, settlements) + await sendAndOpen(adapter, connection, 'client-a') + const olderTurnId = runningTurnId(bodies) + connection.handlers.onMessage?.({ + type: 'result', + subtype: 'success', + uuid: 'result-a', + session_id: PROVIDER_SESSION_ID, + is_error: false, + terminal_reason: 'completed', + duration_ms: 12 + }) + await sendAndOpen(adapter, connection, 'client-b') + const liveTurnId = runningTurnId(bodies) + await send(adapter, 'client-c') + cli.queue(String(connection.sent.at(-1)!.uuid)) + + expect(liveTurnId).not.toBe(olderTurnId) + await expect(namedStop(adapter, olderTurnId)).resolves.toEqual({ cancelled: false }) + expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) + expect(runningTurnId(bodies)).toBe(liveTurnId) + expect(settlements.filter((settlement) => settlement.clientMessageId === 'client-c')).toEqual( + [] + ) }) }) diff --git a/src/main/claude/claude-structured-session-adapter-turns.test.ts b/src/main/claude/claude-structured-session-adapter-turns.test.ts index d410861464a9..3d0512d8b7cc 100644 --- a/src/main/claude/claude-structured-session-adapter-turns.test.ts +++ b/src/main/claude/claude-structured-session-adapter-turns.test.ts @@ -144,41 +144,49 @@ describe('ClaudeStructuredSessionAdapter turns and controls', () => { ) }) - it('does not cancel an acknowledged turn after a later dispatch is still unacknowledged', async () => { - const claude = fakeClaude({ replayUuids: ['turn-T', null] }) - const settled = vi.fn() - const adapter = await acquired(claude, {}, [], settled) + // Claude's interrupt is session-scoped, so naming the running turn asks for what a Stop naming + // no turn does; the later send's missing echo holds neither. + it.each([ + ['names the running turn', { turnId: 'turn-T' }], + ['names no turn', {}] + ])( + 'stops an acknowledged turn when the Stop %s, while a later send is unacknowledged', + async (_label, target) => { + const claude = fakeClaude({ replayUuids: ['turn-T', null] }) + const settled = vi.fn() + const adapter = await acquired(claude, {}, [], settled) - await expect( - adapter.dispatch({ + await expect( + adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-T', + body: USER_MESSAGE, + fence: 7 + }) + ).resolves.toEqual({ state: 'admitted' }) + expect(settled).toHaveBeenCalledWith({ sessionId: 'session-1', clientMessageId: 'client-T', - body: USER_MESSAGE, - fence: 7 - }) - ).resolves.toEqual({ state: 'admitted' }) - expect(settled).toHaveBeenCalledWith({ - sessionId: 'session-1', - clientMessageId: 'client-T', - providerIdentity: { provider: 'claude', sessionId: PROVIDER_SESSION_ID, uuid: 'turn-T' } - }) - await expect( - adapter.dispatch({ - sessionId: 'session-1', - clientMessageId: 'client-U', - body: USER_MESSAGE, - fence: 7 + providerIdentity: { provider: 'claude', sessionId: PROVIDER_SESSION_ID, uuid: 'turn-T' } }) - ).resolves.toEqual({ state: 'admitted' }) - expect(claude.connections[0].sent).toHaveLength(2) + await expect( + adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-U', + body: USER_MESSAGE, + fence: 7 + }) + ).resolves.toEqual({ state: 'admitted' }) + expect(claude.connections[0].sent).toHaveLength(2) - await expect( - adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-T', fence: 7 }) - ).resolves.toEqual({ cancelled: false }) - expect(claude.connections[0].calls.filter((call) => call.subtype === 'interrupt')).toHaveLength( - 0 - ) - }) + await expect( + adapter.cancelTurn({ sessionId: 'session-1', fence: 7, ...target }) + ).resolves.toEqual({ cancelled: true }) + expect( + claude.connections[0].calls.filter((call) => call.subtype === 'interrupt') + ).toHaveLength(1) + } + ) it('classifies provider-declined options without treating timeouts as settled', async () => { const claude = fakeClaude({ diff --git a/src/main/claude/claude-turn-ownership.test.ts b/src/main/claude/claude-turn-ownership.test.ts index 81a84989aa44..19139ad700ec 100644 --- a/src/main/claude/claude-turn-ownership.test.ts +++ b/src/main/claude/claude-turn-ownership.test.ts @@ -7,10 +7,7 @@ import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { StructuredSessionCompaction } from '../native-chat/agent-session-wire/structured-session-compaction' -import { - CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS, - cancelClaudeStructuredTurn -} from './claude-structured-prompt-ownership' +import { cancelClaudeStructuredTurn } from './claude-structured-prompt-ownership' import { sessionFor } from './claude-structured-dispatch-test-support' import { PROVIDER_SESSION_ID, @@ -112,6 +109,19 @@ function sessionHoldingTurn(turnId: string | null): ReturnType Promise<{ cancelled: boolean }>): Promise { + vi.useFakeTimers() + try { + let outcome: unknown = 'still waiting' + void stop().then((result) => (outcome = result)) + await vi.advanceTimersByTimeAsync(0) + return outcome + } finally { + vi.useRealTimers() + } +} + function cancellationOf( session: ReturnType, request: Parameters[0]['request'] @@ -169,56 +179,8 @@ describe('Claude turn ownership', () => { expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) }) - it('keeps the prior dispatch fence after an unknown later send', async () => { - vi.useFakeTimers() - try { - const claude = fakeClaude({ replayUuid: 'echo-turn' }) - 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') - const sendFirst = connection.send - connection.send = async (message) => { - if (connection.sent.length > 0) { - throw new Error('input pump stopped') - } - await sendFirst(message) - } - - await expect( - adapter.dispatch({ - sessionId: 'session-1', - clientMessageId: 'client-2', - body: USER_MESSAGE, - fence: 7 - }) - ).resolves.toMatchObject({ state: 'unknown' }) - - const cancellation = adapter.cancelTurn({ - sessionId: 'session-1', - turnId: 'echo-turn', - fence: 7 - }) - await vi.advanceTimersByTimeAsync(CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS - 1) - expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) - await vi.advanceTimersByTimeAsync(1) - await expect(cancellation).resolves.toEqual({ cancelled: true }) - expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(true) - } finally { - vi.useRealTimers() - } - }) - - it('lets a queued-cancel provider release an unresolved ordinary Stop', async () => { - const claude = fakeClaude({ - replayUuid: 'echo-turn', - capabilities: ['interrupt_cancel_queued_v1'] - }) + it('stops the running turn at once while a later send is unresolved', async () => { + const claude = fakeClaude({ replayUuid: 'echo-turn' }) const { adapter, bodies, connection } = await acquiredWithJournal(claude) await adapter.dispatch({ @@ -228,108 +190,38 @@ describe('Claude turn ownership', () => { fence: 7 }) expect(runningTurnId(bodies)).toBe('echo-turn') + const sendFirst = connection.send + connection.send = async (message) => { + if (connection.sent.length > 0) { + throw new Error('input pump stopped') + } + await sendFirst(message) + } await expect( - adapter.cancelTurn({ - sessionId: 'session-1', - turnId: 'echo-turn', - fence: 7, - dispatchStatus: { state: 'unknown', recovered: false } - }) - ).resolves.toEqual({ cancelled: true }) - expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(true) - }) - - it('lets ordinary Stop proceed after the unresolved delivery fence expires', async () => { - vi.useFakeTimers() - try { - const claude = fakeClaude({ replayUuid: 'echo-turn' }) - const { adapter, bodies, connection } = await acquiredWithJournal(claude) - - await adapter.dispatch({ + adapter.dispatch({ sessionId: 'session-1', - clientMessageId: 'client-1', + clientMessageId: 'client-2', body: USER_MESSAGE, fence: 7 }) - expect(runningTurnId(bodies)).toBe('echo-turn') - - const cancellation = adapter.cancelTurn({ - sessionId: 'session-1', - turnId: 'echo-turn', - fence: 7, - dispatchStatus: { state: 'unknown', recovered: false } - }) - await vi.advanceTimersByTimeAsync(CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS - 1) - expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) - - await vi.advanceTimersByTimeAsync(1) - await expect(cancellation).resolves.toEqual({ cancelled: true }) - expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(true) - } finally { - vi.useRealTimers() - } - }) + ).resolves.toMatchObject({ state: 'unknown' }) - it('releases ordinary Stop as soon as a retired delivery fence settles', async () => { - vi.useFakeTimers() - try { - const session = sessionFor() - session.dispatchSequence = 1 - session.translator = { - handle: vi.fn(), - openTurnInLiveProviderCycle: false, - journalPrompts: { cancel: vi.fn(), resolve: vi.fn() }, - currentTurnId: 'turn-1', - flush: vi.fn(), - contextActivity: 0, - markContextActivity: vi.fn(), - subscribeContextUsageRequests: () => () => {}, - recordContextReport: () => {}, - modelMayHaveChanged: () => {}, - modelWritten: () => {}, - pendingStreamedBlocks: 0, - dispose: vi.fn() - } - session.retiredDispatchWaiters = [ - { - acceptsResult: false, - clientMessageId: 'client-2', - sentUuid: 'uncertain', - dispatchSequence: 1, - requestedAt: null, - replayContentKey: 'ship-it', - resolve: vi.fn(), - retired: true - } - ] - const interrupt = vi.fn().mockResolvedValue(undefined) - session.connection.interrupt = interrupt - const cancellation = cancelClaudeStructuredTurn({ - request: { sessionId: 'session-1', turnId: 'turn-1', fence: 1 }, - sessions: new Map([['session-1', session]]), - compactions: new StructuredSessionCompaction(), - admitPromptCancellation: () => true - }) - await vi.advanceTimersByTimeAsync(100) - session.retiredDispatchWaiters = [] - await vi.advanceTimersByTimeAsync(100) - const settledBeforeDeadline = interrupt.mock.calls.length > 0 - if (!settledBeforeDeadline) { - await vi.advanceTimersByTimeAsync(CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS) - await cancellation - } - expect(settledBeforeDeadline).toBe(true) - await expect(cancellation).resolves.toEqual({ cancelled: true }) - } finally { - vi.useRealTimers() - } + await expect( + outcomeAtOnce(() => + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'echo-turn', fence: 7 }) + ) + ).resolves.toEqual({ cancelled: true }) + expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(true) }) - it('does not wait when the dispatch admission is already current', async () => { - vi.useFakeTimers() - try { - const claude = fakeClaude({ replayUuid: 'echo-turn' }) + it.each([ + ['with cancel_queued', ['interrupt_cancel_queued_v1']], + ['without cancel_queued', []] + ])( + 'stops the running turn at once while its own delivery is unresolved, %s', + async (_label, capabilities) => { + const claude = fakeClaude({ replayUuid: 'echo-turn', capabilities }) const { adapter, bodies, connection } = await acquiredWithJournal(claude) await adapter.dispatch({ @@ -341,36 +233,61 @@ describe('Claude turn ownership', () => { expect(runningTurnId(bodies)).toBe('echo-turn') await expect( - adapter.cancelTurn({ sessionId: 'session-1', turnId: 'echo-turn', fence: 7 }) + outcomeAtOnce(() => + adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'echo-turn', + fence: 7, + dispatchStatus: { state: 'unknown', recovered: false } + }) + ) ).resolves.toEqual({ cancelled: true }) expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(true) - } finally { - vi.useRealTimers() } + ) + + it('stops the in-memory turn at once while a retired send is unresolved', async () => { + const session = sessionHoldingTurn('turn-1') + session.retiredDispatchWaiters = [ + { + acceptsResult: false, + clientMessageId: 'client-2', + sentUuid: 'uncertain', + dispatchSequence: 1, + requestedAt: null, + replayContentKey: 'ship-it', + resolve: vi.fn(), + retired: true + } + ] + const interrupt = vi.fn().mockResolvedValue(undefined) + session.connection.interrupt = interrupt + + await expect( + outcomeAtOnce(() => + cancellationOf(session, { sessionId: 'session-1', turnId: 'turn-1', fence: 1 }) + ) + ).resolves.toEqual({ cancelled: true }) + expect(interrupt).toHaveBeenCalledOnce() }) - it('honors an unresolved journal submission before the first in-memory dispatch', async () => { - vi.useFakeTimers() - try { - const claude = fakeClaude({ replayUuid: null }) - const { adapter, bodies, connection } = await acquiredWithJournal(claude) - providerOutput(connection, 'provider-turn') - expect(runningTurnId(bodies)).toBe('provider-turn') + it('stops a provider-opened turn at once while the first journal submission is unresolved', async () => { + const claude = fakeClaude({ replayUuid: null }) + const { adapter, bodies, connection } = await acquiredWithJournal(claude) + providerOutput(connection, 'provider-turn') + expect(runningTurnId(bodies)).toBe('provider-turn') - const cancellation = adapter.cancelTurn({ - sessionId: 'session-1', - turnId: 'provider-turn', - fence: 7, - dispatchStatus: { state: 'pending', recovered: false } - }) - await vi.advanceTimersByTimeAsync(CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS - 1) - expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) - await vi.advanceTimersByTimeAsync(1) - await expect(cancellation).resolves.toEqual({ cancelled: true }) - expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(true) - } finally { - vi.useRealTimers() - } + await expect( + outcomeAtOnce(() => + adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'provider-turn', + fence: 7, + dispatchStatus: { state: 'pending', recovered: false } + }) + ) + ).resolves.toEqual({ cancelled: true }) + expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(true) }) it('still stops an echo-opened turn', async () => { @@ -444,32 +361,23 @@ describe('Claude turn ownership', () => { expect(interrupt).not.toHaveBeenCalled() }) - // The guard re-checks after the delivery fence may have waited seconds, so the journal read - // has to happen then — a value captured at request time would interrupt whatever ran next. - it('re-reads the published turn after the delivery fence waits', async () => { - vi.useFakeTimers() - try { - let publishedTurnId = 'turn-shown' - const session = sessionHoldingTurn('turn-shown') - const interrupt = vi.fn().mockResolvedValue(undefined) - session.connection.interrupt = interrupt - - const cancellation = cancellationOf(session, { - sessionId: 'session-1', - turnId: 'turn-shown', - fence: 1, - dispatchStatus: { state: 'unknown', recovered: false }, - resolveLiveTurnId: () => publishedTurnId - }) - await vi.advanceTimersByTimeAsync(CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS - 1) - publishedTurnId = 'turn-next' - await vi.advanceTimersByTimeAsync(1) - - await expect(cancellation).resolves.toEqual({ cancelled: false }) - expect(interrupt).not.toHaveBeenCalled() - } finally { - vi.useRealTimers() - } + it('refuses a Stop for an ended turn at once while a later send is unresolved', async () => { + const session = sessionHoldingTurn('turn-next') + const interrupt = vi.fn().mockResolvedValue(undefined) + session.connection.interrupt = interrupt + + await expect( + outcomeAtOnce(() => + cancellationOf(session, { + sessionId: 'session-1', + turnId: 'turn-shown', + fence: 1, + dispatchStatus: { state: 'unknown', recovered: false }, + resolveLiveTurnId: () => 'turn-next' + }) + ) + ).resolves.toEqual({ cancelled: false }) + expect(interrupt).not.toHaveBeenCalled() }) it('refuses a stale turn id once the provider opened a newer turn', async () => { 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 e4ef8eb85f88..071ac6803d33 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 @@ -291,8 +291,8 @@ export type StructuredAgentSessionAdapter = { /** Latest journal submission for this fence, when the host has one. */ dispatchStatus?: { state: AgentJournalDispatchState; recovered: boolean } | null /** Re-reads the turn the published journal says is running — the only turn a client - * could have named. A function, not a value, because the guard re-checks after the - * delivery fence may have waited. Absent for direct callers with no journal. */ + * could have named. A function, not a value, because the guard reads it when the + * interrupt is issued. Absent for direct callers with no journal. */ resolveLiveTurnId?: () => string | null }): Promise /** Changes the provider thread's goal. `rejected` is the provider refusing the From 7ab80f9196f0efeafcfd98de97846d4b7435cd93 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 03:41:09 -0700 Subject: [PATCH 02/10] fix(claude): a Stop naming a turn that just ended withdraws the follow-up behind it The phone names the turn it shows, and its copy of the chat can trail the host's. When that turn ends just before its Stop lands, with a follow-up already written to Claude but not yet started, the Stop was refused and the follow-up ran anyway. With nothing live, a written follow-up whose turn has not opened is work the naming client has not seen start, so the Stop takes the conversation path: it interrupts and Claude's queued follow-up settles as withdrawn. A Stop naming an older turn while a newer one runs is still refused. --- ...laude-structured-conversation-stop.test.ts | 8 +- .../claude-structured-prompt-ownership.ts | 16 ++-- .../claude-structured-queued-stop.test.ts | 55 ++++++++--- ...d-agent-session-claude-queued-stop.test.ts | 91 +++++++++++++++++-- 4 files changed, 142 insertions(+), 28 deletions(-) diff --git a/src/main/claude/claude-structured-conversation-stop.test.ts b/src/main/claude/claude-structured-conversation-stop.test.ts index 840125853ba1..90ca65691516 100644 --- a/src/main/claude/claude-structured-conversation-stop.test.ts +++ b/src/main/claude/claude-structured-conversation-stop.test.ts @@ -33,14 +33,14 @@ describe('Claude Stop that names no turn', () => { expect(interrupts(claude)).toBe(1) }) - it('still refuses the placeholder a client used to name for that gap', async () => { + it('interrupts that gap the same way when the Stop names a turn that is not live', async () => { const claude = fakeClaude({ replayUuid: null }) const adapter = await written(claude) await expect( - adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-none', fence: 7 }) - ).resolves.toEqual({ cancelled: false }) - expect(interrupts(claude)).toBe(0) + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-ended', fence: 7 }) + ).resolves.toEqual({ cancelled: true }) + expect(interrupts(claude)).toBe(1) }) it('interrupts a turn Claude opened on its own echo', async () => { diff --git a/src/main/claude/claude-structured-prompt-ownership.ts b/src/main/claude/claude-structured-prompt-ownership.ts index e53d8156c5c6..78cf435dc20e 100644 --- a/src/main/claude/claude-structured-prompt-ownership.ts +++ b/src/main/claude/claude-structured-prompt-ownership.ts @@ -57,10 +57,11 @@ function claudeLiveTurnId(session: ClaudeSession, request: CancelInput): string } /** - * A Stop that names no turn, or names the live one: the conversation asked to stop whatever this - * child has in flight. Claude's interrupt is session-scoped, so there is no turn identity to check — - * only that this is still the child the host judged, and that it has a turn open or a written - * message whose turn has not opened yet (the gap before its echo, which no client can name). + * A Stop that names no turn, names the live one, or names one that ended before a written + * follow-up opened its own: the conversation asked to stop whatever this child has in flight. + * Claude's interrupt is session-scoped, so there is no turn identity to check — only that this is + * still the child the host judged, and that it has a turn open or a written message whose turn has + * not opened yet (the gap before its echo, which no client can name). */ function cancelClaudeConversation( session: ClaudeSession, @@ -95,13 +96,16 @@ export async function cancelClaudeStructuredTurn(input: { return { cancelled: false } } const requestedTurnId = request.turnId + const liveTurnId = claudeLiveTurnId(session, request) // Naming the live turn asks for exactly what the conversation Stop interrupts, so a follow-up's - // pending handover is no reason to hold it; the queue sweep settles that follow-up. + // pending handover is no reason to hold it; the queue sweep settles that follow-up. With nothing + // live, a written follow-up whose turn has not opened is one the naming client has not seen start. if ( !prompt && (requestedTurnId === undefined || (!compactions.ownsTurn(request.sessionId, requestedTurnId) && - claudeLiveTurnId(session, request) === requestedTurnId)) + (liveTurnId === requestedTurnId || + (liveTurnId === null && session.dispatchWaiters.length > 0)))) ) { return cancelClaudeConversation( session, diff --git a/src/main/claude/claude-structured-queued-stop.test.ts b/src/main/claude/claude-structured-queued-stop.test.ts index f04fb298e4af..6f2f19e5fab7 100644 --- a/src/main/claude/claude-structured-queued-stop.test.ts +++ b/src/main/claude/claude-structured-queued-stop.test.ts @@ -81,14 +81,34 @@ async function acquired( return { adapter, bodies, connection: claude.connections[0]! } } -function runningTurnId(bodies: Map): string { +function openTurnId(bodies: Map): string | null { for (const body of bodies.values()) { const turn = readAgentJournalTurn(body) if (turn?.state === 'running') { return turn.turnId } } - throw new Error('expected a running turn') + return null +} + +function runningTurnId(bodies: Map): string { + const turnId = openTurnId(bodies) + if (turnId === null) { + throw new Error('expected a running turn') + } + return turnId +} + +function completeTurn(connection: FakeConnection, uuid: string): void { + connection.handlers.onMessage?.({ + type: 'result', + subtype: 'success', + uuid, + session_id: PROVIDER_SESSION_ID, + is_error: false, + terminal_reason: 'completed', + duration_ms: 12 + }) } /** Writes a send and opens its turn with the echo, which adopts the client uuid. */ @@ -240,15 +260,7 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => { const { adapter, bodies, connection } = await acquired(cli.claude, settlements) await sendAndOpen(adapter, connection, 'client-a') const olderTurnId = runningTurnId(bodies) - connection.handlers.onMessage?.({ - type: 'result', - subtype: 'success', - uuid: 'result-a', - session_id: PROVIDER_SESSION_ID, - is_error: false, - terminal_reason: 'completed', - duration_ms: 12 - }) + completeTurn(connection, 'result-a') await sendAndOpen(adapter, connection, 'client-b') const liveTurnId = runningTurnId(bodies) await send(adapter, 'client-c') @@ -262,4 +274,25 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => { [] ) }) + + it('withdraws a follow-up whose turn has not opened when the turn the Stop names already ended', async () => { + const settlements: Settlement[] = [] + const cli = claudeCli({ initProof: 'init', capabilities: CAPABILITIES }) + const { adapter, bodies, connection } = await acquired(cli.claude, settlements) + await sendAndOpen(adapter, connection, 'client-a') + const shownTurnId = runningTurnId(bodies) + await send(adapter, 'client-b') + cli.queue(String(connection.sent.at(-1)!.uuid)) + // The phone's copy still shows the first turn, which ends before its Stop lands. + completeTurn(connection, 'result-a') + expect(openTurnId(bodies)).toBeNull() + + await expect(namedStop(adapter, shownTurnId)).resolves.toEqual({ cancelled: true }) + expect(connection.calls.find((call) => call.subtype === 'interrupt')?.params).toEqual({ + cancelQueued: true + }) + expect(settlements.filter((settlement) => settlement.clientMessageId === 'client-b')).toEqual( + WITHDRAWN + ) + }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts index e8b8d0f00442..63493f2ec72f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts @@ -7,6 +7,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, expect, it, vi } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' import { DISPATCH_REJECTED_CANCELLED } from '../../../shared/structured-agent-session-dispatch-rejection' import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-live-turn' import { projectStructuredAgentSessionStatus } from '../../../shared/structured-agent-session-projection' @@ -148,8 +149,12 @@ async function status(): Promise { ) } -it('withdraws a follow-up Claude queued behind the running turn when that turn is stopped', async () => { - const connection = claude.connections[0]! +async function liveTurnId(): Promise { + return activeStructuredAgentSessionTurnId((await host.journalSnapshot(SESSION)).items) +} + +/** Sends the first message and lets Claude open its turn; returns the turn a client reads. */ +async function openFirstTurn(connection: (typeof claude.connections)[number]): Promise { const first = await send('Write a long reply.') await eventually(() => expect(connection.sent).toHaveLength(1)) // Claude opens the turn: its system/init, then the echo of the message it runs. @@ -166,18 +171,38 @@ it('withdraws a follow-up Claude queued behind the running turn when that turn i uuid: connection.sent.at(-1)!.uuid }) await eventually(async () => expect((await dispatch(first)).state).toBe('accepted')) - const turnId = activeStructuredAgentSessionTurnId((await host.journalSnapshot(SESSION)).items) + const turnId = await liveTurnId() expect(turnId).not.toBeNull() + return turnId! +} + +async function endFirstTurn(connection: (typeof claude.connections)[number]): Promise { + connection.handlers.onMessage?.({ + type: 'result', + subtype: 'success', + uuid: 'result-first', + session_id: PROVIDER_SESSION_ID, + is_error: false, + terminal_reason: 'completed', + duration_ms: 12 + }) + await eventually(async () => expect(await liveTurnId()).toBeNull()) +} + +function stop(turnId: string) { + return host.cancel(CALLER, { envelope: envelope('agentSession.cancel', { turnId }), turnId }) +} + +it('withdraws a follow-up Claude queued behind the running turn when that turn is stopped', async () => { + const connection = claude.connections[0]! + const turnId = await openFirstTurn(connection) const followUp = await send('And then this.') await eventually(() => expect(connection.sent).toHaveLength(2)) queued.push(String(connection.sent.at(-1)!.uuid)) await eventually(async () => expect((await dispatch(followUp)).state).toBe('pending')) - const stopped = await host.cancel(CALLER, { - envelope: envelope('agentSession.cancel', { turnId }), - turnId: turnId! - }) + const stopped = await stop(turnId) expect(stopped).toMatchObject({ ok: true, value: { cancelled: true } }) // The interrupted turn closes as the CLI ends it. connection.handlers.onMessage?.({ @@ -194,3 +219,55 @@ it('withdraws a follow-up Claude queued behind the running turn when that turn i ) await eventually(async () => expect(await status()).toBe('idle')) }, 15_000) + +// The phone names the turn from its own copy of the journal, which can trail the host's. +it('withdraws a follow-up Claude holds when the turn the Stop names ended before it landed', async () => { + const connection = claude.connections[0]! + const turnId = await openFirstTurn(connection) + const followUp = await send('And then this.') + await eventually(() => expect(connection.sent).toHaveLength(2)) + queued.push(String(connection.sent.at(-1)!.uuid)) + await eventually(async () => expect((await dispatch(followUp)).state).toBe('pending')) + await endFirstTurn(connection) + + expect(await stop(turnId)).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(connection.calls.find((call) => call.subtype === 'interrupt')?.params).toEqual({ + cancelQueued: true + }) + await eventually(async () => + expect(await dispatch(followUp)).toEqual({ + state: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + ) + await eventually(async () => expect(await status()).toBe('idle')) +}, 15_000) + +it('withdraws a follow-up still queued on the host when the turn the Stop names already ended', async () => { + const connection = claude.connections[0]! + const turnId = await openFirstTurn(connection) + await endFirstTurn(connection) + // Holds the delivery loop between its start check and the handover, with the follow-up queued. + let release!: () => void + const held = new Promise((resolve) => (release = resolve)) + const awaitStarted = vi.spyOn(adapter, 'awaitStarted').mockImplementationOnce(async () => { + await held + }) + const followUp = await send('And then this.') + await eventually(() => expect(awaitStarted).toHaveBeenCalled()) + const submission = (await host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.clientMessageId === followUp + ) + expect(submission && isQueuedAgentJournalSubmission(submission)).toBe(true) + + expect(await stop(turnId)).toMatchObject({ ok: true }) + release() + await eventually(async () => + expect(await dispatch(followUp)).toEqual({ + state: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + ) + expect(connection.sent).toHaveLength(1) + await eventually(async () => expect(await status()).toBe('idle')) +}, 15_000) From ca2355fb968b55a060bf9d0c8e8a4aa32efc6b78 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 03:45:53 -0700 Subject: [PATCH 03/10] fix(native-chat): a named Stop that withdrew a queued message says nothing finished When a Stop named a turn that had already ended, the host still withdrew the message the user had sent after it, yet the Stop reported nothing cancelled and the chat read "The provider had already finished this turn." That withdrawal is the Stop taking effect, so it now reports cancelled and writes no row, as a Stop naming no turn already does in the same case. --- ...uctured-agent-session-claude-queued-stop.test.ts | 7 ++++++- .../structured-agent-session-host-mutations.ts | 13 +++++++++---- .../structured-agent-session-turns.ts | 8 +++++++- 3 files changed, 22 insertions(+), 6 deletions(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts index 63493f2ec72f..ffc8ec88b91a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts @@ -260,7 +260,8 @@ it('withdraws a follow-up still queued on the host when the turn the Stop names ) expect(submission && isQueuedAgentJournalSubmission(submission)).toBe(true) - expect(await stop(turnId)).toMatchObject({ ok: true }) + // Nothing reached Claude, so the host's withdrawal is the whole Stop, as with no turn named. + expect(await stop(turnId)).toMatchObject({ ok: true, value: { cancelled: true } }) release() await eventually(async () => expect(await dispatch(followUp)).toEqual({ @@ -270,4 +271,8 @@ it('withdraws a follow-up still queued on the host when the turn the Stop names ) expect(connection.sent).toHaveLength(1) await eventually(async () => expect(await status()).toBe('idle')) + const rows = (await host.journalSnapshot(SESSION)).items.flatMap((item) => + item.body.kind === 'status' ? [item.body.text] : [] + ) + expect(rows).not.toContain('The provider had already finished this turn.') }, 15_000) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index 1586bd4a1c16..1cf22e4616c4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts @@ -29,6 +29,7 @@ import { import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import type { AgentSessionPromptRequest } from './structured-agent-session-turns-prompt' import { threadGoalPlan } from './structured-agent-session-thread-goal' +import { performCancel } from './structured-agent-session-turns' import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import { admitAndRunAgentSessionMutation, @@ -196,10 +197,14 @@ export function cancelStructuredAgentSessionTurn( ) const record = context.deps.store.getRecord(ctx.sessionId) return child && inFlight - ? plan.run({ - ...ctx, - failureTextContext: structuredAgentSessionFailureWordsContext(record) - }) + ? performCancel( + { ...ctx, failureTextContext: structuredAgentSessionFailureWordsContext(record) }, + { + clientOperationId: params.envelope.clientOperationId, + ...named, + withdrewQueued: withdrawn.length > 0 + } + ) : { ok: true, value: { ...named, cancelled: withdrawn.length > 0 } } } }, 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 46c4638dc4ee..fecae14b88f9 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 @@ -247,6 +247,8 @@ export async function performCancel( scope?: 'background-tasks' taskId?: string prompt?: { itemId: string; expectedRevision: number } + /** The host already withdrew queued messages for this Stop. */ + withdrewQueued?: boolean } ): Promise> { if (input.prompt) { @@ -280,7 +282,11 @@ export async function performCancel( ...(input.prompt ? { prompt: { itemId: input.prompt.itemId } } : {}) }) cancelled = outcome.cancelled - if (!cancelled && input.turnId !== undefined) { + if (!cancelled && input.turnId !== undefined && input.withdrewQueued) { + // The Stop ended what was queued, which is how a Stop naming no turn reports the same case. + cancelled = true + note = null + } else if (!cancelled && input.turnId !== undefined) { note = { kind: 'status', text: 'The provider had already finished this turn.' } } else if (!cancelled && input.prompt) { note = null From 7817f811b85e68d36e89a65446d9e1d5a0b1770d Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 03:53:56 -0700 Subject: [PATCH 04/10] fix(native-chat): a named Stop the provider left unconfirmed is not reported as a success A named Stop that withdrew a host-queued message was reported cancelled, with no row, whatever the provider answered. When the provider took the interrupt without confirming it (or refused it), the turn may still be running, so the Stop now keeps the provider's answer instead of claiming success. --- ...ed-agent-session-conversation-stop.test.ts | 31 +++++++++++++++++++ .../structured-agent-session-turns.ts | 9 +++++- 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index ab0368a6b921..7465f0962cb0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -274,4 +274,35 @@ describe('a Stop that names its turn, as an older client sends it', () => { expect(cancelTurn).toHaveBeenCalledWith(expect.objectContaining({ turnId: 'turn-1' })) expect(await statusRows()).toEqual([ALREADY_FINISHED]) }) + + async function queueOnHost(): Promise<{ id: string; release: () => void }> { + const started = Promise.withResolvers() + awaitStarted.mockImplementationOnce(() => started.promise) + const { id, result } = send('hello') + await result + await eventually(() => expect(awaitStarted).toHaveBeenCalled()) + return { id, release: () => started.resolve(undefined) } + } + + it('reports success with no row when it withdrew a queued message and the turn had ended', async () => { + const queued = await queueOnHost() + cancelTurn.mockResolvedValueOnce({ cancelled: false }) + + expect(await stop('turn-1')).toMatchObject({ ok: true, value: { cancelled: true } }) + queued.release() + + expect(await submission(queued.id)).toMatchObject({ dispatchState: 'rejected' }) + expect(await statusRows()).toEqual([]) + }) + + it('does not report success when the provider left the interrupt unconfirmed', async () => { + const queued = await queueOnHost() + cancelTurn.mockResolvedValueOnce({ cancelled: false, unconfirmed: true }) + + expect(await stop('turn-1')).toMatchObject({ ok: true, value: { cancelled: false } }) + queued.release() + + expect(await submission(queued.id)).toMatchObject({ dispatchState: 'rejected' }) + expect(await statusRows()).not.toEqual([]) + }) }) 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 fecae14b88f9..17a021abaf96 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 @@ -282,8 +282,15 @@ export async function performCancel( ...(input.prompt ? { prompt: { itemId: input.prompt.itemId } } : {}) }) cancelled = outcome.cancelled - if (!cancelled && input.turnId !== undefined && input.withdrewQueued) { + if ( + !cancelled && + input.turnId !== undefined && + input.withdrewQueued && + !outcome.unconfirmed && + !outcome.refusal + ) { // The Stop ended what was queued, which is how a Stop naming no turn reports the same case. + // A provider that took or refused the interrupt still has a turn the withdrawal did not end. cancelled = true note = null } else if (!cancelled && input.turnId !== undefined) { From acacb8ae1444e1a2bcaa578f41b51b8d75545191 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:03:14 -0700 Subject: [PATCH 05/10] fix(native-chat): judge a named Stop's withdrawal by what the journal still runs A named Stop that withdrew a host-queued message was reported as a success unless the provider's answer carried a refusal or was unconfirmed. Providers report a refusal differently: Claude never sets one, so a Stop naming an older turn while a newer one ran came back cancelled with no row; Codex refuses any turn that has ended, so the ended-turn case still wrote "already finished". The success is now decided by the rule the no-turn Stop uses: nothing still reads working on the journal. --- ...d-agent-session-claude-queued-stop.test.ts | 38 +++++++++++++++++++ ...ed-agent-session-conversation-stop.test.ts | 15 ++++++++ .../structured-agent-session-turns.ts | 10 ++++- 3 files changed, 61 insertions(+), 2 deletions(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts index ffc8ec88b91a..c9feabd6dfb2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-queued-stop.test.ts @@ -276,3 +276,41 @@ it('withdraws a follow-up still queued on the host when the turn the Stop names ) expect(rows).not.toContain('The provider had already finished this turn.') }, 15_000) + +it('withdraws a host-queued follow-up but leaves a newer turn running when the Stop names an older one', async () => { + const connection = claude.connections[0]! + const olderTurnId = await openFirstTurn(connection) + await endFirstTurn(connection) + const second = await send('Now this.') + await eventually(() => expect(connection.sent).toHaveLength(2)) + connection.handlers.onMessage?.({ + ...connection.sent.at(-1)!, + uuid: connection.sent.at(-1)!.uuid + }) + await eventually(async () => expect((await dispatch(second)).state).toBe('accepted')) + const newerTurnId = await liveTurnId() + expect(newerTurnId).not.toBeNull() + expect(newerTurnId).not.toBe(olderTurnId) + let release!: () => void + const held = new Promise((resolve) => (release = resolve)) + const awaitStarted = vi.spyOn(adapter, 'awaitStarted').mockImplementationOnce(async () => { + await held + }) + const followUp = await send('And then this.') + await eventually(() => expect(awaitStarted).toHaveBeenCalled()) + + expect(await stop(olderTurnId)).toMatchObject({ ok: true, value: { cancelled: false } }) + release() + await eventually(async () => + expect(await dispatch(followUp)).toEqual({ + state: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + ) + expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false) + expect(await liveTurnId()).toBe(newerTurnId) + const rows = (await host.journalSnapshot(SESSION)).items.flatMap((item) => + item.body.kind === 'status' ? [item.body.text] : [] + ) + expect(rows).toContain('The provider had already finished this turn.') +}, 15_000) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index 7465f0962cb0..b40fa26e823c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -295,6 +295,21 @@ describe('a Stop that names its turn, as an older client sends it', () => { expect(await statusRows()).toEqual([]) }) + // Codex refuses an interrupt for a turn that has ended. + it('reports success with no row when the provider refused a turn that had ended', async () => { + const queued = await queueOnHost() + cancelTurn.mockResolvedValueOnce({ + cancelled: false, + refusal: { detail: { text: 'no such turn', audience: 'person' } } + }) + + expect(await stop('turn-1')).toMatchObject({ ok: true, value: { cancelled: true } }) + queued.release() + + expect(await submission(queued.id)).toMatchObject({ dispatchState: 'rejected' }) + expect(await statusRows()).toEqual([]) + }) + it('does not report success when the provider left the interrupt unconfirmed', async () => { const queued = await queueOnHost() cancelTurn.mockResolvedValueOnce({ cancelled: false, unconfirmed: true }) 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 17a021abaf96..e93cda1319d7 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 @@ -35,6 +35,7 @@ import type { import { structuredAgentSessionStartFailure } from './structured-agent-session-failure-text' import { validatePendingPrompt } from './structured-agent-session-prompt-state' import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' export { performSetOption } from './structured-agent-session-turns-options' export { performPrompt } from './structured-agent-session-turns-prompt' @@ -287,10 +288,15 @@ export async function performCancel( input.turnId !== undefined && input.withdrewQueued && !outcome.unconfirmed && - !outcome.refusal + !isStructuredAgentSessionMainAgentWorking( + ctx.journal.activeTurnId(), + ctx.journal.submissions(), + ctx.fence + ) ) { // The Stop ended what was queued, which is how a Stop naming no turn reports the same case. - // A provider that took or refused the interrupt still has a turn the withdrawal did not end. + // Whether anything still runs is read from the journal: providers differ on refusing a turn + // that has ended, and an unconfirmed interrupt may have left the turn running. cancelled = true note = null } else if (!cancelled && input.turnId !== undefined) { From fee6489f4d701712a396b298031d304d08e33872 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:11:23 -0700 Subject: [PATCH 06/10] refactor(native-chat): one rule for what every conversation Stop reports A conversation Stop that withdrew what was queued and left nothing working on the journal is a success, whether or not it named a turn. The rule was scoped to named Stops, so a Stop naming no turn whose turn ended under it reported that the agent had no turn to stop after it withdrew a message. An unconfirmed interrupt is now reported as unconfirmed before the named-turn row, which said the provider had already finished a turn it had just taken an interrupt for. --- ...ed-agent-session-conversation-stop.test.ts | 2 +- .../structured-agent-session-turns.test.ts | 78 +++++++++++++++++++ .../structured-agent-session-turns.ts | 15 ++-- 3 files changed, 86 insertions(+), 9 deletions(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index b40fa26e823c..d6d16bc86ad4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -318,6 +318,6 @@ describe('a Stop that names its turn, as an older client sends it', () => { queued.release() expect(await submission(queued.id)).toMatchObject({ dispatchState: 'rejected' }) - expect(await statusRows()).not.toEqual([]) + expect(await statusRows()).toEqual(['Cancellation was not confirmed.']) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts index 576743ebac49..edce99685a31 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts @@ -244,3 +244,81 @@ describe('performCancel', () => { expect(journal.snapshot().items).toEqual([]) }) }) + +describe('what a conversation Stop reports when the provider stopped nothing', () => { + async function cancelWith( + outcome: Awaited>, + input: { turnId?: string; withdrewQueued?: boolean }, + running = false + ) { + root = await mkdtemp(join(tmpdir(), 'orca-turn-cancel-report-')) + const journal = await journals.open({ identity: IDENTITY, journalDir: root }) + if (running) { + await journal.appendItem( + { + provider: 'legacy', + agent: 'codex', + sessionId: 'session-1', + recordId: 'turn-lifecycle:turn-1' + }, + { + kind: 'status', + text: 'Agent is working…', + turnLifecycle: { turnId: 'turn-1', state: 'running' } + }, + { fence: 1 } + ) + } + const ctx: AgentSessionTurnContext = { + sessionId: 'session-1', + journal, + fence: 1, + adapter: { + acquire: vi.fn(), + dispatch: vi.fn(), + closeSession: vi.fn(), + cancelTurn: vi.fn(async () => outcome), + answerPrompt: vi.fn(), + setOption: vi.fn() + }, + persistOptions: async () => undefined, + resolvedBy: 'client-1', + publish: vi.fn(), + flushStreamedEvents: async () => undefined, + now: () => 1 + } + const result = await performCancel(ctx, { clientOperationId: 'cancel-report-1', ...input }) + const rows = journal + .snapshot() + .items.flatMap((item) => + item.body.kind === 'status' && !item.body.turnLifecycle ? [item.body.text] : [] + ) + return { cancelled: result.ok && result.value.cancelled, rows } + } + + it('reports a Stop naming no turn that withdrew what was queued, with nothing left working, as a success', async () => { + expect(await cancelWith({ cancelled: false }, { withdrewQueued: true })).toEqual({ + cancelled: true, + rows: [] + }) + }) + + it('keeps a Stop naming no turn not cancelled while the journal still reads working', async () => { + const reported = await cancelWith({ cancelled: false }, { withdrewQueued: true }, true) + expect(reported.cancelled).toBe(false) + expect(reported.rows).toHaveLength(1) + expect(reported.rows).not.toContain('The provider had already finished this turn.') + }) + + it.each([false, true])( + 'says a named Stop the provider left unconfirmed is unconfirmed, not that the turn had finished (withdrew: %s)', + async (withdrewQueued) => { + expect( + await cancelWith( + { cancelled: false, unconfirmed: true }, + { turnId: 'turn-1', withdrewQueued } + ) + ).toEqual({ cancelled: false, rows: ['Cancellation was not confirmed.'] }) + } + ) +}) 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 e93cda1319d7..03d46defcba4 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 @@ -285,7 +285,6 @@ export async function performCancel( cancelled = outcome.cancelled if ( !cancelled && - input.turnId !== undefined && input.withdrewQueued && !outcome.unconfirmed && !isStructuredAgentSessionMainAgentWorking( @@ -294,22 +293,22 @@ export async function performCancel( ctx.fence ) ) { - // The Stop ended what was queued, which is how a Stop naming no turn reports the same case. - // Whether anything still runs is read from the journal: providers differ on refusing a turn - // that has ended, and an unconfirmed interrupt may have left the turn running. + // A Stop that withdrew what was queued and left nothing working ended what it was sent for, + // named or not. The journal judges it: providers differ on refusing a turn that has ended. cancelled = true note = null - } else if (!cancelled && input.turnId !== undefined) { - note = { kind: 'status', text: 'The provider had already finished this turn.' } - } else if (!cancelled && input.prompt) { - note = null } else if (!cancelled && outcome.unconfirmed) { + // Before the named-turn row: the provider took the interrupt, so the turn may still run. note = { kind: 'status', ...agentSessionFailureWords(agentSessionFailureFact('cancelUnconfirmed'), { surface: 'row' }) } + } else if (!cancelled && input.turnId !== undefined) { + note = { kind: 'status', text: 'The provider had already finished this turn.' } + } else if (!cancelled && input.prompt) { + note = null } else if (!cancelled) { // Sent only while the chat reads working, so a Stop that ended nothing must say why. const detail = outcome.refusal?.detail From d325b19e5801e98779851eba2633c8749948c501 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:20:56 -0700 Subject: [PATCH 07/10] fix(native-chat): judge what a Stop left working once its streamed rows land Claude's send echo accepts the send one sink write before the turn row it opens lands, so a Stop that withdrew a queued message could read the journal idle in that gap and report success with no row while the turn then ran. Drain the streamed events before the read, as the prompt Stop already does. --- .../structured-agent-session-turns.test.ts | 28 +++++++++++++++---- .../structured-agent-session-turns.ts | 16 +++++++---- 2 files changed, 34 insertions(+), 10 deletions(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts index edce99685a31..ca34ea551e0d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts @@ -249,12 +249,12 @@ describe('what a conversation Stop reports when the provider stopped nothing', ( async function cancelWith( outcome: Awaited>, input: { turnId?: string; withdrewQueued?: boolean }, - running = false + turnRow: 'none' | 'running' | 'lands-on-flush' = 'none' ) { root = await mkdtemp(join(tmpdir(), 'orca-turn-cancel-report-')) const journal = await journals.open({ identity: IDENTITY, journalDir: root }) - if (running) { - await journal.appendItem( + const openTurn = () => + journal.appendItem( { provider: 'legacy', agent: 'codex', @@ -268,6 +268,8 @@ describe('what a conversation Stop reports when the provider stopped nothing', ( }, { fence: 1 } ) + if (turnRow === 'running') { + await openTurn() } const ctx: AgentSessionTurnContext = { sessionId: 'session-1', @@ -284,7 +286,11 @@ describe('what a conversation Stop reports when the provider stopped nothing', ( persistOptions: async () => undefined, resolvedBy: 'client-1', publish: vi.fn(), - flushStreamedEvents: async () => undefined, + flushStreamedEvents: async () => { + if (turnRow === 'lands-on-flush') { + await openTurn() + } + }, now: () => 1 } const result = await performCancel(ctx, { clientOperationId: 'cancel-report-1', ...input }) @@ -304,12 +310,24 @@ describe('what a conversation Stop reports when the provider stopped nothing', ( }) it('keeps a Stop naming no turn not cancelled while the journal still reads working', async () => { - const reported = await cancelWith({ cancelled: false }, { withdrewQueued: true }, true) + const reported = await cancelWith({ cancelled: false }, { withdrewQueued: true }, 'running') expect(reported.cancelled).toBe(false) expect(reported.rows).toHaveLength(1) expect(reported.rows).not.toContain('The provider had already finished this turn.') }) + it('reads the journal after its streamed rows land: a turn whose send was accepted first is still working', async () => { + const reported = await cancelWith( + { cancelled: false }, + { turnId: 'turn-0', withdrewQueued: true }, + 'lands-on-flush' + ) + expect(reported).toEqual({ + cancelled: false, + rows: ['The provider had already finished this turn.'] + }) + }) + it.each([false, true])( 'says a named Stop the provider left unconfirmed is unconfirmed, not that the turn had finished (withdrew: %s)', async (withdrewQueued) => { 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 03d46defcba4..65733510ac1e 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 @@ -239,6 +239,16 @@ function requireSubmission( return submission } +/** Claude's echo accepts a send one sink write before its turn row lands, so read after the drain. */ +async function isMainAgentWorkingOnceFlushed(ctx: AgentSessionTurnContext): Promise { + await ctx.flushStreamedEvents() + return isStructuredAgentSessionMainAgentWorking( + ctx.journal.activeTurnId(), + ctx.journal.submissions(), + ctx.fence + ) +} + export async function performCancel( ctx: AgentSessionTurnContext, input: { @@ -287,11 +297,7 @@ export async function performCancel( !cancelled && input.withdrewQueued && !outcome.unconfirmed && - !isStructuredAgentSessionMainAgentWorking( - ctx.journal.activeTurnId(), - ctx.journal.submissions(), - ctx.fence - ) + !(await isMainAgentWorkingOnceFlushed(ctx)) ) { // A Stop that withdrew what was queued and left nothing working ended what it was sent for, // named or not. The journal judges it: providers differ on refusing a turn that has ended. From 9242db5f17d686c498d99245f109a4fd12d73532 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:25:04 -0700 Subject: [PATCH 08/10] fix(native-chat): a Stop naming no turn reads what is working once its streamed rows land The no-turn Stop decides whether to interrupt from the journal. Claude's send echo accepts the send one sink write before its turn row lands, so in that gap the journal read idle and the Stop skipped the interrupt while the turn ran. Drain the streamed events first, through the same read the Stop's report uses. A failed drain reads working, so bookkeeping never keeps a Stop from interrupting. --- ...ed-agent-session-conversation-stop.test.ts | 70 ++++++++++++++++--- ...structured-agent-session-host-mutations.ts | 11 +-- .../structured-agent-session-turns.ts | 13 +++- 3 files changed, 72 insertions(+), 22 deletions(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index d6d16bc86ad4..82ab84efb21a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -12,6 +12,7 @@ import { DISPATCH_REJECTED_CANCELLED } from '../../../shared/structured-agent-se import { CancelParams } from '../../../shared/rpc-contract/structured-agent-session-params' import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import type { StructuredAgentSessionEventSink } from './structured-agent-session-event-sink' import { StructuredAgentSessionHost } from './structured-agent-session-host' import { HOST_TEST_NOW as NOW, @@ -32,6 +33,7 @@ let host: StructuredAgentSessionHost let dispatch: Mock let cancelTurn: Mock let awaitStarted: Mock> +let events: StructuredAgentSessionEventSink | undefined function eventually(assertion: () => void | Promise): Promise { return vi.waitFor(assertion, { timeout: 10_000 }) @@ -48,17 +50,25 @@ beforeEach(async () => { host = new StructuredAgentSessionHost({ store, adapter: { - acquire: async ({ fence, spawnToken }) => ({ - process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken }, - acquisitionGeneration: 'generation-1', - link: { - linkId: `link-${fence}`, - handle: { provider: 'codex' as const, threadId: THREAD }, - origin: 'created' as const, - mintedAtFence: fence, - observedAt: NOW + acquire: async ({ fence, spawnToken, events: sink }) => { + events = sink + return { + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: 1_700_000_000_000, + spawnToken + }, + acquisitionGeneration: 'generation-1', + link: { + linkId: `link-${fence}`, + handle: { provider: 'codex' as const, threadId: THREAD }, + origin: 'created' as const, + mintedAtFence: fence, + observedAt: NOW + } } - }), + }, dispatch, awaitStarted, closeSession: vi.fn(async () => true), @@ -256,6 +266,46 @@ describe('a Stop that names no turn', () => { expect(cancelTurn).toHaveBeenCalledOnce() }) + // Claude's echo accepts the send one sink write before the row of the turn it opens. + async function acceptedWithTurnRowUnlanded(): Promise { + dispatch.mockResolvedValueOnce({ + state: 'accepted', + providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-2', ordinal: 1 } + }) + const { id, result } = send('hello') + await result + await eventually(async () => expect((await submission(id))?.dispatchState).toBe('accepted')) + } + + it('interrupts a turn whose accepted send is in the journal before its row lands', async () => { + await acceptedWithTurnRowUnlanded() + const drain = host.flushStreamedEvents + vi.spyOn(host, 'flushStreamedEvents').mockImplementation((sessionId) => { + events!.appendItem( + { provider: 'legacy', agent: 'codex', sessionId, recordId: 'turn-lifecycle:turn-2' }, + { + kind: 'status', + text: 'Agent is working…', + turnLifecycle: { turnId: 'turn-2', state: 'running' } + } + ) + return drain(sessionId) + }) + + expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(cancelTurn).toHaveBeenCalledOnce() + expect(await statusRows()).toContain('Cancellation requested.') + }) + + it('still interrupts when draining the streamed rows fails', async () => { + await acceptedWithTurnRowUnlanded() + vi.spyOn(host, 'flushStreamedEvents').mockRejectedValueOnce(new Error('sink barrier failed')) + + expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(cancelTurn).toHaveBeenCalledOnce() + expect(await statusRows()).toEqual(['Cancellation requested.']) + }) + it('is a quiet no-op with nothing in flight', async () => { expect(await stop()).toMatchObject({ ok: true, value: { cancelled: false } }) expect(cancelTurn).not.toHaveBeenCalled() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index 1cf22e4616c4..e244019390d9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts @@ -21,7 +21,6 @@ import type { AgentSessionThreadGoalChange, AgentSessionThreadGoalResult } from '../../../shared/agent-session-wire' -import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' import { agentSessionFailureWords, type AgentJournalDispatchRejection @@ -29,7 +28,7 @@ import { import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import type { AgentSessionPromptRequest } from './structured-agent-session-turns-prompt' import { threadGoalPlan } from './structured-agent-session-thread-goal' -import { performCancel } from './structured-agent-session-turns' +import { isMainAgentWorkingOnceFlushed, performCancel } from './structured-agent-session-turns' import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import { admitAndRunAgentSessionMutation, @@ -188,13 +187,7 @@ export function cancelStructuredAgentSessionTurn( } // A Stop naming no turn ends nothing more unless the session reads working, by the rule // every session list and the chat's own Stop read it. - const inFlight = - params.turnId !== undefined || - isStructuredAgentSessionMainAgentWorking( - ctx.journal.activeTurnId(), - ctx.journal.submissions(), - ctx.fence - ) + const inFlight = params.turnId !== undefined || (await isMainAgentWorkingOnceFlushed(ctx)) const record = context.deps.store.getRecord(ctx.sessionId) return child && inFlight ? performCancel( 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 65733510ac1e..a9afdb174634 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 @@ -239,9 +239,16 @@ function requireSubmission( return submission } -/** Claude's echo accepts a send one sink write before its turn row lands, so read after the drain. */ -async function isMainAgentWorkingOnceFlushed(ctx: AgentSessionTurnContext): Promise { - await ctx.flushStreamedEvents() +/** Claude's echo accepts a send one sink write before its turn row lands, so read after the drain. + * A failed drain reads working: bookkeeping never talks a Stop out of stopping. */ +export async function isMainAgentWorkingOnceFlushed( + ctx: Pick +): Promise { + try { + await ctx.flushStreamedEvents() + } catch { + return true + } return isStructuredAgentSessionMainAgentWorking( ctx.journal.activeTurnId(), ctx.journal.submissions(), From b7ab836ad72a8d08001baf68cb041ce9f81cbed7 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 13:29:25 -0700 Subject: [PATCH 09/10] test(native-chat): scope the Stop tests' journal rows to the thread after the main merge --- .../structured-agent-session-conversation-stop.test.ts | 4 +++- .../agent-session-wire/structured-agent-session-turns.test.ts | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index 82ab84efb21a..c9df0c59c628 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -2,6 +2,7 @@ // queued, and interrupts a handed-over message even before the provider has opened its turn — // the gap no client can name a turn for. Against the real host, store and journal. +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' @@ -287,7 +288,8 @@ describe('a Stop that names no turn', () => { kind: 'status', text: 'Agent is working…', turnLifecycle: { turnId: 'turn-2', state: 'running' } - } + }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } ) return drain(sessionId) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts index 1801a2752331..cb313c4a4fe9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts @@ -267,7 +267,7 @@ describe('what a conversation Stop reports when the provider stopped nothing', ( text: 'Agent is working…', turnLifecycle: { turnId: 'turn-1', state: 'running' } }, - { fence: 1 } + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } ) if (turnRow === 'running') { await openTurn() From b76df8579f2a535d75d2069565509f3bcce13edc Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 18:27:14 -0700 Subject: [PATCH 10/10] test(native-chat): open the Stop-report journal on the host database after the main merge --- .../agent-session-wire/structured-agent-session-turns.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts index 6ebc6cc3a6d6..2a9ed84012b7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts @@ -253,7 +253,7 @@ describe('what a conversation Stop reports when the provider stopped nothing', ( turnRow: 'none' | 'running' | 'lands-on-flush' = 'none' ) { root = await mkdtemp(join(tmpdir(), 'orca-turn-cancel-report-')) - const journal = await journals.open({ identity: IDENTITY, journalDir: root }) + const journal = await journals.open({ identity: IDENTITY, stateDirectory: root }) const openTurn = () => journal.appendItem( {