From d024af0ff1f9576771ce9d6709ce2cdf9baa8860 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81ngel=20Manuel=20Mart=C3=ADn?= Date: Thu, 16 Apr 2026 14:44:14 +0000 Subject: [PATCH] =?UTF-8?q?feat(loop):=20richer=20Langfuse=20signals=20?= =?UTF-8?q?=E2=80=94=20cost,=20cache=20tokens,=20optional=20input=20previe?= =?UTF-8?q?w?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parse OpenRouter's richer `usage` shape off the SSE wire: top-level `cost`, plus cache tokens in both Anthropic-flat (`cache_read_input_tokens`, `cache_creation_input_tokens`) and OpenAI-nested (`prompt_tokens_details.cached_tokens`) forms. A custom `Deserialize` on `Usage` collapses both into the flat public shape (cache_read = max of the two sources). Previously-dropped data now flows into the Langfuse generation span (`metadata.cost_usd`, `cache_read_tokens`, `cache_creation_tokens`) and into `trace.finalize(...total_cost...)`. Add an opt-in `LANGFUSE_CAPTURE_INPUT` flag that attaches a redacted chat-log snapshot (ported from overfolder's `build_context_snapshot`) to the generation `input` field. Adapted for overloop's `Content` enum: non-text blocks render as `[image]`/`[audio]`/`[media]` placeholders so no base64/data-URI payload leaks. Off by default — the default posture keeps prompt content local. - New `loop/src/observability/preview.rs` + 7 unit tests. - 4 new deser tests covering Anthropic-flat, OpenAI-nested, both absent, and both-present-picks-max. - `test_langfuse.rs`: existing success test now asserts cost_usd round-trip; new test covers string-preview round-trip. Co-Authored-By: Claude Opus 4.6 (1M context) --- loop/src/acp.rs | 3 + loop/src/agentic_loop.rs | 50 ++++-- loop/src/config.rs | 37 +++++ loop/src/llm/types.rs | 118 ++++++++++++++- loop/src/main.rs | 1 + loop/src/observability/langfuse.rs | 1 + loop/src/observability/mod.rs | 2 + loop/src/observability/preview.rs | 235 +++++++++++++++++++++++++++++ loop/tests/test_agentic_loop.rs | 6 + loop/tests/test_compaction.rs | 1 + loop/tests/test_langfuse.rs | 47 +++++- 11 files changed, 487 insertions(+), 14 deletions(-) create mode 100644 loop/src/observability/preview.rs diff --git a/loop/src/acp.rs b/loop/src/acp.rs index c5da7fc..4184684 100644 --- a/loop/src/acp.rs +++ b/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(); diff --git a/loop/src/agentic_loop.rs b/loop/src/agentic_loop.rs index 0255a8e..22680e9 100644 --- a/loop/src/agentic_loop.rs +++ b/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(()) } diff --git a/loop/src/config.rs b/loop/src/config.rs index 0f2c400..f956e3e 100644 --- a/loop/src/config.rs +++ b/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() { diff --git a/loop/src/llm/types.rs b/loop/src/llm/types.rs index df42a8f..021eeb3 100644 --- a/loop/src/llm/types.rs +++ b/loop/src/llm/types.rs @@ -222,12 +222,70 @@ pub struct FunctionCallDelta { pub arguments: Option, } -#[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>(de: D) -> Result { + #[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, + } + + #[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); + } } diff --git a/loop/src/main.rs b/loop/src/main.rs index 9ddc054..a90d93a 100644 --- a/loop/src/main.rs +++ b/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); diff --git a/loop/src/observability/langfuse.rs b/loop/src/observability/langfuse.rs index 11697b9..af20707 100644 --- a/loop/src/observability/langfuse.rs +++ b/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, } } diff --git a/loop/src/observability/mod.rs b/loop/src/observability/mod.rs index af84647..e282f8d 100644 --- a/loop/src/observability/mod.rs +++ b/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; diff --git a/loop/src/observability/preview.rs b/loop/src/observability/preview.rs new file mode 100644 index 0000000..42d6a8a --- /dev/null +++ b/loop/src/observability/preview.rs @@ -0,0 +1,235 @@ +//! Redacted chat-log snapshot used as the Langfuse generation `input`. +//! +//! Ported from the sister `overfolder/agent-runner` harness; adapted for +//! overloop's `Content` enum (which supports multimodal `Blocks`) and for +//! the absence of `` system markers in this stack. Off by default +//! — callers gate on `Config::langfuse_capture_input`. + +use std::fmt::Write as _; + +use serde_json::Value; + +use crate::llm::{Content, Message, Role}; + +/// Number of trailing messages rendered verbatim (after the system +/// prompt at index 0). Anything earlier is summarised as a count. +const DETAIL_TAIL: usize = 3; + +const SYSTEM_TRUNCATE: usize = 200; +const ASSISTANT_TRUNCATE: usize = 300; +const TOOL_TRUNCATE: usize = 200; + +/// Build a redacted plain-text chat-log snapshot suitable for the +/// Langfuse generation `input` field. +/// +/// Shape: +/// - Older messages (past the trailing window) collapse to +/// ``. +/// - Each retained message renders as `[ROLE]: body`, with per-role +/// truncation: system 200, assistant 300, tool 200, user full. +/// - Multimodal image/audio blocks render as `[image]` / `[media]` +/// placeholders so no base64/data-URI payload leaks into the trace. +pub fn build_context_snapshot(messages: &[Message]) -> Value { + // Messages retained in detail: the trailing DETAIL_TAIL items, + // always keeping the system prompt at index 0 hidden from detail + // (it's already captured by Langfuse's trace-level metadata). + let detail_start = messages.len().saturating_sub(DETAIL_TAIL).max(1); + let skipped = detail_start.saturating_sub(1); + + let mut out = String::new(); + if skipped > 0 { + let _ = writeln!(out, "<{skipped} more messages...>"); + } + + for m in messages.iter().skip(detail_start) { + let body = content_to_text(m.content.as_ref()); + match m.role { + Role::System => { + let _ = write!(out, "\n[SYSTEM]: {}", truncate_str(&body, SYSTEM_TRUNCATE)); + } + Role::User => { + let _ = write!(out, "\n[USER]: {body}"); + } + Role::Assistant => { + let _ = write!( + out, + "\n[AGENT]: {}", + truncate_str(&body, ASSISTANT_TRUNCATE) + ); + } + Role::Tool => { + let id = m.tool_call_id.as_deref().unwrap_or("?"); + let _ = write!(out, "\n[TOOL {id}]: {}", truncate_str(&body, TOOL_TRUNCATE)); + } + } + } + + Value::String(out) +} + +/// Flatten a `Content` into a single string, replacing non-text blocks +/// with placeholders so no media payload reaches Langfuse. +fn content_to_text(content: Option<&Content>) -> String { + use crate::llm::{ContentBlock, TypedBlock}; + + let Some(c) = content else { + return String::new(); + }; + match c { + Content::Text(s) => s.clone(), + Content::Blocks(blocks) => { + let mut parts: Vec = Vec::with_capacity(blocks.len()); + for b in blocks { + match b { + TypedBlock::Known(ContentBlock::Text { text }) => parts.push(text.clone()), + TypedBlock::Known(ContentBlock::ImageUrl { .. }) + | TypedBlock::Known(ContentBlock::Image { .. }) => parts.push("[image]".into()), + TypedBlock::Known(ContentBlock::InputAudio { .. }) => { + parts.push("[audio]".into()) + } + TypedBlock::Unknown(_) => parts.push("[media]".into()), + } + } + parts.join("\n") + } + } +} + +fn truncate_str(s: &str, max: usize) -> String { + if s.len() <= max { + return s.to_string(); + } + let mut end = max; + while !s.is_char_boundary(end) { + end -= 1; + } + let mut out = s[..end].to_string(); + out.push_str("…[truncated]"); + out +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + use crate::llm::{ContentBlock, TypedBlock}; + + fn msg(role: Role, text: &str) -> Message { + Message { + role, + content: Some(Content::Text(text.into())), + tool_calls: None, + tool_call_id: None, + } + } + + fn tool_msg(id: &str, text: &str) -> Message { + Message { + role: Role::Tool, + content: Some(Content::Text(text.into())), + tool_calls: None, + tool_call_id: Some(id.into()), + } + } + + #[test] + fn short_transcript_renders_every_message_past_system() { + let messages = vec![ + msg(Role::System, "system prompt"), + msg(Role::User, "hello"), + msg(Role::Assistant, "hi"), + ]; + let snap = build_context_snapshot(&messages); + let s = snap.as_str().unwrap(); + assert!(!s.contains("more messages")); + assert!(s.contains("[USER]: hello")); + assert!(s.contains("[AGENT]: hi")); + // System prompt (index 0) is NOT rendered in the detail window + // when the transcript fits inside the trailing window. + assert!(!s.contains("[SYSTEM]: system prompt")); + } + + #[test] + fn long_transcript_summarises_older_messages() { + let mut messages = vec![msg(Role::System, "sys")]; + for i in 0..10 { + messages.push(msg(Role::User, &format!("u{i}"))); + messages.push(msg(Role::Assistant, &format!("a{i}"))); + } + // 1 system + 20 turns → len=21. detail_start = max(21-3, 1) = 18. + // skipped = 18 - 1 = 17 (system prompt at idx 0 isn't counted). + let snap = build_context_snapshot(&messages); + let s = snap.as_str().unwrap(); + assert!(s.starts_with("<17 more messages...>"), "snapshot: {s}"); + // Last user + last assistant must be present verbatim. + assert!(s.contains("[USER]: u9")); + assert!(s.contains("[AGENT]: a9")); + // First user must NOT be in the detail window. + assert!(!s.contains("[USER]: u0")); + } + + #[test] + fn assistant_truncates_at_300() { + let long = "x".repeat(500); + let messages = vec![msg(Role::System, "s"), msg(Role::Assistant, &long)]; + let snap = build_context_snapshot(&messages); + let s = snap.as_str().unwrap(); + assert!(s.contains("…[truncated]")); + // Body + marker should be strictly less than the original. + let agent_line = s.lines().find(|l| l.starts_with("[AGENT]:")).unwrap(); + assert!(agent_line.len() < long.len() + "[AGENT]: ".len()); + } + + #[test] + fn user_is_not_truncated() { + let long = "y".repeat(5_000); + let messages = vec![msg(Role::System, "s"), msg(Role::User, &long)]; + let snap = build_context_snapshot(&messages); + let s = snap.as_str().unwrap(); + assert!(s.contains(&long)); + } + + #[test] + fn tool_output_truncates_and_labels_with_id() { + let long = "t".repeat(500); + let messages = vec![msg(Role::System, "s"), tool_msg("call_123", &long)]; + let snap = build_context_snapshot(&messages); + let s = snap.as_str().unwrap(); + assert!(s.contains("[TOOL call_123]:")); + assert!(s.contains("…[truncated]")); + } + + #[test] + fn multimodal_blocks_render_as_placeholders() { + let m = Message { + role: Role::User, + content: Some(Content::Blocks(vec![ + TypedBlock::Known(ContentBlock::Text { + text: "see this".into(), + }), + TypedBlock::Known(ContentBlock::ImageUrl { + image_url: json!({"url": "data:image/png;base64,SECRET"}), + }), + ])), + tool_calls: None, + tool_call_id: None, + }; + let messages = vec![msg(Role::System, "s"), m]; + let snap = build_context_snapshot(&messages); + let s = snap.as_str().unwrap(); + assert!(s.contains("[USER]: see this")); + assert!(s.contains("[image]")); + assert!(!s.contains("SECRET")); + assert!(!s.contains("data:image")); + } + + #[test] + fn truncate_respects_char_boundaries() { + // Multi-byte char straddling the cut point. + let s = "aaa🦀bbb"; + let out = truncate_str(s, 4); + assert!(out.ends_with("…[truncated]")); + // No panic means boundary handling is correct. + } +} diff --git a/loop/tests/test_agentic_loop.rs b/loop/tests/test_agentic_loop.rs index 1e54160..db2c76c 100644 --- a/loop/tests/test_agentic_loop.rs +++ b/loop/tests/test_agentic_loop.rs @@ -178,6 +178,7 @@ fn make_text_response(text: &str, stop: StopReason) -> Result prompt_tokens: 10, completion_tokens: 5, total_tokens: 15, + ..Default::default() }), }) } @@ -202,6 +203,7 @@ fn make_tool_call_response(tool_name: &str, arguments: &str) -> Result LoopConfig { max_iterations: 10, timeout: Duration::from_secs(30), model: "test-model".into(), + langfuse_capture_input: false, } } @@ -422,6 +425,7 @@ async fn test_timeout() { max_iterations: 10, timeout: Duration::from_millis(0), model: "test-model".into(), + langfuse_capture_input: false, }; run( @@ -539,6 +543,7 @@ async fn test_wind_down_injects_system_message_when_iterations_remaining_equals_ max_iterations: 5, timeout: Duration::from_secs(30), model: "test-model".into(), + langfuse_capture_input: false, }; run( @@ -587,6 +592,7 @@ async fn test_silence_nudge_injected_after_three_silent_turns() { prompt_tokens: 1, completion_tokens: 0, total_tokens: 1, + ..Default::default() }), }) } diff --git a/loop/tests/test_compaction.rs b/loop/tests/test_compaction.rs index 7a9d68e..e7412a6 100644 --- a/loop/tests/test_compaction.rs +++ b/loop/tests/test_compaction.rs @@ -33,6 +33,7 @@ impl LlmService for MockLlm { prompt_tokens: 100, completion_tokens: 50, total_tokens: 150, + ..Default::default() }), }) } diff --git a/loop/tests/test_langfuse.rs b/loop/tests/test_langfuse.rs index 8c5ac91..78e56d2 100644 --- a/loop/tests/test_langfuse.rs +++ b/loop/tests/test_langfuse.rs @@ -31,6 +31,7 @@ fn test_config(host: &str, public: Option<&str>, secret: Option<&str>) -> Config langfuse_secret_key: secret.map(str::to_string), langfuse_host: host.to_string(), langfuse_environment: "test".into(), + langfuse_capture_input: false, } } @@ -119,7 +120,7 @@ async fn record_generation_success_posts_generation_create() { stop_reason: "stop".into(), prompt_tokens: 100, completion_tokens: 25, - cost: 0.0, + cost: 0.012_345, start_time: now - chrono::Duration::milliseconds(200), end_time: now, cache_read_tokens: 80, @@ -140,11 +141,55 @@ async fn record_generation_success_posts_generation_create() { assert_eq!(body["metadata"]["stop_reason"], "stop"); assert_eq!(body["metadata"]["cache_read_tokens"], 80); assert_eq!(body["metadata"]["cache_creation_tokens"], 5); + // OpenRouter `usage.cost` must round-trip into metadata.cost_usd. + let cost = body["metadata"]["cost_usd"].as_f64().unwrap(); + assert!((cost - 0.012_345).abs() < 1e-9, "unexpected cost: {cost}"); // Success path — no level / statusMessage fields. assert!(body.get("level").is_none()); assert!(body.get("statusMessage").is_none()); } +/// When `LANGFUSE_CAPTURE_INPUT=1` the agentic loop substitutes a +/// `Value::String` chat-log snapshot for the placeholder JSON object. +/// Verify that whichever shape the caller passes round-trips verbatim +/// into the span `input` field. +#[tokio::test] +async fn generation_input_preview_string_roundtrips() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/api/public/ingestion")) + .respond_with(ResponseTemplate::new(207)) + .mount(&server) + .await; + + let cfg = test_config(&server.uri(), Some("pk"), Some("sk")); + let tracer = LangfuseTracer::new(&cfg); + let trace = tracer.start_session("s".into()); + + let snapshot = "<2 more messages...>\n[USER]: hi there\n[AGENT]: hello"; + let now = Utc::now(); + trace.record_generation(GenerationParams { + model: "gpt-4".into(), + message_count: 4, + input_preview: Value::String(snapshot.into()), + output_text: Some("ok".into()), + stop_reason: "stop".into(), + prompt_tokens: 10, + completion_tokens: 2, + cost: 0.0, + start_time: now - chrono::Duration::milliseconds(10), + end_time: now, + cache_read_tokens: 0, + cache_creation_tokens: 0, + level: None, + status_message: None, + }); + + let reqs = wait_for_requests(&server, 1).await; + let body = batch_body(&reqs[0]); + assert_eq!(body["input"], snapshot); +} + #[tokio::test] async fn record_generation_error_sets_level_and_status() { let server = MockServer::start().await;