[feat] langfuse tracing layer

Justfile

@@ -25,3 +25,8 @@ run-ui:
 run-server:
     @echo "Running server..."
     cargo run -p goose-server
+
+# Setup langfuse server
+langfuse-server:
+    #!/usr/bin/env bash
+    ./scripts/setup_langfuse.sh

crates/goose-cli/Cargo.toml

@@ -30,14 +30,16 @@ serde_yaml = "0.9"
 dirs = "4.0"
 strum = "0.26"
 strum_macros = "0.26"
-reqwest = "0.11.27"
+reqwest = { version = "0.11.27", features = ["json"] }
 rand = "0.8.5"
 async-trait = "0.1"
 rustyline = "15.0.0"
 rust_decimal = "1.36.0"
 rust_decimal_macros = "1.36.0"
+tracing = "0.1"
+chrono = "0.4"
+parking_lot = "0.12.3"
 
 [dev-dependencies]
 tempfile = "3"
 temp-env = { version = "0.3.6", features = ["async_closure"] }
-

crates/goose-cli/src/main.rs

@@ -17,6 +17,7 @@ use commands::session::build_session;
 use commands::version::print_version;
 use profile::has_no_profiles;
 use std::io::{self, Read};
+use goose::logging::setup_logging;
 
 mod log_usage;
 
