aboutsummaryrefslogtreecommitdiff
path: root/src/api/mod.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/api/mod.rs')
-rw-r--r--src/api/mod.rs1356
1 files changed, 11 insertions, 1345 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs
index 4893c295..34b37138 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -1,32 +1,23 @@
+pub mod client;
+mod event_hub;
pub mod schema;
+mod server;
+mod status;
+mod subscriptions;
+mod wait;
-use std::fs;
-use std::io::{self, BufRead, BufReader, Read, Write};
-use std::os::unix::net::{UnixListener, UnixStream};
-use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
-use std::time::Duration;
+pub use event_hub::EventHub;
+pub use server::start_server;
+pub use status::{read_runtime_status_at, RuntimeStatus};
-use tracing::{debug, error, info, warn};
+use std::path::PathBuf;
-use regex::Regex;
use tokio::sync::mpsc;
-use crate::api::schema::{
- ErrorBody, ErrorResponse, Method, PaneAgentStatusChangedEvent, PaneOutputMatchedEvent, Request,
- ResponseResult, Subscription, SubscriptionEventData, SubscriptionEventEnvelope,
- SubscriptionEventKind, SuccessResponse,
-};
+use crate::api::schema::{Method, Request};
pub const SOCKET_PATH_ENV_VAR: &str = "HERDR_SOCKET_PATH";
-const SOCKET_PERMISSION_MODE: u32 = 0o600;
-const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100);
-const APP_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
-const INITIAL_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
-const STREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
-
pub(crate) fn request_changes_ui(request: &Request) -> bool {
matches!(
&request.method,
@@ -58,1331 +49,6 @@ pub struct ApiRequestMessage {
pub type ApiRequestSender = mpsc::UnboundedSender<ApiRequestMessage>;
-#[derive(Clone, Default)]
-pub struct EventHub {
- inner: std::sync::Arc<std::sync::Mutex<EventHubState>>,
-}
-
-#[derive(Default)]
-struct EventHubState {
- next_sequence: u64,
- events: Vec<(u64, crate::api::schema::EventEnvelope)>,
-}
-
-impl EventHub {
- const MAX_EVENTS: usize = 512;
-
- pub fn push(&self, event: crate::api::schema::EventEnvelope) {
- let Ok(mut state) = self.inner.lock() else {
- return;
- };
- state.next_sequence += 1;
- let sequence = state.next_sequence;
- state.events.push((sequence, event));
- let overflow = state.events.len().saturating_sub(Self::MAX_EVENTS);
- if overflow > 0 {
- state.events.drain(0..overflow);
- }
- }
-
- pub fn events_after(&self, sequence: u64) -> Vec<(u64, crate::api::schema::EventEnvelope)> {
- let Ok(state) = self.inner.lock() else {
- return Vec::new();
- };
- state
- .events
- .iter()
- .filter(|(event_sequence, _)| *event_sequence > sequence)
- .cloned()
- .collect()
- }
-}
-
pub fn socket_path() -> PathBuf {
crate::session::active_api_socket_path()
}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct RuntimeStatus {
- pub version: Option<String>,
- pub protocol: Option<u32>,
-}
-
-pub fn read_runtime_status_at(
- socket_path: &Path,
- timeout: Duration,
-) -> io::Result<Option<RuntimeStatus>> {
- if !socket_path.exists() {
- return Ok(None);
- }
-
- let mut stream = match UnixStream::connect(socket_path) {
- Ok(stream) => stream,
- Err(err)
- if matches!(
- err.kind(),
- io::ErrorKind::ConnectionRefused
- | io::ErrorKind::NotFound
- | io::ErrorKind::TimedOut
- ) =>
- {
- return Ok(None);
- }
- Err(err) => return Err(err),
- };
-
- stream.set_write_timeout(Some(timeout))?;
- stream.set_read_timeout(Some(timeout))?;
-
- let request = Request {
- id: "runtime:status".into(),
- method: Method::Ping(crate::api::schema::PingParams::default()),
- };
- stream.write_all(serde_json::to_string(&request)?.as_bytes())?;
- stream.write_all(b"\n")?;
- stream.flush()?;
-
- let mut reader = BufReader::new(stream);
- let mut line = String::new();
- let read = reader.read_line(&mut line)?;
- if read == 0 || line.trim().is_empty() {
- return Err(io::Error::new(
- io::ErrorKind::UnexpectedEof,
- "empty server status response",
- ));
- }
-
- let response: serde_json::Value = serde_json::from_str(&line).map_err(io::Error::other)?;
- if response.get("error").is_some() {
- return Err(io::Error::other(format!(
- "server status request failed: {response}"
- )));
- }
-
- let result = &response["result"];
- Ok(Some(RuntimeStatus {
- version: result
- .get("version")
- .and_then(|value| value.as_str())
- .map(str::to_owned),
- protocol: result
- .get("protocol")
- .and_then(|value| value.as_u64())
- .and_then(|value| u32::try_from(value).ok()),
- }))
-}
-
-pub struct ServerHandle {
- _thread: std::thread::JoinHandle<()>,
- path: PathBuf,
- running: Arc<AtomicBool>,
-}
-
-impl Drop for ServerHandle {
- fn drop(&mut self) {
- self.running.store(false, Ordering::Relaxed);
-
- if let Err(err) = fs::remove_file(&self.path) {
- if err.kind() != std::io::ErrorKind::NotFound {
- warn!(path = %self.path.display(), err = %err, "failed to remove api socket on shutdown");
- }
- }
- }
-}
-
-pub fn start_server(
- api_tx: ApiRequestSender,
- event_hub: EventHub,
-) -> std::io::Result<ServerHandle> {
- let path = socket_path();
- prepare_socket_path(&path)?;
-
- let listener = UnixListener::bind(&path)?;
- restrict_socket_permissions(&path)?;
- info!(path = %path.display(), "api server listening");
-
- let running = Arc::new(AtomicBool::new(true));
- let listener_running = Arc::clone(&running);
- let thread = std::thread::spawn(move || {
- for stream in listener.incoming() {
- match stream {
- Ok(stream) => {
- let api_tx = api_tx.clone();
- let event_hub = event_hub.clone();
- let connection_running = Arc::clone(&listener_running);
- std::thread::spawn(move || {
- if let Err(err) =
- handle_connection(stream, &api_tx, &event_hub, &connection_running)
- {
- warn!(err = %err, "api connection failed");
- }
- });
- }
- Err(err) => {
- error!(err = %err, "api listener accept failed");
- break;
- }
- }
- }
- debug!("api server thread exiting");
- });
-
- Ok(ServerHandle {
- _thread: thread,
- path,
- running,
- })
-}
-
-fn prepare_socket_path(path: &Path) -> std::io::Result<()> {
- crate::ipc::prepare_socket_path(path, |path| {
- format!(
- "herdr is already running (socket busy at {})",
- path.display()
- )
- })
-}
-
-fn restrict_socket_permissions(path: &Path) -> std::io::Result<()> {
- crate::ipc::restrict_socket_permissions(path, SOCKET_PERMISSION_MODE)
-}
-
-fn handle_connection(
- mut stream: UnixStream,
- api_tx: &ApiRequestSender,
- event_hub: &EventHub,
- running: &Arc<AtomicBool>,
-) -> std::io::Result<()> {
- stream.set_read_timeout(Some(INITIAL_REQUEST_TIMEOUT))?;
- stream.set_write_timeout(Some(STREAM_WRITE_TIMEOUT))?;
-
- let mut line = String::new();
- {
- let mut reader = BufReader::new(&stream);
- let read = reader.read_line(&mut line)?;
- if read == 0 {
- return Ok(());
- }
- }
-
- stream.set_read_timeout(None)?;
-
- let line = line.trim();
- if line.is_empty() {
- return Ok(());
- }
-
- let request = match serde_json::from_str::<Request>(line) {
- Ok(request) => request,
- Err(err) => {
- write_json_line_allow_disconnect(
- &mut stream,
- &ErrorResponse {
- id: String::new(),
- error: ErrorBody {
- code: "invalid_request".into(),
- message: format!("invalid request: {err}"),
- },
- },
- )?;
- return Ok(());
- }
- };
-
- let request_id = request.id.clone();
- let method = api_method_name(&request.method);
- let changes_ui = request_changes_ui(&request);
- crate::logging::api_request_started(&request_id, method, changes_ui);
-
- match request.method {
- Method::EventsSubscribe(params) => {
- let result = stream_subscriptions(
- stream,
- request_id.clone(),
- params,
- api_tx,
- event_hub,
- running,
- );
- match &result {
- Ok(()) => crate::logging::api_request_completed(
- &request_id,
- method,
- "stream_closed",
- 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)?
- 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_body => {
- let response = handle_request(
- Request {
- id: request_id.clone(),
- method: method_body,
- },
- api_tx,
- );
- 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
- }
- }
-}
-
-fn handle_request(request: Request, api_tx: &ApiRequestSender) -> String {
- match request.method {
- Method::Ping(_) => serde_json::to_string(&SuccessResponse {
- id: request.id,
- result: ResponseResult::Pong {
- version: env!("CARGO_PKG_VERSION").into(),
- protocol: crate::server::protocol::PROTOCOL_VERSION,
- },
- })
- .unwrap_or_else(|_| {
- r#"{"id":"","error":{"code":"internal_error","message":"failed to encode response"}}"#
- .to_string()
- }),
- _ => dispatch_to_app(request, api_tx),
- }
-}
-
-fn api_method_name(method: &Method) -> &'static str {
- match method {
- Method::Ping(_) => "ping",
- Method::ServerStop(_) => "server.stop",
- Method::ServerReloadConfig(_) => "server.reload_config",
- Method::WorkspaceCreate(_) => "workspace.create",
- Method::WorkspaceList(_) => "workspace.list",
- Method::WorkspaceGet(_) => "workspace.get",
- Method::WorkspaceFocus(_) => "workspace.focus",
- Method::WorkspaceRename(_) => "workspace.rename",
- Method::WorkspaceClose(_) => "workspace.close",
- Method::TabCreate(_) => "tab.create",
- Method::TabList(_) => "tab.list",
- Method::TabGet(_) => "tab.get",
- Method::TabFocus(_) => "tab.focus",
- Method::TabRename(_) => "tab.rename",
- Method::TabClose(_) => "tab.close",
- Method::AgentList(_) => "agent.list",
- Method::AgentGet(_) => "agent.get",
- Method::AgentRead(_) => "agent.read",
- Method::AgentSend(_) => "agent.send",
- Method::AgentRename(_) => "agent.rename",
- Method::AgentFocus(_) => "agent.focus",
- Method::AgentStart(_) => "agent.start",
- Method::PaneSplit(_) => "pane.split",
- Method::PaneList(_) => "pane.list",
- Method::PaneGet(_) => "pane.get",
- Method::PaneRename(_) => "pane.rename",
- Method::PaneSendText(_) => "pane.send_text",
- Method::PaneSendKeys(_) => "pane.send_keys",
- Method::PaneSendInput(_) => "pane.send_input",
- Method::PaneRead(_) => "pane.read",
- Method::PaneReportAgent(_) => "pane.report_agent",
- Method::PaneClearAgentAuthority(_) => "pane.clear_agent_authority",
- Method::PaneReleaseAgent(_) => "pane.release_agent",
- Method::PaneClose(_) => "pane.close",
- Method::EventsSubscribe(_) => "events.subscribe",
- Method::EventsWait(_) => "events.wait",
- Method::PaneWaitForOutput(_) => "pane.wait_for_output",
- Method::IntegrationInstall(_) => "integration.install",
- Method::IntegrationUninstall(_) => "integration.uninstall",
- }
-}
-
-fn api_response_outcome(response: &str) -> &'static str {
- let Ok(value) = serde_json::from_str::<serde_json::Value>(response) else {
- return "error";
- };
-
- match value
- .get("error")
- .and_then(|error| error.get("code"))
- .and_then(|code| code.as_str())
- {
- Some("timeout") => "timeout",
- Some(_) => "error",
- None => "ok",
- }
-}
-
-fn output_match_read_source(
- source: &crate::api::schema::ReadSource,
-) -> crate::api::schema::ReadSource {
- match source {
- crate::api::schema::ReadSource::Recent => crate::api::schema::ReadSource::RecentUnwrapped,
- other => *other,
- }
-}
-
-fn wait_for_output(
- request_id: String,
- params: crate::api::schema::PaneWaitForOutputParams,
- stream: &mut UnixStream,
- api_tx: &ApiRequestSender,
- running: &Arc<AtomicBool>,
-) -> std::io::Result<Option<String>> {
- crate::logging::api_wait_started(&request_id, &params.pane_id, params.timeout_ms);
- let deadline = params
- .timeout_ms
- .map(|ms| std::time::Instant::now() + std::time::Duration::from_millis(ms));
-
- let regex = match &params.r#match {
- crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) {
- Ok(regex) => Some(regex),
- Err(err) => {
- return Ok(Some(
- serde_json::to_string(&ErrorResponse {
- id: request_id,
- error: ErrorBody {
- code: "invalid_regex".into(),
- message: err.to_string(),
- },
- })
- .unwrap(),
- ));
- }
- },
- crate::api::schema::OutputMatch::Substring { .. } => None,
- };
-
- loop {
- if should_stop_connection(stream, running)? {
- crate::logging::api_wait_completed(&request_id, &params.pane_id, "client_disconnected");
- return Ok(None);
- }
-
- let read_request = Request {
- id: format!("{request_id}:read"),
- method: Method::PaneRead(crate::api::schema::PaneReadParams {
- pane_id: params.pane_id.clone(),
- source: output_match_read_source(&params.source),
- lines: params.lines,
- format: crate::api::schema::ReadFormat::Text,
- strip_ansi: params.strip_ansi,
- }),
- };
- let response =
- dispatch_to_app_with_timeout(read_request, api_tx, Some(APP_RESPONSE_TIMEOUT));
- let Ok(value) = serde_json::from_str::<serde_json::Value>(&response) else {
- return Ok(Some(response));
- };
- if value.get("error").is_some() {
- let mut value = value;
- value["id"] = serde_json::Value::String(request_id.clone());
- return Ok(Some(serde_json::to_string(&value).unwrap()));
- }
-
- let read_value = value["result"]["read"].clone();
- let Ok(read) = serde_json::from_value::<crate::api::schema::PaneReadResult>(read_value)
- else {
- return Ok(Some(
- serde_json::to_string(&ErrorResponse {
- id: request_id,
- error: ErrorBody {
- code: "internal_error".into(),
- message: "failed to decode pane read result".into(),
- },
- })
- .unwrap(),
- ));
- };
-
- let matched_line = match_output(&read.text, &params.r#match, regex.as_ref());
- if matched_line.is_some() {
- let revision = read.revision;
- crate::logging::api_wait_completed(&request_id, &params.pane_id, "matched");
- return Ok(Some(
- serde_json::to_string(&SuccessResponse {
- id: request_id,
- result: ResponseResult::OutputMatched {
- pane_id: params.pane_id,
- revision,
- matched_line,
- read,
- },
- })
- .unwrap(),
- ));
- }
-
- if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) {
- crate::logging::api_wait_timed_out(&request_id, &params.pane_id);
- return Ok(Some(
- serde_json::to_string(&ErrorResponse {
- id: request_id,
- error: ErrorBody {
- code: "timeout".into(),
- message: "timed out waiting for output match".into(),
- },
- })
- .unwrap(),
- ));
- }
-
- std::thread::sleep(CONNECTION_POLL_INTERVAL);
- }
-}
-
-fn stream_subscriptions(
- mut stream: UnixStream,
- request_id: String,
- params: crate::api::schema::EventsSubscribeParams,
- api_tx: &ApiRequestSender,
- event_hub: &EventHub,
- running: &Arc<AtomicBool>,
-) -> std::io::Result<()> {
- let mut subscriptions = Vec::with_capacity(params.subscriptions.len());
- for (index, subscription) in params.subscriptions.into_iter().enumerate() {
- let active =
- match ActiveSubscription::new(subscription, &request_id, index, api_tx, event_hub) {
- Ok(active) => active,
- Err(response) => {
- if let Err(err) = write_json_line(&mut stream, &response) {
- if is_connection_closed_error(&err) {
- return Ok(());
- }
- return Err(err);
- }
- return Ok(());
- }
- };
- subscriptions.push(active);
- }
-
- if let Err(err) = write_json_line(
- &mut stream,
- &SuccessResponse {
- id: request_id,
- result: ResponseResult::SubscriptionStarted {},
- },
- ) {
- if is_connection_closed_error(&err) {
- return Ok(());
- }
- return Err(err);
- }
-
- loop {
- if should_stop_connection(&mut stream, running)? {
- return Ok(());
- }
-
- for subscription in &mut subscriptions {
- if let Some(event) = subscription.poll(api_tx, event_hub) {
- if let Err(err) = write_json_line(&mut stream, &event) {
- if is_connection_closed_error(&err) {
- return Ok(());
- }
- return Err(err);
- }
- }
- }
- std::thread::sleep(CONNECTION_POLL_INTERVAL);
- }
-}
-
-fn write_text_line(stream: &mut UnixStream, value: &str) -> std::io::Result<()> {
- stream.write_all(value.as_bytes())?;
- stream.write_all(b"\n")?;
- stream.flush()
-}
-
-fn write_text_line_allow_disconnect(stream: &mut UnixStream, value: &str) -> std::io::Result<()> {
- match write_text_line(stream, value) {
- Err(err) if is_connection_closed_error(&err) => Ok(()),
- result => result,
- }
-}
-
-fn write_json_line<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> std::io::Result<()> {
- let encoded = serde_json::to_string(value)
- .map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?;
- write_text_line(stream, &encoded)
-}
-
-fn write_json_line_allow_disconnect<T: serde::Serialize>(
- stream: &mut UnixStream,
- value: &T,
-) -> std::io::Result<()> {
- let encoded = serde_json::to_string(value)
- .map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?;
- write_text_line_allow_disconnect(stream, &encoded)
-}
-
-fn should_stop_connection(
- stream: &mut UnixStream,
- running: &Arc<AtomicBool>,
-) -> std::io::Result<bool> {
- if !running.load(Ordering::Relaxed) {
- return Ok(true);
- }
-
- probe_stream_closed(stream)
-}
-
-fn probe_stream_closed(stream: &mut UnixStream) -> 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
- )
-}
-
-fn match_output(
- text: &str,
- matcher: &crate::api::schema::OutputMatch,
- regex: Option<&Regex>,
-) -> Option<String> {
- match matcher {
- crate::api::schema::OutputMatch::Substring { value } => text
- .lines()
- .find(|line| line.contains(value))
- .map(|line| line.to_string()),
- crate::api::schema::OutputMatch::Regex { .. } => regex.and_then(|re| {
- text.lines()
- .find(|line| re.is_match(line))
- .map(|line| line.to_string())
- }),
- }
-}
-
-struct ActiveOutputMatchedSubscription {
- pane_id: String,
- source: crate::api::schema::ReadSource,
- lines: Option<u32>,
- matcher: crate::api::schema::OutputMatch,
- regex: Option<Regex>,
- strip_ansi: bool,
- currently_matching: bool,
- request_prefix: String,
-}
-
-struct ActiveAgentStatusChangedSubscription {
- pane_id: String,
- status_filter: Option<crate::api::schema::AgentStatus>,
- last_status: Option<crate::api::schema::AgentStatus>,
- request_prefix: String,
-}
-
-struct ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind,
- last_sequence: u64,
-}
-
-enum ActiveSubscription {
- Event(ActiveEventSubscription),
- OutputMatched(ActiveOutputMatchedSubscription),
- AgentStatusChanged(ActiveAgentStatusChangedSubscription),
-}
-
-impl ActiveSubscription {
- fn new(
- subscription: Subscription,
- request_id: &str,
- index: usize,
- api_tx: &ApiRequestSender,
- _event_hub: &EventHub,
- ) -> Result<Self, ErrorResponse> {
- match subscription {
- Subscription::WorkspaceCreated {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::WorkspaceCreated,
- last_sequence: 0,
- })),
- Subscription::WorkspaceClosed {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::WorkspaceClosed,
- last_sequence: 0,
- })),
- Subscription::WorkspaceFocused {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::WorkspaceFocused,
- last_sequence: 0,
- })),
- Subscription::TabCreated {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::TabCreated,
- last_sequence: 0,
- })),
- Subscription::TabClosed {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::TabClosed,
- last_sequence: 0,
- })),
- Subscription::TabFocused {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::TabFocused,
- last_sequence: 0,
- })),
- Subscription::TabRenamed {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::TabRenamed,
- last_sequence: 0,
- })),
- Subscription::PaneCreated {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::PaneCreated,
- last_sequence: 0,
- })),
- Subscription::PaneClosed {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::PaneClosed,
- last_sequence: 0,
- })),
- Subscription::PaneFocused {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::PaneFocused,
- last_sequence: 0,
- })),
- Subscription::PaneExited {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::PaneExited,
- last_sequence: 0,
- })),
- Subscription::PaneAgentDetected {} => Ok(Self::Event(ActiveEventSubscription {
- event_kind: crate::api::schema::EventKind::PaneAgentDetected,
- last_sequence: 0,
- })),
- Subscription::PaneOutputMatched {
- pane_id,
- source,
- lines,
- r#match,
- strip_ansi,
- } => {
- let regex = match &r#match {
- crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) {
- Ok(regex) => Some(regex),
- Err(err) => {
- return Err(ErrorResponse {
- id: request_id.to_string(),
- error: ErrorBody {
- code: "invalid_regex".into(),
- message: err.to_string(),
- },
- });
- }
- },
- crate::api::schema::OutputMatch::Substring { .. } => None,
- };
-
- let probe = pane_read(
- format!("{request_id}:sub:{index}:probe"),
- &pane_id,
- source,
- lines,
- strip_ansi,
- api_tx,
- );
- probe?;
-
- Ok(Self::OutputMatched(ActiveOutputMatchedSubscription {
- pane_id,
- source,
- lines,
- matcher: r#match,
- regex,
- strip_ansi,
- currently_matching: false,
- request_prefix: format!("{request_id}:sub:{index}"),
- }))
- }
- Subscription::PaneAgentStatusChanged {
- pane_id,
- agent_status,
- } => {
- let probe = pane_get(format!("{request_id}:sub:{index}:probe"), &pane_id, api_tx)?;
-
- Ok(Self::AgentStatusChanged(
- ActiveAgentStatusChangedSubscription {
- pane_id,
- status_filter: agent_status,
- last_status: Some(probe.agent_status),
- request_prefix: format!("{request_id}:sub:{index}"),
- },
- ))
- }
- }
- }
-
- fn poll(
- &mut self,
- api_tx: &ApiRequestSender,
- event_hub: &EventHub,
- ) -> Option<serde_json::Value> {
- match self {
- Self::Event(subscription) => subscription.poll(event_hub),
- Self::OutputMatched(subscription) => {
- serde_json::to_value(subscription.poll(api_tx)?).ok()
- }
- Self::AgentStatusChanged(subscription) => {
- serde_json::to_value(subscription.poll(api_tx)?).ok()
- }
- }
- }
-}
-
-impl ActiveEventSubscription {
- fn poll(&mut self, event_hub: &EventHub) -> Option<serde_json::Value> {
- for (sequence, event) in event_hub.events_after(self.last_sequence) {
- self.last_sequence = sequence;
- if event.event == self.event_kind {
- return serde_json::to_value(event).ok();
- }
- }
- None
- }
-}
-
-impl ActiveOutputMatchedSubscription {
- fn poll(&mut self, api_tx: &ApiRequestSender) -> Option<SubscriptionEventEnvelope> {
- let read = pane_read(
- format!("{}:read", self.request_prefix),
- &self.pane_id,
- output_match_read_source(&self.source),
- self.lines,
- self.strip_ansi,
- api_tx,
- )
- .ok()?;
-
- let matched_line = match_output(&read.text, &self.matcher, self.regex.as_ref());
- match matched_line {
- Some(matched_line) => {
- if self.currently_matching {
- return None;
- }
- self.currently_matching = true;
- Some(SubscriptionEventEnvelope {
- event: SubscriptionEventKind::PaneOutputMatched,
- data: SubscriptionEventData::PaneOutputMatched(PaneOutputMatchedEvent {
- pane_id: self.pane_id.clone(),
- matched_line,
- read,
- }),
- })
- }
- None => {
- self.currently_matching = false;
- None
- }
- }
- }
-}
-
-impl ActiveAgentStatusChangedSubscription {
- fn poll(&mut self, api_tx: &ApiRequestSender) -> Option<SubscriptionEventEnvelope> {
- let pane = pane_get(
- format!("{}:pane", self.request_prefix),
- &self.pane_id,
- api_tx,
- )
- .ok()?;
- let current_status = pane.agent_status;
- let previous_status = self.last_status.replace(current_status);
- if previous_status.is_none() || previous_status == Some(current_status) {
- return None;
- }
- if self
- .status_filter
- .is_some_and(|wanted| wanted != current_status)
- {
- return None;
- }
-
- Some(SubscriptionEventEnvelope {
- event: SubscriptionEventKind::PaneAgentStatusChanged,
- data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent {
- pane_id: pane.pane_id,
- workspace_id: pane.workspace_id,
- agent_status: current_status,
- agent: pane.agent,
- custom_status: pane.custom_status,
- }),
- })
- }
-}
-
-fn pane_read(
- request_id: String,
- pane_id: &str,
- source: crate::api::schema::ReadSource,
- lines: Option<u32>,
- strip_ansi: bool,
- api_tx: &ApiRequestSender,
-) -> Result<crate::api::schema::PaneReadResult, ErrorResponse> {
- let response = dispatch_to_app_with_timeout(
- Request {
- id: request_id.clone(),
- method: Method::PaneRead(crate::api::schema::PaneReadParams {
- pane_id: pane_id.to_string(),
- source,
- lines,
- format: crate::api::schema::ReadFormat::Text,
- strip_ansi,
- }),
- },
- api_tx,
- Some(APP_RESPONSE_TIMEOUT),
- );
- let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse {
- id: request_id.clone(),
- error: ErrorBody {
- code: "internal_error".into(),
- message: "failed to decode pane read response".into(),
- },
- })?;
- if value.get("error").is_some() {
- return serde_json::from_value(value).map_err(|_| ErrorResponse {
- id: request_id,
- error: ErrorBody {
- code: "internal_error".into(),
- message: "failed to decode pane read error".into(),
- },
- });
- }
- serde_json::from_value(value["result"]["read"].clone()).map_err(|_| ErrorResponse {
- id: request_id,
- error: ErrorBody {
- code: "internal_error".into(),
- message: "failed to decode pane read result".into(),
- },
- })
-}
-
-fn pane_get(
- request_id: String,
- pane_id: &str,
- api_tx: &ApiRequestSender,
-) -> Result<crate::api::schema::PaneInfo, ErrorResponse> {
- let response = dispatch_to_app_with_timeout(
- Request {
- id: request_id.clone(),
- method: Method::PaneGet(crate::api::schema::PaneTarget {
- pane_id: pane_id.to_string(),
- }),
- },
- api_tx,
- Some(APP_RESPONSE_TIMEOUT),
- );
- let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse {
- id: request_id.clone(),
- error: ErrorBody {
- code: "internal_error".into(),
- message: "failed to decode pane get response".into(),
- },
- })?;
- if value.get("error").is_some() {
- return serde_json::from_value(value).map_err(|_| ErrorResponse {
- id: request_id,
- error: ErrorBody {
- code: "internal_error".into(),
- message: "failed to decode pane get error".into(),
- },
- });
- }
- serde_json::from_value(value["result"]["pane"].clone()).map_err(|_| ErrorResponse {
- id: request_id,
- error: ErrorBody {
- code: "internal_error".into(),
- message: "failed to decode pane get result".into(),
- },
- })
-}
-
-fn dispatch_to_app(request: Request, api_tx: &ApiRequestSender) -> String {
- dispatch_to_app_with_timeout(request, api_tx, None)
-}
-
-fn dispatch_to_app_with_timeout(
- request: Request,
- api_tx: &ApiRequestSender,
- timeout: Option<Duration>,
-) -> String {
- let request_id = request.id.clone();
- let (respond_to, response_rx) = std::sync::mpsc::channel();
- if let Err(err) = api_tx.send(ApiRequestMessage {
- request,
- respond_to,
- }) {
- return error_response_json(
- request_id,
- "server_unavailable",
- format!("failed to dispatch request: {err}"),
- );
- }
-
- let response = match timeout {
- Some(timeout) => response_rx.recv_timeout(timeout).map_err(|err| match err {
- std::sync::mpsc::RecvTimeoutError::Timeout => std::io::Error::new(
- std::io::ErrorKind::TimedOut,
- format!(
- "timed out waiting for app response after {} ms",
- timeout.as_millis()
- ),
- ),
- std::sync::mpsc::RecvTimeoutError::Disconnected => std::io::Error::new(
- std::io::ErrorKind::BrokenPipe,
- "app response channel closed",
- ),
- }),
- None => response_rx
- .recv()
- .map_err(|err| std::io::Error::new(std::io::ErrorKind::BrokenPipe, err)),
- };
-
- match response {
- Ok(response) => response,
- Err(err) => error_response_json(
- request_id,
- "server_unavailable",
- format!("request handling failed: {err}"),
- ),
- }
-}
-
-fn error_response_json(id: String, code: &str, message: String) -> String {
- serde_json::to_string(&ErrorResponse {
- id,
- error: ErrorBody {
- code: code.into(),
- message,
- },
- })
- .unwrap_or_else(|_| {
- r#"{"id":"","error":{"code":"internal_error","message":"failed to encode error response"}}"#
- .to_string()
- })
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use std::os::unix::fs::PermissionsExt;
- use std::sync::{Mutex, OnceLock};
-
- fn env_lock() -> &'static Mutex<()> {
- static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
- LOCK.get_or_init(|| Mutex::new(()))
- }
-
- fn unique_test_path(name: &str) -> PathBuf {
- let nanos = std::time::SystemTime::now()
- .duration_since(std::time::UNIX_EPOCH)
- .unwrap()
- .as_nanos();
- std::env::temp_dir().join(format!("herdr-{name}-{}-{nanos}", std::process::id()))
- }
-
- fn read_line(stream: &mut UnixStream) -> String {
- let mut reader = BufReader::new(stream);
- let mut line = String::new();
- reader.read_line(&mut line).unwrap();
- line
- }
-
- #[test]
- fn socket_path_prefers_explicit_env_override() {
- let _guard = env_lock().lock().unwrap();
- let unique = format!("/tmp/herdr-test-{}.sock", std::process::id());
- std::env::remove_var(crate::session::SESSION_ENV_VAR);
- crate::session::clear_explicit_session_for_test();
- std::env::set_var(SOCKET_PATH_ENV_VAR, &unique);
- assert_eq!(socket_path(), PathBuf::from(&unique));
- std::env::remove_var(SOCKET_PATH_ENV_VAR);
- }
-
- #[test]
- fn socket_path_defaults_to_config_dir_even_when_xdg_runtime_dir_is_set() {
- let _guard = env_lock().lock().unwrap();
- let config_home = unique_test_path("socket-default-config-home");
- let runtime_dir = unique_test_path("socket-default-runtime");
- std::env::remove_var(SOCKET_PATH_ENV_VAR);
- std::env::remove_var(crate::session::SESSION_ENV_VAR);
- crate::session::clear_explicit_session_for_test();
- std::env::set_var("XDG_CONFIG_HOME", &config_home);
- std::env::set_var("XDG_RUNTIME_DIR", &runtime_dir);
-
- let expected = config_home
- .join(crate::config::app_dir_name())
- .join("herdr.sock");
- assert_eq!(socket_path(), expected);
-
- std::env::remove_var("XDG_CONFIG_HOME");
- std::env::remove_var("XDG_RUNTIME_DIR");
- }
-
- #[test]
- fn socket_path_uses_named_session_dir() {
- let _guard = env_lock().lock().unwrap();
- let config_home = unique_test_path("socket-named-config-home");
- std::env::remove_var(SOCKET_PATH_ENV_VAR);
- crate::session::clear_explicit_session_for_test();
- std::env::set_var(crate::session::SESSION_ENV_VAR, "work");
- std::env::set_var("XDG_CONFIG_HOME", &config_home);
-
- let expected = config_home
- .join(crate::config::app_dir_name())
- .join("sessions")
- .join("work")
- .join("herdr.sock");
- assert_eq!(socket_path(), expected);
-
- std::env::remove_var(crate::session::SESSION_ENV_VAR);
- std::env::remove_var("XDG_CONFIG_HOME");
- }
-
- #[test]
- fn restrict_socket_permissions_sets_user_only_mode() {
- let dir = unique_test_path("socket-perms");
- fs::create_dir_all(&dir).unwrap();
- let path = dir.join("api.sock");
- let _listener = UnixListener::bind(&path).unwrap();
-
- restrict_socket_permissions(&path).unwrap();
-
- let mode = fs::metadata(&path).unwrap().permissions().mode() & 0o777;
- assert_eq!(mode, SOCKET_PERMISSION_MODE);
-
- drop(_listener);
- let _ = fs::remove_file(&path);
- let _ = fs::remove_dir_all(&dir);
- }
-
- #[test]
- fn api_response_outcome_uses_top_level_error_shape() {
- let ok_with_error_text = r#"{"id":"req","result":{"read":{"text":"user said \"error\": \"timeout\"","revision":1}}}"#;
- assert_eq!(api_response_outcome(ok_with_error_text), "ok");
-
- let timeout = r#"{"id":"req","error":{"code":"timeout","message":"timed out waiting for output match"}}"#;
- assert_eq!(api_response_outcome(timeout), "timeout");
-
- let generic_error =
- r#"{"id":"req","error":{"code":"server_unavailable","message":"boom"}}"#;
- assert_eq!(api_response_outcome(generic_error), "error");
- }
-
- #[test]
- fn ping_request_returns_pong() {
- let (tx, _rx) = mpsc::unbounded_channel();
- let response = handle_request(
- Request {
- id: "req_1".into(),
- method: Method::Ping(crate::api::schema::PingParams::default()),
- },
- &tx,
- );
-
- let parsed: SuccessResponse = serde_json::from_str(&response).unwrap();
- assert_eq!(parsed.id, "req_1");
- assert!(matches!(parsed.result, ResponseResult::Pong { .. }));
- }
-
- #[test]
- fn request_dispatches_to_app_channel() {
- let (tx, mut rx) = mpsc::unbounded_channel();
- let request = Request {
- id: "req_2".into(),
- method: Method::WorkspaceList(crate::api::schema::EmptyParams::default()),
- };
-
- let request_for_thread = request.clone();
- let thread = std::thread::spawn(move || handle_request(request_for_thread, &tx));
-
- let msg = rx.blocking_recv().unwrap();
- assert_eq!(msg.request.id, "req_2");
- msg.respond_to
- .send(
- serde_json::to_string(&SuccessResponse {
- id: "req_2".into(),
- result: ResponseResult::Ok {},
- })
- .unwrap(),
- )
- .unwrap();
-
- let response = thread.join().unwrap();
- let parsed: SuccessResponse = serde_json::from_str(&response).unwrap();
- assert_eq!(parsed.id, "req_2");
- }
-
- #[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();
- let responder = std::thread::spawn(move || {
- let mut notified = false;
- while let Some(msg) = api_rx.blocking_recv() {
- assert!(matches!(msg.request.method, Method::PaneRead(_)));
- if !notified {
- first_read_tx.send(()).unwrap();
- notified = true;
- }
- msg.respond_to
- .send(
- serde_json::to_string(&SuccessResponse {
- id: msg.request.id,
- result: ResponseResult::PaneRead {
- read: crate::api::schema::PaneReadResult {
- pane_id: "pane_1".into(),
- workspace_id: "ws_1".into(),
- tab_id: "tab_1".into(),
- source: crate::api::schema::ReadSource::RecentUnwrapped,
- format: crate::api::schema::ReadFormat::Text,
- text: String::new(),
- revision: 0,
- truncated: false,
- },
- },
- })
- .unwrap(),
- )
- .unwrap();
- }
- });
-
- let (mut client, server) = UnixStream::pair().unwrap();
- client
- .write_all(br#"{"id":"req_wait","method":"pane.wait_for_output","params":{"pane_id":"pane_1","source":"recent","match":{"type":"substring","value":"never"}}}"#)
- .unwrap();
- client.write_all(b"\n").unwrap();
- client.flush().unwrap();
-
- let running = Arc::new(AtomicBool::new(true));
- let server_running = Arc::clone(&running);
- let event_hub = EventHub::default();
- let (done_tx, done_rx) = std::sync::mpsc::channel();
- let server_thread = std::thread::spawn(move || {
- let result = handle_connection(server, &api_tx, &event_hub, &server_running);
- done_tx.send(result).unwrap();
- });
-
- first_read_rx.recv_timeout(Duration::from_secs(2)).unwrap();
- drop(client);
-
- let result = done_rx.recv_timeout(Duration::from_secs(2)).unwrap();
- assert!(result.is_ok());
-
- server_thread.join().unwrap();
- drop(running);
- responder.join().unwrap();
- }
-
- #[test]
- fn subscriptions_stop_when_client_disconnects() {
- let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
- let (mut client, server) = UnixStream::pair().unwrap();
- client
- .write_all(
- br#"{"id":"sub_1","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#,
- )
- .unwrap();
- client.write_all(b"\n").unwrap();
- client.flush().unwrap();
-
- let running = Arc::new(AtomicBool::new(true));
- let server_running = Arc::clone(&running);
- let event_hub = EventHub::default();
- let (done_tx, done_rx) = std::sync::mpsc::channel();
- let server_thread = std::thread::spawn(move || {
- let result = handle_connection(server, &api_tx, &event_hub, &server_running);
- done_tx.send(result).unwrap();
- });
-
- let ack = read_line(&mut client);
- let ack: serde_json::Value = serde_json::from_str(&ack).unwrap();
- assert_eq!(ack["result"]["type"], "subscription_started");
-
- drop(client);
-
- let result = done_rx.recv_timeout(Duration::from_secs(2)).unwrap();
- assert!(result.is_ok());
- server_thread.join().unwrap();
- }
-
- #[test]
- fn subscriptions_stop_when_server_shuts_down() {
- let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
- let (mut client, server) = UnixStream::pair().unwrap();
- client
- .write_all(
- br#"{"id":"sub_2","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#,
- )
- .unwrap();
- client.write_all(b"\n").unwrap();
- client.flush().unwrap();
-
- let running = Arc::new(AtomicBool::new(true));
- let server_running = Arc::clone(&running);
- let event_hub = EventHub::default();
- let (done_tx, done_rx) = std::sync::mpsc::channel();
- let server_thread = std::thread::spawn(move || {
- let result = handle_connection(server, &api_tx, &event_hub, &server_running);
- done_tx.send(result).unwrap();
- });
-
- let ack = read_line(&mut client);
- let ack: serde_json::Value = serde_json::from_str(&ack).unwrap();
- assert_eq!(ack["result"]["type"], "subscription_started");
-
- running.store(false, Ordering::Relaxed);
-
- let result = done_rx.recv_timeout(Duration::from_secs(2)).unwrap();
- assert!(result.is_ok());
- server_thread.join().unwrap();
- }
-}
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-07 01:58:51 +0000