openlatch-client 0.6.19

//! OpenCode Desktop's request plane, Leg B (AGT-6 C-8 *Leg B — relay-armed service*; plan
//! AGT-6-P2 D-3, D-5, D-16, D-18, D-19, D-21).
//!
//! On the 2.0 line OpenCode Desktop makes its model calls from a detached, shared Bun service,
//! `opencode-cli serve --service`, that Desktop, the CLI and the TUI all adopt when its version
//! matches theirs. Bun reads its CA only from the environment it starts with, and whoever spawns
//! the service owns that environment. So the daemon spawns the service itself, with exactly the
//! four relay keys from `model-relay/env.json`, before Desktop does; Desktop then adopts it.
//!
//! What this module does, on every tick (every 10 s, and on a change to `env.json`, the bundled
//! `opencode-cli.version` or OpenCode's `service.json`):
//!
//! | Situation | Action | Reported |
//! | --------- | ------ | -------- |
//! | No bundled CLI (no Desktop, or the 1.19 line) | nothing | [`ServiceState::Absent`] |
//! | Flatpak or snap Desktop | nothing | [`ServiceState::Unsupported`] |
//! | A registration whose pid the service's own `/api/info` does not confirm, or whose identity cannot be read | nothing is stopped | `Unarmed(identity_unverified)` |
//! | The daemon's own service, current version, keys still valid | nothing | [`ServiceState::Armed`] |
//! | Any other live service while an OpenCode process is attached | left alone | `Unarmed(foreign_attached · update_window · awaiting_respawn)` |
//! | Any other live service, none attached | stopped after a re-read, then the daemon's own started | `Starting` |
//! | No live service | the daemon's own started, under the restart bound | `Starting` |
//! | Restart bound exhausted | nothing until the bundled version changes or the daemon restarts | [`ServiceState::Failed`] |
//!
//! The daemon executes only the bundled CLI, or Desktop's staged copy of it when its SHA-256
//! equals the bundled copy's — never a path read from a config file, `service.json` or the
//! environment, and never through a shell. It writes one file, the ledger
//! (`state/opencode-desktop-service.json`, [`ledger`]). OpenCode's `service.json` is untrusted
//! input: read under a 64 KiB cap, its `url` probed only when it is loopback, its `password`
//! held in memory for that one probe and never logged, written or sent anywhere else.
//!
//! The supervised task is spawned by the daemon (`TaskSpec` from [`task_spec`], future from
//! [`DesktopService::run`]); the attestation reads the state from [`subscribe`] /
//! [`current_state`]; `uninstall` calls [`uninstall`].

pub mod ledger;
pub mod process;

use std::collections::{HashMap, VecDeque};
use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime};

use serde::Deserialize;
use tokio::sync::watch;

use crate::core::supervision::task::{RestartPolicy, TaskSpec};
use crate::hooks::proxy_env_file::{self, EnvJson};

use ledger::{Ledger, VerifiedRegistration};
use process::{ProcId, ProcessTable, ServiceEnv, SystemTable, STOP_GRACE};

/// The supervised task's name in `daemon.log`, `/health` and `openlatch status`.
pub const TASK_NAME: &str = "opencode_desktop_service";

/// Tick period (D-18).
const TICK: Duration = Duration::from_secs(10);

/// Budget for one `/api/info` probe (D-18).
const PROBE_BUDGET: Duration = Duration::from_secs(1);

/// Cap on `service.json` (D-18).
const SERVICE_JSON_MAX_BYTES: u64 = 64 * 1024;

/// Cap on an `/api/info` answer.
const INFO_MAX_BYTES: usize = 64 * 1024;

/// Cap on `opencode-cli.version`.
const VERSION_FILE_MAX_BYTES: u64 = 256;

/// How long a service the daemon started may run without registering before it counts as a
/// failed start. Generous: a Desktop-spawned 2.0 service was seen bootstrapping for 3.5 minutes
/// (Spec Risks).
const STARTING_LIMIT: Duration = Duration::from_secs(5 * 60);

/// Restart bound (D-18): at most this many service starts per rolling hour.
const MAX_STARTS_PER_HOUR: usize = 5;
const BOUND_WINDOW: Duration = Duration::from_secs(60 * 60);
const BACKOFF_INITIAL: Duration = Duration::from_secs(2);
const BACKOFF_MAX: Duration = Duration::from_secs(5 * 60);

/// The Basic-auth user OpenCode's own clients send with the registration's password.
const SERVICE_USER: &str = "opencode";

/// Desktop's app id: its Flatpak id, the Linux executable name and a Windows install folder.
#[cfg(any(target_os = "linux", windows))]
const DESKTOP_APP_ID: &str = "ai.opencode.desktop";

// ---------------------------------------------------------------------------
// State
// ---------------------------------------------------------------------------

/// Why the service is `unarmed` (C-11 `ERR_OPENCODE_DESKTOP_SERVICE_UNARMED`; D-20).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UnarmedCause {
    /// A service the daemon did not start is registered while an OpenCode process is attached.
    ForeignAttached,
    /// The registered service's identity could not be verified: `/api/info` did not confirm
    /// `service.json`'s pid, its URL is not loopback, or its start identity is unreadable.
    IdentityUnverified,
    /// The bundled version changed and the running service is of the old one.
    UpdateWindow,
    /// The daemon's own service is stale after a trust release or a CA rotation, or the relay
    /// keys are not available yet, and the respawn is waiting.
    AwaitingRespawn,
}

impl UnarmedCause {
    /// The wire token the finding names.
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Self::ForeignAttached => "foreign_attached",
            Self::IdentityUnverified => "identity_unverified",
            Self::UpdateWindow => "update_window",
            Self::AwaitingRespawn => "awaiting_respawn",
        }
    }
}