@@ -195,6 +196,8 @@ enum CliProviderVariant {
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    setup_logging()?;
+
     let cli = Cli::parse();
 
     if cli.version {

crates/goose/Cargo.toml

@@ -47,6 +47,9 @@ xcap = "0.0.14"
 libc = "=0.2.167"
 lazy_static = "1.5"
 kill_tree = "0.2.4"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json", "time"] }
+tracing-appender = "0.2"
 
 keyring = { version = "3.6.1", features = [
     "apple-native",

crates/goose/src/agent.rs

@@ -5,6 +5,7 @@ use rust_decimal_macros::dec;
 use serde_json::json;
 use std::collections::HashMap;
 use tokio::sync::Mutex;
+use tracing::{debug, instrument, warn};
 
 use crate::errors::{AgentError, AgentResult};
 use crate::message::{Message, ToolRequest};
@@ -107,11 +108,13 @@ impl Agent {
     }
 
     /// Dispatch a single tool call to the appropriate system
+    #[instrument(skip(self, tool_call), fields(input, output))]
     async fn dispatch_tool_call(
         &self,
         tool_call: AgentResult<ToolCall>,
     ) -> AgentResult<Vec<Content>> {
         let call = tool_call?;
+
         let system = self
             .get_system_for_tool(&call.name)
             .ok_or_else(|| AgentError::ToolNotFound(call.name.clone()))?;
@@ -122,8 +125,13 @@ impl Agent {
             .nth(1)
             .ok_or_else(|| AgentError::InvalidToolName(call.name.clone()))?;
         let system_tool_call = ToolCall::new(tool_name, call.arguments);
+        let result = system.call(system_tool_call.clone()).await;
+
+        tracing::debug!("input"=serde_json::to_string(&system_tool_call).unwrap(),
+                        "output"=serde_json::to_string(&result).unwrap(),
+        );
 
-        system.call(system_tool_call).await
+        result 
     }
 
     fn get_system_prompt(&self) -> AgentResult<String> {
@@ -213,7 +221,12 @@ impl Agent {
         let mut status_content: Vec<String> = Vec::new();
 
         if approx_count > target_limit {
-            println!("[WARNING] Token budget exceeded. Current count: {} \n Difference: {} tokens over buget. Removing context", approx_count, approx_count - target_limit);
+            warn!(
+                approx_count,
+                target_limit,
+                overage = approx_count - target_limit,
+                "Token budget exceeded, removing context"
+            );
 
             // Get token counts for each resourcee
             let mut system_token_counts = HashMap::new();
@@ -326,9 +339,20 @@ impl Agent {
 
     /// Create a stream that yields each message as it's generated by the agent.
     /// This includes both the assistant's responses and any tool responses.
+    #[instrument(skip(self, messages), fields(user_message))]
     pub async fn reply(&self, messages: &[Message]) -> Result<BoxStream<'_, Result<Message>>> {
+        let reply_span = tracing::Span::current();
         let mut messages = messages.to_vec();
         let tools = self.get_prefixed_tools();
+
+        // Set the user_message field in the span instead of creating a new event
+        if let Some(content) = messages.last()
+            .and_then(|msg| msg.content.first())
+            .and_then(|c| c.as_text()) 
+        {
+            reply_span.record("user_message", &content);
+        }
+
         let system_prompt = self.get_system_prompt()?;
         let estimated_limit = self.provider.get_model_config().get_estimated_limit();
 
@@ -344,6 +368,8 @@ impl Agent {
             .await?;
 
         Ok(Box::pin(async_stream::try_stream! {
+            let _reply_guard = reply_span.enter();
+
             loop {
                 // Get completion from provider
                 let (response, usage) = self.provider.complete(
@@ -367,19 +393,26 @@ impl Agent {
                     .filter_map(|content| content.as_tool_request())
                     .collect();
 
+                let request_count = tool_requests.len();
                 if tool_requests.is_empty() {
-                    // No more tool calls, end the reply loop
                     break;
                 }
 
                 // Then dispatch each in parallel
                 let futures: Vec<_> = tool_requests
                     .iter()
-                    .map(|request| self.dispatch_tool_call(request.tool_call.clone()))
+                    .map(|request| {
+                        self.dispatch_tool_call(request.tool_call.clone())
+                    })
                     .collect();
 
+
+
                 // Process all the futures in parallel but wait until all are finished
-                let outputs = futures::future::join_all(futures).await;
+                let outputs = futures::future::join_all(futures)
+                    .await;
+
+                debug!("All tool requests completed");
 
                 // Create a message with the responses
                 let mut message_tool_response = Message::user();

crates/goose/src/lib.rs

@@ -9,3 +9,5 @@ pub mod prompt_template;
 pub mod providers;
 pub mod systems;
 pub mod token_counter;
+pub mod tracing;
+pub mod logging;

crates/goose/src/logging.rs

@@ -0,0 +1,100 @@
+use anyhow::{Context, Result};
+use serde_json::Value;
+use std::fs;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tracing_appender::rolling::Rotation;
+use tracing_subscriber::{fmt, layer::SubscriberExt, EnvFilter, Registry};
+use tracing::dispatcher::set_global_default;
+
+use crate::tracing::{langfuse_layer, observation_layer::{BatchManager, ObservationLayer, SpanTracker}};
+
+struct ConsoleLogger {
+    batch: Vec<Value>,
+}
+
+impl ConsoleLogger {
+    fn new() -> Self {
+        Self {
+            batch: Vec::new(),
+        }
+    }
+}
+
+impl BatchManager for ConsoleLogger {
+    fn add_event(&mut self, _event_type: &str, body: Value) {
+        self.batch.push(body);
+    }
+
+    fn send(&mut self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.batch.clear();
+        Ok(())
+    }
+}
+
+fn get_log_directory() -> Result<PathBuf> {
+    let home = std::env::var("HOME").context("HOME environment variable not set")?;
+    let base_log_dir = PathBuf::from(home).join(".config").join("goose").join("logs");
+
+    // Create date-based subdirectory
+    let now = chrono::Local::now();
+    let date_dir = base_log_dir.join(now.format("%Y-%m-%d").to_string());
+
+    // Ensure log directory exists
+    fs::create_dir_all(&date_dir).context("Failed to create log directory")?;
+
+    Ok(date_dir)
+}
+
+fn create_observation_layer() -> ObservationLayer {
+    let batch_manager = Arc::new(Mutex::new(ConsoleLogger::new()));
+    ObservationLayer {
+        batch_manager,
+        span_tracker: Arc::new(Mutex::new(SpanTracker::new())),
+    }
+}
+
+pub fn setup_logging() -> Result<()> {
+    // Set up file appender for goose module logs
+    let log_dir = get_log_directory()?;
+    let timestamp = chrono::Local::now().format("%Y%m%d_%H%M%S%.3f").to_string();
+
+    // Create non-rolling file appender
+    let file_appender = tracing_appender::rolling::RollingFileAppender::new(
+        Rotation::NEVER,
+        log_dir,
+        &format!("{}.log", timestamp),
+    );
+
+    // Create JSON file logging layer
+    let file_layer = fmt::layer()
+        .with_target(true)
+        .with_level(true)
+        .with_writer(file_appender)
+        .with_ansi(false)
+        .with_file(true);
+
+    // Update filter to include debug level
+    let filter = EnvFilter::try_from_default_env()
+        .unwrap_or_else(|_| EnvFilter::new("goose=debug"));
+
+    // Build the base subscriber
+    let subscriber = Registry::default()
+        .with(file_layer)
+        .with(filter)
+        .with(create_observation_layer());
+
+    // Set up the dispatcher
+    let dispatcher = if let Some(langfuse) = langfuse_layer::create_langfuse_observer() {
+        subscriber.with(langfuse).into()
+    } else {
+        subscriber.into()
+    };
+
+    // Set the subscriber as the default
+    set_global_default(dispatcher)
+        .map_err(|e| anyhow::anyhow!("Failed to set global subscriber: {}", e))?;
+
+    Ok(())
+}

crates/goose/src/providers/databricks.rs

@@ -105,6 +105,7 @@ impl DatabricksProvider {
 
 #[async_trait]
 impl Provider for DatabricksProvider {
+    #[tracing::instrument(skip(self, system, messages, tools), fields(model_config, input, output, input_tokens, output_tokens, total_tokens, cost))]
     async fn complete(
         &self,
         system: &str,
@@ -147,8 +148,7 @@ impl Provider for DatabricksProvider {
                 .collect(),
         );
 
-        // Make request
-        let response = self.post(payload).await?;
+        let response = self.post(payload.clone()).await?;
 
         // Raise specific error if context length is exceeded
         if let Some(error) = response.get("error") {
@@ -165,7 +165,15 @@ impl Provider for DatabricksProvider {
         let usage = Self::get_usage(&response)?;
         let model = get_model(&response);
         let cost = cost(&usage, &model_pricing_for(&model));
-
+        tracing::debug!(
+            model_config = %serde_json::to_string_pretty(&self.config).unwrap_or_default(),
+            input = %serde_json::to_string_pretty(&payload).unwrap_or_default(),
+            output = %serde_json::to_string_pretty(&response).unwrap_or_default(),
+            input_tokens = ?usage.input_tokens.unwrap_or_default(),
+            output_tokens = ?usage.output_tokens.unwrap_or_default(),
+            total_tokens = ?usage.total_tokens.unwrap_or_default(),
+            cost = ?cost.unwrap_or_default()
+        );
         Ok((message, ProviderUsage::new(model, usage, cost)))
     }