From 736d54e46432d4b158ae80bb9a7d46ff333db993 Mon Sep 17 00:00:00 2001 From: duohelingdukele Date: Thu, 22 Oct 2026 10:31:26 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix(ai-opencode):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E4=B8=8E=E5=B7=A5=E5=85=B7=E5=8F=82=E6=95=B0?= =?UTF-8?q?=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 过滤 OpenCode 用户消息的 parts，并等工具输入解析后再开始输出工具调用。 --- .changeset/opencode-stream-parts.md | 5 ++ packages/ai-opencode/src/stream/sdk-types.ts | 23 ++++- packages/ai-opencode/src/stream/translate.ts | 10 ++- packages/ai-opencode/tests/translate.test.ts | 88 ++++++++++++++++++++ 4 files changed, 121 insertions(+), 5 deletions(-) create mode 100644 .changeset/opencode-stream-parts.md 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/packages/ai-opencode/src/stream/sdk-types.ts b/packages/ai-opencode/src/stream/sdk-types.ts index 8deb3ce87a..fca5f8d8cb 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..7bf8736a54 100644 --- a/packages/ai-opencode/src/stream/translate.ts +++ b/packages/ai-opencode/src/stream/translate.ts @@ -123,6 +123,8 @@ 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() /** Accumulated text per text-part id, for delta derivation. */ const textAccumulators = new Map() @@ -308,6 +310,8 @@ export async function* translateOpencodeStream( function* handleToolPart( part: Extract, ): Generator { + if (part.state.status === 'pending') return + yield* closeText() yield* closeReasoning() yield* openToolCall(part) @@ -331,8 +335,12 @@ export async function* translateOpencodeStream( } function* handleEvent(event: OpencodeEvent): Generator { - if (event.type === 'message.part.updated') { + if (event.type === 'message.updated') { + messageRoles.set(event.properties.info.id, event.properties.info.role) + } else if (event.type === 'message.part.updated') { const { part, delta } = event.properties + if (part.messageID && messageRoles.get(part.messageID) === 'user') return + if (isTextPart(part)) { yield* handleTextPart(part, delta) } else if (isReasoningPart(part)) { diff --git a/packages/ai-opencode/tests/translate.test.ts b/packages/ai-opencode/tests/translate.test.ts index f0dc73ca25..3a5b143e2d 100644 --- a/packages/ai-opencode/tests/translate.test.ts +++ b/packages/ai-opencode/tests/translate.test.ts @@ -92,6 +92,73 @@ 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, + { + kind: 'event', + event: { + type: 'message.updated', + properties: { + info: { id: 'user-message', sessionID: 'sess-1', role: 'user' }, + }, + }, + }, + { + kind: 'event', + event: { + type: 'message.part.updated', + properties: { + part: { + id: 'user-part', + sessionID: 'sess-1', + messageID: 'user-message', + type: 'text', + text: 'user prompt', + }, + delta: 'user prompt', + }, + }, + }, + { + kind: 'event', + event: { + type: 'message.updated', + properties: { + info: { + id: 'assistant-message', + sessionID: 'sess-1', + role: 'assistant', + }, + }, + }, + }, + { + kind: 'event', + event: { + type: 'message.part.updated', + properties: { + part: { + id: 'assistant-part', + sessionID: 'sess-1', + messageID: 'assistant-message', + type: 'text', + text: 'assistant answer', + }, + delta: 'assistant answer', + }, + }, + }, + done(), + ]) + + expect( + chunks.filter((chunk) => chunk.type === 'TEXT_MESSAGE_CONTENT'), + ).toMatchObject([ + { delta: 'assistant answer', content: 'assistant answer' }, + ]) + }) + it('accumulates incremental text deltas', async () => { const chunks = await collect([ session, @@ -239,6 +306,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, From df78ecfb9633cfce84d7ff55ee2d57747bc45494 Mon Sep 17 00:00:00 2001 From: duohelingdukele Date: Thu, 22 Oct 2026 11:24:14 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(ai-opencode):=20=E7=AD=89=E5=BE=85?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E8=A7=92=E8=89=B2=E5=90=8E=E8=BD=AC=E6=8D=A2?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E5=88=86=E7=89=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 缓存尚未收到父消息角色的 OpenCode part，确定为 assistant 后再转换，user part 则丢弃；缺少 messageID 时显式失败。 补充 pending 工具参数行为文档，以及覆盖角色乱序和真实工具输入的 E2E 用例。 --- docs/adapters/opencode.md | 4 + packages/ai-opencode/src/stream/sdk-types.ts | 8 +- packages/ai-opencode/src/stream/translate.ts | 63 ++++- packages/ai-opencode/tests/translate.test.ts | 74 ++++- pnpm-lock.yaml | 3 + testing/e2e/README.md | 1 + testing/e2e/package.json | 1 + .../tests/tools-test/opencode-stream.spec.ts | 257 ++++++++++++++++++ 8 files changed, 381 insertions(+), 30 deletions(-) create mode 100644 testing/e2e/tests/tools-test/opencode-stream.spec.ts diff --git a/docs/adapters/opencode.md b/docs/adapters/opencode.md index 88309debc8..af09346e17 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 @@ -188,6 +190,8 @@ Sessions live on the server that ran them, so resuming only works against the sa Two kinds of tools flow through this adapter: +OpenCode can send a tool part with `pending` state before it provides tool input. The adapter waits for a non-pending update before it streams the tool call and its arguments. + 1. **Built-in harness tools** are executed by OpenCode itself and stream back as tool-call events with results already attached: `bash`, `edit`, `write`, `read`, `grep`, and the agent's running todo plan (surfaced as an `opencode.todo` custom event). Your code never executes them. 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). diff --git a/packages/ai-opencode/src/stream/sdk-types.ts b/packages/ai-opencode/src/stream/sdk-types.ts index fca5f8d8cb..f80278a476 100644 --- a/packages/ai-opencode/src/stream/sdk-types.ts +++ b/packages/ai-opencode/src/stream/sdk-types.ts @@ -59,27 +59,27 @@ export type OpencodePart = | { id: string sessionID?: string - messageID?: string + messageID: string type: 'text' text: string } | { id: string sessionID?: string - messageID?: string + messageID: string type: 'reasoning' text: string } | { id: string sessionID?: string - messageID?: string + messageID: string type: 'tool' callID: string tool: string state: OpencodeToolState } - | { id: string; sessionID?: string; messageID?: string; type: string } + | { id: string; sessionID?: string; messageID: string; type: string } /** * The OpenCode events the translator understands. This is a closed diff --git a/packages/ai-opencode/src/stream/translate.ts b/packages/ai-opencode/src/stream/translate.ts index 7bf8736a54..c8757440c7 100644 --- a/packages/ai-opencode/src/stream/translate.ts +++ b/packages/ai-opencode/src/stream/translate.ts @@ -125,6 +125,11 @@ export async function* translateOpencodeStream( 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() @@ -334,22 +339,54 @@ 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.updated') { - messageRoles.set(event.properties.info.id, event.properties.info.role) + yield* setMessageRole( + event.properties.info.id, + event.properties.info.role, + ) } else if (event.type === 'message.part.updated') { const { part, delta } = event.properties - if (part.messageID && messageRoles.get(part.messageID) === 'user') return - - if (isTextPart(part)) { - yield* handleTextPart(part, delta) - } else if (isReasoningPart(part)) { - yield* handleReasoningPart(part, delta) - } else if (isToolPart(part)) { - yield* handleToolPart(part) + if (!part.messageID) { + throw new Error( + `OpenCode message part "${part.id}" is missing its messageID.`, + ) + } + 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, @@ -367,6 +404,10 @@ export async function* translateOpencodeStream( message: OpencodeAssistantMessage, ): Generator { yield* startRun() + if (!messageRoles.has(message.id)) { + yield* setMessageRole(message.id, 'assistant') + } + pendingParts.clear() 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 3a5b143e2d..f8485a7197 100644 --- a/packages/ai-opencode/tests/translate.test.ts +++ b/packages/ai-opencode/tests/translate.test.ts @@ -61,7 +61,13 @@ function textPart( event: { type: 'message.part.updated', properties: { - part: { id, sessionID: 'sess-1', type: 'text', text }, + part: { + id, + sessionID: 'sess-1', + messageID: 'msg-1', + type: 'text', + text, + }, ...(delta !== undefined && { delta }), }, }, @@ -95,15 +101,6 @@ describe('translateOpencodeStream', () => { it('does not expose user message parts as assistant output', async () => { const chunks = await collect([ session, - { - kind: 'event', - event: { - type: 'message.updated', - properties: { - info: { id: 'user-message', sessionID: 'sess-1', role: 'user' }, - }, - }, - }, { kind: 'event', event: { @@ -125,11 +122,7 @@ describe('translateOpencodeStream', () => { event: { type: 'message.updated', properties: { - info: { - id: 'assistant-message', - sessionID: 'sess-1', - role: 'assistant', - }, + info: { id: 'user-message', sessionID: 'sess-1', role: 'user' }, }, }, }, @@ -149,6 +142,19 @@ describe('translateOpencodeStream', () => { }, }, }, + { + kind: 'event', + event: { + type: 'message.updated', + properties: { + info: { + id: 'assistant-message', + sessionID: 'sess-1', + role: 'assistant', + }, + }, + }, + }, done(), ]) @@ -159,6 +165,28 @@ describe('translateOpencodeStream', () => { ]) }) + it('rejects a message part without a message id', async () => { + const missingMessageId = { + kind: 'event', + event: { + type: 'message.part.updated', + properties: { + part: { + id: 'orphan-part', + sessionID: 'sess-1', + type: 'text', + text: 'orphan', + }, + delta: 'orphan', + }, + }, + } as unknown as OpencodeStreamEvent + + await expect(collect([session, missingMessageId])).rejects.toThrow( + 'OpenCode message part "orphan-part" is missing its messageID.', + ) + }) + it('accumulates incremental text deltas', async () => { const chunks = await collect([ session, @@ -229,6 +257,7 @@ describe('translateOpencodeStream', () => { part: { id: 'r-1', sessionID: 'sess-1', + messageID: 'msg-1', type: 'reasoning', text: 'thinking', }, @@ -263,6 +292,7 @@ describe('translateOpencodeStream', () => { part: { id: `part-${callID}`, sessionID: 'sess-1', + messageID: 'msg-1', type: 'tool', callID, tool, @@ -445,6 +475,19 @@ describe('translateOpencodeStream', () => { it('synthesizes results then rethrows when the source stream throws', async () => { async function* failing(): AsyncIterable { yield session + yield { + kind: 'event', + event: { + type: 'message.updated', + properties: { + info: { + id: 'msg-1', + sessionID: 'sess-1', + role: 'assistant', + }, + }, + }, + } yield { kind: 'event', event: { @@ -453,6 +496,7 @@ describe('translateOpencodeStream', () => { part: { id: 'p-c', sessionID: 'sess-1', + messageID: 'msg-1', type: 'tool', callID: 'call-7', tool: 'bash', diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 931cf0e710..98deadac56 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -3867,6 +3867,9 @@ importers: '@tanstack/ai-mcp': specifier: workspace:* version: link:../../packages/ai-mcp + '@tanstack/ai-opencode': + specifier: workspace:* + version: link:../../packages/ai-opencode '@tanstack/ai-memory': specifier: workspace:* version: link:../../packages/ai-memory diff --git a/testing/e2e/README.md b/testing/e2e/README.md index e81b8286ec..ca5c5ed015 100644 --- a/testing/e2e/README.md +++ b/testing/e2e/README.md @@ -53,6 +53,7 @@ Deterministic scenarios covering tool execution flows: | `tests/tools-test/client-tool.spec.ts` | 5 | Single, sequential, parallel, triple, server+client | | `tests/tools-test/race-conditions.spec.ts` | 8 | No blocking, no deadlocks, timing, mixed flows | | `tests/tools-test/server-client-sequence.spec.ts` | 5 | Server→client, parallel server, ordering | +| `tests/tools-test/opencode-stream.spec.ts` | 1 | OpenCode message roles and pending tool input | ### Interrupt playground diff --git a/testing/e2e/package.json b/testing/e2e/package.json index 108d79526a..e32c6545c6 100644 --- a/testing/e2e/package.json +++ b/testing/e2e/package.json @@ -37,6 +37,7 @@ "@tanstack/ai-llmgateway": "workspace:*", "@tanstack/ai-lovable": "workspace:*", "@tanstack/ai-mcp": "workspace:*", + "@tanstack/ai-opencode": "workspace:*", "@tanstack/ai-memory": "workspace:*", "@tanstack/ai-mistral": "workspace:*", "@tanstack/ai-ollama": "workspace:*", diff --git a/testing/e2e/tests/tools-test/opencode-stream.spec.ts b/testing/e2e/tests/tools-test/opencode-stream.spec.ts new file mode 100644 index 0000000000..40d6a0a765 --- /dev/null +++ b/testing/e2e/tests/tools-test/opencode-stream.spec.ts @@ -0,0 +1,257 @@ +import { createServer } from 'node:http' +import type { ServerResponse } from 'node:http' +import type { AddressInfo } from 'node:net' +import { chat, defineChatMiddleware } from '@tanstack/ai' +import type { StreamChunk } from '@tanstack/ai' +import { opencodeText } from '@tanstack/ai-opencode' +import { provideSandbox, SandboxCapability } from '@tanstack/ai-sandbox' +import type { SandboxHandle, SpawnHandle } from '@tanstack/ai-sandbox' +import { expect, test } from '@playwright/test' + +const MODEL = 'anthropic/claude-sonnet-4-5' + +const assistantMessage = { + id: 'assistant-message', + sessionID: 'session-1', + role: 'assistant', + finish: 'stop', + tokens: { input: 2, output: 2 }, +} + +const opencodeEvents = [ + { + type: 'message.part.updated', + properties: { + part: { + id: 'user-part', + sessionID: 'session-1', + messageID: 'user-message', + type: 'text', + text: 'user prompt', + }, + delta: 'user prompt', + }, + }, + { + type: 'message.updated', + properties: { + info: { id: 'user-message', sessionID: 'session-1', role: 'user' }, + }, + }, + { + type: 'message.part.updated', + properties: { + part: { + id: 'assistant-part', + sessionID: 'session-1', + messageID: 'assistant-message', + type: 'text', + text: 'assistant answer', + }, + delta: 'assistant answer', + }, + }, + { + type: 'message.updated', + properties: { + info: assistantMessage, + }, + }, + { + type: 'message.part.updated', + properties: { + part: { + id: 'tool-part', + sessionID: 'session-1', + messageID: 'assistant-message', + type: 'tool', + callID: 'call-1', + tool: 'bash', + state: { status: 'pending', input: {} }, + }, + }, + }, + { + type: 'message.part.updated', + properties: { + part: { + id: 'tool-part', + sessionID: 'session-1', + messageID: 'assistant-message', + type: 'tool', + callID: 'call-1', + tool: 'bash', + state: { status: 'running', input: { command: 'ls' } }, + }, + }, + }, + { + type: 'message.part.updated', + properties: { + part: { + id: 'tool-part', + sessionID: 'session-1', + messageID: 'assistant-message', + type: 'tool', + callID: 'call-1', + tool: 'bash', + state: { + status: 'completed', + input: { command: 'ls' }, + output: 'file.txt', + }, + }, + }, + }, +] + +async function createMockOpenCodeServer(): Promise<{ + url: string + close: () => Promise +}> { + let eventStream: ServerResponse | undefined + let resolveEventStream: () => void = () => {} + const eventStreamReady = new Promise((resolve) => { + resolveEventStream = resolve + }) + + const server = createServer((request, response) => { + const pathname = new URL(request.url ?? '/', 'http://127.0.0.1').pathname + if (request.method === 'POST' && pathname === '/session') { + response.writeHead(200, { 'Content-Type': 'application/json' }) + response.end(JSON.stringify({ id: 'session-1' })) + return + } + + if (request.method === 'GET' && pathname === '/event') { + response.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }) + eventStream = response + resolveEventStream() + return + } + + if ( + request.method === 'POST' && + pathname === '/session/session-1/message' + ) { + void (async () => { + await eventStreamReady + for (const event of opencodeEvents) { + eventStream?.write(`data: ${JSON.stringify(event)}\n\n`) + } + eventStream?.end() + await new Promise((resolve) => setTimeout(resolve, 50)) + response.writeHead(200, { 'Content-Type': 'application/json' }) + response.end( + JSON.stringify({ + info: assistantMessage, + parts: [ + { + id: 'assistant-part', + sessionID: 'session-1', + messageID: 'assistant-message', + type: 'text', + text: 'assistant answer', + }, + ], + }), + ) + })() + return + } + + response.writeHead(404) + response.end() + }) + + await new Promise((resolve, reject) => { + server.once('error', reject) + server.listen(0, '127.0.0.1', resolve) + }) + const address = server.address() as AddressInfo + return { + url: `http://127.0.0.1:${address.port}`, + close: () => + new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())) + }), + } +} + +async function* output(values: Array): AsyncIterable { + yield* values +} + +function mockSandbox(url: string): SandboxHandle { + const spawnHandle: SpawnHandle = { + pid: 1, + stdout: output(['opencode server listening on http://127.0.0.1:4096\n']), + stderr: output([]), + stdin: { write: async () => {}, end: async () => {} }, + wait: async () => 0, + kill: async () => {}, + } + + return { + id: 'opencode-e2e', + provider: 'opencode-e2e', + capabilities: {} as SandboxHandle['capabilities'], + fs: {} as SandboxHandle['fs'], + git: {} as SandboxHandle['git'], + process: { + exec: async () => { + throw new Error('Unexpected process.exec call') + }, + spawn: async () => spawnHandle, + }, + ports: { connect: async () => ({ url }) }, + env: { set: async () => {} }, + destroy: async () => {}, + } +} + +test('filters user parts and waits for tool input before streaming a tool call', async () => { + const mockServer = await createMockOpenCodeServer() + const sandbox = mockSandbox(mockServer.url) + const provideTestSandbox = defineChatMiddleware({ + name: 'opencode-e2e-sandbox', + provides: [SandboxCapability], + setup(ctx) { + provideSandbox(ctx, sandbox) + }, + }) + + try { + const chunks: Array = [] + for await (const chunk of chat({ + adapter: opencodeText(MODEL), + middleware: [provideTestSandbox], + messages: [{ role: 'user', content: 'user prompt' }], + })) { + chunks.push(chunk) + } + + expect( + chunks + .filter((chunk) => chunk.type === 'TEXT_MESSAGE_CONTENT') + .map((chunk) => chunk.delta), + ).toEqual(['assistant answer']) + expect( + chunks.filter((chunk) => chunk.type === 'TOOL_CALL_START'), + ).toHaveLength(1) + expect( + chunks + .filter((chunk) => chunk.type === 'TOOL_CALL_ARGS') + .map((chunk) => chunk.delta), + ).toEqual([JSON.stringify({ command: 'ls' })]) + expect( + chunks.find((chunk) => chunk.type === 'TOOL_CALL_RESULT'), + ).toMatchObject({ content: 'file.txt' }) + } finally { + await mockServer.close() + } +}) From 14ebcab349cc267f1fca121a774fcfeaf7c34748 Mon Sep 17 00:00:00 2001 From: Tom Beckenham <34339192+tombeckenham@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:38:44 +1100 Subject: [PATCH 3/3] fix(ai-opencode): drop dead messageID guard and hand-rolled e2e spec, cover live streaming path --- docs/adapters/opencode.md | 4 +- packages/ai-opencode/src/stream/translate.ts | 17 +- packages/ai-opencode/tests/translate.test.ts | 130 +++------ pnpm-lock.yaml | 3 - testing/e2e/README.md | 1 - testing/e2e/package.json | 1 - .../tests/tools-test/opencode-stream.spec.ts | 257 ------------------ 7 files changed, 45 insertions(+), 368 deletions(-) delete mode 100644 testing/e2e/tests/tools-test/opencode-stream.spec.ts diff --git a/docs/adapters/opencode.md b/docs/adapters/opencode.md index af09346e17..d7cf003ca1 100644 --- a/docs/adapters/opencode.md +++ b/docs/adapters/opencode.md @@ -190,12 +190,12 @@ Sessions live on the server that ran them, so resuming only works against the sa Two kinds of tools flow through this adapter: -OpenCode can send a tool part with `pending` state before it provides tool input. The adapter waits for a non-pending update before it streams the tool call and its arguments. - 1. **Built-in harness tools** are executed by OpenCode itself and stream back as tool-call events with results already attached: `bash`, `edit`, `write`, `read`, `grep`, and the agent's running todo plan (surfaced as an `opencode.todo` custom event). Your code never executes them. 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/translate.ts b/packages/ai-opencode/src/stream/translate.ts index c8757440c7..aff410bf40 100644 --- a/packages/ai-opencode/src/stream/translate.ts +++ b/packages/ai-opencode/src/stream/translate.ts @@ -374,11 +374,6 @@ export async function* translateOpencodeStream( ) } else if (event.type === 'message.part.updated') { const { part, delta } = event.properties - if (!part.messageID) { - throw new Error( - `OpenCode message part "${part.id}" is missing its messageID.`, - ) - } const role = messageRoles.get(part.messageID) if (role === undefined) { const parts = pendingParts.get(part.messageID) ?? [] @@ -396,18 +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() - if (!messageRoles.has(message.id)) { - yield* setMessageRole(message.id, 'assistant') - } - pendingParts.clear() + // 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 f8485a7197..5ec94c31f6 100644 --- a/packages/ai-opencode/tests/translate.test.ts +++ b/packages/ai-opencode/tests/translate.test.ts @@ -55,25 +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', - messageID: 'msg-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([ @@ -101,61 +109,12 @@ describe('translateOpencodeStream', () => { it('does not expose user message parts as assistant output', async () => { const chunks = await collect([ session, - { - kind: 'event', - event: { - type: 'message.part.updated', - properties: { - part: { - id: 'user-part', - sessionID: 'sess-1', - messageID: 'user-message', - type: 'text', - text: 'user prompt', - }, - delta: 'user prompt', - }, - }, - }, - { - kind: 'event', - event: { - type: 'message.updated', - properties: { - info: { id: 'user-message', sessionID: 'sess-1', role: 'user' }, - }, - }, - }, - { - kind: 'event', - event: { - type: 'message.part.updated', - properties: { - part: { - id: 'assistant-part', - sessionID: 'sess-1', - messageID: 'assistant-message', - type: 'text', - text: 'assistant answer', - }, - delta: 'assistant answer', - }, - }, - }, - { - kind: 'event', - event: { - type: 'message.updated', - properties: { - info: { - id: 'assistant-message', - sessionID: 'sess-1', - role: 'assistant', - }, - }, - }, - }, - done(), + 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( @@ -165,25 +124,22 @@ describe('translateOpencodeStream', () => { ]) }) - it('rejects a message part without a message id', async () => { - const missingMessageId = { - kind: 'event', - event: { - type: 'message.part.updated', - properties: { - part: { - id: 'orphan-part', - sessionID: 'sess-1', - type: 'text', - text: 'orphan', - }, - delta: 'orphan', - }, - }, - } as unknown as OpencodeStreamEvent + 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) + } - await expect(collect([session, missingMessageId])).rejects.toThrow( - 'OpenCode message part "orphan-part" is missing its messageID.', + expect(log.indexOf('TEXT_MESSAGE_CONTENT')).toBeLessThan( + log.indexOf('done-pulled'), ) }) @@ -475,19 +431,7 @@ describe('translateOpencodeStream', () => { it('synthesizes results then rethrows when the source stream throws', async () => { async function* failing(): AsyncIterable { yield session - yield { - kind: 'event', - event: { - type: 'message.updated', - properties: { - info: { - id: 'msg-1', - sessionID: 'sess-1', - role: 'assistant', - }, - }, - }, - } + yield messageUpdated('msg-1', 'assistant') yield { kind: 'event', event: { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 98deadac56..931cf0e710 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -3867,9 +3867,6 @@ importers: '@tanstack/ai-mcp': specifier: workspace:* version: link:../../packages/ai-mcp - '@tanstack/ai-opencode': - specifier: workspace:* - version: link:../../packages/ai-opencode '@tanstack/ai-memory': specifier: workspace:* version: link:../../packages/ai-memory diff --git a/testing/e2e/README.md b/testing/e2e/README.md index ca5c5ed015..e81b8286ec 100644 --- a/testing/e2e/README.md +++ b/testing/e2e/README.md @@ -53,7 +53,6 @@ Deterministic scenarios covering tool execution flows: | `tests/tools-test/client-tool.spec.ts` | 5 | Single, sequential, parallel, triple, server+client | | `tests/tools-test/race-conditions.spec.ts` | 8 | No blocking, no deadlocks, timing, mixed flows | | `tests/tools-test/server-client-sequence.spec.ts` | 5 | Server→client, parallel server, ordering | -| `tests/tools-test/opencode-stream.spec.ts` | 1 | OpenCode message roles and pending tool input | ### Interrupt playground diff --git a/testing/e2e/package.json b/testing/e2e/package.json index e32c6545c6..108d79526a 100644 --- a/testing/e2e/package.json +++ b/testing/e2e/package.json @@ -37,7 +37,6 @@ "@tanstack/ai-llmgateway": "workspace:*", "@tanstack/ai-lovable": "workspace:*", "@tanstack/ai-mcp": "workspace:*", - "@tanstack/ai-opencode": "workspace:*", "@tanstack/ai-memory": "workspace:*", "@tanstack/ai-mistral": "workspace:*", "@tanstack/ai-ollama": "workspace:*", diff --git a/testing/e2e/tests/tools-test/opencode-stream.spec.ts b/testing/e2e/tests/tools-test/opencode-stream.spec.ts deleted file mode 100644 index 40d6a0a765..0000000000 --- a/testing/e2e/tests/tools-test/opencode-stream.spec.ts +++ /dev/null @@ -1,257 +0,0 @@ -import { createServer } from 'node:http' -import type { ServerResponse } from 'node:http' -import type { AddressInfo } from 'node:net' -import { chat, defineChatMiddleware } from '@tanstack/ai' -import type { StreamChunk } from '@tanstack/ai' -import { opencodeText } from '@tanstack/ai-opencode' -import { provideSandbox, SandboxCapability } from '@tanstack/ai-sandbox' -import type { SandboxHandle, SpawnHandle } from '@tanstack/ai-sandbox' -import { expect, test } from '@playwright/test' - -const MODEL = 'anthropic/claude-sonnet-4-5' - -const assistantMessage = { - id: 'assistant-message', - sessionID: 'session-1', - role: 'assistant', - finish: 'stop', - tokens: { input: 2, output: 2 }, -} - -const opencodeEvents = [ - { - type: 'message.part.updated', - properties: { - part: { - id: 'user-part', - sessionID: 'session-1', - messageID: 'user-message', - type: 'text', - text: 'user prompt', - }, - delta: 'user prompt', - }, - }, - { - type: 'message.updated', - properties: { - info: { id: 'user-message', sessionID: 'session-1', role: 'user' }, - }, - }, - { - type: 'message.part.updated', - properties: { - part: { - id: 'assistant-part', - sessionID: 'session-1', - messageID: 'assistant-message', - type: 'text', - text: 'assistant answer', - }, - delta: 'assistant answer', - }, - }, - { - type: 'message.updated', - properties: { - info: assistantMessage, - }, - }, - { - type: 'message.part.updated', - properties: { - part: { - id: 'tool-part', - sessionID: 'session-1', - messageID: 'assistant-message', - type: 'tool', - callID: 'call-1', - tool: 'bash', - state: { status: 'pending', input: {} }, - }, - }, - }, - { - type: 'message.part.updated', - properties: { - part: { - id: 'tool-part', - sessionID: 'session-1', - messageID: 'assistant-message', - type: 'tool', - callID: 'call-1', - tool: 'bash', - state: { status: 'running', input: { command: 'ls' } }, - }, - }, - }, - { - type: 'message.part.updated', - properties: { - part: { - id: 'tool-part', - sessionID: 'session-1', - messageID: 'assistant-message', - type: 'tool', - callID: 'call-1', - tool: 'bash', - state: { - status: 'completed', - input: { command: 'ls' }, - output: 'file.txt', - }, - }, - }, - }, -] - -async function createMockOpenCodeServer(): Promise<{ - url: string - close: () => Promise -}> { - let eventStream: ServerResponse | undefined - let resolveEventStream: () => void = () => {} - const eventStreamReady = new Promise((resolve) => { - resolveEventStream = resolve - }) - - const server = createServer((request, response) => { - const pathname = new URL(request.url ?? '/', 'http://127.0.0.1').pathname - if (request.method === 'POST' && pathname === '/session') { - response.writeHead(200, { 'Content-Type': 'application/json' }) - response.end(JSON.stringify({ id: 'session-1' })) - return - } - - if (request.method === 'GET' && pathname === '/event') { - response.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - }) - eventStream = response - resolveEventStream() - return - } - - if ( - request.method === 'POST' && - pathname === '/session/session-1/message' - ) { - void (async () => { - await eventStreamReady - for (const event of opencodeEvents) { - eventStream?.write(`data: ${JSON.stringify(event)}\n\n`) - } - eventStream?.end() - await new Promise((resolve) => setTimeout(resolve, 50)) - response.writeHead(200, { 'Content-Type': 'application/json' }) - response.end( - JSON.stringify({ - info: assistantMessage, - parts: [ - { - id: 'assistant-part', - sessionID: 'session-1', - messageID: 'assistant-message', - type: 'text', - text: 'assistant answer', - }, - ], - }), - ) - })() - return - } - - response.writeHead(404) - response.end() - }) - - await new Promise((resolve, reject) => { - server.once('error', reject) - server.listen(0, '127.0.0.1', resolve) - }) - const address = server.address() as AddressInfo - return { - url: `http://127.0.0.1:${address.port}`, - close: () => - new Promise((resolve, reject) => { - server.close((error) => (error ? reject(error) : resolve())) - }), - } -} - -async function* output(values: Array): AsyncIterable { - yield* values -} - -function mockSandbox(url: string): SandboxHandle { - const spawnHandle: SpawnHandle = { - pid: 1, - stdout: output(['opencode server listening on http://127.0.0.1:4096\n']), - stderr: output([]), - stdin: { write: async () => {}, end: async () => {} }, - wait: async () => 0, - kill: async () => {}, - } - - return { - id: 'opencode-e2e', - provider: 'opencode-e2e', - capabilities: {} as SandboxHandle['capabilities'], - fs: {} as SandboxHandle['fs'], - git: {} as SandboxHandle['git'], - process: { - exec: async () => { - throw new Error('Unexpected process.exec call') - }, - spawn: async () => spawnHandle, - }, - ports: { connect: async () => ({ url }) }, - env: { set: async () => {} }, - destroy: async () => {}, - } -} - -test('filters user parts and waits for tool input before streaming a tool call', async () => { - const mockServer = await createMockOpenCodeServer() - const sandbox = mockSandbox(mockServer.url) - const provideTestSandbox = defineChatMiddleware({ - name: 'opencode-e2e-sandbox', - provides: [SandboxCapability], - setup(ctx) { - provideSandbox(ctx, sandbox) - }, - }) - - try { - const chunks: Array = [] - for await (const chunk of chat({ - adapter: opencodeText(MODEL), - middleware: [provideTestSandbox], - messages: [{ role: 'user', content: 'user prompt' }], - })) { - chunks.push(chunk) - } - - expect( - chunks - .filter((chunk) => chunk.type === 'TEXT_MESSAGE_CONTENT') - .map((chunk) => chunk.delta), - ).toEqual(['assistant answer']) - expect( - chunks.filter((chunk) => chunk.type === 'TOOL_CALL_START'), - ).toHaveLength(1) - expect( - chunks - .filter((chunk) => chunk.type === 'TOOL_CALL_ARGS') - .map((chunk) => chunk.delta), - ).toEqual([JSON.stringify({ command: 'ls' })]) - expect( - chunks.find((chunk) => chunk.type === 'TOOL_CALL_RESULT'), - ).toMatchObject({ content: 'file.txt' }) - } finally { - await mockServer.close() - } -})