diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 0c4e5f158c..65c9dd6203 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -361,50 +361,242 @@ 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 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 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. +/// +/// 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"; -struct ObserverPublishPacer { - next_publish: tokio::time::Instant, - published: VecDeque, +/// 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 ObserverPublishQueue { + coalescer: ObserverChunkCoalescer, + /// `(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, } -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 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 queue in the + // 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(); } - 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)) - { - self.published.pop_front(); + 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, source_events, 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 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, source_events, _)) = self.events.pop_front() { + self.pending_bytes -= bytes; + dropped += source_events; + } else { + dropped += self.coalescer.drop_oldest().expect("guard ensures an item"); } + } + if dropped > 0 { + self.dropped_events += dropped; + tracing::warn!( + dropped, + total_dropped = self.dropped_events, + pending_bytes = self.total_pending_bytes(), + "observer publish queue over byte budget; dropped oldest events" + ); + } + } - 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; - } + /// 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() + } - let published_at = tokio::time::Instant::now(); - self.published.push_back(published_at); - self.next_publish = published_at + OBSERVER_PUBLISH_INTERVAL; - return; + /// 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. + /// + /// 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. + fn next_frame(&mut self) -> Option { + for (source_events, ready) in self.coalescer.flush() { + self.enqueue(source_events, ready); } + let channel = self.events.front()?.2.channel_id.clone(); + + let mut picked: Vec = Vec::new(); + let mut kept: VecDeque<(usize, u64, observer::ObserverEvent)> = + VecDeque::with_capacity(self.events.len()); + let mut gathering = true; + while let Some((bytes, source_events, 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, source_events, 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, source_events, event)); + } + } + self.events = kept; + Some(seal_batch(picked)) + } +} + +/// 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) +} + +/// 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,29 +637,26 @@ 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 queue = ObserverPublishQueue::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; - } + queue.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); + // 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 @@ -475,41 +664,30 @@ 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; - } + queue.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() { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, 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; } } } - _ = flush_interval.tick() => { - // Periodic flush ensures live streaming even during continuous chunk delivery. - for event in coalescer.flush() { + _ = publish_tick.tick() => { + if let Some(frame) = queue.next_frame() { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + &owner_pubkey_hex, &owner_pubkey, frame, ).await; } + if closed && queue.is_empty() { + break; + } } } } @@ -518,12 +696,25 @@ 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, + /// 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)] @@ -544,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; }; @@ -556,25 +750,64 @@ 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(); + pending.source_events += 1; + 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 flush(&mut self) -> Vec { + fn push_pending( + &mut self, + key: ObserverChunkKey, + event: observer::ObserverEvent, + text: String, + ) { + // 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, + event, + text, + bytes, + source_events: 1, + }); + } + + /// Evict the OLDEST pending entry for byte-budget enforcement. Returns + /// 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(removed.source_events) + } + + 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() } @@ -793,10 +1026,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 +5170,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 +5195,865 @@ mod observer_snapshot_race_tests { } #[cfg(test)] -mod observer_publish_pacer_tests { +mod observer_publish_queue_tests { use super::*; + 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 }), + } + } + + 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], + } + } + + /// 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 + } + + /// 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] + fn multiple_events_ship_as_one_envelope_in_order() { + 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")), + ]); + + 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[1]["kind"], "acp_read", "inner events keep their kind"); + } + + /// 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 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); + } + + /// An empty queue yields no frame — a tick with nothing pending must not + /// publish anything. + #[test] + 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 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_gather_queue_wide() { + let mut queue = queue_of(vec![ + event(1, "acp_read", Some("chan-a")), + 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, + "gathered: [1,2,4]@a, [3]@b, [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"); + } + assert_eq!( + 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 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 + /// SUCCESSIVE publish slots — never multiple frames from one slot — with + /// every frame under the cap and no event lost or reordered. + #[test] + fn oversized_backlogs_split_across_publish_slots_under_the_cap() { + let big_text = "x".repeat(30_000); + 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 = drain_frames(&mut queue); + 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" + ); + seen.extend(frame_seqs(frame)); + } + assert_eq!( + seen, + [1, 2, 3, 4, 5, 6], + "no event lost or reordered by splitting" + ); + } + + /// 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] + 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 + } + + let mut queue = ObserverPublishQueue::default(); + queue.ingest(chunk(1, "hello ")); + queue.ingest(chunk(2, "world")); + queue.ingest(event(3, "tool_call", Some("chan-a"))); + + 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!( + inner[0]["payload"]["params"]["update"]["content"]["text"], "hello world", + "chunk text merged before the tool call" + ); + 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!( + 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); + 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" + ); + } + + /// 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. 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); + 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); + } + + 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() >= walked, + "the accumulator must never under-count true retention \ + (accumulator {} < walked {walked})", + 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); + } + + /// 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!( + 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 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" + ); + } + + /// 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] + 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 + /// (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 starts_without_a_burst_and_spaces_frames() { - let started = tokio::time::Instant::now(); - let mut pacer = ObserverPublishPacer::new(); + 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" + ); - pacer.wait().await; - let first = tokio::time::Instant::now(); - pacer.wait().await; - let second = tokio::time::Instant::now(); + // 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" + ); - assert_eq!(first.duration_since(started), OBSERVER_PUBLISH_INTERVAL); - assert_eq!(second.duration_since(first), OBSERVER_PUBLISH_INTERVAL); + // 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]), 2, "a1 + a2 gathered"); + + // 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: 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"); + + // 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 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; + 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..=2 { + 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()), + } } + // 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; - pacer.wait().await; - let ninety_first = tokio::time::Instant::now(); + // 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"); - assert_eq!(ninety_first.duration_since(first), Duration::from_secs(60)); + task.abort(); } } @@ -5058,9 +6127,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] @@ -5081,8 +6155,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"); } } diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index aea5cee077..2cbb82411f 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -106,9 +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 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 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; diff --git a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs index d78d1d21af..e3e3b3e756 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,164 @@ 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), + const turns = getActiveTurnsForAgent(AGENT); + 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("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", ); - assert.ok(!channels.has("c2"), "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) { 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.