/// What the manager last found (plan *Design per Module*: `Absent · Unsupported · Starting ·
/// Armed · Unarmed(cause) · Failed(reason)`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ServiceState {
    /// No bundled OpenCode CLI on this host: no Desktop, or a Desktop line that ships none.
    Absent,
    /// A Desktop the daemon cannot spawn into (Flatpak, snap); the reason names the packaging.
    Unsupported(String),
    /// The daemon started its service and is waiting for it to register.
    Starting,
    /// The daemon's own service is registered and carries the current relay keys.
    Armed,
    Unarmed(UnarmedCause),
    /// The daemon could not start or register the service, or the restart bound is spent.
    Failed(String),
}

fn state_cell() -> &'static watch::Sender<ServiceState> {
    static CELL: OnceLock<watch::Sender<ServiceState>> = OnceLock::new();
    CELL.get_or_init(|| watch::channel(ServiceState::Absent).0)
}

/// Follow the manager's state (the attestation reads it).
#[must_use]
pub fn subscribe() -> watch::Receiver<ServiceState> {
    state_cell().subscribe()
}

/// The manager's last published state.
#[must_use]
pub fn current_state() -> ServiceState {
    state_cell().borrow().clone()
}

fn publish(state: &ServiceState) {
    state_cell().send_if_modified(|s| {
        if s == state {
            return false;
        }
        *s = state.clone();
        true
    });
}

// ---------------------------------------------------------------------------
// Detection
// ---------------------------------------------------------------------------

/// Where OpenCode Desktop's bundled CLI, its staged copy and OpenCode's registration live.
#[derive(Debug, Clone, Default)]
pub struct DesktopPaths {
    /// Candidate `resources` directories holding `opencode-cli` and `opencode-cli.version`.
    pub resource_dirs: Vec<PathBuf>,
    /// `<Desktop user data>/cli`: Desktop stages `<version>/opencode-cli` here.
    pub staged_root: Option<PathBuf>,
    /// Install locations of a sandboxed Desktop, with the packaging they mean.
    pub sandboxed: Vec<(PathBuf, &'static str)>,
    /// OpenCode's `service.json`.
    pub service_json: Option<PathBuf>,
}

/// The service binary's file name on this OS.
fn exe_name() -> &'static str {
    if cfg!(windows) {
        "opencode-cli.exe"
    } else {
        process::SERVICE_BINARY
    }
}

impl DesktopPaths {
    /// The host's paths (C-8 Leg B *Detect*, *Service record*).
    #[must_use]
    pub fn for_host() -> Self {
        let home = dirs::home_dir();
        let mut paths = host_install_paths(home.as_deref());
        paths.service_json = service_json_path(
            std::env::var_os("XDG_STATE_HOME"), // portability-ok: OpenCode reads XDG_STATE_HOME as given on every OS, not only Linux
            home.as_deref(),
        );
        paths
    }
}

/// `$XDG_STATE_HOME/opencode/service.json`, else `~/.local/state/opencode/service.json` — on
/// every OS, as OpenCode resolves it.
fn service_json_path(xdg_state: Option<OsString>, home: Option<&Path>) -> Option<PathBuf> {
    let base = match xdg_state.filter(|v| !v.is_empty()).map(PathBuf::from) {
        Some(p) if p.is_absolute() => p,
        _ => home?.join(".local").join("state"),
    };
    Some(base.join("opencode").join("service.json"))
}

#[cfg(target_os = "macos")]
fn host_install_paths(home: Option<&Path>) -> DesktopPaths {
    let mut resource_dirs = vec![PathBuf::from("/Applications")
        .join("OpenCode.app")
        .join("Contents")
        .join("Resources")];
    if let Some(home) = home {
        resource_dirs.push(
            home.join("Applications")
                .join("OpenCode.app")
                .join("Contents")
                .join("Resources"),
        );
    }
    DesktopPaths {
        resource_dirs,
        staged_root: home.map(|h| {
            h.join("Library")
                .join("Application Support")
                .join("OpenCode")
                .join("cli")
        }),
        sandboxed: Vec::new(),
        service_json: None,
    }
}

#[cfg(target_os = "linux")]
fn host_install_paths(home: Option<&Path>) -> DesktopPaths {
    let mut sandboxed = vec![
        (
            PathBuf::from("/var/lib/flatpak/app").join(DESKTOP_APP_ID),
            "flatpak",
        ),
        (PathBuf::from("/snap/opencode-desktop"), "snap"),
    ];
    if let Some(home) = home {
        sandboxed.push((
            home.join(".local")
                .join("share")
                .join("flatpak")
                .join("app")
                .join(DESKTOP_APP_ID),
            "flatpak",
        ));
    }
    DesktopPaths {
        resource_dirs: vec![PathBuf::from("/opt/OpenCode/resources")],
        staged_root: dirs::config_dir().map(|c| c.join("OpenCode").join("cli")),
        sandboxed,
        service_json: None,
    }
}

#[cfg(windows)]
fn host_install_paths(_home: Option<&Path>) -> DesktopPaths {
    let resource_dirs = dirs::data_local_dir()
        .map(|local| {
            let programs = local.join("Programs");
            ["OpenCode", "opencode", DESKTOP_APP_ID]
                .iter()
                .map(|d| programs.join(d).join("resources"))
                .collect()
        })
        .unwrap_or_default();
    DesktopPaths {
        resource_dirs,
        staged_root: dirs::config_dir().map(|c| c.join("OpenCode").join("cli")),
        sandboxed: Vec::new(),
        service_json: None,
    }
}

#[cfg(not(any(target_os = "macos", target_os = "linux", windows)))]
fn host_install_paths(_home: Option<&Path>) -> DesktopPaths {
    DesktopPaths::default()
}

/// OpenCode Desktop's bundled CLI.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BundledCli {
    pub path: PathBuf,
    pub version: String,
}

/// What [`detect_in`] found.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Detection {
    Absent,
    Unsupported(&'static str),
    Found(BundledCli),
}

/// OpenCode Desktop's bundled CLI on this host, if any (plan API).
#[must_use]
pub fn detect() -> Option<BundledCli> {
    match detect_in(&DesktopPaths::for_host()) {
        Detection::Found(cli) => Some(cli),
        _ => None,
    }
}

