From 6603e38874bb26f8e35fe503b2e4fcb549a9b34b Mon Sep 17 00:00:00 2001
From: "shreya.malpani" <shreya.malpani@datadoghq.com>
Date: Thu, 8 Oct 2026 10:47:12 -0400
Subject: [PATCH 1/5] fix logs payload restitch

---
 bottlecap/src/extension/telemetry/listener.rs |   2 +-
 bottlecap/src/extension/telemetry/stitch.rs   | 240 +++++++++++++++---
 2 files changed, 208 insertions(+), 34 deletions(-)

diff --git a/bottlecap/src/extension/telemetry/listener.rs b/bottlecap/src/extension/telemetry/listener.rs
index 56e96647d..4810178c8 100644
--- a/bottlecap/src/extension/telemetry/listener.rs
+++ b/bottlecap/src/extension/telemetry/listener.rs
@@ -126,7 +126,7 @@ impl TelemetryListener {
             Ok(events) => events,
             // The Telemetry API splits an oversized record across two POSTs, and neither half
             // parses alone. See `stitch`.
-            Err(e) => match state.fragments.stitch(&body, &e) {
+            Err(e) => match state.fragments.stitch(&body) {
                 Stitch::Complete(events) => {
                     debug!(
                         "TELEMETRY API | Reassembled a split payload, recovered {} events",
diff --git a/bottlecap/src/extension/telemetry/stitch.rs b/bottlecap/src/extension/telemetry/stitch.rs
index a375584c2..cd49890a7 100644
--- a/bottlecap/src/extension/telemetry/stitch.rs
+++ b/bottlecap/src/extension/telemetry/stitch.rs
@@ -1,20 +1,23 @@
 //! Reassembles telemetry payloads that the Telemetry API split across two POSTs.
 //!
-//! A record larger than the subscription's `maxBytes` is cut mid-value. The rest of it — plus
-//! the rest of the batch — arrives in the next POST, which repeats the cut record's envelope
-//! ahead of the resumed bytes. Neither half parses on its own:
+//! A log record longer than about 256 KiB is cut mid-value into pieces, each an event of its
+//! own. When a payload ends on a cut piece, the rest of it — plus the rest of the batch —
+//! arrives in the next POST, under the next piece's envelope. Neither half parses on its own:
 //!
 //! ```text
 //! POST 1  [{"time":"..929Z","type":"function","record":{"message":"iVBORw0KGgoAAA}]
 //!         `------------- envelope --------------------'`--- cut here ---'`framing'
 //!
 //! POST 2  [{"time":"..929Z","type":"function","record":QICAgIfAhkiAAA"}},{..},{..runtimeDone..}]
-//!         `------- the same envelope, repeated -------'`-- resumed --'`- rest of the batch -'
+//!         `------ the next piece's envelope ----------'`-- resumed --'`- rest of the batch -'
 //! ```
 //!
-//! So the two are joined by dropping POST 1's framing and POST 2's repeated envelope. That
-//! byte-identical envelope, which carries the cut record's timestamp, is the only thing
-//! pairing them: the API sends no sequence number.
+//! So the two are joined by dropping POST 1's framing and POST 2's envelope. Each piece carries
+//! its own timestamp, so nothing pairs them but the joined bytes parsing.
+//!
+//! Pieces also sit side by side within one POST, where the seam between them — the cut piece's
+//! closing `}` and the next piece's envelope — is dropped the same way. Serde fails at, or just
+//! after, a seam, which tells it apart from the boundary between two whole events.
 //!
 //! Recovering the batch matters beyond the log line itself. `platform.runtimeDone` lands in
 //! the second half, and the on-demand loop waits for it before calling `/next` — so dropping
@@ -22,6 +25,7 @@
 
 use serde_json::error::Category;
 use std::{
+    ops::Range,
     sync::{Arc, Mutex, PoisonError},
     time::{Duration, Instant},
 };
@@ -45,6 +49,22 @@ const RECORD_KEY: &[u8] = b"\"record\":";
 /// record's value short, so a fragment ends with framing that belongs to neither half.
 const FRAMING: &[u8] = b"}]";
 
+/// Opens a seam between two pieces: the cut piece's closing `}`, then the next one's envelope.
+const SEAM: &[u8] = b"},{\"time\":\"";
+
+/// How a piece's envelope ends: only log lines are long enough to be split.
+const PIECE_TYPES: &[&[u8]] = &[
+    b"\"type\":\"function\",\"record\":",
+    b"\"type\":\"extension\",\"record\":",
+];
+
+/// How far past a seam's start serde can fail: a cut inside a string swallows `},{"`, and
+/// the string closes on the `"` before `time`.
+const SEAM_REACH: usize = 4;
+
+/// Bounds the re-parsing of one payload: a 4 MiB payload of 256 KiB pieces has 16 seams.
+const MAX_SEAMS: usize = 64;
+
 /// What came of pairing an unparseable body with a held fragment.
 #[derive(Debug)]
 pub(crate) enum Stitch {
@@ -64,10 +84,7 @@ pub(crate) struct FragmentBuffer {
 
 impl FragmentBuffer {
     /// Joins `body` onto the held fragment, or holds `body` if it opens a split payload.
-    ///
-    /// `error` is the failure `body` produced on its own; it is how a payload cut mid-record
-    /// is told apart from one we simply can't interpret.
-    pub(crate) fn stitch(&self, body: &[u8], error: &serde_json::Error) -> Stitch {
+    pub(crate) fn stitch(&self, body: &[u8]) -> Stitch {
         let mut slot = self.held.lock().unwrap_or_else(PoisonError::into_inner);
 
         if let Some(stale) = slot.take_if(|held| held.received.elapsed() > FRAGMENT_TTL) {
@@ -78,9 +95,9 @@ impl FragmentBuffer {
         }
 
         if let Some(head) = slot.take() {
-            if let Some(resumed) = head.resumed_bytes(body) {
-                let joined = head.join(resumed);
-                return match serde_json::from_slice(&joined) {
+            if let Some(resumed) = Fragment::resumed_bytes(body) {
+                let mut joined = head.join(resumed);
+                return match unsplit(&mut joined) {
                     Ok(events) => Stitch::Complete(events),
                     // A record can be cut more than once, so keep accumulating.
                     Err(e) => hold_or_discard(&mut slot, joined, &e),
@@ -93,7 +110,11 @@ impl FragmentBuffer {
             );
         }
 
-        hold_or_discard(&mut slot, body.to_vec(), error)
+        let mut body = body.to_vec();
+        match unsplit(&mut body) {
+            Ok(events) => Stitch::Complete(events),
+            Err(e) => hold_or_discard(&mut slot, body, &e),
+        }
     }
 }
 
@@ -118,14 +139,17 @@ struct Fragment {
 }
 
 impl Fragment {
-    /// A fragment, if `body` is the leading half of a split payload: an array that ran out of
-    /// input inside its last record's value. Any other parse failure won't be fixed by
-    /// joining, and holding such a body would poison the next stitch.
+    /// A fragment, if `body` is the leading half of a split payload: an array whose last piece
+    /// was cut short, running out of input or into the framing. Any other parse failure won't
+    /// be fixed by joining, and holding such a body would poison the next stitch.
     fn from_cut_payload(body: Vec<u8>, error: &serde_json::Error) -> Option<Self> {
-        if error.classify() != Category::Eof
-            || body.len() > MAX_FRAGMENT_BYTES
-            || body.first() != Some(&b'[')
-        {
+        let cut_short = match error.classify() {
+            Category::Eof => true,
+            // Cut after a `\` or a whole value, the piece runs into the framing instead.
+            Category::Syntax => offset(&body, error) + FRAMING.len() >= body.len(),
+            _ => false,
+        };
+        if !cut_short || body.len() > MAX_FRAGMENT_BYTES || body.first() != Some(&b'[') {
             return None;
         }
 
@@ -137,15 +161,11 @@ impl Fragment {
 
     /// The bytes that resume this fragment, if `body` is its continuation.
     ///
-    /// A continuation opens with the cut record's envelope, so its `record` key is the first
+    /// A continuation opens with the next piece's envelope, so its `record` key is the first
     /// one in the payload — anything the customer nested sits inside the value that follows.
-    /// Finding the same envelope in the fragment is what pairs the two.
-    fn resumed_bytes<'a>(&self, body: &'a [u8]) -> Option<&'a [u8]> {
+    fn resumed_bytes(body: &[u8]) -> Option<&[u8]> {
         let payload = body.strip_prefix(b"[")?;
         let value_start = find(payload, RECORD_KEY)? + RECORD_KEY.len();
-        let envelope = payload.get(..value_start)?;
-
-        find(&self.body, envelope)?;
         payload.get(value_start..)
     }
 
@@ -167,6 +187,59 @@ fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
         .position(|window| window == needle)
 }
 
+/// Parses `body`, dropping the seams between pieces one at a time where serde fails on them.
+fn unsplit(body: &mut Vec<u8>) -> Result<Vec<TelemetryEvent>, serde_json::Error> {
+    let mut seams = 0;
+    loop {
+        let error = match serde_json::from_slice(body) {
+            Ok(events) => return Ok(events),
+            Err(e) => e,
+        };
+        if seams == MAX_SEAMS || error.classify() == Category::Data {
+            return Err(error);
+        }
+        match seam_at(body, offset(body, &error)) {
+            Some(seam) => body.drain(seam),
+            None => return Err(error),
+        };
+        seams += 1;
+    }
+}
+
+/// The seam serde failed on at byte `at`, if there is one.
+fn seam_at(body: &[u8], at: usize) -> Option<Range<usize>> {
+    let start = (at.saturating_sub(SEAM_REACH)..=at)
+        .rev()
+        .find(|&i| body.get(i..).is_some_and(|rest| rest.starts_with(SEAM)))?;
+
+    // The envelope holds only the timestamp and a log type; a brace means the match ran into a
+    // value, and any other type is a whole event — say `platform.runtimeDone` — not a piece.
+    let envelope = body.get(start + SEAM.len()..)?;
+    let len = find(envelope, RECORD_KEY)? + RECORD_KEY.len();
+    let envelope = &envelope[..len];
+    if envelope.iter().any(|b| matches!(b, b'{' | b'}'))
+        || !PIECE_TYPES.iter().any(|tail| envelope.ends_with(tail))
+    {
+        return None;
+    }
+
+    Some(start..start + SEAM.len() + len)
+}
+
+/// The byte serde's one-based line and column point at.
+fn offset(body: &[u8], error: &serde_json::Error) -> usize {
+    let line_start = match error.line() {
+        0 | 1 => 0,
+        line => body
+            .iter()
+            .enumerate()
+            .filter(|&(_, &b)| b == b'\n')
+            .nth(line - 2)
+            .map_or(0, |(i, _)| i + 1),
+    };
+    line_start + error.column().saturating_sub(1)
+}
+
 /// The two halves of a real split payload, trimmed to the bytes that matter.
 #[cfg(test)]
 pub(crate) mod fixtures {
@@ -186,9 +259,9 @@ mod tests {
 
     /// Mirrors the handler: a body only reaches the buffer once it has failed to parse.
     fn stitch(fragments: &FragmentBuffer, body: &str) -> Stitch {
-        let error = serde_json::from_slice::<Vec<TelemetryEvent>>(body.as_bytes())
+        serde_json::from_slice::<Vec<TelemetryEvent>>(body.as_bytes())
             .expect_err("fixture must not parse on its own");
-        fragments.stitch(body.as_bytes(), &error)
+        fragments.stitch(body.as_bytes())
     }
 
     /// The events of a stitch that should have completed, reporting what came back if it did not.
@@ -289,6 +362,108 @@ mod tests {
         );
     }
 
+    /// A record logged as JSON, with the escapes, multi-byte text, nesting, numbers and literals a
+    /// cut can land in.
+    const RECORD: &str = r#"{"level":"info","msg":"{\"a\":[1,true,null],\"b\":\"café caf\u00e9 \\\\ x\"}","n":-1.5e3}"#;
+
+    const RUNTIME_DONE: &str = r#"{"time":"2026-09-03T14:29:52.950Z","type":"platform.runtimeDone","record":{"requestId":"abc123","status":"success"}}"#;
+
+    /// A piece of `RECORD` as Lambda delivers it: raw, under an envelope of its own.
+    fn piece(millis: &str, bytes: &str) -> String {
+        format!(r#"{{"time":"2026-09-03T14:29:52.{millis}Z","type":"function","record":{bytes}}}"#)
+    }
+
+    fn assert_whole_record(events: &[TelemetryEvent]) {
+        let record = serde_json::from_str(RECORD).expect("valid record");
+        assert_eq!(events.len(), 2);
+        assert_eq!(events[0].record, TelemetryRecord::Function(record));
+        assert!(matches!(
+            events[1].record,
+            TelemetryRecord::PlatformRuntimeDone { .. }
+        ));
+    }
+
+    #[test]
+    fn joins_pieces_within_a_payload_at_every_cut() {
+        for cut in (1..RECORD.len()).filter(|&cut| RECORD.is_char_boundary(cut)) {
+            let (first, rest) = RECORD.split_at(cut);
+            let body = format!(
+                "[{},{},{RUNTIME_DONE}]",
+                piece("929", first),
+                piece("930", rest)
+            );
+
+            let Stitch::Complete(events) = stitch(&FragmentBuffer::default(), &body) else {
+                panic!("pieces cut after {first:?} were not joined");
+            };
+            assert_whole_record(&events);
+        }
+    }
+
+    #[test]
+    fn joins_pieces_across_payloads_at_every_cut() {
+        for cut in (1..RECORD.len()).filter(|&cut| RECORD.is_char_boundary(cut)) {
+            let (first, rest) = RECORD.split_at(cut);
+            let fragments = FragmentBuffer::default();
+
+            let head = format!("[{}]", piece("929", first));
+            assert!(
+                matches!(stitch(&fragments, &head), Stitch::Pending),
+                "payload ending after {first:?} was not held"
+            );
+
+            let tail = format!("[{},{RUNTIME_DONE}]", piece("930", rest));
+            let Stitch::Complete(events) = stitch(&fragments, &tail) else {
+                panic!("payloads cut after {first:?} were not joined");
+            };
+            assert_whole_record(&events);
+        }
+    }
+
+    /// The shape seen in production: a record in several pieces, the payload ending between
+    /// two of them.
+    #[test]
+    fn joins_a_record_in_pieces_across_payloads() {
+        let fragments = FragmentBuffer::default();
+
+        let (a, rest) = RECORD.split_at(20);
+        let (b, rest) = rest.split_at(20);
+        let (c, d) = rest.split_at(20);
+
+        let head = format!(
+            r#"[{{"time":"2026-09-03T14:29:52.900Z","type":"extension","record":"ready"}},{},{}]"#,
+            piece("929", a),
+            piece("930", b)
+        );
+        assert!(matches!(stitch(&fragments, &head), Stitch::Pending));
+
+        let tail = format!("[{},{},{RUNTIME_DONE}]", piece("931", c), piece("932", d));
+        let mut events = completed(stitch(&fragments, &tail));
+
+        assert_eq!(events.len(), 3);
+        assert!(matches!(
+            events.remove(0).record,
+            TelemetryRecord::Extension(_)
+        ));
+        assert_whole_record(&events);
+    }
+
+    #[test]
+    fn does_not_mistake_a_broken_record_for_pieces() {
+        let body = r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":{"message":"ok"}},{"time":"2026-09-03T14:29:52.930Z","type":"function","record":{"message":"bad\q"}}]"#;
+        assert!(matches!(
+            stitch(&FragmentBuffer::default(), body),
+            Stitch::Discarded
+        ));
+
+        // Fails where `platform.runtimeDone` begins, which a log piece's seam never does.
+        let body = r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":},{"time":"2026-09-03T14:29:52.930Z","type":"platform.runtimeDone","record":{"requestId":"abc123","status":"success"}}]"#;
+        assert!(matches!(
+            stitch(&FragmentBuffer::default(), body),
+            Stitch::Discarded
+        ));
+    }
+
     #[test]
     fn does_not_hold_a_payload_that_arrived_whole() {
         let fragments = FragmentBuffer::default();
@@ -309,11 +484,10 @@ mod tests {
 
         assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending));
 
-        // A different record's envelope, so the held fragment goes and this one takes its
-        // place.
+        // Another cut record, which doesn't parse joined on, so the held fragment goes with it.
         let other =
             r#"[{"time":"2026-09-03T14:30:11.001Z","type":"function","record":{"message":"CCCC}]"#;
-        assert!(matches!(stitch(&fragments, other), Stitch::Pending));
+        assert!(matches!(stitch(&fragments, other), Stitch::Discarded));
 
         // Proof the first fragment is gone: its own continuation no longer joins.
         assert!(matches!(stitch(&fragments, TAIL), Stitch::Discarded));

From 801454df271a4a49b1ac8751ca16f2261d009fb1 Mon Sep 17 00:00:00 2001
From: "shreya.malpani" <shreya.malpani@datadoghq.com>
Date: Thu, 8 Oct 2026 11:33:47 -0400
Subject: [PATCH 2/5] don't let a failed join drop the incoming telemetry
 payload

---
 bottlecap/src/extension/telemetry/stitch.rs | 93 +++++++++++++++------
 1 file changed, 68 insertions(+), 25 deletions(-)

diff --git a/bottlecap/src/extension/telemetry/stitch.rs b/bottlecap/src/extension/telemetry/stitch.rs
index cd49890a7..83ec96e31 100644
--- a/bottlecap/src/extension/telemetry/stitch.rs
+++ b/bottlecap/src/extension/telemetry/stitch.rs
@@ -8,7 +8,7 @@
 //! POST 1  [{"time":"..929Z","type":"function","record":{"message":"iVBORw0KGgoAAA}]
 //!         `------------- envelope --------------------'`--- cut here ---'`framing'
 //!
-//! POST 2  [{"time":"..929Z","type":"function","record":QICAgIfAhkiAAA"}},{..},{..runtimeDone..}]
+//! POST 2  [{"time":"..930Z","type":"function","record":QICAgIfAhkiAAA"}},{..},{..runtimeDone..}]
 //!         `------ the next piece's envelope ----------'`-- resumed --'`- rest of the batch -'
 //! ```
 //!
@@ -58,6 +58,9 @@ const PIECE_TYPES: &[&[u8]] = &[
     b"\"type\":\"extension\",\"record\":",
 ];
 
+/// Room for an envelope's timestamp and type, bounding the search for its `record` key.
+const MAX_ENVELOPE_BYTES: usize = 128;
+
 /// How far past a seam's start serde can fail: a cut inside a string swallows `},{"`, and
 /// the string closes on the `"` before `time`.
 const SEAM_REACH: usize = 4;
@@ -95,18 +98,24 @@ impl FragmentBuffer {
         }
 
         if let Some(head) = slot.take() {
+            let held = head.body.len();
             if let Some(resumed) = Fragment::resumed_bytes(body) {
                 let mut joined = head.join(resumed);
-                return match unsplit(&mut joined) {
-                    Ok(events) => Stitch::Complete(events),
+                match unsplit(&mut joined) {
+                    Ok(events) => return Stitch::Complete(events),
                     // A record can be cut more than once, so keep accumulating.
-                    Err(e) => hold_or_discard(&mut slot, joined, &e),
-                };
+                    Err(e) => {
+                        if matches!(hold_or_discard(&mut slot, joined, &e), Stitch::Pending) {
+                            return Stitch::Pending;
+                        }
+                    }
+                }
             }
 
+            // Nothing pairs the two but the join parsing, so a failed one means `body` stands
+            // alone.
             debug!(
-                "TELEMETRY API | Dropping {} held bytes, the next payload does not continue it",
-                head.body.len()
+                "TELEMETRY API | Dropping {held} held bytes, the next payload does not continue it"
             );
         }
 
@@ -164,9 +173,8 @@ impl Fragment {
     /// A continuation opens with the next piece's envelope, so its `record` key is the first
     /// one in the payload — anything the customer nested sits inside the value that follows.
     fn resumed_bytes(body: &[u8]) -> Option<&[u8]> {
-        let payload = body.strip_prefix(b"[")?;
-        let value_start = find(payload, RECORD_KEY)? + RECORD_KEY.len();
-        payload.get(value_start..)
+        let payload = body.strip_prefix(b"[{\"time\":\"")?;
+        payload.get(piece_envelope(payload)?..)
     }
 
     /// Joins the resumed bytes on, dropping the framing: left in place it would land inside
@@ -212,18 +220,21 @@ fn seam_at(body: &[u8], at: usize) -> Option<Range<usize>> {
         .rev()
         .find(|&i| body.get(i..).is_some_and(|rest| rest.starts_with(SEAM)))?;
 
+    let len = piece_envelope(body.get(start + SEAM.len()..)?)?;
+    Some(start..start + SEAM.len() + len)
+}
+
+/// Length of the rest of a piece's envelope, if `bytes` — following its `{"time":"` — is one.
+fn piece_envelope(bytes: &[u8]) -> Option<usize> {
+    let window = &bytes[..bytes.len().min(MAX_ENVELOPE_BYTES)];
+    let len = find(window, RECORD_KEY)? + RECORD_KEY.len();
+
     // The envelope holds only the timestamp and a log type; a brace means the match ran into a
     // value, and any other type is a whole event — say `platform.runtimeDone` — not a piece.
-    let envelope = body.get(start + SEAM.len()..)?;
-    let len = find(envelope, RECORD_KEY)? + RECORD_KEY.len();
-    let envelope = &envelope[..len];
-    if envelope.iter().any(|b| matches!(b, b'{' | b'}'))
-        || !PIECE_TYPES.iter().any(|tail| envelope.ends_with(tail))
-    {
-        return None;
-    }
-
-    Some(start..start + SEAM.len() + len)
+    let envelope = &window[..len];
+    let is_piece = !envelope.iter().any(|b| matches!(b, b'{' | b'}'))
+        && PIECE_TYPES.iter().any(|tail| envelope.ends_with(tail));
+    is_piece.then_some(len)
 }
 
 /// The byte serde's one-based line and column point at.
@@ -247,7 +258,7 @@ pub(crate) mod fixtures {
     pub(crate) const HEAD: &str =
         r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":{"message":"AAAA}]"#;
 
-    /// The continuation: the same envelope, the resumed bytes, then the rest of the batch.
+    /// The continuation: the next piece's envelope, the resumed bytes, then the rest of the batch.
     pub(crate) const TAIL: &str = r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":BBBB"}},{"time":"2026-09-03T14:29:52.930Z","type":"platform.runtimeDone","record":{"requestId":"abc123","status":"success","metrics":{"durationMs":18.074,"producedBytes":329814}}}]"#;
 }
 
@@ -484,12 +495,44 @@ mod tests {
 
         assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending));
 
-        // Another cut record, which doesn't parse joined on, so the held fragment goes with it.
+        // Another cut record, which doesn't parse joined on, so the held fragment goes and this
+        // one takes its place.
         let other =
             r#"[{"time":"2026-09-03T14:30:11.001Z","type":"function","record":{"message":"CCCC}]"#;
-        assert!(matches!(stitch(&fragments, other), Stitch::Discarded));
+        assert!(matches!(stitch(&fragments, other), Stitch::Pending));
 
-        // Proof the first fragment is gone: its own continuation no longer joins.
-        assert!(matches!(stitch(&fragments, TAIL), Stitch::Discarded));
+        // Proof the first fragment is gone: the continuation now resumes this one.
+        let events = completed(stitch(&fragments, TAIL));
+        assert_eq!(
+            events[0].record,
+            TelemetryRecord::Function(serde_json::json!({"message": "CCCCBBBB"}))
+        );
+    }
+
+    #[test]
+    fn recovers_a_payload_that_does_not_continue_the_held_fragment() {
+        let fragments = FragmentBuffer::default();
+
+        assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending));
+
+        // Whole once its own seam is dropped, so it must not go down with the held fragment.
+        let (first, rest) = RECORD.split_at(20);
+        let body = format!(
+            "[{},{},{RUNTIME_DONE}]",
+            piece("931", first),
+            piece("932", rest)
+        );
+        assert_whole_record(&completed(stitch(&fragments, &body)));
+    }
+
+    #[test]
+    fn does_not_resume_a_fragment_with_a_whole_event() {
+        let fragments = FragmentBuffer::default();
+
+        assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending));
+
+        // Only a log piece continues one, so this record isn't glued onto the held message.
+        let body = r#"[{"time":"2026-09-03T14:29:52.930Z","type":"platform.runtimeDone","record":BBBB"}}]"#;
+        assert!(matches!(stitch(&fragments, body), Stitch::Discarded));
     }
 }

From b92a519331fd8ee4057bde5809bf367912832b32 Mon Sep 17 00:00:00 2001
From: "shreya.malpani" <shreya.malpani@datadoghq.com>
Date: Tue, 13 Oct 2026 10:58:28 -0400
Subject: [PATCH 3/5] naming fixes

---
 bottlecap/src/extension/telemetry/stitch.rs | 24 +++++++++++----------
 1 file changed, 13 insertions(+), 11 deletions(-)

diff --git a/bottlecap/src/extension/telemetry/stitch.rs b/bottlecap/src/extension/telemetry/stitch.rs
index 83ec96e31..9fee667a6 100644
--- a/bottlecap/src/extension/telemetry/stitch.rs
+++ b/bottlecap/src/extension/telemetry/stitch.rs
@@ -15,9 +15,10 @@
 //! So the two are joined by dropping POST 1's framing and POST 2's envelope. Each piece carries
 //! its own timestamp, so nothing pairs them but the joined bytes parsing.
 //!
-//! Pieces also sit side by side within one POST, where the seam between them — the cut piece's
-//! closing `}` and the next piece's envelope — is dropped the same way. Serde fails at, or just
-//! after, a seam, which tells it apart from the boundary between two whole events.
+//! Pieces can also arrive next to each other in one POST. The seam between two such pieces is the
+//! cut piece's closing `}` followed by the next piece's envelope, and it is dropped the same way.
+//! Serde fails at a seam or a few bytes after it. This is how a seam is told apart from the
+//! boundary between two whole events.
 //!
 //! Recovering the batch matters beyond the log line itself. `platform.runtimeDone` lands in
 //! the second half, and the on-demand loop waits for it before calling `/next` — so dropping
@@ -98,7 +99,7 @@ impl FragmentBuffer {
         }
 
         if let Some(head) = slot.take() {
-            let held = head.body.len();
+            let held_len = head.body.len();
             if let Some(resumed) = Fragment::resumed_bytes(body) {
                 let mut joined = head.join(resumed);
                 match unsplit(&mut joined) {
@@ -115,7 +116,7 @@ impl FragmentBuffer {
             // Nothing pairs the two but the join parsing, so a failed one means `body` stands
             // alone.
             debug!(
-                "TELEMETRY API | Dropping {held} held bytes, the next payload does not continue it"
+                "TELEMETRY API | Dropping {held_len} held bytes, the next payload does not continue it"
             );
         }
 
@@ -152,13 +153,13 @@ impl Fragment {
     /// was cut short, running out of input or into the framing. Any other parse failure won't
     /// be fixed by joining, and holding such a body would poison the next stitch.
     fn from_cut_payload(body: Vec<u8>, error: &serde_json::Error) -> Option<Self> {
-        let cut_short = match error.classify() {
+        let is_cut_short = match error.classify() {
             Category::Eof => true,
             // Cut after a `\` or a whole value, the piece runs into the framing instead.
             Category::Syntax => offset(&body, error) + FRAMING.len() >= body.len(),
             _ => false,
         };
-        if !cut_short || body.len() > MAX_FRAGMENT_BYTES || body.first() != Some(&b'[') {
+        if !is_cut_short || body.len() > MAX_FRAGMENT_BYTES || body.first() != Some(&b'[') {
             return None;
         }
 
@@ -197,20 +198,20 @@ fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
 
 /// Parses `body`, dropping the seams between pieces one at a time where serde fails on them.
 fn unsplit(body: &mut Vec<u8>) -> Result<Vec<TelemetryEvent>, serde_json::Error> {
-    let mut seams = 0;
+    let mut seam_count = 0;
     loop {
         let error = match serde_json::from_slice(body) {
             Ok(events) => return Ok(events),
             Err(e) => e,
         };
-        if seams == MAX_SEAMS || error.classify() == Category::Data {
+        if seam_count == MAX_SEAMS || error.classify() == Category::Data {
             return Err(error);
         }
         match seam_at(body, offset(body, &error)) {
             Some(seam) => body.drain(seam),
             None => return Err(error),
         };
-        seams += 1;
+        seam_count += 1;
     }
 }
 
@@ -224,7 +225,8 @@ fn seam_at(body: &[u8], at: usize) -> Option<Range<usize>> {
     Some(start..start + SEAM.len() + len)
 }
 
-/// Length of the rest of a piece's envelope, if `bytes` — following its `{"time":"` — is one.
+/// Length of a piece's envelope, counted from just after its opening `{"time":"`. Returns
+/// `None` if `bytes` is not the rest of a piece's envelope.
 fn piece_envelope(bytes: &[u8]) -> Option<usize> {
     let window = &bytes[..bytes.len().min(MAX_ENVELOPE_BYTES)];
     let len = find(window, RECORD_KEY)? + RECORD_KEY.len();

From cfdab19d61cdb6f96d82d28e6a0849ddbc0708f8 Mon Sep 17 00:00:00 2001
From: "shreya.malpani" <shreya.malpani@datadoghq.com>
Date: Mon, 19 Oct 2026 16:34:50 -0400
Subject: [PATCH 4/5] fix: detect seams by the record before them, and handle
 events between pieces

---
 bottlecap/src/extension/telemetry/listener.rs |   4 +-
 bottlecap/src/extension/telemetry/stitch.rs   | 356 ++++++++++++------
 2 files changed, 236 insertions(+), 124 deletions(-)

diff --git a/bottlecap/src/extension/telemetry/listener.rs b/bottlecap/src/extension/telemetry/listener.rs
index 4810178c8..7e8884e05 100644
--- a/bottlecap/src/extension/telemetry/listener.rs
+++ b/bottlecap/src/extension/telemetry/listener.rs
@@ -124,8 +124,8 @@ impl TelemetryListener {
 
         let mut telemetry_events: Vec<TelemetryEvent> = match serde_json::from_slice(&body) {
             Ok(events) => events,
-            // The Telemetry API splits an oversized record across two POSTs, and neither half
-            // parses alone. See `stitch`.
+            // Lambda cuts a long log line into pieces, and a payload that holds a piece does not
+            // parse. See `stitch`.
             Err(e) => match state.fragments.stitch(&body) {
                 Stitch::Complete(events) => {
                     debug!(
diff --git a/bottlecap/src/extension/telemetry/stitch.rs b/bottlecap/src/extension/telemetry/stitch.rs
index 9fee667a6..0807ff5fc 100644
--- a/bottlecap/src/extension/telemetry/stitch.rs
+++ b/bottlecap/src/extension/telemetry/stitch.rs
@@ -1,4 +1,4 @@
-//! Reassembles telemetry payloads that the Telemetry API split across two POSTs.
+//! Reassembles telemetry payloads holding a log record that Lambda cut into pieces.
 //!
 //! A log record longer than about 256 KiB is cut mid-value into pieces, each an event of its
 //! own. When a payload ends on a cut piece, the rest of it — plus the rest of the batch —
@@ -17,16 +17,16 @@
 //!
 //! Pieces can also arrive next to each other in one POST. The seam between two such pieces is the
 //! cut piece's closing `}` followed by the next piece's envelope, and it is dropped the same way.
-//! Serde fails at a seam or a few bytes after it. This is how a seam is told apart from the
-//! boundary between two whole events.
+//! A seam has the same bytes as the boundary between two whole events. It is told apart by the
+//! record value before it, which was cut short and so does not parse alone.
 //!
 //! Recovering the batch matters beyond the log line itself. `platform.runtimeDone` lands in
 //! the second half, and the on-demand loop waits for it before calling `/next` — so dropping
 //! the batch holds the invocation open until Lambda times out the sandbox.
 
+use serde::de::IgnoredAny;
 use serde_json::error::Category;
 use std::{
-    ops::Range,
     sync::{Arc, Mutex, PoisonError},
     time::{Duration, Instant},
 };
@@ -50,8 +50,8 @@ const RECORD_KEY: &[u8] = b"\"record\":";
 /// record's value short, so a fragment ends with framing that belongs to neither half.
 const FRAMING: &[u8] = b"}]";
 
-/// Opens a seam between two pieces: the cut piece's closing `}`, then the next one's envelope.
-const SEAM: &[u8] = b"},{\"time\":\"";
+/// Opens every event's envelope.
+const EVENT_START: &[u8] = b"{\"time\":\"";
 
 /// How a piece's envelope ends: only log lines are long enough to be split.
 const PIECE_TYPES: &[&[u8]] = &[
@@ -62,11 +62,7 @@ const PIECE_TYPES: &[&[u8]] = &[
 /// Room for an envelope's timestamp and type, bounding the search for its `record` key.
 const MAX_ENVELOPE_BYTES: usize = 128;
 
-/// How far past a seam's start serde can fail: a cut inside a string swallows `},{"`, and
-/// the string closes on the `"` before `time`.
-const SEAM_REACH: usize = 4;
-
-/// Bounds the re-parsing of one payload: a 4 MiB payload of 256 KiB pieces has 16 seams.
+/// Bounds the re-parsing of one record: a 4 MiB payload of 256 KiB pieces has 16 seams.
 const MAX_SEAMS: usize = 64;
 
 /// What came of pairing an unparseable body with a held fragment.
@@ -101,45 +97,39 @@ impl FragmentBuffer {
         if let Some(head) = slot.take() {
             let held_len = head.body.len();
             if let Some(resumed) = Fragment::resumed_bytes(body) {
-                let mut joined = head.join(resumed);
-                match unsplit(&mut joined) {
-                    Ok(events) => return Stitch::Complete(events),
-                    // A record can be cut more than once, so keep accumulating.
-                    Err(e) => {
-                        if matches!(hold_or_discard(&mut slot, joined, &e), Stitch::Pending) {
-                            return Stitch::Pending;
-                        }
-                    }
+                // A record can be cut more than once, so keep accumulating.
+                match parse_or_hold(&mut slot, head.join(resumed)) {
+                    Stitch::Discarded => {}
+                    stitched => return stitched,
                 }
             }
 
-            // Nothing pairs the two but the join parsing, so a failed one means `body` stands
-            // alone.
             debug!(
                 "TELEMETRY API | Dropping {held_len} held bytes, the next payload does not continue it"
             );
         }
 
-        let mut body = body.to_vec();
-        match unsplit(&mut body) {
-            Ok(events) => Stitch::Complete(events),
-            Err(e) => hold_or_discard(&mut slot, body, &e),
-        }
+        parse_or_hold(&mut slot, body.to_vec())
     }
 }
 
-/// Holds `body` for its continuation, or reports that nothing can be recovered from it.
-fn hold_or_discard(
-    slot: &mut Option<Fragment>,
-    body: Vec<u8>,
-    error: &serde_json::Error,
-) -> Stitch {
-    *slot = Fragment::from_cut_payload(body, error);
-    if slot.is_some() {
-        Stitch::Pending
-    } else {
-        Stitch::Discarded
+/// Parses `body` once its seams are dropped, or holds it if it ends on a cut piece.
+fn parse_or_hold(slot: &mut Option<Fragment>, mut body: Vec<u8>) -> Stitch {
+    match serde_json::from_slice(&body) {
+        Ok(events) => return Stitch::Complete(events),
+        // Valid JSON our types don't accept arrived whole; no seam explains it.
+        Err(e) if e.classify() == Category::Data => return Stitch::Discarded,
+        Err(_) => {}
     }
+
+    if drop_seams(&mut body) && body.len() <= MAX_FRAGMENT_BYTES {
+        *slot = Some(Fragment {
+            body,
+            received: Instant::now(),
+        });
+        return Stitch::Pending;
+    }
+    serde_json::from_slice(&body).map_or(Stitch::Discarded, Stitch::Complete)
 }
 
 /// The leading half of a split payload.
@@ -149,33 +139,17 @@ struct Fragment {
 }
 
 impl Fragment {
-    /// A fragment, if `body` is the leading half of a split payload: an array whose last piece
-    /// was cut short, running out of input or into the framing. Any other parse failure won't
-    /// be fixed by joining, and holding such a body would poison the next stitch.
-    fn from_cut_payload(body: Vec<u8>, error: &serde_json::Error) -> Option<Self> {
-        let is_cut_short = match error.classify() {
-            Category::Eof => true,
-            // Cut after a `\` or a whole value, the piece runs into the framing instead.
-            Category::Syntax => offset(&body, error) + FRAMING.len() >= body.len(),
-            _ => false,
-        };
-        if !is_cut_short || body.len() > MAX_FRAGMENT_BYTES || body.first() != Some(&b'[') {
-            return None;
-        }
-
-        Some(Self {
-            body,
-            received: Instant::now(),
-        })
-    }
-
     /// The bytes that resume this fragment, if `body` is its continuation.
     ///
     /// A continuation opens with the next piece's envelope, so its `record` key is the first
     /// one in the payload — anything the customer nested sits inside the value that follows.
     fn resumed_bytes(body: &[u8]) -> Option<&[u8]> {
-        let payload = body.strip_prefix(b"[{\"time\":\"")?;
-        payload.get(piece_envelope(payload)?..)
+        let event = body.strip_prefix(b"[")?;
+        // Only a log piece continues a fragment; any other type is a whole event.
+        let (len, Some(_)) = envelope(event)? else {
+            return None;
+        };
+        event.get(len..)
     }
 
     /// Joins the resumed bytes on, dropping the framing: left in place it would land inside
@@ -196,47 +170,143 @@ fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
         .position(|window| window == needle)
 }
 
-/// Parses `body`, dropping the seams between pieces one at a time where serde fails on them.
-fn unsplit(body: &mut Vec<u8>) -> Result<Vec<TelemetryEvent>, serde_json::Error> {
+/// Joins the pieces of each cut record in `body`, and returns whether it ends on a cut piece to
+/// hold for the rest.
+///
+/// A cut record continues in the next event of its type. A whole event of another type in
+/// between, such as an extension log line, is moved ahead of it.
+fn drop_seams(body: &mut Vec<u8>) -> bool {
+    let Some(events) = split_events(body) else {
+        return false;
+    };
+
+    let mut joined = Vec::with_capacity(body.len());
+    // A record cut short, and its pieces joined so far.
+    let mut cut: Option<(Event, Vec<u8>)> = None;
     let mut seam_count = 0;
-    loop {
-        let error = match serde_json::from_slice(body) {
-            Ok(events) => return Ok(events),
-            Err(e) => e,
-        };
-        if seam_count == MAX_SEAMS || error.classify() == Category::Data {
-            return Err(error);
+    for event in events {
+        match &mut cut {
+            None => match parse_record(event.value) {
+                Record::Whole => push_event(&mut joined, event.envelope, event.value),
+                Record::Cut if event.kind.is_some() => cut = Some((event, event.value.to_vec())),
+                _ => return false,
+            },
+            Some((record, _)) if event.kind != record.kind => {
+                if !matches!(parse_record(event.value), Record::Whole) {
+                    return false;
+                }
+                push_event(&mut joined, event.envelope, event.value);
+            }
+            Some((record, value)) => {
+                if seam_count == MAX_SEAMS {
+                    return false;
+                }
+                seam_count += 1;
+                value.extend_from_slice(event.value);
+                match parse_record(value) {
+                    Record::Whole => {
+                        push_event(&mut joined, record.envelope, value);
+                        cut = None;
+                    }
+                    Record::Cut => {}
+                    Record::Broken => return false,
+                }
+            }
         }
-        match seam_at(body, offset(body, &error)) {
-            Some(seam) => body.drain(seam),
-            None => return Err(error),
+    }
+
+    let ends_cut = cut.is_some();
+    if let Some((record, value)) = cut {
+        push_event(&mut joined, record.envelope, &value);
+    }
+    joined.push(b']');
+    *body = joined;
+    ends_cut
+}
+
+/// A payload's event: its envelope through `"record":`, its record value, and its type if it is
+/// a log piece.
+#[derive(Clone, Copy)]
+struct Event<'a> {
+    envelope: &'a [u8],
+    value: &'a [u8],
+    kind: Option<&'static [u8]>,
+}
+
+/// Splits a payload into its events.
+fn split_events(body: &[u8]) -> Option<Vec<Event<'_>>> {
+    let mut rest = body.strip_prefix(b"[")?.strip_suffix(FRAMING)?;
+    let mut events = Vec::new();
+    loop {
+        let (len, kind) = envelope(rest)?;
+        let (envelope, tail) = rest.split_at(len);
+        let Some((value, next)) = split_at_next_event(tail) else {
+            events.push(Event {
+                envelope,
+                value: tail,
+                kind,
+            });
+            return Some(events);
         };
-        seam_count += 1;
+        events.push(Event {
+            envelope,
+            value,
+            kind,
+        });
+        rest = next;
     }
 }
 
-/// The seam serde failed on at byte `at`, if there is one.
-fn seam_at(body: &[u8], at: usize) -> Option<Range<usize>> {
-    let start = (at.saturating_sub(SEAM_REACH)..=at)
-        .rev()
-        .find(|&i| body.get(i..).is_some_and(|rest| rest.starts_with(SEAM)))?;
+/// Splits `bytes` at the first `},` that ends an event, which is one followed by an envelope.
+fn split_at_next_event(bytes: &[u8]) -> Option<(&[u8], &[u8])> {
+    (0..bytes.len()).find_map(|i| {
+        let next = bytes[i..].strip_prefix(b"},")?;
+        envelope(next).map(|_| (&bytes[..i], next))
+    })
+}
+
+/// Appends an event to a payload being rebuilt, opening the array before the first one.
+fn push_event(payload: &mut Vec<u8>, envelope: &[u8], value: &[u8]) {
+    payload.push(if payload.is_empty() { b'[' } else { b',' });
+    payload.extend_from_slice(envelope);
+    payload.extend_from_slice(value);
+    payload.push(b'}');
+}
 
-    let len = piece_envelope(body.get(start + SEAM.len()..)?)?;
-    Some(start..start + SEAM.len() + len)
+/// A record value taken on its own.
+enum Record {
+    Whole,
+    /// Valid JSON until it runs out, so the next piece can complete it.
+    Cut,
+    Broken,
 }
 
-/// Length of a piece's envelope, counted from just after its opening `{"time":"`. Returns
-/// `None` if `bytes` is not the rest of a piece's envelope.
-fn piece_envelope(bytes: &[u8]) -> Option<usize> {
-    let window = &bytes[..bytes.len().min(MAX_ENVELOPE_BYTES)];
-    let len = find(window, RECORD_KEY)? + RECORD_KEY.len();
-
-    // The envelope holds only the timestamp and a log type; a brace means the match ran into a
-    // value, and any other type is a whole event — say `platform.runtimeDone` — not a piece.
-    let envelope = &window[..len];
-    let is_piece = !envelope.iter().any(|b| matches!(b, b'{' | b'}'))
-        && PIECE_TYPES.iter().any(|tail| envelope.ends_with(tail));
-    is_piece.then_some(len)
+fn parse_record(value: &[u8]) -> Record {
+    match serde_json::from_slice::<IgnoredAny>(value) {
+        Ok(_) => Record::Whole,
+        // A number cut after its `-`, `.` or `e` is reported as invalid at its last byte rather
+        // than as running out.
+        Err(e) if e.is_eof() || offset(value, &e) + 1 >= value.len() => Record::Cut,
+        Err(_) => Record::Broken,
+    }
+}
+
+/// Length of an event's envelope, through `"record":`, and its type if it is a log piece.
+fn envelope(event: &[u8]) -> Option<(usize, Option<&'static [u8]>)> {
+    let window = &event[..event.len().min(MAX_ENVELOPE_BYTES)];
+    let fields = window.strip_prefix(EVENT_START)?;
+    let len = find(fields, RECORD_KEY)? + RECORD_KEY.len();
+
+    // The envelope holds only the timestamp and type; a brace means the match ran into a value.
+    let fields = &fields[..len];
+    if fields.iter().any(|b| matches!(b, b'{' | b'}')) {
+        return None;
+    }
+    let kind = PIECE_TYPES
+        .iter()
+        .copied()
+        .find(|tail| fields.ends_with(tail));
+    Some((EVENT_START.len() + len, kind))
 }
 
 /// The byte serde's one-based line and column point at.
@@ -320,8 +390,7 @@ mod tests {
     fn joins_a_payload_cut_after_whole_records() {
         let fragments = FragmentBuffer::default();
 
-        // The cut record is the last of several, so the envelope the continuation repeats is
-        // in the middle of the fragment.
+        // The cut record is the last of several.
         let head = format!(
             r#"[{{"time":"2026-09-03T14:29:52.900Z","type":"extension","record":"ready"}},{}"#,
             HEAD.trim_start_matches('[')
@@ -379,15 +448,19 @@ mod tests {
     /// cut can land in.
     const RECORD: &str = r#"{"level":"info","msg":"{\"a\":[1,true,null],\"b\":\"café caf\u00e9 \\\\ x\"}","n":-1.5e3}"#;
 
+    /// A record logged as JSON whose structure, not just its text, holds arrays of objects.
+    const STRUCTURED_RECORD: &str =
+        r#"{"items":[{"id":1,"tags":["a","b"]},{"id":23,"ok":true,"v":null}],"n":-5}"#;
+
     const RUNTIME_DONE: &str = r#"{"time":"2026-09-03T14:29:52.950Z","type":"platform.runtimeDone","record":{"requestId":"abc123","status":"success"}}"#;
 
-    /// A piece of `RECORD` as Lambda delivers it: raw, under an envelope of its own.
+    /// A piece of a record as Lambda delivers it: raw, under an envelope of its own.
     fn piece(millis: &str, bytes: &str) -> String {
         format!(r#"{{"time":"2026-09-03T14:29:52.{millis}Z","type":"function","record":{bytes}}}"#)
     }
 
-    fn assert_whole_record(events: &[TelemetryEvent]) {
-        let record = serde_json::from_str(RECORD).expect("valid record");
+    fn assert_whole_record(record: &str, events: &[TelemetryEvent]) {
+        let record = serde_json::from_str(record).expect("valid record");
         assert_eq!(events.len(), 2);
         assert_eq!(events[0].record, TelemetryRecord::Function(record));
         assert!(matches!(
@@ -396,27 +469,53 @@ mod tests {
         ));
     }
 
+    /// Each record above, cut in two at every char boundary.
+    fn cuts() -> impl Iterator<Item = (&'static str, &'static str, &'static str)> {
+        [RECORD, STRUCTURED_RECORD].into_iter().flat_map(|record| {
+            (1..record.len())
+                .filter(move |&cut| record.is_char_boundary(cut))
+                .map(move |cut| {
+                    let (first, rest) = record.split_at(cut);
+                    (record, first, rest)
+                })
+        })
+    }
+
     #[test]
     fn joins_pieces_within_a_payload_at_every_cut() {
-        for cut in (1..RECORD.len()).filter(|&cut| RECORD.is_char_boundary(cut)) {
-            let (first, rest) = RECORD.split_at(cut);
-            let body = format!(
-                "[{},{},{RUNTIME_DONE}]",
-                piece("929", first),
-                piece("930", rest)
-            );
+        // Another stream's event, such as the extension's own log line, can land between pieces.
+        let extension =
+            r#"{"time":"2026-09-03T14:29:52.931Z","type":"extension","record":"flushing"},"#;
+        for (record, first, rest) in cuts() {
+            for between in ["", extension] {
+                let body = format!(
+                    "[{},{between}{},{RUNTIME_DONE}]",
+                    piece("929", first),
+                    piece("930", rest)
+                );
+                // Known gap: a cut inside a number in an array of objects can leave the body
+                // valid JSON, so it never reaches the stitcher.
+                if serde_json::from_str::<Vec<TelemetryEvent>>(&body).is_ok() {
+                    continue;
+                }
 
-            let Stitch::Complete(events) = stitch(&FragmentBuffer::default(), &body) else {
-                panic!("pieces cut after {first:?} were not joined");
-            };
-            assert_whole_record(&events);
+                let Stitch::Complete(mut events) = stitch(&FragmentBuffer::default(), &body) else {
+                    panic!("pieces cut after {first:?} were not joined");
+                };
+                if !between.is_empty() {
+                    assert!(matches!(
+                        events.remove(0).record,
+                        TelemetryRecord::Extension(_)
+                    ));
+                }
+                assert_whole_record(record, &events);
+            }
         }
     }
 
     #[test]
     fn joins_pieces_across_payloads_at_every_cut() {
-        for cut in (1..RECORD.len()).filter(|&cut| RECORD.is_char_boundary(cut)) {
-            let (first, rest) = RECORD.split_at(cut);
+        for (record, first, rest) in cuts() {
             let fragments = FragmentBuffer::default();
 
             let head = format!("[{}]", piece("929", first));
@@ -429,7 +528,7 @@ mod tests {
             let Stitch::Complete(events) = stitch(&fragments, &tail) else {
                 panic!("payloads cut after {first:?} were not joined");
             };
-            assert_whole_record(&events);
+            assert_whole_record(record, &events);
         }
     }
 
@@ -458,7 +557,7 @@ mod tests {
             events.remove(0).record,
             TelemetryRecord::Extension(_)
         ));
-        assert_whole_record(&events);
+        assert_whole_record(RECORD, &events);
     }
 
     #[test]
@@ -468,13 +567,27 @@ mod tests {
             stitch(&FragmentBuffer::default(), body),
             Stitch::Discarded
         ));
+    }
+
+    /// `platform.runtimeDone` after a cut piece is moved ahead of it rather than glued into it.
+    #[test]
+    fn keeps_runtime_done_whole_after_a_cut_piece() {
+        let fragments = FragmentBuffer::default();
 
-        // Fails where `platform.runtimeDone` begins, which a log piece's seam never does.
-        let body = r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":},{"time":"2026-09-03T14:29:52.930Z","type":"platform.runtimeDone","record":{"requestId":"abc123","status":"success"}}]"#;
+        let head = format!("[{},{RUNTIME_DONE}]", piece("929", r#"{"message":"AA"#));
+        assert!(matches!(stitch(&fragments, &head), Stitch::Pending));
+
+        let tail = format!("[{}]", piece("930", r#"BB"}"#));
+        let events = completed(stitch(&fragments, &tail));
+        assert_eq!(events.len(), 2);
         assert!(matches!(
-            stitch(&FragmentBuffer::default(), body),
-            Stitch::Discarded
+            events[0].record,
+            TelemetryRecord::PlatformRuntimeDone { .. }
         ));
+        assert_eq!(
+            events[1].record,
+            TelemetryRecord::Function(serde_json::json!({"message": "AABB"}))
+        );
     }
 
     #[test]
@@ -497,16 +610,15 @@ mod tests {
 
         assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending));
 
-        // Another cut record, which doesn't parse joined on, so the held fragment goes and this
-        // one takes its place.
-        let other =
-            r#"[{"time":"2026-09-03T14:30:11.001Z","type":"function","record":{"message":"CCCC}]"#;
+        // Another cut record after a whole event, which doesn't parse joined on, so the held
+        // fragment goes and this one takes its place.
+        let other = r#"[{"time":"2026-09-03T14:30:11.000Z","type":"extension","record":"ready"},{"time":"2026-09-03T14:30:11.001Z","type":"function","record":{"message":"CCCC}]"#;
         assert!(matches!(stitch(&fragments, other), Stitch::Pending));
 
         // Proof the first fragment is gone: the continuation now resumes this one.
         let events = completed(stitch(&fragments, TAIL));
         assert_eq!(
-            events[0].record,
+            events[1].record,
             TelemetryRecord::Function(serde_json::json!({"message": "CCCCBBBB"}))
         );
     }
@@ -524,7 +636,7 @@ mod tests {
             piece("931", first),
             piece("932", rest)
         );
-        assert_whole_record(&completed(stitch(&fragments, &body)));
+        assert_whole_record(RECORD, &completed(stitch(&fragments, &body)));
     }
 
     #[test]

From 0e8f43ecd86205fa450c8a51be699fcdc546f32c Mon Sep 17 00:00:00 2001
From: "shreya.malpani" <shreya.malpani@datadoghq.com>
Date: Tue, 20 Oct 2026 12:03:16 -0400
Subject: [PATCH 5/5] fix: rejoin a split record whose payload parses anyway

---
 bottlecap/src/extension/telemetry/listener.rs | 37 ++++++++++++++--
 bottlecap/src/extension/telemetry/stitch.rs   | 42 +++++++++++++++++--
 2 files changed, 72 insertions(+), 7 deletions(-)

diff --git a/bottlecap/src/extension/telemetry/listener.rs b/bottlecap/src/extension/telemetry/listener.rs
index 7e8884e05..488e12c19 100644
--- a/bottlecap/src/extension/telemetry/listener.rs
+++ b/bottlecap/src/extension/telemetry/listener.rs
@@ -2,7 +2,7 @@ use crate::{
     event_bus,
     extension::telemetry::{
         events::TelemetryEvent,
-        stitch::{FragmentBuffer, Stitch},
+        stitch::{FragmentBuffer, Stitch, rejoin_split_that_parsed},
     },
     http::{extract_request_body, handler_not_found},
 };
@@ -122,8 +122,8 @@ impl TelemetryListener {
             }
         };
 
-        let mut telemetry_events: Vec<TelemetryEvent> = match serde_json::from_slice(&body) {
-            Ok(events) => events,
+        let mut telemetry_events = match serde_json::from_slice::<Vec<TelemetryEvent>>(&body) {
+            Ok(events) => rejoin_split_that_parsed(&body, events.len()).unwrap_or(events),
             // Lambda cuts a long log line into pieces, and a payload that holds a piece does not
             // parse. See `stitch`.
             Err(e) => match state.fragments.stitch(&body) {
@@ -185,6 +185,19 @@ mod tests {
             .expect("failed to build request")
     }
 
+    /// A record cut after a full 262,118-byte piece, inside the number `23`, so its payload still
+    /// parses: the next piece's envelope reads as an item of the array. Returns the payload and
+    /// the whole record.
+    fn split_that_parses() -> (String, String) {
+        let pad = "x".repeat(262_118 - r#"{"items":[{"pad":""},{"id":2"#.len());
+        let first = format!(r#"{{"items":[{{"pad":"{pad}"}},{{"id":2"#);
+        let rest = r#"3,"ok":true}],"n":5}"#;
+        let body = format!(
+            r#"[{{"time":"2026-09-03T14:29:52.929Z","type":"function","record":{first}}},{{"time":"2026-09-03T14:29:52.930Z","type":"function","record":{rest}}},{{"time":"2026-09-03T14:29:52.950Z","type":"platform.runtimeDone","record":{{"requestId":"abc123","status":"success"}}}}]"#
+        );
+        (body, format!("{first}{rest}"))
+    }
+
     #[tokio::test]
     #[allow(clippy::unwrap_used)]
     async fn test_handle() {
@@ -243,4 +256,22 @@ mod tests {
             TelemetryRecord::PlatformRuntimeDone { .. }
         ));
     }
+
+    /// A seam that happens to be valid JSON still has its pieces rejoined.
+    #[tokio::test]
+    async fn test_handle_split_payload_that_parses() {
+        let (body, record) = split_that_parses();
+        serde_json::from_str::<Vec<TelemetryEvent>>(&body).expect("the seam parses");
+
+        let (tx, mut rx) = tokio::sync::mpsc::channel(2);
+        let response =
+            TelemetryListener::handle(axum::extract::State(state(tx)), post(&body)).await;
+        assert_eq!(response.status(), StatusCode::OK);
+
+        let record = serde_json::from_str(&record).expect("valid record");
+        assert_eq!(
+            rx.try_recv().expect("function record").record,
+            TelemetryRecord::Function(record)
+        );
+    }
 }
diff --git a/bottlecap/src/extension/telemetry/stitch.rs b/bottlecap/src/extension/telemetry/stitch.rs
index 0807ff5fc..4a7a0f7d0 100644
--- a/bottlecap/src/extension/telemetry/stitch.rs
+++ b/bottlecap/src/extension/telemetry/stitch.rs
@@ -65,6 +65,10 @@ const MAX_ENVELOPE_BYTES: usize = 128;
 /// Bounds the re-parsing of one record: a 4 MiB payload of 256 KiB pieces has 16 seams.
 const MAX_SEAMS: usize = 64;
 
+/// Lambda cuts a long log line at 262,118 bytes (256 KiB less 26 bytes of per-event overhead), so
+/// a split record's first piece is at least this long, with a margin.
+const MIN_FIRST_PIECE_BYTES: usize = 255 * 1024;
+
 /// What came of pairing an unparseable body with a held fragment.
 #[derive(Debug)]
 pub(crate) enum Stitch {
@@ -122,7 +126,7 @@ fn parse_or_hold(slot: &mut Option<Fragment>, mut body: Vec<u8>) -> Stitch {
         Err(_) => {}
     }
 
-    if drop_seams(&mut body) && body.len() <= MAX_FRAGMENT_BYTES {
+    if drop_seams(&mut body, 0) && body.len() <= MAX_FRAGMENT_BYTES {
         *slot = Some(Fragment {
             body,
             received: Instant::now(),
@@ -132,6 +136,21 @@ fn parse_or_hold(slot: &mut Option<Fragment>, mut body: Vec<u8>) -> Stitch {
     serde_json::from_slice(&body).map_or(Stitch::Discarded, Stitch::Complete)
 }
 
+/// Rejoins a split record whose payload parsed anyway: a cut inside a number in an array of
+/// objects leaves valid JSON, with the next piece's envelope read as part of the record.
+pub(crate) fn rejoin_split_that_parsed(body: &[u8], parsed: usize) -> Option<Vec<TelemetryEvent>> {
+    if body.len() < MIN_FIRST_PIECE_BYTES || split_events(body)?.len() <= parsed {
+        return None;
+    }
+    let mut rebuilt = body.to_vec();
+    drop_seams(&mut rebuilt, MIN_FIRST_PIECE_BYTES);
+    // Dropping a seam shortens the payload; if it didn't, there was nothing to rejoin.
+    if rebuilt.len() == body.len() {
+        return None;
+    }
+    serde_json::from_slice(&rebuilt).ok()
+}
+
 /// The leading half of a split payload.
 struct Fragment {
     body: Vec<u8>,
@@ -174,8 +193,9 @@ fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
 /// hold for the rest.
 ///
 /// A cut record continues in the next event of its type. A whole event of another type in
-/// between, such as an extension log line, is moved ahead of it.
-fn drop_seams(body: &mut Vec<u8>) -> bool {
+/// between, such as an extension log line, is moved ahead of it. A record only counts as cut if
+/// its first piece is at least `min_first_piece` bytes.
+fn drop_seams(body: &mut Vec<u8>, min_first_piece: usize) -> bool {
     let Some(events) = split_events(body) else {
         return false;
     };
@@ -188,7 +208,9 @@ fn drop_seams(body: &mut Vec<u8>) -> bool {
         match &mut cut {
             None => match parse_record(event.value) {
                 Record::Whole => push_event(&mut joined, event.envelope, event.value),
-                Record::Cut if event.kind.is_some() => cut = Some((event, event.value.to_vec())),
+                Record::Cut if event.kind.is_some() && event.value.len() >= min_first_piece => {
+                    cut = Some((event, event.value.to_vec()));
+                }
                 _ => return false,
             },
             Some((record, _)) if event.kind != record.kind => {
@@ -590,6 +612,18 @@ mod tests {
         );
     }
 
+    /// A log line can hold telemetry-shaped JSON as data. In a payload that parses, that's only
+    /// taken for a split record after a full first piece.
+    #[test]
+    fn leaves_telemetry_shaped_data_alone_in_a_payload_that_parses() {
+        let filler = vec![piece("900", &format!(r#""{}""#, "y".repeat(1024))); 256].join(",");
+        let nested = r#"{"events":[{"time":"2026-09-03T14:29:52.900Z","type":"function","record":1},{"time":"2026-09-03T14:29:52.901Z","type":"function","record":2}]}"#;
+        let body = format!("[{filler},{},{RUNTIME_DONE}]", piece("929", nested));
+
+        let events: Vec<TelemetryEvent> = serde_json::from_str(&body).expect("valid payload");
+        assert!(rejoin_split_that_parsed(body.as_bytes(), events.len()).is_none());
+    }
+
     #[test]
     fn does_not_hold_a_payload_that_arrived_whole() {
         let fragments = FragmentBuffer::default();