feat(loop): richer Langfuse signals — OpenRouter cost/cache + optional input preview

loop/src/acp.rs

@@ -624,6 +624,7 @@ mod tests {
             prompt_tokens: 10,
             completion_tokens: 5,
             total_tokens: 15,
+            ..Default::default()
         };
         acp.turn_end(&messages, &usage).unwrap();
 
@@ -679,6 +680,7 @@ mod tests {
             prompt_tokens: 100,
             completion_tokens: 50,
             total_tokens: 150,
+            ..Default::default()
         };
         acp.turn_end(&messages, &usage).unwrap();
 
@@ -706,6 +708,7 @@ mod tests {
             prompt_tokens: 100,
             completion_tokens: 50,
             total_tokens: 150,
+            ..Default::default()
         };
         acp.turn_end(&messages, &usage).unwrap();
 

loop/src/agentic_loop.rs

@@ -9,7 +9,9 @@ use crate::llm::{
     parse_tool_arguments, Content, ContentBlock, Message, ParsedArguments, Role, StopReason,
     ToolContent, TypedBlock, Usage,
 };
-use crate::observability::{GenerationParams, SessionTrace, ToolSpanParams};
+use crate::observability::{
+    build_context_snapshot, GenerationParams, SessionTrace, ToolSpanParams,
+};
 use crate::tools::{ToolOutput, ToolRegistry};
 use crate::traits::{AcpService, LlmService};
 