/// The first resource directory holding both `opencode-cli` and a readable
/// `opencode-cli.version`; else a sandboxed install; else nothing.
#[must_use]
pub fn detect_in(paths: &DesktopPaths) -> Detection {
    for dir in &paths.resource_dirs {
        let bin = dir.join(exe_name());
        if !bin.is_file() {
            continue;
        }
        if let Some(version) = read_version(&dir.join("opencode-cli.version")) {
            return Detection::Found(BundledCli { path: bin, version });
        }
    }
    paths
        .sandboxed
        .iter()
        .find(|(p, _)| p.exists())
        .map_or(Detection::Absent, |(_, why)| Detection::Unsupported(why))
}

/// `opencode-cli.version`, trimmed: 1 to 64 characters of `[A-Za-z0-9._+-]`, or `None`.
fn read_version(path: &Path) -> Option<String> {
    use std::io::Read;
    let mut raw = String::new();
    std::fs::File::open(path)
        .ok()?
        .take(VERSION_FILE_MAX_BYTES)
        .read_to_string(&mut raw)
        .ok()?;
    let v = raw.trim();
    let ok = (1..=64).contains(&v.len())
        && v.bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'+' | b'-'));
    ok.then(|| v.to_string())
}

/// Desktop's staged directory name for a version (`cli-version` sanitising in OpenCode).
fn staged_dir_name(version: &str) -> String {
    version
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') {
                c
            } else {
                '-'
            }
        })
        .collect()
}

/// SHA-256 digests by file fingerprint, so a 200 MB binary is hashed once per change.
#[derive(Debug, Default)]
struct DigestCache(HashMap<PathBuf, (u64, Option<SystemTime>, String)>);

impl DigestCache {
    fn sha256(&mut self, path: &Path) -> std::io::Result<String> {
        use sha2::{Digest, Sha256};
        use std::io::Read;
        let meta = std::fs::metadata(path)?;
        let key = (meta.len(), meta.modified().ok());
        if let Some((len, modified, digest)) = self.0.get(path) {
            if (*len, *modified) == key {
                return Ok(digest.clone());
            }
        }
        let mut file = std::fs::File::open(path)?;
        let mut hasher = Sha256::new();
        let mut buf = vec![0u8; 64 * 1024];
        loop {
            let n = file.read(&mut buf)?;
            if n == 0 {
                break;
            }
            hasher.update(&buf[..n]);
        }
        let digest = hex::encode(hasher.finalize());
        self.0
            .insert(path.to_path_buf(), (key.0, key.1, digest.clone()));
        Ok(digest)
    }
}

/// The binary to run and its digest: Desktop's staged copy when its SHA-256 equals the bundled
/// copy's, else the bundled copy (C-8 Leg B *Detect*).
fn resolve_executable(
    cli: &BundledCli,
    staged_root: Option<&Path>,
    digests: &mut DigestCache,
) -> std::io::Result<(PathBuf, String)> {
    let bundled = digests.sha256(&cli.path)?;
    if let Some(root) = staged_root {
        let staged = root.join(staged_dir_name(&cli.version)).join(exe_name());
        if staged.is_file() && digests.sha256(&staged).ok().as_deref() == Some(bundled.as_str()) {
            return Ok((staged, bundled));
        }
    }
    Ok((cli.path.clone(), bundled))
}

// ---------------------------------------------------------------------------
// OpenCode's registration and its `/api/info` probe
// ---------------------------------------------------------------------------

/// The registration password. Held in memory for the `/api/info` probe only; never logged,
/// recorded, written or sent anywhere else, and never printed by `Debug`.
#[derive(Clone, PartialEq, Eq)]
pub struct ServicePassword(String);

impl std::fmt::Debug for ServicePassword {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("ServicePassword(<redacted>)")
    }
}

/// OpenCode's `service.json` (`url, pid, version, password, id`), as untrusted input.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Registration {
    pub url: String,
    pub pid: u32,
    pub version: Option<String>,
    password: Option<ServicePassword>,
}

#[derive(Deserialize)]
struct RawRegistration {
    url: String,
    pid: u64,
    #[serde(default)]
    version: Option<String>,
    #[serde(default)]
    password: Option<String>,
}

/// Read `service.json` under its 64 KiB cap. `None` when absent, oversized or malformed.
#[must_use]
pub fn read_registration(path: &Path) -> Option<Registration> {
    use std::io::Read;
    let file = std::fs::File::open(path).ok()?;
    let mut raw = Vec::new();
    file.take(SERVICE_JSON_MAX_BYTES + 1)
        .read_to_end(&mut raw)
        .ok()?;
    if raw.len() as u64 > SERVICE_JSON_MAX_BYTES {
        return None;
    }
    let r: RawRegistration = serde_json::from_slice(&raw).ok()?;
    Some(Registration {
        url: r.url,
        pid: u32::try_from(r.pid).ok().filter(|p| *p > 0)?,
        version: r.version,
        password: r.password.map(ServicePassword),
    })
}

/// A loopback `http://` URL: host `127.0.0.1`, `[::1]` or `localhost`, a port, at most a `/`
/// after it, no userinfo. Returns the `Host` header value and the addresses to try.
fn loopback_target(url: &str) -> Option<(String, Vec<std::net::SocketAddr>)> {
    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
    let rest = url.strip_prefix("http://")?;
    let authority = rest.strip_suffix('/').unwrap_or(rest);
    if authority.contains(['/', '@', '?', '#']) {
        return None;
    }
    let (host, port) = authority.rsplit_once(':')?;
    let port: u16 = port
        .bytes()
        .all(|b| b.is_ascii_digit())
        .then(|| port.parse().ok())
        .flatten()
        .filter(|p| *p > 0)?;
    let v4 = IpAddr::V4(Ipv4Addr::LOCALHOST);
    let v6 = IpAddr::V6(Ipv6Addr::LOCALHOST);
    let ips = match host {
        "127.0.0.1" => vec![v4],
        "[::1]" => vec![v6],
        "localhost" => vec![v4, v6],
        _ => return None,
    };
    Some((
        authority.to_string(),
        ips.into_iter()
            .map(|ip| SocketAddr::new(ip, port))
            .collect(),
    ))
}

