diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index 7bf70ecda3d2..cf20e98efeb3 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -73,6 +73,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-attachment-admission.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-background-stop-unreadable.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-claude-compact-stop.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-claude-echo-working.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts', @@ -189,6 +190,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-transition.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-unnamed-stop-races.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-unopened-tab.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-unsent-send-hold.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-view-start-after-failed-start.test.ts', diff --git a/mobile/src/session/mobile-native-chat-merged-snapshot-parity-hooks.test.tsx b/mobile/src/session/mobile-native-chat-merged-snapshot-parity-hooks.test.tsx index 5e933c94e49b..19dfa70f8cb3 100644 --- a/mobile/src/session/mobile-native-chat-merged-snapshot-parity-hooks.test.tsx +++ b/mobile/src/session/mobile-native-chat-merged-snapshot-parity-hooks.test.tsx @@ -90,7 +90,8 @@ const HOST_SUPPORT = { queuedMessages: true, queuedCommands: false, statusFeed: true, - quietRepeatedStop: true + quietRepeatedStop: true, + conversationStop: false } type Phone = { diff --git a/mobile/src/session/mobile-structured-agent-session-cancel.ts b/mobile/src/session/mobile-structured-agent-session-cancel.ts index 49232afeb326..9077c0238eef 100644 --- a/mobile/src/session/mobile-structured-agent-session-cancel.ts +++ b/mobile/src/session/mobile-structured-agent-session-cancel.ts @@ -30,6 +30,8 @@ export async function requestMobileStructuredAgentSessionCancel(args: { prompt?: PromptIdentity /** Whether the host answers a repeated Stop of a turn quietly; null until the status probe answers. */ hostAnswersRepeatedStops: boolean | null + /** Whether the host takes a Stop naming no turn; null until the status probe answers. */ + hostStopsConversation: boolean | null /** Stops still on their way, by what they stop; against a host that does not answer a repeat * quietly, a press for the same one joins it here. */ inFlight: Map> @@ -38,15 +40,16 @@ export async function requestMobileStructuredAgentSessionCancel(args: { const { client, enabled, inFlight, onSendError, sessionId, stateRef } = args const current = stateRef.current const turnId = runningStructuredAgentSessionTurnId(current) - if (!client || !sessionId || !enabled || current.fence === null || !turnId) { + // Check the capability before fields enter the fingerprint. + const prompt = args.prompt && args.promptCancelSupported === true ? args.prompt : undefined + // Such a host stops whatever the chat has in flight, a send the agent has not echoed included; + // a card's Cancel names the turn that raised it, prompt identity sent or not. + const unnamed = args.hostStopsConversation === true && !args.prompt + if (!client || !sessionId || !enabled || current.fence === null || (!turnId && !unnamed)) { onSendError('Stop not sent') return false } - // Check the capability before fields enter the fingerprint. - const fields = { - turnId, - ...(args.prompt && args.promptCancelSupported === true ? { prompt: args.prompt } : {}) - } + const fields = unnamed ? {} : { turnId, ...(prompt ? { prompt } : {}) } // Every press is its own Stop: a kept id would be answered from the last one and stop nothing. const fence = current.fence const stop = () => sendStop({ client, sessionId, fence, fields, onSendError }) diff --git a/mobile/src/session/mobile-structured-agent-session-host-support.test.ts b/mobile/src/session/mobile-structured-agent-session-host-support.test.ts index e5debf14376a..12283903ad3e 100644 --- a/mobile/src/session/mobile-structured-agent-session-host-support.test.ts +++ b/mobile/src/session/mobile-structured-agent-session-host-support.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest' import { + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY, AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY, AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY, @@ -17,7 +18,8 @@ describe('structuredAgentSessionHostSupport', () => { queuedMessages: false, queuedCommands: false, quietRepeatedStop: false, - statusFeed: false + statusFeed: false, + conversationStop: false } expect(structuredAgentSessionHostSupport([])).toEqual(none) expect( @@ -38,5 +40,8 @@ describe('structuredAgentSessionHostSupport', () => { expect( structuredAgentSessionHostSupport([AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY]) ).toEqual({ ...none, statusFeed: true }) + expect( + structuredAgentSessionHostSupport([AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY]) + ).toEqual({ ...none, conversationStop: true }) }) }) diff --git a/mobile/src/session/mobile-structured-agent-session-host-support.ts b/mobile/src/session/mobile-structured-agent-session-host-support.ts index 84c329e63738..967cea4341c9 100644 --- a/mobile/src/session/mobile-structured-agent-session-host-support.ts +++ b/mobile/src/session/mobile-structured-agent-session-host-support.ts @@ -1,4 +1,5 @@ import { + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY, AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY, AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY, @@ -20,6 +21,8 @@ export type StructuredAgentSessionHostSupport = { statusFeed: boolean /** A Stop that stopped nothing adds no row, so a repeated Stop is quiet. */ quietRepeatedStop: boolean + /** A Stop naming no turn stops what the chat has in flight; an older host needs the turn. */ + conversationStop: boolean } export function structuredAgentSessionHostSupport( @@ -31,6 +34,7 @@ export function structuredAgentSessionHostSupport( queuedMessages: capabilities.includes(AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY), queuedCommands: capabilities.includes(AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY), statusFeed: capabilities.includes(AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY), - quietRepeatedStop: capabilities.includes(AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY) + quietRepeatedStop: capabilities.includes(AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY), + conversationStop: capabilities.includes(AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY) } } diff --git a/mobile/src/session/mobile-structured-agent-session-unnamed-stop.test.ts b/mobile/src/session/mobile-structured-agent-session-unnamed-stop.test.ts new file mode 100644 index 000000000000..b5efabfec46a --- /dev/null +++ b/mobile/src/session/mobile-structured-agent-session-unnamed-stop.test.ts @@ -0,0 +1,102 @@ +import { describe, expect, it, vi } from 'vitest' +import type { StructuredAgentSessionState } from '../../../src/shared/structured-agent-session-reducer' +import type { RpcClient } from '../transport/rpc-client' +import { requestMobileStructuredAgentSessionCancel } from './mobile-structured-agent-session-cancel' + +type SentParams = Record & { envelope: { clientOperationId: string } } + +function recordingClient(sent: SentParams[]): RpcClient { + const sendRequest = async (_method: string, params: SentParams) => { + sent.push(params) + return { ok: true, result: { ok: true, value: { cancelled: true } }, _meta: { runtimeId: 'r' } } + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: cancel reaches only `sendRequest`. + return { sendRequest } as unknown as RpcClient +} + +function state(turnId: string | null): StructuredAgentSessionState { + const items = turnId + ? [ + { + itemId: 'status-1', + revision: 1, + sequence: 1, + observedAt: 10, + body: { kind: 'status', text: 'Working', turnLifecycle: { turnId, state: 'running' } } + } + ] + : [] + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: cancel reads only the fence and the running turn. + return { fence: 3, items } as unknown as StructuredAgentSessionState +} + +function press(input: { + sent: SentParams[] + turnId: string | null + hostStopsConversation: boolean | null + prompt?: { itemId: string; expectedRevision: number } + promptCancelSupported?: boolean +}) { + const onSendError = vi.fn() + const pressed = requestMobileStructuredAgentSessionCancel({ + client: recordingClient(input.sent), + sessionId: 'session-1', + enabled: true, + stateRef: { current: state(input.turnId) }, + promptCancelSupported: input.promptCancelSupported ?? true, + ...(input.prompt ? { prompt: input.prompt } : {}), + hostAnswersRepeatedStops: true, + hostStopsConversation: input.hostStopsConversation, + inFlight: new Map(), + onSendError + }) + return { pressed, onSendError } +} + +describe('phone Stop against a host that stops what the chat has in flight', () => { + it('stops a send the agent has not answered yet, naming no turn', async () => { + const sent: SentParams[] = [] + const { pressed, onSendError } = press({ sent, turnId: null, hostStopsConversation: true }) + expect(await pressed).toBe(true) + expect(onSendError).not.toHaveBeenCalled() + expect(sent).toHaveLength(1) + expect(sent[0]).not.toHaveProperty('turnId') + }) + + it('names no turn while a turn runs either', async () => { + const sent: SentParams[] = [] + expect(await press({ sent, turnId: 'turn-1', hostStopsConversation: true }).pressed).toBe(true) + expect(sent[0]).not.toHaveProperty('turnId') + }) + + it("keeps naming a card's turn", async () => { + const sent: SentParams[] = [] + const prompt = { itemId: 'prompt-1', expectedRevision: 2 } + await press({ sent, turnId: 'turn-1', hostStopsConversation: true, prompt }).pressed + expect(sent[0]).toMatchObject({ turnId: 'turn-1', prompt }) + }) + + it("keeps naming a card's turn when the host cannot take the card's identity", async () => { + const sent: SentParams[] = [] + const prompt = { itemId: 'prompt-1', expectedRevision: 2 } + await press({ + sent, + turnId: 'turn-1', + hostStopsConversation: true, + prompt, + promptCancelSupported: false + }).pressed + expect(sent[0]).toMatchObject({ turnId: 'turn-1' }) + expect(sent[0]).not.toHaveProperty('prompt') + }) + + it('names the turn to an older host, and sends nothing there before one opens', async () => { + const sent: SentParams[] = [] + await press({ sent, turnId: 'turn-1', hostStopsConversation: false }).pressed + expect(sent[0]).toMatchObject({ turnId: 'turn-1' }) + const before = press({ sent, turnId: null, hostStopsConversation: null }) + expect(await before.pressed).toBe(false) + expect(before.onSendError).toHaveBeenCalledWith('Stop not sent') + expect(sent).toHaveLength(1) + }) +}) diff --git a/mobile/src/session/mobile-structured-operation-id-retirement.test.ts b/mobile/src/session/mobile-structured-operation-id-retirement.test.ts index 03181db4100b..9d285f782af5 100644 --- a/mobile/src/session/mobile-structured-operation-id-retirement.test.ts +++ b/mobile/src/session/mobile-structured-operation-id-retirement.test.ts @@ -57,6 +57,7 @@ function cancelArgs(client: RpcClient, hostAnswersRepeatedStops: boolean | null stateRef: { current: runningState() }, promptCancelSupported: null, hostAnswersRepeatedStops, + hostStopsConversation: null, inFlight: new Map>(), onSendError: vi.fn() } diff --git a/mobile/src/session/use-mobile-native-chat-controller.test.ts b/mobile/src/session/use-mobile-native-chat-controller.test.ts index 20ebc232e60d..55c3d4e15331 100644 --- a/mobile/src/session/use-mobile-native-chat-controller.test.ts +++ b/mobile/src/session/use-mobile-native-chat-controller.test.ts @@ -58,6 +58,7 @@ const structuredQuestion = { optionTokens: ['choice-a', 'choice-b'] } const structuredActivity = { isWorking: false, turnId: null as string | null } +let harnessHostSupport: ReturnType | null = null const structuredSessionState = { messages: [] as unknown[], status: 'ready', @@ -162,6 +163,7 @@ import { } from './use-mobile-native-chat-controller' import type { MobileNativeChatStatus } from './use-mobile-native-chat-session' import { mobileNativeChatPromptDismissals } from './mobile-native-chat-prompt-dismissals' +import { structuredAgentSessionHostSupport } from './mobile-structured-agent-session-host-support' const sendWithOutcome = vi.mocked(sendMobileNativeChatMessageWithOutcome) @@ -219,6 +221,7 @@ describe('useMobileNativeChatController handleNativeChatSend', () => { deviceTokenRef: { current: null }, nativeChatTranscriptIsLocalReadable: true, nativeChatInputLeaseReady: inputLeaseReady, + agentSessionHostSupport: harnessHostSupport, onSendError, onSendResolved }) @@ -385,7 +388,14 @@ describe('useMobileNativeChatController handleNativeChatSend', () => { }) expect(controller?.nativeChatCanStop).toBe(true) expect(controller?.handleNativeChatCancelPrompt).toEqual(expect.any(Function)) + // A host that stops what the chat has in flight takes Stop before a turn; a card still can't. + structuredActivity.turnId = null + harnessHostSupport = structuredAgentSessionHostSupport(['agent-session.conversation-stop.v1']) + await act(async () => renderer?.update(createElement(Harness, props))) + expect(controller?.nativeChatCanStop).toBe(true) + expect(controller?.handleNativeChatCancelPrompt).toBeUndefined() } finally { + harnessHostSupport = null structuredActivity.isWorking = false structuredActivity.turnId = null } diff --git a/mobile/src/session/use-mobile-native-chat-controller.ts b/mobile/src/session/use-mobile-native-chat-controller.ts index c97abb27a515..e525d52fa721 100644 --- a/mobile/src/session/use-mobile-native-chat-controller.ts +++ b/mobile/src/session/use-mobile-native-chat-controller.ts @@ -302,8 +302,10 @@ export function useMobileNativeChatController(args: { nativeChatTurnJournal: activeChatStructured ? structuredNativeChat.turnJournal : null, // Only the structured lane names a refusal's cause; the bridge lane's starved ones drop none. nativeChatCommandRefusalCauses: structuredNativeChat.commandRefusalCauses, + // A host that stops what the chat has in flight takes Stop before the agent opens a turn. nativeChatCanStop: activeChatStructured - ? structuredNativeChat.turnId !== null + ? structuredNativeChat.turnId !== null || + (agentSessionHostSupport?.conversationStop === true && structuredNativeChat.isWorking) : nativeChatAgentWorking, nativeChatStreamingText, nativeChatStreamLive, diff --git a/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx index 7a14895fd66e..e45c4c11aa0c 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx @@ -124,7 +124,8 @@ function Harness({ queuedMessages: false, queuedCommands: false, statusFeed: false, - quietRepeatedStop: false + quietRepeatedStop: false, + conversationStop: false }, onSendError: vi.fn() }) diff --git a/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts b/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts index 7a0c57b9b352..68d323c81970 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts +++ b/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts @@ -18,7 +18,8 @@ export const CAPABLE: StructuredAgentSessionHostSupport = { queuedMessages: true, queuedCommands: false, statusFeed: false, - quietRepeatedStop: false + quietRepeatedStop: false, + conversationStop: false } export const LEGACY: StructuredAgentSessionHostSupport = { ...CAPABLE, queuedMessages: false } diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts index 4b11ee4fd582..f4d2e7073de7 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.ts +++ b/mobile/src/session/use-mobile-structured-agent-session.ts @@ -77,6 +77,7 @@ export function useMobileStructuredAgentSession(args: { const commandsWait = queueCapable && hostSupport?.queuedCommands === true const promptCancelSupported = hostSupport?.promptCancel ?? null const hostAnswersRepeatedStops = hostSupport?.quietRepeatedStop ?? null + const hostStopsConversation = hostSupport?.conversationStop ?? null const sessionKey = encodeNativeChatTranscriptIdentity([sourceIdentity, agent, sessionId]) const commandPendingRef = useRef(false) // Against a host that predates the quiet repeated Stop, a Stop of a turn still being stopped joins it. @@ -248,6 +249,7 @@ export function useMobileStructuredAgentSession(args: { client, enabled, hostAnswersRepeatedStops, + hostStopsConversation, inFlight: inFlightStopsRef.current, onSendError, prompt, @@ -259,6 +261,7 @@ export function useMobileStructuredAgentSession(args: { client, enabled, hostAnswersRepeatedStops, + hostStopsConversation, onSendError, promptCancelSupported, sessionId, diff --git a/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx b/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx index 79f761f5511e..61debb60c019 100644 --- a/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx +++ b/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx @@ -267,6 +267,7 @@ describe("useMobileStructuredAgentSession and the host's Stopping", () => { queuedMessages: false, queuedCommands: false, quietRepeatedStop: false, + conversationStop: false, statusFeed }, agent: 'codex', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-stop-unreadable.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-stop-unreadable.test.ts new file mode 100644 index 000000000000..9a9be3fd46e2 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-background-stop-unreadable.test.ts @@ -0,0 +1,102 @@ +import { afterEach, expect, it, vi } from 'vitest' +import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestOperationId +} from './structured-agent-session-host-test-data' + +const child: AgentChildWorkView = { + id: 'child-1', + providerId: 'task-1', + kind: 'agent', + state: 'working', + membership: 'live', + firstObservedAt: NOW, + observedAt: NOW, + stoppable: true, + invocation: { invocationId: 'invocation-1', generation: 1 } +} + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +function taskStop(testRig: QueuedMessageTestRig, taskId?: string) { + const fields = { + turnId: 'turn-1', + scope: 'background-tasks' as const, + ...(taskId ? { taskId } : {}) + } + return { + ...fields, + envelope: testRig.envelope(fields, 'agentSession.cancel', hostTestOperationId()) + } +} + +async function rigWithTaskList(): Promise<{ + testRig: QueuedMessageTestRig + stopBackgroundTasks: ReturnType + setReadFails: (fails: boolean) => void +}> { + const testRig = await createQueuedMessageTestRig() + rig = testRig + let readFails = false + testRig.host.deps.statusSink = { + publish: () => {}, + forget: () => {}, + readChildWork: () => { + if (readFails) { + throw new Error('host child store temporarily unreadable') + } + return [child] + } + } + await testRig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + testRig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + return { testRig, stopBackgroundTasks, setReadFails: (fails) => void (readFails = fails) } +} + +async function unconfirmedNotes(testRig: QueuedMessageTestRig) { + return (await testRig.host.journalSnapshot(SESSION)).items.filter( + (item) => item.body.kind === 'status' && item.body.failure?.kind === 'cancelUnconfirmed' + ) +} + +for (const taskId of ['task-1', undefined]) { + it(`answers ${taskId ? 'one task' : 'every task'}'s Stop unconfirmed while the task list cannot be read`, async () => { + const { testRig, stopBackgroundTasks, setReadFails } = await rigWithTaskList() + setReadFails(true) + expect(await testRig.host.cancel(QUEUED_RIG_CALLER, taskStop(testRig, taskId))).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + // Saved, so the answer is its receipt; the chat says the Stop went unconfirmed. + expect(await unconfirmedNotes(testRig)).toHaveLength(1) + setReadFails(false) + expect(await testRig.host.cancel(QUEUED_RIG_CALLER, taskStop(testRig, taskId))).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(stopBackgroundTasks).toHaveBeenCalledOnce() + }) +} + +it('still answers a task the readable list no longer offers as already ended', async () => { + const { testRig, stopBackgroundTasks } = await rigWithTaskList() + expect( + await testRig.host.cancel(QUEUED_RIG_CALLER, taskStop(testRig, 'task-gone')) + ).toMatchObject({ ok: true, value: { cancelled: false } }) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + expect(await unconfirmedNotes(testRig)).toHaveLength(0) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-stop.ts index 2aef5001aa16..edd946fb2417 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-background-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-background-stop.ts @@ -26,6 +26,8 @@ export async function stopStructuredAgentSessionBackgroundWork( input: Pick & { /** Captured before the stop goes out: its acknowledgement may end only these runs. */ targets: readonly (AgentChildWorkView & { providerId: string })[] + /** The host could not read the task list the targets came from. */ + unreadable?: true }, atRest: () => boolean ): Promise<{ cancelled: boolean; stillRunning?: true }> { @@ -59,5 +61,9 @@ export async function stopStructuredAgentSessionBackgroundWork( await input.unlisted.stop() return { cancelled: true } } + // A list the host could not read proves no task ended: unconfirmed, never "nothing to stop". + if (input.unreadable) { + throw new Error('the background task list could not be read') + } return { cancelled: background?.cancelled === true } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-targeted-cancel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-targeted-cancel.ts index 57d9196ead34..400d7f4c0ce3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-targeted-cancel.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-targeted-cancel.ts @@ -35,8 +35,10 @@ export async function runTargetedCancel( const { clientOperationId, prompt } = params if (params.scope) { // Resolved once, here: a retry is answered from the receipt and never reaches a newer task. - const targets = agentChildWorkStopTargets(params.childWork?.(), params.taskId) - if (targets.length === 0 && !(params.unlisted?.present() && atRest(ctx))) { + const children = params.childWork?.() + const targets = agentChildWorkStopTargets(children, params.taskId) + const unreadable = params.childWork !== undefined && children === undefined + if (targets.length === 0 && !unreadable && !(params.unlisted?.present() && atRest(ctx))) { return { ok: true, value: { ...named, cancelled: false } } } const accepted = await acceptStopTarget(ctx) @@ -47,6 +49,7 @@ export async function runTargetedCancel( operationId: clientOperationId, work: { targets, + ...(unreadable ? { unreadable: true as const } : {}), ...(params.acknowledgeStops ? { acknowledgeStops: params.acknowledgeStops } : {}), ...(params.unlisted ? { unlisted: params.unlisted } : {}) }, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unnamed-stop-races.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unnamed-stop-races.test.ts new file mode 100644 index 000000000000..6764b0ccc45e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unnamed-stop-races.test.ts @@ -0,0 +1,195 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { publishCodexTurnLifecycle } from '../../codex/codex-structured-journal-translation-turns' +import { + adapterFor, + fakeCodex, + THREAD_ID as CODEX_THREAD +} from '../../codex/codex-structured-session-adapter-fixture' +import { codexTurnLifecycleFake } from '../../codex/codex-turn-lifecycle-fake' +import type { CodexStructuredSessionAdapter } from '../../codex/codex-structured-session-adapter' +import { + createRestTestRig, + REST_TEST_CALLER as CALLER, + restTestSend, + type RestTestRig +} from './structured-agent-session-rest-test-rig' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION as SESSION, + HOST_TEST_THREAD as THREAD, + hostTestAttachParams, + hostTestOperationId +} from './structured-agent-session-host-test-data' + +// The Stop names no turn: the host stops whatever the chat has in flight, including a send the +// agent has not echoed yet. + +let restRig: RestTestRig | undefined +let adapter: CodexStructuredSessionAdapter | undefined +let queuedRig: QueuedMessageTestRig | undefined +afterEach(async () => { + vi.restoreAllMocks() + await adapter?.closeAll() + await restRig?.dispose() + await queuedRig?.dispose() + restRig = undefined + adapter = undefined + queuedRig = undefined +}) + +function unnamedStop(clientOperationId = hostTestOperationId()) { + return { + envelope: { + sessionId: SESSION, + clientOperationId, + expectedRuntimeFence: 1, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.cancel', + sessionId: SESSION, + fields: {} + }) + } + } +} + +/** A Codex send whose `turn/start` Codex answered, with no `turn/started` or user echo yet. */ +async function admittedCodexSend() { + const codex = fakeCodex() + const notify = (method: string, params: unknown) => + codex.connections.at(-1)?.handlers.onNotification?.(method, params) + const turns = codexTurnLifecycleFake(CODEX_THREAD, () => notify) + Object.assign(codex.routes, turns.routes) + const provider = adapterFor(codex, {}, [], { + onDispatchSettledLate: (settlement) => void restRig?.host.settleLateDispatch(settlement), + onEvent: (event) => { + if (event.type === 'ended' && 'cause' in event) { + void restRig?.host.handleAdapterEvent(event) + } + } + }) + adapter = provider + const current = await createRestTestRig({ adapter: provider }) + restRig = current + expect(await current.host.attach(CALLER, hostTestAttachParams(null))).toMatchObject({ ok: true }) + const send = restTestSend('work on this') + expect(await current.host.send(CALLER, send)).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(turns.turnId).toBe('turn-1')) + const id = send.envelope.clientOperationId + const interrupts = () => + codex.connections[0]!.calls.filter((call) => call.method === 'turn/interrupt') + return { current, turns, id, interrupts } +} + +it('Codex: reaches a turn that started before the user echo', async () => { + const { current, turns, id, interrupts } = await admittedCodexSend() + turns.start() + await current.host.flushStreamedEvents(SESSION) + expect( + (await current.host.journalSnapshot(SESSION)).submissions.find( + (row) => row.clientMessageId === id + ) + ).toMatchObject({ providerItemId: null, dispatchState: 'pending' }) + expect(await current.host.cancel(CALLER, unnamedStop())).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(interrupts()).toMatchObject([{ params: { threadId: CODEX_THREAD, turnId: 'turn-1' } }]) +}) + +it('Codex: reaches a turn admitted before its started notification or user echo', async () => { + const { current, turns, interrupts } = await admittedCodexSend() + turns.run() + expect(await current.host.cancel(CALLER, unnamedStop())).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(interrupts()).toMatchObject([{ params: { turnId: 'turn-1' } }]) +}) + +it('Codex: never answers a quiet success while an admitted turn cannot yet take an interrupt', async () => { + const { current, interrupts } = await admittedCodexSend() + const response = await current.host.cancel(CALLER, unnamedStop()) + expect(interrupts().length).toBeGreaterThan(0) + const child = current.host.collaboratorsForTests().sessions.get(SESSION)?.child + if (response.ok) { + // Success only once the agent's end is proven: the host lets go of a child only then. + expect(response.value.cancelled).toBe(true) + expect(child ?? null).toBeNull() + } else { + // While the agent may still run the turn, the Stop says it is unconfirmed. + expect(response.refusal.code).toBe('agent_session_operation_unknown') + } +}) + +it('Codex: a retried Stop replays and never interrupts a later turn', async () => { + const { current, turns, interrupts } = await admittedCodexSend() + turns.start() + await current.host.flushStreamedEvents(SESSION) + const stop = unnamedStop() + expect(await current.host.cancel(CALLER, stop)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + await vi.waitFor(() => expect(turns.turnId).toBeNull()) + expect(await current.host.send(CALLER, restTestSend('later work'))).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(turns.turnId).toBe('turn-2')) + turns.start() + await current.host.flushStreamedEvents(SESSION) + expect(await current.host.cancel(CALLER, stop)).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(interrupts()).toHaveLength(1) + expect(turns.turnId).toBe('turn-2') +}) + +it('reaches a turn opened before the user echo links it to the send', async () => { + queuedRig = await createQueuedMessageTestRig() + await queuedRig.workingSend() + publishCodexTurnLifecycle({ + sink: queuedRig.providerEvents(), + primaryThreadId: THREAD, + sessionId: SESSION, + threadId: THREAD, + turnId: 'turn-opening', + state: 'running' + }) + await queuedRig.host.flushStreamedEvents(SESSION) + expect(await queuedRig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(queuedRig.cancelTurn).toHaveBeenCalled() +}) + +it('stops the waiting send when a later send reaches the host just before the Stop', async () => { + queuedRig = await createQueuedMessageTestRig({ restartable: true }) + await queuedRig.workingSend() + const later = queuedRig.send('sent just before Stop') + expect(await later.result).toMatchObject({ ok: true }) + expect(await queuedRig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + // The waiting send's turn is interrupted, and the later one never reaches the agent. + expect(queuedRig.cancelTurn).toHaveBeenCalledOnce() + expect(await queuedRig.submission(later.id)).toMatchObject({ dispatchState: 'rejected' }) +}) + +it('a Stop after a Stop that failed still tries to stop the waiting send', async () => { + queuedRig = await createQueuedMessageTestRig({ restartable: true }) + await queuedRig.workingSend() + // The interrupt fails and the child's exit cannot be proven: the send may run on. + queuedRig.cancelTurn.mockImplementationOnce(async () => { + throw new Error('interrupt timed out') + }) + queuedRig.closeSession.mockImplementationOnce(async () => { + throw new Error('exit not proven') + }) + await queuedRig.host.cancel(QUEUED_RIG_CALLER, unnamedStop()) + expect(queuedRig.host.collaboratorsForTests().sessions.get(SESSION)?.child).toBeTruthy() + const tries = queuedRig.cancelTurn.mock.calls.length + queuedRig.closeSession.mock.calls.length + await queuedRig.host.cancel(QUEUED_RIG_CALLER, unnamedStop()) + expect( + queuedRig.cancelTurn.mock.calls.length + queuedRig.closeSession.mock.calls.length + ).toBeGreaterThan(tries) +})