pitchfork-cli 2.23.0

use crate::daemon::{Daemon, RunOptions};
use crate::daemon_id::DaemonId;
use crate::error::IpcError;
use crate::ipc::batch::RunResult;
use crate::ipc::{IpcRequest, IpcResponse, deserialize, fs_name, serialize};
use crate::settings::settings;
use crate::{Result, supervisor};
use exponential_backoff::Backoff;
use interprocess::local_socket::tokio::{RecvHalf, SendHalf};
use interprocess::local_socket::traits::tokio::Stream;
use miette::Context;
use std::path::PathBuf;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::Mutex;
use uuid::Uuid;

/// Returns whether the version-mismatch warning was already emitted for this
/// supervisor version, marking it as emitted. Parallel batch tasks each open
/// their own connection, so without this a mismatched supervisor would warn
/// once per daemon instead of once. Keyed by the supervisor version so a
/// long-lived process (TUI, web, MCP) warns again if a *different* mismatched
/// supervisor appears later.
fn version_mismatch_already_warned(supervisor_version: &str) -> bool {
    static WARNED: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
    let mut warned = WARNED.lock().unwrap_or_else(|e| e.into_inner());
    if warned.as_deref() == Some(supervisor_version) {
        true
    } else {
        *warned = Some(supervisor_version.to_string());
        false
    }
}

pub struct IpcClient {
    _id: String,
    recv: Mutex<BufReader<RecvHalf>>,
    send: Mutex<SendHalf>,
    /// Held across a full request/response exchange. The wire protocol has no
    /// request IDs, so responses are attributed purely by read order — if two
    /// tasks interleave request/read on the same connection, each can consume
    /// the other's response. Callers that need parallel requests must use
    /// dedicated connections (one in-flight request per connection).
    exchange: Mutex<()>,
    /// Set when an exchange fails mid-stream (timeout or I/O error). The
    /// abandoned response may still arrive on the connection later, so reading
    /// again would attribute it to the next request (permanent off-by-one
    /// desync). The next exchange reconnects instead of reading stale data.
    desynced: std::sync::atomic::AtomicBool,
}

impl IpcClient {
    pub async fn connect(autostart: bool) -> Result<Self> {
        if autostart && settings().supervisor.auto_start {
            supervisor::start_if_not_running()?;
        }
        let id = Uuid::new_v4().to_string();
        let client = Self::connect_(&id, "main").await?;
        trace!("Connected to IPC socket");
        let client_version = env!("CARGO_PKG_VERSION").to_string();

        // Try ConnectV2 first (supervisor that knows about it will return ConnectOk with its version).
        // If the supervisor is older and doesn't recognize ConnectV2, it will return Error,
        // and we fall back to the legacy Connect handshake.
        let rsp = client
            .request(IpcRequest::ConnectV2 {
                version: client_version.clone(),
            })
            .await?;
        match rsp {
            IpcResponse::ConnectOk {
                version: supervisor_version,
            } => {
                if supervisor_version != client_version
                    && !version_mismatch_already_warned(&supervisor_version)
                {
                    warn!(
                        "CLI version {client_version} differs from supervisor version {supervisor_version}. \
                         Restart the supervisor with: pitchfork supervisor start --force"
                    );
                }
            }
            IpcResponse::Error(_) => {
                // Old supervisor doesn't recognize ConnectV2 — fall back to legacy Connect
                debug!("Supervisor did not recognize ConnectV2, falling back to legacy Connect");
                let rsp = client.request(IpcRequest::Connect).await?;
                if !rsp.is_ok() {
                    return Err(IpcError::UnexpectedResponse {
                        expected: "Ok".to_string(),
                        actual: format!("{rsp:?}"),
                    }
                    .into());
                }
                if !version_mismatch_already_warned("legacy") {
                    warn!(
                        "Supervisor is running an older version. \
                         Restart the supervisor with: pitchfork supervisor start --force"
                    );
                }
            }
            _ => {
                return Err(IpcError::UnexpectedResponse {
                    expected: "ConnectOk or Error".to_string(),
                    actual: format!("{rsp:?}"),
                }
                .into());
            }
        }
        debug!("Connected to IPC main");
        Ok(client)
    }

