fix(sight): improve SSL capture robustness and SSE large payload handling

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
       }
     ]
   }

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));

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};

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\"}";

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<u8>) -> Rc<SslEvent> {
+        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()
+        ));
+    }
+}

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/<pid>/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)
     }
 

src/agentsight/src/probes/sslsniff.rs

@@ -315,12 +315,19 @@ impl SslSniff {
                 .collect::<Vec<_>>()
         );
 
-        let mut attached_inodes = Vec::new();
+        let mut attached_inodes: Vec<u64> = 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<Vec<(String, u64, SslLibKind)>> {
             // 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/<pid>/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}")
             };