diff --git a/mobile/src/session/mobile-session-route-parity.test.ts b/mobile/src/session/mobile-session-route-parity.test.ts index 76435561664..6285f468735 100644 --- a/mobile/src/session/mobile-session-route-parity.test.ts +++ b/mobile/src/session/mobile-session-route-parity.test.ts @@ -62,8 +62,8 @@ const HOST_COMPONENT_NAMES = new Set([ 'View' ]) -const HEAD_MAIN_HOOK_SHA256 = 'c3e33699e3e3fa7e24408f3d4946fcc451e9b9419442d985c4ccde01782e5114' -const HEAD_HOOK_BINDING_SHA256 = '7f907e028893721d662eeee0aa9002ad1e00359948f39fb148d274596cd9b3c0' +const HEAD_MAIN_HOOK_SHA256 = '3e97fba5684419f936b7e49986d6c0a4accdd5b342d2dbde25e1f38bbada5dab' +const HEAD_HOOK_BINDING_SHA256 = 'a0f140869570f13c811e0e241581691d9ef03b23477bbedfd8d1ad72f0a21a3a' const HEAD_CALLBACK_IDENTITY_SHA256 = '2a9e4825df007f6ef53b81aa5004991d6318eee7507b44d625c07e630be432eb' const HEAD_CALLBACK_BODY_SHA256 = 'af7f3c62954250d4be7ee432ecd10dc2689792aad8230fed2d1d68bbc892d776' @@ -472,7 +472,7 @@ describe('mobile session route extraction parity', () => { const contentBindings = CONTENT_COMPONENT_NAMES.flatMap( (name) => readHookFacts(name, definitions).bindings ) - expect(main.hooks).toHaveLength(267) + expect(main.hooks).toHaveLength(268) expect(hash(main.hooks)).toBe(HEAD_MAIN_HOOK_SHA256) expect(hash(main.bindings)).toBe(HEAD_HOOK_BINDING_SHA256) expect(main.callbacks).toHaveLength(77) diff --git a/mobile/src/session/mobile-structured-agent-session-test-fixtures.ts b/mobile/src/session/mobile-structured-agent-session-test-fixtures.ts new file mode 100644 index 00000000000..14061baee3f --- /dev/null +++ b/mobile/src/session/mobile-structured-agent-session-test-fixtures.ts @@ -0,0 +1,130 @@ +import type { + AgentJournalRenderItem, + AgentJournalResolution +} from '../../../src/shared/agent-session-journal-types' +import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire' + +type SnapshotEvent = Extract + +export function snapshotEvent(fence = 3): SnapshotEvent { + return { + type: 'snapshot', + sessionId: 'session-1', + fence, + page: { + sessionId: 'session-1', + epoch: 'epoch-1', + fence, + direction: 'tail', + items: [], + removedItemIds: [], + submissions: [], + window: { + oldest: null, + newest: null, + nextCursor: { epoch: 'epoch-1', sequence: 0 } + }, + liveCursor: { epoch: 'epoch-1', sequence: 0 }, + hasOlder: false, + hasNewer: false + } + } +} + +export function snapshotWithMessage(): SnapshotEvent { + const event = snapshotEvent() + return { + ...event, + page: { + ...event.page, + items: [ + { + itemId: 'msg-1', + revision: 1, + sequence: 1, + observedAt: 10, + body: { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'sent before the blip' }] + } + } + ], + window: { + oldest: { epoch: 'epoch-1', sequence: 1 }, + newest: { epoch: 'epoch-1', sequence: 1 }, + nextCursor: { epoch: 'epoch-1', sequence: 2 } + }, + liveCursor: { epoch: 'epoch-1', sequence: 1 } + } + } +} + +function pendingResolution(): AgentJournalResolution { + return { + state: 'pending', + selectedOptionId: null, + resolvedBy: null, + resolvedAt: null + } +} + +export function approvalItem(): AgentJournalRenderItem { + return { + itemId: 'approval-1', + revision: 2, + sequence: 1, + observedAt: 10, + body: { + kind: 'approval', + title: 'Allow Bash?', + detail: 'rm -rf build', + options: [ + { id: 'allow-once', label: 'Allow once' }, + { id: 'deny', label: 'Deny' } + ], + resolution: pendingResolution() + } + } +} + +export function approvalItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem { + return { ...approvalItem(), itemId, revision } +} + +export function questionItem(): AgentJournalRenderItem { + return { + itemId: 'question-1', + revision: 7, + sequence: 2, + observedAt: 12, + body: { + kind: 'question', + question: 'Pick destination', + freeTextQuestionId: 'free-q', + options: [ + { id: 'choice-a', label: 'Choice A' }, + { id: 'choice-b', label: 'Choice B' } + ], + resolution: pendingResolution() + } + } +} + +export function questionItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem { + return { ...questionItem(), itemId, revision } +} + +export function runningStatusItem(): AgentJournalRenderItem { + return { + itemId: 'status-1', + revision: 1, + sequence: 3, + observedAt: 14, + body: { + kind: 'status', + text: 'Working', + turnLifecycle: { turnId: 'turn-1', state: 'running' } + } + } +} diff --git a/mobile/src/session/use-mobile-native-chat-controller.ts b/mobile/src/session/use-mobile-native-chat-controller.ts index 4087694b567..121137a4146 100644 --- a/mobile/src/session/use-mobile-native-chat-controller.ts +++ b/mobile/src/session/use-mobile-native-chat-controller.ts @@ -35,6 +35,7 @@ export function useMobileNativeChatController(args: { nativeChatInputLeaseReady: boolean /** Live socket state; the lease collapses on disconnect but one render later. */ connState: ConnectionState + promptCancelSupported?: boolean onSendError: (message: string) => void /** Retires a held failure banner. Any accepted chat write clears it — a delivered * answer or permission reply must not sit under a stale "not sent". */ @@ -51,6 +52,7 @@ export function useMobileNativeChatController(args: { nativeChatTranscriptIsLocalReadable, nativeChatInputLeaseReady, connState, + promptCancelSupported, onSendError, onSendResolved } = args @@ -89,7 +91,9 @@ export function useMobileNativeChatController(args: { sourceIdentity, enabled: showNativeChat, connState, - onSendError + promptCancelSupported, + onSendError, + onCancelResolved: onSendResolved }) const { composerText: chatComposerText, diff --git a/mobile/src/session/use-mobile-native-chat-session-lane.ts b/mobile/src/session/use-mobile-native-chat-session-lane.ts index fc465de902b..99872c310d4 100644 --- a/mobile/src/session/use-mobile-native-chat-session-lane.ts +++ b/mobile/src/session/use-mobile-native-chat-session-lane.ts @@ -16,7 +16,9 @@ export function useMobileNativeChatSessionLane({ sourceIdentity, enabled, connState, - onSendError + promptCancelSupported, + onSendError, + onCancelResolved }: { client: RpcClient | null structured: boolean @@ -29,7 +31,9 @@ export function useMobileNativeChatSessionLane({ sourceIdentity: Parameters[0]['sourceIdentity'] enabled: boolean connState: ConnectionState + promptCancelSupported?: boolean onSendError: (message: string) => void + onCancelResolved: () => void }): { structuredSession: ReturnType session: ReturnType @@ -50,7 +54,9 @@ export function useMobileNativeChatSessionLane({ // reacquire the provider without clearing the cached transcript. connected: connState === 'connected', agent: structured ? agent : null, - onSendError + promptCancelSupported, + onSendError, + onCancelResolved }) return { structuredSession, diff --git a/mobile/src/session/use-mobile-session-native-chat-dictation.ts b/mobile/src/session/use-mobile-session-native-chat-dictation.ts index 6046cba1059..a555d32bbb0 100644 --- a/mobile/src/session/use-mobile-session-native-chat-dictation.ts +++ b/mobile/src/session/use-mobile-session-native-chat-dictation.ts @@ -1,5 +1,7 @@ import { useEffect, useCallback } from 'react' import { useFocusEffect } from 'expo-router' +import { AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version' +import { useHostProtocolGates } from '../components/HostProtocolGate' import { useMobileDictation } from '../hooks/use-mobile-dictation' import { triggerError } from '../platform/haptics' import { @@ -22,6 +24,7 @@ export function useMobileSessionNativeChatDictation( scope: MobileSessionFeedbackCapabilitiesModel, sendLiveTerminalInput: (handle: string, bytes: string) => Promise ) { + const { hostCapabilities, statusPending } = useHostProtocolGates() const { hostId, worktreeId, @@ -72,6 +75,9 @@ export function useMobileSessionNativeChatDictation( nativeChatTranscriptIsLocalReadable, nativeChatInputLeaseReady, connState, + promptCancelSupported: statusPending + ? undefined + : hostCapabilities.includes(AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY), onSendError: nativeChatSendError.show, onSendResolved: nativeChatSendError.clear }) diff --git a/mobile/src/session/use-mobile-structured-agent-session-cancel.ts b/mobile/src/session/use-mobile-structured-agent-session-cancel.ts new file mode 100644 index 00000000000..5cb6cc79c4d --- /dev/null +++ b/mobile/src/session/use-mobile-structured-agent-session-cancel.ts @@ -0,0 +1,90 @@ +import { useCallback } from 'react' +import type { AgentSessionCancelResult } from '../../../src/shared/agent-session-wire' +import { activeStructuredAgentSessionTurnId } from '../../../src/shared/structured-agent-session-live-turn' +import { AGENT_SESSION_PROMPT_CANCEL_UPDATE_REQUIRED_MESSAGE } from '../../../src/shared/protocol-version' +import type { RpcClient } from '../transport/rpc-client' +import { + requestStructuredAgentSessionMutation, + retainStructuredSessionOperationId +} from './mobile-structured-agent-session-rpc' +import { + pendingStructuredApproval, + pendingStructuredQuestion +} from './mobile-structured-agent-prompts' +import type { useMobileStructuredAgentState } from './use-mobile-structured-agent-state' + +export function useMobileStructuredAgentSessionCancel(args: { + client: RpcClient | null + sessionId: string | null + sessionKey: string + enabled: boolean + promptCancelSupported?: boolean + stateRef: ReturnType['stateRef'] + operationIdsRef: { current: Map } + onSendError: (message: string) => void + onCancelResolved?: () => void +}): () => void { + const { + client, + sessionId, + sessionKey, + enabled, + promptCancelSupported, + stateRef, + operationIdsRef, + onSendError, + onCancelResolved + } = args + return useCallback(() => { + const current = stateRef.current + const turnId = activeStructuredAgentSessionTurnId(current.items) + const prompt = + current.items.find(pendingStructuredApproval) ?? current.items.find(pendingStructuredQuestion) + if (!client || !sessionId || !enabled || current.fence === null || (!turnId && !prompt)) { + onSendError('Stop not sent') + return + } + if (prompt && promptCancelSupported !== true) { + onSendError( + promptCancelSupported === false + ? AGENT_SESSION_PROMPT_CANCEL_UPDATE_REQUIRED_MESSAGE + : 'Checking desktop capabilities — try again in a moment' + ) + return + } + const fields = { + ...(turnId ? { turnId } : {}), + ...(promptCancelSupported && prompt + ? { prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } } + : {}) + } + const key = `${sessionKey}:agentSession.cancel:${JSON.stringify(fields)}` + const clientOperationId = retainStructuredSessionOperationId( + operationIdsRef.current, + key, + operationIdsRef.current.get(key) + ) + void requestStructuredAgentSessionMutation({ + client, + method: 'agentSession.cancel', + fingerprintMethod: 'agentSession.cancel', + sessionId, + expectedRuntimeFence: current.fence, + fields, + clientOperationId + }).then((result) => { + if (result.status !== 'unknown') { + operationIdsRef.current.delete(key) + } + if (result.status === 'unknown') { + onSendError('Stop unconfirmed — check chat before retrying') + } else if (result.status === 'refused') { + onSendError(result.message) + } else if (result.status === 'failed') { + onSendError(result.message === 'Request not sent' ? 'Stop not sent' : result.message) + } else { + onCancelResolved?.() + } + }) + }, [client, enabled, onCancelResolved, onSendError, promptCancelSupported, sessionId, sessionKey]) +} diff --git a/mobile/src/session/use-mobile-structured-agent-session.test.tsx b/mobile/src/session/use-mobile-structured-agent-session.test.tsx index d888e74c921..c25f6163c03 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session.test.tsx @@ -1,143 +1,24 @@ import { createElement } from 'react' import { act, create, type ReactTestRenderer } from 'react-test-renderer' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import type { - AgentJournalRenderItem, - AgentJournalResolution -} from '../../../src/shared/agent-session-journal-types' -import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire' import type { RpcClient } from '../transport/rpc-client' import { markRpcDeliveryUnknown } from '../transport/rpc-delivery-ambiguity' import { formatQuestionFreeTextAnswer } from './mobile-native-chat-question' import { useMobileStructuredAgentSession } from './use-mobile-structured-agent-session' +import { + approvalItem, + approvalItemWithIdentity, + questionItem, + questionItemWithIdentity, + runningStatusItem, + snapshotEvent, + snapshotWithMessage +} from './mobile-structured-agent-session-test-fixtures' function ok(result: unknown) { return { ok: true, result, _meta: { runtimeId: 'runtime-1' } } } -function snapshotEvent(fence = 3): AgentSessionSubscribeEvent { - return { - type: 'snapshot', - sessionId: 'session-1', - fence, - page: { - sessionId: 'session-1', - epoch: 'epoch-1', - fence, - direction: 'tail', - items: [], - removedItemIds: [], - submissions: [], - window: { - oldest: null, - newest: null, - nextCursor: { epoch: 'epoch-1', sequence: 0 } - }, - liveCursor: { epoch: 'epoch-1', sequence: 0 }, - hasOlder: false, - hasNewer: false - } - } -} - -function snapshotWithMessage(): AgentSessionSubscribeEvent { - const event = snapshotEvent() - return { - ...event, - page: { - ...event.page, - items: [ - { - itemId: 'msg-1', - revision: 1, - sequence: 1, - observedAt: 10, - body: { - kind: 'message', - role: 'user', - blocks: [{ type: 'text', text: 'sent before the blip' }] - } - } - ], - window: { - oldest: { epoch: 'epoch-1', sequence: 1 }, - newest: { epoch: 'epoch-1', sequence: 1 }, - nextCursor: { epoch: 'epoch-1', sequence: 2 } - }, - liveCursor: { epoch: 'epoch-1', sequence: 1 } - } - } as AgentSessionSubscribeEvent -} - -function pendingResolution(): AgentJournalResolution { - return { - state: 'pending', - selectedOptionId: null, - resolvedBy: null, - resolvedAt: null - } -} - -function approvalItem(): AgentJournalRenderItem { - return { - itemId: 'approval-1', - revision: 2, - sequence: 1, - observedAt: 10, - body: { - kind: 'approval', - title: 'Allow Bash?', - detail: 'rm -rf build', - options: [ - { id: 'allow-once', label: 'Allow once' }, - { id: 'deny', label: 'Deny' } - ], - resolution: pendingResolution() - } - } -} - -function approvalItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem { - return { ...approvalItem(), itemId, revision } -} - -function questionItem(): AgentJournalRenderItem { - return { - itemId: 'question-1', - revision: 7, - sequence: 2, - observedAt: 12, - body: { - kind: 'question', - question: 'Pick destination', - freeTextQuestionId: 'free-q', - options: [ - { id: 'choice-a', label: 'Choice A' }, - { id: 'choice-b', label: 'Choice B' } - ], - resolution: pendingResolution() - } - } -} - -function questionItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem { - return { ...questionItem(), itemId, revision } -} - -function runningStatusItem(): AgentJournalRenderItem { - return { - itemId: 'status-1', - revision: 1, - sequence: 3, - observedAt: 14, - body: { - kind: 'status', - text: 'Working', - turnLifecycle: { turnId: 'turn-1', state: 'running' } - } - } -} - async function defaultSendRequest(method: string, params?: Record) { if (method === 'agentSession.send') { return ok({ @@ -217,6 +98,7 @@ describe('useMobileStructuredAgentSession', () => { let hook: ReturnType | null = null let listener: ((value: unknown) => void) | null = null const onSendError = vi.fn() + const onCancelResolved = vi.fn() const unsubscribe = vi.fn() const sendRequest = vi.fn(defaultSendRequest) const subscribe = vi.fn((_method: string, _params: unknown, onData: (value: unknown) => void) => { @@ -232,11 +114,13 @@ describe('useMobileStructuredAgentSession', () => { sessionId = 'session-1', agent = 'codex', connected = true, + promptCancelSupported = false, sourceIdentity = 'host-a\0workspace-a' }: { sessionId?: string | null agent?: string | null connected?: boolean + promptCancelSupported?: boolean sourceIdentity?: string }): null { hook = useMobileStructuredAgentSession({ @@ -246,7 +130,9 @@ describe('useMobileStructuredAgentSession', () => { enabled: true, connected, agent, - onSendError + promptCancelSupported, + onSendError, + onCancelResolved } as never) return null } @@ -760,6 +646,92 @@ describe('useMobileStructuredAgentSession', () => { expect(onSendError).toHaveBeenCalledWith('Stop unconfirmed — check chat before retrying') }) + it('targets the displayed prompt revision when the host supports prompt cancellation', async () => { + act(() => { + renderer = create(createElement(Harness, { promptCancelSupported: true })) + }) + await vi.waitFor(() => expect(listener).toEqual(expect.any(Function))) + act(() => + listener?.({ + ...snapshotEvent(3), + page: { + ...snapshotEvent(3).page, + items: [approvalItem()] + } + }) + ) + + act(() => hook!.cancel()) + + await vi.waitFor(() => + expect(sendRequest).toHaveBeenCalledWith( + 'agentSession.cancel', + expect.objectContaining({ + prompt: { itemId: 'approval-1', expectedRevision: 2 } + }), + expect.any(Object) + ) + ) + const cancelParams = sendRequest.mock.calls.find( + ([method]) => method === 'agentSession.cancel' + )?.[1] + expect(cancelParams).not.toHaveProperty('turnId') + }) + + it('retires a held error after the host accepts structured Stop', async () => { + act(() => { + renderer = create(createElement(Harness)) + }) + await vi.waitFor(() => expect(listener).toEqual(expect.any(Function))) + act(() => + listener?.({ + ...snapshotEvent(3), + page: { ...snapshotEvent(3).page, items: [runningStatusItem()] } + }) + ) + sendRequest.mockImplementation(async (method, params) => + method === 'agentSession.cancel' + ? ok({ + ok: true, + replayed: false, + fence: 3, + cursor: { epoch: 'epoch-1', sequence: 4 }, + value: { turnId: 'turn-1', cancelled: true } + }) + : defaultSendRequest(method, params) + ) + + await act(async () => { + hook!.cancel() + await Promise.resolve() + }) + + expect(onCancelResolved).toHaveBeenCalledOnce() + }) + + it('refuses prompt cancellation for a host without prompt cancellation', async () => { + act(() => { + renderer = create(createElement(Harness)) + }) + await vi.waitFor(() => expect(listener).toEqual(expect.any(Function))) + act(() => + listener?.({ + ...snapshotEvent(3), + page: { + ...snapshotEvent(3).page, + items: [runningStatusItem(), approvalItem()] + } + }) + ) + + act(() => hook!.cancel()) + + expect(sendRequest.mock.calls.some(([method]) => method === 'agentSession.cancel')).toBe(false) + expect(onSendError).toHaveBeenCalledWith( + 'Cancelling a pending prompt requires a newer Orca server. Update the server and try again.' + ) + }) + it('releases a landed hold when the structured tab unmounts', async () => { act(() => { renderer = create(createElement(Harness)) diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts index 938e881aec7..b4e98ed9041 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.ts +++ b/mobile/src/session/use-mobile-structured-agent-session.ts @@ -1,9 +1,6 @@ import { useCallback, useEffect, useMemo, useRef } from 'react' import { dispatchMobileStructuredCommand } from './mobile-structured-composer-command' -import type { - AgentSessionCancelResult, - AgentSessionSendResult -} from '../../../src/shared/agent-session-wire' +import type { AgentSessionSendResult } from '../../../src/shared/agent-session-wire' import { structuredAgentSessionSendBody, type StructuredAgentSessionAttachment @@ -38,6 +35,7 @@ import { useMobileStructuredAgentState } from './use-mobile-structured-agent-sta import { useMobileStructuredPromptResponses } from './use-mobile-structured-prompt-responses' import { useMobileStructuredAgentOptions } from './use-mobile-structured-agent-options' import { useMobileStructuredAgentTurnTiming } from './use-mobile-structured-agent-turn-timing' +import { useMobileStructuredAgentSessionCancel } from './use-mobile-structured-agent-session-cancel' type StructuredMobileAttachment = StructuredAgentSessionAttachment & { id?: string } @@ -70,9 +68,21 @@ export function useMobileStructuredAgentSession(args: { /** Live transport only; gates the connection-scoped hold, nothing else. */ connected: boolean agent: string | null + promptCancelSupported?: boolean onSendError: (message: string) => void + onCancelResolved?: () => void }): StructuredMobileSession { - const { agent, client, connected, sessionId, sourceIdentity = '', enabled, onSendError } = args + const { + agent, + client, + connected, + sessionId, + sourceIdentity = '', + enabled, + promptCancelSupported, + onSendError, + onCancelResolved + } = args const sessionKey = encodeNativeChatTranscriptIdentity([sourceIdentity, agent, sessionId]) const operationIdsRef = useRef(new Map()) const commandPendingRef = useRef(false) @@ -237,37 +247,17 @@ export function useMobileStructuredAgentSession(args: { onSendError }) - const cancel = useCallback(() => { - const current = stateRef.current - const turnId = activeStructuredAgentSessionTurnId(current.items) - if (!client || !sessionId || !enabled || current.fence === null || !turnId) { - onSendError('Stop not sent') - return - } - const fields = { turnId } - const key = `${sessionKey}:agentSession.cancel:${JSON.stringify(fields)}` - const clientOperationId = retainOperationId(key, operationIdsRef.current.get(key)) - void requestStructuredAgentSessionMutation({ - client, - method: 'agentSession.cancel', - fingerprintMethod: 'agentSession.cancel', - sessionId, - expectedRuntimeFence: current.fence, - fields, - clientOperationId - }).then((result) => { - if (result.status !== 'unknown') { - operationIdsRef.current.delete(key) - } - if (result.status === 'unknown') { - onSendError('Stop unconfirmed — check chat before retrying') - } else if (result.status === 'refused') { - onSendError(result.message) - } else if (result.status === 'failed') { - onSendError(result.message === 'Request not sent' ? 'Stop not sent' : result.message) - } - }) - }, [client, enabled, onSendError, sessionId, sessionKey]) + const cancel = useMobileStructuredAgentSessionCancel({ + client, + sessionId, + sessionKey, + enabled, + promptCancelSupported, + stateRef, + operationIdsRef, + onSendError, + onCancelResolved + }) const messages = useMemo( () => projectStructuredAgentSessionMessages(state.items, [], state.submissions), diff --git a/mobile/src/transport/host-status-gates.ts b/mobile/src/transport/host-status-gates.ts index 91f0205a5c7..77602562e85 100644 --- a/mobile/src/transport/host-status-gates.ts +++ b/mobile/src/transport/host-status-gates.ts @@ -1,9 +1,10 @@ import { useEffect, useState } from 'react' import type { RpcClient } from './rpc-client' -import type { ConnectionState, RpcSuccess } from './types' +import type { ConnectionState } from './types' import { evaluateCompat, type CompatVerdict } from './protocol-compat' import type { DesktopStatus } from '../worktree/host-worktree-rpc-types' import { normalizeHostAppVersion, recordHostAppVersion } from './host-app-version-store' +import { startRuntimeStatusProbe } from './runtime-capability-probe' export type HostStatusGates = { hostCapabilities: string[] @@ -39,68 +40,39 @@ export function useHostStatusGates(args: { setUnverified(true) return } - let cancelled = false const requestClient = client const settle = (gates: Omit) => { setLoaded({ hostId, client: requestClient, ...gates }) setUnverified(false) } - void (async () => { - try { - const response = await requestClient.sendRequest('status.get') - if (cancelled) { - return - } - if (!response.ok) { - settle({ - hostCapabilities: [], - floatingWorkspaceEnabled: false, - desktopAppVersion: null, - compatVerdict: { kind: 'ok' } - }) - return - } - const status = (response as RpcSuccess).result as DesktopStatus & { - capabilities?: string[] - } - const verdict = evaluateCompat({ - desktopProtocolVersion: status.protocolVersion, - desktopMinCompatibleMobileVersion: status.minCompatibleMobileVersion - }) - const desktopAppVersion = normalizeHostAppVersion(status.appVersion) - if (hostId && desktopAppVersion) { - void recordHostAppVersion(hostId, desktopAppVersion) - } - settle({ - hostCapabilities: status.capabilities ?? [], - floatingWorkspaceEnabled: status.floatingWorkspaceEnabled === true, - desktopAppVersion, - compatVerdict: verdict - }) - if (verdict.kind === 'blocked') { - // Why: support breadcrumb to confirm a block fired vs a render bug; no PII, just version ints. - console.warn('[protocol-compat] blocked', { - reason: verdict.reason, - desktopVersion: verdict.desktopVersion, - requiredMobileVersion: verdict.requiredMobileVersion, - requiredDesktopVersion: verdict.requiredDesktopVersion - }) - } - } catch { - // Why: a transient status failure must not trap navigation; conservative feature gates remain disabled. - if (!cancelled) { - settle({ - hostCapabilities: [], - floatingWorkspaceEnabled: false, - desktopAppVersion: null, - compatVerdict: { kind: 'ok' } - }) - } + return startRuntimeStatusProbe(requestClient, (result) => { + const status = result as DesktopStatus & { + capabilities?: string[] } - })() - return () => { - cancelled = true - } + const verdict = evaluateCompat({ + desktopProtocolVersion: status.protocolVersion, + desktopMinCompatibleMobileVersion: status.minCompatibleMobileVersion + }) + const desktopAppVersion = normalizeHostAppVersion(status.appVersion) + if (hostId && desktopAppVersion) { + void recordHostAppVersion(hostId, desktopAppVersion) + } + settle({ + hostCapabilities: status.capabilities ?? [], + floatingWorkspaceEnabled: status.floatingWorkspaceEnabled === true, + desktopAppVersion, + compatVerdict: verdict + }) + if (verdict.kind === 'blocked') { + // Why: support breadcrumb to confirm a block fired vs a render bug; no PII, just version ints. + console.warn('[protocol-compat] blocked', { + reason: verdict.reason, + desktopVersion: verdict.desktopVersion, + requiredMobileVersion: verdict.requiredMobileVersion, + requiredDesktopVersion: verdict.requiredDesktopVersion + }) + } + }) }, [client, connState, hostId]) // Why: effects run after render, so key loaded gates by host and client to fail closed during route reuse. diff --git a/mobile/src/transport/runtime-capability-probe.test.ts b/mobile/src/transport/runtime-capability-probe.test.ts index 2272c25610f..e1a0a1d34e3 100644 --- a/mobile/src/transport/runtime-capability-probe.test.ts +++ b/mobile/src/transport/runtime-capability-probe.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { startRuntimeCapabilityProbe } from './runtime-capability-probe' +import { startRuntimeCapabilityProbe, startRuntimeStatusProbe } from './runtime-capability-probe' import { LogicalClientCutoverError } from './stable-logical-rpc-client' import type { RpcClient } from './rpc-client' import type { RpcResponse } from './types' @@ -48,6 +48,26 @@ describe('startRuntimeCapabilityProbe', () => { cancel() }) + it('shares retry semantics with full host-status consumers', async () => { + const status = { + ok: true as const, + id: '1', + result: { appVersion: '1.2.3', capabilities: ['a.v1'] }, + _meta: { runtimeId: 'r1' } + } + const { client, calls } = makeClient([new Error('timeout'), status]) + const seen: unknown[] = [] + const cancel = startRuntimeStatusProbe(client, (result) => seen.push(result)) + await flushMicrotasks() + expect(seen).toEqual([]) + + await vi.advanceTimersByTimeAsync(1_000) + + expect(seen).toEqual([status.result]) + expect(calls()).toBe(2) + cancel() + }) + it('treats malformed capabilities as unsupported', async () => { const response: RpcResponse = { ok: true, diff --git a/mobile/src/transport/runtime-capability-probe.ts b/mobile/src/transport/runtime-capability-probe.ts index 6bec0ca05bd..33fe5a61439 100644 --- a/mobile/src/transport/runtime-capability-probe.ts +++ b/mobile/src/transport/runtime-capability-probe.ts @@ -12,6 +12,23 @@ const FAILURE_RETRY_MAX_DELAY_MS = 15_000 export function startRuntimeCapabilityProbe( client: Pick, onCapabilities: (capabilities: readonly string[]) => void +): () => void { + return startRuntimeStatusProbe(client, (result) => { + const rawCapabilities = + result && typeof result === 'object' + ? (result as { capabilities?: unknown }).capabilities + : null + const capabilities = + Array.isArray(rawCapabilities) && rawCapabilities.every((value) => typeof value === 'string') + ? rawCapabilities + : [] + onCapabilities(capabilities) + }) +} + +export function startRuntimeStatusProbe( + client: Pick, + onStatus: (status: unknown) => void ): () => void { let cancelled = false let retryTimer: ReturnType | null = null @@ -27,17 +44,7 @@ export function startRuntimeCapabilityProbe( scheduleRetry(false) return } - const result = (response as RpcSuccess).result - const rawCapabilities = - result && typeof result === 'object' - ? (result as { capabilities?: unknown }).capabilities - : null - const capabilities = - Array.isArray(rawCapabilities) && - rawCapabilities.every((value) => typeof value === 'string') - ? rawCapabilities - : [] - onCapabilities(capabilities) + onStatus((response as RpcSuccess).result) }, (error: unknown) => { if (cancelled) { diff --git a/src/main/claude/claude-host-prompt-cancellation.ts b/src/main/claude/claude-host-prompt-cancellation.ts new file mode 100644 index 00000000000..af6fed03e74 --- /dev/null +++ b/src/main/claude/claude-host-prompt-cancellation.ts @@ -0,0 +1,46 @@ +import type { ClaudePendingPrompt } from './claude-structured-prompt-types' + +export class ClaudeHostPromptCancellation { + private readonly active = new Set() + private readonly suppressedProviderEvents = new Set() + + begin(prompt: ClaudePendingPrompt | null, turnId: string): ClaudePendingPrompt | null { + if (!prompt || prompt.turnId !== turnId) { + return null + } + this.active.add(prompt) + return prompt + } + + finish( + prompt: ClaudePendingPrompt, + confirmed: boolean, + isPending: () => boolean, + settle: () => void + ): boolean { + if (confirmed && isPending()) { + settle() + prompt.settle(null) + } + this.active.delete(prompt) + const suppressed = this.suppressedProviderEvents.delete(prompt) + return suppressed && !confirmed + } + + consume(prompt: ClaudePendingPrompt): boolean { + const consumed = this.active.delete(prompt) + if (consumed) { + this.suppressedProviderEvents.add(prompt) + } + return consumed + } + + forget(prompt: ClaudePendingPrompt): void { + this.active.delete(prompt) + } + + clear(): void { + this.active.clear() + this.suppressedProviderEvents.clear() + } +} diff --git a/src/main/claude/claude-structured-control-actions.test.ts b/src/main/claude/claude-structured-control-actions.test.ts index 168ce558f53..a80f6ed04fb 100644 --- a/src/main/claude/claude-structured-control-actions.test.ts +++ b/src/main/claude/claude-structured-control-actions.test.ts @@ -87,6 +87,26 @@ describe('cancelClaudeTurn', () => { }) describe('answerClaudePrompt', () => { + it('groups journal rows only under the live turn that owns the prompt', () => { + const prompts = new ClaudePromptRegistry() + const prompt = prompts.register({ + requestId: 'question-1', + turnId: 'turn-1', + toolName: 'AskUserQuestion', + toolUseId: 'tool-1', + input: { questions: [{ question: 'One?' }, { question: 'Two?' }] }, + suggestions: [], + settle: vi.fn() + })! + prompts.bindJournalItemId('journal-q1', prompt.promptKey, 'One?') + prompts.bindJournalItemId('journal-q2', prompt.promptKey, 'Two?') + + expect(prompts.cancellation('journal-q1')).toEqual({ + turnId: 'turn-1', + itemIds: ['journal-q1', 'journal-q2'] + }) + }) + it('settles the pending prompt callback and forgets it', async () => { const prompts = new ClaudePromptRegistry() const settle = vi.fn() diff --git a/src/main/claude/claude-structured-inbound-control.test.ts b/src/main/claude/claude-structured-inbound-control.test.ts index 07be4bbb516..bda35272a6b 100644 --- a/src/main/claude/claude-structured-inbound-control.test.ts +++ b/src/main/claude/claude-structured-inbound-control.test.ts @@ -30,6 +30,7 @@ function callbacksFor() { const { canUseTool, onUserDialog } = buildClaudePermissionCallbacks({ sessionId: 'session-1', prompts, + activeTurnId: () => 'turn-1', emit }) return { prompts, emit, canUseTool, onUserDialog } @@ -95,6 +96,24 @@ describe('Claude permission callbacks', () => { expect(control.prompts.find('perm-3')).toBeNull() }) + it('does not tombstone a prompt whose abort belongs to confirmed host cancellation', async () => { + const control = callbacksFor() + const controller = new AbortController() + const answered = control.canUseTool( + 'Bash', + { command: 'ls' }, + permissionOptions('perm-host-cancel', 'tool-host-cancel', controller.signal) + ) + control.prompts.bindJournalItemId('journal-host-cancel', 'perm-host-cancel') + expect(control.prompts.beginHostCancellation('journal-host-cancel', 'turn-1')).not.toBeNull() + + controller.abort() + + await expect(answered).resolves.toBeNull() + expect(control.emit).toHaveBeenCalledTimes(1) + expect(control.emit).toHaveBeenCalledWith(expect.objectContaining({ type: 'prompt' })) + }) + it('cancels a request whose abort raced ahead of delivery without emitting a prompt', async () => { const control = callbacksFor() const controller = new AbortController() diff --git a/src/main/claude/claude-structured-inbound-control.ts b/src/main/claude/claude-structured-inbound-control.ts index 343e76d4ea5..94e5cd9bcff 100644 --- a/src/main/claude/claude-structured-inbound-control.ts +++ b/src/main/claude/claude-structured-inbound-control.ts @@ -24,6 +24,7 @@ export type ClaudeBlockingControlSubtype = keyof typeof CLAUDE_BLOCKING_CONTROL_ export type ClaudePermissionCallbackDeps = { sessionId: string prompts: ClaudePromptRegistry + activeTurnId?: () => string | null emit: (event: ClaudeStructuredSessionEvent) => void } @@ -53,6 +54,7 @@ export function buildClaudePermissionCallbacks(deps: ClaudePermissionCallbackDep new Promise((resolve) => { const prompt = deps.prompts.register({ requestId: options.requestId, + turnId: deps.activeTurnId?.() ?? null, toolName, toolUseId: options.toolUseID, input, @@ -64,12 +66,15 @@ export function buildClaudePermissionCallbacks(deps: ClaudePermissionCallbackDep return } const cancel = (): void => { + const hostCancellation = deps.prompts.consumeHostCancellation(prompt) if (deps.prompts.forgetIfPending(prompt)) { - deps.emit({ - type: 'prompt-cancelled', - sessionId: deps.sessionId, - promptKey: prompt.promptKey - }) + if (!hostCancellation) { + deps.emit({ + type: 'prompt-cancelled', + sessionId: deps.sessionId, + promptKey: prompt.promptKey + }) + } // Null is the SDK's "no response written" sentinel: a cancelled request must not // be answered, only forgotten. resolve(null) diff --git a/src/main/claude/claude-structured-prompt-replies.ts b/src/main/claude/claude-structured-prompt-replies.ts index deec74b7308..5c6bf0fd70b 100644 --- a/src/main/claude/claude-structured-prompt-replies.ts +++ b/src/main/claude/claude-structured-prompt-replies.ts @@ -1,32 +1,17 @@ import { decodeAgentSessionQuestionAnswers } from '../../shared/agent-session-question-answer' +import { ClaudeHostPromptCancellation } from './claude-host-prompt-cancellation' +import type { + ClaudePendingPrompt, + ClaudePromptRegistration +} from './claude-structured-prompt-types' export const CLAUDE_APPROVAL_DECISIONS = ['allow', 'allowForSession', 'deny', 'cancel'] as const export type ClaudeApprovalDecision = (typeof CLAUDE_APPROVAL_DECISIONS)[number] - -/** Settles the SDK's `canUseTool` promise; `null` is the SDK's "no response written" sentinel. */ -export type ClaudePromptSettle = (response: Record | null) => void - -export type ClaudePendingPrompt = { - requestId: string - promptKey: string - toolUseId: string - toolName: string - kind: 'approval' | 'question' - input: Record - suggestions: unknown[] - questionIds: readonly string[] - answers: Map - settle: ClaudePromptSettle -} - -export type ClaudePromptRegistration = { - requestId: string - toolName: string - toolUseId: string - input: Record - suggestions: unknown[] - settle: ClaudePromptSettle -} +export type { + ClaudePendingPrompt, + ClaudePromptRegistration, + ClaudePromptSettle +} from './claude-structured-prompt-types' type PromptBinding = { address: string @@ -108,6 +93,7 @@ export function decodeClaudeQuestionOptionId( export class ClaudePromptRegistry { private readonly prompts = new Map() private readonly journalBindings = new Map() + private readonly hostCancellation = new ClaudeHostPromptCancellation() register(registration: ClaudePromptRegistration): ClaudePendingPrompt | null { const toolUseId = readString(registration.toolUseId) @@ -120,6 +106,7 @@ export class ClaudePromptRegistry { const prompt: ClaudePendingPrompt = { requestId: registration.requestId, promptKey: registration.requestId, + turnId: registration.turnId ?? null, toolUseId, toolName, kind: questions.length > 0 ? 'question' : 'approval', @@ -133,7 +120,6 @@ export class ClaudePromptRegistry { return prompt } - /** True only if the prompt was still pending; lets an abort and an answer race settle once. */ forgetIfPending(prompt: ClaudePendingPrompt): boolean { if (!this.prompts.has(prompt.promptKey)) { return false @@ -142,6 +128,23 @@ export class ClaudePromptRegistry { return true } + beginHostCancellation(itemId: string, turnId: string): ClaudePendingPrompt | null { + return this.hostCancellation.begin(this.find(itemId)?.prompt ?? null, turnId) + } + + finishHostCancellation(prompt: ClaudePendingPrompt, confirmed: boolean): boolean { + return this.hostCancellation.finish( + prompt, + confirmed, + () => this.prompts.get(prompt.promptKey) === prompt, + () => this.forget(prompt) + ) + } + + consumeHostCancellation(prompt: ClaudePendingPrompt): boolean { + return this.hostCancellation.consume(prompt) + } + bindJournalItemId(journalItemId: string, promptKey: string, questionIdForItem?: string): void { this.journalBindings.set(journalItemId, { address: promptKey, @@ -157,6 +160,20 @@ export class ClaudePromptRegistry { : null } + cancellation(itemId: string): { turnId: string; itemIds: readonly string[] } | null { + const binding = this.journalBindings.get(itemId) + const prompt = binding ? this.prompts.get(binding.address) : undefined + if (!binding || !prompt?.turnId) { + return null + } + return { + turnId: prompt.turnId, + itemIds: [...this.journalBindings] + .filter(([, candidate]) => candidate.address === binding.address) + .map(([journalItemId]) => journalItemId) + } + } + cancel(requestId: string): ClaudePendingPrompt | null { const prompt = this.prompts.get(requestId) ?? null if (prompt) { @@ -167,6 +184,7 @@ export class ClaudePromptRegistry { forget(prompt: ClaudePendingPrompt): void { this.prompts.delete(prompt.promptKey) + this.hostCancellation.forget(prompt) for (const [itemId, binding] of this.journalBindings) { if (binding.address === prompt.promptKey) { this.journalBindings.delete(itemId) @@ -178,6 +196,7 @@ export class ClaudePromptRegistry { const pending = [...this.prompts.values()] this.prompts.clear() this.journalBindings.clear() + this.hostCancellation.clear() return pending } } diff --git a/src/main/claude/claude-structured-prompt-types.ts b/src/main/claude/claude-structured-prompt-types.ts new file mode 100644 index 00000000000..6e1b718de04 --- /dev/null +++ b/src/main/claude/claude-structured-prompt-types.ts @@ -0,0 +1,25 @@ +export type ClaudePromptSettle = (response: Record | null) => void + +export type ClaudePendingPrompt = { + requestId: string + promptKey: string + turnId?: string | null + toolUseId: string + toolName: string + kind: 'approval' | 'question' + input: Record + suggestions: unknown[] + questionIds: readonly string[] + answers: Map + settle: ClaudePromptSettle +} + +export type ClaudePromptRegistration = { + requestId: string + turnId?: string | null + toolName: string + toolUseId: string + input: Record + suggestions: unknown[] + settle: ClaudePromptSettle +} diff --git a/src/main/claude/claude-structured-session-acquisition.ts b/src/main/claude/claude-structured-session-acquisition.ts index 57bad1f1873..3aa90445840 100644 --- a/src/main/claude/claude-structured-session-acquisition.ts +++ b/src/main/claude/claude-structured-session-acquisition.ts @@ -137,6 +137,7 @@ export async function acquireClaudeSession({ const { canUseTool, onUserDialog } = buildClaudePermissionCallbacks({ sessionId, prompts, + activeTurnId: () => liveSession?.activeTurnId ?? null, emit: (event) => callbacks.deliver(attempt, sessionId, () => callbacks.emit(liveSession, input.events, event)) }) 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 fc5eaa6b66e..8250709c285 100644 --- a/src/main/claude/claude-structured-session-adapter-turns.test.ts +++ b/src/main/claude/claude-structured-session-adapter-turns.test.ts @@ -7,11 +7,116 @@ import { claudeUnwrittenUserMessageError } from './claude-agent-sdk-user-message import { acquired, fakeClaude, + invokeCanUseTool, PROVIDER_SESSION_ID, USER_MESSAGE } from './claude-structured-session-test-support' describe('ClaudeStructuredSessionAdapter turns and controls', () => { + it('binds a prompt cancellation to the provider turn that created it', async () => { + const claude = fakeClaude({ replayUuid: 'turn-1' }) + const adapter = await acquired(claude) + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + invokeCanUseTool(claude.connections[0], 'Bash', 'permission-1', 'tool-1') + adapter.bindPromptItemId('session-1', 'journal-approval', 'permission-1') + + expect( + adapter.promptCancellation?.({ + sessionId: 'session-1', + itemId: 'journal-approval', + fence: 7 + }) + ).toEqual({ turnId: 'turn-1', itemIds: ['journal-approval'] }) + await expect( + adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7, + promptItemId: 'unrelated-prompt' + }) + ).resolves.toEqual({ cancelled: false }) + expect(claude.connections[0].calls.some((call) => call.subtype === 'interrupt')).toBe(false) + }) + + it('keeps the cancelled receipt when the matching SDK prompt aborts during interrupt', async () => { + const controller = new AbortController() + const events: Parameters[2] = [] + const claude = fakeClaude({ + replayUuid: 'turn-1', + routes: { interrupt: () => controller.abort() } + }) + const adapter = await acquired(claude, {}, events) + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + const pending = invokeCanUseTool(claude.connections[0], 'Bash', 'permission-1', 'tool-1', { + signal: controller.signal + }) + adapter.bindPromptItemId('session-1', 'journal-approval', 'permission-1') + + await expect( + adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7, + promptItemId: 'journal-approval' + }) + ).resolves.toEqual({ cancelled: true }) + await expect(pending.promise).resolves.toBeNull() + expect(events.some((event) => event.type === 'prompt-cancelled')).toBe(false) + expect( + adapter.promptCancellation?.({ + sessionId: 'session-1', + itemId: 'journal-approval', + fence: 7 + }) + ).toBeNull() + }) + + it('publishes an SDK prompt abort when provider interruption is refused', async () => { + const controller = new AbortController() + const events: Parameters[2] = [] + const claude = fakeClaude({ + replayUuid: 'turn-1', + routes: { + interrupt: () => { + controller.abort() + throw new ClaudeControlRequestError('interrupt', 'not running') + } + } + }) + const adapter = await acquired(claude, {}, events) + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + const pending = invokeCanUseTool(claude.connections[0], 'Bash', 'permission-1', 'tool-1', { + signal: controller.signal + }) + adapter.bindPromptItemId('session-1', 'journal-approval', 'permission-1') + + await expect( + adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7, + promptItemId: 'journal-approval' + }) + ).resolves.toEqual({ cancelled: false }) + await expect(pending.promise).resolves.toBeNull() + expect(events.at(-1)).toMatchObject({ type: 'prompt-cancelled', promptKey: 'permission-1' }) + }) + it("admits a dispatch on the write and names it from Claude's replay", async () => { const claude = fakeClaude({ replayUuid: 'user-provider-uuid' }) const settled = vi.fn() diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index d613a175917..d8be1d7925b 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -4,11 +4,7 @@ import type { StructuredAgentSessionAcquireInput, StructuredAgentSessionAdapter } from '../native-chat/agent-session-wire/structured-agent-session-adapter' -import { - answerClaudePrompt, - cancelClaudeTurn, - stopClaudeBackgroundTasks -} from './claude-structured-control-actions' +import { answerClaudePrompt, stopClaudeBackgroundTasks } from './claude-structured-control-actions' import { dispatchClaudeTurn } from './claude-structured-dispatch' import { StructuredSessionCompaction } from '../native-chat/agent-session-wire/structured-session-compaction' import { releaseClaudeAcquisition } from './claude-structured-acquisition-release' @@ -32,6 +28,7 @@ import { } from './claude-structured-session-close' import { readClaudeTranscriptLeafWithReproof } from './claude-transcript-branch-proof' import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire' +import { cancelClaudeStructuredTurn } from './claude-structured-session-cancel' export type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution' export type { @@ -223,25 +220,35 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda compact: NonNullable = (input) => compactClaudeSession(this.session(input.sessionId), this.compactions, input) - cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (input) => { + cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = async (input) => { const session = this.session(input.sessionId) const acquisitionGeneration = session.acquisitionGeneration - return cancelClaudeTurn(session, this.deps.requestTimeoutMs, () => { - // Keep every ownership check adjacent to the provider interrupt. The - // session map check fences a replaced child; the turn check fences a - // delayed cancel after a newer turn was admitted on the same child. - return ( - this.sessions.get(input.sessionId) === session && - session.fence === input.fence && - session.acquisitionGeneration === acquisitionGeneration && - (this.compactions.ownsTurn(input.sessionId, input.turnId) || - (session.activeTurnId === undefined - ? session.dispatchSequence === 0 - : session.activeTurnId === input.turnId && - session.activeTurnSequence === session.dispatchSequence)) - ) + return cancelClaudeStructuredTurn({ + session, + request: input, + requestTimeoutMs: this.deps.requestTimeoutMs, + stillOwnsTurn: () => { + // Keep every ownership check adjacent to the provider interrupt. The + // session map check fences a replaced child; the turn check fences a + // delayed cancel after a newer turn was admitted on the same child. + return ( + this.sessions.get(input.sessionId) === session && + session.fence === input.fence && + session.acquisitionGeneration === acquisitionGeneration && + (this.compactions.ownsTurn(input.sessionId, input.turnId) || + (session.activeTurnId === undefined + ? session.dispatchSequence === 0 + : session.activeTurnId === input.turnId && + session.activeTurnSequence === session.dispatchSequence)) + ) + }, + emitPromptCancelled: (promptKey) => { + this.emit(session, { type: 'prompt-cancelled', sessionId: input.sessionId, promptKey }) + } }) } + promptCancellation: NonNullable = (input) => + this.session(input.sessionId).prompts.cancellation(input.itemId) stopBackgroundTasks: StructuredAgentSessionAdapter['stopBackgroundTasks'] = (input) => { const session = this.session(input.sessionId) const acquisitionGeneration = session.acquisitionGeneration diff --git a/src/main/claude/claude-structured-session-cancel.ts b/src/main/claude/claude-structured-session-cancel.ts new file mode 100644 index 00000000000..f0e57e49fba --- /dev/null +++ b/src/main/claude/claude-structured-session-cancel.ts @@ -0,0 +1,37 @@ +import type { StructuredAgentSessionAdapter } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import { cancelClaudeTurn } from './claude-structured-control-actions' +import type { ClaudeSession } from './claude-structured-session-state' + +type CancelTurnInput = Parameters[0] + +export async function cancelClaudeStructuredTurn(input: { + session: ClaudeSession + request: CancelTurnInput + requestTimeoutMs?: number + stillOwnsTurn: () => boolean + emitPromptCancelled: (promptKey: string) => void +}): Promise<{ cancelled: boolean }> { + const { session, request } = input + const prompt = request.promptItemId + ? session.prompts.beginHostCancellation(request.promptItemId, request.turnId) + : null + if (request.promptItemId && !prompt) { + return { cancelled: false } + } + let cancelled = false + try { + const result = await cancelClaudeTurn(session, input.requestTimeoutMs, () => { + return ( + input.stillOwnsTurn() && + (!request.promptItemId || + session.prompts.cancellation(request.promptItemId)?.turnId === request.turnId) + ) + }) + cancelled = result.cancelled + return result + } finally { + if (prompt && session.prompts.finishHostCancellation(prompt, cancelled)) { + input.emitPromptCancelled(prompt.promptKey) + } + } +} diff --git a/src/main/codex/codex-persistent-command-retention.test.ts b/src/main/codex/codex-persistent-command-retention.test.ts index 949e3156ee2..37ed0997557 100644 --- a/src/main/codex/codex-persistent-command-retention.test.ts +++ b/src/main/codex/codex-persistent-command-retention.test.ts @@ -116,7 +116,8 @@ describe('persistent command retention', () => { turnLifecycle: null, sink, streams: items.streams, - activeItems: items.activeItems + activeItems: items.activeItems, + pendingPrompts: new Map() }) ).toEqual({ accepted: true }) } @@ -211,7 +212,8 @@ describe('persistent command retention', () => { turnLifecycle: null, sink, streams: items.streams, - activeItems: items.activeItems + activeItems: items.activeItems, + pendingPrompts: new Map() }) ).toEqual({ accepted: true }) expect(items.activeItems.size).toBe(1) diff --git a/src/main/codex/codex-structured-journal-contracts.ts b/src/main/codex/codex-structured-journal-contracts.ts index d7be380446f..8140d436494 100644 --- a/src/main/codex/codex-structured-journal-contracts.ts +++ b/src/main/codex/codex-structured-journal-contracts.ts @@ -8,7 +8,12 @@ export type CodexJournalTranslatorDeps = { /** Keys restored lifecycle rows to the live identity; without it history restore skips them. */ sessionId?: string now?: () => number - bindPromptItemId?: (journalItemId: string, threadId: string, promptKey: string) => void + bindPromptItemId?: ( + journalItemId: string, + threadId: string, + promptKey: string, + turnId: string | null + ) => void primaryThreadId?: () => string | null subagentExecutions?: CodexSubagentExecutions coalesceMs?: number diff --git a/src/main/codex/codex-structured-journal-prompts.ts b/src/main/codex/codex-structured-journal-prompts.ts index 93ecec77f77..4c8bc8fd8ab 100644 --- a/src/main/codex/codex-structured-journal-prompts.ts +++ b/src/main/codex/codex-structured-journal-prompts.ts @@ -18,13 +18,15 @@ import { publishCodexLifecycle } from './codex-structured-journal-sink' import type { CodexPendingJournalPrompt } from './codex-structured-journal-settlement' +import { readCodexTurnId } from './codex-structured-thread-facts' export class CodexJournalPrompts { readonly pending = new Map() constructor( private readonly deps: Pick, - private readonly detailFor: (threadId: string, itemId: string) => string | null + private readonly detailFor: (threadId: string, itemId: string) => string | null, + private readonly activeTurn: (threadId: string) => string | null ) {} handle(event: { @@ -47,12 +49,22 @@ export class CodexJournalPrompts { } for (const question of promptItems) { const itemId = agentJournalItemKey(question.identity) - this.pending.set(itemId, { identity: question.identity, body: question.body }) + this.pending.set(itemId, { + threadId: event.threadId, + turnId: readCodexTurnId(event.params) ?? this.activeTurn(event.threadId), + identity: question.identity, + body: question.body + }) const trimAdmission = this.trim() if (!trimAdmission.accepted) { return trimAdmission } - this.deps.bindPromptItemId?.(itemId, event.threadId, event.promptKey) + this.deps.bindPromptItemId?.( + itemId, + event.threadId, + event.promptKey, + this.pending.get(itemId)?.turnId ?? null + ) } return CODEX_JOURNAL_ADMITTED } @@ -70,12 +82,22 @@ export class CodexJournalPrompts { return admission } const itemId = agentJournalItemKey(identity) - this.pending.set(itemId, { identity, body }) + this.pending.set(itemId, { + threadId: event.threadId, + turnId: readCodexTurnId(event.params) ?? this.activeTurn(event.threadId), + identity, + body + }) const trimAdmission = this.trim() if (!trimAdmission.accepted) { return trimAdmission } - this.deps.bindPromptItemId?.(itemId, event.threadId, event.promptKey) + this.deps.bindPromptItemId?.( + itemId, + event.threadId, + event.promptKey, + this.pending.get(itemId)?.turnId ?? null + ) return CODEX_JOURNAL_ADMITTED } @@ -89,7 +111,7 @@ export class CodexJournalPrompts { private admit( event: { method: string; threadId: string; promptKey: string }, - items: readonly CodexPendingJournalPrompt[] + items: readonly Pick[] ): CodexJournalTranslationAdmission { return admitCodexLifecycleItems( this.deps.sink, diff --git a/src/main/codex/codex-structured-journal-settlement.ts b/src/main/codex/codex-structured-journal-settlement.ts index 5785b273e92..518170f748d 100644 --- a/src/main/codex/codex-structured-journal-settlement.ts +++ b/src/main/codex/codex-structured-journal-settlement.ts @@ -26,6 +26,10 @@ import { codexTurnLifecycleBody, codexTurnLifecycleIdentity } from './codex-structured-journal-translation-turns' +import { collectCodexTurnPromptCancellations } from './codex-structured-prompt-turn-settlement' +import type { CodexPendingJournalPrompt } from './codex-structured-prompt-turn-settlement' + +export type { CodexPendingJournalPrompt } from './codex-structured-prompt-turn-settlement' export type CodexActiveJournalItem = { threadId: string @@ -34,11 +38,6 @@ export type CodexActiveJournalItem = { item: CodexThreadItem } -export type CodexPendingJournalPrompt = { - identity: AgentJournalItemIdentity - body: AgentJournalItemBody -} - const ADMITTED: StructuredAgentSessionSinkAdmission = { accepted: true } export function settleCodexJournalSession(input: { @@ -50,7 +49,6 @@ export function settleCodexJournalSession(input: { currentTurnIds: ReadonlyMap> primaryThreadId: string | null ordinals: CodexTurnOrdinals - /** Terminal lifecycle for a turn the provider left running when it ended. */ settledTurnLifecycle: (threadId: string, turnId: string) => AgentJournalTurnLifecycle }): StructuredAgentSessionSinkAdmission { const mutations: JournalLifecycleMutationInput[] = [] @@ -112,11 +110,11 @@ export function settleCodexJournalTurn(input: { sessionId: string threadId: string turnId: string - /** Null off the primary thread: only the primary turn owns a lifecycle row. */ turnLifecycle: AgentJournalTurnLifecycle | null sink: StructuredAgentSessionEventSink streams: CodexStructuredItemStreams activeItems: Map + pendingPrompts: Map }): StructuredAgentSessionSinkAdmission { const mutations: JournalLifecycleMutationInput[] = [] const activeItemsToForget: { key: string; threadId: string; itemId: string }[] = [] @@ -137,7 +135,8 @@ export function settleCodexJournalTurn(input: { } activeItemsToForget.push({ key, threadId: active.threadId, itemId: active.item.id }) } - // Revised, never tombstoned: the terminal row keeps the turn's duration durable. + const promptCancellations = collectCodexTurnPromptCancellations(input) + mutations.push(...promptCancellations.mutations) if (input.turnLifecycle) { mutations.push({ kind: 'item', @@ -160,6 +159,9 @@ export function settleCodexJournalTurn(input: { input.streams.forget(active.threadId, active.itemId) input.activeItems.delete(active.key) } + for (const itemId of promptCancellations.itemIds) { + input.pendingPrompts.delete(itemId) + } return ADMITTED } diff --git a/src/main/codex/codex-structured-journal-sink.ts b/src/main/codex/codex-structured-journal-sink.ts index 7da381def41..d5b6e5e4a36 100644 --- a/src/main/codex/codex-structured-journal-sink.ts +++ b/src/main/codex/codex-structured-journal-sink.ts @@ -57,7 +57,7 @@ export function publishCodexLifecycle( export function admitCodexLifecycleItems( sink: StructuredAgentSessionEventSink, settlementId: string, - items: readonly CodexPendingJournalPrompt[] + items: readonly Pick[] ): CodexJournalTranslationAdmission { if (items.length === 0) { return { accepted: false, reason: 'untranslated' } diff --git a/src/main/codex/codex-structured-journal-translation-settlement.test.ts b/src/main/codex/codex-structured-journal-translation-settlement.test.ts index b5e60699ce1..2840a0e195a 100644 --- a/src/main/codex/codex-structured-journal-translation-settlement.test.ts +++ b/src/main/codex/codex-structured-journal-translation-settlement.test.ts @@ -566,6 +566,36 @@ describe('codex journal translation', () => { ]) }) + it('cancels only prompts owned by the completed turn', () => { + const { translator, tap } = translatorWith() + const ask = (turnId: string, promptKey: string): void => { + translator.handle({ + type: 'prompt', + sessionId: SESSION_ID, + threadId: THREAD_ID, + method: CODEX_COMMAND_APPROVAL_METHOD, + params: { turnId, availableDecisions: ['accept', 'decline'] }, + codexItemId: `command-${turnId}`, + promptKey + }) + } + + translator.handle(TURN_STARTED) + ask(TURN_ID, 'approval-turn-1') + translator.handle(notification('turn/started', { turn: { id: 'turn-2' } })) + ask('turn-2', 'approval-turn-2') + translator.handle(notification('turn/completed', { turn: { id: TURN_ID } })) + + const promptRows = tap.rows.filter((row) => row.body.kind === 'approval') + const rowsFor = (key: string) => promptRows.filter((row) => row.key.includes(key)) + const completed = rowsFor('approval-turn-1') + const active = rowsFor('approval-turn-2') + expect(completed).toHaveLength(2) + expect(completed.at(-1)?.body).toMatchObject({ resolution: { state: 'cancelled' } }) + expect(active).toHaveLength(1) + expect(active[0]?.body).toMatchObject({ resolution: { state: 'pending' } }) + }) + it('journals an approval naming the command the item already announced, and binds it', () => { const { translator, tap } = translatorWith() diff --git a/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts b/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts index a4aa3200aca..7dcb3d9b5f3 100644 --- a/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts +++ b/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts @@ -5,6 +5,7 @@ import { type CodexJournalTranslationAdmission } from './codex-structured-journal-contracts' import type { CodexJournalItems } from './codex-structured-journal-items' +import type { CodexJournalPrompts } from './codex-structured-journal-prompts' import { settleCodexJournalTurn } from './codex-structured-journal-settlement' import type { CodexJournalActiveTurns } from './codex-structured-journal-translation-turn-state' import { @@ -33,6 +34,7 @@ export class CodexJournalTurnBoundaries { primaryThreadId: () => string | null activeTurns: CodexJournalActiveTurns items: Pick + prompts: Pick flushSuppression: () => CodexJournalTranslationAdmission resetActivity: (threadId: string) => void now?: () => number @@ -93,7 +95,8 @@ export class CodexJournalTurnBoundaries { ) : null, streams: this.deps.items.streams, - activeItems: this.deps.items.activeItems + activeItems: this.deps.items.activeItems, + pendingPrompts: this.deps.prompts.pending }) if (admission.accepted) { this.deps.items.ordinals.forgetTurn(event.threadId, turnId) diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 5c1e97310bc..1cd249ac596 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -59,8 +59,10 @@ export function createCodexJournalTranslator( (threadId, turnId) => genericFrames.suppress(threadId, turnId) ) const settleOversizedNotification = createCodexOversizedNotificationSettler(deps, items) - const prompts = new CodexJournalPrompts(deps, (threadId, itemId) => - items.detailFor(threadId, itemId) + const prompts = new CodexJournalPrompts( + deps, + (threadId, itemId) => items.detailFor(threadId, itemId), + (threadId) => activeTurns.current(threadId) ) const subagents = new CodexSubagentRoster({ sink: deps.sink, @@ -82,6 +84,7 @@ export function createCodexJournalTranslator( primaryThreadId: () => deps.primaryThreadId?.() ?? null, activeTurns, items, + prompts, flushSuppression: () => genericFrames.flush(), resetActivity, ...(deps.now ? { now: deps.now } : {}) diff --git a/src/main/codex/codex-structured-prompt-replies.test.ts b/src/main/codex/codex-structured-prompt-replies.test.ts index 831124c1c48..57f4446928a 100644 --- a/src/main/codex/codex-structured-prompt-replies.test.ts +++ b/src/main/codex/codex-structured-prompt-replies.test.ts @@ -107,6 +107,39 @@ describe('CodexPromptRegistry', () => { expect(registry.find('codex-item-1')).toBeNull() }) + it('returns every journal row for one prompt only when it belongs to the cancelled turn', () => { + const registry = new CodexPromptRegistry() + registry.register(userInputRequest(['q1', 'q2'])) + registry.bindJournalItemId('journal-q1', 'thread-1', 'codex-item-1') + registry.bindJournalItemId('journal-q2', 'thread-1', 'codex-item-1') + + expect(registry.cancellation('journal-q1')).toEqual({ + turnId: 'turn-1', + itemIds: ['journal-q1', 'journal-q2'] + }) + expect(registry.cancellation('unbound')).toBeNull() + }) + + it('reads nested turn identity and adopts the translator fallback when absent', () => { + const nested = new CodexPromptRegistry() + nested.register({ + id: 6, + method: 'item/commandExecution/requestApproval', + params: { itemId: 'item-nested', threadId: 'thread-1', turn: { id: 'turn-nested' } } + }) + nested.bindJournalItemId('journal-nested', 'thread-1', 'item-nested') + expect(nested.cancellation('journal-nested')?.turnId).toBe('turn-nested') + + const fallback = new CodexPromptRegistry() + fallback.register({ + id: 7, + method: 'item/commandExecution/requestApproval', + params: { itemId: 'item-fallback', threadId: 'thread-1' } + }) + fallback.bindJournalItemId('journal-fallback', 'thread-1', 'item-fallback', 'turn-active') + expect(fallback.cancellation('journal-fallback')?.turnId).toBe('turn-active') + }) + it('keeps identical item ids on different threads independently answerable', () => { const registry = new CodexPromptRegistry() const register = (id: number, threadId: string) => diff --git a/src/main/codex/codex-structured-prompt-replies.ts b/src/main/codex/codex-structured-prompt-replies.ts index 1c96a95c5f2..7d5804d007c 100644 --- a/src/main/codex/codex-structured-prompt-replies.ts +++ b/src/main/codex/codex-structured-prompt-replies.ts @@ -1,4 +1,5 @@ import type { CodexAppServerConnection } from './codex-app-server-connection' +import { readCodexTurnId } from './codex-structured-thread-facts' import { CODEX_PROMPT_MAX_ANSWER_BYTES, MAX_CODEX_PROMPT_JOURNAL_BINDINGS, @@ -186,7 +187,7 @@ export class CodexPromptRegistry { requestId: request.id, method: request.method, threadId, - turnId: readString(request.params, 'turnId'), + turnId: readCodexTurnId(request.params), codexItemId, promptKey: readString(request.params, 'approvalId') ?? codexItemId, questionIds, @@ -222,7 +223,12 @@ export class CodexPromptRegistry { } /** Called by the translation module once the prompt has a journal id. */ - bindJournalItemId(journalItemId: string, threadId: string, promptKey: string): void { + bindJournalItemId( + journalItemId: string, + threadId: string, + promptKey: string, + turnId?: string | null + ): void { const existing = this.journalItemIds.get(journalItemId) if (existing) { this.boundPrompts.delete(journalItemId) @@ -233,6 +239,9 @@ export class CodexPromptRegistry { if (!prompt) { return } + if (prompt.turnId === null && turnId && prompt.threadId === threadId) { + prompt.turnId = turnId + } this.journalItemIds.set(journalItemId, address) this.boundPrompts.set(journalItemId, prompt) this.trim() @@ -251,6 +260,19 @@ export class CodexPromptRegistry { return matches.length === 1 ? matches[0]! : null } + cancellation(journalItemId: string): { turnId: string; itemIds: readonly string[] } | null { + const prompt = this.find(journalItemId) + if (!prompt?.turnId || this.boundPrompts.get(journalItemId) !== prompt) { + return null + } + return { + turnId: prompt.turnId, + itemIds: [...this.boundPrompts] + .filter(([, candidate]) => candidate === prompt) + .map(([itemId]) => itemId) + } + } + forget(prompt: CodexPendingPrompt): void { const address = this.address(prompt.threadId, prompt.promptKey) if (this.byAddress.get(address) === prompt) { diff --git a/src/main/codex/codex-structured-prompt-turn-settlement.ts b/src/main/codex/codex-structured-prompt-turn-settlement.ts new file mode 100644 index 00000000000..212e31eba29 --- /dev/null +++ b/src/main/codex/codex-structured-prompt-turn-settlement.ts @@ -0,0 +1,33 @@ +import type { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { cancelledJournalPromptBody } from '../native-chat/agent-session-journal/journal-prompt-body-bounds' +import type { JournalLifecycleMutationInput } from '../native-chat/agent-session-journal/journal-row-builders' + +export type CodexPendingJournalPrompt = { + threadId: string + turnId: string | null + identity: AgentJournalItemIdentity + body: AgentJournalItemBody +} + +export function collectCodexTurnPromptCancellations(input: { + threadId: string + turnId: string + pendingPrompts: ReadonlyMap +}): { mutations: JournalLifecycleMutationInput[]; itemIds: string[] } { + const mutations: JournalLifecycleMutationInput[] = [] + const itemIds: string[] = [] + for (const [itemId, prompt] of input.pendingPrompts) { + if (prompt.threadId !== input.threadId || prompt.turnId !== input.turnId) { + continue + } + const body = cancelledJournalPromptBody(prompt.body) + if (body) { + mutations.push({ kind: 'item', identity: prompt.identity, body }) + } + itemIds.push(itemId) + } + return { mutations, itemIds } +} diff --git a/src/main/codex/codex-structured-session-acquire.ts b/src/main/codex/codex-structured-session-acquire.ts index c9a306128b6..c6359e4d0f8 100644 --- a/src/main/codex/codex-structured-session-acquire.ts +++ b/src/main/codex/codex-structured-session-acquire.ts @@ -84,8 +84,8 @@ export async function acquireCodexStructuredSession(input: { ...(deps.now ? { now: deps.now } : {}), primaryThreadId: () => primaryThreadId, subagentExecutions, - bindPromptItemId: (journalItemId, threadId, promptKey) => - acquisition.prompts.bindJournalItemId(journalItemId, threadId, promptKey) + bindPromptItemId: (journalItemId, threadId, promptKey, turnId) => + acquisition.prompts.bindJournalItemId(journalItemId, threadId, promptKey, turnId) }) : null const open = deps.openConnection ?? openCodexAppServerConnection diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 061626f9724..52ab9744546 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -204,12 +204,24 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap sessionId: string turnId: string fence: number + promptItemId?: string }): Promise<{ cancelled: boolean }> { const session = this.session(input.sessionId) const turnId = this.compactions.providerTurnId(input.sessionId, input.turnId) - return turnId ? this.turnCancellation.cancel(session, turnId) : { cancelled: false } + const prompt = input.promptItemId ? session.prompts.find(input.promptItemId) : null + if (!turnId || (input.promptItemId && prompt?.turnId !== turnId)) { + return { cancelled: false } + } + const result = await this.turnCancellation.cancel(session, turnId) + if (result.cancelled && prompt) { + session.prompts.forget(prompt) + } + return result } + promptCancellation: NonNullable = (input) => + this.session(input.sessionId).prompts.cancellation(input.itemId) + rewindSupport: NonNullable = (sessionId) => this.sessions.get(sessionId)?.historyMode === 'legacy' ? { supported: false, reason: 'history-not-paginated' } diff --git a/src/main/codex/codex-structured-session-cancel.test.ts b/src/main/codex/codex-structured-session-cancel.test.ts index 2ea81d44786..54ebaf74c41 100644 --- a/src/main/codex/codex-structured-session-cancel.test.ts +++ b/src/main/codex/codex-structured-session-cancel.test.ts @@ -1,8 +1,10 @@ import { describe, expect, it, vi } from 'vitest' import type { + AgentJournalItemBody, AgentJournalMessageItem, AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { CodexAppServerRequestError, type CodexAppServerConnection, @@ -84,7 +86,8 @@ async function acquired( CodexStructuredSessionAdapterDeps, 'captureTurnProcesses' | 'terminateTurnProcesses' | 'now' > - > = {} + > = {}, + eventSink?: StructuredAgentSessionEventSink ): Promise { const adapter = new CodexStructuredSessionAdapter({ resolveLaunch: async () => ({ @@ -101,7 +104,12 @@ async function acquired( terminateTurnProcesses: async () => true, ...processControl }) - await adapter.acquire({ identity: identity(), fence: 7, spawnToken: 'spawn-9' }) + await adapter.acquire({ + identity: identity(), + fence: 7, + spawnToken: 'spawn-9', + ...(eventSink ? { events: eventSink } : {}) + }) return adapter } @@ -113,6 +121,40 @@ function completeTurn(codex: ReturnType, turnId = 'turn-1'): v } describe('CodexStructuredSessionAdapter.cancelTurn', () => { + it('binds prompt cancellation to the live logical request and owning turn', async () => { + const codex = fakeCodex() + const adapter = await acquired(codex) + codex.connections[0].handlers.onServerRequest?.({ + id: 1, + method: 'item/tool/requestUserInput', + params: { + itemId: 'questions-1', + threadId: THREAD_ID, + turnId: 'turn-1', + questions: [{ id: 'q1' }, { id: 'q2' }] + } + }) + adapter.bindPromptItemId('session-1', 'journal-q1', 'questions-1') + adapter.bindPromptItemId('session-1', 'journal-q2', 'questions-1') + + expect( + adapter.promptCancellation?.({ + sessionId: 'session-1', + itemId: 'journal-q1', + fence: 7 + }) + ).toEqual({ turnId: 'turn-1', itemIds: ['journal-q1', 'journal-q2'] }) + await expect( + adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7, + promptItemId: 'unrelated-prompt' + }) + ).resolves.toEqual({ cancelled: false }) + expect(codex.connections[0].calls.some((call) => call.method === 'turn/interrupt')).toBe(false) + }) + it('confirms an interrupt Codex acknowledged', async () => { const codex = fakeCodex() const adapter = await acquired(codex) @@ -291,18 +333,50 @@ describe('CodexStructuredSessionAdapter.cancelTurn', () => { it('does not strand a deferred completion when the interrupt receipt fails', async () => { const events: CodexStructuredSessionEvent[] = [] + const bodies: AgentJournalItemBody[] = [] + const sink: StructuredAgentSessionEventSink = { + appendItem: (_identity, body) => bodies.push(body), + appendTombstone: vi.fn(), + publish: vi.fn() + } const codex = fakeCodex() codex.routes['turn/interrupt'] = () => { completeTurn(codex) throw new Error('interrupt receipt lost') } - const adapter = await acquired(codex, events, { - terminateTurnProcesses: async () => true + const adapter = await acquired( + codex, + events, + { + terminateTurnProcesses: async () => true + }, + sink + ) + codex.connections[0].handlers.onNotification?.('turn/started', { + threadId: THREAD_ID, + turn: { id: 'turn-1' } + }) + codex.connections[0].handlers.onServerRequest?.({ + id: 1, + method: 'item/commandExecution/requestApproval', + params: { + itemId: 'command-1', + approvalId: 'approval-1', + threadId: THREAD_ID, + turnId: 'turn-1', + availableDecisions: ['accept', 'decline'] + } }) await expect( adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) ).rejects.toThrow('interrupt receipt lost') expect(events).toContainEqual(expect.objectContaining({ method: 'turn/completed' })) + expect(bodies).toContainEqual( + expect.objectContaining({ + kind: 'approval', + resolution: expect.objectContaining({ state: 'cancelled' }) + }) + ) }) }) diff --git a/src/main/native-chat/agent-session-journal/journal-prompt-cancellation.ts b/src/main/native-chat/agent-session-journal/journal-prompt-cancellation.ts new file mode 100644 index 00000000000..1a972b4c8fd --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-prompt-cancellation.ts @@ -0,0 +1,71 @@ +import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { JournalReducerState } from './journal-reducer' +import { journalLifecycleBatchRowBuilder } from './journal-row-builders' +import type { JournalRowWriter } from './journal-row-writer' + +export type JournalPromptCancellationInput = { + prompts: readonly { itemId: string; expectedRevision: number }[] + settlementId: string + resolvedBy: string + resolvedAt: number + fence: number +} + +/** Atomically revises only prompt revisions that still match the cancellation snapshot. */ +export function cancelJournalPromptsAtRevisions( + rowWriter: JournalRowWriter, + state: () => JournalReducerState, + input: JournalPromptCancellationInput +): Promise { + const identities = input.prompts.map((prompt) => parseAgentJournalItemKey(prompt.itemId)) + if ( + input.prompts.length === 0 || + new Set(input.prompts.map((prompt) => prompt.itemId)).size !== input.prompts.length || + identities.some((identity) => !identity) + ) { + return Promise.resolve(0) + } + return rowWriter + .enqueueIf((sequence, timestamp) => { + const currentState = state() + const mutations = input.prompts.flatMap((prompt, index) => { + const resolved = currentState.aliases.get(prompt.itemId) ?? prompt.itemId + const current = currentState.items.get(resolved) + const identity = identities[index] + if ( + !identity || + !current || + current.revision !== prompt.expectedRevision || + (current.body.kind !== 'approval' && current.body.kind !== 'question') || + current.body.resolution.state !== 'pending' + ) { + return [] + } + return [ + { + kind: 'item' as const, + identity, + body: { + ...current.body, + resolution: { + state: 'cancelled' as const, + selectedOptionId: null, + resolvedBy: input.resolvedBy, + resolvedAt: input.resolvedAt + } + } + } + ] + }) + return mutations.length === 0 + ? null + : journalLifecycleBatchRowBuilder(state, input.settlementId, mutations, { + fence: input.fence + })(sequence, timestamp) + }) + .then((row) => + row?.kind === 'lifecycle-batch' + ? row.mutations.filter((mutation) => mutation.kind === 'item').length + : 0 + ) +} diff --git a/src/main/native-chat/agent-session-journal/journal-row-writer.ts b/src/main/native-chat/agent-session-journal/journal-row-writer.ts index 85ff7da7a3f..33e0edc4fec 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-writer.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-writer.ts @@ -18,9 +18,21 @@ export class JournalRowWriter { constructor(private readonly deps: JournalRowWriterDeps) {} enqueue(build: (seq: number, ts: number) => JournalRow): Promise { + return this.enqueueIf(build).then((row) => { + if (!row) { + throw new Error('journal_row_builder_skipped_unconditionally') + } + return row + }) + } + + enqueueIf(build: (seq: number, ts: number) => JournalRow | null): Promise { return this.deps.serialize(async () => { assertJournalWritable(this.deps.readOnly(), this.deps.sessionId) const row = build(this.deps.nextSequence(), this.deps.now()) + if (!row) { + return null + } assertJournalFence(row.fence, this.deps.highestFence()) const { db } = this.deps.database() db.exec('BEGIN IMMEDIATE') diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index 2e372f9abae..9a8b8939732 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -6,6 +6,7 @@ import type { AgentJournalCursor, AgentJournalItemBody, AgentJournalItemIdentity, + AgentJournalRenderItem, AgentJournalSnapshot, AgentJournalSubmission, AgentSessionJournalIdentity @@ -49,6 +50,10 @@ import { createJournalStoreCollaborators } from './journal-store-collaborators' import { ensureJournalDir, journalStoreLoadedFields } from './journal-store-open' import type { JournalItemAppender } from './journal-item-appender' import type { JournalLifecycleBatchAppender } from './journal-lifecycle-batch-appender' +import { + cancelJournalPromptsAtRevisions, + type JournalPromptCancellationInput +} from './journal-prompt-cancellation' export { AgentSessionJournalError } from './journal-write-guards' @@ -162,6 +167,30 @@ export class AgentSessionJournal { snapshot = (): AgentJournalSnapshot => renderJournalState(this.state) + /** Orders a CAS preflight behind provider writes already admitted to the journal queue. */ + readItem(itemId: string): Promise { + return this.queue.serialize(async () => { + const resolved = this.state.aliases.get(itemId) ?? itemId + return this.state.items.get(resolved) ?? null + }) + } + + /** Orders a bounded prompt-group snapshot behind admitted provider writes. */ + readItems(itemIds: readonly string[]): Promise { + return this.queue.serialize(async () => { + const items: AgentJournalRenderItem[] = [] + for (const itemId of itemIds) { + const resolved = this.state.aliases.get(itemId) ?? itemId + const item = this.state.items.get(resolved) + if (!item) { + return null + } + items.push(item) + } + return items + }) + } + /** Visits reduced items without allocating and sorting a full snapshot. */ visitItems = (visit: (itemId: string, sequence: number) => void): void => { for (const item of this.state.items.values()) { @@ -213,6 +242,10 @@ export class AgentSessionJournal { return this.itemAppender.append(identity, body, options) } + cancelPromptsAtRevisions(input: JournalPromptCancellationInput): Promise { + return cancelJournalPromptsAtRevisions(this.rowWriter, () => this.state, input) + } + appendTombstone( identity: AgentJournalItemIdentity, options: JournalTombstoneInput diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts index c6566083eac..69120a90dec 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts @@ -32,6 +32,33 @@ describe('StructuredAgentSessionAdapterRouter.releaseAcquisition', () => { }) }) +describe('StructuredAgentSessionAdapterRouter prompt cancellation', () => { + it('reads the logical prompt group from the session owner', async () => { + const claude = adapterOf(vi.fn(async () => true)) + const promptCancellation = vi.fn(() => ({ + turnId: 'turn-1', + itemIds: ['prompt-1', 'prompt-2'] + })) + claude.promptCancellation = promptCancellation + const router = new StructuredAgentSessionAdapterRouter( + { claude, codex: adapterOf(vi.fn(async () => false)) }, + async () => {} + ) + await router.acquire({ + identity: { sessionId: 'session-1', agent: 'claude' } as never, + fence: 1, + spawnToken: 'spawn-1' + }) + const input = { sessionId: 'session-1', turnId: 'turn-1', itemId: 'prompt-1', fence: 1 } + + expect(router.promptCancellation(input)).toEqual({ + turnId: 'turn-1', + itemIds: ['prompt-1', 'prompt-2'] + }) + expect(promptCancellation).toHaveBeenCalledExactlyOnceWith(input) + }) +}) + describe('StructuredAgentSessionAdapterRouter.closeSession', () => { it('retains the owner after an unproven close so a later retry reaches the same adapter', async () => { const claude = adapterOf(vi.fn(async () => true)) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts index 42f9289783a..3a524d33ffb 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts @@ -71,6 +71,9 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (input) => this.owner(input.sessionId).cancelTurn(input) + promptCancellation: NonNullable = (input) => + this.owner(input.sessionId).promptCancellation?.(input) ?? null + stopBackgroundTasks: NonNullable = ( input ) => { 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 3bd14a057aa..bbe98d57600 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 @@ -188,10 +188,16 @@ export type StructuredAgentSessionAdapter = { }): Promise<{ error?: string }> /** Cancels one turn, not the session: a session-wide interrupt would also kill * a turn the client never asked to stop. */ + promptCancellation?(input: { + sessionId: string + itemId: string + fence: number + }): { turnId: string; itemIds: readonly string[] } | null cancelTurn(input: { sessionId: string turnId: string fence: number + promptItemId?: string }): Promise<{ cancelled: boolean }> stopBackgroundTasks?(input: { sessionId: string 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 9d50adaaa1a..50110132ae5 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 @@ -106,9 +106,10 @@ export function cancelStructuredAgentSessionTurn( caller: StructuredAgentSessionCaller, params: { envelope: AgentSessionMutationEnvelope - turnId: string + turnId?: string scope?: 'background-tasks' taskId?: string + prompt?: { itemId: string; expectedRevision: number } } ): Promise> { const command = context.deps.store.getRecord(params.envelope.sessionId)?.conversationCommand diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts index c67a8fabf58..8fe74bf1463 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts @@ -7,6 +7,7 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire' import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import { journalDirectoryFor } from '../agent-session-journal/journal-paths' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' import type { AgentSessionDispatchOutcome, @@ -58,6 +59,7 @@ let acquire: Mock let releaseAcquisition: Mock> let dispatch: Mock let cancelTurn: Mock +let promptCancellation: Mock> let answerPrompt: Mock let setOption: Mock let ordinal = 0 @@ -75,6 +77,7 @@ function adapter(): StructuredAgentSessionAdapter { acquire, releaseAcquisition, dispatch, + promptCancellation, cancelTurn, answerPrompt, setOption @@ -87,12 +90,15 @@ async function attach(): Promise { return store.getRecord(SESSION) } -/** Puts a pending approval in the journal BEFORE attach, which is the only way - * 1d can stage one: the adapter that would emit it is phase 2's. */ -async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revision: number }> { - const identity = { provider: 'codex' as const, threadId: THREAD, turnId: 'turn-1', ordinal: 99 } +async function promptJournal(): Promise { + const live = ( + host as unknown as { sessions: Map } + ).sessions.get(SESSION)?.journal + if (live) { + return live + } const journalDir = journalDirectoryFor(root, { workspaceId: 'workspace-1', sessionId: SESSION }) - const journal = await journals.open({ + return journals.open({ identity: { sessionId: SESSION, workspaceId: 'workspace-1', @@ -102,6 +108,14 @@ async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revis }, journalDir }) +} + +async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revision: number }> { + const identity = { + provider: 'orca' as const, + clientMessageId: `codex-prompt:${THREAD}:approval-1` + } + const journal = await promptJournal() const appended = await journal.appendItem( identity, { @@ -111,11 +125,33 @@ async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revis options: [{ id: optionId, label: 'Allow' }], resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } }, - { fence: 1 } + { fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1 } ) return { itemId: appended.itemId, revision: appended.revision } } +async function seedQuestionGroup(): Promise<{ itemId: string; revision: number }[]> { + const journal = await promptJournal() + const prompts: { itemId: string; revision: number }[] = [] + for (const questionId of ['q1', 'q2']) { + const appended = await journal.appendItem( + { + provider: 'orca', + clientMessageId: `codex-prompt:${THREAD}:questions-1:${questionId}` + }, + { + kind: 'question', + question: `Question ${questionId}`, + options: [{ id: `${questionId}:yes`, label: 'Yes' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1 } + ) + prompts.push({ itemId: appended.itemId, revision: appended.revision }) + } + return prompts +} + beforeEach(async () => { root = await mkdtemp(join(tmpdir(), 'orca-wire-host-')) resetHostTestOperationIds() @@ -138,6 +174,7 @@ beforeEach(async () => { releaseAcquisition = vi.fn(async () => true) dispatch = vi.fn(async () => accepted()) cancelTurn = vi.fn(async () => ({ cancelled: true })) + promptCancellation = vi.fn(({ itemId }) => ({ turnId: 'turn-1', itemIds: [itemId] })) answerPrompt = vi.fn(async () => undefined) setOption = vi.fn(async () => undefined) store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) @@ -178,6 +215,7 @@ export function hostTestState() { acquire, releaseAcquisition, dispatch, + promptCancellation, cancelTurn, answerPrompt, setOption @@ -193,5 +231,6 @@ export { ensureParams, envelope, journals, - seedApproval + seedApproval, + seedQuestionGroup } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index d53c3c30e50..b0465e52256 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -17,7 +17,8 @@ import { envelope, hostTestState, replaceHostTestState, - seedApproval + seedApproval, + seedQuestionGroup } from './structured-agent-session-host-test-harness' import { HOST_TEST_NOW as NOW, @@ -33,6 +34,7 @@ let acquire: Mock let releaseAcquisition: Mock> let dispatch: Mock let cancelTurn: Mock +let promptCancellation: Mock> let answerPrompt: Mock let setOption: Mock @@ -44,6 +46,7 @@ beforeEach(() => { acquire, releaseAcquisition, dispatch, + promptCancellation, cancelTurn, answerPrompt, setOption @@ -230,16 +233,237 @@ describe('cancel', () => { expect(await host.cancel(CALLER, params)).toMatchObject({ ok: true, replayed: true, - value: { cancelled: false } + value: { cancelled: true } }) expect(cancelTurn).toHaveBeenCalledTimes(1) }) + + it('recovers a confirmed prompt interruption when later status persistence fails', async () => { + await attach() + const prompt = await seedApproval() + const journal = ( + host as unknown as { sessions: Map } + ).sessions.get(SESSION)!.journal + const appendItem = journal.appendItem.bind(journal) + const statusFailure = vi + .spyOn(journal, 'appendItem') + .mockImplementation((identity, body, options) => + body.kind === 'status' + ? Promise.reject(new Error('status persistence failed')) + : appendItem(identity, body, options) + ) + const fields = { + prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } + } + const params = { + envelope: envelope('agentSession.cancel', fields), + ...fields + } + + await expect(host.cancel(CALLER, params)).rejects.toThrow('status persistence failed') + statusFailure.mockRestore() + + await expect(host.cancel(CALLER, params)).resolves.toMatchObject({ + ok: true, + replayed: true, + value: { cancelled: true } + }) + expect(cancelTurn).toHaveBeenCalledTimes(1) + }) + + it('terminalizes the displayed prompt only after the matching turn interruption succeeds', async () => { + await attach() + const prompt = await seedApproval() + const fields = { + turnId: 'turn-1', + prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } + } + + expect( + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', fields), + ...fields + }) + ).toMatchObject({ ok: true, value: { cancelled: true } }) + + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + const items = page.ok ? page.page.items : [] + expect(items.find((item) => item.itemId === prompt.itemId)).toMatchObject({ + revision: prompt.revision + 1, + body: { + kind: 'approval', + resolution: { + state: 'cancelled', + selectedOptionId: null, + resolvedBy: CALLER.callerKey, + resolvedAt: NOW + } + } + }) + expect( + items.filter( + (item) => + (item.body.kind === 'approval' || item.body.kind === 'question') && + item.body.resolution.state === 'pending' + ) + ).toEqual([]) + }) + + it('terminalizes every pending row for the cancelled logical question request', async () => { + await attach() + const prompts = await seedQuestionGroup() + promptCancellation.mockReturnValue({ + turnId: 'provider-child-turn', + itemIds: prompts.map((prompt) => prompt.itemId) + }) + const displayed = prompts[0]! + const fields = { + prompt: { itemId: displayed.itemId, expectedRevision: displayed.revision } + } + + expect( + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', fields), + ...fields + }) + ).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(cancelTurn).toHaveBeenCalledWith( + expect.objectContaining({ + turnId: 'provider-child-turn', + promptItemId: displayed.itemId + }) + ) + + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + const group = page.ok + ? page.page.items.filter((item) => prompts.some((prompt) => prompt.itemId === item.itemId)) + : [] + expect(group).toHaveLength(2) + expect(group.every((item) => item.revision === 2)).toBe(true) + expect( + group.every( + (item) => + (item.body.kind === 'question' || item.body.kind === 'approval') && + item.body.resolution.state === 'cancelled' + ) + ).toBe(true) + }) + + it('does not interrupt or settle when the displayed prompt revision is stale', async () => { + await attach() + const prompt = await seedApproval() + const fields = { + turnId: 'turn-1', + prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision + 1 } + } + + expect( + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', fields), + ...fields + }) + ).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_item_revision_stale', currentRevision: prompt.revision } + }) + expect(cancelTurn).not.toHaveBeenCalled() + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(page.ok && page.page.items.find((item) => item.itemId === prompt.itemId)).toMatchObject({ + revision: prompt.revision, + body: { resolution: { state: 'pending' } } + }) + }) + + it('does not let a restored prompt interrupt an unrelated live turn', async () => { + const prompt = await seedApproval() + promptCancellation.mockReturnValue(null) + await attach() + const fields = { + turnId: 'turn-1', + prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } + } + + expect( + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', fields), + ...fields + }) + ).toMatchObject({ ok: false, refusal: { code: 'agent_session_item_revision_stale' } }) + expect(cancelTurn).not.toHaveBeenCalled() + }) + + it('leaves the displayed prompt pending when the provider refuses cancellation', async () => { + await attach() + const prompt = await seedApproval() + cancelTurn.mockResolvedValueOnce({ cancelled: false }) + const fields = { + turnId: 'turn-1', + prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } + } + + expect( + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', fields), + ...fields + }) + ).toMatchObject({ ok: true, value: { cancelled: false } }) + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(page.ok && page.page.items.find((item) => item.itemId === prompt.itemId)).toMatchObject({ + revision: prompt.revision, + body: { resolution: { state: 'pending' } } + }) + }) + + it('preserves a provider-settled prompt that races the confirmed interruption', async () => { + await attach() + const prompt = await seedApproval() + const journal = ( + host as unknown as { sessions: Map } + ).sessions.get(SESSION)!.journal + cancelTurn.mockImplementationOnce(async () => { + await journal.appendItem( + { provider: 'orca', clientMessageId: `codex-prompt:${THREAD}:approval-1` }, + { + kind: 'approval', + title: 'Run the command?', + detail: null, + options: [{ id: 'allow', label: 'Allow' }], + resolution: { + state: 'resolved', + selectedOptionId: 'allow', + resolvedBy: 'provider-race', + resolvedAt: NOW - 1 + } + }, + { fence: 1 } + ) + return { cancelled: true } + }) + const fields = { + turnId: 'turn-1', + prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } + } + + expect( + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', fields), + ...fields + }) + ).toMatchObject({ ok: true, value: { cancelled: true } }) + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(page.ok && page.page.items.find((item) => item.itemId === prompt.itemId)).toMatchObject({ + revision: prompt.revision + 1, + body: { + resolution: { state: 'resolved', selectedOptionId: 'allow', resolvedBy: 'provider-race' } + } + }) + }) }) describe('respondToPrompt', () => { it('commits the answer before the provider callback', async () => { - const prompt = await seedApproval() await attach() + const prompt = await seedApproval() const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' } const result = await host.respondToPrompt(CALLER, { envelope: envelope('agentSession.respondTo:approval', fields), @@ -254,8 +478,8 @@ describe('respondToPrompt', () => { }) it('refuses a second answer to one prompt and says which answer won', async () => { - const prompt = await seedApproval() await attach() + const prompt = await seedApproval() const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' } await host.respondToPrompt(CALLER, { envelope: envelope('agentSession.respondTo:approval', fields), @@ -281,8 +505,8 @@ describe('respondToPrompt', () => { }) it('refuses an option the prompt does not offer', async () => { - const prompt = await seedApproval() await attach() + const prompt = await seedApproval() const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'deny' } expect( await host.respondToPrompt(CALLER, { @@ -295,8 +519,8 @@ describe('respondToPrompt', () => { }) it("does not turn a recorded refusal into another client's successful answer", async () => { - const prompt = await seedApproval() await attach() + const prompt = await seedApproval() const rejectedFields = { itemId: prompt.itemId, expectedRevision: prompt.revision, @@ -326,8 +550,8 @@ describe('respondToPrompt', () => { }) it('keeps the answer and reports it undelivered when the provider callback throws', async () => { - const prompt = await seedApproval() await attach() + const prompt = await seedApproval() answerPrompt.mockRejectedValueOnce(new Error('pipe closed')) const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' } const result = await host.respondToPrompt(CALLER, { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts index 5a40816ef82..83cc7c5499f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts @@ -76,27 +76,53 @@ export function sendPlan(params: { export function cancelPlan(params: { envelope: AgentSessionMutationEnvelope - turnId: string + turnId?: string scope?: 'background-tasks' taskId?: string + prompt?: { itemId: string; expectedRevision: number } }): MutationPlan { return { method: 'agentSession.cancel', fields: { - turnId: params.turnId, + ...(params.turnId ? { turnId: params.turnId } : {}), ...(params.scope ? { scope: params.scope } : {}), - ...(params.taskId ? { taskId: params.taskId } : {}) + ...(params.taskId ? { taskId: params.taskId } : {}), + ...(params.prompt ? { prompt: params.prompt } : {}) }, run: (ctx) => performCancel(ctx, { clientOperationId: params.envelope.clientOperationId, turnId: params.turnId, ...(params.scope ? { scope: params.scope } : {}), - ...(params.taskId ? { taskId: params.taskId } : {}) + ...(params.taskId ? { taskId: params.taskId } : {}), + ...(params.prompt ? { prompt: params.prompt } : {}) }), - // Interrupting twice would kill a turn the client never asked to stop, so a - // replay reports the turn as already handled instead. - replay: () => ({ turnId: params.turnId, cancelled: false }) + settledOutcome: (value) => ({ + status: 'succeeded', + sessionId: params.envelope.sessionId, + cancelled: value.cancelled, + cancelledTurnId: value.turnId + }), + recoverUnknownFromDurableState: params.prompt !== undefined, + // Interrupting twice would kill a turn the client never asked to stop. + replay: (ctx, outcome) => { + if (outcome.status === 'unknown' && params.prompt) { + const item = ctx.journal + .snapshot() + .items.find((candidate) => candidate.itemId === params.prompt?.itemId) + const cancelled = + (item?.body.kind === 'approval' || item?.body.kind === 'question') && + item.body.resolution.state === 'cancelled' + return cancelled ? { turnId: params.turnId ?? params.prompt.itemId, cancelled: true } : null + } + return { + turnId: + outcome.status === 'succeeded' && outcome.cancelledTurnId + ? outcome.cancelledTurnId + : (params.turnId ?? params.prompt?.itemId ?? 'unknown'), + cancelled: outcome.status === 'succeeded' && outcome.cancelled === true + } + } } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancellation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancellation.ts new file mode 100644 index 00000000000..8aa5b07fc61 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancellation.ts @@ -0,0 +1,111 @@ +import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { + AgentJournalRenderItem, + AgentJournalResolution +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire' +import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' + +type PromptTarget = { itemId: string; expectedRevision: number } +type PromptCancellationRefusal = Extract, { ok: false }> + +function invalid(message: string): PromptCancellationRefusal { + return { ok: false, refusal: { code: 'agent_session_operation_invalid', message } } +} + +export async function preparePromptCancellation( + ctx: AgentSessionTurnContext, + target: PromptTarget +): Promise< + | PromptCancellationRefusal + | { + ok: true + turnId: string + prompts: PromptTarget[] + } +> { + const displayed = await ctx.journal.readItem(target.itemId) + const displayedRefusal = promptCancellationRefusal(target, displayed) + if (displayedRefusal) { + return displayedRefusal + } + const cancellation = ctx.adapter.promptCancellation?.({ + sessionId: ctx.sessionId, + itemId: target.itemId, + fence: ctx.fence + }) + if ( + !cancellation || + !cancellation.itemIds.includes(target.itemId) || + new Set(cancellation.itemIds).size !== cancellation.itemIds.length + ) { + return invalid(`Item ${target.itemId} is not pending on a cancellable turn.`) + } + const items = await ctx.journal.readItems(cancellation.itemIds) + if (!items) { + return invalid(`The prompt group for ${target.itemId} is incomplete.`) + } + const prompts: PromptTarget[] = [] + for (const item of items) { + if ( + !parseAgentJournalItemKey(item.itemId) || + (item.body.kind !== 'approval' && item.body.kind !== 'question') + ) { + return invalid(`Item ${item.itemId} is not a pending prompt.`) + } + if (item.body.resolution.state === 'pending') { + prompts.push({ itemId: item.itemId, expectedRevision: item.revision }) + } + } + const currentDisplayed = prompts.find((prompt) => prompt.itemId === target.itemId) + if (!currentDisplayed || currentDisplayed.expectedRevision !== target.expectedRevision) { + return ( + promptCancellationRefusal( + target, + items.find((item) => item.itemId === target.itemId) ?? null + ) ?? invalid(`Item ${target.itemId} is not a pending prompt.`) + ) + } + return { ok: true, turnId: cancellation.turnId, prompts } +} + +function promptCancellationRefusal( + target: PromptTarget, + item: AgentJournalRenderItem | null +): PromptCancellationRefusal | null { + if (!item || (item.body.kind !== 'approval' && item.body.kind !== 'question')) { + return invalid(`Item ${target.itemId} is not a pending prompt.`) + } + if (!parseAgentJournalItemKey(target.itemId)) { + return invalid(`Item id ${target.itemId} is not a well-formed item key.`) + } + if (item.revision !== target.expectedRevision) { + return refusal( + 'agent_session_item_revision_stale', + `Item ${target.itemId} has moved on.`, + item.revision, + item.body.resolution + ) + } + if (item.body.resolution.state !== 'pending') { + return refusal( + 'agent_session_already_resolved', + `Item ${target.itemId} was already ${item.body.resolution.state}.`, + item.revision, + item.body.resolution + ) + } + return null +} + +function refusal( + code: AgentSessionWireRefusal['code'], + message: string, + currentRevision: number, + resolution: AgentJournalResolution +): PromptCancellationRefusal { + return { + ok: false, + refusal: { code, message, currentRevision, resolution } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts index 8ffb7acf6ce..250c407fb9f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts @@ -5,6 +5,7 @@ import type { AgentSessionJournal } from '../agent-session-journal/journal-store import { runningTurnLifecycleRevisions, settleStaleRunningTurnsOnAcquire, + stalePromptCancellationRevisions, turnVerdictFromDeathEvidence } from './structured-agent-session-stale-turn-verdict' @@ -43,6 +44,35 @@ function legacyLifecycleItem(turnId: string, startedAt: number): AgentJournalRen } } +function approvalItem(state: 'pending' | 'resolved'): AgentJournalRenderItem { + return { + itemId: agentJournalItemKey({ + provider: 'codex', + threadId: THREAD, + turnId: 'turn-2', + ordinal: 3 + }), + revision: 1, + sequence: 3, + observedAt: 3, + body: { + kind: 'approval', + title: 'Run command?', + detail: null, + options: [], + resolution: + state === 'pending' + ? { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + : { + state: 'resolved', + selectedOptionId: 'allow', + resolvedBy: 'client-1', + resolvedAt: 2 + } + } + } +} + describe('turn verdict from death evidence', () => { it('earns an end time only from an observed exit', () => { expect( @@ -105,6 +135,25 @@ describe('running turn lifecycle revisions', () => { }) }) +describe('stale prompt revisions', () => { + it('cancels only pending prompts that cannot belong to the replacement child', () => { + expect( + stalePromptCancellationRevisions([approvalItem('pending'), approvalItem('resolved')]) + ).toEqual([ + expect.objectContaining({ + body: expect.objectContaining({ + resolution: { + state: 'cancelled', + selectedOptionId: null, + resolvedBy: null, + resolvedAt: null + } + }) + }) + ]) + }) +}) + describe('stale running turns on a cold acquire', () => { function journalWith(items: AgentJournalRenderItem[]) { const appendLifecycleBatch = vi.fn(async () => ({ epoch: 'epoch-1', sequence: 9 })) @@ -145,6 +194,32 @@ describe('stale running turns on a cold acquire', () => { }) }) + it('terminalizes a pending prompt restored from the prior provider child', async () => { + const prompt = approvalItem('pending') + const { journal, appendLifecycleBatch } = journalWith([prompt]) + + await expect( + settleStaleRunningTurnsOnAcquire({ + journal, + sessionId: 'session-1', + fence: 14, + acquisitionGeneration: 'generation-2' + }) + ).resolves.toBe(1) + + expect(appendLifecycleBatch).toHaveBeenCalledWith( + expect.objectContaining({ + mutations: [ + expect.objectContaining({ + body: expect.objectContaining({ + resolution: expect.objectContaining({ state: 'cancelled' }) + }) + }) + ] + }) + ) + }) + it('writes nothing when no turn is running and keys on the journal position without a generation', async () => { const idle = journalWith([ lifecycleItem('turn-1', 'completed', 1, { startedAt: 10, completedAt: 20 }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts index 940b0c8be17..abfa0a6d4cd 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts @@ -17,6 +17,7 @@ import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-re import { partitionJournalLifecycleMutations } from '../agent-session-journal/journal-lifecycle-batch-partition' import type { JournalLifecycleMutationInput } from '../agent-session-journal/journal-row-builders' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { cancelledJournalPromptBody } from '../agent-session-journal/journal-prompt-body-bounds' export type StructuredAgentSessionTurnVerdict = | { state: 'interrupted'; completedAt: number } @@ -58,6 +59,27 @@ export function runningTurnLifecycleRevisions( return revisions } +/** Pending prompts cannot belong to the replacement provider child. */ +export function stalePromptCancellationRevisions( + items: readonly AgentJournalRenderItem[] +): JournalLifecycleMutationInput[] { + const revisions: JournalLifecycleMutationInput[] = [] + for (const item of items) { + if ( + (item.body.kind !== 'approval' && item.body.kind !== 'question') || + item.body.resolution.state !== 'pending' + ) { + continue + } + const identity = parseAgentJournalItemKey(item.itemId) + const body = cancelledJournalPromptBody(item.body) + if (identity && body) { + revisions.push({ kind: 'item', identity, body }) + } + } + return revisions +} + function settledLifecycle( lifecycle: AgentJournalTurnLifecycle, verdict: StructuredAgentSessionTurnVerdict @@ -75,8 +97,7 @@ function settledLifecycle( return settled } -/** A running row found when a NEW child is acquired belongs to a generation whose exit nobody - * observed. Must run before that child's buffered events land, or a live turn would be judged. */ +/** Rows owned by the prior child must settle before the replacement's buffered events land. */ export async function settleStaleRunningTurnsOnAcquire(input: { journal: AgentSessionJournal sessionId: string @@ -84,10 +105,11 @@ export async function settleStaleRunningTurnsOnAcquire(input: { acquisitionGeneration: string | null }): Promise { const { journal } = input - const revisions = runningTurnLifecycleRevisions( - journal.snapshot().items, - UNVERIFIABLE_TURN_VERDICT - ) + const items = journal.snapshot().items + const revisions = [ + ...runningTurnLifecycleRevisions(items, UNVERIFIABLE_TURN_VERDICT), + ...stalePromptCancellationRevisions(items) + ] const generation = input.acquisitionGeneration ?? `seq-${journal.cursor().sequence}` const settlementId = `stale-turn:${input.sessionId}:${input.fence}:${generation}` for (const chunk of partitionJournalLifecycleMutations(settlementId, revisions)) { 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 35ca56a451c..5748a37a764 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 @@ -25,6 +25,7 @@ import type { AgentSessionDispatchOutcome, StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { preparePromptCancellation } from './structured-agent-session-prompt-cancellation' export { performSetOption } from './structured-agent-session-turns-options' export { performPrompt } from './structured-agent-session-turns-prompt' @@ -207,11 +208,21 @@ export async function performCancel( ctx: AgentSessionTurnContext, input: { clientOperationId: string - turnId: string + turnId?: string scope?: 'background-tasks' taskId?: string + prompt?: { itemId: string; expectedRevision: number } } ): Promise> { + const promptCancellation = + input.prompt && !input.scope ? await preparePromptCancellation(ctx, input.prompt) : null + if (promptCancellation && !promptCancellation.ok) { + return promptCancellation + } + const turnId = promptCancellation?.ok ? promptCancellation.turnId : input.turnId + if (!turnId) { + return invalid('Cancellation requires a turn or pending prompt.') + } let cancelled = false let note = 'Cancellation requested.' try { @@ -226,8 +237,9 @@ export async function performCancel( : ( await ctx.adapter.cancelTurn({ sessionId: ctx.sessionId, - turnId: input.turnId, - fence: ctx.fence + turnId, + fence: ctx.fence, + ...(input.prompt ? { promptItemId: input.prompt.itemId } : {}) }) ).cancelled if (!cancelled) { @@ -239,9 +251,21 @@ export async function performCancel( }` } if (input.scope) { - return { ok: true, value: { turnId: input.turnId, cancelled } } + return { ok: true, value: { turnId, cancelled } } + } + if (cancelled && promptCancellation?.ok) { + const settled = await ctx.journal.cancelPromptsAtRevisions({ + prompts: promptCancellation.prompts, + settlementId: `prompt-cancel:${input.clientOperationId}`, + resolvedBy: ctx.resolvedBy, + resolvedAt: ctx.now(), + fence: ctx.fence + }) + if (settled > 0) { + ctx.publish() + } } // Keyed by the operation id so a replayed cancel upserts one item, not two. await appendStatus(ctx, input.clientOperationId, note) - return { ok: true, value: { turnId: input.turnId, cancelled } } + return { ok: true, value: { turnId, cancelled } } } diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index c2d46b09818..7bbd70c8791 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry' import { + AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY, AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY, RUNTIME_CAPABILITIES, RUNTIME_PROTOCOL_VERSION, @@ -148,6 +149,7 @@ describe('capability gating', () => { it('advertises the capability without bumping the protocol version', () => { expect(RUNTIME_CAPABILITIES).toContain(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY) expect(RUNTIME_CAPABILITIES).toContain(AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY) + expect(RUNTIME_CAPABILITIES).toContain(AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY) expect(RUNTIME_CAPABILITIES).toContain(STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY) expect(RUNTIME_CAPABILITIES).toContain(STRUCTURED_AGENT_SESSION_REVEAL_RUNTIME_CAPABILITY) // Additive methods do not break an old client; bumping would strand every @@ -606,6 +608,16 @@ describe('method routing', () => { expect(hostCalls.cancel).toHaveBeenCalledWith(expect.anything(), params) }) + it('routes a prompt compare-and-set target without a client turn id', async () => { + const params = { + envelope: envelope(), + prompt: { itemId: 'item-1', expectedRevision: 2 } + } + + expect(await call('agentSession.cancel', params, STRUCTURED_CLIENT)).toMatchObject({ ok: true }) + expect(hostCalls.cancel).toHaveBeenCalledWith(expect.anything(), params) + }) + it('routes the structured handoff mutation through the host', async () => { const response = await call('agentSession.requestHandoff', { envelope: envelope(), @@ -651,6 +663,21 @@ describe('parameter validation', () => { expect(hostCalls.cancel).not.toHaveBeenCalled() }) + it('rejects prompt targets on background-task cancellation', async () => { + await rejects('agentSession.cancel', { + envelope: envelope(), + turnId: 'background-tasks', + scope: 'background-tasks', + prompt: { itemId: 'item-1', expectedRevision: 1 } + }) + expect(hostCalls.cancel).not.toHaveBeenCalled() + }) + + it('requires either a turn id or a prompt target for cancellation', async () => { + await rejects('agentSession.cancel', { envelope: envelope() }) + expect(hostCalls.cancel).not.toHaveBeenCalled() + }) + it('refuses to let a client author anything but a user turn', async () => { await rejects( 'agentSession.send', diff --git a/src/renderer/src/components/native-chat/NativeChatApprovalCard.tsx b/src/renderer/src/components/native-chat/NativeChatApprovalCard.tsx index 3c5766b7f2f..f8ff041b0a9 100644 --- a/src/renderer/src/components/native-chat/NativeChatApprovalCard.tsx +++ b/src/renderer/src/components/native-chat/NativeChatApprovalCard.tsx @@ -40,7 +40,7 @@ export function NativeChatApprovalCard({