diff --git a/crates/proxy/src/integrations/langfuse.rs b/crates/proxy/src/integrations/langfuse.rs index 567fa2c..8838fc5 100644 --- a/crates/proxy/src/integrations/langfuse.rs +++ b/crates/proxy/src/integrations/langfuse.rs @@ -1,13 +1,351 @@ -// Langfuse integration stub (implementation in Task 3). +// Langfuse named callback integration. +// +// Sends LLM generation events to Langfuse's batch ingestion API. +// Activated when LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY are set, +// either explicitly via env vars or when "langfuse" appears in +// litellm_settings.callbacks. +// +// Fire-and-forget: send() spawns a tokio task and returns immediately. +use crate::admin::state::RequestLogEntry; use std::sync::Arc; -pub struct LangfuseClient; +pub struct LangfuseClient { + pub(crate) public_key: String, + pub(crate) secret_key: String, + pub(crate) host: String, + client: reqwest::Client, +} impl LangfuseClient { + /// Construct from env vars: LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST. + /// Returns None when either required key is absent or empty. pub fn from_env() -> Option> { - None + let public_key = std::env::var("LANGFUSE_PUBLIC_KEY").ok()?; + let secret_key = std::env::var("LANGFUSE_SECRET_KEY").ok()?; + if public_key.is_empty() || secret_key.is_empty() { + return None; + } + let host = std::env::var("LANGFUSE_HOST") + .unwrap_or_else(|_| "https://cloud.langfuse.com".to_string()); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(5)) + .build() + .expect("langfuse http client"); + Some(Arc::new(Self { + public_key, + secret_key, + host, + client, + })) } - pub fn send(&self, _entry: &crate::admin::state::RequestLogEntry) {} + /// Fire-and-forget: POST generation event to Langfuse ingestion API. + /// Returns immediately; the HTTP call happens in a spawned tokio task. + pub fn send(&self, entry: &RequestLogEntry) { + let payload = build_generation_payload(entry); + let url = format!("{}/api/public/ingestion", self.host.trim_end_matches('/')); + let auth = format!( + "Basic {}", + base64_encode(format!("{}:{}", self.public_key, self.secret_key).as_bytes()) + ); + let client = self.client.clone(); + tokio::spawn(async move { + match client + .post(&url) + .header("Authorization", &auth) + .json(&payload) + .send() + .await + { + Ok(resp) => { + if !resp.status().is_success() { + tracing::debug!( + url = %url, + status = %resp.status(), + "langfuse ingestion returned non-2xx" + ); + } + } + Err(e) => { + tracing::debug!(error = %e, "langfuse ingestion request failed"); + } + } + }); + } +} + +/// Build the Langfuse batch ingestion payload for a single generation. +pub(crate) fn build_generation_payload(entry: &RequestLogEntry) -> serde_json::Value { + let end_time = &entry.timestamp; + let start_time = iso8601_to_epoch(end_time) + .map(|epoch| { + crate::admin::db::epoch_to_iso8601(epoch.saturating_sub(entry.latency_ms / 1000)) + }) + .unwrap_or_else(|| end_time.clone()); + + let model = entry + .model_mapped + .as_deref() + .or(entry.model_requested.as_deref()) + .unwrap_or("unknown"); + + let mut usage = serde_json::json!({ "unit": "TOKENS" }); + if let Some(input) = entry.input_tokens { + usage["input"] = serde_json::json!(input); + } + if let Some(output) = entry.output_tokens { + usage["output"] = serde_json::json!(output); + } + if let Some(cost) = entry.cost_usd { + usage["totalCost"] = serde_json::json!(cost); + } + + let level = if entry.error_message.is_some() { "ERROR" } else { "DEFAULT" }; + + let mut metadata = serde_json::json!({ + "backend": entry.backend, + "status_code": entry.status_code, + "latency_ms": entry.latency_ms, + }); + if let Some(ref msg) = entry.error_message { + metadata["error"] = serde_json::json!(msg); + } + + serde_json::json!({ + "batch": [{ + "id": entry.request_id, + "type": "generation-create", + "timestamp": end_time, + "body": { + "id": entry.request_id, + "name": model, + "startTime": start_time, + "endTime": end_time, + "model": model, + "usage": usage, + "level": level, + "metadata": metadata, + } + }] + }) +} + +/// Encode bytes as standard base64 (RFC 4648). No external dependencies. +pub(crate) fn base64_encode(input: &[u8]) -> String { + const CHARS: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut out = String::with_capacity(input.len().div_ceil(3) * 4); + let mut i = 0; + while i + 2 < input.len() { + let b = ((input[i] as u32) << 16) + | ((input[i + 1] as u32) << 8) + | (input[i + 2] as u32); + out.push(CHARS[((b >> 18) & 0x3f) as usize] as char); + out.push(CHARS[((b >> 12) & 0x3f) as usize] as char); + out.push(CHARS[((b >> 6) & 0x3f) as usize] as char); + out.push(CHARS[(b & 0x3f) as usize] as char); + i += 3; + } + let rem = input.len() - i; + if rem == 1 { + let b = (input[i] as u32) << 16; + out.push(CHARS[((b >> 18) & 0x3f) as usize] as char); + out.push(CHARS[((b >> 12) & 0x3f) as usize] as char); + out.push('='); + out.push('='); + } else if rem == 2 { + let b = ((input[i] as u32) << 16) | ((input[i + 1] as u32) << 8); + out.push(CHARS[((b >> 18) & 0x3f) as usize] as char); + out.push(CHARS[((b >> 12) & 0x3f) as usize] as char); + out.push(CHARS[((b >> 6) & 0x3f) as usize] as char); + out.push('='); + } + out +} + +/// Parse ISO 8601 UTC timestamp ("2026-03-27T10:15:30Z") to Unix epoch seconds. +/// Returns None on parse failure. +pub(crate) fn iso8601_to_epoch(s: &str) -> Option { + if s.len() < 20 { + return None; + } + let year: i64 = s[0..4].parse().ok()?; + let month: i64 = s[5..7].parse().ok()?; + let day: i64 = s[8..10].parse().ok()?; + let hour: i64 = s[11..13].parse().ok()?; + let min: i64 = s[14..16].parse().ok()?; + let sec: i64 = s[17..19].parse().ok()?; + if !(1..=12).contains(&month) || !(1..=31).contains(&day) { + return None; + } + let days = days_from_civil(year, month, day); + let total = days * 86400 + hour * 3600 + min * 60 + sec; + u64::try_from(total).ok() +} + +/// Days from 1970-01-01 to the given date (may be negative for dates before epoch). +/// Algorithm: http://howardhinnant.github.io/date_algorithms.html +fn days_from_civil(year: i64, month: i64, day: i64) -> i64 { + // Shift March 1 to start of year so Feb (with leap day) is last month. + let y = if month <= 2 { year - 1 } else { year }; + let m = month; + let d = day; + let era = y.div_euclid(400); + let yoe = y - era * 400; // [0, 399] + let doy = (153 * (m + if m > 2 { -3 } else { 9 }) / 5) + d - 1; // [0, 365] + let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; // [0, 146096] + era * 146097 + doe - 719468 +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + // Serialize env-var tests to avoid races with parallel test runner. + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + #[test] + fn from_env_returns_none_when_keys_absent() { + let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + unsafe { + std::env::remove_var("LANGFUSE_PUBLIC_KEY"); + std::env::remove_var("LANGFUSE_SECRET_KEY"); + } + assert!(LangfuseClient::from_env().is_none()); + } + + #[test] + fn from_env_returns_some_when_keys_present() { + let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + unsafe { + std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test"); + std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test"); + } + let client = LangfuseClient::from_env(); + unsafe { + std::env::remove_var("LANGFUSE_PUBLIC_KEY"); + std::env::remove_var("LANGFUSE_SECRET_KEY"); + } + assert!(client.is_some()); + } + + #[test] + fn from_env_uses_custom_host() { + let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + unsafe { + std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test"); + std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test"); + std::env::set_var("LANGFUSE_HOST", "https://my-langfuse.example.com"); + } + let client = LangfuseClient::from_env(); + unsafe { + std::env::remove_var("LANGFUSE_PUBLIC_KEY"); + std::env::remove_var("LANGFUSE_SECRET_KEY"); + std::env::remove_var("LANGFUSE_HOST"); + } + assert_eq!(client.unwrap().host, "https://my-langfuse.example.com"); + } + + #[test] + fn base64_encode_empty() { + assert_eq!(base64_encode(b""), ""); + } + + #[test] + fn base64_encode_produces_expected_output() { + let encoded = base64_encode(b"pk-test:sk-test"); + assert_eq!(encoded, "cGstdGVzdDpzay10ZXN0"); + } + + #[test] + fn base64_encode_one_byte_padding() { + assert_eq!(base64_encode(b"A"), "QQ=="); + } + + #[test] + fn base64_encode_two_byte_padding() { + assert_eq!(base64_encode(b"AB"), "QUI="); + } + + #[test] + fn base64_encode_three_bytes_no_padding() { + assert_eq!(base64_encode(b"ABC"), "QUJD"); + } + + #[test] + fn iso8601_to_epoch_unix_epoch() { + let epoch = iso8601_to_epoch("1970-01-01T00:00:00Z").unwrap(); + assert_eq!(epoch, 0); + } + + #[test] + fn iso8601_to_epoch_parses_2026() { + let epoch = iso8601_to_epoch("2026-03-27T10:00:00Z").unwrap(); + assert!(epoch > 1_735_689_600); // > 2025-01-01 + assert!(epoch < 1_900_000_000); // < 2030 + } + + #[test] + fn iso8601_to_epoch_returns_none_on_short_string() { + assert!(iso8601_to_epoch("not-a-date").is_none()); + assert!(iso8601_to_epoch("").is_none()); + } + + #[test] + fn build_payload_structure() { + let entry = RequestLogEntry { + request_id: "req-123".to_string(), + timestamp: "2026-03-27T10:00:01Z".to_string(), + backend: "openai".to_string(), + model_requested: Some("claude-3-haiku-20240307".to_string()), + model_mapped: Some("gpt-4o-mini".to_string()), + status_code: 200, + latency_ms: 500, + input_tokens: Some(100), + output_tokens: Some(50), + is_streaming: false, + error_message: None, + key_id: None, + cost_usd: Some(0.001), + }; + let payload = build_generation_payload(&entry); + let batch = payload["batch"].as_array().unwrap(); + assert_eq!(batch.len(), 1); + let event = &batch[0]; + assert_eq!(event["type"], "generation-create"); + let body = &event["body"]; + assert_eq!(body["model"], "gpt-4o-mini"); + assert_eq!(body["usage"]["input"], 100); + assert_eq!(body["usage"]["output"], 50); + assert_eq!(body["usage"]["unit"], "TOKENS"); + assert!(body["startTime"].as_str().is_some()); + assert!(body["endTime"].as_str().is_some()); + assert_eq!(body["level"], "DEFAULT"); + assert_eq!(body["metadata"]["backend"], "openai"); + assert_eq!(body["metadata"]["latency_ms"], 500); + } + + #[test] + fn build_payload_error_entry() { + let entry = RequestLogEntry { + request_id: "req-err".to_string(), + timestamp: "2026-03-27T10:00:01Z".to_string(), + backend: "openai".to_string(), + model_requested: Some("gpt-4o".to_string()), + model_mapped: None, + status_code: 500, + latency_ms: 100, + input_tokens: None, + output_tokens: None, + is_streaming: false, + error_message: Some("internal error".to_string()), + key_id: None, + cost_usd: None, + }; + let payload = build_generation_payload(&entry); + let body = &payload["batch"][0]["body"]; + assert_eq!(body["level"], "ERROR"); + assert_eq!(body["metadata"]["error"], "internal error"); + } }