feat(llm): add xAI Build affinity and smarter prompt-cache compaction

_docs/prompt-caching.mdx

@@ -30,14 +30,38 @@ What crabcode does today for each routing path:
 
 ## What gets marked (direct Anthropic)
 
-Aligned with OpenCode's `auto` policy:
+Hybrid of OpenCode `auto` + Grok Build placement:
 
 1. **Last tool** — tool schemas are large and stable across a tool loop  
 2. **Last system block** — instructions / project context  
-3. **Latest user content block** — conversation prefix through the current turn (including tool-result groups)
+3. **Transcript tip** — last markable content block (skips `thinking` / `redacted_thinking`)  
+4. **Previous user tip** — where the prior request ended, so turns past the 20-block lookback still hit cache  
+
+At most **4** breakpoints; the 4th slot stays free when tools or previous-user are missing so gateways can auto-mark.
 
 Breakpoints are applied **after** message regrouping so adjacent tool_use / tool_result blocks stay valid.
 
+### xAI Grok Build (cli-chat-proxy)
+
+In addition to sticky `prompt_cache_key = session_id`, crabcode stamps Grok Build–style affinity headers:
+
+| Header | Value |
+| --- | --- |
+| `x-grok-session-id` | Session id (sticky) |
+| `x-grok-conv-id` | Same as session for main turns (sticky) |
+| `x-grok-req-id` | Unique per model invocation |
+| `x-grok-turn-idx` | 0-based user-turn index |
+| `x-grok-agent-id` | Process-stable agent id |
+
+**Parent-cached aux** (e.g. max-steps text-only summary): keeps parent `prompt_cache_key` + session/conv so the conversation prefix can reuse the main turn's KV cache; assigns a fresh `aux-…` req id.
+
+**Subagents** use the **child** session id on purpose. They have a different system prompt and tool set, so parent-prefix reuse would miss and can pollute sticky routing. Isolation beats a false shared key.
+
+### Prefix stability (anti-bust)
+
+- **Empty system/user rows** are dropped before the wire (they pad the sticky prefix).  
+- **Images** use hysteresis: compact only above a trigger (~6 MiB total data-urls), then reclaim to a lower target (~3 MiB) so later turns stay cache-warm instead of re-evicting every step.
+
 ---
 
 ## How to verify it works
@@ -53,21 +77,27 @@ Logging writes to `app.log` in the working directory when `--emit-logs` is set.
 ### Direct Anthropic
 
 ```
-[prompt-cache] anthropic input=… output=… cache_read=… cache_creation=…
+[prompt-cache] anthropic input=… output=… cache_read=… cache_creation=… total_input=… hit_pct=…
 ```
 
 ### AI Gateway / OpenAI-compatible
 
 ```
-[prompt-cache] openai-compatible prompt=… completion=… cached_tokens=… cache_read=… cache_creation=…
+[prompt-cache] openai-compatible prompt=… completion=… cached_tokens=… cache_read=… cache_creation=… hit_pct=…
+```
+
+### OpenAI / xAI Responses
+
+```
+[prompt-cache] openai-responses input=… output=… cached_tokens=… hit_pct=…
 ```
 
 ### Healthy multi-step session
 
 | Step | What you want to see |
 | --- | --- |
-| First request | `cache_creation` / write > 0, or full input billed once |
-| Later tool steps (same tools + system + prefix) | `cache_read` / `cached_tokens` > 0 and growing |
+| First request | `cache_creation` / write > 0, or full input billed once; `hit_pct` near 0 |
+| Later tool steps (same tools + system + prefix) | `cache_read` / `cached_tokens` > 0 and `hit_pct` climbing (often 70%+) |
 
 Notes:
 

src/agent/subagent.rs

