feat(acp): session lifecycle + prompt driving (U3)

Implements the real AgentRuntime: createSession spawns + handshakes (U2)
then opens session/new (empty mcpServers, KTD5), persisting sessionId, cwd,
and the engine-provided actionGateContext (KTD3) plus the live connection
on the session. promptWithFallback builds ContentBlocks and drives one
prompt turn to its terminal stopReason. cancel/loadSession/resume helpers;
dispose does best-effort cancel then registry-authoritative teardown (KTD4a).
prompt-builder.ts builds text/image ContentBlock[]. 8 files / 53 tests green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

plugins/fusion-plugin-acp-runtime/src/__tests__/fixtures/echo-agent.mjs

@@ -21,6 +21,9 @@ class EchoAgent {
   constructor(connection) {
     this.connection = connection;
     this.sessions = new Map();
+    // Resolver for the in-flight prompt when ACP_FIXTURE_HANG_PROMPT is set:
+    // the turn stays open until cancel() fires, then resolves "cancelled".
+    this._cancelTurn = undefined;
   }
 
   async initialize(_params) {
@@ -38,7 +41,7 @@ class EchoAgent {
       versionOverride !== undefined ? Number(versionOverride) : PROTOCOL_VERSION;
     const response = {
       protocolVersion,
-      agentCapabilities: { loadSession: false },
+      agentCapabilities: { loadSession: process.env.ACP_FIXTURE_LOAD_SESSION === "1" },
     };
     if (process.env.ACP_FIXTURE_REQUIRE_AUTH === "1") {
       response.authMethods = [{ id: "api-key", name: "API Key", description: null }];
@@ -58,6 +61,13 @@ class EchoAgent {
     return { sessionId };
   }
 
+  async loadSession(params) {
+    // Resume path: acknowledge the existing session id (history replay would
+    // happen here in a real agent). Mark that this session was loaded, not new.
+    this.sessions.set(params.sessionId, { loaded: true });
+    return {};
+  }
+
   async setSessionMode(_params) {
     return {};
   }
@@ -70,11 +80,23 @@ class EchoAgent {
         content: { type: "text", text: "echo: hello" },
       },
     });
+    // Cancel-mid-prompt test: keep the turn open until cancel() arrives, then
+    // resolve with the "cancelled" stop reason (mirrors a real agent).
+    if (process.env.ACP_FIXTURE_HANG_PROMPT === "1") {
+      return await new Promise((resolve) => {
+        this._cancelTurn = () => resolve({ stopReason: "cancelled" });
+      });
+    }
     return { stopReason: "end_turn" };
   }
 
   async cancel(_params) {
-    // no-op for the trivial turn
+    // Release any in-flight hung turn with a "cancelled" stop reason.
+    if (this._cancelTurn) {
+      const release = this._cancelTurn;
+      this._cancelTurn = undefined;
+      release();
+    }
   }
 }
 

plugins/fusion-plugin-acp-runtime/src/__tests__/index.test.ts

@@ -1,8 +1,12 @@
-import { describe, it, expect } from "vitest";
+import { describe, it, expect, afterEach } from "vitest";
 import plugin, { AcpRuntimeAdapter, acpRuntimeFactory, acpRuntimeMetadata, resolveCliSettings } from "../index.js";
-import { ACP_NOT_IMPLEMENTED } from "../runtime-adapter.js";
+import { killAllProcesses } from "../process-manager.js";
 import type { AgentRuntime } from "../types.js";
 