/// The registration's URL is loopback (C-8 Leg B *Service record*).
#[must_use]
pub fn is_loopback_url(url: &str) -> bool {
    loopback_target(url).is_some()
}

/// What `/api/info` answered.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
pub struct InfoAnswer {
    pub pid: u32,
    #[serde(default)]
    pub version: Option<String>,
}

/// The `/api/info` probe, behind a trait so the manager's rules are tested without a server.
pub trait InfoProbe: Send + Sync {
    /// `GET <url>/api/info` with the registration's password, within `budget`. An `Err` never
    /// carries the password.
    fn probe(&self, reg: &Registration, budget: Duration) -> Result<InfoAnswer, String>;
}

/// The real probe: one HTTP/1.1 request over a loopback TCP connection, no proxy.
#[derive(Debug, Default, Clone, Copy)]
pub struct HttpInfoProbe;

impl InfoProbe for HttpInfoProbe {
    fn probe(&self, reg: &Registration, budget: Duration) -> Result<InfoAnswer, String> {
        use base64::Engine;
        use std::io::{Read, Write};
        let deadline = Instant::now() + budget;
        let (host, addrs) = loopback_target(&reg.url).ok_or("registration URL is not loopback")?;
        let remaining = || deadline.saturating_duration_since(Instant::now());
        let mut stream = None;
        for addr in addrs {
            let left = remaining();
            if left.is_zero() {
                break;
            }
            if let Ok(s) = std::net::TcpStream::connect_timeout(&addr, left) {
                stream = Some(s);
                break;
            }
        }
        let mut stream = stream.ok_or("could not connect to the registered service")?;
        let mut request = format!(
            "GET /api/info HTTP/1.1\r\nHost: {host}\r\nAccept: application/json\r\nConnection: close\r\n"
        );
        if let Some(pw) = &reg.password {
            let token = base64::engine::general_purpose::STANDARD
                .encode(format!("{SERVICE_USER}:{}", pw.0));
            request.push_str("Authorization: Basic ");
            request.push_str(&token);
            request.push_str("\r\n");
        }
        request.push_str("\r\n");
        let io = |e: std::io::Error| format!("probe I/O: {}", e.kind());
        stream
            .set_write_timeout(Some(remaining().max(Duration::from_millis(1))))
            .map_err(io)?;
        stream.write_all(request.as_bytes()).map_err(io)?;
        drop(request);
        let mut raw = Vec::new();
        let mut buf = [0u8; 4096];
        loop {
            let left = remaining();
            if left.is_zero() {
                return Err("probe timed out".into());
            }
            stream.set_read_timeout(Some(left)).map_err(io)?;
            match stream.read(&mut buf) {
                Ok(0) => break,
                Ok(n) => {
                    raw.extend_from_slice(&buf[..n]);
                    if raw.len() > INFO_MAX_BYTES {
                        return Err("probe answer too large".into());
                    }
                    if response_complete(&raw) {
                        break;
                    }
                }
                Err(e) => return Err(io(e)),
            }
        }
        parse_info_response(&raw)
    }
}

/// Response headers, names lower-cased.
type Headers = Vec<(String, String)>;

/// Split an HTTP response into `(status, headers, body)`.
fn split_response(raw: &[u8]) -> Option<(u16, Headers, &[u8])> {
    let end = raw.windows(4).position(|w| w == b"\r\n\r\n")?;
    let head = std::str::from_utf8(&raw[..end]).ok()?;
    let mut lines = head.split("\r\n");
    let status = lines.next()?.split_whitespace().nth(1)?.parse().ok()?;
    let headers = lines
        .filter_map(|l| l.split_once(':'))
        .map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string()))
        .collect();
    Some((status, headers, &raw[end + 4..]))
}

/// A response is complete once its `Content-Length` body has arrived (a chunked or unsized one
/// is read to the close).
fn response_complete(raw: &[u8]) -> bool {
    let Some((_, headers, body)) = split_response(raw) else {
        return false;
    };
    headers
        .iter()
        .find(|(k, _)| k == "content-length")
        .and_then(|(_, v)| v.parse::<usize>().ok())
        .is_some_and(|len| body.len() >= len)
}

/// Decode a chunked body. `None` on a malformed one.
fn dechunk(mut body: &[u8]) -> Option<Vec<u8>> {
    let mut out = Vec::new();
    loop {
        let line_end = body.windows(2).position(|w| w == b"\r\n")?;
        let size_text = std::str::from_utf8(&body[..line_end]).ok()?;
        let size = usize::from_str_radix(size_text.split(';').next()?.trim(), 16).ok()?;
        body = &body[line_end + 2..];
        if size == 0 {
            return Some(out);
        }
        if body.len() < size.checked_add(2)? {
            return None;
        }
        out.extend_from_slice(&body[..size]);
        body = &body[size + 2..];
    }
}

fn parse_info_response(raw: &[u8]) -> Result<InfoAnswer, String> {
    let (status, headers, body) = split_response(raw).ok_or("malformed probe answer")?;
    if status != 200 {
        return Err(format!("probe answered HTTP {status}"));
    }
    let chunked = headers
        .iter()
        .any(|(k, v)| k == "transfer-encoding" && v.to_ascii_lowercase().contains("chunked"));
    let body = if chunked {
        dechunk(body).ok_or("malformed chunked probe answer")?
    } else {
        let len = headers
            .iter()
            .find(|(k, _)| k == "content-length")
            .and_then(|(_, v)| v.parse::<usize>().ok())
            .unwrap_or(body.len())
            .min(body.len());
        body[..len].to_vec()
    };
    serde_json::from_slice(&body).map_err(|_| "probe answer is not an info object".to_string())
}

// ---------------------------------------------------------------------------
// Restart bound (D-18)
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Permit {
    Go,
    Wait,
    Exhausted,
}