    async fn connect_(id: &str, name: &str) -> Result<Self> {
        let (recv, send) = Self::open_stream(name).await?;
        Ok(Self {
            _id: id.to_string(),
            recv: Mutex::new(recv),
            send: Mutex::new(send),
            exchange: Mutex::new(()),
            desynced: std::sync::atomic::AtomicBool::new(false),
        })
    }

    /// Open a raw socket stream to the supervisor with connect retry/backoff.
    /// Used both for the initial connection and to re-establish a connection
    /// whose framing was lost after a failed exchange (see `desynced`).
    async fn open_stream(name: &str) -> Result<(BufReader<RecvHalf>, SendHalf)> {
        let s = settings();
        let connection_help = || {
            if s.supervisor.auto_start {
                "ensure the supervisor is running with: pitchfork supervisor start".to_string()
            } else {
                "supervisor auto-start is disabled; start it with your service manager or run: \
                 pitchfork supervisor start"
                    .to_string()
            }
        };
        let connect_attempts = u32::try_from(s.ipc.connect_attempts).unwrap_or_else(|_| {
            warn!(
                "ipc.connect_attempts value {} is out of range (0-{}), clamping to 5",
                s.ipc.connect_attempts,
                u32::MAX
            );
            5
        });
        let connect_attempts = if connect_attempts == 0 {
            warn!("ipc.connect_attempts is 0; defaulting to 1");
            1
        } else {
            connect_attempts
        };
        let connect_min_delay = s.ipc_connect_min_delay();
        let connect_max_delay = s.ipc_connect_max_delay();

        // Compute timeout from backoff parameters: sum the worst-case delays
        // for each attempt (exponential backoff capped at connect_max_delay),
        // plus a 1s buffer for connection overhead.
        let connect_timeout = {
            let mut total = Duration::from_secs(1); // buffer
            let mut delay = connect_min_delay;
            for _ in 0..connect_attempts {
                total += delay;
                delay = (delay * 2).min(connect_max_delay);
            }
            total
        };

        tokio::time::timeout(connect_timeout, async {
            for duration in Backoff::new(connect_attempts, connect_min_delay, connect_max_delay) {
                match interprocess::local_socket::tokio::Stream::connect(fs_name(name)?).await {
                    Ok(conn) => {
                        let (recv, send) = conn.split();
                        return Ok((BufReader::new(recv), send));
                    }
                    Err(err) => {
                        if let Some(duration) = duration {
                            debug!(
                                "Failed to connect to IPC socket: {err:?}, retrying in {duration:?}"
                            );
                            tokio::time::sleep(duration).await;
                            continue;
                        } else {
                            return Err(IpcError::ConnectionFailed {
                                attempts: connect_attempts,
                                source: Some(err),
                                help: connection_help(),
                            }
                            .into());
                        }
                    }
                }
            }
            Err(IpcError::ConnectionFailed {
                attempts: connect_attempts,
                source: None,
                help: connection_help(),
            }
            .into())
        })
        .await
        .unwrap_or_else(|_| {
            Err(IpcError::ConnectionFailed {
                attempts: connect_attempts,
                source: None,
                help: format!(
                    "connection timed out after {connect_timeout:?}; {}",
                    connection_help()
                ),
            }
            .into())
        })
    }

    pub async fn send(&self, msg: IpcRequest) -> Result<()> {
        let mut msg = serialize(&msg)?;
        if msg.contains(&0) {
            return Err(IpcError::InvalidMessage {
                reason: "message contains null byte".to_string(),
            }
            .into());
        }
        msg.push(0);
        let mut send = self.send.lock().await;
        send.write_all(&msg)
            .await
            .map_err(|e| IpcError::SendFailed { source: e })?;
        Ok(())
    }

