chore: prototype opt-in RFD 2452 background shell bridge

This commit is contained in:
Codex
2026-10-23 16:21:27 +02:00
parent 59e292e29f
commit 98e7fe3231
7 changed files with 703 additions and 5 deletions

View File

@@ -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.

View File

@@ -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

View File

@@ -0,0 +1,362 @@
/** Opt-in wire prototype for RFD 2452 over the installed native task adapters. */
type Json = Record<string, unknown>;
type Task = {
id: string;
toolCallId?: string;
canStop: boolean;
final: boolean;
supported: boolean;
};
type Session = {
state: string;
known: boolean;
level?: Set<string>;
tasks: Map<string, Task>;
tools: Set<string>;
finalTools: Set<string>;
children: Set<string>;
};
function object(value: unknown): Json {
return value !== null && typeof value === "object" && !Array.isArray(value)
? (value as Json)
: {};
}
export class BackgroundWorkExperiment {
private sessions = new Map<string, Session>();
private requests = new Map<unknown, { method: string; sessionId?: string }>();
private cancellations = new Set<unknown>();
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<T>(stream: {
readable: ReadableStream<T>;
writable: WritableStream<T>;
}): 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<T, T>({
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<T>({
write: (message) => send(experiment.outbound(object(message))),
close: () => writer.close(),
abort: (reason) => writer.abort(reason),
}),
};
}

View File

@@ -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,

View File

@@ -0,0 +1,244 @@
import { describe, expect, it } from "vitest";
import { BackgroundWorkExperiment, backgroundWorkStream } from "../background-work-experiment.js";
const update = (value: Record<string, unknown>, 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);
});
});

View File

@@ -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. */

View File

@@ -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`,
);