feat(acp): stream acp-next tool updates

nexxeln · nexxeln · commit b0f7068801e2 · 2026-05-26T12:32:45.000+05:30
diff --git a/packages/opencode/src/acp-next/event.ts b/packages/opencode/src/acp-next/event.ts
@@ -7,9 +7,18 @@ import type {
   OpencodeClient,
   Part,
   SessionMessageResponse,
+  ToolPart,
 } from "@opencode-ai/sdk/v2"
 import { Effect } from "effect"
 import { ACPNextSession } from "./session"
+import {
+  duplicateRunningToolUpdate,
+  errorToolUpdate,
+  pendingToolCall,
+  runningToolUpdate,
+  shellOutputSnapshot,
+  completedToolUpdate,
+} from "./tool"
 
 const log = Log.create({ service: "acp-next-event" })
 
@@ -29,6 +38,8 @@ export function start(input: { sdk: OpencodeClient; connection: Connection; sess
 
 export class Subscription {
   private readonly abort = new AbortController()
+  private readonly shellSnapshots = new Map<string, string>()
+  private readonly toolStarts = new Set<string>()
   private started = false
 
   constructor(
@@ -61,6 +72,17 @@ export class Subscription {
     }
   }
 
+  async replayMessage(message: SessionMessageResponse) {
+    if (message.info.role !== "assistant" && message.info.role !== "user") return
+
+    for (const part of message.parts) {
+      await this.recordFetchedPart(message.info.sessionID, message, part)
+      if (part.type === "tool") {
+        await this.handleToolPart(message.info.sessionID, part)
+      }
+    }
+  }
+
   private async run() {
     while (!this.abort.signal.aborted) {
       const events = (await this.input.sdk.global.event({
@@ -96,6 +118,9 @@ export class Subscription {
         metadata: "metadata" in part ? part.metadata : undefined,
       }),
     )
+    if (part.type === "tool") {
+      await this.handleToolPart(session.id, part)
+    }
   }
 
   private async handlePartDelta(event: EventMessagePartDelta) {
@@ -181,6 +206,106 @@ export class Subscription {
       }),
     )
   }
+
+  private async handleToolPart(sessionId: string, part: ToolPart) {
+    await this.toolStart(sessionId, part)
+
+    switch (part.state.status) {
+      case "pending":
+        this.shellSnapshots.delete(part.callID)
+        return
+
+      case "running":
+        await this.runningTool(sessionId, part)
+        return
+
+      case "completed":
+        this.clearTool(part.callID)
+        await this.input.connection.sessionUpdate({
+          sessionId,
+          update: {
+            sessionUpdate: "tool_call_update",
+            ...completedToolUpdate({
+              toolCallId: part.callID,
+              toolName: part.tool,
+              state: part.state,
+            }),
+          },
+        })
+        return
+
+      case "error":
+        this.clearTool(part.callID)
+        await this.input.connection.sessionUpdate({
+          sessionId,
+          update: {
+            sessionUpdate: "tool_call_update",
+            ...errorToolUpdate({
+              toolCallId: part.callID,
+              toolName: part.tool,
+              state: part.state,
+            }),
+          },
+        })
+        return
+    }
+  }
+
+  private async runningTool(sessionId: string, part: ToolPart) {
+    if (part.state.status !== "running") return
+
+    const output = part.tool === "bash" ? shellOutputSnapshot(part.state) : undefined
+    if (output !== undefined) {
+      if (this.shellSnapshots.get(part.callID) === output) {
+        await this.input.connection.sessionUpdate({
+          sessionId,
+          update: {
+            sessionUpdate: "tool_call_update",
+            ...duplicateRunningToolUpdate({
+              toolCallId: part.callID,
+              toolName: part.tool,
+              state: part.state,
+            }),
+          },
+        })
+        return
+      }
+      this.shellSnapshots.set(part.callID, output)
+    }
+
+    await this.input.connection.sessionUpdate({
+      sessionId,
+      update: {
+        sessionUpdate: "tool_call_update",
+        ...runningToolUpdate({
+          toolCallId: part.callID,
+          toolName: part.tool,
+          state: part.state,
+          output,
+        }),
+      },
+    })
+  }
+
+  private async toolStart(sessionId: string, part: ToolPart) {
+    if (this.toolStarts.has(part.callID)) return
+    this.toolStarts.add(part.callID)
+    await this.input.connection.sessionUpdate({
+      sessionId,
+      update: {
+        sessionUpdate: "tool_call",
+        ...pendingToolCall({
+          toolCallId: part.callID,
+          toolName: part.tool,
+        }),
+      },
+    })
+  }
+
+  private clearTool(toolCallId: string) {
+    this.toolStarts.delete(toolCallId)
+    this.shellSnapshots.delete(toolCallId)
+  }
 }
 
 export * as ACPNextEvent from "./event"
diff --git a/packages/opencode/src/acp-next/service.ts b/packages/opencode/src/acp-next/service.ts
@@ -31,7 +31,7 @@ import {
 } from "@agentclientprotocol/sdk"
 import { InstallationVersion } from "@opencode-ai/core/installation/version"
 import * as Log from "@opencode-ai/core/util/log"
-import type { Message, OpencodeClient } from "@opencode-ai/sdk/v2"
+import type { Message, OpencodeClient, SessionMessageResponse } from "@opencode-ai/sdk/v2"
 import { Context, Effect, Layer, ManagedRuntime } from "effect"
 import * as ACPNextError from "./error"
 import { buildConfigOptions, parseModelSelection } from "./config-option"
@@ -77,10 +77,10 @@ export function make(input: {
   const session = input.session ?? makeSessionService()
   const directoryService = input.directory ?? makeDirectoryService(input.sdk)
   const registeredMcp = new Map<string, Set<string>>()
-  if (input.connection) {
-    const subscription = ACPNextEvent.start({ sdk: input.sdk, connection: input.connection, session })
-    input.eventSubscription?.(subscription)
-  }
+  const events = input.connection
+    ? ACPNextEvent.start({ sdk: input.sdk, connection: input.connection, session })
+    : undefined
+  if (events) input.eventSubscription?.(events)
 
   const initialize = Effect.fn("ACPNext.initialize")(function* (params: InitializeRequest) {
     const authMethod: AuthMethod = {
@@ -207,6 +207,7 @@ export function make(input: {
 
     yield* registerMcpServers(input.sdk, registeredMcp, params.cwd, state.id, params.mcpServers)
     yield* sendAvailableCommands(input.connection, state.id, snapshot)
+    yield* replayMessages(events, messages)
 
     return {
       configOptions: configOptions(snapshot, {
@@ -276,6 +277,7 @@ export function make(input: {
 
     yield* registerMcpServers(input.sdk, registeredMcp, params.cwd, state.id, params.mcpServers ?? [])
     yield* sendAvailableCommands(input.connection, state.id, snapshot)
+    yield* replayMessages(events, messages)
 
     return {
       configOptions: configOptions(snapshot, {
@@ -335,6 +337,7 @@ export function make(input: {
 
     yield* registerMcpServers(input.sdk, registeredMcp, params.cwd, state.id, params.mcpServers ?? [])
     yield* sendAvailableCommands(input.connection, state.id, snapshot)
+    yield* replayMessages(events, messages)
 
     return {
       sessionId: state.id,
@@ -470,6 +473,14 @@ function makeDirectoryService(sdk: OpencodeClient) {
   ).runSync(Directory.Service.use((service) => Effect.succeed(service)))
 }
 
+function replayMessages(subscription: ACPNextEvent.Subscription | undefined, messages: SessionMessageResponse[]) {
+  if (!subscription) return Effect.void
+  return Effect.tryPromise({
+    try: () => Promise.all(messages.map((message) => subscription.replayMessage(message))).then(() => undefined),
+    catch: (error) => fromUnknownError(error, "event"),
+  })
+}
+
 type ConfigState = {
   readonly model: Directory.DefaultModel
   readonly variant?: string
diff --git a/packages/opencode/src/acp-next/tool.ts b/packages/opencode/src/acp-next/tool.ts
@@ -1,4 +1,4 @@
-import type { ToolCallContent, ToolCallLocation, ToolKind } from "@agentclientprotocol/sdk"
+import type { ToolCall, ToolCallContent, ToolCallLocation, ToolCallUpdate, ToolKind } from "@agentclientprotocol/sdk"
 
 export type ToolInput = Record<string, unknown>
 
@@ -16,6 +16,19 @@ export type CompletedToolState = {
   readonly attachments?: ReadonlyArray<ToolAttachment>
 }
 
+export type RunningToolState = {
+  readonly status: "running"
+  readonly input: ToolInput
+  readonly title?: string
+}
+
+export type ErrorToolState = {
+  readonly status: "error"
+  readonly input: ToolInput
+  readonly error: string
+  readonly metadata?: unknown
+}
+
 export type ImageAttachment = {
   readonly mimeType: string
   readonly data: string
@@ -100,6 +113,104 @@ export function completedToolContent(toolName: string, state: CompletedToolState
   return content
 }
 
+export function pendingToolCall(input: { readonly toolCallId: string; readonly toolName: string }): ToolCall {
+  return {
+    toolCallId: input.toolCallId,
+    title: input.toolName,
+    kind: toToolKind(input.toolName),
+    status: "pending",
+    locations: [],
+    rawInput: {},
+  }
+}
+
+export function runningToolUpdate(input: {
+  readonly toolCallId: string
+  readonly toolName: string
+  readonly state: RunningToolState
+  readonly output?: string
+}): ToolCallUpdate {
+  const content = input.output
+    ? [
+        {
+          type: "content" as const,
+          content: {
+            type: "text" as const,
+            text: input.output,
+          },
+        },
+      ]
+    : undefined
+
+  return {
+    toolCallId: input.toolCallId,
+    status: "in_progress",
+    kind: toToolKind(input.toolName),
+    title: input.state.title ?? input.toolName,
+    locations: toLocations(input.toolName, input.state.input),
+    rawInput: input.state.input,
+    ...(content ? { content } : {}),
+  }
+}
+
+export function duplicateRunningToolUpdate(input: {
+  readonly toolCallId: string
+  readonly toolName: string
+  readonly state: RunningToolState
+}): ToolCallUpdate {
+  return {
+    toolCallId: input.toolCallId,
+    status: "in_progress",
+    kind: toToolKind(input.toolName),
+    title: input.state.title ?? input.toolName,
+    locations: toLocations(input.toolName, input.state.input),
+    rawInput: input.state.input,
+  }
+}
+
+export function completedToolUpdate(input: {
+  readonly toolCallId: string
+  readonly toolName: string
+  readonly state: CompletedToolState & { readonly title: string }
+}): ToolCallUpdate {
+  return {
+    toolCallId: input.toolCallId,
+    status: "completed",
+    kind: toToolKind(input.toolName),
+    title: input.state.title,
+    content: completedToolContent(input.toolName, input.state),
+    rawInput: input.state.input,
+    rawOutput: completedToolRawOutput(input.state),
+  }
+}
+
+export function errorToolUpdate(input: {
+  readonly toolCallId: string
+  readonly toolName: string
+  readonly state: ErrorToolState
+}): ToolCallUpdate {
+  return {
+    toolCallId: input.toolCallId,
+    status: "failed",
+    kind: toToolKind(input.toolName),
+    title: input.toolName,
+    rawInput: input.state.input,
+    content: [
+      {
+        type: "content",
+        content: {
+          type: "text",
+          text: input.state.error,
+        },
+      },
+    ],
+    rawOutput: {
+      error: input.state.error,
+      metadata: input.state.metadata,
+    },
+  }
+}
+
 export function completedToolRawOutput(state: CompletedToolState) {
   return {
     output: state.output,
@@ -138,6 +249,11 @@ export const extractLocations = toLocations
 export const buildCompletedToolContent = completedToolContent
 export const buildCompletedRawOutput = completedToolRawOutput
 export const extractShellOutputSnapshot = shellOutputSnapshot
+export const buildPendingToolCall = pendingToolCall
+export const buildRunningToolUpdate = runningToolUpdate
+export const buildDuplicateRunningToolUpdate = duplicateRunningToolUpdate
+export const buildCompletedToolUpdate = completedToolUpdate
+export const buildErrorToolUpdate = errorToolUpdate
 
 function locationFrom(value: unknown): ToolCallLocation[] {
   const path = stringValue(value)
diff --git a/packages/opencode/test/acp-next/event.test.ts b/packages/opencode/test/acp-next/event.test.ts