    async fn read(&self, timeout: Duration) -> Result<IpcResponse> {
        let mut recv = self.recv.lock().await;
        let mut bytes = Vec::new();
        match tokio::time::timeout(timeout, recv.read_until(0, &mut bytes)).await {
            Ok(Ok(_)) => {}
            Ok(Err(err)) => {
                return Err(IpcError::ReadFailed { source: err }.into());
            }
            Err(_) => {
                return Err(IpcError::Timeout {
                    seconds: timeout.as_secs(),
                }
                .into());
            }
        }
        if bytes.is_empty() {
            return Err(IpcError::ConnectionClosed.into());
        }
        deserialize(&bytes).wrap_err("failed to deserialize IPC response")
    }

    pub(crate) async fn request(&self, msg: IpcRequest) -> Result<IpcResponse> {
        self.request_with_timeout(msg, settings().ipc_request_timeout())
            .await
    }

    pub(crate) fn unexpected_response(expected: &str, actual: &IpcResponse) -> IpcError {
        IpcError::UnexpectedResponse {
            expected: expected.to_string(),
            actual: format!("{actual:?}"),
        }
    }

    pub(crate) async fn request_with_timeout(
        &self,
        msg: IpcRequest,
        timeout: Duration,
    ) -> Result<IpcResponse> {
        use std::sync::atomic::Ordering;
        let _exchange = self.exchange.lock().await;
        // A previous exchange failed mid-stream, so its response may still be
        // queued (or half-read) on the socket. Reconnect to restore framing —
        // reading would attribute the stale response to this request.
        if self.desynced.load(Ordering::Acquire) {
            let (recv, send) = Self::open_stream("main").await?;
            *self.recv.lock().await = recv;
            *self.send.lock().await = send;
            self.desynced.store(false, Ordering::Release);
        }
        let result = async {
            self.send(msg).await?;
            self.read(timeout).await
        }
        .await;
        if result.is_err() {
            self.desynced.store(true, Ordering::Release);
        }
        result
    }

    // =========================================================================
    // Low-level IPC operations
    // =========================================================================

    pub async fn enable(&self, id: DaemonId) -> Result<bool> {
        let id_str = id.qualified();
        let rsp = self.request(IpcRequest::Enable { id: id.clone() }).await?;
        match rsp {
            IpcResponse::Yes => {
                info!("Enabled daemon {id_str}");
                Ok(true)
            }
            IpcResponse::No => {
                info!("Daemon {id_str} already enabled");
                Ok(false)
            }
            IpcResponse::Error(error) => Err(miette::miette!(error)),
            rsp => Err(Self::unexpected_response("Yes or No", &rsp).into()),
        }
    }

    pub async fn disable(&self, id: DaemonId) -> Result<bool> {
        let id_str = id.qualified();
        let rsp = self.request(IpcRequest::Disable { id: id.clone() }).await?;
        match rsp {
            IpcResponse::Yes => {
                info!("Disabled daemon {id_str}");
                Ok(true)
            }
            IpcResponse::No => {
                info!("Daemon {id_str} already disabled");
                Ok(false)
            }
            IpcResponse::Error(error) => Err(miette::miette!(error)),
            rsp => Err(Self::unexpected_response("Yes or No", &rsp).into()),
        }
    }

