diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/event_hub.rs | 81 | ||||
| -rw-r--r-- | src/api/server.rs | 23 | ||||
| -rw-r--r-- | src/api/server/subscription_socket_tests.rs | 355 | ||||
| -rw-r--r-- | src/api/subscriptions.rs | 255 |
4 files changed, 670 insertions, 44 deletions
diff --git a/src/api/event_hub.rs b/src/api/event_hub.rs index 42cf47a2..29843313 100644 --- a/src/api/event_hub.rs +++ b/src/api/event_hub.rs @@ -9,6 +9,12 @@ struct EventHubState { events: Vec<(u64, crate::api::schema::EventEnvelope)>, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum EventHistoryError { + Lost, + Unavailable, +} + impl EventHub { const MAX_EVENTS: usize = 512; @@ -37,6 +43,29 @@ impl EventHub { .collect() } + pub(super) fn events_after_checked( + &self, + sequence: u64, + ) -> Result<Vec<(u64, crate::api::schema::EventEnvelope)>, EventHistoryError> { + let state = self + .inner + .lock() + .map_err(|_| EventHistoryError::Unavailable)?; + if state + .events + .first() + .is_some_and(|(first, _)| sequence < first.saturating_sub(1)) + { + return Err(EventHistoryError::Lost); + } + Ok(state + .events + .iter() + .filter(|(event_sequence, _)| *event_sequence > sequence) + .cloned() + .collect()) + } + pub fn current_sequence(&self) -> u64 { let Ok(state) = self.inner.lock() else { return 0; @@ -44,3 +73,55 @@ impl EventHub { state.next_sequence } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::api::schema::{EventData, EventEnvelope, EventKind}; + + fn event() -> EventEnvelope { + EventEnvelope { + event: EventKind::WorkspaceFocused, + data: EventData::WorkspaceFocused { + workspace_id: "workspace_1".into(), + }, + } + } + + #[test] + fn checked_history_distinguishes_retained_boundary_from_lost_events() { + let hub = EventHub::default(); + assert!(hub.events_after_checked(0).unwrap().is_empty()); + for _ in 0..EventHub::MAX_EVENTS { + hub.push(event()); + } + assert_eq!( + hub.events_after_checked(0).unwrap().len(), + EventHub::MAX_EVENTS + ); + hub.push(event()); + assert_eq!(hub.events_after_checked(0), Err(EventHistoryError::Lost)); + let retained = hub.events_after_checked(1).unwrap(); + assert_eq!(retained.len(), EventHub::MAX_EVENTS); + assert_eq!(retained.first().unwrap().0, 2); + assert_eq!(retained.last().unwrap().0, hub.current_sequence()); + assert!(hub + .events_after_checked(hub.current_sequence()) + .unwrap() + .is_empty()); + } + + #[test] + fn checked_history_reports_unavailable_instead_of_empty_after_poison() { + let hub = EventHub::default(); + assert!(std::panic::catch_unwind(|| { + let _guard = hub.inner.lock().unwrap(); + panic!("poison the test event history"); + }) + .is_err()); + assert_eq!( + hub.events_after_checked(0), + Err(EventHistoryError::Unavailable) + ); + } +} diff --git a/src/api/server.rs b/src/api/server.rs index 03724813..44a2a0df 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -24,6 +24,9 @@ use crate::ipc::{ mod pane_graphics_stream; +#[cfg(test)] +mod subscription_socket_tests; + const SOCKET_PERMISSION_MODE: u32 = 0o600; pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100); pub(super) const APP_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); @@ -749,7 +752,7 @@ fn stream_subscriptions( if let Err(err) = write_json_line( &mut stream, &SuccessResponse { - id: request_id, + id: request_id.clone(), result: ResponseResult::SubscriptionStarted {}, }, ) { @@ -765,7 +768,23 @@ fn stream_subscriptions( } for subscription in &mut subscriptions { - if let Some(event) = subscription.poll(api_tx, event_hub) { + let events = match subscription.poll_batch(api_tx, event_hub) { + Ok(events) => events, + Err(error) => { + write_json_line_allow_disconnect( + &mut stream, + &ErrorResponse { + id: request_id, + error, + }, + )?; + return Ok(()); + } + }; + for event in events { + if should_stop_connection(&mut stream, running)? { + return Ok(()); + } if let Err(err) = write_json_line(&mut stream, &event) { if is_connection_closed_error(&err) { return Ok(()); diff --git a/src/api/server/subscription_socket_tests.rs b/src/api/server/subscription_socket_tests.rs new file mode 100644 index 00000000..5cfbc706 --- /dev/null +++ b/src/api/server/subscription_socket_tests.rs @@ -0,0 +1,355 @@ +use super::*; +use crate::api::schema::{ + AgentStatus, EventData, EventEnvelope, EventKind, PaneInfo, PaneReadResult, ReadFormat, + ReadSource, +}; +use crate::ipc::{poll_local_stream_read_count, LocalStreamReadCount}; +use interprocess::local_socket::traits::Listener as _; +use serde_json::{json, Value}; +use std::sync::atomic::AtomicU64; +use tokio::sync::mpsc; + +const RESPONSE_TIMEOUT: Duration = Duration::from_secs(2); + +struct SocketTest { + hub: EventHub, + running: Arc<AtomicBool>, + api_tx: ApiRequestSender, + api_rx: mpsc::UnboundedReceiver<ApiRequestMessage>, + workers: Vec<std::thread::JoinHandle<io::Result<()>>>, + paths: Vec<PathBuf>, +} + +impl SocketTest { + fn new() -> Self { + let (api_tx, api_rx) = mpsc::unbounded_channel(); + Self { + hub: EventHub::default(), + running: Arc::new(AtomicBool::new(true)), + api_tx, + api_rx, + workers: Vec::new(), + paths: Vec::new(), + } + } + + fn connect(&mut self) -> Client { + static NEXT_SOCKET: AtomicU64 = AtomicU64::new(0); + let path = std::env::temp_dir().join(format!( + "herdr-sub-{}-{}", + std::process::id(), + NEXT_SOCKET.fetch_add(1, Ordering::Relaxed) + )); + let listener = bind_local_listener(&path).unwrap(); + self.paths.push(path.clone()); + let mut stream = crate::ipc::connect_local_stream(&path).unwrap(); + let server = listener.accept().unwrap(); + set_local_stream_polling(&mut stream, true).unwrap(); + let api_tx = self.api_tx.clone(); + let hub = self.hub.clone(); + let running = Arc::clone(&self.running); + let worker = + std::thread::spawn(move || handle_connection(server, &api_tx, &hub, &running, None)); + self.workers.push(worker); + Client { + stream, + buffered: Vec::new(), + } + } + + fn app_request(&mut self) -> ApiRequestMessage { + let deadline = Instant::now() + RESPONSE_TIMEOUT; + loop { + match self.api_rx.try_recv() { + Ok(request) => return request, + Err(mpsc::error::TryRecvError::Empty) => { + assert!( + Instant::now() < deadline, + "timed out waiting for app request" + ); + std::thread::sleep(Duration::from_millis(1)); + } + Err(error) => panic!("app request channel closed: {error}"), + } + } + } +} + +impl Drop for SocketTest { + fn drop(&mut self) { + self.running.store(false, Ordering::Relaxed); + self.api_rx.close(); + while self.api_rx.try_recv().is_ok() {} + let deadline = Instant::now() + APP_RESPONSE_TIMEOUT + Duration::from_secs(1); + let mut failures = Vec::new(); + for worker in self.workers.drain(..) { + while !worker.is_finished() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(1)); + } + if worker.is_finished() { + match worker.join() { + Ok(Ok(())) => {} + Ok(Err(error)) => failures.push(format!("connection failed: {error}")), + Err(_) => failures.push("connection panicked".into()), + } + } else { + failures.push("subscription connection did not stop".into()); + } + } + // Also remove Windows listener marker files, including after a setup failure. + for path in self.paths.drain(..) { + if let Err(error) = std::fs::remove_file(path) { + if error.kind() != io::ErrorKind::NotFound { + failures.push(format!("socket cleanup failed: {error}")); + } + } + } + if !std::thread::panicking() { + assert!(failures.is_empty(), "{failures:?}"); + } + } +} + +struct Client { + stream: LocalStream, + buffered: Vec<u8>, +} + +impl Client { + fn send(&mut self, request: Value) { + writeln!(self.stream, "{request}").unwrap(); + } + + fn subscribe(&mut self, id: &str, subscriptions: Value) { + self.send(json!({ + "id": id, + "method": "events.subscribe", + "params": {"subscriptions": subscriptions} + })); + } + + fn next_line(&mut self, deadline: Instant) -> Option<Value> { + // LocalStream recv timeouts are unsupported on Windows. Use the same bounded + // nonblocking/PeekNamedPipe reads as the API client, retaining partial JSON lines. + loop { + if let Some(end) = self.buffered.iter().position(|byte| *byte == b'\n') { + let line: Vec<_> = self.buffered.drain(..=end).collect(); + return Some(serde_json::from_slice(&line).expect("subscription JSON")); + } + assert!( + Instant::now() < deadline, + "timed out waiting for socket response" + ); + let mut bytes = [0; 4096]; + match poll_local_stream_read_count(&mut self.stream, &mut bytes).unwrap() { + LocalStreamReadCount::Data(count) => { + self.buffered.extend_from_slice(&bytes[..count]) + } + LocalStreamReadCount::Pending => std::thread::sleep(Duration::from_millis(1)), + LocalStreamReadCount::Closed => { + assert!(self.buffered.is_empty(), "incomplete JSON at EOF"); + return None; + } + } + } + } + + fn response(&mut self) -> Value { + self.next_line(Instant::now() + RESPONSE_TIMEOUT) + .expect("socket response before EOF") + } + + fn assert_started(&mut self, id: &str) { + let response = self.response(); + assert_eq!(response["id"], id); + assert_eq!(response["result"]["type"], "subscription_started"); + } + + fn assert_renames(&mut self, indices: std::ops::Range<usize>, deadline: Instant) { + for index in indices { + let event = self.next_line(deadline).expect("rename before EOF"); + assert_eq!(event["event"], "workspace_renamed"); + assert_eq!(event["data"]["label"], format!("flood-{index}")); + } + } + + fn assert_history_lost(&mut self, id: &str) { + let response = self.response(); + assert_eq!(response["id"], id); + assert_eq!(response["error"]["code"], "events_lost", "{response}"); + assert_eq!(self.next_line(Instant::now() + RESPONSE_TIMEOUT), None); + } +} + +fn renamed_event(index: usize) -> EventEnvelope { + EventEnvelope { + event: EventKind::WorkspaceRenamed, + data: EventData::WorkspaceRenamed { + workspace_id: "workspace_1".into(), + label: format!("flood-{index}"), + }, + } +} + +fn output_subscription() -> Value { + json!({ + "type": "pane.output_matched", + "pane_id": "pane_1", + "source": "recent", + "match": {"type": "substring", "value": "never"} + }) +} + +fn reply(request: ApiRequestMessage, result: ResponseResult) { + request + .respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: request.request.id, + result, + }) + .unwrap(), + ) + .unwrap(); +} + +fn reply_to_probe(request: ApiRequestMessage) { + let result = match request.request.method { + Method::PaneGet(_) => ResponseResult::PaneInfo { + pane: PaneInfo { + pane_id: "pane_1".into(), + terminal_id: "term_1".into(), + workspace_id: "workspace_1".into(), + tab_id: "tab_1".into(), + focused: true, + cwd: None, + foreground_cwd: None, + label: None, + agent: Some("pi".into()), + title: None, + terminal_title: None, + terminal_title_stripped: None, + display_agent: None, + agent_status: AgentStatus::Working, + state_labels: Default::default(), + tokens: Default::default(), + agent_session: None, + scroll: None, + revision: 0, + }, + }, + Method::PaneRead(_) => ResponseResult::PaneRead { + read: PaneReadResult { + pane_id: "pane_1".into(), + workspace_id: "workspace_1".into(), + tab_id: "tab_1".into(), + source: ReadSource::RecentUnwrapped, + format: ReadFormat::Text, + text: String::new(), + revision: 0, + truncated: false, + }, + }, + ref other => panic!("unexpected subscription probe: {other:?}"), + }; + reply(request, result); +} + +#[test] +fn subscriptions_drain_retained_bursts_without_per_event_poll_delay() { + let mut test = SocketTest::new(); + let mut client = test.connect(); + client.subscribe("burst", json!([{"type": "workspace.renamed"}])); + client.assert_started("burst"); + for index in 0..128 { + test.hub.push(renamed_event(index)); + } + // One deadline for the entire batch detects a 100 ms delay per event. + client.assert_renames(0..128, Instant::now() + RESPONSE_TIMEOUT); +} + +#[test] +fn subscriptions_report_history_loss_before_sending_a_partial_stream() { + assert_subscription_history_loss(false); +} + +#[test] +fn subscriptions_report_history_loss_before_initial_agent_status() { + assert_subscription_history_loss(true); +} + +fn assert_subscription_history_loss(agent_status: bool) { + let mut test = SocketTest::new(); + let mut client = test.connect(); + let subscriptions = if agent_status { + json!([{ + "type": "pane.agent_status_changed", + "pane_id": "pane_1", + "agent_status": "working" + }]) + } else { + json!([{"type": "workspace.renamed"}, output_subscription()]) + }; + client.subscribe("history-gap", subscriptions); + // Hold the setup probe after the server pins its subscription cursor. + let probe = test.app_request(); + assert!(probe.request.id.ends_with(":probe")); + for index in 0..600 { + test.hub.push(renamed_event(index)); + } + reply_to_probe(probe); + client.assert_started("history-gap"); + client.assert_history_lost("history-gap"); +} + +#[test] +fn lagging_subscription_closes_without_interrupting_other_clients() { + let mut test = SocketTest::new(); + let mut healthy = test.connect(); + healthy.subscribe("healthy", json!([{"type": "workspace.renamed"}])); + healthy.assert_started("healthy"); + + let mut slow = test.connect(); + slow.subscribe( + "slow", + json!([{"type": "workspace.renamed"}, output_subscription()]), + ); + let probe = test.app_request(); + assert_eq!(probe.request.id, "slow:sub:1:probe"); + reply_to_probe(probe); + slow.assert_started("slow"); + // Pause only this connection in an existing app request, rather than depending + // on OS socket buffer sizes or sleeping to make its event cursor fall behind. + let paused_read = test.app_request(); + assert_eq!(paused_read.request.id, "slow:sub:1:read"); + assert!(matches!(paused_read.request.method, Method::PaneRead(_))); + // All five batches must finish before the held app request can time out. + let deadline = Instant::now() + RESPONSE_TIMEOUT; + for batch in 0..5 { + let indices = batch * 128..(batch + 1) * 128; + for index in indices.clone() { + test.hub.push(renamed_event(index)); + } + healthy.assert_renames(indices, deadline); + } + + reply_to_probe(paused_read); + slow.assert_history_lost("slow"); + test.hub.push(renamed_event(640)); + healthy.assert_renames(640..641, Instant::now() + RESPONSE_TIMEOUT); + + let mut ordinary = test.connect(); + ordinary.send(json!({"id": "ordinary", "method": "workspace.list", "params": {}})); + let request = test.app_request(); + assert_eq!(request.request.id, "ordinary"); + assert!(matches!(request.request.method, Method::WorkspaceList(_))); + reply( + request, + ResponseResult::WorkspaceList { + workspaces: Vec::new(), + }, + ); + let response = ordinary.response(); + assert_eq!(response["id"], "ordinary"); + assert_eq!(response["result"]["type"], "workspace_list"); +} diff --git a/src/api/subscriptions.rs b/src/api/subscriptions.rs index 0888250e..5be2d523 100644 --- a/src/api/subscriptions.rs +++ b/src/api/subscriptions.rs @@ -1,5 +1,6 @@ use regex::Regex; +use crate::api::event_hub::EventHistoryError; use crate::api::schema::{ ErrorBody, ErrorResponse, EventKind, Method, PaneAgentStatusChangedEvent, PaneOutputMatchedEvent, PaneScrollChangedEvent, PaneScrollInfo, Request, Subscription, @@ -277,6 +278,72 @@ impl ActiveSubscription { _ => Ok(self.poll(api_tx, event_hub)), } } + + pub(super) fn poll_batch( + &mut self, + api_tx: &ApiRequestSender, + event_hub: &EventHub, + ) -> Result<Vec<serde_json::Value>, ErrorBody> { + match self { + Self::Event(subscription) => { + let events = subscription_events_after(event_hub, subscription.last_sequence)?; + let mut matching = Vec::new(); + for (sequence, event) in events { + subscription.last_sequence = sequence; + if event.event == subscription.event_kind { + matching.push(serde_json::to_value(event).map_err(event_encoding_error)?); + } + } + Ok(matching) + } + Self::AgentStatusChanged(subscription) => { + let events = subscription_events_after(event_hub, subscription.last_sequence)?; + let mut matching = Vec::new(); + for (sequence, event) in events { + subscription.last_sequence = sequence; + if let Some(event) = subscription.event_from_history(event) { + matching.push(serde_json::to_value(event).map_err(event_encoding_error)?); + } + } + if matching.is_empty() { + if let Some(event) = + subscription.poll_snapshot(api_tx, event_hub).ok().flatten() + { + matching.push(serde_json::to_value(event).map_err(event_encoding_error)?); + } + } + Ok(matching) + } + // These subscriptions sample current state, not retained event history. + // Keep their existing cadence even when a lifecycle batch was nonempty. + Self::OutputMatched(_) | Self::ScrollChanged(_) => { + Ok(self.poll(api_tx, event_hub).into_iter().collect()) + } + } + } +} + +fn subscription_events_after( + event_hub: &EventHub, + sequence: u64, +) -> Result<Vec<(u64, crate::api::schema::EventEnvelope)>, ErrorBody> { + event_hub.events_after_checked(sequence).map_err(|error| match error { + EventHistoryError::Lost => ErrorBody { + code: "events_lost".into(), + message: "event subscription fell behind retained history; resubscribe and resync with session.snapshot".into(), + }, + EventHistoryError::Unavailable => ErrorBody { + code: "server_unavailable".into(), + message: "event history is unavailable".into(), + }, + }) +} + +fn event_encoding_error(error: serde_json::Error) -> ErrorBody { + ErrorBody { + code: "internal_error".into(), + message: format!("failed to encode subscription event: {error}"), + } } impl ActiveEventSubscription { @@ -341,10 +408,55 @@ impl ActiveAgentStatusChangedSubscription { api_tx: &ApiRequestSender, event_hub: &EventHub, ) -> Result<Option<SubscriptionEventEnvelope>, ErrorResponse> { - let mut saw_status_event = false; for (sequence, event) in event_hub.events_after(self.last_sequence) { self.last_sequence = sequence; - let crate::api::schema::EventData::PaneAgentStatusChanged { + if let Some(event) = self.event_from_history(event) { + return Ok(Some(event)); + } + } + + self.poll_snapshot(api_tx, event_hub) + } + + fn event_from_history( + &mut self, + event: crate::api::schema::EventEnvelope, + ) -> Option<SubscriptionEventEnvelope> { + if event.event != EventKind::PaneAgentStatusChanged { + return None; + } + let crate::api::schema::EventData::PaneAgentStatusChanged { + pane_id, + workspace_id, + agent_status, + agent, + title, + display_agent, + state_labels, + } = event.data + else { + return None; + }; + if pane_id != self.pane_id { + return None; + } + self.last_status = Some(agent_status); + self.last_presentation = Some(PanePresentationSnapshot::from_event( + &title, + &display_agent, + &state_labels, + )); + self.initial_event = None; + if self + .status_filter + .is_some_and(|wanted| wanted != agent_status) + { + return None; + } + + Some(SubscriptionEventEnvelope { + event: SubscriptionEventKind::PaneAgentStatusChanged, + data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent { pane_id, workspace_id, agent_status, @@ -352,47 +464,16 @@ impl ActiveAgentStatusChangedSubscription { title, display_agent, state_labels, - } = event.data - else { - continue; - }; - if event.event != crate::api::schema::EventKind::PaneAgentStatusChanged { - continue; - } - if pane_id != self.pane_id { - continue; - } - saw_status_event = true; - - let current_presentation = - PanePresentationSnapshot::from_event(&title, &display_agent, &state_labels); - self.last_status = Some(agent_status); - self.last_presentation = Some(current_presentation); - if self - .status_filter - .is_some_and(|wanted| wanted != agent_status) - { - continue; - } - - self.initial_event = None; - return Ok(Some(SubscriptionEventEnvelope { - event: SubscriptionEventKind::PaneAgentStatusChanged, - data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent { - pane_id, - workspace_id, - agent_status, - agent, - title, - display_agent, - state_labels, - }), - })); - } + }), + }) + } - if saw_status_event { - self.initial_event = None; - } else if event_hub.current_sequence() != self.last_sequence { + fn poll_snapshot( + &mut self, + api_tx: &ApiRequestSender, + event_hub: &EventHub, + ) -> Result<Option<SubscriptionEventEnvelope>, ErrorResponse> { + if event_hub.current_sequence() != self.last_sequence { return Ok(None); } else if let Some(event) = self.initial_event.take() { return Ok(Some(SubscriptionEventEnvelope { @@ -688,6 +769,96 @@ mod tests { } #[test] + fn lifecycle_batch_drains_in_order_and_advances_past_unmatched_events() { + let event_hub = EventHub::default(); + event_hub.push(workspace_focused_event("old")); + let start = event_hub.current_sequence(); + event_hub.push(workspace_focused_event("setup")); + let (api_tx, _api_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut subscription = ActiveSubscription::new( + Subscription::WorkspaceFocused {}, + "batch", + 0, + &api_tx, + &event_hub, + start, + ) + .unwrap(); + event_hub.push(presentation_event(None)); + event_hub.push(workspace_focused_event("live")); + let events = subscription.poll_batch(&api_tx, &event_hub).unwrap(); + assert_eq!(events.len(), 2); + assert_eq!(events[0]["data"]["workspace_id"], "setup"); + assert_eq!(events[1]["data"]["workspace_id"], "live"); + assert!(subscription + .poll_batch(&api_tx, &event_hub) + .unwrap() + .is_empty()); + let ActiveSubscription::Event(subscription) = subscription else { + panic!("expected lifecycle subscription"); + }; + assert_eq!(subscription.last_sequence, event_hub.current_sequence()); + } + + #[test] + fn agent_status_batch_preserves_transitions_filters_and_initial_state_ordering() { + for filtered in [false, true] { + let event_hub = EventHub::default(); + let mut subscription = ActiveSubscription::AgentStatusChanged(Box::new( + ActiveAgentStatusChangedSubscription { + pane_id: "pane_1".into(), + status_filter: filtered.then_some(AgentStatus::Working), + last_status: Some(AgentStatus::Working), + last_presentation: None, + last_sequence: event_hub.current_sequence(), + initial_event: Some(PaneAgentStatusChangedEvent { + pane_id: "pane_1".into(), + workspace_id: "workspace_1".into(), + agent_status: AgentStatus::Working, + agent: Some("pi".into()), + title: Some("stale initial snapshot".into()), + display_agent: None, + state_labels: HashMap::new(), + }), + request_prefix: "batch".into(), + }, + )); + for (status, title) in [ + (AgentStatus::Working, "started"), + (AgentStatus::Blocked, "approval"), + (AgentStatus::Idle, "finished"), + (AgentStatus::Working, "restarted"), + ] { + let mut event = presentation_event(Some(title)); + let EventData::PaneAgentStatusChanged { agent_status, .. } = &mut event.data else { + panic!("expected status data"); + }; + *agent_status = status; + event_hub.push(event); + } + let (api_tx, _api_rx) = tokio::sync::mpsc::unbounded_channel(); + let events = subscription.poll_batch(&api_tx, &event_hub).unwrap(); + let titles = events + .iter() + .map(|event| event["data"]["title"].as_str().unwrap()) + .collect::<Vec<_>>(); + assert_eq!( + titles, + if filtered { + vec!["started", "restarted"] + } else { + vec!["started", "approval", "finished", "restarted"] + } + ); + let ActiveSubscription::AgentStatusChanged(subscription) = subscription else { + panic!("expected agent subscription"); + }; + assert_eq!(subscription.last_sequence, event_hub.current_sequence()); + assert!(subscription.initial_event.is_none()); + } + } + + #[test] fn scroll_subscription_emits_when_scroll_snapshot_changes() { let at_bottom = PaneScrollInfo { offset_from_bottom: 0, |