mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-24 10:12:44 +00:00
Ni Xiaoandnistream
d6fa231894
ACP agents replay a loaded session's session/update notifications before session/load resolves, but AcpRuntime.open registered nativeSessionId only after the response, so the session filter dropped every replayed update and restored sessions opened with an empty timeline. Register the native id before the request, drain the notification lane once the response arrives so the replay completes before session.opened, and follow the open request's history contract: replayed updates rebuild the timeline under history "replay" and are dropped under history "skip". Replayed user_message_chunk updates now restore user messages through the chunk fallback scheme, so multi-chunk messages keep their boundaries; live user chunks stay dropped because admitPrompt already echoes the prompt. Co-authored-by: nistream <nistream@users.noreply.github.com>
1687 lines
61 KiB
TypeScript
1687 lines
61 KiB
TypeScript
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import path from "node:path";
|
|
import {
|
|
AgentSideConnection,
|
|
PROTOCOL_VERSION,
|
|
type Agent,
|
|
type SessionUpdate,
|
|
} from "@agentclientprotocol/sdk";
|
|
import type { ProviderConnection, ProviderEvent, ProviderInput } from "./provider.js";
|
|
import { afterEach, describe, expect, it } from "vitest";
|
|
import { runAcpProvider, type AcpStreamMessage } from "./acp.js";
|
|
|
|
const temporaryDirectories: string[] = [];
|
|
|
|
afterEach(async () => {
|
|
await Promise.all(
|
|
temporaryDirectories.splice(0).map((directory) => rm(directory, { recursive: true })),
|
|
);
|
|
});
|
|
|
|
async function fakeAcp(source: string): Promise<string> {
|
|
const directory = await mkdtemp(path.join(tmpdir(), "paseo-acp-test-"));
|
|
temporaryDirectories.push(directory);
|
|
const executable = path.join(directory, "agent.cjs");
|
|
await writeFile(executable, source, "utf8");
|
|
return executable;
|
|
}
|
|
|
|
async function connect(
|
|
executable: string,
|
|
capabilities: string[],
|
|
transformers?: Parameters<typeof runAcpProvider>[0]["transformers"],
|
|
): Promise<{ connection: ProviderConnection; events: ProviderEvent[] }> {
|
|
const registration = runAcpProvider({
|
|
id: "test-acp",
|
|
label: "Test ACP",
|
|
command: [process.execPath, executable],
|
|
transformers,
|
|
});
|
|
const connection = await registration.connect({ versions: [1], capabilities });
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
return { connection, events };
|
|
}
|
|
|
|
function openInput(
|
|
settings: Record<string, string> = {},
|
|
): Extract<ProviderInput, { type: "session.open" }> {
|
|
return {
|
|
type: "session.open",
|
|
requestId: "open-1",
|
|
sessionId: "session-1",
|
|
config: { cwd: "/repo", env: {}, mcpServers: {}, settings, persist: true },
|
|
history: "skip",
|
|
};
|
|
}
|
|
|
|
async function waitForEvent(
|
|
events: ProviderEvent[],
|
|
predicate: (event: ProviderEvent) => boolean,
|
|
): Promise<ProviderEvent> {
|
|
await expect.poll(() => events.find(predicate), { timeout: 3_000 }).toBeTruthy();
|
|
return events.find(predicate)!;
|
|
}
|
|
|
|
type AcpRequestMessage = Extract<AcpStreamMessage, { id: string | number | null; method: string }>;
|
|
|
|
interface ConnectorInstance {
|
|
readonly requests: AcpRequestMessage[];
|
|
readsCanceled: boolean;
|
|
writesClosed: boolean;
|
|
closeFromAgent(): void;
|
|
request(method: string, params: unknown): Promise<unknown>;
|
|
respond(request: AcpRequestMessage, result: unknown): void;
|
|
reject(request: AcpRequestMessage, message: string): void;
|
|
notify(method: string, params: unknown): void;
|
|
}
|
|
|
|
interface ConnectorResponse {
|
|
resolve(value: unknown): void;
|
|
reject(error: Error): void;
|
|
}
|
|
|
|
interface ConnectorHarnessOptions {
|
|
capabilities?: Record<string, unknown>;
|
|
configOptions?: unknown[];
|
|
/** Session updates the agent streams for a prompt (keyed by its text) before ending the turn. */
|
|
promptUpdates?(promptText: string): unknown[];
|
|
handleMessage?(instance: ConnectorInstance, message: AcpStreamMessage): boolean;
|
|
handleRequest?(instance: ConnectorInstance, request: AcpRequestMessage): boolean;
|
|
}
|
|
|
|
function countRequests(instance: ConnectorInstance, method: string): number {
|
|
return instance.requests.filter((request) => request.method === method).length;
|
|
}
|
|
|
|
function receiveConnectorMessage(
|
|
instance: ConnectorInstance,
|
|
message: AcpStreamMessage,
|
|
options?: ConnectorHarnessOptions,
|
|
): void {
|
|
if (!("method" in message)) return;
|
|
if ("id" in message) instance.requests.push(message as AcpRequestMessage);
|
|
if (options?.handleMessage?.(instance, message)) return;
|
|
if (!("id" in message)) return;
|
|
const request = message as AcpRequestMessage;
|
|
if (options?.handleRequest?.(instance, request)) return;
|
|
respondToConnectorRequest(instance, request, options);
|
|
}
|
|
|
|
function respondToConnectorRequest(
|
|
instance: ConnectorInstance,
|
|
request: AcpRequestMessage,
|
|
options?: ConnectorHarnessOptions,
|
|
): void {
|
|
if (request.method === "initialize") {
|
|
instance.respond(request, {
|
|
protocolVersion: request.params
|
|
? (request.params as { protocolVersion: number }).protocolVersion
|
|
: PROTOCOL_VERSION,
|
|
agentCapabilities: options?.capabilities ?? {},
|
|
});
|
|
} else if (request.method === "session/new" || request.method === "session/load") {
|
|
instance.respond(request, {
|
|
sessionId: "connector-session",
|
|
modes: null,
|
|
configOptions: options?.configOptions ?? [],
|
|
});
|
|
} else if (request.method === "session/prompt") {
|
|
const { prompt } = request.params as { prompt: Array<{ type: string; text?: string }> };
|
|
const promptText = prompt.find((part) => part.type === "text")?.text ?? "";
|
|
for (const update of options?.promptUpdates?.(promptText) ?? []) {
|
|
instance.notify("session/update", { sessionId: "connector-session", update });
|
|
}
|
|
instance.respond(request, { stopReason: "end_turn" });
|
|
} else if (request.method === "session/close") {
|
|
instance.respond(request, {});
|
|
} else if (request.method === "session/set_config_option") {
|
|
const params = request.params as { configId: string; value: string | boolean };
|
|
instance.respond(request, {
|
|
configOptions: (options?.configOptions ?? []).map((option) =>
|
|
(option as { id?: string }).id === params.configId
|
|
? Object.assign({}, option, { currentValue: params.value })
|
|
: option,
|
|
),
|
|
});
|
|
}
|
|
}
|
|
|
|
function connectorHarness(options?: ConnectorHarnessOptions) {
|
|
const instances: ConnectorInstance[] = [];
|
|
const connector = () => {
|
|
let controller!: ReadableStreamDefaultController<AcpStreamMessage>;
|
|
let requestSequence = 0;
|
|
const responses = new Map<number, ConnectorResponse>();
|
|
const instance: ConnectorInstance = {
|
|
requests: [],
|
|
readsCanceled: false,
|
|
writesClosed: false,
|
|
closeFromAgent: () => controller.close(),
|
|
request: (method, params) => {
|
|
const response = new Promise<unknown>((resolve, reject) => {
|
|
requestSequence += 1;
|
|
responses.set(requestSequence, { resolve, reject });
|
|
controller.enqueue({ jsonrpc: "2.0", id: requestSequence, method, params });
|
|
});
|
|
void response.catch(() => undefined);
|
|
return response;
|
|
},
|
|
respond: (request, result) => controller.enqueue({ jsonrpc: "2.0", id: request.id, result }),
|
|
reject: (request, message) =>
|
|
controller.enqueue({ jsonrpc: "2.0", id: request.id, error: { code: -32603, message } }),
|
|
notify: (method, params) => controller.enqueue({ jsonrpc: "2.0", method, params }),
|
|
};
|
|
instances.push(instance);
|
|
const readable = new ReadableStream<AcpStreamMessage>({
|
|
start(value) {
|
|
controller = value;
|
|
},
|
|
cancel() {
|
|
instance.readsCanceled = true;
|
|
},
|
|
});
|
|
const writable = new WritableStream<AcpStreamMessage>({
|
|
write(message) {
|
|
if (!("method" in message)) {
|
|
if (typeof message.id !== "number") return;
|
|
const response = responses.get(message.id);
|
|
if (!response) return;
|
|
responses.delete(message.id);
|
|
if ("error" in message) {
|
|
response.reject(
|
|
new Error(
|
|
message.error.data === undefined
|
|
? message.error.message
|
|
: `${message.error.message}: ${JSON.stringify(message.error.data)}`,
|
|
),
|
|
);
|
|
} else response.resolve(message.result);
|
|
return;
|
|
}
|
|
receiveConnectorMessage(instance, message, options);
|
|
},
|
|
close() {
|
|
instance.writesClosed = true;
|
|
},
|
|
abort() {
|
|
instance.writesClosed = true;
|
|
},
|
|
});
|
|
return { readable, writable };
|
|
};
|
|
return { connector, instances };
|
|
}
|
|
|
|
const basicAgent = `const readline = require("node:readline");
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
if (message.method === "initialize") send({ jsonrpc: "2.0", id: message.id, result: { protocolVersion: message.params.protocolVersion, agentCapabilities: {} } });
|
|
else if (message.method === "session/new") setTimeout(() => send({ jsonrpc: "2.0", id: message.id, result: { sessionId: "native-1", modes: null, configOptions: [] } }), 150);
|
|
else if (message.method === "session/prompt") send({ jsonrpc: "2.0", id: message.id, result: { stopReason: "end_turn" } });
|
|
});`;
|
|
|
|
describe("runAcpProvider", () => {
|
|
it("spawns the daemon-resolved launch for probes, catalogues, listing, and sessions", async () => {
|
|
const executable = await fakeAcp(`
|
|
if (process.env.LAUNCH_TOKEN !== "resolved" || process.env.CLAUDECODE || process.env.PASEO_NODE_ENV) process.exit(1);
|
|
const readline = require("node:readline");
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
let result;
|
|
if (message.method === "initialize") result = { protocolVersion: message.params.protocolVersion, agentCapabilities: { sessionCapabilities: { list: {} } } };
|
|
else if (message.method === "session/new") result = { sessionId: "native", configOptions: [{ id: "model", name: "Model", category: "model", type: "select", currentValue: "resolved", options: [{ value: "resolved", name: process.env.SESSION_TOKEN || "probe" }] }] };
|
|
else if (message.method === "session/list") result = { sessions: [] };
|
|
else return;
|
|
send({ jsonrpc: "2.0", id: message.id, result });
|
|
});`);
|
|
const registration = runAcpProvider({
|
|
id: "launch-acp",
|
|
label: "Launch ACP",
|
|
command: ["missing-default-command"],
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message", "session.list"],
|
|
launch: { command: process.execPath, args: [executable], env: { LAUNCH_TOKEN: "resolved" } },
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
try {
|
|
await connection.send({ type: "catalog", requestId: "catalog" });
|
|
expect(await waitForEvent(events, (event) => event.type === "catalog")).toMatchObject({
|
|
catalog: { models: [{ id: "resolved", label: "probe" }] },
|
|
});
|
|
await connection.send({ type: "sessions", requestId: "list" });
|
|
expect(await waitForEvent(events, (event) => event.type === "sessions")).toMatchObject({
|
|
sessions: [],
|
|
});
|
|
const input = openInput();
|
|
input.config.env = { SESSION_TOKEN: "session" };
|
|
await connection.send(input);
|
|
expect(await waitForEvent(events, (event) => event.type === "session.config")).toMatchObject({
|
|
config: { models: [{ id: "resolved", label: "session" }] },
|
|
});
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
});
|
|
|
|
it("still reports initialization failures for workspace-only global discovery", async () => {
|
|
let unavailable = false;
|
|
const { connector } = connectorHarness({
|
|
handleRequest(instance, request) {
|
|
if (unavailable && request.method === "initialize") {
|
|
instance.reject(request, "Bridge unavailable");
|
|
return true;
|
|
}
|
|
return false;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "workspace-acp",
|
|
label: "Workspace ACP",
|
|
connector,
|
|
acpOptions: { workspaceOnlyCatalog: true },
|
|
});
|
|
const connection = await registration.connect({ versions: [1], capabilities: [] });
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
try {
|
|
unavailable = true;
|
|
await connection.send({ type: "catalog", requestId: "global" });
|
|
expect(await waitForEvent(events, (event) => event.type === "request.failed")).toMatchObject({
|
|
requestId: "global",
|
|
error: { message: expect.stringContaining("Bridge unavailable") },
|
|
});
|
|
expect(events.some((event) => event.type === "catalog")).toBe(false);
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
});
|
|
|
|
it("rejects connection cleanly when the ACP executable does not exist", async () => {
|
|
const registration = runAcpProvider({
|
|
id: "missing-acp",
|
|
label: "Missing ACP",
|
|
command: [path.join(tmpdir(), `missing-acp-${process.pid}`)],
|
|
acpOptions: { startupTimeoutMs: 250 },
|
|
});
|
|
|
|
await expect(
|
|
registration.connect({ versions: [1], capabilities: ["prompt.message"] }),
|
|
).rejects.toThrow(/ENOENT|spawn/u);
|
|
});
|
|
|
|
it("accepts an official ACP SDK stream connector instead of a command", async () => {
|
|
const connector = () => {
|
|
const clientToAgent = new TransformStream();
|
|
const agentToClient = new TransformStream();
|
|
const agent: Agent = {
|
|
initialize: async () => ({ protocolVersion: PROTOCOL_VERSION, agentCapabilities: {} }),
|
|
newSession: async () => ({ sessionId: "sdk-session", modes: null, configOptions: [] }),
|
|
prompt: async () => ({ stopReason: "end_turn" }),
|
|
cancel: async () => undefined,
|
|
};
|
|
const agentConnection = new AgentSideConnection(() => agent, {
|
|
writable: agentToClient.writable,
|
|
readable: clientToAgent.readable,
|
|
});
|
|
void agentConnection;
|
|
return {
|
|
writable: clientToAgent.writable,
|
|
readable: agentToClient.readable,
|
|
};
|
|
};
|
|
const registration = runAcpProvider({
|
|
id: "sdk-acp",
|
|
label: "SDK ACP",
|
|
connector,
|
|
});
|
|
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
expect(connection.capabilities).toEqual(["prompt.message"]);
|
|
await connection.close();
|
|
});
|
|
|
|
it("maps permission behavior to the first matching ACP option", async () => {
|
|
const harness = connectorHarness();
|
|
const registration = runAcpProvider({
|
|
id: "sdk-acp",
|
|
label: "SDK ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message", "permission"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
const allow = harness.instances[1]!.request("session/request_permission", {
|
|
sessionId: "connector-session",
|
|
toolCall: {
|
|
toolCallId: "tool-allow",
|
|
title: "Allow tool",
|
|
kind: "execute",
|
|
status: "pending",
|
|
content: [],
|
|
rawInput: {},
|
|
},
|
|
options: [
|
|
{ optionId: "allow-always", name: "Always allow", kind: "allow_always" },
|
|
{ optionId: "allow-once", name: "Allow once", kind: "allow_once" },
|
|
{ optionId: "reject-once", name: "Reject", kind: "reject_once" },
|
|
],
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.permission" && event.request.id === "permission:tool-allow",
|
|
);
|
|
await connection.send({
|
|
type: "session.permission",
|
|
sessionId: "session-1",
|
|
permissionId: "permission:tool-allow",
|
|
response: { behavior: "allow" },
|
|
});
|
|
await expect(allow).resolves.toEqual({
|
|
outcome: { outcome: "selected", optionId: "allow-always" },
|
|
});
|
|
|
|
const deny = harness.instances[1]!.request("session/request_permission", {
|
|
sessionId: "connector-session",
|
|
toolCall: {
|
|
toolCallId: "tool-deny",
|
|
title: "Deny tool",
|
|
kind: "execute",
|
|
status: "pending",
|
|
content: [],
|
|
rawInput: {},
|
|
},
|
|
options: [
|
|
{ optionId: "allow-once", name: "Allow", kind: "allow_once" },
|
|
{ optionId: "reject-always", name: "Always reject", kind: "reject_always" },
|
|
{ optionId: "reject-once", name: "Reject once", kind: "reject_once" },
|
|
],
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) => event.type === "session.permission" && event.request.id === "permission:tool-deny",
|
|
);
|
|
await connection.send({
|
|
type: "session.permission",
|
|
sessionId: "session-1",
|
|
permissionId: "permission:tool-deny",
|
|
response: { behavior: "deny" },
|
|
});
|
|
await expect(deny).resolves.toEqual({
|
|
outcome: { outcome: "selected", optionId: "reject-always" },
|
|
});
|
|
await connection.close();
|
|
});
|
|
|
|
it("rejects an explicit ACP permission option with the wrong behavior", async () => {
|
|
const harness = connectorHarness();
|
|
const registration = runAcpProvider({
|
|
id: "sdk-acp",
|
|
label: "SDK ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message", "permission"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
const permission = harness.instances[1]!.request("session/request_permission", {
|
|
sessionId: "connector-session",
|
|
toolCall: {
|
|
toolCallId: "tool-explicit",
|
|
title: "Explicit tool",
|
|
kind: "execute",
|
|
status: "pending",
|
|
content: [],
|
|
rawInput: {},
|
|
},
|
|
options: [
|
|
{ optionId: "allow-once", name: "Allow", kind: "allow_once" },
|
|
{ optionId: "reject-once", name: "Reject", kind: "reject_once" },
|
|
],
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.permission" && event.request.id === "permission:tool-explicit",
|
|
);
|
|
await connection.send({
|
|
type: "session.permission",
|
|
sessionId: "session-1",
|
|
permissionId: "permission:tool-explicit",
|
|
response: { behavior: "allow", selectedActionId: "missing" },
|
|
});
|
|
await expect(
|
|
waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.runtime_failed" && event.error.message.includes("'missing'"),
|
|
),
|
|
).resolves.toBeDefined();
|
|
|
|
await connection.send({
|
|
type: "session.permission",
|
|
sessionId: "session-1",
|
|
permissionId: "permission:tool-explicit",
|
|
response: { behavior: "allow", selectedActionId: "reject-once" },
|
|
});
|
|
await expect(
|
|
waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.runtime_failed" &&
|
|
event.error.message.includes("'reject-once'") &&
|
|
event.error.message.includes("does not match 'allow' behavior"),
|
|
),
|
|
).resolves.toBeDefined();
|
|
|
|
await connection.send({
|
|
type: "session.permission",
|
|
sessionId: "session-1",
|
|
permissionId: "permission:tool-explicit",
|
|
response: { behavior: "allow", selectedActionId: "allow-once" },
|
|
});
|
|
await expect(permission).resolves.toEqual({
|
|
outcome: { outcome: "selected", optionId: "allow-once" },
|
|
});
|
|
await connection.close();
|
|
});
|
|
|
|
it("closes both connector directions for the probe and live session", async () => {
|
|
const harness = connectorHarness();
|
|
const registration = runAcpProvider({
|
|
id: "owned-acp",
|
|
label: "Owned ACP",
|
|
connector: harness.connector,
|
|
});
|
|
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
expect(harness.instances).toHaveLength(1);
|
|
expect(harness.instances[0]).toMatchObject({ readsCanceled: true, writesClosed: true });
|
|
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
await connection.close();
|
|
|
|
expect(harness.instances).toHaveLength(2);
|
|
expect(harness.instances[1]).toMatchObject({ readsCanceled: true, writesClosed: true });
|
|
});
|
|
|
|
it("publishes runtime failure on unexpected connector EOF", async () => {
|
|
const harness = connectorHarness();
|
|
const registration = runAcpProvider({
|
|
id: "eof-acp",
|
|
label: "EOF ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
harness.instances[1]!.closeFromAgent();
|
|
|
|
await expect(
|
|
waitForEvent(events, (event) => event.type === "session.runtime_failed"),
|
|
).resolves.toMatchObject({
|
|
sessionId: "session-1",
|
|
error: { message: expect.stringContaining("closed") },
|
|
});
|
|
await connection.close();
|
|
});
|
|
|
|
it("settles an active auto prompt before admitting its auto replacement", async () => {
|
|
const order: string[] = [];
|
|
let firstPrompt: AcpRequestMessage | null = null;
|
|
let activePrompts = 0;
|
|
let maximumActivePrompts = 0;
|
|
const harness = connectorHarness({
|
|
handleMessage(instance, message) {
|
|
if (!("method" in message)) return false;
|
|
const request = "id" in message ? (message as AcpRequestMessage) : null;
|
|
if (message.method === "session/prompt" && request) {
|
|
const promptNumber = firstPrompt ? "second" : "first";
|
|
order.push(`prompt:${promptNumber}`);
|
|
activePrompts += 1;
|
|
maximumActivePrompts = Math.max(maximumActivePrompts, activePrompts);
|
|
if (!firstPrompt) {
|
|
firstPrompt = request;
|
|
} else {
|
|
activePrompts -= 1;
|
|
instance.respond(request, { stopReason: "end_turn" });
|
|
}
|
|
return true;
|
|
}
|
|
if (message.method === "session/cancel") {
|
|
order.push("cancel");
|
|
activePrompts -= 1;
|
|
instance.respond(firstPrompt!, { stopReason: "cancelled" });
|
|
return true;
|
|
}
|
|
return false;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "overlap-acp",
|
|
label: "Overlap ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
await connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId: "first",
|
|
delivery: "auto",
|
|
input: { type: "message", content: [{ type: "text", text: "first" }] },
|
|
},
|
|
});
|
|
await expect.poll(() => countRequests(harness.instances[1]!, "session/prompt")).toBe(1);
|
|
await connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId: "second",
|
|
delivery: "auto",
|
|
input: { type: "message", content: [{ type: "text", text: "second" }] },
|
|
},
|
|
});
|
|
|
|
await waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.turn" &&
|
|
event.turnId === "acp:second" &&
|
|
event.state === "completed",
|
|
);
|
|
expect(order).toEqual(["prompt:first", "cancel", "prompt:second"]);
|
|
expect(maximumActivePrompts).toBe(1);
|
|
const firstTerminal = events.findIndex(
|
|
(event) =>
|
|
event.type === "session.turn" && event.turnId === "acp:first" && event.state === "canceled",
|
|
);
|
|
const secondAdmission = events.findIndex(
|
|
(event) => event.type === "session.prompt_result" && event.clientMessageId === "second",
|
|
);
|
|
expect(firstTerminal).toBeGreaterThan(-1);
|
|
expect(firstTerminal).toBeLessThan(secondAdmission);
|
|
await connection.close();
|
|
});
|
|
|
|
it.each([
|
|
["end_turn", "completed", "completed"],
|
|
["cancelled", "canceled", "canceled"],
|
|
] as const)(
|
|
"closes a transformer's running tool row when the turn ends with %s",
|
|
async (stopReason, turnState, rowStatus) => {
|
|
const harness = connectorHarness({
|
|
handleMessage(instance, message) {
|
|
if (!("method" in message) || !("id" in message)) return false;
|
|
if (message.method !== "session/prompt") return false;
|
|
instance.notify("vendor/phase", { id: "phase-1", label: "Preparing" });
|
|
instance.respond(message as AcpRequestMessage, { stopReason });
|
|
return true;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "vendor-row-acp",
|
|
label: "Vendor row ACP",
|
|
connector: harness.connector,
|
|
transformers: [
|
|
{
|
|
notification(notification) {
|
|
if (notification.method !== "vendor/phase") return null;
|
|
return {
|
|
type: "timeline",
|
|
item: {
|
|
type: "tool_call",
|
|
id: "phase-1",
|
|
callId: "phase-1",
|
|
name: "Vendor phase",
|
|
detail: { type: "plain_text", label: "Preparing" },
|
|
status: "running",
|
|
error: null,
|
|
},
|
|
};
|
|
},
|
|
},
|
|
],
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
await connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId: "first",
|
|
delivery: "auto",
|
|
input: { type: "message", content: [{ type: "text", text: "first" }] },
|
|
},
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) => event.type === "session.turn" && event.state === turnState,
|
|
);
|
|
|
|
const rows = events.flatMap((event) =>
|
|
event.type === "timeline.item" && event.item.id === "phase-1" ? [event.item] : [],
|
|
);
|
|
expect(rows.map((row) => row.type === "tool_call" && row.status)).toEqual([
|
|
"running",
|
|
rowStatus,
|
|
]);
|
|
const closed = events.findIndex(
|
|
(event) =>
|
|
event.type === "timeline.item" &&
|
|
event.item.id === "phase-1" &&
|
|
event.item.type === "tool_call" &&
|
|
event.item.status === rowStatus,
|
|
);
|
|
const terminal = events.findIndex(
|
|
(event) => event.type === "session.turn" && event.state === turnState,
|
|
);
|
|
expect(closed).toBeLessThan(terminal);
|
|
// The wire contract only accepts a failed tool call with a non-null error.
|
|
expect(rows.at(-1)).toMatchObject({ error: null });
|
|
await connection.close();
|
|
},
|
|
);
|
|
|
|
it("cancels an in-progress ACP tool call with the canceled turn instead of failing it without an error", async () => {
|
|
const harness = connectorHarness({
|
|
handleMessage(instance, message) {
|
|
if (!("method" in message) || !("id" in message)) return false;
|
|
if (message.method !== "session/prompt") return false;
|
|
instance.notify("session/update", {
|
|
sessionId: "connector-session",
|
|
update: {
|
|
sessionUpdate: "tool_call",
|
|
toolCallId: "tool-1",
|
|
title: "Run tests",
|
|
kind: "execute",
|
|
status: "in_progress",
|
|
},
|
|
});
|
|
instance.respond(message as AcpRequestMessage, { stopReason: "cancelled" });
|
|
return true;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "cancel-tool-acp",
|
|
label: "Cancel tool ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
await connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId: "first",
|
|
delivery: "auto",
|
|
input: { type: "message", content: [{ type: "text", text: "first" }] },
|
|
},
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) => event.type === "session.turn" && event.state === "canceled",
|
|
);
|
|
|
|
const rows = events.flatMap((event) =>
|
|
event.type === "timeline.item" && event.item.id === "tool-1" ? [event.item] : [],
|
|
);
|
|
expect(rows.at(-1)).toMatchObject({ type: "tool_call", status: "canceled", error: null });
|
|
await connection.close();
|
|
});
|
|
|
|
it("serializes configuration mutations", async () => {
|
|
let releaseMutation: (() => void) | null = null;
|
|
const configOptions = [
|
|
{
|
|
id: "model",
|
|
name: "Model",
|
|
category: "model",
|
|
type: "select",
|
|
currentValue: "a",
|
|
options: [
|
|
{ name: "A", value: "a" },
|
|
{ name: "B", value: "b" },
|
|
],
|
|
},
|
|
];
|
|
const harness = connectorHarness({
|
|
capabilities: { loadSession: true },
|
|
configOptions,
|
|
handleRequest(instance, request) {
|
|
if (request.method !== "session/set_config_option") return false;
|
|
const value = (request.params as { value: string }).value;
|
|
if (value !== "first") return false;
|
|
releaseMutation = () =>
|
|
instance.respond(request, {
|
|
configOptions: configOptions.map((option) =>
|
|
Object.assign({}, option, { currentValue: value }),
|
|
),
|
|
});
|
|
return true;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "lane-acp",
|
|
label: "Lane ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message", "session.configure"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
await connection.send({
|
|
type: "session.configure",
|
|
requestId: "configure-first",
|
|
sessionId: "session-1",
|
|
changes: { model: "first" },
|
|
});
|
|
await connection.send({
|
|
type: "session.configure",
|
|
requestId: "configure-second",
|
|
sessionId: "session-1",
|
|
changes: { model: "second" },
|
|
});
|
|
await expect
|
|
.poll(() => countRequests(harness.instances[1]!, "session/set_config_option"))
|
|
.toBe(1);
|
|
releaseMutation!();
|
|
await waitForEvent(
|
|
events,
|
|
(event) => event.type === "request.completed" && event.requestId === "configure-second",
|
|
);
|
|
expect(
|
|
events
|
|
.filter(
|
|
(event) => event.type === "request.completed" && event.requestId.startsWith("configure-"),
|
|
)
|
|
.map((event) => (event as Extract<ProviderEvent, { type: "request.completed" }>).requestId),
|
|
).toEqual(["configure-first", "configure-second"]);
|
|
|
|
await connection.close();
|
|
});
|
|
|
|
it("does not claim exact permission policy, listing, image, or configuration support", async () => {
|
|
const executable = await fakeAcp(basicAgent);
|
|
const { connection, events } = await connect(executable, [
|
|
"prompt.message",
|
|
"prompt.image",
|
|
"session.configure",
|
|
"session.list",
|
|
"permission.tool_policy",
|
|
]);
|
|
|
|
expect(connection.capabilities).toEqual(["prompt.message", "session.configure"]);
|
|
await connection.send(openInput());
|
|
const opened = await waitForEvent(events, (event) => event.type === "session.opened");
|
|
expect(opened).toMatchObject({ capabilities: ["prompt.message"] });
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
await expect(
|
|
connection.send({
|
|
type: "session.configure",
|
|
requestId: "configure-unsupported",
|
|
sessionId: "session-1",
|
|
changes: { model: "not-advertised" },
|
|
}),
|
|
).rejects.toThrow("session.configure");
|
|
await connection.close();
|
|
});
|
|
|
|
it("derives optional connection capabilities from ACP initialization", async () => {
|
|
const executable = await fakeAcp(`const readline = require("node:readline");
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
if (message.method === "initialize") send({ jsonrpc: "2.0", id: message.id, result: {
|
|
protocolVersion: message.params.protocolVersion,
|
|
agentCapabilities: {
|
|
loadSession: true,
|
|
promptCapabilities: { image: true },
|
|
sessionCapabilities: { list: {} },
|
|
},
|
|
} });
|
|
});`);
|
|
const { connection } = await connect(executable, [
|
|
"prompt.message",
|
|
"prompt.image",
|
|
"session.list",
|
|
"session.persistence",
|
|
]);
|
|
|
|
expect(connection.capabilities).toEqual([
|
|
"prompt.message",
|
|
"prompt.image",
|
|
"session.list",
|
|
"session.persistence",
|
|
]);
|
|
await connection.close();
|
|
});
|
|
|
|
it("acknowledges admission before session creation completes", async () => {
|
|
const executable = await fakeAcp(basicAgent);
|
|
const { connection, events } = await connect(executable, ["prompt.message"]);
|
|
let ready = false;
|
|
const admission = connection.send(openInput()).then(() => {
|
|
ready = true;
|
|
return undefined;
|
|
});
|
|
|
|
await admission;
|
|
expect(ready).toBe(true);
|
|
expect(events).toEqual([]);
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
await connection.close();
|
|
});
|
|
|
|
it("rejects provider inputs after the ACP connection closes", async () => {
|
|
const harness = connectorHarness();
|
|
const registration = runAcpProvider({
|
|
id: "closed-acp",
|
|
label: "Closed ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({ versions: [1], capabilities: [] });
|
|
|
|
await connection.close();
|
|
await expect(connection.send({ type: "catalog", requestId: "too-late" })).rejects.toThrow(
|
|
"ACP provider connection is closed",
|
|
);
|
|
expect(harness.instances).toHaveLength(1);
|
|
});
|
|
|
|
it("waits for delayed session creation and closes its connector after connection close", async () => {
|
|
let delayedNew: AcpRequestMessage | null = null;
|
|
const harness = connectorHarness({
|
|
handleRequest(_instance, request) {
|
|
if (request.method !== "session/new") return false;
|
|
delayedNew = request;
|
|
return true;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "delayed-open-acp",
|
|
label: "Delayed open ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await expect.poll(() => delayedNew).not.toBeNull();
|
|
|
|
let closeSettled = false;
|
|
const closing = connection.close().then(() => {
|
|
closeSettled = true;
|
|
return undefined;
|
|
});
|
|
await Promise.resolve();
|
|
expect(closeSettled).toBe(false);
|
|
harness.instances[1]!.respond(delayedNew!, {
|
|
sessionId: "late-session",
|
|
modes: null,
|
|
configOptions: [],
|
|
});
|
|
await closing;
|
|
|
|
expect(harness.instances[1]).toMatchObject({ readsCanceled: true, writesClosed: true });
|
|
expect(events).toEqual([]);
|
|
});
|
|
|
|
it("refuses unsupported steering before sending an ACP prompt", async () => {
|
|
const executable = await fakeAcp(basicAgent);
|
|
const { connection, events } = await connect(executable, ["prompt.message"]);
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
await expect(
|
|
connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId: "steer-1",
|
|
delivery: "steer",
|
|
input: { type: "message", content: [{ type: "text", text: "change course" }] },
|
|
},
|
|
}),
|
|
).rejects.toThrow("prompt.steer");
|
|
expect(events.some((event) => event.type === "session.prompt_result")).toBe(false);
|
|
await connection.close();
|
|
});
|
|
|
|
it("keeps configuration observable state atomic when a later field fails", async () => {
|
|
const executable = await fakeAcp(`const readline = require("node:readline");
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
|
|
let model = "a";
|
|
const options = () => [
|
|
{ id: "model", name: "Model", category: "model", type: "select", currentValue: model, options: [{ name: "A", value: "a" }, { name: "B", value: "b" }] },
|
|
{ id: "verbosity", name: "Verbosity", type: "select", currentValue: "quiet", options: [{ name: "Quiet", value: "quiet" }, { name: "Loud", value: "loud" }] },
|
|
];
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
if (message.method === "initialize") send({ jsonrpc: "2.0", id: message.id, result: { protocolVersion: message.params.protocolVersion, agentCapabilities: {} } });
|
|
else if (message.method === "session/new") send({ jsonrpc: "2.0", id: message.id, result: { sessionId: "native-1", modes: null, configOptions: options() } });
|
|
else if (message.method === "session/set_config_option" && message.params.configId === "verbosity") send({ jsonrpc: "2.0", id: message.id, error: { code: -32602, message: "verbosity rejected" } });
|
|
else if (message.method === "session/set_config_option") { model = message.params.value; send({ jsonrpc: "2.0", id: message.id, result: { configOptions: options() } }); }
|
|
});`);
|
|
const { connection, events } = await connect(executable, [
|
|
"prompt.message",
|
|
"session.configure",
|
|
]);
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
await connection.send({
|
|
type: "session.configure",
|
|
requestId: "configure-1",
|
|
sessionId: "session-1",
|
|
changes: { model: "b", settings: { verbosity: "loud" } },
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) => event.type === "request.failed" && event.requestId === "configure-1",
|
|
);
|
|
const configs = events.filter((event) => event.type === "session.config");
|
|
expect(configs).toHaveLength(1);
|
|
expect(configs[0]).toMatchObject({ config: { model: "a" } });
|
|
await connection.close();
|
|
});
|
|
|
|
it.each([1, 64])(
|
|
"commits transformer config notifications once and discards them on rollback (%i per response)",
|
|
async (notificationCount) => {
|
|
const executable = await fakeAcp(`const readline = require("node:readline");
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (...messages) => process.stdout.write(messages.map((message) => JSON.stringify(message) + "\\n").join(""));
|
|
let model = "a";
|
|
const options = () => [
|
|
{ id: "model", name: "Model", category: "model", type: "select", currentValue: model, options: [{ name: "A", value: "a" }, { name: "B", value: "b" }] },
|
|
{ id: "late", name: "Late", type: "select", currentValue: "ok", options: [{ name: "Ok", value: "ok" }, { name: "Fail", value: "fail" }] },
|
|
];
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
if (message.method === "initialize") send({ jsonrpc: "2.0", id: message.id, result: { protocolVersion: message.params.protocolVersion, agentCapabilities: {} } });
|
|
else if (message.method === "session/new") send({ jsonrpc: "2.0", id: message.id, result: { sessionId: "native-1", modes: null, configOptions: options() } });
|
|
else if (message.method === "session/set_config_option" && message.params.configId === "late") send({ jsonrpc: "2.0", id: message.id, error: { code: -32602, message: "late rejected" } });
|
|
else if (message.method === "session/set_config_option") {
|
|
model = message.params.value;
|
|
send(
|
|
...Array.from({ length: ${notificationCount} }, () => ({ jsonrpc: "2.0", method: "vendor/config", params: { model } })),
|
|
{ jsonrpc: "2.0", id: message.id, result: { configOptions: options() } },
|
|
);
|
|
}
|
|
});`);
|
|
const transformer = {
|
|
notification(notification: { method: string; params: unknown }) {
|
|
if (notification.method !== "vendor/config") return null;
|
|
const params = notification.params as { model: string };
|
|
return {
|
|
type: "config" as const,
|
|
config: {
|
|
model: params.model,
|
|
models: [{ id: params.model, label: `Transformer ${params.model}` }],
|
|
modes: [],
|
|
thinkingOptions: [],
|
|
settings: [],
|
|
},
|
|
};
|
|
},
|
|
};
|
|
const { connection, events } = await connect(
|
|
executable,
|
|
["prompt.message", "session.configure"],
|
|
[transformer],
|
|
);
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
const initialConfigCount = events.filter((event) => event.type === "session.config").length;
|
|
|
|
await connection.send({
|
|
type: "session.configure",
|
|
requestId: "transformer-initial-failure",
|
|
sessionId: "session-1",
|
|
changes: { model: "b", settings: { late: "fail" } },
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "request.failed" && event.requestId === "transformer-initial-failure",
|
|
);
|
|
expect(events.filter((event) => event.type === "session.config")).toHaveLength(
|
|
initialConfigCount,
|
|
);
|
|
|
|
await connection.send({
|
|
type: "session.configure",
|
|
requestId: "transformer-success",
|
|
sessionId: "session-1",
|
|
changes: { model: "b" },
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) => event.type === "request.completed" && event.requestId === "transformer-success",
|
|
);
|
|
let configs = events.filter((event) => event.type === "session.config");
|
|
expect(configs).toHaveLength(initialConfigCount + 1);
|
|
expect(configs.at(-1)).toMatchObject({
|
|
config: { model: "b", models: [{ label: "Transformer b" }] },
|
|
});
|
|
|
|
await connection.send({
|
|
type: "session.configure",
|
|
requestId: "transformer-failure",
|
|
sessionId: "session-1",
|
|
changes: { model: "a", settings: { late: "fail" } },
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) => event.type === "request.failed" && event.requestId === "transformer-failure",
|
|
);
|
|
configs = events.filter((event) => event.type === "session.config");
|
|
expect(configs).toHaveLength(initialConfigCount + 1);
|
|
expect(configs.at(-1)).toMatchObject({ config: { model: "b" } });
|
|
await connection.close();
|
|
},
|
|
);
|
|
|
|
it("gives discovery transformers the discovered session configuration facade", async () => {
|
|
const executable = await fakeAcp(`const readline = require("node:readline");
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
if (message.method === "initialize") send({ jsonrpc: "2.0", id: message.id, result: { protocolVersion: message.params.protocolVersion, agentCapabilities: {} } });
|
|
else if (message.method === "session/new") send({ jsonrpc: "2.0", id: message.id, result: { sessionId: "catalog-native", modes: null, configOptions: [{ id: "flavor", name: "Flavor", type: "select", currentValue: "mint", options: [{ name: "Mint", value: "mint" }] }] } });
|
|
});`);
|
|
let discovered: Readonly<Record<string, unknown>> | null = null;
|
|
const { connection, events } = await connect(
|
|
executable,
|
|
[],
|
|
[
|
|
{
|
|
async discover(catalog, context) {
|
|
discovered = await context.config.read();
|
|
return catalog;
|
|
},
|
|
},
|
|
],
|
|
);
|
|
|
|
await connection.send({ type: "catalog", requestId: "catalog-1" });
|
|
await waitForEvent(events, (event) => event.type === "catalog");
|
|
expect(discovered).toEqual({ flavor: "mint" });
|
|
await connection.close();
|
|
});
|
|
|
|
it("drains heavy ACP stderr while initializing", async () => {
|
|
const executable = await fakeAcp(
|
|
`process.stderr.write("x".repeat(2 * 1024 * 1024));\n${basicAgent}`,
|
|
);
|
|
const { connection, events } = await connect(executable, ["prompt.message"]);
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
await connection.close();
|
|
});
|
|
|
|
it("publishes runtime failure when the ACP process dies", async () => {
|
|
const executable = await fakeAcp(`const readline = require("node:readline");
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
if (message.method === "initialize") send({ jsonrpc: "2.0", id: message.id, result: { protocolVersion: message.params.protocolVersion, agentCapabilities: {} } });
|
|
else if (message.method === "session/new") {
|
|
send({ jsonrpc: "2.0", id: message.id, result: { sessionId: "native-1", modes: null, configOptions: [] } });
|
|
setTimeout(() => process.exit(17), 20);
|
|
}
|
|
});`);
|
|
const { connection, events } = await connect(executable, ["prompt.message"]);
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
const failure = await waitForEvent(events, (event) => event.type === "session.runtime_failed");
|
|
expect(failure).toMatchObject({
|
|
sessionId: "session-1",
|
|
error: { message: expect.stringContaining("17") },
|
|
});
|
|
await connection.close();
|
|
});
|
|
|
|
it("escalates shutdown when an ACP process ignores SIGTERM", async () => {
|
|
const executable = await fakeAcp(`const readline = require("node:readline");
|
|
process.on("SIGTERM", () => {});
|
|
const lines = readline.createInterface({ input: process.stdin });
|
|
const send = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
|
|
lines.on("line", (line) => {
|
|
const message = JSON.parse(line);
|
|
if (message.method === "initialize") send({ jsonrpc: "2.0", id: message.id, result: { protocolVersion: message.params.protocolVersion, agentCapabilities: {} } });
|
|
else if (message.method === "session/new") send({ jsonrpc: "2.0", id: message.id, result: { sessionId: "native-1", modes: null, configOptions: [] } });
|
|
});`);
|
|
const { connection, events } = await connect(executable, ["prompt.message"]);
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
const startedAt = Date.now();
|
|
await connection.close();
|
|
expect(Date.now() - startedAt).toBeGreaterThanOrEqual(900);
|
|
expect(Date.now() - startedAt).toBeLessThan(2_500);
|
|
});
|
|
});
|
|
|
|
describe("runAcpProvider streamed chunks without messageId (#4699)", () => {
|
|
interface Chunk {
|
|
kind: "agent_message_chunk" | "agent_thought_chunk";
|
|
text: string;
|
|
messageId?: string;
|
|
}
|
|
|
|
it("continues one timeline item per turn instead of one per chunk", async () => {
|
|
const chunksByPrompt: Record<string, Chunk[]> = {
|
|
first: [
|
|
{ kind: "agent_message_chunk", text: "- **Current tem" },
|
|
{ kind: "agent_message_chunk", text: "perature**: 25°C" },
|
|
],
|
|
second: [{ kind: "agent_message_chunk", text: "next turn" }],
|
|
// An explicit-id text chunk between two id-less thoughts separates them.
|
|
third: [
|
|
{ kind: "agent_thought_chunk", text: "think A" },
|
|
{ kind: "agent_message_chunk", text: "say", messageId: "m-explicit" },
|
|
{ kind: "agent_thought_chunk", text: "think B" },
|
|
],
|
|
};
|
|
const harness = connectorHarness({
|
|
promptUpdates: (promptText) =>
|
|
(chunksByPrompt[promptText] ?? []).map((chunk) => {
|
|
const update: Record<string, unknown> = {
|
|
sessionUpdate: chunk.kind,
|
|
content: { type: "text", text: chunk.text },
|
|
};
|
|
if (chunk.messageId) update.messageId = chunk.messageId;
|
|
return update;
|
|
}),
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "chunk-acp",
|
|
label: "Chunk ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
for (const clientMessageId of ["first", "second", "third"]) {
|
|
await connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId,
|
|
delivery: "auto",
|
|
input: { type: "message", content: [{ type: "text", text: clientMessageId }] },
|
|
},
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.turn" &&
|
|
event.turnId === `acp:${clientMessageId}` &&
|
|
event.state === "completed",
|
|
);
|
|
}
|
|
|
|
const items = events.flatMap((event) =>
|
|
event.type === "timeline.item" &&
|
|
(event.item.type === "assistant_message" || event.item.type === "reasoning")
|
|
? [{ id: event.item.id, text: event.item.text }]
|
|
: [],
|
|
);
|
|
expect(items).toEqual([
|
|
{ id: "agent_message_chunk:1", text: "- **Current tem" },
|
|
{ id: "agent_message_chunk:1", text: "- **Current temperature**: 25°C" },
|
|
{ id: "agent_message_chunk:2", text: "next turn" },
|
|
{ id: "agent_thought_chunk:3", text: "think A" },
|
|
{ id: "m-explicit", text: "say" },
|
|
{ id: "agent_thought_chunk:4", text: "think B" },
|
|
]);
|
|
await connection.close();
|
|
});
|
|
});
|
|
|
|
function textChunk(
|
|
text: string,
|
|
kind: "agent_message_chunk" | "agent_thought_chunk" = "agent_message_chunk",
|
|
messageId?: string,
|
|
): SessionUpdate {
|
|
return {
|
|
sessionUpdate: kind,
|
|
content: { type: "text", text },
|
|
...(messageId === undefined ? {} : { messageId }),
|
|
};
|
|
}
|
|
|
|
async function streamChunkTurns(
|
|
updates: SessionUpdate[],
|
|
ending: "completed" | "canceled" | "failed" = "completed",
|
|
) {
|
|
let promptCount = 0;
|
|
const harness = connectorHarness({
|
|
handleRequest(instance, request) {
|
|
if (request.method !== "session/prompt") return false;
|
|
promptCount++;
|
|
for (const update of promptCount === 1 ? updates : [textChunk("Next turn")]) {
|
|
instance.notify("session/update", { sessionId: "connector-session", update });
|
|
}
|
|
if (promptCount === 1 && ending === "failed") instance.reject(request, "Prompt failed");
|
|
else
|
|
instance.respond(request, {
|
|
stopReason: promptCount === 1 && ending === "canceled" ? "cancelled" : "end_turn",
|
|
});
|
|
return true;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "chunk-boundaries",
|
|
label: "Chunk boundaries",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
try {
|
|
await connection.send(openInput());
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
for (const clientMessageId of ["first", "second"]) {
|
|
await connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId,
|
|
delivery: "auto",
|
|
input: { type: "message", content: [{ type: "text", text: clientMessageId }] },
|
|
},
|
|
});
|
|
await waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.turn" &&
|
|
event.turnId === `acp:${clientMessageId}` &&
|
|
event.state === (clientMessageId === "first" ? ending : "completed"),
|
|
);
|
|
}
|
|
const messages = new Map<string, { type: string; text: string }>();
|
|
for (const event of events) {
|
|
if (
|
|
event.type === "timeline.item" &&
|
|
(event.item.type === "assistant_message" || event.item.type === "reasoning")
|
|
) {
|
|
messages.set(event.item.id, { type: event.item.type, text: event.item.text });
|
|
}
|
|
}
|
|
return [...messages.values()];
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
}
|
|
|
|
describe("ACP streamed message boundaries", () => {
|
|
it.each([
|
|
{
|
|
name: "tool call",
|
|
update: {
|
|
sessionUpdate: "tool_call",
|
|
toolCallId: "tool",
|
|
title: "Read",
|
|
kind: "read",
|
|
status: "in_progress",
|
|
},
|
|
},
|
|
{ name: "plan", update: { sessionUpdate: "plan", entries: [] } },
|
|
{
|
|
name: "user message",
|
|
update: { sessionUpdate: "user_message_chunk", content: { type: "text", text: "User" } },
|
|
},
|
|
] satisfies Array<{ name: string; update: SessionUpdate }>)(
|
|
"starts a new message after a $name",
|
|
async ({ update }) => {
|
|
expect(
|
|
await streamChunkTurns([
|
|
textChunk("Before"),
|
|
update,
|
|
textChunk("After "),
|
|
textChunk("boundary"),
|
|
]),
|
|
).toEqual([
|
|
{ type: "assistant_message", text: "Before" },
|
|
{ type: "assistant_message", text: "After boundary" },
|
|
{ type: "assistant_message", text: "Next turn" },
|
|
]);
|
|
},
|
|
);
|
|
|
|
it("continues text while an existing tool receives a status update", async () => {
|
|
expect(
|
|
await streamChunkTurns([
|
|
{
|
|
sessionUpdate: "tool_call",
|
|
toolCallId: "tool",
|
|
title: "Read",
|
|
kind: "read",
|
|
status: "in_progress",
|
|
},
|
|
textChunk("Still "),
|
|
{ sessionUpdate: "tool_call_update", toolCallId: "tool", status: "completed" },
|
|
textChunk("speaking"),
|
|
]),
|
|
).toEqual([
|
|
{ type: "assistant_message", text: "Still speaking" },
|
|
{ type: "assistant_message", text: "Next turn" },
|
|
]);
|
|
});
|
|
|
|
it("keeps explicit messages separate from neighboring anonymous chunks", async () => {
|
|
expect(
|
|
await streamChunkTurns([
|
|
textChunk("Before"),
|
|
textChunk("Explicit ", "agent_message_chunk", "message"),
|
|
textChunk("message", "agent_message_chunk", "message"),
|
|
textChunk("After"),
|
|
]),
|
|
).toEqual([
|
|
{ type: "assistant_message", text: "Before" },
|
|
{ type: "assistant_message", text: "Explicit message" },
|
|
{ type: "assistant_message", text: "After" },
|
|
{ type: "assistant_message", text: "Next turn" },
|
|
]);
|
|
});
|
|
|
|
it("continues reasoning but starts a new segment after nontext assistant content", async () => {
|
|
expect(
|
|
await streamChunkTurns([
|
|
textChunk("Think ", "agent_thought_chunk"),
|
|
textChunk("first", "agent_thought_chunk"),
|
|
{
|
|
sessionUpdate: "agent_message_chunk",
|
|
content: { type: "image", data: "", mimeType: "image/png" },
|
|
},
|
|
textChunk("Then think", "agent_thought_chunk"),
|
|
]),
|
|
).toEqual([
|
|
{ type: "reasoning", text: "Think first" },
|
|
{ type: "reasoning", text: "Then think" },
|
|
{ type: "assistant_message", text: "Next turn" },
|
|
]);
|
|
});
|
|
|
|
it.each(["completed", "canceled", "failed"] as const)(
|
|
"keeps the next turn separate after a %s prompt",
|
|
async (ending) => {
|
|
expect(await streamChunkTurns([textChunk("First "), textChunk("turn")], ending)).toEqual([
|
|
{ type: "assistant_message", text: "First turn" },
|
|
{ type: "assistant_message", text: "Next turn" },
|
|
]);
|
|
},
|
|
);
|
|
});
|
|
|
|
function userChunk(text: string, messageId?: string): SessionUpdate {
|
|
return {
|
|
sessionUpdate: "user_message_chunk",
|
|
content: { type: "text", text },
|
|
...(messageId === undefined ? {} : { messageId }),
|
|
};
|
|
}
|
|
|
|
async function openReplaySession(
|
|
replay: SessionUpdate[],
|
|
history: "replay" | "skip",
|
|
): Promise<{ connection: ProviderConnection; events: ProviderEvent[] }> {
|
|
const harness = connectorHarness({
|
|
capabilities: { loadSession: true },
|
|
handleRequest(instance, request) {
|
|
if (request.method !== "session/load") return false;
|
|
for (const update of replay) {
|
|
instance.notify("session/update", { sessionId: "connector-session", update });
|
|
}
|
|
instance.respond(request, {
|
|
sessionId: "connector-session",
|
|
modes: null,
|
|
configOptions: [],
|
|
});
|
|
return true;
|
|
},
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "replay-acp",
|
|
label: "Replay ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message", "session.persistence"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
const input = openInput();
|
|
input.persistence = { version: 1, data: { sessionId: "connector-session" } };
|
|
input.history = history;
|
|
await connection.send(input);
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
return { connection, events };
|
|
}
|
|
|
|
function timelineProjection(events: ProviderEvent[]): Array<Record<string, string>> {
|
|
return events.flatMap((event) => {
|
|
if (event.type !== "timeline.item") return [];
|
|
const item = event.item;
|
|
if (
|
|
item.type === "user_message" ||
|
|
item.type === "assistant_message" ||
|
|
item.type === "reasoning"
|
|
) {
|
|
return [{ type: item.type, id: item.id, text: item.text }];
|
|
}
|
|
if (item.type === "tool_call") return [{ type: item.type, id: item.id, status: item.status }];
|
|
return [{ type: item.type, id: item.id }];
|
|
});
|
|
}
|
|
|
|
describe("ACP session history replay", () => {
|
|
it("restores the timeline replayed before session/load resolves", async () => {
|
|
const { connection, events } = await openReplaySession(
|
|
[
|
|
userChunk("What is the ", "user-1"),
|
|
userChunk("temperature?", "user-1"),
|
|
textChunk("Checking the sensor", "agent_thought_chunk"),
|
|
{
|
|
sessionUpdate: "tool_call",
|
|
toolCallId: "tool-1",
|
|
title: "Read sensor",
|
|
kind: "read",
|
|
status: "in_progress",
|
|
},
|
|
{ sessionUpdate: "tool_call_update", toolCallId: "tool-1", status: "completed" },
|
|
textChunk("It is 25°C"),
|
|
],
|
|
"replay",
|
|
);
|
|
try {
|
|
expect(timelineProjection(events)).toEqual([
|
|
{ type: "user_message", id: "user-1", text: "What is the " },
|
|
{ type: "user_message", id: "user-1", text: "What is the temperature?" },
|
|
{ type: "reasoning", id: "agent_thought_chunk:1", text: "Checking the sensor" },
|
|
{ type: "tool_call", id: "tool-1", status: "running" },
|
|
{ type: "tool_call", id: "tool-1", status: "completed" },
|
|
{ type: "assistant_message", id: "agent_message_chunk:2", text: "It is 25°C" },
|
|
]);
|
|
const lastReplayed = events.findLastIndex((event) => event.type === "timeline.item");
|
|
const opened = events.findIndex((event) => event.type === "session.opened");
|
|
expect(lastReplayed).toBeLessThan(opened);
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
});
|
|
|
|
it("continues anonymous user chunks into one message and starts a new one after an agent reply", async () => {
|
|
const { connection, events } = await openReplaySession(
|
|
[
|
|
userChunk("first "),
|
|
userChunk("question"),
|
|
textChunk("answer"),
|
|
userChunk("second question"),
|
|
],
|
|
"replay",
|
|
);
|
|
try {
|
|
expect(timelineProjection(events)).toEqual([
|
|
{ type: "user_message", id: "user_message_chunk:1", text: "first " },
|
|
{ type: "user_message", id: "user_message_chunk:1", text: "first question" },
|
|
{ type: "assistant_message", id: "agent_message_chunk:2", text: "answer" },
|
|
{ type: "user_message", id: "user_message_chunk:3", text: "second question" },
|
|
]);
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
});
|
|
|
|
it("does not duplicate the live prompt echo when the agent echoes it as a user chunk", async () => {
|
|
const harness = connectorHarness({
|
|
capabilities: { loadSession: true },
|
|
promptUpdates: (promptText) => [
|
|
{ sessionUpdate: "user_message_chunk", content: { type: "text", text: promptText } },
|
|
{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "Acknowledged" } },
|
|
],
|
|
});
|
|
const registration = runAcpProvider({
|
|
id: "echo-acp",
|
|
label: "Echo ACP",
|
|
connector: harness.connector,
|
|
});
|
|
const connection = await registration.connect({
|
|
versions: [1],
|
|
capabilities: ["prompt.message", "session.persistence"],
|
|
});
|
|
const events: ProviderEvent[] = [];
|
|
connection.onEvent((event) => events.push(event));
|
|
const input = openInput();
|
|
input.persistence = { version: 1, data: { sessionId: "connector-session" } };
|
|
input.history = "replay";
|
|
await connection.send(input);
|
|
await waitForEvent(events, (event) => event.type === "session.ready");
|
|
|
|
await connection.send({
|
|
type: "session.prompt",
|
|
sessionId: "session-1",
|
|
prompt: {
|
|
clientMessageId: "echo-1",
|
|
delivery: "auto",
|
|
input: { type: "message", content: [{ type: "text", text: "hello" }] },
|
|
},
|
|
});
|
|
try {
|
|
await waitForEvent(
|
|
events,
|
|
(event) =>
|
|
event.type === "session.turn" &&
|
|
event.turnId === "acp:echo-1" &&
|
|
event.state === "completed",
|
|
);
|
|
expect(timelineProjection(events)).toEqual([
|
|
{ type: "user_message", id: "echo-1", text: "hello" },
|
|
{ type: "assistant_message", id: "agent_message_chunk:1", text: "Acknowledged" },
|
|
]);
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
});
|
|
|
|
it("drops replayed updates when the open request asks to skip history", async () => {
|
|
const { connection, events } = await openReplaySession(
|
|
[userChunk("What is in the README?", "user-1"), textChunk("It describes the project.")],
|
|
"skip",
|
|
);
|
|
try {
|
|
expect(timelineProjection(events)).toEqual([]);
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
});
|
|
});
|