fix: keep windows wait connections open

refs #963

7 files changed, 493 insertions, 59 deletions

diff --git a/Cargo.toml b/Cargo.toml
index 06f08d1e..3b96c7b4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -53,6 +53,7 @@ windows-sys = { version = "0.61.2", features = [
     "Win32_System_Diagnostics_ToolHelp",
     "Win32_System_Console",
     "Win32_System_Kernel",
+    "Win32_System_Pipes",
     "Win32_System_Threading",
     "Win32_UI_Shell",
     "Win32_UI_WindowsAndMessaging",
diff --git a/src/api/server.rs b/src/api/server.rs
index 37b464ac..df0f8fc3 100644
--- a/src/api/server.rs
+++ b/src/api/server.rs
@@ -14,11 +14,11 @@ use crate::api::schema::{
     ErrorBody, ErrorResponse, Method, Request, ResponseResult, ServerCapabilities, SuccessResponse,
 };
 use crate::api::subscriptions::ActiveSubscription;
-use crate::api::wait::wait_for_output;
+use crate::api::wait::{wait_for_event, wait_for_output};
 use crate::api::{request_changes_ui, socket_path, ApiRequestMessage, ApiRequestSender, EventHub};
 use crate::ipc::{
-    bind_local_listener, remove_socket_file_if_owned, socket_file_identity, LocalStream,
-    SocketFileIdentity,
+    bind_local_listener, is_connection_closed_error, local_stream_peer_closed,
+    remove_socket_file_if_owned, socket_file_identity, LocalStream, SocketFileIdentity,
 };
 
 const SOCKET_PERMISSION_MODE: u32 = 0o600;
@@ -197,6 +197,38 @@ fn handle_connection(
             }
             result
         }
+        Method::EventsWait(params) => {
+            let Some(response) = wait_for_event(
+                request_id.clone(),
+                params,
+                &mut stream,
+                api_tx,
+                event_hub,
+                running,
+            )?
+            else {
+                crate::logging::api_request_completed(
+                    &request_id,
+                    method,
+                    "client_disconnected",
+                    changes_ui,
+                );
+                return Ok(());
+            };
+            let result = write_text_line_allow_disconnect(&mut stream, &response);
+            match &result {
+                Ok(()) => crate::logging::api_request_completed(
+                    &request_id,
+                    method,
+                    api_response_outcome(&response),
+                    changes_ui,
+                ),
+                Err(err) => {
+                    crate::logging::api_request_failed(&request_id, method, &err.to_string())
+                }
+            }
+            result
+        }
         Method::PaneWaitForOutput(params) => {
             let Some(response) =
                 wait_for_output(request_id.clone(), params, &mut stream, api_tx, running)?
@@ -515,40 +547,7 @@ pub(super) fn should_stop_connection(
         return Ok(true);
     }
 
-    probe_stream_closed(stream)
-}
-
-fn probe_stream_closed(stream: &mut LocalStream) -> std::io::Result<bool> {
-    stream.set_nonblocking(true)?;
-    let mut probe = [0u8; 1];
-    let status = match stream.read(&mut probe) {
-        Ok(0) => Ok(true),
-        Ok(_) => Ok(true),
-        Err(err)
-            if matches!(
-                err.kind(),
-                std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
-            ) =>
-        {
-            Ok(false)
-        }
-        Err(err) if is_connection_closed_error(&err) => Ok(true),
-        Err(err) => Err(err),
-    };
-    stream.set_nonblocking(false)?;
-    status
-}
-
-fn is_connection_closed_error(err: &std::io::Error) -> bool {
-    matches!(
-        err.kind(),
-        std::io::ErrorKind::BrokenPipe
-            | std::io::ErrorKind::ConnectionAborted
-            | std::io::ErrorKind::ConnectionReset
-            | std::io::ErrorKind::NotConnected
-            | std::io::ErrorKind::UnexpectedEof
-            | std::io::ErrorKind::WriteZero
-    )
+    local_stream_peer_closed(stream)
 }
 
 fn dispatch_to_app(request: Request, api_tx: &ApiRequestSender) -> String {
@@ -620,6 +619,7 @@ fn error_response_json(id: String, code: &str, message: String) -> String {
 mod tests {
     use super::*;
     use interprocess::local_socket::traits::Listener as _;
+    use std::collections::HashMap;
     use std::io::{BufRead, BufReader};
     use std::os::unix::fs::PermissionsExt;
     use std::os::unix::net::UnixListener;
@@ -654,6 +654,64 @@ mod tests {
         (client, server, path)
     }
 
+    fn pane_info(
+        pane_id: &str,
+        agent_status: crate::api::schema::AgentStatus,
+    ) -> crate::api::schema::PaneInfo {
+        crate::api::schema::PaneInfo {
+            pane_id: pane_id.into(),
+            terminal_id: "term_1".into(),
+            workspace_id: "ws_1".into(),
+            tab_id: "tab_1".into(),
+            focused: true,
+            cwd: None,
+            foreground_cwd: None,
+            label: None,
+            agent: Some("pi".into()),
+            title: None,
+            display_agent: None,
+            agent_status,
+            custom_status: None,
+            state_labels: HashMap::new(),
+            agent_session: None,
+            revision: 0,
+        }
+    }
+
+    fn spawn_pane_get_responder(
+        agent_status: crate::api::schema::AgentStatus,
+    ) -> (ApiRequestSender, std::thread::JoinHandle<()>) {
+        let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
+        let responder = std::thread::spawn(move || {
+            while let Some(msg) = api_rx.blocking_recv() {
+                match msg.request.method {
+                    Method::PaneGet(_) => msg
+                        .respond_to
+                        .send(
+                            serde_json::to_string(&SuccessResponse {
+                                id: msg.request.id,
+                                result: ResponseResult::PaneInfo {
+                                    pane: pane_info("pane_1", agent_status),
+                                },
+                            })
+                            .unwrap(),
+                        )
+                        .unwrap(),
+                    Method::EventsWait(_) => msg
+                        .respond_to
+                        .send(error_response_json(
+                            msg.request.id,
+                            "unexpected_dispatch",
+                            "events.wait should be handled by the api server".into(),
+                        ))
+                        .unwrap(),
+                    other => panic!("unexpected request: {other:?}"),
+                }
+            }
+        });
+        (api_tx, responder)
+    }
+
     #[test]
     fn socket_path_prefers_explicit_env_override() {
         let _guard = env_lock().lock().unwrap();
@@ -784,6 +842,60 @@ mod tests {
     }
 
     #[test]
+    fn events_wait_agent_status_returns_initial_match() {
+        let (api_tx, responder) =
+            spawn_pane_get_responder(crate::api::schema::AgentStatus::Blocked);
+
+        let (mut client, server, _path) = local_stream_pair("api-events-wait-initial");
+        client
+            .write_all(br#"{"id":"wait_1","method":"events.wait","params":{"match_event":{"event":"pane_agent_status_changed","pane_id":"pane_1","agent_status":"blocked"},"timeout_ms":1000}}"#)
+            .unwrap();
+        client.write_all(b"\n").unwrap();
+        client.flush().unwrap();
+
+        let running = Arc::new(AtomicBool::new(true));
+        let event_hub = EventHub::default();
+        handle_connection(server, &api_tx, &event_hub, &running, None).unwrap();
+
+        let response: serde_json::Value = serde_json::from_str(&read_line(&mut client)).unwrap();
+        assert_eq!(response["id"], "wait_1");
+        assert_eq!(response["result"]["type"], "wait_matched");
+        assert_eq!(
+            response["result"]["event"]["data"]["agent_status"],
+            "blocked"
+        );
+        drop(api_tx);
+        responder.join().unwrap();
+    }
+
+    #[test]
+    fn events_wait_agent_status_times_out_server_side() {
+        let (api_tx, responder) =
+            spawn_pane_get_responder(crate::api::schema::AgentStatus::Unknown);
+
+        let (mut client, server, _path) = local_stream_pair("api-events-wait-timeout");
+        client
+            .write_all(br#"{"id":"wait_2","method":"events.wait","params":{"match_event":{"event":"pane_agent_status_changed","pane_id":"pane_1","agent_status":"blocked"},"timeout_ms":30}}"#)
+            .unwrap();
+        client.write_all(b"\n").unwrap();
+        client.flush().unwrap();
+
+        let running = Arc::new(AtomicBool::new(true));
+        let event_hub = EventHub::default();
+        handle_connection(server, &api_tx, &event_hub, &running, None).unwrap();
+
+        let response: serde_json::Value = serde_json::from_str(&read_line(&mut client)).unwrap();
+        assert_eq!(response["id"], "wait_2");
+        assert_eq!(response["error"]["code"], "timeout");
+        assert_eq!(
+            response["error"]["message"],
+            "timed out waiting for event match"
+        );
+        drop(api_tx);
+        responder.join().unwrap();
+    }
+
+    #[test]
     fn wait_for_output_stops_when_client_disconnects() {
         let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
         let (first_read_tx, first_read_rx) = std::sync::mpsc::channel();
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()
+}
diff --git a/src/cli.rs b/src/cli.rs
index 70420b09..2f9ef0d7 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -4,8 +4,10 @@ use serde::Serialize;
 
 use crate::api::client::{ApiClient, ApiClientError};
 use crate::api::schema::{
-    AgentStatus, ClientWindowTitleSetParams, EmptyParams, Method, OutputMatch, PaneAgentState,
-    PaneWaitForOutputParams, ReadFormat, ReadSource, Request, SplitDirection, Subscription,
+    AgentStatus, ClientWindowTitleSetParams, EmptyParams, EventData, EventMatch, EventsWaitParams,
+    Method, OutputMatch, PaneAgentState, PaneWaitForOutputParams, ReadFormat, ReadSource, Request,
+    ResponseResult, SplitDirection, SubscriptionEventData, SubscriptionEventEnvelope,
+    SubscriptionEventKind,
 };
 
 mod agent;
@@ -820,19 +822,77 @@ fn wait_agent_status(args: &[String]) -> std::io::Result<i32> {
         return Ok(2);
     };
 
-    wait_for_agent_change(
-        Request {
-            id: "cli:wait:agent-status".into(),
-            method: Method::EventsSubscribe(crate::api::schema::EventsSubscribeParams {
-                subscriptions: vec![Subscription::PaneAgentStatusChanged {
-                    pane_id,
-                    agent_status: Some(agent_status),
-                }],
-            }),
-        },
-        timeout_ms,
-        "timed out waiting for agent status change",
-    )
+    wait_for_agent_status_change(pane_id, agent_status, timeout_ms)
+}
+
+fn wait_for_agent_status_change(
+    pane_id: String,
+    agent_status: AgentStatus,
+    timeout_ms: Option<u64>,
+) -> std::io::Result<i32> {
+    let request = Request {
+        id: "cli:wait:agent-status".into(),
+        method: Method::EventsWait(EventsWaitParams {
+            match_event: EventMatch::PaneAgentStatusChanged {
+                pane_id,
+                agent_status,
+            },
+            timeout_ms,
+        }),
+    };
+    let response = send_request(&request)?;
+    match crate::api::client::parse_response_value(response) {
+        Ok(success) => {
+            let ResponseResult::WaitMatched { event } = success.result else {
+                return Err(std::io::Error::other("unexpected wait response result"));
+            };
+            let EventData::PaneAgentStatusChanged {
+                pane_id,
+                workspace_id,
+                agent_status,
+                agent,
+                title,
+                display_agent,
+                custom_status,
+                state_labels,
+            } = event.data
+            else {
+                return Err(std::io::Error::other("unexpected wait event data"));
+            };
+            let event = SubscriptionEventEnvelope {
+                event: SubscriptionEventKind::PaneAgentStatusChanged,
+                data: SubscriptionEventData::PaneAgentStatusChanged(
+                    crate::api::schema::PaneAgentStatusChangedEvent {
+                        pane_id,
+                        workspace_id,
+                        agent_status,
+                        agent,
+                        custom_status,
+                        title,
+                        display_agent,
+                        state_labels,
+                    },
+                ),
+            };
+            println!(
+                "{}",
+                serde_json::to_string(&event).map_err(std::io::Error::other)?
+            );
+            Ok(0)
+        }
+        Err(ApiClientError::ErrorResponse(response)) => {
+            if response.error.code == "timeout" {
+                eprintln!("timed out waiting for agent status change");
+            } else {
+                eprintln!(
+                    "{}",
+                    serde_json::to_string(&response).map_err(std::io::Error::other)?
+                );
+            }
+            Ok(1)
+        }
+        Err(err) => Err(api_client_error_to_io(err)),
+    }
 }
 
 pub(super) fn wait_for_agent_change(
diff --git a/src/ipc.rs b/src/ipc.rs
index 4c92fec8..8b72dd4c 100644
--- a/src/ipc.rs
+++ b/src/ipc.rs
@@ -1,9 +1,14 @@
 use std::fs;
 use std::io;
 #[cfg(unix)]
+use std::io::Read;
+#[cfg(unix)]
 use std::os::unix::fs::{MetadataExt, PermissionsExt};
 use std::path::Path;
 
+#[cfg(unix)]
+use interprocess::local_socket::traits::Stream as _;
+
 pub(crate) type LocalListener = interprocess::local_socket::Listener;
 pub(crate) type LocalStream = interprocess::local_socket::Stream;
 
@@ -99,6 +104,75 @@ fn stale_socket_connect_error(kind: io::ErrorKind) -> bool {
     ) || (cfg!(windows) && kind == io::ErrorKind::WouldBlock)
 }
 
+pub(crate) fn local_stream_peer_closed(stream: &mut LocalStream) -> io::Result<bool> {
+    probe_stream_closed(stream)
+}
+
+#[cfg(unix)]
+fn probe_stream_closed(stream: &mut LocalStream) -> io::Result<bool> {
+    stream.set_nonblocking(true)?;
+    let mut probe = [0u8; 1];
+    let status = match stream.read(&mut probe) {
+        Ok(0) => Ok(true),
+        Ok(_) => Ok(true),
+        Err(err)
+            if matches!(
+                err.kind(),
+                io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
+            ) =>
+        {
+            Ok(false)
+        }
+        Err(err) if is_connection_closed_error(&err) => Ok(true),
+        Err(err) => Err(err),
+    };
+    stream.set_nonblocking(false)?;
+    status
+}
+
+#[cfg(windows)]
+fn probe_stream_closed(stream: &mut LocalStream) -> io::Result<bool> {
+    use std::os::windows::io::{AsHandle, AsRawHandle};
+
+    let LocalStream::NamedPipe(pipe) = stream;
+    let ok = unsafe {
+        windows_sys::Win32::System::Pipes::PeekNamedPipe(
+            pipe.as_handle().as_raw_handle(),
+            std::ptr::null_mut(),
+            0,
+            std::ptr::null_mut(),
+            std::ptr::null_mut(),
+            std::ptr::null_mut(),
+        )
+    };
+    if ok != 0 {
+        return Ok(false);
+    }
+
+    let err = io::Error::last_os_error();
+    if is_connection_closed_error(&err) || windows_named_pipe_closed_error(&err) {
+        return Ok(true);
+    }
+    Err(err)
+}
+
+pub(crate) fn is_connection_closed_error(err: &io::Error) -> bool {
+    matches!(
+        err.kind(),
+        io::ErrorKind::BrokenPipe
+            | io::ErrorKind::ConnectionAborted
+            | io::ErrorKind::ConnectionReset
+            | io::ErrorKind::NotConnected
+            | io::ErrorKind::UnexpectedEof
+            | io::ErrorKind::WriteZero
+    )
+}
+
+#[cfg(windows)]
+fn windows_named_pipe_closed_error(err: &io::Error) -> bool {
+    matches!(err.raw_os_error(), Some(6 | 109 | 232 | 233))
+}
+
 pub(crate) fn socket_file_identity(path: &Path) -> io::Result<SocketFileIdentity> {
     #[cfg(windows)]
     {
@@ -163,6 +237,8 @@ pub(crate) fn restrict_socket_permissions(_path: &Path, _mode: u32) -> io::Resul
 mod tests {
     use super::*;
     #[cfg(windows)]
+    use interprocess::local_socket::traits::Listener as _;
+    #[cfg(windows)]
     use std::path::PathBuf;
 
     #[test]
@@ -194,6 +270,34 @@ mod tests {
     }
 
     #[cfg(windows)]
+    #[test]
+    fn idle_named_pipe_peer_is_not_treated_as_closed() {
+        let path = temp_socket_marker_path("idle-pipe");
+        let listener = bind_local_listener(&path).unwrap();
+        let _client = connect_local_stream(&path).unwrap();
+        let mut server = listener.accept().unwrap();
+
+        assert!(!local_stream_peer_closed(&mut server).unwrap());
+
+        let _ = fs::remove_file(path);
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn disconnected_named_pipe_peer_is_treated_as_closed() {
+        let path = temp_socket_marker_path("disconnected-pipe");
+        let listener = bind_local_listener(&path).unwrap();
+        let client = connect_local_stream(&path).unwrap();
+        let mut server = listener.accept().unwrap();
+
+        drop(client);
+
+        assert!(local_stream_peer_closed(&mut server).unwrap());
+
+        let _ = fs::remove_file(path);
+    }
+
+    #[cfg(windows)]
     fn temp_socket_marker_path(name: &str) -> PathBuf {
         std::env::temp_dir().join(format!("herdr-{name}-{}.sock", std::process::id()))
     }
diff --git a/src/server/headless.rs b/src/server/headless.rs
index f0363ba1..d0c90f6c 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -193,7 +193,8 @@ pub struct HeadlessServer {
     app: app::App,
     #[cfg(unix)]
     api_tx: Option<api::ApiRequestSender>,
-    #[cfg(unix)]
+    // Kept on every platform so dropping HeadlessServer owns API server shutdown.
+    #[cfg_attr(windows, allow(dead_code))]
     api_server: Option<api::ServerHandle>,
     #[cfg(unix)]
     client_listener: LocalListener,
@@ -386,13 +387,11 @@ impl HeadlessServer {
         let (server_config_diagnostic, server_config_diagnostic_without_keybindings) =
             server_config_diagnostic_summaries(config_diagnostics);
         #[cfg(not(unix))]
-        let _ = (&api_tx, &api_server);
-
+        let _ = api_tx;
         Ok(Self {
             app,
             #[cfg(unix)]
             api_tx,
-            #[cfg(unix)]
             api_server,
             #[cfg(unix)]
             client_listener: listener,
@@ -4173,7 +4172,6 @@ mod tests {
             app,
             #[cfg(unix)]
             api_tx: None,
-            #[cfg(unix)]
             api_server: None,
             #[cfg(unix)]
             client_listener: listener,
diff --git a/tests/cli_wrapper.rs b/tests/cli_wrapper.rs
index 99d4f107..41e924ee 100644
--- a/tests/cli_wrapper.rs
+++ b/tests/cli_wrapper.rs
@@ -4218,6 +4218,48 @@ fn wait_agent_status_exits_immediately_when_status_already_matches() {
 }
 
 #[test]
+fn wait_agent_status_times_out_when_status_does_not_match() {
+    let base = unique_test_dir();
+    let config_home = base.join("config");
+    let runtime_dir = base.join("runtime");
+    let socket_path = runtime_dir.join("herdr.sock");
+
+    let herdr = spawn_herdr(&config_home, &runtime_dir, &socket_path);
+    wait_for_socket(&socket_path, Duration::from_secs(5));
+
+    let created = send_request(
+        &socket_path,
+        &format!(
+            r#"{{"id":"req_cli_timeout_1","method":"workspace.create","params":{{"cwd":"{}","focus":true}}}}"#,
+            base.display()
+        ),
+    );
+    assert_eq!(created["result"]["type"], "workspace_created");
+
+    let waited = run_cli(
+        &socket_path,
+        &[
+            "wait",
+            "agent-status",
+            "1-1",
+            "--status",
+            "blocked",
+            "--timeout",
+            "100",
+        ],
+    );
+    assert!(!waited.status.success());
+    assert!(
+        String::from_utf8_lossy(&waited.stderr)
+            .contains("timed out waiting for agent status change"),
+        "stderr: {}",
+        String::from_utf8_lossy(&waited.stderr)
+    );
+
+    cleanup_spawned_herdr(herdr, base);
+}
+
+#[test]
 fn wait_agent_status_exits_when_done_status_matches() {
     let base = unique_test_dir();
     let config_home = base.join("config");