1 files changed, 119 insertions, 2 deletions
diff --git a/src/api/wait.rs b/src/api/wait.rs
index 3ed08d70..e70472c3 100644
--- a/src/api/wait.rs
+++ b/src/api/wait.rs
@@ -4,14 +4,17 @@ use std::sync::Arc;
 use regex::Regex;
 
 use crate::api::schema::{
-    ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse,
+    ErrorBody, ErrorResponse, EventData, EventEnvelope, EventKind, EventMatch, EventsWaitParams,
+    Method, Request, ResponseResult, Subscription, SubscriptionEventData,
+    SubscriptionEventEnvelope, SuccessResponse,
 };
 use crate::api::server::{
     dispatch_to_app_with_timeout, should_stop_connection, APP_RESPONSE_TIMEOUT,
     CONNECTION_POLL_INTERVAL,
 };
+use crate::api::subscriptions::ActiveSubscription;
 use crate::api::subscriptions::{match_output, output_match_read_source};
-use crate::api::ApiRequestSender;
+use crate::api::{ApiRequestSender, EventHub};
 use crate::ipc::LocalStream;
 
 pub(super) fn wait_for_output(
@@ -122,3 +125,117 @@ pub(super) fn wait_for_output(
         std::thread::sleep(CONNECTION_POLL_INTERVAL);
     }
 }
+
+pub(super) fn wait_for_event(
+    request_id: String,
+    params: EventsWaitParams,
+    stream: &mut LocalStream,
+    api_tx: &ApiRequestSender,
+    event_hub: &EventHub,
+    running: &Arc<AtomicBool>,
+) -> std::io::Result<Option<String>> {
+    let deadline = params
+        .timeout_ms
+        .map(|ms| std::time::Instant::now() + std::time::Duration::from_millis(ms));
+
+    let subscription = match event_match_subscription(&request_id, params.match_event) {
+        Ok(subscription) => subscription,
+        Err(response) => return Ok(Some(serde_json::to_string(&response).unwrap())),
+    };
+    let mut active = match ActiveSubscription::new(subscription, &request_id, 0, api_tx, event_hub)
+    {
+        Ok(active) => active,
+        Err(response) => return Ok(Some(serde_json::to_string(&response).unwrap())),
+    };
+
+    loop {
+        if should_stop_connection(stream, running)? {
+            return Ok(None);
+        }
+
+        if let Some(event) = active.poll(api_tx, event_hub) {
+            return Ok(Some(wait_matched_response(&request_id, event)));
+        }
+
+        if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) {
+            return Ok(Some(
+                serde_json::to_string(&ErrorResponse {
+                    id: request_id,
+                    error: ErrorBody {
+                        code: "timeout".into(),
+                        message: "timed out waiting for event match".into(),
+                    },
+                })
+                .unwrap(),
+            ));
+        }
+
+        std::thread::sleep(CONNECTION_POLL_INTERVAL);
+    }
+}
+
+fn event_match_subscription(
+    request_id: &str,
+    match_event: EventMatch,
+) -> Result<Subscription, ErrorResponse> {
+    match match_event {
+        EventMatch::PaneAgentStatusChanged {
+            pane_id,
+            agent_status,
+        } => Ok(Subscription::PaneAgentStatusChanged {
+            pane_id,
+            agent_status: Some(agent_status),
+        }),
+        _ => Err(ErrorResponse {
+            id: request_id.into(),
+            error: ErrorBody {
+                code: "unsupported_event_wait_match".into(),
+                message: "events.wait currently supports pane agent status matches".into(),
+            },
+        }),
+    }
+}
+
+fn wait_matched_response(request_id: &str, event: serde_json::Value) -> String {
+    let Ok(event) = serde_json::from_value::<SubscriptionEventEnvelope>(event) else {
+        return serde_json::to_string(&ErrorResponse {
+            id: request_id.into(),
+            error: ErrorBody {
+                code: "internal_error".into(),
+                message: "failed to decode matched event".into(),
+            },
+        })
+        .unwrap();
+    };
+
+    let SubscriptionEventData::PaneAgentStatusChanged(data) = event.data else {
+        return serde_json::to_string(&ErrorResponse {
+            id: request_id.into(),
+            error: ErrorBody {
+                code: "unsupported_event_wait_match".into(),
+                message: "events.wait currently supports pane agent status matches".into(),
+            },
+        })
+        .unwrap();
+    };
+
+    serde_json::to_string(&SuccessResponse {
+        id: request_id.into(),
+        result: ResponseResult::WaitMatched {
+            event: EventEnvelope {
+                event: EventKind::PaneAgentStatusChanged,
+                data: EventData::PaneAgentStatusChanged {
+                    pane_id: data.pane_id,
+                    workspace_id: data.workspace_id,
+                    agent_status: data.agent_status,
+                    agent: data.agent,
+                    title: data.title,
+                    display_agent: data.display_agent,
+                    custom_status: data.custom_status,
+                    state_labels: data.state_labels,
+                },
+            },
+        },
+    })
+    .unwrap()
+}