fix(buzz-acp): trim oversized observer frames to fit instead of dropping (#1072)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>

crates/buzz-acp/src/lib.rs

@@ -21,6 +21,7 @@ use buzz_core::kind::{
 };
 use buzz_core::observer::{
     decrypt_observer_payload, encrypt_observer_payload, OBSERVER_FRAME_TELEMETRY,
+    OBSERVER_MAX_PLAINTEXT_LEN,
 };
 use clap::Parser;
 use config::{Config, DedupMode, ModelsArgs, MultipleEventHandling, RespondTo, SubscribeMode};
@@ -355,7 +356,10 @@ struct ObserverChunkKey {
 }
 
 /// Flush coalesced chunks before they exceed the NIP-44 plaintext limit (65,535 bytes).
-/// Leave headroom for the JSON envelope wrapping the text.
+/// Leave headroom for the JSON envelope wrapping the text. This is a SOFT pre-flush
+/// of raw text below the hard cap; `fit_observer_event_to_budget` (the final ceiling,
+/// keyed to `OBSERVER_MAX_PLAINTEXT_LEN` in buzz-core/observer.rs:25) is what actually
+/// guarantees the serialized frame fits. Edit one of these two and review the other.
 const OBSERVER_CHUNK_MAX_TEXT_BYTES: usize = 60_000;
 
 impl ObserverChunkCoalescer {
@@ -440,14 +444,179 @@ fn set_observer_chunk_text(payload: &mut serde_json::Value, text: String) {
     }
 }
 
+/// Bytes of head and tail to retain from an elided string leaf — the value
+/// shown to the renderer at each end. The ONLY tuning knob here: large enough
+/// that a clipped diff/tool-result still shows real content, small enough that
+/// eliding actually shrinks the frame.
+const OBSERVER_LEAF_RETAIN_BYTES: usize = 3_000;
+
+/// Trim an oversized observer telemetry frame so its SERIALIZED form fits under
+/// `OBSERVER_MAX_PLAINTEXT_LEN`, instead of dropping the whole frame (silent
+/// telemetry loss). The common case — a frame already under budget — is left
+/// byte-identical.
+///
+/// The cap is measured in SERIALIZED bytes (JSON escaping makes serialized
+/// length differ from raw), so the stop condition is always a full reserialize
+/// of the whole frame: that counts the envelope, the variable `Option<String>`
+/// IDs, and any elision markers exactly. No separate margin constant is needed.
+///
+/// Termination is provable: each iteration elides the largest string leaf that
+/// would STRICTLY shrink the serialized frame, then reserializes. Shrinkability
+/// is re-evaluated against each leaf's CURRENT value, so a leaf already at its
+/// retained floor can never be re-elided — the loop strictly decreases the
+/// serialized length each pass and is bounded by the leaf count. When no leaf
+/// can shrink the frame and it still overflows, the payload is replaced with a
+/// tiny stub, which trivially fits. Monotone decrease, bounded below by the stub.
+///
+/// **Signature choice (`&mut`, double-serialize accepted):** on the common
+/// under-budget path this serializes the frame once to decide it fits, then
+/// `encrypt_observer_payload` serializes it again — one extra `to_string` of an
+/// already-small frame. Reusing that string would mean changing buzz-core's
+/// `encrypt_observer_payload` signature or adding a parallel encrypt path; both
+/// are out of this change's scope (buzz-core stays untouched). The clean `&mut`
+/// signature with one cheap redundant serialize is the deliberate tradeoff.
+fn fit_observer_event_to_budget(event: &mut observer::ObserverEvent) {
+    if serialized_len(event) <= OBSERVER_MAX_PLAINTEXT_LEN {
+        return;
+    }
+
+    // Raw size of the payload we are about to trim, captured before mutation so
+    // the stub's `originalBytes` reports source bytes discarded, not serialized
+    // overflow — consistent with the per-leaf marker's raw byte count.
+    let original_payload_bytes = serde_json::to_string(&event.payload)
+        .map(|s| s.len())
+        .unwrap_or(0);
+
+    // Elide the largest shrinkable leaf, reserialize, repeat. Each successful
+    // elision strictly shrinks the serialized frame, and a floored leaf can
+    // never be re-elided, so the loop is bounded by the leaf count.
+    while let Some(leaf) = largest_shrinkable_leaf(&mut event.payload) {
+        elide_leaf(leaf);
+        if serialized_len(event) <= OBSERVER_MAX_PLAINTEXT_LEN {
+            return;
+        }
+    }
+
+    // No leaf can shrink the frame further and it still overflows: replace the
+    // whole payload with a stub that is trivially under-cap.
+    event.payload = serde_json::json!({
+        "elided": format!("{} payload too large", event.kind),
+        "originalBytes": original_payload_bytes,
+    });
+}
+
+fn serialized_len(event: &observer::ObserverEvent) -> usize {
+    serde_json::to_string(event).map(|s| s.len()).unwrap_or(0)
+}
+
+/// Find the longest string leaf that would STRICTLY shrink if elided, returning
+/// a mutable handle to it. A leaf shrinks only if `head + marker + tail` is
+/// shorter than its current value (the marker-pushback guard); a leaf already at
+/// its retained floor fails this test and is skipped, which is what bounds the
+/// loop. Returns `None` when no leaf can shrink.
+fn largest_shrinkable_leaf(value: &mut serde_json::Value) -> Option<&mut serde_json::Value> {
+    // First pass: find the byte length of the best candidate without holding a
+    // borrow, then re-descend to return the matching mutable reference. Two
+    // immutable-style passes keep the borrow checker happy without unsafe.
+    let best_len = max_shrinkable_len(value)?;
+    find_leaf_with_len(value, best_len)
+}
+
+/// Largest current length among string leaves that can strictly shrink.
+fn max_shrinkable_len(value: &serde_json::Value) -> Option<usize> {
+    match value {
+        serde_json::Value::String(s) if leaf_shrinks(s) => Some(s.len()),
+        serde_json::Value::String(_) => None,
+        serde_json::Value::Array(items) => items.iter().filter_map(max_shrinkable_len).max(),
+        serde_json::Value::Object(map) => map.values().filter_map(max_shrinkable_len).max(),
+        _ => None,
+    }
+}
+
+/// Return the first string leaf whose current length equals `target` and that
+/// can strictly shrink. Used after `max_shrinkable_len` to re-acquire a mutable
+/// borrow of the chosen leaf.
+fn find_leaf_with_len(
+    value: &mut serde_json::Value,
+    target: usize,
+) -> Option<&mut serde_json::Value> {
+    match value {
+        serde_json::Value::String(s) if s.len() == target && leaf_shrinks(s) => Some(value),
+        serde_json::Value::Array(items) => items
+            .iter_mut()
+            .find_map(|item| find_leaf_with_len(item, target)),
+        serde_json::Value::Object(map) => map
+            .values_mut()
+            .find_map(|item| find_leaf_with_len(item, target)),
+        _ => None,
+    }
+}
+
+/// True when eliding `s` to head + marker + tail yields a strictly shorter raw
+/// string. The marker width grows with `N` (bytes removed), so a leaf only
+/// marginally larger than the retained ends must NOT be touched.
+fn leaf_shrinks(s: &str) -> bool {
+    let (head_end, tail_start) = elision_boundaries(s);
+    tail_start > head_end && {
+        let removed = tail_start - head_end;
+        let marker = elision_marker(removed);
+        head_end + marker.len() + (s.len() - tail_start) < s.len()
+    }
+}
+
+/// Replace the middle of a string leaf with `…[elided N bytes]…`, keeping a head
+/// and tail slice on UTF-8 char boundaries. `N` is RAW bytes removed.
+fn elide_leaf(leaf: &mut serde_json::Value) {
+    let serde_json::Value::String(s) = leaf else {
+        return;
+    };
+    let (head_end, tail_start) = elision_boundaries(s);
+    let removed = tail_start - head_end;
+    let mut elided = String::with_capacity(head_end + 32 + (s.len() - tail_start));
+    elided.push_str(&s[..head_end]);
+    elided.push_str(&elision_marker(removed));
+    elided.push_str(&s[tail_start..]);
+    *s = elided;
+}
+
+fn elision_marker(removed_bytes: usize) -> String {
+    format!("…[elided {removed_bytes} bytes]…")
+}
+
+/// Byte offsets bounding the elided middle, snapped to char boundaries so we
+/// never split a multi-byte char. Returns `(head_end, tail_start)` with
+/// `head_end <= tail_start`.
+fn elision_boundaries(s: &str) -> (usize, usize) {
+    let head_end = floor_char_boundary(s, OBSERVER_LEAF_RETAIN_BYTES.min(s.len()));
+    let tail_start = ceil_char_boundary(s, s.len().saturating_sub(OBSERVER_LEAF_RETAIN_BYTES));
+    (head_end, tail_start.max(head_end))
+}
+
+fn floor_char_boundary(s: &str, mut i: usize) -> usize {
+    while i > 0 && !s.is_char_boundary(i) {
+        i -= 1;
+    }
+    i
+}
+
+fn ceil_char_boundary(s: &str, mut i: usize) -> usize {
+    while i < s.len() && !s.is_char_boundary(i) {
+        i += 1;
+    }
+    i
+}
+
 async fn publish_relay_observer_event(
     publisher: &RelayEventPublisher,
     keys: &nostr::Keys,
     agent_pubkey_hex: &str,
     owner_pubkey_hex: &str,
     owner_pubkey: &PublicKey,
-    event: observer::ObserverEvent,
+    mut event: observer::ObserverEvent,
 ) {
+    // Trim oversized frames to fit the plaintext cap rather than letting
+    // encrypt_observer_payload reject and drop them whole (silent telemetry loss).
+    fit_observer_event_to_budget(&mut event);
     let encrypted = match encrypt_observer_payload(keys, owner_pubkey, &event) {
         Ok(encrypted) => encrypted,
         Err(error) => {
@@ -3289,3 +3458,180 @@ mod error_outcome_emission_tests {
         assert_eq!(turn_errors_emitted_for(PromptOutcome::Error(app)).await, 1);
     }
 }
+
+#[cfg(test)]
+mod observer_payload_trim_tests {
+    use super::*;
+
+    fn event_with_payload(kind: &str, payload: serde_json::Value) -> observer::ObserverEvent {
+        observer::ObserverEvent {
+            seq: 1,
+            timestamp: "2026-06-16T00:00:00Z".to_string(),
+            kind: kind.to_string(),
+            agent_index: Some(0),
+            channel_id: Some("11111111-1111-1111-1111-111111111111".to_string()),
+            session_id: Some("sess-1".to_string()),
+            turn_id: Some("turn-1".to_string()),
+            payload,
+        }
+    }
+
+    fn serialized(event: &observer::ObserverEvent) -> String {
+        serde_json::to_string(event).unwrap()
+    }
+
+    #[test]
+    fn test_under_budget_frame_passes_through_byte_identical() {
+        let mut event = event_with_payload("acp_read", serde_json::json!({ "body": "small" }));
+        let before = serialized(&event);
+        fit_observer_event_to_budget(&mut event);
+        assert_eq!(
+            serialized(&event),
+            before,
+            "under-budget frame must not be mutated"
+        );
+    }
+
+    #[test]
+    fn test_single_giant_leaf_is_elided_to_fit_with_envelope_intact() {
+        let big = "x".repeat(100_000);
+        let mut event = event_with_payload("acp_read", serde_json::json!({ "body": big }));
+        fit_observer_event_to_budget(&mut event);
+
+        assert!(
+            serialized(&event).len() <= OBSERVER_MAX_PLAINTEXT_LEN,
+            "frame must fit after trimming"
+        );
+        // Envelope intact.
+        assert_eq!(event.kind, "acp_read");
+        assert_eq!(event.turn_id.as_deref(), Some("turn-1"));
+        assert_eq!(
+            event.channel_id.as_deref(),
+            Some("11111111-1111-1111-1111-111111111111")
+        );
+        assert_eq!(event.seq, 1);
+
+        let leaf = event.payload["body"].as_str().unwrap();
+        assert!(
+            leaf.starts_with(&"x".repeat(OBSERVER_LEAF_RETAIN_BYTES)),
+            "head retained"
+        );
+        assert!(
+            leaf.ends_with(&"x".repeat(OBSERVER_LEAF_RETAIN_BYTES)),
+            "tail retained"
+        );
+        // N in the marker is RAW bytes removed: original len minus retained len.
+        let removed = 100_000 - leaf.chars().filter(|c| *c == 'x').count();
+        assert!(
+            leaf.contains(&format!("…[elided {removed} bytes]…")),
+            "marker reports raw bytes removed"
+        );
+    }
+
+    #[test]
+    fn test_multi_leaf_elides_largest_shrinkable_first_and_stops_when_it_fits() {
+        // One leaf alone over the cap; a second smaller-but-still-large leaf.
+        // Eliding the biggest should suffice, leaving the smaller intact.
+        let mut event = event_with_payload(
+            "acp_write",
+            serde_json::json!({
+                "huge": "a".repeat(90_000),
+                "medium": "b".repeat(20_000),
+            }),
+        );
+        fit_observer_event_to_budget(&mut event);
+
+        assert!(serialized(&event).len() <= OBSERVER_MAX_PLAINTEXT_LEN);
+        assert!(
+            event.payload["huge"].as_str().unwrap().contains("…[elided"),
+            "the largest leaf is elided"
+        );
+        assert_eq!(
+            event.payload["medium"].as_str().unwrap().len(),
+            20_000,
+            "the smaller leaf is left untouched once the frame fits"
+        );
+    }
+
+    #[test]
+    fn test_coalesced_chunk_nested_leaf_is_reached_by_recursive_walk() {
+        // The coalesced-chunk big leaf lives at params.update.content.text,
+        // not a top-level field — the walk must recurse to reach it.
+        let big = "z".repeat(80_000);
+        let mut event = event_with_payload(
+            "session_update",
+            serde_json::json!({
+                "params": {
+                    "update": {
+                        "sessionUpdate": "agent_message_chunk",
+                        "content": { "text": big }
+                    }
+                }
+            }),
+        );
+        fit_observer_event_to_budget(&mut event);
+
+        assert!(serialized(&event).len() <= OBSERVER_MAX_PLAINTEXT_LEN);
+        let text = event.payload["params"]["update"]["content"]["text"]
+            .as_str()
+            .unwrap();
+        assert!(text.contains("…[elided"), "nested leaf was elided");
+    }
+
+    #[test]
+    fn test_many_medium_leaves_terminate_via_stub() {
+        // Many leaves each too small to shrink on their own (below 2x retain),
+        // collectively over the cap. No leaf can strictly shrink, so the trimmer
+        // must terminate via the stub rather than loop forever.
+        let leaf = "m".repeat(OBSERVER_LEAF_RETAIN_BYTES); // shorter than head+tail → cannot shrink
+        let items: Vec<serde_json::Value> = (0..40)
+            .map(|_| serde_json::Value::String(leaf.clone()))
+            .collect();
+        let mut event = event_with_payload("acp_read", serde_json::json!({ "items": items }));
+        assert!(
+            serialized(&event).len() > OBSERVER_MAX_PLAINTEXT_LEN,
+            "precondition: frame is over the cap"
+        );
+
+        fit_observer_event_to_budget(&mut event);
+
+        assert!(serialized(&event).len() <= OBSERVER_MAX_PLAINTEXT_LEN);
+        assert_eq!(
+            event.payload["elided"].as_str().unwrap(),
+            "acp_read payload too large",
+            "fell back to the stub"
+        );
+        assert!(event.payload.get("originalBytes").is_some());
+    }
+
+    #[test]
+    fn test_leaf_too_small_to_shrink_is_not_mutated() {
+        // A frame already under budget whose only leaf is below the shrink floor:
+        // nothing should change. (Under-budget short-circuits, and even if forced,
+        // leaf_shrinks would reject it.)
+        let short = "s".repeat(OBSERVER_LEAF_RETAIN_BYTES); // == head; cannot strictly shrink
+        assert!(
+            !leaf_shrinks(&short),
+            "a leaf at the retain floor must not shrink"
+        );
+        let longer = "L".repeat(OBSERVER_LEAF_RETAIN_BYTES * 2 + 100);
+        assert!(leaf_shrinks(&longer), "a clearly larger leaf must shrink");
+    }
+
+    #[test]
+    fn test_utf8_multibyte_leaf_elides_on_char_boundary() {
+        // A leaf of 3-byte chars (… = U+2026) — eliding must land on char
+        // boundaries and never panic or produce invalid UTF-8.
+        let big: String = "…".repeat(40_000); // 120_000 bytes
+        let mut event = event_with_payload("acp_read", serde_json::json!({ "body": big }));
+        fit_observer_event_to_budget(&mut event);
+
+        assert!(serialized(&event).len() <= OBSERVER_MAX_PLAINTEXT_LEN);
+        let leaf = event.payload["body"].as_str().unwrap();
+        // Valid UTF-8 by construction (it's a &str); confirm head/tail are whole
+        // multi-byte chars and the marker is present.
+        assert!(leaf.starts_with('…'));
+        assert!(leaf.ends_with('…'));
+        assert!(leaf.contains("[elided"));
+    }
+}