feat: support agent governance (#11446)

* feat: add AiAgentConfig for AI agent governance

* feat: add BIZ_TYPE_AI_AGENT constant

* feat: add ai_agent stub in enterprise-utils

* feat: add AI agent URL detection hook in the HTTP parser

* feat: add biz_type to ProcessData and ProcessInfo

* feat: exempt AI agent traffic from flow reassembly limits

* feat: add access_permission and AI agent hook points for eBPF

* feat: integrate AiAgentRegistry into the agent lifecycle

* fix: add missing L7ProtocolInfoInterface import and remove the unused variable

* fix: add a null statement after skip_latency_filter

* feat: complete PRD 2.2 updates for AI agent governance

* fix: resolve blockers in the AI agent governance event pipeline

* feat: add AI agent reassembly and process biz_type handling

* feat: add ai_agent config and bump the database version

* feat: extend endpoints for AI agent identification

* fix: align AI agent tracepoint hooks

* feat: add chmod, chown, and unlink tracepoints for AI agents

* fix: clean up AI agents with full process scanning

* fix: correct AI agent pid_tgid usage in socket_trace

* fix: reduce AI agent stack usage in data submission

* fix: sync biz_type for gprocess in multi-controller mode

* fix: add missing is_ai_agent handling in socket_trace

* fix: avoid BPF stack usage for the ai_agent flag

* fix: address AI agent governance review feedback

* fix: resolve the proc scan hook warning and HTTP endpoint borrow issue

* feat: auto-sync AI agent gprocess_info

* feat: mark AI agents with biz_type in gprocess

* feat: support gprocess.biz_type tag queries

* feat: log AI agent PIDs during gprocess sync

* feat: inherit child process lifecycle for AI agents

* feat: expose process event start time for AI agents

* fix: correct gprocess fallback and captured bytes for proc lifecycle

* fix: guard AI reassembly bytes on invalid socket info

* fix: refresh proc.gprocess_info on process change

* fix: enable reassembly after protocol inference

* fix: enable reassembly for existing sockets after protocol inference

* fix: inherit gprocess_id for AI agent descendants

* fix: preserve gprocess inheritance on proc exec and add tests

* fix: include AI agent PIDs in socket list sync

* fix: propagate reasm_bytes during eBPF merge

* fix: remove the record_endpoint_hit stub from AiAgentRegistry

* feat: add ai_agent_root_pid for child gprocess_id resolution

* fix: normalize file_op output to match IoEvent

* fix: bypass collect_mode filtering for child AI agent file I/O events

* chore: add deepflow-ctl support for ai-agent process show

* fix: keep AI endpoint extraction independent of http_endpoint_disabled

* feat: split AI agent event tables

* fix: register metrics for AI agent event tables

* fix: add localized tag descriptions for AI agent event tables

* fix: compact management events and disable 1-second aggregation

* fix: reduce AI agent governance sync log noise

* fix: scope event reducers by agent and gpid

* fix: guard AI agent proc scan hook for non-Linux builds

* fix: refine file agg event windowing

* fix: clarify AI agent event semantics

* fix: route biz_type queries through gprocess tags

* fix: avoid duplicate AI agent exec and exit probes

* fix: correct captured byte accounting for ebpf reassembly

* fix: address latest pr11446 review feedback

* fix: restrict file agg events to ai agents

* feat: link AI event data source retention updates

* fix: add raw biz_type fallback for enum filters

* fix: preserve ai agent endpoint and retention sync

* fix: harden ai agent ebpf reassembly and limits

* docs: clarify gprocess id sync timing

* test: harden ai agent review regressions

* fix: harden ai agent event reducers and cache

* fix: clean up ai agent review follow-ups

* fix: harden ai agent ebpf and proc scan follow-ups

* fix: tighten ai agent endpoint and pid fetch paths

* fix: avoid endpoint borrow conflicts during ai-agent matching

* fix: align gprocess biz_type query translators

* refactor: avoid extra endpoint copies in http parser

* fix(cli): align Go version with server requirement

* fix(cli): tidy module metadata for Go 1.26

Author: kylewanginchina

.github/workflows/cli-build.yml

@@ -32,7 +32,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@master
         with:
-          go-version: 1.24.x
+          go-version: 1.26.x
 
       - name: Set up GOPATH env
         run: echo "GOPATH=$(go env GOPATH)" >> "$GITHUB_ENV"

.github/workflows/cli-verify.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@master
         with:
-          go-version: 1.24.x
+          go-version: 1.26.x
 
       - name: Set up GOPATH env
         run: echo "GOPATH=$(go env GOPATH)" >> "$GITHUB_ENV"

agent/crates/enterprise-utils/src/lib.rs

@@ -509,3 +509,89 @@ pub mod rpc {
         }
     }
 }
