aboutsummaryrefslogtreecommitdiff
path: root/src/api/subscriptions.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/api/subscriptions.rs')
-rw-r--r--src/api/subscriptions.rs306
1 files changed, 283 insertions, 23 deletions
diff --git a/src/api/subscriptions.rs b/src/api/subscriptions.rs
index aaa7e5ac..9c92a41b 100644
--- a/src/api/subscriptions.rs
+++ b/src/api/subscriptions.rs
@@ -49,8 +49,9 @@ pub(super) struct ActiveAgentStatusChangedSubscription {
pane_id: String,
status_filter: Option<crate::api::schema::AgentStatus>,
last_status: Option<crate::api::schema::AgentStatus>,
- emit_initial_match: bool,
last_presentation: Option<PanePresentationSnapshot>,
+ last_sequence: u64,
+ initial_event: Option<PaneAgentStatusChangedEvent>,
request_prefix: String,
}
@@ -71,6 +72,20 @@ impl PanePresentationSnapshot {
state_labels: pane.state_labels.clone(),
}
}
+
+ fn from_event(
+ title: &Option<String>,
+ display_agent: &Option<String>,
+ custom_status: &Option<String>,
+ state_labels: &std::collections::HashMap<String, String>,
+ ) -> Self {
+ Self {
+ title: title.clone(),
+ display_agent: display_agent.clone(),
+ custom_status: custom_status.clone(),
+ state_labels: state_labels.clone(),
+ }
+ }
}
pub(super) struct ActiveEventSubscription {
@@ -81,7 +96,7 @@ pub(super) struct ActiveEventSubscription {
pub(super) enum ActiveSubscription {
Event(ActiveEventSubscription),
OutputMatched(ActiveOutputMatchedSubscription),
- AgentStatusChanged(ActiveAgentStatusChangedSubscription),
+ AgentStatusChanged(Box<ActiveAgentStatusChangedSubscription>),
}
impl ActiveSubscription {
@@ -90,7 +105,7 @@ impl ActiveSubscription {
request_id: &str,
index: usize,
api_tx: &ApiRequestSender,
- _event_hub: &EventHub,
+ event_hub: &EventHub,
) -> Result<Self, ErrorResponse> {
match subscription {
Subscription::WorkspaceCreated {} => Ok(Self::Event(ActiveEventSubscription {
@@ -197,20 +212,34 @@ impl ActiveSubscription {
pane_id,
agent_status,
} => {
+ let last_sequence = event_hub.current_sequence();
let probe = pane_get(format!("{request_id}:sub:{index}:probe"), &pane_id, api_tx)?;
- let emit_initial_match =
- agent_status.is_some_and(|wanted| wanted == probe.agent_status);
+ let last_status = probe.agent_status;
+ let last_presentation = PanePresentationSnapshot::from(&probe);
+ let initial_event = agent_status
+ .is_some_and(|wanted| wanted == probe.agent_status)
+ .then_some(PaneAgentStatusChangedEvent {
+ pane_id: probe.pane_id,
+ workspace_id: probe.workspace_id,
+ agent_status: probe.agent_status,
+ agent: probe.agent,
+ title: probe.title,
+ display_agent: probe.display_agent,
+ custom_status: probe.custom_status,
+ state_labels: probe.state_labels,
+ });
- Ok(Self::AgentStatusChanged(
+ Ok(Self::AgentStatusChanged(Box::new(
ActiveAgentStatusChangedSubscription {
pane_id,
status_filter: agent_status,
- last_status: Some(probe.agent_status),
- emit_initial_match,
- last_presentation: Some(PanePresentationSnapshot::from(&probe)),
+ last_status: Some(last_status),
+ last_presentation: Some(last_presentation),
+ last_sequence,
+ initial_event,
request_prefix: format!("{request_id}:sub:{index}"),
},
- ))
+ )))
}
}
}
@@ -226,7 +255,7 @@ impl ActiveSubscription {
serde_json::to_value(subscription.poll(api_tx)?).ok()
}
Self::AgentStatusChanged(subscription) => {
- serde_json::to_value(subscription.poll(api_tx)?).ok()
+ serde_json::to_value(subscription.poll(api_tx, event_hub)?).ok()
}
}
}
@@ -281,13 +310,100 @@ impl ActiveOutputMatchedSubscription {
}
impl ActiveAgentStatusChangedSubscription {
- fn poll(&mut self, api_tx: &ApiRequestSender) -> Option<SubscriptionEventEnvelope> {
+ fn poll(
+ &mut self,
+ api_tx: &ApiRequestSender,
+ event_hub: &EventHub,
+ ) -> Option<SubscriptionEventEnvelope> {
+ 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 {
+ pane_id,
+ workspace_id,
+ agent_status,
+ agent,
+ title,
+ display_agent,
+ custom_status,
+ 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,
+ &custom_status,
+ &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 Some(SubscriptionEventEnvelope {
+ event: SubscriptionEventKind::PaneAgentStatusChanged,
+ data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent {
+ pane_id,
+ workspace_id,
+ agent_status,
+ agent,
+ title,
+ display_agent,
+ custom_status,
+ state_labels,
+ }),
+ });
+ }
+
+ if saw_status_event {
+ self.initial_event = None;
+ } else if event_hub.current_sequence() != self.last_sequence {
+ return None;
+ } else if let Some(event) = self.initial_event.take() {
+ return Some(SubscriptionEventEnvelope {
+ event: SubscriptionEventKind::PaneAgentStatusChanged,
+ data: SubscriptionEventData::PaneAgentStatusChanged(event),
+ });
+ }
+
+ let before_snapshot_sequence = self.last_sequence;
let pane = pane_get(
format!("{}:pane", self.request_prefix),
&self.pane_id,
api_tx,
)
.ok()?;
+ let after_snapshot_sequence = event_hub.current_sequence();
+ if after_snapshot_sequence != before_snapshot_sequence {
+ return None;
+ }
+
+ let event = self.event_from_snapshot(pane);
+ if event.is_some() {
+ self.last_sequence = after_snapshot_sequence;
+ }
+ event
+ }
+
+ fn event_from_snapshot(
+ &mut self,
+ pane: crate::api::schema::PaneInfo,
+ ) -> Option<SubscriptionEventEnvelope> {
let current_status = pane.agent_status;
let current_presentation = PanePresentationSnapshot::from(&pane);
let previous_status = self.last_status.replace(current_status);
@@ -296,17 +412,13 @@ impl ActiveAgentStatusChangedSubscription {
.as_ref()
.is_some_and(|previous| previous != &current_presentation);
let status_changed = previous_status.is_some_and(|previous| previous != current_status);
- let should_emit = if self.emit_initial_match {
- self.emit_initial_match = false;
- self.status_filter
- .is_some_and(|wanted| wanted == current_status)
- } else {
- (status_changed || presentation_changed)
- && !self
- .status_filter
- .is_some_and(|wanted| wanted != current_status)
- };
- if !should_emit {
+ if !(status_changed || presentation_changed) {
+ return None;
+ }
+ if self
+ .status_filter
+ .is_some_and(|wanted| wanted != current_status)
+ {
return None;
}
@@ -412,3 +524,151 @@ fn pane_get(
},
})
}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+
+ use super::*;
+ use crate::api::schema::{AgentStatus, EventData, EventEnvelope, EventKind};
+
+ fn status_event(custom_status: Option<&str>) -> EventEnvelope {
+ EventEnvelope {
+ event: EventKind::PaneAgentStatusChanged,
+ data: EventData::PaneAgentStatusChanged {
+ pane_id: "pane_1".into(),
+ workspace_id: "workspace_1".into(),
+ agent_status: AgentStatus::Working,
+ agent: Some("pi".into()),
+ title: None,
+ display_agent: None,
+ custom_status: custom_status.map(str::to_string),
+ state_labels: HashMap::new(),
+ },
+ }
+ }
+
+ #[test]
+ fn agent_status_subscription_replays_queued_metadata_set_and_expiry_events() {
+ let event_hub = EventHub::default();
+ let mut subscription = ActiveAgentStatusChangedSubscription {
+ pane_id: "pane_1".into(),
+ status_filter: None,
+ last_status: Some(AgentStatus::Working),
+ last_presentation: Some(PanePresentationSnapshot {
+ title: None,
+ display_agent: None,
+ custom_status: None,
+ state_labels: HashMap::new(),
+ }),
+ last_sequence: event_hub.current_sequence(),
+ initial_event: None,
+ request_prefix: "test".into(),
+ };
+
+ event_hub.push(status_event(Some("short lived")));
+ event_hub.push(status_event(None));
+
+ let set_event = subscription
+ .poll(&tokio::sync::mpsc::unbounded_channel().0, &event_hub)
+ .expect("set event");
+ let SubscriptionEventData::PaneAgentStatusChanged(set_data) = set_event.data else {
+ panic!("wrong event data");
+ };
+ assert_eq!(set_data.custom_status.as_deref(), Some("short lived"));
+
+ let expiry_event = subscription
+ .poll(&tokio::sync::mpsc::unbounded_channel().0, &event_hub)
+ .expect("expiry event");
+ let SubscriptionEventData::PaneAgentStatusChanged(expiry_data) = expiry_event.data else {
+ panic!("wrong event data");
+ };
+ assert_eq!(expiry_data.custom_status, None);
+ }
+
+ #[test]
+ fn agent_status_subscription_prefers_setup_window_events_over_initial_snapshot() {
+ let event_hub = EventHub::default();
+ let mut subscription = ActiveAgentStatusChangedSubscription {
+ pane_id: "pane_1".into(),
+ status_filter: Some(AgentStatus::Working),
+ last_status: Some(AgentStatus::Working),
+ last_presentation: Some(PanePresentationSnapshot {
+ title: None,
+ display_agent: None,
+ custom_status: None,
+ state_labels: HashMap::new(),
+ }),
+ 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: None,
+ display_agent: None,
+ custom_status: None,
+ state_labels: HashMap::new(),
+ }),
+ request_prefix: "test".into(),
+ };
+
+ event_hub.push(status_event(Some("short lived")));
+ event_hub.push(status_event(None));
+
+ let set_event = subscription
+ .poll(&tokio::sync::mpsc::unbounded_channel().0, &event_hub)
+ .expect("set event");
+ let SubscriptionEventData::PaneAgentStatusChanged(set_data) = set_event.data else {
+ panic!("wrong event data");
+ };
+ assert_eq!(set_data.custom_status.as_deref(), Some("short lived"));
+
+ let expiry_event = subscription
+ .poll(&tokio::sync::mpsc::unbounded_channel().0, &event_hub)
+ .expect("expiry event");
+ let SubscriptionEventData::PaneAgentStatusChanged(expiry_data) = expiry_event.data else {
+ panic!("wrong event data");
+ };
+ assert_eq!(expiry_data.custom_status, None);
+ }
+
+ #[test]
+ fn agent_status_subscription_emits_setup_window_event_already_reflected_by_probe() {
+ let event_hub = EventHub::default();
+ let mut subscription = ActiveAgentStatusChangedSubscription {
+ pane_id: "pane_1".into(),
+ status_filter: Some(AgentStatus::Working),
+ last_status: Some(AgentStatus::Working),
+ last_presentation: Some(PanePresentationSnapshot {
+ title: None,
+ display_agent: None,
+ custom_status: Some("short lived".into()),
+ state_labels: HashMap::new(),
+ }),
+ 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: None,
+ display_agent: None,
+ custom_status: Some("short lived".into()),
+ state_labels: HashMap::new(),
+ }),
+ request_prefix: "test".into(),
+ };
+
+ event_hub.push(status_event(Some("short lived")));
+
+ let event = subscription
+ .poll(&tokio::sync::mpsc::unbounded_channel().0, &event_hub)
+ .expect("setup-window event");
+ let SubscriptionEventData::PaneAgentStatusChanged(data) = event.data else {
+ panic!("wrong event data");
+ };
+ assert_eq!(data.custom_status.as_deref(), Some("short lived"));
+ assert!(subscription.initial_event.is_none());
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-06 22:47:11 +0000