diff --git a/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx b/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx index 5d582b111a3f..e20551af489b 100644 --- a/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx +++ b/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx @@ -49,13 +49,23 @@ function view(id: string, overrides: Partial = {}): AgentChi } type Props = Parameters[0] +type StopResult = Awaited> + +const REJECTED: StopResult = { status: 'rejected' } +const UNCONFIRMED: StopResult = { status: 'unknown' } +const STOPPED: StopResult = { status: 'accepted', value: { cancelled: true }, sameFence: true } +const NOTHING_STOPPED: StopResult = { + status: 'accepted', + value: { cancelled: false }, + sameFence: true +} function tasksFor( state: AgentSessionBackgroundTaskState | null, options: { streamLive?: boolean hostClockOffsetMs?: number - stop?: (taskId?: string) => Promise + stop?: Props['tasks']['stop'] } = {} ): Props['tasks'] { return { @@ -65,7 +75,7 @@ function tasksFor( options.hostClockOffsetMs ?? 0 ), sessionKey: 'session-a', - stop: options.stop ?? vi.fn(async () => undefined) + stop: options.stop ?? vi.fn(async () => REJECTED) } } @@ -264,8 +274,8 @@ describe('MobileNativeChatBackgroundTasks', () => { }) it('stops one row by its provider id and holds its button while the Stop is on its way', async () => { - let finish: () => void = () => {} - const stop = vi.fn(() => new Promise((resolve) => (finish = resolve))) + let finish: (result: StopResult) => void = () => {} + const stop = vi.fn(() => new Promise((resolve) => (finish = resolve))) const mounted = mount( tasksFor( { @@ -284,13 +294,65 @@ describe('MobileNativeChatBackgroundTasks', () => { expect(stop).toHaveBeenCalledTimes(1) expect(stop).toHaveBeenCalledWith('task-a') expect(stopButtons(mounted)[0]!.props.disabled).toBe(true) - await act(async () => finish()) + // Unconfirmed: the row is the verdict, so its Stop is offered again. + await act(async () => finish(UNCONFIRMED)) expect(stopButtons(mounted)[0]!.props.disabled).toBe(false) }) + it('keeps a confirmed Stop held until its row leaves, and offers it again if the row comes back', async () => { + let finish: (result: StopResult) => void = () => {} + const stop = vi.fn(() => new Promise((resolve) => (finish = resolve))) + const running = { state: 'monitoring' as const, supportsTaskStop: true } + const mounted = mount(tasksFor({ ...running, children: [view('a'), view('b')] }, { stop })) + const show = (children: AgentChildWorkView[]) => + act(() => { + mounted.update( + createElement(MobileNativeChatBackgroundTasks, { + tasks: tasksFor({ ...running, children }, { stop }) + }) + ) + }) + expand(mounted) + act(() => stopButtons(mounted)[0]!.props.onPress()) + await act(async () => finish(STOPPED)) + expect(stopButtons(mounted).map((button) => button.props.disabled)).toEqual([true, false]) + show([view('b')]) + show([view('a'), view('b')]) + expect(stopButtons(mounted).map((button) => button.props.disabled)).toEqual([false, false]) + act(() => stopButtons(mounted)[0]!.props.onPress()) + await act(async () => finish(NOTHING_STOPPED)) + expect(stopButtons(mounted)[0]!.props.disabled).toBe(false) + }) + + it('stops a backgrounded Codex command by its own id, and offers none on the subagent beside it', () => { + const stop = vi.fn(async () => REJECTED) + const mounted = mount( + tasksFor( + { + state: 'monitoring', + supportsTaskStop: true, + children: [ + view('agent', { providerId: 'codex-agent:child-1', stoppable: false }), + view('cmd', { + kind: 'command', + providerId: 'codex-command:exec-1', + description: 'pnpm dev' + }) + ] + }, + { stop } + ) + ) + expand(mounted) + const buttons = stopButtons(mounted) + expect(buttons.map((button) => button.props.accessibilityLabel)).toEqual(['Stop pnpm dev']) + act(() => buttons[0]!.props.onPress()) + expect(stop).toHaveBeenCalledWith('codex-command:exec-1') + }) + it('offers Stop all only to a host with no per-row stop that still accepts one', async () => { - let finish: () => void = () => {} - const stop = vi.fn(() => new Promise((resolve) => (finish = resolve))) + let finish: (result: StopResult) => void = () => {} + const stop = vi.fn(() => new Promise((resolve) => (finish = resolve))) const fallback = mount(tasksFor({ state: 'monitoring', children: [view('a')] }, { stop })) expand(fallback) expect(stopButtons(fallback).map((button) => button.props.accessibilityLabel)).toEqual([ @@ -305,7 +367,7 @@ describe('MobileNativeChatBackgroundTasks', () => { expect(stop).toHaveBeenCalledTimes(1) expect(stop).toHaveBeenCalledWith(undefined) expect(stopButtons(fallback)[0]!.props.disabled).toBe(true) - await act(async () => finish()) + await act(async () => finish(STOPPED)) expect(stopButtons(fallback)[0]!.props.disabled).toBe(false) const none = mount( tasksFor({ state: 'monitoring', supportsStopAll: false, children: [view('a')] }) diff --git a/mobile/src/session/MobileNativeChatBackgroundTasks.tsx b/mobile/src/session/MobileNativeChatBackgroundTasks.tsx index 33dd4807ede5..0b1e53e52621 100644 --- a/mobile/src/session/MobileNativeChatBackgroundTasks.tsx +++ b/mobile/src/session/MobileNativeChatBackgroundTasks.tsx @@ -1,4 +1,4 @@ -import { memo, useMemo, useRef, useState } from 'react' +import { memo, useMemo, useState } from 'react' import { Dimensions, type LayoutChangeEvent, @@ -36,6 +36,10 @@ import { AGENT_WORKING_COLOR, AgentStateDot } from '../components/AgentStateDot' import { useNow } from '../hooks/use-now' import { colors, radii, spacing, typography } from '../theme/mobile-theme' import type { MobileStructuredBackgroundTasks } from './use-mobile-structured-background-tasks' +import { + useMobileBackgroundTaskStops, + type MobileBackgroundTaskStopping as Stopping +} from './use-mobile-background-task-stops' import { mobileAgentChildRowName, mobileAgentChildRowText } from './mobile-agent-child-row-text' type TaskKind = AgentSessionBackgroundTask['kind'] @@ -64,9 +68,6 @@ function kindIconColor(kind: TaskKind): string { return kind === 'monitor' ? AGENT_WORKING_COLOR : colors.textMuted } -type Stopping = { taskIds: ReadonlySet; all: boolean } -const NOT_STOPPING: Stopping = { taskIds: new Set(), all: false } - function TaskRow({ row, now, @@ -149,14 +150,13 @@ function MobileNativeChatBackgroundTasksImpl({ }: { tasks: MobileStructuredBackgroundTasks }): React.JSX.Element | null { - const { view, rowContext, stop } = tasks + const { view, rowContext } = tasks const [expanded, setExpanded] = useState(false) // Before the first layout, the window less the strip's own margins, so it opens in its final form. const [narrow, setNarrow] = useState( () => Dimensions.get('window').width - 2 * spacing.lg < NARROW_STRIP_WIDTH ) - const [stopping, setStopping] = useState(NOT_STOPPING) - const stoppingRef = useRef(NOT_STOPPING) + const { stopping, onStop } = useMobileBackgroundTaskStops(tasks) const groups = useMemo( () => view.children !== undefined @@ -172,29 +172,6 @@ function MobileNativeChatBackgroundTasksImpl({ const hostNow = now - rowContext.hostClockOffsetMs const header = backgroundTasksHeaderContent(groups, { narrow, now: hostNow }, say) - const updateStopping = (next: Stopping): void => { - stoppingRef.current = next - setStopping(next) - } - // A press already on its way holds its button; the ref closes the same-frame double tap. - const onStop = (taskId?: string): void => { - const current = stoppingRef.current - if (taskId ? current.taskIds.has(taskId) : current.all) { - return - } - updateStopping({ - taskIds: taskId ? new Set([...current.taskIds, taskId]) : current.taskIds, - all: taskId ? current.all : true - }) - void stop(taskId).finally(() => { - const latest = stoppingRef.current - const taskIds = new Set(latest.taskIds) - if (taskId) { - taskIds.delete(taskId) - } - updateStopping({ taskIds, all: taskId ? latest.all : false }) - }) - } const onLayout = (event: LayoutChangeEvent): void => { setNarrow(event.nativeEvent.layout.width < NARROW_STRIP_WIDTH) } diff --git a/mobile/src/session/use-mobile-background-task-stops.ts b/mobile/src/session/use-mobile-background-task-stops.ts new file mode 100644 index 000000000000..7f989d61d5dd --- /dev/null +++ b/mobile/src/session/use-mobile-background-task-stops.ts @@ -0,0 +1,61 @@ +import { useRef, useState } from 'react' +import { structuredSessionConfirmedStopsStillListed } from '../../../src/shared/structured-session-background-tasks-view' +import type { MobileStructuredBackgroundTasks } from './use-mobile-structured-background-tasks' + +export type MobileBackgroundTaskStopping = { taskIds: ReadonlySet; all: boolean } + +const NO_TASKS: ReadonlySet = new Set() +const NOT_STOPPING: MobileBackgroundTaskStopping = { taskIds: NO_TASKS, all: false } + +/** The strip's Stop presses: which buttons hold, and the press that sends one. */ +export function useMobileBackgroundTaskStops( + tasks: Pick +): { stopping: MobileBackgroundTaskStopping; onStop: (taskId?: string) => void } { + const { view, stop } = tasks + const [stopping, setStopping] = useState(NOT_STOPPING) + const stoppingRef = useRef(NOT_STOPPING) + // Stops the host confirmed: each row keeps its stopping button until it leaves the strip. + const [confirmed, setConfirmed] = useState>(NO_TASKS) + const confirmedListed = structuredSessionConfirmedStopsStillListed(view, confirmed) + if (confirmedListed !== confirmed) { + // A row that left and came back is a task the Stop did not end, so it offers Stop again. + setConfirmed(confirmedListed) + } + + const updateStopping = (next: MobileBackgroundTaskStopping): void => { + stoppingRef.current = next + setStopping(next) + } + // A press already on its way holds its button; the ref closes the same-frame double tap. + const onStop = (taskId?: string): void => { + const current = stoppingRef.current + if (taskId ? current.taskIds.has(taskId) : current.all) { + return + } + updateStopping({ + taskIds: taskId ? new Set([...current.taskIds, taskId]) : current.taskIds, + all: taskId ? current.all : true + }) + void stop(taskId) + .then((result) => { + if (taskId && result.status === 'accepted' && result.value.cancelled) { + setConfirmed((held) => new Set([...held, taskId])) + } + }) + .finally(() => { + const latest = stoppingRef.current + const taskIds = new Set(latest.taskIds) + if (taskId) { + taskIds.delete(taskId) + } + updateStopping({ taskIds, all: taskId ? latest.all : false }) + }) + } + + return { + stopping: confirmedListed.size + ? { ...stopping, taskIds: new Set([...stopping.taskIds, ...confirmedListed]) } + : stopping, + onStop + } +} diff --git a/mobile/src/session/use-mobile-native-chat-composer-tray.test.tsx b/mobile/src/session/use-mobile-native-chat-composer-tray.test.tsx index eae7cbb8bec5..f51088beb25a 100644 --- a/mobile/src/session/use-mobile-native-chat-composer-tray.test.tsx +++ b/mobile/src/session/use-mobile-native-chat-composer-tray.test.tsx @@ -78,7 +78,7 @@ function tasks( sessionKey, view: structuredSessionBackgroundTasksView(state, null), rowContext: agentChildRowContextForSessionStream(true, 0), - stop: vi.fn(async () => undefined) + stop: vi.fn(async () => ({ status: 'rejected' as const })) } } diff --git a/mobile/src/session/use-mobile-structured-background-tasks.ts b/mobile/src/session/use-mobile-structured-background-tasks.ts index 5b32d82a7391..bab8597b30c5 100644 --- a/mobile/src/session/use-mobile-structured-background-tasks.ts +++ b/mobile/src/session/use-mobile-structured-background-tasks.ts @@ -3,12 +3,14 @@ import { useCallback, useMemo } from 'react' import type { AgentChildRowContext } from '../../../src/shared/agent-child-row-model' +import type { AgentSessionCancelResult } from '../../../src/shared/agent-session-wire' import { agentChildRowContextForSessionStream } from '../../../src/shared/agent-child-row-stream-context' import type { StructuredAgentSessionState } from '../../../src/shared/structured-agent-session-reducer' import { structuredSessionBackgroundTasksView, type StructuredSessionBackgroundTasksView } from '../../../src/shared/structured-session-background-tasks-view' +import type { StructuredAgentSessionMutationResult } from './mobile-structured-agent-session-rpc' import type { MobileStructuredAgentMutate } from './use-mobile-structured-agent-mutation' export type MobileStructuredBackgroundTasks = { @@ -19,7 +21,7 @@ export type MobileStructuredBackgroundTasks = { * host clock offset also moves the rows' host-stamped clocks onto the phone's. */ rowContext: AgentChildRowContext /** One child by its provider id, or, with none, every background task the host can stop. */ - stop: (taskId?: string) => Promise + stop: (taskId?: string) => Promise> } /** Coarser than one frame's delivery latency, finer than any clock a row shows. */ @@ -53,7 +55,7 @@ export function useMobileStructuredBackgroundTasks(args: { ) const stop = useCallback( (taskId?: string) => - mutate('agentSession.cancel', 'agentSession.cancel', { + mutate('agentSession.cancel', 'agentSession.cancel', { turnId: 'background-tasks', scope: 'background-tasks', ...(taskId ? { taskId } : {}) diff --git a/src/main/acp/acp-structured-child-stop.ts b/src/main/acp/acp-structured-child-stop.ts index 5836975ba6d9..c4994cefa0b9 100644 --- a/src/main/acp/acp-structured-child-stop.ts +++ b/src/main/acp/acp-structured-child-stop.ts @@ -35,20 +35,31 @@ export async function stopAcpChildren( fence: number, taskIds: readonly string[], at: () => number -): Promise<{ cancelled: boolean }> { +): Promise<{ cancelled: boolean; stillRunning?: true }> { const control = session.spec.dialect.subagentStop if (!control || !session.subagentStopSupported || session.fence !== fence) { throw new Error('ACP child cancellation is unavailable') } let cancelled = false + let stillRunning = false for (const id of taskIds) { const request = control.request(session.lane.translator.providerSessionId, id) - const value = await session.connection.requestExtension(request.method, request.params) + let value: unknown + try { + value = await session.connection.requestExtension(request.method, request.params) + } catch (error) { + // Grok answers an ended or unknown subagent normally, so its refusal means it did nothing; anything else leaves the effect unknown. + if (error instanceof AcpAgentError) { + stillRunning = true + continue + } + throw error + } const result = control.response(value, id) if (result.state && result.state !== 'working') { session.lane.apply(session.lane.translator.reconcileSubagent(id, result.state, at())) } cancelled ||= result.cancelled } - return { cancelled } + return { cancelled, ...(stillRunning ? { stillRunning: true as const } : {}) } } diff --git a/src/main/acp/acp-structured-child-work.test.ts b/src/main/acp/acp-structured-child-work.test.ts index 768b0bd1a355..c97843b1e6d5 100644 --- a/src/main/acp/acp-structured-child-work.test.ts +++ b/src/main/acp/acp-structured-child-work.test.ts @@ -358,9 +358,13 @@ describe('Grok roster evidence through the existing host child store', () => { } }) - it.each(['rpc-error', 'mismatched-id'])( + // Grok's own refusal means it did nothing; a reply about another subagent proves nothing, so it is unknown. + it.each([ + ['rpc-error', { ok: false, refusal: { code: 'agent_session_operation_invalid' } }], + ['mismatched-id', { ok: false, refusal: { code: 'agent_session_operation_unknown' } }] + ] as const)( 'does not settle or cancel the parent after a targeted %s', - async (failure) => { + async (failure, answer) => { const f = await fixture() await f.spawn('first') await f.spawn('sibling') @@ -371,7 +375,7 @@ describe('Grok roster evidence through the existing host child store', () => { f.agent.reply(frame, { result: { subagentId: 'other', cancelled: true } }) } }) - expect(await f.targetedStop('first')).toMatchObject({ ok: true, value: { cancelled: false } }) + expect(await f.targetedStop('first')).toMatchObject(answer) expect(f.childWork.views().every((view) => view.membership === 'live')).toBe(true) expect(f.agent.frames.filter((frame) => frame.method === 'session/cancel')).toEqual([]) expect(f.rig.child().closed).toBe(false) diff --git a/src/main/codex/codex-background-command-tracker.ts b/src/main/codex/codex-background-command-tracker.ts index 35027c823988..4327ce1f2621 100644 --- a/src/main/codex/codex-background-command-tracker.ts +++ b/src/main/codex/codex-background-command-tracker.ts @@ -18,6 +18,8 @@ type Command = { turnId: string | null backgrounded: boolean task: AgentSessionBackgroundTask + /** Codex's id for the command's process, which a background-terminal stop names. */ + processId?: string bytes: number } @@ -29,6 +31,7 @@ export type CodexBackgroundCommandChange = turnId: string | null backgrounded: boolean task: AgentSessionBackgroundTask + processId?: string } | { type: 'ended'; threadId: string; taskId: string } @@ -97,8 +100,15 @@ export class CodexBackgroundCommandTracker { } private liveChange(command: Command): Extract { - const { threadId, turnId, backgrounded, task } = command - return { type: 'started', threadId, turnId, backgrounded, task } + const { threadId, turnId, backgrounded, task, processId } = command + return { + type: 'started', + threadId, + turnId, + backgrounded, + task, + ...(processId !== undefined ? { processId } : {}) + } } /** The thread closed: Codex stops its processes first, so none of them can report an exit. */ @@ -130,6 +140,24 @@ export class CodexBackgroundCommandTracker { }) } + /** Whether a running command carries a process id a background-terminal stop could name. */ + get holdsProcess(): boolean { + return [...this.commands.values()].some((command) => command.processId !== undefined) + } + + /** Every live process, on every thread. */ + liveCommands(): CodexBackgroundCommandChange[] { + return [...this.commands.values()].map((command) => this.liveChange(command)) + } + + /** The processes the named tasks run: only backgrounded commands, the ones the strip offers. */ + backgroundProcesses(taskIds: readonly string[]): { threadId: string; processId: string }[] { + const named = new Set(taskIds) + return [...this.commands.values()].flatMap(({ threadId, backgrounded, task, processId }) => + backgrounded && processId !== undefined && named.has(task.id) ? [{ threadId, processId }] : [] + ) + } + /** Live processes, including foreground tools that are absent from the strip. */ threadCommands(threadId: string): CodexBackgroundCommandChange[] { return [...this.commands.values()] @@ -213,6 +241,7 @@ export class CodexBackgroundCommandTracker { .replace(/\s+/g, ' ') .trim() const turnId = readCodexTurnId(event.params) + const processId = readString(item, 'processId') ?? undefined const value = { threadId: event.threadId, turnId, @@ -224,7 +253,8 @@ export class CodexBackgroundCommandTracker { : `codex-command:thread:${encodeURIComponent(event.threadId)}:${encodeURIComponent(item.id)}`, kind: 'command' as const, ...(description ? { description } : {}) - } + }, + ...(processId !== undefined ? { processId } : {}) } return { key, diff --git a/src/main/codex/codex-background-task-tracker.ts b/src/main/codex/codex-background-task-tracker.ts index 83213242fff2..06853e36f506 100644 --- a/src/main/codex/codex-background-task-tracker.ts +++ b/src/main/codex/codex-background-task-tracker.ts @@ -30,12 +30,17 @@ export function codexChildWorkSink( } } +/** What a probe learned: `unknown` when the app-server gave no answer, so the next chance asks again. */ +export type CodexTerminalStopSupport = 'supported' | 'unsupported' | 'unknown' + /** Projects the same child execution facts the durable roster consumes. */ export class CodexBackgroundTaskTracker { private publishedFingerprint = '[]' private publishedState: AgentSessionBackgroundTaskState | null = null private readonly commands: CodexBackgroundCommandTracker private readonly childWork: CodexChildWorkEvidence + /** Whether this app-server can terminate a background command, settled by its first answer. */ + private terminalStops: CodexTerminalStopSupport | 'probing' = 'unknown' constructor( private readonly primaryThreadId: string, @@ -43,11 +48,45 @@ export class CodexBackgroundTaskTracker { private readonly childWorkSink?: CodexChildWorkSink ) { this.commands = new CodexBackgroundCommandTracker(primaryThreadId) - this.childWork = new CodexChildWorkEvidence(primaryThreadId, executions, (threadId) => - this.commands.threadCommands(threadId) + this.childWork = new CodexChildWorkEvidence( + primaryThreadId, + executions, + (threadId) => this.commands.threadCommands(threadId), + () => this.stopsTerminals ) } + get stopsTerminals(): boolean { + return this.terminalStops === 'supported' + } + + /** True when a running command has a process a stop could name and the app-server has not + * answered yet: the caller probes then, and settles the answer here. */ + beginTerminalStopProbe(): boolean { + if (this.terminalStops !== 'unknown' || !this.commands.holdsProcess) { + return false + } + this.terminalStops = 'probing' + return true + } + + /** On yes, restates every running command so its record says it can be stopped; callers publish + * the child work after. */ + settleTerminalStopProbe(support: CodexTerminalStopSupport): void { + if (this.terminalStops !== 'probing') { + return + } + this.terminalStops = support + if (support === 'supported') { + this.childWork.restateCommands(this.commands.liveCommands()) + } + } + + /** The processes behind the named tasks that a stop reaches. */ + backgroundProcesses(taskIds: readonly string[]): { threadId: string; processId: string }[] { + return this.stopsTerminals ? this.commands.backgroundProcesses(taskIds) : [] + } + get state(): AgentSessionBackgroundTaskState | null { // Journal admission precedes observe; readers must not see its pending facts. return this.publishedState diff --git a/src/main/codex/codex-background-terminals.test.ts b/src/main/codex/codex-background-terminals.test.ts new file mode 100644 index 000000000000..1fcc28855c26 --- /dev/null +++ b/src/main/codex/codex-background-terminals.test.ts @@ -0,0 +1,442 @@ +import { describe, expect, it } from 'vitest' +import type { AgentChildWorkEvidence } from '../../shared/agent-status-child-work-evidence' +import { codexPrimaryCommandTaskId } from '../../shared/structured-session-foreground-commands' +import { + CodexAppServerRequestError, + type CodexAppServerConnection +} from './codex-app-server-connection' +import { + CodexAppServerTimeoutError, + CodexAppServerUnsupportedError +} from './codex-app-server-session' +import { + probeCodexBackgroundTerminals, + terminateCodexBackgroundTerminals +} from './codex-background-terminals' +import { + adapterFor, + fakeCodex, + identityFor, + THREAD_ID, + type Route +} from './codex-structured-session-adapter-fixture' + +const TURN_ID = 'turn-1' +const PROCESS_ID = '71831' +const TASK_ID = codexPrimaryCommandTaskId('exec-1') + +type Call = { method: string; params?: Record } + +function unknownVariant(method: string): CodexAppServerRequestError { + const message = `Invalid request: unknown variant \`${method}\`` + return new CodexAppServerRequestError( + method, + -32600, + `codex app-server ${method} failed: ${message}`, + message + ) +} + +function threadNotFound(method: string): CodexAppServerRequestError { + const message = `thread not found: ${THREAD_ID}` + return new CodexAppServerRequestError( + method, + -32600, + `codex app-server ${method} failed: ${message}`, + message + ) +} + +/** Answers each request from `routes`, as a live app-server would, and records it. */ +function fakeRpc(routes: Record): { + rpc: Pick + calls: Call[] +} { + const calls: Call[] = [] + return { + calls, + rpc: { + request: async (method, params) => { + calls.push({ method, params }) + const route = routes[method] + return route ? route(params) : {} + } + } + } +} + +function codexWith(routes: Record) { + const codex = fakeCodex(routes) + const evidence: AgentChildWorkEvidence[] = [] + const adapter = adapterFor(codex, {}, [], { + onChildWorkEvidence: (_sessionId, edges) => evidence.push(...edges) + }) + const live = () => codex.connections.at(-1)! + const calls = (): Call[] => live().calls + const notify = (method: string, params: unknown): void => + live().handlers.onNotification?.(method, params) + return { adapter, calls, evidence, notify } +} + +const identity = identityFor('session-1') + +function commandItem(status: 'inProgress' | 'completed') { + return { + type: 'commandExecution', + id: 'exec-1', + processId: PROCESS_ID, + source: 'unifiedExecStartup', + command: 'pnpm dev', + status + } +} + +/** A turn that starts a dev server and ends with it still running. */ +async function backgroundedDevServer(codex: ReturnType): Promise { + await codex.adapter.acquire({ identity, fence: 7, spawnToken: 'spawn-1' }) + codex.notify('turn/started', { threadId: THREAD_ID, turn: { id: TURN_ID, status: 'inProgress' } }) + codex.notify('item/started', { + threadId: THREAD_ID, + turnId: TURN_ID, + item: commandItem('inProgress') + }) + await new Promise((resolve) => setImmediate(resolve)) + codex.notify('turn/completed', { + threadId: THREAD_ID, + turn: { id: TURN_ID, status: 'completed' } + }) +} + +function lastCommandStoppable(evidence: AgentChildWorkEvidence[]): boolean | undefined { + const live = evidence.filter( + (edge): edge is Extract => + edge.type === 'live' && edge.child.handle.id === TASK_ID + ) + return live.at(-1)?.child.stoppable +} + +describe('Codex background-command Stop', () => { + it('offers a Stop once the app-server lists background terminals, and terminates by process id', async () => { + let running = [PROCESS_ID] + const codex = codexWith({ + 'thread/backgroundTerminals/list': () => ({ + data: running.map((processId) => ({ itemId: 'exec-1', processId, command: 'pnpm dev' })), + nextCursor: null + }), + 'thread/backgroundTerminals/terminate': (params) => { + running = running.filter((processId) => processId !== params?.processId) + return { terminated: true } + } + }) + await backgroundedDevServer(codex) + + expect( + codex.calls().filter((call) => call.method === 'thread/backgroundTerminals/list') + ).toEqual([ + { method: 'thread/backgroundTerminals/list', params: { threadId: THREAD_ID, limit: 1 } } + ]) + expect(codex.adapter.backgroundTaskStops('session-1')).toEqual({ + supportsTaskStop: true, + supportsStopAll: true + }) + expect(lastCommandStoppable(codex.evidence)).toBe(true) + + await expect( + codex.adapter.stopBackgroundTasks({ sessionId: 'session-1', fence: 7, taskIds: [TASK_ID] }) + ).resolves.toEqual({ cancelled: true }) + expect(codex.calls().slice(-2)).toEqual([ + { + method: 'thread/backgroundTerminals/terminate', + params: { threadId: THREAD_ID, processId: PROCESS_ID } + }, + { method: 'thread/backgroundTerminals/list', params: { threadId: THREAD_ID } } + ]) + }) + + it('keeps an older Codex without background terminals at no Stop', async () => { + const codex = codexWith({ + 'thread/backgroundTerminals/list': () => { + throw unknownVariant('thread/backgroundTerminals/list') + } + }) + + await backgroundedDevServer(codex) + + expect(codex.adapter.backgroundTaskStops('session-1')).toEqual({ + supportsTaskStop: false, + supportsStopAll: false + }) + expect(lastCommandStoppable(codex.evidence)).toBe(false) + await expect( + codex.adapter.stopBackgroundTasks({ sessionId: 'session-1', fence: 7, taskIds: [TASK_ID] }) + ).resolves.toEqual({ cancelled: false }) + expect( + codex.calls().some((call) => call.method === 'thread/backgroundTerminals/terminate') + ).toBe(false) + }) + + it('restates an already backgrounded command as stoppable when the probe answers late', async () => { + let answer: (value: unknown) => void = () => {} + const codex = codexWith({ + 'thread/backgroundTerminals/list': () => + new Promise((resolve) => { + answer = resolve + }) + }) + await backgroundedDevServer(codex) + expect(lastCommandStoppable(codex.evidence)).toBe(false) + expect(codex.adapter.backgroundTaskStops('session-1')).toEqual({ + supportsTaskStop: false, + supportsStopAll: false + }) + + answer({ data: [], nextCursor: null }) + await new Promise((resolve) => setImmediate(resolve)) + + expect(lastCommandStoppable(codex.evidence)).toBe(true) + expect(codex.adapter.backgroundTaskStops('session-1')?.supportsTaskStop).toBe(true) + }) + + it('probes the app-server once, not per command', async () => { + const codex = codexWith({ + 'thread/backgroundTerminals/list': () => ({ data: [], nextCursor: null }) + }) + await backgroundedDevServer(codex) + codex.notify('item/started', { + threadId: THREAD_ID, + turnId: 'turn-2', + item: { ...commandItem('inProgress'), id: 'exec-2', processId: '71832' } + }) + await new Promise((resolve) => setImmediate(resolve)) + expect( + codex.calls().filter((call) => call.method === 'thread/backgroundTerminals/list') + ).toHaveLength(1) + }) + + it('asks again after a probe that timed out, instead of hiding Stop for good', async () => { + let probes = 0 + const codex = codexWith({ + 'thread/backgroundTerminals/list': () => { + probes += 1 + if (probes === 1) { + throw new CodexAppServerTimeoutError('codex app-server timed out') + } + return { data: [], nextCursor: null } + } + }) + await codex.adapter.acquire({ identity, fence: 7, spawnToken: 'spawn-1' }) + codex.notify('turn/started', { + threadId: THREAD_ID, + turn: { id: TURN_ID, status: 'inProgress' } + }) + codex.notify('item/started', { + threadId: THREAD_ID, + turnId: TURN_ID, + item: commandItem('inProgress') + }) + await new Promise((resolve) => setImmediate(resolve)) + expect(probes).toBe(1) + expect(codex.adapter.backgroundTaskStops('session-1')?.supportsTaskStop).toBe(false) + + codex.notify('turn/completed', { + threadId: THREAD_ID, + turn: { id: TURN_ID, status: 'completed' } + }) + await new Promise((resolve) => setImmediate(resolve)) + + expect(probes).toBe(2) + expect(codex.adapter.backgroundTaskStops('session-1')?.supportsTaskStop).toBe(true) + expect(lastCommandStoppable(codex.evidence)).toBe(true) + }) + + it('stops nothing for a stale fence', async () => { + const codex = codexWith({ + 'thread/backgroundTerminals/list': () => ({ data: [], nextCursor: null }) + }) + await backgroundedDevServer(codex) + await expect( + codex.adapter.stopBackgroundTasks({ sessionId: 'session-1', fence: 6, taskIds: [TASK_ID] }) + ).resolves.toEqual({ cancelled: false }) + expect( + codex.calls().some((call) => call.method === 'thread/backgroundTerminals/terminate') + ).toBe(false) + }) +}) + +describe('probeCodexBackgroundTerminals', () => { + const probe = (error: Error | null, reply: unknown = { data: [] }) => + probeCodexBackgroundTerminals( + fakeRpc({ + 'thread/backgroundTerminals/list': () => { + if (error) { + throw error + } + return reply + } + }).rpc, + THREAD_ID, + 5_000 + ) + + it('reads a listing as support and a refused method as none', async () => { + await expect(probe(null)).resolves.toBe('supported') + await expect(probe(unknownVariant('thread/backgroundTerminals/list'))).resolves.toBe( + 'unsupported' + ) + await expect( + probe(new CodexAppServerUnsupportedError('codex app-server does not support it')) + ).resolves.toBe('unsupported') + await expect(probe(null, {})).resolves.toBe('unsupported') + }) + + it('leaves the answer open when the app-server did not answer', async () => { + await expect(probe(new CodexAppServerTimeoutError('codex app-server timed out'))).resolves.toBe( + 'unknown' + ) + await expect(probe(new Error('codex app-server connection closed'))).resolves.toBe('unknown') + await expect(probe(threadNotFound('thread/backgroundTerminals/list'))).resolves.toBe('unknown') + }) +}) + +describe('terminateCodexBackgroundTerminals', () => { + const always = { timeoutMs: 5_000, isCurrent: () => true } + + it('counts a process Codex no longer holds as stopped once the list confirms it', async () => { + const { rpc } = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => ({ terminated: false }), + 'thread/backgroundTerminals/list': () => ({ data: [], nextCursor: null }) + }) + await expect( + terminateCodexBackgroundTerminals( + rpc, + [{ threadId: THREAD_ID, processId: PROCESS_ID }], + always + ) + ).resolves.toEqual({ stopped: 1, survived: false }) + }) + + it('answers a survivor the list shows, after trying the others', async () => { + const { rpc, calls } = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => ({ terminated: true }), + 'thread/backgroundTerminals/list': () => ({ + data: [{ processId: PROCESS_ID }], + nextCursor: null + }) + }) + await expect( + terminateCodexBackgroundTerminals( + rpc, + [ + { threadId: THREAD_ID, processId: PROCESS_ID }, + { threadId: THREAD_ID, processId: '71832' } + ], + always + ) + ).resolves.toEqual({ stopped: 1, survived: true }) + expect( + calls.filter((call) => call.method === 'thread/backgroundTerminals/terminate') + ).toHaveLength(2) + }) + + it('reads every page of the list', async () => { + const { rpc } = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => ({ terminated: true }), + 'thread/backgroundTerminals/list': (params) => + params?.cursor === 'page-2' + ? { data: [{ processId: PROCESS_ID }], nextCursor: null } + : { data: [{ processId: '1' }], nextCursor: 'page-2' } + }) + await expect( + terminateCodexBackgroundTerminals( + rpc, + [{ threadId: THREAD_ID, processId: PROCESS_ID }], + always + ) + ).resolves.toEqual({ stopped: 0, survived: true }) + }) + + it('answers a listed survivor even when another request timed out', async () => { + const { rpc } = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => ({ terminated: true }), + 'thread/backgroundTerminals/list': (params) => { + if (params?.threadId === 'thread-2') { + throw new CodexAppServerTimeoutError('codex app-server timed out') + } + return { data: [{ processId: PROCESS_ID }], nextCursor: null } + } + }) + await expect( + terminateCodexBackgroundTerminals( + rpc, + [ + { threadId: THREAD_ID, processId: PROCESS_ID }, + { threadId: 'thread-2', processId: '71832' } + ], + always + ) + ).resolves.toEqual({ stopped: 0, survived: true }) + }) + + it('counts a thread Codex unloaded as stopped', async () => { + const { rpc } = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => { + throw threadNotFound('thread/backgroundTerminals/terminate') + } + }) + await expect( + terminateCodexBackgroundTerminals( + rpc, + [{ threadId: THREAD_ID, processId: PROCESS_ID }], + always + ) + ).resolves.toEqual({ stopped: 1, survived: false }) + }) + + it('never counts a dropped connection as stopped', async () => { + const closed = new Error('codex app-server connection closed') + const onTerminate = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => { + throw closed + } + }) + await expect( + terminateCodexBackgroundTerminals( + onTerminate.rpc, + [{ threadId: THREAD_ID, processId: PROCESS_ID }], + always + ) + ).rejects.toBe(closed) + const onConfirm = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => ({ terminated: true }), + 'thread/backgroundTerminals/list': () => { + throw closed + } + }) + await expect( + terminateCodexBackgroundTerminals( + onConfirm.rpc, + [{ threadId: THREAD_ID, processId: PROCESS_ID }], + always + ) + ).rejects.toBe(closed) + }) + + it('stops asking an app-server that timed out', async () => { + const { rpc, calls } = fakeRpc({ + 'thread/backgroundTerminals/terminate': () => { + throw new CodexAppServerTimeoutError('codex app-server timed out') + } + }) + await expect( + terminateCodexBackgroundTerminals( + rpc, + [ + { threadId: THREAD_ID, processId: PROCESS_ID }, + { threadId: THREAD_ID, processId: '71832' } + ], + always + ) + ).rejects.toThrow('timed out') + expect(calls).toHaveLength(1) + }) +}) diff --git a/src/main/codex/codex-background-terminals.ts b/src/main/codex/codex-background-terminals.ts new file mode 100644 index 000000000000..a70205183eaa --- /dev/null +++ b/src/main/codex/codex-background-terminals.ts @@ -0,0 +1,237 @@ +// Stopping a command Codex left running, by the `processId` on its `commandExecution` item, through +// the app-server's experimental background-terminal methods (Codex 0.140+). Each chat has its own +// app-server, so one probe per app-server answers for the Codex binary on the host that runs it. The +// killed process still ends its item with `item/completed`, which removes the command's record. + +import type { CodexAppServerConnection } from './codex-app-server-connection' +import { isCodexAppServerRequestError } from './codex-app-server-request-error' +import { + CodexAppServerTimeoutError, + isCodexAppServerUnsupportedError +} from './codex-app-server-session' +import type { CodexTerminalStopSupport } from './codex-background-task-tracker' +import { readRecord } from './codex-item-field-readers' +import { + requireLiveCodexSession, + type CodexSession, + type CodexStructuredSessionAdapterDeps +} from './codex-structured-session-state' +import type { AgentSessionBackgroundTaskStops } from '../../shared/agent-child-work-stop-targets' + +type CodexTerminalRpc = Pick + +/** One command process a Stop names: the thread that ran it and Codex's id for its process. */ +export type CodexBackgroundTerminal = { threadId: string; processId: string } + +/** Why bounded: a Stop the app-server never answers must not hold the session's other actions. */ +const CODEX_BACKGROUND_TERMINAL_TIMEOUT_MS = 10_000 +const MAX_LIST_PAGES = 32 + +function boundedTimeout(timeoutMs: number | undefined): number { + return Math.min( + timeoutMs ?? CODEX_BACKGROUND_TERMINAL_TIMEOUT_MS, + CODEX_BACKGROUND_TERMINAL_TIMEOUT_MS + ) +} + +/** An older Codex refuses the method by name: unknown variant, or method not found. */ +function refusesMethod(error: unknown): boolean { + return ( + isCodexAppServerUnsupportedError(error) || + (isCodexAppServerRequestError(error) && + error.code === -32600 && + error.message.includes('unknown variant')) + ) +} + +/** Whether this app-server lists background terminals, and so can stop them. Only a refusal or + * a malformed answer is no; a timeout or dropped request proves nothing either way. */ +export async function probeCodexBackgroundTerminals( + rpc: CodexTerminalRpc, + threadId: string, + timeoutMs: number | undefined +): Promise { + try { + const reply = readRecord( + await rpc.request( + 'thread/backgroundTerminals/list', + { threadId, limit: 1 }, + { timeoutMs: boundedTimeout(timeoutMs) } + ) + ) + return Array.isArray(reply.data) ? 'supported' : 'unsupported' + } catch (error) { + return refusesMethod(error) ? 'unsupported' : 'unknown' + } +} + +/** Codex unloaded the thread, which ends its terminals. A connection that dropped is no such + * proof: losing contact is never evidence the process ended. */ +function terminalsAreGone(error: unknown): boolean { + return ( + isCodexAppServerRequestError(error) && + error.code === -32600 && + error.message.includes('failed: thread not found') + ) +} + +async function stillRunning( + rpc: CodexTerminalRpc, + threadId: string, + timeoutMs: number +): Promise> { + const running = new Set() + let cursor: string | null = null + for (let page = 0; page < MAX_LIST_PAGES; page++) { + const reply = readRecord( + await rpc.request( + 'thread/backgroundTerminals/list', + { threadId, ...(cursor === null ? {} : { cursor }) }, + { timeoutMs } + ) + ) + for (const terminal of Array.isArray(reply.data) ? reply.data : []) { + const processId = readRecord(terminal).processId + if (typeof processId === 'string') { + running.add(processId) + } + } + cursor = typeof reply.nextCursor === 'string' ? reply.nextCursor : null + if (cursor === null) { + return running + } + } + throw new Error('codex app-server listed more background terminals than Orca reads') +} + +/** Terminates each terminal, then confirms against the thread's own list. Resolves with how many + * are gone and whether the list shows one still running; throws when a request failed with no + * survivor listed, after trying every other, since the effect is then unknown. */ +export async function terminateCodexBackgroundTerminals( + rpc: CodexTerminalRpc, + terminals: readonly CodexBackgroundTerminal[], + options: { timeoutMs: number | undefined; isCurrent: () => boolean } +): Promise<{ stopped: number; survived: boolean }> { + const timeoutMs = boundedTimeout(options.timeoutMs) + let failure: { error: unknown } | undefined + // An app-server that is not answering would make every later request wait out its own deadline. + const fail = (error: unknown): boolean => { + failure ??= { error } + return error instanceof CodexAppServerTimeoutError + } + const unconfirmed = new Map>() + let stopped = 0 + let survived = false + let timedOut = false + for (const { threadId, processId } of terminals) { + if (timedOut || !options.isCurrent()) { + break + } + try { + await rpc.request( + 'thread/backgroundTerminals/terminate', + { threadId, processId }, + { timeoutMs } + ) + // Answered either way: `terminated: false` is a process Codex no longer holds, which the + // list below tells apart from one that survived. + unconfirmed.set(threadId, (unconfirmed.get(threadId) ?? new Set()).add(processId)) + } catch (error) { + if (terminalsAreGone(error)) { + stopped += 1 + } else { + timedOut = fail(error) + } + } + } + for (const [threadId, processIds] of unconfirmed) { + if (timedOut) { + break + } + try { + const running = await stillRunning(rpc, threadId, timeoutMs) + for (const processId of processIds) { + if (running.has(processId)) { + survived = true + } else { + stopped += 1 + } + } + } catch (error) { + if (terminalsAreGone(error)) { + stopped += processIds.size + } else { + timedOut = fail(error) + } + } + } + if (failure && !survived) { + throw failure.error + } + return { stopped, survived } +} + +/** A backgrounded command stops through its app-server's background terminals, once that + * app-server proved it has them; a child thread has no stop here. */ +export function codexBackgroundTaskStops( + session: CodexSession | undefined +): AgentSessionBackgroundTaskStops | undefined { + const stops = session?.backgroundTasks.stopsTerminals + return stops === undefined ? undefined : { supportsTaskStop: stops, supportsStopAll: stops } +} + +/** Probes the session's app-server off the frame path when a running command has a process a stop + * could name, until it answers; a yes restates the running commands as stoppable. */ +export function startCodexTerminalStopProbe( + sessions: ReadonlyMap, + sessionId: string, + session: CodexSession, + deps: Pick +): void { + if (!session.backgroundTasks.beginTerminalStopProbe()) { + return + } + void probeCodexBackgroundTerminals( + session.connection, + session.threadId, + deps.requestTimeoutMs + ).then((support) => { + if (sessions.get(sessionId) !== session || session.ended) { + return + } + try { + session.backgroundTasks.settleTerminalStopProbe(support) + session.backgroundTasks.publishChildWork() + } catch (error) { + deps.logger?.warn('Codex background-terminal stops could not be published', { + scope: 'codex-background-terminals', + sessionId, + error + }) + } + }) +} + +/** Stops the backgrounded commands the named tasks run, while `session` is still the one the host + * asked about. Cancelled when at least one is gone. */ +export async function stopCodexBackgroundCommands( + sessions: Map, + input: { sessionId: string; fence: number; taskIds: readonly string[] }, + timeoutMs: number | undefined +): Promise<{ cancelled: boolean; stillRunning?: true }> { + const session = requireLiveCodexSession(sessions, input.sessionId) + const terminals = session.backgroundTasks.backgroundProcesses(input.taskIds) + if (terminals.length === 0) { + return { cancelled: false } + } + const { acquisitionGeneration } = session + const { stopped, survived } = await terminateCodexBackgroundTerminals(session.connection, terminals, { + timeoutMs, + isCurrent: () => + sessions.get(input.sessionId) === session && + !session.ended && + session.fence === input.fence && + session.acquisitionGeneration === acquisitionGeneration + }) + return { cancelled: stopped > 0, ...(survived ? { stillRunning: true as const } : {}) } +} diff --git a/src/main/codex/codex-child-work-evidence.ts b/src/main/codex/codex-child-work-evidence.ts index 3b6d146b2e58..a55c7a5b3fc7 100644 --- a/src/main/codex/codex-child-work-evidence.ts +++ b/src/main/codex/codex-child-work-evidence.ts @@ -66,7 +66,8 @@ function ofTurn(fact: T | undefined, turnId function commandLive( command: Extract, - ownerId: string | null + ownerId: string | null, + stopsTerminals: boolean ) { const { task } = command return (observedAt: number): AgentChildWorkEvidence => ({ @@ -79,7 +80,8 @@ function commandLive( state: 'working', ...(task.description ? { description: task.description } : {}), ...(ownerId !== null ? { ownerId } : {}), - stoppable: false + // As the strip offers it: a backgrounded process this app-server can terminate. + stoppable: stopsTerminals && command.backgrounded && command.processId !== undefined } }) } @@ -91,9 +93,16 @@ export class CodexChildWorkEvidence { constructor( private readonly primaryThreadId: string, private readonly executions: CodexSubagentExecutions, - private readonly liveCommands: (threadId: string) => readonly CodexBackgroundCommandChange[] + private readonly liveCommands: (threadId: string) => readonly CodexBackgroundCommandChange[], + /** Whether this app-server proved it can terminate a background command. */ + private readonly stopsTerminals: () => boolean = () => false ) {} + /** Restates these commands, as when whether they can be stopped changed. */ + restateCommands(commands: readonly CodexBackgroundCommandChange[]): void { + this.queueCommands(commands) + } + /** After the tracker applied the frame: which command processes it saw start or stop, and the * child the frame is about. */ observe( @@ -144,7 +153,7 @@ export class CodexChildWorkEvidence { for (const command of commands) { if (command.type === 'started') { const ownerId = command.threadId === this.primaryThreadId ? null : command.threadId - this.pending.push(commandLive(command, ownerId)) + this.pending.push(commandLive(command, ownerId, this.stopsTerminals())) continue } const { taskId } = command diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index a5fbf4628a2d..73c3ad14a09b 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -40,6 +40,11 @@ import { import { createCodexStructuredNotificationRetry } from './codex-structured-notification-retry' import { acquireCodexStructuredSession } from './codex-structured-session-acquire' import { changeCodexThreadGoal } from './codex-structured-thread-goal' +import { + codexBackgroundTaskStops, + startCodexTerminalStopProbe, + stopCodexBackgroundCommands +} from './codex-background-terminals' import { answerCodexStructuredPrompt, cancelCodexStructuredTurn @@ -142,6 +147,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap session.backgroundTasks.observe(event, session.prompts.takeAbandonedCommands()) // After the journal and the parent's republished row, never ahead of either. session.backgroundTasks.publishChildWork() + startCodexTerminalStopProbe(this.sessions, event.sessionId, session, this.deps) } this.deps.onEvent?.(event) return admission @@ -182,11 +188,13 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap ) } - // Codex exposes no honest stop for a child thread or a persistent command. backgroundTaskStops: NonNullable = ( sessionId - ) => - this.sessions.has(sessionId) ? { supportsTaskStop: false, supportsStopAll: false } : undefined + ) => codexBackgroundTaskStops(this.sessions.get(sessionId)) + + stopBackgroundTasks: NonNullable = ( + input + ) => stopCodexBackgroundCommands(this.sessions, input, this.deps.requestTimeoutMs) bindPromptItemId = ( sessionId: string, 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 98aab1b6ba28..c449383d50f4 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 @@ -379,12 +379,14 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & * start a new goal rather than rewrite that one's objective in place. */ replacesGoal: boolean }): Promise<{ ok: true } | { ok: false; rejected: string }> - /** Stops exactly the tasks `taskIds` names, which the host resolves from its child records. */ + /** Stops exactly the tasks `taskIds` names, which the host resolves from its child records. + * `stillRunning` is the provider's own answer that one survived its stop; a throw leaves the + * effect unknown. */ stopBackgroundTasks?(input: { sessionId: string fence: number taskIds: readonly string[] - }): Promise<{ cancelled: boolean }> + }): Promise<{ cancelled: boolean; stillRunning?: true }> /** The stops this provider honours for a live session's background work; undefined when the * adapter holds no live session for it. */ backgroundTaskStops?(sessionId: string): AgentSessionBackgroundTaskStops | undefined diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts index 19fac5d20b8c..a588fa7168af 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts @@ -7,6 +7,7 @@ import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import type { AgentSessionCancelResult } from '../../../shared/agent-session-wire' +import { refuseUnclassified } from '../../../shared/agent-session-wire-refusals' import { latestJournalDispatchObservation } from '../agent-session-journal/journal-dispatch-observation' import type { AgentSessionCancelOutcome } from './structured-agent-session-adapter' import { @@ -194,19 +195,29 @@ async function cancelAndNote( let stoppedTurn: string | undefined try { const dispatchStatus = latestJournalDispatchObservation(ctx.journal, ctx.fence) + const background = + input.scope && !stoppedBefore + ? await ctx.adapter.stopBackgroundTasks?.({ + sessionId: ctx.sessionId, + fence: ctx.fence, + taskIds: agentChildWorkStopTargets(input.childWork?.(), input.taskId) + }) + : undefined + // A background Stop writes no row, so a task the agent shows still running after it goes back + // to the client that pressed it. + if (background?.stillRunning) { + return { + ok: false, + refusal: refuseUnclassified( + 'agent_session_operation_invalid', + 'a background task still runs after its stop' + ) + } + } const outcome: AgentSessionCancelOutcome = stoppedBefore ? { cancelled: false } : input.scope - ? { - cancelled: - ( - await ctx.adapter.stopBackgroundTasks?.({ - sessionId: ctx.sessionId, - fence: ctx.fence, - taskIds: agentChildWorkStopTargets(input.childWork?.(), input.taskId) - }) - )?.cancelled === true - } + ? { cancelled: background?.cancelled === true } : await ctx.adapter.cancelTurn({ sessionId: ctx.sessionId, ...(input.turnId !== undefined ? { turnId: input.turnId } : {}), @@ -238,6 +249,16 @@ async function cancelAndNote( if (input.prompt) { throw error } + // A timeout or lost contact proves nothing about the task: unconfirmed, never "not stopped". + if (input.scope) { + return { + ok: false, + refusal: refuseUnclassified( + 'agent_session_operation_unknown', + `background task stop unconfirmed: ${error instanceof Error ? error.message : String(error)}` + ) + } + } interruptFailed = true // The adapter's error is Orca's; the row says only that the stop is unconfirmed. note = { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts index b79191c9ac39..3fe9542e7b68 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts @@ -267,6 +267,66 @@ describe('performCancel', () => { expect(cancelTurn).not.toHaveBeenCalled() expect(journal.snapshot().items).toEqual([]) }) + + async function backgroundStopWith( + stopBackgroundTasks: StructuredAgentSessionAdapter['stopBackgroundTasks'], + operation: string + ) { + root = await mkdtemp(join(tmpdir(), 'orca-background-task-failed-cancel-')) + const journal = await journals.open({ identity: IDENTITY, stateDirectory: root }) + const cancelTurn = vi.fn(async () => ({ cancelled: true })) + const ctx: AgentSessionTurnContext = { + logger: createStructuredAgentSessionLogger(), + sessionId: 'session-1', + journal, + fence: 1, + agents: NO_STRUCTURED_AGENTS, + agent: 'codex', + adapter: { cancelTurn, stopBackgroundTasks } as unknown as StructuredAgentSessionAdapter, + persistOptions: async () => undefined, + resolvedBy: 'client-1', + publish: vi.fn(), + now: () => 1 + } + const result = await performCancel(ctx, { + clientOperationId: operation, + turnId: 'background-tasks', + scope: 'background-tasks', + taskId: 'task-1', + childWork: () => [liveTask('task-1')] + }) + expect(cancelTurn).not.toHaveBeenCalled() + expect(journal.snapshot().items).toEqual([]) + return result + } + + it('refuses a background Stop the agent shows still running, so the person hears it failed', async () => { + const result = await backgroundStopWith( + async () => ({ cancelled: false, stillRunning: true }), + 'cancel-background-task-survived' + ) + + expect(result).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid' } + }) + expect(result.ok ? undefined : result.refusal.details).toBeUndefined() + }) + + it.each([ + ['timed out', 'codex app-server request timed out'], + ['lost the agent', 'codex app-server connection closed'] + ])('answers a background Stop that %s as unconfirmed, never not stopped', async (_, message) => { + const result = await backgroundStopWith(async () => { + throw new Error(message) + }, 'cancel-background-task-unconfirmed') + + expect(result).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(result.ok ? undefined : result.refusal.details).toBeUndefined() + }) }) /** A live child record the strip would offer a stop, named by the provider as `providerId`. */ diff --git a/src/main/runtime/structured-foreground-command-strip.test.ts b/src/main/runtime/structured-foreground-command-strip.test.ts index f21054ef1b64..50c0c9417727 100644 --- a/src/main/runtime/structured-foreground-command-strip.test.ts +++ b/src/main/runtime/structured-foreground-command-strip.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire' +import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' import { AgentHookServer, _internals } from '../agent-hooks/server' import { fakeCodex, THREAD_ID } from '../codex/codex-structured-session-adapter-fixture' import { @@ -33,8 +34,9 @@ describe('foreground commands in the background-task channel', () => { await rm(root, { recursive: true, force: true }) }) - async function rig() { + async function rig(routes: Parameters[0] = {}) { const codex = fakeCodex({ + ...routes, 'model/list': () => ({ data: [ { @@ -94,19 +96,47 @@ describe('foreground commands in the background-task channel', () => { } const turn = (method: string, id: string) => notify(method, { threadId: THREAD_ID, turn: { id, status: 'completed' } }) - const command = (method: string, id: string, turnId = 'turn-1') => + const command = (method: string, id: string, turnId = 'turn-1', processId?: string) => notify(method, { threadId: THREAD_ID, turnId, item: { type: 'commandExecution', id, + ...(processId ? { processId } : {}), command: id === 'pwd' ? 'pwd' : 'git status --short --branch', source: 'unifiedExecStartup', status: method === 'item/completed' ? 'completed' : 'inProgress' } }) - return { host, rosters, turn, command } + return { host, rosters, turn, command, fence: attached.ok ? attached.value.fence : null } + } + + type StripHost = Awaited>['host'] + + /** The strip's Stop on the `server` command, as a client sends it. */ + function stopFromStrip(host: StripHost, fence: number | null, operation: string) { + const fields = { + turnId: 'background-tasks', + scope: 'background-tasks' as const, + taskId: 'codex-command:primary:server' + } + return host.cancel( + { callerKey: 'strip-test' }, + { + envelope: { + sessionId: SESSION, + clientOperationId: `${Date.now()}-${operation.padStart(32, '0')}`, + expectedRuntimeFence: fence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.cancel', + sessionId: SESSION, + fields + }) + }, + ...fields + } + ) } it('never publishes a strip between a quick command starting and completing', async () => { @@ -144,4 +174,76 @@ describe('foreground commands in the background-task channel', () => { await command('item/completed', 'server') expect(rosters.at(-1)).toBeNull() }) + + it('stops a surviving command from the strip through Codex background terminals', async () => { + let running = ['4242'] + const { host, rosters, turn, command, fence } = await rig({ + 'thread/backgroundTerminals/list': () => ({ + data: running.map((processId) => ({ processId })), + nextCursor: null + }), + 'thread/backgroundTerminals/terminate': (params) => { + running = running.filter((processId) => processId !== params?.processId) + return { terminated: true } + } + }) + await turn('turn/started', 'turn-1') + await command('item/started', 'server', 'turn-1', '4242') + await turn('turn/completed', 'turn-1') + expect(rosters.at(-1)).toMatchObject({ + supportsTaskStop: true, + children: [{ providerId: 'codex-command:primary:server', stoppable: true }] + }) + + const stopped = await stopFromStrip(host, fence, '2') + expect(stopped).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(running).toEqual([]) + // The killed process's own end is what takes its row off the strip. + await command('item/completed', 'server', 'turn-1', '4242') + expect(rosters.at(-1)).toBeNull() + }) + + it('refuses a Stop whose command still runs, and keeps the row stoppable', async () => { + const { host, rosters, turn, command, fence } = await rig({ + 'thread/backgroundTerminals/list': () => ({ + data: [{ processId: '4242' }], + nextCursor: null + }), + 'thread/backgroundTerminals/terminate': () => ({ terminated: false }) + }) + await turn('turn/started', 'turn-1') + await command('item/started', 'server', 'turn-1', '4242') + await turn('turn/completed', 'turn-1') + + await expect(stopFromStrip(host, fence, '3')).resolves.toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid' } + }) + expect(rosters.at(-1)).toMatchObject({ + children: [{ providerId: 'codex-command:primary:server', stoppable: true }] + }) + }) + + it('answers a Stop that lost contact with Codex as unconfirmed, and keeps the row stoppable', async () => { + const { host, rosters, turn, command, fence } = await rig({ + 'thread/backgroundTerminals/list': () => ({ + data: [{ processId: '4242' }], + nextCursor: null + }), + 'thread/backgroundTerminals/terminate': () => { + throw new Error('codex app-server connection closed') + } + }) + await turn('turn/started', 'turn-1') + await command('item/started', 'server', 'turn-1', '4242') + await turn('turn/completed', 'turn-1') + + await expect(stopFromStrip(host, fence, '4')).resolves.toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(rosters.at(-1)).toMatchObject({ + children: [{ providerId: 'codex-command:primary:server', stoppable: true }] + }) + }) }) diff --git a/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx b/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx index 5d32aacf2b3b..9ad245fde3e6 100644 --- a/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx @@ -6,6 +6,7 @@ import { act, cleanup, fireEvent, render, screen, within } from '@testing-librar import { Profiler, useState } from 'react' import { afterEach, describe, expect, it, vi } from 'vitest' import type { AgentSessionBackgroundTask } from '../../../../shared/agent-session-wire' +import type { AgentChildWorkView } from '../../../../shared/agent-status-child-work-view' import { NativeChatBackgroundTasksStatus } from './NativeChatBackgroundTasksStatus' afterEach(() => { @@ -96,8 +97,55 @@ describe('NativeChatBackgroundTasksStatus stop affordances', () => { expect(screen.getByLabelText('Stop backgrounded subagent')).toBeInTheDocument() }) + it('stops a backgrounded Codex command by its own id, and offers none on the subagent beside it', () => { + const view = (overrides: Partial): AgentChildWorkView => ({ + id: 'child', + kind: 'agent', + state: 'working', + membership: 'live', + firstObservedAt: Date.now() - 65_000, + observedAt: Date.now() - 1_000, + stoppable: true, + invocation: { invocationId: 'spawn-1', generation: 1 }, + ...overrides + }) + const onStop = vi.fn() + render( + + ) + fireEvent.click(screen.getByRole('button', { expanded: false })) + + expect(screen.queryByLabelText('Stop count_a')).not.toBeInTheDocument() + fireEvent.click(screen.getByLabelText('Stop pnpm dev')) + expect(onStop).toHaveBeenCalledWith('codex-command:exec-1') + }) + it('offers no stop at all when the provider exposes none', () => { - // Codex: a Stop button here would be a control that cannot act. + // An older Codex: a Stop button here would be a control that cannot act. renderStrip({ supportsTaskStop: false, supportsStopAll: false }) expect(screen.queryByLabelText('Stop background tasks')).not.toBeInTheDocument() expect(screen.queryByLabelText('Stop count_a')).not.toBeInTheDocument() diff --git a/src/renderer/src/components/native-chat/NativeChatStructuredSession.test.tsx b/src/renderer/src/components/native-chat/NativeChatStructuredSession.test.tsx index 5fd922be40b5..41741aea8b22 100644 --- a/src/renderer/src/components/native-chat/NativeChatStructuredSession.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatStructuredSession.test.tsx @@ -552,7 +552,7 @@ describe('NativeChatStructuredSession', () => { expect(screen.getByText('watcher')).toBeTruthy() }) - it('tracks concurrent task stops independently and clears each pending result', async () => { + it('tracks concurrent task stops independently: a confirmed one holds its row, a failed one clears', async () => { mocks.monitoringBackgroundTasks = true mocks.supportsBackgroundTaskStop = true mocks.backgroundTasks = [ @@ -572,7 +572,7 @@ describe('NativeChatStructuredSession', () => { }) ) - render( + const { rerender } = render( claudeSessionView('structured-tab-concurrent-background', 'session-concurrent-background') ) fireEvent.click(screen.getByRole('button', { name: '2 shells — 2 working' })) @@ -584,12 +584,30 @@ describe('NativeChatStructuredSession', () => { expect((firstStop as HTMLButtonElement).disabled).toBe(true) expect((secondStop as HTMLButtonElement).disabled).toBe(true) + // Confirmed before its row leaves: the button stays in its stopping state, never Stop again. await act(async () => finishFirst({ cancelled: true })) - await waitFor(() => expect((firstStop as HTMLButtonElement).disabled).toBe(false)) + expect((firstStop as HTMLButtonElement).disabled).toBe(true) expect((secondStop as HTMLButtonElement).disabled).toBe(true) await act(async () => finishSecond(null)) await waitFor(() => expect((secondStop as HTMLButtonElement).disabled).toBe(false)) + expect((firstStop as HTMLButtonElement).disabled).toBe(true) + + mocks.backgroundTasks = [{ id: 'task-two', kind: 'command', description: 'Second task' }] + rerender( + claudeSessionView('structured-tab-concurrent-background', 'session-concurrent-background') + ) + expect(screen.queryByRole('button', { name: 'Stop First task' })).toBeNull() + // The same task listed again is one the Stop did not end. + mocks.backgroundTasks = [ + { id: 'task-one', kind: 'command', description: 'First task' }, + { id: 'task-two', kind: 'command', description: 'Second task' } + ] + rerender( + claudeSessionView('structured-tab-concurrent-background', 'session-concurrent-background') + ) + const firstAgain = screen.getByRole('button', { name: 'Stop First task' }) + expect(firstAgain.hasAttribute('disabled')).toBe(false) }) it('keeps a stale session stop result from clearing the current session pending state', async () => { @@ -622,7 +640,7 @@ describe('NativeChatStructuredSession', () => { await act(async () => finishOld({ cancelled: true })) expect((currentStop as HTMLButtonElement).disabled).toBe(true) - await act(async () => finishCurrent({ cancelled: true })) + await act(async () => finishCurrent({ cancelled: false })) await waitFor(() => expect((currentStop as HTMLButtonElement).disabled).toBe(false)) }) diff --git a/src/renderer/src/components/native-chat/NativeChatStructuredSessionStatus.tsx b/src/renderer/src/components/native-chat/NativeChatStructuredSessionStatus.tsx index 7705c4cd4bd2..e5fb813cb445 100644 --- a/src/renderer/src/components/native-chat/NativeChatStructuredSessionStatus.tsx +++ b/src/renderer/src/components/native-chat/NativeChatStructuredSessionStatus.tsx @@ -1,6 +1,10 @@ import { useState } from 'react' import { NativeChatBackgroundTasksStatus } from './NativeChatBackgroundTasksStatus' -import type { StructuredSessionBackgroundTasksView } from '../../../../shared/structured-session-background-tasks-view' +import type { AgentSessionCancelResult } from '../../../../shared/agent-session-wire' +import { + structuredSessionConfirmedStopsStillListed, + type StructuredSessionBackgroundTasksView +} from '../../../../shared/structured-session-background-tasks-view' import { useStructuredSessionChildRowContext } from './use-structured-session-child-row-context' type StoppingBackgroundTasks = { @@ -9,6 +13,8 @@ type StoppingBackgroundTasks = { all: boolean } +type ConfirmedStops = { sessionId: string; taskIds: ReadonlySet } + const NO_STOPPING_TASKS: ReadonlySet = new Set() export function NativeChatStructuredSessionStatus(props: { @@ -17,11 +23,26 @@ export function NativeChatStructuredSessionStatus(props: { paneKey: string isVisible: boolean backgroundTasks: StructuredSessionBackgroundTasksView - stopBackgroundTask: (taskId?: string) => Promise + stopBackgroundTask: (taskId?: string) => Promise }): React.JSX.Element { const [stopping, setStopping] = useState(null) + // Stops the host confirmed: each row keeps its stopping button until it leaves the strip. + const [confirmed, setConfirmed] = useState(null) const [expanded, setExpanded] = useState<{ sessionId: string; expanded: boolean } | null>(null) const activeStopping = stopping?.sessionId === props.sessionId ? stopping : null + const activeConfirmed = confirmed?.sessionId === props.sessionId ? confirmed.taskIds : null + const confirmedListed = activeConfirmed + ? structuredSessionConfirmedStopsStillListed(props.backgroundTasks, activeConfirmed) + : null + if (confirmedListed !== activeConfirmed) { + // A row that left and came back is a task the Stop did not end, so it offers Stop again. + setConfirmed( + confirmedListed?.size ? { sessionId: props.sessionId, taskIds: confirmedListed } : null + ) + } + const stoppingTaskIds = confirmedListed?.size + ? new Set([...(activeStopping?.taskIds ?? NO_STOPPING_TASKS), ...confirmedListed]) + : (activeStopping?.taskIds ?? NO_STOPPING_TASKS) const childRowContext = useStructuredSessionChildRowContext(props.paneKey) const onStop = (taskId?: string) => { @@ -39,19 +60,32 @@ export function NativeChatStructuredSessionStatus(props: { all: taskId ? current?.sessionId === sessionId && current.all : true } }) - void props.stopBackgroundTask(taskId).finally(() => { - setStopping((current) => { - if (current?.sessionId !== sessionId) { - return current - } - const taskIds = new Set(current.taskIds) - if (taskId) { - taskIds.delete(taskId) + void props + .stopBackgroundTask(taskId) + .then((result) => { + if (taskId && result?.cancelled) { + setConfirmed((current) => ({ + sessionId, + taskIds: new Set([ + ...(current?.sessionId === sessionId ? current.taskIds : NO_STOPPING_TASKS), + taskId + ]) + })) } - const all = taskId ? current.all : false - return taskIds.size === 0 && !all ? null : { sessionId, taskIds, all } }) - }) + .finally(() => { + setStopping((current) => { + if (current?.sessionId !== sessionId) { + return current + } + const taskIds = new Set(current.taskIds) + if (taskId) { + taskIds.delete(taskId) + } + const all = taskId ? current.all : false + return taskIds.size === 0 && !all ? null : { sessionId, taskIds, all } + }) + }) } return ( @@ -67,7 +101,7 @@ export function NativeChatStructuredSessionStatus(props: { indicatorActive={props.backgroundTasks.isMonitoring} supportsTaskStop={props.backgroundTasks.supportsStop} supportsStopAll={props.backgroundTasks.supportsStopAll} - stoppingTaskIds={activeStopping?.taskIds ?? NO_STOPPING_TASKS} + stoppingTaskIds={stoppingTaskIds} stoppingAll={activeStopping?.all ?? false} expanded={expanded?.sessionId === props.sessionId && expanded.expanded} onExpandedChange={(value) => setExpanded({ sessionId: props.sessionId, expanded: value })} diff --git a/src/renderer/src/components/native-chat/structured-child-records-switch.test.tsx b/src/renderer/src/components/native-chat/structured-child-records-switch.test.tsx index b262e2a9ecfb..ad5fb2080996 100644 --- a/src/renderer/src/components/native-chat/structured-child-records-switch.test.tsx +++ b/src/renderer/src/components/native-chat/structured-child-records-switch.test.tsx @@ -201,7 +201,7 @@ function Surfaces(props: { roster: AgentSessionBackgroundTaskState | null }): Re paneKey={PANE_KEY} isVisible backgroundTasks={structuredSessionBackgroundTasksView(props.roster, null)} - stopBackgroundTask={async () => undefined} + stopBackgroundTask={async () => null} /> ) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session.ts b/src/renderer/src/components/native-chat/use-structured-agent-session.ts index 5c5f4615057e..ef5397cc79a1 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session.ts @@ -3,6 +3,7 @@ import { useStructuredAgentSessionSends } from './use-structured-agent-session-s import { useStructuredAgentSessionCommandWrite } from './use-structured-agent-session-command-write' import { structuredAgentSessionNewSendsQueue } from './structured-agent-session-queue-request' import type { AgentType } from '../../../../shared/agent-status-types' +import type { AgentSessionCancelResult } from '../../../../shared/agent-session-wire' import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client' import { structuredAgentLabel } from '@/lib/structured-agent-session-launch-label' import { takeBackStructuredLaunchPrompts } from '@/lib/structured-agent-session-launch-prompt' @@ -298,7 +299,7 @@ export function useStructuredAgentSession(args: { }) }, stopBackgroundTask: (taskId?: string) => - mutate('agentSession.cancel', 'agentSession.cancel', { + mutate('agentSession.cancel', 'agentSession.cancel', { turnId: 'background-tasks', scope: 'background-tasks', ...(taskId ? { taskId } : {}) diff --git a/src/shared/structured-session-background-tasks-view.ts b/src/shared/structured-session-background-tasks-view.ts index 9786239f6452..66f27129707b 100644 --- a/src/shared/structured-session-background-tasks-view.ts +++ b/src/shared/structured-session-background-tasks-view.ts @@ -76,3 +76,23 @@ export function structuredSessionBackgroundTasksView( supportsStopAll: backgroundTasks?.supportsStopAll !== false } } + +/** Whether the strip still lists the task a Stop named, live or settled. */ +function listsTask(view: StructuredSessionBackgroundTasksView, providerId: string): boolean { + return ( + view.show && + (view.children + ? view.children.some((child) => child.providerId === providerId) + : view.tasks.some((task) => task.id === providerId)) + ) +} + +/** The host-confirmed Stops whose rows the strip still lists, each holding its row's button; the + * same set when none has left, so a caller can tell when to drop the rest. */ +export function structuredSessionConfirmedStopsStillListed( + view: StructuredSessionBackgroundTasksView, + confirmed: ReadonlySet +): ReadonlySet { + const listed = [...confirmed].filter((providerId) => listsTask(view, providerId)) + return listed.length === confirmed.size ? confirmed : new Set(listed) +}