car-proto 0.55.0

//! Mechanism helpers for the CAR daemon — URL parsing, probing,
//! and spawning the `car-server` binary. Hoisted out of
//! `car-cli` and `car-ffi-common::proxy` (#139-fu1) so both sides
//! share one implementation.
//!
//! These functions encode **mechanism only**, not policy:
//!
//! - "Where is the daemon?" — `daemon_ws_url`, `daemon_port`,
//!   `parse_host_port`.
//! - "Is something listening on its port?" — `probe_daemon_port`.
//! - "Try to start one." — `try_spawn_daemon`.
//!
//! Policy decisions ("should we probe? should we auto-spawn? what
//! happens on failure?") live in the callers — `car-cli` for
//! per-request retry policy, `car-ffi-common::proxy::RuntimeMode`
//! for FFI construction-time daemon-or-embedded resolution. Each
//! has a different shape; they share these primitives.

use std::path::PathBuf;
use std::time::Duration;

/// JSON-RPC methods whose daemon handlers require host-management authority.
///
/// This is the single shared allowlist used by the daemon gate. The manifest
/// generator independently derives the same set from the handler's real gate
/// call sites and a drift test requires exact equality. The role describes an
/// auth-enabled deployment: the underlying daemon gates deliberately degrade
/// to no-ops when no host token is configured, except `session.clear_halt`,
/// whose connection-local latch always requires an already host-bound session.
pub const HOST_MANAGEMENT_METHODS: &[&str] = &[
    "agent_permissions.evaluate_tool",
    "agent_permissions.reset",
    "agent_permissions.reset_tool",
    "agent_permissions.set",
    "agent_permissions.set_default",
    "agent_permissions.set_tool",
    "agents.install",
    "agents.remove",
    "agents.upsert",
    "assistant.identity.set",
    "auth.accounts",
    "auth.authority_hint",
    "auth.complete",
    "auth.completion_status",
    "auth.logout",
    "auth.remove_account",
    "auth.snapshot",
    "auth.start",
    "auth.status",
    "auth.switch_account",
    "auth.switch_org",
    "diagnostics.secret_store_activity",
    "declagents.remove",
    "declagents.set_enabled",
    "messaging.config.get",
    "messaging.config.set",
    "messaging.pairing.start",
    "messaging.pairing.status",
    "messaging.status",
    "messaging.test_send",
    "models.adopt",
    "models.install",
    "models.pull",
    "models.remove",
    "models.resource_policy.set",
    "models.storage_roots",
    "openrouter.auth_cancel",
    "openrouter.auth_start",
    "openrouter.disconnect",
    "openrouter.status",
    "permission.approve",
    "permission.reject",
    "permission.set_tier",
    "session.clear_halt",
    "tasks.schedule",
    "tasks.unschedule",
];

/// Methods a bound supervised agent may invoke for itself and a host may
/// invoke for any managed agent. These are not host-only manifest roles, so
/// they stay outside [`HOST_MANAGEMENT_METHODS`] and its daemon-wide gate.
const AGENT_SELF_OR_HOST_METHODS: &[&str] = &["agents.restart", "agents.start", "agents.stop"];

/// Whether a generic host-management client may attach its host credential to
/// this method. This includes both host-only management operations and the
/// three self-or-host lifecycle operations without duplicating either list in
/// the CLI or daemon client.
pub fn method_accepts_host_authority(method: &str) -> bool {
    HOST_MANAGEMENT_METHODS.contains(&method) || AGENT_SELF_OR_HOST_METHODS.contains(&method)
}

/// WebSocket URL the FFI proxy and CLI both target. Honors
/// `CAR_DAEMON_URL` for cross-host or alt-port setups; defaults
/// to the loopback singleton at port 9100 (matches `car-server`'s
/// default bind).
pub fn daemon_ws_url() -> String {
    std::env::var("CAR_DAEMON_URL").unwrap_or_else(|_| "ws://127.0.0.1:9100".to_string())
}

/// Port number for the daemon. IPv4 and IPv6 forms both accepted.
/// Returns 9100 when the URL is malformed or has no port.
pub fn daemon_port() -> u16 {
    parse_port(&daemon_ws_url()).unwrap_or(9100)
}

/// Extract `host:port` (or `[ipv6]:port`) from a `ws://...` /
/// `wss://...` / bare URL. Returns `None` for malformed inputs
/// and for missing-port URLs (the probe wouldn't know where to
/// look without one).
pub fn parse_host_port(url: &str) -> Option<String> {
    let after_scheme = url.split_once("://").map(|(_, rest)| rest).unwrap_or(url);
    let host_port = after_scheme.split('/').next().unwrap_or("");
    if host_port.starts_with('[') {
        // IPv6 — must have `]:port` to be addressable.
        if host_port.contains("]:") {
            Some(host_port.to_string())
        } else {
            None
        }
    } else if host_port.contains(':') {
        Some(host_port.to_string())
    } else {
        None
    }
}

