diff --git a/docs/user/getting-started-and-configuration.md b/docs/user/getting-started-and-configuration.md index 8289f8f..24c0a8d 100644 --- a/docs/user/getting-started-and-configuration.md +++ b/docs/user/getting-started-and-configuration.md @@ -157,6 +157,7 @@ provider = "openai-subscription" # or "openrouter" or "speakeasy" model = "gpt-5.4" reasoning_effort = "medium" # low, medium, or high a2a = "127.0.0.1:7331" +capture_error_spans = false # optional local context in fatal error logs otel_endpoint = "http://localhost:4317" otel_protocol = "grpc" # grpc, http/protobuf, or http/json otel_capture_message_content = false @@ -210,6 +211,14 @@ For settings exposed by a command, precedence is: 2. values in `~/.kit/config.toml`; 3. built-in defaults. +Set `capture_error_spans = true` when troubleshooting unexpected failures or +preparing a bug report. Kit adds diagnostic context about recent operations to +error logs in `~/.kit/errors//`, which can help explain a failure. + +It is disabled by default to avoid additional collection overhead and does not +require OpenTelemetry. The extra context excludes prompts and tool inputs and +outputs; it is not a complete execution history. + The OpenTelemetry endpoint follows the same CLI-over-TOML precedence, then falls back to the standard `OTEL_EXPORTER_OTLP_ENDPOINT` environment variable. If none is set, trace export is disabled. The trace protocol precedence is diff --git a/src/fatal.rs b/src/fatal.rs index dd0e5e3..996e24d 100644 --- a/src/fatal.rs +++ b/src/fatal.rs @@ -50,6 +50,22 @@ struct FatalRecord { message: String, #[serde(default, skip_serializing_if = "Option::is_none")] diagnostics: Option, + #[serde( + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "deserialize_span_context" + )] + span_context: Option, +} + +fn deserialize_span_context<'de, D: serde::Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + let snapshot = Option::::deserialize(deserializer)?; + if snapshot.as_ref().is_some_and(|snapshot| !snapshot.valid()) { + return Err(serde::de::Error::custom("invalid span context")); + } + Ok(snapshot) } #[derive(Clone, Copy, Debug, Deserialize, Serialize)] @@ -447,7 +463,7 @@ fn write_in_with_diagnostics( std::process::id(), NEXT_EVENT.fetch_add(1, Ordering::Relaxed) ); - let record = FatalRecord { + let mut record = FatalRecord { schema_version: SCHEMA_VERSION, event_id: event_id.clone(), occurred_at_ms, @@ -458,9 +474,15 @@ fn write_in_with_diagnostics( code: canonical_code(code).into(), message: bounded(message), diagnostics: diagnostics.filter(|value| value.valid()).cloned(), + span_context: crate::telemetry::error_spans::snapshot(&tracing::Span::current()), }; let mut bytes = serde_json::to_vec_pretty(&record) .map_err(|error| format!("could not encode fatal error log: {error}"))?; + if bytes.len() >= MAX_RECORD_BYTES && record.span_context.take().is_some() { + // Optional diagnostics must not displace an otherwise valid ordinary error. + bytes = serde_json::to_vec_pretty(&record) + .map_err(|error| format!("could not encode fatal error log: {error}"))?; + } bytes.push(b'\n'); if bytes.len() > MAX_RECORD_BYTES { return Err("fatal error log exceeds size limit".into()); @@ -620,6 +642,84 @@ mod tests { assert_eq!(record.code, "stream_transport"); } + #[test] + fn schema_two_readers_preserve_optional_span_context() { + use tracing_subscriber::prelude::*; + // Frozen shipped schema-v2 shape: unknown top-level fields are ignored. + #[derive(serde::Deserialize, serde::Serialize)] + struct ShippedV2 { + schema_version: u64, + event_id: String, + occurred_at_ms: u64, + kit_version: String, + session_id: String, + surface: String, + kind: String, + code: String, + message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + diagnostics: Option, + } + let root = tempfile::tempdir().unwrap(); + tracing::subscriber::with_default( + tracing_subscriber::registry().with(crate::telemetry::error_spans::ErrorSpanLayer), + || { + let operation = crate::telemetry::error_spans::operation("prompt"); + operation.in_scope(|| { + { let _child = tracing::info_span!(target: "agentkit_loop", "agent.execute_tool", launch_kind = "plain"); } + let path = write_in_with_diagnostics(root.path(), "session-context", Surface::Prompt, "provider", "stream_transport", "openai-subscription stream transport failed", Some(&sample_diagnostics())).unwrap(); + let bytes = fs::read(&path).unwrap(); + let mut value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value["schema_version"], 2); + assert_eq!(value["span_context"]["fragments"][1]["fields"]["launch_kind"], "plain"); + let current: FatalRecord = serde_json::from_slice(&bytes).unwrap(); + let legacy: ShippedV2 = serde_json::from_slice(&bytes).unwrap(); + let mut known = serde_json::to_value(¤t).unwrap(); + known.as_object_mut().unwrap().remove("span_context"); + assert_eq!(serde_json::to_value(&legacy).unwrap(), known); + assert_eq!(current.message, "openai-subscription stream transport failed"); + let supplied = value["span_context"].clone(); + for marker in [1, 2, 3, 4] { + value["schema_version"] = json!(marker); + let parsed: FatalRecord = serde_json::from_value(value.clone()).unwrap(); + assert_eq!(serde_json::to_value(parsed.span_context).unwrap(), supplied); + } + value.as_object_mut().unwrap().remove("span_context"); + assert!(serde_json::from_value::(value.clone()).unwrap().span_context.is_none()); + value["span_context"] = supplied; + value["span_context"]["fragments"][1]["fields"]["launch_kind"] = json!("SECRET"); + assert!(serde_json::from_value::(value).is_err()); + assert_eq!(fs::read(path).unwrap(), bytes); + assert!(bytes.len() < super::MAX_RECORD_BYTES); + }); + }, + ); + } + + #[test] + fn ordinary_writer_omits_disabled_context() { + tracing::subscriber::with_default(tracing_subscriber::registry(), || { + let root = tempfile::tempdir().unwrap(); + let operation = crate::telemetry::error_spans::operation("prompt"); + let path = operation + .in_scope(|| { + write_in( + root.path(), + "session-disabled", + Surface::Prompt, + "runtime", + "runtime_error", + "ordinary error", + ) + }) + .unwrap(); + let value: serde_json::Value = + serde_json::from_slice(&fs::read(path).unwrap()).unwrap(); + assert!(value.get("span_context").is_none()); + assert_eq!(value["message"], "ordinary error"); + }); + } + #[test] fn schema_v1_records_remain_readable() { let record: FatalRecord = serde_json::from_value(json!({ diff --git a/src/main.rs b/src/main.rs index 4d06877..7c0a466 100644 --- a/src/main.rs +++ b/src/main.rs @@ -52,6 +52,9 @@ fn resolve_openrouter_api_key( #[derive(Args)] struct TelemetryArgs { + /// Resolved local diagnostic setting inherited by built-in Kit children. + #[arg(long, hide = true, global = true, value_name = "BOOL", action = clap::ArgAction::Set)] + internal_capture_error_spans: Option, /// OTLP collector endpoint for OpenTelemetry trace export. #[arg(long, global = true)] otel_endpoint: Option, @@ -232,6 +235,7 @@ struct Config { provider: Option, reasoning_effort: Option, a2a: Option, + capture_error_spans: Option, otel_endpoint: Option, otel_protocol: Option, otel_capture_message_content: Option, @@ -409,6 +413,13 @@ impl Config { max_messages, max_bytes, ) + .map(|mut settings| { + settings.capture_error_spans = args + .internal_capture_error_spans + .or(self.capture_error_spans) + .unwrap_or(false); + settings + }) .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error)) } @@ -1679,6 +1690,61 @@ credential_store = "keychain" assert!(toml::from_str::("otel_protocol = 'http'").is_err()); } + #[test] + fn inherited_error_capture_overrides_config_without_rewriting_it() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("config.toml"); + for inherited in [false, true] { + let text = format!( + "# User-owned settings\ncapture_error_spans = {}\n", + !inherited + ); + fs::write(&path, &text).unwrap(); + let config = Config::load(&path).unwrap(); + let cli = Cli::try_parse_from([ + "kit", + "prompt", + "--internal-capture-error-spans", + &inherited.to_string(), + "hello", + ]) + .unwrap(); + let settings = config + .telemetry_settings(&cli.telemetry, None, None, None, None) + .unwrap(); + assert_eq!(settings.capture_error_spans, inherited); + assert_eq!(fs::read_to_string(&path).unwrap(), text); + } + } + + #[test] + fn error_span_capture_defaults_off_and_is_independent_of_export() { + let cli = Cli::try_parse_from(["kit", "prompt", "hello"]).unwrap(); + for (text, expected) in [ + ("", false), + ("capture_error_spans = false", false), + ("capture_error_spans = true", true), + ] { + let config: Config = toml::from_str(text).unwrap(); + for endpoint in [None, Some("http://localhost:4317".to_owned())] { + let settings = config + .telemetry_settings( + &cli.telemetry, + endpoint.clone(), + Some("true".into()), + None, + None, + ) + .unwrap(); + assert_eq!(settings.capture_error_spans, expected); + assert_eq!(settings.endpoint, endpoint); + assert!(settings.capture_message_content); + } + } + assert!(toml::from_str::("capture_error_spans = 'true'").is_err()); + assert!(toml::from_str::("capture_error_spans = 1").is_err()); + } + #[test] fn telemetry_environment_is_strict_and_settings_are_bounded() { let config = Config::default(); diff --git a/src/protocols/a2a.rs b/src/protocols/a2a.rs index b690c56..2c7e908 100644 --- a/src/protocols/a2a.rs +++ b/src/protocols/a2a.rs @@ -13,6 +13,7 @@ use a2a_protocol_types::{ }; use sha2::{Digest as _, Sha256}; +use tracing::Instrument as _; use crate::runtime::Runtime; @@ -24,67 +25,70 @@ impl AgentExecutor for KitAgent { context: &'a RequestContext, queue: &'a dyn EventQueueWriter, ) -> Pin> + Send + 'a>> { - Box::pin(async move { - let emit = EventEmitter::new(context, queue); - emit.status(TaskState::Working).await?; - let prompt = context - .message - .parts - .iter() - .filter_map(Part::text_content) - .collect::>() - .join("\n"); - if prompt.trim().is_empty() { - emit.artifact( - "error", - vec![Part::text("A2A request must contain a text part")], - None, - Some(true), - ) - .await?; - emit.status(TaskState::Failed).await?; - return Ok(()); - } - match self - .0 - .run_cancelled(prompt, 0, Some(context.cancellation_token.clone())) - .await - { - Ok(output) => { - emit.artifact("result", vec![Part::text(output)], None, Some(true)) - .await?; - emit.status(TaskState::Completed).await?; - } - Err(error) => { - let session_id = a2a_session_id(context); - let rendered = crate::fatal::render_loop_error(&error); - let rendered = match crate::fatal::record_loop_error( - &session_id, - crate::fatal::Surface::A2a, - &error, - ) { - Ok(Some(path)) => { - eprintln!( - "stored fatal error log for {session_id}: {}", - path.display() - ); - rendered - } - Ok(None) => rendered, - Err(log_error) => { - eprintln!( - "could not store fatal error log for {session_id}: {log_error}" - ); - rendered - } - }; - emit.artifact("error", vec![Part::text(rendered)], None, Some(true)) - .await?; + Box::pin( + async move { + let emit = EventEmitter::new(context, queue); + emit.status(TaskState::Working).await?; + let prompt = context + .message + .parts + .iter() + .filter_map(Part::text_content) + .collect::>() + .join("\n"); + if prompt.trim().is_empty() { + emit.artifact( + "error", + vec![Part::text("A2A request must contain a text part")], + None, + Some(true), + ) + .await?; emit.status(TaskState::Failed).await?; + return Ok(()); } + match self + .0 + .run_cancelled(prompt, 0, Some(context.cancellation_token.clone())) + .await + { + Ok(output) => { + emit.artifact("result", vec![Part::text(output)], None, Some(true)) + .await?; + emit.status(TaskState::Completed).await?; + } + Err(error) => { + let session_id = a2a_session_id(context); + let rendered = crate::fatal::render_loop_error(&error); + let rendered = match crate::fatal::record_loop_error( + &session_id, + crate::fatal::Surface::A2a, + &error, + ) { + Ok(Some(path)) => { + eprintln!( + "stored fatal error log for {session_id}: {}", + path.display() + ); + rendered + } + Ok(None) => rendered, + Err(log_error) => { + eprintln!( + "could not store fatal error log for {session_id}: {log_error}" + ); + rendered + } + }; + emit.artifact("error", vec![Part::text(rendered)], None, Some(true)) + .await?; + emit.status(TaskState::Failed).await?; + } + } + Ok(()) } - Ok(()) - }) + .instrument(crate::telemetry::error_spans::operation("a2a")), + ) } } diff --git a/src/protocols/acp.rs b/src/protocols/acp.rs index 4893ba6..bf01f4a 100644 --- a/src/protocols/acp.rs +++ b/src/protocols/acp.rs @@ -45,6 +45,7 @@ use tokio::{ task::{AbortHandle, JoinSet}, time::timeout, }; +use tracing::Instrument as _; mod activity; mod skill_catalog; @@ -1650,7 +1651,9 @@ async fn session_actor(actor: SessionActor) { &tasks, &background_jobs, structured_completion, - ), |reason| Some(reason.clone())).await; + ), |reason| Some(reason.clone())) + .instrument(crate::telemetry::error_spans::operation("acp")) + .await; let response = result.and_then(|reason| { agentkit_acp::finish_reason_to_stop_reason(&reason).map(PromptResponse::new) }); @@ -2080,6 +2083,7 @@ async fn drive_unsolicited( drive_finalized(session_id, integration, driver, false, None), |reason| Some(reason.clone()), ) + .instrument(crate::telemetry::error_spans::operation("acp_autonomous")) .await .map(|_| ()) } diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index eb923cb..548f894 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -26,6 +26,7 @@ use agentkit_loop::{ use agentkit_task_manager::{TaskEvent, TaskManagerHandle}; use async_trait::async_trait; use tokio::sync::{mpsc, oneshot, watch}; +use tracing::Instrument as _; use crate::{ provider::{ProviderKind, SelectableAdapter, authentication_method_id}, @@ -1160,6 +1161,7 @@ async fn session_actor(actor: SessionActor) &background_jobs, structured_completion, &activity,) + .instrument(crate::telemetry::error_spans::operation("acp")) .await; busy.store(false, Ordering::Release); if let Err(error) = result { @@ -1185,17 +1187,22 @@ async fn session_actor(actor: SessionActor) }, event = mcp_events.recv() => { if let Some(event) = event { - let result = match driver.submit_input(vec![Item::notification(event.message)]) { - Ok(()) => drive_autonomous( - &session_id, - &integration, - &handle, - &busy, - &mut driver, - &sink, - &activity,).await, - Err(error) => Err(map_loop_error(&session_id, &error)), - }; + let result = async { + match driver.submit_input(vec![Item::notification(event.message)]) { + Ok(()) => drive_autonomous( + &session_id, + &integration, + &handle, + &busy, + &mut driver, + &sink, + &activity, + ).await, + Err(error) => Err(map_loop_error(&session_id, &error)), + } + } + .instrument(crate::telemetry::error_spans::operation("acp_autonomous")) + .await; if let Err(error) = result { eprintln!("ACP v2 autonomous turn failed for {session_id}: {error}"); } @@ -1212,7 +1219,9 @@ async fn session_actor(actor: SessionActor) &busy, &mut driver, &sink, - &activity,).await + &activity,) + .instrument(crate::telemetry::error_spans::operation("acp_autonomous")) + .await { eprintln!("ACP v2 autonomous turn failed for {session_id}: {error}"); } diff --git a/src/runtime.rs b/src/runtime.rs index 6134f64..5e9ea4c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -30,6 +30,7 @@ use async_trait::async_trait; use serde_json::{Value, json}; use tokio::sync::watch; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::{ acp_child::{AcpHarnesses, BUILTIN_HARNESS, ChildConfig}, @@ -1007,11 +1008,20 @@ impl Runtime { } pub async fn run(self: &Arc, prompt: String, depth: usize) -> Result { - self.run_interruptible(prompt, depth, None).await + self.run_interruptible(prompt, depth, None) + .instrument(crate::telemetry::error_spans::operation("prompt")) + .await } /// Runs one prompt in the configured durable session. pub async fn run_persistent(self: &Arc, prompt: String) -> Result { + // Keep the operation current through startup and the existing fatal writes. + self.run_persistent_inner(prompt) + .instrument(crate::telemetry::error_spans::operation("prompt")) + .await + } + + async fn run_persistent_inner(self: &Arc, prompt: String) -> Result { let request = self .session .lock() diff --git a/src/runtime/tests.rs b/src/runtime/tests.rs index 37dc149..7d80cb5 100644 --- a/src/runtime/tests.rs +++ b/src/runtime/tests.rs @@ -1648,6 +1648,179 @@ async fn persistent_startup_failure_does_not_commit_new_session() { assert!(crate::session::load(root.path(), &session_id).is_err()); } +/// Run with a private HOME in a subprocess: fatal logs and durable sessions both +/// use HOME, and changing it in this process would race unrelated tests. +#[tokio::test] +async fn persistent_provider_failure_retains_private_span_context() { + use tracing::instrument::WithSubscriber as _; + use tracing_subscriber::{Layer, layer::SubscriberExt as _, registry::LookupSpan}; + + const CHILD: &str = "KIT_TEST_PERSISTENT_ERROR_SPANS"; + if std::env::var_os(CHILD).is_none() { + let home = tempfile::tempdir().unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let app = axum::Router::new().fallback(|| async { + ( + axum::http::StatusCode::BAD_REQUEST, + axum::Json(json!({"error": { + "message": "response-secret-sentinel", + "type": "invalid_request_error", + "code": 400 + }})), + ) + }); + let server = tokio::spawn( + async move { axum::serve(listener, app).await.unwrap() }.with_current_subscriber(), + ); + let mut command = tokio::process::Command::new(std::env::current_exe().unwrap()); + command + .args([ + "--exact", + "runtime::tests::persistent_provider_failure_retains_private_span_context", + "--nocapture", + ]) + .env(CHILD, "1") + .env("HOME", home.path()) + .env( + "OPENROUTER_BASE_URL", + format!("http://{address}/endpoint-secret-sentinel"), + ) + .env("NO_PROXY", "127.0.0.1") + .env_remove("OPENROUTER_MAX_COMPLETION_TOKENS") + .env_remove("OPENROUTER_TEMPERATURE") + .kill_on_drop(true); + let output = tokio::time::timeout(Duration::from_secs(30), command.output()) + .await + .expect("isolated runtime test timed out") + .unwrap(); + server.abort(); + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + return; + } + + struct ClosedBeforeLog { + directory: std::path::PathBuf, + names: Arc>>, + } + impl Layer for ClosedBeforeLog + where + S: tracing::Subscriber + for<'a> LookupSpan<'a>, + { + fn on_close(&self, id: tracing::span::Id, ctx: tracing_subscriber::layer::Context<'_, S>) { + let span = ctx.span(&id).unwrap(); + let name = span.metadata().name(); + if matches!(name, "chat" | "agent.turn") && !self.directory.exists() { + self.names.lock().unwrap().push(name); + } + } + } + + let home = std::path::PathBuf::from(std::env::var_os("HOME").unwrap()); + let root = tempfile::tempdir().unwrap(); + std::fs::write(root.path().join("AGENTS.md"), "context-secret-sentinel").unwrap(); + let mut original_error = None; + for (session_id, capture, block_log) in [ + ("span-enabled", true, false), + ("span-default-disabled", false, false), + ("span-write-failed", true, true), + ] { + let directory = home.join(".kit/errors").join(session_id); + if block_log { + // A file where the log directory belongs is a permanent write error. + std::fs::write(&directory, "not a directory").unwrap(); + } + let closed = Arc::new(std::sync::Mutex::new(Vec::new())); + let mut settings = crate::telemetry::Settings { + capture_message_content: true, + ..Default::default() + }; + assert!(!settings.capture_error_spans); + settings.capture_error_spans = capture; + assert!(settings.endpoint.is_none()); + let subscriber = tracing_subscriber::registry() + .with( + settings + .capture_error_spans + .then_some(crate::telemetry::error_spans::ErrorSpanLayer), + ) + .with(ClosedBeforeLog { + directory: directory.clone(), + names: closed.clone(), + }); + let runtime = Runtime::with_session_provider_credentials_effort_and_openrouter_key( + root.path(), + "private/model-secret-sentinel", + crate::provider::ProviderKind::OpenRouter, + SessionRequest { + id: session_id.into(), + resume: false, + force: false, + }, + crate::credentials::CredentialStorage::Memory, + None, + Some(crate::provider::OpenRouterApiKey::new( + "api-secret-sentinel", + )), + ) + .unwrap(); + let runtime = Runtime::with_telemetry(runtime, settings).unwrap(); + let error = runtime + .run_persistent("prompt-secret-sentinel".into()) + .with_subscriber(subscriber) + .await + .unwrap_err(); + let rendered = error.split("; fatal log: ").next().unwrap(); + assert!(rendered.starts_with("provider error:"), "{error}"); + if let Some(original) = &original_error { + assert_eq!(rendered, original); + } else { + original_error = Some(rendered.to_owned()); + } + if block_log { + assert!(!error.contains("; fatal log: ")); + assert_eq!( + std::fs::read_to_string(&directory).unwrap(), + "not a directory" + ); + continue; + } + let (_, path) = error.split_once("; fatal log: ").unwrap(); + let encoded = std::fs::read_to_string(path).unwrap(); + let record: Value = serde_json::from_str(&encoded).unwrap(); + assert_eq!(record["session_id"], session_id); + assert_eq!(record["surface"], "prompt"); + assert_eq!(record["kind"], "provider"); + assert_eq!(record["code"], "provider_error"); + assert_eq!(record["message"], "provider request failed"); + assert!(!encoded.contains("secret-sentinel"), "{encoded}"); + assert!(!encoded.contains(&root.path().display().to_string())); + assert!(!encoded.contains(&home.display().to_string())); + if capture { + let fragments = record["span_context"]["fragments"].as_array().unwrap(); + assert_eq!(fragments[0]["name"], "kit.operation"); + assert_eq!(fragments[0]["fields"]["surface"], "prompt"); + for name in ["agent.turn", "chat"] { + assert!( + closed.lock().unwrap().contains(&name), + "{name} did not close before logging" + ); + assert!( + fragments.iter().any(|fragment| fragment["name"] == name), + "{encoded}" + ); + } + } else { + assert!(record.get("span_context").is_none(), "{encoded}"); + } + } +} + #[tokio::test] async fn persistent_missing_openai_credentials_do_not_commit_new_session() { let root = tempfile::tempdir().unwrap(); diff --git a/src/telemetry.rs b/src/telemetry.rs index 8b06213..03c705f 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -1,4 +1,6 @@ -//! Optional OpenTelemetry trace export and resolved host settings. +//! Independent opt-in local error context and OpenTelemetry trace export. + +pub(crate) mod error_spans; use agentkit_loop::{MessageCapture, TelemetryConfig}; use opentelemetry::trace::TracerProvider as _; @@ -57,6 +59,8 @@ impl FromStr for Protocol { #[derive(Clone, Debug, PartialEq, Eq)] pub struct Settings { pub endpoint: Option, + /// Collect bounded local span context in fatal error logs (independent of OTLP). + pub capture_error_spans: bool, pub protocol: Protocol, pub capture_message_content: bool, pub message_content_max_messages: usize, @@ -104,6 +108,7 @@ impl Settings { (endpoint, Protocol::Grpc) | (endpoint @ None, _) => endpoint, }; Ok(Self { + capture_error_spans: false, endpoint, protocol, capture_message_content, @@ -143,6 +148,8 @@ impl Settings { /// re-enabling export or message capture. pub fn append_cli_args(&self, command: &mut tokio::process::Command) { command + .arg("--internal-capture-error-spans") + .arg(self.capture_error_spans.to_string()) .env_remove("OTEL_EXPORTER_OTLP_ENDPOINT") .env_remove("OTEL_EXPORTER_OTLP_PROTOCOL") .env_remove("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL") @@ -163,6 +170,7 @@ impl Default for Settings { fn default() -> Self { Self { endpoint: None, + capture_error_spans: false, protocol: Protocol::default(), capture_message_content: false, message_content_max_messages: DEFAULT_MESSAGE_CONTENT_MAX_MESSAGES, @@ -311,23 +319,44 @@ fn build_provider( } } -/// Installs OTLP trace export when an endpoint is configured. +/// Installs local span collection and OTLP export independently. pub fn init(settings: &Settings) -> Result, Box> { - let Some(endpoint) = settings.endpoint.as_deref() else { + if settings.endpoint.is_none() && !settings.capture_error_spans { return Ok(None); - }; - let exporter = build_exporter(endpoint, settings.protocol)?; - let provider = build_provider(exporter, settings.protocol); - let tracer = provider.tracer(env!("CARGO_PKG_NAME")); - let layer = tracing_opentelemetry::layer() - .with_tracer(tracer) - .with_location(false) - .with_threads(false) - .with_tracked_inactivity(false) - .with_target(false) - .with_filter(exported_targets()); - tracing_subscriber::registry().with(layer).try_init()?; - Ok(Some(Guard { + } + let provider = settings + .endpoint + .as_deref() + .map(|endpoint| { + build_exporter(endpoint, settings.protocol) + .map(|exporter| build_provider(exporter, settings.protocol)) + }) + .transpose()?; + let export_layer = provider.as_ref().map(|provider| { + tracing_opentelemetry::layer() + .with_tracer(provider.tracer(env!("CARGO_PKG_NAME"))) + .with_location(false) + .with_threads(false) + .with_tracked_inactivity(false) + .with_target(false) + .with_filter(exported_targets()) + }); + let result = tracing_subscriber::registry() + .with(export_layer) + .with( + settings + .capture_error_spans + .then_some(error_spans::ErrorSpanLayer), + ) + .try_init(); + // Local diagnostics are best-effort, including when a host owns the subscriber. + // Preserve the existing exporter initialization failure behavior. + if let Err(error) = result + && provider.is_some() + { + return Err(error.into()); + } + Ok(provider.map(|provider| Guard { provider: Some(provider), protocol: settings.protocol, })) @@ -538,6 +567,75 @@ mod tests { assert!(invalid.agentkit_config().is_err()); } + #[test] + fn child_args_propagate_both_local_capture_values() { + for enabled in [false, true] { + let settings = Settings { + capture_error_spans: enabled, + ..Settings::default() + }; + let mut command = tokio::process::Command::new("kit"); + settings.append_cli_args(&mut command); + let args: Vec<_> = command + .as_std() + .get_args() + .map(|value| value.to_string_lossy().into_owned()) + .collect(); + assert_eq!( + &args[..2], + &[ + "--internal-capture-error-spans".to_owned(), + enabled.to_string() + ] + ); + } + } + + #[test] + fn error_span_initialization_matrix() { + // init owns a global subscriber, so exercise each independent combination + // in a fresh test process rather than leaking state to parallel tests. + const CHILD: &str = "KIT_TEST_ERROR_SPAN_INIT"; + let Ok(mode) = std::env::var(CHILD) else { + for mode in ["00", "01", "10", "11", "occupied"] { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "telemetry::tests::error_span_initialization_matrix", + "--nocapture", + ]) + .env(CHILD, mode) + .output() + .unwrap(); + assert!( + output.status.success(), + "{mode}: {} {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + return; + }; + let runtime = tokio::runtime::Runtime::new().unwrap(); + let _runtime = runtime.enter(); + if mode == "occupied" { + tracing::subscriber::set_global_default(tracing_subscriber::registry()).unwrap(); + } + let settings = Settings { + capture_error_spans: mode.starts_with('1') || mode == "occupied", + endpoint: mode.ends_with('1').then(|| "http://127.0.0.1:1".into()), + ..Settings::default() + }; + let guard = super::init(&settings).unwrap(); + assert_eq!(guard.is_some(), settings.endpoint.is_some()); + let operation = super::error_spans::operation("prompt"); + operation.in_scope(|| { let _child = tracing::info_span!(target: "agentkit_loop", "chat", "gen_ai.operation.name" = "chat"); }); + assert_eq!( + super::error_spans::snapshot(&operation).is_some(), + settings.capture_error_spans && mode != "occupied" + ); + } + #[test] fn child_args_propagate_protocol_endpoint_explicit_false_and_bounds() { let settings = Settings::try_new_with_protocol( @@ -581,6 +679,8 @@ mod tests { assert_eq!( args, [ + "--internal-capture-error-spans", + "false", "--otel-endpoint", "http://collector:4318/v1/traces", "--otel-protocol", @@ -607,7 +707,7 @@ mod tests { .collect(); assert_eq!( - &args[..4], + &args[2..6], ["--otel-endpoint", "", "--otel-protocol", "grpc"] ); assert!( diff --git a/src/telemetry/error_spans.rs b/src/telemetry/error_spans.rs new file mode 100644 index 0000000..500c42a --- /dev/null +++ b/src/telemetry/error_spans.rs @@ -0,0 +1,426 @@ +//! Opt-in, operation-local tracing history for existing fatal diagnostics. +//! This is a partial history, not an effects ledger or a causal error chain. + +use std::{ + collections::BTreeMap, + fmt, + sync::{Arc, Mutex}, +}; + +use serde::{Deserialize, Serialize}; +use tracing::{ + Span, Subscriber, + field::{Field, Visit}, + span::{Attributes, Id, Record}, +}; +use tracing_subscriber::{Layer, Registry, layer::Context, registry::LookupSpan}; + +const MAX_FRAGMENTS: usize = 24; +const MAX_DEPTH: usize = 8; +const MAX_FIELDS: usize = 6; +const MAX_VALUE_BYTES: usize = 32; +const MAX_SNAPSHOT_BYTES: usize = 12 * 1024; +const TARGET: &str = "kit::telemetry::error_spans"; + +#[cfg(test)] +mod task_manager_tests; + +/// The future instrumented with this span must include execution AND error logging. +/// Each call starts a fresh history even when nested within another operation. +pub(crate) fn operation(surface: &'static str) -> Span { + tracing::info_span!(target: TARGET, parent: Span::current(), "kit.operation", surface) +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Snapshot { + fragments: Vec, + /// Indicates a collection bound, not whether observations are complete. + truncated: bool, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct Fragment { + name: String, + parent: Option, + fields: BTreeMap, +} + +impl Snapshot { + pub(crate) fn valid(&self) -> bool { + !self.fragments.is_empty() + && self.fragments.len() <= MAX_FRAGMENTS + && self.fragments.iter().enumerate().all(|(index, fragment)| { + (if index == 0 { + fragment.name == "kit.operation" && fragment.parent.is_none() + } else { + approved_name(&fragment.name) + && fragment.parent.is_some_and(|parent| parent < index) + }) && { + let mut parent = fragment.parent; + let mut depth = 0; + while let Some(index) = parent { + depth += 1; + if depth > MAX_DEPTH { + return false; + } + parent = self.fragments[index].parent; + } + true + } && fragment.fields.len() <= MAX_FIELDS + && fragment + .fields + .iter() + .all(|(key, value)| approved_value(key, value)) + }) + && serde_json::to_vec_pretty(self).is_ok_and(|bytes| bytes.len() <= MAX_SNAPSHOT_BYTES) + } +} + +#[derive(Clone)] +struct Capture { + history: Arc>, + index: usize, + depth: usize, +} + +pub(crate) struct ErrorSpanLayer; + +impl Layer for ErrorSpanLayer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) { + let Some(span) = ctx.span(id) else { return }; + let metadata = attrs.metadata(); + let root = metadata.target() == TARGET && metadata.name() == "kit.operation"; + let capture = if root { + Capture { + history: Arc::new(Mutex::new(Snapshot { + fragments: Vec::new(), + truncated: false, + })), + index: 0, + depth: 0, + } + } else { + let parent = if attrs.is_contextual() { + ctx.lookup_current() + } else { + attrs.parent().and_then(|parent| ctx.span(parent)) + }; + let Some(mut capture) = + parent.and_then(|parent| parent.extensions().get::().cloned()) + else { + return; + }; + capture.depth = capture.depth.saturating_add(1); + if capture.depth > MAX_DEPTH { + if let Ok(mut history) = capture.history.try_lock() { + history.truncated = true; + } + return; + } + capture + }; + let approved = + root || (metadata.target() == "agentkit_loop" && approved_name(metadata.name())); + let mut capture = capture; + if approved { + // No extension lock is held while locking the operation store. + let Ok(mut history) = capture.history.try_lock() else { + return; + }; + if history.fragments.len() < MAX_FRAGMENTS { + let mut visitor = Fields::default(); + attrs.record(&mut visitor); + let parent = (!root).then_some(capture.index); + capture.index = history.fragments.len(); + history.fragments.push(Fragment { + name: metadata.name().into(), + parent, + fields: visitor.values, + }); + history.truncated |= visitor.truncated; + } else { + history.truncated = true; + // Do not let records on an omitted span update its parent's fields. + return; + } + } + span.extensions_mut().insert(capture); + } + + fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) { + let Some(span) = ctx.span(id) else { return }; + let metadata = span.metadata(); + if !(metadata.target() == TARGET && metadata.name() == "kit.operation" + || metadata.target() == "agentkit_loop" && approved_name(metadata.name())) + { + return; + } + let capture = span.extensions().get::().cloned(); + let Some(capture) = capture else { return }; + let mut visitor = Fields::default(); + values.record(&mut visitor); + let Ok(mut history) = capture.history.try_lock() else { + return; + }; + let mut truncated = visitor.truncated; + if let Some(fragment) = history.fragments.get_mut(capture.index) { + for (key, value) in visitor.values { + if fragment.fields.contains_key(&key) || fragment.fields.len() < MAX_FIELDS { + fragment.fields.insert(key, value); + } else { + truncated = true; + } + } + } + history.truncated |= truncated; + } +} + +/// Uses the retained operation's extensions, including children already closed. +/// Without the layer there are no buffers and no traversal/serialization. +pub(crate) fn snapshot(span: &Span) -> Option { + span.with_subscriber(|(id, dispatch)| { + dispatch.downcast_ref::()?; + let registry = dispatch.downcast_ref::()?; + let span = registry.span(id)?; + let capture = span.extensions().get::().cloned()?; + let mut snapshot = capture.history.try_lock().ok()?.clone(); + // Keep encoding outside all locks. A failed/contended capture is optional. + while serde_json::to_vec_pretty(&snapshot).ok()?.len() > MAX_SNAPSHOT_BYTES { + snapshot.fragments.pop()?; + snapshot.truncated = true; + } + Some(snapshot) + }) + .flatten() +} + +fn approved_name(name: &str) -> bool { + matches!(name, "agent.turn" | "agent.execute_tool" | "chat") +} + +fn approved_value(key: &str, value: &serde_json::Value) -> bool { + match value { + serde_json::Value::String(value) if value.len() <= MAX_VALUE_BYTES => match key { + "surface" => matches!(value.as_str(), "prompt" | "a2a" | "acp" | "acp_autonomous"), + "gen_ai.operation.name" => { + matches!(value.as_str(), "invoke_agent" | "execute_tool" | "chat") + } + "launch_kind" => matches!(value.as_str(), "plain" | "approved"), + "error.type" => matches!(value.as_str(), "tool_error" | "provider_error"), + _ => false, + }, + serde_json::Value::Number(value) => { + matches!( + key, + "transcript.len" | "gen_ai.usage.input_tokens" | "gen_ai.usage.output_tokens" + ) && value.as_u64().is_some_and(|value| value <= u32::MAX.into()) + } + serde_json::Value::Bool(_) => key == "saw_tool_call", + _ => false, + } +} + +#[derive(Default)] +struct Fields { + values: BTreeMap, + truncated: bool, +} + +impl Fields { + fn insert(&mut self, field: &Field, value: serde_json::Value) { + if !approved_value(field.name(), &value) { + return; + } + if self.values.contains_key(field.name()) || self.values.len() < MAX_FIELDS { + self.values.insert(field.name().into(), value); + } else { + self.truncated = true; + } + } +} + +impl Visit for Fields { + fn record_str(&mut self, field: &Field, value: &str) { + // Reject before allocation; identifiers/content are intentionally not collected. + if value.len() <= MAX_VALUE_BYTES { + self.insert(field, value.into()); + } + } + fn record_u64(&mut self, field: &Field, value: u64) { + self.insert(field, value.into()); + } + fn record_i64(&mut self, field: &Field, value: i64) { + self.insert(field, value.into()); + } + fn record_bool(&mut self, field: &Field, value: bool) { + self.insert(field, value.into()); + } + fn record_debug(&mut self, _: &Field, _: &dyn fmt::Debug) { + // Includes Display wrappers: never format arbitrary user/provider payloads. + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tracing::Instrument as _; + use tracing_subscriber::prelude::*; + + #[test] + fn disabled_layer_has_no_capture() { + tracing::subscriber::with_default(tracing_subscriber::registry(), || { + let operation = operation("prompt"); + assert!(snapshot(&operation).is_none()); + operation.with_subscriber(|(id, dispatch)| { + let registry = dispatch.downcast_ref::().unwrap(); + assert!( + registry + .span(id) + .unwrap() + .extensions() + .get::() + .is_none() + ); + }); + }); + } + + #[test] + fn closed_children_and_late_records_survive_without_exporter() { + tracing::subscriber::with_default( + tracing_subscriber::registry().with(ErrorSpanLayer), + || { + let operation = operation("prompt"); + operation.in_scope(|| { + let child = tracing::info_span!(target: "agentkit_loop", "agent.execute_tool", + launch_kind = "plain", "error.type" = tracing::field::Empty); + child.record("error.type", "tool_error"); + }); + let context = snapshot(&operation).unwrap(); + assert!(context.valid()); + assert_eq!(context.fragments.len(), 2); + assert_eq!(context.fragments[1].parent, Some(0)); + assert_eq!(context.fragments[1].fields["error.type"], "tool_error"); + assert_eq!(context.fragments[1].fields["launch_kind"], "plain"); + }, + ); + } + + #[test] + fn privacy_rejects_content_identifiers_debug_and_untrusted_targets() { + struct NeverFormat; + impl fmt::Debug for NeverFormat { + fn fmt(&self, _: &mut fmt::Formatter<'_>) -> fmt::Result { + panic!("must not format"); + } + } + tracing::subscriber::with_default( + tracing_subscriber::registry().with(ErrorSpanLayer), + || { + let operation = operation("prompt"); + operation.in_scope(|| { + let child = tracing::info_span!(target: "agentkit_loop", "chat", + "gen_ai.input.messages" = ?NeverFormat, + "gen_ai.output.messages" = "SECRET", + "gen_ai.conversation.id" = ?NeverFormat, + "gen_ai.operation.name" = "chat", + "gen_ai.usage.input_tokens" = u64::MAX, + "error.type" = ?NeverFormat); + child.record("gen_ai.output.messages", "SECRET".repeat(100_000)); + child.record("gen_ai.operation.name", "https://secret.invalid/token"); + child.record("gen_ai.conversation.id", "../../SECRET"); + let _untrusted = tracing::info_span!(target: "untrusted", "chat", "gen_ai.operation.name" = "chat"); + }); + let context = snapshot(&operation).unwrap(); + assert_eq!(context.fragments.len(), 2); + assert_eq!(context.fragments[1].fields.len(), 1); + assert_eq!(context.fragments[1].fields["gen_ai.operation.name"], "chat"); + let encoded = serde_json::to_string(&context).unwrap(); + assert!(!encoded.contains("SECRET")); + assert!(!encoded.contains("secret.invalid")); + }, + ); + } + + #[test] + fn bounds_depth_count_and_snapshot_size() { + tracing::subscriber::with_default( + tracing_subscriber::registry().with(ErrorSpanLayer), + || { + let operation = operation("prompt"); + operation.in_scope(|| { + for _ in 0..1000 { + let _child = tracing::info_span!(target: "agentkit_loop", "chat", "gen_ai.operation.name" = "chat"); + } + }); + let context = snapshot(&operation).unwrap(); + assert_eq!(context.fragments.len(), MAX_FRAGMENTS); + assert!(context.truncated); + assert!(context.valid()); + + let deep = super::operation("acp"); + let mut parent = deep.clone(); + for _ in 0..100 { + parent = tracing::info_span!(target: "agentkit_loop", parent: &parent, "chat"); + } + let context = snapshot(&deep).unwrap(); + assert!(context.truncated); + assert_eq!(context.fragments.len(), MAX_DEPTH + 1); + }, + ); + } + + #[test] + fn unavailable_capture_is_omitted() { + tracing::subscriber::with_default( + tracing_subscriber::registry().with(ErrorSpanLayer), + || { + let operation = operation("prompt"); + operation.with_subscriber(|(id, dispatch)| { + let registry = dispatch.downcast_ref::().unwrap(); + let capture = registry + .span(id) + .unwrap() + .extensions() + .get::() + .unwrap() + .clone(); + let _lock = capture.history.lock().unwrap(); + assert!(snapshot(&operation).is_none()); + operation.record("surface", "a2a"); // contended collection cannot block or fail + }); + assert!(snapshot(&operation).is_some()); + }, + ); + } + + #[tokio::test] + async fn explicitly_instrumented_spawns_keep_separate_operation_histories() { + use tracing::instrument::WithSubscriber as _; + let subscriber = tracing_subscriber::registry().with(ErrorSpanLayer); + async { + let first = operation("prompt"); + let second = operation("a2a"); + let one = tokio::spawn(async { + tokio::task::yield_now().await; + let _child = tracing::info_span!(target: "agentkit_loop", "chat", "gen_ai.operation.name" = "chat"); + }.instrument(first.clone()).with_current_subscriber()); + let two = tokio::spawn(async { + let _child = tracing::info_span!(target: "agentkit_loop", "agent.execute_tool", launch_kind = "approved"); + tokio::task::yield_now().await; + }.instrument(second.clone()).with_current_subscriber()); + one.await.unwrap(); + two.await.unwrap(); + assert_eq!(snapshot(&first).unwrap().fragments[1].name, "chat"); + assert_eq!(snapshot(&second).unwrap().fragments[1].name, "agent.execute_tool"); + assert_eq!(snapshot(&first).unwrap().fragments.len(), 2); + assert_eq!(snapshot(&second).unwrap().fragments.len(), 2); + }.with_subscriber(subscriber).await; + } +} diff --git a/src/telemetry/error_spans/task_manager_tests.rs b/src/telemetry/error_spans/task_manager_tests.rs new file mode 100644 index 0000000..181bbd8 --- /dev/null +++ b/src/telemetry/error_spans/task_manager_tests.rs @@ -0,0 +1,207 @@ +//! Real async-manager execution documents the upstream spawn boundary without +//! wrapping or patching it. Set KIT_TEST_REQUIRE_TASK_MANAGER_ANCESTRY=1 to turn +//! the known limitation assertion into the desired (currently failing) contract. + +use std::{ + collections::VecDeque, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, +}; + +use agentkit_core::{ + FinishReason, Item, ItemKind, MetadataMap, Part, SessionId, ToolCallPart, ToolOutput, + ToolResultPart, TurnCancellation, +}; +use agentkit_loop::{ + Agent, LoopError, ModelAdapter, ModelSession, ModelTurn, ModelTurnEvent, ModelTurnResult, + SessionConfig, TurnRequest, +}; +use agentkit_task_manager::AsyncTaskManager; +use agentkit_tools_core::{ + Tool, ToolContext, ToolError, ToolName, ToolRegistry, ToolRequest, ToolResult, ToolSpec, +}; +use async_trait::async_trait; +use serde_json::json; +use tracing::Instrument; +use tracing_subscriber::prelude::*; + +use super::{ErrorSpanLayer, operation, snapshot}; + +struct FixtureAdapter; +struct FixtureSession(bool); +struct FixtureTurn(VecDeque); + +#[async_trait] +impl ModelAdapter for FixtureAdapter { + type Session = FixtureSession; + + async fn start_session(&self, _: SessionConfig) -> Result { + Ok(FixtureSession(false)) + } +} + +#[async_trait] +impl ModelSession for FixtureSession { + type Turn = FixtureTurn; + + async fn begin_turn( + &mut self, + _: TurnRequest, + _: Option, + ) -> Result { + if self.0 { + return Err(LoopError::Provider("fixture final failure".into())); + } + self.0 = true; + let call = ToolCallPart::new("probe-call", "probe", json!({})); + Ok(FixtureTurn(VecDeque::from([ + ModelTurnEvent::ToolCall(call.clone()), + ModelTurnEvent::Finished(ModelTurnResult { + finish_reason: FinishReason::ToolCall, + output_items: vec![Item::new(ItemKind::Assistant, vec![Part::ToolCall(call)])], + usage: None, + metadata: MetadataMap::new(), + model: None, + response_id: None, + }), + ]))) + } +} + +#[async_trait] +impl ModelTurn for FixtureTurn { + async fn next_event( + &mut self, + _: Option, + ) -> Result, LoopError> { + Ok(self.0.pop_front()) + } +} + +struct Probe { + spec: ToolSpec, + executed: Arc, +} + +#[async_trait] +impl Tool for Probe { + fn spec(&self) -> &ToolSpec { + &self.spec + } + + async fn invoke( + &self, + request: ToolRequest, + _: &mut ToolContext<'_>, + ) -> Result { + // Same allowlisted shape as the control span, but a distinct numeric + // marker so the loop's own inference spans cannot satisfy the assertion. + let span = tracing::info_span!(target: "agentkit_loop", "chat", + gen_ai.operation.name = "chat", gen_ai.usage.output_tokens = 4242_u64); + assert!( + !span.is_disabled(), + "global subscriber must reach the spawned task" + ); + async { + self.executed.store(true, Ordering::SeqCst); + Ok(ToolResult::new(ToolResultPart::success( + request.call_id, + ToolOutput::text("probe completed"), + ))) + } + .instrument(span) + .await + } +} + +#[tokio::test] +async fn async_manager_does_not_retain_invocation_ancestry() { + const TEST: &str = "telemetry::error_spans::task_manager_tests::async_manager_does_not_retain_invocation_ancestry"; + const CHILD: &str = "KIT_TASK_MANAGER_ANCESTRY_TEST_CHILD"; + if std::env::var(CHILD).as_deref() != Ok(TEST) { + // A global subscriber in a fresh process tests span propagation, not + // the separate failure to propagate a thread-local default subscriber. + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", TEST, "--nocapture"]) + .env(CHILD, TEST) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + String::from_utf8_lossy(&output.stdout).contains("running 1 test"), + "isolated child must run the exact ancestry test" + ); + return; + } + tracing_subscriber::registry().with(ErrorSpanLayer).init(); + let executed = Arc::new(AtomicBool::new(false)); + let tools = ToolRegistry::new().with(Probe { + spec: ToolSpec::new(ToolName::new("probe"), "probe", json!({"type": "object"})), + executed: executed.clone(), + }); + let agent = Agent::builder() + .model(FixtureAdapter) + .add_tool_source(tools) + // Default routing is foreground: detachment is not needed to lose ancestry. + .task_manager(AsyncTaskManager::new()) + .input(vec![Item::text(ItemKind::User, "probe")]) + .build() + .unwrap(); + let root = operation("prompt"); + let error = async { + tracing::info_span!(target: "agentkit_loop", "chat", + gen_ai.operation.name = "chat", gen_ai.usage.output_tokens = 4241_u64) + .in_scope(|| {}); + let mut driver = agent + .start(SessionConfig::new(SessionId::new("ancestry-test")).without_cache()) + .await + .unwrap(); + for _ in 0..8 { + if let Err(error) = driver.next().await { + return error; + } + } + panic!("fixture did not reach the final provider failure"); + } + .instrument(root.clone()) + .await; + assert!(matches!(error, LoopError::Provider(message) if message == "fixture final failure")); + assert!(executed.load(Ordering::SeqCst)); + let snapshot = snapshot(&root).expect("operation retains history at the actual error boundary"); + assert!(snapshot.valid()); + assert!(!snapshot.truncated); + assert!( + snapshot + .fragments + .iter() + .any(|fragment| fragment.name == "agent.execute_tool") + ); + let has_marker = |marker| { + snapshot.fragments.iter().any(|fragment| { + fragment.name == "chat" + && fragment.fields.get("gen_ai.usage.output_tokens") == Some(&json!(marker)) + }) + }; + assert!( + has_marker(4241), + "allowlisted control span must be captured" + ); + if std::env::var("KIT_TEST_REQUIRE_TASK_MANAGER_ANCESTRY").as_deref() == Ok("1") { + assert!( + has_marker(4242), + "actual tool invocation lost its dispatch/operation ancestry" + ); + } else { + assert!( + !has_marker(4242), + "upstream propagation changed; revisit the documented scope" + ); + } +} diff --git a/src/tools/observed.rs b/src/tools/observed.rs index bf60407..f67674e 100644 --- a/src/tools/observed.rs +++ b/src/tools/observed.rs @@ -9,7 +9,8 @@ use std::{sync::Arc, time::Instant}; use agentkit_core::ToolOutput; use agentkit_tools_core::{ - PermissionRequest, Tool, ToolContext, ToolError, ToolRequest, ToolResult, ToolSpec, + PermissionRequest, Tool, ToolContext, ToolError, ToolExecutionOutcome, ToolRequest, ToolResult, + ToolSpec, }; use async_trait::async_trait; @@ -57,6 +58,14 @@ impl Tool for SharedTool { ) -> Result { self.0.invoke(request, context).await } + + async fn invoke_outcome( + &self, + request: ToolRequest, + context: &mut ToolContext<'_>, + ) -> ToolExecutionOutcome { + self.0.invoke_outcome(request, context).await + } } #[async_trait] @@ -81,8 +90,44 @@ impl Tool for Observed { request: ToolRequest, context: &mut ToolContext<'_>, ) -> Result { + let display = DisplayInvocation::start(&request); + let outcome = self.0.invoke(request, context).await; + if let Some(display) = display { + display.finish(outcome.as_ref()); + } + outcome + } + + async fn invoke_outcome( + &self, + request: ToolRequest, + context: &mut ToolContext<'_>, + ) -> ToolExecutionOutcome { + let display = DisplayInvocation::start(&request); + let outcome = self.0.invoke_outcome(request, context).await; + if let Some(display) = display { + match &outcome { + ToolExecutionOutcome::Completed(result) => display.finish(Ok(result)), + ToolExecutionOutcome::Failed(error) + | ToolExecutionOutcome::FailedBeforeInvocation(error) => display.finish(Err(error)), + // An approval interruption is not a completed invocation. + ToolExecutionOutcome::Interrupted(_) => {} + } + } + outcome + } +} + +struct DisplayInvocation { + call: String, + tool: String, + started: Instant, +} + +impl DisplayInvocation { + fn start(request: &ToolRequest) -> Option { if !events::enabled() { - return self.0.invoke(request, context).await; + return None; } let call = request.call_id.0.clone(); let tool = request.tool_name.0.to_string(); @@ -92,9 +137,15 @@ impl Tool for Observed { summary: summarize_input(&request.input), at: events::now_millis(), }); - let started = Instant::now(); - let outcome = self.0.invoke(request, context).await; - let (ok, summary) = match &outcome { + Some(Self { + call, + tool, + started: Instant::now(), + }) + } + + fn finish(self, result: Result<&ToolResult, &ToolError>) { + let (ok, summary) = match result { Ok(result) => ( !result.result.is_error, summarize_output(&output_value(&result.result.output)), @@ -102,13 +153,12 @@ impl Tool for Observed { Err(error) => (false, summarize_output(&json!(error.to_string()))), }; events::emit(&RuntimeEvent::ChildFinished { - call, - tool, + call: self.call, + tool: self.tool, ok, summary, - millis: u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), + millis: u64::try_from(self.started.elapsed().as_millis()).unwrap_or(u64::MAX), }); - outcome } } @@ -120,3 +170,129 @@ fn output_value(output: &ToolOutput) -> Value { ToolOutput::Files(files) => json!(format!("{} files", files.len())), } } + +#[cfg(test)] +mod tests { + use super::*; + use agentkit_core::{MetadataMap, SessionId, ToolCallId, ToolResultPart, TurnId}; + use agentkit_tools_core::{ + AllowAllPermissions, ApprovalReason, ApprovalRequest, OwnedToolContext, ToolInterruption, + ToolName, + }; + + #[derive(Clone, Copy, Debug)] + enum Mode { + Completed, + Failed, + FailedBeforeInvocation, + Cancelled, + Interrupted, + } + + struct NativeTool { + spec: ToolSpec, + mode: Mode, + } + + #[async_trait] + impl Tool for NativeTool { + fn spec(&self) -> &ToolSpec { + &self.spec + } + + async fn invoke( + &self, + _: ToolRequest, + _: &mut ToolContext<'_>, + ) -> Result { + panic!("wrapper must forward invoke_outcome, not use the invoke fallback") + } + + async fn invoke_outcome( + &self, + request: ToolRequest, + _: &mut ToolContext<'_>, + ) -> ToolExecutionOutcome { + match self.mode { + Mode::Completed => ToolExecutionOutcome::Completed(ToolResult::new( + ToolResultPart::success(request.call_id, ToolOutput::text("done")), + )), + Mode::Failed => { + ToolExecutionOutcome::Failed(ToolError::ExecutionFailed("failed".into())) + } + Mode::FailedBeforeInvocation => ToolExecutionOutcome::FailedBeforeInvocation( + ToolError::Unavailable("not started".into()), + ), + Mode::Cancelled => ToolExecutionOutcome::Failed(ToolError::Cancelled), + Mode::Interrupted => ToolExecutionOutcome::Interrupted( + ToolInterruption::ApprovalRequired(ApprovalRequest::new( + "approval", + "native", + ApprovalReason::PolicyRequiresConfirmation, + "approval", + )), + ), + } + } + } + + #[tokio::test] + async fn both_wrappers_preserve_native_outcomes() { + for mode in [ + Mode::Completed, + Mode::Failed, + Mode::FailedBeforeInvocation, + Mode::Cancelled, + Mode::Interrupted, + ] { + for dynamic in [false, true] { + let native = NativeTool { + spec: ToolSpec::new(ToolName::new("native"), "native", json!({})), + mode, + }; + let tool: Box = if dynamic { + Box::new(shared(Arc::new(native))) + } else { + Box::new(Observed::new(native)) + }; + let context = OwnedToolContext { + session_id: SessionId::new("session"), + turn_id: TurnId::new("turn"), + metadata: MetadataMap::new(), + permissions: Arc::new(AllowAllPermissions), + resources: Arc::new(()), + cancellation: None, + execution_scope: None, + approved_request: None, + }; + let request = ToolRequest::new( + ToolCallId::new("call"), + ToolName::new("native"), + json!({}), + context.session_id.clone(), + context.turn_id.clone(), + ); + let outcome = tool.invoke_outcome(request, &mut context.borrowed()).await; + let preserved = match (mode, outcome) { + (Mode::Completed, ToolExecutionOutcome::Completed(result)) => { + result.result.output == ToolOutput::text("done") + } + ( + Mode::Failed, + ToolExecutionOutcome::Failed(ToolError::ExecutionFailed(message)), + ) => message == "failed", + ( + Mode::FailedBeforeInvocation, + ToolExecutionOutcome::FailedBeforeInvocation(ToolError::Unavailable( + message, + )), + ) => message == "not started", + (Mode::Cancelled, ToolExecutionOutcome::Failed(ToolError::Cancelled)) => true, + (Mode::Interrupted, ToolExecutionOutcome::Interrupted(_)) => true, + _ => false, + }; + assert!(preserved, "{mode:?}, shared={dynamic}"); + } + } + } +} diff --git a/src/tools/subagent.rs b/src/tools/subagent.rs index 4872986..8e989e0 100644 --- a/src/tools/subagent.rs +++ b/src/tools/subagent.rs @@ -12,6 +12,7 @@ use async_trait::async_trait; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use tokio::sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, Semaphore, oneshot}; +use tracing::Instrument as _; const MAX_LIVE_SUBAGENTS: usize = 120; const MAX_DISPLAY_NAME_LEN: usize = 32; @@ -608,18 +609,21 @@ impl Subagents { let (reply, response) = oneshot::channel(); let manager = self.clone(); - tokio::spawn(async move { - let source_state = Arc::clone(&operation.source_state); - let reservation = operation.id.clone(); - let result = manager.run_fork(operation, &reply).await; - manager.finish_forking(&source_state, &reservation).await; - match result { - Ok(value) => manager.handoff_fork_success(reply, value).await, - Err(error) => { - let _ = reply.send(Err(error)); + tokio::spawn( + async move { + let source_state = Arc::clone(&operation.source_state); + let reservation = operation.id.clone(); + let result = manager.run_fork(operation, &reply).await; + manager.finish_forking(&source_state, &reservation).await; + match result { + Ok(value) => manager.handoff_fork_success(reply, value).await, + Err(error) => { + let _ = reply.send(Err(error)); + } } } - }); + .instrument(tracing::Span::current()), + ); match response.await.map_err(|_| { ChildError::Failed("subagent fork task stopped before returning a result".into()) })? { @@ -1087,11 +1091,14 @@ impl Subagents { error: ChildError, ) -> ChildError { let manager = self.clone(); - match tokio::spawn(async move { - manager - .cleanup_installed_child(&id, &state, &child, error) - .await - }) + match tokio::spawn( + async move { + manager + .cleanup_installed_child(&id, &state, &child, error) + .await + } + .instrument(tracing::Span::current()), + ) .await { Ok(error) => error, diff --git a/src/tui/mod.rs b/src/tui/mod.rs index f06fdce..d4c33ab 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -4105,6 +4105,8 @@ a = [still text] assert_eq!( args, [ + "--internal-capture-error-spans", + "false", "--otel-endpoint", "http://collector:4318/v1/traces", "--otel-protocol", diff --git a/tests/runtime.rs b/tests/runtime.rs index d5b1b9e..d49cdaa 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -1,6 +1,9 @@ use std::{ collections::BTreeMap, - sync::Arc, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, time::{Duration, Instant}, }; @@ -626,6 +629,13 @@ async fn execute_compose_cancelled( script: &str, cancellation: Option, ) -> ToolExecutionOutcome { + // HOME/session/call-scoped artifacts must not share a spill directory with + // another parallel invocation that may remove it during cleanup. + static NEXT_CALL: AtomicUsize = AtomicUsize::new(0); + let call_id = ToolCallId::new(format!( + "compose-test-{}", + NEXT_CALL.fetch_add(1, Ordering::Relaxed) + )); let source: Arc = Arc::new(runtime.compose(0)); let executor = Arc::new(BasicToolExecutor::new([source])); let scope = ToolExecutionScope { @@ -638,7 +648,7 @@ async fn execute_compose_cancelled( }; scope .execute_child(ToolRequest { - call_id: ToolCallId::new("compose-test"), + call_id, tool_name: ToolName::new("compose"), input: json!({"script": script}), session_id: SessionId::new("test"),