/// At most [`MAX_STARTS_PER_HOUR`] service starts per rolling hour — every spawn counts, a first
/// start, a restart after an exit and a replacement after a proxy or CA change alike — with a
/// failure backoff of 2 s doubling to 5 min kept apart from that window; past the bound, nothing
/// until the bundled version changes or the daemon restarts. The bound is the manager's own
/// state, not the supervisor's (which never gives up by design).
#[derive(Debug)]
struct RestartBound {
    starts: VecDeque<Instant>,
    backoff: Duration,
    not_before: Option<Instant>,
    exhausted_for: Option<String>,
}

impl Default for RestartBound {
    fn default() -> Self {
        Self {
            starts: VecDeque::new(),
            backoff: BACKOFF_INITIAL,
            not_before: None,
            exhausted_for: None,
        }
    }
}

impl RestartBound {
    /// A new bundled version lifts an exhausted bound.
    fn observe_version(&mut self, version: &str) {
        if self.exhausted_for.as_deref().is_some_and(|v| v != version) {
            *self = Self::default();
        }
    }

    fn exhausted(&self) -> bool {
        self.exhausted_for.is_some()
    }

    fn permit(&mut self, now: Instant, version: &str) -> Permit {
        if self.exhausted() {
            return Permit::Exhausted;
        }
        while self
            .starts
            .front()
            .is_some_and(|t| now.saturating_duration_since(*t) >= BOUND_WINDOW)
        {
            self.starts.pop_front();
        }
        if self.starts.len() >= MAX_STARTS_PER_HOUR {
            self.exhausted_for = Some(version.to_string());
            return Permit::Exhausted;
        }
        if self.not_before.is_some_and(|t| now < t) {
            return Permit::Wait;
        }
        Permit::Go
    }

    /// A spawn is about to happen: charged to the hourly window whatever its outcome.
    fn note_start(&mut self, now: Instant) {
        self.starts.push_back(now);
    }

    /// The daemon's own service exited, or failed to start or to register: the backoff only.
    fn note_exit(&mut self, now: Instant) {
        self.not_before = Some(now + self.backoff);
        self.backoff = (self.backoff * 2).min(BACKOFF_MAX);
    }

    /// A healthy, armed service resets the backoff curve (the hourly window stands).
    fn note_armed(&mut self) {
        self.backoff = BACKOFF_INITIAL;
        self.not_before = None;
    }

    fn exhausted_reason(&self) -> String {
        format!(
            "the OpenLatch daemon started OpenCode's service {MAX_STARTS_PER_HOUR} times within an hour"
        )
    }
}

// ---------------------------------------------------------------------------
// The manager
// ---------------------------------------------------------------------------

/// The registered service, as the manager classified it.
enum Live {
    /// No live service is registered (none, or a registration whose pid is gone or is not an
    /// OpenCode process), or the daemon's own service has not answered yet.
    None,
    /// A registration naming a live OpenCode process whose pid `/api/info` did not confirm, whose
    /// URL is not loopback, or whose identity is unreadable: never signalled.
    Unverified,
    /// Confirmed by `/api/info`, with a fresh identity read.
    Verified(VerifiedRegistration, ProcId),
}

type CaDigest = dyn Fn(&Path) -> Option<String> + Send + Sync;
type InheritedEnv = dyn Fn() -> Vec<(OsString, OsString)> + Send + Sync;
type Elevated = dyn Fn() -> bool + Send + Sync;

/// The Leg B rules (module doc), one [`Manager::tick`] at a time.
pub struct Manager {
    ol_dir: PathBuf,
    paths: DesktopPaths,
    table: Arc<dyn ProcessTable>,
    probe: Arc<dyn InfoProbe>,
    ca_digest: Box<CaDigest>,
    inherited_env: Box<InheritedEnv>,
    elevated: Box<Elevated>,
    bound: RestartBound,
    digests: DigestCache,
    /// The daemon's own service this manager last saw running (or just started): only its
    /// disappearance counts as an exit. A ledger left from before a reboot does not.
    own_seen: Option<u32>,
}

impl std::fmt::Debug for Manager {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Manager")
            .field("ol_dir", &self.ol_dir)
            .field("paths", &self.paths)
            .field("bound", &self.bound)
            .finish_non_exhaustive()
    }
}

/// The relay CA's SHA-256 (lowercase hex), from the CA on disk.
fn relay_ca_digest(ol_dir: &Path) -> Option<String> {
    let dir = crate::model_relay::ca::ca_dir(ol_dir);
    crate::model_relay::ca::inspect(&dir).map(|info| info.sha256_hex.to_ascii_lowercase())
}

impl Manager {
    /// The host's manager for `ol_dir` (the OpenLatch directory).
    #[must_use]
    pub fn for_host(ol_dir: PathBuf) -> Self {
        Self::with_parts(
            ol_dir,
            DesktopPaths::for_host(),
            Arc::new(SystemTable),
            Arc::new(HttpInfoProbe),
        )
    }

    /// A manager over explicit parts; the CA digest, inherited environment and elevation check
    /// are the host's.
    #[must_use]
    pub fn with_parts(
        ol_dir: PathBuf,
        paths: DesktopPaths,
        table: Arc<dyn ProcessTable>,
        probe: Arc<dyn InfoProbe>,
    ) -> Self {
        Self {
            ol_dir,
            paths,
            table,
            probe,
            ca_digest: Box::new(relay_ca_digest),
            inherited_env: Box::new(|| std::env::vars_os().collect()),
            elevated: Box::new(process::running_elevated),
            bound: RestartBound::default(),
            digests: DigestCache::default(),
            own_seen: None,
        }
    }

    /// The files whose change should trigger a tick: `env.json`, the bundled version file and
    /// OpenCode's `service.json`. Their parent directories are what a watcher watches.
    #[must_use]
    pub fn watched_files(&self) -> Vec<PathBuf> {
        let mut out = vec![proxy_env_file::env_json_path(&self.ol_dir)];
        out.extend(
            self.paths
                .resource_dirs
                .iter()
                .map(|d| d.join("opencode-cli.version")),
        );
        out.extend(self.paths.service_json.clone());
        out
    }

    /// One pass of the Leg B rules at `now` (monotonic) and `wall_ms` (milliseconds since the
    /// epoch, recorded as the ledger's `started_at`).
    pub fn tick(&mut self, now: Instant, wall_ms: u64) -> ServiceState {
        let cli = match detect_in(&self.paths) {
            Detection::Absent => return ServiceState::Absent,
            Detection::Unsupported(why) => return ServiceState::Unsupported(why.to_string()),
            Detection::Found(cli) => cli,
        };
        self.bound.observe_version(&cli.version);
        if self.bound.exhausted() {
            return ServiceState::Failed(self.bound.exhausted_reason());
        }
        let env = proxy_env_file::read_env_json(&self.ol_dir);
        let ca = (self.ca_digest)(&self.ol_dir);

        let previous = ledger::read(&self.ol_dir);
        let own = match ledger::current(&self.ol_dir, &*self.table) {
            ledger::Current::Own(l, p) => Some((l, p)),
            // The daemon's own pid runs with an identity it cannot read: never acted on, and
            // never taken for an exit (D-19, fail closed).
            ledger::Current::Unverified(_) => {
                return ServiceState::Unarmed(UnarmedCause::IdentityUnverified);
            }
            ledger::Current::None => None,
        };
        if own.is_none() && previous.is_some_and(|l| self.own_seen == Some(l.pid)) {
            // The service the daemon started is gone without the daemon stopping it.
            self.bound.note_exit(now);
        }
        self.own_seen = own.as_ref().map(|(l, _)| l.pid);

        let registration = self
            .paths
            .service_json
            .as_deref()
            .and_then(read_registration);
        let own_pid = own.as_ref().map(|(l, _)| l.pid);
        let live = registration.map_or(Live::None, |r| self.classify(&r));

        match live {
            Live::Unverified => {
                return ServiceState::Unarmed(UnarmedCause::IdentityUnverified);
            }
            Live::Verified(reg, proc_id) => {
                let ours = own
                    .as_ref()
                    .is_some_and(|(l, _)| l.owns(&reg, Some(&proc_id)));
                let cause = match own.as_ref().filter(|_| ours) {
                    Some((l, _)) if l.version != cli.version => UnarmedCause::UpdateWindow,
                    Some((l, _)) if l.armed(env.as_ref(), ca.as_deref()) => {
                        self.bound.note_armed();
                        return ServiceState::Armed;
                    }
                    Some(_) => UnarmedCause::AwaitingRespawn,
                    None => UnarmedCause::ForeignAttached,
                };
                // A foreign service is only replaced when the daemon can start its own right
                // after: stopping it otherwise leaves OpenCode with no service at all. The
                // daemon's own stale service is broken (old keys), so it goes either way.
                if !ours && self.readiness(now, &cli, env.is_some() && ca.is_some()) != Permit::Go {
                    return ServiceState::Unarmed(cause);
                }
                let mut except = vec![proc_id.pid];
                except.extend(own_pid);
                if let Some(blocked) = self.attached_cause(&except, cause) {
                    return ServiceState::Unarmed(blocked);
                }
                if !self.at_detect_path(&proc_id.exe, &cli) {
                    return ServiceState::Unarmed(UnarmedCause::IdentityUnverified);
                }
                let reason = match cause {
                    UnarmedCause::UpdateWindow => "the bundled OpenCode version changed",
                    UnarmedCause::AwaitingRespawn => "its relay keys no longer hold",
                    _ => "a service the daemon did not start, with no OpenCode process attached",
                };
                tracing::info!(
                    pid = proc_id.pid,
                    binary = %proc_id.exe.display(),
                    reason,
                    "stopping OpenCode's service"
                );
                let outcome = self.table.stop(&proc_id, STOP_GRACE);
                if !outcome.is_gone() {
                    tracing::warn!(
                        pid = proc_id.pid,
                        binary = %proc_id.exe.display(),
                        outcome = ?outcome,
                        "OpenCode's service was not stopped"
                    );
                    return ServiceState::Unarmed(UnarmedCause::IdentityUnverified);
                }
                if ours {
                    self.remove_ledger();
                } else if own.is_some() {
                    // The daemon's own service is still starting; with the foreign one gone it
                    // can register now.
                    return ServiceState::Starting;
                }
            }
            Live::None => {
                if let Some((l, p)) = &own {
                    if wall_ms.saturating_sub(l.started_at)
                        < u64::try_from(STARTING_LIMIT.as_millis()).unwrap_or(u64::MAX)
                    {
                        return ServiceState::Starting;
                    }
                    // The attached rule holds before any timeout-based stop, as before any other.
                    if let Some(blocked) =
                        self.attached_cause(&[l.pid], UnarmedCause::AwaitingRespawn)
                    {
                        return ServiceState::Unarmed(blocked);
                    }
                    tracing::info!(
                        pid = p.pid,
                        binary = %p.exe.display(),
                        reason = "it never registered",
                        "stopping OpenCode's service"
                    );
                    if !self.table.stop(p, STOP_GRACE).is_gone() {
                        return ServiceState::Failed(format!(
                            "the OpenCode service the daemon started (pid {}) never registered and could not be stopped",
                            p.pid
                        ));
                    }
                    self.remove_ledger();
                    self.bound.note_exit(now);
                }
            }
        }
        self.start(&cli, env, ca, now, wall_ms)
    }

    /// Whether a start may happen now (keys present, not elevated, the bound permits).
    fn readiness(&mut self, now: Instant, cli: &BundledCli, keys: bool) -> Permit {
        if !keys || (self.elevated)() {
            return Permit::Wait;
        }
        self.bound.permit(now, &cli.version)
    }

    /// The attached rule (D-19) as the cause a stop is held back with: `cause` while an OpenCode
    /// process other than `except` is attached, `identity_unverified` while the process table
    /// cannot be read (fail closed), `None` when nothing is attached.
    fn attached_cause(&self, except: &[u32], cause: UnarmedCause) -> Option<UnarmedCause> {
        match process::attached(&*self.table, except) {
            process::Attached::No => None,
            process::Attached::Yes => Some(cause),
            process::Attached::Unknown => Some(UnarmedCause::IdentityUnverified),
        }
    }

    /// Classify a registration (D-19). A live OpenCode process whose registration the probe did
    /// not confirm is unverified — the daemon's own service included: a registration naming its
    /// pid means it registered, so a failed or mismatched probe is never "not registered yet".
    fn classify(&self, reg: &Registration) -> Live {
        let fresh = self.table.read(reg.pid);
        let opencode_alive = || {
            fresh
                .as_ref()
                .is_some_and(|p| process::is_opencode_exe(&p.exe))
        };
        if !is_loopback_url(&reg.url) {
            return if opencode_alive() {
                Live::Unverified
            } else {
                Live::None
            };
        }
        match self.probe.probe(reg, PROBE_BUDGET) {
            Ok(answer) if answer.pid == reg.pid => match fresh {
                Some(p) => Live::Verified(
                    VerifiedRegistration {
                        pid: reg.pid,
                        version: answer.version,
                    },
                    p,
                ),
                // An unreadable start identity fails closed.
                None => Live::Unverified,
            },
            _ if opencode_alive() => Live::Unverified,
            // The pid is gone, or now belongs to something that is not OpenCode: a stale
            // registration, so no live service.
            _ => Live::None,
        }
    }

    /// `exe` is an `opencode-cli` at a Detect path: a resource directory, or under the staged
    /// root.
    fn at_detect_path(&self, exe: &Path, cli: &BundledCli) -> bool {
        if !process::is_service_binary(exe) {
            return false;
        }
        let canon = |p: &Path| std::fs::canonicalize(p).unwrap_or_else(|_| p.to_path_buf());
        let exe = canon(exe);
        let key = |p: &Path| crate::core::path_compat::dedup_key(p);
        let parent = exe.parent().map(key);
        let in_resources = self
            .paths
            .resource_dirs
            .iter()
            .chain(cli.path.parent().map(Path::to_path_buf).iter())
            .any(|d| parent.as_ref() == Some(&key(&canon(d))));
        let in_staged = self
            .paths
            .staged_root
            .as_deref()
            .is_some_and(|root| key(&exe).starts_with(key(&canon(root))));
        in_resources || in_staged
    }

    fn remove_ledger(&self) {
        if let Err(e) = ledger::remove(&self.ol_dir) {
            tracing::warn!(error = %e, "could not remove the OpenCode service ledger");
        }
    }

    /// Start the daemon's own service with the four relay keys and write its ledger.
    fn start(
        &mut self,
        cli: &BundledCli,
        env: Option<EnvJson>,
        ca: Option<String>,
        now: Instant,
        wall_ms: u64,
    ) -> ServiceState {
        if (self.elevated)() {
            return ServiceState::Failed(
                "the OpenLatch daemon runs with elevated rights, so it does not start OpenCode's service".into(),
            );
        }
        let (Some(env), Some(ca)) = (env, ca) else {
            return ServiceState::Unarmed(UnarmedCause::AwaitingRespawn);
        };
        match self.bound.permit(now, &cli.version) {
            Permit::Go => {}
            Permit::Wait => return ServiceState::Unarmed(UnarmedCause::AwaitingRespawn),
            Permit::Exhausted => return ServiceState::Failed(self.bound.exhausted_reason()),
        }
        let (exe, sha256) =
            match resolve_executable(cli, self.paths.staged_root.as_deref(), &mut self.digests) {
                Ok(found) => found,
                Err(e) => {
                    return ServiceState::Failed(format!("reading {}: {e}", cli.path.display()));
                }
            };
        let service_env = ServiceEnv::new((self.inherited_env)(), &env);
        self.bound.note_start(now);
        let spawned = match self.table.spawn(&exe, &service_env) {
            Ok(p) => p,
            Err(e) => {
                self.bound.note_exit(now);
                return ServiceState::Failed(format!("starting {}: {e}", exe.display()));
            }
        };
        let entry = Ledger {
            pid: spawned.pid,
            start: spawned.start.clone(),
            binary: spawned.exe.clone(),
            sha256,
            version: cli.version.clone(),
            started_at: wall_ms,
            proxy_url: env.https_proxy.clone(),
            ca_sha256: ca,
        };
        if let Err(e) = ledger::write(&self.ol_dir, &entry) {
            // A service the ledger cannot name must not keep running.
            let _ = self.table.stop(&spawned, STOP_GRACE);
            self.bound.note_exit(now);
            return ServiceState::Failed(format!("writing the OpenCode service ledger: {e}"));
        }
        self.own_seen = Some(spawned.pid);
        tracing::info!(
            pid = spawned.pid,
            binary = %spawned.exe.display(),
            version = %cli.version,
            reason = "no live OpenCode service was registered",
            "started OpenCode's service through the OpenLatch local model relay"
        );
        ServiceState::Starting
    }
}

// ---------------------------------------------------------------------------
// The supervised task
// ---------------------------------------------------------------------------

/// The `TaskSpec` the daemon supervises Leg B under (D-18): restarted on every exit.
#[must_use]
pub fn task_spec() -> TaskSpec {
    TaskSpec::new(TASK_NAME, RestartPolicy::Always)
}

/// The Leg B task. The manager lives behind a shared lock, outside the supervised future, so a
/// panic and restart keep the restart bound (the factory produces a fresh future per run).
#[derive(Debug, Clone)]
pub struct DesktopService {
    manager: Arc<Mutex<Manager>>,
}

impl DesktopService {
    /// The host's Leg B task for `ol_dir`.
    #[must_use]
    pub fn for_host(ol_dir: PathBuf) -> Self {
        Self::new(Manager::for_host(ol_dir))
    }

    #[must_use]
    pub fn new(manager: Manager) -> Self {
        Self {
            manager: Arc::new(Mutex::new(manager)),
        }
    }

    /// Run one tick on a blocking thread and publish its state.
    async fn tick(&self) -> ServiceState {
        let manager = Arc::clone(&self.manager);
        let state = tokio::task::spawn_blocking(move || {
            let mut m = manager.lock().unwrap_or_else(|e| e.into_inner());
            m.tick(Instant::now(), crate::core::cloud::now_unix_ms())
        })
        .await
        .unwrap_or_else(|e| ServiceState::Failed(format!("the service manager tick failed: {e}")));
        publish(&state);
        state
    }