    /// Run a single daemon with the given options (low-level operation)
    pub async fn run(&self, opts: RunOptions) -> Result<RunResult> {
        let start_time = chrono::Local::now();
        // If any configured readiness check is unbounded (no timeout), the
        // supervisor may wait indefinitely. Use a generous cap so the client
        // does not disconnect prematurely — e.g. when a bounded ready_cmd
        // is paired with an unbounded ready_port. When all checks are bounded,
        // wait through the longest deadline plus a response buffer.
        let has_unbounded_check = opts
            .ready_output
            .as_ref()
            .is_some_and(|o| o.timeout.is_none())
            || opts
                .ready_port
                .as_ref()
                .is_some_and(|p| p.timeout.is_none())
            || opts
                .ready_http
                .as_ref()
                .is_some_and(|h| h.timeout.is_none())
            || opts.ready_cmd.as_ref().is_some_and(|c| c.timeout.is_none());
        let timeout = if has_unbounded_check {
            Duration::from_secs(3600)
        } else {
            let max_deadline = opts
                .ready_output
                .as_ref()
                .and_then(|o| o.timeout)
                .map(|d| d.as_secs())
                .unwrap_or(0)
                .max(
                    opts.ready_http
                        .as_ref()
                        .and_then(|h| h.timeout)
                        .map(|d| d.as_secs())
                        .unwrap_or(0),
                )
                .max(
                    opts.ready_cmd
                        .as_ref()
                        .and_then(|c| c.timeout)
                        .map(|d| d.as_secs())
                        .unwrap_or(0),
                )
                .max(
                    opts.ready_port
                        .as_ref()
                        .and_then(|p| p.timeout)
                        .map(|d| d.as_secs())
                        .unwrap_or(0),
                )
                .max(opts.ready_delay.unwrap_or(3));
            Duration::from_secs(max_deadline + 60)
        };
        let rsp = self
            .request_with_timeout(IpcRequest::Run(opts.clone()), timeout)
            .await?;

        match rsp {
            IpcResponse::DaemonStart { daemon } => {
                debug!("Started {}", daemon.id);
                Ok(RunResult {
                    started: true,
                    exit_code: None,
                    start_time,
                    resolved_ports: daemon.resolved_port.clone(),
                    error_message: None,
                })
            }
            IpcResponse::DaemonReady { daemon } => {
                debug!("Started {}", daemon.id);
                Ok(RunResult {
                    started: true,
                    exit_code: None,
                    start_time,
                    resolved_ports: daemon.resolved_port.clone(),
                    error_message: None,
                })
            }
            IpcResponse::DaemonFailedWithCode { exit_code } => {
                let code = exit_code.unwrap_or(1);
                Ok(RunResult {
                    started: false,
                    exit_code: Some(code),
                    start_time,
                    resolved_ports: Vec::new(),
                    error_message: Some(format!(
                        "Daemon {} failed with exit code {}",
                        opts.id, code
                    )),
                })
            }
            IpcResponse::DaemonAlreadyRunning => {
                warn!("Daemon {} already running", opts.id);
                Ok(RunResult {
                    started: false,
                    exit_code: None,
                    start_time,
                    resolved_ports: Vec::new(),
                    error_message: None,
                })
            }
            IpcResponse::DaemonFailed { error } => Ok(RunResult {
                started: false,
                exit_code: Some(1),
                start_time,
                resolved_ports: Vec::new(),
                error_message: Some(format!("Failed to start daemon {}: {}", opts.id, error)),
            }),
            IpcResponse::PortConflict { port, process, pid } => Ok(RunResult {
                started: false,
                exit_code: Some(1),
                start_time,
                resolved_ports: Vec::new(),
                error_message: Some(format!(
                    "Failed to start daemon {}: port {} is already in use by process '{}' (PID: {})",
                    opts.id, port, process, pid
                )),
            }),
            IpcResponse::NoAvailablePort {
                start_port,
                attempts,
            } => Ok(RunResult {
                started: false,
                exit_code: Some(1),
                start_time,
                resolved_ports: Vec::new(),
                error_message: Some(format!(
                    "Failed to start daemon {}: could not find an available port after {} attempts starting from {}",
                    opts.id, attempts, start_port
                )),
            }),
            rsp => Err(Self::unexpected_response("DaemonStart or DaemonReady", &rsp).into()),
        }
    }

    pub async fn active_daemons(&self) -> Result<Vec<Daemon>> {
        let rsp = self.request(IpcRequest::GetActiveDaemons).await?;
        match rsp {
            IpcResponse::ActiveDaemons(daemons) => Ok(daemons),
            rsp => Err(Self::unexpected_response("ActiveDaemons", &rsp).into()),
        }
    }

    pub async fn update_shell_dir(&self, shell_pid: u32, dir: PathBuf) -> Result<()> {
        let rsp = self
            .request(IpcRequest::UpdateShellDir {
                shell_pid,
                dir: dir.clone(),
            })
            .await?;
        match rsp {
            IpcResponse::Ok => {
                trace!("updated shell dir for pid {shell_pid} to {}", dir.display());
            }
            rsp => return Err(Self::unexpected_response("Ok", &rsp).into()),
        }
        Ok(())
    }

