From 045f506ec8a333cde84718304a7d21bcdb93a1fc Mon Sep 17 00:00:00 2001 From: Ogulcan Celik Date: Fri, 3 Jul 2026 20:38:05 +0300 Subject: [PATCH] fix: keep windows wait connections open refs #963 --- Cargo.toml | 1 + src/api/server.rs | 186 +++++++++++++++++++++++++++++++++-------- src/api/wait.rs | 121 ++++++++++++++++++++++++++- src/cli.rs | 90 ++++++++++++++++---- src/ipc.rs | 104 +++++++++++++++++++++++ src/server/headless.rs | 8 +- tests/cli_wrapper.rs | 42 ++++++++++ 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 { - 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::(); + 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(); @@ -783,6 +841,60 @@ mod tests { assert_eq!(parsed.id, "req_2"); } + #[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::(); 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, +) -> std::io::Result> { + 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 { + 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::(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 { 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, +) -> std::io::Result { + 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 { + probe_stream_closed(stream) +} + +#[cfg(unix)] +fn probe_stream_closed(stream: &mut LocalStream) -> io::Result { + 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 { + 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 { #[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] @@ -193,6 +269,34 @@ mod tests { let _ = fs::remove_file(&path); } + #[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, - #[cfg(unix)] + // Kept on every platform so dropping HeadlessServer owns API server shutdown. + #[cfg_attr(windows, allow(dead_code))] api_server: Option, #[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 @@ -4217,6 +4217,48 @@ fn wait_agent_status_exits_immediately_when_status_already_matches() { cleanup_spawned_herdr(herdr, base); } +#[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();