diff --git a/src/agentsight/build.rs b/src/agentsight/build.rs index f6b420a64..b6691c6c6 100644 --- a/src/agentsight/build.rs +++ b/src/agentsight/build.rs @@ -61,7 +61,11 @@ fn main() { // Generate tcpsniff skeleton (no header — reuses sslsniff.h event format) generate_skeleton(&mut out, "tcpsniff"); - + + // Generate schedmon skeleton and bindings + generate_skeleton(&mut out, "schedmon"); + generate_header(&mut out, "schedmon"); + // generate_header(&mut out, "frametypes"); // generate_header(&mut out, "errors"); // generate_header(&mut out, "stackdeltatypes"); diff --git a/src/agentsight/src/bin/cli/trace.rs b/src/agentsight/src/bin/cli/trace.rs index 8a5dec188..e68864c1b 100644 --- a/src/agentsight/src/bin/cli/trace.rs +++ b/src/agentsight/src/bin/cli/trace.rs @@ -22,6 +22,10 @@ pub struct TraceCommand { #[structopt(long)] pub enable_filewatch: bool, + /// Enable scheduler (idle-burst-idle cgroup CPU weight management) + #[structopt(long)] + pub enable_scheduler: bool, + /// Path to JSON configuration file #[structopt(short, long, default_value = "/etc/agentsight/config.json")] pub config: String, @@ -72,6 +76,7 @@ impl TraceCommand { let config = AgentsightConfig::new() .set_verbose(self.verbose) .set_enable_filewatch(self.enable_filewatch) + .set_enable_scheduler(self.enable_scheduler) .set_config_path(std::path::PathBuf::from(&self.config)); // Create AgentSight (auto-attaches probes and starts polling) diff --git a/src/agentsight/src/bpf/common.h b/src/agentsight/src/bpf/common.h index c99dbb79a..fc8b025e5 100644 --- a/src/agentsight/src/bpf/common.h +++ b/src/agentsight/src/bpf/common.h @@ -23,6 +23,7 @@ typedef enum { EVENT_SOURCE_FILEWATCH = 4, // File watch events (filewatch) EVENT_SOURCE_FILEWRITE = 5, // File write events (filewrite) EVENT_SOURCE_UDPDNS = 6, // UDP DNS query events (udpdns) + EVENT_SOURCE_SCHED = 7, // Scheduler state events (schedmon) } event_source_t; // Common event header - every ringbuffer event MUST start with this diff --git a/src/agentsight/src/bpf/procmon.bpf.c b/src/agentsight/src/bpf/procmon.bpf.c index c5e60d3ac..6161f79a2 100644 --- a/src/agentsight/src/bpf/procmon.bpf.c +++ b/src/agentsight/src/bpf/procmon.bpf.c @@ -44,7 +44,7 @@ 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; @@ -78,7 +78,7 @@ 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; diff --git a/src/agentsight/src/bpf/schedmon.bpf.c b/src/agentsight/src/bpf/schedmon.bpf.c new file mode 100644 index 000000000..e05d3e256 --- /dev/null +++ b/src/agentsight/src/bpf/schedmon.bpf.c @@ -0,0 +1,115 @@ +// SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) +// Copyright (c) 2025 AgentSight Project +// +// Scheduler monitor BPF program +// Detects idle/active state transitions for traced Agent processes via the +// BTF-typed sched_switch / sched_wakeup tracepoints. +// +// Using tp_btf (rather than the format-struct tracepoints) gives direct typed +// access to the task_struct, which lets us: +// 1. read the *raw* prev->__state (TASK_RUNNING == 0) and the `preempt` flag, +// so a task that is merely preempted while still runnable is NOT treated as +// going to sleep — only a genuinely blocking task emits SLEEP; +// 2. filter by tgid (thread-group id) so we only watch traced Agent families, +// but emit per-tid (thread id), so userspace can tell a family is ACTIVE +// while ANY of its threads is runnable (a multithreaded process commonly has +// some threads sleeping while others run). +// +// Events are rate-limited per-tid to avoid flooding the shared ring buffer. +#include "vmlinux.h" +#include +#include +#include +#include "schedmon.h" +#include "common.h" + +struct pid_sched_state { + u8 last_state; // sched_event_type +}; + +// Per-tid last emitted state, used to deduplicate consecutive same-state events +// (a running thread does not re-emit WAKEUP). We deliberately do NOT apply a +// time-based cooldown: a thread can legitimately wake and immediately block again +// (WAKEUP then SLEEP within microseconds), and dropping that SLEEP would strand +// the thread in the userspace runnable set, pinning its family ACTIVE forever. +// LRU so it self-evicts (schedmon has no thread-exit hook) and update never fails. +struct { + __uint(type, BPF_MAP_TYPE_LRU_HASH); + __uint(max_entries, MAX_TRACED_PROCESSES); + __type(key, u32); + __type(value, struct pid_sched_state); +} pid_sched_state SEC(".maps"); + +static __always_inline int +emit_sched_event(u32 tgid, u32 tid, u64 now, u8 event_type) +{ + struct pid_sched_state *st = bpf_map_lookup_elem(&pid_sched_state, &tid); + if (st) { + if (st->last_state == event_type) + return 0; + st->last_state = event_type; + } else { + struct pid_sched_state new_st = { + .last_state = event_type, + }; + bpf_map_update_elem(&pid_sched_state, &tid, &new_st, BPF_ANY); + } + + struct sched_event *ev = bpf_ringbuf_reserve(&rb, sizeof(*ev), 0); + if (!ev) + return 0; + + ev->source = EVENT_SOURCE_SCHED; + ev->tgid = tgid; + ev->tid = tid; + ev->timestamp_ns = now; + ev->event_type = event_type; + ev->pad[0] = 0; + ev->pad[1] = 0; + ev->pad[2] = 0; + + bpf_ringbuf_submit(ev, 0); + return 0; +} + +// sched_switch: prev (one thread) is going off-CPU. +// Emit SLEEP only when prev is actually blocking (not merely preempted while +// still runnable). Filtered by tgid, emitted per-tid. +SEC("tp_btf/sched_switch") +int BPF_PROG(handle_sched_switch, bool preempt, struct task_struct *prev, + struct task_struct *next) +{ + // Preempted tasks stay runnable — they are not going to sleep. + if (preempt) + return 0; + + // prev->__state has already been set to the target state by the scheduler. + // TASK_RUNNING (0) means it is still runnable (e.g. yielded); skip it. + unsigned int state = BPF_CORE_READ(prev, __state); + if (state == 0) + return 0; + + u32 tgid = BPF_CORE_READ(prev, tgid); + u32 *traced = bpf_map_lookup_elem(&traced_processes, &tgid); + if (!traced) + return 0; + + u32 tid = BPF_CORE_READ(prev, pid); + return emit_sched_event(tgid, tid, bpf_ktime_get_ns(), SCHED_EVENT_SLEEP); +} + +// sched_wakeup: task p (one thread) is being woken up. Any thread waking makes +// its family ACTIVE. Filtered by tgid, emitted per-tid. +SEC("tp_btf/sched_wakeup") +int BPF_PROG(handle_sched_wakeup, struct task_struct *p) +{ + u32 tgid = BPF_CORE_READ(p, tgid); + u32 *traced = bpf_map_lookup_elem(&traced_processes, &tgid); + if (!traced) + return 0; + + u32 tid = BPF_CORE_READ(p, pid); + return emit_sched_event(tgid, tid, bpf_ktime_get_ns(), SCHED_EVENT_WAKEUP); +} + +char LICENSE[] SEC("license") = "GPL"; diff --git a/src/agentsight/src/bpf/schedmon.h b/src/agentsight/src/bpf/schedmon.h new file mode 100644 index 000000000..46b6167d5 --- /dev/null +++ b/src/agentsight/src/bpf/schedmon.h @@ -0,0 +1,35 @@ +// SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) +// Copyright (c) 2025 AgentSight Project +// +// Scheduler monitor BPF program header +// Detects idle/active state transitions for traced Agent processes +#ifndef __SCHEDMON_H +#define __SCHEDMON_H + +#define TASK_COMM_LEN 16 + +typedef signed char s8; +typedef unsigned char u8; +typedef signed short s16; +typedef unsigned short u16; +typedef signed int s32; +typedef unsigned int u32; +typedef signed long long s64; +typedef unsigned long long u64; + +enum sched_event_type { + SCHED_EVENT_SLEEP = 1, + SCHED_EVENT_WAKEUP = 2, +}; + +struct sched_event { + u32 source; // EVENT_SOURCE_SCHED + u32 tgid; // thread-group id — identifies the Agent family + u32 tid; // thread id — tracked individually so a family is ACTIVE + // while ANY of its threads is runnable + u64 timestamp_ns; + u8 event_type; // enum sched_event_type + u8 pad[3]; +}; + +#endif /* __SCHEDMON_H */ diff --git a/src/agentsight/src/config.rs b/src/agentsight/src/config.rs index 1bf37f436..a6569f3c2 100644 --- a/src/agentsight/src/config.rs +++ b/src/agentsight/src/config.rs @@ -235,6 +235,8 @@ struct JsonFullConfig { encryption: Option, #[serde(default)] runtime: Option, + #[serde(default)] + scheduler: Option, } /// Runtime 动态配置区段（支持热加载，无需重启） @@ -245,6 +247,15 @@ pub struct JsonRuntime { pub sls_logtail_path: Option, } +/// Activity monitor config (idle/active detection for observability) +#[derive(serde::Deserialize)] +struct JsonScheduler { + #[serde(default)] + enabled: Option, + #[serde(default)] + idle_threshold_ms: Option, +} + /// 加密配置：可选公钥（PEM 字符串）或公钥文件路径 #[derive(serde::Deserialize)] struct JsonEncryption { @@ -418,6 +429,10 @@ pub struct AgentsightConfig { pub poll_timeout_ms: u64, /// Enable file watch probe (monitors .jsonl file opens from traced processes) pub enable_filewatch: bool, + /// Enable activity monitor (agent idle/active state detection via schedmon BPF) + pub enable_scheduler: bool, + /// Activity monitor configuration + pub activity_config: crate::scheduler::ActivityConfig, /// TCP capture targets for plain HTTP capture (empty = disabled). /// Each entry specifies destination IP, port, or both. pub tcp_targets: Vec, @@ -490,6 +505,8 @@ impl Default for AgentsightConfig { target_uid: None, poll_timeout_ms: DEFAULT_POLL_TIMEOUT_MS, enable_filewatch: false, + enable_scheduler: false, + activity_config: crate::scheduler::ActivityConfig::default(), tcp_targets: Vec::new(), // HTTP/Aggregation defaults @@ -580,6 +597,13 @@ impl AgentsightConfig { self } + /// Set enable_scheduler (activity monitor) + pub fn set_enable_scheduler(mut self, enable: bool) -> Self { + self.enable_scheduler = enable; + self.activity_config.enabled = enable; + self + } + /// Set connection capacity pub fn set_connection_capacity(mut self, capacity: usize) -> Self { self.connection_capacity = capacity; @@ -643,6 +667,23 @@ impl AgentsightConfig { } } + // Load scheduler config + if let Some(sched) = parsed.scheduler.take() { + if let Some(enabled) = sched.enabled { + if enabled != self.enable_scheduler { + log::warn!( + "config scheduler.enabled={} overrides --enable-scheduler={}", + enabled, self.enable_scheduler + ); + } + self.enable_scheduler = enabled; + self.activity_config.enabled = enabled; + } + if let Some(t) = sched.idle_threshold_ms { + self.activity_config.idle_threshold_ms = t; + } + } + let (cmdline_rules, https_rules, http_targets) = extract_rules(&parsed); self.cmdline_rules.extend(cmdline_rules); self.https_rules.extend(https_rules); diff --git a/src/agentsight/src/discovery/scanner.rs b/src/agentsight/src/discovery/scanner.rs index f1873db90..2fdb97bb0 100644 --- a/src/agentsight/src/discovery/scanner.rs +++ b/src/agentsight/src/discovery/scanner.rs @@ -282,6 +282,29 @@ impl AgentScanner { } } +/// Read the parent PID of a process from /proc/[pid]/stat +pub fn read_ppid(pid: u32) -> Option { + let data = fs::read_to_string(format!("/proc/{}/stat", pid)).ok()?; + // Format: pid (comm) state ppid ... — ppid is the 4th field. + // comm may contain spaces/parens, so find the closing ')' first. + let after_comm = data.rsplit_once(')')?.1; + let ppid_str = after_comm.split_whitespace().nth(1)?; + ppid_str.parse().ok() +} + +/// Check if a process has AGENT_MODE=1 in its environment +/// +/// Reads /proc/[pid]/environ and looks for the AGENT_MODE=1 variable. +pub fn has_agent_mode(pid: u32) -> bool { + let path = format!("/proc/{}/environ", pid); + match fs::read(&path) { + Ok(data) => data + .split(|&b| b == 0) + .any(|entry| entry == b"AGENT_MODE=1"), + Err(_) => false, + } +} + /// Read and parse cmdline file /// /// The cmdline file contains arguments separated by null bytes. diff --git a/src/agentsight/src/event.rs b/src/agentsight/src/event.rs index fde41b4c1..449deff03 100644 --- a/src/agentsight/src/event.rs +++ b/src/agentsight/src/event.rs @@ -4,6 +4,7 @@ use crate::probes::procmon::Event as ProcMonEvent; use crate::probes::filewatch::FileWatchEvent; use crate::probes::filewrite::FileWriteEvent; use crate::probes::udpdns::UdpDnsEvent; +use crate::probes::schedmon::SchedEvent; /// Unified event type that can represent any probe event /// @@ -16,6 +17,7 @@ pub enum Event { FileWatch(FileWatchEvent), FileWrite(FileWriteEvent), UdpDns(UdpDnsEvent), + Sched(SchedEvent), } impl Event { @@ -28,6 +30,7 @@ impl Event { Event::FileWatch(_) => "FileWatch", Event::FileWrite(_) => "FileWrite", Event::UdpDns(_) => "UdpDns", + Event::Sched(_) => "Sched", } } } @@ -110,6 +113,19 @@ impl Event { _ => None, } } + + /// Check if this is a scheduler state event + pub fn is_sched(&self) -> bool { + matches!(self, Event::Sched(_)) + } + + /// Get scheduler event if this is one + pub fn as_sched(&self) -> Option<&SchedEvent> { + match self { + Event::Sched(e) => Some(e), + _ => None, + } + } } #[cfg(test)] diff --git a/src/agentsight/src/ffi.rs b/src/agentsight/src/ffi.rs index ee0d9433b..d8fd9302c 100644 --- a/src/agentsight/src/ffi.rs +++ b/src/agentsight/src/ffi.rs @@ -680,6 +680,8 @@ fn ffi_background_thread( if sight.try_process().is_none() { // No event available — flush any timed-out pending GenAI events sight.flush_expired_pending_genai(); + // Finalize debounced scheduler idle transitions (same as run()). + sight.on_idle_tick(); std::thread::sleep(std::time::Duration::from_millis(10)); } } diff --git a/src/agentsight/src/lib.rs b/src/agentsight/src/lib.rs index c730a3d66..0a99ebf9a 100644 --- a/src/agentsight/src/lib.rs +++ b/src/agentsight/src/lib.rs @@ -37,11 +37,13 @@ pub mod storage; pub mod chrome_trace; pub mod discovery; pub mod health; +pub mod lineage; pub mod tokenizer; pub mod genai; pub mod atif; pub mod response_map; pub mod interruption; +pub mod scheduler; pub mod skill_metrics; pub mod utils; #[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..90cf9cbbf --- /dev/null +++ b/src/agentsight/src/lineage/mod.rs @@ -0,0 +1,358 @@ +//! 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 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ProcessType { + Unknown, + Agent, + SubAgent, + Tool, + 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, + } + } +} + +/// 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 +pub struct LineageTree { + nodes: HashMap, +} + +impl LineageTree { + pub fn new() -> Self { + Self { + nodes: HashMap::new(), + } + } + + /// Insert or update a node. Automatically maintains parent→child links. + pub fn insert(&mut self, node: LineageNode) { + let pid = node.pid; + let ppid = node.ppid; + + // Add this pid as a child of its parent + if let Some(parent) = self.nodes.get_mut(&ppid) { + if !parent.children.contains(&pid) { + parent.children.push(pid); + } + } + + 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); + parent.children.extend(node.children.iter()); + } + + 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) + } + + /// 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 + AGENT_MODE=1 in env → Agent (new root) + /// 3. 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. + 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 has_agent_mode_env { + ProcessType::Agent + } else { + ProcessType::Unknown + } + } + }; + + if let Some(node) = self.nodes.get_mut(&pid) { + node.process_type = process_type; + } + } + + /// 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 (for REST API) + 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() + } + + /// Walk up from `pid` to find the Agent root of its family. + /// Returns the PID of the nearest Agent ancestor, or `pid` itself if it is an Agent. + /// Returns None if no Agent is found in the ancestry chain. + pub fn find_root(&self, pid: u32) -> Option { + let mut current = pid; + for _ in 0..64 { + let node = self.nodes.get(¤t)?; + if node.process_type == ProcessType::Agent { + return Some(current); + } + if !self.nodes.contains_key(&node.ppid) { + return None; + } + current = node.ppid; + } + None + } +} + +/// 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, +} + +#[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)); + } + + #[test] + fn test_find_root_agent() { + 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.find_root(100), Some(100)); + assert_eq!(tree.find_root(200), Some(100)); + assert_eq!(tree.find_root(300), Some(100)); + } + + #[test] + fn test_find_root_no_agent() { + let mut tree = LineageTree::new(); + tree.insert(make_node(100, 1, ProcessType::Unknown)); + tree.insert(make_node(200, 100, ProcessType::Unknown)); + + assert_eq!(tree.find_root(100), None); + assert_eq!(tree.find_root(200), None); + } + + #[test] + fn test_find_root_nonexistent_pid() { + let tree = LineageTree::new(); + assert_eq!(tree.find_root(999), None); + } +} diff --git a/src/agentsight/src/parser/unified.rs b/src/agentsight/src/parser/unified.rs index 70e62ff8b..b88e2544b 100644 --- a/src/agentsight/src/parser/unified.rs +++ b/src/agentsight/src/parser/unified.rs @@ -141,6 +141,7 @@ impl Parser { Event::FileWatch(_) => ParseResult { messages: Vec::new() }, Event::FileWrite(_) => ParseResult { messages: Vec::new() }, Event::UdpDns(_) => ParseResult { messages: Vec::new() }, + Event::Sched(_) => ParseResult { messages: Vec::new() }, } } diff --git a/src/agentsight/src/probes/mod.rs b/src/agentsight/src/probes/mod.rs index ebd022ea1..defa83384 100644 --- a/src/agentsight/src/probes/mod.rs +++ b/src/agentsight/src/probes/mod.rs @@ -7,6 +7,7 @@ pub mod filewatch; pub mod filewrite; pub mod udpdns; pub mod tcpsniff; +pub mod schedmon; pub mod probes; // Re-export commonly used types @@ -17,4 +18,5 @@ pub use procmon::{ProcMon, ProcMonEvent, Event as ProcMonEventExt}; pub use filewatch::{FileWatch, FileWatchEvent}; pub use filewrite::{FileWrite as FileWriteProbe, FileWriteEvent}; pub use udpdns::{UdpDns, UdpDnsEvent}; -pub use tcpsniff::TcpSniff; \ No newline at end of file +pub use tcpsniff::TcpSniff; +pub use schedmon::{SchedMon, SchedEvent}; \ No newline at end of file diff --git a/src/agentsight/src/probes/probes.rs b/src/agentsight/src/probes/probes.rs index 626e1e2c0..4e99a0f12 100644 --- a/src/agentsight/src/probes/probes.rs +++ b/src/agentsight/src/probes/probes.rs @@ -24,6 +24,7 @@ use super::filewrite::{FileWrite as FileWriteProbe, RawFileWriteEvent}; use super::udpdns::{UdpDns, RawUdpDnsEvent}; use crate::config::TcpTarget; use super::tcpsniff::TcpSniff; +use super::schedmon::{SchedMon, RawSchedEvent}; const POLL_TIMEOUT_MS: u64 = 100; @@ -34,6 +35,7 @@ const EVENT_SOURCE_PROCMON: u32 = 3; const EVENT_SOURCE_FILEWATCH: u32 = 4; const EVENT_SOURCE_FILEWRITE: u32 = 5; const EVENT_SOURCE_UDPDNS: u32 = 6; +const EVENT_SOURCE_SCHED: u32 = 7; /// Unified probe manager that coordinates sslsniff and proctrace /// @@ -57,6 +59,8 @@ pub struct Probes { udpdns: Option, /// TCP sniff probe (captures plain HTTP traffic on configured ports, optional) tcpsniff: Option, + /// Scheduler monitor probe (detects idle/active transitions, optional) + schedmon: Option, /// Shared ring buffer handle (cloned from proctrace) for polling rb_handle: MapHandle, /// Unified event channel - events are converted to Event type inside the poller @@ -70,7 +74,7 @@ impl Probes { /// # Arguments /// * `target_pids` - Initial PIDs to trace (empty means trace all matching UID) /// * `target_uid` - Optional UID filter - pub fn new(target_pids: &[u32], target_uid: Option, enable_filewatch: bool, enable_udpdns: bool, tcp_targets: &[TcpTarget]) -> Result { + pub fn new(target_pids: &[u32], target_uid: Option, enable_filewatch: bool, enable_udpdns: bool, tcp_targets: &[TcpTarget], enable_schedmon: bool) -> Result { // Create proctrace first - it will own the traced_processes map and ring buffer let proctrace = ProcTrace::new_with_target(target_pids, target_uid) .context("failed to create proctrace")?; @@ -126,8 +130,22 @@ impl Probes { None }; + // Optionally create schedmon - detects idle/active transitions for traced processes + let schedmon = if enable_schedmon { + match SchedMon::new_with_maps(&map_handle, &rb_handle) { + Ok(sm) => Some(sm), + Err(e) => { + log::warn!("SchedMon probe unavailable (requires BTF sched tracepoints): {e}"); + None + } + } + } else { + log::info!("SchedMon probe disabled (scheduler not enabled)"); + None + }; + let (event_tx, event_rx) = crossbeam_channel::unbounded(); - + Ok(Self { proctrace, sslsniff, @@ -136,6 +154,7 @@ impl Probes { filewrite, udpdns, tcpsniff, + schedmon, rb_handle, event_tx, event_rx, @@ -166,6 +185,11 @@ impl Probes { tcp.attach() .context("failed to attach tcpsniff")?; } + // Attach schedmon for idle/active transition detection (if enabled) + if let Some(ref mut sm) = self.schedmon { + sm.attach() + .context("failed to attach schedmon")?; + } // sslsniff uses uprobes attached per-process via attach_process() Ok(()) } @@ -193,6 +217,7 @@ impl Probes { let filewatch_event_size = mem::size_of::(); let filewrite_event_size = mem::size_of::(); let udpdns_event_size = mem::size_of::(); + let sched_event_size = mem::size_of::(); let event_tx = self.event_tx.clone(); let stop_flag = Arc::new(AtomicBool::new(false)); @@ -259,6 +284,14 @@ impl Probes { None } } + EVENT_SOURCE_SCHED => { + // Scheduler state event (sleep/wakeup) + if data.len() >= sched_event_size { + super::schedmon::SchedEvent::from_bytes(data).map(Event::Sched) + } else { + None + } + } _ => { // Unknown source - ignore log::warn!("probes: unknown event source {source}"); diff --git a/src/agentsight/src/probes/schedmon.rs b/src/agentsight/src/probes/schedmon.rs new file mode 100644 index 000000000..f9d3c954f --- /dev/null +++ b/src/agentsight/src/probes/schedmon.rs @@ -0,0 +1,161 @@ +// SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) +// Copyright (c) 2025 AgentSight Project +// +// Scheduler monitor probe — detects idle/active state transitions for traced +// Agent processes via the BTF-typed sched_switch / sched_wakeup tracepoints. + +use crate::config; +use anyhow::{Context, Result}; +use libbpf_rs::{ + Link, MapHandle, + skel::{OpenSkel, SkelBuilder}, +}; +use std::{ + mem::MaybeUninit, + os::fd::AsFd, +}; + +mod bpf { + include!(concat!(env!("OUT_DIR"), "/schedmon.skel.rs")); + include!(concat!(env!("OUT_DIR"), "/schedmon.rs")); +} +use bpf::*; + +pub type RawSchedEvent = bpf::sched_event; + +pub const SCHED_EVENT_SLEEP: u8 = 1; +pub const SCHED_EVENT_WAKEUP: u8 = 2; + +#[derive(Debug, Clone)] +pub struct SchedEvent { + /// Thread-group id — identifies the Agent family. + pub tgid: u32, + /// Thread id — tracked individually (a family is ACTIVE while any thread runs). + pub tid: u32, + pub timestamp_ns: u64, + pub event_type: u8, +} + +impl SchedEvent { + pub fn from_bytes(data: &[u8]) -> Option { + if data.len() < std::mem::size_of::() { + return None; + } + + let raw = unsafe { &*(data.as_ptr() as *const RawSchedEvent) }; + + Some(SchedEvent { + tgid: raw.tgid, + tid: raw.tid, + timestamp_ns: config::ktime_to_unix_ns(raw.timestamp_ns), + event_type: raw.event_type, + }) + } + + pub fn is_sleep(&self) -> bool { + self.event_type == SCHED_EVENT_SLEEP + } + + pub fn is_wakeup(&self) -> bool { + self.event_type == SCHED_EVENT_WAKEUP + } +} + +pub struct SchedMon { + // Vestigial under libbpf-rs 0.23.x: `builder.open()` takes no OpenObject and + // owns its bpf_object internally, so this field is unused (kept for parity + // with the sibling probes). TODO: on upgrade to libbpf-rs 0.24+, `open()` + // requires `&mut MaybeUninit` and the 'static lifetime laundering + // below must be revisited across all probes. + _open_object: Box>, + skel: Box>, + _links: Vec, +} + +impl SchedMon { + pub fn new_with_maps(traced_processes: &MapHandle, rb: &MapHandle) -> Result { + let mut builder = SchedmonSkelBuilder::default(); + builder.obj_builder.debug(config::verbose()); + + let open_object = Box::new(MaybeUninit::::uninit()); + let mut open_skel = builder.open().context("failed to open schedmon BPF object")?; + + open_skel + .maps_mut() + .traced_processes() + .reuse_fd(traced_processes.as_fd()) + .context("failed to reuse traced_processes map")?; + + open_skel + .maps_mut() + .rb() + .reuse_fd(rb.as_fd()) + .context("failed to reuse rb map")?; + + let skel = open_skel.load().context("failed to load schedmon BPF object")?; + + let skel = + unsafe { Box::from_raw(Box::into_raw(Box::new(skel)) as *mut SchedmonSkel<'static>) }; + + Ok(Self { + _open_object: open_object, + skel, + _links: Vec::new(), + }) + } + + pub fn attach(&mut self) -> Result<()> { + let mut links = Vec::new(); + + let link = self + .skel + .progs_mut() + .handle_sched_switch() + .attach() + .context("failed to attach sched_switch tracepoint")?; + links.push(link); + + let link = self + .skel + .progs_mut() + .handle_sched_wakeup() + .attach() + .context("failed to attach sched_wakeup tracepoint")?; + links.push(link); + + self._links = links; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_sched_event_from_bytes_too_short() { + let data = [0u8; 4]; + assert!(SchedEvent::from_bytes(&data).is_none()); + } + + #[test] + fn test_sched_event_is_sleep_wakeup() { + let ev = SchedEvent { + tgid: 100, + tid: 100, + timestamp_ns: 12345, + event_type: SCHED_EVENT_SLEEP, + }; + assert!(ev.is_sleep()); + assert!(!ev.is_wakeup()); + + let ev2 = SchedEvent { + tgid: 100, + tid: 101, + timestamp_ns: 12345, + event_type: SCHED_EVENT_WAKEUP, + }; + assert!(ev2.is_wakeup()); + assert!(!ev2.is_sleep()); + } +} diff --git a/src/agentsight/src/scheduler/mod.rs b/src/agentsight/src/scheduler/mod.rs new file mode 100644 index 000000000..e83dd0402 --- /dev/null +++ b/src/agentsight/src/scheduler/mod.rs @@ -0,0 +1,337 @@ +//! Agent activity monitor. +//! +//! Tracks the scheduling state of Agent process families using per-thread +//! sleep/wakeup events from the schedmon BPF probe. Exports idle/active +//! transitions as observability signals (log + metrics). Does NOT actuate +//! cgroup changes — CPU scheduling policy belongs in the container spec. + +use std::collections::{HashMap, HashSet}; +use std::time::Instant; + +use crate::probes::schedmon::{SCHED_EVENT_SLEEP, SCHED_EVENT_WAKEUP}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ActivityState { + Active, + Idle, +} + +#[derive(Debug, Clone)] +pub struct ActivityConfig { + pub enabled: bool, + pub idle_threshold_ms: u64, +} + +impl Default for ActivityConfig { + fn default() -> Self { + Self { + enabled: false, + idle_threshold_ms: 50, + } + } +} + +/// Per-family activity metrics. +#[derive(Debug, Clone, Default)] +pub struct FamilyMetrics { + pub idle_to_active_count: u64, + pub active_to_idle_count: u64, + pub last_active_duration_ns: u64, + pub last_idle_duration_ns: u64, +} + +struct FamilyState { + member_pids: HashSet, + active_tids: HashSet, + state: ActivityState, + idle_since: Option, + last_transition: Instant, + metrics: FamilyMetrics, +} + +pub struct ActivityMonitor { + config: ActivityConfig, + families: HashMap, + pid_to_root: HashMap, +} + +impl ActivityMonitor { + pub fn new(config: ActivityConfig) -> Self { + Self { + config, + families: HashMap::new(), + pid_to_root: HashMap::new(), + } + } + + pub fn add_process(&mut self, pid: u32, root_pid: u32) { + self.pid_to_root.insert(pid, root_pid); + + let family = self.families.entry(root_pid).or_insert_with(|| FamilyState { + member_pids: HashSet::new(), + active_tids: HashSet::new(), + state: ActivityState::Active, + idle_since: None, + last_transition: Instant::now(), + metrics: FamilyMetrics::default(), + }); + + family.member_pids.insert(pid); + family.active_tids.insert(pid); + family.idle_since = None; + } + + pub fn remove_process(&mut self, pid: u32) { + let root_pid = match self.pid_to_root.remove(&pid) { + Some(r) => r, + None => return, + }; + + let should_remove = if let Some(family) = self.families.get_mut(&root_pid) { + family.member_pids.remove(&pid); + family.active_tids.remove(&pid); + family.member_pids.is_empty() + } else { + false + }; + + if should_remove { + if let Some(family) = self.families.remove(&root_pid) { + log::debug!( + "activity: family {root_pid} removed (idle_to_active={}, active_to_idle={})", + family.metrics.idle_to_active_count, + family.metrics.active_to_idle_count, + ); + } + } + } + + pub fn on_sched_event(&mut self, tgid: u32, tid: u32, event_type: u8) { + let root_pid = match self.pid_to_root.get(&tgid) { + Some(&r) => r, + None => return, + }; + + let family = match self.families.get_mut(&root_pid) { + Some(f) => f, + None => return, + }; + + match event_type { + SCHED_EVENT_WAKEUP => { + family.active_tids.insert(tid); + } + SCHED_EVENT_SLEEP => { + family.active_tids.remove(&tid); + } + _ => return, + } + + if family.active_tids.is_empty() { + if family.state == ActivityState::Active && family.idle_since.is_none() { + family.idle_since = Some(Instant::now()); + } + return; + } + + // A member is runnable: cancel pending idle. + family.idle_since = None; + if family.state == ActivityState::Idle { + let idle_duration = family.last_transition.elapsed(); + family.state = ActivityState::Active; + family.last_transition = Instant::now(); + family.metrics.idle_to_active_count += 1; + family.metrics.last_idle_duration_ns = idle_duration.as_nanos() as u64; + log::debug!( + "activity: family {root_pid} IDLE -> ACTIVE (idle_ms={})", + idle_duration.as_millis(), + ); + } + } + + pub fn tick(&mut self) { + let threshold = std::time::Duration::from_millis(self.config.idle_threshold_ms); + + for (&root_pid, family) in &mut self.families { + if family.state == ActivityState::Idle || !family.active_tids.is_empty() { + continue; + } + match family.idle_since { + Some(t) if t.elapsed() >= threshold => {} + _ => continue, + } + + let active_duration = family.last_transition.elapsed(); + family.state = ActivityState::Idle; + family.last_transition = Instant::now(); + family.metrics.active_to_idle_count += 1; + family.metrics.last_active_duration_ns = active_duration.as_nanos() as u64; + log::debug!( + "activity: family {root_pid} ACTIVE -> IDLE (active_ms={})", + active_duration.as_millis(), + ); + } + } + + pub fn family_count(&self) -> usize { + self.families.len() + } + + pub fn family_state(&self, root_pid: u32) -> Option { + self.families.get(&root_pid).map(|f| f.state) + } + + pub fn family_metrics(&self, root_pid: u32) -> Option<&FamilyMetrics> { + self.families.get(&root_pid).map(|f| &f.metrics) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn test_config() -> ActivityConfig { + ActivityConfig { + enabled: true, + idle_threshold_ms: 10, + } + } + + fn make_monitor() -> ActivityMonitor { + ActivityMonitor::new(test_config()) + } + + fn force_idle_window(monitor: &mut ActivityMonitor, root_pid: u32) { + if let Some(family) = monitor.families.get_mut(&root_pid) { + family.idle_since = Some(Instant::now() - Duration::from_millis(20)); + } + } + + #[test] + fn test_add_remove_process() { + let mut mon = make_monitor(); + + mon.add_process(100, 100); + assert_eq!(mon.family_count(), 1); + assert_eq!(mon.pid_to_root.get(&100), Some(&100)); + + mon.add_process(200, 100); + assert_eq!(mon.family_count(), 1); + + mon.remove_process(200); + assert_eq!(mon.family_count(), 1); + + mon.remove_process(100); + assert_eq!(mon.family_count(), 0); + } + + #[test] + fn test_sched_event_active() { + let mut mon = make_monitor(); + mon.add_process(100, 100); + + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + + mon.on_sched_event(100, 100, SCHED_EVENT_SLEEP); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + + mon.on_sched_event(100, 100, SCHED_EVENT_WAKEUP); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + } + + #[test] + fn test_sched_event_idle_after_threshold() { + let mut mon = make_monitor(); + mon.add_process(100, 100); + + mon.on_sched_event(100, 100, SCHED_EVENT_SLEEP); + force_idle_window(&mut mon, 100); + + mon.tick(); + assert_eq!(mon.family_state(100), Some(ActivityState::Idle)); + assert_eq!(mon.family_metrics(100).unwrap().active_to_idle_count, 1); + } + + #[test] + fn test_wakeup_cancels_pending_idle() { + let mut mon = make_monitor(); + mon.add_process(100, 100); + + mon.on_sched_event(100, 100, SCHED_EVENT_SLEEP); + mon.on_sched_event(100, 100, SCHED_EVENT_WAKEUP); + force_idle_window(&mut mon, 100); + mon.tick(); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + } + + #[test] + fn test_multithreaded_active_while_any_thread_runs() { + let mut mon = make_monitor(); + mon.add_process(100, 100); + + mon.on_sched_event(100, 101, SCHED_EVENT_WAKEUP); + mon.on_sched_event(100, 100, SCHED_EVENT_SLEEP); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + + mon.on_sched_event(100, 101, SCHED_EVENT_SLEEP); + force_idle_window(&mut mon, 100); + mon.tick(); + assert_eq!(mon.family_state(100), Some(ActivityState::Idle)); + + mon.on_sched_event(100, 101, SCHED_EVENT_WAKEUP); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + assert_eq!(mon.family_metrics(100).unwrap().idle_to_active_count, 1); + } + + #[test] + fn test_family_active_if_any_active() { + let mut mon = make_monitor(); + mon.add_process(100, 100); + mon.add_process(200, 100); + + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + + mon.on_sched_event(100, 100, SCHED_EVENT_SLEEP); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + + mon.on_sched_event(200, 200, SCHED_EVENT_SLEEP); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + + force_idle_window(&mut mon, 100); + mon.tick(); + assert_eq!(mon.family_state(100), Some(ActivityState::Idle)); + + mon.on_sched_event(200, 200, SCHED_EVENT_WAKEUP); + assert_eq!(mon.family_state(100), Some(ActivityState::Active)); + } + + #[test] + fn test_remove_nonexistent_process() { + let mut mon = make_monitor(); + mon.remove_process(999); + assert_eq!(mon.family_count(), 0); + } + + #[test] + fn test_sched_event_unknown_pid() { + let mut mon = make_monitor(); + mon.on_sched_event(999, 999, SCHED_EVENT_WAKEUP); + assert_eq!(mon.family_count(), 0); + } + + #[test] + fn test_multiple_families() { + let mut mon = make_monitor(); + mon.add_process(100, 100); + mon.add_process(200, 200); + assert_eq!(mon.family_count(), 2); + + mon.on_sched_event(100, 100, SCHED_EVENT_SLEEP); + force_idle_window(&mut mon, 100); + mon.tick(); + + assert_eq!(mon.family_state(100), Some(ActivityState::Idle)); + assert_eq!(mon.family_state(200), Some(ActivityState::Active)); + } +} diff --git a/src/agentsight/src/unified.rs b/src/agentsight/src/unified.rs index fcca60c30..923d9a608 100644 --- a/src/agentsight/src/unified.rs +++ b/src/agentsight/src/unified.rs @@ -99,6 +99,10 @@ pub struct AgentSight { sls_activated: Arc, /// Mailbox for watcher thread to deposit a dynamically-created LogtailExporter pending_logtail: Arc>>>, + /// Blood lineage tree tracking process parent-child relationships and type classification + lineage_tree: Arc>, + /// Agent activity monitor — tracks idle/active transitions (observability only) + activity_monitor: Option, } /// GenAI events waiting for session_id resolution via ResponseSessionMapper. @@ -210,7 +214,7 @@ impl AgentSight { // Create probes - agent discovery is handled by AgentScanner via ProcMon events let enable_udpdns = !config.https_rules.is_empty() || !http_domains.is_empty(); let mut probes = - Probes::new(&[], config.target_uid, config.enable_filewatch, enable_udpdns, &tcp_targets).context("Failed to create probes")?; + Probes::new(&[], config.target_uid, config.enable_filewatch, enable_udpdns, &tcp_targets, config.enable_scheduler).context("Failed to create probes")?; // Attach procmon for process monitoring probes.attach().context("Failed to attach probes")?; @@ -439,6 +443,16 @@ impl AgentSight { .ok(); } + let activity_monitor = if config.enable_scheduler { + log::info!( + "Activity monitor enabled: idle_threshold={}ms", + config.activity_config.idle_threshold_ms, + ); + Some(crate::scheduler::ActivityMonitor::new(config.activity_config.clone())) + } else { + None + }; + Ok(AgentSight { probes, parser: Parser::new(), @@ -463,6 +477,8 @@ impl AgentSight { http_domains, sls_activated, pending_logtail, + lineage_tree: Arc::new(std::sync::RwLock::new(crate::lineage::LineageTree::new())), + activity_monitor, }) } @@ -541,6 +557,19 @@ impl AgentSight { return None; } + // Handle activity monitor events (idle/active detection) + if let Event::Sched(ref sched_event) = event { + if let Some(ref mut activity) = self.activity_monitor { + activity.on_sched_event(sched_event.tgid, sched_event.tid, sched_event.event_type); + } + 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); @@ -717,7 +746,7 @@ 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/{}/cmdline", pid)); @@ -732,11 +761,45 @@ 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); + let matches_agent_pattern = matched.is_some(); + + if let Some(agent) = matched { let agent_name = agent.agent_info.name.clone(); self.pid_agent_name_cache.insert(*pid, agent_name.clone()); self.attach_process(*pid, &agent_name); } + + // Phase 3: userspace AGENT_MODE=1 detection (complements BPF layer) + // Skip if already attached by cmdline rules or previously attached. + // Also skip if the parent is already tracked — that means this + // process inherited AGENT_MODE from its parent, not set it itself. + if !matches_agent_pattern + && !self.pid_agent_name_cache.contains_key(pid) + && crate::discovery::scanner::has_agent_mode(*pid) + { + let parent_tracked = crate::discovery::scanner::read_ppid(*pid) + .is_some_and(|ppid| self.pid_agent_name_cache.contains_key(&ppid)); + if parent_tracked { + log::debug!( + "ProcMon: pid={} has AGENT_MODE=1 but parent is already tracked, skipping", + pid + ); + } else { + log::info!( + "ProcMon: pid={} detected AGENT_MODE=1 via /proc environ, auto-attaching", + pid + ); + let agent_name = format!("agent-mode-{}", comm); + self.pid_agent_name_cache.insert(*pid, agent_name.clone()); + self.attach_process(*pid, &agent_name); + // Insert into the lineage tree directly. proctrace does not + // emit an exec event for an AGENT_MODE root (it was not yet in + // traced_processes when it execed), so without this the root + // would never appear in the tree. + self.ensure_lineage_node(*pid, *ppid, comm, Some(agent_name), true, false); + } + } } ProcMonEvent::Exit { pid, .. } => { // Remove from tracking if it was an agent @@ -744,10 +807,141 @@ impl AgentSight { let agent_name = agent.agent_info.name.clone(); self.detach_process(*pid, &agent_name); } + // Clean activity monitor (root agents may not be in proctrace child_pids) + if let Some(ref mut activity) = self.activity_monitor { + activity.remove_process(*pid); + } + // Clean lineage tree + 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, + has_agent_mode: bool, + matches_agent: bool, + ) { + if let Ok(mut tree) = self.lineage_tree.write() { + if tree.get(pid).is_none() { + let node = crate::lineage::LineageNode { + pid, + ppid, + process_type: crate::lineage::ProcessType::Unknown, + flags: if has_agent_mode { + crate::lineage::LINEAGE_FLAG_AGENT_MODE + } else { + 0 + }, + create_time_ns: 0, + comm: comm.to_string(), + agent_name, + children: Vec::new(), + }; + tree.insert(node); + } + tree.classify(pid, has_agent_mode, matches_agent); + + // Register a classified process with the activity monitor. + if let Some(ref mut activity) = self.activity_monitor { + let ptype = tree.get(pid).map(|n| n.process_type); + if ptype.is_some_and(|t| t != crate::lineage::ProcessType::Unknown) { + if let Some(root_pid) = tree.find_root(pid) { + activity.add_process(pid, root_pid); + } + } + } + } + } + + /// 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_bytes: Vec = header + .comm + .iter() + .map(|&c| c as u8) + .take_while(|&b| b != 0) + .collect(); + let comm = String::from_utf8_lossy(&comm_bytes).into_owned(); + + let agent_name = self.pid_agent_name_cache.get(&header.pid).cloned(); + let matches_agent = agent_name + .as_ref() + .is_some_and(|n| !n.starts_with("agent-mode-")); + // Infer AGENT_MODE from cache: if procmon already detected it, the + // name starts with "agent-mode-". Avoids redundant /proc/pid/environ I/O. + let has_agent_mode = agent_name + .as_ref() + .is_some_and(|n| n.starts_with("agent-mode-")); + + let node = crate::lineage::LineageNode { + pid: header.pid, + ppid: header.ppid, + process_type: crate::lineage::ProcessType::Unknown, + flags: if has_agent_mode { + crate::lineage::LINEAGE_FLAG_AGENT_MODE + } else { + 0 + }, + create_time_ns: header.timestamp_ns, + comm, + agent_name, + children: Vec::new(), + }; + + if let Ok(mut tree) = self.lineage_tree.write() { + tree.insert(node); + tree.classify(header.pid, has_agent_mode, matches_agent); + + let ptype = tree.get(header.pid).map(|n| n.process_type); + log::debug!( + "Lineage: added pid={} ppid={} type={:?}", + header.pid, + header.ppid, + ptype, + ); + + // Register a classified process with the activity monitor. + if let Some(ref mut activity) = self.activity_monitor { + if ptype.is_some_and(|t| t != crate::lineage::ProcessType::Unknown) { + if let Some(root_pid) = tree.find_root(header.pid) { + activity.add_process(header.pid, root_pid); + } + } + } + } + } + VariableEvent::Exit { header, .. } => { + if let Some(ref mut activity) = self.activity_monitor { + activity.remove_process(header.pid); + } + if let Ok(mut tree) = self.lineage_tree.write() { + if let Some(removed) = tree.remove(header.pid) { + log::debug!( + "Lineage: removed pid={} type={:?}", + header.pid, + removed.process_type, + ); + } + } + } + _ => {} + } + } + /// Handle FileWatch event via registered callback fn handle_filewatch_event(&self, event: &FileWatchEvent) { log::debug!("FileWatch: pid={} file={}", event.pid, event.filename); @@ -764,6 +958,15 @@ impl AgentSight { self.filewatch_callback = Some(Box::new(callback)); } + /// Idle-loop hook: finalize debounced activity transitions. Must be called + /// from every driver loop's idle branch (run() and the FFI loop) so the + /// monitor is not stuck never transitioning families to idle. + pub fn on_idle_tick(&mut self) { + if let Some(ref mut activity) = self.activity_monitor { + activity.tick(); + } + } + /// Handle FileWrite event: extract responseId→sessionId mapping, then call callback fn handle_filewrite_event(&mut self, event: &FileWriteEvent) { log::debug!( @@ -790,6 +993,7 @@ impl AgentSight { self.drain_and_persist_dead_connections(); // Check if config watcher deposited a new LogtailExporter self.check_pending_logtail(); + self.on_idle_tick(); std::thread::sleep(std::time::Duration::from_millis(10)); } }