/// Connect deadline for a daemon on loopback. **Windows only** — see
/// [`connect_deadline`] for why it does not apply elsewhere.
///
/// **Windows refuses a closed loopback port slowly.** Measured on Windows 11:
/// connecting to a *listening* 127.0.0.1 port returns in ~3ms, but every closed
/// one takes ~2050ms to come back `ConnectionRefused` — the SYN is retransmitted
/// before the RST is surfaced. macOS and Linux refuse in microseconds, which is
/// why the cost is invisible where CAR is developed (and why `probe_daemon_port`
/// above can name only Darwin and Linux in its sub-millisecond claim).
///
/// That 2050ms is longer than the 2s budget `car info` gives each of its daemon
/// probes, so on Windows an absent daemon was reported as an unresponsive one —
/// "daemon reachability probe timed out" rather than "daemon not running" — and
/// commands that probe more than once paid the refusal repeatedly: `car info`
/// took ~6.4s and `car models list` ~5.2s with no daemon up, against ~50ms for
/// the daemon-free `car doctor` beside them.
///
/// A local daemon accepts in single-digit milliseconds, so 400ms is a ~100x
/// margin over the observed accept while cutting the not-running verdict well
/// inside every caller's budget. On macOS and Linux nothing changes: refusal
/// already returns long before either deadline.
pub const LOOPBACK_CONNECT_TIMEOUT: Duration = Duration::from_millis(400);

/// Connect deadline for a daemon that is NOT on loopback. Generous, because a
/// remote daemon legitimately takes longer than a local one to accept.
pub const REMOTE_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// Whether `url`'s host is loopback.
///
/// Deliberately conservative: anything that does not parse, or that names a host
/// we cannot confirm is local, is treated as remote. Being wrong in that
/// direction only costs a slow failure; being wrong the other way would cut off
/// a legitimately slow remote connect.
pub fn is_loopback_url(url: &str) -> bool {
    let after_scheme = url.split_once("://").map(|(_, r)| r).unwrap_or(url);
    let authority = after_scheme.split('/').next().unwrap_or("");
    // Drop any userinfo before reading the host.
    let authority = authority.rsplit('@').next().unwrap_or(authority);
    let host = if let Some(v6) = authority.strip_prefix('[') {
        v6.split(']').next().unwrap_or("")
    } else {
        authority.split(':').next().unwrap_or("")
    };
    match host.parse::<std::net::IpAddr>() {
        Ok(ip) => ip.is_loopback(),
        // `localhost` is loopback by definition on every platform CAR targets.
        Err(_) => host.eq_ignore_ascii_case("localhost"),
    }
}

fn connect_deadline_for_platform(url: &str, windows: bool) -> Duration {
    if windows && is_loopback_url(url) {
        LOOPBACK_CONNECT_TIMEOUT
    } else {
        REMOTE_CONNECT_TIMEOUT
    }
}

/// The connect deadline to apply when dialing `url`.
///
/// The short loopback deadline is **Windows-only**, because the problem it
/// solves is Windows-only: there, a *closed* loopback port takes ~2050ms to
/// come back `ConnectionRefused`, which overshoots every caller's budget.
/// macOS and Linux refuse in microseconds, so the short deadline never changes
/// the not-running verdict there — [`LOOPBACK_CONNECT_TIMEOUT`]'s own
/// documentation says as much ("On macOS and Linux nothing changes").
///
/// What it *did* do on Unix was cap the success path. These callers bound
/// `connect_async`, which is the TCP connect **plus the full WebSocket
/// upgrade** — so a daemon that is up and answering is reported absent
/// whenever the local machine cannot schedule both ends of that exchange
/// inside 400ms. That is not hypothetical: it is a recurring CI flake in
/// `car-cli`'s `info_degrades_cleanly_against_a_daemon_without_selfheal_status`,
/// where the fixture daemon shares a loaded runner with the `car info` process
/// it is answering, and both of that command's probes abandon the connection
/// after sending the upgrade request but before the JSON-RPC handshake.
///
/// An accept-but-never-upgrade socket is bounded by this connect deadline
/// itself, not by the later 30s read timeout. On Unix, callers with no outer
/// budget — the reachability probe, generic daemon RPC path, and UniFFI lazy
/// connect (which blocks the JS event loop) — can therefore block for 5 s
/// instead of 400 ms. A caller with its own shorter budget still returns sooner;
/// for example, `car info` gives each daemon probe 2s.
pub fn connect_deadline(url: &str) -> Duration {
    connect_deadline_for_platform(url, cfg!(windows))
}

