fix(session): preserve reasoning across empty tool-call deltas

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<LanguageModelV3StreamPart>({
+          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([

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<void>()
+            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 }) =>

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({