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..edf7b92a9 100644 --- a/src/agentsight/src/analyzer/unified.rs +++ b/src/agentsight/src/analyzer/unified.rs @@ -1501,6 +1501,18 @@ data:{"usage":{"input_tokens":57,"output_tokens":3}}"#; assert_eq!(record.output_tokens, 3); } + #[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..60bdfc5ca 100644 --- a/src/agentsight/src/parser/unified.rs +++ b/src/agentsight/src/parser/unified.rs @@ -80,18 +80,58 @@ 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) { + Some(buf.len() - TERMINATOR.len()) + } 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 +218,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..34630d0f7 100644 --- a/src/agentsight/src/probes/sslsniff.rs +++ b/src/agentsight/src/probes/sslsniff.rs @@ -315,12 +315,19 @@ impl SslSniff { .collect::>() ); - let mut attached_inodes = Vec::new(); + let mut attached_inodes: Vec = Vec::new(); for (path, inode, kind) in libs { // Skip libraries whose inode we already traced. - // Now using pid=-1 for global attach, so each library only needs to be attached once. + // 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; } @@ -417,7 +424,8 @@ impl SslSniff { let still_used = self .pid_inodes .values() - .any(|other_inodes| other_inodes.contains(inode)); + .flatten() + .any(|other_inode| other_inode == inode); if !still_used { self.traced_files.remove(inode); removed += 1; @@ -776,6 +784,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 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}") }; 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); } }