+afterEach(() => {
+  killAllProcesses();
+});
+
 describe("fusion-plugin-acp-runtime", () => {
   it("declares the acp runtime in its manifest", () => {
     expect(plugin.manifest.id).toBe("fusion-plugin-acp-runtime");
@@ -27,12 +31,21 @@ describe("fusion-plugin-acp-runtime", () => {
     expect(desc).toBe("acp/gemini-2.0");
   });
 
-  it("session-driving stubs reject with the not-implemented marker (until U2/U3)", async () => {
-    const runtime = new AcpRuntimeAdapter({});
+  it("createSession against a non-spawnable binary rejects (ENOENT), no orphan", async () => {
+    const runtime = new AcpRuntimeAdapter({
+      acpBinaryPath: "/nonexistent/acp-agent-does-not-exist",
+      acpArgs: [],
+    });
     await expect(
-      runtime.createSession({ cwd: "/tmp", systemPrompt: "" } as never),
-    ).rejects.toThrow(ACP_NOT_IMPLEMENTED);
-    await expect(runtime.promptWithFallback({} as never, "hi")).rejects.toThrow(ACP_NOT_IMPLEMENTED);
+      runtime.createSession({ cwd: process.cwd(), systemPrompt: "" } as never),
+    ).rejects.toMatchObject({ code: "ENOENT" });
+  });
+
+  it("promptWithFallback on a session with no live connection rejects cleanly", async () => {
+    const runtime = new AcpRuntimeAdapter({});
+    await expect(runtime.promptWithFallback({ sessionId: "x" } as never, "hi")).rejects.toThrow(
+      /no live connection/,
+    );
   });
 });
 

plugins/fusion-plugin-acp-runtime/src/__tests__/prompt-builder.test.ts

@@ -0,0 +1,31 @@
+import { describe, it, expect } from "vitest";
+import { buildPromptBlocks } from "../prompt-builder.js";
+
+describe("buildPromptBlocks", () => {
+  it("turns a plain string into a single text block", () => {
+    const blocks = buildPromptBlocks("hello world");
+    expect(blocks).toEqual([{ type: "text", text: "hello world" }]);
+  });
+
+  it("emits no text block for an empty string", () => {
+    expect(buildPromptBlocks("")).toEqual([]);
+  });
+
+  it("appends image blocks after the text block", () => {
+    const blocks = buildPromptBlocks("describe this", {
+      images: [{ data: "AAAA", mimeType: "image/png", uri: "file:///a.png" }],
+    });
+    expect(blocks).toEqual([
+      { type: "text", text: "describe this" },
+      { type: "image", data: "AAAA", mimeType: "image/png", uri: "file:///a.png" },
+    ]);
+  });
+
+  it("omits the uri field when not provided on an image", () => {
+    const blocks = buildPromptBlocks("", {
+      images: [{ data: "BBBB", mimeType: "image/jpeg" }],
+    });
+    expect(blocks).toEqual([{ type: "image", data: "BBBB", mimeType: "image/jpeg" }]);
+    expect(blocks[0]).not.toHaveProperty("uri");
+  });
+});

plugins/fusion-plugin-acp-runtime/src/__tests__/provider-session.test.ts

@@ -0,0 +1,114 @@
+import { describe, it, expect, afterEach } from "vitest";
+import { fileURLToPath } from "node:url";
+import {
+  connect,
+  newAcpSession,
+  promptAcpSession,
+  cancelAcpSession,
+  loadAcpSession,
+  type AcpConnection,
+} from "../provider.js";
+import { buildPromptBlocks } from "../prompt-builder.js";
+import { killAllProcesses } from "../process-manager.js";
+
+const FIXTURE = fileURLToPath(new URL("./fixtures/echo-agent.mjs", import.meta.url));
+
+afterEach(() => {
+  killAllProcesses();
+});
+
+function baseOpts(extraEnv: Record<string, string> = {}) {
+  return {
+    binaryPath: process.execPath,
+    args: [FIXTURE],
+    cwd: process.cwd(),
+    env: extraEnv as NodeJS.ProcessEnv,
+    advertiseFs: { read: false, write: false },
+    initializeTimeoutMs: 10_000,
+  };
+}
+
+async function open(extraEnv: Record<string, string> = {}): Promise<AcpConnection> {
+  return connect(baseOpts(extraEnv));
+}
+
+describe("session driving helpers", () => {
+  it("newAcpSession opens a session and returns a sessionId", async () => {
+    const conn = await open();
+    try {
+      const { sessionId } = await newAcpSession(conn, { cwd: process.cwd() });
+      expect(typeof sessionId).toBe("string");
+      expect(sessionId.length).toBeGreaterThan(0);
+    } finally {
+      conn.dispose();
+    }
+  });
+
+  it("promptAcpSession resolves with end_turn for a normal turn", async () => {
+    const conn = await open();
+    try {
+      const { sessionId } = await newAcpSession(conn, { cwd: process.cwd() });
+      const stopReason = await promptAcpSession(conn, sessionId, buildPromptBlocks("hello"));
+      expect(stopReason).toBe("end_turn");
+    } finally {
+      conn.dispose();
+    }
+  });
+
+  it("cancelAcpSession releases a mid-turn prompt with the cancelled stop reason", async () => {
+    const conn = await open({ ACP_FIXTURE_HANG_PROMPT: "1" });
+    try {
+      const { sessionId } = await newAcpSession(conn, { cwd: process.cwd() });
+      const promptPromise = promptAcpSession(conn, sessionId, buildPromptBlocks("hello"));
+      // Give the turn a tick to register the hang before cancelling.
+      await new Promise((r) => setImmediate(r));
+      await cancelAcpSession(conn, sessionId);
+      const stopReason = await promptPromise;
+      expect(stopReason).toBe("cancelled");
+    } finally {
+      conn.dispose();
+    }
+  });
+
+  it("cancelAcpSession swallows errors (fire-and-forget)", async () => {
+    const conn = await open();
+    try {
+      const { sessionId } = await newAcpSession(conn, { cwd: process.cwd() });
+      conn.dispose(); // kill the child so cancel cannot round-trip
+      await expect(cancelAcpSession(conn, sessionId)).resolves.toBeUndefined();
+    } finally {
+      conn.dispose();
+    }
+  });
+
+  it("loadAcpSession uses session/load when the agent advertises loadSession", async () => {
+    const conn = await open({ ACP_FIXTURE_LOAD_SESSION: "1" });
+    try {
+      expect(conn.agentCapabilities).toMatchObject({ loadSession: true });
+      const result = await loadAcpSession(conn, {
+        sessionId: "prior-session-id",
+        cwd: process.cwd(),
+      });
+      // session/load echoes back the requested id (no fresh id minted).
+      expect(result.sessionId).toBe("prior-session-id");
+    } finally {
+      conn.dispose();
+    }
+  });
+
+  it("loadAcpSession falls back to newSession when loadSession is not advertised", async () => {
+    const conn = await open(); // loadSession defaults false
+    try {
+      expect(conn.agentCapabilities).toMatchObject({ loadSession: false });
+      const result = await loadAcpSession(conn, {
+        sessionId: "prior-session-id",
+        cwd: process.cwd(),
+      });
+      // Fresh session: a new id is minted, not the prior one.
+      expect(result.sessionId).not.toBe("prior-session-id");
+      expect(result.sessionId.length).toBeGreaterThan(0);
+    } finally {
+      conn.dispose();
+    }
+  });
+});

plugins/fusion-plugin-acp-runtime/src/__tests__/runtime-adapter.test.ts

@@ -0,0 +1,98 @@
+import { describe, it, expect, afterEach } from "vitest";
+import os from "node:os";
+import { fileURLToPath } from "node:url";
+import { AcpRuntimeAdapter } from "../runtime-adapter.js";
+import { killAllProcesses, activeProcessCount } from "../process-manager.js";
+import type { AcpSession, AgentRuntimeOptions } from "../types.js";
+
+const FIXTURE = fileURLToPath(new URL("./fixtures/echo-agent.mjs", import.meta.url));
+
+afterEach(() => {
+  killAllProcesses();
+});
+
+function makeAdapter(extra: Record<string, unknown> = {}) {
+  return new AcpRuntimeAdapter({
+    acpBinaryPath: process.execPath,
+    acpArgs: [FIXTURE],
+    acpModel: "echo-agent",
+    ...extra,
+  });
+}
+
+function makeOptions(over: Partial<AgentRuntimeOptions> = {}): AgentRuntimeOptions {
+  return {
+    cwd: process.cwd(),
+    systemPrompt: "be helpful",
+    ...over,
+  };
+}
+
+describe("AcpRuntimeAdapter (U3)", () => {
+  it("createSession spawns + opens a session with a real sessionId", async () => {
+    const adapter = makeAdapter();
+    const { session } = await adapter.createSession(makeOptions());
+    try {
+      expect(session.sessionId.length).toBeGreaterThan(0);
+      expect((session as AcpSession).connection).toBeDefined();
+      expect(session.lastModelDescription).toBe("acp/echo-agent");
+    } finally {
+      await adapter.dispose(session);
+    }
+  });
+
+  it("createSession persists actionGateContext and cwd on the session", async () => {
+    const adapter = makeAdapter();
+    const gate = { permissionPolicy: { preset: "unrestricted" } };
+    // cwd must be a real, spawnable directory (it is the subprocess cwd too).
+    const cwd = os.tmpdir();
+    const { session } = await adapter.createSession(
+      makeOptions({ cwd, actionGateContext: gate }),
+    );
+    try {
+      // Both reachable from the session object for the U5/U7 handlers to read.
+      expect((session as AcpSession).gate).toBe(gate);
+      expect(session.cwd).toBe(cwd);
+    } finally {
+      await adapter.dispose(session);
+    }
+  });
+
+  it("promptWithFallback drives a full turn to completion", async () => {
+    const adapter = makeAdapter();
+    const { session } = await adapter.createSession(makeOptions());
+    try {
+      await expect(adapter.promptWithFallback(session, "hello")).resolves.toBeUndefined();
+    } finally {
+      await adapter.dispose(session);
+    }
+  });
+
+  it("dispose tears down the subprocess and is idempotent", async () => {
+    const adapter = makeAdapter();
+    const { session } = await adapter.createSession(makeOptions());
+    expect(activeProcessCount()).toBe(1);
+    await adapter.dispose(session);
+    expect(activeProcessCount()).toBe(0);
+    // second dispose must not throw
+    await expect(adapter.dispose(session)).resolves.toBeUndefined();
+    expect(activeProcessCount()).toBe(0);
+  });
+
+  it("promptWithFallback rejects when the session has no live connection", async () => {
+    const adapter = makeAdapter();
+    await expect(
+      adapter.promptWithFallback({ sessionId: "x" } as never, "hi"),
+    ).rejects.toThrow(/no live connection/);
+  });
+
+  it("describeModel returns the session model description", async () => {
+    const adapter = makeAdapter();
+    const { session } = await adapter.createSession(makeOptions());
+    try {
+      expect(adapter.describeModel(session)).toBe("acp/echo-agent");
+    } finally {
+      await adapter.dispose(session);
+    }
+  });
+});

plugins/fusion-plugin-acp-runtime/src/prompt-builder.ts

@@ -0,0 +1,49 @@
+// Builds ACP `ContentBlock[]` from a Fusion prompt.
+//
+// U3 core path: a plain string prompt becomes a single `{ type: "text", text }`
+// block. The runtime may later pass structured content (e.g. an attached image);
+// when present we emit the matching block. Keep this small and pure.
+
+import type { ContentBlock } from "@agentclientprotocol/sdk";
+
+/** Optional structured content the runtime may attach alongside the text prompt. */
+export interface PromptImage {
+  /** Base64-encoded image data (no data: prefix). */
+  data: string;
+  /** MIME type, e.g. "image/png". */
+  mimeType: string;
+  /** Optional source URI for the image. */
+  uri?: string;
+}
+
+export interface BuildPromptOptions {
+  /** Image content to append as image block(s) after the text. */
+  images?: PromptImage[];
+}
+
+/**
+ * Build the ACP prompt content blocks for a turn.
+ *
+ * A non-empty string yields one text block. An empty/whitespace-only string
+ * yields no text block (but any attached images are still included), so we never
+ * send a meaningless empty text block. Images, when supplied, are appended as
+ * `image` blocks (passthrough — KTD ContentBlock image variant).
+ */
+export function buildPromptBlocks(prompt: string, opts?: BuildPromptOptions): ContentBlock[] {
+  const blocks: ContentBlock[] = [];
+
+  if (typeof prompt === "string" && prompt.length > 0) {
+    blocks.push({ type: "text", text: prompt });
+  }
+
+  for (const image of opts?.images ?? []) {
+    blocks.push({
+      type: "image",
+      data: image.data,
+      mimeType: image.mimeType,
+      ...(image.uri ? { uri: image.uri } : {}),
+    });
+  }
+
+  return blocks;
+}

plugins/fusion-plugin-acp-runtime/src/provider.ts

@@ -19,6 +19,8 @@ import {
   PROTOCOL_VERSION,
   type Agent,
   type Client,
+  type ContentBlock,
+  type StopReason,
 } from "@agentclientprotocol/sdk";
 import { spawnAgent, captureStderr, forceKill, unregisterProcess } from "./process-manager.js";
 
@@ -198,3 +200,92 @@ export async function connect(opts: ConnectOptions): Promise<AcpConnection> {
     dispose,
   };
 }
+
+// --- U3: session driving on top of connect() -------------------------------
+//
+// These helpers wrap the `ClientSideConnection` session methods so the runtime
+// adapter drives one shape (open → prompt → cancel/resume) without touching SDK
+// types directly. v1 always sends an empty `mcpServers` (KTD5).
+
+/** Narrow view of the agent capabilities we read for resume routing. */
+interface AgentCapabilitiesView {
+  loadSession?: boolean;
+}
+
+function readsLoadSession(connection: AcpConnection): boolean {
+  const caps = connection.agentCapabilities as AgentCapabilitiesView | undefined;
+  return caps?.loadSession === true;
+}
+
+export interface NewAcpSessionResult {
+  sessionId: string;
+  /** Initial session mode state, when the agent reports one. */
+  modes?: unknown;
+}
+
+/**
+ * Open a fresh ACP session via `session/new`. Always passes an empty
+ * `mcpServers` (KTD5 — Fusion custom-tool forwarding is deferred).
+ */
+export async function newAcpSession(
+  connection: AcpConnection,
+  opts: { cwd: string },
+): Promise<NewAcpSessionResult> {
+  const res = await connection.conn.newSession({ cwd: opts.cwd, mcpServers: [] });
+  return { sessionId: res.sessionId, modes: res.modes ?? undefined };
+}
+
+/**
+ * Send a prompt turn via `session/prompt` and return the terminal `stopReason`.
+ *
+ * The SDK prompt promise resolves only AFTER every `session/update` for the turn
+ * has been delivered to the client handler — so resolving here is the correct
+ * "turn complete" signal (no extra draining required).
+ */
+export async function promptAcpSession(
+  connection: AcpConnection,
+  sessionId: string,
+  blocks: ContentBlock[],
+): Promise<StopReason> {
+  const res = await connection.conn.prompt({ sessionId, prompt: blocks });
+  return res.stopReason;
+}
+
+/**
+ * Best-effort cancel of the active turn via the `session/cancel` notification.
+ *
+ * This is fire-and-forget (no ack in the protocol). Errors are swallowed — it
+ * runs during teardown where the registry SIGKILL is the authoritative guarantee
+ * (KTD4a).
+ */
+export async function cancelAcpSession(
+  connection: AcpConnection,
+  sessionId: string,
+): Promise<void> {
+  try {
+    await connection.conn.cancel({ sessionId });
+  } catch {
+    // fire-and-forget; teardown's SIGKILL is authoritative
+  }
+}
+
+/**
+ * Resume a session. Prefers `session/load` (history replay) when the agent
+ * advertised the `loadSession` capability; otherwise falls back to opening a
+ * fresh `session/new`. There is no separate `resume` method in this SDK build —
+ * `loadSession` IS the resume path.
+ */
+export async function loadAcpSession(
+  connection: AcpConnection,
+  opts: { sessionId: string; cwd: string },
+): Promise<NewAcpSessionResult> {
+  if (readsLoadSession(connection)) {
+    const res = await connection.conn.loadSession({
+      sessionId: opts.sessionId,
+      cwd: opts.cwd,
+      mcpServers: [],
+    });
+    return { sessionId: opts.sessionId, modes: res.modes ?? undefined };
+  }
+  return newAcpSession(connection, { cwd: opts.cwd });
+}

plugins/fusion-plugin-acp-runtime/src/runtime-adapter.ts

@@ -1,12 +1,22 @@
 // AgentRuntime adapter for the ACP runtime.
 //
-// U1 scaffold: implements the full `AgentRuntime` contract shape (including the
-// required `describeModel`) with stubs that throw `not_implemented` until the
-// session driver lands in U2/U3. The skeleton exists so the plugin loads,
-// registers as `runtimeId: "acp"`, and conforms to the interface the engine
-// resolves via `getRuntimeById`.
+// U3 implements the real session lifecycle: createSession spawns + handshakes
+// (U2 connect()) then opens a `session/new`; promptWithFallback drives one
+// prompt turn to its terminal stopReason; dispose tears down the connection
+// (KTD4a — registry SIGKILL is authoritative). The `session/update` event
+// bridge (U4) and the permission gate (U5) are wired in later units; for U3 the
+// default client handler from U2 is used and a turn still resolves with a
+// stopReason.
 
 import { resolveCliSettings, type AcpCliSettings } from "./cli-spawn.js";
+import {
+  connect,
+  newAcpSession,
+  promptAcpSession,
+  cancelAcpSession,
+} from "./provider.js";
+import { buildSpawnEnv } from "./process-manager.js";
+import { buildPromptBlocks } from "./prompt-builder.js";
 import type {
   AgentRuntime,
   AgentRuntimeOptions,
@@ -15,6 +25,10 @@ import type {
   AcpSession,
 } from "./types.js";
 
+/**
+ * Retained for back-compat: earlier units' tests imported this marker. The real
+ * adapter no longer throws it; it remains exported so external references resolve.
+ */
 export const ACP_NOT_IMPLEMENTED = "acp_not_implemented";
 
 export class AcpRuntimeAdapter implements AgentRuntime {
@@ -27,13 +41,35 @@ export class AcpRuntimeAdapter implements AgentRuntime {
   }
 
   async createSession(options: AgentRuntimeOptions): Promise<AgentSessionResult> {
-    // Session establishment (spawn + initialize + session/new) lands in U2/U3.
-    // The skeleton constructs the session shell so the contract is observable.
     const model = this.settings.model ?? options.defaultModelId ?? "acp";
+
+    // Spawn + initialize (U2). fs capabilities are advertised only where the
+    // resolved settings enable them (KTD6); the subprocess env is built from the
+    // allow-list, never inherited process.env (KTD6b).
+    const connection = await connect({
+      binaryPath: this.settings.binaryPath,
+      args: this.settings.args,
+      cwd: options.cwd,
+      env: buildSpawnEnv(this.settings.envAllowList),
+      advertiseFs: { read: this.settings.fsRead, write: this.settings.fsWrite },
+    });
+
+    // Open the ACP session over the task worktree (empty mcpServers — KTD5).
+    let sessionId: string;
+    try {
+      const opened = await newAcpSession(connection, { cwd: options.cwd });
+      sessionId = opened.sessionId;
+    } catch (err) {
+      // Don't leak the subprocess if session/new fails after a good handshake.
+      connection.dispose();
+      throw err;
+    }
+
+    let disposed = false;
     const session: AcpSession = {
       model,
       systemPrompt: options.systemPrompt,
-      sessionId: "",
+      sessionId,
       cwd: options.cwd,
       lastModelDescription: `acp/${model}`,
       callbacks: {
@@ -42,14 +78,34 @@ export class AcpRuntimeAdapter implements AgentRuntime {
         onToolStart: options.onToolStart,
         onToolEnd: options.onToolEnd,
       },
+      // Persist the per-run gate (KTD3) so U5/U7 can reach the live action gate.
       gate: options.actionGateContext,
-      dispose: () => undefined,
+      connection,
+      dispose: () => {
+        if (disposed) return;
+        disposed = true;
+        connection.dispose();
+      },
     };
-    throw new Error(`${ACP_NOT_IMPLEMENTED}: createSession lands in U2/U3 (session=${session.lastModelDescription})`);
+
+    return { session };
   }
 
-  async promptWithFallback(_session: AgentSession, _prompt: string, _options?: unknown): Promise<void> {
-    throw new Error(`${ACP_NOT_IMPLEMENTED}: promptWithFallback lands in U3`);
+  async promptWithFallback(
+    session: AgentSession,
+    prompt: string,
+    _options?: unknown,
+  ): Promise<void> {
+    const acp = session as AcpSession;
+    if (!acp.connection) {
+      throw new Error("ACP session has no live connection (createSession not completed)");
+    }
+    const blocks = buildPromptBlocks(prompt);
+    // Resolve when the SDK prompt promise resolves — it already drains all
+    // session/update notifications for the turn before reporting the stopReason.
+    // TODO(U4): wire a bridging client handler so streamed text/tool updates
+    // surface onto session.callbacks; for U3 the turn simply completes.
+    await promptAcpSession(acp.connection, acp.sessionId, blocks);
   }
 
   describeModel(session: AgentSession): string {
@@ -57,7 +113,13 @@ export class AcpRuntimeAdapter implements AgentRuntime {
   }
 
   async dispose(session: AgentSession): Promise<void> {
-    // Best-effort teardown; the authoritative kill is the process registry (KTD4a).
+    // KTD4a teardown: best-effort cancel of any in-flight turn, then force the
+    // connection down. The process-registry SIGKILL is the authoritative
+    // no-orphan guarantee, not the cancel round-trip. Idempotent.
+    const acp = session as AcpSession;
+    if (acp.connection && acp.sessionId) {
+      await cancelAcpSession(acp.connection, acp.sessionId);
+    }
     session.dispose();
   }
 }

plugins/fusion-plugin-acp-runtime/src/types.ts

@@ -10,6 +10,8 @@
 // per-run permission gate — see `PermissionGate` below, the narrow structural
 // view this plugin couples to instead of importing `@fusion/engine` internals.
 
+import type { AcpConnection } from "./provider.js";
+
 /** Callbacks the engine wires to surface streamed agent output into Fusion's UI/logs. */
 export interface AcpCallbacks {
   onText?: (text: string) => void;
@@ -65,6 +67,11 @@ export interface AcpSession {
   callbacks: AcpCallbacks;
   /** Per-run permission gate captured at createSession (U5/U7 read this). */
   gate?: PermissionGate;
+  /**
+   * Live ACP connection backing this session (U3). Prompt/dispose reach the
+   * agent through it. Undefined only for the bare session shell used in tests.
+   */
+  connection?: AcpConnection;
   dispose(): void;
 }