+
+pub mod ai_agent {
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    #[derive(Debug, Clone, Default)]
+    pub struct AgentMeta {
+        pub first_seen: Duration,
+        pub last_seen: Duration,
+        pub matched_endpoint: String,
+        pub root_pid: u32,
+    }
+
+    #[derive(Debug, Clone, Default)]
+    pub struct AiAgentRegistry;
+
+    impl AiAgentRegistry {
+        pub fn new() -> Self {
+            AiAgentRegistry
+        }
+
+        pub fn register(&self, _pid: u32, _endpoint: &str, _now: Duration) -> bool {
+            false
+        }
+
+        pub fn is_ai_agent(&self, _pid: u32) -> bool {
+            false
+        }
+
+        pub fn get_root_pid(&self, _pid: u32) -> u32 {
+            0
+        }
+
+        pub fn register_child(&self, _parent_pid: u32, _child_pid: u32, _now: Duration) -> bool {
+            false
+        }
+
+        pub fn get_all_pids(&self) -> Vec<u32> {
+            vec![]
+        }
+
+        pub fn cleanup_dead_pids(&self, _alive_pids: &[u32]) -> Vec<u32> {
+            vec![]
+        }
+
+        pub fn len(&self) -> usize {
+            0
+        }
+
+        pub fn is_empty(&self) -> bool {
+            true
+        }
+
+        pub fn sync_bpf_map_add(&self, _pid: u32) {}
+
+        pub fn sync_bpf_map_remove(&self, _pid: u32) {}
+
+        #[cfg(target_os = "linux")]
+        pub fn set_bpf_map_fd(&self, _fd: i32) {}
+
+        pub fn set_file_io_enabled(&self, _enabled: bool) {}
+    }
+
+    /// Check if a URL path matches an AI Agent endpoint pattern.
+    pub fn match_ai_agent_endpoint(
+        _endpoints: &[String],
+        _path: &str,
+        _pid: u32,
+        _socket_role: u8,
+        _now: Duration,
+    ) -> Option<String> {
+        None
+    }
+
+    /// Initialize the global AI Agent registry. Returns the registry Arc.
+    /// Stub: returns a no-op registry.
+    pub fn init_global_registry() -> Arc<AiAgentRegistry> {
+        Arc::new(AiAgentRegistry::new())
+    }
+
+    /// Get a reference to the global AI Agent registry.
+    /// Stub: always returns None.
+    pub fn global_registry() -> Option<&'static Arc<AiAgentRegistry>> {
+        None
+    }
+}

agent/src/common/ebpf.rs

@@ -40,6 +40,10 @@ pub const GO_HTTP2_UPROBE_DATA: u8 = 5;
 pub const SOCKET_CLOSE_EVENT: u8 = 6;
 // unix socket
 pub const UNIX_SOCKET: u8 = 8;
+// AI Agent governance event types
+pub const FILE_OP_EVENT: u8 = 9;
+pub const PERM_OP_EVENT: u8 = 10;
+pub const PROC_LIFECYCLE_EVENT: u8 = 11;
 
 const EBPF_TYPE_TRACEPOINT: u8 = 0;
 const EBPF_TYPE_TLS_UPROBE: u8 = 1;

agent/src/common/flow.rs

@@ -537,6 +537,10 @@ impl From<FlowPerfStats> for flow_log::FlowPerfStats {
     }
 }
 
+// Business type constants for process classification
+pub const BIZ_TYPE_DEFAULT: u8 = 0;
+pub const BIZ_TYPE_AI_AGENT: u8 = 1;
+
 #[derive(Clone, Debug, Default)]
 pub struct L7Stats {
     pub stats: L7PerfStats,

agent/src/common/l7_protocol_log.rs

@@ -24,6 +24,8 @@ use std::sync::{
 };
 use std::time::Duration;
 
+use crate::common::l7_protocol_info::L7ProtocolInfoInterface;
+
 use enum_dispatch::enum_dispatch;
 use log::debug;
 use lru::LruCache;
@@ -252,6 +254,16 @@ impl L7ParseResult {
             L7ParseResult::None => panic!("parse result is none but unwrap multi"),
         }
     }
+
+    /// Check if any parsed result has the given biz_type.
+    /// Used to detect AI Agent flows after parsing.
+    pub fn has_biz_type(&self, biz_type: u8) -> bool {
+        match self {
+            L7ParseResult::Single(info) => info.get_biz_type() == biz_type,
+            L7ParseResult::Multi(infos) => infos.iter().any(|i| i.get_biz_type() == biz_type),
+            L7ParseResult::None => false,
+        }
+    }
 }
 
 #[enum_dispatch]
