From 1275556c12fd3c55c3b88624a16875abc4c53b8c Mon Sep 17 00:00:00 2001 From: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 12:27:44 -0400 Subject: [PATCH 1/6] fix(acp): pace observer telemetry at 1/s with per-channel batch envelopes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The observer publish pacer (167ms spacing + 90/min rolling cap) still let a busy session bill up to 6 relay events per second against the owner's quota. Replace it with a 1-second publish tick: everything the session emits between ticks ships in a single kind:"batch" envelope per channel, so telemetry costs at most one event per channel per second and is never silently dropped. - ObserverBatcher wraps ObserverChunkCoalescer; batch_observer_events groups strictly per channel_id (the desktop archive indexes by top-level channelId, so batches never mix channels) and splits over OBSERVER_MAX_PLAINTEXT_LEN. Singletons ship unwrapped — the wire shape for a quiet session is unchanged. - Per-event fit_observer_event_to_budget runs before batching so one oversized leaf can't force elision of the whole envelope. - Envelope metadata mirrors the last inner event; final drain on Closed so shutdown never strands buffered frames. - Desktop: unwrapObserverBatch expands envelopes on both the live relay path and the archive-ingest seam; a malformed envelope degrades to itself. - Rolling-minute cap removed as unreachable under the 1/s tick; updated the GATED_OBSERVER_QUEUE_CAP rationale accordingly. Verified: cargo test -p buzz-acp 673+9 pass, desktop node tests 4288 pass (includes new batcher + unwrap tests), tsc --noEmit and biome clean. Co-authored-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Signed-off-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 364 +++++++++++++----- crates/buzz-acp/src/relay.rs | 7 +- .../ingestArchivedObserverEvents.test.mjs | 80 ++++ .../src/features/agents/observerRelayStore.ts | 134 ++++--- 4 files changed, 437 insertions(+), 148 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 811253e4ac..3b4d74b468 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -14,7 +14,7 @@ mod usage; pub use usage::TurnUsage; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::time::Duration; @@ -361,50 +361,120 @@ async fn check_sibling_via_profile( false } -const OBSERVER_PUBLISH_INTERVAL: Duration = Duration::from_millis(167); -const OBSERVER_PUBLISH_LIMIT_PER_MINUTE: usize = 90; +/// Observer frames are published once per tick; everything that accumulated in +/// between ships as a single batched frame per channel. One update per second +/// is smooth enough for a human watching the session viewer, and the batch +/// keeps a tool-heavy turn (hundreds of events) from either flooding the relay +/// or backing up behind a per-event pacer. This replaces the old per-frame +/// pacer (167ms interval + 90/min rolling cap): the tick IS the pacer, and at +/// one flush per second a rolling-minute cap is unreachable. +const OBSERVER_PUBLISH_TICK: Duration = Duration::from_secs(1); -struct ObserverPublishPacer { - next_publish: tokio::time::Instant, - published: VecDeque, +/// Observer event kind for a batch envelope wrapping multiple events. +/// +/// The payload is `{"events": [, ...]}` with every inner event +/// carrying its own `seq`/`timestamp`, so consumers process inner events +/// exactly as they would unbatched ones. Single pending events are published +/// unwrapped, so the envelope only appears when there is something to batch. +const OBSERVER_BATCH_KIND: &str = "batch"; + +/// Collects observer events between publish ticks. +/// +/// Chunk-type events ride the [`ObserverChunkCoalescer`]; everything else is +/// appended in arrival order, force-flushing pending chunks first — the same +/// ordering rule the pre-batching publisher enforced, so merged chunk text can +/// never leapfrog a tool call that arrived mid-stream. +#[derive(Default)] +struct ObserverBatcher { + coalescer: ObserverChunkCoalescer, + ready: Vec, } -impl ObserverPublishPacer { - fn new() -> Self { - Self { - // No initial burst: even the first snapshot frame waits for its slot. - next_publish: tokio::time::Instant::now() + OBSERVER_PUBLISH_INTERVAL, - published: VecDeque::with_capacity(OBSERVER_PUBLISH_LIMIT_PER_MINUTE), +impl ObserverBatcher { + fn ingest(&mut self, event: observer::ObserverEvent) { + // ObserverChunkCoalescer::ingest returns immediately-publishable events + // (force-flushed pending chunks + non-chunk passthrough, or a pending + // set displaced by the 60KB pre-flush); they join the ready queue in + // the order the coalescer emitted them. + self.ready.extend(self.coalescer.ingest(event)); + } + + /// Drain everything pending — coalesced chunks included — in order. + fn drain(&mut self) -> Vec { + self.ready.extend(self.coalescer.flush()); + std::mem::take(&mut self.ready) + } +} + +/// Pack drained events into publishable frames: one batch envelope per +/// channel (splitting when a batch would exceed the plaintext budget), with +/// singletons left unwrapped. +/// +/// Grouping by `channel_id` is load-bearing: the desktop archive indexes a +/// frame under its decrypted top-level `channelId`, so a frame must never mix +/// events from different channels. Groups preserve first-seen channel order +/// and arrival order within each channel. +fn batch_observer_events(events: Vec) -> Vec { + let mut groups: Vec<(Option, Vec)> = Vec::new(); + for mut event in events { + // Pre-trim each inner event so one oversized leaf cannot force every + // batch it touches into whole-envelope elision downstream. + fit_observer_event_to_budget(&mut event); + match groups.iter_mut().find(|(key, _)| *key == event.channel_id) { + Some((_, group)) => group.push(event), + None => groups.push((event.channel_id.clone(), vec![event])), } } - async fn wait(&mut self) { - loop { - let now = tokio::time::Instant::now(); - while self - .published - .front() - .is_some_and(|sent| now.duration_since(*sent) >= Duration::from_secs(60)) + let mut frames = Vec::new(); + for (_, group) in groups { + let mut pending: Vec = Vec::new(); + for event in group { + pending.push(event); + if serialized_len(&batch_envelope(&pending)) > OBSERVER_MAX_PLAINTEXT_LEN + && pending.len() > 1 { - self.published.pop_front(); + let overflow = pending.pop().expect("len > 1"); + frames.push(seal_batch(pending)); + pending = vec![overflow]; } + } + if !pending.is_empty() { + frames.push(seal_batch(pending)); + } + } + frames +} - let minute_slot = self.published.front().and_then(|sent| { - (self.published.len() >= OBSERVER_PUBLISH_LIMIT_PER_MINUTE) - .then_some(*sent + Duration::from_secs(60)) - }); - let publish_at = - minute_slot.map_or(self.next_publish, |slot| slot.max(self.next_publish)); - if publish_at > now { - tokio::time::sleep_until(publish_at).await; - continue; - } +/// A single event ships unwrapped; two or more get the batch envelope. +fn seal_batch(mut events: Vec) -> observer::ObserverEvent { + if events.len() == 1 { + return events.pop().expect("len == 1"); + } + batch_envelope(&events) +} - let published_at = tokio::time::Instant::now(); - self.published.push_back(published_at); - self.next_publish = published_at + OBSERVER_PUBLISH_INTERVAL; - return; - } +/// Build the batch envelope for a set of same-channel events. +/// +/// Envelope metadata mirrors the LAST inner event — the same convention the +/// chunk coalescer uses for merged chunks — so `(timestamp, seq)` ordering and +/// the desktop's latest-live-session tracking see the newest state. +fn batch_envelope(events: &[observer::ObserverEvent]) -> observer::ObserverEvent { + let last = events + .last() + .expect("batch envelope needs at least 1 event"); + observer::ObserverEvent { + seq: last.seq, + timestamp: last.timestamp.clone(), + kind: OBSERVER_BATCH_KIND.to_string(), + agent_index: last.agent_index, + channel_id: last.channel_id.clone(), + session_id: last.session_id.clone(), + turn_id: last.turn_id.clone(), + started_at: last.started_at.clone(), + payload: serde_json::json!({ + "events": serde_json::to_value(events).unwrap_or_default(), + }), } } @@ -445,26 +515,17 @@ async fn run_relay_observer_publisher( owner_pubkey_hex: String, owner_pubkey: PublicKey, ) { - let mut coalescer = ObserverChunkCoalescer::default(); - let mut pacer = ObserverPublishPacer::new(); + let mut batcher = ObserverBatcher::default(); let max_snapshot_seq = snapshot.iter().map(|event| event.seq).max().unwrap_or(0); for event in snapshot { - for event in coalescer.ingest(event) { - publish_relay_observer_event( - &publisher, - &keys, - &agent_pubkey_hex, - &owner_pubkey_hex, - &owner_pubkey, - &mut pacer, - event, - ) - .await; - } + batcher.ingest(event); } - let mut flush_interval = tokio::time::interval(std::time::Duration::from_millis(500)); - flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // One publish per tick: drain the batcher and ship the accumulated events + // as (at most a few) batched frames. The snapshot drains on the first tick, + // so startup pays the same 1s latency as steady state instead of bursting. + let mut publish_tick = tokio::time::interval(OBSERVER_PUBLISH_TICK); + publish_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { tokio::select! { result = rx.recv() => { @@ -475,39 +536,28 @@ async fn run_relay_observer_publisher( if event.seq <= max_snapshot_seq { continue; } - for event in coalescer.ingest(event) { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, - ).await; - } + batcher.ingest(event); } Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { - for event in coalescer.flush() { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, - ).await; - } tracing::warn!(dropped = count, "relay observer publisher lagged"); } Err(tokio::sync::broadcast::error::RecvError::Closed) => { - for event in coalescer.flush() { + // Final drain so shutdown never strands buffered events. + for event in batch_observer_events(batcher.drain()) { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + &owner_pubkey_hex, &owner_pubkey, event, ).await; } break; } } } - _ = flush_interval.tick() => { - // Periodic flush ensures live streaming even during continuous chunk delivery. - for event in coalescer.flush() { + _ = publish_tick.tick() => { + for event in batch_observer_events(batcher.drain()) { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + &owner_pubkey_hex, &owner_pubkey, event, ).await; } } @@ -793,10 +843,8 @@ async fn publish_relay_observer_event( agent_pubkey_hex: &str, owner_pubkey_hex: &str, owner_pubkey: &PublicKey, - pacer: &mut ObserverPublishPacer, mut event: observer::ObserverEvent, ) { - pacer.wait().await; // 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); @@ -4939,12 +4987,21 @@ mod observer_snapshot_race_tests { // The run loop has exited, dropping the publisher; drain the forwarded // events until the channel closes (deterministic — no try_recv race - // with the test_pair forwarding task). + // with the test_pair forwarding task). With per-tick batching the three + // events arrive inside batch envelopes (or unwrapped when a drain held + // exactly one event); unwrap both shapes. let mut markers = Vec::new(); while let Some(event) = published_rx.recv().await { let payload: serde_json::Value = decrypt_observer_payload(&owner_keys, &event).expect("decrypt published frame"); - markers.push(payload["payload"]["marker"].as_str().unwrap().to_string()); + match payload["payload"]["events"].as_array() { + Some(inner) => markers.extend( + inner + .iter() + .map(|e| e["payload"]["marker"].as_str().unwrap().to_string()), + ), + None => markers.push(payload["payload"]["marker"].as_str().unwrap().to_string()), + } } assert_eq!( markers, @@ -4955,36 +5012,155 @@ mod observer_snapshot_race_tests { } #[cfg(test)] -mod observer_publish_pacer_tests { +mod observer_batcher_tests { use super::*; - #[tokio::test(start_paused = true)] - async fn starts_without_a_burst_and_spaces_frames() { - let started = tokio::time::Instant::now(); - let mut pacer = ObserverPublishPacer::new(); + fn event(seq: u64, kind: &str, channel: Option<&str>) -> observer::ObserverEvent { + observer::ObserverEvent { + seq, + timestamp: format!("2026-04-29T04:00:{:02}Z", seq.min(59)), + kind: kind.to_string(), + agent_index: Some(0), + channel_id: channel.map(ToOwned::to_owned), + session_id: Some("session-1".to_string()), + turn_id: Some("turn-1".to_string()), + started_at: None, + payload: serde_json::json!({ "seq": seq }), + } + } + + /// Two or more pending events for one channel ship as a single batch + /// envelope whose payload carries every inner event in arrival order. + #[test] + fn multiple_events_ship_as_one_envelope_in_order() { + let frames = batch_observer_events(vec![ + event(1, "turn_started", Some("chan-a")), + event(2, "acp_read", Some("chan-a")), + event(3, "acp_write", Some("chan-a")), + ]); - pacer.wait().await; - let first = tokio::time::Instant::now(); - pacer.wait().await; - let second = tokio::time::Instant::now(); + assert_eq!(frames.len(), 1, "one channel, one frame"); + let frame = &frames[0]; + assert_eq!(frame.kind, OBSERVER_BATCH_KIND); + assert_eq!(frame.seq, 3, "envelope mirrors the last inner event"); + let inner = frame.payload["events"].as_array().expect("events array"); + assert_eq!( + inner + .iter() + .map(|e| e["seq"].as_u64().unwrap()) + .collect::>(), + [1, 2, 3], + "inner events preserve arrival order" + ); + assert_eq!(inner[1]["kind"], "acp_read", "inner events keep their kind"); + } - assert_eq!(first.duration_since(started), OBSERVER_PUBLISH_INTERVAL); - assert_eq!(second.duration_since(first), OBSERVER_PUBLISH_INTERVAL); + /// A single pending event is published unwrapped — no envelope, so + /// consumers that predate batching still understand quiet periods. + #[test] + fn a_single_event_stays_unwrapped() { + let frames = batch_observer_events(vec![event(7, "turn_started", Some("chan-a"))]); + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].kind, "turn_started"); + assert_eq!(frames[0].seq, 7); } - #[tokio::test(start_paused = true)] - async fn limits_frames_in_each_rolling_minute() { - let mut pacer = ObserverPublishPacer::new(); - pacer.wait().await; - let first = tokio::time::Instant::now(); - for _ in 1..OBSERVER_PUBLISH_LIMIT_PER_MINUTE { - pacer.wait().await; + /// Events from different channels never share an envelope: the desktop + /// archive indexes a frame under its top-level `channelId`, so a mixed + /// batch would mis-file every foreign event. First-seen channel order and + /// per-channel arrival order are both preserved. + #[test] + fn batches_never_mix_channels() { + let frames = batch_observer_events(vec![ + event(1, "acp_read", Some("chan-a")), + event(2, "acp_read", Some("chan-b")), + event(3, "acp_write", Some("chan-a")), + event(4, "acp_read", None), + ]); + + assert_eq!( + frames.len(), + 3, + "chan-a batch, chan-b singleton, None singleton" + ); + assert_eq!(frames[0].kind, OBSERVER_BATCH_KIND); + assert_eq!(frames[0].channel_id.as_deref(), Some("chan-a")); + let inner = frames[0].payload["events"].as_array().unwrap(); + assert_eq!(inner.len(), 2); + assert_eq!(frames[1].channel_id.as_deref(), Some("chan-b")); + assert_eq!(frames[1].kind, "acp_read", "singleton unwrapped"); + assert_eq!(frames[2].channel_id, None); + assert_eq!(frames[2].seq, 4); + } + + /// A batch that would exceed the plaintext budget splits into multiple + /// envelopes, each independently under the cap, with no event lost. + #[test] + fn oversized_batches_split_under_the_plaintext_cap() { + let big_text = "x".repeat(30_000); + let events: Vec<_> = (1..=6) + .map(|seq| { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ "seq": seq, "text": big_text }); + e + }) + .collect(); + + let frames = batch_observer_events(events); + assert!( + frames.len() > 1, + "six 30KB events cannot fit one 64KB frame" + ); + let mut seen = Vec::new(); + for frame in &frames { + assert!( + serialized_len(frame) <= OBSERVER_MAX_PLAINTEXT_LEN, + "every emitted frame must fit the plaintext cap" + ); + match frame.payload.get("events").and_then(|v| v.as_array()) { + Some(inner) => { + seen.extend(inner.iter().map(|e| e["seq"].as_u64().unwrap())); + } + None => seen.push(frame.seq), + } + } + assert_eq!( + seen, + [1, 2, 3, 4, 5, 6], + "no event lost or reordered by splitting" + ); + } + + /// The batcher preserves the coalescer's ordering rule: a non-chunk event + /// force-flushes pending chunk text ahead of itself, so merged chunks can + /// never leapfrog a tool call that arrived after them. + #[test] + fn non_chunk_events_flush_pending_chunks_ahead_of_themselves() { + fn chunk(seq: u64, text: &str) -> observer::ObserverEvent { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ + "params": { "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "m1", + "content": { "text": text }, + }} + }); + e } - pacer.wait().await; - let ninety_first = tokio::time::Instant::now(); + let mut batcher = ObserverBatcher::default(); + batcher.ingest(chunk(1, "hello ")); + batcher.ingest(chunk(2, "world")); + batcher.ingest(event(3, "tool_call", Some("chan-a"))); + let drained = batcher.drain(); - assert_eq!(ninety_first.duration_since(first), Duration::from_secs(60)); + assert_eq!(drained.len(), 2, "two chunks coalesce into one event"); + assert_eq!( + drained[0].payload["params"]["update"]["content"]["text"], "hello world", + "chunk text merged before the tool call" + ); + assert_eq!(drained[1].kind, "tool_call"); + assert!(drained[0].seq < drained[1].seq); } } diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index aea5cee077..78b9e877df 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -106,9 +106,10 @@ const REQ_PACING_INTERVAL: Duration = Duration::from_millis(125); /// blocked for more than one REQ's worth of I/O between drain ticks. const DRAIN_BUDGET_PER_ITER: usize = 1; /// Maximum observer telemetry frames parked while the rate-limit gate is armed -/// (or the socket is down). The upstream pacer feeds at most ~6 frames/s, so -/// this covers ~40 s of gating; beyond that the oldest frames are dropped with -/// visible accounting (`gated_observer_dropped`). +/// (or the socket is down). The upstream publisher ships at most ~1 frame/s +/// per active channel (one batched frame per publish tick), so this covers +/// minutes of gating even with several concurrent channels; beyond that the +/// oldest frames are dropped with visible accounting (`gated_observer_dropped`). const GATED_OBSERVER_QUEUE_CAP: usize = 256; use std::time::Instant; diff --git a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs index c44c2a88e0..343ce24133 100644 --- a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs +++ b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs @@ -1106,4 +1106,84 @@ describe("raw-event-level merge: stateful aggregates across live/archive boundar ); assert.equal(latestMs, Date.parse(TIMESTAMP)); }); + + it("test_batch_envelope_inner_events_route_by_channel_id", async () => { + // A kind:"batch" envelope from the 1/s observer pacer must be unwrapped on + // the archive-ingest seam: inner events with a channelId land in the + // channel-scoped archive window, inner events without one fall through to + // the per-agent live store — and the envelope itself is never stored. + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const CHANNEL = "chan-batch"; + + const channelInner = makeObserverEvent({ + seq: 21, + timestamp: "2026-01-01T00:21:00.000Z", + channelId: CHANNEL, + }); + const liveInner = makeObserverEvent({ + seq: 22, + timestamp: "2026-01-01T00:21:01.000Z", + channelId: null, + sessionId: null, + }); + // Envelope metadata mirrors the last inner event (production shape). + const envelope = makeObserverEvent({ + seq: 22, + timestamp: "2026-01-01T00:21:01.000Z", + kind: "batch", + channelId: CHANNEL, + payload: { events: [channelInner, liveInner] }, + }); + + await ingestArchivedObserverEvents([makeRawEvent()], makeDecrypt(envelope)); + + const archived = _testGetArchivedChannelEvents(AGENT_PUBKEY, CHANNEL); + assert.equal( + archived.length, + 1, + "channelId inner event must land in the archive window", + ); + assert.equal(archived[0].seq, 21); + assert.equal(archived[0].kind, "acp_write"); + + const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true); + assert.equal( + snap.events.length, + 1, + "null-channelId inner event must land in the live store", + ); + assert.equal(snap.events[0].seq, 22); + assert.notEqual( + snap.events[0].kind, + "batch", + "the batch envelope itself must never be stored", + ); + }); + + it("test_malformed_batch_envelope_degrades_to_itself", async () => { + // A kind:"batch" envelope with no events array (harness bug shape) must + // degrade to the envelope itself rather than being silently dropped, so a + // batching defect stays visible in the session viewer. + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const CHANNEL = "chan-malformed-batch"; + + const envelope = makeObserverEvent({ + seq: 31, + timestamp: "2026-01-01T00:31:00.000Z", + kind: "batch", + channelId: CHANNEL, + payload: {}, + }); + + await ingestArchivedObserverEvents([makeRawEvent()], makeDecrypt(envelope)); + + const archived = _testGetArchivedChannelEvents(AGENT_PUBKEY, CHANNEL); + assert.equal( + archived.length, + 1, + "malformed envelope must be stored as itself, not dropped", + ); + assert.equal(archived[0].kind, "batch"); + assert.equal(archived[0].seq, 31); + }); }); diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 56c69f915a..611fdd489d 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -338,6 +338,71 @@ export function isObserverEventAfter( return candidate.seq > stored.seq; } +// Observer event kind for a batch envelope wrapping multiple events. The ACP +// harness publishes one frame per second; everything that accumulated between +// ticks arrives as `{ kind: "batch", payload: { events: [...] } }` with every +// inner event carrying its own seq/timestamp. Inner events are processed +// exactly as unbatched ones; the envelope itself is never stored. +const OBSERVER_BATCH_KIND = "batch"; + +// Expand a decrypted observer event into its inner events when it is a batch +// envelope; a non-batch event passes through as a single-element array. A +// malformed envelope (no events array) degrades to the envelope itself so a +// harness bug cannot silently blank the session viewer. +function unwrapObserverBatch(parsed: ObserverEvent): ObserverEvent[] { + if (parsed.kind !== OBSERVER_BATCH_KIND) { + return [parsed]; + } + const payload = parsed.payload as { events?: unknown } | null; + const events = Array.isArray(payload?.events) + ? (payload.events as ObserverEvent[]) + : null; + return events && events.length > 0 ? events : [parsed]; +} + +// Per-event processing shared by every event a live frame carries (one for a +// plain frame, many for a batch envelope). +function processLiveObserverEvent(agentPubkey: string, parsed: ObserverEvent) { + // Track the latest-live-session-id per (agent, channel) on the live path. + // Only set when the parsed event carries both a sessionId and channelId, + // so we never attribute a session to the wrong channel. + if (parsed.sessionId && parsed.channelId) { + const key = liveSessionKey(agentPubkey, parsed.channelId); + const stored = latestLiveSessionByAgentChannel.get(key); + // Advance only when this event sorts strictly AFTER the stored one via + // isObserverEventAfter (timestamp then seq — same ordering as + // compareObserverEvents). This prevents late-arriving live frames from + // older sessions from regressing the latest-live id, while also + // correctly advancing on a same-timestamp frame with a higher seq. + if (!stored || isObserverEventAfter(parsed, stored)) { + latestLiveSessionByAgentChannel.set(key, { + sessionId: parsed.sessionId, + timestamp: parsed.timestamp, + seq: parsed.seq, + }); + } + } + appendAgentEvent(agentPubkey, parsed); + const managementRequest = parseAgentManagementRequest(parsed.payload); + if (managementRequest) { + for (const listener of agentManagementListeners) { + listener(agentPubkey, managementRequest); + } + } + if (parsed.kind === "session_config_captured") { + void putAgentSessionConfig(agentPubkey, parsed.payload); + onSessionConfigCaptured?.(agentPubkey); + } else if (parsed.kind === "control_result") { + dispatchControlResult(agentPubkey, parsed.payload); + } else if (parsed.kind === "managed_agent_runtime_lifecycle") { + void putManagedAgentRuntimeLifecycle(agentPubkey, parsed.payload).catch( + (error) => { + console.debug("Late/untracked lifecycle frame dropped:", error); + }, + ); + } +} + async function handleRelayObserverEvent( event: RelayEvent, activeGeneration: number, @@ -372,43 +437,8 @@ async function handleRelayObserverEvent( if (activeGeneration !== generation) { return; } - // Track the latest-live-session-id per (agent, channel) on the live path. - // Only set when the parsed event carries both a sessionId and channelId, - // so we never attribute a session to the wrong channel. - if (parsed.sessionId && parsed.channelId) { - const key = liveSessionKey(agentPubkey, parsed.channelId); - const stored = latestLiveSessionByAgentChannel.get(key); - // Advance only when this event sorts strictly AFTER the stored one via - // isObserverEventAfter (timestamp then seq — same ordering as - // compareObserverEvents). This prevents late-arriving live frames from - // older sessions from regressing the latest-live id, while also - // correctly advancing on a same-timestamp frame with a higher seq. - if (!stored || isObserverEventAfter(parsed, stored)) { - latestLiveSessionByAgentChannel.set(key, { - sessionId: parsed.sessionId, - timestamp: parsed.timestamp, - seq: parsed.seq, - }); - } - } - appendAgentEvent(agentPubkey, parsed); - const managementRequest = parseAgentManagementRequest(parsed.payload); - if (managementRequest) { - for (const listener of agentManagementListeners) { - listener(agentPubkey, managementRequest); - } - } - if (parsed.kind === "session_config_captured") { - void putAgentSessionConfig(agentPubkey, parsed.payload); - onSessionConfigCaptured?.(agentPubkey); - } else if (parsed.kind === "control_result") { - dispatchControlResult(agentPubkey, parsed.payload); - } else if (parsed.kind === "managed_agent_runtime_lifecycle") { - void putManagedAgentRuntimeLifecycle(agentPubkey, parsed.payload).catch( - (error) => { - console.debug("Late/untracked lifecycle frame dropped:", error); - }, - ); + for (const inner of unwrapObserverBatch(parsed)) { + processLiveObserverEvent(agentPubkey, inner); } } catch (error) { if (activeGeneration !== generation) { @@ -676,20 +706,22 @@ export async function ingestArchivedObserverEvents( } try { const parsed = (await _decryptFn(event)) as ObserverEvent; - // Route archived events to the channel-scoped archive window (no cap) - // rather than the per-agent live-relay store (MAX_OBSERVER_EVENTS cap). - // Events without a channelId fall through to the live store so they - // remain visible in the agent's general transcript. - if (parsed.channelId) { - const added = appendArchivedChannelEvent( - agentPubkey, - parsed.channelId, - parsed, - ); - if (added) archiveChanged = true; - } else { - // Live path already calls notifyListeners() inside appendAgentEvent. - appendAgentEvent(agentPubkey, parsed); + for (const inner of unwrapObserverBatch(parsed)) { + // Route archived events to the channel-scoped archive window (no cap) + // rather than the per-agent live-relay store (MAX_OBSERVER_EVENTS cap). + // Events without a channelId fall through to the live store so they + // remain visible in the agent's general transcript. + if (inner.channelId) { + const added = appendArchivedChannelEvent( + agentPubkey, + inner.channelId, + inner, + ); + if (added) archiveChanged = true; + } else { + // Live path already calls notifyListeners() inside appendAgentEvent. + appendAgentEvent(agentPubkey, inner); + } } } catch { // Silently drop decrypt failures — same as live path error handling. From cb90a38ed91354e3ba784490143f0931589b5211 Mon Sep 17 00:00:00 2001 From: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 16:34:40 -0400 Subject: [PATCH 2/6] observer: pace publishes at one global frame per second with a bounded queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 1s batcher fixed the per-event pacer but published EVERY pending frame back-to-back on each tick — with the 1,000-event replay snapshot or several busy channels that is a burst, not a pacer (Max's blocker), and the first interval tick fired immediately, so reconnect replayed the whole snapshot at t=0 (Sami's Finding 1). Per-channel frames also scale linearly with channel count against the flat 120/min Messages quota the observer shares with real chat (Sami's Finding 2). Replace ObserverBatcher with ObserverPublishQueue and make the pacer global: - AT MOST ONE relay frame per tick, regardless of channel count or backlog. interval_at(now + tick) restores the no-startup-burst property. At 1 frame/s telemetry spends at most 60/min of the shared 120/min quota; the tick constant documents that tradeoff as the knob. - Events queue as events (byte-accounted FIFO), not pre-sealed frames, so a backlog keeps compacting into full frames. next_frame() packs the maximal same-channel run at the FRONT of the queue: frames never mix channels (desktop archive indexes by top-level channelId) and never reorder across channels (the desktop per-agent (timestamp,seq) watermark silently skips events that sort stale, so cross-channel inversion loses data). - The queue is bounded at 4 MiB of serialized events (~64s of sustained over-production at the ~64KB/s drain rate). Over budget, the OLDEST events drop with accounting (warn + counter) — designed, visible loss instead of the old silent 90/min drop. Sustained losslessness is therefore one frame's worth of events per second, e.g. ~6 ev/s at 10KB payloads. - Producer close no longer bypasses the pacer: the drain stays one frame per tick and the loop exits only when the queue is empty. (The binary's shutdown path aborts the publisher task, so the paced tail is best-effort at process exit — telemetry, not chat.) Tests: publish-queue packing/order/never-mix/ceiling-with-accounting suite plus paused-time cadence tests asserting no startup burst, exactly one frame per tick, and a paced lossless shutdown drain. Co-authored-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Signed-off-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 647 ++++++++++++++++++++++++++++------- crates/buzz-acp/src/relay.rs | 10 +- 2 files changed, 522 insertions(+), 135 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 3b4d74b468..1a6a09f5e9 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -14,7 +14,7 @@ mod usage; pub use usage::TurnUsage; -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::time::Duration; @@ -361,15 +361,31 @@ async fn check_sibling_via_profile( false } -/// Observer frames are published once per tick; everything that accumulated in -/// between ships as a single batched frame per channel. One update per second -/// is smooth enough for a human watching the session viewer, and the batch -/// keeps a tool-heavy turn (hundreds of events) from either flooding the relay -/// or backing up behind a per-event pacer. This replaces the old per-frame -/// pacer (167ms interval + 90/min rolling cap): the tick IS the pacer, and at -/// one flush per second a rolling-minute cap is unreachable. +/// Observer frames are published at a global rate of AT MOST ONE relay frame +/// per tick — not one per channel, and not one per drain. Everything that +/// accumulates between ticks waits in [`ObserverPublishQueue`] as events and +/// is packed greedily into that single frame. One update per second is smooth +/// enough for a human watching the session viewer, and the global budget is +/// what makes the relay cost model flat: observer frames bill the agent's +/// `LimitType::Messages` quota (`agent_standard_messages_per_min` = 120, +/// enforced in relay `connection.rs::enforce_ws_admission`), shared with the +/// agent's real chat messages. At 1 frame/s telemetry spends at most 60/min — +/// half that budget — regardless of how many channels are active. A slower +/// tick (e.g. 2s → 30/min) would leave more quota headroom for chat at the +/// price of doubled viewer latency; this constant is the knob. const OBSERVER_PUBLISH_TICK: Duration = Duration::from_secs(1); +/// Byte budget for events queued awaiting a publish slot, measured in +/// serialized (post-`fit_observer_event_to_budget`) bytes. Lossless-ness is +/// bounded by this budget: at one ~64KB frame per second the queue drains at +/// most ~64KB/s, so 4 MiB buys roughly **64 seconds** of sustained +/// over-production (at any event size — both the cap and the drain rate are +/// bytes) before the oldest events are dropped WITH accounting (a warn +/// carrying the dropped-event count). Beyond-budget floods therefore degrade +/// to designed, visible loss — strictly better than the pre-batching pacer's +/// silent 90/min drop. +const OBSERVER_PENDING_QUEUE_MAX_BYTES: usize = 4 * 1024 * 1024; + /// Observer event kind for a batch envelope wrapping multiple events. /// /// The payload is `{"events": [, ...]}` with every inner event @@ -378,72 +394,123 @@ const OBSERVER_PUBLISH_TICK: Duration = Duration::from_secs(1); /// unwrapped, so the envelope only appears when there is something to batch. const OBSERVER_BATCH_KIND: &str = "batch"; -/// Collects observer events between publish ticks. +/// Collects observer events awaiting a publish slot. /// /// Chunk-type events ride the [`ObserverChunkCoalescer`]; everything else is /// appended in arrival order, force-flushing pending chunks first — the same /// ordering rule the pre-batching publisher enforced, so merged chunk text can /// never leapfrog a tool call that arrived mid-stream. +/// +/// Events wait here as EVENTS, not pre-sealed frames: each publish slot packs +/// one frame at publish time ([`Self::next_frame`]), so a backlog keeps +/// compacting into full frames instead of freezing into a frame queue. +/// +/// The queue is bounded by [`OBSERVER_PENDING_QUEUE_MAX_BYTES`]. When a +/// sustained flood outruns the one-frame-per-tick drain for longer than the +/// budget, the OLDEST events are dropped (the viewer wants recent state) with +/// accounting: a warning carrying the dropped-event count, and +/// `dropped_events` for tests. #[derive(Default)] -struct ObserverBatcher { +struct ObserverPublishQueue { coalescer: ObserverChunkCoalescer, - ready: Vec, + /// `(serialized_len, event)`, oldest first. Length is captured at enqueue + /// (post-fit) so byte accounting never re-serializes on eviction. + events: VecDeque<(usize, observer::ObserverEvent)>, + pending_bytes: usize, + dropped_events: u64, } -impl ObserverBatcher { +impl ObserverPublishQueue { fn ingest(&mut self, event: observer::ObserverEvent) { // ObserverChunkCoalescer::ingest returns immediately-publishable events // (force-flushed pending chunks + non-chunk passthrough, or a pending - // set displaced by the 60KB pre-flush); they join the ready queue in - // the order the coalescer emitted them. - self.ready.extend(self.coalescer.ingest(event)); + // set displaced by the 60KB pre-flush); they join the queue in the + // order the coalescer emitted them. + for ready in self.coalescer.ingest(event) { + self.enqueue(ready); + } } - /// Drain everything pending — coalesced chunks included — in order. - fn drain(&mut self) -> Vec { - self.ready.extend(self.coalescer.flush()); - std::mem::take(&mut self.ready) - } -} - -/// Pack drained events into publishable frames: one batch envelope per -/// channel (splitting when a batch would exceed the plaintext budget), with -/// singletons left unwrapped. -/// -/// Grouping by `channel_id` is load-bearing: the desktop archive indexes a -/// frame under its decrypted top-level `channelId`, so a frame must never mix -/// events from different channels. Groups preserve first-seen channel order -/// and arrival order within each channel. -fn batch_observer_events(events: Vec) -> Vec { - let mut groups: Vec<(Option, Vec)> = Vec::new(); - for mut event in events { - // Pre-trim each inner event so one oversized leaf cannot force every - // batch it touches into whole-envelope elision downstream. + fn enqueue(&mut self, mut event: observer::ObserverEvent) { + // 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); - match groups.iter_mut().find(|(key, _)| *key == event.channel_id) { - Some((_, group)) => group.push(event), - None => groups.push((event.channel_id.clone(), vec![event])), + let bytes = serialized_len(&event); + self.pending_bytes += bytes; + self.events.push_back((bytes, event)); + + let mut dropped = 0u64; + // `len() > 1`: never drop the event just enqueued (a fitted event is + // ≤ OBSERVER_MAX_PLAINTEXT_LEN, far under the queue budget). + while self.pending_bytes > OBSERVER_PENDING_QUEUE_MAX_BYTES && self.events.len() > 1 { + let (bytes, _) = self.events.pop_front().expect("len > 1"); + self.pending_bytes -= bytes; + dropped += 1; + } + if dropped > 0 { + self.dropped_events += dropped; + tracing::warn!( + dropped, + total_dropped = self.dropped_events, + pending_bytes = self.pending_bytes, + "observer publish queue over byte budget; dropped oldest events" + ); } } - let mut frames = Vec::new(); - for (_, group) in groups { - let mut pending: Vec = Vec::new(); - for event in group { - pending.push(event); - if serialized_len(&batch_envelope(&pending)) > OBSERVER_MAX_PLAINTEXT_LEN - && pending.len() > 1 + /// True when nothing is waiting anywhere — the event queue AND the + /// coalescer's pending chunk buffer. + fn is_empty(&self) -> bool { + self.events.is_empty() && self.coalescer.pending.is_empty() + } + + /// Pack and remove AT MOST ONE publishable frame: the maximal run of + /// same-channel events at the FRONT of the queue, packed greedily until + /// adding the next event would push the envelope over + /// `OBSERVER_MAX_PLAINTEXT_LEN`. Singletons ship unwrapped. + /// + /// Strict FIFO-prefix packing is load-bearing twice over: + /// - A frame must never mix channels (the desktop archive indexes a frame + /// under its decrypted top-level `channelId`). + /// - Frames must publish in GLOBAL event order, not per-channel order: + /// the desktop's `activeAgentTurnsStore` keeps one `(timestamp, seq)` + /// watermark per agent and SKIPS events that sort at or before it, so + /// publishing channel A's newer events ahead of channel B's older ones + /// would make the watermark silently discard B's turn-state events. + /// Packing only the front run — never reaching past a foreign-channel + /// event — makes cross-channel inversion structurally impossible, and + /// makes fairness automatic (pure FIFO, no channel can starve another). + /// + /// Pending coalesced chunks are flushed into the queue first, so a + /// publish slot never leaves merged chunk text stranded behind the tick. + fn next_frame(&mut self) -> Option { + for ready in self.coalescer.flush() { + self.enqueue(ready); + } + let channel = self.events.front()?.1.channel_id.clone(); + + let mut picked: Vec = Vec::new(); + while let Some((bytes, event)) = self.events.front() { + if event.channel_id != channel { + break; + } + let bytes = *bytes; + let (_, event) = self.events.pop_front().expect("front exists"); + picked.push(event); + if picked.len() > 1 + && serialized_len(&batch_envelope(&picked)) > OBSERVER_MAX_PLAINTEXT_LEN { - let overflow = pending.pop().expect("len > 1"); - frames.push(seal_batch(pending)); - pending = vec![overflow]; + // Frame full: the overflow event returns to the front and + // leads the next slot's packing. + let event = picked.pop().expect("len > 1"); + self.events.push_front((bytes, event)); + break; } + self.pending_bytes -= bytes; } - if !pending.is_empty() { - frames.push(seal_batch(pending)); - } + Some(seal_batch(picked)) } - frames } /// A single event ships unwrapped; two or more get the batch envelope. @@ -515,20 +582,26 @@ async fn run_relay_observer_publisher( owner_pubkey_hex: String, owner_pubkey: PublicKey, ) { - let mut batcher = ObserverBatcher::default(); + let mut queue = ObserverPublishQueue::default(); let max_snapshot_seq = snapshot.iter().map(|event| event.seq).max().unwrap_or(0); for event in snapshot { - batcher.ingest(event); + queue.ingest(event); } - // One publish per tick: drain the batcher and ship the accumulated events - // as (at most a few) batched frames. The snapshot drains on the first tick, - // so startup pays the same 1s latency as steady state instead of bursting. - let mut publish_tick = tokio::time::interval(OBSERVER_PUBLISH_TICK); + // Global pacer: AT MOST ONE relay frame per tick, no matter how many + // channels are active or how large the backlog is. `interval_at` starts + // the first tick a full period out, so a pre-loaded snapshot (up to the + // 1,000-event replay buffer on reconnect) cannot burst at t=0 — the old + // pacer's explicit "no initial burst" property, restored. + let mut publish_tick = tokio::time::interval_at( + tokio::time::Instant::now() + OBSERVER_PUBLISH_TICK, + OBSERVER_PUBLISH_TICK, + ); publish_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut closed = false; loop { tokio::select! { - result = rx.recv() => { + result = rx.recv(), if !closed => { match result { Ok(event) => { // Skip live events already delivered via the snapshot @@ -536,30 +609,30 @@ async fn run_relay_observer_publisher( if event.seq <= max_snapshot_seq { continue; } - batcher.ingest(event); + queue.ingest(event); } Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { tracing::warn!(dropped = count, "relay observer publisher lagged"); } Err(tokio::sync::broadcast::error::RecvError::Closed) => { - // Final drain so shutdown never strands buffered events. - for event in batch_observer_events(batcher.drain()) { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, - ).await; - } - break; + // Producer gone: stop selecting on the receiver and let + // the tick arm drain what remains — still one frame per + // tick. An unpaced final drain would be a burst bypass + // around everything the pacer exists to prevent. + closed = true; } } } _ = publish_tick.tick() => { - for event in batch_observer_events(batcher.drain()) { + if let Some(frame) = queue.next_frame() { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, + &owner_pubkey_hex, &owner_pubkey, frame, ).await; } + if closed && queue.is_empty() { + break; + } } } } @@ -5012,7 +5085,7 @@ mod observer_snapshot_race_tests { } #[cfg(test)] -mod observer_batcher_tests { +mod observer_publish_queue_tests { use super::*; fn event(seq: u64, kind: &str, channel: Option<&str>) -> observer::ObserverEvent { @@ -5029,29 +5102,48 @@ mod observer_batcher_tests { } } + fn queue_of(events: Vec) -> ObserverPublishQueue { + let mut queue = ObserverPublishQueue::default(); + for event in events { + queue.ingest(event); + } + queue + } + + /// Collect every frame the queue will produce, one publish slot at a time. + fn drain_frames(queue: &mut ObserverPublishQueue) -> Vec { + let mut frames = Vec::new(); + while !queue.is_empty() { + frames.push(queue.next_frame().expect("queue not empty")); + } + frames + } + + /// Inner seqs of a frame, whether it is an envelope or an unwrapped + /// singleton. + fn frame_seqs(frame: &observer::ObserverEvent) -> Vec { + match frame.payload.get("events").and_then(|v| v.as_array()) { + Some(inner) => inner.iter().map(|e| e["seq"].as_u64().unwrap()).collect(), + None => vec![frame.seq], + } + } + /// Two or more pending events for one channel ship as a single batch /// envelope whose payload carries every inner event in arrival order. #[test] fn multiple_events_ship_as_one_envelope_in_order() { - let frames = batch_observer_events(vec![ + let mut queue = queue_of(vec![ event(1, "turn_started", Some("chan-a")), event(2, "acp_read", Some("chan-a")), event(3, "acp_write", Some("chan-a")), ]); - assert_eq!(frames.len(), 1, "one channel, one frame"); - let frame = &frames[0]; + let frame = queue.next_frame().expect("one frame"); + assert!(queue.is_empty(), "one channel, one publish slot"); assert_eq!(frame.kind, OBSERVER_BATCH_KIND); assert_eq!(frame.seq, 3, "envelope mirrors the last inner event"); + assert_eq!(frame_seqs(&frame), [1, 2, 3], "arrival order preserved"); let inner = frame.payload["events"].as_array().expect("events array"); - assert_eq!( - inner - .iter() - .map(|e| e["seq"].as_u64().unwrap()) - .collect::>(), - [1, 2, 3], - "inner events preserve arrival order" - ); assert_eq!(inner[1]["kind"], "acp_read", "inner events keep their kind"); } @@ -5059,54 +5151,82 @@ mod observer_batcher_tests { /// consumers that predate batching still understand quiet periods. #[test] fn a_single_event_stays_unwrapped() { - let frames = batch_observer_events(vec![event(7, "turn_started", Some("chan-a"))]); - assert_eq!(frames.len(), 1); - assert_eq!(frames[0].kind, "turn_started"); - assert_eq!(frames[0].seq, 7); + let mut queue = queue_of(vec![event(7, "turn_started", Some("chan-a"))]); + let frame = queue.next_frame().expect("one frame"); + assert!(queue.is_empty()); + assert_eq!(frame.kind, "turn_started"); + assert_eq!(frame.seq, 7); } - /// Events from different channels never share an envelope: the desktop - /// archive indexes a frame under its top-level `channelId`, so a mixed - /// batch would mis-file every foreign event. First-seen channel order and - /// per-channel arrival order are both preserved. + /// An empty queue yields no frame — a tick with nothing pending must not + /// publish anything. #[test] - fn batches_never_mix_channels() { - let frames = batch_observer_events(vec![ + fn empty_queue_yields_no_frame() { + let mut queue = ObserverPublishQueue::default(); + assert!(queue.next_frame().is_none()); + assert!(queue.is_empty()); + } + + /// Frames never mix channels AND always publish in global arrival order. + /// Interleaved channels therefore produce one frame per same-channel run: + /// a frame must not reach past a foreign-channel event, or the desktop's + /// per-agent (timestamp, seq) watermark would skip the delayed channel's + /// events as stale. + #[test] + fn frames_never_mix_channels_and_never_reorder_across_channels() { + let mut queue = queue_of(vec![ event(1, "acp_read", Some("chan-a")), - event(2, "acp_read", Some("chan-b")), - event(3, "acp_write", Some("chan-a")), - event(4, "acp_read", None), + event(2, "acp_write", Some("chan-a")), + event(3, "acp_read", Some("chan-b")), + event(4, "acp_read", Some("chan-a")), + event(5, "acp_read", None), ]); + let frames = drain_frames(&mut queue); assert_eq!( frames.len(), - 3, - "chan-a batch, chan-b singleton, None singleton" + 4, + "runs: [1,2]@a, [3]@b, [4]@a, [5]@None — one frame each" + ); + for frame in &frames { + let channels: HashSet> = match frame.payload.get("events") { + Some(serde_json::Value::Array(inner)) => inner + .iter() + .map(|e| e["channelId"].as_str().map(ToOwned::to_owned)) + .collect(), + _ => std::iter::once(frame.channel_id.clone()).collect(), + }; + assert_eq!(channels.len(), 1, "a frame never mixes channels"); + } + let published: Vec = frames.iter().flat_map(frame_seqs).collect(); + assert_eq!( + published, + [1, 2, 3, 4, 5], + "global arrival order survives packing" ); - assert_eq!(frames[0].kind, OBSERVER_BATCH_KIND); assert_eq!(frames[0].channel_id.as_deref(), Some("chan-a")); - let inner = frames[0].payload["events"].as_array().unwrap(); - assert_eq!(inner.len(), 2); - assert_eq!(frames[1].channel_id.as_deref(), Some("chan-b")); - assert_eq!(frames[1].kind, "acp_read", "singleton unwrapped"); - assert_eq!(frames[2].channel_id, None); - assert_eq!(frames[2].seq, 4); + assert_eq!(frames[1].kind, "acp_read", "singleton run stays unwrapped"); + assert_eq!(frames[2].channel_id.as_deref(), Some("chan-a")); + assert_eq!(frames[3].channel_id, None); } - /// A batch that would exceed the plaintext budget splits into multiple - /// envelopes, each independently under the cap, with no event lost. + /// A same-channel backlog that cannot fit one 64KB frame splits across + /// SUCCESSIVE publish slots — never multiple frames from one slot — with + /// every frame under the cap and no event lost or reordered. #[test] - fn oversized_batches_split_under_the_plaintext_cap() { + fn oversized_backlogs_split_across_publish_slots_under_the_cap() { let big_text = "x".repeat(30_000); - let events: Vec<_> = (1..=6) - .map(|seq| { - let mut e = event(seq, "acp_read", Some("chan-a")); - e.payload = serde_json::json!({ "seq": seq, "text": big_text }); - e - }) - .collect(); + let mut queue = queue_of( + (1..=6) + .map(|seq| { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ "seq": seq, "text": big_text }); + e + }) + .collect(), + ); - let frames = batch_observer_events(events); + let frames = drain_frames(&mut queue); assert!( frames.len() > 1, "six 30KB events cannot fit one 64KB frame" @@ -5117,12 +5237,7 @@ mod observer_batcher_tests { serialized_len(frame) <= OBSERVER_MAX_PLAINTEXT_LEN, "every emitted frame must fit the plaintext cap" ); - match frame.payload.get("events").and_then(|v| v.as_array()) { - Some(inner) => { - seen.extend(inner.iter().map(|e| e["seq"].as_u64().unwrap())); - } - None => seen.push(frame.seq), - } + seen.extend(frame_seqs(frame)); } assert_eq!( seen, @@ -5131,7 +5246,7 @@ mod observer_batcher_tests { ); } - /// The batcher preserves the coalescer's ordering rule: a non-chunk event + /// The queue preserves the coalescer's ordering rule: a non-chunk event /// force-flushes pending chunk text ahead of itself, so merged chunks can /// never leapfrog a tool call that arrived after them. #[test] @@ -5148,19 +5263,289 @@ mod observer_batcher_tests { e } - let mut batcher = ObserverBatcher::default(); - batcher.ingest(chunk(1, "hello ")); - batcher.ingest(chunk(2, "world")); - batcher.ingest(event(3, "tool_call", Some("chan-a"))); - let drained = batcher.drain(); + let mut queue = ObserverPublishQueue::default(); + queue.ingest(chunk(1, "hello ")); + queue.ingest(chunk(2, "world")); + queue.ingest(event(3, "tool_call", Some("chan-a"))); - assert_eq!(drained.len(), 2, "two chunks coalesce into one event"); + let frame = queue.next_frame().expect("one frame"); + assert!(queue.is_empty()); + let inner = frame.payload["events"].as_array().expect("batch of 2"); + assert_eq!(inner.len(), 2, "two chunks coalesce into one event"); assert_eq!( - drained[0].payload["params"]["update"]["content"]["text"], "hello world", + inner[0]["payload"]["params"]["update"]["content"]["text"], "hello world", "chunk text merged before the tool call" ); - assert_eq!(drained[1].kind, "tool_call"); - assert!(drained[0].seq < drained[1].seq); + assert_eq!(inner[1]["kind"], "tool_call"); + assert!(inner[0]["seq"].as_u64() < inner[1]["seq"].as_u64()); + } + + /// Chunks still pending inside the coalescer (no non-chunk flushed them) + /// are picked up by the publish slot itself, not stranded. + #[test] + fn a_publish_slot_flushes_pending_coalesced_chunks() { + let mut e = event(1, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ + "params": { "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "m1", + "content": { "text": "buffered" }, + }} + }); + let mut queue = ObserverPublishQueue::default(); + queue.ingest(e); + assert!(!queue.is_empty(), "pending chunk counts as queued work"); + + let frame = queue.next_frame().expect("chunk must ship"); + assert!(queue.is_empty()); + assert_eq!( + frame.payload["params"]["update"]["content"]["text"], + "buffered" + ); + } + + /// Sami's ceiling assertion: when sustained input outruns the one-frame + /// drain budget for longer than the queue's byte budget, the OLDEST events + /// drop with accounting — never silently — and everything that survives + /// publishes in order with nothing else lost. + #[test] + fn over_budget_floods_drop_oldest_with_accounting() { + let big_text = "y".repeat(10_000); + let total = 500usize; // ~5MB of ~10KB events > 4MiB budget + let mut queue = ObserverPublishQueue::default(); + for seq in 1..=total as u64 { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ "seq": seq, "text": big_text }); + queue.ingest(e); + } + + assert!( + queue.dropped_events > 0, + "a 5MB backlog must overflow the 4MiB budget" + ); + assert!( + queue.pending_bytes <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + "eviction must restore the byte budget" + ); + + let frames = drain_frames(&mut queue); + let published: Vec = frames.iter().flat_map(frame_seqs).collect(); + let expected: Vec = (queue.dropped_events + 1..=total as u64).collect(); + assert_eq!( + published, expected, + "exactly the oldest `dropped_events` events are missing; the rest \ + publish in order" + ); + assert_eq!( + published.len() as u64 + queue.dropped_events, + total as u64, + "accounting: published + dropped == ingested" + ); + } + + /// Under the byte budget the queue is lossless: every ingested event + /// publishes exactly once. + #[test] + fn under_budget_backlogs_are_lossless() { + let mut queue = queue_of( + (1..=200) + .map(|seq| event(seq, "acp_read", Some("chan-a"))) + .collect(), + ); + let frames = drain_frames(&mut queue); + let published: Vec = frames.iter().flat_map(frame_seqs).collect(); + assert_eq!(published, (1..=200).collect::>()); + assert_eq!(queue.dropped_events, 0); + } +} + +#[cfg(test)] +mod observer_publish_cadence_tests { + use super::*; + use nostr::Keys; + + /// Let every spawned task (publisher loop, test_pair forwarder) run to + /// quiescence WITHOUT advancing paused time. `yield_now` keeps this task + /// runnable, so tokio's auto-advance never fires here — time only moves + /// when the test says so. + async fn settle() { + for _ in 0..64 { + tokio::task::yield_now().await; + } + } + + fn recv_all(rx: &mut tokio::sync::mpsc::Receiver) -> Vec { + let mut out = Vec::new(); + while let Ok(event) = rx.try_recv() { + out.push(event); + } + out + } + + fn count_inner(owner: &Keys, event: &nostr::Event) -> usize { + let payload: serde_json::Value = + decrypt_observer_payload(owner, event).expect("decrypt frame"); + match payload["payload"]["events"].as_array() { + Some(inner) => inner.len(), + None => 1, + } + } + + fn emit_on(observer: &observer::ObserverHandle, channel: Option, marker: &str) { + observer.emit( + "test_event", + None, + &observer::context_for(channel, None, None), + serde_json::json!({ "marker": marker }), + ); + } + + /// THE regression Max demanded: with a backlog needing multiple frames + /// (multiple channels — the interleaving forces one frame per run), no + /// frame publishes before its tick. Startup publishes NOTHING at t=0 + /// (Sami's Finding 1: a full replay buffer must not burst on reconnect), + /// frame 1 arrives at +1s, frame 2 no earlier than +2s, and so on. + #[tokio::test(start_paused = true)] + async fn one_frame_per_second_and_no_startup_burst() { + let observer = observer::ObserverHandle::in_process(); + let agent_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let (publisher, mut published_rx) = RelayEventPublisher::test_pair(); + + // Interleave channels so the backlog cannot fit one frame: each run + // boundary forces a new publish slot. + let chan_a = uuid::Uuid::new_v4(); + let chan_b = uuid::Uuid::new_v4(); + emit_on(&observer, Some(chan_a), "a1"); + emit_on(&observer, Some(chan_b), "b1"); + emit_on(&observer, Some(chan_a), "a2"); + + let rx = observer.subscribe(); + let snapshot = observer.snapshot(); + assert_eq!(snapshot.len(), 3, "all three preloaded in the snapshot"); + + let task = tokio::spawn(run_relay_observer_publisher( + snapshot, + rx, + publisher, + agent_keys.clone(), + agent_keys.public_key().to_hex(), + owner_keys.public_key().to_hex(), + owner_keys.public_key(), + )); + + // t=0: nothing may publish, no matter how full the snapshot was. + settle().await; + assert_eq!( + recv_all(&mut published_rx).len(), + 0, + "startup must not burst at t=0" + ); + + // t=0.999s: still nothing. + tokio::time::advance(Duration::from_millis(999)).await; + settle().await; + assert_eq!( + recv_all(&mut published_rx).len(), + 0, + "no frame may publish before the first tick" + ); + + // t=1s: exactly ONE frame (the chan-a run: a1 only — b1 arrived + // before a2, so the front run is a1 alone). + tokio::time::advance(Duration::from_millis(1)).await; + settle().await; + let frames = recv_all(&mut published_rx); + assert_eq!(frames.len(), 1, "tick 1 publishes exactly one frame"); + assert_eq!(count_inner(&owner_keys, &frames[0]), 1); + + // t=1.5s: between ticks, nothing. + tokio::time::advance(Duration::from_millis(500)).await; + settle().await; + assert_eq!( + recv_all(&mut published_rx).len(), + 0, + "frame 2 must wait for tick 2" + ); + + // t=2s and t=3s: one frame per tick until the backlog drains. + tokio::time::advance(Duration::from_millis(500)).await; + settle().await; + assert_eq!(recv_all(&mut published_rx).len(), 1, "tick 2: one frame"); + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + assert_eq!(recv_all(&mut published_rx).len(), 1, "tick 3: one frame"); + + // Backlog drained; a quiet tick publishes nothing. + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + assert_eq!(recv_all(&mut published_rx).len(), 0, "quiet tick is quiet"); + + task.abort(); + } + + /// Shutdown is NOT a burst bypass: when the producer closes with a + /// backlog, the remaining frames still publish one per tick, and the loop + /// exits only after the queue is empty — paced, lossless, in order. + #[tokio::test(start_paused = true)] + async fn shutdown_drain_is_paced_and_lossless() { + let observer = observer::ObserverHandle::in_process(); + let agent_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let (publisher, mut published_rx) = RelayEventPublisher::test_pair(); + + let chan_a = uuid::Uuid::new_v4(); + let chan_b = uuid::Uuid::new_v4(); + emit_on(&observer, Some(chan_a), "a1"); + emit_on(&observer, Some(chan_b), "b1"); + emit_on(&observer, Some(chan_a), "a2"); + + let rx = observer.subscribe(); + let snapshot = observer.snapshot(); + // Close the broadcast channel immediately: the entire drain happens + // in "shutdown" mode. + drop(observer); + + let task = tokio::spawn(run_relay_observer_publisher( + snapshot, + rx, + publisher, + agent_keys.clone(), + agent_keys.public_key().to_hex(), + owner_keys.public_key().to_hex(), + owner_keys.public_key(), + )); + + settle().await; + assert_eq!( + recv_all(&mut published_rx).len(), + 0, + "shutdown drain must not burst at t=0" + ); + + let mut markers = Vec::new(); + for tick in 1..=3 { + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + let frames = recv_all(&mut published_rx); + assert_eq!(frames.len(), 1, "shutdown tick {tick}: exactly one frame"); + let payload: serde_json::Value = + decrypt_observer_payload(&owner_keys, &frames[0]).expect("decrypt"); + match payload["payload"]["events"].as_array() { + Some(inner) => markers.extend( + inner + .iter() + .map(|e| e["payload"]["marker"].as_str().unwrap().to_string()), + ), + None => markers.push(payload["payload"]["marker"].as_str().unwrap().to_string()), + } + } + assert_eq!(markers, ["a1", "b1", "a2"], "paced drain loses nothing"); + + // Queue empty + closed: the loop must have exited on its own. + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + assert!(task.is_finished(), "publisher exits after paced drain"); } } diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 78b9e877df..2cbb82411f 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -106,10 +106,12 @@ const REQ_PACING_INTERVAL: Duration = Duration::from_millis(125); /// blocked for more than one REQ's worth of I/O between drain ticks. const DRAIN_BUDGET_PER_ITER: usize = 1; /// Maximum observer telemetry frames parked while the rate-limit gate is armed -/// (or the socket is down). The upstream publisher ships at most ~1 frame/s -/// per active channel (one batched frame per publish tick), so this covers -/// minutes of gating even with several concurrent channels; beyond that the -/// oldest frames are dropped with visible accounting (`gated_observer_dropped`). +/// (or the socket is down). The upstream publisher ships at most ONE batched +/// frame per second GLOBALLY (one publish slot per tick, regardless of how +/// many channels are active), so this covers ~4 minutes of gating; beyond that +/// the oldest frames are dropped with visible accounting +/// (`gated_observer_dropped`). Note each dropped frame may carry a whole batch +/// of events, so event-level loss is larger than the frame count. const GATED_OBSERVER_QUEUE_CAP: usize = 256; use std::time::Instant; From d55a3e8a7fdc704b73420328e88b80c888c817f2 Mon Sep 17 00:00:00 2001 From: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 17:52:36 -0400 Subject: [PATCH 3/6] observer batching: gather-packing, coalescer byte cap, per-channel desktop watermark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes from round-2 review of the 1 frame/s observer pacer: 1. Gather-packing (Sami's drain-rate blocker): next_frame now gathers the front event's channel QUEUE-WIDE in FIFO order instead of only the contiguous front run. Front-run packing degraded to ~1 event/s under 2-channel interleave (275 B/s vs 63.5 KB/s measured) — silent unbounded viewer latency with no drop counter moving. Null-channel events are packing barriers nothing gathers across, so agent_panic-class events keep exact global order. 2. Coalescer byte cap (Max's bypass): ObserverChunkCoalescer.pending now carries byte accounting and counts against OBSERVER_PENDING_QUEUE_MAX_BYTES alongside the event FIFO — a distinct-messageId 50KB chunk flood previously retained ~48MB against the 4 MiB cap with zero drops. Eviction stays oldest-first across both stores (queue front, then coalescer front; structural age order). 3. Desktop watermark per (agent, channel): gather-packing intentionally reorders frames across channels, so activeAgentTurnsStore's lastProcessed gate is re-keyed from per-agent to per-(agent, channel), with a dedicated null-channel bucket. Every turn-mutating path (endTurn's null-turnId channel fallback, resurrectTurn) is channel-scoped, so per-channel serialization preserves each existing guard; the tombstone-cap justification is rewritten for the new keying. Community-switch save/restore deep-clones the nested map. Also pins MissedTickBehavior::Skip (surviving mutant M6): a stalled tick arm fires one catch-up frame, not one per missed deadline. New regressions: interleaved 2-channel drain at bytes/slot, null-channel barrier, distinct-key chunk flood bounded with event accounting, cross-channel-delayed null-turnId turn_error evicting only its own channel's turn, null-bucket replay idempotency. Co-authored-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Signed-off-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 404 ++++++++++++++---- .../agents/activeAgentTurnsStore.test.mjs | 126 +++++- .../features/agents/activeAgentTurnsStore.ts | 105 +++-- 3 files changed, 518 insertions(+), 117 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 45519880a5..6e758598d7 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -375,15 +375,23 @@ async fn check_sibling_via_profile( /// price of doubled viewer latency; this constant is the knob. const OBSERVER_PUBLISH_TICK: Duration = Duration::from_secs(1); -/// Byte budget for events queued awaiting a publish slot, measured in -/// serialized (post-`fit_observer_event_to_budget`) bytes. Lossless-ness is -/// bounded by this budget: at one ~64KB frame per second the queue drains at -/// most ~64KB/s, so 4 MiB buys roughly **64 seconds** of sustained -/// over-production (at any event size — both the cap and the drain rate are -/// bytes) before the oldest events are dropped WITH accounting (a warn -/// carrying the dropped-event count). Beyond-budget floods therefore degrade -/// to designed, visible loss — strictly better than the pre-batching pacer's -/// silent 90/min drop. +/// Byte budget for EVERYTHING retained while awaiting a publish slot: the +/// event FIFO (serialized, post-`fit_observer_event_to_budget` bytes) PLUS +/// the chunk coalescer's pending buffer (serialized event skeletons + raw +/// accumulated text). Both stores count against this one cap — a +/// high-cardinality chunk flood (many distinct coalescer keys) is bounded +/// exactly like a plain event flood; neither buffer is a bypass around the +/// other. Lossless-ness is bounded by this budget: each publish slot packs +/// one ~64KB frame, gathered queue-wide for the front channel, so a single +/// channel drains at ~64KB/s and 4 MiB buys roughly **64 seconds** of +/// sustained over-production before the oldest items are dropped WITH +/// accounting (a warn carrying the dropped-event count). With C channels +/// producing concurrently the slots round-robin between them, so the +/// per-channel drain is ~64KB/Cs and the budget shortens accordingly — +/// still bytes-per-slot, never events-per-slot (see +/// [`ObserverPublishQueue::next_frame`]). Beyond-budget floods therefore +/// degrade to designed, visible loss — strictly better than the +/// pre-batching pacer's silent 90/min drop. const OBSERVER_PENDING_QUEUE_MAX_BYTES: usize = 4 * 1024 * 1024; /// Observer event kind for a batch envelope wrapping multiple events. @@ -429,6 +437,7 @@ impl ObserverPublishQueue { for ready in self.coalescer.ingest(event) { self.enqueue(ready); } + self.enforce_byte_budget(); } fn enqueue(&mut self, mut event: observer::ObserverEvent) { @@ -439,13 +448,33 @@ impl ObserverPublishQueue { let bytes = serialized_len(&event); self.pending_bytes += bytes; self.events.push_back((bytes, event)); + } + + /// Total bytes retained across BOTH stores — the event FIFO and the + /// coalescer's pending chunk buffer. The budget binds this sum; counting + /// only the FIFO would let a high-cardinality chunk flood (many distinct + /// coalescer keys, nothing ever flushing) grow unbounded outside the cap. + fn total_pending_bytes(&self) -> usize { + self.pending_bytes + self.coalescer.pending_bytes + } + /// Enforce [`OBSERVER_PENDING_QUEUE_MAX_BYTES`] over the total, dropping + /// OLDEST items first with accounting. Global age order across the two + /// stores is structural: every enqueue path flushes the coalescer first, + /// so every pending coalescer entry is strictly newer than every queued + /// event — eviction is queue front, then coalescer front. The `> 1` + /// guard never drops the sole remaining item (any single fitted event or + /// pre-flush-capped chunk entry is far under the budget). + fn enforce_byte_budget(&mut self) { let mut dropped = 0u64; - // `len() > 1`: never drop the event just enqueued (a fitted event is - // ≤ OBSERVER_MAX_PLAINTEXT_LEN, far under the queue budget). - while self.pending_bytes > OBSERVER_PENDING_QUEUE_MAX_BYTES && self.events.len() > 1 { - let (bytes, _) = self.events.pop_front().expect("len > 1"); - self.pending_bytes -= bytes; + while self.total_pending_bytes() > OBSERVER_PENDING_QUEUE_MAX_BYTES + && self.events.len() + self.coalescer.pending.len() > 1 + { + if let Some((bytes, _)) = self.events.pop_front() { + self.pending_bytes -= bytes; + } else { + self.coalescer.drop_oldest().expect("guard ensures an item"); + } dropped += 1; } if dropped > 0 { @@ -453,7 +482,7 @@ impl ObserverPublishQueue { tracing::warn!( dropped, total_dropped = self.dropped_events, - pending_bytes = self.pending_bytes, + pending_bytes = self.total_pending_bytes(), "observer publish queue over byte budget; dropped oldest events" ); } @@ -465,22 +494,30 @@ impl ObserverPublishQueue { self.events.is_empty() && self.coalescer.pending.is_empty() } - /// Pack and remove AT MOST ONE publishable frame: the maximal run of - /// same-channel events at the FRONT of the queue, packed greedily until + /// Pack and remove AT MOST ONE publishable frame: the front event's + /// channel, gathered queue-wide in FIFO order (packed greedily until /// adding the next event would push the envelope over - /// `OBSERVER_MAX_PLAINTEXT_LEN`. Singletons ship unwrapped. + /// `OBSERVER_MAX_PLAINTEXT_LEN`). Singletons ship unwrapped. /// - /// Strict FIFO-prefix packing is load-bearing twice over: - /// - A frame must never mix channels (the desktop archive indexes a frame - /// under its decrypted top-level `channelId`). - /// - Frames must publish in GLOBAL event order, not per-channel order: - /// the desktop's `activeAgentTurnsStore` keeps one `(timestamp, seq)` - /// watermark per agent and SKIPS events that sort at or before it, so - /// publishing channel A's newer events ahead of channel B's older ones - /// would make the watermark silently discard B's turn-state events. - /// Packing only the front run — never reaching past a foreign-channel - /// event — makes cross-channel inversion structurally impossible, and - /// makes fairness automatic (pure FIFO, no channel can starve another). + /// Two invariants bound the gather: + /// - A frame never mixes channels (the desktop archive indexes a frame + /// under its decrypted top-level `channelId`), and events keep their + /// FIFO order *within* each channel. Cross-channel frame order MAY + /// differ from arrival order — the desktop tolerates that everywhere: + /// the transcript store sorts + rebuilds on out-of-order arrival, the + /// archive is per-channel by construction, and the turn store's + /// watermark is keyed per (agent, channel). + /// - A NULL-channel event is a BARRIER nothing gathers across: null-scope + /// events (`agent_panic`-class) can causally couple to any channel, so + /// their relative order against every channel is preserved exactly. + /// Null-channel events themselves ship only as their contiguous front + /// run. + /// + /// Gathering queue-wide (not just the front run) is what keeps the drain + /// rate in BYTES per slot rather than front-run-length events per slot: + /// with round-robin producers (channel A, B, A, B, ...) a front-run + /// packer degrades to ~1 event per slot regardless of size, silently + /// growing latency without ever tripping the byte budget. /// /// Pending coalesced chunks are flushed into the queue first, so a /// publish slot never leaves merged chunk text stranded behind the tick. @@ -491,24 +528,33 @@ impl ObserverPublishQueue { let channel = self.events.front()?.1.channel_id.clone(); let mut picked: Vec = Vec::new(); - while let Some((bytes, event)) = self.events.front() { - if event.channel_id != channel { - break; - } - let bytes = *bytes; - let (_, event) = self.events.pop_front().expect("front exists"); - picked.push(event); - if picked.len() > 1 - && serialized_len(&batch_envelope(&picked)) > OBSERVER_MAX_PLAINTEXT_LEN - { - // Frame full: the overflow event returns to the front and - // leads the next slot's packing. - let event = picked.pop().expect("len > 1"); - self.events.push_front((bytes, event)); - break; + let mut kept: VecDeque<(usize, observer::ObserverEvent)> = + VecDeque::with_capacity(self.events.len()); + let mut gathering = true; + while let Some((bytes, event)) = self.events.pop_front() { + if gathering && event.channel_id == channel { + picked.push(event); + if picked.len() > 1 + && serialized_len(&batch_envelope(&picked)) > OBSERVER_MAX_PLAINTEXT_LEN + { + // Frame full: the overflow event stays queued and leads + // its channel's next slot. + let event = picked.pop().expect("len > 1"); + kept.push_back((bytes, event)); + gathering = false; + } else { + self.pending_bytes -= bytes; + } + } else { + if gathering && (channel.is_none() || event.channel_id.is_none()) { + // Null-channel barrier (or, for a null-channel frame, the + // end of its contiguous front run): stop gathering. + gathering = false; + } + kept.push_back((bytes, event)); } - self.pending_bytes -= bytes; } + self.events = kept; Some(seal_batch(picked)) } } @@ -641,12 +687,21 @@ async fn run_relay_observer_publisher( #[derive(Default)] struct ObserverChunkCoalescer { pending: Vec, + /// Approximate serialized bytes retained in `pending` (each entry's + /// serialized skeleton at creation plus appended chunk text). Counted + /// against [`OBSERVER_PENDING_QUEUE_MAX_BYTES`] by the owning + /// [`ObserverPublishQueue`] so this buffer can never grow outside the + /// queue's byte budget (a distinct-key chunk flood parks everything here + /// and nothing would otherwise bound it). + pending_bytes: usize, } struct PendingObserverChunk { key: ObserverChunkKey, event: observer::ObserverEvent, text: String, + /// Bytes this entry contributes to `pending_bytes`. + bytes: usize, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -679,20 +734,54 @@ impl ObserverChunkCoalescer { if pending.text.len() + text.len() >= OBSERVER_CHUNK_MAX_TEXT_BYTES { let events = self.flush(); // Start a new pending entry with the current chunk. - self.pending.push(PendingObserverChunk { key, event, text }); + self.push_pending(key, event, text); return events; } pending.text.push_str(&text); + pending.bytes += text.len(); + self.pending_bytes += text.len(); pending.event.seq = event.seq; pending.event.timestamp = event.timestamp; return Vec::new(); } - self.pending.push(PendingObserverChunk { key, event, text }); + self.push_pending(key, event, text); Vec::new() } + fn push_pending( + &mut self, + key: ObserverChunkKey, + event: observer::ObserverEvent, + text: String, + ) { + // The serialized skeleton already carries this entry's first chunk + // text (`text` is an extracted copy, later written back at flush), so + // the entry starts at its full serialized weight and appends add + // exactly their text length. + let bytes = serialized_len(&event); + self.pending_bytes += bytes; + self.pending.push(PendingObserverChunk { + key, + event, + text, + bytes, + }); + } + + /// Evict the OLDEST pending entry for byte-budget enforcement. Returns + /// `None` when there is nothing to drop. + fn drop_oldest(&mut self) -> Option<()> { + if self.pending.is_empty() { + return None; + } + let removed = self.pending.remove(0); + self.pending_bytes -= removed.bytes; + Some(()) + } + fn flush(&mut self) -> Vec { + self.pending_bytes = 0; self.pending .drain(..) .map(|mut pending| { @@ -5167,13 +5256,14 @@ mod observer_publish_queue_tests { assert!(queue.is_empty()); } - /// Frames never mix channels AND always publish in global arrival order. - /// Interleaved channels therefore produce one frame per same-channel run: - /// a frame must not reach past a foreign-channel event, or the desktop's - /// per-agent (timestamp, seq) watermark would skip the delayed channel's - /// events as stale. + /// Frames never mix channels, and each channel's events keep their FIFO + /// order. Gathering is QUEUE-WIDE: the front event's channel collects its + /// events from anywhere in the queue (that is what keeps the drain rate + /// in bytes per slot under interleaving), so cross-channel frame order + /// MAY differ from arrival order — but a null-channel event is a barrier + /// nothing gathers across. #[test] - fn frames_never_mix_channels_and_never_reorder_across_channels() { + fn frames_never_mix_channels_and_gather_queue_wide() { let mut queue = queue_of(vec![ event(1, "acp_read", Some("chan-a")), event(2, "acp_write", Some("chan-a")), @@ -5185,8 +5275,8 @@ mod observer_publish_queue_tests { let frames = drain_frames(&mut queue); assert_eq!( frames.len(), - 4, - "runs: [1,2]@a, [3]@b, [4]@a, [5]@None — one frame each" + 3, + "gathered: [1,2,4]@a, [3]@b, [5]@None — one frame each" ); for frame in &frames { let channels: HashSet> = match frame.payload.get("events") { @@ -5198,16 +5288,76 @@ mod observer_publish_queue_tests { }; assert_eq!(channels.len(), 1, "a frame never mixes channels"); } - let published: Vec = frames.iter().flat_map(frame_seqs).collect(); assert_eq!( - published, - [1, 2, 3, 4, 5], - "global arrival order survives packing" + frame_seqs(&frames[0]), + [1, 2, 4], + "chan-a gathers queue-wide, FIFO within the channel" ); assert_eq!(frames[0].channel_id.as_deref(), Some("chan-a")); - assert_eq!(frames[1].kind, "acp_read", "singleton run stays unwrapped"); - assert_eq!(frames[2].channel_id.as_deref(), Some("chan-a")); - assert_eq!(frames[3].channel_id, None); + assert_eq!(frames[1].kind, "acp_read", "singleton stays unwrapped"); + assert_eq!(frames[1].channel_id.as_deref(), Some("chan-b")); + assert_eq!(frames[2].channel_id, None); + } + + /// A NULL-channel event is a barrier: channel events queued BEHIND it + /// must not gather into a frame ahead of it, so causally-global events + /// (`agent_panic`-class) keep their exact order against every channel. + /// The null event itself ships only its contiguous front run. + #[test] + fn null_channel_events_are_gather_barriers() { + let mut queue = queue_of(vec![ + event(1, "acp_read", Some("chan-a")), + event(2, "acp_read", Some("chan-b")), + event(3, "agent_panic", None), + event(4, "acp_write", Some("chan-a")), + ]); + + let frames = drain_frames(&mut queue); + let published: Vec> = frames.iter().map(frame_seqs).collect(); + assert_eq!( + published, + [vec![1], vec![2], vec![3], vec![4]], + "seq 4 must not gather past the null barrier into frame 1" + ); + } + + /// The drain-rate regression Sami measured: with two channels strictly + /// alternating, a front-run packer degrades to ONE event per slot + /// (~275 B/s regardless of the 64KB frame budget). Queue-wide gathering + /// must drain an interleaved backlog in ~ceil(events / per-frame-fit) + /// slots per channel, not one slot per event. + #[test] + fn interleaved_channels_drain_at_bytes_per_slot_not_events_per_slot() { + let mut events = Vec::new(); + for i in 0..100u64 { + events.push(event(2 * i + 1, "acp_read", Some("chan-a"))); + events.push(event(2 * i + 2, "acp_read", Some("chan-b"))); + } + let mut queue = queue_of(events); + + let frames = drain_frames(&mut queue); + assert!( + frames.len() <= 4, + "200 tiny alternating events must gather into a few full frames, \ + got {} (front-run packing would need 200 slots)", + frames.len() + ); + for frame in &frames { + assert!(serialized_len(frame) <= OBSERVER_MAX_PLAINTEXT_LEN); + } + // Within each channel, FIFO order survives the gather. + let mut seqs_a = Vec::new(); + let mut seqs_b = Vec::new(); + for frame in &frames { + match frame.channel_id.as_deref() { + Some("chan-a") => seqs_a.extend(frame_seqs(frame)), + Some("chan-b") => seqs_b.extend(frame_seqs(frame)), + other => panic!("unexpected channel {other:?}"), + } + } + assert!(seqs_a.windows(2).all(|w| w[0] < w[1]), "chan-a FIFO"); + assert!(seqs_b.windows(2).all(|w| w[0] < w[1]), "chan-b FIFO"); + assert_eq!(seqs_a.len() + seqs_b.len(), 200, "nothing lost"); } /// A same-channel backlog that cannot fit one 64KB frame splits across @@ -5324,7 +5474,7 @@ mod observer_publish_queue_tests { "a 5MB backlog must overflow the 4MiB budget" ); assert!( - queue.pending_bytes <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + queue.total_pending_bytes() <= OBSERVER_PENDING_QUEUE_MAX_BYTES, "eviction must restore the byte budget" ); @@ -5343,6 +5493,57 @@ mod observer_publish_queue_tests { ); } + /// Max's coalescer-bypass regression: a flood of chunks with DISTINCT + /// messageIds never flushes on its own, so every chunk sits in the + /// coalescer's pending buffer. Those bytes MUST count against the byte + /// budget and be evicted with accounting — pre-fix this retained ~25MB + /// against the 4 MiB cap with `pending_bytes == 0` and zero drops. + #[test] + fn distinct_key_chunk_floods_are_bounded_by_the_byte_budget() { + let big_text = "z".repeat(50_000); + let total = 500u64; // ~25MB pending chunk text vs a 4MiB budget + let mut queue = ObserverPublishQueue::default(); + for seq in 1..=total { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "session-1", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": format!("message-{seq}"), + "content": { "type": "text", "text": big_text }, + }, + }, + }); + queue.ingest(e); + } + + assert!( + queue.total_pending_bytes() <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + "total retained bytes (FIFO + coalescer pending) must respect the \ + cap, got {}", + queue.total_pending_bytes() + ); + assert!( + queue.dropped_events > 0, + "a ~25MB distinct-key chunk flood must record drops" + ); + // Event-level accounting: everything that survives publishes, and + // survivors + dropped == ingested. + let frames = drain_frames(&mut queue); + let survived: u64 = frames.iter().map(|f| frame_seqs(f).len() as u64).sum(); + assert_eq!( + survived + queue.dropped_events, + total, + "accounting: published + dropped == ingested" + ); + // The survivors are the NEWEST events (drop-oldest). + let last_frame_seqs = frame_seqs(frames.last().expect("frames")); + assert_eq!(*last_frame_seqs.last().expect("seqs"), total); + } + /// Under the byte budget the queue is lossless: every ingested event /// publishes exactly once. #[test] @@ -5401,10 +5602,10 @@ mod observer_publish_cadence_tests { } /// THE regression Max demanded: with a backlog needing multiple frames - /// (multiple channels — the interleaving forces one frame per run), no - /// frame publishes before its tick. Startup publishes NOTHING at t=0 - /// (Sami's Finding 1: a full replay buffer must not burst on reconnect), - /// frame 1 arrives at +1s, frame 2 no earlier than +2s, and so on. + /// (two channels — a frame never mixes channels, so the backlog takes two + /// publish slots), no frame publishes before its tick. Startup publishes + /// NOTHING at t=0 (Sami's Finding 1: a full replay buffer must not burst + /// on reconnect), frame 1 arrives at +1s, frame 2 no earlier than +2s. #[tokio::test(start_paused = true)] async fn one_frame_per_second_and_no_startup_burst() { let observer = observer::ObserverHandle::in_process(); @@ -5451,13 +5652,13 @@ mod observer_publish_cadence_tests { "no frame may publish before the first tick" ); - // t=1s: exactly ONE frame (the chan-a run: a1 only — b1 arrived - // before a2, so the front run is a1 alone). + // t=1s: exactly ONE frame — chan-a gathered queue-wide, so a1 AND a2 + // ride the first slot together. tokio::time::advance(Duration::from_millis(1)).await; settle().await; let frames = recv_all(&mut published_rx); assert_eq!(frames.len(), 1, "tick 1 publishes exactly one frame"); - assert_eq!(count_inner(&owner_keys, &frames[0]), 1); + assert_eq!(count_inner(&owner_keys, &frames[0]), 2, "a1 + a2 gathered"); // t=1.5s: between ticks, nothing. tokio::time::advance(Duration::from_millis(500)).await; @@ -5468,13 +5669,10 @@ mod observer_publish_cadence_tests { "frame 2 must wait for tick 2" ); - // t=2s and t=3s: one frame per tick until the backlog drains. + // t=2s: the chan-b frame drains on its own tick. tokio::time::advance(Duration::from_millis(500)).await; settle().await; assert_eq!(recv_all(&mut published_rx).len(), 1, "tick 2: one frame"); - tokio::time::advance(Duration::from_secs(1)).await; - settle().await; - assert_eq!(recv_all(&mut published_rx).len(), 1, "tick 3: one frame"); // Backlog drained; a quiet tick publishes nothing. tokio::time::advance(Duration::from_secs(1)).await; @@ -5524,7 +5722,7 @@ mod observer_publish_cadence_tests { ); let mut markers = Vec::new(); - for tick in 1..=3 { + for tick in 1..=2 { tokio::time::advance(Duration::from_secs(1)).await; settle().await; let frames = recv_all(&mut published_rx); @@ -5540,13 +5738,67 @@ mod observer_publish_cadence_tests { None => markers.push(payload["payload"]["marker"].as_str().unwrap().to_string()), } } - assert_eq!(markers, ["a1", "b1", "a2"], "paced drain loses nothing"); + // Gather-packing: chan-a (a1+a2) ships tick 1, chan-b tick 2. + assert_eq!(markers, ["a1", "a2", "b1"], "paced drain loses nothing"); // Queue empty + closed: the loop must have exited on its own. tokio::time::advance(Duration::from_secs(1)).await; settle().await; assert!(task.is_finished(), "publisher exits after paced drain"); } + + /// Pins `MissedTickBehavior::Skip` (Sami's M6 mutant): when the publisher + /// misses ticks — relay backpressure can stall the tick arm past several + /// deadlines, since `publish_event` awaits a bounded mpsc — the interval + /// must fire ONE catch-up tick and realign, not fire once per missed + /// deadline. With `Burst`, a 10s stall against a multi-frame backlog + /// would replay all 10 missed ticks back-to-back: an unpaced burst that + /// bypasses exactly what the pacer exists to prevent. + #[tokio::test(start_paused = true)] + async fn missed_ticks_skip_instead_of_bursting() { + let observer = observer::ObserverHandle::in_process(); + let agent_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let (publisher, mut published_rx) = RelayEventPublisher::test_pair(); + + // Three channels => three frames pending (a frame never mixes + // channels), so a bursting interval would have work for every + // spurious catch-up tick. + for chan in 0..3 { + emit_on(&observer, Some(uuid::Uuid::new_v4()), &format!("c{chan}")); + } + let rx = observer.subscribe(); + let snapshot = observer.snapshot(); + + let task = tokio::spawn(run_relay_observer_publisher( + snapshot, + rx, + publisher, + agent_keys.clone(), + agent_keys.public_key().to_hex(), + owner_keys.public_key().to_hex(), + owner_keys.public_key(), + )); + settle().await; + + // Jump 10 seconds in ONE advance — the loop was never polled in + // between, exactly like a stall across 10 deadlines. + tokio::time::advance(Duration::from_secs(10)).await; + settle().await; + assert_eq!( + recv_all(&mut published_rx).len(), + 1, + "Skip: one catch-up frame after a stall — Burst would publish \ + one per missed deadline" + ); + + // The interval realigned: the remaining backlog stays paced. + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + assert_eq!(recv_all(&mut published_rx).len(), 1, "paced after realign"); + + task.abort(); + } } #[cfg(test)] diff --git a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs index d78d1d21af..73d83a1a4f 100644 --- a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs +++ b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs @@ -59,30 +59,47 @@ describe("activeAgentTurnsStore", () => { assert.ok(channels.has("c1")); }); - it("skips events at or below the watermark", () => { + it("skips events at or below their channel's watermark", () => { + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ seq: 5, turnId: "t1", channelId: "c1" }), + ]); + // An older event on the SAME channel is stale — ignored. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ seq: 3, turnId: "t1b", channelId: "c1" }), + ]); + const turns = getActiveTurnsForAgent(AGENT); + assert.equal(turns.length, 1); + assert.ok(channelIdsOf(turns).has("c1")); + }); + + it("processes an older event on a DIFFERENT channel (cross-channel reorder)", () => { + // The harness's batching packer may publish channel frames out of + // arrival order; each channel gates independently. syncAgentTurnsFromEvents(AGENT, [ makeEvent({ seq: 5, turnId: "t1", channelId: "c1" }), ]); - // Try to process an older event — should be ignored syncAgentTurnsFromEvents(AGENT, [ makeEvent({ seq: 3, turnId: "t2", channelId: "c2" }), ]); const channels = channelIdsOf(getActiveTurnsForAgent(AGENT)); - assert.equal(channels.size, 1); + assert.equal(channels.size, 2); assert.ok(channels.has("c1")); - assert.ok(!channels.has("c2")); + assert.ok( + channels.has("c2"), + "a delayed channel's events must not be skipped as stale", + ); }); - it("skips duplicate seq", () => { + it("skips duplicate seq on the same channel", () => { syncAgentTurnsFromEvents(AGENT, [ makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }), ]); syncAgentTurnsFromEvents(AGENT, [ - makeEvent({ seq: 1, turnId: "t2", channelId: "c2" }), + makeEvent({ seq: 1, turnId: "t1-dup", channelId: "c1" }), ]); - const channels = channelIdsOf(getActiveTurnsForAgent(AGENT)); - assert.equal(channels.size, 1); - assert.ok(channels.has("c1")); + const turns = getActiveTurnsForAgent(AGENT); + assert.equal(turns.length, 1); + assert.ok(channelIdsOf(turns).has("c1")); }); }); @@ -631,6 +648,85 @@ describe("activeAgentTurnsStore", () => { assert.equal(notified, 0, "replayed evictions must not notify listeners"); }); + + it("cross-channel-delayed null-turnId turn_error evicts only its own channel's turn", () => { + // Sami's redteam target for gather-packing: channel frames may publish + // out of arrival order, so an OLDER turn_error on channel B can arrive + // AFTER newer events on channel A. Its null-turnId fallback must evict + // only B's turn — never A's live one — and it must not be skipped as + // stale either (B's own watermark hasn't seen it). + syncAgentTurnsFromEvents(AGENT, [ + // Channel A's frame arrives first, carrying the NEWER events. + makeEvent({ + seq: 10, + turnId: "t-a", + channelId: "c-a", + timestamp: "2024-01-01T00:01:00Z", + }), + ]); + syncAgentTurnsFromEvents(AGENT, [ + // Channel B's frame arrives second with OLDER events, in B's own + // FIFO order (the harness preserves within-channel order). + makeEvent({ + seq: 1, + turnId: "t-b", + channelId: "c-b", + timestamp: "2024-01-01T00:00:00Z", + }), + makeEvent({ + seq: 2, + kind: "turn_error", + turnId: null, + channelId: "c-b", + timestamp: "2024-01-01T00:00:01Z", + }), + ]); + + const channels = channelIdsOf(getActiveTurnsForAgent(AGENT)); + assert.ok( + channels.has("c-a"), + "the delayed channel's eviction must not kill the other channel's live turn", + ); + assert.ok(!channels.has("c-b"), "B's own errored turn must be evicted"); + }); + + it("null-channel events gate against their own per-agent bucket", () => { + // A null-channel agent_panic has no channel watermark; it lands in the + // dedicated null bucket. Replaying it must be a no-op, and it must not + // advance (or be gated by) any real channel's watermark. + const panic = makeEvent({ + seq: 20, + kind: "agent_panic", + turnId: null, + channelId: null, + timestamp: "2024-01-01T00:02:00Z", + }); + syncAgentTurnsFromEvents(AGENT, [panic]); + + // An older event on a real channel still processes — the panic's newer + // timestamp lives in the null bucket, not the channel's. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 1, + turnId: "t1", + channelId: "c1", + timestamp: "2024-01-01T00:00:00Z", + }), + ]); + assert.ok( + channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"), + "a null-channel event must not raise real channels' watermarks", + ); + + // Replaying the panic is gated by the null bucket: no notification. + let notified = 0; + const unsub = subscribeActiveAgentTurns(() => { + notified++; + }); + syncAgentTurnsFromEvents(AGENT, [panic]); + unsub(); + assert.equal(notified, 0, "replayed null-channel event must be a no-op"); + }); }); describe("getActiveTurnsForAgent", () => { @@ -1674,7 +1770,7 @@ describe("community-switch save / restore", () => { resetActiveAgentTurnsStore(); restoreActiveAgentTurnsForCommunity("ws-a"); - // Replaying an event at or below the watermark must be a no-op. + // Replaying an event at or below its channel's watermark must be a no-op. let notified = 0; const unsub = subscribeActiveAgentTurns(() => { notified++; @@ -1682,18 +1778,16 @@ describe("community-switch save / restore", () => { syncAgentTurnsFromEvents(AGENT, [ makeEvent({ seq: 3, - turnId: "t2", - channelId: "c2", + turnId: "t1-stale", + channelId: "c1", timestamp: "2024-01-01T00:00:00Z", }), ]); unsub(); assert.equal(notified, 0, "stale event after restore must not notify"); - const channels = new Set( - getActiveTurnsForAgent(AGENT).map((s) => s.channelId), - ); - assert.ok(!channels.has("c2"), "stale event must not add a turn"); + const turns = getActiveTurnsForAgent(AGENT); + assert.equal(turns.length, 1, "stale event must not add a turn"); }); it("terminal tombstones are preserved — a stale liveness cannot resurrect a completed turn", () => { diff --git a/desktop/src/features/agents/activeAgentTurnsStore.ts b/desktop/src/features/agents/activeAgentTurnsStore.ts index 3af41ae078..15789f2c3c 100644 --- a/desktop/src/features/agents/activeAgentTurnsStore.ts +++ b/desktop/src/features/agents/activeAgentTurnsStore.ts @@ -29,10 +29,19 @@ const PRUNE_PAUSE_MAX_MS = 3 * 60_000; * Desktop default of 24 — any lower value silently evicts a live turn, dropping * its working badge. */ const MAX_TURNS_PER_AGENT = 32; -/** Cap on per-agent terminal tombstones (A's resurrection guard). Only the - * most recently completed turns can be raced by a late liveness frame; older - * ones are already below the watermark, so a small multiple of the live cap is - * ample and keeps the map from growing across a long session. */ +/** Cap on per-agent terminal tombstones (A's resurrection guard). For + * terminals that carry the turn's channelId (turn_completed / turn_error), + * the (agent, channel) watermark already gates any liveness emitted before + * the terminal — both gate on the same channel key and within-channel order + * is preserved end-to-end — so those tombstones only matter briefly. The + * tombstones doing durable work are the ones whose creation the channel + * watermark does NOT witness: null-channel terminals (agent_panic gates on + * the null bucket) and desktop-initiated clears (`clearActiveTurnsForAgent`). + * Those are always the newest entries, so evicting the oldest past a small + * multiple of the live cap preserves them. Worst case for an evicted + * tombstone (a >128-completions-late liveness beating a quiet channel's + * watermark) is a ghost badge that the prune reaps within its normal bounds — + * bounded cosmetic staleness, not state corruption. */ const MAX_TERMINAL_TOMBSTONES = MAX_TURNS_PER_AGENT * 4; /** Interval for pruning stale/expired turns. */ const PRUNE_INTERVAL_MS = 5_000; @@ -85,11 +94,36 @@ const clockOffsetByAgent = new Map(); const cachedTurnSummaries = new Map(); let cachedChannelTurnSummaries: ActiveChannelTurnSummary[] | null = null; -// Composite watermark per agent: the newest observer event processed, by -// (timestamp, seq) ordering. An event is processed only if it is strictly -// newer than this — making full-buffer replays idempotent and post-restart -// streams (seq resets to 1, timestamp keeps climbing) handled for free. -const lastProcessed = new Map(); +// Composite watermark per (agent, channel): the newest observer event +// processed for that channel, by (timestamp, seq) ordering. An event is +// processed only if it is strictly newer than its channel's watermark — +// making full-buffer replays idempotent and post-restart streams (seq resets +// to 1, timestamp keeps climbing) handled for free. +// +// Keyed per channel, not per agent, because the harness's batching packer +// preserves FIFO order only WITHIN a channel: frames for different channels +// may publish in an order that differs from arrival order. A per-agent +// watermark would silently skip a delayed channel's events as stale. This is +// safe because every turn-mutating path is channel-scoped by the event's own +// channelId (endTurn's fallback matches `turn.channelId`, resurrectTurn keys +// on `event.channelId`), so per-channel serialization gives each channel +// exactly the protection per-agent serialization gave the whole agent. +// +// Events with a null channelId (agent_panic-class) gate against a dedicated +// per-agent bucket: the harness treats null-channel events as packing +// barriers, so they never reorder against any channel's events. +// +// The other per-agent maps stay agent-keyed because they are reorder-safe: +// `clockOffsetByAgent` is a running MINIMUM (order-insensitive by +// construction), and `activeTurnsByAgent` / `terminalAtByAgent` are mutated +// only through channel-scoped paths this gate serializes (see the +// MAX_TERMINAL_TOMBSTONES doc for the tombstone-eviction analysis). +const NULL_CHANNEL_KEY = "\u0000null-channel"; +const lastProcessed = new Map>(); + +function watermarkChannelKey(event: ObserverEvent): string { + return event.channelId ?? NULL_CHANNEL_KEY; +} // Per-agent record of when each turn terminally ended (turnId → // terminal-event timestamp, in agent-host clock ms). endTurn hard-deletes a @@ -324,21 +358,36 @@ function pruneExpired() { function processEvent(agentPubkey: string, event: ObserverEvent) { const key = normalizePubkey(agentPubkey); - // Gate every event kind on the watermark uniformly: process only events - // strictly newer than the last one seen for this agent. With sorted buffers - // (the documented invariant), this makes full-buffer replays a complete - // no-op. Evictions must be gated too — replaying a stale turn_error/ - // agent_panic (emitted with a null turnId) would otherwise fall back to - // deleting the first turn in the channel, killing the live turn. Resurrection + // Gate every event kind on its (agent, channel) watermark uniformly: + // process only events strictly newer than the last one seen for this + // agent+channel. With sorted buffers (the documented invariant), this makes + // full-buffer replays a complete no-op. Evictions must be gated too — + // replaying a stale turn_error/agent_panic (emitted with a null turnId) + // would otherwise fall back to deleting the first turn in the channel, + // killing the live turn; the gate is per-channel, but so is that fallback + // (it matches the event's own channelId), so each channel's stream is + // serialized against exactly the mutations it can perform. Resurrection // (the turn_liveness/acp case below) is gated here too: it runs only for a - // frame that passes the watermark, so replayed stale frames cannot revive a - // pruned turn, and the per-turn terminal tombstone blocks reviving a turn - // that already completed. - const last = lastProcessed.get(key); + // frame that passes its channel's watermark, so replayed stale frames + // cannot revive a pruned turn, and the per-turn terminal tombstone blocks + // reviving a turn that already completed. + // + // The per-agent map grows one entry per channel the agent has ever emitted + // on — bounded by real channel membership. Deliberately NOT capped: + // evicting a channel's watermark would reopen replay-processing for that + // channel, and a re-run stale eviction is exactly the badge-killing hazard + // this gate exists to prevent. + const channelKey = watermarkChannelKey(event); + let agentWatermarks = lastProcessed.get(key); + const last = agentWatermarks?.get(channelKey); if (last && compareObserverEvents(event, last) <= 0) { return; } - lastProcessed.set(key, event); + if (!agentWatermarks) { + agentWatermarks = new Map(); + lastProcessed.set(key, agentWatermarks); + } + agentWatermarks.set(channelKey, event); // Refine the clock offset from every fresh event. A tighter offset shifts // every live anchor for this agent, so a change must reach the UI even when @@ -630,7 +679,7 @@ export function resetActiveAgentTurnsStore() { type TurnsStoreSnapshot = { turns: Map>; offsets: Map; - watermarks: Map; + watermarks: Map>; terminals: Map>; }; @@ -663,9 +712,15 @@ export function saveActiveAgentTurnsForCommunity(communityId: string): void { turns.set(agentKey, clonedAgent); } - // Shallow-clone scalar maps (primitives as values). + // Shallow-clone the offsets map (primitives as values). const offsets = new Map(clockOffsetByAgent); - const watermarks = new Map(lastProcessed); + + // Deep-clone the per-(agent, channel) watermark map: outer map + inner + // per-agent maps (ObserverEvent values are treated as immutable). + const watermarks = new Map>(); + for (const [agentKey, channelMarks] of lastProcessed) { + watermarks.set(agentKey, new Map(channelMarks)); + } // Deep-clone terminalAtByAgent: outer map + inner per-agent maps. const terminals = new Map>(); @@ -719,8 +774,8 @@ export function restoreActiveAgentTurnsForCommunity(communityId: string): void { clockOffsetByAgent.set(agentKey, offset); } - for (const [agentKey, event] of snap.watermarks) { - lastProcessed.set(agentKey, event); + for (const [agentKey, channelMarks] of snap.watermarks) { + lastProcessed.set(agentKey, new Map(channelMarks)); } for (const [agentKey, tombstones] of snap.terminals) { From cc9333b7cfdd75f9f85a4e558aeaecac547af55f Mon Sep 17 00:00:00 2001 From: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 18:21:35 -0400 Subject: [PATCH 4/6] observer batching: count merged chunk sources in drop accounting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Max's round-3 blocker: a PendingObserverChunk that merged N same-messageId chunks counted as 1 in dropped_events when byte-budget eviction removed it — his probe (50 merged 1KB chunks evicted under distinct-key pressure) read dropped_events == 1 while 50 generated observer events vanished from the accounting, so published + dropped could not reconcile with ingested. dropped_events is now denominated in SOURCE (generated) observer events end to end: - PendingObserverChunk carries source_events (1 at creation, +1 per absorbed chunk); drop_oldest returns it and eviction charges it. - The count survives flush: coalescer ingest/flush return (source_events, event) pairs and the queue FIFO stores the count per entry, so a flushed-but-unpublished merged entry still accounts for every source on eviction. - Published side of the ledger: a published merged entry delivers all N sources' text in ONE event, so the invariant is ingested == dropped_events + sum(source_events over published events) (documented on the field). New regression per Max's probe shape: 50 x 1KB chunks merged under one messageId, then a distinct-key 50KB flood forcing eviction of the merged entry — accounting must close exactly (all 50 sources charged). Verified the test bites: reverting the coalescer eviction to += 1 fails it (left: 101, right: 150). Co-authored-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Signed-off-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 164 +++++++++++++++++++++++++++++-------- 1 file changed, 128 insertions(+), 36 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 6e758598d7..0243a4615a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -421,10 +421,18 @@ const OBSERVER_BATCH_KIND: &str = "batch"; #[derive(Default)] struct ObserverPublishQueue { coalescer: ObserverChunkCoalescer, - /// `(serialized_len, event)`, oldest first. Length is captured at enqueue - /// (post-fit) so byte accounting never re-serializes on eviction. - events: VecDeque<(usize, observer::ObserverEvent)>, + /// `(serialized_len, source_events, event)`, oldest first. Length is + /// captured at enqueue (post-fit) so byte accounting never re-serializes + /// on eviction; `source_events` is how many GENERATED observer events the + /// entry represents (a merged chunk carries every chunk it absorbed), so + /// eviction accounting stays in source units after flush. + events: VecDeque<(usize, u64, observer::ObserverEvent)>, pending_bytes: usize, + /// SOURCE observer events lost to byte-budget eviction. Counted in + /// generated-event units, not retained entries: a coalesced entry that + /// merged N chunks accounts for N when evicted. A PUBLISHED merged entry + /// delivers all N sources' text in one event, so the invariant is + /// `ingested == dropped_events + Σ source_events over published events`. dropped_events: u64, } @@ -433,21 +441,22 @@ impl ObserverPublishQueue { // ObserverChunkCoalescer::ingest returns immediately-publishable events // (force-flushed pending chunks + non-chunk passthrough, or a pending // set displaced by the 60KB pre-flush); they join the queue in the - // order the coalescer emitted them. - for ready in self.coalescer.ingest(event) { - self.enqueue(ready); + // order the coalescer emitted them, each carrying the count of source + // events it represents. + for (source_events, ready) in self.coalescer.ingest(event) { + self.enqueue(source_events, ready); } self.enforce_byte_budget(); } - fn enqueue(&mut self, mut event: observer::ObserverEvent) { + fn enqueue(&mut self, source_events: u64, mut event: observer::ObserverEvent) { // 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); self.pending_bytes += bytes; - self.events.push_back((bytes, event)); + self.events.push_back((bytes, source_events, event)); } /// Total bytes retained across BOTH stores — the event FIFO and the @@ -459,23 +468,23 @@ impl ObserverPublishQueue { } /// Enforce [`OBSERVER_PENDING_QUEUE_MAX_BYTES`] over the total, dropping - /// OLDEST items first with accounting. Global age order across the two - /// stores is structural: every enqueue path flushes the coalescer first, - /// so every pending coalescer entry is strictly newer than every queued - /// event — eviction is queue front, then coalescer front. The `> 1` - /// guard never drops the sole remaining item (any single fitted event or - /// pre-flush-capped chunk entry is far under the budget). + /// OLDEST items first with accounting in SOURCE-event units. Global age + /// order across the two stores is structural: every enqueue path flushes + /// the coalescer first, so every pending coalescer entry is strictly newer + /// than every queued event — eviction is queue front, then coalescer + /// front. The `> 1` guard never drops the sole remaining item (any single + /// fitted event or pre-flush-capped chunk entry is far under the budget). fn enforce_byte_budget(&mut self) { let mut dropped = 0u64; while self.total_pending_bytes() > OBSERVER_PENDING_QUEUE_MAX_BYTES && self.events.len() + self.coalescer.pending.len() > 1 { - if let Some((bytes, _)) = self.events.pop_front() { + if let Some((bytes, source_events, _)) = self.events.pop_front() { self.pending_bytes -= bytes; + dropped += source_events; } else { - self.coalescer.drop_oldest().expect("guard ensures an item"); + dropped += self.coalescer.drop_oldest().expect("guard ensures an item"); } - dropped += 1; } if dropped > 0 { self.dropped_events += dropped; @@ -522,16 +531,16 @@ impl ObserverPublishQueue { /// Pending coalesced chunks are flushed into the queue first, so a /// publish slot never leaves merged chunk text stranded behind the tick. fn next_frame(&mut self) -> Option { - for ready in self.coalescer.flush() { - self.enqueue(ready); + for (source_events, ready) in self.coalescer.flush() { + self.enqueue(source_events, ready); } - let channel = self.events.front()?.1.channel_id.clone(); + let channel = self.events.front()?.2.channel_id.clone(); let mut picked: Vec = Vec::new(); - let mut kept: VecDeque<(usize, observer::ObserverEvent)> = + let mut kept: VecDeque<(usize, u64, observer::ObserverEvent)> = VecDeque::with_capacity(self.events.len()); let mut gathering = true; - while let Some((bytes, event)) = self.events.pop_front() { + while let Some((bytes, source_events, event)) = self.events.pop_front() { if gathering && event.channel_id == channel { picked.push(event); if picked.len() > 1 @@ -540,7 +549,7 @@ impl ObserverPublishQueue { // Frame full: the overflow event stays queued and leads // its channel's next slot. let event = picked.pop().expect("len > 1"); - kept.push_back((bytes, event)); + kept.push_back((bytes, source_events, event)); gathering = false; } else { self.pending_bytes -= bytes; @@ -551,7 +560,7 @@ impl ObserverPublishQueue { // end of its contiguous front run): stop gathering. gathering = false; } - kept.push_back((bytes, event)); + kept.push_back((bytes, source_events, event)); } } self.events = kept; @@ -702,6 +711,10 @@ struct PendingObserverChunk { text: String, /// Bytes this entry contributes to `pending_bytes`. bytes: usize, + /// GENERATED observer events merged into this entry (1 at creation, +1 + /// per absorbed chunk). Evicting the entry loses this many source events, + /// so drop accounting must charge this count, not 1. + source_events: u64, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -722,10 +735,13 @@ struct ObserverChunkKey { const OBSERVER_CHUNK_MAX_TEXT_BYTES: usize = 60_000; impl ObserverChunkCoalescer { - fn ingest(&mut self, event: observer::ObserverEvent) -> Vec { + /// Returns immediately-publishable events, each paired with the number of + /// SOURCE observer events it represents (merged chunks carry the count of + /// every chunk they absorbed; passthrough events are always 1). + fn ingest(&mut self, event: observer::ObserverEvent) -> Vec<(u64, observer::ObserverEvent)> { let Some((key, text)) = observer_chunk_key_and_text(&event) else { let mut events = self.flush(); - events.push(event); + events.push((1, event)); return events; }; @@ -739,6 +755,7 @@ impl ObserverChunkCoalescer { } pending.text.push_str(&text); pending.bytes += text.len(); + pending.source_events += 1; self.pending_bytes += text.len(); pending.event.seq = event.seq; pending.event.timestamp = event.timestamp; @@ -766,27 +783,29 @@ impl ObserverChunkCoalescer { event, text, bytes, + source_events: 1, }); } /// Evict the OLDEST pending entry for byte-budget enforcement. Returns - /// `None` when there is nothing to drop. - fn drop_oldest(&mut self) -> Option<()> { + /// the number of SOURCE events the entry represented (its merged chunk + /// count), or `None` when there is nothing to drop. + fn drop_oldest(&mut self) -> Option { if self.pending.is_empty() { return None; } let removed = self.pending.remove(0); self.pending_bytes -= removed.bytes; - Some(()) + Some(removed.source_events) } - fn flush(&mut self) -> Vec { + fn flush(&mut self) -> Vec<(u64, observer::ObserverEvent)> { self.pending_bytes = 0; self.pending .drain(..) .map(|mut pending| { set_observer_chunk_text(&mut pending.event.payload, pending.text); - pending.event + (pending.source_events, pending.event) }) .collect() } @@ -5544,6 +5563,74 @@ mod observer_publish_queue_tests { assert_eq!(*last_frame_seqs.last().expect("seqs"), total); } + /// Max's merged-chunk accounting regression: one coalescer entry can + /// represent MANY generated observer events (same-messageId chunks merge + /// in place), so evicting it must charge every merged source event to + /// `dropped_events`, not 1 per retained entry. Pre-fix, evicting an entry + /// that merged 50 chunks recorded `dropped_events == 1` and 49 generated + /// events vanished from the accounting. + #[test] + fn evicting_a_merged_chunk_entry_accounts_every_source_event() { + fn chunk(seq: u64, message_id: &str, text: &str) -> observer::ObserverEvent { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "session-1", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": message_id, + "content": { "type": "text", "text": text }, + }, + }, + }); + e + } + + let mut queue = ObserverPublishQueue::default(); + // 50 × 1KB chunks under ONE messageId merge into a single pending + // coalescer entry — the oldest item anywhere in the queue. + let merged_text = "m".repeat(1_000); + let merged_sources = 50u64; + for seq in 1..=merged_sources { + queue.ingest(chunk(seq, "message-merged", &merged_text)); + } + // Flood with distinct-key 50KB chunks until the byte budget evicts + // the oldest entries — the merged entry goes first. + let flood_text = "f".repeat(50_000); + let flood = 100u64; + for seq in 1..=flood { + queue.ingest(chunk( + merged_sources + seq, + &format!("message-{seq}"), + &flood_text, + )); + } + + assert!( + queue.total_pending_bytes() <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + "eviction must restore the byte budget" + ); + let frames = drain_frames(&mut queue); + assert!( + !frames + .iter() + .flat_map(frame_seqs) + .any(|seq| seq <= merged_sources), + "the merged entry (globally oldest) must have been evicted" + ); + // Every survivor is an unmerged distinct-key chunk (1 source each), + // so source-event accounting must close exactly: the merged entry's + // eviction charges all 50 sources. + let survived: u64 = frames.iter().map(|f| frame_seqs(f).len() as u64).sum(); + assert_eq!( + survived + queue.dropped_events, + merged_sources + flood, + "accounting: published sources + dropped sources == ingested" + ); + } + /// Under the byte budget the queue is lossless: every ingested event /// publishes exactly once. #[test] @@ -5871,9 +5958,14 @@ mod observer_chunk_coalescer_tests { let events = coalescer.ingest(non_chunk_event(3)); assert_eq!(events.len(), 2); - assert_eq!(events[0].seq, 2); - assert_eq!(chunk_text(&events[0]), "hello world"); - assert_eq!(events[1].kind, "turn_started"); + assert_eq!(events[0].1.seq, 2); + assert_eq!(chunk_text(&events[0].1), "hello world"); + assert_eq!( + events[0].0, 2, + "a merged entry reports every source chunk it absorbed" + ); + assert_eq!(events[1].1.kind, "turn_started"); + assert_eq!(events[1].0, 1); } #[test] @@ -5894,8 +5986,8 @@ mod observer_chunk_coalescer_tests { let events = coalescer.flush(); assert_eq!(events.len(), 2); - assert_eq!(chunk_text(&events[0]), "answer"); - assert_eq!(chunk_text(&events[1]), "thinking"); + assert_eq!(chunk_text(&events[0].1), "answer"); + assert_eq!(chunk_text(&events[1].1), "thinking"); } } From fa529ac9945bed2ce34dd996b258be563ba7d1c1 Mon Sep 17 00:00:00 2001 From: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 18:37:05 -0400 Subject: [PATCH 5/6] observer batching: true coalescer byte accounting + eviction-arm coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-4 review items (Sami round-3 re-review + Max's independent reproductions), all tests-and-arithmetic — no behavioural change to the pacer, gather, barriers, or watermark: 1. Coalescer byte counter undercounted ~2x (Sami blocker, Max confirmed): a pending entry retains its first chunk's text TWICE until flush — in the serialized skeleton (event.payload still carries it) and as the extracted `text` copy — but push_pending charged only serialized_len. Each entry omitted exactly its first text.len() from the budget, so a high-cardinality 50KB flood retained 8.33 MiB against the 4 MiB cap (1.99x) while the accumulator read under cap. Fix: initialize bytes = serialized_len(&event) + text.len(). 2. Cap regressions now assert on walked_retained_bytes() — retained bytes computed by walking the entries — never on total_pending_bytes(): asserting the accumulator against itself is how 1.99x overshoot passed. The distinct-key regression also pins accumulator >= walked so the counter can never under-count true retention again. 3. Sami's M13 (queue-side `dropped += source_events` -> `+= 1` survived all 687 tests): new regression drives a merged 50-chunk entry THROUGH a forced flush into the publish FIFO, then evicts it under distinct-key pressure. Verified it bites: the M13 mutation fails it 102 vs 151 — exactly Sami's and Max's probe numbers. 4. Desktop save-side snapshot aliasing (Sami gap): deleting the watermark deep-clone in saveActiveAgentTurnsForCommunity survived all 4363 desktop tests. New test advances the live watermark after save and asserts a post-restore event fresh vs the SAVED watermark still processes. Verified it kills the aliasing mutant and passes clean. (Restore-side clone is a genuinely equivalent mutant — snapshot is consumed on restore, no second owner.) Mutation receipts (each mutant applied in-place, test observed failing, clean code restored): byte-fix revert -> both walked-bytes regressions fail (8,328,386 vs 4 MiB cap); M13 -> FIFO-arm regression fails 102/151; save-clone delete -> aliasing test fails. Co-authored-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Signed-off-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 144 ++++++++++++++++-- .../agents/activeAgentTurnsStore.test.mjs | 51 +++++++ 2 files changed, 180 insertions(+), 15 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 0243a4615a..95ad64e497 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -772,11 +772,13 @@ impl ObserverChunkCoalescer { event: observer::ObserverEvent, text: String, ) { - // The serialized skeleton already carries this entry's first chunk - // text (`text` is an extracted copy, later written back at flush), so - // the entry starts at its full serialized weight and appends add - // exactly their text length. - let bytes = serialized_len(&event); + // The entry RETAINS the first chunk's text twice until flush: once + // inside the serialized skeleton (`event.payload` still carries it) + // and once as the extracted `text` copy that appends grow. Both are + // real memory, so both count — charging only `serialized_len` lets a + // high-cardinality flood retain up to 2x the byte budget (each entry + // undercounts by exactly its first chunk's length). + let bytes = serialized_len(&event) + text.len(); self.pending_bytes += bytes; self.pending.push(PendingObserverChunk { key, @@ -5236,6 +5238,27 @@ mod observer_publish_queue_tests { } } + /// Retained bytes computed by WALKING the entries, independently of the + /// queue's own accumulator. Cap regressions must assert on this, not on + /// `total_pending_bytes()` — asserting the counter against itself passed + /// while the process retained ~2x the budget (Sami/Max round 3: each + /// pending coalescer entry holds the first chunk's text twice, in the + /// serialized skeleton AND the extracted `text` copy). + fn walked_retained_bytes(queue: &ObserverPublishQueue) -> usize { + let fifo: usize = queue + .events + .iter() + .map(|(_, _, event)| serialized_len(event)) + .sum(); + let coalescer: usize = queue + .coalescer + .pending + .iter() + .map(|pending| serialized_len(&pending.event) + pending.text.len()) + .sum(); + fifo + coalescer + } + /// Two or more pending events for one channel ship as a single batch /// envelope whose payload carries every inner event in arrival order. #[test] @@ -5493,8 +5516,9 @@ mod observer_publish_queue_tests { "a 5MB backlog must overflow the 4MiB budget" ); assert!( - queue.total_pending_bytes() <= OBSERVER_PENDING_QUEUE_MAX_BYTES, - "eviction must restore the byte budget" + walked_retained_bytes(&queue) <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + "eviction must restore the byte budget (entry-walked), got {}", + walked_retained_bytes(&queue) ); let frames = drain_frames(&mut queue); @@ -5514,9 +5538,12 @@ mod observer_publish_queue_tests { /// Max's coalescer-bypass regression: a flood of chunks with DISTINCT /// messageIds never flushes on its own, so every chunk sits in the - /// coalescer's pending buffer. Those bytes MUST count against the byte - /// budget and be evicted with accounting — pre-fix this retained ~25MB - /// against the 4 MiB cap with `pending_bytes == 0` and zero drops. + /// coalescer's pending buffer. TRUE retained bytes — walked from the + /// entries, never the queue's own accumulator — MUST respect the byte + /// budget with event-level drop accounting. Pre-fix this retained ~25MB + /// against the 4 MiB cap with `pending_bytes == 0` and zero drops; the + /// round-3 refinement (Sami/Max) caught the accumulator itself reading + /// under cap while true retention was 1.99x over. #[test] fn distinct_key_chunk_floods_are_bounded_by_the_byte_budget() { let big_text = "z".repeat(50_000); @@ -5539,10 +5566,16 @@ mod observer_publish_queue_tests { queue.ingest(e); } + let walked = walked_retained_bytes(&queue); + assert!( + walked <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + "TRUE retained bytes (walked from entries) must respect the cap, \ + got {walked}" + ); assert!( - queue.total_pending_bytes() <= OBSERVER_PENDING_QUEUE_MAX_BYTES, - "total retained bytes (FIFO + coalescer pending) must respect the \ - cap, got {}", + queue.total_pending_bytes() >= walked, + "the accumulator must never under-count true retention \ + (accumulator {} < walked {walked})", queue.total_pending_bytes() ); assert!( @@ -5609,8 +5642,9 @@ mod observer_publish_queue_tests { } assert!( - queue.total_pending_bytes() <= OBSERVER_PENDING_QUEUE_MAX_BYTES, - "eviction must restore the byte budget" + walked_retained_bytes(&queue) <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + "eviction must restore the byte budget (entry-walked), got {}", + walked_retained_bytes(&queue) ); let frames = drain_frames(&mut queue); assert!( @@ -5631,6 +5665,86 @@ mod observer_publish_queue_tests { ); } + /// Sami's M13 / Max's forced-flush probe: the OTHER eviction arm. A + /// merged entry FLUSHED into the publish FIFO (by a non-chunk event) must + /// still charge every absorbed source on eviction — the FIFO stores the + /// per-entry count precisely so the ledger survives flush. The + /// coalescer-side regression above never exercises this arm; mutating the + /// FIFO eviction to `dropped += 1` survived all 687 tests until this one. + #[test] + fn evicting_a_flushed_merged_entry_from_the_fifo_accounts_every_source_event() { + fn chunk(seq: u64, message_id: &str, text: &str) -> observer::ObserverEvent { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "session-1", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": message_id, + "content": { "type": "text", "text": text }, + }, + }, + }); + e + } + + let mut queue = ObserverPublishQueue::default(); + // 50 × 1KB chunks merge under one messageId in the coalescer… + let merged_text = "m".repeat(1_000); + let merged_sources = 50u64; + for seq in 1..=merged_sources { + queue.ingest(chunk(seq, "message-merged", &merged_text)); + } + // …then a non-chunk event force-flushes the merged entry into the + // publish FIFO. From here eviction happens on the FIFO arm. + queue.ingest(event(merged_sources + 1, "tool_call", Some("chan-a"))); + assert!( + queue.coalescer.pending.is_empty(), + "the non-chunk event must have flushed the merged entry" + ); + assert_eq!( + queue.events.front().expect("flushed entry queued").1, + merged_sources, + "the FIFO front must carry the merged source count" + ); + + // Distinct-key flood forces byte-budget eviction of the FIFO front. + let flood_text = "f".repeat(50_000); + let flood = 100u64; + for seq in 1..=flood { + queue.ingest(chunk( + merged_sources + 1 + seq, + &format!("message-{seq}"), + &flood_text, + )); + } + + assert!( + walked_retained_bytes(&queue) <= OBSERVER_PENDING_QUEUE_MAX_BYTES, + "eviction must restore the byte budget (entry-walked), got {}", + walked_retained_bytes(&queue) + ); + let frames = drain_frames(&mut queue); + assert!( + !frames + .iter() + .flat_map(frame_seqs) + .any(|seq| seq <= merged_sources), + "the flushed merged entry (globally oldest) must have been evicted" + ); + // Ledger in source units: survivors are unmerged (1 source each), the + // evicted merged FIFO entry must charge all 50 sources. + let survived: u64 = frames.iter().map(|f| frame_seqs(f).len() as u64).sum(); + let ingested = merged_sources + 1 + flood; + assert_eq!( + survived + queue.dropped_events, + ingested, + "accounting: published sources + dropped sources == ingested" + ); + } + /// Under the byte budget the queue is lossless: every ingested event /// publishes exactly once. #[test] diff --git a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs index 73d83a1a4f..208fa42029 100644 --- a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs +++ b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs @@ -1790,6 +1790,57 @@ describe("community-switch save / restore", () => { assert.equal(turns.length, 1, "stale event must not add a turn"); }); + it("snapshot watermarks are isolated — live advances after save must not leak into the snapshot", () => { + // Save with the channel watermark at seq 5. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 5, + turnId: "t1", + channelId: "c1", + timestamp: "2024-01-01T00:00:00Z", + }), + ]); + saveActiveAgentTurnsForCommunity("ws-a"); + + // Keep working in the live store AFTER the save: the watermark advances + // to seq 10 by mutating the same inner per-agent map the snapshot cloned. + // If save aliased instead of deep-cloning, this leaks into the snapshot. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 10, + turnId: "t10", + channelId: "c1", + timestamp: "2024-01-01T00:00:20Z", + }), + ]); + + resetActiveAgentTurnsStore(); + restoreActiveAgentTurnsForCommunity("ws-a"); + + // seq 7 is FRESH relative to the saved watermark (5) and must be + // processed — processing notifies listeners (stale events do not, per + // the gate). Under save-side aliasing the leaked watermark (10) wrongly + // skips it as stale and no notification fires. + let notified = 0; + const unsub = subscribeActiveAgentTurns(() => { + notified++; + }); + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 7, + turnId: "t7", + channelId: "c1", + timestamp: "2024-01-01T00:00:10Z", + }), + ]); + unsub(); + assert.ok( + notified > 0, + "event fresh vs the SAVED watermark must be processed — a live " + + "watermark advancing after save must not leak into the snapshot", + ); + }); + it("terminal tombstones are preserved — a stale liveness cannot resurrect a completed turn", () => { // Complete a turn before saving. syncAgentTurnsFromEvents(AGENT, [ From 63d821620d3513505e8766ac691a8002f9d4a96f Mon Sep 17 00:00:00 2001 From: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 19:13:12 -0400 Subject: [PATCH 6/6] observer batching: pin the walker instrument + two pre-existing snapshot-clone gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-5 review items (Sami, 9/9/9) — tests only, no production change: 1. walked_retained_bytes() was itself unverified: every cap test asks it only <= CAP, so a blinded walker (missing the text-copy arm, missing the FIFO arm, or returning literally 0) satisfied all of them — and pairing a blinded walker with a reverted push_pending fix cancelled out, hiding exactly the 8.3 MB overshoot it exists to detect (Sami's M17/M18/M19/M20 all survived 688 tests). New two-sided pin: the walker must SEE the first chunk's text twice, and must agree with the accumulator EXACTLY while both stores are non-empty. Kills all four. 2. Desktop snapshot clone family (both pre-existing at merge-base 7334ad1e1; this PR only made the class visible): aliasing the inner turns map leaks a turn started after save into the snapshot (D5b); aliasing the inner tombstones map leaks a post-save terminal that then blocks a legitimate post-restore resurrection (D5c). Two isolation tests, each with an in-test control, one per clone site — all three inner-map clones in saveActiveAgentTurnsForCommunity are now pinned. Mutation receipts (each applied in-place with a single-anchor edit, test observed failing, file restored byte-identical to the pinned copy): M17/M19/M20/M18 -> walker-agreement test fails; D5b -> turns-isolation test fails; D5c -> tombstone-isolation test fails. Co-authored-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Signed-off-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 55 +++++++++++ .../agents/activeAgentTurnsStore.test.mjs | 97 +++++++++++++++++++ 2 files changed, 152 insertions(+) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 95ad64e497..65c9dd6203 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -5259,6 +5259,61 @@ mod observer_publish_queue_tests { fifo + coalescer } + /// The walker above is itself an instrument, and every cap test asks it + /// only for `<= CAP` — a blinded walker (missing an arm, or returning 0) + /// would satisfy all of them while hiding exactly the 2x overshoot it was + /// added to catch (Sami round 5, M17-M20). Pin it two-sided: it must SEE + /// the double retention, and it must agree with the accumulator EXACTLY + /// while both stores are non-empty — neither may drift. + #[test] + fn walked_retained_bytes_agrees_with_the_accumulator_exactly() { + fn chunk(seq: u64, message_id: &str, text: &str) -> observer::ObserverEvent { + let mut e = event(seq, "acp_read", Some("chan-a")); + e.payload = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "session-1", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": message_id, + "content": { "type": "text", "text": text }, + }, + }, + }); + e + } + + let text = "w".repeat(7_000); + let mut queue = ObserverPublishQueue::default(); + // One pending chunk: its text lives in the serialized skeleton AND + // the extracted copy, so a walker blind to either arm reads short. + queue.ingest(chunk(1, "message-a", &text)); + assert!( + walked_retained_bytes(&queue) >= 2 * text.len(), + "the walker must SEE the first chunk's text twice \ + (skeleton + extracted copy), got {}", + walked_retained_bytes(&queue) + ); + + // Populate BOTH stores: the non-chunk event flushes message-a into + // the FIFO and queues itself; fresh pending keys (plus a same-key + // append) rebuild the coalescer side. + queue.ingest(event(2, "tool_call", Some("chan-a"))); + queue.ingest(chunk(3, "message-b", &text)); + queue.ingest(chunk(4, "message-b", &text)); + queue.ingest(chunk(5, "message-c", &text)); + assert!( + !queue.events.is_empty() && !queue.coalescer.pending.is_empty(), + "both arms must be non-empty for the agreement check to bind" + ); + assert_eq!( + queue.total_pending_bytes(), + walked_retained_bytes(&queue), + "accumulator and entry-walk must agree exactly: neither may drift" + ); + } + /// Two or more pending events for one channel ship as a single batch /// envelope whose payload carries every inner event in arrival order. #[test] diff --git a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs index 208fa42029..e3e3b3e756 100644 --- a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs +++ b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs @@ -1841,6 +1841,103 @@ describe("community-switch save / restore", () => { ); }); + it("snapshot turns are isolated — a turn started after save must not leak into the snapshot", () => { + // Save with one live turn on c1. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }), + ]); + saveActiveAgentTurnsForCommunity("ws-a"); + + // Keep working AFTER the save: a new turn on c2 mutates the same inner + // per-agent turns map the snapshot cloned. If save aliased that inner + // map instead of deep-cloning it, t2 leaks into the snapshot. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ seq: 2, turnId: "t2", channelId: "c2" }), + ]); + + resetActiveAgentTurnsStore(); + restoreActiveAgentTurnsForCommunity("ws-a"); + + const channels = channelIdsOf(getActiveTurnsForAgent(AGENT)); + assert.ok(channels.has("c1"), "the saved turn must restore"); + assert.ok( + !channels.has("c2"), + "a turn started after the save must NOT appear in the restored snapshot", + ); + }); + + it("snapshot tombstones are isolated — a terminal recorded after save must not block a legitimate post-restore resurrection", () => { + // Complete t1 on c1 before saving: the snapshot legitimately carries its + // tombstone (terminal at 00:00:10). + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 1, + turnId: "t1", + channelId: "c1", + timestamp: "2024-01-01T00:00:00Z", + }), + makeEvent({ + seq: 2, + kind: "turn_completed", + turnId: "t1", + channelId: "c1", + timestamp: "2024-01-01T00:00:10Z", + }), + ]); + saveActiveAgentTurnsForCommunity("ws-a"); + + // AFTER the save, t2 on c2 starts and completes (terminal at 00:00:30). + // That records a tombstone in the same inner per-agent tombstone map the + // snapshot cloned — if save aliased it, t2's tombstone leaks in. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 3, + turnId: "t2", + channelId: "c2", + timestamp: "2024-01-01T00:00:20Z", + }), + makeEvent({ + seq: 4, + kind: "turn_completed", + turnId: "t2", + channelId: "c2", + timestamp: "2024-01-01T00:00:30Z", + }), + ]); + + resetActiveAgentTurnsStore(); + restoreActiveAgentTurnsForCommunity("ws-a"); + + // Recovered liveness frames: t1's is strictly newer than its (snapshot) + // terminal so it revives either way — the control. t2 has NO tombstone + // in a properly isolated snapshot, so its frame (00:00:25, before its + // live-store terminal at 00:00:30) must also revive; a leaked tombstone + // wrongly blocks exactly this legitimate resurrection. + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 5, + kind: "turn_liveness", + turnId: "t1", + channelId: "c1", + timestamp: "2024-01-01T00:00:15Z", + }), + makeEvent({ + seq: 6, + kind: "turn_liveness", + turnId: "t2", + channelId: "c2", + timestamp: "2024-01-01T00:00:25Z", + }), + ]); + const channels = channelIdsOf(getActiveTurnsForAgent(AGENT)); + assert.ok(channels.has("c1"), "control: t1 revives past its own terminal"); + assert.ok( + channels.has("c2"), + "a tombstone recorded after the save must NOT leak into the snapshot " + + "and block a legitimate resurrection", + ); + }); + it("terminal tombstones are preserved — a stale liveness cannot resurrect a completed turn", () => { // Complete a turn before saving. syncAgentTurnsFromEvents(AGENT, [