fix(ai-opencode): avoid streaming user parts and empty tool args

.changeset/opencode-stream-parts.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-opencode': patch
+---
+
+Only translate assistant message parts and wait for OpenCode to provide tool input before emitting a tool call.

docs/adapters/opencode.md

@@ -15,6 +15,8 @@ keywords:
 
 The OpenCode adapter runs [OpenCode](https://opencode.ai) as a chat backend, driving it over its local HTTP server (`@opencode-ai/sdk`). Unlike HTTP provider adapters, this is a **harness adapter**: OpenCode runs its own agent loop and executes its own tools — shell commands, file reads and edits, search — locally on your server. Each `chat()` call runs one full harness turn; assistant text and reasoning stream as true token-level deltas, and the harness's tool activity streams back as already-resolved tool-call events your UI can render.
 
+The adapter buffers each message part until it knows the parent message role. It streams assistant parts and skips user parts.
+
 > **Server-only.** The adapter spawns (or attaches to) an `opencode serve` process, so it only works in a Node.js server environment — never in the browser. Treat it like giving OpenCode a shell on the machine it runs on, and configure permissions accordingly.
 
 ## Installation
@@ -192,6 +194,8 @@ Two kinds of tools flow through this adapter:
 
 2. **Your TanStack tools** are bridged *into* the harness: the adapter starts a short-lived Streamable-HTTP MCP server on `127.0.0.1` for the duration of the turn and registers it with OpenCode. Define tools as usual with `toolDefinition().server()`; tool-call events come back under the names you registered (OpenCode prefixes MCP tools `tanstack_…` internally, which the adapter strips).
 
+OpenCode can send a tool part in `pending` state before the tool input is ready. The adapter waits for the next update, so a tool call always streams with its arguments.
+
 ```typescript group=tools
 import { z } from "zod";
 import { chat, toolDefinition } from "@tanstack/ai";

packages/ai-opencode/src/stream/sdk-types.ts

@@ -56,17 +56,30 @@ export type OpencodeToolState =
  * the `is*Part` type guards, so those kinds simply match no guard.
  */
 export type OpencodePart =
-  | { id: string; sessionID?: string; type: 'text'; text: string }
-  | { id: string; sessionID?: string; type: 'reasoning'; text: string }
   | {
       id: string
       sessionID?: string
+      messageID: string
+      type: 'text'
+      text: string
+    }
+  | {
+      id: string
+      sessionID?: string
+      messageID: string
+      type: 'reasoning'
+      text: string
+    }
+  | {
+      id: string
+      sessionID?: string
+      messageID: string
       type: 'tool'
       callID: string
       tool: string
       state: OpencodeToolState
     }
-  | { id: string; sessionID?: string; type: string }
+  | { id: string; sessionID?: string; messageID: string; type: string }
 
 /**
  * The OpenCode events the translator understands. This is a closed
@@ -81,7 +94,9 @@ export type OpencodeEvent =
     }
   | {
       type: 'message.updated'
-      properties: { info: { sessionID?: string } }
+      properties: {
+        info: { id: string; sessionID?: string; role: 'user' | 'assistant' }
+      }
     }
   | { type: 'session.idle'; properties: { sessionID: string } }
   | {

packages/ai-opencode/src/stream/translate.ts

@@ -123,6 +123,13 @@ export async function* translateOpencodeStream(
   const openedToolCalls = new Set<string>()
   /** Tool call ids that already emitted a TOOL_CALL_RESULT. */
   const resolvedToolCalls = new Set<string>()
+  /** Message roles from message.updated events, keyed by message id. */
+  const messageRoles = new Map<string, 'user' | 'assistant'>()
+  /** Parts held until their parent message role arrives. */
+  const pendingParts = new Map<
+    string,
+    Array<{ part: OpencodePart; delta?: string }>
+  >()
 
   /** Accumulated text per text-part id, for delta derivation. */
   const textAccumulators = new Map<string, string>()
@@ -308,6 +315,8 @@ export async function* translateOpencodeStream(
   function* handleToolPart(
     part: Extract<OpencodePart, { type: 'tool' }>,
   ): Generator<AdapterYieldChunk> {
+    if (part.state.status === 'pending') return
+
     yield* closeText()
     yield* closeReasoning()
     yield* openToolCall(part)
@@ -330,18 +339,49 @@ export async function* translateOpencodeStream(
     }
   }
 
+  function* handlePart(
+    part: OpencodePart,
+    delta: string | undefined,
+  ): Generator<AdapterYieldChunk> {
+    if (isTextPart(part)) {
+      yield* handleTextPart(part, delta)
+    } else if (isReasoningPart(part)) {
+      yield* handleReasoningPart(part, delta)
+    } else if (isToolPart(part)) {
+      yield* handleToolPart(part)
+    }
+    // Other part kinds (file, step-start/finish, snapshot, ...) carry no
+    // state the chunk stream needs.
+  }
+
+  function* setMessageRole(
+    messageId: string,
+    role: 'user' | 'assistant',
+  ): Generator<AdapterYieldChunk> {
+    messageRoles.set(messageId, role)
+    const parts = pendingParts.get(messageId)
+    if (parts === undefined) return
+    pendingParts.delete(messageId)
+    if (role === 'user') return
+    for (const { part, delta } of parts) yield* handlePart(part, delta)
+  }
+
   function* handleEvent(event: OpencodeEvent): Generator<AdapterYieldChunk> {
-    if (event.type === 'message.part.updated') {
+    if (event.type === 'message.updated') {
+      yield* setMessageRole(
+        event.properties.info.id,
+        event.properties.info.role,
+      )
+    } else if (event.type === 'message.part.updated') {
       const { part, delta } = event.properties
-      if (isTextPart(part)) {
-        yield* handleTextPart(part, delta)
-      } else if (isReasoningPart(part)) {
-        yield* handleReasoningPart(part, delta)
-      } else if (isToolPart(part)) {
-        yield* handleToolPart(part)
+      const role = messageRoles.get(part.messageID)
+      if (role === undefined) {
+        const parts = pendingParts.get(part.messageID) ?? []
+        parts.push({ part, delta })
+        pendingParts.set(part.messageID, parts)
+      } else if (role === 'assistant') {
+        yield* handlePart(part, delta)
       }
-      // Other part kinds (file, step-start/finish, snapshot, ...) carry no
-      // state the chunk stream needs.
     } else if (event.type === 'todo.updated') {
       yield {
         type: EventType.CUSTOM,
@@ -351,14 +391,18 @@ export async function* translateOpencodeStream(
         value: { todos: event.properties.todos },
       }
     }
-    // session.idle / session.status / message.updated are redundant with the
-    // terminal `done` event and are ignored.
+    // session.idle / session.error are redundant with the terminal `done`
+    // event and are ignored.
   }
 
   function* finish(
     message: OpencodeAssistantMessage,
   ): Generator<AdapterYieldChunk> {
     yield* startRun()
+    // The final message is the assistant reply, so flush any parts that beat
+    // its message.updated. OpenCode sends message.updated when it creates a
+    // message, so parts of any other message still waiting here are dropped.
+    yield* setMessageRole(message.id, 'assistant')
     yield* closeText()
     yield* closeReasoning()
     yield* synthesizeUnresolvedResults()

packages/ai-opencode/tests/translate.test.ts

@@ -55,19 +55,33 @@ function textPart(
   id: string,
   text: string,
   delta?: string,
+  messageID = 'msg-1',
 ): OpencodeStreamEvent {
   return {
     kind: 'event',
     event: {
       type: 'message.part.updated',
       properties: {
-        part: { id, sessionID: 'sess-1', type: 'text', text },
+        part: { id, sessionID: 'sess-1', messageID, type: 'text', text },
         ...(delta !== undefined && { delta }),
       },
     },
   }
 }
 
+function messageUpdated(
+  id: string,
+  role: 'user' | 'assistant',
+): OpencodeStreamEvent {
+  return {
+    kind: 'event',
+    event: {
+      type: 'message.updated',
+      properties: { info: { id, sessionID: 'sess-1', role } },
+    },
+  }
+}
+
 describe('translateOpencodeStream', () => {
   it('translates a simple text turn', async () => {
     const chunks = await collect([
@@ -92,6 +106,43 @@ describe('translateOpencodeStream', () => {
     expect(chunks.at(-1)).toMatchObject({ finishReason: 'stop' })
   })
 
+  it('does not expose user message parts as assistant output', async () => {
+    const chunks = await collect([
+      session,
+      textPart('user-part', 'user prompt', 'user prompt', 'user-message'),
+      messageUpdated('user-message', 'user'),
+      textPart('late-user-part', 'more', 'more', 'user-message'),
+      textPart('a-part', 'assistant answer', 'assistant answer', 'a-message'),
+      messageUpdated('a-message', 'assistant'),
+      done({ id: 'a-message' }),
+    ])
+
+    expect(
+      chunks.filter((chunk) => chunk.type === 'TEXT_MESSAGE_CONTENT'),
+    ).toMatchObject([
+      { delta: 'assistant answer', content: 'assistant answer' },
+    ])
+  })
+
+  it('streams parts as they arrive once the assistant role is known', async () => {
+    const log: Array<string> = []
+    async function* source(): AsyncIterable<OpencodeStreamEvent> {
+      yield session
+      yield messageUpdated('msg-1', 'assistant')
+      yield textPart('part-1', 'hi', 'hi')
+      log.push('done-pulled')
+      yield done()
+    }
+
+    for await (const chunk of translateOpencodeStream(source(), makeCtx())) {
+      log.push(chunk.type)
+    }
+
+    expect(log.indexOf('TEXT_MESSAGE_CONTENT')).toBeLessThan(
+      log.indexOf('done-pulled'),
+    )
+  })
+
   it('accumulates incremental text deltas', async () => {
     const chunks = await collect([
       session,
@@ -162,6 +213,7 @@ describe('translateOpencodeStream', () => {
             part: {
               id: 'r-1',
               sessionID: 'sess-1',
+              messageID: 'msg-1',
               type: 'reasoning',
               text: 'thinking',
             },
@@ -196,6 +248,7 @@ describe('translateOpencodeStream', () => {
           part: {
             id: `part-${callID}`,
             sessionID: 'sess-1',
+            messageID: 'msg-1',
             type: 'tool',
             callID,
             tool,
@@ -239,6 +292,27 @@ describe('translateOpencodeStream', () => {
     expect((chunks[5] as { state?: string }).state).toBeUndefined()
   })
 
+  it('waits for tool input before opening a pending tool call', async () => {
+    const chunks = await collect([
+      session,
+      toolEvent('call-pending', 'bash', { status: 'pending', input: {} }),
+      toolEvent('call-pending', 'bash', {
+        status: 'running',
+        input: { command: 'ls' },
+      }),
+      toolEvent('call-pending', 'bash', {
+        status: 'completed',
+        input: { command: 'ls' },
+        output: 'file.txt',
+      }),
+      done(),
+    ])
+
+    expect(
+      chunks.filter((chunk) => chunk.type === 'TOOL_CALL_ARGS'),
+    ).toMatchObject([{ args: JSON.stringify({ command: 'ls' }) }])
+  })
+
   it('marks tool errors as output-error', async () => {
     const chunks = await collect([
       session,
@@ -357,6 +431,7 @@ describe('translateOpencodeStream', () => {
   it('synthesizes results then rethrows when the source stream throws', async () => {
     async function* failing(): AsyncIterable<OpencodeStreamEvent> {
       yield session
+      yield messageUpdated('msg-1', 'assistant')
       yield {
         kind: 'event',
         event: {
@@ -365,6 +440,7 @@ describe('translateOpencodeStream', () => {
             part: {
               id: 'p-c',
               sessionID: 'sess-1',
+              messageID: 'msg-1',
               type: 'tool',
               callID: 'call-7',
               tool: 'bash',