feat(acp): session/update event bridge (U4)

Maps ACP session/update notifications to AgentRuntime callbacks using the
authoritative SDK 0.24.0 vocabulary: agent_message_chunk->onText,
agent_thought_chunk->onThinking, tool_call->onToolStart, tool_call_update
(completed/failed)->onToolEnd correlated by toolCallId, plan as full
replacement. tool-mapping.ts derives display names + normalizes args.
createSession now passes a bridging client handler into connect() so
streamed updates reach the engine callbacks. +24 tests (77 total).

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

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

@@ -0,0 +1,272 @@
+import { describe, it, expect, vi } from "vitest";
+import type { SessionUpdate } from "@agentclientprotocol/sdk";
+import { createEventBridge } from "../event-bridge.js";
+import type { AcpCallbacks } from "../types.js";
+
+function makeCallbacks() {
+  const onText = vi.fn<(text: string) => void>();
+  const onThinking = vi.fn<(text: string) => void>();
+  const onToolStart = vi.fn<(name: string, args?: unknown) => void>();
+  const onToolEnd = vi.fn<(name: string, isError: boolean, result?: unknown) => void>();
+  const callbacks: AcpCallbacks = { onText, onThinking, onToolStart, onToolEnd };
+  return { callbacks, onText, onThinking, onToolStart, onToolEnd };
+}
+
+describe("event bridge: text/thinking", () => {
+  it("agent_message_chunk sequence reconstructs the full message via successive onText", () => {
+    const { callbacks, onText, onThinking } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: "Hello" },
+    } as SessionUpdate);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: " world." },
+    } as SessionUpdate);
+
+    expect(onText).toHaveBeenCalledTimes(2);
+    expect(onText.mock.calls.map((c) => c[0]).join("")).toBe("Hello world.");
+    expect(onThinking).not.toHaveBeenCalled();
+  });
+
+  it("repairs a dropped inter-chunk space between sentence end and capitalized start", () => {
+    const { callbacks, onText } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: "Done." },
+    } as SessionUpdate);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: "Next step." },
+    } as SessionUpdate);
+
+    expect(onText.mock.calls.map((c) => c[0]).join("")).toBe("Done. Next step.");
+  });
+
+  it("agent_thought_chunk routes to onThinking, not onText", () => {
+    const { callbacks, onText, onThinking } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_thought_chunk",
+      content: { type: "text", text: "thinking..." },
+    } as SessionUpdate);
+
+    expect(onThinking).toHaveBeenCalledTimes(1);
+    expect(onThinking).toHaveBeenCalledWith("thinking...");
+    expect(onText).not.toHaveBeenCalled();
+  });
+
+  it("ignores user_message_chunk", () => {
+    const { callbacks, onText, onThinking } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "user_message_chunk",
+      content: { type: "text", text: "user echo" },
+    } as SessionUpdate);
+    expect(onText).not.toHaveBeenCalled();
+    expect(onThinking).not.toHaveBeenCalled();
+  });
+
+  it("ignores non-text content blocks for text extraction", () => {
+    const { callbacks, onText } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "image", data: "abc", mimeType: "image/png" },
+    } as unknown as SessionUpdate);
+    expect(onText).not.toHaveBeenCalled();
+  });
+});
+
+describe("event bridge: tool call lifecycle", () => {
+  it("tool_call → onToolStart with mapped name + normalized args", () => {
+    const { callbacks, onToolStart } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call",
+      toolCallId: "t1",
+      title: "Run tests",
+      kind: "execute",
+      rawInput: { command: "pnpm test" },
+    } as SessionUpdate);
+
+    expect(onToolStart).toHaveBeenCalledTimes(1);
+    expect(onToolStart).toHaveBeenCalledWith("Run tests", { command: "pnpm test" });
+  });
+
+  it("tool_call_update(status:failed) → onToolEnd(isError=true), correlated by toolCallId", () => {
+    const { callbacks, onToolStart, onToolEnd } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call",
+      toolCallId: "t1",
+      title: "Run tests",
+      kind: "execute",
+    } as SessionUpdate);
+    // partial update omits title/kind — bridge must carry them forward
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call_update",
+      toolCallId: "t1",
+      status: "failed",
+      rawOutput: { exitCode: 1 },
+    } as SessionUpdate);
+
+    expect(onToolStart).toHaveBeenCalledWith("Run tests", {});
+    expect(onToolEnd).toHaveBeenCalledTimes(1);
+    expect(onToolEnd).toHaveBeenCalledWith("Run tests", true, { exitCode: 1 });
+  });
+
+  it("intermediate statuses do not fire onToolEnd; completed fires isError=false", () => {
+    const { callbacks, onToolEnd } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call",
+      toolCallId: "t1",
+      title: "Read file",
+      kind: "read",
+    } as SessionUpdate);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call_update",
+      toolCallId: "t1",
+      status: "in_progress",
+    } as SessionUpdate);
+    expect(onToolEnd).not.toHaveBeenCalled();
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call_update",
+      toolCallId: "t1",
+      status: "completed",
+      rawOutput: "ok",
+    } as SessionUpdate);
+    expect(onToolEnd).toHaveBeenCalledTimes(1);
+    expect(onToolEnd).toHaveBeenCalledWith("Read file", false, "ok");
+  });
+
+  it("does not fire onToolEnd twice for repeated terminal updates", () => {
+    const { callbacks, onToolEnd } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call",
+      toolCallId: "t1",
+      title: "X",
+    } as SessionUpdate);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call_update",
+      toolCallId: "t1",
+      status: "completed",
+    } as SessionUpdate);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call_update",
+      toolCallId: "t1",
+      status: "completed",
+    } as SessionUpdate);
+    expect(onToolEnd).toHaveBeenCalledTimes(1);
+  });
+
+  it("tool_call_update for an unknown id still resolves a display name (no prior start)", () => {
+    const { callbacks, onToolEnd } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "tool_call_update",
+      toolCallId: "orphan",
+      kind: "edit",
+      status: "completed",
+    } as SessionUpdate);
+    expect(onToolEnd).toHaveBeenCalledWith("Edit", false, undefined);
+  });
+});
+
+describe("event bridge: plan (full replacement)", () => {
+  it("two successive plan updates → second fully replaces (no accumulation)", () => {
+    const { callbacks, onThinking } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "plan",
+      entries: [{ content: "Step A", priority: "high", status: "pending" }],
+    } as SessionUpdate);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "plan",
+      entries: [
+        { content: "Step B", priority: "high", status: "completed" },
+        { content: "Step C", priority: "low", status: "pending" },
+      ],
+    } as SessionUpdate);
+
+    expect(onThinking).toHaveBeenCalledTimes(2);
+    const second = onThinking.mock.calls[1][0];
+    // Second snapshot reflects only the new entries — no Step A carried over.
+    expect(second).toContain("Step B");
+    expect(second).toContain("Step C");
+    expect(second).not.toContain("Step A");
+  });
+});
+
+describe("event bridge: tolerance", () => {
+  it("ignores an unknown/forward-compat sessionUpdate tag without throwing", () => {
+    const { callbacks, onText, onThinking, onToolStart, onToolEnd } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    expect(() =>
+      bridge.handleSessionUpdate({ sessionUpdate: "totally_new_thing" } as unknown as SessionUpdate),
+    ).not.toThrow();
+    expect(onText).not.toHaveBeenCalled();
+    expect(onThinking).not.toHaveBeenCalled();
+    expect(onToolStart).not.toHaveBeenCalled();
+    expect(onToolEnd).not.toHaveBeenCalled();
+  });
+
+  it("ignores store-only update tags", () => {
+    const { callbacks, onText, onThinking } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    for (const tag of [
+      "available_commands_update",
+      "current_mode_update",
+      "config_option_update",
+      "session_info_update",
+      "usage_update",
+    ]) {
+      expect(() =>
+        bridge.handleSessionUpdate({ sessionUpdate: tag } as unknown as SessionUpdate),
+      ).not.toThrow();
+    }
+    expect(onText).not.toHaveBeenCalled();
+    expect(onThinking).not.toHaveBeenCalled();
+  });
+
+  it("does not throw on a malformed tool_call missing toolCallId", () => {
+    const { callbacks, onToolStart } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    expect(() =>
+      bridge.handleSessionUpdate({
+        sessionUpdate: "tool_call",
+        title: "no id",
+      } as unknown as SessionUpdate),
+    ).not.toThrow();
+    expect(onToolStart).not.toHaveBeenCalled();
+  });
+
+  it("reset() clears correlation state between turns", () => {
+    const { callbacks, onText } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: "End." },
+    } as SessionUpdate);
+    bridge.reset();
+    // After reset, leading-capital repair has no prior text to key off — the
+    // next chunk emits unmodified.
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: "Start." },
+    } as SessionUpdate);
+    expect(onText.mock.calls.map((c) => c[0])).toEqual(["End.", "Start."]);
+  });
+});

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

