fix(anthropic-vertex): stop re-marking cache_control on transport-budgeted payloads

openperf · openperf · commit 6ef19602bfb4 · 2026-06-12T16:00:52.000+08:00
diff --git a/extensions/anthropic-vertex/stream-runtime.test.ts b/extensions/anthropic-vertex/stream-runtime.test.ts
@@ -3,8 +3,6 @@ import { createAssistantMessageEventStream, type Model } from "openclaw/plugin-s
 import { beforeAll, describe, expect, it, vi } from "vitest";
 import type { AnthropicVertexStreamDeps } from "./stream-runtime.js";
 
-const SYSTEM_PROMPT_CACHE_BOUNDARY = "\n<!-- OPENCLAW_CACHE_BOUNDARY -->\n";
-
 function createStreamDeps(): {
   deps: AnthropicVertexStreamDeps;
   streamAnthropicMock: ReturnType<typeof vi.fn>;
@@ -50,8 +48,6 @@ function makeModel(params: {
   } as Model<"anthropic-messages">;
 }
 
-const CACHE_BOUNDARY_PROMPT = `Stable prefix${SYSTEM_PROMPT_CACHE_BOUNDARY}Dynamic suffix`;
-
 type PayloadHook = (payload: unknown, payloadModel: unknown) => Promise<unknown>;
 
 function streamAnthropicCall(streamAnthropicMock: ReturnType<typeof vi.fn>): unknown[] {
@@ -72,8 +68,8 @@ function streamTransportOptions(
   return options as Record<string, unknown>;
 }
 
-function captureCacheBoundaryPayloadHook(
-  onPayload: PayloadHook,
+function captureTransportPayloadHook(
+  onPayload: PayloadHook | undefined,
   deps: AnthropicVertexStreamDeps,
   streamAnthropicMock: ReturnType<typeof vi.fn>,
 ) {
@@ -82,41 +78,39 @@ function captureCacheBoundaryPayloadHook(
 
   void streamFn(
     model,
-    {
-      systemPrompt: CACHE_BOUNDARY_PROMPT,
-      messages: [{ role: "user", content: "Hello" }],
-    } as never,
-    {
-      cacheRetention: "short",
-      onPayload,
-    } as never,
+    { messages: [{ role: "user", content: "Hello" }] } as never,
+    { cacheRetention: "short", ...(onPayload ? { onPayload } : {}) } as never,
   );
 
   const transportOptions = streamTransportOptions(streamAnthropicMock);
 
   return { model, onPayload: transportOptions.onPayload as PayloadHook | undefined };
 }
 
-function buildExpectedCacheBoundaryPayload(messageText: string) {
+// Mirrors the shared anthropic-messages transport output: cache boundary already
+// split (uncached dynamic suffix) and all four cache_control markers allocated.
+function buildBudgetedTransportPayload() {
   return {
     system: [
-      {
-        type: "text",
-        text: "Stable prefix",
-        cache_control: { type: "ephemeral" },
-      },
-      {
-        type: "text",
-        text: "Dynamic suffix",
-      },
+      { type: "text", text: "Stable prefix", cache_control: { type: "ephemeral" } },
+      { type: "text", text: "Dynamic suffix" },
+    ],
+    tools: [
+      { name: "exec", input_schema: { type: "object" }, cache_control: { type: "ephemeral" } },
     ],
     messages: [
+      {
+        role: "user",
+        content: [{ type: "text", text: "Hello", cache_control: { type: "ephemeral" } }],
+      },
+      { role: "assistant", content: [{ type: "tool_use", id: "t1", name: "exec", input: {} }] },
       {
         role: "user",
         content: [
           {
-            type: "text",
-            text: messageText,
+            type: "tool_result",
+            tool_use_id: "t1",
+            content: [],
             cache_control: { type: "ephemeral" },
           },
         ],
@@ -125,6 +119,29 @@ function buildExpectedCacheBoundaryPayload(messageText: string) {
   };
 }
 
+function countCacheControlMarkers(payload: unknown): number {
+  let count = 0;
+  const visit = (value: unknown) => {
+    if (Array.isArray(value)) {
+      value.forEach(visit);
+      return;
+    }
+    if (!value || typeof value !== "object") {
+      return;
+    }
+    const record = value as Record<string, unknown>;
+    if (record.cache_control !== undefined) {
+      count += 1;
+    }
+    visit(record.content);
+  };
+  const record = payload as Record<string, unknown>;
+  visit(record.system);
+  visit(record.tools);
+  visit(record.messages);
+  return count;
+}
+
 describe("createAnthropicVertexStreamFn", () => {
   beforeAll(async () => {
     ({ createAnthropicVertexStreamFn, createAnthropicVertexStreamFnForModel } =
@@ -343,63 +360,35 @@ describe("createAnthropicVertexStreamFn", () => {
     expect(transportOptions).not.toHaveProperty("temperature");
   });
 
-  it("applies Anthropic cache-boundary shaping before forwarding payload hooks", async () => {
+  it("keeps already-budgeted cache_control markers intact when forwarding payload hooks", async () => {
     const { deps, streamAnthropicMock } = createStreamDeps();
     const onPayload = vi.fn(async (payload: unknown) => payload);
-    const { model, onPayload: transportPayloadHook } = captureCacheBoundaryPayloadHook(
+    const { model, onPayload: transportPayloadHook } = captureTransportPayloadHook(
       onPayload,
       deps,
       streamAnthropicMock,
     );
-    const payload = {
-      system: [
-        {
-          type: "text",
-          text: CACHE_BOUNDARY_PROMPT,
-          cache_control: { type: "ephemeral" },
-        },
-      ],
-      messages: [{ role: "user", content: "Hello" }],
-    };
+    const payload = buildBudgetedTransportPayload();
 
     const nextPayload = await transportPayloadHook?.(payload, model);
 
-    const expectedPayload = buildExpectedCacheBoundaryPayload("Hello");
-    expect(onPayload).toHaveBeenCalledWith(expectedPayload, model);
-    expect(nextPayload).toEqual(expectedPayload);
+    expect(onPayload).toHaveBeenCalledWith(payload, model);
+    expect(countCacheControlMarkers(nextPayload)).toBe(4);
+    expect((nextPayload as ReturnType<typeof buildBudgetedTransportPayload>).system[1]).toEqual({
+      type: "text",
+      text: "Dynamic suffix",
+    });
   });
 
-  it("reapplies Anthropic cache-boundary shaping when payload hooks return a fresh payload", async () => {
+  it("omits the transport payload hook when the caller provides none", () => {
     const { deps, streamAnthropicMock } = createStreamDeps();
-    const onPayload = vi.fn(async () => ({
-      system: [
-        {
-          type: "text",
-          text: CACHE_BOUNDARY_PROMPT,
-        },
-      ],
-      messages: [{ role: "user", content: "Hello again" }],
-    }));
-    const { model, onPayload: transportPayloadHook } = captureCacheBoundaryPayloadHook(
-      onPayload,
+    const { onPayload: transportPayloadHook } = captureTransportPayloadHook(
+      undefined,
       deps,
       streamAnthropicMock,
     );
 
-    const nextPayload = await transportPayloadHook?.(
-      {
-        system: [
-          {
-            type: "text",
-            text: CACHE_BOUNDARY_PROMPT,
-          },
-        ],
-        messages: [{ role: "user", content: "Hello" }],
-      },
-      model,
-    );
-
-    expect(nextPayload).toEqual(buildExpectedCacheBoundaryPayload("Hello again"));
+    expect(transportPayloadHook).toBeUndefined();
   });
 
   it("omits maxTokens when neither the model nor request provide a finite limit", () => {
diff --git a/extensions/anthropic-vertex/stream-runtime.ts b/extensions/anthropic-vertex/stream-runtime.ts
@@ -1,6 +1,6 @@
 /**
  * Anthropic Vertex stream runtime. It constructs Vertex SDK clients and adapts
- * OpenClaw stream options into Anthropic Messages payload policy.
+ * OpenClaw stream options for the shared Anthropic Messages transport.
  */
 import { AnthropicVertex as AnthropicVertexSdk } from "@anthropic-ai/vertex-sdk";
 import type { StreamFn } from "openclaw/plugin-sdk/agent-core";
@@ -18,10 +18,6 @@ import {
   supportsClaudeNativeMaxEffort,
   supportsClaudeNativeXhighEffort,
 } from "openclaw/plugin-sdk/provider-model-shared";
-import {
-  applyAnthropicPayloadPolicyToParams,
-  resolveAnthropicPayloadPolicy,
-} from "openclaw/plugin-sdk/provider-stream-shared";
 import { resolveAnthropicVertexClientRegion, resolveAnthropicVertexProjectId } from "./region.js";
 
 type AnthropicVertexTransportOptions = ProviderStreamOptions & {
@@ -120,36 +116,6 @@ function resolveAnthropicVertexMaxTokens(params: {
   return requested ?? modelMax;
 }
 
-function createAnthropicVertexOnPayload(params: {
-  model: { api: string; baseUrl?: string; provider: string };
-  cacheRetention: ProviderStreamOptions["cacheRetention"] | undefined;
-  onPayload: ProviderStreamOptions["onPayload"] | undefined;
-}): NonNullable<ProviderStreamOptions["onPayload"]> {
-  const policy = resolveAnthropicPayloadPolicy({
-    provider: params.model.provider,
-    api: params.model.api,
-    baseUrl: params.model.baseUrl,
-    cacheRetention: params.cacheRetention,
-    enableCacheControl: true,
-  });
-
-  function applyPolicy(payload: unknown): unknown {
-    if (payload && typeof payload === "object" && !Array.isArray(payload)) {
-      applyAnthropicPayloadPolicyToParams(payload as Record<string, unknown>, policy);
-    }
-    return payload;
-  }
-
-  return async (payload, model) => {
-    const shapedPayload = applyPolicy(payload);
-    const nextPayload = await params.onPayload?.(shapedPayload, model);
-    if (nextPayload === undefined || nextPayload === shapedPayload) {
-      return shapedPayload;
-    }
-    return applyPolicy(nextPayload);
-  };
-}
-
 /**
  * Create a StreamFn that routes through OpenClaw's generic model stream with an
  * injected `AnthropicVertex` client.  All streaming, message conversion, and
@@ -200,11 +166,10 @@ export function createAnthropicVertexStreamFn(
       cacheRetention: options?.cacheRetention,
       sessionId: options?.sessionId,
       headers: options?.headers,
-      onPayload: createAnthropicVertexOnPayload({
-        model: transportModel,
-        cacheRetention: options?.cacheRetention,
-        onPayload: options?.onPayload,
-      }),
+      // The shared anthropic-messages transport already splits the system prompt
+      // cache boundary and budgets all cache_control markers; re-applying the
+      // payload policy here marked the uncached suffix and breached the 4-marker cap.
+      onPayload: options?.onPayload,
       maxRetryDelayMs: options?.maxRetryDelayMs,
       metadata: options?.metadata,
     };