    /// The supervised future: tick every 10 s and on a change to a watched file, until
    /// `shutdown` turns true. A fresh future per call.
    pub async fn run(self, mut shutdown: watch::Receiver<bool>) {
        let wake = Arc::new(tokio::sync::Notify::new());
        let files = self
            .manager
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .watched_files();
        let _watcher = watch_files(&files, Arc::clone(&wake));
        loop {
            if *shutdown.borrow() {
                return;
            }
            self.tick().await;
            // The settle sleep runs after the `select!`, never inside an arm: an `.await` there
            // holds the select's output, whose `watch::Ref` is not `Send`, and a supervised
            // future must be.
            let woke = tokio::select! {
                () = tokio::time::sleep(TICK) => false,
                () = wake.notified() => true,
                _ = shutdown.wait_for(|stop| *stop) => return,
            };
            if woke {
                // Let a burst of writes (atomic rename, then a second file) settle.
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
        }
    }
}

/// Watch the parent directories of `files` (non-recursively) and wake the loop on any event in
/// them that names one of `files`. A directory that does not exist is skipped; the 10 s tick
/// covers it.
fn watch_files(
    files: &[PathBuf],
    wake: Arc<tokio::sync::Notify>,
) -> Option<notify::RecommendedWatcher> {
    use notify::Watcher;
    let names: Vec<PathBuf> = files.to_vec();
    let mut watcher = notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
        if let Ok(event) = res {
            if event
                .paths
                .iter()
                .any(|p| names.iter().any(|n| p.file_name() == n.file_name()))
            {
                wake.notify_one();
            }
        }
    })
    .ok()?;
    let mut dirs: Vec<&Path> = files.iter().filter_map(|f| f.parent()).collect();
    dirs.sort();
    dirs.dedup();
    for dir in dirs {
        if dir.is_dir() {
            let _ = watcher.watch(dir, notify::RecursiveMode::NonRecursive);
        }
    }
    Some(watcher)
}

// ---------------------------------------------------------------------------
// Uninstall (C-8 Leg B *Reversal*)
// ---------------------------------------------------------------------------

/// What [`uninstall`] did.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct UninstallOutcome {
    /// The pid of the daemon's own service, stopped.
    pub stopped: Option<u32>,
    /// A process the ledger named that failed the ownership check, left running: pid and binary.
    pub left_running: Option<(u32, PathBuf)>,
    /// The daemon's own service passed the ownership check but could not be stopped: pid, binary
    /// and why. Its ledger is kept, so a later uninstall or daemon can still prove ownership.
    pub stop_failed: Option<(u32, PathBuf, String)>,
    /// The ledger was removed (or there was none). `false` when it was kept for `stop_failed`.
    pub ledger_removed: bool,
    /// An OpenCode process is still running (or the table could not be read): OpenCode Desktop
    /// must be restarted.
    pub opencode_attached: bool,
}

/// Stop the service the daemon started — only after the ledger's ownership check — and remove
/// the ledger (C-8 Leg B *Reversal*).
#[must_use]
pub fn uninstall(ol_dir: &Path) -> UninstallOutcome {
    uninstall_with(
        ol_dir,
        DesktopPaths::for_host().service_json.as_deref(),
        &SystemTable,
        &HttpInfoProbe,
    )
}

/// [`uninstall`] over explicit parts.
pub fn uninstall_with(
    ol_dir: &Path,
    service_json: Option<&Path>,
    table: &dyn ProcessTable,
    probe: &dyn InfoProbe,
) -> UninstallOutcome {
    let mut out = UninstallOutcome::default();
    if let Some(entry) = ledger::read(ol_dir) {
        let fresh = table.read(entry.pid);
        let identity_matches = fresh.as_ref().is_some_and(|p| entry.names(p));
        // When OpenCode's registration names the ledger's pid, its own `/api/info` must confirm
        // it too (D-19); a service that has not registered yet is checked on identity alone.
        let registration = service_json
            .and_then(read_registration)
            .filter(|r| r.pid == entry.pid);
        let registration_ok = match &registration {
            None => true,
            Some(r) => {
                is_loopback_url(&r.url)
                    && probe
                        .probe(r, PROBE_BUDGET)
                        .is_ok_and(|a| a.pid == entry.pid)
            }
        };
        match fresh {
            Some(p) if identity_matches && registration_ok => {
                tracing::info!(
                    pid = p.pid,
                    binary = %p.exe.display(),
                    reason = "uninstall",
                    "stopping OpenCode's service"
                );
                match table.stop(&p, STOP_GRACE) {
                    outcome if outcome.is_gone() => out.stopped = Some(p.pid),
                    process::StopOutcome::Failed(why) => {
                        tracing::warn!(pid = p.pid, error = %why, "could not stop OpenCode's service");
                        out.stop_failed = Some((p.pid, p.exe, why));
                    }
                    // Its identity moved or became unreadable at the re-read: not ours to signal.
                    _ => out.left_running = Some((p.pid, p.exe)),
                }
            }
            Some(p) if identity_matches => out.left_running = Some((p.pid, p.exe)),
            Some(p) if process::is_opencode_exe(&p.exe) => {
                // The pid is reused by another OpenCode process: named, never signalled.
                out.left_running = Some((p.pid, p.exe));
            }
            _ => {}
        }
    }
    // Removed whatever `read` made of it: an absent ledger is success, and a malformed, oversized
    // or unreadable one — whose ownership could not be established, so nothing was signalled —
    // goes too, never reported removed while it stays.
    // An owned service that survived its stop keeps its ledger: without it nothing could later
    // prove the running process is ours.
    out.ledger_removed = out.stop_failed.is_none() && ledger::remove(ol_dir).is_ok();
    out.opencode_attached = process::attached(table, &[]).blocks_replacement();
    out
}

#[cfg(test)]
mod tests;