@@ -13,6 +13,10 @@
 //   ACP_FIXTURE_LEAK_TOKEN=1      — write a fake auth token to stderr (redaction
 //                                   test).
 //   ACP_FIXTURE_REQUIRE_AUTH=1    — advertise a non-empty authMethods list.
+//   ACP_FIXTURE_RICH_PROMPT=1     — prompt emits the full U4 update vocabulary
+//                                   (agent_message_chunk, agent_thought_chunk,
+//                                   tool_call, tool_call_update[completed], plan)
+//                                   before resolving the turn.
 
 import { AgentSideConnection, ndJsonStream, PROTOCOL_VERSION } from "@agentclientprotocol/sdk";
 import { Readable, Writable } from "node:stream";
@@ -73,6 +77,54 @@ class EchoAgent {
   }
 
   async prompt(params) {
+    if (process.env.ACP_FIXTURE_RICH_PROMPT === "1") {
+      const sessionId = params.sessionId;
+      await this.connection.sessionUpdate({
+        sessionId,
+        update: {
+          sessionUpdate: "agent_message_chunk",
+          content: { type: "text", text: "Working on it." },
+        },
+      });
+      await this.connection.sessionUpdate({
+        sessionId,
+        update: {
+          sessionUpdate: "agent_thought_chunk",
+          content: { type: "text", text: "Let me think about this." },
+        },
+      });
+      await this.connection.sessionUpdate({
+        sessionId,
+        update: {
+          sessionUpdate: "tool_call",
+          toolCallId: "call-1",
+          title: "Run tests",
+          kind: "execute",
+          status: "in_progress",
+          rawInput: { command: "pnpm test" },
+        },
+      });
+      await this.connection.sessionUpdate({
+        sessionId,
+        update: {
+          sessionUpdate: "tool_call_update",
+          toolCallId: "call-1",
+          status: "completed",
+          rawOutput: { exitCode: 0 },
+        },
+      });
+      await this.connection.sessionUpdate({
+        sessionId,
+        update: {
+          sessionUpdate: "plan",
+          entries: [
+            { content: "Read the code", priority: "high", status: "completed" },
+            { content: "Fix the bug", priority: "medium", status: "pending" },
+          ],
+        },
+      });
+      return { stopReason: "end_turn" };
+    }
     await this.connection.sessionUpdate({
       sessionId: params.sessionId,
       update: {

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

@@ -1,4 +1,4 @@
-import { describe, it, expect, afterEach } from "vitest";
+import { describe, it, expect, afterEach, vi } from "vitest";
 import { fileURLToPath } from "node:url";
 import {
   connect,
@@ -6,6 +6,7 @@ import {
   promptAcpSession,
   cancelAcpSession,
   loadAcpSession,
+  createBridgingClientHandler,
   type AcpConnection,
 } from "../provider.js";
 import { buildPromptBlocks } from "../prompt-builder.js";
@@ -96,6 +97,33 @@ describe("session driving helpers", () => {
     }
   });
 
+  it("bridging client handler surfaces a rich prompt turn's updates onto callbacks (U4)", async () => {
+    const onText = vi.fn<(t: string) => void>();
+    const onThinking = vi.fn<(t: string) => void>();
+    const onToolStart = vi.fn<(name: string, args?: unknown) => void>();
+    const onToolEnd = vi.fn<(name: string, isError: boolean, result?: unknown) => void>();
+
+    const conn = await connect({
+      ...baseOpts({ ACP_FIXTURE_RICH_PROMPT: "1" }),
+      clientHandler: createBridgingClientHandler({ onText, onThinking, onToolStart, onToolEnd }),
+    });
+    try {
+      const { sessionId } = await newAcpSession(conn, { cwd: process.cwd() });
+      const stopReason = await promptAcpSession(conn, sessionId, buildPromptBlocks("go"));
+      expect(stopReason).toBe("end_turn");
+
+      // The SDK prompt promise resolves only after all updates are delivered.
+      expect(onText.mock.calls.map((c) => c[0]).join("")).toBe("Working on it.");
+      expect(onThinking).toHaveBeenCalledWith("Let me think about this.");
+      expect(onToolStart).toHaveBeenCalledWith("Run tests", { command: "pnpm test" });
+      expect(onToolEnd).toHaveBeenCalledWith("Run tests", false, { exitCode: 0 });
+      // The plan surfaces as a thinking line.
+      expect(onThinking.mock.calls.some((c) => String(c[0]).includes("Fix the bug"))).toBe(true);
+    } finally {
+      conn.dispose();
+    }
+  });
+
   it("loadAcpSession falls back to newSession when loadSession is not advertised", async () => {
     const conn = await open(); // loadSession defaults false
     try {

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

@@ -0,0 +1,45 @@
+import { describe, it, expect } from "vitest";
+import { toolDisplayName, normalizeToolArgs } from "../tool-mapping.js";
+
+describe("toolDisplayName", () => {
+  it("prefers an explicit title", () => {
+    expect(toolDisplayName({ title: "Run tests", kind: "execute" })).toBe("Run tests");
+  });
+
+  it("falls back to a label derived from kind when title is missing", () => {
+    expect(toolDisplayName({ kind: "execute" })).toBe("Execute");
+    expect(toolDisplayName({ kind: "read" })).toBe("Read");
+    expect(toolDisplayName({ kind: "switch_mode" })).toBe("Switch Mode");
+  });
+
+  it("treats an empty/whitespace title as missing", () => {
+    expect(toolDisplayName({ title: "   ", kind: "edit" })).toBe("Edit");
+    expect(toolDisplayName({ title: "", kind: "fetch" })).toBe("Fetch");
+  });
+
+  it("falls back to 'tool' when both title and kind are absent", () => {
+    expect(toolDisplayName({})).toBe("tool");
+    expect(toolDisplayName({ title: null, kind: null })).toBe("tool");
+  });
+
+  it("falls back to 'tool' for an unknown kind", () => {
+    expect(toolDisplayName({ kind: "mystery" as never })).toBe("tool");
+  });
+});
+
+describe("normalizeToolArgs", () => {
+  it("returns the object when rawInput is a plain object", () => {
+    expect(normalizeToolArgs({ command: "ls" })).toEqual({ command: "ls" });
+  });
+
+  it("returns {} for undefined / null", () => {
+    expect(normalizeToolArgs(undefined)).toEqual({});
+    expect(normalizeToolArgs(null)).toEqual({});
+  });
+
+  it("returns {} for non-object / array inputs", () => {
+    expect(normalizeToolArgs("string")).toEqual({});
+    expect(normalizeToolArgs(42)).toEqual({});
+    expect(normalizeToolArgs([1, 2, 3])).toEqual({});
+  });
+});

plugins/fusion-plugin-acp-runtime/src/event-bridge.ts

@@ -0,0 +1,190 @@
+// Event bridge: translate ACP `session/update` notifications into Fusion's
+// `AgentRuntime` callbacks (onText / onThinking / onToolStart / onToolEnd) so an
+// ACP agent renders identically to existing runtimes.
+//
+// Scope (U4): mapping only. Output BYTE bounds + string sanitization are U6 — no
+// caps are applied here. Permission requests are U5.
+//
+// Design notes:
+// - Tolerant: every field except the `sessionUpdate` discriminator and
+//   `toolCallId` is optional/partial. The handler NEVER throws on a malformed or
+//   partial update; unknown/forward-compat tags are ignored silently.
+// - Tool start/end correlation: a `tool_call` records `{ title, kind }` keyed by
+//   `toolCallId`; a later `tool_call_update` carries that metadata forward when
+//   the update omits it, then fires `onToolEnd` once the status reaches a
+//   terminal value (`completed` / `failed`).
+// - Plans are FULL REPLACEMENTS: each `plan` (or `plan_update`) update replaces
+//   the prior snapshot wholesale; we never accumulate across updates.
+
+import type {
+  SessionUpdate,
+  ContentBlock,
+  ToolKind,
+  PlanEntry,
+} from "@agentclientprotocol/sdk";
+import type { AcpCallbacks } from "./types.js";
+import { toolDisplayName, normalizeToolArgs } from "./tool-mapping.js";
+
+/** Tracked metadata for an in-flight tool call, keyed by `toolCallId`. */
+interface TrackedToolCall {
+  title?: string | null;
+  kind?: ToolKind | null;
+  /** Whether onToolEnd has already fired (terminal status seen). */
+  ended: boolean;
+}
+
+export interface EventBridge {
+  /** Process one `session/update` payload (`params.update`). Never throws. */
+  handleSessionUpdate(update: SessionUpdate): void;
+  /** Clear per-turn correlation state (tool calls, plan snapshot, last text). */
+  reset(): void;
+}
+
+/** Extract plain text from a `ContentBlock`, or `undefined` for non-text blocks. */
+function extractText(content: ContentBlock | undefined): string | undefined {
+  if (content && content.type === "text" && typeof content.text === "string") {
+    return content.text;
+  }
+  return undefined;
+}
+
+/**
+ * Repair the specific "sentence punctuation + capitalized next sentence" case
+ * where an agent splits adjacent sentences across chunks without the separating
+ * space. Mirrors the droid runtime's `normalizeStreamingDelta` — conservative so
+ * code, domains, and lowercase continuations are left untouched.
+ */
+function normalizeStreamingDelta(previousText: string, nextDelta: string): string {
+  if (!previousText || !nextDelta) return nextDelta;
+  const previousChar = previousText.slice(-1);
+  const nextChar = nextDelta[0] ?? "";
+  if (/\s/.test(previousChar) || /\s/.test(nextChar)) return nextDelta;
+  if (/[.!?]/.test(previousChar) && /[A-Z0-9"'([]/.test(nextChar)) {
+    return ` ${nextDelta}`;
+  }
+  return nextDelta;
+}
+
+/** Format a plan snapshot into a single thinking/log line. */
+function formatPlan(entries: PlanEntry[]): string {
+  const lines = entries.map((entry) => {
+    const status = typeof entry.status === "string" ? entry.status : "pending";
+    const text = typeof entry.content === "string" ? entry.content : "";
+    return `- [${status}] ${text}`;
+  });
+  return `Plan:\n${lines.join("\n")}`;
+}
+
+export function createEventBridge(callbacks: AcpCallbacks): EventBridge {
+  // Start/end correlation across `tool_call` → `tool_call_update`.
+  const toolCalls = new Map<string, TrackedToolCall>();
+  // Running text/thinking accumulators for delta-space repair across chunks.
+  let textSoFar = "";
+  let thinkingSoFar = "";
+
+  function reset(): void {
+    toolCalls.clear();
+    textSoFar = "";
+    thinkingSoFar = "";
+  }
+
+  function emitText(content: ContentBlock | undefined): void {
+    const raw = extractText(content);
+    if (raw === undefined || raw === "") return;
+    const delta = normalizeStreamingDelta(textSoFar, raw);
+    textSoFar += delta;
+    callbacks.onText?.(delta);
+  }
+
+  function emitThinking(content: ContentBlock | undefined): void {
+    const raw = extractText(content);
+    if (raw === undefined || raw === "") return;
+    const delta = normalizeStreamingDelta(thinkingSoFar, raw);
+    thinkingSoFar += delta;
+    callbacks.onThinking?.(delta);
+  }
+
+  function handleToolCall(update: Extract<SessionUpdate, { sessionUpdate: "tool_call" }>): void {
+    const id = update.toolCallId;
+    if (typeof id !== "string" || id === "") return;
+    toolCalls.set(id, { title: update.title, kind: update.kind, ended: false });
+    const name = toolDisplayName({ title: update.title, kind: update.kind });
+    callbacks.onToolStart?.(name, normalizeToolArgs(update.rawInput));
+  }
+
+  function handleToolCallUpdate(
+    update: Extract<SessionUpdate, { sessionUpdate: "tool_call_update" }>,
+  ): void {
+    const id = update.toolCallId;
+    if (typeof id !== "string" || id === "") return;
+    const tracked = toolCalls.get(id) ?? { ended: false };
+    // Carry forward title/kind from the prior `tool_call` when this update omits
+    // them (a partial update may only set status/output).
+    if (update.title != null) tracked.title = update.title;
+    if (update.kind != null) tracked.kind = update.kind;
+    toolCalls.set(id, tracked);
+
+    const status = update.status;
+    if (status !== "completed" && status !== "failed") {
+      // Intermediate (pending/in_progress) — tracking updated, no callback.
+      return;
+    }
+    if (tracked.ended) return; // already fired a terminal callback
+    tracked.ended = true;
+    const name = toolDisplayName({ title: tracked.title, kind: tracked.kind });
+    callbacks.onToolEnd?.(name, status === "failed", update.rawOutput);
+  }
+
+  function handlePlan(entries: PlanEntry[] | undefined): void {
+    // FULL REPLACEMENT: drop any prior snapshot, surface the new one once.
+    const list = Array.isArray(entries) ? entries : [];
+    callbacks.onThinking?.(formatPlan(list));
+  }
+
+  function handleSessionUpdate(update: SessionUpdate): void {
+    if (!update || typeof update !== "object") return;
+    try {
+      switch (update.sessionUpdate) {
+        case "agent_message_chunk":
+          emitText(update.content);
+          break;
+        case "agent_thought_chunk":
+          emitThinking(update.content);
+          break;
+        case "user_message_chunk":
+          // Echo of user input — ignored in v1.
+          break;
+        case "tool_call":
+          handleToolCall(update);
+          break;
+        case "tool_call_update":
+          handleToolCallUpdate(update);
+          break;
+        case "plan":
+          handlePlan(update.entries);
+          break;
+        case "plan_update":
+          // Treat an incremental plan op as a plan refresh for v1.
+          handlePlan((update as { entries?: PlanEntry[] }).entries);
+          break;
+        case "plan_removed":
+          // Clearing the plan: surface nothing.
+          break;
+        case "available_commands_update":
+        case "current_mode_update":
+        case "config_option_update":
+        case "session_info_update":
+        case "usage_update":
+          // Stored/ignored in v1 — no callback surface.
+          break;
+        default:
+          // Unknown/forward-compat tag — ignore without throwing.
+          break;
+      }
+    } catch {
+      // Tolerant: a malformed/partial update must never break the stream.
+    }
+  }
+
+  return { handleSessionUpdate, reset };
+}

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

@@ -23,6 +23,8 @@ import {
   type StopReason,
 } from "@agentclientprotocol/sdk";
 import { spawnAgent, captureStderr, forceKill, unregisterProcess } from "./process-manager.js";
+import { createEventBridge } from "./event-bridge.js";
+import type { AcpCallbacks } from "./types.js";
 
 /** Default bound for the `initialize` handshake. */
 export const DEFAULT_INITIALIZE_TIMEOUT_MS = 30_000;
@@ -67,6 +69,25 @@ export function createDefaultClientHandler(): Client {
   };
 }
 
+/**
+ * The real client handler (U4): bridges every `session/update` notification into
+ * the engine callbacks via an event bridge so streamed agent text/thinking/tool
+ * activity surfaces in Fusion. The permission floor is still the safe default —
+ * U5 replaces `requestPermission` with the per-category action gate.
+ */
+export function createBridgingClientHandler(callbacks: AcpCallbacks): Client {
+  const bridge = createEventBridge(callbacks);
+  return {
+    async sessionUpdate(params) {
+      bridge.handleSessionUpdate(params.update);
+    },
+    async requestPermission() {
+      // U5 replaces this with the per-category gate; default-cancel for now.
+      return { outcome: { outcome: "cancelled" } };
+    },
+  };
+}
+
 export interface AcpConnection {
   /** Live ACP connection — later units drive session/new, prompt, cancel, load. */
   conn: ClientSideConnection;

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

@@ -14,6 +14,7 @@ import {
   newAcpSession,
   promptAcpSession,
   cancelAcpSession,
+  createBridgingClientHandler,
 } from "./provider.js";
 import { buildSpawnEnv } from "./process-manager.js";
 import { buildPromptBlocks } from "./prompt-builder.js";
@@ -43,6 +44,15 @@ export class AcpRuntimeAdapter implements AgentRuntime {
   async createSession(options: AgentRuntimeOptions): Promise<AgentSessionResult> {
     const model = this.settings.model ?? options.defaultModelId ?? "acp";
 
+    // Bridge streamed `session/update` notifications onto the engine callbacks
+    // (U4) so ACP agents render like existing runtimes.
+    const callbacks = {
+      onText: options.onText,
+      onThinking: options.onThinking,
+      onToolStart: options.onToolStart,
+      onToolEnd: options.onToolEnd,
+    };
+
     // 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).
@@ -52,6 +62,7 @@ export class AcpRuntimeAdapter implements AgentRuntime {
       cwd: options.cwd,
       env: buildSpawnEnv(this.settings.envAllowList),
       advertiseFs: { read: this.settings.fsRead, write: this.settings.fsWrite },
+      clientHandler: createBridgingClientHandler(callbacks),
     });
 
     // Open the ACP session over the task worktree (empty mcpServers — KTD5).
@@ -72,12 +83,7 @@ export class AcpRuntimeAdapter implements AgentRuntime {
       sessionId,
       cwd: options.cwd,
       lastModelDescription: `acp/${model}`,
-      callbacks: {
-        onText: options.onText,
-        onThinking: options.onThinking,
-        onToolStart: options.onToolStart,
-        onToolEnd: options.onToolEnd,
-      },
+      callbacks,
       // Persist the per-run gate (KTD3) so U5/U7 can reach the live action gate.
       gate: options.actionGateContext,
       connection,
@@ -103,8 +109,8 @@ export class AcpRuntimeAdapter implements AgentRuntime {
     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.
+    // The bridging client handler installed at createSession (U4) has already
+    // surfaced streamed text/thinking/tool updates onto session.callbacks.
     await promptAcpSession(acp.connection, acp.sessionId, blocks);
   }
 

plugins/fusion-plugin-acp-runtime/src/tool-mapping.ts

@@ -0,0 +1,46 @@
+// Pure helpers mapping ACP `ToolCall` metadata into the display name + args
+// shape Fusion's `onToolStart`/`onToolEnd` callbacks expect.
+//
+// ACP's `kind` is agent-defined, optional, and partial (U4). These helpers must
+// never throw on missing/odd input — a missing title falls back to a label
+// derived from `kind`, and a missing/non-object `rawInput` normalizes to `{}`.
+
+import type { ToolKind } from "@agentclientprotocol/sdk";
+
+/** Human-readable labels for each ACP `ToolKind`. */
+const KIND_LABELS: Record<ToolKind, string> = {
+  read: "Read",
+  edit: "Edit",
+  delete: "Delete",
+  move: "Move",
+  search: "Search",
+  execute: "Execute",
+  think: "Think",
+  fetch: "Fetch",
+  switch_mode: "Switch Mode",
+  other: "Tool",
+};
+
+/**
+ * Resolve a display name for a tool call. Prefers the agent-supplied `title`;
+ * falls back to a label derived from `kind`; final fallback is `"tool"`.
+ */
+export function toolDisplayName(toolCall: { title?: string | null; kind?: ToolKind | null }): string {
+  const title = typeof toolCall.title === "string" ? toolCall.title.trim() : "";
+  if (title) return title;
+  const kind = toolCall.kind;
+  if (kind && kind in KIND_LABELS) return KIND_LABELS[kind];
+  return "tool";
+}
+
+/**
+ * Normalize a tool call's `rawInput` to a plain object. Returns `{}` when the
+ * input is undefined, null, or any non-object (arrays included) so downstream
+ * code can always treat args as a record.
+ */
+export function normalizeToolArgs(rawInput: unknown): Record<string, unknown> {
+  if (rawInput === null || typeof rawInput !== "object" || Array.isArray(rawInput)) {
+    return {};
+  }
+  return rawInput as Record<string, unknown>;
+}