feat(agent): fix Anthropic prompt caching with Databricks (+ MCP proxy/TLS passthrough)

crates/buzz-acp/src/acp.rs

@@ -1848,6 +1848,11 @@ impl AcpClient {
                         session_id = %notif.session_id,
                         input = payload.accumulated_input_tokens,
                         output = payload.accumulated_output_tokens,
+                        // A subset of `input`, logged so downstream accounting can
+                        // price it at the provider's cached rate. Always emitted,
+                        // including as 0, so a parser can tell "no cache hits"
+                        // apart from "this build predates the field".
+                        cached = payload.accumulated_cached_input_tokens,
                         "goose usage update"
                     );
                     self.goose_usage.record(&notif.session_id, payload);

crates/buzz-acp/src/usage.rs

@@ -85,6 +85,12 @@ pub(crate) struct UsageUpdatePayload {
     pub context_limit: u64,
     pub accumulated_input_tokens: u64,
     pub accumulated_output_tokens: u64,
+    /// The cache-served subset of `accumulated_input_tokens`. Optional — goose
+    /// does not send it, and buzz-agent only reports a non-zero value when the
+    /// provider returned a cache split, so `0` legitimately means either "no
+    /// cache hits" or "provider reported none".
+    #[serde(default)]
+    pub accumulated_cached_input_tokens: u64,
     pub accumulated_cost: Option<f64>,
     /// Effective model id for this turn. Optional — goose payloads that
     /// predate this field deserialize cleanly as `None`.
@@ -323,12 +329,44 @@ impl UsageTracker {
 mod tests {
     use super::*;
 
+    /// The camelCase key buzz-agent actually puts on the wire must land on the
+    /// field. A rename mismatch here would deserialize to the serde default of
+    /// 0, and every trial would price as if nothing had ever been cached — the
+    /// exact silent failure this field was added to remove.
+    #[test]
+    fn cached_input_tokens_deserialize_from_the_wire_key() {
+        let p: UsageUpdatePayload = serde_json::from_value(serde_json::json!({
+            "used": 15_247,
+            "contextLimit": 0,
+            "accumulatedInputTokens": 15_091,
+            "accumulatedOutputTokens": 156,
+            "accumulatedCachedInputTokens": 5_033,
+        }))
+        .expect("payload must deserialize");
+        assert_eq!(p.accumulated_cached_input_tokens, 5_033);
+        assert!(p.accumulated_cached_input_tokens <= p.accumulated_input_tokens);
+    }
+
+    /// goose does not send the field; its payloads must still deserialize.
+    #[test]
+    fn a_payload_without_the_cache_field_defaults_to_zero() {
+        let p: UsageUpdatePayload = serde_json::from_value(serde_json::json!({
+            "used": 500,
+            "contextLimit": 200_000,
+            "accumulatedInputTokens": 400,
+            "accumulatedOutputTokens": 100,
+        }))
+        .expect("payload must deserialize without the cache field");
+        assert_eq!(p.accumulated_cached_input_tokens, 0);
+    }
+
     fn payload(input: u64, output: u64, cost: Option<f64>) -> UsageUpdatePayload {
         UsageUpdatePayload {
             used: input + output,
             context_limit: 200_000,
             accumulated_input_tokens: input,
             accumulated_output_tokens: output,
+            accumulated_cached_input_tokens: 0,
             accumulated_cost: cost,
             model: None,
         }
@@ -340,6 +378,7 @@ mod tests {
             context_limit: 0,
             accumulated_input_tokens: input,
             accumulated_output_tokens: output,
+            accumulated_cached_input_tokens: 0,
             accumulated_cost: cost,
             model: None,
         }
@@ -836,6 +875,7 @@ mod tests {
             context_limit: 200_000,
             accumulated_input_tokens: input,
             accumulated_output_tokens: output,
+            accumulated_cached_input_tokens: 0,
             accumulated_cost: cost,
             model: model.map(str::to_string),
         }

crates/buzz-agent/src/agent.rs

@@ -60,6 +60,11 @@ pub struct RunCtx<'a> {
     /// Accumulated output tokens across all LLM rounds in this turn, for
     /// NIP-AM metric publishing. Reset to `None` at turn start in `run()`.
     pub turn_output_tokens: &'a mut Option<u64>,
+    /// The cache-served subset of `turn_input_tokens`, accumulated across all
+    /// LLM rounds in this turn. Reset to `None` at turn start in `run()`.
+    /// Consumers price this slice at the provider's cached rate; without it
+    /// every round of a growing conversation is billed at full price.
+    pub turn_cached_input_tokens: &'a mut Option<u64>,
 }
 
 impl RunCtx<'_> {
@@ -78,6 +83,7 @@ impl RunCtx<'_> {
         // Reset per-turn token accumulators for this prompt.
         *self.turn_input_tokens = None;
         *self.turn_output_tokens = None;
+        *self.turn_cached_input_tokens = None;
 
         let mut round = 0u32;
         // Per-prompt `_Stop` objection count. Bounded per prompt (not per
@@ -175,6 +181,17 @@ impl RunCtx<'_> {
                 *self.turn_output_tokens =
                     Some(self.turn_output_tokens.unwrap_or(0).saturating_add(out));
             }
+            // Accumulate the cache-served subset of this turn's input. Tracked
+            // separately from `turn_input_tokens` rather than subtracted from
+            // it: the input total must stay inclusive for the handoff gate,
+            // which cares how much context was sent, not what it cost.
+            if let Some(cached) = response.cached_input_tokens {
+                *self.turn_cached_input_tokens = Some(
+                    self.turn_cached_input_tokens
+                        .unwrap_or(0)
+                        .saturating_add(cached),
+                );
+            }
 
             if !response.reasoning.is_empty() {
                 wire::send(

crates/buzz-agent/src/config.rs

@@ -738,6 +738,13 @@ pub struct Config {
     /// Thinking/reasoning effort level. `None` = use provider default (no
     /// thinking config sent). Set via `BUZZ_AGENT_THINKING_EFFORT`.
     pub thinking_effort: Option<ThinkingEffort>,
+    /// Emit Anthropic `cache_control` breakpoints on the stable prefix
+    /// (tools + system prompt) and the rolling conversation tail. Default on;
+    /// disable with `BUZZ_AGENT_PROMPT_CACHING=0`. Only consulted on Anthropic
+    /// Messages routes (first-party Anthropic and the DatabricksV2 Claude
+    /// route) — the Databricks gateway does not auto-cache, so without this the
+    /// surfaced `cache_read_input_tokens` is structurally always 0.
+    pub prompt_caching: bool,
 }
 
 impl Config {
@@ -833,6 +840,7 @@ impl Config {
             hook_servers: parse_hook_servers_env("MCP_HOOK_SERVERS"),
             hints_enabled: parse_env("BUZZ_AGENT_NO_HINTS", 0u8)? == 0,
             thinking_effort: parse_thinking_effort(env("BUZZ_AGENT_THINKING_EFFORT").as_deref())?,
+            prompt_caching: parse_env("BUZZ_AGENT_PROMPT_CACHING", 1u8)? != 0,
         };
         cfg.validate()?;
         Ok(cfg)
@@ -874,6 +882,7 @@ impl Config {
             hook_servers: HookServers::None,
             hints_enabled: false,
             thinking_effort: None,
+            prompt_caching: false,
         }
     }
 

crates/buzz-agent/src/lib.rs

@@ -100,6 +100,11 @@ struct Session {
     accumulated_input_tokens: u64,
     /// Session-cumulative output tokens across all turns.
     accumulated_output_tokens: u64,
+    /// Session-cumulative cache-served input tokens across all turns — a subset
+    /// of `accumulated_input_tokens`, not an addition to it. Emitted alongside
+    /// it so a consumer can price the cached slice at the provider's discounted
+    /// rate instead of assuming every input token cost full price.
+    accumulated_cached_input_tokens: u64,
 }
 
 fn die(msg: String) -> ! {
@@ -426,6 +431,7 @@ async fn session_new(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSen
             effective_model: None,
             accumulated_input_tokens: 0,
             accumulated_output_tokens: 0,
+            accumulated_cached_input_tokens: 0,
         },
     );
     drop(sessions);
@@ -672,6 +678,7 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
         .unwrap_or(&app.cfg.model);
     let mut turn_input_tokens: Option<u64> = None;
     let mut turn_output_tokens: Option<u64> = None;
+    let mut turn_cached_input_tokens: Option<u64> = None;
     let mut ctx = RunCtx {
         cfg: &app.cfg,
         effective_model: effective_model_str,
@@ -690,6 +697,7 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
         last_request_history_bytes: &mut last_request_history_bytes,
         turn_input_tokens: &mut turn_input_tokens,
         turn_output_tokens: &mut turn_output_tokens,
+        turn_cached_input_tokens: &mut turn_cached_input_tokens,
     };
     let result = ctx.run(p.prompt).await;
     if let Some(s) = app.sessions.lock().await.get_mut(&sid) {
@@ -722,14 +730,21 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
                 s.accumulated_output_tokens = s
                     .accumulated_output_tokens
                     .saturating_add(turn_output_tokens.unwrap_or(0));
-                Some((s.accumulated_input_tokens, s.accumulated_output_tokens))
+                s.accumulated_cached_input_tokens = s
+                    .accumulated_cached_input_tokens
+                    .saturating_add(turn_cached_input_tokens.unwrap_or(0));
+                Some((
+                    s.accumulated_input_tokens,
+                    s.accumulated_output_tokens,
+                    s.accumulated_cached_input_tokens,
+                ))
             } else {
                 // Session is gone — the accumulated baseline no longer exists, so
                 // there is nothing correct to emit. Skip the usage notification.
                 None
             }
         };
-        if let Some((accumulated_in, accumulated_out)) = accumulated {
+        if let Some((accumulated_in, accumulated_out, accumulated_cached)) = accumulated {
             wire::send(
                 &wire_tx,
                 goose_session_update(
@@ -742,6 +757,11 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
                         "contextLimit": 0u64,
                         "accumulatedInputTokens": accumulated_in,
                         "accumulatedOutputTokens": accumulated_out,
+                        // A subset of accumulatedInputTokens, not an addition to
+                        // it. Extends goose's usage_update shape; a consumer that
+                        // does not know the field ignores it and prices exactly as
+                        // it did before.
+                        "accumulatedCachedInputTokens": accumulated_cached,
                         "model": effective_model_str,
                     }),
                 ),