daemon: advance stream watermark before batch processing

src/daemon.rs

@@ -2607,13 +2607,53 @@ fn process_alive(pid: u32) -> bool {
     unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
 }
 
-fn read_json_line<R: BufRead>(reader: &mut R) -> Result<Option<String>, GitAiError> {
-    let mut line = String::new();
-    let read = reader.read_line(&mut line)?;
+/// Upper bound on a single control/trace socket line. Both sockets speak
+/// line-delimited JSON with bounded frames (trace2 events, control request
+/// headers; checkpoint bodies travel separately after the header line), but
+/// `read_line` is otherwise unbounded, so one runaway client line can
+/// balloon daemon RSS past the memory watchdog (#2244).
+const MAX_SOCKET_LINE_BYTES: u64 = 4 * 1024 * 1024;
+
+/// A line read from a daemon socket.
+enum SocketLine {
+    Line(String),
+    /// Line exceeded [`MAX_SOCKET_LINE_BYTES`]; the content was discarded and
+    /// the reader advanced past its newline. The control loop answers these
+    /// with an error response (a silently-eaten request would stall the
+    /// client until its socket timeout and then be resent forever); the trace
+    /// loop just keeps reading.
+    Oversized,
+}
+
+fn read_json_line<R: BufRead>(reader: &mut R) -> Result<Option<SocketLine>, GitAiError> {
+    // Byte-based read, not read_line: the cap can slice a multi-byte
+    // character, and read_line's UTF-8 validation would turn that into an
+    // InvalidData error (dropping the connection) instead of an Oversized
+    // classification.
+    let mut buf = Vec::new();
+    let read = reader
+        .by_ref()
+        .take(MAX_SOCKET_LINE_BYTES)
+        .read_until(b'\n', &mut buf)?;
     if read == 0 {
         return Ok(None);
     }
-    Ok(Some(line))
+    if buf.last() != Some(&b'\n') && read as u64 == MAX_SOCKET_LINE_BYTES {
+        drop(buf);
+        let skipped = reader.skip_until(b'\n')?;
+        tracing::warn!(
+            component = "daemon",
+            phase = "socket_read",
+            consumed_bytes = read as u64 + skipped as u64,
+            max_bytes = MAX_SOCKET_LINE_BYTES,
+            "skipping oversized socket line"
+        );
+        return Ok(Some(SocketLine::Oversized));
+    }
+    let line = String::from_utf8(buf).map_err(|error| {
+        GitAiError::IoError(std::io::Error::new(std::io::ErrorKind::InvalidData, error))
+    })?;
+    Ok(Some(SocketLine::Line(line)))
 }
 
 fn read_checkpoint_body<R: BufRead>(
@@ -7862,7 +7902,20 @@ fn handle_control_connection_actor_reader<R: ControlConnection>(
     let mut uses_idle_timeout = false;
     loop {
         let line = match read_json_line(reader) {
-            Ok(Some(line)) => line,
+            Ok(Some(SocketLine::Line(line))) => line,
+            Ok(Some(SocketLine::Oversized)) => {
+                // Answer instead of silently eating the request: an
+                // unanswered client blocks until its socket timeout, then
+                // reconnects and resends the same oversized request forever.
+                if let Err(error) = write_control_response(
+                    reader.get_mut(),
+                    &ControlResponse::err("control request exceeds the maximum line size"),
+                ) {
+                    tracing::warn!(%error, "failed responding to an oversized control request");
+                    break;
+                }
+                continue;
+            }
             Ok(None) => break,
             Err(error) if control_receive_timed_out(&error) => break,
             Err(error) => return Err(error),
@@ -8471,7 +8524,11 @@ fn handle_trace_connection_actor_reader<R: Read>(
     mut observed_roots: std::collections::BTreeSet<String>,
 ) -> Result<(), GitAiError> {
     let read_result = (|| {
-        while let Some(line) = read_json_line(&mut reader)? {
+        while let Some(socket_line) = read_json_line(&mut reader)? {
+            let line = match socket_line {
+                SocketLine::Line(line) => line,
+                SocketLine::Oversized => continue,
+            };
             if process_trace_connection_line(&line, coordinator.clone(), &mut observed_roots)?
                 .is_some_and(|outcome| !outcome.continue_reading)
             {

src/daemon/stream_worker.rs

@@ -1127,8 +1127,26 @@ impl StreamWorker {
 
             let batch_count = batch.events.len();
 
+            // Advance the watermark BEFORE converting/persisting the batch.
+            // Transcript metrics are best-effort telemetry; when processing a
+            // batch OOM-aborts the daemon (memory watchdog), a watermark that
+            // only advances after processing makes the respawned daemon
+            // re-read the same bytes and die again — an abort loop (#2244).
+            // Skipping one batch of diagnostics on crash is the safer
+            // failure mode.
+            db.update_watermark(
+                &stream.session_id,
+                &task.stream_kind,
+                &stream.stream_path,
+                batch.new_watermark.as_ref(),
+            )?;
+
             let is_otel_stream = task.stream_kind == "otel_traces";
-            let metric_events: Vec<MetricEvent> = batch
+            // Serialize each event as it is built and let the MetricEvent
+            // (which still holds the redacted JSON tree) drop before the next
+            // one is constructed. Collecting Vec<MetricEvent> and serializing
+            // afterwards kept two full copies of the batch alive at once (#2244).
+            let metric_event_jsons: Vec<String> = batch
                 .events
                 .into_iter()
                 .enumerate()
@@ -1164,7 +1182,7 @@ impl StreamWorker {
 
                     let attrs_sparse = event_attrs.to_sparse();
                     let raw_event = redact_json_secrets(raw_event);
-                    Some(if is_otel_stream {
+                    let metric_event = if is_otel_stream {
                         MetricEvent::from_values_with_timestamp(
                             OtelTraceValues::with_ids(raw_event, eid, pid, tid),
                             attrs_sparse,
@@ -1176,11 +1194,18 @@ impl StreamWorker {
                             attrs_sparse,
                             Some(event_ts),
                         )
-                    })
+                    };
+                    match serde_json::to_string(&metric_event) {
+                        Ok(json) => Some(json),
+                        Err(e) => {
+                            tracing::warn!(%e, "telemetry: failed to serialize transcript metric event; dropping");
+                            None
+                        }
+                    }
                 })
                 .collect();
 
-            if let Err(e) = telemetry.persist_metrics_blocking(&metric_events) {
+            if let Err(e) = telemetry.persist_metric_jsons_blocking(&metric_event_jsons) {
                 tracing::warn!(%e, "telemetry: failed to persist transcript metrics locally");
             }
 
@@ -1202,12 +1227,6 @@ impl StreamWorker {
             }
 
             total_events += batch_count;
-            db.update_watermark(
-                &stream.session_id,
-                &task.stream_kind,
-                &stream.stream_path,
-                batch.new_watermark.as_ref(),
-            )?;
             current_watermark = batch.new_watermark;
         }
 

src/daemon/telemetry_worker.rs

@@ -435,6 +435,19 @@ impl DaemonTelemetryWorkerHandle {
         store_metrics_in_db(events)
     }
 
+    /// Persist pre-serialized metric-event JSON rows.
+    ///
+    /// Streaming variant of [`Self::persist_metrics_blocking`] for large
+    /// transcript batches: the caller serializes each event as it is built
+    /// and drops the event immediately, so a batch's `MetricEvent` trees and
+    /// their JSON copies are never all resident at the same time (#2244).
+    pub fn persist_metric_jsons_blocking(
+        &self,
+        event_jsons: &[String],
+    ) -> Result<Vec<i64>, GitAiError> {
+        store_metric_jsons_in_db(event_jsons)
+    }
+
     /// Submit telemetry envelopes synchronously (best-effort, non-blocking).
     ///
     /// Used by the daemon process's own `observability::log_*()` calls which
@@ -970,6 +983,10 @@ fn store_metrics_in_db(events: &[MetricEvent]) -> Result<Vec<i64>, GitAiError> {
         .map(serde_json::to_string)
         .collect::<Result<_, _>>()?;
 
+    store_metric_jsons_in_db(&event_jsons)
+}
+
+fn store_metric_jsons_in_db(event_jsons: &[String]) -> Result<Vec<i64>, GitAiError> {
     if event_jsons.is_empty() {
         return Ok(Vec::new());
     }
@@ -978,7 +995,7 @@ fn store_metrics_in_db(events: &[MetricEvent]) -> Result<Vec<i64>, GitAiError> {
     let mut db_lock = db
         .lock()
         .map_err(|_| GitAiError::Generic("metrics DB lock poisoned".to_string()))?;
-    db_lock.insert_events(&event_jsons)
+    db_lock.insert_events(event_jsons)
 }
 
 #[derive(Debug, Default, PartialEq, Eq)]

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

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

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

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

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<chrono::DateTime<chrono::Utc>> =
@@ -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;
             }
         }