@@ -47,7 +47,24 @@ pub async fn run_subagent(
     use futures::StreamExt;
     use std::collections::HashMap;
 
-    let session = resolve_subagent_session(&agent, parent_session, sender.as_ref()).await?;
+    let mut session = resolve_subagent_session(&agent, parent_session, sender.as_ref()).await?;
+    // Child cache key on purpose: subagents have a different system prompt and tool set,
+    // so parent-prefix reuse would miss (and risk sticky-routing pollution). See
+    // SessionAffinity::child_session docs.
+    session.prompt_cache_key = Some(session_id.clone());
+    session.openai_options.prompt_cache_key = Some(session_id.clone());
+    if crate::llm::xai_build::is_build_transport(&session.openai_options.additional_headers) {
+        let affinity = crate::llm::xai_build::SessionAffinity::child_session(&session_id);
+        crate::llm::xai_build::inject_session_affinity_headers(
+            &mut session.openai_options.additional_headers,
+            &affinity,
+        );
+        crate::emit_log!(
+            "[prompt-cache] xai-build affinity kind=child session_id={} req_id={}",
+            affinity.session_id,
+            affinity.req_id
+        );
+    }
 
     let scoped_registry = build_scoped_registry(full_registry, &agent).await;
 

src/aisdk/providers/anthropic.rs

@@ -269,12 +269,24 @@ fn log_anthropic_usage(usage: &serde_json::Value) {
         return;
     }
 
+    let input_v = input.unwrap_or(0);
+    let total_input = input_v
+        .saturating_add(cache_read)
+        .saturating_add(cache_creation);
+    let hit_pct = if total_input > 0 {
+        (cache_read as f64 * 100.0) / total_input as f64
+    } else {
+        0.0
+    };
+
     crate::emit_log!(
-        "[prompt-cache] anthropic input={} output={} cache_read={} cache_creation={}",
+        "[prompt-cache] anthropic input={} output={} cache_read={} cache_creation={} total_input={} hit_pct={:.1}",
         input.map(|v| v.to_string()).unwrap_or_else(|| "-".into()),
         output.map(|v| v.to_string()).unwrap_or_else(|| "-".into()),
         cache_read,
-        cache_creation
+        cache_creation,
+        total_input,
+        hit_pct
     );
 }
 
@@ -407,9 +419,13 @@ fn anthropic_tool_input_is_empty(value: &serde_json::Value) -> bool {
     }
 }
 
-/// Anthropic prompt caching (opencode `auto` parity):
-/// mark last tool + last system block + latest user content block.
-/// Cap at 4 breakpoints; each is `{"type":"ephemeral"}`.
+/// Anthropic prompt caching (Grok Build / OpenCode hybrid):
+/// 1. last tool (stable schemas — high value in tool loops)
+/// 2. last system block
+/// 3. tip of transcript (last markable block; skips thinking)
+/// 4. previous user tip (covers turns past the 20-block lookback)
+/// Cap at 4 breakpoints; each is `{"type":"ephemeral"}`. The 4th slot stays
+/// free when tools or previous-user are missing so gateways can still auto-mark.
 fn apply_anthropic_prompt_caching(body: &mut serde_json::Value) {
     let mut remaining = 4usize;
 
@@ -439,27 +455,48 @@ fn apply_anthropic_prompt_caching(body: &mut serde_json::Value) {
         }
     }
 
-    // 3. Latest user message's last content block (after anthropic_messages
-    // regrouping so tool_result groups keep the marker).
+    // 3–4. Transcript tip + previous user tip (skip thinking blocks).
+    if remaining == 0 {
+        return;
+    }
+    let Some(messages) = body.get_mut("messages").and_then(|v| v.as_array_mut()) else {
+        return;
+    };
+
+    let tip = (0..messages.len())
+        .rev()
+        .find(|&i| mark_message_cache_breakpoint(&mut messages[i]));
+    if tip.is_some() {
+        remaining = remaining.saturating_sub(1);
+    }
+
+    // Where the previous request ended: skip the whole trailing user run after
+    // the last assistant, then mark that earlier user tip (Grok Build placement).
     if remaining > 0 {
-        if let Some(messages) = body.get_mut("messages").and_then(|v| v.as_array_mut()) {
-            if let Some(user) = messages
-                .iter_mut()
-                .rev()
-                .find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user"))
+        if let Some(tip) = tip {
+            if let Some(prev) = messages[..tip]
+                .iter()
+                .rposition(|m| m.get("role").and_then(|r| r.as_str()) == Some("assistant"))
+                .and_then(|assistant| {
+                    messages[..assistant]
+                        .iter()
+                        .rposition(|m| m.get("role").and_then(|r| r.as_str()) == Some("user"))
+                })
             {
-                mark_latest_user_cache_control(user);
+                let _ = mark_message_cache_breakpoint(&mut messages[prev]);
             }
         }
     }
 }
 
