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();
    }
  });
});