From d92d68ac7766a5b5f33606f77872daf9f0fbe144 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 15:29:26 -0700 Subject: [PATCH] fix(providers): canonicalize model families and stream outcomes (#116998) * fix(providers): canonicalize live model families and stream finalization * test(openrouter): use canonical typed stream fixture --------- Co-authored-by: Peter Steinberger --- .../stream.runtime.lifecycle.test.ts | 110 ++++++++++++++++ extensions/amazon-bedrock/stream.runtime.ts | 12 ++ .../openrouter/chat-owner-regression.test.ts | 117 ++++++++++++++++++ extensions/openrouter/index.ts | 21 ++-- extensions/openrouter/models.ts | 17 +-- extensions/openrouter/stream.ts | 18 ++- 6 files changed, 260 insertions(+), 35 deletions(-) create mode 100644 extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts create mode 100644 extensions/openrouter/chat-owner-regression.test.ts diff --git a/extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts b/extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts new file mode 100644 index 000000000000..ff58b5026a9a --- /dev/null +++ b/extensions/amazon-bedrock/stream.runtime.lifecycle.test.ts @@ -0,0 +1,110 @@ +import { + BedrockRuntimeClient, + ConversationRole, + StopReason as BedrockStopReason, +} from "@aws-sdk/client-bedrock-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { streamSimpleBedrock } from "./stream.runtime.js"; + +const model = { + api: "bedrock-converse-stream", + provider: "amazon-bedrock", + id: "amazon.nova-micro-v1:0", + name: "Nova Micro", + baseUrl: "https://bedrock-runtime.us-east-1.amazonaws.com", + reasoning: true, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128_000, + maxTokens: 4096, +} as const; + +async function* events(items: unknown[]) { + yield* items; +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("Bedrock provider-owned stream lifecycle", () => { + it.each([ + { + label: "text", + blocks: [{ contentBlockDelta: { contentBlockIndex: 0, delta: { text: "ready" } } }], + endEvent: "text_end", + stopReason: BedrockStopReason.END_TURN, + }, + { + label: "thinking", + blocks: [ + { + contentBlockDelta: { + contentBlockIndex: 0, + delta: { reasoningContent: { text: "considered" } }, + }, + }, + ], + endEvent: "thinking_end", + stopReason: BedrockStopReason.END_TURN, + }, + { + label: "redacted thinking", + blocks: [ + { + contentBlockDelta: { + contentBlockIndex: 0, + delta: { reasoningContent: { redactedContent: new Uint8Array([1, 2, 3]) } }, + }, + }, + ], + endEvent: "thinking_end", + stopReason: BedrockStopReason.END_TURN, + }, + { + label: "tool call", + blocks: [ + { + contentBlockStart: { + contentBlockIndex: 0, + start: { toolUse: { toolUseId: "call_lookup", name: "lookup" } }, + }, + }, + { + contentBlockDelta: { + contentBlockIndex: 0, + delta: { toolUse: { input: '{"query":"ready"}' } }, + }, + }, + ], + endEvent: "toolcall_end", + stopReason: BedrockStopReason.TOOL_USE, + }, + ])("finalizes the active $label block at the provider terminal boundary", async (scenario) => { + vi.spyOn(BedrockRuntimeClient.prototype, "send").mockResolvedValue({ + $metadata: { httpStatusCode: 200 }, + stream: events([ + { messageStart: { role: ConversationRole.ASSISTANT } }, + ...scenario.blocks, + { messageStop: { stopReason: scenario.stopReason } }, + ]), + } as never); + + const stream = streamSimpleBedrock(model as never, { + messages: [{ role: "user", content: "Continue", timestamp: 0 }], + }); + const observed = []; + for await (const event of stream) { + observed.push(event.type); + } + const output = await stream.result(); + + expect(observed.at(-2)).toBe(scenario.endEvent); + expect(observed.at(-1)).toBe("done"); + expect(output.content[0]).not.toHaveProperty("index"); + expect(output.content[0]).not.toHaveProperty("partialJson"); + if (scenario.label === "redacted thinking") { + expect(output.content[0]).toMatchObject({ redacted: true, thinkingSignature: "AQID" }); + } + }); +}); diff --git a/extensions/amazon-bedrock/stream.runtime.ts b/extensions/amazon-bedrock/stream.runtime.ts index 688e9e6f1b81..a0bedf3609aa 100644 --- a/extensions/amazon-bedrock/stream.runtime.ts +++ b/extensions/amazon-bedrock/stream.runtime.ts @@ -357,6 +357,18 @@ const streamBedrock: StreamFunction<"bedrock-converse-stream", BedrockOptions> = throw new Error(output.errorMessage ?? "An unknown error occurred"); } + // Some valid provider streams omit contentBlockStop; never persist their scratch state. + for (const block of blocks) { + if (block.index !== undefined) { + handleContentBlockStop( + { contentBlockIndex: block.index }, + blocks, + output, + eventSink, + redactedReasoningChunks, + ); + } + } refusalBuffer?.flush(); stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); diff --git a/extensions/openrouter/chat-owner-regression.test.ts b/extensions/openrouter/chat-owner-regression.test.ts new file mode 100644 index 000000000000..2fd508d11072 --- /dev/null +++ b/extensions/openrouter/chat-owner-regression.test.ts @@ -0,0 +1,117 @@ +import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; +import { createAssistantMessageEventStream } from "openclaw/plugin-sdk/llm"; +import { registerSingleProviderPlugin } from "openclaw/plugin-sdk/plugin-test-runtime"; +import { describe, expect, it, vi } from "vitest"; +import openrouterPlugin from "./index.js"; + +async function captureProviderPayload( + modelId: string, + thinkingLevel: string, + payload: Record, +) { + const provider = await registerSingleProviderPlugin(openrouterPlugin); + const baseStreamFn = vi.fn((...args: Parameters): ReturnType => { + void args[2]?.onPayload?.(payload, args[0]); + return createAssistantMessageEventStream(); + }); + const wrapped = provider.wrapStreamFn?.({ + provider: "openrouter", + modelId, + thinkingLevel, + streamFn: baseStreamFn, + } as never); + + void wrapped?.( + { + provider: "openrouter", + api: "openai-completions", + id: modelId, + baseUrl: "https://openrouter.ai/api/v1", + compat: {}, + } as never, + { messages: [] }, + {}, + ); + + expect(baseStreamFn).toHaveBeenCalledOnce(); + return payload; +} + +describe("OpenRouter chat owner invariants", () => { + it.each([ + "openrouter/anthropic/claude-sonnet-5", + "openrouter/deepseek/deepseek-v4-pro", + "openrouter/moonshotai/kimi-k3", + "z-ai/glm-5.2", + "openrouter/z-ai/glm-5.2", + "~anthropic/claude-opus-latest", + "~moonshotai/kimi-latest", + ])("recognizes the live upstream cacheable model reference %s", async (modelId) => { + const provider = await registerSingleProviderPlugin(openrouterPlugin); + + expect(provider.isCacheTtlEligible?.({ provider: "openrouter", modelId } as never)).toBe(true); + }); + + it.each(["openai/gpt-5.4", "~openai/gpt-5.4", "openrouter/openai/gpt-5.4"])( + "does not infer provider cache support for unrelated model reference %s", + async (modelId) => { + const provider = await registerSingleProviderPlugin(openrouterPlugin); + + expect(provider.isCacheTtlEligible?.({ provider: "openrouter", modelId } as never)).toBe( + false, + ); + }, + ); + + it("preserves assistant prefill when the provider reasoning object disables reasoning", async () => { + const payload = await captureProviderPayload("anthropic/claude-opus-5", "off", { + reasoning: { enabled: false }, + messages: [ + { role: "user", content: "Return JSON." }, + { role: "assistant", content: "{" }, + ], + }); + + expect(payload.messages).toEqual([ + { role: "user", content: "Return JSON." }, + { role: "assistant", content: "{" }, + ]); + }); + + it.each(["~anthropic/claude-opus-latest", "openrouter/~anthropic/claude-opus-latest"])( + "removes unsupported assistant prefill for reasoning model alias %s", + async (modelId) => { + const payload = await captureProviderPayload(modelId, "high", { + reasoning: { effort: "high" }, + messages: [ + { role: "user", content: "Continue." }, + { role: "assistant", content: "{" }, + ], + }); + + expect(payload.messages).toEqual([{ role: "user", content: "Continue." }]); + }, + ); + + it("recognizes the live dated DeepSeek V4 model in its owner thinking profile", async () => { + const provider = await registerSingleProviderPlugin(openrouterPlugin); + + expect( + provider.resolveThinkingProfile?.({ + provider: "openrouter", + modelId: "deepseek/deepseek-v4-flash-0731", + } as never), + ).toMatchObject({ defaultLevel: "high" }); + }); + + it("backfills reasoning replay for the live dated DeepSeek V4 model", async () => { + const payload = await captureProviderPayload("deepseek/deepseek-v4-flash-0731", "high", { + messages: [{ role: "assistant", content: "done" }], + }); + + expect(payload.reasoning).toEqual({ effort: "high" }); + expect(payload.messages).toEqual([ + { role: "assistant", content: "done", reasoning_content: "" }, + ]); + }); +}); diff --git a/extensions/openrouter/index.ts b/extensions/openrouter/index.ts index d011c419fc18..e9ac92123f62 100644 --- a/extensions/openrouter/index.ts +++ b/extensions/openrouter/index.ts @@ -18,7 +18,11 @@ import { asOptionalRecord as readRecord } from "openclaw/plugin-sdk/string-coerc import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { buildOpenRouterImageGenerationProvider } from "./image-generation-provider.js"; import { openrouterMediaUnderstandingProvider } from "./media-understanding-provider.js"; -import { isOpenRouterMistralModelId, normalizeOpenRouterApiModelId } from "./models.js"; +import { + isOpenRouterMistralModelId, + normalizeOpenRouterApiModelId, + normalizeOpenRouterModelFamilyId, +} from "./models.js"; import { buildOpenRouterMusicGenerationProvider } from "./music-generation-provider.js"; import { createOpenRouterOAuthAuthMethod } from "./oauth.js"; import { applyOpenrouterConfig, OPENROUTER_DEFAULT_MODEL_REF } from "./onboard.js"; @@ -43,13 +47,7 @@ import { const PROVIDER_ID = "openrouter"; const OPENROUTER_DEFAULT_MAX_TOKENS = 8192; const OPENROUTER_FUSION_MODEL_ID = "openrouter/fusion"; -const OPENROUTER_CACHE_TTL_MODEL_PREFIXES = [ - "anthropic/", - "deepseek/", - "moonshot/", - "moonshotai/", - "zai/", -] as const; +const OPENROUTER_CACHE_TTL_MODEL_FAMILY = /^(?:anthropic|deepseek|moonshot(?:ai)?|z-?ai)\//; const MAX_PROMPT_MODEL_ID_DISPLAY_CHARS = 256; type OpenRouterFusionPromptContext = { @@ -253,10 +251,6 @@ export default defineSingleProviderPluginEntry({ }; } - function isOpenRouterCacheTtlModel(modelId: string): boolean { - return OPENROUTER_CACHE_TTL_MODEL_PREFIXES.some((prefix) => modelId.startsWith(prefix)); - } - const passthroughGeminiReplayHooks = buildProviderReplayFamilyHooks({ family: "passthrough-gemini", }); @@ -335,7 +329,8 @@ export default defineSingleProviderPluginEntry({ resolveSystemPromptContribution: resolveOpenRouterFusionPromptContribution, extraParamsForTransport: resolveOpenRouterExtraParamsForTransport, wrapStreamFn: wrapOpenRouterProviderStream, - isCacheTtlEligible: (ctx) => isOpenRouterCacheTtlModel(ctx.modelId), + isCacheTtlEligible: ({ modelId }) => + OPENROUTER_CACHE_TTL_MODEL_FAMILY.test(normalizeOpenRouterModelFamilyId(modelId) ?? ""), resolveUsageAuth: async (ctx) => { const apiKey = ctx.resolveApiKeyFromConfigAndStore({ envDirect: [ctx.env.OPENROUTER_API_KEY], diff --git a/extensions/openrouter/models.ts b/extensions/openrouter/models.ts index 1c17d32bf338..50965e30e010 100644 --- a/extensions/openrouter/models.ts +++ b/extensions/openrouter/models.ts @@ -21,14 +21,12 @@ const OPENROUTER_SHORT_TO_API_MODEL_ID = new Map([ ["deepseek-v4-pro", "deepseek/deepseek-v4-pro"], ]); -function normalizeOpenRouterModelId(modelId: unknown): string | undefined { +export function normalizeOpenRouterModelFamilyId(modelId: unknown): string | undefined { if (typeof modelId !== "string") { return undefined; } const normalized = normalizeLowercaseStringOrEmpty(modelId); - return normalized.startsWith(OPENROUTER_MODEL_PREFIX) - ? normalized.slice(OPENROUTER_MODEL_PREFIX.length) - : normalized; + return normalized.replace(/^openrouter\//, "").replace(/^~/, ""); } export function normalizeOpenRouterApiModelId(modelId: unknown): string | undefined { @@ -50,17 +48,14 @@ export function normalizeOpenRouterApiModelId(modelId: unknown): string | undefi } export function isOpenRouterMistralModelId(modelId: unknown): boolean { - const normalized = normalizeOpenRouterModelId(modelId); + const normalized = normalizeOpenRouterModelFamilyId(modelId); return Boolean( normalized && OPENROUTER_MISTRAL_MODEL_PREFIXES.some((prefix) => normalized.startsWith(prefix)), ); } export function isOpenRouterDeepSeekV4ModelId(modelId: unknown): boolean { - const normalized = normalizeOpenRouterModelId(modelId); - if (!normalized?.startsWith("deepseek/")) { - return false; - } - const deepSeekModelId = normalized.slice("deepseek/".length).split(":", 1)[0]; - return deepSeekModelId === "deepseek-v4-flash" || deepSeekModelId === "deepseek-v4-pro"; + return /^deepseek\/deepseek-v4-(?:flash|pro)(?:-\d{4,8})?(?::[^/]*)?$/.test( + normalizeOpenRouterModelFamilyId(modelId) ?? "", + ); } diff --git a/extensions/openrouter/stream.ts b/extensions/openrouter/stream.ts index 976884a022a2..900b2bbe192a 100644 --- a/extensions/openrouter/stream.ts +++ b/extensions/openrouter/stream.ts @@ -8,7 +8,7 @@ import { normalizeOpenAICompatibleReasoningReplay, } from "openclaw/plugin-sdk/provider-stream-shared"; import { createSubsystemLogger } from "openclaw/plugin-sdk/runtime-env"; -import { isOpenRouterDeepSeekV4ModelId } from "./models.js"; +import { isOpenRouterDeepSeekV4ModelId, normalizeOpenRouterModelFamilyId } from "./models.js"; import { isOpenRouterProxyReasoningUnsupportedModel, normalizeOpenRouterBaseUrl, @@ -22,14 +22,6 @@ function readString(value: unknown): string | undefined { return typeof value === "string" ? value.trim() : undefined; } -function isOpenRouterAnthropicModelId(modelId: unknown): boolean { - const normalized = readString(modelId)?.toLowerCase(); - return ( - normalized?.startsWith("anthropic/") === true || - normalized?.startsWith("openrouter/anthropic/") === true - ); -} - function isVerifiedOpenRouterRoute(model: Parameters[0]): boolean { const provider = readString(model.provider)?.toLowerCase(); const baseUrl = readString(model.baseUrl); @@ -43,7 +35,7 @@ function shouldPatchAnthropicOpenRouterPayload(model: Parameters[0]): const api = readString(model.api); return ( (api === undefined || api === "openai-completions") && - isOpenRouterAnthropicModelId(model.id) && + normalizeOpenRouterModelFamilyId(model.id)?.startsWith("anthropic/") === true && isVerifiedOpenRouterRoute(model) ); } @@ -154,7 +146,11 @@ function isEnabledReasoningValue(value: unknown): boolean { return normalized !== "" && normalized !== "off" && normalized !== "none"; } if (typeof value === "object" && !Array.isArray(value)) { - const effort = (value as Record).effort; + const reasoning = value as Record; + if (reasoning.enabled === false) { + return false; + } + const effort = reasoning.effort; if (typeof effort === "string") { const normalized = effort.trim().toLowerCase(); return normalized !== "" && normalized !== "off" && normalized !== "none";