From b658e6949aefdec183b63e095d0db36160812809 Mon Sep 17 00:00:00 2001 From: colt Date: Wed, 2 Sep 2026 17:22:02 +0300 Subject: [PATCH] fix(cli): route openai-compatible providers through streamText for memory consolidation --- .changeset/memory-streamtext-openai.md | 5 + .../opencode/src/kilocode/memory/ports.ts | 48 +++-- .../test/kilocode/memory/memory-ports.test.ts | 173 +++++++++++++++++- 3 files changed, 206 insertions(+), 20 deletions(-) create mode 100644 .changeset/memory-streamtext-openai.md diff --git a/.changeset/memory-streamtext-openai.md b/.changeset/memory-streamtext-openai.md new file mode 100644 index 00000000000..a7f0bdf1ccb --- /dev/null +++ b/.changeset/memory-streamtext-openai.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": patch +--- + +Fix memory consolidation failing with `Invalid JSON response` on OpenAI-compatible providers. diff --git a/packages/opencode/src/kilocode/memory/ports.ts b/packages/opencode/src/kilocode/memory/ports.ts index 9f6fc42d884..6199aebbfaa 100644 --- a/packages/opencode/src/kilocode/memory/ports.ts +++ b/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[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() + +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 | undefined const timeout = new Promise((_, reject) => { diff --git a/packages/opencode/test/kilocode/memory/memory-ports.test.ts b/packages/opencode/test/kilocode/memory/memory-ports.test.ts index c316dee12f4..0fa75b5643e 100644 --- a/packages/opencode/test/kilocode/memory/memory-ports.test.ts +++ b/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 + 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 + 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"]) + }) })