From c5ae5572a201d0534e93d5856f41637e2f6d674b Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 08:55:20 -0700 Subject: [PATCH] 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) --- .../src/__tests__/event-bridge.test.ts | 272 ++++++++++++++++++ .../src/__tests__/fixtures/echo-agent.mjs | 52 ++++ .../src/__tests__/provider-session.test.ts | 30 +- .../src/__tests__/tool-mapping.test.ts | 45 +++ .../src/event-bridge.ts | 190 ++++++++++++ .../fusion-plugin-acp-runtime/src/provider.ts | 21 ++ .../src/runtime-adapter.ts | 22 +- .../src/tool-mapping.ts | 46 +++ 8 files changed, 669 insertions(+), 9 deletions(-) create mode 100644 plugins/fusion-plugin-acp-runtime/src/__tests__/event-bridge.test.ts create mode 100644 plugins/fusion-plugin-acp-runtime/src/__tests__/tool-mapping.test.ts create mode 100644 plugins/fusion-plugin-acp-runtime/src/event-bridge.ts create mode 100644 plugins/fusion-plugin-acp-runtime/src/tool-mapping.ts diff --git a/plugins/fusion-plugin-acp-runtime/src/__tests__/event-bridge.test.ts b/plugins/fusion-plugin-acp-runtime/src/__tests__/event-bridge.test.ts new file mode 100644 index 0000000000..f42c1d50ca --- /dev/null +++ b/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."]); + }); +}); diff --git a/plugins/fusion-plugin-acp-runtime/src/__tests__/fixtures/echo-agent.mjs b/plugins/fusion-plugin-acp-runtime/src/__tests__/fixtures/echo-agent.mjs index cbf1fa17c4..a93a85ce8b 100644 --- a/plugins/fusion-plugin-acp-runtime/src/__tests__/fixtures/echo-agent.mjs +++ b/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: { diff --git a/plugins/fusion-plugin-acp-runtime/src/__tests__/provider-session.test.ts b/plugins/fusion-plugin-acp-runtime/src/__tests__/provider-session.test.ts index 90364e053e..a6d75d656b 100644 --- a/plugins/fusion-plugin-acp-runtime/src/__tests__/provider-session.test.ts +++ b/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 { diff --git a/plugins/fusion-plugin-acp-runtime/src/__tests__/tool-mapping.test.ts b/plugins/fusion-plugin-acp-runtime/src/__tests__/tool-mapping.test.ts new file mode 100644 index 0000000000..498c7d5d20 --- /dev/null +++ b/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({}); + }); +}); diff --git a/plugins/fusion-plugin-acp-runtime/src/event-bridge.ts b/plugins/fusion-plugin-acp-runtime/src/event-bridge.ts new file mode 100644 index 0000000000..f278adfaf0 --- /dev/null +++ b/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(); + // 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): 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, + ): 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 }; +} diff --git a/plugins/fusion-plugin-acp-runtime/src/provider.ts b/plugins/fusion-plugin-acp-runtime/src/provider.ts index 10020d4306..ade90f09aa 100644 --- a/plugins/fusion-plugin-acp-runtime/src/provider.ts +++ b/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; diff --git a/plugins/fusion-plugin-acp-runtime/src/runtime-adapter.ts b/plugins/fusion-plugin-acp-runtime/src/runtime-adapter.ts index 8bc2c951bd..1348472463 100644 --- a/plugins/fusion-plugin-acp-runtime/src/runtime-adapter.ts +++ b/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 { 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); } diff --git a/plugins/fusion-plugin-acp-runtime/src/tool-mapping.ts b/plugins/fusion-plugin-acp-runtime/src/tool-mapping.ts new file mode 100644 index 0000000000..ad5e0a620c --- /dev/null +++ b/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 = { + 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 { + if (rawInput === null || typeof rawInput !== "object" || Array.isArray(rawInput)) { + return {}; + } + return rawInput as Record; +}