    /// Report a line of a daemon's output the supervisor must act on.
    ///
    /// Called by a log sink, which reads the daemon's output in the
    /// supervisor's place and so is the only process in a position to see it.
    /// `fires_hook` is the sink's answer to whether the line passed the
    /// `on_output` hook's filter and debounce; the supervisor cannot re-derive
    /// it without redoing the debounce on a clock of its own.
    pub async fn sink_output_line(
        &self,
        id: DaemonId,
        token: u64,
        fires_hook: bool,
        line: String,
    ) -> Result<()> {
        let rsp = self
            .request(IpcRequest::SinkOutputLine {
                id,
                token,
                fires_hook,
                line,
            })
            .await?;
        match rsp {
            IpcResponse::Ok => Ok(()),
            rsp => Err(Self::unexpected_response("Ok", &rsp).into()),
        }
    }

    pub async fn clean(&self) -> Result<()> {
        let rsp = self.request(IpcRequest::Clean).await?;
        match rsp {
            IpcResponse::Ok => {
                info!("Cleaned up stopped/failed daemons");
            }
            rsp => return Err(Self::unexpected_response("Ok", &rsp).into()),
        }
        Ok(())
    }

    pub async fn clean_filtered(
        &self,
        namespaces: Vec<String>,
        daemons: Vec<DaemonId>,
        prune: bool,
    ) -> Result<u64> {
        let rsp = self
            .request(IpcRequest::CleanFiltered {
                namespaces,
                daemons,
                prune,
            })
            .await?;
        match rsp {
            IpcResponse::Cleaned { count } => Ok(count),
            rsp => Err(Self::unexpected_response("Cleaned", &rsp).into()),
        }
    }

    pub async fn get_disabled_daemons(&self) -> Result<Vec<DaemonId>> {
        let rsp = self.request(IpcRequest::GetDisabledDaemons).await?;
        match rsp {
            IpcResponse::DisabledDaemons(daemons) => Ok(daemons),
            rsp => Err(Self::unexpected_response("DisabledDaemons", &rsp).into()),
        }
    }

    pub async fn get_notifications(&self) -> Result<Vec<(log::LevelFilter, String)>> {
        let rsp = self.request(IpcRequest::GetNotifications).await?;
        match rsp {
            IpcResponse::Notifications(notifications) => Ok(notifications),
            rsp => Err(Self::unexpected_response("Notifications", &rsp).into()),
        }
    }

    /// Notify the supervisor that the slug registry has changed.
    ///
    /// The supervisor will re-read slugs and update mDNS records.  This is a
    /// best-effort notification — if the supervisor is not running, the call
    /// silently succeeds (mDNS is not needed without a running supervisor).
    pub async fn sync_mdns(&self) -> Result<()> {
        let rsp = self.request(IpcRequest::SyncMdns).await?;
        match rsp {
            IpcResponse::MdnsSynced => Ok(()),
            IpcResponse::Error(e) => {
                // Old supervisor doesn't recognize SyncMdns — not an error.
                info!("mDNS sync skipped: {e}");
                Ok(())
            }
            rsp => Err(Self::unexpected_response("MdnsSynced", &rsp).into()),
        }
    }

    /// Notify the supervisor that settings have changed.
    ///
    /// The supervisor will reload settings from config files. This is a
    /// best-effort notification — if the supervisor is not running, the call
    /// silently succeeds (settings will be fresh on next supervisor start).
    pub async fn reload_config(&self) -> Result<()> {
        match self.request(IpcRequest::ReloadConfig).await {
            Ok(IpcResponse::ConfigReloaded) => Ok(()),
            Ok(IpcResponse::Error(e)) => {
                debug!("config reload skipped: {e}");
                Ok(())
            }
            Ok(rsp) => Err(Self::unexpected_response("ConfigReloaded", &rsp).into()),
            Err(err) => {
                debug!("config reload skipped: {err:?}");
                Ok(())
            }
        }
    }

