feat(llm): cache hint TTL, breakpoint cap, and tool placement (#26779)

2026-05-10 21:17:38 -04:00
parent fed716ada5
commit 77e6c0d329
12 changed files with 555 additions and 39 deletions
--- a/packages/llm/src/protocols/anthropic-messages.ts
+++ b/packages/llm/src/protocols/anthropic-messages.ts
@@ -16,6 +16,7 @@ import {
  type ToolResultPart,
 } from "../schema"
 import { JsonObject, optionalArray, optionalNull, ProviderShared } from "./shared"
+import * as Cache from "./utils/cache"
 import { ToolStream } from "./utils/tool-stream"

 const ADAPTER = "anthropic-messages"
@@ -25,7 +26,10 @@ export const PATH = "/messages"
 // =============================================================================
 // Request Body Schema
 // =============================================================================
-const AnthropicCacheControl = Schema.Struct({ type: Schema.tag("ephemeral") })
+const AnthropicCacheControl = Schema.Struct({
+  type: Schema.tag("ephemeral"),
+  ttl: Schema.optional(Schema.Literals(["5m", "1h"])),
+})

 const AnthropicTextBlock = Schema.Struct({
  type: Schema.tag("text"),
@@ -193,8 +197,24 @@ const invalid = ProviderShared.invalidRequest
 // =============================================================================
 // Request Lowering
 // =============================================================================
-const cacheControl = (cache: CacheHint | undefined) =>
-  cache?.type === "ephemeral" ? { type: "ephemeral" as const } : undefined
+// Anthropic accepts at most 4 explicit cache_control breakpoints per request,
+// across `tools`, `system`, and `messages`. Beyond the cap the API returns a
+// 400 — so the lowering layer counts emitted markers and silently drops any
+// that exceed it.
+const ANTHROPIC_BREAKPOINT_CAP = 4
+
+const EPHEMERAL_5M = { type: "ephemeral" as const }
+const EPHEMERAL_1H = { type: "ephemeral" as const, ttl: "1h" as const }
+
+const cacheControl = (breakpoints: Cache.Breakpoints, cache: CacheHint | undefined) => {
+  if (cache?.type !== "ephemeral" && cache?.type !== "persistent") return undefined
+  if (breakpoints.remaining <= 0) {
+    breakpoints.dropped += 1
+    return undefined
+  }
+  breakpoints.remaining -= 1
+  return Cache.ttlBucket(cache.ttlSeconds) === "1h" ? EPHEMERAL_1H : EPHEMERAL_5M
+}

 const anthropicMetadata = (metadata: Record<string, unknown>): ProviderMetadata => ({ anthropic: metadata })

@@ -204,10 +224,11 @@ const signatureFromMetadata = (metadata: ProviderMetadata | undefined): string |
  return typeof anthropic.signature === "string" ? anthropic.signature : undefined
 }

-const lowerTool = (tool: ToolDefinition): AnthropicTool => ({
+const lowerTool = (breakpoints: Cache.Breakpoints, tool: ToolDefinition): AnthropicTool => ({
  name: tool.name,
  description: tool.description,
  input_schema: tool.inputSchema,
+  cache_control: cacheControl(breakpoints, tool.cache),
 })

 const lowerToolChoice = (toolChoice: NonNullable<LLMRequest["toolChoice"]>) =>
@@ -249,7 +270,10 @@ const lowerServerToolResult = Effect.fn("AnthropicMessages.lowerServerToolResult
  return { type: wireType, tool_use_id: part.id, content: part.result.value } satisfies AnthropicServerToolResultBlock
 })

-const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (request: LLMRequest) {
+const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (
+  request: LLMRequest,
+  breakpoints: Cache.Breakpoints,
+) {
  const messages: AnthropicMessage[] = []

  for (const message of request.messages) {
@@ -258,7 +282,7 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (re
      for (const part of message.content) {
        if (!ProviderShared.supportsContent(part, ["text"]))
          return yield* ProviderShared.unsupportedContent("Anthropic Messages", "user", ["text"])
-        content.push({ type: "text", text: part.text, cache_control: cacheControl(part.cache) })
+        content.push({ type: "text", text: part.text, cache_control: cacheControl(breakpoints, part.cache) })
      }
      messages.push({ role: "user", content })
      continue
@@ -268,7 +292,7 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (re
      const content: AnthropicAssistantBlock[] = []
      for (const part of message.content) {
        if (part.type === "text") {
-          content.push({ type: "text", text: part.text, cache_control: cacheControl(part.cache) })
+          content.push({ type: "text", text: part.text, cache_control: cacheControl(breakpoints, part.cache) })
          continue
        }
        if (part.type === "reasoning") {
@@ -304,6 +328,7 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (re
        tool_use_id: part.id,
        content: ProviderShared.toolResultText(part),
        is_error: part.result.type === "error" ? true : undefined,
+        cache_control: cacheControl(breakpoints, part.cache),
      })
    }
    messages.push({ role: "user", content })
@@ -330,18 +355,33 @@ const lowerThinking = Effect.fn("AnthropicMessages.lowerThinking")(function* (re
 const fromRequest = Effect.fn("AnthropicMessages.fromRequest")(function* (request: LLMRequest) {
  const toolChoice = request.toolChoice ? yield* lowerToolChoice(request.toolChoice) : undefined
  const generation = request.generation
+  // Allocate the 4-breakpoint budget in invalidation order: tools → system →
+  // messages. Tools live highest in the cache hierarchy, so when callers
+  // over-mark we keep their tool hints and shed the message-tail ones first.
+  const breakpoints = Cache.newBreakpoints(ANTHROPIC_BREAKPOINT_CAP)
+  const tools =
+    request.tools.length === 0 || request.toolChoice?.type === "none"
+      ? undefined
+      : request.tools.map((tool) => lowerTool(breakpoints, tool))
+  const system =
+    request.system.length === 0
+      ? undefined
+      : request.system.map((part) => ({
+          type: "text" as const,
+          text: part.text,
+          cache_control: cacheControl(breakpoints, part.cache),
+        }))
+  const messages = yield* lowerMessages(request, breakpoints)
+  if (breakpoints.dropped > 0) {
+    yield* Effect.logWarning(
+      `Anthropic Messages: dropped ${breakpoints.dropped} cache breakpoint(s); the API allows at most ${ANTHROPIC_BREAKPOINT_CAP} per request.`,
+    )
+  }
  return {
    model: request.model.id,
-    system:
-      request.system.length === 0
-        ? undefined
-        : request.system.map((part) => ({
-            type: "text" as const,
-            text: part.text,
-            cache_control: cacheControl(part.cache),
-          })),
-    messages: yield* lowerMessages(request),
-    tools: request.tools.length === 0 || request.toolChoice?.type === "none" ? undefined : request.tools.map(lowerTool),
+    system,
+    messages,
+    tools,
    tool_choice: toolChoice,
    stream: true as const,
    max_tokens: generation?.maxTokens ?? request.model.limits.output ?? 4096,
--- a/packages/llm/src/protocols/bedrock-converse.ts
+++ b/packages/llm/src/protocols/bedrock-converse.ts
@@ -108,7 +108,7 @@ type BedrockMessage = Schema.Schema.Type<typeof BedrockMessage>
 const BedrockSystemBlock = Schema.Union([BedrockTextBlock, BedrockCache.CachePointBlock])
 type BedrockSystemBlock = Schema.Schema.Type<typeof BedrockSystemBlock>

-const BedrockTool = Schema.Struct({
+const BedrockToolSpec = Schema.Struct({
  toolSpec: Schema.Struct({
    name: Schema.String,
    description: Schema.String,
@@ -117,6 +117,9 @@ const BedrockTool = Schema.Struct({
    }),
  }),
 })
+type BedrockToolSpec = Schema.Schema.Type<typeof BedrockToolSpec>
+
+const BedrockTool = Schema.Union([BedrockToolSpec, BedrockCache.CachePointBlock])
 type BedrockTool = Schema.Schema.Type<typeof BedrockTool>

 const BedrockToolChoice = Schema.Union([
@@ -214,7 +217,7 @@ type BedrockEvent = Schema.Schema.Type<typeof BedrockEvent>
 // =============================================================================
 // Request Lowering
 // =============================================================================
-const lowerTool = (tool: ToolDefinition): BedrockTool => ({
+const lowerToolSpec = (tool: ToolDefinition): BedrockToolSpec => ({
  toolSpec: {
    name: tool.name,
    description: tool.description,
@@ -222,11 +225,25 @@ const lowerTool = (tool: ToolDefinition): BedrockTool => ({
  },
 })

+const lowerTools = (
+  breakpoints: BedrockCache.Breakpoints,
+  tools: ReadonlyArray<ToolDefinition>,
+): BedrockTool[] => {
+  const result: BedrockTool[] = []
+  for (const tool of tools) {
+    result.push(lowerToolSpec(tool))
+    const cachePoint = BedrockCache.block(breakpoints, tool.cache)
+    if (cachePoint) result.push(cachePoint)
+  }
+  return result
+}
+
 const textWithCache = (
+  breakpoints: BedrockCache.Breakpoints,
  text: string,
  cache: CacheHint | undefined,
 ): Array<BedrockTextBlock | BedrockCache.CachePointBlock> => {
-  const cachePoint = BedrockCache.block(cache)
+  const cachePoint = BedrockCache.block(breakpoints, cache)
  return cachePoint ? [{ text }, cachePoint] : [{ text }]
 }

@@ -257,7 +274,10 @@ const lowerToolResult = (part: ToolResultPart): BedrockToolResultBlock => ({
  },
 })

-const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (request: LLMRequest) {
+const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (
+  request: LLMRequest,
+  breakpoints: BedrockCache.Breakpoints,
+) {
  const messages: BedrockMessage[] = []

  for (const message of request.messages) {
@@ -267,7 +287,7 @@ const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (requ
        if (!ProviderShared.supportsContent(part, ["text", "media"]))
          return yield* ProviderShared.unsupportedContent("Bedrock Converse", "user", ["text", "media"])
        if (part.type === "text") {
-          content.push(...textWithCache(part.text, part.cache))
+          content.push(...textWithCache(breakpoints, part.text, part.cache))
          continue
        }
        if (part.type === "media") {
@@ -289,7 +309,7 @@ const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (requ
            "tool-call",
          ])
        if (part.type === "text") {
-          content.push(...textWithCache(part.text, part.cache))
+          content.push(...textWithCache(breakpoints, part.text, part.cache))
          continue
        }
        if (part.type === "reasoning") {
@@ -309,11 +329,13 @@ const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (requ
      continue
    }

-    const content: BedrockToolResultBlock[] = []
+    const content: BedrockUserBlock[] = []
    for (const part of message.content) {
      if (!ProviderShared.supportsContent(part, ["tool-result"]))
        return yield* ProviderShared.unsupportedContent("Bedrock Converse", "tool", ["tool-result"])
      content.push(lowerToolResult(part))
+      const cachePoint = BedrockCache.block(breakpoints, part.cache)
+      if (cachePoint) content.push(cachePoint)
    }
    messages.push({ role: "user", content })
  }
@@ -323,16 +345,32 @@ const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (requ

 // System prompts share the cache-point convention: emit the text block, then
 // optionally a positional `cachePoint` marker.
-const lowerSystem = (system: ReadonlyArray<LLMRequest["system"][number]>): BedrockSystemBlock[] =>
-  system.flatMap((part) => textWithCache(part.text, part.cache))
+const lowerSystem = (
+  breakpoints: BedrockCache.Breakpoints,
+  system: ReadonlyArray<LLMRequest["system"][number]>,
+): BedrockSystemBlock[] => system.flatMap((part) => textWithCache(breakpoints, part.text, part.cache))

 const fromRequest = Effect.fn("BedrockConverse.fromRequest")(function* (request: LLMRequest) {
  const toolChoice = request.toolChoice ? yield* lowerToolChoice(request.toolChoice) : undefined
  const generation = request.generation
+  // Bedrock-Claude shares Anthropic's 4-breakpoint cap. Spend the budget in
+  // tools → system → messages order to favour the highest-impact prefixes.
+  const breakpoints = BedrockCache.breakpoints()
+  const toolConfig =
+    request.tools.length > 0 && request.toolChoice?.type !== "none"
+      ? { tools: lowerTools(breakpoints, request.tools), toolChoice }
+      : undefined
+  const system = request.system.length === 0 ? undefined : lowerSystem(breakpoints, request.system)
+  const messages = yield* lowerMessages(request, breakpoints)
+  if (breakpoints.dropped > 0) {
+    yield* Effect.logWarning(
+      `Bedrock Converse: dropped ${breakpoints.dropped} cache breakpoint(s); the API allows at most ${BedrockCache.BEDROCK_BREAKPOINT_CAP} per request.`,
+    )
+  }
  return {
    modelId: request.model.id,
-    messages: yield* lowerMessages(request),
-    system: request.system.length === 0 ? undefined : lowerSystem(request.system),
+    messages,
+    system,
    inferenceConfig:
      generation?.maxTokens === undefined &&
      generation?.temperature === undefined &&
@@ -345,10 +383,7 @@ const fromRequest = Effect.fn("BedrockConverse.fromRequest")(function* (request:
            topP: generation?.topP,
            stopSequences: generation?.stop,
          },
-    toolConfig:
-      request.tools.length > 0 && request.toolChoice?.type !== "none"
-        ? { tools: request.tools.map(lowerTool), toolChoice }
-        : undefined,
+    toolConfig,
  }
 })

--- a/packages/llm/src/protocols/utils/bedrock-cache.ts
+++ b/packages/llm/src/protocols/utils/bedrock-cache.ts
@@ -1,20 +1,37 @@
 import { Schema } from "effect"
 import type { CacheHint } from "../../schema"
+import { newBreakpoints, ttlBucket, type Breakpoints } from "./cache"

 // Bedrock cache markers are positional: emit a `cachePoint` block immediately
-// after the content the caller wants treated as a cacheable prefix.
+// after the content the caller wants treated as a cacheable prefix. Bedrock
+// accepts optional `ttl: "5m" | "1h"` on cachePoint, mirroring Anthropic.
 export const CachePointBlock = Schema.Struct({
-  cachePoint: Schema.Struct({ type: Schema.tag("default") }),
+  cachePoint: Schema.Struct({
+    type: Schema.tag("default"),
+    ttl: Schema.optional(Schema.Literals(["5m", "1h"])),
+  }),
 })
 export type CachePointBlock = Schema.Schema.Type<typeof CachePointBlock>

-// Bedrock recently added optional `ttl: "5m" | "1h"` on cachePoint. Map
-// `CacheHint.ttlSeconds` here once a recorded cassette validates the wire shape.
-const DEFAULT: CachePointBlock = { cachePoint: { type: "default" } }
+// Bedrock-Claude enforces the same 4-breakpoint cap as the Anthropic Messages
+// API. Callers pass a shared counter through every `block()` call site so the
+// budget is respected across `system`, `messages`, and `tools`.
+export const BEDROCK_BREAKPOINT_CAP = 4

-export const block = (cache: CacheHint | undefined): CachePointBlock | undefined => {
+export type { Breakpoints } from "./cache"
+export const breakpoints = () => newBreakpoints(BEDROCK_BREAKPOINT_CAP)
+
+const DEFAULT_5M: CachePointBlock = { cachePoint: { type: "default" } }
+const DEFAULT_1H: CachePointBlock = { cachePoint: { type: "default", ttl: "1h" } }
+
+export const block = (breakpoints: Breakpoints, cache: CacheHint | undefined): CachePointBlock | undefined => {
  if (cache?.type !== "ephemeral" && cache?.type !== "persistent") return undefined
-  return DEFAULT
+  if (breakpoints.remaining <= 0) {
+    breakpoints.dropped += 1
+    return undefined
+  }
+  breakpoints.remaining -= 1
+  return ttlBucket(cache.ttlSeconds) === "1h" ? DEFAULT_1H : DEFAULT_5M
 }

 export * as BedrockCache from "./bedrock-cache"
--- a/packages/llm/src/protocols/utils/cache.ts
+++ b/packages/llm/src/protocols/utils/cache.ts
@@ -0,0 +1,16 @@
+// Shared helpers for provider cache-marker lowering. Anthropic and Bedrock
+// both enforce a 4-breakpoint cap per request and accept the same `5m`/`1h`
+// TTL buckets, so the counter and TTL mapping live here.
+
+export interface Breakpoints {
+  remaining: number
+  dropped: number
+}
+
+export const newBreakpoints = (cap: number): Breakpoints => ({ remaining: cap, dropped: 0 })
+
+// Returns `"1h"` for any `ttlSeconds >= 3600`, otherwise `undefined` (the
+// provider default 5m). Anthropic & Bedrock both treat anything shorter than
+// an hour as 5m.
+export const ttlBucket = (ttlSeconds: number | undefined): "1h" | undefined =>
+  ttlSeconds !== undefined && ttlSeconds >= 3600 ? "1h" : undefined
--- a/packages/llm/src/schema/messages.ts
+++ b/packages/llm/src/schema/messages.ts
@@ -79,6 +79,7 @@ export const ToolResultPart = Object.assign(
    name: Schema.String,
    result: ToolResultValue,
    providerExecuted: Schema.optional(Schema.Boolean),
+    cache: Schema.optional(CacheHint),
    metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
    providerMetadata: Schema.optional(ProviderMetadata),
  }).annotate({ identifier: "LLM.Content.ToolResult" }),
@@ -94,6 +95,7 @@ export const ToolResultPart = Object.assign(
      name: input.name,
      result: ToolResultValue.make(input.result, input.resultType),
      providerExecuted: input.providerExecuted,
+      cache: input.cache,
      metadata: input.metadata,
      providerMetadata: input.providerMetadata,
    }),
@@ -151,6 +153,7 @@ export class ToolDefinition extends Schema.Class<ToolDefinition>("LLM.ToolDefini
  name: Schema.String,
  description: Schema.String,
  inputSchema: JsonSchema,
+  cache: Schema.optional(CacheHint),
  metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
  native: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
 }) {}