From b54578cb2e4ff20c8eb964b83093994796ca8a26 Mon Sep 17 00:00:00 2001 From: Jiangtian Feng Date: Sat, 27 Jun 2026 17:09:41 +0800 Subject: [PATCH] feat(sight): agent crash full-stack observability BPF exit_code capture + classify_exit replaces pending-calls heuristic. Lineage tree, enriched crash detail, CLI crash report, pid cache fix. --- .../scripts/check-arch-boundaries.py | 1 + src/agentsight/src/background.rs | 6 +- src/agentsight/src/bin/cli/interruption.rs | 19 +- src/agentsight/src/bin/cli/token.rs | 54 +- src/agentsight/src/bpf/procmon.bpf.c | 7 +- src/agentsight/src/bpf/procmon.h | 1 + src/agentsight/src/bpf/proctrace.bpf.c | 3 +- src/agentsight/src/genai/builder.rs | 2 + src/agentsight/src/genai/call_builder.rs | 1 + src/agentsight/src/genai/logtail.rs | 3 + src/agentsight/src/genai/semantic.rs | 4 + src/agentsight/src/interruption/detector.rs | 1 + src/agentsight/src/lib.rs | 1 + src/agentsight/src/lineage/mod.rs | 1457 +++++++++++++++++ src/agentsight/src/probes/codex_offsets.rs | 2 +- src/agentsight/src/probes/elf_buildid.rs | 2 +- src/agentsight/src/probes/procmon.rs | 2 + src/agentsight/src/server/token_savings.rs | 14 +- .../src/storage/sqlite/genai/events.rs | 43 +- .../src/storage/sqlite/genai/pending.rs | 7 +- .../src/storage/sqlite/genai/schema.rs | 3 + .../src/storage/sqlite/genai/tests.rs | 131 ++ src/agentsight/src/unified.rs | 275 +++- src/agentsight/tests/common/mod.rs | 1 + src/agentsight/tests/memory_storage.rs | 10 +- 25 files changed, 1978 insertions(+), 72 deletions(-) create mode 100644 src/agentsight/src/lineage/mod.rs diff --git a/src/agentsight/scripts/check-arch-boundaries.py b/src/agentsight/scripts/check-arch-boundaries.py index af6295d5a..811f9f8c9 100755 --- a/src/agentsight/scripts/check-arch-boundaries.py +++ b/src/agentsight/scripts/check-arch-boundaries.py @@ -63,6 +63,7 @@ "logging", # logging init "utils", # utility functions "discovery", # process discovery (Cross in ARCHITECTURE.md) + "lineage", # process lineage tree (Cross: L8 populates, L7 reads) "skill_metrics", # metric helpers "token_breakdown", # token analysis helpers } diff --git a/src/agentsight/src/background.rs b/src/agentsight/src/background.rs index 7c674e62d..d9a2b8182 100644 --- a/src/agentsight/src/background.rs +++ b/src/agentsight/src/background.rs @@ -251,7 +251,10 @@ mod tests { fn tmp_dir(tag: &str) -> PathBuf { static C: AtomicU32 = AtomicU32::new(0); let n = C.fetch_add(1, Ordering::SeqCst); - let dir = std::env::temp_dir().join(format!("bg-test-{tag}-{n}")); + let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("target") + .join("test-tmp") + .join(format!("bg-test-{tag}-{n}")); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap(); dir @@ -297,6 +300,7 @@ mod tests { model: None, provider: None, call_kind: "main".to_string(), + process_type: None, }) .unwrap(); diff --git a/src/agentsight/src/bin/cli/interruption.rs b/src/agentsight/src/bin/cli/interruption.rs index 4e849f860..53bd2c3ae 100644 --- a/src/agentsight/src/bin/cli/interruption.rs +++ b/src/agentsight/src/bin/cli/interruption.rs @@ -568,19 +568,26 @@ fn print_record_detail(r: &InterruptionRecord) { ); println!(" Agent: {}", r.agent_name.as_deref().unwrap_or("-")); if let Some(ref detail) = r.detail { - // Pretty-print JSON detail if let Ok(v) = serde_json::from_str::(detail) { - println!(" Detail:"); - println!( - "{}", - serde_json::to_string_pretty(&v).unwrap_or_else(|_| detail.clone()) - ); + if r.interruption_type == "agent_crash" && v.get("signal").is_some() { + print_crash_report(&v); + } else { + println!(" Detail:"); + println!( + "{}", + serde_json::to_string_pretty(&v).unwrap_or_else(|_| detail.clone()) + ); + } } else { println!(" Detail: {detail}"); } } } +fn print_crash_report(detail: &serde_json::Value) { + print!("{}", agentsight::lineage::format_crash_report(detail)); +} + /// Print any Serialize value as JSON. fn print_json(value: &T) { match serde_json::to_string_pretty(value) { diff --git a/src/agentsight/src/bin/cli/token.rs b/src/agentsight/src/bin/cli/token.rs index 50ecb6bcd..c82ce6066 100644 --- a/src/agentsight/src/bin/cli/token.rs +++ b/src/agentsight/src/bin/cli/token.rs @@ -24,6 +24,10 @@ pub struct TokenCommand { #[structopt(long)] pub json: bool, + /// Group token usage by process type (agent/sub_agent/tool) + #[structopt(long)] + pub by_type: bool, + /// Custom data file path #[structopt(long)] pub data_file: Option, @@ -31,9 +35,10 @@ pub struct TokenCommand { impl TokenCommand { pub fn execute(&self) { - // Determine data file path - // Use the unified database path (agentsight.db) as default, - // which is where Storage writes all tables. + if self.by_type { + self.execute_by_type(); + return; + } let data_path = self .data_file .as_ref() @@ -43,6 +48,49 @@ impl TokenCommand { self.execute_summary(&data_path); } + fn execute_by_type(&self) { + use agentsight::storage::sqlite::GenAISqliteStore; + + let genai_path = self + .data_file + .as_ref() + .map(std::path::PathBuf::from) + .unwrap_or_else(GenAISqliteStore::default_path); + let store = match GenAISqliteStore::new_with_path(&genai_path) { + Ok(s) => s, + Err(e) => { + eprintln!("Failed to open GenAI store: {e}"); + return; + } + }; + + let hours = self.hours.unwrap_or(24); + let now_ns = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as i64) + .unwrap_or(0); + let start_ns = now_ns - (hours as i64) * 3_600_000_000_000; + + match store.token_usage_by_process_type(start_ns, now_ns) { + Ok(rows) if rows.is_empty() => { + println!("No LLM calls with process_type in the last {hours} hours."); + } + Ok(rows) => { + if self.json { + println!("{}", agentsight::lineage::format_token_by_type_json(&rows)); + } else { + print!( + "{}", + agentsight::lineage::format_token_by_type_table(&rows, hours) + ); + } + } + Err(e) => { + eprintln!("Query failed: {e}"); + } + } + } + fn execute_summary(&self, data_path: &std::path::Path) { // Open token store let store = TokenStore::new(data_path); diff --git a/src/agentsight/src/bpf/procmon.bpf.c b/src/agentsight/src/bpf/procmon.bpf.c index dcdb5618b..d16e246f9 100644 --- a/src/agentsight/src/bpf/procmon.bpf.c +++ b/src/agentsight/src/bpf/procmon.bpf.c @@ -45,11 +45,12 @@ int trace_execve_exit(struct trace_event_raw_sys_exit *ctx) // Fill event event->source = EVENT_SOURCE_PROCMON; event->timestamp_ns = ts; - event->pid = get_task_ns_pid(task); + event->pid = pid; event->tid = tid; event->ppid = ppid; event->uid = uid; event->event_type = PROCMON_EVENT_EXEC; + event->exit_code = 0; bpf_get_current_comm(&event->comm, sizeof(event->comm)); bpf_ringbuf_submit(event, 0); @@ -79,11 +80,13 @@ int trace_process_exit(void *ctx) // Fill event event->source = EVENT_SOURCE_PROCMON; event->timestamp_ns = ts; - event->pid = current_ns_pid(); + event->pid = pid; event->tid = tid; event->ppid = 0; event->uid = uid; event->event_type = PROCMON_EVENT_EXIT; + struct task_struct *exit_task = (struct task_struct *)bpf_get_current_task(); + event->exit_code = BPF_CORE_READ(exit_task, exit_code); bpf_get_current_comm(&event->comm, sizeof(event->comm)); bpf_ringbuf_submit(event, 0); diff --git a/src/agentsight/src/bpf/procmon.h b/src/agentsight/src/bpf/procmon.h index 588a8a981..71cdf8bfb 100644 --- a/src/agentsight/src/bpf/procmon.h +++ b/src/agentsight/src/bpf/procmon.h @@ -32,6 +32,7 @@ struct procmon_event { u32 ppid; // Parent PID u32 uid; u32 event_type; // enum procmon_event_type + s32 exit_code; // kernel wait-status (signal|coredump|exit_status) char comm[TASK_COMM_LEN]; }; diff --git a/src/agentsight/src/bpf/proctrace.bpf.c b/src/agentsight/src/bpf/proctrace.bpf.c index 84f31c462..2b3e4bfd9 100644 --- a/src/agentsight/src/bpf/proctrace.bpf.c +++ b/src/agentsight/src/bpf/proctrace.bpf.c @@ -378,7 +378,8 @@ int trace_process_exit(void *ctx) // Fill exit_data struct proc_exit_data *exit_data = (void *)(event + 1); - exit_data->exit_code = 0; // TODO: Get actual exit code from sched_process_exit + struct task_struct *exit_task = (struct task_struct *)bpf_get_current_task(); + exit_data->exit_code = BPF_CORE_READ(exit_task, exit_code); bpf_ringbuf_submit(event, 0); return 0; diff --git a/src/agentsight/src/genai/builder.rs b/src/agentsight/src/genai/builder.rs index 30be4c0fa..cc58be704 100644 --- a/src/agentsight/src/genai/builder.rs +++ b/src/agentsight/src/genai/builder.rs @@ -156,6 +156,7 @@ impl GenAIBuilder { .get("call_kind") .cloned() .unwrap_or_else(|| "main".to_string()), + process_type: None, }); events.push(GenAISemanticEvent::LLMCall(llm_call)); @@ -364,6 +365,7 @@ impl GenAIBuilder { model, provider, call_kind: call_kind.to_string(), + process_type: None, }) } diff --git a/src/agentsight/src/genai/call_builder.rs b/src/agentsight/src/genai/call_builder.rs index 1dc45e08d..914e3bdd3 100644 --- a/src/agentsight/src/genai/call_builder.rs +++ b/src/agentsight/src/genai/call_builder.rs @@ -207,6 +207,7 @@ impl GenAIBuilder { pid: pid_i32, process_name: http.comm.clone(), agent_name: Some(agent_name.clone()), + process_type: None, metadata: { let mut meta = HashMap::new(); meta.insert("method".to_string(), http.method); diff --git a/src/agentsight/src/genai/logtail.rs b/src/agentsight/src/genai/logtail.rs index bc698c664..4014d8204 100644 --- a/src/agentsight/src/genai/logtail.rs +++ b/src/agentsight/src/genai/logtail.rs @@ -517,6 +517,9 @@ pub fn events_to_flat_records( if let Some(ref name) = call.agent_name { m.insert("agentsight.agent.name".to_string(), name.clone()); } + if let Some(ref ptype) = call.process_type { + m.insert("agentsight.process_type".to_string(), ptype.clone()); + } m.insert( "agentsight.duration_ns".to_string(), call.duration_ns.to_string(), diff --git a/src/agentsight/src/genai/semantic.rs b/src/agentsight/src/genai/semantic.rs index 404fcceac..c9b11c858 100644 --- a/src/agentsight/src/genai/semantic.rs +++ b/src/agentsight/src/genai/semantic.rs @@ -48,6 +48,9 @@ pub struct LLMCall { pub process_name: String, /// Resolved agent name from discovery registry (e.g. "OpenClaw", "Cosh") pub agent_name: Option, + /// Process type from lineage tree (agent/sub_agent/tool) + #[serde(skip_serializing_if = "Option::is_none")] + pub process_type: Option, /// Additional metadata pub metadata: HashMap, } @@ -304,6 +307,7 @@ impl LLMCall { pid, process_name, agent_name: None, + process_type: None, metadata: HashMap::new(), } } diff --git a/src/agentsight/src/interruption/detector.rs b/src/agentsight/src/interruption/detector.rs index 76a862794..27c207b17 100644 --- a/src/agentsight/src/interruption/detector.rs +++ b/src/agentsight/src/interruption/detector.rs @@ -430,6 +430,7 @@ mod tests { pid: 1234, process_name: "agent".to_string(), agent_name: Some("TestAgent".to_string()), + process_type: None, metadata: HashMap::from([("status_code".to_string(), "200".to_string())]), } } diff --git a/src/agentsight/src/lib.rs b/src/agentsight/src/lib.rs index d79944b0c..4d791c80c 100644 --- a/src/agentsight/src/lib.rs +++ b/src/agentsight/src/lib.rs @@ -54,6 +54,7 @@ pub mod ffi; pub mod genai; pub mod health; pub mod interruption; +pub mod lineage; pub mod parser; pub mod response_map; #[cfg(feature = "server")] diff --git a/src/agentsight/src/lineage/mod.rs b/src/agentsight/src/lineage/mod.rs new file mode 100644 index 000000000..0997cb966 --- /dev/null +++ b/src/agentsight/src/lineage/mod.rs @@ -0,0 +1,1457 @@ +//! Blood lineage tree for AI agent process tracking +//! +//! Maintains a userspace mirror of the BPF lineage_tree map, enriched with +//! process type classification (Agent / SubAgent / Tool / Skill). + +use std::collections::HashMap; + +use serde::Serialize; + +/// Process type classification for lineage tree nodes. +/// +/// `Skill` is forward-declared for the follow-up scoring task — `classify()` +/// does NOT currently produce it (today's classifier covers Agent / SubAgent / +/// Tool only). The variant exists in `from_u32`/`as_u32` so a future BPF-side +/// scorer can emit it without breaking userspace decoding. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ProcessType { + Unknown, + Agent, + SubAgent, + Tool, + /// Forward-declared. Not produced by the current `classify()`. + Skill, +} + +impl ProcessType { + pub fn from_u32(v: u32) -> Self { + match v { + 1 => Self::Agent, + 2 => Self::SubAgent, + 3 => Self::Tool, + 4 => Self::Skill, + _ => Self::Unknown, + } + } + + pub fn as_u32(&self) -> u32 { + match self { + Self::Unknown => 0, + Self::Agent => 1, + Self::SubAgent => 2, + Self::Tool => 3, + Self::Skill => 4, + } + } + + pub fn as_str(&self) -> &'static str { + match self { + Self::Unknown => "unknown", + Self::Agent => "agent", + Self::SubAgent => "sub_agent", + Self::Tool => "tool", + Self::Skill => "skill", + } + } +} + +/// Flags on a lineage node (mirrors LINEAGE_FLAG_* from BPF) +pub const LINEAGE_FLAG_AGENT_MODE: u32 = 1 << 0; + +/// A single node in the lineage tree +#[derive(Debug, Clone, Serialize)] +pub struct LineageNode { + pub pid: u32, + pub ppid: u32, + pub process_type: ProcessType, + pub flags: u32, + pub create_time_ns: u64, + pub comm: String, + pub agent_name: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub children: Vec, +} + +impl LineageNode { + pub fn has_agent_mode(&self) -> bool { + self.flags & LINEAGE_FLAG_AGENT_MODE != 0 + } +} + +/// Userspace lineage tree — mirrors BPF lineage_tree map with classification +#[derive(Default)] +pub struct LineageTree { + nodes: HashMap, +} + +impl LineageTree { + pub fn new() -> Self { + Self::default() + } + + /// Insert or update a node. Automatically maintains parent→child links. + /// + /// Also handles PID re-parenting: if the pid already exists under a + /// different parent (e.g. PID reuse where the old Exit was missed under + /// ringbuf pressure), the stale child entry is removed from the old + /// parent's children list before inserting under the new parent. + pub fn insert(&mut self, mut node: LineageNode) { + let pid = node.pid; + let ppid = node.ppid; + + // PID-reuse / re-parent: detach from old parent first. + if let Some(old_node) = self.nodes.get(&pid) { + let old_ppid = old_node.ppid; + if old_ppid != ppid { + if let Some(old_parent) = self.nodes.get_mut(&old_ppid) { + old_parent.children.retain(|&c| c != pid); + } + } + } + + // Add this pid as a child of its (possibly new) parent. + if let Some(parent) = self.nodes.get_mut(&ppid) { + if !parent.children.contains(&pid) { + parent.children.push(pid); + } + } + + if let Some(old) = self.nodes.get(&pid) { + node.children = old.children.clone(); + } + + self.nodes.insert(pid, node); + } + + /// Remove a node, reparent its children to grandparent, and clean up links. + pub fn remove(&mut self, pid: u32) -> Option { + let node = self.nodes.remove(&pid)?; + + // Reparent children to the removed node's parent (mirrors kernel subreaper) + for &child_pid in &node.children { + if let Some(child) = self.nodes.get_mut(&child_pid) { + child.ppid = node.ppid; + } + } + + // Update parent's children list: remove this node, add its children + if let Some(parent) = self.nodes.get_mut(&node.ppid) { + parent.children.retain(|&c| c != pid); + for child_pid in &node.children { + if !parent.children.contains(child_pid) { + parent.children.push(*child_pid); + } + } + } + + Some(node) + } + + /// Get a reference to a node + pub fn get(&self, pid: u32) -> Option<&LineageNode> { + self.nodes.get(&pid) + } + + /// Get a mutable reference to a node + pub fn get_mut(&mut self, pid: u32) -> Option<&mut LineageNode> { + self.nodes.get_mut(&pid) + } + + /// Insert a process (if not already present) and classify it in one step. + /// Returns the classified ProcessType. + /// + /// This is the primary entry point for unified.rs glue code: it combines + /// node creation + insertion + classification into a single pure-function + /// call on the tree, enabling full unit-test coverage without BPF. + pub fn insert_and_classify( + &mut self, + pid: u32, + ppid: u32, + comm: &str, + agent_name: Option, + has_agent_mode: bool, + matches_agent: bool, + ) -> ProcessType { + if self.get(pid).is_none() { + let node = LineageNode { + pid, + ppid, + process_type: ProcessType::Unknown, + flags: if has_agent_mode { + LINEAGE_FLAG_AGENT_MODE + } else { + 0 + }, + create_time_ns: 0, + comm: comm.to_string(), + agent_name, + children: vec![], + }; + self.insert(node); + } + self.classify(pid, has_agent_mode, matches_agent); + self.get(pid) + .map(|n| n.process_type) + .unwrap_or(ProcessType::Unknown) + } + + /// Classify a newly added process based on its ancestry and environment. + /// + /// Rules (evaluated in order): + /// 1. Parent is Agent/SubAgent → child inherits lineage: + /// - matches agent pattern → SubAgent + /// - otherwise → Tool + /// 2. No tracked parent + matches agent pattern → Agent (cmdline rule match) + /// 3. No tracked parent + AGENT_MODE=1 in env → Agent (reserved for future + /// agent frameworks that self-declare; currently no component sets this) + /// 4. Otherwise → Unknown + /// + /// Note: child processes inherit AGENT_MODE=1 via environment, but that + /// does NOT make them Agents — only top-level processes (without a tracked + /// parent) are classified as Agent via AGENT_MODE. Cmdline pattern matching + /// takes precedence for root classification. + pub fn classify(&mut self, pid: u32, has_agent_mode_env: bool, matches_agent_pattern: bool) { + let ppid = match self.nodes.get(&pid) { + Some(n) => n.ppid, + None => return, + }; + + let parent_in_tree = self.nodes.get(&ppid); + let parent_type = parent_in_tree + .map(|p| p.process_type) + .unwrap_or(ProcessType::Unknown); + + let process_type = match parent_type { + ProcessType::Agent | ProcessType::SubAgent => { + if matches_agent_pattern { + ProcessType::SubAgent + } else { + ProcessType::Tool + } + } + _ => { + if matches_agent_pattern || has_agent_mode_env { + ProcessType::Agent + } else { + ProcessType::Unknown + } + } + }; + + if let Some(node) = self.nodes.get_mut(&pid) { + node.process_type = process_type; + } + } + + /// Record a process exec: classify it and stamp its create time. + /// + /// Derives the classification inputs the way the live event pipeline does: + /// - `has_agent_mode` is read from any pre-existing node (AGENT_MODE is + /// inherited via env and preserved across re-exec), defaulting to false. + /// - `matches_agent` is true when discovery resolved an `agent_name` AND the + /// process is not already an AGENT_MODE process (cmdline match wins for + /// roots; an AGENT_MODE child stays a Tool — see `classify`). + /// + /// Returns the resulting `ProcessType`. This is the pure tree-side of + /// `unified::update_lineage_from_proc`, extracted so it can be unit-tested + /// without the BPF event pipeline. + pub fn record_exec( + &mut self, + pid: u32, + ppid: u32, + comm: &str, + agent_name: Option, + create_time_ns: u64, + ) -> ProcessType { + let has_agent_mode = self.get(pid).map(|n| n.has_agent_mode()).unwrap_or(false); + let matches_agent = agent_name.is_some() && !has_agent_mode; + let ptype = + self.insert_and_classify(pid, ppid, comm, agent_name, has_agent_mode, matches_agent); + if let Some(node) = self.get_mut(pid) { + node.create_time_ns = create_time_ns; + } + ptype + } + + /// Snapshot of direct child PIDs at exit time. + /// + /// Not a liveness check — returns whichever children are in the tree when + /// called, regardless of whether they are still running. + pub fn children_at_exit(&self, pid: u32) -> Vec { + self.get(pid) + .map(|n| n.children.clone()) + .unwrap_or_default() + } + + /// Walk the ppid chain from `pid` upward to find the root Agent ancestor. + /// Returns `None` if `pid` is itself an Agent, is absent, or has no Agent + /// ancestor in the tree. + pub fn root_agent_ancestor(&self, pid: u32) -> Option<&LineageNode> { + let node = self.nodes.get(&pid)?; + if node.process_type == ProcessType::Agent { + return None; + } + let mut current_ppid = node.ppid; + for _ in 0..64 { + let parent = self.nodes.get(¤t_ppid)?; + if parent.process_type == ProcessType::Agent { + return Some(parent); + } + current_ppid = parent.ppid; + } + None + } + + /// Get the full subtree rooted at `pid` as a serializable structure + pub fn subtree(&self, pid: u32) -> Option { + self.subtree_inner(pid, 0) + } + + fn subtree_inner(&self, pid: u32, depth: u32) -> Option { + if depth > 64 { + return None; + } + let node = self.nodes.get(&pid)?; + let children = node + .children + .iter() + .filter_map(|&cpid| self.subtree_inner(cpid, depth + 1)) + .collect(); + Some(LineageSubtree { + pid: node.pid, + ppid: node.ppid, + process_type: node.process_type, + flags: node.flags, + create_time_ns: node.create_time_ns, + comm: node.comm.clone(), + agent_name: node.agent_name.clone(), + children, + }) + } + + /// Return all root nodes (nodes whose ppid is not in the tree) + pub fn roots(&self) -> Vec { + self.nodes + .values() + .filter(|n| !self.nodes.contains_key(&n.ppid)) + .map(|n| n.pid) + .collect() + } + + /// Snapshot the entire tree as a flat list. + pub fn snapshot(&self) -> Vec<&LineageNode> { + self.nodes.values().collect() + } + + pub fn len(&self) -> usize { + self.nodes.len() + } + + pub fn is_empty(&self) -> bool { + self.nodes.is_empty() + } +} + +/// Recursive subtree for JSON serialization +#[derive(Debug, Clone, Serialize)] +pub struct LineageSubtree { + pub pid: u32, + pub ppid: u32, + pub process_type: ProcessType, + pub flags: u32, + pub create_time_ns: u64, + pub comm: String, + pub agent_name: Option, + pub children: Vec, +} + +// ─── Exit classification (crash detection) ─────────────────────────────────── + +/// Kernel wait-status decoded into a crash classification. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ExitClassification { + /// Fatal signal (SIGSEGV, SIGKILL, SIGABRT, SIGBUS, etc.) + SignalCrash { signal: u8, coredump: bool }, + /// Graceful stop (SIGTERM, SIGINT, SIGHUP) + GracefulStop { signal: u8 }, + /// Benign signal (SIGPIPE) + BenignSignal { signal: u8 }, + /// Clean exit (exit(0)) + NormalExit, + /// Non-zero exit status (exit(N), N != 0) + AbnormalExit { status: u8 }, +} + +impl ExitClassification { + pub fn is_crash(&self) -> bool { + matches!(self, Self::SignalCrash { .. } | Self::AbnormalExit { .. }) + } +} + +/// Classify a raw kernel exit_code into a crash category. +/// +/// exit_code layout (wait(2) encoding): +/// bits [6:0] = terminating signal (0 if normal exit) +/// bit [7] = core dump produced +/// bits [15:8] = exit status (only meaningful when signal == 0) +pub fn classify_exit(exit_code: i32) -> ExitClassification { + let signal = (exit_code & 0x7f) as u8; + let coredump = (exit_code >> 7) & 1 != 0; + let status = ((exit_code >> 8) & 0xff) as u8; + + if signal == 0 { + if status == 0 { + ExitClassification::NormalExit + } else { + ExitClassification::AbnormalExit { status } + } + } else { + match signal { + // SIGTERM(15), SIGINT(2), SIGHUP(1): graceful stop + 15 | 2 | 1 => ExitClassification::GracefulStop { signal }, + // SIGPIPE(13): benign (broken pipe) + 13 => ExitClassification::BenignSignal { signal }, + // Everything else: crash (SIGSEGV=11, SIGKILL=9, SIGABRT=6, SIGBUS=7, ...) + _ => ExitClassification::SignalCrash { signal, coredump }, + } + } +} + +/// Map a signal number to its conventional name. +pub fn signal_name(signal: u8) -> &'static str { + match signal { + 1 => "SIGHUP", + 2 => "SIGINT", + 6 => "SIGABRT", + 7 => "SIGBUS", + 9 => "SIGKILL", + 11 => "SIGSEGV", + 13 => "SIGPIPE", + 15 => "SIGTERM", + _ => "SIG?", + } +} + +/// Format a one-line crash cause from a raw exit_code. +pub fn format_crash_cause(exit_code: i32) -> String { + let classification = classify_exit(exit_code); + match classification { + ExitClassification::SignalCrash { signal, coredump } => { + format!( + "{} (signal {signal}, coredump: {coredump})", + signal_name(signal) + ) + } + ExitClassification::AbnormalExit { status } => format!("exit({status})"), + ExitClassification::NormalExit => "exit(0)".to_string(), + ExitClassification::GracefulStop { signal } => { + format!("{} (graceful)", signal_name(signal)) + } + ExitClassification::BenignSignal { signal } => { + format!("{} (benign)", signal_name(signal)) + } + } +} + +/// Build a structured crash report string from the detail JSON of an agent_crash event. +pub fn format_crash_report(detail: &serde_json::Value) -> String { + let mut out = String::new(); + out.push_str("\n --- Crash Report ---\n"); + let signal = detail.get("signal").and_then(|v| v.as_u64()).unwrap_or(0); + let exit_status = detail + .get("exit_status") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + let coredump = detail + .get("coredump") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let oom = detail.get("oom").and_then(|v| v.as_bool()).unwrap_or(false); + + if signal > 0 { + let sig_name = signal_name(signal as u8); + out.push_str(&format!( + " Cause: {sig_name} (signal {signal}, coredump: {coredump})\n" + )); + } else { + out.push_str(&format!(" Cause: exit({exit_status})\n")); + } + out.push_str(&format!( + " OOM Killed: {}\n", + if oom { "yes" } else { "no" } + )); + if let Some(ptype) = detail.get("process_type").and_then(|v| v.as_str()) { + out.push_str(&format!(" Process Type: {ptype}\n")); + } + if let Some(blast) = detail.get("blast_radius").and_then(|v| v.as_str()) { + out.push_str(&format!(" Blast Radius: {blast}\n")); + } + + if let Some(calls) = detail.get("call_ids").and_then(|v| v.as_array()) { + if !calls.is_empty() { + out.push_str(&format!(" Pending LLM Calls ({}):\n", calls.len())); + for c in calls { + if let Some(s) = c.as_str() { + out.push_str(&format!(" - {s}\n")); + } + } + } + } + + if let Some(children) = detail.get("children_at_exit").and_then(|v| v.as_array()) { + if !children.is_empty() { + let pids: Vec = children + .iter() + .filter_map(|v| v.as_u64().map(|p| p.to_string())) + .collect(); + out.push_str(&format!(" Children at Exit: [{}]\n", pids.join(", "))); + } + } + + if let Some(tree) = detail.get("process_tree") { + out.push_str("\n Process Tree at Crash:\n"); + render_tree_to_string(tree, " ", true, &mut out); + } + out +} + +/// Render a process tree JSON node into a string with tree connectors. +pub fn render_tree_to_string( + node: &serde_json::Value, + prefix: &str, + is_last: bool, + out: &mut String, +) { + let pid = node.get("pid").and_then(|v| v.as_u64()).unwrap_or(0); + let comm = node.get("comm").and_then(|v| v.as_str()).unwrap_or("?"); + let ptype = node + .get("process_type") + .and_then(|v| v.as_str()) + .unwrap_or("unknown"); + let connector = if is_last { "`-- " } else { "|-- " }; + out.push_str(&format!("{prefix}{connector}{comm} [{pid}] ({ptype})\n")); + if let Some(children) = node.get("children").and_then(|v| v.as_array()) { + let child_prefix = format!("{prefix}{}", if is_last { " " } else { "| " }); + for (i, child) in children.iter().enumerate() { + render_tree_to_string(child, &child_prefix, i == children.len() - 1, out); + } + } +} + +/// Format token usage by process type as JSON string. +pub fn format_token_by_type_json(rows: &[(String, i64, i64, i64, i64)]) -> String { + let data: Vec<_> = rows + .iter() + .map(|(ptype, count, input, output, total)| { + serde_json::json!({ + "process_type": ptype, + "call_count": count, + "input_tokens": input, + "output_tokens": output, + "total_tokens": total, + }) + }) + .collect(); + serde_json::to_string_pretty(&data).unwrap_or_default() +} + +/// Format token usage by process type as a table string. +pub fn format_token_by_type_table(rows: &[(String, i64, i64, i64, i64)], hours: u64) -> String { + use crate::format_tokens_with_commas; + let mut out = format!("Token Usage by Process Type (last {hours}h)\n\n"); + out.push_str(&format!( + "{:<12} {:>8} {:>12} {:>12} {:>12}\n", + "TYPE", "CALLS", "INPUT", "OUTPUT", "TOTAL" + )); + out.push_str(&format!("{}\n", "-".repeat(60))); + let mut grand_total = 0i64; + for (ptype, count, input, output, total) in rows { + out.push_str(&format!( + "{:<12} {:>8} {:>12} {:>12} {:>12}\n", + ptype, + count, + format_tokens_with_commas(*input as u64), + format_tokens_with_commas(*output as u64), + format_tokens_with_commas(*total as u64), + )); + grand_total += total; + } + out.push_str(&format!("{}\n", "-".repeat(60))); + out.push_str(&format!( + "{:<12} {:>8} {:>12} {:>12} {:>12}\n", + "TOTAL", + "", + "", + "", + format_tokens_with_commas(grand_total as u64) + )); + out +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_node(pid: u32, ppid: u32, ptype: ProcessType) -> LineageNode { + LineageNode { + pid, + ppid, + process_type: ptype, + flags: 0, + create_time_ns: 0, + comm: format!("proc-{pid}"), + agent_name: None, + children: Vec::new(), + } + } + + #[test] + fn test_insert_and_parent_link() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + + let parent = tree.get(100).unwrap(); + assert_eq!(parent.children, vec![200]); + } + + #[test] + fn test_remove_cleans_parent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + tree.remove(200); + + let parent = tree.get(100).unwrap(); + assert!(parent.children.is_empty()); + } + + #[test] + fn test_classify_agent_mode() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Unknown)); + tree.classify(100, true, false); + assert_eq!(tree.get(100).unwrap().process_type, ProcessType::Agent); + } + + #[test] + fn test_classify_tool_under_agent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Unknown)); + tree.classify(200, false, false); + assert_eq!(tree.get(200).unwrap().process_type, ProcessType::Tool); + } + + #[test] + fn test_classify_subagent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Unknown)); + tree.classify(200, false, true); + assert_eq!(tree.get(200).unwrap().process_type, ProcessType::SubAgent); + } + + #[test] + fn test_roots() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + tree.insert(make_node(300, 2, ProcessType::Agent)); + + let roots = tree.roots(); + assert_eq!(roots.len(), 2); + assert!(roots.contains(&100)); + assert!(roots.contains(&300)); + } + + /// PID-reuse / re-parent: a pid that re-execs (or whose Exit was dropped) + /// under a different parent must NOT leave a phantom child entry on the + /// old parent. Discriminating: without the cleanup branch in insert(), the + /// final assertion below fails — old parent still lists pid 200. + #[test] + fn test_insert_cleans_old_parent_on_reparent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(300, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + // Re-insert pid=200 under a different parent (e.g. PID reuse). + tree.insert(make_node(200, 300, ProcessType::Tool)); + + assert!( + tree.get(100).unwrap().children.is_empty(), + "old parent 100 must not retain a phantom child 200" + ); + assert_eq!(tree.get(300).unwrap().children, vec![200]); + assert_eq!(tree.get(200).unwrap().ppid, 300); + } + + /// AGENT_MODE precedence pin (1/2): when the parent is already an Agent, + /// a child with has_agent_mode_env=true must classify as Tool — NOT + /// promote itself to Agent. The env var is inherited through fork; only + /// top-level processes (no tracked parent) are eligible for Agent. + /// Discriminating: reordering the match arms in classify() so that + /// AGENT_MODE wins over parent_type would flip this case to Agent. + #[test] + fn test_classify_agent_mode_inherited_under_agent_stays_tool() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Unknown)); + tree.classify( + 200, /* has_agent_mode_env */ true, /* matches_agent */ false, + ); + assert_eq!(tree.get(200).unwrap().process_type, ProcessType::Tool); + } + + /// AGENT_MODE precedence pin (2/2): same rule but parent is SubAgent. + /// Inherited AGENT_MODE under a SubAgent is still a Tool. + #[test] + fn test_classify_agent_mode_inherited_under_subagent_stays_tool() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::SubAgent)); + tree.insert(make_node(300, 200, ProcessType::Unknown)); + tree.classify( + 300, /* has_agent_mode_env */ true, /* matches_agent */ false, + ); + assert_eq!(tree.get(300).unwrap().process_type, ProcessType::Tool); + } + + /// `ProcessType::Skill` is forward-declared. Document the current + /// behaviour: classify() never produces Skill regardless of inputs. + /// Tightens the contract so a future scorer change is explicit. + #[test] + fn test_classify_never_produces_skill_today() { + // Try a variety of parent types + flags; none should yield Skill. + for parent_type in [ + ProcessType::Unknown, + ProcessType::Agent, + ProcessType::SubAgent, + ProcessType::Tool, + ProcessType::Skill, + ] { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, parent_type)); + tree.insert(make_node(200, 100, ProcessType::Unknown)); + for has_env in [false, true] { + for matches in [false, true] { + tree.classify(200, has_env, matches); + assert_ne!( + tree.get(200).unwrap().process_type, + ProcessType::Skill, + "classify() must not produce Skill (forward-declared)", + ); + } + } + } + } + + #[test] + fn test_classify_cmdline_match_root_becomes_agent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Unknown)); + // cmdline-matched root with no AGENT_MODE — should still become Agent + tree.classify( + 100, /* has_agent_mode_env */ false, /* matches_agent */ true, + ); + assert_eq!( + tree.get(100).unwrap().process_type, + ProcessType::Agent, + "cmdline-matched root process must be classified as Agent" + ); + } + + #[test] + fn test_reinsert_preserves_children() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + tree.insert(make_node(300, 200, ProcessType::Tool)); + + assert_eq!(tree.get(200).unwrap().children, vec![300]); + + // Re-insert 200 (simulates re-execve or procmon+proctrace double-insert) + tree.insert(make_node(200, 100, ProcessType::Tool)); + + assert_eq!( + tree.get(200).unwrap().children, + vec![300], + "re-insert must preserve accumulated children", + ); + } + + #[test] + fn test_lineage_node_fields_and_flags() { + let node = LineageNode { + pid: 1, + ppid: 0, + process_type: ProcessType::Agent, + flags: LINEAGE_FLAG_AGENT_MODE, + create_time_ns: 12345, + comm: "test".to_string(), + agent_name: Some("TestAgent".to_string()), + children: vec![2, 3], + }; + assert!(node.has_agent_mode()); + assert_eq!(node.pid, 1); + assert_eq!(node.ppid, 0); + assert_eq!(node.comm, "test"); + assert_eq!(node.agent_name, Some("TestAgent".to_string())); + assert_eq!(node.children, vec![2, 3]); + + // without flag + let node2 = LineageNode { + pid: 2, + ppid: 1, + process_type: ProcessType::Tool, + flags: 0, + create_time_ns: 0, + comm: "bash".to_string(), + agent_name: None, + children: vec![], + }; + assert!(!node2.has_agent_mode()); + } + + #[test] + fn test_snapshot() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + tree.insert(make_node(300, 100, ProcessType::SubAgent)); + + let snap = tree.snapshot(); + assert_eq!(snap.len(), 3); + } + + #[test] + fn test_subtree() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + tree.insert(make_node(300, 100, ProcessType::SubAgent)); + tree.insert(make_node(400, 200, ProcessType::Tool)); + + // subtree from root + let sub = tree.subtree(100); + assert!(sub.is_some()); + let sub = sub.unwrap(); + assert_eq!(sub.pid, 100); + assert_eq!(sub.children.len(), 2); + + // subtree from leaf + let leaf = tree.subtree(400); + assert!(leaf.is_some()); + assert_eq!(leaf.unwrap().children.len(), 0); + + // subtree from non-existent + let none = tree.subtree(999); + assert!(none.is_none()); + } + + #[test] + fn test_remove_reparents_children_to_grandparent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(1, 0, ProcessType::Agent)); + tree.insert(make_node(2, 1, ProcessType::Tool)); + tree.insert(make_node(3, 2, ProcessType::Tool)); + tree.insert(make_node(4, 2, ProcessType::Tool)); + + // Remove middle node (pid=2), children should reparent to pid=1 + let removed = tree.remove(2); + assert!(removed.is_some()); + assert_eq!(removed.unwrap().pid, 2); + + // Children 3,4 should now have ppid=1 + assert_eq!(tree.get(3).unwrap().ppid, 1); + assert_eq!(tree.get(4).unwrap().ppid, 1); + + // Parent 1 should have children 3,4 + let parent = tree.get(1).unwrap(); + assert!(parent.children.contains(&3)); + assert!(parent.children.contains(&4)); + assert!(!parent.children.contains(&2)); + } + + #[test] + fn test_remove_reparent_deduplicates_existing_child_link() { + let mut tree = LineageTree::new(); + tree.insert(make_node(1, 0, ProcessType::Agent)); + tree.insert(make_node(2, 1, ProcessType::Tool)); + tree.insert(make_node(3, 2, ProcessType::Tool)); + // Simulate an out-of-order / duplicate parent edge that can exist after + // missed exits or PID reuse. Removing 2 should not duplicate child 3. + tree.get_mut(1).unwrap().children.push(3); + + tree.remove(2); + + let children = &tree.get(1).unwrap().children; + assert_eq!( + children.iter().filter(|&&pid| pid == 3).count(), + 1, + "reparent must not duplicate an existing child edge" + ); + assert_eq!(tree.get(3).unwrap().ppid, 1); + } + + #[test] + fn test_len_and_is_empty() { + let mut tree = LineageTree::new(); + assert!(tree.is_empty()); + assert_eq!(tree.len(), 0); + + tree.insert(make_node(100, 1, ProcessType::Unknown)); + assert!(!tree.is_empty()); + assert_eq!(tree.len(), 1); + + tree.remove(100); + assert!(tree.is_empty()); + } + + #[test] + fn test_classify_nonexistent_pid_is_noop() { + let mut tree = LineageTree::new(); + // Should not panic, just return early + tree.classify(999, true, true); + assert!(tree.get(999).is_none()); + } + + // --- insert_and_classify tests --- + + #[test] + fn test_insert_and_classify_root_agent() { + let mut tree = LineageTree::new(); + let pt = tree.insert_and_classify(100, 1, "cosh", Some("Cosh".to_string()), false, true); + assert_eq!(pt, ProcessType::Agent); + assert_eq!(tree.get(100).unwrap().comm, "cosh"); + assert_eq!(tree.get(100).unwrap().agent_name, Some("Cosh".to_string())); + } + + #[test] + fn test_insert_and_classify_tool_under_agent() { + let mut tree = LineageTree::new(); + tree.insert_and_classify(100, 1, "cosh", Some("Cosh".to_string()), false, true); + let pt = tree.insert_and_classify(200, 100, "bash", None, false, false); + assert_eq!(pt, ProcessType::Tool); + } + + #[test] + fn test_insert_and_classify_with_agent_mode() { + let mut tree = LineageTree::new(); + let pt = tree.insert_and_classify( + 100, + 1, + "python", + Some("agent-mode-python".to_string()), + true, + false, + ); + assert_eq!(pt, ProcessType::Agent); + assert!(tree.get(100).unwrap().has_agent_mode()); + } + + #[test] + fn test_insert_and_classify_skips_existing() { + let mut tree = LineageTree::new(); + tree.insert_and_classify(100, 1, "cosh", Some("Cosh".to_string()), false, true); + // Second call should not overwrite + let pt = tree.insert_and_classify(100, 1, "different", None, false, false); + // Should re-classify based on current state (no parent in tree -> Unknown) + // but original comm is preserved + assert_eq!(tree.get(100).unwrap().comm, "cosh"); + // Re-classify: no tracked parent, matches_agent=false, has_agent_mode=false -> Unknown + assert_eq!(pt, ProcessType::Unknown); + } + + #[test] + fn test_insert_and_classify_subagent_under_agent() { + let mut tree = LineageTree::new(); + tree.insert_and_classify(100, 1, "cosh", Some("Cosh".to_string()), false, true); + let pt = + tree.insert_and_classify(200, 100, "sub-agent", Some("Sub".to_string()), false, true); + assert_eq!(pt, ProcessType::SubAgent); + } + + #[test] + fn test_record_exec_classifies_root_agent_and_stamps_time() { + let mut tree = LineageTree::new(); + let pt = tree.record_exec(100, 1, "cosh", Some("Cosh".to_string()), 42); + assert_eq!(pt, ProcessType::Agent); + assert_eq!(tree.get(100).unwrap().create_time_ns, 42); + } + + #[test] + fn test_record_exec_tool_child_under_agent() { + let mut tree = LineageTree::new(); + tree.record_exec(100, 1, "cosh", Some("Cosh".to_string()), 1); + // Child with no agent_name under an Agent classifies as Tool. + let pt = tree.record_exec(200, 100, "bash", None, 2); + assert_eq!(pt, ProcessType::Tool); + } + + /// AGENT_MODE precedence pin: `record_exec` must derive `matches_agent` as + /// false when the node already carries AGENT_MODE, so an inherited-env child + /// with an agent_name still classifies as Tool — not promote to SubAgent. + /// Discriminating: dropping the `!has_agent_mode` term in `record_exec` + /// would flip this to SubAgent. + #[test] + fn test_record_exec_agent_mode_node_stays_tool() { + let mut tree = LineageTree::new(); + tree.record_exec(100, 1, "cosh", Some("Cosh".to_string()), 1); + // Seed an AGENT_MODE child node under the agent. + tree.insert(LineageNode { + pid: 200, + ppid: 100, + process_type: ProcessType::Unknown, + flags: LINEAGE_FLAG_AGENT_MODE, + create_time_ns: 0, + comm: "child".to_string(), + agent_name: None, + children: vec![], + }); + // Re-exec the AGENT_MODE node with an agent_name: matches_agent must be + // suppressed by has_agent_mode, so it stays Tool (not SubAgent). + let pt = tree.record_exec(200, 100, "child", Some("X".to_string()), 5); + assert_eq!(pt, ProcessType::Tool); + } + + #[test] + fn test_children_at_exit_returns_children() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + tree.insert(make_node(300, 100, ProcessType::Tool)); + let mut orphans = tree.children_at_exit(100); + orphans.sort(); + assert_eq!(orphans, vec![200, 300]); + } + + #[test] + fn test_children_at_exit_empty_for_leaf_and_missing() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + assert!(tree.children_at_exit(100).is_empty()); + assert!(tree.children_at_exit(999).is_empty()); + } + + // ─── classify_exit tests ───────────────────────────────────────── + + #[test] + fn test_classify_exit_normal() { + assert_eq!(classify_exit(0), ExitClassification::NormalExit); + } + + #[test] + fn test_classify_exit_abnormal_status() { + assert_eq!( + classify_exit(1 << 8), + ExitClassification::AbnormalExit { status: 1 } + ); + assert_eq!( + classify_exit(2 << 8), + ExitClassification::AbnormalExit { status: 2 } + ); + } + + #[test] + fn test_classify_exit_sigsegv() { + let code = 11; // SIGSEGV, no coredump + assert_eq!( + classify_exit(code), + ExitClassification::SignalCrash { + signal: 11, + coredump: false + } + ); + } + + #[test] + fn test_classify_exit_sigsegv_with_coredump() { + let code = 11 | 0x80; // SIGSEGV + coredump bit + assert_eq!( + classify_exit(code), + ExitClassification::SignalCrash { + signal: 11, + coredump: true + } + ); + } + + #[test] + fn test_classify_exit_sigkill() { + assert_eq!( + classify_exit(9), + ExitClassification::SignalCrash { + signal: 9, + coredump: false + } + ); + } + + #[test] + fn test_classify_exit_sigterm_graceful() { + assert_eq!( + classify_exit(15), + ExitClassification::GracefulStop { signal: 15 } + ); + } + + #[test] + fn test_classify_exit_sigint_graceful() { + assert_eq!( + classify_exit(2), + ExitClassification::GracefulStop { signal: 2 } + ); + } + + #[test] + fn test_classify_exit_sigpipe_benign() { + assert_eq!( + classify_exit(13), + ExitClassification::BenignSignal { signal: 13 } + ); + } + + #[test] + fn test_classify_exit_sigabrt() { + assert_eq!( + classify_exit(6), + ExitClassification::SignalCrash { + signal: 6, + coredump: false + } + ); + } + + #[test] + fn test_is_crash() { + assert!( + ExitClassification::SignalCrash { + signal: 11, + coredump: false + } + .is_crash() + ); + assert!(ExitClassification::AbnormalExit { status: 1 }.is_crash()); + assert!(!ExitClassification::NormalExit.is_crash()); + assert!(!ExitClassification::GracefulStop { signal: 15 }.is_crash()); + assert!(!ExitClassification::BenignSignal { signal: 13 }.is_crash()); + } + + #[test] + fn test_classify_exit_sighup_graceful() { + assert_eq!( + classify_exit(1), + ExitClassification::GracefulStop { signal: 1 } + ); + } + + #[test] + fn test_classify_exit_sigbus() { + assert_eq!( + classify_exit(7), + ExitClassification::SignalCrash { + signal: 7, + coredump: false + } + ); + } + + #[test] + fn test_classify_exit_max_status() { + assert_eq!( + classify_exit(255 << 8), + ExitClassification::AbnormalExit { status: 255 } + ); + } + + #[test] + fn test_classify_exit_coredump_bit_without_signal() { + // exit_code=0x80: coredump bit set but signal=0. Kernel won't produce + // this, but classify_exit treats signal=0 as normal exit (coredump bit + // is only meaningful with a terminating signal). + assert_eq!(classify_exit(0x80), ExitClassification::NormalExit); + } + + // ─── signal_name / format_crash_cause tests ────────────────────── + + #[test] + fn test_signal_name_known() { + assert_eq!(signal_name(11), "SIGSEGV"); + assert_eq!(signal_name(9), "SIGKILL"); + assert_eq!(signal_name(6), "SIGABRT"); + assert_eq!(signal_name(15), "SIGTERM"); + assert_eq!(signal_name(2), "SIGINT"); + assert_eq!(signal_name(1), "SIGHUP"); + assert_eq!(signal_name(7), "SIGBUS"); + assert_eq!(signal_name(13), "SIGPIPE"); + } + + #[test] + fn test_signal_name_unknown() { + assert_eq!(signal_name(99), "SIG?"); + assert_eq!(signal_name(0), "SIG?"); + } + + #[test] + fn test_format_crash_cause_sigsegv() { + assert_eq!( + format_crash_cause(11), + "SIGSEGV (signal 11, coredump: false)" + ); + } + + #[test] + fn test_format_crash_cause_sigsegv_coredump() { + assert_eq!( + format_crash_cause(11 | 0x80), + "SIGSEGV (signal 11, coredump: true)" + ); + } + + #[test] + fn test_format_crash_cause_abnormal_exit() { + assert_eq!(format_crash_cause(1 << 8), "exit(1)"); + } + + #[test] + fn test_format_crash_cause_normal_exit() { + assert_eq!(format_crash_cause(0), "exit(0)"); + } + + #[test] + fn test_format_crash_cause_sigterm_graceful() { + assert_eq!(format_crash_cause(15), "SIGTERM (graceful)"); + } + + #[test] + fn test_format_crash_cause_sigpipe_benign() { + assert_eq!(format_crash_cause(13), "SIGPIPE (benign)"); + } + + // ─── format_crash_report / render_tree_to_string tests ──────────── + + #[test] + fn test_format_crash_report_sigsegv() { + let detail = serde_json::json!({ + "signal": 11, "exit_code": 11, "coredump": true, + "exit_status": 0, "oom": false, + "process_type": "agent", "blast_radius": "total_session_loss", + "call_ids": ["call-1", "call-2"], + "children_at_exit": [200, 300], + }); + let report = format_crash_report(&detail); + assert!(report.contains("SIGSEGV")); + assert!(report.contains("coredump: true")); + assert!(report.contains("OOM Killed: no")); + assert!(report.contains("Process Type: agent")); + assert!(report.contains("Blast Radius: total_session_loss")); + assert!(report.contains("Pending LLM Calls (2)")); + assert!(report.contains("call-1")); + assert!(report.contains("Children at Exit: [200, 300]")); + } + + #[test] + fn test_format_crash_report_abnormal_exit() { + let detail = serde_json::json!({ + "signal": 0, "exit_status": 1, "coredump": false, "oom": false, + }); + let report = format_crash_report(&detail); + assert!(report.contains("exit(1)")); + assert!(!report.contains("SIGSEGV")); + } + + #[test] + fn test_format_crash_report_with_process_tree() { + let detail = serde_json::json!({ + "signal": 9, "exit_code": 9, "coredump": false, + "exit_status": 0, "oom": true, + "process_tree": { + "pid": 100, "comm": "claude", "process_type": "agent", + "children": [ + {"pid": 200, "comm": "bash", "process_type": "tool", "children": []} + ] + } + }); + let report = format_crash_report(&detail); + assert!(report.contains("SIGKILL")); + assert!(report.contains("OOM Killed: yes")); + assert!(report.contains("claude [100] (agent)")); + assert!(report.contains("bash [200] (tool)")); + } + + #[test] + fn test_render_tree_to_string_nested() { + let tree = serde_json::json!({ + "pid": 1, "comm": "root", "process_type": "agent", + "children": [ + {"pid": 2, "comm": "child1", "process_type": "tool", "children": []}, + {"pid": 3, "comm": "child2", "process_type": "sub_agent", "children": [ + {"pid": 4, "comm": "grandchild", "process_type": "tool", "children": []} + ]} + ] + }); + let mut out = String::new(); + render_tree_to_string(&tree, "", true, &mut out); + assert!(out.contains("`-- root [1] (agent)")); + assert!(out.contains("|-- child1 [2] (tool)")); + assert!(out.contains("`-- child2 [3] (sub_agent)")); + assert!(out.contains("`-- grandchild [4] (tool)")); + } + + #[test] + fn test_format_token_by_type_table() { + let rows = vec![ + ("agent".to_string(), 10, 5000, 3000, 8000), + ("tool".to_string(), 2, 200, 100, 300), + ]; + let table = format_token_by_type_table(&rows, 24); + assert!(table.contains("Token Usage by Process Type (last 24h)")); + assert!(table.contains("agent")); + assert!(table.contains("tool")); + assert!(table.contains("8,000")); + assert!(table.contains("TOTAL")); + } + + #[test] + fn test_format_token_by_type_json() { + let rows = vec![("agent".to_string(), 5, 1000, 500, 1500)]; + let json = format_token_by_type_json(&rows); + assert!(json.contains("\"process_type\": \"agent\"")); + assert!(json.contains("\"total_tokens\": 1500")); + } + + // ─── root_agent_ancestor tests ─────────────────────────────────── + + #[test] + fn test_root_agent_ancestor_tool_under_agent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::Tool)); + let root = tree.root_agent_ancestor(200).unwrap(); + assert_eq!(root.pid, 100); + } + + #[test] + fn test_root_agent_ancestor_nested() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + tree.insert(make_node(200, 100, ProcessType::SubAgent)); + tree.insert(make_node(300, 200, ProcessType::Tool)); + let root = tree.root_agent_ancestor(300).unwrap(); + assert_eq!(root.pid, 100); + } + + #[test] + fn test_root_agent_ancestor_none_for_agent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Agent)); + assert!(tree.root_agent_ancestor(100).is_none()); + } + + #[test] + fn test_root_agent_ancestor_none_for_orphan() { + let mut tree = LineageTree::new(); + tree.insert(make_node(200, 999, ProcessType::Tool)); + assert!(tree.root_agent_ancestor(200).is_none()); + } + + #[test] + fn test_root_agent_ancestor_none_for_missing() { + let tree = LineageTree::new(); + assert!(tree.root_agent_ancestor(999).is_none()); + } + + #[test] + fn test_process_type_as_str() { + assert_eq!(ProcessType::Agent.as_str(), "agent"); + assert_eq!(ProcessType::SubAgent.as_str(), "sub_agent"); + assert_eq!(ProcessType::Tool.as_str(), "tool"); + assert_eq!(ProcessType::Unknown.as_str(), "unknown"); + assert_eq!(ProcessType::Skill.as_str(), "skill"); + } + + #[test] + fn test_process_type_from_u32() { + assert_eq!(ProcessType::from_u32(0), ProcessType::Unknown); + assert_eq!(ProcessType::from_u32(1), ProcessType::Agent); + assert_eq!(ProcessType::from_u32(2), ProcessType::SubAgent); + assert_eq!(ProcessType::from_u32(3), ProcessType::Tool); + assert_eq!(ProcessType::from_u32(4), ProcessType::Skill); + assert_eq!(ProcessType::from_u32(99), ProcessType::Unknown); + } + + #[test] + fn test_process_type_as_u32_roundtrip() { + for v in 0..=4 { + assert_eq!(ProcessType::from_u32(v).as_u32(), v); + } + } + + #[test] + fn test_format_crash_report_no_optional_fields() { + let detail = serde_json::json!({ + "signal": 0, "exit_status": 0, "coredump": false, "oom": false, + }); + let report = format_crash_report(&detail); + assert!(report.contains("exit(0)")); + assert!(!report.contains("Pending LLM Calls")); + assert!(!report.contains("Children at Exit")); + assert!(!report.contains("Process Tree")); + assert!(!report.contains("Process Type")); + assert!(!report.contains("Blast Radius")); + } + + #[test] + fn test_format_crash_report_oom() { + let detail = serde_json::json!({ + "signal": 9, "exit_status": 0, "coredump": false, "oom": true, + "process_type": "tool", "blast_radius": "recoverable", + }); + let report = format_crash_report(&detail); + assert!(report.contains("SIGKILL")); + assert!(report.contains("OOM Killed: yes")); + assert!(report.contains("Process Type: tool")); + assert!(report.contains("Blast Radius: recoverable")); + } + + #[test] + fn test_format_token_by_type_table_empty() { + let rows: Vec<(String, i64, i64, i64, i64)> = vec![]; + let table = format_token_by_type_table(&rows, 12); + assert!(table.contains("last 12h")); + assert!(table.contains("TOTAL")); + } + + #[test] + fn test_insert_preserves_children_on_reinsert() { + let mut tree = LineageTree::new(); + tree.insert(make_node(1, 0, ProcessType::Agent)); + tree.insert(make_node(2, 1, ProcessType::Tool)); + assert_eq!(tree.get(1).unwrap().children, vec![2]); + let mut replacement = make_node(1, 0, ProcessType::Agent); + replacement.comm = "updated".to_string(); + tree.insert(replacement); + assert_eq!(tree.get(1).unwrap().children, vec![2]); + assert_eq!(tree.get(1).unwrap().comm, "updated"); + } + + #[test] + fn test_snapshot_returns_all_nodes() { + let mut tree = LineageTree::new(); + tree.insert(make_node(1, 0, ProcessType::Agent)); + tree.insert(make_node(2, 1, ProcessType::Tool)); + tree.insert(make_node(3, 1, ProcessType::SubAgent)); + assert_eq!(tree.snapshot().len(), 3); + } + + #[test] + fn test_roots_returns_parentless_nodes() { + let mut tree = LineageTree::new(); + tree.insert(make_node(1, 999, ProcessType::Agent)); + tree.insert(make_node(2, 1, ProcessType::Tool)); + let roots = tree.roots(); + assert_eq!(roots, vec![1]); + } +} diff --git a/src/agentsight/src/probes/codex_offsets.rs b/src/agentsight/src/probes/codex_offsets.rs index fb146522a..bca63b2cf 100644 --- a/src/agentsight/src/probes/codex_offsets.rs +++ b/src/agentsight/src/probes/codex_offsets.rs @@ -115,7 +115,7 @@ fn compute_head_sha256(path: &str) -> Option { let n = f.read(&mut buf).ok()?; buf.truncate(n); let hash = Sha256::digest(&buf); - Some(hash.iter().map(|b| format!("{:02x}", b)).collect()) + Some(hash.iter().map(|b| format!("{b:02x}")).collect()) } #[cfg(test)] diff --git a/src/agentsight/src/probes/elf_buildid.rs b/src/agentsight/src/probes/elf_buildid.rs index 28bf7a02a..84aac0f99 100644 --- a/src/agentsight/src/probes/elf_buildid.rs +++ b/src/agentsight/src/probes/elf_buildid.rs @@ -106,7 +106,7 @@ pub fn read_buildid(path: &str) -> Option { } fn hex_encode(bytes: &[u8]) -> String { - bytes.iter().map(|b| format!("{:02x}", b)).collect() + bytes.iter().map(|b| format!("{b:02x}")).collect() } #[cfg(test)] diff --git a/src/agentsight/src/probes/procmon.rs b/src/agentsight/src/probes/procmon.rs index ff46f176e..f90465507 100644 --- a/src/agentsight/src/probes/procmon.rs +++ b/src/agentsight/src/probes/procmon.rs @@ -49,6 +49,7 @@ pub enum Event { tid: u32, uid: u32, timestamp_ns: u64, + exit_code: i32, comm: String, }, } @@ -87,6 +88,7 @@ impl Event { tid: raw.tid, uid: raw.uid, timestamp_ns: config::ktime_to_unix_ns(raw.timestamp_ns), + exit_code: raw.exit_code, comm, }), _ => None, diff --git a/src/agentsight/src/server/token_savings.rs b/src/agentsight/src/server/token_savings.rs index ee5a1a6bc..5d88af1ec 100644 --- a/src/agentsight/src/server/token_savings.rs +++ b/src/agentsight/src/server/token_savings.rs @@ -351,13 +351,11 @@ pub(crate) fn build_explanation( ) -> String { if category == "mcp_response" { format!( - "MCP响应压缩: 原始 {} tokens → {} tokens，压缩率 {:.1}%。后续 {} 轮LLM调用均受益，复合节省 {} tokens。", - before_tokens, after_tokens, compression_ratio, compounding_turns, compounded + "MCP响应压缩: 原始 {before_tokens} tokens → {after_tokens} tokens，压缩率 {compression_ratio:.1}%。后续 {compounding_turns} 轮LLM调用均受益，复合节省 {compounded} tokens。" ) } else { format!( - "工具输出优化: 原始 {} tokens → {} tokens，压缩率 {:.1}%。后续 {} 轮LLM调用均受益，复合节省 {} tokens。", - before_tokens, after_tokens, compression_ratio, compounding_turns, compounded + "工具输出优化: 原始 {before_tokens} tokens → {after_tokens} tokens，压缩率 {compression_ratio:.1}%。后续 {compounding_turns} 轮LLM调用均受益，复合节省 {compounded} tokens。" ) } } @@ -414,7 +412,7 @@ pub(crate) fn generate_optimization_tips( if zero_savings_sessions > 0 { tips.push(OptimizationTip { level: "info".to_string(), - title: format!("发现 {} 个未优化会话", zero_savings_sessions), + title: format!("发现 {zero_savings_sessions} 个未优化会话"), description: "部分会话消耗较高但无优化记录，可能是对应 Agent 未启用 tokenless 或工具调用较少。建议检查这些会话的 Agent 配置。".to_string(), }); } @@ -424,8 +422,7 @@ pub(crate) fn generate_optimization_tips( level: "success".to_string(), title: "节省效果优秀".to_string(), description: format!( - "当前复合节省率 {:.1}%，表现优秀！继续保持当前配置。", - grand_compounded_rate + "当前复合节省率 {grand_compounded_rate:.1}%，表现优秀！继续保持当前配置。" ), }); } else if grand_compounded_rate >= 15.0 { @@ -433,8 +430,7 @@ pub(crate) fn generate_optimization_tips( level: "success".to_string(), title: "节省效果良好".to_string(), description: format!( - "当前复合节省率 {:.1}%，已达到良好水平。可尝试调整压缩策略以进一步提升。", - grand_compounded_rate + "当前复合节省率 {grand_compounded_rate:.1}%，已达到良好水平。可尝试调整压缩策略以进一步提升。" ), }); } diff --git a/src/agentsight/src/storage/sqlite/genai/events.rs b/src/agentsight/src/storage/sqlite/genai/events.rs index ce7e1957d..0f1775893 100644 --- a/src/agentsight/src/storage/sqlite/genai/events.rs +++ b/src/agentsight/src/storage/sqlite/genai/events.rs @@ -328,6 +328,43 @@ impl GenAISqliteStore { } } + /// Aggregate token usage grouped by process_type within a time range. + pub fn token_usage_by_process_type( + &self, + start_ns: i64, + end_ns: i64, + ) -> Result, Box> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT COALESCE(process_type, 'unknown'), \ + COUNT(*), \ + COALESCE(SUM(input_tokens), 0), \ + COALESCE(SUM(output_tokens), 0), \ + COALESCE(SUM(total_tokens), 0) \ + FROM genai_events \ + WHERE event_type = 'llm_call' \ + AND status = 'complete' \ + AND start_timestamp_ns >= ?1 \ + AND start_timestamp_ns <= ?2 \ + GROUP BY COALESCE(process_type, 'unknown') \ + ORDER BY COALESCE(SUM(total_tokens), 0) DESC", + )?; + let rows = stmt.query_map(params![start_ns, end_ns], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, i64>(3)?, + row.get::<_, i64>(4)?, + )) + })?; + let mut result = Vec::new(); + for row in rows { + result.push(row?); + } + Ok(result) + } + /// Try to insert an event without size check fn try_insert_event( &self, @@ -448,12 +485,13 @@ impl GenAISqliteStore { cache_creation_tokens, cache_read_tokens, system_instructions, input_messages, output_messages, user_query, http_method, http_path, status_code, - is_sse, sse_event_count, event_json, tool_call_ids, call_kind + is_sse, sse_event_count, event_json, tool_call_ids, call_kind, + process_type ) VALUES ( ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27, ?28, ?29, ?30, ?31, ?32, - ?33, ?34, ?35, ?36, ?37, ?38, ?39, ?40, ?41 + ?33, ?34, ?35, ?36, ?37, ?38, ?39, ?40, ?41, ?42 )", params![ "llm_call", @@ -497,6 +535,7 @@ impl GenAISqliteStore { event_json, tool_call_ids, call.metadata.get("call_kind").map(|s| s.as_str()).unwrap_or("main"), + call.process_type, ], )?; } diff --git a/src/agentsight/src/storage/sqlite/genai/pending.rs b/src/agentsight/src/storage/sqlite/genai/pending.rs index 7a1f85cdb..0fc6345b2 100644 --- a/src/agentsight/src/storage/sqlite/genai/pending.rs +++ b/src/agentsight/src/storage/sqlite/genai/pending.rs @@ -43,6 +43,8 @@ pub struct PendingCallInfo { pub provider: Option, /// Call kind classification: "main" | "recap" | "web_search" pub call_kind: String, + /// Process type from lineage tree (agent/sub_agent/tool) + pub process_type: Option, } /// Data extracted from captured SSE events for enriching a pending record. @@ -73,14 +75,14 @@ impl GenAISqliteStore { start_timestamp_ns, pid, process_name, agent_name, http_method, http_path, is_sse, input_messages, system_instructions, user_query, - model, provider, call_kind, + model, provider, call_kind, process_type, event_json ) VALUES ( 'llm_call', 'pending', ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, - ?16, ?17, ?18, + ?16, ?17, ?18, ?19, '{}' )", params![ @@ -102,6 +104,7 @@ impl GenAISqliteStore { info.model, info.provider, info.call_kind, + info.process_type, ], )?; Ok(()) diff --git a/src/agentsight/src/storage/sqlite/genai/schema.rs b/src/agentsight/src/storage/sqlite/genai/schema.rs index 8c55b5665..3826c0f98 100644 --- a/src/agentsight/src/storage/sqlite/genai/schema.rs +++ b/src/agentsight/src/storage/sqlite/genai/schema.rs @@ -185,6 +185,9 @@ impl GenAISqliteStore { "idx_genai_call_kind" ); + // v7: process type from lineage tree (agent/sub_agent/tool) + ensure_col!("process_type", "TEXT"); + Ok(()) } diff --git a/src/agentsight/src/storage/sqlite/genai/tests.rs b/src/agentsight/src/storage/sqlite/genai/tests.rs index fd977e47a..259ef650f 100644 --- a/src/agentsight/src/storage/sqlite/genai/tests.rs +++ b/src/agentsight/src/storage/sqlite/genai/tests.rs @@ -659,6 +659,7 @@ fn test_insert_pending() { model: Some("gpt-4".to_string()), provider: Some("openai".to_string()), call_kind: "main".to_string(), + process_type: None, }; store.insert_pending(&info).unwrap(); let conn = store.conn.lock().unwrap(); @@ -830,6 +831,7 @@ fn make_store_with_pending(records: &[(&str, &str, &str, &str, i64)]) -> GenAISq model: Some("gpt-4".to_string()), provider: Some("openai".to_string()), call_kind: kind.to_string(), + process_type: None, }; store.insert_pending(&info).unwrap(); } @@ -906,3 +908,132 @@ fn test_list_traces_call_kind_filter_with_time_range() { .unwrap(); assert_eq!(traces_all.len(), 2); } + +#[test] +fn test_token_usage_by_process_type() { + let path = std::env::temp_dir().join(format!( + "test_token_by_ptype_{}.db", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let store = GenAISqliteStore::new_with_path(&path).unwrap(); + + // Insert events with different process_types + let mut call1 = LLMCall::new( + "c1".into(), + 1000, + "openai".into(), + "gpt-4".into(), + LLMRequest { + messages: vec![], + temperature: None, + max_tokens: None, + frequency_penalty: None, + presence_penalty: None, + top_p: None, + top_k: None, + seed: None, + stop_sequences: None, + stream: false, + tools: None, + raw_body: None, + }, + 100, + "agent".into(), + ); + call1.process_type = Some("agent".into()); + call1.token_usage = Some(crate::genai::semantic::TokenUsage { + input_tokens: 100, + output_tokens: 50, + total_tokens: 150, + cache_creation_input_tokens: None, + cache_read_input_tokens: None, + }); + call1.end_timestamp_ns = 2000; + call1.duration_ns = 1000; + store + .store_event(&GenAISemanticEvent::LLMCall(call1)) + .unwrap(); + + let mut call2 = LLMCall::new( + "c2".into(), + 1500, + "openai".into(), + "gpt-4".into(), + LLMRequest { + messages: vec![], + temperature: None, + max_tokens: None, + frequency_penalty: None, + presence_penalty: None, + top_p: None, + top_k: None, + seed: None, + stop_sequences: None, + stream: false, + tools: None, + raw_body: None, + }, + 200, + "tool".into(), + ); + call2.process_type = Some("tool".into()); + call2.token_usage = Some(crate::genai::semantic::TokenUsage { + input_tokens: 30, + output_tokens: 10, + total_tokens: 40, + cache_creation_input_tokens: None, + cache_read_input_tokens: None, + }); + call2.end_timestamp_ns = 2500; + call2.duration_ns = 1000; + store + .store_event(&GenAISemanticEvent::LLMCall(call2)) + .unwrap(); + + // Insert a pending-status event that should be EXCLUDED by the query's + // `status = 'complete'` filter. Without this negative case, deleting the + // filter from the SQL would not cause the test to fail. + let pending = super::pending::PendingCallInfo { + call_id: "pending-1".into(), + trace_id: None, + conversation_id: None, + session_id: None, + start_timestamp_ns: 2000, + pid: 300, + process_name: "agent".into(), + agent_name: Some("pending-agent".into()), + http_method: Some("POST".into()), + http_path: Some("/v1/chat".into()), + is_sse: true, + input_messages: None, + system_instructions: None, + user_query: None, + model: None, + provider: None, + call_kind: "main".into(), + process_type: Some("agent".into()), + }; + store.insert_pending(&pending).unwrap(); + + let result = store.token_usage_by_process_type(0, 10000).unwrap(); + assert_eq!( + result.len(), + 2, + "pending row must be excluded by status='complete' filter" + ); + // Sorted by total_tokens DESC: agent first + assert_eq!(result[0].0, "agent"); + assert_eq!( + result[0].1, 1, + "call_count for agent must be 1 (pending row excluded)" + ); + assert_eq!(result[0].4, 150); // total_tokens + assert_eq!(result[1].0, "tool"); + assert_eq!(result[1].1, 1); + assert_eq!(result[1].4, 40); + + std::fs::remove_file(&path).ok(); +} diff --git a/src/agentsight/src/unified.rs b/src/agentsight/src/unified.rs index e7a5798a6..a257af2f7 100644 --- a/src/agentsight/src/unified.rs +++ b/src/agentsight/src/unified.rs @@ -110,6 +110,8 @@ pub struct AgentSight { process_killer: Arc, /// Cached feature flags so runtime paths can check them without the config. features: crate::config::FeatureFlags, + /// Blood lineage tree tracking process parent-child relationships and type classification + lineage_tree: Arc>, } /// GenAI events waiting for session_id resolution via ResponseSessionMapper. @@ -331,8 +333,7 @@ impl AgentSight { if let Some(ref path) = sysom_logtail_path { log::info!( - "SLS sysom mode detected (path={}), skipping SQLite and default SLS exporter", - path + "SLS sysom mode detected (path={path}), skipping SQLite and default SLS exporter" ); if logtail_currently_enabled { let exporter = LogtailExporter::new_with_fixed_path( @@ -544,6 +545,23 @@ impl AgentSight { deadloop_kill_after_count: config.deadloop_kill_after_count, process_killer: Arc::new(crate::utils::process::LibcProcessKiller), features: config.features.clone(), + lineage_tree: { + let tree = crate::lineage::LineageTree::new(); + let tree = Arc::new(std::sync::RwLock::new(tree)); + if let Ok(mut t) = tree.write() { + for agent in &existing_agents { + t.insert_and_classify( + agent.pid, + 1, + &agent.agent_info.name, + Some(agent.agent_info.name.clone()), + false, + true, + ); + } + } + tree + }, }) } @@ -634,6 +652,11 @@ impl AgentSight { return None; } + // Maintain lineage tree from proctrace exec/exit events + if let Event::Proc(ref proc_event) = event { + self.update_lineage_from_proc(proc_event); + } + // Handle FileWatch events via callback (not through the pipeline) if let Event::FileWatch(ref fw_event) = event { self.handle_filewatch_event(fw_event); @@ -715,12 +738,28 @@ impl AgentSight { let mut analysis_results = self.analyzer.analyze_aggregated(agg_result); // Build GenAI semantic events AND pending info in one pass - let (output, pending_info) = self.genai_builder.build_with_pending( + let (mut output, mut pending_info) = self.genai_builder.build_with_pending( &analysis_results, &self.response_mapper, &self.pid_agent_name_cache, ); + // Enrich events with process_type from lineage tree + if let Ok(tree) = self.lineage_tree.read() { + for event in &mut output.events { + if let crate::genai::semantic::GenAISemanticEvent::LLMCall(call) = event { + call.process_type = tree + .get(call.pid as u32) + .map(|n| n.process_type.as_str().to_string()); + } + } + if let Some(ref mut info) = pending_info { + info.process_type = tree + .get(info.pid as u32) + .map(|n| n.process_type.as_str().to_string()); + } + } + // Backfill TokenRecord.agent from pid_agent_name_cache, falling back to comm for ar in &mut analysis_results { if let crate::analyzer::AnalysisResult::Token(t) = ar { @@ -840,7 +879,9 @@ impl AgentSight { use crate::probes::procmon::Event as ProcMonEvent; match event { - ProcMonEvent::Exec { pid, comm, .. } => { + ProcMonEvent::Exec { + pid, ppid, comm, .. + } => { // Read cmdline for deny-check and custom matching let cmdline_args = crate::discovery::scanner::read_cmdline(&format!("/proc/{pid}/cmdline")); @@ -852,20 +893,112 @@ impl AgentSight { } // Phase 2: check if this is a known agent and start tracking - if let Some(agent) = self.scanner.on_process_create(*pid, comm) { + let matched = self.scanner.on_process_create(*pid, comm); + + if let Some(agent) = matched { let agent_name = agent.agent_info.name.clone(); self.pid_agent_name_cache.put(*pid, agent_name.clone()); self.attach_process(*pid, &agent_name); + // proctrace does not emit an exec event for a scanner-matched + // root (it was not yet in traced_processes when it execed). + self.ensure_lineage_node(*pid, *ppid, comm, Some(agent_name), true); + } else { + // PID reuse: new non-agent process inherited a recycled PID. + // Clear stale cache so record_exec won't misclassify it. + self.pid_agent_name_cache.pop(pid); } } - ProcMonEvent::Exit { pid, .. } => { - // Remove from tracking if it was an agent + ProcMonEvent::Exit { pid, exit_code, .. } => { if let Some(agent) = self.scanner.on_process_exit(*pid) { let agent_name = agent.agent_info.name.clone(); self.detach_process(*pid, &agent_name); - self.handle_agent_crash_detection(*pid, &agent_name); + self.handle_agent_crash_detection(*pid, &agent_name, *exit_code); + } else if matches!( + crate::lineage::classify_exit(*exit_code), + crate::lineage::ExitClassification::SignalCrash { .. } + ) { + // SubAgent/Tool signal crash only (not AbnormalExit — child + // exit(1) after parent dies is normal cleanup, not a crash). + let child_crash_name = self.lineage_tree.read().ok().and_then(|tree| { + tree.get(*pid).and_then(|node| { + if matches!( + node.process_type, + crate::lineage::ProcessType::SubAgent + | crate::lineage::ProcessType::Tool + ) { + Some( + tree.root_agent_ancestor(*pid) + .and_then(|a| a.agent_name.clone()) + .unwrap_or_else(|| node.comm.clone()), + ) + } else { + None + } + }) + }); + if let Some(agent_name) = child_crash_name { + self.handle_agent_crash_detection(*pid, &agent_name, *exit_code); + } } + // Clean lineage tree (root agents may not be in proctrace child_pids) + if let Ok(mut tree) = self.lineage_tree.write() { + tree.remove(*pid); + } + } + } + } + + /// Ensure a lineage node exists and is correctly classified after procmon + /// provides agent info. Handles the race where the proctrace exec event is + /// missed because the process was not yet in traced_processes when it execed. + fn ensure_lineage_node( + &mut self, + pid: u32, + ppid: u32, + comm: &str, + agent_name: Option, + matches_agent: bool, + ) { + if let Ok(mut tree) = self.lineage_tree.write() { + tree.insert_and_classify(pid, ppid, comm, agent_name, false, matches_agent); + } + } + + /// Update lineage tree from proctrace exec/exit events + fn update_lineage_from_proc(&mut self, event: &crate::probes::proctrace::VariableEvent) { + use crate::probes::proctrace::VariableEvent; + + match event { + VariableEvent::Exec { header, .. } => { + let comm = event.comm_str(); + let agent_name = self.pid_agent_name_cache.get(&header.pid).cloned(); + if let Ok(mut tree) = self.lineage_tree.write() { + let ptype = tree.record_exec( + header.pid, + header.ppid, + &comm, + agent_name, + header.timestamp_ns, + ); + log::debug!( + "Lineage: added pid={} ppid={} type={ptype:?}", + header.pid, + header.ppid, + ); + } + } + VariableEvent::Exit { header, .. } => { + // Do NOT remove from lineage tree here — procmon Exit handler + // does the authoritative remove (line ~884). Removing here first + // would race: if proctrace exit arrives before procmon exit for + // the same PID, tree.get(pid) in the procmon SubAgent/Tool crash + // detection branch returns None, silently dropping the crash event. + log::debug!( + "Lineage: proctrace exit pid={} (deferred to procmon handler)", + header.pid, + ); } + _ => {} } } @@ -1159,12 +1292,17 @@ impl AgentSight { /// Immediate crash detection when a tracked agent process exits. /// - /// Called from `ProcMon::Exit` handler. Drains in-flight connections for - /// the PID, persists them as pending calls, then generates an `agent_crash` - /// interruption event if any pending calls exist. - fn handle_agent_crash_detection(&mut self, pid: u32, agent_name: &str) { + /// Uses the kernel exit_code (signal/status) as ground truth instead of + /// the pending-calls heuristic. Signal-killed or non-zero-exit processes + /// produce an `agent_crash` event regardless of in-flight LLM calls. + fn handle_agent_crash_detection(&mut self, pid: u32, agent_name: &str, exit_code: i32) { use crate::aggregator::ConnectionState; - use crate::interruption::{InterruptionEvent, InterruptionType, was_pid_oom_killed}; + use crate::interruption::{ + InterruptionEvent, InterruptionType, Severity, was_pid_oom_killed, + }; + use crate::lineage::classify_exit; + + let classification = classify_exit(exit_code); // 1. Drain in-flight connections for this PID from the aggregator let drained = self.aggregator.drain_connections_for_pid(pid); @@ -1192,7 +1330,15 @@ impl AgentSight { } } - // 3. Query all pending calls for this PID (including any persisted earlier) + // 3. Exit classification decides crash vs normal — not pending calls + if !classification.is_crash() { + log::debug!( + "[CrashDetect] Agent {agent_name} (pid={pid}) exited normally (exit_code={exit_code})", + ); + return; + } + + // 4. Query pending calls for context (not for crash decision) let pending_calls = if let Some(ref store) = self.genai_sqlite_store { store .list_pending_for_pids(&[pid as i32]) @@ -1201,45 +1347,90 @@ impl AgentSight { vec![] }; - if pending_calls.is_empty() { - log::debug!( - "[CrashDetect] Agent {agent_name} (pid={pid}) exited with no pending calls — normal shutdown", - ); - return; - } - - // 4. Generate agent_crash interruption event + // 5. Generate agent_crash interruption event if let Some(ref istore) = self.interruption_store { let now_ns = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_nanos() as i64) .unwrap_or(0); - let is_oom = was_pid_oom_killed(pid as i32); + let signal = (exit_code & 0x7f) as u32; + let coredump = (exit_code >> 7) & 1 != 0; + let exit_status = ((exit_code >> 8) & 0xff) as u32; + let is_oom = signal == 9 && was_pid_oom_killed(pid as i32); - // Group by (session_id, conversation_id) to produce one event per conversation + let mut severity = match &classification { + crate::lineage::ExitClassification::SignalCrash { .. } => Severity::Critical, + crate::lineage::ExitClassification::AbnormalExit { .. } => Severity::High, + _ => Severity::Medium, + }; + // Downgrade severity for SubAgent/Tool crashes (recoverable) + if let Ok(tree) = self.lineage_tree.read() { + if let Some(node) = tree.get(pid) { + match node.process_type { + crate::lineage::ProcessType::Tool => { + severity = Severity::Medium; + } + crate::lineage::ProcessType::SubAgent => { + if severity == Severity::Critical { + severity = Severity::High; + } + } + _ => {} + } + } + } + + // Group by (session_id, conversation_id) or use a single "no-context" group let mut by_conv: std::collections::HashMap< (Option, Option), Vec, > = std::collections::HashMap::new(); - for (call_id, session_id, _trace_id, conversation_id) in &pending_calls { - by_conv - .entry((session_id.clone(), conversation_id.clone())) - .or_default() - .push(call_id.clone()); + if pending_calls.is_empty() { + by_conv.insert((None, None), vec![]); + } else { + for (call_id, session_id, _trace_id, conversation_id) in &pending_calls { + by_conv + .entry((session_id.clone(), conversation_id.clone())) + .or_default() + .push(call_id.clone()); + } } for ((session_id, conversation_id), call_ids) in &by_conv { let mut detail = serde_json::json!({ "pid": pid, "agent_name": agent_name, + "signal": signal, + "exit_code": exit_code, + "coredump": coredump, + "exit_status": exit_status, "call_ids": call_ids, "source": "trace_procmon_exit", }); if is_oom { detail["oom"] = serde_json::json!(true); } - let event = InterruptionEvent::new( + if let Ok(tree) = self.lineage_tree.read() { + if let Some(subtree) = tree.subtree(pid) { + detail["process_tree"] = serde_json::to_value(&subtree).unwrap_or_default(); + } + let children = tree.children_at_exit(pid); + if !children.is_empty() { + detail["children_at_exit"] = serde_json::json!(children); + } + if let Some(node) = tree.get(pid) { + let pt = node.process_type; + detail["process_type"] = serde_json::json!(pt.as_str()); + detail["blast_radius"] = serde_json::json!(match pt { + crate::lineage::ProcessType::Agent => "total_session_loss", + crate::lineage::ProcessType::SubAgent => "partial", + crate::lineage::ProcessType::Tool => "recoverable", + _ => "unknown", + }); + } + } + let mut event = InterruptionEvent::new( InterruptionType::AgentCrash, session_id.clone(), None, @@ -1250,15 +1441,17 @@ impl AgentSight { now_ns, Some(detail), ); + event.severity = severity.clone(); if let Err(e) = istore.insert(&event) { log::warn!("[CrashDetect] Failed to record agent_crash for pid={pid}: {e}"); } else { log::info!( - "[CrashDetect] Recorded agent_crash for {} (pid={}, session={:?}, conversation={:?}, {} call(s), oom={})", + "[CrashDetect] Recorded agent_crash for {} (pid={}, signal={}, exit_status={}, session={:?}, {} call(s), oom={})", agent_name, pid, + signal, + exit_status, session_id, - conversation_id, call_ids.len(), is_oom, ); @@ -1266,13 +1459,15 @@ impl AgentSight { crate::genai::logtail::export_interruption_events(std::slice::from_ref(&event)); } - // Mark all pending calls for this PID as interrupted - if let Some(ref store) = self.genai_sqlite_store { - let itype = if is_oom { "oom_crash" } else { "agent_crash" }; - if let Err(e) = store.mark_pending_interrupted_for_pid(pid as i32, itype) { - log::warn!( - "[CrashDetect] Failed to mark pending interrupted for pid={pid}: {e}" - ); + // Mark pending calls as interrupted + if !pending_calls.is_empty() { + if let Some(ref store) = self.genai_sqlite_store { + let itype = if is_oom { "oom_crash" } else { "agent_crash" }; + if let Err(e) = store.mark_pending_interrupted_for_pid(pid as i32, itype) { + log::warn!( + "[CrashDetect] Failed to mark pending interrupted for pid={pid}: {e}" + ); + } } } } @@ -1968,6 +2163,7 @@ mod tests { pid: 1234, process_name: "test".to_string(), agent_name: Some("test-agent".to_string()), + process_type: None, metadata: std::collections::HashMap::new(), } } @@ -1991,6 +2187,7 @@ mod tests { model: Some("gpt-4".to_string()), provider: Some("openai".to_string()), call_kind: "main".to_string(), + process_type: None, } } diff --git a/src/agentsight/tests/common/mod.rs b/src/agentsight/tests/common/mod.rs index 63184aa42..4a24aebd8 100644 --- a/src/agentsight/tests/common/mod.rs +++ b/src/agentsight/tests/common/mod.rs @@ -244,6 +244,7 @@ pub fn make_test_llm_call(call_id: &str) -> agentsight::genai::LLMCall { pid: 1234, process_name: "test".to_string(), agent_name: Some("test-agent".to_string()), + process_type: None, metadata: HashMap::new(), } } diff --git a/src/agentsight/tests/memory_storage.rs b/src/agentsight/tests/memory_storage.rs index 88676023b..2a07a56b9 100644 --- a/src/agentsight/tests/memory_storage.rs +++ b/src/agentsight/tests/memory_storage.rs @@ -23,7 +23,7 @@ fn make_llm_call(idx: usize) -> LLMCall { messages: vec![InputMessage { role: "user".to_string(), parts: vec![MessagePart::Text { - content: format!("hello world request number {}", idx), + content: format!("hello world request number {idx}"), }], name: None, }], @@ -44,7 +44,7 @@ fn make_llm_call(idx: usize) -> LLMCall { messages: vec![OutputMessage { role: "assistant".to_string(), parts: vec![MessagePart::Text { - content: format!("hello world response number {}", idx), + content: format!("hello world response number {idx}"), }], name: None, finish_reason: Some("stop".to_string()), @@ -54,7 +54,7 @@ fn make_llm_call(idx: usize) -> LLMCall { }; let mut call = LLMCall::new( - format!("call-{}", idx), + format!("call-{idx}"), 0, "openai".to_string(), "gpt-4o".to_string(), @@ -142,8 +142,8 @@ fn sqlite_batch_lowers_write_overhead() { }), ); - println!("no batch: {:?}", no_batch); - println!("with batch: {:?}", with_batch); + println!("no batch: {no_batch:?}"); + println!("with batch: {with_batch:?}"); // NOTE: We do NOT assert batch is faster than no-batch because in CI // environments (shared runners, cold disk cache) the difference can be