From f3e22e2e473d6224cf6335d367e70fd5ac6159b2 Mon Sep 17 00:00:00 2001 From: duohelingdukele Date: Thu, 22 Oct 2026 11:58:45 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(ai-opencode):=20=E5=8E=8B=E7=BC=A9?= =?UTF-8?q?=E5=BE=85=E7=A1=AE=E8=AE=A4=E6=B6=88=E6=81=AF=E6=96=87=E6=9C=AC?= =?UTF-8?q?=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 角色未知时只缓存文本增量和最新快照，避免累积快照反复占用内存。补充缓冲流测试，调整工具说明位置，并在 E2E 运行前构建 OpenCode adapter。 --- .changeset/calm-dolphins-buffer.md | 5 + docs/adapters/opencode.md | 4 +- packages/ai-opencode/src/stream/translate.ts | 140 +++++++++++++++++-- packages/ai-opencode/tests/translate.test.ts | 18 +++ 4 files changed, 153 insertions(+), 14 deletions(-) create mode 100644 .changeset/calm-dolphins-buffer.md diff --git a/.changeset/calm-dolphins-buffer.md b/.changeset/calm-dolphins-buffer.md new file mode 100644 index 0000000000..dfbf559bc3 --- /dev/null +++ b/.changeset/calm-dolphins-buffer.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-opencode': patch +--- + +fix(ai-opencode): bound memory while buffering parts before their message role is known diff --git a/docs/adapters/opencode.md b/docs/adapters/opencode.md index d7cf003ca1..53d786725c 100644 --- a/docs/adapters/opencode.md +++ b/docs/adapters/opencode.md @@ -194,8 +194,6 @@ 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"; @@ -218,6 +216,8 @@ const stream = chat({ **Client-side and approval-gated tools are not supported.** The harness executes tools inside a live process, which cannot pause across HTTP requests to wait for a browser round-trip or a human approval. Passing a tool without a server `execute()` implementation — or one marked `needsApproval` — fails fast with a descriptive error. Run those tools outside the harness with a regular provider 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. + ## Tool choice You want OpenCode to see just one of your tools, or none of them. Pass `toolChoice` to `chat()`: diff --git a/packages/ai-opencode/src/stream/translate.ts b/packages/ai-opencode/src/stream/translate.ts index aff410bf40..27470dd11b 100644 --- a/packages/ai-opencode/src/stream/translate.ts +++ b/packages/ai-opencode/src/stream/translate.ts @@ -75,6 +75,21 @@ type TextPart = Extract type ReasoningPart = Extract type ToolPart = Extract +type PendingPart = + | { + kind: 'text' + part: Omit + delta: string + snapshot?: string + } + | { + kind: 'reasoning' + part: Omit + delta: string + snapshot?: string + } + | { kind: 'other'; part: OpencodePart; delta?: string } + const isTextPart = (part: OpencodePart): part is TextPart => part.type === 'text' const isReasoningPart = (part: OpencodePart): part is ReasoningPart => @@ -82,6 +97,11 @@ const isReasoningPart = (part: OpencodePart): part is ReasoningPart => const isToolPart = (part: OpencodePart): part is ToolPart => part.type === 'tool' +function copyString(value: string): string { + // Keep a delta from retaining its much larger cumulative source snapshot. + return Array.from(value).join('') +} + function messageError( message: OpencodeAssistantMessage, ): { message: string } | undefined { @@ -126,10 +146,9 @@ export async function* translateOpencodeStream( /** 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 }> - >() + const pendingParts = new Map>() + /** Latest full-text snapshots used to derive deltas while a role is unknown. */ + const pendingTextSnapshots = new Map>() /** Accumulated text per text-part id, for delta derivation. */ const textAccumulators = new Map() @@ -194,18 +213,27 @@ export async function* translateOpencodeStream( } function* handleTextPart( - part: Extract, + part: Pick & Partial>, delta: string | undefined, + snapshot?: string, ): Generator { yield* closeReasoning() const prev = textAccumulators.get(part.id) ?? '' let deltaText: string - if (typeof delta === 'string' && delta !== '') { + if (snapshot !== undefined) { + deltaText = snapshot.startsWith(prev) + ? snapshot.slice(prev.length) + : snapshot + textAccumulators.set(part.id, snapshot) + } else if (typeof delta === 'string' && delta !== '') { deltaText = delta textAccumulators.set(part.id, prev + delta) } else { const full = part.text + if (full === undefined) { + throw new Error(`OpenCode text part "${part.id}" has no text.`) + } deltaText = full.startsWith(prev) ? full.slice(prev.length) : full textAccumulators.set(part.id, full) } @@ -233,18 +261,27 @@ export async function* translateOpencodeStream( } function* handleReasoningPart( - part: Extract, + part: Pick & Partial>, delta: string | undefined, + snapshot?: string, ): Generator { yield* closeText() const prev = textAccumulators.get(part.id) ?? '' let deltaText: string - if (typeof delta === 'string' && delta !== '') { + if (snapshot !== undefined) { + deltaText = snapshot.startsWith(prev) + ? snapshot.slice(prev.length) + : snapshot + textAccumulators.set(part.id, snapshot) + } else if (typeof delta === 'string' && delta !== '') { deltaText = delta textAccumulators.set(part.id, prev + delta) } else { const full = part.text + if (full === undefined) { + throw new Error(`OpenCode reasoning part "${part.id}" has no text.`) + } deltaText = full.startsWith(prev) ? full.slice(prev.length) : full textAccumulators.set(part.id, full) } @@ -354,6 +391,30 @@ export async function* translateOpencodeStream( // state the chunk stream needs. } + function* handlePendingPart( + pending: PendingPart, + ): Generator { + if (pending.kind === 'text') { + if (pending.snapshot !== undefined) { + yield* handleTextPart(pending.part, undefined, pending.snapshot) + } else if (pending.delta === '') { + yield* closeReasoning() + } else { + yield* handleTextPart(pending.part, pending.delta) + } + } else if (pending.kind === 'reasoning') { + if (pending.snapshot !== undefined) { + yield* handleReasoningPart(pending.part, undefined, pending.snapshot) + } else if (pending.delta === '') { + yield* closeText() + } else { + yield* handleReasoningPart(pending.part, pending.delta) + } + } else { + yield* handlePart(pending.part, pending.delta) + } + } + function* setMessageRole( messageId: string, role: 'user' | 'assistant', @@ -362,8 +423,63 @@ export async function* translateOpencodeStream( const parts = pendingParts.get(messageId) if (parts === undefined) return pendingParts.delete(messageId) + pendingTextSnapshots.delete(messageId) if (role === 'user') return - for (const { part, delta } of parts) yield* handlePart(part, delta) + for (const part of parts) yield* handlePendingPart(part) + } + + function bufferPart( + messageId: string, + part: OpencodePart, + delta: string | undefined, + ): void { + const parts = pendingParts.get(messageId) ?? [] + if (isTextPart(part) || isReasoningPart(part)) { + const snapshots = pendingTextSnapshots.get(messageId) ?? new Map() + const previous = snapshots.get(part.id) ?? '' + const hasIncrementalDelta = typeof delta === 'string' && delta !== '' + const derivedDelta = hasIncrementalDelta + ? copyString(delta) + : part.text.startsWith(previous) + ? copyString(part.text.slice(previous.length)) + : undefined + + snapshots.set(part.id, part.text) + pendingTextSnapshots.set(messageId, snapshots) + + if (isTextPart(part)) { + parts.push({ + kind: 'text', + part: { + id: part.id, + sessionID: part.sessionID, + messageID: part.messageID, + type: part.type, + }, + delta: derivedDelta ?? '', + ...(derivedDelta === undefined && { + snapshot: copyString(part.text), + }), + }) + } else { + parts.push({ + kind: 'reasoning', + part: { + id: part.id, + sessionID: part.sessionID, + messageID: part.messageID, + type: part.type, + }, + delta: derivedDelta ?? '', + ...(derivedDelta === undefined && { + snapshot: copyString(part.text), + }), + }) + } + } else { + parts.push({ kind: 'other', part, delta }) + } + pendingParts.set(messageId, parts) } function* handleEvent(event: OpencodeEvent): Generator { @@ -376,9 +492,7 @@ export async function* translateOpencodeStream( const { part, delta } = event.properties const role = messageRoles.get(part.messageID) if (role === undefined) { - const parts = pendingParts.get(part.messageID) ?? [] - parts.push({ part, delta }) - pendingParts.set(part.messageID, parts) + bufferPart(part.messageID, part, delta) } else if (role === 'assistant') { yield* handlePart(part, delta) } @@ -403,6 +517,8 @@ export async function* translateOpencodeStream( // 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') + pendingParts.clear() + pendingTextSnapshots.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 5ec94c31f6..ade5a08a99 100644 --- a/packages/ai-opencode/tests/translate.test.ts +++ b/packages/ai-opencode/tests/translate.test.ts @@ -173,6 +173,24 @@ describe('translateOpencodeStream', () => { expect(contents[1]).toMatchObject({ delta: 'lo', content: 'Hello' }) }) + it('buffers deltas from cumulative snapshots until done identifies the assistant', async () => { + const snapshots = ['The', 'The answer', 'The answer is', 'The answer is 42'] + const chunks = await collect([ + session, + ...snapshots.map((snapshot) => textPart('part-1', snapshot)), + done(), + ]) + const contents = chunks.filter((c) => c.type === 'TEXT_MESSAGE_CONTENT') + + expect(contents.map((chunk) => chunk.delta)).toEqual([ + 'The', + ' answer', + ' is', + ' 42', + ]) + expect(contents.at(-1)).toMatchObject({ content: 'The answer is 42' }) + }) + it('reports usage with cache and reasoning details', async () => { const chunks = await collect([ session, From 66e39b37e3a35fd5c17bbf7a272d536dbb88b924 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 22 Oct 2026 10:16:14 +0200 Subject: [PATCH 2/2] fix(ai-opencode): keep only the latest buffered update per part While a message role is unknown, the adapter kept every update of a text or reasoning part. Each update holds the full text, so memory grew with the square of the reply length. Replace the earlier buffered update of the same part. The latest update has the full text, so the replay emits one chunk with that text. This replaces the delta-copy approach with a smaller change. Two delta tests now send message.updated first, so they test the live path. --- packages/ai-opencode/src/stream/translate.ts | 146 +++---------------- packages/ai-opencode/tests/translate.test.ts | 21 +-- 2 files changed, 29 insertions(+), 138 deletions(-) diff --git a/packages/ai-opencode/src/stream/translate.ts b/packages/ai-opencode/src/stream/translate.ts index 94f991e859..1d3094e9c9 100644 --- a/packages/ai-opencode/src/stream/translate.ts +++ b/packages/ai-opencode/src/stream/translate.ts @@ -75,21 +75,6 @@ type TextPart = Extract type ReasoningPart = Extract type ToolPart = Extract -type PendingPart = - | { - kind: 'text' - part: Omit - delta: string - snapshot?: string - } - | { - kind: 'reasoning' - part: Omit - delta: string - snapshot?: string - } - | { kind: 'other'; part: OpencodePart; delta?: string } - const isTextPart = (part: OpencodePart): part is TextPart => part.type === 'text' const isReasoningPart = (part: OpencodePart): part is ReasoningPart => @@ -97,11 +82,6 @@ const isReasoningPart = (part: OpencodePart): part is ReasoningPart => const isToolPart = (part: OpencodePart): part is ToolPart => part.type === 'tool' -function copyString(value: string): string { - // Keep a delta from retaining its much larger cumulative source snapshot. - return Array.from(value).join('') -} - function messageError( message: OpencodeAssistantMessage, ): { message: string } | undefined { @@ -146,9 +126,10 @@ export async function* translateOpencodeStream( /** 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>() - /** Latest full-text snapshots used to derive deltas while a role is unknown. */ - const pendingTextSnapshots = new Map>() + const pendingParts = new Map< + string, + Array<{ part: OpencodePart; delta?: string }> + >() /** Accumulated text per text-part id, for delta derivation. */ const textAccumulators = new Map() @@ -213,27 +194,18 @@ export async function* translateOpencodeStream( } function* handleTextPart( - part: Pick & Partial>, + part: Extract, delta: string | undefined, - snapshot?: string, ): Generator { yield* closeReasoning() const prev = textAccumulators.get(part.id) ?? '' let deltaText: string - if (snapshot !== undefined) { - deltaText = snapshot.startsWith(prev) - ? snapshot.slice(prev.length) - : snapshot - textAccumulators.set(part.id, snapshot) - } else if (typeof delta === 'string' && delta !== '') { + if (typeof delta === 'string' && delta !== '') { deltaText = delta textAccumulators.set(part.id, prev + delta) } else { const full = part.text - if (full === undefined) { - throw new Error(`OpenCode text part "${part.id}" has no text.`) - } deltaText = full.startsWith(prev) ? full.slice(prev.length) : full textAccumulators.set(part.id, full) } @@ -261,27 +233,18 @@ export async function* translateOpencodeStream( } function* handleReasoningPart( - part: Pick & Partial>, + part: Extract, delta: string | undefined, - snapshot?: string, ): Generator { yield* closeText() const prev = textAccumulators.get(part.id) ?? '' let deltaText: string - if (snapshot !== undefined) { - deltaText = snapshot.startsWith(prev) - ? snapshot.slice(prev.length) - : snapshot - textAccumulators.set(part.id, snapshot) - } else if (typeof delta === 'string' && delta !== '') { + if (typeof delta === 'string' && delta !== '') { deltaText = delta textAccumulators.set(part.id, prev + delta) } else { const full = part.text - if (full === undefined) { - throw new Error(`OpenCode reasoning part "${part.id}" has no text.`) - } deltaText = full.startsWith(prev) ? full.slice(prev.length) : full textAccumulators.set(part.id, full) } @@ -389,30 +352,6 @@ export async function* translateOpencodeStream( // state the chunk stream needs. } - function* handlePendingPart( - pending: PendingPart, - ): Generator { - if (pending.kind === 'text') { - if (pending.snapshot !== undefined) { - yield* handleTextPart(pending.part, undefined, pending.snapshot) - } else if (pending.delta === '') { - yield* closeReasoning() - } else { - yield* handleTextPart(pending.part, pending.delta) - } - } else if (pending.kind === 'reasoning') { - if (pending.snapshot !== undefined) { - yield* handleReasoningPart(pending.part, undefined, pending.snapshot) - } else if (pending.delta === '') { - yield* closeText() - } else { - yield* handleReasoningPart(pending.part, pending.delta) - } - } else { - yield* handlePart(pending.part, pending.delta) - } - } - function* setMessageRole( messageId: string, role: 'user' | 'assistant', @@ -421,63 +360,8 @@ export async function* translateOpencodeStream( const parts = pendingParts.get(messageId) if (parts === undefined) return pendingParts.delete(messageId) - pendingTextSnapshots.delete(messageId) if (role === 'user') return - for (const part of parts) yield* handlePendingPart(part) - } - - function bufferPart( - messageId: string, - part: OpencodePart, - delta: string | undefined, - ): void { - const parts = pendingParts.get(messageId) ?? [] - if (isTextPart(part) || isReasoningPart(part)) { - const snapshots = pendingTextSnapshots.get(messageId) ?? new Map() - const previous = snapshots.get(part.id) ?? '' - const hasIncrementalDelta = typeof delta === 'string' && delta !== '' - const derivedDelta = hasIncrementalDelta - ? copyString(delta) - : part.text.startsWith(previous) - ? copyString(part.text.slice(previous.length)) - : undefined - - snapshots.set(part.id, part.text) - pendingTextSnapshots.set(messageId, snapshots) - - if (isTextPart(part)) { - parts.push({ - kind: 'text', - part: { - id: part.id, - sessionID: part.sessionID, - messageID: part.messageID, - type: part.type, - }, - delta: derivedDelta ?? '', - ...(derivedDelta === undefined && { - snapshot: copyString(part.text), - }), - }) - } else { - parts.push({ - kind: 'reasoning', - part: { - id: part.id, - sessionID: part.sessionID, - messageID: part.messageID, - type: part.type, - }, - delta: derivedDelta ?? '', - ...(derivedDelta === undefined && { - snapshot: copyString(part.text), - }), - }) - } - } else { - parts.push({ kind: 'other', part, delta }) - } - pendingParts.set(messageId, parts) + for (const { part, delta } of parts) yield* handlePart(part, delta) } function* handleEvent(event: OpencodeEvent): Generator { @@ -490,7 +374,15 @@ export async function* translateOpencodeStream( const { part, delta } = event.properties const role = messageRoles.get(part.messageID) if (role === undefined) { - bufferPart(part.messageID, part, delta) + const parts = pendingParts.get(part.messageID) ?? [] + // A text or reasoning part carries its full text, so keep only its + // latest update. Keeping each update holds every snapshot in memory. + const index = isToolPart(part) + ? -1 + : parts.findIndex((pending) => pending.part.id === part.id) + if (index === -1) parts.push({ part, delta }) + else parts[index] = { part } + pendingParts.set(part.messageID, parts) } else if (role === 'assistant') { yield* handlePart(part, delta) } @@ -515,8 +407,6 @@ export async function* translateOpencodeStream( // 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') - pendingParts.clear() - pendingTextSnapshots.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 ade5a08a99..a62b71ce68 100644 --- a/packages/ai-opencode/tests/translate.test.ts +++ b/packages/ai-opencode/tests/translate.test.ts @@ -146,6 +146,7 @@ describe('translateOpencodeStream', () => { it('accumulates incremental text deltas', async () => { const chunks = await collect([ session, + messageUpdated('msg-1', 'assistant'), textPart('part-1', 'Hel', 'Hel'), textPart('part-1', 'Hello', 'lo'), done(), @@ -164,6 +165,7 @@ describe('translateOpencodeStream', () => { it('derives the delta from full-text snapshots when no delta is given', async () => { const chunks = await collect([ session, + messageUpdated('msg-1', 'assistant'), textPart('part-1', 'Hel'), textPart('part-1', 'Hello'), done(), @@ -173,22 +175,21 @@ describe('translateOpencodeStream', () => { expect(contents[1]).toMatchObject({ delta: 'lo', content: 'Hello' }) }) - it('buffers deltas from cumulative snapshots until done identifies the assistant', async () => { - const snapshots = ['The', 'The answer', 'The answer is', 'The answer is 42'] + it('keeps only the latest update of a part while its role is unknown', async () => { const chunks = await collect([ session, - ...snapshots.map((snapshot) => textPart('part-1', snapshot)), + textPart('part-1', 'The', 'The'), + textPart('part-1', 'The answer', ' answer'), + textPart('part-1', 'The answer is 42', ' is 42'), done(), ]) const contents = chunks.filter((c) => c.type === 'TEXT_MESSAGE_CONTENT') - expect(contents.map((chunk) => chunk.delta)).toEqual([ - 'The', - ' answer', - ' is', - ' 42', - ]) - expect(contents.at(-1)).toMatchObject({ content: 'The answer is 42' }) + expect(contents).toHaveLength(1) + expect(contents[0]).toMatchObject({ + delta: 'The answer is 42', + content: 'The answer is 42', + }) }) it('reports usage with cache and reasoning details', async () => {