    /// Enter or replace a project session for a host PID in a directory.
    pub async fn project_enter(&self, pid: u32, dir: PathBuf) -> Result<()> {
        let rsp = self
            .request(IpcRequest::ProjectEnter {
                pid,
                dir: dir.clone(),
            })
            .await?;
        match rsp {
            IpcResponse::Ok => {
                trace!("entered project session pid {pid} in {}", dir.display());
                Ok(())
            }
            rsp => Err(Self::unexpected_response("Ok", &rsp).into()),
        }
    }

    /// Leave a project session for a host PID in a directory.
    pub async fn project_leave(&self, pid: u32, dir: PathBuf) -> Result<()> {
        let rsp = self
            .request(IpcRequest::ProjectLeave {
                pid,
                dir: dir.clone(),
            })
            .await?;
        match rsp {
            IpcResponse::Ok => {
                trace!("left project session pid {pid} in {}", dir.display());
                Ok(())
            }
            rsp => Err(Self::unexpected_response("Ok", &rsp).into()),
        }
    }

    /// List all tracked project sessions with live liveness status.
    pub async fn get_project_sessions(&self) -> Result<Vec<crate::ipc::ProjectSessionInfo>> {
        let rsp = self.request(IpcRequest::GetProjectSessions).await?;
        match rsp {
            IpcResponse::ProjectSessions(sessions) => Ok(sessions),
            rsp => Err(Self::unexpected_response("ProjectSessions", &rsp).into()),
        }
    }

    /// Get the URL of the web UI from the supervisor, if it is running.
    ///
    /// Returns the actual bound address (which may differ from static config
    /// due to port bumping or settings changed since the supervisor started).
    /// Returns `None` if the web UI is not running or the supervisor is too
    /// old to recognize the request.
    pub async fn get_web_url(&self) -> Result<Option<String>> {
        match self.request(IpcRequest::GetWebUrl).await? {
            IpcResponse::WebUrl { url } => Ok(url),
            IpcResponse::Error(e) => {
                // Old supervisor doesn't recognize GetWebUrl — not an error.
                debug!("web url lookup skipped: {e}");
                Ok(None)
            }
            rsp => Err(Self::unexpected_response("WebUrl", &rsp).into()),
        }
    }

    /// Stop a single daemon (low-level operation)
    pub async fn stop(&self, id: DaemonId) -> Result<bool> {
        let id_str = id.qualified();
        // The supervisor's Stop handler blocks until the daemon's ENTIRE
        // process group has exited (stop signal -> stop budget -> SIGKILL ->
        // bounded verification), which can far exceed the flat IPC request
        // timeout — a correct, in-progress stop would otherwise be reported
        // as a failure. Budget the request from the daemon's own stop
        // configuration: its graceful window, plus the supervisor's ~2s
        // post-SIGKILL verification, plus the normal request timeout as slack.
        let stop_budget = crate::state_file::StateFile::get()
            .daemons
            .get(&id)
            .and_then(|d| d.stop_signal.as_ref())
            .and_then(|s| s.timeout)
            .unwrap_or_else(|| settings().supervisor_stop_timeout());
        let timeout = stop_budget + Duration::from_secs(2) + settings().ipc_request_timeout();
        let rsp = self
            .request_with_timeout(IpcRequest::Stop { id: id.clone() }, timeout)
            .await?;
        match rsp {
            IpcResponse::Ok => {
                info!("Stopped daemon {id_str}");
                Ok(true)
            }
            IpcResponse::DaemonNotRunning => {
                warn!("Daemon {id_str} is not running");
                Ok(false)
            }
            IpcResponse::DaemonNotFound => {
                warn!("Daemon {id_str} not found");
                Ok(false)
            }
            IpcResponse::DaemonWasNotRunning => {
                warn!("Daemon {id_str} was not running (process may have exited unexpectedly)");
                Ok(false)
            }
            IpcResponse::DaemonStopFailed { error } => {
                error!("Failed to stop daemon {id_str}: {error}");
                Err(crate::error::DaemonError::StopFailed {
                    id: id_str.clone(),
                    error,
                }
                .into())
            }
            rsp => Err(Self::unexpected_response(
                "Ok, DaemonNotRunning, DaemonNotFound, DaemonWasNotRunning, or DaemonStopFailed",
                &rsp,
            )
            .into()),
        }
    }
}