diff --git a/src/streams/agents/claude.rs b/src/streams/agents/claude.rs index b7f4e9632a..8ea28613a9 100644 --- a/src/streams/agents/claude.rs +++ b/src/streams/agents/claude.rs @@ -186,6 +186,7 @@ impl Agent for ClaudeAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -203,6 +204,17 @@ impl Agent for ClaudeAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -222,8 +234,10 @@ impl Agent for ClaudeAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/codex.rs b/src/streams/agents/codex.rs index eecb76a2b9..a074f88ec4 100644 --- a/src/streams/agents/codex.rs +++ b/src/streams/agents/codex.rs @@ -250,6 +250,7 @@ impl Agent for CodexAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; let mut line = String::new(); @@ -267,6 +268,17 @@ impl Agent for CodexAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -286,8 +298,10 @@ impl Agent for CodexAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/copilot.rs b/src/streams/agents/copilot.rs index dacd97dc88..9359854373 100644 --- a/src/streams/agents/copilot.rs +++ b/src/streams/agents/copilot.rs @@ -527,6 +527,7 @@ pub(super) fn read_event_stream( })?; let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -545,6 +546,17 @@ pub(super) fn read_event_stream( line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -564,8 +576,9 @@ pub(super) fn read_event_stream( } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES { break; } } diff --git a/src/streams/agents/cursor.rs b/src/streams/agents/cursor.rs index 606fc6634a..a9b64a1e95 100644 --- a/src/streams/agents/cursor.rs +++ b/src/streams/agents/cursor.rs @@ -158,6 +158,7 @@ impl Agent for CursorAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -175,6 +176,17 @@ impl Agent for CursorAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -194,8 +206,10 @@ impl Agent for CursorAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/droid.rs b/src/streams/agents/droid.rs index 6dc37a4276..1c4fccac5d 100644 --- a/src/streams/agents/droid.rs +++ b/src/streams/agents/droid.rs @@ -165,6 +165,7 @@ impl Agent for DroidAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; let mut latest_timestamp: Option> = @@ -185,6 +186,17 @@ impl Agent for DroidAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } // Skip empty lines @@ -225,8 +237,10 @@ impl Agent for DroidAgent { } // Push raw JSON entry + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/gemini.rs b/src/streams/agents/gemini.rs index 449414fe0f..4f59b7f3dd 100644 --- a/src/streams/agents/gemini.rs +++ b/src/streams/agents/gemini.rs @@ -160,6 +160,7 @@ impl Agent for GeminiAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -177,6 +178,17 @@ impl Agent for GeminiAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -196,8 +208,10 @@ impl Agent for GeminiAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/pi.rs b/src/streams/agents/pi.rs index 2aad9ceb81..ce60c88eb9 100644 --- a/src/streams/agents/pi.rs +++ b/src/streams/agents/pi.rs @@ -91,6 +91,7 @@ impl Agent for PiAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -108,6 +109,17 @@ impl Agent for PiAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -127,8 +139,10 @@ impl Agent for PiAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/windsurf.rs b/src/streams/agents/windsurf.rs index cae5a720c3..65ebdc1716 100644 --- a/src/streams/agents/windsurf.rs +++ b/src/streams/agents/windsurf.rs @@ -91,6 +91,7 @@ impl Agent for WindsurfAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -108,6 +109,17 @@ impl Agent for WindsurfAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -127,8 +139,10 @@ impl Agent for WindsurfAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/types.rs b/src/streams/types.rs index 8a799b5b54..a3d7637551 100644 --- a/src/streams/types.rs +++ b/src/streams/types.rs @@ -1,9 +1,24 @@ //! Core types for transcript processing. -use std::io::BufRead; +use std::io::{BufRead, Read}; use std::time::Duration; +/// Upper bound on a single JSONL line. Lines beyond this are skipped rather +/// than buffered: `read_line` is otherwise unbounded, and a single multi-hundred-MB +/// transcript line can balloon daemon RSS past the memory watchdog's hard limit +/// (git-ai#2244). 8 MiB comfortably fits any legitimate transcript event. +pub const MAX_JSONL_LINE_BYTES: u64 = 8 * 1024 * 1024; + +/// Upper bound on the total raw bytes carried by one transcript batch. +/// Batches were previously capped by event count only (1000), so a transcript +/// whose events embed file contents could put hundreds of MB into a single +/// batch — which downstream redaction + metrics conversion amplifies several +/// times over (git-ai#2244). The batch loop stops early once this budget is +/// spent; remaining events arrive in later batches. +pub const MAX_BATCH_BYTES: usize = 8 * 1024 * 1024; + /// Result of reading a single line from a JSONL reader. +#[derive(Debug)] pub enum JsonlLineState { /// End of file reached. Eof, @@ -11,21 +26,47 @@ pub enum JsonlLineState { Partial, /// Complete line ready for processing. Contains bytes read. Complete(usize), + /// Line exceeded [`MAX_JSONL_LINE_BYTES`] and was skipped without being + /// buffered. Contains total bytes consumed (cap + remainder up to and + /// including the newline) so callers can advance their watermark past it. + Oversized(usize), } /// Read a line from a BufReader, detecting partial writes from concurrent writers. /// /// Returns `Eof` if no more data, `Partial` if the line lacks a trailing newline, -/// or `Complete(bytes)` on success. +/// `Complete(bytes)` on success, or `Oversized(bytes)` when the line exceeded +/// [`MAX_JSONL_LINE_BYTES`] (content is discarded, reader advanced past the newline). pub fn read_jsonl_line( reader: &mut impl BufRead, line: &mut String, ) -> std::io::Result { line.clear(); - let bytes_read = reader.read_line(line)?; + // Read raw bytes, not via read_line: the byte cap can slice a multi-byte + // character, and read_line's UTF-8 validation would then return + // InvalidData instead of letting us classify the line as Oversized — + // wedging the stream at a fixed watermark. + let mut buf = Vec::new(); + let bytes_read = reader + .by_ref() + .take(MAX_JSONL_LINE_BYTES) + .read_until(b'\n', &mut buf)?; if bytes_read == 0 { return Ok(JsonlLineState::Eof); } + if buf.last() != Some(&b'\n') && bytes_read as u64 == MAX_JSONL_LINE_BYTES { + // The cap was hit mid-line: discard what we buffered (no UTF-8 + // conversion is attempted on discarded content) and skip the rest of + // the physical line without storing it. If EOF arrives before the + // newline (giant line still being written), we still report + // Oversized — re-reading it later would OOM anyway. + drop(buf); + let skipped = reader.skip_until(b'\n')?; + return Ok(JsonlLineState::Oversized(bytes_read + skipped)); + } + // Same contract as read_line: non-UTF-8 content is an InvalidData error. + *line = String::from_utf8(buf) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; if !line.ends_with('\n') { return Ok(JsonlLineState::Partial); } @@ -195,4 +236,71 @@ mod tests { let r2 = read_jsonl_line(&mut reader, &mut line).unwrap(); assert!(matches!(r2, JsonlLineState::Partial)); } + + #[test] + fn test_read_jsonl_line_oversized_is_skipped_and_reader_recovers() { + let big = "x".repeat(MAX_JSONL_LINE_BYTES as usize + 100); + let data = format!("{big}\n{{\"ok\":1}}\n"); + let mut reader = std::io::BufReader::new(data.as_bytes()); + let mut line = String::new(); + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Oversized(consumed) => { + assert_eq!(consumed, big.len() + 1); + assert!(line.is_empty(), "oversized content must not be retained"); + } + _ => panic!("expected Oversized for a line beyond MAX_JSONL_LINE_BYTES"), + } + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Complete(_) => assert_eq!(line.trim_end(), "{\"ok\":1}"), + _ => panic!("expected the next line to parse normally"), + } + } + + #[test] + fn test_read_jsonl_line_oversized_without_newline_at_eof() { + let big = "x".repeat(MAX_JSONL_LINE_BYTES as usize + 50); + let mut reader = std::io::BufReader::new(big.as_bytes()); + let mut line = String::new(); + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Oversized(consumed) => assert_eq!(consumed, big.len()), + _ => panic!("expected Oversized even when the giant line lacks a newline"), + } + } + + #[test] + fn test_read_jsonl_line_oversized_multibyte_at_cap_boundary() { + // 8 MiB is not divisible by 3, so a line of 3-byte characters + // guarantees the byte cap slices mid-character. The reader must + // classify Oversized — a read_line-based implementation returns + // InvalidData here and wedges the stream at a fixed watermark. + let euro = "€"; // 3 bytes in UTF-8 + let big = euro.repeat((MAX_JSONL_LINE_BYTES as usize / 3) + 50); + let data = format!("{big}\n{{\"ok\":1}}\n"); + let mut reader = std::io::BufReader::new(data.as_bytes()); + let mut line = String::new(); + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Oversized(consumed) => assert_eq!(consumed, big.len() + 1), + _ => panic!("expected Oversized for a multi-byte giant line"), + } + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Complete(_) => assert_eq!(line.trim_end(), "{\"ok\":1}"), + _ => panic!("expected the next line to parse normally"), + } + } + + #[test] + fn test_read_jsonl_line_invalid_utf8_within_cap_still_errors() { + // Non-UTF-8 content on a normal-sized line keeps read_line's + // InvalidData contract. + let data: &[u8] = b"\xff\xfe bad bytes\n"; + let mut reader = std::io::BufReader::new(data); + let mut line = String::new(); + let err = read_jsonl_line(&mut reader, &mut line).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + } }