From c7256fc5180c0f55eb79950459e369b76fc7aad5 Mon Sep 17 00:00:00 2001 From: liyuqing Date: Thu, 9 Jul 2026 19:03:20 +0800 Subject: [PATCH 1/2] fix(sight): fix codex SSL capture and SSE token extraction BoringSSL (codex) uses /proc//exe for uprobe attach path because overlayfs containers resolve /proc//root/ to a different inode than the kernel uses for uprobe matching. BoringSSL inodes are tracked separately from shared library inodes: detached on process exit to allow re-attach for new processes, while shared libraries keep the still_used check to avoid duplicate uprobe fds. The HTTP chunked terminator (0\r\n\r\n) detection now checks if the buffer ends with the terminator rather than requiring an exact 5-byte match. The terminator is typically appended to the last SSE data chunk, not a standalone read. is_done() no longer terminates the SSE stream on the event field alone for OpenAI Responses API. The response.completed data field routinely exceeds 16 KB and spans multiple TLS records; terminating early causes the usage-bearing tail to be lost. The stream now terminates via the chunked terminator or when the full JSON is parseable. The done event's source chunk is always appended to the continuation buffer, bypassing the source_event pointer dedup, so usage data in the final chunk is not lost. SSEParser's remaining field is also checked for partial events that lack a trailing \n\n. attach_process registers the pid in the BPF traced map even when SSL attach fails, so global uprobe events are not silently dropped. Signed-off-by: liyuqing --- src/agentsight/agentsight.json | 6 + .../src/aggregator/http/aggregator.rs | 51 ++++++- src/agentsight/src/analyzer/unified.rs | 45 +++++- src/agentsight/src/parser/sse/event.rs | 58 ++++++-- src/agentsight/src/parser/unified.rs | 128 ++++++++++++++++-- src/agentsight/src/probes/probes.rs | 10 +- src/agentsight/src/probes/sslsniff.rs | 64 ++++++--- src/agentsight/src/unified.rs | 12 +- 8 files changed, 324 insertions(+), 50 deletions(-) diff --git a/src/agentsight/agentsight.json b/src/agentsight/agentsight.json index 674bcbf6b..07afe5726 100644 --- a/src/agentsight/agentsight.json +++ b/src/agentsight/agentsight.json @@ -57,6 +57,7 @@ {"rule": ["*python*", "-m", "*hermes*"], "agent_name": "Hermes"}, {"rule": ["*codex*"], "agent_name": "Codex"}, {"rule": ["*node*", "*codex*"], "agent_name": "Codex"}, + {"rule": ["*/node*", "*", "*", "*", "*", "*server/index.ts*"], "agent_name": "Runloop"}, {"rule": ["node*", "*/bin/co*"], "agent_name": "Cosh"}, {"rule": ["node*", "*/bin/cosh*"], "agent_name": "Cosh"}, {"rule": ["node*", "*/bin/copliot*"], "agent_name": "Cosh"}, @@ -84,6 +85,11 @@ "codex_version": "0.137.0", "fingerprint": { "file_size": 227758400, "head_64k_sha256": "49b0a1c3a831071e9766cd0db83537a5270a374f8ac11783264a5a354c3b544a" }, "offsets": { "ssl_write": 172964768, "ssl_read": 172964176, "ssl_do_handshake": 172962496, "write_is_ex": true, "read_is_ex": true } + }, + { + "codex_version": "0.50.0", + "fingerprint": { "build_id": "bb5d7e7c56d11f3877c23081359ad9b94faa85fe", "file_size": 46101568, "head_64k_sha256": "8a223d7f6dfcbc6f8a21a15cb349ae81ec4b82fc3aeb37383cf3331f4b95dbac" }, + "offsets": null } ] } diff --git a/src/agentsight/src/aggregator/http/aggregator.rs b/src/agentsight/src/aggregator/http/aggregator.rs index 25a0a943f..608483fe3 100644 --- a/src/agentsight/src/aggregator/http/aggregator.rs +++ b/src/agentsight/src/aggregator/http/aggregator.rs @@ -802,15 +802,20 @@ impl HttpConnectionAggregator { // Responses API — other providers emit usage in single // small events. Dedup by source_event pointer so a single // SSL_read producing multiple SSE events contributes only - // once. + // once — EXCEPT for the done event, which we always append + // because its source chunk carries usage data that must not + // be lost to dedup. if needs_sse_continuation_buffer(request.as_ref()) { const MAX_CONTINUATION_BYTES: usize = 1 << 20; // 1 MiB cap let src = sse_event.source_event(); let src_ptr = src as *const _ as usize; let src_buf_len = src.buf_size() as usize; let last_ptr = self.last_appended_src_ptr.get(connection_id).copied(); - if last_ptr != Some(src_ptr) && src_buf_len > 0 && src_buf_len <= src.buf.len() - { + let should_append = is_done + || (last_ptr != Some(src_ptr) + && src_buf_len > 0 + && src_buf_len <= src.buf.len()); + if should_append && src_buf_len > 0 && src_buf_len <= src.buf.len() { let buf = self .sse_continuation_buffers .get_or_insert_mut(*connection_id, Vec::new); @@ -2338,6 +2343,46 @@ mod tests { ); } + #[test] + fn test_sse_continuation_buffer_done_event_bypasses_dedup() { + // The done event's source chunk must always be appended to the + // continuation buffer, even when it shares the same source_event + // pointer as a prior event. Without this bypass, the usage data + // in the done event's chunk would be lost to dedup. + let mut aggregator = HttpConnectionAggregator::new(); + let conn_id = enter_uncompressed_responses_sse_active(&mut aggregator, 32, 0xE000); + + // Two events sharing the same source SslEvent — first is non-done, + // second is done. The done event carries usage data in its source + // buffer that must not be skipped. + let payload = b"data: {\"type\":\"response.output_text.delta\",\"delta\":\"x\"}"; + let usage_payload = b"data: {\"usage\":{\"input_tokens\":42,\"output_tokens\":7}}"; + + // Non-done event from a different source + let src1 = create_mock_ssl_event_with_buf(32, 0xE000, payload.to_vec(), 0); + let event1 = ParsedSseEvent::new(None, None, None, 0, payload.len(), src1); + let _ = aggregator.process_sse_event(&conn_id, event1); + + // Done event sharing the SAME source pointer as event1 — its buffer + // contains usage data that differs from event1's payload. + let src_done = create_mock_ssl_event_with_buf(32, 0xE000, usage_payload.to_vec(), 0); + let done = ParsedSseEvent::new_done_marker(src_done); + let result = aggregator.process_sse_event(&conn_id, done); + let pair = match result { + Some(AggregatedResult::SseComplete(pair)) => pair, + other => panic!("expected SseComplete, got {other:?}"), + }; + let buf = pair + .response + .sse_continuation_bytes + .expect("continuation buffer should exist"); + // The done event's payload must appear in the buffer despite dedup. + assert!( + buf.windows(usage_payload.len()).any(|w| w == usage_payload), + "done event source chunk must be in continuation buffer even with dedup" + ); + } + #[test] fn test_with_limits_sets_fields() { let agg = HttpConnectionAggregator::with_limits(10, 4096, Duration::from_secs(30)); diff --git a/src/agentsight/src/analyzer/unified.rs b/src/agentsight/src/analyzer/unified.rs index 34fefaf59..9ed480740 100644 --- a/src/agentsight/src/analyzer/unified.rs +++ b/src/agentsight/src/analyzer/unified.rs @@ -630,12 +630,24 @@ impl Analyzer { if from_events.is_some() { return from_events; } + // The final response.completed event may be incomplete (no + // trailing \n\n) because its data field spans multiple TLS + // records and later chunks arrived after the stream closed. + // SSEParser puts such partial events in `remaining`; try to + // extract usage from that too. + if !reassembled.remaining.is_empty() { + let from_remaining = self.token.parse_data(&reassembled.remaining); + if from_remaining.is_some() { + return from_remaining; + } + } let from_scan = self.token.parse_data(&text); if from_scan.is_none() { log::debug!( - "[extract_token_from_sse] continuation buffer scan miss: len={} reassembled_events={}", + "[extract_token_from_sse] continuation buffer scan miss: len={} reassembled_events={} remaining_len={}", extra.len(), reassembled.events.len(), + reassembled.remaining.len(), ); } from_scan @@ -1501,6 +1513,37 @@ data:{"usage":{"input_tokens":57,"output_tokens":3}}"#; assert_eq!(record.output_tokens, 3); } + #[test] + fn test_extract_token_from_sse_remaining_field() { + // Covers the SSEParser `remaining` fallback: usage extracted from + // a partial SSE event without trailing \n\n. + let analyzer = Analyzer::new(); + let events = vec![create_test_event( + "data: {\"type\":\"response.output_text.delta\"}", + )]; + // Simulate a continuation buffer where the SSE event is incomplete + // (no trailing double-newline), so SSEParser puts it in `remaining` + // rather than in `events`. + let continuation = b"data: {\"usage\":{\"input_tokens\":100,\"output_tokens\":42}}"; + let record = analyzer + .extract_token_from_sse(&events, Some(continuation), 1234, "test") + .expect("should recover from remaining field"); + assert_eq!(record.input_tokens, 100); + assert_eq!(record.output_tokens, 42); + } + + #[test] + fn test_extract_token_from_sse_continuation_no_usage_returns_none() { + // Continuation buffer with no parseable token data at all. + let analyzer = Analyzer::new(); + let events = vec![create_test_event( + "data: {\"type\":\"response.output_text.delta\"}", + )]; + let continuation = b"event: response.completed\ndata: {\"id\":\"resp_001\"}\n\n"; + let result = analyzer.extract_token_from_sse(&events, Some(continuation), 1234, "test"); + assert!(result.is_none(), "should return None when no usage found"); + } + #[test] fn test_analyze_aggregated_response_only_with_sse() { use crate::aggregator::{AggregatedResponse, ConnectionId}; diff --git a/src/agentsight/src/parser/sse/event.rs b/src/agentsight/src/parser/sse/event.rs index 267a9895a..081999c4b 100644 --- a/src/agentsight/src/parser/sse/event.rs +++ b/src/agentsight/src/parser/sse/event.rs @@ -98,8 +98,12 @@ impl ParsedSseEvent { /// Recognizes: /// - OpenAI style: data is `[DONE]` or `[END]` /// - Anthropic style: event field is `message_stop`, or data is `{"type":"message_stop"}` - /// - OpenAI Responses API: event field is `response.completed`/`response.failed`/`response.incomplete`, - /// or data is `{"type":"response.completed",...}`. + /// - OpenAI Responses API: data contains parseable JSON with + /// `"type":"response.completed"` (or `.failed`/`.incomplete`). + /// The `event:` field alone does NOT terminate — the data payload of + /// `response.completed` routinely exceeds 16 KB and spans multiple TLS + /// records, so the stream must stay open until the full JSON is available + /// or the HTTP chunked terminator (`0\r\n\r\n`) fires a synthetic done. pub fn is_done(&self) -> bool { if self.is_synthetic_done { return true; @@ -108,14 +112,13 @@ impl ParsedSseEvent { if self.event.as_deref() == Some("message_stop") { return true; } - // OpenAI Responses API: event field flags a terminal frame. - // Use this even when the data payload is too large to parse as JSON. - if matches!( - self.event.as_deref(), - Some("response.completed") | Some("response.failed") | Some("response.incomplete") - ) { - return true; - } + // OpenAI Responses API: do NOT terminate based on event field alone. + // The `response.completed` data field routinely exceeds 16KB (echoes + // instructions + tools + output), so the first TLS chunk only has + // the event header + truncated data. The stream must stay open to + // receive remaining chunks; it will terminate via the HTTP chunked + // end marker `0\r\n\r\n` (synthetic done) or when the data-field + // JSON is fully parseable (check below). let data = self.data(); let text = String::from_utf8_lossy(data); let trimmed = text.trim(); @@ -575,6 +578,41 @@ mod tests { assert!(!parsed.is_done()); } + #[test] + fn test_is_done_responses_api_event_field_alone_not_done() { + // The `event: response.completed` field alone must NOT terminate the + // stream — the data field may be truncated across TLS records and the + // full JSON (containing usage) has not arrived yet. The stream + // terminates via the HTTP chunked terminator `0\r\n\r\n`. + let data = b"{\"sequence_number\":10,\"type\":\"response.completed\",\"response\":{\"instructions\":\"trunc"; + let ev = make_event(data); + let parsed = ParsedSseEvent::new( + None, + Some("response.completed".to_string()), + None, + 0, + data.len(), + ev, + ); + assert!(!parsed.is_done()); + } + + #[test] + fn test_is_done_responses_api_complete_json_is_done() { + // When the full JSON IS available in data, is_done should fire. + let data = b"{\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}"; + let ev = make_event(data); + let parsed = ParsedSseEvent::new( + None, + Some("response.completed".to_string()), + None, + 0, + data.len(), + ev, + ); + assert!(parsed.is_done()); + } + #[test] fn test_parsed_sse_event_json_body() { let data = b"{\"key\":\"value\"}"; diff --git a/src/agentsight/src/parser/unified.rs b/src/agentsight/src/parser/unified.rs index f68fa833d..7abcfe3e1 100644 --- a/src/agentsight/src/parser/unified.rs +++ b/src/agentsight/src/parser/unified.rs @@ -80,18 +80,62 @@ impl Parser { } } - // 2.5. Detect HTTP chunked transfer encoding end marker "0\r\n\r\n" - // This signals end of a chunked SSE stream (e.g., SysOM POP API) - // Synthesize a [DONE] event so the SSE aggregator can complete the stream + // 2.5. Detect HTTP chunked transfer encoding end marker "0\r\n\r\n". + // This signals end of a chunked SSE stream (e.g., OpenAI Responses API). + // The terminator is often appended to the end of the last SSE data + // chunk rather than arriving as a standalone 5-byte read, so we check + // whether the buffer ends with (or equals) `0\r\n\r\n`. + // + // When found, strip the terminator bytes, parse any remaining SSE data + // from the prefix, and synthesize a [DONE] event so the aggregator can + // complete the stream. { let buf_size = ssl_event.buf_size() as usize; let buf = &ssl_event.buf[..buf_size]; - if buf == b"0\r\n\r\n" { - return ParseResult { - messages: vec![ParsedMessage::SseEvent(ParsedSseEvent::new_done_marker( - Rc::clone(&ssl_event), - ))], - }; + const TERMINATOR: &[u8] = b"0\r\n\r\n"; + + let terminator_pos = if buf.len() >= TERMINATOR.len() && buf.ends_with(TERMINATOR) { + // Buffer ends with `0\r\n\r\n` — could be standalone or + // appended to SSE data. Strip it and keep the prefix for SSE + // parsing. + Some(buf.len() - TERMINATOR.len()) + } else { + // Also handle the case where `0\r\n\r\n` is the entire buffer. + if buf == TERMINATOR { Some(0) } else { None } + }; + + if let Some(prefix_len) = terminator_pos { + let prefix = &buf[..prefix_len]; + let mut messages = Vec::new(); + + // Parse SSE events from the data before the terminator + if !prefix.is_empty() { + let trimmed = SslEvent { + source: ssl_event.source, + timestamp_ns: ssl_event.timestamp_ns, + delta_ns: ssl_event.delta_ns, + pid: ssl_event.pid, + tid: ssl_event.tid, + uid: ssl_event.uid, + len: prefix_len as u32, + rw: ssl_event.rw, + comm: ssl_event.comm.clone(), + buf: ssl_event.buf[..prefix_len].to_vec(), + is_handshake: ssl_event.is_handshake, + ssl_ptr: ssl_event.ssl_ptr, + }; + let prefix_events = self.sse_parser.parse(Rc::new(trimmed)); + for ev in prefix_events { + messages.push(ParsedMessage::SseEvent(ev)); + } + } + + // Always append a synthetic done marker + messages.push(ParsedMessage::SseEvent(ParsedSseEvent::new_done_marker( + Rc::clone(&ssl_event), + ))); + + return ParseResult { messages }; } } @@ -178,3 +222,69 @@ impl Parser { &self.http2_parser } } + +#[cfg(test)] +mod tests { + use super::*; + + fn make_ssl_event(data: Vec) -> Rc { + let len = data.len(); + Rc::new(SslEvent { + source: 0, + timestamp_ns: 0, + delta_ns: 0, + pid: 1, + tid: 1, + uid: 0, + len: len as u32, + rw: 0, + comm: String::new(), + buf: data, + is_handshake: false, + ssl_ptr: 0x1000, + }) + } + + #[test] + fn test_chunked_terminator_standalone() { + let parser = Parser::new(); + let event = make_ssl_event(b"0\r\n\r\n".to_vec()); + let result = parser.parse_ssl_event(event); + assert_eq!(result.messages.len(), 1); + assert!(matches!( + result.messages[0], + ParsedMessage::SseEvent(ref e) if e.is_done() + )); + } + + #[test] + fn test_chunked_terminator_appended_to_sse_data() { + let parser = Parser::new(); + let data = b"data: {\"type\":\"response.output_text.delta\"}\n\n0\r\n\r\n".to_vec(); + let event = make_ssl_event(data); + let result = parser.parse_ssl_event(event); + // Should produce: 1 SSE event from the data + 1 done marker + assert_eq!(result.messages.len(), 2); + assert!(matches!( + result.messages[0], + ParsedMessage::SseEvent(ref e) if !e.is_done() + )); + assert!(matches!( + result.messages[1], + ParsedMessage::SseEvent(ref e) if e.is_done() + )); + } + + #[test] + fn test_no_chunked_terminator_normal_sse() { + let parser = Parser::new(); + let data = b"data: hello\n\n".to_vec(); + let event = make_ssl_event(data); + let result = parser.parse_ssl_event(event); + assert_eq!(result.messages.len(), 1); + assert!(matches!( + result.messages[0], + ParsedMessage::SseEvent(ref e) if !e.is_done() + )); + } +} diff --git a/src/agentsight/src/probes/probes.rs b/src/agentsight/src/probes/probes.rs index 89da014af..ddc62756a 100644 --- a/src/agentsight/src/probes/probes.rs +++ b/src/agentsight/src/probes/probes.rs @@ -237,7 +237,15 @@ impl Probes { } pub fn attach_process(&mut self, pid: i32) -> Result<()> { - self.attach_ssl_to_process(pid)?; + if let Err(e) = self.attach_ssl_to_process(pid) { + // SSL attach may fail when the process exits before /proc//maps is + // readable, but a previously-attached global uprobe (pid=-1) is still + // valid. Register the pid in the traced map so BPF events from the + // global uprobe are not silently dropped. + log::warn!( + "[attach_process] pid={pid}: SSL attach failed ({e:#}); registering pid anyway for global uprobe coverage" + ); + } self.add_traced_pid(pid as u32) } diff --git a/src/agentsight/src/probes/sslsniff.rs b/src/agentsight/src/probes/sslsniff.rs index 6d00ded54..6fddad89b 100644 --- a/src/agentsight/src/probes/sslsniff.rs +++ b/src/agentsight/src/probes/sslsniff.rs @@ -236,9 +236,12 @@ pub struct SslSniff { skel: Box>, _links: Vec, traced_files: HashSet, - /// Maps pid -> inodes that were attached for this pid. - /// Used to clean up traced_files when the process exits. - pid_inodes: HashMap>, + /// Maps pid -> inodes that were attached for this pid, with a flag + /// indicating whether the inode is a BoringSSL static binary (e.g. codex). + /// BoringSSL inodes are always removed on detach to force re-attach with + /// the new process's /proc path; shared library inodes use a still_used + /// check to avoid duplicate uprobe fds. + pid_inodes: HashMap>, // Channel for user-space SslEvent (lightweight, no need for Box) tx: crossbeam_channel::Sender, rx: crossbeam_channel::Receiver, @@ -315,14 +318,28 @@ impl SslSniff { .collect::>() ); - let mut attached_inodes = Vec::new(); + let mut attached_inodes: Vec<(u64, bool)> = Vec::new(); for (path, inode, kind) in libs { + let is_boring = matches!(kind, SslLibKind::Boring); // Skip libraries whose inode we already traced. // Now using pid=-1 for global attach, so each library only needs to be attached once. - if !self.traced_files.insert(inode) { + // Exception: BoringSSL static binaries (codex) must always re-attach + // because libbpf's perf_event_open(pid=-1) only covers VMAs that + // existed at attach time — new processes need their own attach. + if !is_boring && !self.traced_files.insert(inode) { log::debug!("[attach_process] pid={pid}: skipping already-traced {path}"); continue; } + // For BoringSSL, still track inode to avoid duplicate attach for the + // SAME process (multiple libs from same binary), but allow re-attach + // for DIFFERENT processes. + if is_boring && !self.traced_files.insert(inode) { + // Same binary already attached for a different process — + // re-attach anyway for this new process's VMAs. + log::debug!( + "[attach_process] pid={pid}: re-attaching BoringSSL {path} (new process VMA)" + ); + } log::debug!("[attach_process] pid={pid}: attaching {kind:?} → {path}"); @@ -383,7 +400,7 @@ impl SslSniff { match result { Ok(ls) => { self._links.extend(ls); - attached_inodes.push(inode); + attached_inodes.push((inode, is_boring)); } Err(e) => { // Attach failed: remove inode from traced_files so retries can succeed @@ -403,24 +420,26 @@ impl SslSniff { /// Detach SSL probes for a process and clean up traced inodes. /// - /// When a process exits, its inodes are removed from `traced_files` **only - /// if no other traced pid still references the same inode**. Uprobes are - /// attached globally (`pid=-1`), so the link remains valid for other - /// processes using the same library; removing the inode prematurely would - /// cause the scanner to re-attach on the next sweep, producing duplicate - /// uprobe fds. + /// BoringSSL (static binary) inodes are always removed so the next process + /// re-attaches with its own /proc path. Shared library inodes use a + /// still_used check to avoid duplicate uprobe fds. pub fn detach_process(&mut self, pid: u32) { if let Some(inodes) = self.pid_inodes.remove(&pid) { let mut removed = 0; - for inode in &inodes { - // Check whether another pid still maps this inode. - let still_used = self - .pid_inodes - .values() - .any(|other_inodes| other_inodes.contains(inode)); - if !still_used { + for (inode, is_boring) in &inodes { + if *is_boring { self.traced_files.remove(inode); removed += 1; + } else { + let still_used = self + .pid_inodes + .values() + .flatten() + .any(|(other_inode, _)| other_inode == inode); + if !still_used { + self.traced_files.remove(inode); + removed += 1; + } } } log::debug!( @@ -776,6 +795,13 @@ fn ssl_libs_from_maps(pid: i32) -> Result> { // both host and container processes. let path_str = if path_str.ends_with(" (deleted)") { format!("/proc/{pid}/exe") + } else if matches!(kind, SslLibKind::Boring) { + // BoringSSL is typically a static binary (codex, node, etc). + // /proc//exe is a kernel-maintained symlink that always + // resolves to the inode the kernel uses for uprobe matching. + // /proc//root/ can resolve to a different inode + // in overlayfs containers and some host configurations. + format!("/proc/{pid}/exe") } else { format!("/proc/{pid}/root{path_str}") }; diff --git a/src/agentsight/src/unified.rs b/src/agentsight/src/unified.rs index e7a5798a6..a14ed44dd 100644 --- a/src/agentsight/src/unified.rs +++ b/src/agentsight/src/unified.rs @@ -584,15 +584,13 @@ impl AgentSight { /// Internal helper to attach SSL probes to a process fn attach_process_internal(probes: &mut Probes, pid: u32, agent_name: &str) { - log::debug!("Attaching to pid {pid}, agent name: {agent_name}"); if let Err(e) = probes.add_traced_pid(pid) { log::warn!("Failed to add pid {pid} to traced_processes map: {e}"); } - if let Err(e) = probes.attach_process(pid as i32) { - log::error!("Failed to attach SSL probe to pid {pid}: {e}"); - } else { - log::info!("Attached to agent: {agent_name} (pid={pid})"); - } + // attach_process logs WARN internally if SSL attach degrades to + // global-uprobe-only coverage; always succeeds for BPF map registration. + let _ = probes.attach_process(pid as i32); + log::info!("Attached to agent: {agent_name} (pid={pid})"); } /// Detach SSL probes from a specific agent process @@ -862,7 +860,7 @@ impl AgentSight { // Remove from tracking if it was an agent 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.probes.detach_ssl_probes(*pid); self.handle_agent_crash_detection(*pid, &agent_name); } } From e86838a20128a648d1ddf2d71067a4d9a830d2e1 Mon Sep 17 00:00:00 2001 From: liyuqing Date: Fri, 10 Jul 2026 13:40:16 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(sight):=20address=20PR=20review=20?= =?UTF-8?q?=E2=80=94=20unify=20inode=20dedup=20and=20remove=20dead=20code?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unify BoringSSL and shared library dedup logic: BoringSSL no longer re-attaches on duplicate inodes. The kernel's uprobe_mmap mechanism automatically installs breakpoints for new processes that map an already-registered inode, so per-process re-attach was creating duplicate perf_event consumers — each SSL call produced N ring buffer events after N process cycles. Fix the still_used guard: dedup-skipped processes now push their inode to attached_inodes so pid_inodes tracks all referencing pids. This makes the still_used check in detach_process effective — previously dedup-skipped pids were never recorded, causing premature inode removal on first-pid exit. Remove the SSEParser remaining field check (dead code): parse_stream's .lines() fully consumes input, so remaining is always empty or a single trailing character. The from_scan fallback already covers this. Remove the unreachable terminator else branch in parser/unified.rs: when buf == TERMINATOR, the first condition (ends_with) is already true. Fix the /proc//exe comment to accurately describe the rationale: both paths go through d_real_inode() for overlayfs, so the real reason is handling replaced/unlinked binaries, not inode mismatch. Signed-off-by: liyuqing --- src/agentsight/src/analyzer/unified.rs | 33 +----------- src/agentsight/src/parser/unified.rs | 6 +-- src/agentsight/src/probes/sslsniff.rs | 75 +++++++++++--------------- 3 files changed, 34 insertions(+), 80 deletions(-) diff --git a/src/agentsight/src/analyzer/unified.rs b/src/agentsight/src/analyzer/unified.rs index 9ed480740..edf7b92a9 100644 --- a/src/agentsight/src/analyzer/unified.rs +++ b/src/agentsight/src/analyzer/unified.rs @@ -630,24 +630,12 @@ impl Analyzer { if from_events.is_some() { return from_events; } - // The final response.completed event may be incomplete (no - // trailing \n\n) because its data field spans multiple TLS - // records and later chunks arrived after the stream closed. - // SSEParser puts such partial events in `remaining`; try to - // extract usage from that too. - if !reassembled.remaining.is_empty() { - let from_remaining = self.token.parse_data(&reassembled.remaining); - if from_remaining.is_some() { - return from_remaining; - } - } let from_scan = self.token.parse_data(&text); if from_scan.is_none() { log::debug!( - "[extract_token_from_sse] continuation buffer scan miss: len={} reassembled_events={} remaining_len={}", + "[extract_token_from_sse] continuation buffer scan miss: len={} reassembled_events={}", extra.len(), reassembled.events.len(), - reassembled.remaining.len(), ); } from_scan @@ -1513,25 +1501,6 @@ data:{"usage":{"input_tokens":57,"output_tokens":3}}"#; assert_eq!(record.output_tokens, 3); } - #[test] - fn test_extract_token_from_sse_remaining_field() { - // Covers the SSEParser `remaining` fallback: usage extracted from - // a partial SSE event without trailing \n\n. - let analyzer = Analyzer::new(); - let events = vec![create_test_event( - "data: {\"type\":\"response.output_text.delta\"}", - )]; - // Simulate a continuation buffer where the SSE event is incomplete - // (no trailing double-newline), so SSEParser puts it in `remaining` - // rather than in `events`. - let continuation = b"data: {\"usage\":{\"input_tokens\":100,\"output_tokens\":42}}"; - let record = analyzer - .extract_token_from_sse(&events, Some(continuation), 1234, "test") - .expect("should recover from remaining field"); - assert_eq!(record.input_tokens, 100); - assert_eq!(record.output_tokens, 42); - } - #[test] fn test_extract_token_from_sse_continuation_no_usage_returns_none() { // Continuation buffer with no parseable token data at all. diff --git a/src/agentsight/src/parser/unified.rs b/src/agentsight/src/parser/unified.rs index 7abcfe3e1..60bdfc5ca 100644 --- a/src/agentsight/src/parser/unified.rs +++ b/src/agentsight/src/parser/unified.rs @@ -95,13 +95,9 @@ impl Parser { const TERMINATOR: &[u8] = b"0\r\n\r\n"; let terminator_pos = if buf.len() >= TERMINATOR.len() && buf.ends_with(TERMINATOR) { - // Buffer ends with `0\r\n\r\n` — could be standalone or - // appended to SSE data. Strip it and keep the prefix for SSE - // parsing. Some(buf.len() - TERMINATOR.len()) } else { - // Also handle the case where `0\r\n\r\n` is the entire buffer. - if buf == TERMINATOR { Some(0) } else { None } + None }; if let Some(prefix_len) = terminator_pos { diff --git a/src/agentsight/src/probes/sslsniff.rs b/src/agentsight/src/probes/sslsniff.rs index 6fddad89b..34630d0f7 100644 --- a/src/agentsight/src/probes/sslsniff.rs +++ b/src/agentsight/src/probes/sslsniff.rs @@ -236,12 +236,9 @@ pub struct SslSniff { skel: Box>, _links: Vec, traced_files: HashSet, - /// Maps pid -> inodes that were attached for this pid, with a flag - /// indicating whether the inode is a BoringSSL static binary (e.g. codex). - /// BoringSSL inodes are always removed on detach to force re-attach with - /// the new process's /proc path; shared library inodes use a still_used - /// check to avoid duplicate uprobe fds. - pid_inodes: HashMap>, + /// Maps pid -> inodes that were attached for this pid. + /// Used to clean up traced_files when the process exits. + pid_inodes: HashMap>, // Channel for user-space SslEvent (lightweight, no need for Box) tx: crossbeam_channel::Sender, rx: crossbeam_channel::Receiver, @@ -318,28 +315,21 @@ impl SslSniff { .collect::>() ); - let mut attached_inodes: Vec<(u64, bool)> = Vec::new(); + let mut attached_inodes: Vec = Vec::new(); for (path, inode, kind) in libs { - let is_boring = matches!(kind, SslLibKind::Boring); // Skip libraries whose inode we already traced. - // Now using pid=-1 for global attach, so each library only needs to be attached once. - // Exception: BoringSSL static binaries (codex) must always re-attach - // because libbpf's perf_event_open(pid=-1) only covers VMAs that - // existed at attach time — new processes need their own attach. - if !is_boring && !self.traced_files.insert(inode) { + // Uprobes are attached globally (pid=-1), and the kernel's + // uprobe_mmap mechanism automatically installs breakpoints for + // new processes that map an already-registered inode, so each + // library only needs to be attached once — including BoringSSL + // static binaries (codex, node, etc). + if !self.traced_files.insert(inode) { log::debug!("[attach_process] pid={pid}: skipping already-traced {path}"); + // Still record the pid→inode association so detach_process + // can track all pids referencing this inode. + attached_inodes.push(inode); continue; } - // For BoringSSL, still track inode to avoid duplicate attach for the - // SAME process (multiple libs from same binary), but allow re-attach - // for DIFFERENT processes. - if is_boring && !self.traced_files.insert(inode) { - // Same binary already attached for a different process — - // re-attach anyway for this new process's VMAs. - log::debug!( - "[attach_process] pid={pid}: re-attaching BoringSSL {path} (new process VMA)" - ); - } log::debug!("[attach_process] pid={pid}: attaching {kind:?} → {path}"); @@ -400,7 +390,7 @@ impl SslSniff { match result { Ok(ls) => { self._links.extend(ls); - attached_inodes.push((inode, is_boring)); + attached_inodes.push(inode); } Err(e) => { // Attach failed: remove inode from traced_files so retries can succeed @@ -420,26 +410,25 @@ impl SslSniff { /// Detach SSL probes for a process and clean up traced inodes. /// - /// BoringSSL (static binary) inodes are always removed so the next process - /// re-attaches with its own /proc path. Shared library inodes use a - /// still_used check to avoid duplicate uprobe fds. + /// When a process exits, its inodes are removed from `traced_files` **only + /// if no other traced pid still references the same inode**. Uprobes are + /// attached globally (`pid=-1`), so the link remains valid for other + /// processes using the same library; removing the inode prematurely would + /// cause the scanner to re-attach on the next sweep, producing duplicate + /// uprobe fds. pub fn detach_process(&mut self, pid: u32) { if let Some(inodes) = self.pid_inodes.remove(&pid) { let mut removed = 0; - for (inode, is_boring) in &inodes { - if *is_boring { + for inode in &inodes { + // Check whether another pid still maps this inode. + let still_used = self + .pid_inodes + .values() + .flatten() + .any(|other_inode| other_inode == inode); + if !still_used { self.traced_files.remove(inode); removed += 1; - } else { - let still_used = self - .pid_inodes - .values() - .flatten() - .any(|(other_inode, _)| other_inode == inode); - if !still_used { - self.traced_files.remove(inode); - removed += 1; - } } } log::debug!( @@ -797,10 +786,10 @@ fn ssl_libs_from_maps(pid: i32) -> Result> { format!("/proc/{pid}/exe") } else if matches!(kind, SslLibKind::Boring) { // BoringSSL is typically a static binary (codex, node, etc). - // /proc//exe is a kernel-maintained symlink that always - // resolves to the inode the kernel uses for uprobe matching. - // /proc//root/ can resolve to a different inode - // in overlayfs containers and some host configurations. + // /proc//exe is a kernel-maintained symlink that stays + // valid even when the backing file has been replaced or + // unlinked, which is common for npm-installed binaries that + // get updated while old processes are still running. format!("/proc/{pid}/exe") } else { format!("/proc/{pid}/root{path_str}")