@@ -690,12 +702,14 @@ pub struct ParseParam<'a> {
 
     // the config of `l7_log_packet_size`, must set in parse_payload and check_payload
     pub buf_size: u16,
-    pub captured_byte: u16,
+    pub captured_byte: u32,
 
     pub oracle_parse_conf: OracleConfig,
     pub iso8583_parse_conf: Iso8583ParseConfig,
     pub web_sphere_mq_parse_conf: WebSphereMqParseConfig,
     pub net_sign_parse_conf: NetSignParseConfig,
+    pub process_id: u32,
+    pub socket_role: u8,
 }
 
 impl<'a> fmt::Debug for ParseParam<'a> {
@@ -729,6 +743,8 @@ impl<'a> fmt::Debug for ParseParam<'a> {
             .field("iso8583_parse_conf", &self.iso8583_parse_conf)
             .field("web_sphere_mq_parse_conf", &self.web_sphere_mq_parse_conf)
             .field("net_sign_parse_conf", &self.net_sign_parse_conf)
+            .field("process_id", &self.process_id)
+            .field("socket_role", &self.socket_role)
             .finish()
     }
 }
@@ -803,6 +819,8 @@ impl<'a> ParseParam<'a> {
             iso8583_parse_conf: Iso8583ParseConfig::default(),
             web_sphere_mq_parse_conf: WebSphereMqParseConfig::default(),
             net_sign_parse_conf: NetSignParseConfig::default(),
+            process_id: packet.process_id,
+            socket_role: packet.socket_role,
         }
     }
 }