@@ -30,6 +32,9 @@ pub struct LoopConfig {
     /// Model name recorded on Langfuse generation spans. Free-form label;
     /// the trace is untouched when the `SessionTrace` is disabled.
     pub model: String,
+    /// When true, attach a redacted chat-log snapshot to Langfuse
+    /// generation `input`. Off by default.
+    pub langfuse_capture_input: bool,
 }
 
 pub async fn run(
@@ -41,11 +46,7 @@ pub async fn run(
     trace: &SessionTrace,
 ) -> Result<()> {
     let start = Instant::now();
-    let mut total_usage = Usage {
-        prompt_tokens: 0,
-        completion_tokens: 0,
-        total_tokens: 0,
-    };
+    let mut total_usage = Usage::default();
     let mut silent_turns = 0u32;
     let mut last_heartbeat = Instant::now();
     let mut response_text = String::new();
@@ -109,6 +110,15 @@ pub async fn run(
         let mut text_deltas = Vec::new();
         let tools = registry.definitions();
         let message_count = messages.len();
+        // Pre-build the input preview once per LLM call so both the
+        // error and success branches below share it without cloning
+        // messages twice. Cheap when the flag is off (one `json!`
+        // literal); only touches message content when opted in.
+        let input_preview = if config.langfuse_capture_input {
+            build_context_snapshot(messages)
+        } else {
+            json!({"message_count": message_count})
+        };
         let lf_start = Utc::now();
         let streamed = llm
             .stream_completion(messages, &tools, &mut |text| {
@@ -137,7 +147,7 @@ pub async fn run(
                 trace.record_generation(GenerationParams {
                     model: config.model.clone(),
                     message_count,
-                    input_preview: json!({"message_count": message_count}),
+                    input_preview: input_preview.clone(),
                     output_text: None,
                     stop_reason: "error".into(),
                     prompt_tokens: 0,
@@ -167,7 +177,7 @@ pub async fn run(
         trace.record_generation(GenerationParams {
             model: config.model.clone(),
             message_count,
-            input_preview: json!({"message_count": message_count}),
+            input_preview,
             output_text: (!streamed_text.is_empty()).then(|| streamed_text.clone()),
             stop_reason: stop_reason_str.into(),
             prompt_tokens: streamed
@@ -180,11 +190,19 @@ pub async fn run(
                 .as_ref()
                 .map(|u| u.completion_tokens)
                 .unwrap_or(0),
-            cost: 0.0,
+            cost: streamed.usage.as_ref().map(|u| u.cost).unwrap_or(0.0),
             start_time: lf_start,
             end_time: lf_end,
-            cache_read_tokens: 0,
-            cache_creation_tokens: 0,
+            cache_read_tokens: streamed
+                .usage
+                .as_ref()
+                .map(|u| u.cache_read_tokens)
+                .unwrap_or(0),
+            cache_creation_tokens: streamed
+                .usage
+                .as_ref()
+                .map(|u| u.cache_creation_tokens)
+                .unwrap_or(0),
             level: None,
             status_message: None,
         });
@@ -197,6 +215,9 @@ pub async fn run(
             total_usage.prompt_tokens += usage.prompt_tokens;
             total_usage.completion_tokens += usage.completion_tokens;
             total_usage.total_tokens += usage.total_tokens;
+            total_usage.cost += usage.cost;
+            total_usage.cache_read_tokens += usage.cache_read_tokens;
+            total_usage.cache_creation_tokens += usage.cache_creation_tokens;
             let _ = acp.quota_update(usage.prompt_tokens, usage.completion_tokens);
         }
 
@@ -318,7 +339,12 @@ pub async fn run(
         error!("Failed to emit turn/end: {}", e);
     }
 
-    trace.finalize(total_usage.total_tokens, 0.0, tool_count, &response_text);
+    trace.finalize(
+        total_usage.total_tokens,
+        total_usage.cost,
+        tool_count,
+        &response_text,
+    );
 
     Ok(())
 }

loop/src/config.rs

@@ -24,6 +24,10 @@ pub struct Config {
     pub langfuse_host: String,
     /// Tag attached to every Langfuse observation. Defaults to `local`.
     pub langfuse_environment: String,
+    /// Opt-in: attach a redacted chat-log snapshot to the Langfuse
+    /// generation `input` field. Off by default so prompt content
+    /// stays local unless the operator explicitly opts in.
+    pub langfuse_capture_input: bool,
 }
 
 impl Config {
@@ -88,6 +92,11 @@ impl Config {
         let langfuse_environment =
             env::var("LANGFUSE_ENVIRONMENT").unwrap_or_else(|_| "local".into());
 
+        let langfuse_capture_input = env::var("LANGFUSE_CAPTURE_INPUT")
+            .ok()
+            .map(|s| matches!(s.trim().to_ascii_lowercase().as_str(), "1" | "true" | "yes"))
+            .unwrap_or(false);
+
         Ok(Self {
             llm_api_key,
             llm_api_url,
@@ -104,6 +113,7 @@ impl Config {
             langfuse_secret_key,
             langfuse_host,
             langfuse_environment,
+            langfuse_capture_input,
         })
     }
 }
@@ -131,6 +141,7 @@ mod tests {
             "LANGFUSE_SECRET_KEY",
             "LANGFUSE_HOST",
             "LANGFUSE_ENVIRONMENT",
+            "LANGFUSE_CAPTURE_INPUT",
         ] {
             env::remove_var(key);
         }
@@ -166,6 +177,7 @@ mod tests {
         assert!(cfg.langfuse_secret_key.is_none());
         assert_eq!(cfg.langfuse_host, "https://cloud.langfuse.com");
         assert_eq!(cfg.langfuse_environment, "local");
+        assert!(!cfg.langfuse_capture_input);
         cleanup_env();
     }
 
@@ -186,6 +198,31 @@ mod tests {
         cleanup_env();
     }
 
+    #[test]
+    #[serial]
+    fn test_langfuse_capture_input_toggle() {
+        cleanup_env();
+        env::set_var("LLM_API_KEY", "test");
+        for (val, expected) in [
+            ("1", true),
+            ("true", true),
+            ("TRUE", true),
+            ("yes", true),
+            ("0", false),
+            ("false", false),
+            ("bogus", false),
+            ("", false),
+        ] {
+            env::set_var("LANGFUSE_CAPTURE_INPUT", val);
+            let cfg = Config::from_env().unwrap();
+            assert_eq!(
+                cfg.langfuse_capture_input, expected,
+                "LANGFUSE_CAPTURE_INPUT={val:?}"
+            );
+        }
+        cleanup_env();
+    }
+
     #[test]
     #[serial]
     fn test_langfuse_empty_keys_treated_as_unset() {

loop/src/llm/types.rs

@@ -222,12 +222,70 @@ pub struct FunctionCallDelta {
     pub arguments: Option<String>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Default)]
 pub struct Usage {
     pub prompt_tokens: u64,
     pub completion_tokens: u64,
     #[serde(default)]
     pub total_tokens: u64,
+    /// USD cost reported by OpenRouter (top-level `usage.cost`).
+    /// Zero when the provider doesn't populate it.
+    #[serde(default)]
+    pub cost: f64,
+    /// Tokens served from prompt cache. Merged from Anthropic's flat
+    /// `cache_read_input_tokens` and OpenAI's nested
+    /// `prompt_tokens_details.cached_tokens` — we take the max since
+    /// different providers populate different fields.
+    #[serde(default)]
+    pub cache_read_tokens: u64,
+    /// Tokens written to prompt cache (Anthropic-only field
+    /// `cache_creation_input_tokens`).
+    #[serde(default)]
+    pub cache_creation_tokens: u64,
+}
+
+// Custom Deserialize so both Anthropic-flat and OpenAI-nested cache
+// shapes collapse into the flat public `Usage`.
+impl<'de> Deserialize<'de> for Usage {
+    fn deserialize<D: serde::Deserializer<'de>>(de: D) -> Result<Self, D::Error> {
+        #[derive(Deserialize, Default)]
+        struct Raw {
+            #[serde(default)]
+            prompt_tokens: u64,
+            #[serde(default)]
+            completion_tokens: u64,
+            #[serde(default)]
+            total_tokens: u64,
+            #[serde(default)]
+            cost: f64,
+            #[serde(default)]
+            cache_read_input_tokens: u64,
+            #[serde(default)]
+            cache_creation_input_tokens: u64,
+            #[serde(default)]
+            prompt_tokens_details: Option<PromptTokensDetails>,
+        }
+
+        #[derive(Deserialize, Default)]
+        struct PromptTokensDetails {
+            #[serde(default)]
+            cached_tokens: u64,
+        }
+
+        let raw = Raw::deserialize(de)?;
+        let nested_cached = raw
+            .prompt_tokens_details
+            .as_ref()
+            .map_or(0, |d| d.cached_tokens);
+        Ok(Usage {
+            prompt_tokens: raw.prompt_tokens,
+            completion_tokens: raw.completion_tokens,
+            total_tokens: raw.total_tokens,
+            cost: raw.cost,
+            cache_read_tokens: raw.cache_read_input_tokens.max(nested_cached),
+            cache_creation_tokens: raw.cache_creation_input_tokens,
+        })
+    }
 }
 
 #[derive(Debug, Clone, Deserialize, PartialEq)]
@@ -435,4 +493,62 @@ mod tests {
             serde_json::from_value(json!({"prompt_tokens":10,"completion_tokens":5})).unwrap();
         assert_eq!(without.total_tokens, 0);
     }
+
+    #[test]
+    fn usage_deser_cost_and_cache_absent() {
+        let u: Usage =
+            serde_json::from_value(json!({"prompt_tokens": 10, "completion_tokens": 5})).unwrap();
+        assert_eq!(u.cost, 0.0);
+        assert_eq!(u.cache_read_tokens, 0);
+        assert_eq!(u.cache_creation_tokens, 0);
+    }
+
+    #[test]
+    fn usage_deser_anthropic_flat_cache() {
+        let u: Usage = serde_json::from_value(json!({
+            "prompt_tokens": 100,
+            "completion_tokens": 20,
+            "cost": 0.0123,
+            "cache_read_input_tokens": 80,
+            "cache_creation_input_tokens": 30,
+        }))
+        .unwrap();
+        assert!((u.cost - 0.0123).abs() < 1e-9);
+        assert_eq!(u.cache_read_tokens, 80);
+        assert_eq!(u.cache_creation_tokens, 30);
+    }
+
+    #[test]
+    fn usage_deser_openai_nested_cache() {
+        let u: Usage = serde_json::from_value(json!({
+            "prompt_tokens": 100,
+            "completion_tokens": 20,
+            "prompt_tokens_details": { "cached_tokens": 42 },
+        }))
+        .unwrap();
+        assert_eq!(u.cache_read_tokens, 42);
+        assert_eq!(u.cache_creation_tokens, 0);
+    }
+
+    #[test]
+    fn usage_deser_both_shapes_picks_max() {
+        let u: Usage = serde_json::from_value(json!({
+            "prompt_tokens": 100,
+            "completion_tokens": 20,
+            "cache_read_input_tokens": 10,
+            "prompt_tokens_details": { "cached_tokens": 50 },
+        }))
+        .unwrap();
+        // max(flat=10, nested=50) = 50
+        assert_eq!(u.cache_read_tokens, 50);
+
+        let u2: Usage = serde_json::from_value(json!({
+            "prompt_tokens": 100,
+            "completion_tokens": 20,
+            "cache_read_input_tokens": 77,
+            "prompt_tokens_details": { "cached_tokens": 50 },
+        }))
+        .unwrap();
+        assert_eq!(u2.cache_read_tokens, 77);
+    }
 }

loop/src/main.rs

@@ -177,6 +177,7 @@ async fn run(config: Config) -> Result<()> {
         max_iterations: config.max_iterations,
         timeout: Duration::from_secs(config.timeout_minutes * 60),
         model: config.model.clone(),
+        langfuse_capture_input: config.langfuse_capture_input,
     };
 
     let tracer = LangfuseTracer::new(&config);

loop/src/observability/langfuse.rs

@@ -338,6 +338,7 @@ mod tests {
             langfuse_secret_key: secret.map(str::to_string),
             langfuse_host: "https://cloud.langfuse.com".into(),
             langfuse_environment: "test".into(),
+            langfuse_capture_input: false,
         }
     }
 

loop/src/observability/mod.rs

@@ -1,3 +1,5 @@
 pub mod langfuse;
+pub mod preview;
 
 pub use langfuse::{GenerationParams, LangfuseTracer, SessionTrace, ToolSpanParams};
+pub use preview::build_context_snapshot;