-fn mark_latest_user_cache_control(user: &mut serde_json::Value) {
-    let Some(obj) = user.as_object_mut() else {
-        return;
+/// Marks the last content block that can carry a breakpoint, scanning back past
+/// `thinking` / `redacted_thinking` which the API rejects.
+fn mark_message_cache_breakpoint(message: &mut serde_json::Value) -> bool {
+    let Some(obj) = message.as_object_mut() else {
+        return false;
     };
 
-    // String content must become a text block to host cache_control.
+    // Plain string content must become a text block to host cache_control.
     if let Some(serde_json::Value::String(text)) = obj.get("content").cloned() {
         obj.insert(
             "content".to_string(),
@@ -469,17 +506,28 @@ fn mark_latest_user_cache_control(user: &mut serde_json::Value) {
                 "cache_control": { "type": "ephemeral" }
             }]),
         );
-        return;
+        return true;
     }
 
-    if let Some(blocks) = obj.get_mut("content").and_then(|c| c.as_array_mut()) {
-        if let Some(last) = blocks.last_mut().and_then(|b| b.as_object_mut()) {
-            last.insert(
-                "cache_control".to_string(),
-                serde_json::json!({ "type": "ephemeral" }),
-            );
+    let Some(blocks) = obj.get_mut("content").and_then(|c| c.as_array_mut()) else {
+        return false;
+    };
+
+    for block in blocks.iter_mut().rev() {
+        let Some(block_obj) = block.as_object_mut() else {
+            continue;
+        };
+        let block_type = block_obj.get("type").and_then(|t| t.as_str()).unwrap_or("");
+        if block_type == "thinking" || block_type == "redacted_thinking" {
+            continue;
         }
+        block_obj.insert(
+            "cache_control".to_string(),
+            serde_json::json!({ "type": "ephemeral" }),
+        );
+        return true;
     }
+    false
 }
 
 fn anthropic_user_content(user: &crate::message::UserMessage) -> serde_json::Value {
@@ -738,7 +786,7 @@ mod tests {
     }
 
     #[test]
-    fn prompt_caching_marks_last_tool_system_and_latest_user() {
+    fn prompt_caching_marks_last_tool_system_tip_and_previous_user() {
         let mut body = serde_json::json!({
             "system": [
                 { "type": "text", "text": "sys a" },
@@ -767,17 +815,15 @@ mod tests {
             body["system"][1]["cache_control"],
             serde_json::json!({ "type": "ephemeral" })
         );
-        // string user content is wrapped so cache_control can attach
+        // tip (latest user) wrapped with cache_control
         assert_eq!(
             body["messages"][2]["content"][0]["cache_control"],
             serde_json::json!({ "type": "ephemeral" })
         );
-        assert!(
-            body["messages"][0]["content"]
-                .as_str()
-                .map(|s| s == "first")
-                .unwrap_or(false)
-                || body["messages"][0].get("cache_control").is_none()
+        // previous user tip also marked (Grok Build placement)
+        assert_eq!(
+            body["messages"][0]["content"][0]["cache_control"],
+            serde_json::json!({ "type": "ephemeral" })
         );
     }
 
@@ -803,6 +849,29 @@ mod tests {
             serde_json::json!({ "type": "ephemeral" })
         );
     }
+
+    #[test]
+    fn prompt_caching_skips_thinking_blocks_when_marking_tip() {
+        let mut body = serde_json::json!({
+            "messages": [{
+                "role": "assistant",
+                "content": [
+                    { "type": "text", "text": "answer" },
+                    { "type": "thinking", "thinking": "secret" },
+                ]
+            }],
+        });
+
+        apply_anthropic_prompt_caching(&mut body);
+
+        assert_eq!(
+            body["messages"][0]["content"][0]["cache_control"],
+            serde_json::json!({ "type": "ephemeral" })
+        );
+        assert!(body["messages"][0]["content"][1]
+            .get("cache_control")
+            .is_none());
+    }
 }
 
 fn anthropic_tool_output_content(tool: &crate::message::ToolOutputMessage) -> serde_json::Value {

src/aisdk/providers/compatible.rs

@@ -425,15 +425,32 @@ fn log_openai_compatible_usage(usage: &serde_json::Value) {
         return;
     }
 
+    // Prefer OpenAI-style cached_tokens; fall back to Anthropic-style cache_read.
+    let effective_cached = if cached > 0 { cached } else { cache_read };
+    let prompt_v = prompt.unwrap_or(0);
+    let hit_pct = if prompt_v > 0 {
+        (effective_cached as f64 * 100.0) / prompt_v as f64
+    } else if effective_cached > 0 || cache_creation > 0 {
+        let total = effective_cached.saturating_add(cache_creation);
+        if total > 0 {
+            (effective_cached as f64 * 100.0) / total as f64
+        } else {
+            0.0
+        }
+    } else {
+        0.0
+    };
+
     crate::emit_log!(
-        "[prompt-cache] openai-compatible prompt={} completion={} cached_tokens={} cache_read={} cache_creation={}",
+        "[prompt-cache] openai-compatible prompt={} completion={} cached_tokens={} cache_read={} cache_creation={} hit_pct={:.1}",
         prompt.map(|v| v.to_string()).unwrap_or_else(|| "-".into()),
         completion
             .map(|v| v.to_string())
             .unwrap_or_else(|| "-".into()),
         cached,
         cache_read,
-        cache_creation
+        cache_creation,
+        hit_pct
     );
 }
 

src/aisdk/providers/openai.rs

@@ -1522,6 +1522,9 @@ fn response_sse_data_to_chunk(data: &str) -> Option<Result<ChunkType>> {
                 return Some(Ok(responses_error_chunk(&value, event_type)));
             }
             let resp = &value["response"];
+            if let Some(usage) = resp.get("usage") {
+                log_openai_responses_usage(usage);
+            }
             Some(Ok(ChunkType::ResponseCompleted {
                 end_turn: resp.get("end_turn").and_then(|value| value.as_bool()),
             }))
@@ -1544,6 +1547,44 @@ fn response_sse_data_to_chunk(data: &str) -> Option<Result<ChunkType>> {
     }
 }
 
+/// Log Responses API usage for prompt-cache visibility.
+/// Looks for `input_tokens_details.cached_tokens` (OpenAI/xAI shape).
+fn log_openai_responses_usage(usage: &serde_json::Value) {
+    let input = usage
+        .get("input_tokens")
+        .or_else(|| usage.get("prompt_tokens"))
+        .and_then(|v| v.as_u64());
+    let output = usage
+        .get("output_tokens")
+        .or_else(|| usage.get("completion_tokens"))
+        .and_then(|v| v.as_u64());
+    let cached = usage
+        .pointer("/input_tokens_details/cached_tokens")
+        .or_else(|| usage.pointer("/prompt_tokens_details/cached_tokens"))
+        .and_then(|v| v.as_u64())
+        .or_else(|| usage.get("cached_tokens").and_then(|v| v.as_u64()))
+        .unwrap_or(0);
+
+    if input.is_none() && output.is_none() && cached == 0 {
+        return;
+    }
+
+    let input_v = input.unwrap_or(0);
+    let hit_pct = if input_v > 0 {
+        (cached as f64 * 100.0) / input_v as f64
+    } else {
+        0.0
+    };
+
+    crate::emit_log!(
+        "[prompt-cache] openai-responses input={} output={} cached_tokens={} hit_pct={:.1}",
+        input.map(|v| v.to_string()).unwrap_or_else(|| "-".into()),
+        output.map(|v| v.to_string()).unwrap_or_else(|| "-".into()),
+        cached,
+        hit_pct
+    );
+}
+
 fn responses_provider_error_message(value: &serde_json::Value, fallback: &str) -> String {
     let code = response_error_field(value, "code");
     let message = response_error_field(value, "message");