fix(anthropic): restore Fable 5 Vertex simple completions (#98932)

steipete · web-flow · commit 2913d3aee90c · 2026-07-02T07:16:39.000+01:00
* fix(anthropic): restore Fable 5 Vertex simple completions

* test(agents): satisfy custom API model types

* test(ci): route reliability test from temp helper

* test(agents): satisfy custom API model types
diff --git a/extensions/anthropic-vertex/stream-runtime.test.ts b/extensions/anthropic-vertex/stream-runtime.test.ts
@@ -177,6 +177,23 @@ describe("createAnthropicVertexStreamFn", () => {
     });
   });
 
+  it("restores the canonical API before calling the shared Anthropic transport", () => {
+    const { deps, streamAnthropicMock } = createStreamDeps();
+    const streamFn = createAnthropicVertexStreamFn("vertex-project", "us-east5", undefined, deps);
+    const model = {
+      ...makeModel({ id: "claude-fable-5", maxTokens: 128000 }),
+      api: "openclaw-anthropic-vertex-simple:default",
+    };
+
+    void streamFn(model as never, { messages: [] }, {});
+
+    expect(streamAnthropicCall(streamAnthropicMock)[0]).toMatchObject({
+      api: "anthropic-messages",
+      provider: "anthropic-vertex",
+      id: "claude-fable-5",
+    });
+  });
+
   it("defaults maxTokens to the model limit instead of the old 32000 cap", () => {
     const { deps, streamAnthropicMock } = createStreamDeps();
     const streamFn = createAnthropicVertexStreamFn("vertex-project", "us-east5", undefined, deps);
diff --git a/extensions/anthropic-vertex/stream-runtime.ts b/extensions/anthropic-vertex/stream-runtime.ts
@@ -135,8 +135,11 @@ export function createAnthropicVertexStreamFn(
   });
 
   return (model, context, options) => {
-    const transportModel = model as Model<"anthropic-messages"> & {
-      api: string;
+    // Simple completions use a synthetic registry API to select this plugin.
+    // The shared Anthropic transport must receive its canonical API or it recurses.
+    const transportModel = (
+      model.api === "anthropic-messages" ? model : { ...model, api: "anthropic-messages" as const }
+    ) as Model<"anthropic-messages"> & {
       baseUrl?: string;
       provider: string;
     };
diff --git a/src/agents/custom-api-registry.test.ts b/src/agents/custom-api-registry.test.ts
@@ -12,6 +12,7 @@ import {
 } from "../llm/providers/register-builtins.js";
 import { createAssistantMessageEventStream } from "../llm/utils/event-stream.js";
 import { ensureCustomApiRegistered } from "./custom-api-registry.js";
+import { buildAssistantMessageWithZeroUsage } from "./stream-message-shared.js";
 
 function getRegisteredTestProvider() {
   const provider = getApiProvider("test-custom-api");
@@ -56,6 +57,50 @@ describe("ensureCustomApiRegistered", () => {
     expect(streamFn).toHaveBeenCalledTimes(2);
   });
 
+  it("adapts async stream factories to the synchronous provider contract", async () => {
+    const message = buildAssistantMessageWithZeroUsage({
+      model: { api: "test-custom-api", provider: "custom", id: "m" },
+      content: [{ type: "text", text: "done" }],
+      stopReason: "stop",
+    });
+    const streamFn = vi.fn(async () => {
+      await Promise.resolve();
+      const stream = createAssistantMessageEventStream();
+      stream.push({ type: "done", reason: "stop", message });
+      return stream;
+    });
+    ensureCustomApiRegistered("test-custom-api", streamFn);
+
+    const provider = getRegisteredTestProvider();
+    const stream = provider.stream(
+      { api: "test-custom-api", provider: "custom", id: "m" } as never,
+      { messages: [] },
+      {},
+    );
+
+    expect(stream).not.toBeInstanceOf(Promise);
+    await expect(stream.result()).resolves.toBe(message);
+  });
+
+  it("converts async stream factory failures into terminal stream errors", async () => {
+    const streamFn = vi.fn(async () => {
+      throw new Error("factory failed");
+    });
+    ensureCustomApiRegistered("test-custom-api", streamFn);
+
+    const provider = getRegisteredTestProvider();
+    const stream = provider.stream(
+      { api: "test-custom-api", provider: "custom", id: "m" } as never,
+      { messages: [] },
+      {},
+    );
+
+    await expect(stream.result()).resolves.toMatchObject({
+      stopReason: "error",
+      errorMessage: "factory failed",
+    });
+  });
+
   it("keeps plugin api providers when refreshing built-ins", () => {
     // Built-in refresh should preserve plugin-owned API providers while
     // repopulating core providers.
diff --git a/src/agents/custom-api-registry.ts b/src/agents/custom-api-registry.ts
@@ -2,8 +2,15 @@
  * Registers caller-supplied custom API stream functions with the LLM registry.
  */
 import { getApiProvider, registerApiProvider } from "../llm/api-registry.js";
-import type { Api, StreamOptions } from "../llm/types.js";
+import type {
+  Api,
+  AssistantMessageEventStreamContract,
+  Model,
+  StreamOptions,
+} from "../llm/types.js";
+import { createAssistantMessageEventStream } from "../llm/utils/event-stream.js";
 import type { StreamFn } from "./runtime/index.js";
+import { buildStreamErrorAssistantMessage } from "./stream-message-shared.js";
 
 const CUSTOM_API_SOURCE_PREFIX = "openclaw-custom-api:";
 
@@ -12,6 +19,33 @@ function getCustomApiRegistrySourceId(api: Api): string {
   return `${CUSTOM_API_SOURCE_PREFIX}${api}`;
 }
 
+function adaptCustomStream(
+  model: Model,
+  stream: ReturnType<StreamFn>,
+): AssistantMessageEventStreamContract {
+  if (!(stream instanceof Promise)) {
+    return stream as AssistantMessageEventStreamContract;
+  }
+
+  const adapted = createAssistantMessageEventStream();
+  void (async () => {
+    try {
+      // Registry providers must return a stream immediately, while plugin
+      // hooks may resolve one lazily. Bridge that lifecycle at the boundary.
+      const resolved = await stream;
+      for await (const event of resolved) {
+        adapted.push(event);
+      }
+      adapted.end(await resolved.result());
+    } catch (error) {
+      const errorMessage = error instanceof Error ? error.message : String(error);
+      const message = buildStreamErrorAssistantMessage({ model, errorMessage });
+      adapted.push({ type: "error", reason: "error", error: message });
+    }
+  })();
+  return adapted;
+}
+
 /** Registers a custom API stream function when no provider already owns it. */
 export function ensureCustomApiRegistered(api: Api, streamFn: StreamFn): boolean {
   if (getApiProvider(api)) {
@@ -22,13 +56,9 @@ export function ensureCustomApiRegistered(api: Api, streamFn: StreamFn): boolean
     {
       api,
       stream: (model, context, options) =>
-        streamFn(model, context, options) as unknown as ReturnType<
-          NonNullable<ReturnType<typeof getApiProvider>>["stream"]
-        >,
+        adaptCustomStream(model, streamFn(model, context, options)),
       streamSimple: (model, context, options) =>
-        streamFn(model, context, options as StreamOptions) as unknown as ReturnType<
-          NonNullable<ReturnType<typeof getApiProvider>>["stream"]
-        >,
+        adaptCustomStream(model, streamFn(model, context, options as StreamOptions)),
     },
     getCustomApiRegistrySourceId(api),
   );