/// Pull just the trailing port from a URL. `None` when the URL
/// has no port.
pub fn parse_port(url: &str) -> Option<u16> {
    let host_port = parse_host_port(url)?;
    let port_str = if host_port.starts_with('[') {
        host_port.rsplit("]:").next()?
    } else {
        host_port.rsplit(':').next()?
    };
    port_str.parse().ok()
}

/// TCP-probe the daemon's host:port from `daemon_ws_url()`. True
/// iff a TCP connect completes within `timeout`. No WebSocket
/// handshake — just port reachability. Localhost connect to a
/// listening port resolves sub-millisecond on Darwin and Linux,
/// so callers can use a 100ms timeout for the "is it up?"
/// question without waiting on hung-daemon detection (a separate
/// failure mode handled by the WS-handshake timeout downstream).
pub fn probe_daemon_port(timeout: Duration) -> bool {
    let url = daemon_ws_url();
    let host_port = match parse_host_port(&url) {
        Some(hp) => hp,
        None => return false,
    };
    let addrs: Vec<_> = match std::net::ToSocketAddrs::to_socket_addrs(&host_port) {
        Ok(it) => it.collect(),
        Err(_) => return false,
    };
    addrs
        .into_iter()
        .any(|addr| std::net::TcpStream::connect_timeout(&addr, timeout).is_ok())
}

/// Best-effort `car-server` spawn. Returns `Ok(())` on successful
/// fork (the daemon may still take ~500ms-2s to bind its port; the
/// caller is expected to re-probe). Returns `Err` describing the
/// failure mode — `binary not found` for missing `car-server`,
/// `spawn failed: <io error>` when fork itself returned an error
/// (`EACCES`, `ENOMEM`, sandbox refusal, etc.). Callers today
/// only check `is_ok()`, but the structured error string is here
/// for when someone surfaces it to the user.
///
/// Lookup order: sibling-of-current-exe (so `target/release/car`
/// finds its sibling `car-server` without needing PATH) → bare
/// `car-server` (PATH lookup).
pub fn try_spawn_daemon() -> Result<(), String> {
    validate_auto_start_target(&daemon_ws_url())?;
    let mut candidates: Vec<PathBuf> = Vec::new();
    if let Ok(exe) = std::env::current_exe() {
        if let Some(dir) = exe.parent() {
            candidates.push(dir.join("car-server"));
        }
    }
    candidates.push(PathBuf::from("car-server"));

    let port = daemon_port().to_string();
    let mut last_err: Option<std::io::Error> = None;
    for candidate in candidates {
        let mut cmd = std::process::Command::new(&candidate);
        cmd.args(["--port", &port])
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null());
        match cmd.spawn() {
            Ok(_) => return Ok(()),
            Err(e) => last_err = Some(e),
        }
    }

    // `candidates` is never empty (sibling-of-current-exe or PATH
    // lookup always pushes at least the bare "car-server" path),
    // so the loop above touched `last_err` at least once. The
    // `None` arm here is defensive; the optimizer drops it.
    Err(match last_err {
        Some(e) if e.kind() == std::io::ErrorKind::NotFound => {
            "car-server binary not found".to_string()
        }
        Some(e) => format!("car-server spawn failed: {e}"),
        None => unreachable!("candidates is never empty"),
    })
}

