fix(cli): route openai-compatible providers through streamText for memory consolidation

.changeset/memory-streamtext-openai.md

@@ -0,0 +1,5 @@
+---
+"@kilocode/cli": patch
+---
+
+Fix memory consolidation failing with `Invalid JSON response` on OpenAI-compatible providers.

packages/opencode/src/kilocode/memory/ports.ts

@@ -6,7 +6,7 @@ import type { MemoryPorts } from "@kilocode/kilo-memory/effect/ports"
 import { MemoryRedact } from "@kilocode/kilo-memory/redact"
 import { MemoryShared } from "@kilocode/kilo-memory/shared"
 import * as Log from "@opencode-ai/core/util/log"
-import type { LanguageModelV3 } from "@ai-sdk/provider"
+import { APICallError, JSONParseError, type LanguageModelV3 } from "@ai-sdk/provider"
 import { Provider } from "@/provider/provider"
 import { ProviderTransform } from "@/provider/transform"
 import type { MessageV2 } from "@/session/message-v2"
@@ -167,6 +167,25 @@ function consolidationPrompt(input: { model: Provider.Model; options: Record<str
   }
 }
 
+async function consumeStream(common: Parameters<typeof streamText>[0]) {
+  const result = streamText(common)
+  const text: string[] = []
+  let usage: unknown
+  for await (const part of result.fullStream) {
+    if (part.type === "text-delta" && part.text) text.push(part.text)
+    if (part.type === "finish-step") usage = part.usage
+    if (part.type === "finish") usage = part.totalUsage
+    if (part.type === "error") throw part.error
+  }
+  return { text: text.join(""), usage }
+}
+
+const streamNeeded = new Set<string>()
+
+export function resetStreamNeeded() {
+  streamNeeded.clear()
+}
+
 async function memoryText(input: {
   source: Provider.Model
   language: LanguageModelV3
@@ -182,7 +201,6 @@ async function memoryText(input: {
   const ctl = new AbortController()
   const ms = Math.max(1, input.timeoutMs)
   const params = consolidationPrompt({ model: input.source, options: input.options, system: input.system })
-  const openai = input.source.providerID === "openai" && input.source.api.npm === "@ai-sdk/openai"
   const common = {
     model: input.language,
     ...(params.system ? { system: params.system } : {}),
@@ -194,19 +212,23 @@ async function memoryText(input: {
     topK: input.topK,
     maxRetries: 1,
   }
+  const native = input.source.providerID === "openai" && input.source.api.npm === "@ai-sdk/openai"
+  const compat = input.source.api.npm === "@ai-sdk/openai-compatible"
+  const key = `${input.source.providerID}/${input.source.id}`
   const work = async () => {
-    if (!openai) return generateText(common)
-
-    const result = streamText(common)
-    const text: string[] = []
-    let usage: unknown
-    for await (const part of result.fullStream) {
-      if (part.type === "text-delta" && part.text) text.push(part.text)
-      if (part.type === "finish-step") usage = part.usage
-      if (part.type === "finish") usage = part.totalUsage
-      if (part.type === "error") throw part.error
+    if (native) return consumeStream(common)
+    if (!compat) return generateText(common)
+    if (streamNeeded.has(key)) return consumeStream(common)
+    try {
+      return await generateText(common)
+    } catch (err) {
+      if (JSONParseError.isInstance(err) || (APICallError.isInstance(err) && JSONParseError.isInstance(err.cause))) {
+        const out = await consumeStream(common)
+        streamNeeded.add(key)
+        return out
+      }
+      throw err
     }
-    return { text: text.join(""), usage }
   }
   let timer: ReturnType<typeof setTimeout> | undefined
   const timeout = new Promise<never>((_, reject) => {

packages/opencode/test/kilocode/memory/memory-ports.test.ts

@@ -1,7 +1,6 @@
-import { describe, expect, test } from "bun:test"
+import { beforeEach, describe, expect, test } from "bun:test"
 import { createOpenAICompatible } from "@ai-sdk/openai-compatible"
-import type { LanguageModelV3 } from "@ai-sdk/provider"
-import { APICallError } from "ai"
+import { APICallError, JSONParseError, type LanguageModelV3 } from "@ai-sdk/provider"
 import { Effect } from "effect"
 import { ModelNotFoundError, type Provider } from "../../../src/provider/provider"
 import { ProviderV2 } from "@opencode-ai/core/provider"
@@ -11,7 +10,7 @@ import { MessageID, PartID, SessionID } from "../../../src/session/schema"
 import type { Session } from "../../../src/session/session"
 import type { SessionSummary } from "../../../src/session/summary"
 import type { Snapshot } from "../../../src/snapshot"
-import { MemoryModel, MemorySession } from "../../../src/kilocode/memory/ports"
+import { MemoryModel, MemorySession, resetStreamNeeded } from "../../../src/kilocode/memory/ports"
 
 const pid = ProviderV2.ID.make("test")
 const mid = ModelV2.ID.make("fake-memory-model")
@@ -105,6 +104,71 @@ function provider(
   }
 }
 
+function compatLanguage(input: {
+  generate: () => Promise<unknown>
+  streamed?: string[]
+  called?: string[]
+}): LanguageModelV3 {
+  return {
+    specificationVersion: "v3",
+    provider: "test",
+    modelId: "compat",
+    supportedUrls: {},
+    doGenerate: input.generate,
+    doStream: input.streamed
+      ? async () => ({
+          stream: new ReadableStream({
+            start: (controller: ReadableStreamDefaultController) => {
+              input.called?.push("stream")
+              controller.enqueue({ type: "text-start", id: "1" })
+              for (const chunk of input.streamed!) controller.enqueue({ type: "text-delta", id: "1", delta: chunk })
+              controller.enqueue({ type: "text-end", id: "1" })
+              controller.enqueue({
+                type: "finish",
+                finishReason: "stop" as const,
+                usage: { inputTokens: 12, outputTokens: 8 },
+              })
+              controller.close()
+            },
+          }),
+          request: {},
+          response: {},
+        })
+      : async () => {
+          input.called?.push("stream")
+          throw new Error("streamText should not run")
+        },
+  } as unknown as LanguageModelV3
+}
+
+function compatProvider(input: {
+  generate: () => Promise<unknown>
+  streamed?: string[]
+  called?: string[]
+}): Provider.Interface {
+  const source = mdl(ModelV2.ID.make("compat"), "@ai-sdk/openai-compatible")
+  const info = {
+    id: pid,
+    name: "Test",
+    source: "config",
+    env: [],
+    options: {},
+    models: { [source.id]: source },
+  } satisfies Provider.Info
+  return {
+    list: () => Effect.succeed({ [pid]: info }),
+    getProvider: () => Effect.succeed(info),
+    getModel: (_providerID, modelID) =>
+      info.models[modelID]
+        ? Effect.succeed(info.models[modelID])
+        : Effect.fail(new ModelNotFoundError({ providerID: pid, modelID })),
+    getLanguage: () => Effect.succeed(compatLanguage(input)),
+    closest: () => Effect.succeed({ providerID: pid, modelID: source.id }),
+    getSmallModel: () => Effect.succeed(source),
+    defaultModel: () => Effect.succeed({ providerID: pid, modelID: source.id }),
+  }
+}
+
 function text(sessionID: SessionID, messageID: MessageID, body: string): MessageV2.TextPart {
   return {
     id: PartID.make(`prt_${messageID}_text`),
@@ -203,6 +267,7 @@ function summary(input: { seen: string[]; diffs: Snapshot.FileDiff[] }): Session
 const ref = { providerID: "test", modelID: "fake-memory-model" }
 
 describe("memory ports", () => {
+  beforeEach(() => resetStreamNeeded())
   test("session port extracts the latest turn, recall markers, and all assistant steps", async () => {
     const sessionID = SessionID.make("ses_memory_adapter")
     const uid = MessageID.make("msg_user")
@@ -289,9 +354,7 @@ describe("memory ports", () => {
     const seen: string[] = []
     const port = MemoryModel.port({ provider: provider({ seen }) })
 
-    const configured = await Effect.runPromise(
-      port.resolve({ configured: "test/memory-config-model", session: ref }),
-    )
+    const configured = await Effect.runPromise(port.resolve({ configured: "test/memory-config-model", session: ref }))
     const fallback = await Effect.runPromise(port.resolve({ configured: "test/missing-memory-model", session: ref }))
 
     expect(configured.fallback).toBeUndefined()
@@ -454,4 +517,100 @@ describe("memory ports", () => {
     expect(handles.size).toBe(1)
     expect(cleared.size).toBe(1)
   })
+
+  test("openai-compatible providers fall back to streamText when generateText returns invalid JSON", async () => {
+    const called: string[] = []
+    const port = MemoryModel.port({
+      provider: compatProvider({
+        generate: async () => {
+          throw new APICallError({
+            message: "Invalid JSON response",
+            url: "test://invalid-json",
+            requestBodyValues: {},
+            cause: new JSONParseError({ text: "event: data, not json", cause: undefined }),
+          })
+        },
+        streamed: ["streamed result"],
+        called,
+      }),
+    })
+    const resolved = await Effect.runPromise(port.resolve({ configured: "test/compat", session: ref }))
+    const out = await port.run({ handle: resolved.handle, system: "s", prompt: "p", timeoutMs: 30_000 })
+    expect(out.text).toBe("streamed result")
+    expect(called).toEqual(["stream"])
+  })
+
+  test("openai-compatible providers fall back to streamText on a bare JSONParseError", async () => {
+    const called: string[] = []
+    const port = MemoryModel.port({
+      provider: compatProvider({
+        generate: async () => {
+          throw new JSONParseError({ text: "event: data, not json", cause: undefined })
+        },
+        streamed: ["streamed result"],
+        called,
+      }),
+    })
+    const resolved = await Effect.runPromise(port.resolve({ configured: "test/compat", session: ref }))
+    const out = await port.run({ handle: resolved.handle, system: "s", prompt: "p", timeoutMs: 30_000 })
+    expect(out.text).toBe("streamed result")
+    expect(called).toEqual(["stream"])
+  })
+
+  test("openai-compatible providers stay on generateText when it succeeds", async () => {
+    const called: string[] = []
+    const port = MemoryModel.port({
+      provider: compatProvider({
+        generate: async () => ({
+            content: [{ type: "text", text: '{"topic":"t","summary":"s"}' }],
+            finishReason: { unified: "stop" },
+            usage: { inputTokens: { total: 12 }, outputTokens: { total: 8 }, raw: {} },
+            warnings: [],
+            providerMetadata: {},
+            request: {},
+          response: {},
+        }),
+        called,
+      }),
+    })
+    const resolved = await Effect.runPromise(port.resolve({ configured: "test/compat", session: ref }))
+    const out = await port.run({ handle: resolved.handle, system: "s", prompt: "p", timeoutMs: 30_000 })
+    expect(out.text).toBe('{"topic":"t","summary":"s"}')
+    expect(called).toEqual([])
+  })
+
+  test("openai-compatible providers stream directly once a model is known to need it", async () => {
+    const called: string[] = []
+    const first = MemoryModel.port({
+      provider: compatProvider({
+        generate: async () => {
+          throw new APICallError({
+            message: "Invalid JSON response",
+            url: "test://invalid-json",
+            requestBodyValues: {},
+            cause: new JSONParseError({ text: "event: data, not json", cause: undefined }),
+          })
+        },
+        streamed: ["first"],
+        called,
+      }),
+    })
+    const resolved = await Effect.runPromise(first.resolve({ configured: "test/compat", session: ref }))
+    const out = await first.run({ handle: resolved.handle, system: "s", prompt: "p", timeoutMs: 30_000 })
+    expect(out.text).toBe("first")
+
+    const again = MemoryModel.port({
+      provider: compatProvider({
+        generate: async () => {
+          throw new Error("generateText must not run again")
+        },
+        streamed: ["second"],
+        called,
+      }),
+    })
+    const resolved2 = await Effect.runPromise(again.resolve({ configured: "test/compat", session: ref }))
+    const out2 = await again.run({ handle: resolved2.handle, system: "s", prompt: "p", timeoutMs: 30_000 })
+    expect(out2.text).toBe("second")
+    expect(called).toEqual(["stream", "stream"])
+  })
 })