1 files changed, 245 insertions, 0 deletions
diff --git a/src/api/client.rs b/src/api/client.rs
new file mode 100644
index 00000000..7929f937
--- /dev/null
+++ b/src/api/client.rs
@@ -0,0 +1,245 @@
+use std::fmt;
+use std::io::{self, BufRead, BufReader, Write};
+use std::os::unix::net::UnixStream;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use serde::de::DeserializeOwned;
+
+use crate::api::schema::{
+    ErrorResponse, EventsSubscribeParams, Method, PingParams, Request, ResponseResult,
+    SubscriptionEventEnvelope, SuccessResponse,
+};
+
+/// API connection target resolved by clients at the process edge.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ConnectionTarget {
+    LocalSession(Option<String>),
+    SocketPath(PathBuf),
+}
+
+impl ConnectionTarget {
+    fn socket_path(&self) -> PathBuf {
+        match self {
+            Self::LocalSession(None) => crate::api::socket_path(),
+            Self::LocalSession(Some(name)) => crate::session::api_socket_path_for(Some(name)),
+            Self::SocketPath(path) => path.clone(),
+        }
+    }
+}
+
+/// Reusable client for Herdr's newline-delimited JSON API.
+#[derive(Debug, Clone)]
+pub struct ApiClient {
+    target: ConnectionTarget,
+}
+
+impl ApiClient {
+    pub fn local() -> Self {
+        Self::for_target(ConnectionTarget::LocalSession(None))
+    }
+
+    pub fn for_target(target: ConnectionTarget) -> Self {
+        Self { target }
+    }
+
+    pub fn socket_path(&self) -> PathBuf {
+        self.target.socket_path()
+    }
+
+    pub fn request(&self, request: Request) -> Result<SuccessResponse, ApiClientError> {
+        let value = self.request_value(&request)?;
+        parse_response_value(value)
+    }
+
+    pub fn request_value(&self, request: &Request) -> Result<serde_json::Value, ApiClientError> {
+        let mut stream = self.connect()?;
+        write_request(&mut stream, request)?;
+
+        let mut reader = BufReader::new(stream);
+        read_json_line(&mut reader)
+    }
+
+    pub fn request_value_with_timeout(
+        &self,
+        request: &Request,
+        timeout: Duration,
+    ) -> Result<serde_json::Value, ApiClientError> {
+        let mut stream = self.connect()?;
+        stream.set_write_timeout(Some(timeout))?;
+        stream.set_read_timeout(Some(timeout))?;
+        write_request(&mut stream, request)?;
+
+        let mut reader = BufReader::new(stream);
+        read_json_line(&mut reader)
+    }
+
+    #[allow(dead_code)] // Kept as the typed subscription API; CLI wait paths use subscribe_value to preserve raw ack errors.
+    pub fn subscribe(
+        &self,
+        id: impl Into<String>,
+        params: EventsSubscribeParams,
+        read_timeout: Option<Duration>,
+    ) -> Result<(SuccessResponse, EventStream), ApiClientError> {
+        let request = Request {
+            id: id.into(),
+            method: Method::EventsSubscribe(params),
+        };
+        let (ack, stream) = self.subscribe_value(&request, read_timeout)?;
+        Ok((parse_response_value(ack)?, stream))
+    }
+
+    pub fn subscribe_value(
+        &self,
+        request: &Request,
+        read_timeout: Option<Duration>,
+    ) -> Result<(serde_json::Value, EventStream), ApiClientError> {
+        let mut stream = self.connect()?;
+        write_request(&mut stream, request)?;
+        if let Some(timeout) = read_timeout {
+            stream.set_read_timeout(Some(timeout))?;
+        }
+
+        let mut reader = BufReader::new(stream);
+        let ack = read_json_line(&mut reader)?;
+        Ok((ack, EventStream { reader }))
+    }
+
+    pub fn status(&self) -> Result<crate::api::RuntimeStatus, ApiClientError> {
+        let response = self.request(Request {
+            id: "api-client:status".into(),
+            method: Method::Ping(PingParams::default()),
+        })?;
+        match response.result {
+            ResponseResult::Pong { version, protocol } => Ok(crate::api::RuntimeStatus {
+                version: Some(version),
+                protocol: Some(protocol),
+            }),
+            result => Err(ApiClientError::UnexpectedResult(format!("{result:?}"))),
+        }
+    }
+
+    fn connect(&self) -> io::Result<UnixStream> {
+        UnixStream::connect(self.socket_path())
+    }
+}
+
+pub struct EventStream {
+    reader: BufReader<UnixStream>,
+}
+
+impl EventStream {
+    pub fn next_value(&mut self) -> Result<Option<serde_json::Value>, ApiClientError> {
+        read_optional_json_line(&mut self.reader)
+    }
+
+    pub fn next_event(&mut self) -> Result<Option<SubscriptionEventEnvelope>, ApiClientError> {
+        self.next_value()?
+            .map(serde_json::from_value)
+            .transpose()
+            .map_err(ApiClientError::Json)
+    }
+}
+
+#[derive(Debug)]
+pub enum ApiClientError {
+    Io(io::Error),
+    Json(serde_json::Error),
+    ErrorResponse(ErrorResponse),
+    EmptyResponse,
+    UnexpectedResult(String),
+}
+
+impl fmt::Display for ApiClientError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Io(err) => write!(f, "{err}"),
+            Self::Json(err) => write!(f, "{err}"),
+            Self::ErrorResponse(response) => write!(f, "{}", response.error.message),
+            Self::EmptyResponse => write!(f, "empty api response"),
+            Self::UnexpectedResult(result) => write!(f, "unexpected api result: {result}"),
+        }
+    }
+}
+
+impl std::error::Error for ApiClientError {}
+
+impl From<io::Error> for ApiClientError {
+    fn from(err: io::Error) -> Self {
+        Self::Io(err)
+    }
+}
+
+impl From<serde_json::Error> for ApiClientError {
+    fn from(err: serde_json::Error) -> Self {
+        Self::Json(err)
+    }
+}
+
+fn write_request(stream: &mut UnixStream, request: &Request) -> Result<(), ApiClientError> {
+    stream.write_all(serde_json::to_string(request)?.as_bytes())?;
+    stream.write_all(b"\n")?;
+    stream.flush()?;
+    Ok(())
+}
+
+fn read_json_line<T: DeserializeOwned>(
+    reader: &mut BufReader<UnixStream>,
+) -> Result<T, ApiClientError> {
+    let mut line = String::new();
+    let read = reader.read_line(&mut line)?;
+    if read == 0 || line.trim().is_empty() {
+        return Err(ApiClientError::EmptyResponse);
+    }
+    serde_json::from_str(&line).map_err(ApiClientError::Json)
+}
+
+fn read_optional_json_line<T: DeserializeOwned>(
+    reader: &mut BufReader<UnixStream>,
+) -> Result<Option<T>, ApiClientError> {
+    let mut line = String::new();
+    let read = reader.read_line(&mut line)?;
+    if read == 0 {
+        return Ok(None);
+    }
+    if line.trim().is_empty() {
+        return Err(ApiClientError::EmptyResponse);
+    }
+    serde_json::from_str(&line)
+        .map(Some)
+        .map_err(ApiClientError::Json)
+}
+
+#[derive(serde::Deserialize)]
+#[serde(untagged)]
+enum WireResponse {
+    Success(Box<SuccessResponse>),
+    Error(ErrorResponse),
+}
+
+pub(crate) fn parse_response_value(
+    value: serde_json::Value,
+) -> Result<SuccessResponse, ApiClientError> {
+    match serde_json::from_value(value)? {
+        WireResponse::Success(response) => Ok(*response),
+        WireResponse::Error(response) => Err(ApiClientError::ErrorResponse(response)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn local_session_target_resolves_named_session_socket() {
+        let client = ApiClient::for_target(ConnectionTarget::LocalSession(Some("work".into())));
+        assert!(client.socket_path().ends_with("sessions/work/herdr.sock"));
+    }
+
+    #[test]
+    fn socket_path_target_uses_explicit_path() {
+        let path = PathBuf::from("/tmp/herdr-test.sock");
+        let client = ApiClient::for_target(ConnectionTarget::SocketPath(path.clone()));
+        assert_eq!(client.socket_path(), path);
+    }
+}