// A local process cannot satisfy a configured remote endpoint. In particular,
// never create an extra local daemon as a side effect of a remote outage.
fn validate_auto_start_target(url: &str) -> Result<(), String> {
    if is_loopback_url(url) {
        Ok(())
    } else {
        Err("The configured CAR daemon is not local. Start or reconnect that daemon, or set CAR_DAEMON_URL to a local endpoint. No local daemon was started.".into())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn auto_start_requires_a_local_destination() {
        for url in [
            "ws://127.0.0.1:9100",
            "ws://localhost:19384",
            "ws://[::1]:9100",
        ] {
            assert!(validate_auto_start_target(url).is_ok(), "{url}");
        }
        for url in [
            "wss://remote.example:9100",
            "ws://192.0.2.1:19384",
            "invalid",
            "ws://localhost.example:9100",
        ] {
            let error = validate_auto_start_target(url).unwrap_err();
            assert!(error.contains("No local daemon was started"));
            assert!(!error.contains(url), "do not echo URL credentials");
        }
    }

    #[test]
    fn parse_host_port_basic() {
        assert_eq!(
            parse_host_port("ws://127.0.0.1:9100").as_deref(),
            Some("127.0.0.1:9100")
        );
        assert_eq!(
            parse_host_port("wss://other:1234/json-rpc").as_deref(),
            Some("other:1234")
        );
        assert!(parse_host_port("ws://localhost").is_none());
        assert_eq!(
            parse_host_port("127.0.0.1:9100").as_deref(),
            Some("127.0.0.1:9100")
        );
        // IPv6 + TLS combination.
        assert_eq!(
            parse_host_port("wss://[::1]:9100/").as_deref(),
            Some("[::1]:9100")
        );
        assert!(parse_host_port("ws://[::1]").is_none());
    }

    /// The classifier must cover the spellings CAR actually dials, and must
    /// NOT cover a remote daemon, whose connect is legitimately slower than
    /// 400ms. Getting the second half wrong would cut off real remote use, so the
    /// classifier fails closed on anything it cannot confirm is local.
    ///
    /// Exercise both platform branches even when this test runs on Unix. The
    /// explicit inequality is the positive control: swapping the Windows gate
    /// makes this fail instead of leaving loopback and remote at the same Unix
    /// value.
    #[test]
    fn connect_deadline_is_windows_only_for_loopback() {
        for url in [
            "ws://127.0.0.1:9100",
            "ws://127.0.0.1:9100/",
            "ws://127.0.0.1",
            "ws://localhost:9100",
            "ws://LOCALHOST:9100",
            "wss://127.0.0.1:9100/rpc",
            "ws://[::1]:9100",
            "ws://user:pw@127.0.0.1:9100",
            "ws://127.0.0.53:9100",
        ] {
            assert!(is_loopback_url(url), "expected loopback: {url}");
            let windows_deadline = connect_deadline_for_platform(url, true);
            let unix_deadline = connect_deadline_for_platform(url, false);
            assert_eq!(windows_deadline, LOOPBACK_CONNECT_TIMEOUT, "{url}");
            assert_eq!(unix_deadline, REMOTE_CONNECT_TIMEOUT, "{url}");
            assert_ne!(windows_deadline, unix_deadline, "{url}");
            assert_eq!(
                connect_deadline(url),
                if cfg!(windows) {
                    windows_deadline
                } else {
                    unix_deadline
                },
                "{url}"
            );
        }
        for url in [
            "ws://10.0.0.4:9100",
            "ws://car.internal:9100",
            "wss://daemon.example.com/rpc",
            "ws://[2001:db8::1]:9100",
            "not a url",
            "",
        ] {
            assert!(!is_loopback_url(url), "expected non-loopback: {url}");
            assert_eq!(
                connect_deadline_for_platform(url, true),
                REMOTE_CONNECT_TIMEOUT,
                "{url}"
            );
            assert_eq!(
                connect_deadline_for_platform(url, false),
                REMOTE_CONNECT_TIMEOUT,
                "{url}"
            );
            assert_eq!(connect_deadline(url), REMOTE_CONNECT_TIMEOUT, "{url}");
        }
    }

    /// The whole point of the shorter deadline: far enough under the ~2050ms
    /// Windows loopback refusal to beat every caller's budget, far enough over a
    /// real local accept (~3ms measured) to never truncate one.
    #[test]
    fn loopback_deadline_sits_between_a_local_accept_and_windows_refusal() {
        assert!(LOOPBACK_CONNECT_TIMEOUT > Duration::from_millis(100));
        assert!(LOOPBACK_CONNECT_TIMEOUT < Duration::from_millis(2000));
        assert!(LOOPBACK_CONNECT_TIMEOUT < REMOTE_CONNECT_TIMEOUT);
    }

    #[test]
    fn parse_port_basic() {
        assert_eq!(parse_port("ws://127.0.0.1:9100"), Some(9100));
        assert_eq!(parse_port("wss://other:1234"), Some(1234));
        assert_eq!(parse_port("ws://[::1]:9101"), Some(9101));
        assert_eq!(parse_port("ws://localhost"), None);
        assert_eq!(parse_port("ws://[::1]"), None);
    }

    #[test]
    fn probe_dead_port_returns_false() {
        let prev = std::env::var("CAR_DAEMON_URL").ok();
        std::env::set_var("CAR_DAEMON_URL", "ws://127.0.0.1:1");
        assert!(!probe_daemon_port(Duration::from_millis(100)));
        match prev {
            Some(v) => std::env::set_var("CAR_DAEMON_URL", v),
            None => std::env::remove_var("CAR_DAEMON_URL"),
        }
    }
}