diff --git a/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx b/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx index 332aa29512b7..a71179df178c 100644 --- a/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx +++ b/mobile/src/session/MobileNativeChatBackgroundTasks.test.tsx @@ -376,6 +376,37 @@ describe('MobileNativeChatBackgroundTasks', () => { expect(stop).toHaveBeenCalledWith('codex-command:exec-1') }) + it('stops a Grok background command and the Grok subagent beside it by their own ids', () => { + const stop = vi.fn(async () => REJECTED) + const mounted = mount( + tasksFor( + { + state: 'monitoring', + supportsTaskStop: true, + supportsStopAll: false, + children: [ + view('agent', { providerId: 'subagent-1', description: 'Review the diff' }), + view('cmd', { + kind: 'command', + providerId: 'acp-task:01a10366', + description: 'Run the dev server' + }) + ] + }, + { stop } + ) + ) + expand(mounted) + const buttons = stopButtons(mounted) + const labels = buttons.map((button) => button.props.accessibilityLabel) + expect(labels).toEqual( + expect.arrayContaining(['Stop Run the dev server', 'Stop Review the diff']) + ) + expect(labels).not.toContain('Stop background tasks') + act(() => buttons[labels.indexOf('Stop Run the dev server')]!.props.onPress()) + expect(stop).toHaveBeenCalledWith('acp-task:01a10366') + }) + it('offers Stop all only to a host with no per-row stop that still accepts one', async () => { let finish: (result: StopResult) => void = () => {} const stop = vi.fn(() => new Promise((resolve) => (finish = resolve))) diff --git a/src/main/acp/acp-background-task-child-work.ts b/src/main/acp/acp-background-task-child-work.ts new file mode 100644 index 000000000000..ec44fc57c134 --- /dev/null +++ b/src/main/acp/acp-background-task-child-work.ts @@ -0,0 +1,63 @@ +import type { + AgentChildWorkEvidence, + AgentChildWorkEvidenceHandle +} from '../../shared/agent-status-child-work-evidence' +import { isSettledBackgroundTaskState } from '../../shared/native-chat-background-task-row' +import { isBackgroundTaskBlock } from '../../shared/native-chat-types' +import { AGENT_CHILD_WORK_ID_MAX_LENGTH } from '../../shared/agent-status-child-work-value-guards' +import type { AcpTimelineEvent } from './acp-timeline-event' + +/** Keeps a task's handle apart from a subagent's, so a Stop knows which route it takes. */ +const TASK_ID_PREFIX = 'acp-task:' + +function acpBackgroundTaskHandle(taskId: string): AgentChildWorkEvidenceHandle { + return { idKind: 'task_id', id: `${TASK_ID_PREFIX}${taskId}` } +} + +/** The agent's own task id behind a child handle, when the handle names a background task. */ +export function acpBackgroundTaskStopId(childId: string): string | undefined { + return childId.startsWith(TASK_ID_PREFIX) ? childId.slice(TASK_ID_PREFIX.length) : undefined +} + +function backgroundTaskBlocks(event: AcpTimelineEvent) { + return event.type === 'item.update' && 'body' in event && event.body.kind === 'message' + ? event.body.blocks.filter(isBackgroundTaskBlock) + : [] +} + +export function isAcpBackgroundTaskChildWorkEvent(event: AcpTimelineEvent): boolean { + return backgroundTaskBlocks(event).length > 0 +} + +/** A running task is a live record until it ends, and then its record goes, like a command's. */ +export function acpBackgroundTaskChildWork( + event: AcpTimelineEvent, + observedAt: number, + stoppable: boolean +): AgentChildWorkEvidence[] { + return backgroundTaskBlocks(event).flatMap((block): AgentChildWorkEvidence[] => { + const handle = acpBackgroundTaskHandle(block.taskId) + // A handle too long to record could not address the agent's own task either. + if (handle.id.length > AGENT_CHILD_WORK_ID_MAX_LENGTH) { + return [] + } + if (isSettledBackgroundTaskState(block.state)) { + return [{ type: 'removed', observedAt, handle }] + } + return [ + { + type: 'live', + observedAt, + child: { + handle, + kind: block.kind, + residency: 'background', + state: + block.state === 'monitoring' || block.state === 'waiting' ? block.state : 'working', + ...(block.label ? { description: block.label } : {}), + stoppable + } + } + ] + }) +} diff --git a/src/main/acp/acp-dialects/acp-dialect.ts b/src/main/acp/acp-dialects/acp-dialect.ts index 2657f10d1b08..e0bd89d70e04 100644 --- a/src/main/acp/acp-dialects/acp-dialect.ts +++ b/src/main/acp/acp-dialects/acp-dialect.ts @@ -71,6 +71,15 @@ export type AcpDialect = { request(sessionId: string, id: string): { method: string; params: unknown } response(value: unknown, id: string): { cancelled: boolean; state?: NativeChatSubagentState } } + backgroundTaskStop?: { + /** Missing required target: proves the route exists without addressing any task. */ + probe: { method: string; params: unknown } + /** The agent refusing the route by name: the only answer that hides Stop. */ + lacksRoute(error: AcpAgentError): boolean + request(sessionId: string, taskId: string): { method: string; params: unknown } + /** `gone`: the agent no longer runs a task by that id; its own ending may still follow. */ + response(value: unknown, taskId: string): 'killed' | 'gone' | 'refused' + } injectedPromptIdentity?: true /** A tool update in the shared shape (`rawOutput.stdout`, `rawOutput.exitCode`), read first. */ normalizeToolUpdate?(update: ToolCallUpdate): ToolCallUpdate @@ -98,4 +107,7 @@ export type AcpDialect = { compactionReply?(text: string): AcpCompactionReply | undefined } +/** Which of the dialect's stop routes this agent process proved it answers. */ +export type AcpChildStops = { subagents: boolean; backgroundTasks: boolean } + export const GENERIC_ACP_DIALECT: AcpDialect = {} diff --git a/src/main/acp/acp-dialects/grok-background-task-stop.ts b/src/main/acp/acp-dialects/grok-background-task-stop.ts new file mode 100644 index 000000000000..fe2bcf523e96 --- /dev/null +++ b/src/main/acp/acp-dialects/grok-background-task-stop.ts @@ -0,0 +1,29 @@ +import { z } from 'zod' +import type { AcpDialect } from './acp-dialect' + +const response = z.union([ + z.object({ + error: z.null().optional(), + result: z.object({ + taskId: z.string(), + outcome: z.enum(['killed', 'already_exited', 'not_found']) + }) + }), + // Grok's envelope for a kill it could not attempt, such as an unknown session. + z.object({ result: z.null(), error: z.string() }) +]) + +export const grokBackgroundTaskStop: NonNullable = { + probe: { method: '_x.ai/task/kill', params: {} }, + lacksRoute: (error) => error.code === -32601, + // `source` is left to Grok's default: its own per-task UI kill. + request: (sessionId, taskId) => ({ method: '_x.ai/task/kill', params: { sessionId, taskId } }), + response: (value, taskId) => { + const parsed = response.safeParse(value) + if (!parsed.success || (parsed.data.result && parsed.data.result.taskId !== taskId)) { + throw new Error('Grok returned an invalid background task kill response') + } + const { result } = parsed.data + return !result ? 'refused' : result.outcome === 'killed' ? 'killed' : 'gone' + } +} diff --git a/src/main/acp/acp-dialects/grok-dialect.ts b/src/main/acp/acp-dialects/grok-dialect.ts index 569bca94ebf4..b38eea156b2c 100644 --- a/src/main/acp/acp-dialects/grok-dialect.ts +++ b/src/main/acp/acp-dialects/grok-dialect.ts @@ -6,6 +6,7 @@ import { grokBackgroundTaskNotification, grokToolBackgroundTasks } from './grok- import { grokModelEfforts } from './grok-model-catalog' import { grokSubagentNotification, grokToolSubagents } from './grok-subagents' import { grokSubagentStop } from './grok-subagent-stop' +import { grokBackgroundTaskStop } from './grok-background-task-stop' const tokenCount = z.number().int().nonnegative() const toolMetaSchema = z.object({ 'x.ai/tool': z.object({ name: z.string().min(1) }) }) @@ -194,6 +195,7 @@ function notification( export const GROK_ACP_DIALECT: AcpDialect = { subagentStop: grokSubagentStop, + backgroundTaskStop: grokBackgroundTaskStop, injectedPromptIdentity: true, toolName: (update) => { const parsed = toolMetaSchema.safeParse(update._meta) diff --git a/src/main/acp/acp-structured-acquire.ts b/src/main/acp/acp-structured-acquire.ts index 3f8e9555e9f8..bef2d4c3a166 100644 --- a/src/main/acp/acp-structured-acquire.ts +++ b/src/main/acp/acp-structured-acquire.ts @@ -31,7 +31,7 @@ import type { AcpSessionEvent } from './acp-session-runtime' import type { AcpStructuredConnection } from './acp-structured-connection' import { ACP_HANDLE_TRANSPORT, acpAgentName } from './acp-structured-agent-definitions' import { AcpStructuredLane, acpLaneChildWorkDelivery } from './acp-structured-lane' -import { probeAcpChildStop } from './acp-structured-child-stop' +import { NO_ACP_CHILD_STOPS, probeAcpChildStops } from './acp-structured-child-stop' import type { AcpStructuredLaunch } from './acp-structured-launch-resolution' import { AcpStructuredOptions, restoreAcpSessionOptions } from './acp-structured-options' import { AcpStructuredPrompts } from './acp-structured-prompts' @@ -185,7 +185,7 @@ export async function acquireAcpStructuredSession(input: { deps.readProcessStartTime ) /** `attaching`: the lane opens inside the attach window, before any frame queued so far. */ - let subagentStopSupported = false + let childStops = NO_ACP_CHILD_STOPS const makeLane = (providerSessionId: string, attaching = false): AcpStructuredLane => { const lane = new AcpStructuredLane({ sink, @@ -196,7 +196,7 @@ export async function acquireAcpStructuredSession(input: { providerSessionId, dialect: spec.dialect, now, - canStopSubagents: () => subagentStopSupported, + childStops: () => childStops, ...acpLaneChildWorkDelivery(deps, sessionId), logger: deps.logger ?? createStructuredAgentSessionLogger(), onInputAccepted: (clientMessageId) => session?.turns.accept(clientMessageId), @@ -261,7 +261,7 @@ export async function acquireAcpStructuredSession(input: { started = await connection.start({ cwd: launch.cwd, mcpServers: [], ...auth }) liveLane = makeLane(started.sessionId) } - subagentStopSupported = await probeAcpChildStop(connection, spec.dialect) + childStops = await probeAcpChildStops(connection, spec.dialect) options.adoptSession(started.response, started.kind === 'new' ? 'new' : 'loaded') liveLane.apply(liveLane.translator.contextModels(started.response.models, now())) const restoreSkipped = await restoreAcpSessionOptions(connection, options, acquire.options) @@ -280,7 +280,7 @@ export async function acquireAcpStructuredSession(input: { fence: acquire.fence, acquisitionGeneration: generation, spec, - subagentStopSupported, + childStops, connection, lane: liveLane, prompts, diff --git a/src/main/acp/acp-structured-adapter.test-support.ts b/src/main/acp/acp-structured-adapter.test-support.ts index ca3e3bd7f66e..b2bd9a3976ed 100644 --- a/src/main/acp/acp-structured-adapter.test-support.ts +++ b/src/main/acp/acp-structured-adapter.test-support.ts @@ -191,6 +191,9 @@ export async function openAcpAdapterRig( agent.on('_x.ai/subagent/cancel', (frame) => agent.fail(frame, -32602, 'Invalid params', 'invalid params: missing field `subagentId`') ) + agent.on('_x.ai/task/kill', (frame) => + agent.fail(frame, -32602, 'Invalid params', 'invalid params: missing field `sessionId`') + ) agent.on('initialize', (frame) => { spawned.push('initialize') agent.reply(frame, { diff --git a/src/main/acp/acp-structured-background-task-stop.test.ts b/src/main/acp/acp-structured-background-task-stop.test.ts new file mode 100644 index 000000000000..e9a723970387 --- /dev/null +++ b/src/main/acp/acp-structured-background-task-stop.test.ts @@ -0,0 +1,305 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { _internals } from '../agent-hooks/server' +import { isBackgroundTaskBlock } from '../../shared/native-chat-types' +import { + closeProviderTimelineRigs, + SESSION as ADAPTER_SESSION +} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' +import { + CALLER, + envelope +} from '../native-chat/agent-session-wire/structured-agent-session-host-test-harness' +import { HOST_TEST_SESSION as SESSION } from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { grokBackgroundTaskStop } from './acp-dialects/grok-background-task-stop' +import type { AcpScriptedAgent, FakeFrame } from './acp-scripted-agent.test-support' +import type { AgentChildWorkEvidence } from '../../shared/agent-status-child-work-evidence' +import { + openAcpAdapterRig, + PROVIDER_SESSION, + replyChunk +} from './acp-structured-adapter.test-support' +import { openAttachedHostRig, promptIdOf, send } from './acp-structured-host.test-support' +import { acpChildWorkStatusSink } from './acp-structured-child-work.test-support' + +beforeEach(() => _internals.resetCachesForTests()) +const cleanup: (() => Promise)[] = [] +afterEach(async () => { + for (const close of cleanup.splice(0)) { + await close() + } + await closeProviderTimelineRigs() +}) + +const TASK = '01a10366-1e7f-7191-bcd1-fac24585851f' +const CHILD = `acp-task:${TASK}` + +async function fixture() { + const childWork = acpChildWorkStatusSink() + const hosted = await openAttachedHostRig({}, childWork.sink) + cleanup.push(() => hosted.host.close(SESSION, 'user-close')) + await send(hosted.host, 'Start the dev server') + const prompt = await hosted.rig.frame('session/prompt') + const agent = hosted.rig.child().agent + agent.notify('session/update', replyChunk(promptIdOf(prompt), 'Starting it')) + await hosted.rows() + // As the recorded s6-background Grok run sends them. + const background = async (taskId = TASK, extra: Record = {}) => { + agent.notify('_x.ai/task_backgrounded', { + sessionId: PROVIDER_SESSION, + update: { + sessionUpdate: 'task_backgrounded', + tool_call_id: 'call-1', + task_id: taskId, + command: 'pnpm dev', + cwd: '/workspace/project', + output_file: '/workspace/file-1', + description: 'Run the dev server', + ...extra + } + }) + await hosted.rows() + } + const complete = async (snapshot: Record, taskId = TASK) => { + agent.notify('_x.ai/task_completed', { + sessionId: PROVIDER_SESSION, + update: { + sessionUpdate: 'task_completed', + task_snapshot: { + task_id: taskId, + command: 'pnpm dev', + completed: true, + kind: 'bash', + description: 'Run the dev server', + is_backgrounded: true, + ...snapshot + }, + will_wake: false + } + }) + await hosted.rows() + } + const strip = async () => + (await hosted.host.history({ sessionId: SESSION, direction: 'tail' })).page.backgroundTasks + const rowState = async () => + (await hosted.rows()) + .flatMap((row) => + row.body.kind === 'message' ? row.body.blocks.filter(isBackgroundTaskBlock) : [] + ) + .find((block) => block.taskId === TASK)?.state + const targetedStop = (taskId: string) => { + const fields = { turnId: 'background-tasks', scope: 'background-tasks' as const, taskId } + return hosted.host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', fields), + ...fields + }) + } + const kills = () => agent.frames.filter((frame) => frame.method === '_x.ai/task/kill') + return { + ...hosted, + childWork, + prompt, + agent, + background, + complete, + strip, + rowState, + targetedStop, + kills + } +} + +describe('Grok kill reply contract', () => { + it.each([ + [{ result: { taskId: 'task', outcome: 'killed' } }, 'killed'], + [{ result: { taskId: 'task', outcome: 'already_exited' } }, 'gone'], + [{ result: { taskId: 'task', outcome: 'not_found' } }, 'gone'], + [{ result: null, error: 'session not found' }, 'refused'] + ])('reads %j as %s', (reply, outcome) => { + expect(grokBackgroundTaskStop.response(reply, 'task')).toBe(outcome) + }) + + it.each([ + { result: { taskId: 'other', outcome: 'killed' } }, + { result: { taskId: 'task', outcome: 'future_outcome' } }, + { result: { taskId: 'task' } }, + { result: null }, + { killed: true } + ])('rejects a malformed or mismatched reply: %j', (reply) => { + expect(() => grokBackgroundTaskStop.response(reply, 'task')).toThrow( + 'invalid background task kill response' + ) + }) +}) + +describe('Grok background commands in the background-tasks strip', () => { + it('lists a backgrounded command with its own Stop and removes it when Grok says it ended', async () => { + const f = await fixture() + await f.background() + expect(await f.strip()).toMatchObject({ + supportsTaskStop: true, + supportsStopAll: false, + tasks: [{ id: CHILD, kind: 'command', description: 'Run the dev server', stoppable: true }] + }) + expect(f.childWork.views()).toMatchObject([ + { providerId: CHILD, kind: 'command', membership: 'live' } + ]) + f.agent.reply(f.prompt, { stopReason: 'end_turn' }) + await f.rows() + expect((await f.strip())?.tasks).toMatchObject([{ id: CHILD }]) + await f.complete({ exit_code: 0 }) + expect(await f.strip()).toBeNull() + expect(f.childWork.views()).toEqual([]) + expect(await f.rowState()).toBe('done') + }) + + it('lists a monitor as monitoring', async () => { + const f = await fixture() + await f.background(TASK, { monitor_description: 'Watch the build log' }) + expect((await f.strip())?.tasks).toMatchObject([ + { id: CHILD, kind: 'monitor', description: 'Watch the build log', stoppable: true } + ]) + expect(f.childWork.views()).toMatchObject([{ state: 'monitoring' }]) + }) + + it("kills only the named task, and the row leaves on Grok's own completion", async () => { + const f = await fixture() + await f.background() + await f.background('sibling-task') + f.agent.on('_x.ai/task/kill', (frame) => + f.agent.reply(frame, { result: { taskId: TASK, outcome: 'killed' } }) + ) + expect(await f.targetedStop(CHILD)).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(f.kills().at(-1)?.params).toEqual({ sessionId: PROVIDER_SESSION, taskId: TASK }) + expect(f.agent.frames.filter((frame) => frame.method === 'session/cancel')).toEqual([]) + expect(f.rig.child().closed).toBe(false) + // Held until Grok's own ending arrives, so the strip keeps the row in its stopping state. + expect((await f.strip())?.tasks?.map((task) => task.id)).toEqual([ + CHILD, + 'acp-task:sibling-task' + ]) + await f.complete({ signal: 'killed', explicitly_killed: true }) + expect((await f.strip())?.tasks?.map((task) => task.id)).toEqual(['acp-task:sibling-task']) + expect(await f.rowState()).toBe('idle') + }) + + it.each(['already_exited', 'not_found'])( + 'treats %s as gone: the transcript row settles, and a restated live task stays gone', + async (outcome) => { + const f = await fixture() + await f.background() + f.agent.on('_x.ai/task/kill', (frame) => + f.agent.reply(frame, { result: { taskId: TASK, outcome } }) + ) + expect(await f.targetedStop(CHILD)).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(await f.rowState()).toBe('unverifiable') + expect(await f.strip()).toBeNull() + expect(f.childWork.views()).toEqual([]) + await f.background() + expect(await f.rowState()).toBe('unverifiable') + expect(await f.strip()).toBeNull() + expect(f.childWork.views()).toEqual([]) + } + ) + + it("still writes Grok's result when its completion follows already_exited", async () => { + const f = await fixture() + await f.background() + f.agent.on('_x.ai/task/kill', (frame) => + f.agent.reply(frame, { result: { taskId: TASK, outcome: 'already_exited' } }) + ) + expect(await f.targetedStop(CHILD)).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(await f.rowState()).toBe('unverifiable') + await f.complete({ exit_code: 0 }) + expect(await f.rowState()).toBe('done') + expect(await f.strip()).toBeNull() + }) + + // As for Codex and Claude: running background work must be stopped before the chat is cleared. + it('refuses /clear while a background command runs', async () => { + const f = await fixture() + await f.background() + f.agent.reply(f.prompt, { stopReason: 'end_turn' }) + await f.rows() + const fields = { command: 'clear' as const } + expect( + await f.host.conversationCommand(CALLER, { + ...fields, + envelope: envelope('agentSession.conversationCommand', fields) + }) + ).toMatchObject({ + ok: false, + refusal: { + code: 'agent_session_operation_invalid', + message: 'Stop background tasks before using this command.', + details: { reason: 'backgroundTasksRunning' } + } + }) + }) + + // Grok's own refusal means it killed nothing; a reply about another task proves nothing. + it.each([ + ['rpc-error', { ok: false, refusal: { code: 'agent_session_operation_invalid' } }], + ['envelope-error', { ok: false, refusal: { code: 'agent_session_operation_invalid' } }], + ['mismatched-id', { ok: false, refusal: { code: 'agent_session_operation_unknown' } }] + ] as const)('keeps the row after a %s', async (failure, answer) => { + const f = await fixture() + await f.background() + f.agent.on('_x.ai/task/kill', (frame) => { + if (failure === 'rpc-error') { + f.agent.fail(frame, -32603, 'Internal error') + } else if (failure === 'envelope-error') { + f.agent.reply(frame, { result: null, error: 'session actor died' }) + } else { + f.agent.reply(frame, { result: { taskId: 'other', outcome: 'killed' } }) + } + }) + expect(await f.targetedStop(CHILD)).toMatchObject(answer) + expect((await f.strip())?.tasks).toMatchObject([{ id: CHILD, stoppable: true }]) + expect(f.rig.child().closed).toBe(false) + }) +}) + +describe('Grok kill route check', () => { + async function checked(answer: (agent: AcpScriptedAgent, frame: FakeFrame) => void) { + const evidence: AgentChildWorkEvidence[] = [] + const rig = await openAcpAdapterRig({ + script: (agent) => agent.on('_x.ai/task/kill', (frame) => answer(agent, frame)), + deps: { onChildWorkEvidence: (_session, edges) => evidence.push(...edges) } + }) + cleanup.push(() => rig.adapter.closeAll()) + await rig.acquire() + rig.child().agent.notify('_x.ai/task_backgrounded', { + sessionId: PROVIDER_SESSION, + update: { sessionUpdate: 'task_backgrounded', task_id: TASK, command: 'pnpm dev' } + }) + await rig.settle() + return { rig, evidence } + } + + it('hides Stop only when Grok names the route unknown, and still lists the command', async () => { + const { rig, evidence } = await checked((agent, frame) => { + expect(frame.params).toEqual({}) + agent.fail(frame, -32601, 'Method not found') + }) + // The session still offers Stop for its subagents, whose route answered. + expect(rig.adapter.backgroundTaskStops(ADAPTER_SESSION)?.supportsTaskStop).toBe(true) + expect(evidence).toMatchObject([ + { type: 'live', child: { handle: { id: CHILD }, kind: 'command', stoppable: false } } + ]) + await expect( + rig.adapter.stopBackgroundTasks({ sessionId: ADAPTER_SESSION, fence: 1, taskIds: [CHILD] }) + ).rejects.toThrow('unavailable') + expect(rig.sent('_x.ai/task/kill')).toHaveLength(1) + }) + + it.each([ + [ + 'another refusal', + (agent: AcpScriptedAgent, frame: FakeFrame) => agent.fail(frame, -32603, 'Internal error') + ], + ['no answer', () => {}] + ])('leaves Stop to answer for itself after %s', async (_label, answer) => { + const { evidence } = await checked(answer) + expect(evidence).toMatchObject([{ type: 'live', child: { stoppable: true } }]) + }) +}) diff --git a/src/main/acp/acp-structured-child-stop.test.ts b/src/main/acp/acp-structured-child-stop.test.ts index c39fe5fad2f3..8a5d9b51c97d 100644 --- a/src/main/acp/acp-structured-child-stop.test.ts +++ b/src/main/acp/acp-structured-child-stop.test.ts @@ -53,11 +53,13 @@ describe('Grok targeted child control contract', () => { 'hides targeted Stop when the non-mutating route check fails with %s', async (code) => { const rig = await openAcpAdapterRig({ - script: (agent) => + script: (agent) => { agent.on('_x.ai/subagent/cancel', (frame) => { expect(frame.params).toEqual({}) agent.fail(frame, code, 'Unavailable') }) + agent.on('_x.ai/task/kill', (frame) => agent.fail(frame, -32601, 'Method not found')) + } }) cleanup.push(() => rig.adapter.closeAll()) await rig.acquire() @@ -92,7 +94,10 @@ describe('Grok targeted child control contract', () => { it('bounds an unanswered route check without ending the parent connection', async () => { const rig = await openAcpAdapterRig({ - script: (agent) => agent.on('_x.ai/subagent/cancel', () => {}) + script: (agent) => { + agent.on('_x.ai/subagent/cancel', () => {}) + agent.on('_x.ai/task/kill', (frame) => agent.fail(frame, -32601, 'Method not found')) + } }) cleanup.push(() => rig.adapter.closeAll()) await rig.acquire() diff --git a/src/main/acp/acp-structured-child-stop.ts b/src/main/acp/acp-structured-child-stop.ts index c4994cefa0b9..acabe6ba6b65 100644 --- a/src/main/acp/acp-structured-child-stop.ts +++ b/src/main/acp/acp-structured-child-stop.ts @@ -1,18 +1,29 @@ import type { AcpStructuredSession } from './acp-structured-session' import type { AcpStructuredConnection } from './acp-structured-connection' -import type { AcpDialect } from './acp-dialects/acp-dialect' +import type { AcpChildStops, AcpDialect } from './acp-dialects/acp-dialect' import { AcpAgentError } from './acp-errors' import type { AgentSessionBackgroundTaskStops } from '../../shared/agent-child-work-stop-targets' +import { acpBackgroundTaskStopId } from './acp-background-task-child-work' + +type ChildStopResult = { cancelled: boolean; stillRunning?: true } +type SubagentStop = NonNullable +type BackgroundTaskStop = NonNullable + +export const NO_ACP_CHILD_STOPS: AcpChildStops = { subagents: false, backgroundTasks: false } export function acpChildStopCapabilities( session?: AcpStructuredSession ): AgentSessionBackgroundTaskStops | undefined { return session?.journalClosed === null - ? { supportsTaskStop: session.subagentStopSupported === true, supportsStopAll: false } + ? { + supportsTaskStop: + session.childStops?.subagents === true || session.childStops?.backgroundTasks === true, + supportsStopAll: false + } : undefined } -export async function probeAcpChildStop( +async function probeAcpSubagentStop( connection: AcpStructuredConnection, dialect: AcpDialect ): Promise { @@ -29,37 +40,115 @@ export async function probeAcpChildStop( } } +/** Only the agent naming the route unknown hides Stop; any other answer leaves the Stop itself to + * report what happened. */ +async function probeAcpBackgroundTaskStop( + connection: AcpStructuredConnection, + dialect: AcpDialect +): Promise { + const control = dialect.backgroundTaskStop + if (!control) { + return false + } + try { + await connection.requestExtension(control.probe.method, control.probe.params, 2_000) + return true + } catch (error) { + return !(error instanceof AcpAgentError && control.lacksRoute(error)) + } +} + +export async function probeAcpChildStops( + connection: AcpStructuredConnection, + dialect: AcpDialect +): Promise { + const [subagents, backgroundTasks] = await Promise.all([ + probeAcpSubagentStop(connection, dialect), + probeAcpBackgroundTaskStop(connection, dialect) + ]) + return { subagents, backgroundTasks } +} + +async function stopAcpSubagent( + session: AcpStructuredSession, + control: SubagentStop, + id: string, + at: () => number +): Promise { + const request = control.request(session.lane.translator.providerSessionId, id) + 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) { + return { cancelled: false, stillRunning: true } + } + 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())) + } + return { cancelled: result.cancelled } +} + +async function stopAcpBackgroundTask( + session: AcpStructuredSession, + control: BackgroundTaskStop, + taskId: string +): Promise { + const request = control.request(session.lane.translator.providerSessionId, taskId) + let value: unknown + try { + value = await session.connection.requestExtension(request.method, request.params) + } catch (error) { + // Grok answers a task it no longer holds normally, so a refusal means it killed nothing. + if (error instanceof AcpAgentError) { + return { cancelled: false, stillRunning: true } + } + throw error + } + const outcome = control.response(value, taskId) + if (outcome === 'refused') { + return { cancelled: false, stillRunning: true } + } + // A killed task's own completion ends its row. A gone one settles here, and any completion + // Grok still sends for it (exited but not yet swept) overwrites this with the real result. + if (outcome === 'gone') { + session.lane.apply(session.lane.translator.reconcileBackgroundTask(taskId)) + } + return { cancelled: true } +} + /** The host has already resolved these handles from this parent's stoppable child records. */ export async function stopAcpChildren( session: AcpStructuredSession, fence: number, taskIds: readonly string[], at: () => number -): Promise<{ cancelled: boolean; stillRunning?: true }> { - const control = session.spec.dialect.subagentStop - if (!control || !session.subagentStopSupported || session.fence !== fence) { +): Promise { + const subagentStop = session.childStops?.subagents ? session.spec.dialect.subagentStop : undefined + const taskStop = session.childStops?.backgroundTasks + ? session.spec.dialect.backgroundTaskStop + : undefined + const routed = (id: string) => + acpBackgroundTaskStopId(id) === undefined ? subagentStop !== undefined : taskStop !== undefined + if (session.fence !== fence || (!subagentStop && !taskStop) || !taskIds.every(routed)) { 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) - 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())) - } + const taskId = acpBackgroundTaskStopId(id) + const result = + taskId !== undefined && taskStop + ? await stopAcpBackgroundTask(session, taskStop, taskId) + : subagentStop + ? await stopAcpSubagent(session, subagentStop, id, at) + : { cancelled: false } cancelled ||= result.cancelled + stillRunning ||= result.stillRunning === true } 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 dd999aa42715..98a234e2d39a 100644 --- a/src/main/acp/acp-structured-child-work.test.ts +++ b/src/main/acp/acp-structured-child-work.test.ts @@ -330,8 +330,10 @@ describe('Grok roster evidence through the existing host child store', () => { const childWork = acpChildWorkStatusSink() const hosted = await openHostRig({ statusSink: childWork.sink, - script: (agent) => + script: (agent) => { agent.on('_x.ai/subagent/cancel', (frame) => agent.fail(frame, -32601, 'Method not found')) + agent.on('_x.ai/task/kill', (frame) => agent.fail(frame, -32601, 'Method not found')) + } }) const cancelOnCleanup = cleanupScriptedHost(hosted) expect(await hosted.host.attach(CALLER, attachParams())).toMatchObject({ ok: true }) diff --git a/src/main/acp/acp-structured-host-subagent-stop.test.ts b/src/main/acp/acp-structured-host-subagent-stop.test.ts index f2c47a4ecb4c..509af927d023 100644 --- a/src/main/acp/acp-structured-host-subagent-stop.test.ts +++ b/src/main/acp/acp-structured-host-subagent-stop.test.ts @@ -58,6 +58,9 @@ async function stoppingChild(reportsOutcome: boolean) { agent.on('_x.ai/subagent/cancel', (frame) => agent.fail(frame, -32602, 'Invalid params', 'invalid params: missing field `subagentId`') ) + agent.on('_x.ai/task/kill', (frame) => + agent.fail(frame, -32602, 'Invalid params', 'invalid params: missing field `sessionId`') + ) agent.on('session/new', (frame) => agent.reply(frame, { sessionId: PROVIDER_SESSION, configOptions: GROK_CONFIG_OPTIONS }) ) diff --git a/src/main/acp/acp-structured-lane.ts b/src/main/acp/acp-structured-lane.ts index 6dac885d6ea0..045e1b544d5e 100644 --- a/src/main/acp/acp-structured-lane.ts +++ b/src/main/acp/acp-structured-lane.ts @@ -11,9 +11,13 @@ import type { AcpTimelineEvent } from './acp-timeline-event' import { createLegacyProviderTimelineIdentityScheme } from '../native-chat/agent-session-timeline/provider-timeline-identity' import type { ProviderTimelineSink } from '../native-chat/agent-session-timeline/provider-timeline-plan' import type { StructuredAgentSessionCommandRun } from '../native-chat/agent-session-wire/structured-agent-session-adapter' -import type { AcpDialect } from './acp-dialects/acp-dialect' +import type { AcpChildStops, AcpDialect } from './acp-dialects/acp-dialect' import { AcpTimelineTranslator } from './acp-timeline-translator' import { acpSubagentChildWork, isAcpSubagentChildWorkEvent } from './acp-subagent-child-work' +import { + acpBackgroundTaskChildWork, + isAcpBackgroundTaskChildWorkEvent +} from './acp-background-task-child-work' import type { AgentChildWorkEvidence } from '../../shared/agent-status-child-work-evidence' import type { AcpStructuredSessionAdapterDeps } from './acp-structured-session-adapter-deps' import { @@ -42,7 +46,7 @@ export type AcpStructuredLaneDeps = { onChildWorkEvidence?: (evidence: AgentChildWorkEvidence[]) => void onChildWorkFailure?: (error: unknown) => void now?: () => number - canStopSubagents?: () => boolean + childStops?: () => AcpChildStops } export class AcpStructuredLane { @@ -163,7 +167,9 @@ export class AcpStructuredLane { const event = this.backlog[0] const { admission, dropped } = this.assembler.apply( event, - isAcpSubagentChildWorkEvent(event) ? () => this.publishChildWork(event) : undefined + isAcpSubagentChildWorkEvent(event) || isAcpBackgroundTaskChildWorkEvent(event) + ? () => this.publishChildWork(event) + : undefined ) if (dropped === 'stream-mismatch' && event.type === 'text.delta') { this.reportTextDrop({ @@ -195,11 +201,12 @@ export class AcpStructuredLane { } private publishChildWork(event: AcpTimelineEvent): void { - const evidence = acpSubagentChildWork( - event, - this.deps.now?.() ?? Date.now(), - this.deps.canStopSubagents?.() === true - ) + const observedAt = this.deps.now?.() ?? Date.now() + const stops = this.deps.childStops?.() + const evidence = [ + ...acpSubagentChildWork(event, observedAt, stops?.subagents === true), + ...acpBackgroundTaskChildWork(event, observedAt, stops?.backgroundTasks === true) + ] if (!evidence.length) { return } diff --git a/src/main/acp/acp-structured-session-adapter.test.ts b/src/main/acp/acp-structured-session-adapter.test.ts index ad3dbcd2e3ff..7299c13c6e30 100644 --- a/src/main/acp/acp-structured-session-adapter.test.ts +++ b/src/main/acp/acp-structured-session-adapter.test.ts @@ -298,7 +298,7 @@ describe('ACP structured session adapter: requests a Stop cancels', () => { cancelled: true }) expect(await question).toMatchObject({ result: { outcome: 'cancelled' } }) - expect(agent.frames.filter((frame) => frame.id === 5)).toHaveLength(1) + expect(agent.frames.filter((frame) => frame.id === 5 && !frame.method)).toHaveLength(1) }) it("shows Grok's plan in the plan row and answers its plan-mode exit at once, with no card", async () => { diff --git a/src/main/acp/acp-structured-session.ts b/src/main/acp/acp-structured-session.ts index 629145cf483d..c27179fa9b89 100644 --- a/src/main/acp/acp-structured-session.ts +++ b/src/main/acp/acp-structured-session.ts @@ -8,6 +8,7 @@ import { type StructuredAgentSessionTurnVerdict } from '../native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict' import type { AcpLaunchSpec } from './acp-launch-specs' +import type { AcpChildStops } from './acp-dialects/acp-dialect' import type { AcpSessionEvent } from './acp-session-runtime' import type { AcpStructuredConnection } from './acp-structured-connection' import type { AcpStructuredLane } from './acp-structured-lane' @@ -20,7 +21,7 @@ export type AcpStructuredSession = { fence: number acquisitionGeneration: string spec: AcpLaunchSpec - subagentStopSupported?: boolean + childStops?: AcpChildStops /** The agent's process and its protocol, one owner. */ connection: AcpStructuredConnection lane: AcpStructuredLane diff --git a/src/main/acp/acp-timeline-translator.ts b/src/main/acp/acp-timeline-translator.ts index fbfdfba9dfb0..a5bd8746b3b8 100644 --- a/src/main/acp/acp-timeline-translator.ts +++ b/src/main/acp/acp-timeline-translator.ts @@ -161,6 +161,12 @@ export class AcpTimelineTranslator { ) } + /** Settles a task the agent says it no longer runs; its own later ending may still overwrite this. */ + reconcileBackgroundTask(taskId: string): ProviderTimelineEvent[] { + const update = { taskId, state: 'unverifiable' as const, summary: '', error: '' } + return this.backgroundTasks.translate([update], { thread: this.options.sessionId }) + } + sessionEvent(event: AcpSessionEvent, at: number): ProviderTimelineEvent[] { return this.notification( 'session/update', diff --git a/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx b/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx index 9ad245fde3e6..e3da66df4448 100644 --- a/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatBackgroundTasksStatus.test.tsx @@ -144,6 +144,49 @@ describe('NativeChatBackgroundTasksStatus stop affordances', () => { expect(onStop).toHaveBeenCalledWith('codex-command:exec-1') }) + it('stops a Grok background command and the Grok subagent beside it by their own ids', () => { + 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 })) + + fireEvent.click(screen.getByLabelText('Stop Run the dev server')) + fireEvent.click(screen.getByLabelText('Stop Review the diff')) + expect(onStop.mock.calls).toEqual([['acp-task:01a10366'], ['subagent-1']]) + expect(screen.queryByLabelText('Stop background tasks')).not.toBeInTheDocument() + }) + it('offers no stop at all when the provider exposes none', () => { // An older Codex: a Stop button here would be a control that cannot act. renderStrip({ supportsTaskStop: false, supportsStopAll: false })