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 6d45e77bbbb2..78c63af80752 100644 --- a/src/main/claude/claude-structured-prompt-ownership.ts +++ b/src/main/claude/claude-structured-prompt-ownership.ts @@ -15,11 +15,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] @@ -55,42 +50,14 @@ 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. + * 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). @@ -107,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) } @@ -128,10 +94,27 @@ 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. 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 || + (session.translator?.commandTurnId !== requestedTurnId && + (liveTurnId === requestedTurnId || + (liveTurnId === null && session.dispatchWaiters.length > 0)))) + ) { + 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 } @@ -151,7 +134,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 @@ -166,27 +149,12 @@ 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 => session.translator !== null && session.translator.commandTurnId === 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 && @@ -194,10 +162,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..6f2f19e5fab7 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, @@ -82,50 +81,90 @@ 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 } -/** 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') +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. */ +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 +191,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 +214,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 +238,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 +251,48 @@ 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) + completeTurn(connection, 'result-a') + 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( + [] + ) + }) + + 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/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 c1915d9bf947..6bfcb32cb863 100644 --- a/src/main/claude/claude-turn-ownership.test.ts +++ b/src/main/claude/claude-turn-ownership.test.ts @@ -6,10 +6,7 @@ import type { AgentJournalItemBody } from '../../shared/agent-session-journal-ty 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 { - 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, @@ -115,6 +112,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'] @@ -171,56 +181,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({ @@ -230,111 +192,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', - commandTurnId: null, - beginCommand: vi.fn(), - forgetCommand: vi.fn(), - commandInterruptRequested: vi.fn(), - 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]]), - 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({ @@ -346,36 +235,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 () => { @@ -449,32 +363,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 6620139bd26d..a689334c8b43 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 @@ -307,8 +307,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 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 ec9bc17e6964..dfbe77c68ba6 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' @@ -149,8 +150,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. @@ -167,18 +172,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?.({ @@ -195,3 +220,98 @@ 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) + + // 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({ + state: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + ) + 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) + +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 7bb05480f774..5793f9677d8b 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' @@ -12,6 +13,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, @@ -33,6 +35,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 }) @@ -49,17 +52,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), @@ -258,6 +269,47 @@ 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' } + }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + 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() @@ -276,4 +328,39 @@ 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([]) + }) + + // 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([]) + }) }) 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 2a4e282ec231..01143eb4bb47 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,6 +28,10 @@ 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 { + isMainAgentWorkingOnceFlushed, + performCancel +} from './structured-agent-session-turns-cancel' import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import { admitAndRunAgentSessionMutation, @@ -197,13 +200,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) if (!child || !inFlight) { if (withdrawn.length > 0) { @@ -212,10 +209,15 @@ export function cancelStructuredAgentSessionTurn( return { ok: true, value: { ...named, cancelled: withdrawn.length > 0 } } } await tookEffect() - return plan.run({ - ...ctx, - failureTextContext: structuredAgentSessionFailureWordsContext(record) - }) + return performCancel( + { ...ctx, failureTextContext: structuredAgentSessionFailureWordsContext(record) }, + { + clientOperationId: params.envelope.clientOperationId, + ...named, + stopChild: () => context.stopAgent(params.envelope.sessionId), + withdrewQueued: withdrawn.length > 0 + } + ) }) }, openForWrite(context, params.envelope) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts index 3fe70a0f4a30..eb3784b0a1d7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts @@ -10,8 +10,26 @@ import { structuredAgentSessionStopNoteIdentity } from './structured-agent-session-command-turn' import { validatePendingPrompt } from './structured-agent-session-prompt-state' +import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' +/** 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(), + ctx.fence + ) +} + export async function performCancel( ctx: AgentSessionTurnContext, input: { @@ -23,6 +41,8 @@ export async function performCancel( prompt?: { itemId: string; expectedRevision: number } /** Ends the provider child, for a running command the provider did not take the Stop on. */ stopChild?: () => Promise + /** The host already withdrew queued messages for this Stop. */ + withdrewQueued?: boolean } ): Promise> { if (input.prompt) { @@ -71,7 +91,12 @@ export async function performCancel( ...(input.prompt ? { prompt: { itemId: input.prompt.itemId } } : {}) }) cancelled = outcome.cancelled - if (!cancelled && input.turnId !== undefined) { + if (!cancelled && input.withdrewQueued && !(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. + 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 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 f5d378d8ce86..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 @@ -245,3 +245,87 @@ 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 }, + turnRow: 'none' | 'running' | 'lands-on-flush' = 'none' + ) { + root = await mkdtemp(join(tmpdir(), 'orca-turn-cancel-report-')) + const journal = await journals.open({ identity: IDENTITY, stateDirectory: root }) + const openTurn = () => + 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, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + if (turnRow === 'running') { + await openTurn() + } + 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 () => { + if (turnRow === 'lands-on-flush') { + await openTurn() + } + }, + 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 }, '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.'] + }) + }) +})