From 3ffd6326ba15163af5e888a3300145cf320396d7 Mon Sep 17 00:00:00 2001 From: Xiao Yi Date: Tue, 8 Sep 2026 11:29:43 +0800 Subject: [PATCH] fix(session): preserve reasoning across empty tool-call deltas Backport the nonempty tool-call guard from vercel/ai#20253 to the pinned @ai-sdk/openai-compatible@2.0.41 source, ESM and CJS targets. Preserve the existing structured-error patch and real text/tool boundaries without changing processor or adapter production code. Add 26 regression and boundary tests through the real SDK, stream adapter and session persistence, including error/interruption cleanup. Upstream: https://github.com/vercel/ai/pull/20253 Refs: https://github.com/anomalyco/opencode/issues/47875 Co-Authored-By: Claude --- packages/opencode/test/session/llm.test.ts | 113 ++++++++++- .../test/session/processor-effect.test.ts | 191 +++++++++++++++++- .../@ai-sdk%2Fopenai-compatible@2.0.41.patch | 24 +++ 3 files changed, 326 insertions(+), 2 deletions(-) diff --git a/packages/opencode/test/session/llm.test.ts b/packages/opencode/test/session/llm.test.ts index fcb536f46d91..79df3d4e50f4 100644 --- a/packages/opencode/test/session/llm.test.ts +++ b/packages/opencode/test/session/llm.test.ts @@ -3,7 +3,9 @@ import { ConfigV1 } from "@opencode-ai/core/v1/config/config" import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test" import { SessionV1 } from "@opencode-ai/core/v1/session" import path from "path" -import { tool, type ModelMessage } from "ai" +import { streamText, tool, type ModelMessage } from "ai" +import { createOpenAICompatible } from "@ai-sdk/openai-compatible" +import type { LanguageModelV3StreamPart } from "@ai-sdk/provider" import { Cause, Effect, Exit, Fiber, Layer, Stream } from "effect" import { InstanceRef } from "../../src/effect/instance-ref" import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http" @@ -185,6 +187,115 @@ describe("session.llm.ai-sdk adapter", () => { // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- tests defensive adapter branches outside AI SDK's current typed surface const uncheckedAdapterEvent = (input: unknown) => input as AISDKAdapterEvent + for (const scenario of [ + { name: "empty arrays", calls: [], end: "text" }, + { name: "omitted arrays", calls: undefined, end: "text" }, + { name: "null arrays", calls: null, end: "text" }, + { name: "reasoning-only empty arrays", calls: [], end: "flush" }, + { name: "nonempty tool boundary", calls: undefined, end: "tool" }, + { name: "mixed content boundary", calls: [], end: "text" }, + { name: "empty arrays before tool boundary", calls: [], end: "tool" }, + { name: "empty array heartbeats", calls: undefined, end: "text" }, + ] as const) { + const deltas: unknown[] = [ + { reasoning_content: "r1", tool_calls: scenario.calls }, + { + reasoning_content: "r2", + tool_calls: scenario.calls, + ...(scenario.name === "mixed content boundary" ? { content: "done" } : {}), + }, + ] + if (scenario.name === "empty array heartbeats") deltas.splice(1, 0, { tool_calls: [] }, { tool_calls: [] }) + if (scenario.end === "text" && scenario.name !== "mixed content boundary") deltas.push({ content: "done" }) + if (scenario.end === "tool") + deltas.push({ + tool_calls: [ + { index: 0, id: "call-1", type: "function", function: { name: "lookup", arguments: '{"query":"weather"}' } }, + ], + }) + const chunks = [ + ...deltas.map((delta) => ({ + id: "chatcmpl-reasoning", + object: "chat.completion.chunk", + choices: [{ index: 0, delta }], + })), + { + id: "chatcmpl-reasoning", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: {}, finish_reason: scenario.end === "tool" ? "tool_calls" : "stop" }], + }, + ] + const model = () => + createOpenAICompatible({ + name: "test", + baseURL: "https://example.test/v1", + fetch: Object.assign(async () => createEventResponse(chunks, true), { preconnect() {} }), + }).chatModel("test-model") + const lifecycle = [ + "reasoning-start", + "reasoning-delta", + "reasoning-delta", + "reasoning-end", + ] satisfies AISDKAdapterEvent["type"][] + + test(`compatible reasoning doStream preserves one lifecycle: ${scenario.name}`, async () => { + const result = await model().doStream({ prompt: [{ role: "user", content: [{ type: "text", text: "reason" }] }] }) + const events: LanguageModelV3StreamPart[] = [] + await result.stream.pipeTo( + new WritableStream({ + write(event) { + events.push(event) + }, + }), + ) + expect(events.filter((event) => event.type === "error")).toEqual([]) + if (scenario.end !== "tool") expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([]) + expect(events.flatMap((event) => (event.type === "reasoning-delta" ? [event.delta] : [])).join("")).toBe("r1r2") + expect(events.filter((event) => event.type.startsWith("reasoning-")).map((event) => event.type)).toEqual( + lifecycle, + ) + const end = events.findIndex((event) => event.type === "reasoning-end") + expect(events[end + 1]?.type).toBe( + scenario.end === "tool" ? "tool-input-start" : scenario.end === "text" ? "text-start" : "finish", + ) + expect(events.filter((event) => event.type === "tool-call")).toEqual( + scenario.end === "tool" + ? [{ type: "tool-call", toolCallId: "call-1", toolName: "lookup", input: '{"query":"weather"}' }] + : [], + ) + expect(events.flatMap((event) => (event.type === "text-delta" ? [event.delta] : [])).join("")).toBe( + scenario.end === "text" ? "done" : "", + ) + }) + + test(`compatible reasoning streamText adapter preserves one lifecycle: ${scenario.name}`, async () => { + const result = streamText({ + model: model(), + prompt: "reason", + maxRetries: 0, + tools: { lookup: tool({ inputSchema: z.object({ query: z.string() }) }) }, + }) + const events = await adapt(await Array.fromAsync(result.fullStream)) + expect(events.filter((event) => event.type === "provider-error" || event.type === "tool-error")).toEqual([]) + if (scenario.end !== "tool") expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([]) + expect(events.flatMap((event) => (event.type === "reasoning-delta" ? [event.text] : [])).join("")).toBe("r1r2") + expect(events.filter((event) => event.type.startsWith("reasoning-")).map((event) => event.type)).toEqual( + lifecycle, + ) + const end = events.findIndex((event) => event.type === "reasoning-end") + expect(events[end + 1]?.type).toBe( + scenario.end === "tool" ? "tool-input-start" : scenario.end === "text" ? "text-start" : "step-finish", + ) + const calls = events.filter((event) => event.type === "tool-call") + expect(calls).toHaveLength(scenario.end === "tool" ? 1 : 0) + if (scenario.end === "tool") + expect(calls[0]).toMatchObject({ id: "call-1", name: "lookup", input: { query: "weather" } }) + expect(events.flatMap((event) => (event.type === "text-delta" ? [event.text] : [])).join("")).toBe( + scenario.end === "text" ? "done" : "", + ) + }) + } + test("maps AI SDK stream chunks without losing session-visible fields", async () => { const metadata = { openai: { itemID: "item-1" } } const events = await adapt([ diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index c67f82d9c71b..68e02388cc80 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -4,7 +4,7 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2Bridge } from "@/event-v2-bridge" import { expect } from "bun:test" import { tool } from "ai" -import { Cause, Effect, Exit, Fiber, Layer, Stream } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, Layer, Stream } from "effect" import path from "path" import z from "zod" import type { Agent } from "../../src/agent/agent" @@ -467,6 +467,195 @@ it.live("session.processor effect tests capture reasoning from http mock", () => ), ) +for (const scenario of [ + { name: "empty arrays", calls: [], end: "text" }, + { name: "omitted arrays", calls: undefined, end: "text" }, + { name: "null arrays", calls: null, end: "text" }, + { name: "reasoning-only empty arrays", calls: [], end: "flush" }, + { name: "nonempty tool boundary", calls: undefined, end: "tool" }, + { name: "mixed content boundary", calls: [], end: "text" }, + { name: "empty arrays before tool boundary", calls: [], end: "tool" }, + { name: "empty array heartbeats", calls: undefined, end: "text" }, +] as const) { + it.live(`session.processor compatible reasoning persists one part: ${scenario.name}`, () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + const deltas: unknown[] = [ + { reasoning_content: "r1", tool_calls: scenario.calls }, + { + reasoning_content: "r2", + tool_calls: scenario.calls, + ...(scenario.name === "mixed content boundary" ? { content: "done" } : {}), + }, + ] + if (scenario.name === "empty array heartbeats") deltas.splice(1, 0, { tool_calls: [] }, { tool_calls: [] }) + if (scenario.end === "text" && scenario.name !== "mixed content boundary") deltas.push({ content: "done" }) + if (scenario.end === "tool") + deltas.push({ + tool_calls: [ + { + index: 0, + id: "call-1", + type: "function", + function: { name: "lookup", arguments: '{"query":"weather"}' }, + }, + ], + }) + yield* llm.push( + raw({ + chunks: [ + ...deltas.map((delta) => ({ + id: "chatcmpl-reasoning", + object: "chat.completion.chunk", + choices: [{ index: 0, delta }], + })), + { + id: "chatcmpl-reasoning", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: {}, finish_reason: scenario.end === "tool" ? "tool_calls" : "stop" }], + }, + ], + }), + ) + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "reason") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl }) + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: ref, + }, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "reason" }], + tools: { + lookup: tool({ + inputSchema: z.object({ query: z.string() }), + execute: async (input) => ({ title: "Lookup", output: `result:${input.query}`, metadata: {} }), + }), + }, + }) + const parts = yield* MessageV2.parts(msg.id) + const reasoning = parts.filter((part): part is SessionV1.ReasoningPart => part.type === "reasoning") + const calls = parts.filter((part): part is SessionV1.ToolPart => part.type === "tool") + expect(value).toBe("continue") + expect(yield* llm.calls).toBe(1) + expect( + parts + .filter((part) => part.type === "text") + .map((part) => part.text) + .join(""), + ).toBe(scenario.end === "text" ? "done" : "") + expect(calls).toHaveLength(scenario.end === "tool" ? 1 : 0) + if (scenario.end === "tool") + expect(calls[0]).toMatchObject({ + callID: "call-1", + tool: "lookup", + state: { status: "completed", input: { query: "weather" }, output: "result:weather" }, + }) + expect(reasoning.map((part) => part.text).join("")).toBe("r1r2") + expect(reasoning).toHaveLength(1) + expect(reasoning[0]).toMatchObject({ + text: "r1r2", + time: { start: expect.any(Number), end: expect.any(Number) }, + }) + }), + { config: (url) => providerCfg(url) }, + ), + ) +} + +for (const failure of ["provider error", "interruption"] as const) { + it.live( + `session.processor compatible reasoning closes active parts after ${failure}`, + () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + const bridge = yield* EventV2Bridge.Service + const seen = yield* Deferred.make() + const chunks: unknown[] = ["r1", "r2"].map((text) => ({ + id: "chatcmpl-reasoning", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { reasoning_content: text, tool_calls: [] } }], + })) + if (failure === "provider error") + chunks.push({ + error: { type: "invalid_request_error", code: "invalid_request_error", message: "invalid request" }, + }) + yield* llm.push(raw({ chunks, hang: failure === "interruption" })) + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "reason") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model: mdl }) + const off = yield* bridge.listen((event) => { + if (event.type !== MessageV2.Event.PartDelta.type) return Effect.void + const data = event.data as typeof MessageV2.Event.PartDelta.data.Type + if (data.messageID !== msg.id || data.delta !== "r2") return Effect.void + return Deferred.succeed(seen, undefined).pipe(Effect.asVoid) + }) + yield* Effect.addFinalizer(() => off) + const run = yield* handle + .process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: ref, + }, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "reason" }], + tools: {}, + }) + .pipe(Effect.forkChild) + if (failure === "interruption") { + // Wait for actual reasoning consumption, not just receipt of the HTTP request. + const reached = yield* Effect.raceFirst( + Deferred.await(seen).pipe(Effect.as("delta")), + Fiber.await(run).pipe(Effect.as("completed")), + ) + expect(reached).toBe("delta") + yield* Fiber.interrupt(run) + const exit = yield* Fiber.await(run) + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true) + expect(handle.message.error?.name).toBe("MessageAbortedError") + } else { + expect(yield* Fiber.join(run)).toBe("stop") + expect(JSON.stringify(handle.message.error)).toContain("invalid request") + } + expect(yield* llm.calls).toBe(1) + const parts = yield* MessageV2.parts(msg.id) + const reasoning = parts.filter((part): part is SessionV1.ReasoningPart => part.type === "reasoning") + expect(reasoning).toHaveLength(1) + expect(reasoning[0]).toMatchObject({ + text: "r1r2", + time: { start: expect.any(Number), end: expect.any(Number) }, + }) + }), + { config: (url) => providerCfg(url) }, + ), + 30000, + ) +} + it.live("session.processor effect tests reset reasoning state across retries", () => provideTmpdirServer( ({ dir, llm }) => diff --git a/patches/@ai-sdk%2Fopenai-compatible@2.0.41.patch b/patches/@ai-sdk%2Fopenai-compatible@2.0.41.patch index 9f03ec95732a..54d5bac27e87 100644 --- a/patches/@ai-sdk%2Fopenai-compatible@2.0.41.patch +++ b/patches/@ai-sdk%2Fopenai-compatible@2.0.41.patch @@ -11,6 +11,15 @@ index dca128d3a790378c51a24a16d92585178343b278..da75f9d64acd2b607abd15079ce2d03b }); return; } +@@ -755,7 +755,7 @@ var OpenAICompatibleChatLanguageModel = class { + delta: delta.content + }); + } +- if (delta.tool_calls != null) { ++ if (delta.tool_calls != null && delta.tool_calls.length > 0) { + if (isActiveReasoning) { + controller.enqueue({ + type: "reasoning-end", diff --git a/dist/index.mjs b/dist/index.mjs index 3b1e1b6bdec5032e3b4fa5ffbcc8cdf3dfe1cc40..eaffc446f80552ea0b26573b89daa4dfc7776e6e 100644 --- a/dist/index.mjs @@ -24,6 +33,15 @@ index 3b1e1b6bdec5032e3b4fa5ffbcc8cdf3dfe1cc40..eaffc446f80552ea0b26573b89daa4df }); return; } +@@ -742,7 +742,7 @@ var OpenAICompatibleChatLanguageModel = class { + delta: delta.content + }); + } +- if (delta.tool_calls != null) { ++ if (delta.tool_calls != null && delta.tool_calls.length > 0) { + if (isActiveReasoning) { + controller.enqueue({ + type: "reasoning-end", diff --git a/src/chat/openai-compatible-chat-language-model.ts b/src/chat/openai-compatible-chat-language-model.ts index 8c622db23c2d9a7373701f5a1b0c2ba109e24602..643c3db68a6043e097edc1122e0eb53fd13495c5 100644 --- a/src/chat/openai-compatible-chat-language-model.ts @@ -37,3 +55,9 @@ index 8c622db23c2d9a7373701f5a1b0c2ba109e24602..643c3db68a6043e097edc1122e0eb53f }); return; } +@@ -522,4 +522,4 @@ export class OpenAICompatibleChatLanguageModel implements LanguageModelV3 { +- if (delta.tool_calls != null) { ++ if (delta.tool_calls != null && delta.tool_calls.length > 0) { + // end active reasoning block before tool calls start + if (isActiveReasoning) { + controller.enqueue({