diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 8e50a7ce76..923e132167 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -13,7 +13,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; @@ -287,6 +287,53 @@ async fn check_sibling_via_profile( false } +const OBSERVER_PUBLISH_INTERVAL: Duration = Duration::from_millis(167); +const OBSERVER_PUBLISH_LIMIT_PER_MINUTE: usize = 90; + +struct ObserverPublishPacer { + next_publish: tokio::time::Instant, + published: VecDeque, +} + +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), + } + } + + 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(); + } + + 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; + } + + let published_at = tokio::time::Instant::now(); + self.published.push_back(published_at); + self.next_publish = published_at + OBSERVER_PUBLISH_INTERVAL; + return; + } + } +} + fn spawn_relay_observer_publisher( observer: observer::ObserverHandle, publisher: RelayEventPublisher, @@ -296,68 +343,102 @@ fn spawn_relay_observer_publisher( owner_pubkey: PublicKey, ) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { - let mut coalescer = ObserverChunkCoalescer::default(); - for event in observer.snapshot() { - for event in coalescer.ingest(event) { - publish_relay_observer_event( - &publisher, - &keys, - &agent_pubkey_hex, - &owner_pubkey_hex, - &owner_pubkey, - event, - ) - .await; - } + // Subscribe BEFORE snapshotting so an event emitted between the two + // calls is never lost: it lands in the snapshot, the live receiver, or + // both. The overlap is deduped in the run loop via the snapshot's + // high-water `seq` (monotonic, assigned at emit). + let rx = observer.subscribe(); + let snapshot = observer.snapshot(); + run_relay_observer_publisher( + snapshot, + rx, + publisher, + keys, + agent_pubkey_hex, + owner_pubkey_hex, + owner_pubkey, + ) + .await; + }) +} + +async fn run_relay_observer_publisher( + snapshot: Vec, + mut rx: tokio::sync::broadcast::Receiver, + publisher: RelayEventPublisher, + keys: nostr::Keys, + agent_pubkey_hex: String, + owner_pubkey_hex: String, + owner_pubkey: PublicKey, +) { + let mut coalescer = ObserverChunkCoalescer::default(); + let mut pacer = ObserverPublishPacer::new(); + 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; } + } - let mut rx = observer.subscribe(); - let mut flush_interval = tokio::time::interval(std::time::Duration::from_millis(500)); - flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - loop { - tokio::select! { - result = rx.recv() => { - match result { - Ok(event) => { - for event in coalescer.ingest(event) { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, - ).await; - } + let mut flush_interval = tokio::time::interval(std::time::Duration::from_millis(500)); + flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + result = rx.recv() => { + match result { + Ok(event) => { + // Skip live events already delivered via the snapshot + // (the subscribe-before-snapshot overlap). + if event.seq <= max_snapshot_seq { + continue; } - 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, event, - ).await; - } - tracing::warn!(dropped = count, "relay observer publisher lagged"); + for event in coalescer.ingest(event) { + publish_relay_observer_event( + &publisher, &keys, &agent_pubkey_hex, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + ).await; } - 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, event, - ).await; - } - break; + } + 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"); } - } - _ = flush_interval.tick() => { - // Periodic flush ensures live streaming even during continuous chunk delivery. - for event in coalescer.flush() { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, - ).await; + 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; } } } + _ = flush_interval.tick() => { + // Periodic flush ensures live streaming even during continuous chunk delivery. + for event in coalescer.flush() { + publish_relay_observer_event( + &publisher, &keys, &agent_pubkey_hex, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + ).await; + } + } } - }) + } } #[derive(Default)] @@ -638,8 +719,10 @@ 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); @@ -4070,6 +4153,105 @@ mod author_gate_tests { } } +#[cfg(test)] +mod observer_snapshot_race_tests { + use super::*; + use nostr::Keys; + + fn emit_marker(observer: &observer::ObserverHandle, marker: &str) { + observer.emit( + "test_event", + None, + &observer::context_for(None, None, None), + serde_json::json!({ "marker": marker }), + ); + } + + /// An event emitted between `subscribe()` and `snapshot()` lands in BOTH + /// the snapshot and the live receiver; the seq high-water dedupe must + /// deliver it exactly once — and never lose events on either side of it. + #[tokio::test(start_paused = true)] + async fn overlap_between_subscribe_and_snapshot_publishes_exactly_once() { + let observer = observer::ObserverHandle::in_process(); + let agent_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let (publisher, mut published_rx) = RelayEventPublisher::test_pair(); + + // Before the publisher starts: replay-buffer only. + emit_marker(&observer, "before"); + // The race window: emitted after subscribe() but before snapshot(), + // so it is present in the snapshot AND queued on the receiver. + let rx = observer.subscribe(); + emit_marker(&observer, "overlap"); + let snapshot = observer.snapshot(); + assert_eq!(snapshot.len(), 2, "overlap event must be in the snapshot"); + // After the snapshot: live receiver only. + emit_marker(&observer, "after"); + // Close the broadcast channel so the run loop drains and exits. + drop(observer); + + 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(), + ) + .await; + + // 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). + 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()); + } + assert_eq!( + markers, + ["before", "overlap", "after"], + "each event must be published exactly once, in order" + ); + } +} + +#[cfg(test)] +mod observer_publish_pacer_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(); + + pacer.wait().await; + let first = tokio::time::Instant::now(); + pacer.wait().await; + let second = tokio::time::Instant::now(); + + assert_eq!(first.duration_since(started), OBSERVER_PUBLISH_INTERVAL); + assert_eq!(second.duration_since(first), OBSERVER_PUBLISH_INTERVAL); + } + + #[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; + } + + pacer.wait().await; + let ninety_first = tokio::time::Instant::now(); + + assert_eq!(ninety_first.duration_since(first), Duration::from_secs(60)); + } +} + #[cfg(test)] mod observer_chunk_coalescer_tests { use super::*; diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index b1497b6be7..ce7d565f12 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -105,6 +105,11 @@ const REQ_PACING_INTERVAL: Duration = Duration::from_millis(125); /// below the relay's 50-frames/5s budget, and ensures the select! loop is never /// 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`). +const GATED_OBSERVER_QUEUE_CAP: usize = 256; use std::time::Instant; @@ -551,6 +556,24 @@ impl RelayEventPublisher { .await .map_err(|_| RelayError::ConnectionClosed) } + + /// Test-only publisher pair: published events are forwarded to the + /// returned receiver instead of a live relay socket. + #[cfg(test)] + pub(crate) fn test_pair() -> (Self, mpsc::Receiver) { + let (cmd_tx, mut cmd_rx) = mpsc::channel::(64); + let (event_tx, event_rx) = mpsc::channel(64); + tokio::spawn(async move { + while let Some(cmd) = cmd_rx.recv().await { + if let RelayCommand::PublishEvent { event } = cmd { + if event_tx.send(*event).await.is_err() { + break; + } + } + } + }); + (Self { cmd_tx }, event_rx) + } } impl HarnessRelay { @@ -998,6 +1021,19 @@ struct BgState { /// subscription. The main-loop drain re-sends the REQ once the gate clears, /// even when `rate_limited_pending` is empty. observer_resub_needed: bool, + /// Observer telemetry frames (kind 24200) parked while the rate-limit gate + /// is armed. Unlike typing indicators, these frames are durable telemetry: + /// dropping them silently loses turn history in the Desktop observer. + /// Bounded at `GATED_OBSERVER_QUEUE_CAP` (drop-oldest); drained by the + /// main loop one frame per pacing tick once the gate clears. + gated_observer_pending: VecDeque>, + /// Observer frames written to the socket but not yet acknowledged. The + /// relay's rate-limit NOTICE does not carry an event ID, so all unresolved + /// observer writes are moved back ahead of the parked FIFO when one arrives. + observer_in_flight: VecDeque>, + /// Frames evicted from the bounded pending/in-flight observer buffers since + /// summary log. Makes overflow loss visible instead of silent. + gated_observer_dropped: u64, /// Channels whose REQ failed during `resubscribe_after_reconnect`. /// /// A single failed channel REQ is parked here instead of aborting the whole @@ -1030,6 +1066,9 @@ impl BgState { rate_limited_pending: HashMap::new(), membership_resub_needed: false, observer_resub_needed: false, + gated_observer_pending: VecDeque::new(), + observer_in_flight: VecDeque::new(), + gated_observer_dropped: 0, resubscribe_retry: HashSet::new(), backoff_step: 0, } @@ -1128,13 +1167,65 @@ impl BgState { } None } + + /// Park an observer telemetry frame while the rate-limit gate is armed. + /// + /// Bounded drop-oldest queue: overflow evicts the oldest frame and counts + /// it in `gated_observer_dropped` so the loss is visible, never silent. + fn park_gated_observer_frame(&mut self, event: Box) { + if self.gated_observer_pending.len() >= GATED_OBSERVER_QUEUE_CAP { + self.gated_observer_pending.pop_front(); + self.gated_observer_dropped += 1; + warn!( + dropped_total = self.gated_observer_dropped, + "gated observer queue full — dropped oldest frame" + ); + } + self.gated_observer_pending.push_back(event); + } + + /// Restore unresolved observer writes ahead of frames parked after the + /// gate armed. NOTICE has no event ID, so conservatively retry every frame + /// without an OK; duplicate IDs are harmless at the relay. + fn requeue_observer_in_flight(&mut self) { + while let Some(event) = self.observer_in_flight.pop_back() { + self.gated_observer_pending.push_front(event); + } + while self.gated_observer_pending.len() > GATED_OBSERVER_QUEUE_CAP { + self.gated_observer_pending.pop_front(); + self.gated_observer_dropped += 1; + } + } + + fn track_observer_in_flight(&mut self, event: Box) { + if self.observer_in_flight.len() >= GATED_OBSERVER_QUEUE_CAP { + self.observer_in_flight.pop_front(); + self.gated_observer_dropped += 1; + warn!( + dropped_total = self.gated_observer_dropped, + "observer acknowledgment window full — dropped oldest frame" + ); + } + self.observer_in_flight.push_back(event); + } + + fn acknowledge_observer_frame(&mut self, event_id: &str) { + if let Some(index) = self + .observer_in_flight + .iter() + .position(|event| event.id.to_hex() == event_id) + { + self.observer_in_flight.remove(index); + } + } } /// Record a command's intent in state while disconnected (no WebSocket). /// /// Subscribe/Unsubscribe/SubscribeMembership record intent so reconnect /// restores the right subscriptions. SetStartupWatermark floors the replay -/// window. PublishEvent and Reconnect are no-ops while disconnected. +/// window. Observer telemetry publishes are parked for post-reconnect drain; +/// other PublishEvent and Reconnect are no-ops while disconnected. /// /// Callers MUST handle `Shutdown` before calling — reaching the Shutdown /// arm here is a logic error. @@ -1174,8 +1265,15 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) { state.membership_last_seen = Some(ts); } } - // Ephemeral events are meaningless while disconnected. - RelayCommand::PublishEvent { .. } => {} + // Observer telemetry frames are durable: park them (bounded, visible + // overflow) so they are delivered by the post-reconnect drain. Other + // ephemeral publishes (typing indicators) are meaningless while + // disconnected and are dropped. + RelayCommand::PublishEvent { event } => { + if event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME { + state.park_gated_observer_frame(event); + } + } // Already reconnecting — redundant. RelayCommand::Reconnect => {} // Callers MUST handle Shutdown before calling this function. @@ -1190,11 +1288,17 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) { /// Retain command intent after a live send failure. /// -/// Subscription state must survive reconnect; ephemeral publishes are deliberately +/// Subscription state must survive reconnect. Observer telemetry publishes are +/// parked for post-reconnect drain; other ephemeral publishes are deliberately /// discarded because replaying a typing indicator after reconnect is meaningless. /// `Shutdown` and `Reconnect` are handled by the caller. fn retain_failed_command_intent(state: &mut BgState, cmd: RelayCommand) { match cmd { + RelayCommand::PublishEvent { event } + if event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME => + { + state.park_gated_observer_frame(event); + } RelayCommand::PublishEvent { .. } => {} cmd => apply_command_to_state(state, cmd), } @@ -1350,28 +1454,44 @@ async fn execute_connected_command( } } RelayCommand::PublishEvent { event } => { - // Drop ephemeral publishes while rate-gated. Stale typing indicators - // are worthless and sending them would consume admission budget the - // relay already rejected us on. + // Observer telemetry frames (kind 24200) are durable telemetry, not + // droppable ephemera: park them while the rate-limit gate is armed — + // and while earlier parked frames are still draining, so relative + // order is preserved — then let the main-loop drain deliver them + // one per pacing tick once the gate clears. + if event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME + && (state.check_rate_gate().is_some() || !state.gated_observer_pending.is_empty()) + { + debug!( + pending = state.gated_observer_pending.len(), + "rate-gated: parking observer frame for paced drain" + ); + state.park_gated_observer_frame(event); + return true; + } + // Drop remaining ephemeral publishes while rate-gated. Stale typing + // indicators are worthless and sending them would consume admission + // budget the relay already rejected us on. // - // INVARIANT: the WS publish path carries only ephemeral kinds (typing - // indicators). The silent drop-while-gated relies on that invariant. If a - // future caller publishes durable events through this path, it must add a - // kind guard before this branch to avoid silently discarding user data. + // INVARIANT: apart from observer frames (parked above), the WS publish + // path carries only ephemeral kinds (typing indicators). The silent + // drop-while-gated relies on that invariant. If a future caller + // publishes durable events through this path, it must extend the + // kind guard above to avoid silently discarding user data. if state.check_rate_gate().is_some() { debug!("rate-gated: dropping ephemeral PublishEvent (typing indicator)"); return true; } - let msg = json!(["EVENT", event]); - if let Ok(text) = serde_json::to_string(&msg) { - if let Err(e) = - ws_send_timeout(ws, Message::Text(text.into()), WS_SEND_TIMEOUT_SECS).await - { - // Ephemeral events (typing indicators) are best-effort. - // Log the failure but don't trigger reconnect — the next - // ping or read will detect the dead socket. - warn!("failed to publish event: {e}"); + // Best-effort: log a send failure but don't trigger reconnect — the + // next ping or read will detect the dead socket. A failed observer + // frame is parked so the post-reconnect drain redelivers it. + let is_observer = event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME; + if send_publish_event_frame(ws, &event).await { + if is_observer { + state.track_observer_in_flight(event); } + } else if is_observer { + state.park_gated_observer_frame(event); } true } @@ -1624,6 +1744,14 @@ async fn run_background_task( if budget > 0 && !state.resubscribe_retry.is_empty() { let sent = drain_resubscribe_retry(&mut ws, &mut state, &agent_pubkey_hex, budget).await; + budget = budget.saturating_sub(sent); + if sent > 0 { + any_sent = true; + } + } + + if budget > 0 && !state.gated_observer_pending.is_empty() { + let sent = drain_gated_observer_pending(&mut ws, &mut state, budget).await; if sent > 0 { any_sent = true; } @@ -1631,6 +1759,13 @@ async fn run_background_task( if any_sent { drain_pacing_next = Some(tokio::time::Instant::now() + REQ_PACING_INTERVAL); + } else if !state.gated_observer_pending.is_empty() { + // Nothing sent because the gate is still armed. Arm the pacing + // timer to the gate deadline so parked observer frames drain + // promptly even when no other traffic wakes the select loop. + drain_pacing_next = state + .check_rate_gate() + .or_else(|| Some(tokio::time::Instant::now() + REQ_PACING_INTERVAL)); } } @@ -2050,6 +2185,7 @@ async fn handle_ws_message( if message.starts_with("rate-limited:") { let secs = parse_rate_limit_retry_secs(&message).unwrap_or(0); let deadline = state.set_rate_limit_gate(secs); + state.requeue_observer_in_flight(); warn!( "rate-limit gate armed via NOTICE until ~{:.1}s from now", deadline @@ -2213,6 +2349,7 @@ async fn handle_ws_message( warn!("mid-session AUTH rejected (event {event_id}): {message} — triggering reconnect"); return false; } + state.acknowledge_observer_frame(&event_id); debug!("OK for event {event_id}: accepted={accepted} message={message}"); } } @@ -2455,6 +2592,60 @@ async fn resubscribe_after_reconnect( } } +/// Send a signed EVENT frame on the live socket. Returns `false` on send failure. +/// +/// Best-effort at the socket level: a failure is logged but does not trigger +/// reconnect — the next ping or read will detect the dead socket. +async fn send_publish_event_frame(ws: &mut WsStream, event: &Event) -> bool { + let msg = json!(["EVENT", event]); + if let Ok(text) = serde_json::to_string(&msg) { + if let Err(e) = ws_send_timeout(ws, Message::Text(text.into()), WS_SEND_TIMEOUT_SECS).await + { + warn!("failed to publish event: {e}"); + return false; + } + } + true +} + +/// Drain parked observer telemetry frames once the rate-limit gate clears. +/// +/// Called by the main loop pacing timer. Sends at most `budget` frames without +/// sleeping — pacing is enforced by the caller via `drain_pacing_next`. Stops +/// immediately if the gate re-arms mid-drain. When the queue empties, any +/// overflow loss is summarized in one warning. Returns the number of frames sent. +async fn drain_gated_observer_pending( + ws: &mut WsStream, + state: &mut BgState, + budget: usize, +) -> usize { + let mut sent = 0; + while sent < budget { + if state.check_rate_gate().is_some() { + break; + } + let Some(event) = state.gated_observer_pending.pop_front() else { + break; + }; + if !send_publish_event_frame(ws, &event).await { + // Socket may be dead — re-park at the front so the frame survives + // reconnect (the post-reconnect drain will retry it in order). + state.gated_observer_pending.push_front(event); + break; + } + state.track_observer_in_flight(event); + sent += 1; + } + if state.gated_observer_pending.is_empty() && state.gated_observer_dropped > 0 { + warn!( + observer_frames_dropped = state.gated_observer_dropped, + "observer frames lost to gated-queue overflow" + ); + state.gated_observer_dropped = 0; + } + sent +} + /// Drain `rate_limited_pending` channels whose retry deadline has passed. /// /// Called by the main loop pacing timer. Sends at most `budget` REQs without @@ -2698,6 +2889,7 @@ async fn try_autonomous_reconnect( observer_control_tx: &mpsc::Sender, auth_tag: Option<&nostr::Tag>, ) -> ReconnectOutcome { + state.requeue_observer_in_flight(); // 5 attempts, up to 16s base backoff. Shares delay values with the // initial-connect retry in `HarnessRelay::connect()` (STARTUP_CONNECT_BACKOFFS) — // see its doc comment for how the two loops consume the array differently. @@ -2827,6 +3019,7 @@ async fn wait_for_reconnect( skip_drain: bool, auth_tag: Option<&nostr::Tag>, ) -> ReconnectOutcome { + state.requeue_observer_in_flight(); if !skip_drain { // Drain commands until we get Reconnect (or Shutdown). // Other commands update state so reconnect reflects latest intent. @@ -5564,6 +5757,214 @@ mod tests { ); } + /// Build a signed observer telemetry frame (kind 24200) for gate tests. + fn make_observer_frame(keys: &Keys) -> Event { + let recipient = Keys::generate(); + let encrypted = buzz_core::observer::encrypt_observer_payload( + keys, + &recipient.public_key(), + &json!({"type": "test"}), + ) + .expect("encrypt test observer payload"); + buzz_sdk::build_agent_observer_frame( + &recipient.public_key().to_hex(), + &keys.public_key().to_hex(), + "telemetry", + &encrypted, + ) + .expect("build test observer frame") + .sign_with_keys(keys) + .expect("sign test observer frame") + } + + /// While the rate-limit gate is armed, an observer frame (kind 24200) is + /// parked — not silently dropped — and delivered by the drain once the + /// gate clears. A typing indicator in the same window stays dropped. + #[tokio::test] + async fn gated_observer_frame_is_parked_then_drained_not_dropped() { + let (mut client, mut server) = test_ws_pair().await; + let mut state = BgState::new(); + let keys = Keys::generate(); + state.rate_limit_gate = Some(tokio::time::Instant::now() + Duration::from_millis(150)); + + // Observer frame while gated: parked, nothing on the wire. + let observer_frame = make_observer_frame(&keys); + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(observer_frame.clone()), + }, + ) + .await; + assert!(ok); + assert_eq!( + state.gated_observer_pending.len(), + 1, + "observer frame must be parked while gated" + ); + + // Typing indicator while gated: still dropped, not parked. + let typing = EventBuilder::new(Kind::Custom(KIND_TYPING_INDICATOR as u16), "") + .tags([Tag::parse(["h", &Uuid::new_v4().to_string()]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign typing indicator"); + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(typing), + }, + ) + .await; + assert!(ok); + assert_eq!( + state.gated_observer_pending.len(), + 1, + "typing indicators must not be parked" + ); + assert!( + timeout(Duration::from_millis(50), server.next()) + .await + .is_err(), + "nothing may reach the wire while the gate is armed" + ); + + // Gate expires — the drain delivers the parked frame. + tokio::time::sleep(Duration::from_millis(160)).await; + assert_eq!( + drain_gated_observer_pending(&mut client, &mut state, 1).await, + 1 + ); + assert!(state.gated_observer_pending.is_empty()); + let frame = next_test_frame(&mut server).await; + assert_eq!(frame[0], "EVENT"); + assert_eq!(frame[1]["id"], observer_frame.id.to_hex()); + assert_eq!( + frame[1]["kind"], + u64::from(KIND_AGENT_OBSERVER_FRAME), + "delivered frame must be the parked observer frame" + ); + } + + /// Observer frames arriving while earlier parked frames are still queued + /// are appended behind them (order preserved), even if the gate has + /// already expired. + #[tokio::test] + async fn observer_frames_queue_behind_parked_backlog_in_order() { + let (mut client, mut server) = test_ws_pair().await; + let mut state = BgState::new(); + let keys = Keys::generate(); + state.rate_limit_gate = Some(tokio::time::Instant::now() + Duration::from_millis(50)); + + let first = make_observer_frame(&keys); + let second = make_observer_frame(&keys); + for event in [&first, &second] { + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(event.clone()), + }, + ) + .await; + assert!(ok); + } + assert_eq!(state.gated_observer_pending.len(), 2); + + // Gate expires but the backlog is not drained yet — a third frame must + // queue behind it rather than jumping ahead on the wire. + tokio::time::sleep(Duration::from_millis(60)).await; + let third = make_observer_frame(&keys); + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(third.clone()), + }, + ) + .await; + assert!(ok); + assert_eq!( + state.gated_observer_pending.len(), + 3, + "frame must queue behind undrained backlog to preserve order" + ); + + for expected in [&first, &second, &third] { + assert_eq!( + drain_gated_observer_pending(&mut client, &mut state, 1).await, + 1 + ); + let frame = next_test_frame(&mut server).await; + assert_eq!(frame[1]["id"], expected.id.to_hex(), "order preserved"); + } + assert!(state.gated_observer_pending.is_empty()); + } + + #[test] + fn observer_notice_requeues_unacknowledged_frames_and_ok_retires_them() { + let mut state = BgState::new(); + let keys = Keys::generate(); + let accepted = make_observer_frame(&keys); + let rejected = make_observer_frame(&keys); + let later = make_observer_frame(&keys); + + state.track_observer_in_flight(Box::new(accepted.clone())); + state.track_observer_in_flight(Box::new(rejected.clone())); + state.acknowledge_observer_frame(&accepted.id.to_hex()); + state.park_gated_observer_frame(Box::new(later.clone())); + state.requeue_observer_in_flight(); + + let ids: Vec<_> = state + .gated_observer_pending + .iter() + .map(|event| event.id) + .collect(); + assert_eq!(ids, [rejected.id, later.id]); + assert!(state.observer_in_flight.is_empty()); + } + + /// The parked-frame queue is bounded: overflow evicts the oldest frame and + /// counts it; the drain resets the counter after logging the summary. + #[tokio::test] + async fn gated_observer_queue_drops_oldest_on_overflow() { + let mut state = BgState::new(); + let keys = Keys::generate(); + let first = make_observer_frame(&keys); + state.park_gated_observer_frame(Box::new(first.clone())); + for _ in 1..GATED_OBSERVER_QUEUE_CAP { + state.park_gated_observer_frame(Box::new(make_observer_frame(&keys))); + } + assert_eq!(state.gated_observer_pending.len(), GATED_OBSERVER_QUEUE_CAP); + assert_eq!(state.gated_observer_dropped, 0); + + let overflow = make_observer_frame(&keys); + state.park_gated_observer_frame(Box::new(overflow.clone())); + assert_eq!( + state.gated_observer_pending.len(), + GATED_OBSERVER_QUEUE_CAP, + "queue must stay bounded" + ); + assert_eq!(state.gated_observer_dropped, 1, "loss must be counted"); + assert!( + !state + .gated_observer_pending + .iter() + .any(|e| e.id == first.id), + "oldest frame must be the one evicted" + ); + assert_eq!( + state.gated_observer_pending.back().map(|e| e.id), + Some(overflow.id), + "newest frame must be retained" + ); + } + /// is_dns_error correctly classifies platform resolver strings, including /// the production shape: a WebSocket I/O error wrapping the OS message. #[test]