@@ -821,11 +839,17 @@ impl<'a> ParseParam<'a> {
     }
 
     pub fn set_buf_size(&mut self, buf_size: usize) {
-        self.buf_size = buf_size as u16;
+        // Saturate to u16::MAX to avoid overflow when AI Agent flows use larger payload sizes.
+        // buf_size is informational for plugins; actual payload truncation uses the usize value directly.
+        self.buf_size = if buf_size > u16::MAX as usize {
+            u16::MAX
+        } else {
+            buf_size as u16
+        };
     }
 
     pub fn set_captured_byte(&mut self, captured_byte: usize) {
-        self.captured_byte = captured_byte as u16;
+        self.captured_byte = u32::try_from(captured_byte).unwrap_or(u32::MAX);
     }
 
     pub fn set_rrt_timeout(&mut self, t: usize) {
@@ -955,3 +979,28 @@ impl fmt::Debug for L7ProtocolBitmap {
         f.write_str(format!("{:#?}", p).as_str())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
+    #[test]
+    fn captured_byte_should_not_truncate_large_payloads() {
+        let packet = MetaPacket::default();
+        let mut param = ParseParam::new(
+            &packet,
+            None,
+            Rc::new(RefCell::new(None)),
+            #[cfg(any(target_os = "linux", target_os = "android"))]
+            Rc::new(RefCell::new(None)),
+            false,
+            false,
+        );
+
+        let captured: u32 = 200_000;
+        param.set_captured_byte(captured as usize);
+        assert_eq!(param.captured_byte as u32, captured);
+    }
+}

agent/src/common/meta_packet.rs

@@ -245,6 +245,7 @@ pub struct MetaPacket<'a> {
     pub socket_id: u64,
     pub cap_start_seq: u64,
     pub cap_end_seq: u64,
+    pub reasm_bytes: u32,
     pub l7_protocol_from_ebpf: L7Protocol,
     //  流结束标识, 目前只有 go http2 uprobe 用到
     pub is_request_end: bool,
@@ -1035,6 +1036,16 @@ impl<'a> MetaPacket<'a> {
     #[inline]
     pub fn get_captured_byte(&self) -> usize {
         if self.tap_port.is_from(TapPort::FROM_EBPF) {
+            // For eBPF reassembly segments, upper layers merge multiple packets
+            // by accumulating each segment's captured length. `reasm_bytes` is a
+            // cumulative counter on the socket, so using it here would double
+            // count across segment merges (100 + 200 + 300 ...).
+            if self.is_reassembly_segment() {
+                return self.l4_payload_len as usize;
+            }
+            if self.reasm_bytes > 0 {
+                return self.reasm_bytes as usize;
+            }
             return self.packet_len as usize - 54;
         }
 
@@ -1061,6 +1072,16 @@ impl<'a> MetaPacket<'a> {
         0
     }
 
+    #[cfg(all(unix, feature = "libtrace"))]
+    fn is_reassembly_segment(&self) -> bool {
+        self.segment_flags != SegmentFlags::None
+    }
+
+    #[cfg(not(all(unix, feature = "libtrace")))]
+    fn is_reassembly_segment(&self) -> bool {
+        false
+    }
+
     #[inline]
     pub fn merge(&mut self, packet: &mut MetaPacket) {
         if self.ebpf_type == EbpfType::None {
@@ -1083,6 +1104,13 @@ impl<'a> MetaPacket<'a> {
         self.payload_len += packet.payload_len;
         self.l4_payload_len += packet.l4_payload_len;
         self.cap_end_seq = packet.cap_start_seq;
+        // eBPF reassembly: propagate the latest cumulative reassembly bytes.
+        // `reasm_bytes` reflects the total bytes reassembled in the kernel for
+        // this flow. When we merge multiple MSG_REASM_* segments, we must keep
+        // the newest cumulative value, otherwise `get_captured_byte()` will
+        // stay at the first segment's size (e.g., HTTP headers only) and
+        // `captured_request_byte` will be incorrect for large bodies.
+        self.reasm_bytes = self.reasm_bytes.max(packet.reasm_bytes);
     }
 
     #[cfg(all(unix, feature = "libtrace"))]
@@ -1147,6 +1175,7 @@ impl<'a> MetaPacket<'a> {
         packet.signal_source = SignalSource::EBPF;
         packet.cap_start_seq = data.cap_seq;
         packet.cap_end_seq = data.cap_seq;
+        packet.reasm_bytes = data.reasm_bytes;
         packet.process_id = data.process_id;
         packet.thread_id = data.thread_id;
         packet.coroutine_id = data.coroutine_id;
@@ -1424,12 +1453,12 @@ impl CacheItem for MetaPacket<'static> {
         self.l7_protocol_from_ebpf
     }
 
-    #[cfg(feature = "libtrace")]
+    #[cfg(all(unix, feature = "libtrace"))]
     fn is_segment_start(&self) -> bool {
         self.segment_flags == SegmentFlags::Start
     }
 
-    #[cfg(not(feature = "libtrace"))]
+    #[cfg(not(all(unix, feature = "libtrace")))]
     fn is_segment_start(&self) -> bool {
         false
     }
@@ -1598,4 +1627,62 @@ mod tests {
             pkt
         );
     }
+
+    #[test]
+    fn get_captured_byte_prefers_reasm_bytes_for_ebpf() {
+        let mut pkt = MetaPacket::default();
+        pkt.tap_port = TapPort::from_ebpf(1, 0);
+        pkt.packet_len = 54 + 16;
+        pkt.reasm_bytes = 200_000;
+
+        assert_eq!(pkt.get_captured_byte(), 200_000);
+    }
+
+    #[cfg(all(unix, feature = "libtrace"))]
+    #[test]
+    fn get_captured_byte_for_ebpf_reasm_segment_should_use_current_segment_len() {
+        let mut pkt = MetaPacket::default();
+        pkt.tap_port = TapPort::from_ebpf(1, 0);
+        pkt.segment_flags = SegmentFlags::Seg;
+        pkt.reasm_bytes = 200_000;
+        pkt.raw_from_ebpf = vec![0u8; 4096];
+        pkt.l4_payload_len = 4096;
+
+        assert_eq!(pkt.get_captured_byte(), 4096);
+    }
+
+    #[cfg(all(unix, feature = "libtrace"))]
+    #[test]
+    fn get_captured_byte_for_merged_ebpf_reasm_start_should_use_merged_payload_len() {
+        let mut start = MetaPacket::default();
+        start.tap_port = TapPort::from_ebpf(1, 0);
+        start.ebpf_type = EbpfType::GoHttp2Uprobe;
+        start.segment_flags = SegmentFlags::Start;
+        start.reasm_bytes = 16_384;
+        start.raw_from_ebpf = vec![0u8; 16_384];
+        start.l4_payload_len = 16_384;
+        start.payload_len = 16_384;
+        start.packet_len = 16_384 + 54;
+        start.cap_start_seq = 10;
+        start.cap_end_seq = 10;
+        start.sub_packets.push(SubPacket::default());
+
+        let mut seg = MetaPacket::default();
+        seg.tap_port = TapPort::from_ebpf(1, 0);
+        seg.ebpf_type = EbpfType::GoHttp2Uprobe;
+        seg.segment_flags = SegmentFlags::Seg;
+        seg.reasm_bytes = 32_768;
+        seg.raw_from_ebpf = vec![0u8; 16_384];
+        seg.l4_payload_len = 16_384;
+        seg.payload_len = 16_384;
+        seg.packet_len = 16_384 + 54;
+        seg.cap_start_seq = 11;
+        seg.cap_end_seq = 11;
+
+        start.merge(&mut seg);
+
+        assert_eq!(start.l4_payload_len, 32_768);
+        assert_eq!(start.reasm_bytes, 32_768);
+        assert_eq!(start.get_captured_byte(), 32_768);
+    }
 }