diff --git a/docs/background-work-experiment.md b/docs/background-work-experiment.md new file mode 100644 index 00000000..41b2e381 --- /dev/null +++ b/docs/background-work-experiment.md @@ -0,0 +1,47 @@ +# Background work RFD experiment + +This branch prototypes [RFD 2452](https://github.com/agentclientprotocol/agent-client-protocol/pull/2452) +at revision `33c473ba0723a0f6ffcfcbb54fd780c05df7b208`. It is an opt-in local +experiment, not a released protocol implementation. + +Build with `npm ci --ignore-scripts` and `npm run build`. Launch with +`ACP_BACKGROUND_WORK_EXPERIMENT=1 node dist/index.js`. Omit the environment +variable to restore normal behavior. No package publication is needed. + +The message-stream bridge works on ACP v1 and v2. It reuses the installed +native asynchronous-task adapter and its private negotiation internally. Known +background shells keep the existing tool call `in_progress`; state snapshots +reference that tool. `tool_call/cancel` is translated to the existing native +task stop request. A successful response is acceptance only. A native terminal +event finalizes the tool. ACP v1 represents cancellation as `failed` because its +tool status enum has no `cancelled`; ACP v2 uses `cancelled`. + +Native replace-semantics task levels establish whether the list is complete. +Before a native level arrives, after a malformed level, or while any native task +cannot be mapped to an announced tool, `background` is `null`. This includes +unmapped monitors, scheduled tasks and subagents. The prototype does not claim +their full RFD support. Forks start with `background: []`. Reconnects remain +unreported until native current state and referenced entities are available. + +This bridge does not upgrade the SDK schema. A client must accept the draft +fields and raw `tool_call/cancel`, and must treat repeated `idle` snapshots +without a stop reason as list changes rather than new completions. The private +negotiation can also select the adapter's existing client-specific tool rendering; +this is a bounded experiment, not general client capability negotiation. + +Suggested manual evaluation: start a bounded background shell, finish the +foreground turn, cancel one shell, and separately allow a shell to finish while +idle. Verify that the tool remains in progress and listed until native evidence +ends it. Send foreground `session/cancel` while background work is listed, then +close the session. Reconnect and fork must not invent live tools from history. +Exercise pending permissions separately; deterministic protocol tests are not +evidence of live native permission or session-close behavior. + +A tool already reported final before native background discovery is never revived. +That task remains unreported by this bridge. Native subagents also remain unreported +while associated, including an idle child whose own background state is unknown. + +Codex list disappearance proves the process ended but supplies no exit outcome. +The existing native adapter reports this as stopped, represented here as cancelled +(v2) or failed (v1); only an explicit native command completion supplies a reliable +completed/failed outcome. This is a known limit of the experiment. diff --git a/src/acp-agent.ts b/src/acp-agent.ts index e9049160..a77638fc 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -4410,7 +4410,10 @@ export class ClaudeAcpAgent { } else { await session.query.stopTask(params.asyncTaskId); } - await asyncTasks.taskStopped(params.asyncTaskId); + // Acceptance is separate from the native terminal event in the RFD experiment. + if (process.env.ACP_BACKGROUND_WORK_EXPERIMENT !== "1") { + await asyncTasks.taskStopped(params.asyncTaskId); + } return { stopped: true }; } catch (error) { asyncTasks.releaseStop(params.asyncTaskId); @@ -6799,6 +6802,19 @@ export class ClaudeAcpAgent { case "control_request_progress": break; case "background_tasks_changed": + if (process.env.ACP_BACKGROUND_WORK_EXPERIMENT === "1") { + await this.client.sessionUpdate({ + sessionId: params.sessionId, + update: { + sessionUpdate: "session_info_update", + _meta: { + backgroundWorkExperiment: { + taskIds: message.tasks.map((task) => task.task_id), + }, + }, + }, + }); + } await asyncTasks.backgroundTasksChanged(message.tasks); // A level signal: the full live background-task set on every // membership change, with REPLACE semantics. Used only to @@ -8596,7 +8612,13 @@ export class ClaudeAcpAgent { session.cancelled = true; // The interrupt below makes Claude Code withdraw the requests that are open // now (see `withPendingUserInput`). - for (const input of session.openUserInputs ?? []) input.cancelled = true; + for (const input of session.openUserInputs ?? []) { + if ( + process.env.ACP_BACKGROUND_WORK_EXPERIMENT !== "1" || + blocksForeground(session, input.agentID) + ) + input.cancelled = true; + } // Before anything awaits, so a model request that the consumer reads in // the meantime is already judged against these prompts. session.cancelledPrompts ??= new Set(); @@ -11396,7 +11418,10 @@ export class ClaudeAcpAgent { // `stopBackgroundWorkflows` stops instead. Background commands survive an // interrupt either way. Before `userProvidedOptions`, so a client's own // `perTaskStopAffordance` wins. - ...(subagentForm(this.clientCapabilities) === "rfd" ? { perTaskStopAffordance: true } : {}), + ...(subagentForm(this.clientCapabilities) === "rfd" || + process.env.ACP_BACKGROUND_WORK_EXPERIMENT === "1" + ? { perTaskStopAffordance: true } + : {}), // One predicted follow-up prompt per turn, published as // `session_info_update._meta.promptSuggestions`. It rides the turn's own // prompt cache, and the user can turn it off in settings.json diff --git a/src/background-work-experiment.ts b/src/background-work-experiment.ts new file mode 100644 index 00000000..45217f04 --- /dev/null +++ b/src/background-work-experiment.ts @@ -0,0 +1,362 @@ +/** Opt-in wire prototype for RFD 2452 over the installed native task adapters. */ +type Json = Record; +type Task = { + id: string; + toolCallId?: string; + canStop: boolean; + final: boolean; + supported: boolean; +}; +type Session = { + state: string; + known: boolean; + level?: Set; + tasks: Map; + tools: Set; + finalTools: Set; + children: Set; +}; +function object(value: unknown): Json { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Json) + : {}; +} + +export class BackgroundWorkExperiment { + private sessions = new Map(); + private requests = new Map(); + private cancellations = new Set(); + private protocol = 1; + private session(id: string): Session { + let session = this.sessions.get(id); + if (!session) { + session = { + state: "idle", + known: false, + tasks: new Map(), + tools: new Set(), + finalTools: new Set(), + children: new Set(), + }; + this.sessions.set(id, session); + } + return session; + } + private notification(sessionId: string, update: Json): Json { + return { + jsonrpc: "2.0", + method: "session/update", + params: { sessionId, update }, + }; + } + snapshot(sessionId: string): Json { + const session = this.session(sessionId); + const live = [...session.tasks.values()].filter( + (task) => !task.final && session.level?.has(task.id), + ); + const complete = + session.known && + session.children.size === 0 && + [...(session.level ?? [])].every((id) => { + const task = session.tasks.get(id); + return ( + task && + task.supported && + !task.final && + task.toolCallId !== undefined && + session.tools.has(task.toolCallId) && + !session.finalTools.has(task.toolCallId) + ); + }); + return this.notification(sessionId, { + sessionUpdate: "state_update", + state: session.state, + background: complete + ? live.map((task) => ({ + type: "tool_call", + toolCallId: task.toolCallId, + })) + : null, + }); + } + inbound(message: Json): { forward?: Json; send: Json[] } { + const params = object(message["params"]); + const sessionId = typeof params["sessionId"] === "string" ? params["sessionId"] : undefined; + const send: Json[] = []; + if (message["method"] === "initialize") { + this.protocol = typeof params["protocolVersion"] === "number" ? params["protocolVersion"] : 1; + const capabilities = object(params["clientCapabilities"]), + meta = object(capabilities["_meta"]); + const jetbrains = object(meta["jetbrains"]), + air = object(jetbrains["air"]); + const old = Array.isArray(air["capabilities"]) ? air["capabilities"] : []; + // The native task adapters already implement these private capabilities. + message = { + ...message, + params: { + ...params, + clientCapabilities: { + ...capabilities, + _meta: { + ...meta, + jetbrains: { + ...jetbrains, + air: { + ...air, + version: 1, + capabilities: [...new Set([...old, "asyncTasks", "backgroundSubagents"])], + }, + }, + }, + }, + }, + }; + } + if (["session/resume", "session/load"].includes(String(message["method"])) && sessionId) { + const session = this.session(sessionId); + session.known = false; + delete session.level; + session.tools.clear(); + } + if (message["method"] === "tool_call/cancel") { + const session = sessionId ? this.session(sessionId) : undefined; + const task = + session && + [...session.tasks.values()].find( + (task) => + !task.final && + task.supported && + task.canStop && + task.toolCallId === params["toolCallId"] && + task.toolCallId !== undefined && + session.tools.has(task.toolCallId) && + !session.finalTools.has(task.toolCallId), + ); + if (!task) + return { + send: [ + { + jsonrpc: "2.0", + id: message["id"], + error: { + code: -32602, + message: "Tool call is not cancellable background work", + }, + }, + ], + }; + this.cancellations.add(message["id"]); + return { + forward: { + ...message, + method: "_session/async_task/stop", + params: { sessionId, asyncTaskId: task.id }, + }, + send, + }; + } + if (message["id"] !== undefined && typeof message["method"] === "string") + this.requests.set(message["id"], { + method: message["method"], + sessionId, + }); + if (message["method"] === "session/prompt" && sessionId && this.protocol < 2) { + this.session(sessionId).state = "running"; + send.push(this.snapshot(sessionId)); + } + return { forward: message, send }; + } + outbound(message: Json): Json[] { + if (!message["method"] && this.cancellations.delete(message["id"])) { + if (message["error"]) return [message]; + return object(message["result"])["stopped"] === true + ? [{ ...message, result: {} }] + : [ + { + jsonrpc: "2.0", + id: message["id"], + error: { + code: -32602, + message: "Native cancellation was not accepted", + }, + }, + ]; + } + const request = !message["method"] ? this.requests.get(message["id"]) : undefined; + if (request) { + this.requests.delete(message["id"]); + const result = object(message["result"]); + const id = typeof result["sessionId"] === "string" ? result["sessionId"] : request.sessionId; + if (id && !message["error"]) { + if (request.method === "session/new" || request.method === "session/fork") { + if (request.method === "session/fork") { + this.session(id).known = true; + this.session(id).level = new Set(); + } + return [message, this.snapshot(id)]; + } + if (request.method === "session/resume" || request.method === "session/load") + return [this.snapshot(id), message]; + if (request.method === "session/close") this.sessions.delete(id); + } + if (id && request.method === "session/prompt" && this.protocol < 2) { + this.session(id).state = "idle"; + return [this.snapshot(id), message]; + } + } + const params = object(message["params"]), + update = object(params["update"]); + const sessionId = typeof params["sessionId"] === "string" ? params["sessionId"] : undefined; + if (message["method"] !== "session/update" || !sessionId) return [message]; + const session = this.session(sessionId); + const marker = object(object(update["_meta"])["backgroundWorkExperiment"]); + if (update["sessionUpdate"] === "session_info_update" && "taskIds" in marker) { + if ( + !Array.isArray(marker["taskIds"]) || + !marker["taskIds"].every((id) => typeof id === "string" && id.length > 0) + ) { + session.known = false; + } else { + session.level = new Set(marker["taskIds"] as string[]); + session.known = true; + } + return [this.snapshot(sessionId)]; + } + if (update["sessionUpdate"] === "state_update") { + if (typeof update["state"] === "string") session.state = update["state"]; + return [ + { + ...message, + params: { + ...params, + update: { + ...update, + background: object(object(this.snapshot(sessionId)["params"])["update"])[ + "background" + ], + }, + }, + }, + ]; + } + if ( + ["subagent_update", "_subagent_update", "subagent_spawned", "subagent_state_update"].includes( + String(update["sessionUpdate"]), + ) + ) { + const id = String(update["sessionId"] ?? update["subagentSessionId"] ?? "unknown"); + const state = + typeof update["state"] === "string" ? update["state"] : object(update["state"])["state"]; + if (["completed", "failed", "cancelled"].includes(String(state))) session.children.delete(id); + else session.children.add(id); + return [message, this.snapshot(sessionId)]; + } + if (update["sessionUpdate"] === "tool_call" || update["sessionUpdate"] === "tool_call_update") { + const id = typeof update["toolCallId"] === "string" ? update["toolCallId"] : ""; + if (id && (this.protocol >= 2 || update["sessionUpdate"] === "tool_call")) + session.tools.add(id); + const task = [...session.tasks.values()].find( + (task) => + !task.final && task.supported && task.toolCallId === id && !session.finalTools.has(id), + ); + if (!task && ["completed", "failed", "cancelled"].includes(String(update["status"]))) + session.finalTools.add(id); + return task + ? [ + { + ...message, + params: { + ...params, + update: { + ...update, + status: "in_progress", + capabilities: task.canStop ? { cancel: {} } : {}, + }, + }, + }, + ] + : [message]; + } + if ( + typeof update["sessionUpdate"] !== "string" || + !update["sessionUpdate"].replace(/^_/, "").startsWith("async_task_") + ) + return [message]; + const id = typeof update["asyncTaskId"] === "string" ? update["asyncTaskId"] : undefined; + if (!id) { + session.known = false; + return [message, this.snapshot(sessionId)]; + } + let task = session.tasks.get(id); + if (!task) { + task = { id, canStop: false, final: false, supported: false }; + session.tasks.set(id, task); + } + if (String(update["sessionUpdate"]).replace(/^_/, "") === "async_task_spawned") + task.supported = update["taskType"] === "shell"; + if (typeof update["toolCallId"] === "string") task.toolCallId = update["toolCallId"]; + if (update["canStop"] === true) task.canStop = true; + const terminal = ["completed", "failed", "stopped"].includes(String(update["state"])); + if (!terminal && session.level && !session.level.has(id)) session.known = false; + const wasFinal = task.final; + if (terminal) { + task.final = true; + session.level?.delete(id); + } + if (wasFinal) return [this.snapshot(sessionId)]; + if ( + !task.supported || + !task.toolCallId || + !session.tools.has(task.toolCallId) || + session.finalTools.has(task.toolCallId) + ) + return [message, this.snapshot(sessionId)]; + const updates: Json[] = []; + if (!wasFinal) + updates.push( + this.notification(sessionId, { + sessionUpdate: "tool_call_update", + toolCallId: task.toolCallId, + status: terminal + ? update["state"] === "stopped" + ? this.protocol < 2 + ? "failed" + : "cancelled" + : update["state"] + : "in_progress", + capabilities: terminal ? null : task.canStop ? { cancel: {} } : {}, + }), + ); + if (terminal) session.finalTools.add(task.toolCallId); + return [...updates, this.snapshot(sessionId)]; + } +} + +/** Draft messages are added at the stream boundary, outside the installed SDK schema. */ +export function backgroundWorkStream(stream: { + readable: ReadableStream; + writable: WritableStream; +}): typeof stream { + if (process.env["ACP_BACKGROUND_WORK_EXPERIMENT"] !== "1") return stream; + const experiment = new BackgroundWorkExperiment(), + writer = stream.writable.getWriter(); + const send = async (messages: Json[]) => { + for (const message of messages) await writer.write(message as T); + }; + return { + readable: stream.readable.pipeThrough( + new TransformStream({ + async transform(message, controller) { + const result = experiment.inbound(object(message)); + await send(result["send"]); + if (result["forward"]) controller.enqueue(result["forward"] as T); + }, + }), + ), + writable: new WritableStream({ + write: (message) => send(experiment.outbound(object(message))), + close: () => writer.close(), + abort: (reason) => writer.abort(reason), + }), + }; +} diff --git a/src/serve.ts b/src/serve.ts index 3f11699d..a67fd6bf 100644 --- a/src/serve.ts +++ b/src/serve.ts @@ -1,3 +1,4 @@ +import { backgroundWorkStream } from "./background-work-experiment.js"; import * as v2 from "@agentclientprotocol/sdk/experimental/v2"; import { v1AgentApp, type ClaudeAcpAgent, type Logger } from "./acp-agent.js"; import { nodeToWebReadable, nodeToWebWritable } from "./utils.js"; @@ -37,7 +38,7 @@ export function serveAcp( agent = created; }; const { closed } = acpProtocolRouter(logger, onAgent, options.protocols).connect( - v2.ndJsonStream(output, input), + backgroundWorkStream(v2.ndJsonStream(output, input)), ); return { closed, diff --git a/src/tests/background-work-experiment.test.ts b/src/tests/background-work-experiment.test.ts new file mode 100644 index 00000000..918d3e89 --- /dev/null +++ b/src/tests/background-work-experiment.test.ts @@ -0,0 +1,244 @@ +import { describe, expect, it } from "vitest"; +import { BackgroundWorkExperiment, backgroundWorkStream } from "../background-work-experiment.js"; + +const update = (value: Record, sessionId = "s") => ({ + method: "session/update", + params: { sessionId, update: value }, +}); +const level = (taskIds: unknown) => + update({ + sessionUpdate: "session_info_update", + _meta: { backgroundWorkExperiment: { taskIds } }, + }); +const expected = (background: unknown, state = "idle") => ({ + params: { + sessionId: "s", + update: { sessionUpdate: "state_update", state, background }, + }, +}); + +describe.each([1, 2])("background work experiment v%s", (version) => { + function fixture() { + const bridge = new BackgroundWorkExperiment(); + bridge.inbound({ + id: 1, + method: "initialize", + params: { protocolVersion: version }, + }); + bridge.inbound({ id: 2, method: "session/new", params: {} }); + bridge.outbound({ id: 2, result: { sessionId: "s" } }); + return bridge; + } + function spawn(bridge: BackgroundWorkExperiment) { + bridge.outbound( + update({ + sessionUpdate: version === 1 ? "tool_call" : "tool_call_update", + toolCallId: "tool", + status: "in_progress", + }), + ); + bridge.outbound(level(["native"])); + return bridge.outbound( + update({ + sessionUpdate: version === 1 ? "async_task_spawned" : "_async_task_spawned", + asyncTaskId: "native", + taskType: "shell", + toolCallId: "tool", + canStop: true, + }), + ); + } + it("does not infer empty from a silent backend or unmapped monitor", () => { + const bridge = fixture(); + expect(bridge.snapshot("s")).toMatchObject(expected(null)); + expect(bridge.outbound(level(["monitor"]))).toMatchObject([expected(null)]); + expect(bridge.outbound(level([]))).toMatchObject([expected([])]); + expect(bridge.outbound(level([null]))).toMatchObject([expected(null)]); + expect(bridge.outbound(level("damaged"))).toMatchObject([expected(null)]); + }); + it("references announced tools and keeps them running after foreground completion", () => { + const bridge = fixture(); + expect(spawn(bridge)).toMatchObject([ + { + params: { + update: { + sessionUpdate: "tool_call_update", + toolCallId: "tool", + status: "in_progress", + capabilities: { cancel: {} }, + }, + }, + }, + expected([{ type: "tool_call", toolCallId: "tool" }]), + ]); + expect( + bridge.outbound( + update({ + sessionUpdate: "tool_call_update", + toolCallId: "tool", + status: "completed", + }), + ), + ).toMatchObject([{ params: { update: { status: "in_progress" } } }]); + bridge.outbound( + update({ + sessionUpdate: "state_update", + state: "idle", + stopReason: "end_turn", + }), + ); + expect(bridge.outbound(level(["native"]))).toMatchObject([ + expected([{ type: "tool_call", toolCallId: "tool" }]), + ]); + expect(bridge.outbound(level(["native"]))[0]).not.toHaveProperty("params.update.stopReason"); + bridge.inbound({ + method: "session/cancel", + params: { sessionId: "s" }, + }); + expect(bridge.snapshot("s")).toMatchObject( + expected([{ type: "tool_call", toolCallId: "tool" }]), + ); + }); + it("acceptance does not finalize, and completion wins a cancellation race", () => { + const bridge = fixture(); + spawn(bridge); + expect( + bridge.inbound({ + id: 3, + method: "tool_call/cancel", + params: { sessionId: "s", toolCallId: "tool" }, + }), + ).toMatchObject({ + forward: { + method: "_session/async_task/stop", + params: { asyncTaskId: "native" }, + }, + }); + expect(bridge.outbound({ id: 3, result: { stopped: true } })).toMatchObject([{ result: {} }]); + expect(bridge.snapshot("s")).toMatchObject( + expected([{ type: "tool_call", toolCallId: "tool" }]), + ); + expect( + bridge.outbound( + update({ + sessionUpdate: "async_task_state_update", + asyncTaskId: "native", + state: "completed", + }), + ), + ).toMatchObject([ + { params: { update: { status: "completed", capabilities: null } } }, + expected([]), + ]); + expect( + bridge.outbound( + update({ + sessionUpdate: "async_task_state_update", + asyncTaskId: "native", + state: "stopped", + }), + ), + ).toMatchObject([expected([])]); + expect( + bridge.inbound({ + id: 4, + method: "tool_call/cancel", + params: { sessionId: "s", toolCallId: "tool" }, + }), + ).toMatchObject({ send: [{ error: { code: -32602 } }] }); + }); + it("does not revive a tool finalized before late native discovery", () => { + const bridge = fixture(); + bridge.outbound( + update({ + sessionUpdate: version === 1 ? "tool_call" : "tool_call_update", + toolCallId: "late", + status: "completed", + }), + ); + bridge.outbound(level(["native"])); + const messages = bridge.outbound( + update({ + sessionUpdate: "async_task_spawned", + asyncTaskId: "native", + taskType: "shell", + toolCallId: "late", + canStop: true, + }), + ); + expect(messages).toContainEqual(expect.objectContaining(expected(null))); + expect(messages).not.toContainEqual( + expect.objectContaining({ + params: expect.objectContaining({ + update: expect.objectContaining({ status: "in_progress" }), + }), + }), + ); + expect( + bridge.inbound({ + id: 30, + method: "tool_call/cancel", + params: { sessionId: "s", toolCallId: "late" }, + }), + ).toMatchObject({ send: [{ error: { code: -32602 } }] }); + }); + it("does not claim scheduled workflows even when they name an announced tool", () => { + const bridge = fixture(); + bridge.outbound( + update({ + sessionUpdate: version === 1 ? "tool_call" : "tool_call_update", + toolCallId: "scheduled", + status: "pending", + }), + ); + bridge.outbound(level(["workflow"])); + const messages = bridge.outbound( + update({ + sessionUpdate: "async_task_spawned", + asyncTaskId: "workflow", + taskType: "workflow", + toolCallId: "scheduled", + canStop: true, + }), + ); + expect(messages.at(-1)).toMatchObject(expected(null)); + expect( + bridge.inbound({ + id: 31, + method: "tool_call/cancel", + params: { sessionId: "s", toolCallId: "scheduled" }, + }), + ).toMatchObject({ send: [{ error: { code: -32602 } }] }); + expect( + bridge.outbound( + update({ sessionUpdate: "tool_call_update", toolCallId: "scheduled", status: "pending" }), + )[0], + ).toMatchObject({ params: { update: { status: "pending" } } }); + }); + it("does not fabricate referenced tools during resume and starts forks empty", () => { + const bridge = fixture(); + bridge.inbound({ + id: 8, + method: "session/resume", + params: { sessionId: "restored" }, + }); + expect(bridge.outbound({ id: 8, result: {} })[0]).toMatchObject({ + params: { sessionId: "restored", update: { background: null } }, + }); + bridge.inbound({ + id: 9, + method: "session/fork", + params: { sessionId: "s" }, + }); + expect(bridge.outbound({ id: 9, result: { sessionId: "fork" } })[1]).toMatchObject({ + params: { sessionId: "fork", update: { background: [] } }, + }); + }); + it("leaves the stream untouched when disabled", () => { + const stream = { + readable: new ReadableStream(), + writable: new WritableStream(), + }; + expect(backgroundWorkStream(stream)).toBe(stream); + }); +}); diff --git a/src/v2/agent.ts b/src/v2/agent.ts index 9d1101be..a9b26247 100644 --- a/src/v2/agent.ts +++ b/src/v2/agent.ts @@ -14,6 +14,7 @@ import type { RequestPermissionResponse, } from "@agentclientprotocol/sdk/unstable"; import * as v2 from "@agentclientprotocol/sdk/experimental/v2/unstable"; +import { z } from "zod"; import { ClaudeAcpAgent, type AcpClient, type Logger } from "../acp-agent.js"; import type { AcpSessionNotification } from "../acp-subagents.js"; import type { AcpPermissionRequest } from "../permissions/presentation.js"; @@ -129,7 +130,16 @@ export function v2AgentApp( void client.send({ sessionId: params.sessionId, update }); }), ) - .onNotification(v2.methods.agent.session.cancel, ({ params }) => agent.cancel(params)); + .onNotification(v2.methods.agent.session.cancel, ({ params }) => agent.cancel(params)) + .onRequest( + "_session/async_task/stop", + z.object({ sessionId: z.string(), asyncTaskId: z.string() }), + ({ params }) => { + if (process.env.ACP_BACKGROUND_WORK_EXPERIMENT !== "1") + throw new Error("Background work experiment is disabled"); + return agent.stopAsyncTask(params); + }, + ); } /** The `state_update` that reports `state` as the current foreground state. */ diff --git a/src/v2/session-update.ts b/src/v2/session-update.ts index 6de793f9..e41d2b35 100644 --- a/src/v2/session-update.ts +++ b/src/v2/session-update.ts @@ -76,6 +76,15 @@ export function v2SessionUpdate( // `configOptions` of a `session/set_config_option` response. return undefined; default: + if ( + process.env.ACP_BACKGROUND_WORK_EXPERIMENT === "1" && + update.sessionUpdate.startsWith("async_task_") + ) { + return { + ...update, + sessionUpdate: `_${update.sessionUpdate}`, + } as unknown as v2.SessionUpdate; + } throw new Error( `The ACP v2 surface does not translate ${update.sessionUpdate} session updates yet`, );