diff --git a/crates/buzz-acp/TESTING.md b/crates/buzz-acp/TESTING.md index 528bac2138..00f3225aaf 100644 --- a/crates/buzz-acp/TESTING.md +++ b/crates/buzz-acp/TESTING.md @@ -59,3 +59,26 @@ Goose and uses its provider to invoke the native developer shell. Both operate only on temporary local repositories, verify commit/tag signatures and identity, check unrelated-remote credential scoping, and assert keyfile removal. They do not replace authenticated relay clone/push/readback testing. + +## Observer performance measurement + +An ignored local probe exercises the production publish queue and its JSON +fitter. The default workload is 200 events, each with 1 MiB across 64 string +leaves. It prints elapsed milliseconds and retained queue bytes/events; there +is no timing threshold or additional CI job. + +```sh +cargo test --release -p buzz-acp --lib observer::perf_tests::publish_queue_probe -- \ + --exact --ignored --nocapture + +# Under-budget control: +BUZZ_OBSERVER_PROBE_EVENTS=10000 BUZZ_OBSERVER_PROBE_BYTES=1024 \ + BUZZ_OBSERVER_PROBE_FIELDS=1 \ + cargo test --release -p buzz-acp --lib observer::perf_tests::publish_queue_probe -- \ + --exact --ignored --nocapture +``` + +Elapsed time includes cloning the template payload and queue ingestion; it +excludes template construction and later publication/encryption. Use release +builds, fresh processes, identical workloads, and repeated alternating runs +when comparing revisions. Queue bytes are serialized JSON, not heap or RSS. diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 026b2344a0..d01127dc15 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -660,12 +660,16 @@ impl AcpClient { /// Emit a semantic event to the local observer feed, if enabled. pub fn observe(&self, kind: impl Into, payload: serde_json::Value) { + self.observe_with(kind, || payload); + } + + fn observe_with(&self, kind: impl Into, payload: impl FnOnce() -> serde_json::Value) { if let Some(observer) = &self.observer { observer.emit( kind, self.observer_agent_index, &self.observer_context, - payload, + payload(), ); } } @@ -1157,7 +1161,7 @@ impl AcpClient { .await .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))? .map_err(AcpError::Io)?; - self.observe("acp_write", value.clone()); + self.observe_with("acp_write", || value.clone()); Ok(()) } @@ -1314,7 +1318,7 @@ impl AcpClient { continue; } }; - self.observe("acp_read", msg.clone()); + self.observe_with("acp_read", || msg.clone()); // Check if this is a response to our expected request (has matching id // AND no `method` field — a `method` field means it's an agent-initiated @@ -1637,7 +1641,7 @@ impl AcpClient { continue; } }; - self.observe("acp_read", msg.clone()); + self.observe_with("acp_read", || msg.clone()); let activity_now = Instant::now(); idle_deadline = activity_now + idle_timeout; @@ -3399,6 +3403,8 @@ mod tests { let mut client = spawn_script(r#"while IFS= read -r line; do printf '%s\n' "$line"; done"#).await; let mut stdin = client.stdin.take().expect("test stdin"); + // Raw wire payload copies must be lazy when observation is disabled. + client.observe_with("acp_read", || panic!("disabled observer copied payload")); let observer = ObserverHandle::in_process(); let mut events = observer.subscribe(); client.set_observer(Some(observer), 0); @@ -3441,6 +3447,10 @@ mod tests { ); if let Ok(event) = events.try_recv() { assert_eq!(event.kind, "acp_read"); + assert_eq!( + event.payload, + serde_json::from_str::(&frame).unwrap() + ); break; } assert!( diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index e9f2eb7dbc..b9aa0cba96 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1221,8 +1221,7 @@ impl ObserverPublishQueue { // Pre-trim at enqueue so (a) byte accounting reflects what will ship // and (b) one oversized leaf cannot force every frame it touches into // whole-envelope elision downstream. - fit_observer_event_to_budget(&mut event); - let bytes = serialized_len(&event); + let bytes = fit_observer_event_to_budget(&mut event); self.pending_bytes += bytes; self.events.push_back((bytes, source_events, event)); } @@ -1638,44 +1637,55 @@ const OBSERVER_LEAF_RETAIN_BYTES: usize = 3_000; /// 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` -/// IDs, and any elision markers exactly. No separate margin constant is needed. +/// length differ from raw). For small leaves, subtract the old serialized +/// size and add the replacement's size. JSON delimiters and the envelope stay +/// unchanged, so this accounting is exact. For a dominant leaf, recount the +/// fitted frame instead of traversing the large original leaf again. /// /// 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. +/// would STRICTLY shrink the serialized frame, then updates its size. +/// Shrinkability is re-evaluated against each leaf's CURRENT value: a shorter +/// removed-byte count can shorten an existing marker once more. Every pass +/// strictly decreases serialized length, so the loop terminates. When no leaf +/// can shrink the frame and it still overflows, replace the payload with a +/// small stub. Callers must still reject an oversized metadata envelope. /// -/// **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; +/// Returns the fitted serialized size so retention can account for bytes +/// without another traversal. Counting writes avoid allocating a second copy +/// of a potentially multi-megabyte source frame just to measure its size. +fn fit_observer_event_to_budget(event: &mut observer::ObserverEvent) -> usize { + let mut bytes = serialized_len(event); + if bytes <= OBSERVER_MAX_PLAINTEXT_LEN { + return bytes; } - // 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); + // Derive the original payload size from the measured frame and its small + // envelope, avoiding a second traversal of the oversized source payload. + // JSON null occupies four bytes in the temporary envelope. + let payload = event.payload.take(); + let envelope_bytes = serialized_len(event).saturating_sub(4); + event.payload = payload; + let original_payload_bytes = bytes.saturating_sub(envelope_bytes); - // 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. + // Each successful elision strictly shrinks the serialized frame, so the + // loop terminates even when a shortened marker permits another elision. while let Some(leaf) = largest_shrinkable_leaf(&mut event.payload) { - elide_leaf(leaf); - if serialized_len(event) <= OBSERVER_MAX_PLAINTEXT_LEN { - return; + // A JSON string uses at most six bytes per source byte (\u0000), + // plus quotes. Use deltas only when even that upper bound is small + // relative to the frame; otherwise recount the now-smaller frame. + let leaf_bound = leaf + .as_str() + .map_or(0, |s| s.len().saturating_mul(6).saturating_add(2)); + if leaf_bound < bytes / 2 { + let old_leaf_bytes = serialized_len(leaf); + elide_leaf(leaf); + bytes = bytes - old_leaf_bytes + serialized_len(leaf); + } else { + elide_leaf(leaf); + bytes = serialized_len(event); + } + if bytes <= OBSERVER_MAX_PLAINTEXT_LEN { + return bytes; } } @@ -1685,10 +1695,27 @@ fn fit_observer_event_to_budget(event: &mut observer::ObserverEvent) { "elided": format!("{} payload too large", event.kind), "originalBytes": original_payload_bytes, }); + serialized_len(event) } -fn serialized_len(event: &observer::ObserverEvent) -> usize { - serde_json::to_string(event).map(|s| s.len()).unwrap_or(0) +fn serialized_len(value: &impl serde::Serialize) -> usize { + let mut counter = ObserverByteCounter(0); + serde_json::to_writer(&mut counter, value) + .map(|()| counter.0) + .unwrap_or(usize::MAX) +} + +struct ObserverByteCounter(usize); + +impl std::io::Write for ObserverByteCounter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 = self.0.saturating_add(bytes.len()); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } } /// Find the longest string leaf that would STRICTLY shrink if elided, returning @@ -13566,6 +13593,38 @@ mod observer_payload_trim_tests { 20_000, "the smaller leaf is left untouched once the frame fits" ); + + // Many escaped leaves exercise incremental size accounting; enough + // shrink to fit while untouched leaves and section order survive. + for text in ["x", "\n\"\\", "🐝\u{0}"] { + let leaf = format!( + "{}{}{}", + "x".repeat(3_000), + text.repeat(4_000), + "y".repeat(3_000) + ); + let mut event = event_with_payload( + "acp_read", + serde_json::json!({"items": vec![leaf.clone(); 9], "small": "keep"}), + ); + let fitted_bytes = fit_observer_event_to_budget(&mut event); + assert_eq!(fitted_bytes, serialized(&event).len()); + assert!(fitted_bytes <= OBSERVER_MAX_PLAINTEXT_LEN); + let items = event.payload["items"] + .as_array() + .expect("trimmed sections survive"); + assert_eq!(items.len(), 9); + assert_eq!(event.payload["small"], "keep"); + let mut saw_untouched = false; + for item in items { + if item == &leaf { + saw_untouched = true; + } else { + assert!(!saw_untouched, "equal-size leaves elide in traversal order"); + assert!(item.as_str().unwrap().contains("…[elided")); + } + } + } } #[test] @@ -13617,6 +13676,19 @@ mod observer_payload_trim_tests { "fell back to the stub" ); assert!(event.payload.get("originalBytes").is_some()); + + // An escaped leaf just above its retained floor is small relative to + // a non-shrinkable numeric array. Delta accounting must include the + // replacement, or it would incorrectly accept this oversized frame. + let mut event = event_with_payload( + "acp_read", + serde_json::json!({"numbers": vec![0; 18_500], "edge": "\u{0}".repeat(6_100)}), + ); + let original_payload_bytes = serialized_len(&event.payload); + let fitted_bytes = fit_observer_event_to_budget(&mut event); + assert_eq!(fitted_bytes, serialized(&event).len()); + assert!(fitted_bytes <= OBSERVER_MAX_PLAINTEXT_LEN); + assert_eq!(event.payload["originalBytes"], original_payload_bytes); } #[test] diff --git a/crates/buzz-acp/src/observer.rs b/crates/buzz-acp/src/observer.rs index 7029e5af6d..1861c55e13 100644 --- a/crates/buzz-acp/src/observer.rs +++ b/crates/buzz-acp/src/observer.rs @@ -15,6 +15,9 @@ use std::{ use serde::Serialize; use tokio::sync::broadcast; +#[cfg(test)] +mod perf_tests; + const OBSERVER_BUFFER_CAP: usize = 1_000; /// Best-effort metadata attached to observer events. @@ -38,7 +41,7 @@ pub struct ObserverHandle { struct ObserverInner { tx: broadcast::Sender, - buffer: Mutex>, + buffer: Mutex>>, seq: AtomicU64, } @@ -91,13 +94,15 @@ impl ObserverHandle { /// Return the current replay buffer. pub fn snapshot(&self) -> Vec { - match self.inner.buffer.lock() { + // Pin the replay window under the lock, then copy JSON after releasing it. + let events: Vec<_> = match self.inner.buffer.lock() { Ok(buffer) => buffer.iter().cloned().collect(), Err(error) => { tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); Vec::new() } - } + }; + events.into_iter().map(|event| (*event).clone()).collect() } /// Emit a local observer event. @@ -108,9 +113,9 @@ impl ObserverHandle { context: &ObserverContext, payload: serde_json::Value, ) { - let event = ObserverEvent { - seq: self.inner.seq.fetch_add(1, Ordering::Relaxed), - timestamp: chrono::Utc::now().to_rfc3339(), + let mut event = ObserverEvent { + seq: 0, + timestamp: String::new(), kind: kind.into(), agent_index, channel_id: context.channel_id.clone(), @@ -120,17 +125,25 @@ impl ObserverHandle { payload, }; - match self.inner.buffer.lock() { - Ok(mut buffer) => { - if buffer.len() >= OBSERVER_BUFFER_CAP { - buffer.pop_front(); - } - buffer.push_back(event.clone()); - } + // Copy the payload before locking so other emitters can proceed. + let mut replay = event.clone(); + let mut buffer = match self.inner.buffer.lock() { + Ok(buffer) => buffer, Err(error) => { tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); + return; } + }; + // Assign both fields with publication: desktop sorts by timestamp, + // and the publisher deduplicates snapshots by sequence high-water mark. + event.seq = self.inner.seq.fetch_add(1, Ordering::Relaxed); + event.timestamp = chrono::Utc::now().to_rfc3339(); + replay.seq = event.seq; + replay.timestamp.clone_from(&event.timestamp); + if buffer.len() >= OBSERVER_BUFFER_CAP { + buffer.pop_front(); } + buffer.push_back(Arc::new(replay)); let _ = self.inner.tx.send(event); } diff --git a/crates/buzz-acp/src/observer/perf_tests.rs b/crates/buzz-acp/src/observer/perf_tests.rs new file mode 100644 index 0000000000..1f5e44228d --- /dev/null +++ b/crates/buzz-acp/src/observer/perf_tests.rs @@ -0,0 +1,134 @@ +use super::*; +use serde_json::json; +use tokio::sync::broadcast::error::TryRecvError; + +struct SlowKind { + started: std::sync::mpsc::Sender<()>, + resume: std::sync::mpsc::Receiver<()>, +} + +impl From for String { + fn from(kind: SlowKind) -> Self { + kind.started.send(()).unwrap(); + kind.resume.recv().unwrap(); + "slow".into() + } +} + +#[test] +fn concurrent_emitters_keep_live_and_replay_sequence_order() { + let observer = ObserverHandle::in_process(); + let mut live = observer.subscribe(); + let context = context_for_turn( + None, + Some("session".into()), + "turn".into(), + "2026-11-01T00:00:00Z".into(), + ); + let payload = json!({"text":"hello 🐝\n\"world\"", "body":"x".repeat(70_000)}); + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let (resume_tx, resume_rx) = std::sync::mpsc::channel(); + let (observer_ref, context_ref, payload_ref) = (&observer, &context, &payload); + let (mut handoff_live, mut handoff_snapshot) = std::thread::scope(move |scope| { + scope.spawn(move || { + observer_ref.emit( + SlowKind { + started: started_tx, + resume: resume_rx, + }, + Some(0), + context_ref, + payload_ref.clone(), + ) + }); + // Pause one real emit during preparation, then let another publish. + // No timing assertion: the channel controls this interleaving exactly. + started_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .unwrap(); + observer_ref.emit("fast", Some(1), context_ref, payload_ref.clone()); + let handoff_live = observer_ref.subscribe(); + let snapshot = observer_ref.snapshot(); + resume_tx.send(()).unwrap(); + (handoff_live, snapshot) + }); + let snapshot = observer.snapshot(); + assert_eq!(snapshot.len(), 2); + for (index, kind) in ["fast", "slow"].into_iter().enumerate() { + let event = live.try_recv().unwrap(); + assert_eq!(event.seq, index as u64 + 1); + assert_eq!(event.kind, kind); + assert!(chrono::DateTime::parse_from_rfc3339(&event.timestamp).is_ok()); + assert_eq!(event.payload, payload); + assert_eq!(event.session_id, context.session_id); + assert_eq!(event.turn_id, context.turn_id); + assert_eq!(event.started_at, context.started_at); + assert_eq!( + serde_json::to_value(&event).unwrap(), + serde_json::to_value(&snapshot[index]).unwrap() + ); + } + assert!(matches!(live.try_recv(), Err(TryRecvError::Empty))); + let high_water = handoff_snapshot.last().map_or(0, |event| event.seq); + while let Ok(event) = handoff_live.try_recv() { + if event.seq > high_water { + handoff_snapshot.push(event); + } + } + assert_eq!( + serde_json::to_value(&handoff_snapshot).unwrap(), + serde_json::to_value(&snapshot).unwrap(), + "snapshot/live handoff must neither skip nor duplicate the slow event" + ); + for n in 0..1_010 { + observer.emit("marker", None, &context, json!({"n":n})); + } + let snapshot = observer.snapshot(); + assert_eq!(snapshot.len(), 1_000); + assert_eq!(snapshot.first().unwrap().payload["n"], 10); + assert_eq!(snapshot.last().unwrap().payload["n"], 1_009); +} + +/// Local measurement only: the production publish queue, with no timing gate. +#[test] +#[ignore = "manual observer fitting measurement"] +fn publish_queue_probe() { + let count: usize = std::env::var("BUZZ_OBSERVER_PROBE_EVENTS") + .unwrap_or_else(|_| "200".into()) + .parse() + .unwrap(); + let payload_bytes: usize = std::env::var("BUZZ_OBSERVER_PROBE_BYTES") + .unwrap_or_else(|_| "1048576".into()) + .parse() + .unwrap(); + let fields: usize = std::env::var("BUZZ_OBSERVER_PROBE_FIELDS") + .unwrap_or_else(|_| "64".into()) + .parse() + .unwrap(); + assert!(fields > 0); + let mut template = ObserverEvent { + seq: 0, + timestamp: "2026-11-01T00:00:00Z".into(), + kind: "acp_read".into(), + agent_index: None, + channel_id: None, + session_id: None, + turn_id: None, + started_at: None, + payload: json!({"body": (0..fields).map(|n| + "x".repeat(payload_bytes / fields + usize::from(n < payload_bytes % fields)) + ).collect::>() }), + }; + let mut queue = crate::ObserverPublishQueue::default(); + let start = std::time::Instant::now(); + for n in 0..count { + template.seq = n as u64 + 1; + queue.ingest(template.clone()); + } + let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0; + println!( + "{}", + json!({"events": count, "payloadBytes": payload_bytes, "fields": fields, + "elapsedMs": elapsed_ms, "queuedEvents": queue.events.len(), "queuedBytes": queue.pending_bytes}) + ); +}