diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index a99f8acff20c..6be09c145775 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -369,7 +369,7 @@ const live: Layer.Layer< // Adapter seam: both runtimes expose the same LLMEvent stream. Native // already returns one; AI SDK streams are converted here. - const state = LLMAISDK.adapterState() + const state = LLMAISDK.adapterState(input.model.providerID) return Stream.fromAsyncIterable(result.result.fullStream, (e) => e instanceof Error ? e : new Error(String(e)), ).pipe( diff --git a/packages/opencode/src/session/llm/ai-sdk.ts b/packages/opencode/src/session/llm/ai-sdk.ts index 13d427aab62c..77bb9343f4e1 100644 --- a/packages/opencode/src/session/llm/ai-sdk.ts +++ b/packages/opencode/src/session/llm/ai-sdk.ts @@ -7,8 +7,9 @@ import { ProviderError } from "@/provider/error" type Result = Awaited> type AISDKEvent = Result["fullStream"] extends AsyncIterable ? T : never -export function adapterState() { +export function adapterState(providerID?: string) { return { + providerID, step: 0, text: 0, reasoning: 0, @@ -16,6 +17,7 @@ export function adapterState() { currentReasoningID: undefined as string | undefined, toolNames: {} as Record, copilotTotalNanoAiu: undefined as number | undefined, + hasOutput: false, } } @@ -88,6 +90,16 @@ export function toLLMEvents( case "finish-step": if (event.rawFinishReason === "network_error") return Effect.fail(new ProviderError.ResponseStreamError("Provider finish_reason: network_error")) + if ( + event.finishReason === "stop" && + state.providerID === "openrouter" && + state.step === 0 && + !state.hasOutput && + !event.usage.inputTokens && + !event.usage.outputTokens && + !event.usage.totalTokens + ) + return Effect.fail(new ProviderError.ResponseStreamError("Provider returned an empty response")) return Effect.sync(() => { const original = providerMetadata(event.providerMetadata) const metadata = @@ -101,6 +113,7 @@ export function toLLMEvents( }, } state.copilotTotalNanoAiu = undefined + state.hasOutput = false return [ LLMEvent.stepFinish({ index: state.step++, @@ -122,7 +135,7 @@ export function toLLMEvents( ] // Reset so the adapter can be reused for a follow-up stream without leaking // counters or block IDs. adapterState() is the single source of truth for shape. - Object.assign(state, adapterState()) + Object.assign(state, adapterState(state.providerID)) return events }) @@ -138,13 +151,16 @@ export function toLLMEvents( }) case "text-delta": - return Effect.succeed([ - LLMEvent.textDelta({ - id: currentTextID(state, event.id), - text: event.text, - providerMetadata: providerMetadata(event.providerMetadata), - }), - ]) + return Effect.sync(() => { + state.hasOutput ||= event.text.length > 0 + return [ + LLMEvent.textDelta({ + id: currentTextID(state, event.id), + text: event.text, + providerMetadata: providerMetadata(event.providerMetadata), + }), + ] + }) case "text-end": return Effect.sync(() => { @@ -170,13 +186,16 @@ export function toLLMEvents( }) case "reasoning-delta": - return Effect.succeed([ - LLMEvent.reasoningDelta({ - id: currentReasoningID(state, event.id), - text: event.text, - providerMetadata: providerMetadata(event.providerMetadata), - }), - ]) + return Effect.sync(() => { + state.hasOutput ||= event.text.length > 0 + return [ + LLMEvent.reasoningDelta({ + id: currentReasoningID(state, event.id), + text: event.text, + providerMetadata: providerMetadata(event.providerMetadata), + }), + ] + }) case "reasoning-end": return Effect.sync(() => { @@ -222,6 +241,7 @@ export function toLLMEvents( case "tool-call": return Effect.sync(() => { + state.hasOutput = true state.toolNames[event.toolCallId] = event.toolName return [ LLMEvent.toolCall({ diff --git a/packages/opencode/test/fixtures/openrouter-ox-duplicate-stop.json b/packages/opencode/test/fixtures/openrouter-ox-duplicate-stop.json new file mode 100644 index 000000000000..3a6d110d8a6c --- /dev/null +++ b/packages/opencode/test/fixtures/openrouter-ox-duplicate-stop.json @@ -0,0 +1,64 @@ +{ + "chunks": [ + { + "id": "chatcmpl-ox", + "model": "stealth/ox-alpha", + "provider": "Stealth", + "choices": [ + { + "index": 0, + "finish_reason": null, + "delta": { + "role": "assistant", + "content": "", + "reasoning": "thinking", + "reasoning_details": [{ "type": "reasoning.text", "text": "thinking" }] + } + } + ] + }, + { + "id": "chatcmpl-ox", + "model": "stealth/ox-alpha", + "provider": "Stealth", + "choices": [ + { + "index": 0, + "finish_reason": null, + "delta": { "role": "assistant", "content": "OX_STREAM_OK" } + } + ] + }, + { + "id": "chatcmpl-ox", + "model": "stealth/ox-alpha", + "provider": "Stealth", + "choices": [ + { + "index": 0, + "finish_reason": "stop", + "delta": { "role": "assistant", "content": "", "reasoning": "" } + } + ] + }, + { + "id": "chatcmpl-ox", + "model": "stealth/ox-alpha", + "provider": "Stealth", + "choices": [ + { + "index": 0, + "finish_reason": "stop", + "delta": { "role": "assistant", "content": "" } + } + ], + "usage": { + "prompt_tokens": 96, + "completion_tokens": 52, + "total_tokens": 148, + "prompt_tokens_details": { "cached_tokens": 64 }, + "completion_tokens_details": { "reasoning_tokens": 0 } + } + } + ] +} diff --git a/packages/opencode/test/session/llm.test.ts b/packages/opencode/test/session/llm.test.ts index 635e697517c3..223d00059079 100644 --- a/packages/opencode/test/session/llm.test.ts +++ b/packages/opencode/test/session/llm.test.ts @@ -9,6 +9,7 @@ import { InstanceRef } from "../../src/effect/instance-ref" import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http" import z from "zod" import { LLM } from "../../src/session/llm" +import type { LLMEvent } from "@opencode-ai/llm" import { LLMClient, RequestExecutor } from "@opencode-ai/llm/route" import { Provider } from "@/provider/provider" import { ProviderTransform } from "@/provider/transform" @@ -57,6 +58,8 @@ const it = testEffect(AppNodeBuilder.build(LayerNode.group([LLM.node, Provider.n // LLM.stream returns a Stream, not an Effect, so we can't use the serviceUse proxy. const drain = (input: LLM.StreamInput) => LLM.Service.use((svc) => svc.stream(input).pipe(Stream.runDrain)) +const collect = (input: LLM.StreamInput) => + LLM.Service.use((svc) => svc.stream(input).pipe(Stream.runCollect, Effect.map(Array.from))) // drainWith builds an isolated runtime so custom replacements fully own LLM and // its transitive deps. @@ -362,6 +365,7 @@ describe("session.llm.ai-sdk adapter", () => { // telemetry distinguishes "missing" from "zero," so emitting an empty object causes // false positives ("usage was tracked, just empty") instead of correct nulls. const events = await adapt([ + uncheckedAdapterEvent({ type: "text-delta", text: "ok" }), { type: "finish-step", response: { id: "response-1", timestamp: new Date(0), modelId: "gpt-test" }, @@ -380,8 +384,8 @@ describe("session.llm.ai-sdk adapter", () => { }, ]) - expect(events).toHaveLength(1) - const stepFinish = events[0] + expect(events).toHaveLength(2) + const stepFinish = events[1] if (stepFinish.type !== "step-finish") throw new Error("expected step-finish") expect(stepFinish.usage).toBeUndefined() }) @@ -897,6 +901,73 @@ describe("session.llm.stream", () => { }, ) + const openrouterFixture = { providerID: "openrouter", modelID: "inclusionai/ling-2.6-1t" } + it.instance( + "preserves OX Alpha text and usage across duplicate stop chunks", + () => + Effect.gen(function* () { + const fixture = loadFixture(openrouterFixture.providerID, openrouterFixture.modelID) + const stream = (yield* Effect.promise(() => + Bun.file(path.join(import.meta.dir, "../fixtures/openrouter-ox-duplicate-stop.json")).json(), + )) as { chunks: unknown[] } + const request = waitRequest("/chat/completions", createEventResponse(stream.chunks, true)) + const resolved = yield* Provider.use.getModel( + ProviderV2.ID.make(openrouterFixture.providerID), + ModelV2.ID.make(fixture.model.id), + ) + const sessionID = SessionID.make("session-test-ox-duplicate-stop") + const agent = { + name: "test", + mode: "primary", + options: {}, + permission: [{ permission: "*", pattern: "*", action: "allow" }], + } satisfies Agent.Info + const user = { + id: MessageID.make("msg_user-ox-duplicate-stop"), + sessionID, + role: "user", + time: { created: Date.now() }, + agent: agent.name, + model: { providerID: ProviderV2.ID.make(openrouterFixture.providerID), modelID: resolved.id }, + } satisfies SessionV1.User + + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- LLM.Service.use erases the stream item type + const events = (yield* collect({ + user, + sessionID, + model: resolved, + agent, + system: ["You are a helpful assistant."], + messages: [{ role: "user", content: "Reply with exactly OX_STREAM_OK" }], + tools: {}, + })) as LLMEvent[] + yield* Effect.promise(() => request) + + expect( + events + .filter((event) => event.type === "text-delta") + .map((event) => event.text) + .join(""), + ).toBe("OX_STREAM_OK") + expect(events.find((event) => event.type === "step-finish")?.usage).toMatchObject({ + inputTokens: 96, + outputTokens: 52, + totalTokens: 148, + cacheReadInputTokens: 64, + }) + }), + { + config: () => ({ + enabled_providers: [openrouterFixture.providerID], + provider: { + [openrouterFixture.providerID]: { + options: { apiKey: "test-key", baseURL: `${state.server!.url.origin}/v1` }, + }, + }, + }), + }, + ) + const cerebrasFixture = { providerID: "cerebras", modelID: "gpt-oss-120b" } it.instance( "replays Cerebras assistant reasoning using the provider-supported field", diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index c67f82d9c71b..36b6ff71d8f3 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -41,6 +41,11 @@ const ref = { modelID: ModelV2.ID.make("test-model"), } +const openrouterRef = { + providerID: ProviderV2.ID.make("openrouter"), + modelID: ModelV2.ID.make("test-model"), +} + const cfg = { provider: { test: { @@ -86,6 +91,22 @@ function providerCfg(url: string) { } } +function openrouterCfg(url: string) { + return { + provider: { + openrouter: { + ...cfg.provider.test, + id: "openrouter", + npm: "@openrouter/ai-sdk-provider", + options: { + ...cfg.provider.test.options, + baseURL: url, + }, + }, + }, + } +} + function agent(): Agent.Info { return { name: "build", @@ -708,6 +729,69 @@ it.live("session.processor effect tests retry network_error finish reasons", () ), ) +it.live("session.processor effect tests retry empty successful streams", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + + yield* llm.push( + raw({ + chunks: [ + { + id: "chatcmpl-empty-stop", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { role: "assistant", content: "" }, finish_reason: "stop" }], + }, + { + id: "chatcmpl-empty-stop", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { role: "assistant", content: "" }, finish_reason: "stop" }], + usage: { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, + }, + ], + }), + ) + yield* llm.text("after empty retry") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "retry empty response") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(openrouterRef.providerID, openrouterRef.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: { providerID: openrouterRef.providerID, modelID: openrouterRef.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "retry empty response" }], + tools: {}, + }) + + const parts = yield* MessageV2.parts(msg.id) + + expect(value).toBe("continue") + expect(yield* llm.calls).toBe(2) + expect(parts.some((part) => part.type === "text" && part.text === "after empty retry")).toBe(true) + expect(handle.message.error).toBeUndefined() + }), + { config: (url) => openrouterCfg(url) }, + ), +) + it.live("session.processor effect tests publish retry status updates", () => provideTmpdirServer( ({ dir, llm }) =>