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 <steipete@macos.shared>

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" });
+    }
+  });
+});

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();

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<string, unknown>,
+) {
+  const provider = await registerSingleProviderPlugin(openrouterPlugin);
+  const baseStreamFn = vi.fn((...args: Parameters<StreamFn>): ReturnType<StreamFn> => {
+    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: "" },
+    ]);
+  });
+});

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],

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) ?? "",
+  );
 }

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<StreamFn>[0]): boolean {
   const provider = readString(model.provider)?.toLowerCase();
   const baseUrl = readString(model.baseUrl);
@@ -43,7 +35,7 @@ function shouldPatchAnthropicOpenRouterPayload(model: Parameters<StreamFn>[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<string, unknown>).effort;
+    const reasoning = value as Record<string, unknown>;
+    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";