diff --git a/.changeset/opencode-stream-parts.md b/.changeset/opencode-stream-parts.md new file mode 100644 index 0000000000..a437b79ad2 --- /dev/null +++ b/.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. diff --git a/docs/adapters/opencode.md b/docs/adapters/opencode.md index 88309debc8..d7cf003ca1 100644 --- a/docs/adapters/opencode.md +++ b/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"; diff --git a/packages/ai-opencode/src/stream/sdk-types.ts b/packages/ai-opencode/src/stream/sdk-types.ts index 8deb3ce87a..f80278a476 100644 --- a/packages/ai-opencode/src/stream/sdk-types.ts +++ b/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 } } | { diff --git a/packages/ai-opencode/src/stream/translate.ts b/packages/ai-opencode/src/stream/translate.ts index a20c901f53..aff410bf40 100644 --- a/packages/ai-opencode/src/stream/translate.ts +++ b/packages/ai-opencode/src/stream/translate.ts @@ -123,6 +123,13 @@ export async function* translateOpencodeStream( const openedToolCalls = new Set() /** Tool call ids that already emitted a TOOL_CALL_RESULT. */ const resolvedToolCalls = new Set() + /** Message roles from message.updated events, keyed by message id. */ + const messageRoles = new Map() + /** 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() @@ -308,6 +315,8 @@ export async function* translateOpencodeStream( function* handleToolPart( part: Extract, ): Generator { + 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 { + 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 { + 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 { - 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 { 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() diff --git a/packages/ai-opencode/tests/translate.test.ts b/packages/ai-opencode/tests/translate.test.ts index f0dc73ca25..5ec94c31f6 100644 --- a/packages/ai-opencode/tests/translate.test.ts +++ b/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 = [] + async function* source(): AsyncIterable { + 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 { 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',