openlatch-client 0.6.19

//! The Leg B rules against a fake process table and a fake `/api/info` probe, plus the real HTTP
//! probe against an in-process listener.
#![cfg(test)]

use super::*;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};

const INSTANCE: &str = "0123456789abcdef0123456789abcdef";
const PROXY: &str = "http://127.0.0.1:47811";

// ---------------------------------------------------------------------------
// Fakes
// ---------------------------------------------------------------------------

#[derive(Default)]
struct FakeTable {
    procs: Mutex<HashMap<u32, ProcId>>,
    /// Pids that run but whose identity cannot be read.
    unreadable: Mutex<std::collections::HashSet<u32>>,
    unreadable_table: AtomicBool,
    signalled: Mutex<Vec<u32>>,
    spawns: Mutex<Vec<(PathBuf, ServiceEnv)>>,
    next_pid: AtomicU32,
    spawn_fails: AtomicBool,
    /// A pid whose start identity changes between the caller's read and the signal.
    reuse_on_stop: Mutex<Option<u32>>,
    /// Every stop of a verified process fails, leaving it running.
    stop_fails: AtomicBool,
}

impl FakeTable {
    fn new() -> Arc<Self> {
        let t = Self::default();
        t.next_pid.store(1000, Ordering::SeqCst);
        Arc::new(t)
    }

    fn add(&self, pid: u32, exe: &Path) -> ProcId {
        let id = ProcId {
            pid,
            start: format!("start-{pid}"),
            exe: exe.to_path_buf(),
        };
        self.procs.lock().unwrap().insert(pid, id.clone());
        id
    }

    fn kill(&self, pid: u32) {
        self.procs.lock().unwrap().remove(&pid);
    }

    fn alive(&self, pid: u32) -> bool {
        self.procs.lock().unwrap().contains_key(&pid)
            || self.unreadable.lock().unwrap().contains(&pid)
    }

    fn signalled(&self) -> Vec<u32> {
        self.signalled.lock().unwrap().clone()
    }

    fn spawn_count(&self) -> usize {
        self.spawns.lock().unwrap().len()
    }

    fn last_spawn(&self) -> (PathBuf, ServiceEnv) {
        self.spawns
            .lock()
            .unwrap()
            .last()
            .cloned()
            .expect("a spawn")
    }
}

impl ProcessTable for FakeTable {
    fn read(&self, pid: u32) -> Option<ProcId> {
        self.procs.lock().unwrap().get(&pid).cloned()
    }

    fn exists(&self, pid: u32) -> bool {
        self.alive(pid)
    }

    fn opencode_processes(&self) -> Result<Vec<(u32, PathBuf)>, String> {
        if self.unreadable_table.load(Ordering::SeqCst) {
            return Err("denied".into());
        }
        Ok(self
            .procs
            .lock()
            .unwrap()
            .values()
            .filter(|p| process::is_opencode_exe(&p.exe))
            .map(|p| (p.pid, p.exe.clone()))
            .collect())
    }

    fn stop(&self, expected: &ProcId, _grace: Duration) -> process::StopOutcome {
        let mut procs = self.procs.lock().unwrap();
        if *self.reuse_on_stop.lock().unwrap() == Some(expected.pid) {
            if let Some(p) = procs.get_mut(&expected.pid) {
                p.start = "reused".into();
            }
        }
        match procs.get(&expected.pid) {
            None => process::StopOutcome::Gone,
            Some(p) if !p.same_process(expected) => process::StopOutcome::Changed,
            Some(_) if self.stop_fails.load(Ordering::SeqCst) => {
                process::StopOutcome::Failed(format!("pid {} survived SIGKILL", expected.pid))
            }
            Some(_) => {
                procs.remove(&expected.pid);
                self.signalled.lock().unwrap().push(expected.pid);
                process::StopOutcome::Stopped
            }
        }
    }

    fn spawn(&self, cli: &Path, env: &ServiceEnv) -> std::io::Result<ProcId> {
        if self.spawn_fails.load(Ordering::SeqCst) {
            return Err(std::io::Error::other("exec failed"));
        }
        let pid = self.next_pid.fetch_add(1, Ordering::SeqCst);
        self.spawns
            .lock()
            .unwrap()
            .push((cli.to_path_buf(), env.clone()));
        Ok(self.add(pid, cli))
    }
}

/// Answers `/api/info` per registered pid while that pid runs; anything else fails.
struct FakeProbe {
    table: Arc<FakeTable>,
    answers: Mutex<HashMap<u32, InfoAnswer>>,
    calls: AtomicU32,
}

impl FakeProbe {
    fn answer(&self, reg_pid: u32, pid: u32) {
        self.answers.lock().unwrap().insert(
            reg_pid,
            InfoAnswer {
                pid,
                version: Some("2.0.26".into()),
            },
        );
    }

    fn silence(&self, reg_pid: u32) {
        self.answers.lock().unwrap().remove(&reg_pid);
    }
}

impl InfoProbe for FakeProbe {
    fn probe(&self, reg: &Registration, _budget: Duration) -> Result<InfoAnswer, String> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        if !self.table.alive(reg.pid) {
            return Err("connection refused".into());
        }
        self.answers
            .lock()
            .unwrap()
            .get(&reg.pid)
            .cloned()
            .ok_or_else(|| "connection refused".to_string())
    }
}

// ---------------------------------------------------------------------------
// Fixture
// ---------------------------------------------------------------------------

struct Fx {
    _dir: tempfile::TempDir,
    root: PathBuf,
    ol: PathBuf,
    resources: PathBuf,
    staged_root: PathBuf,
    service_json: PathBuf,
    table: Arc<FakeTable>,
    probe: Arc<FakeProbe>,
    ca: Arc<Mutex<Option<String>>>,
    elevated: Arc<AtomicBool>,
}

impl Fx {
    fn new() -> Self {
        let dir = tempfile::tempdir().unwrap();
        let root = std::fs::canonicalize(dir.path()).unwrap();
        let ol = root.join("ol");
        let resources = root.join("app").join("resources");
        let staged_root = root.join("userdata").join("cli");
        let service_json = root.join("xdg-state").join("opencode").join("service.json");
        std::fs::create_dir_all(&ol).unwrap();
        std::fs::create_dir_all(&resources).unwrap();
        std::fs::write(resources.join(exe_name()), b"bundled opencode-cli 2.0.26").unwrap();
        std::fs::write(resources.join("opencode-cli.version"), "2.0.26\n").unwrap();
        let table = FakeTable::new();
        let fx = Self {
            _dir: dir,
            root,
            ol,
            resources,
            staged_root,
            service_json,
            probe: Arc::new(FakeProbe {
                table: Arc::clone(&table),
                answers: Mutex::new(HashMap::new()),
                calls: AtomicU32::new(0),
            }),
            table,
            ca: Arc::new(Mutex::new(Some("aa".repeat(32)))),
            elevated: Arc::new(AtomicBool::new(false)),
        };
        fx.write_env_json();
        fx
    }

    fn paths(&self) -> DesktopPaths {
        DesktopPaths {
            resource_dirs: vec![self.resources.clone()],
            staged_root: Some(self.staged_root.clone()),
            sandboxed: Vec::new(),
            service_json: Some(self.service_json.clone()),
        }
    }

    /// A manager as a freshly started daemon would build it.
    fn manager(&self) -> Manager {
        let mut m = Manager::with_parts(
            self.ol.clone(),
            self.paths(),
            self.table.clone(),
            self.probe.clone(),
        );
        let ca = Arc::clone(&self.ca);
        m.ca_digest = Box::new(move |_| ca.lock().unwrap().clone());
        m.inherited_env = Box::new(|| {
            vec![
                (OsString::from("PATH"), OsString::from("/usr/bin")),
                (
                    OsString::from("OPENLATCH_API_KEY"),
                    OsString::from("secret"),
                ),
                (OsString::from("OPENLATCH_TOKEN"), OsString::from("secret")),
                (
                    OsString::from("ALL_PROXY"),
                    OsString::from("socks5://corp:1"),
                ),
                (
                    OsString::from("all_proxy"),
                    OsString::from("socks5://corp:1"),
                ),
            ]
        });
        let elevated = Arc::clone(&self.elevated);
        m.elevated = Box::new(move || elevated.load(Ordering::SeqCst));
        m
    }

    fn env_json(&self) -> EnvJson {
        EnvJson {
            https_proxy: PROXY.into(),
            http_proxy: PROXY.into(),
            no_proxy: "localhost,127.0.0.1,::1,[::1]".into(),
            node_extra_ca_certs: self
                .ol
                .join("model-relay")
                .join("ca")
                .join("ca.pem")
                .to_string_lossy()
                .into_owned(),
            instance_id: INSTANCE.into(),
        }
    }

    fn write_env_json(&self) {
        let path = proxy_env_file::env_json_path(&self.ol);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, serde_json::to_string(&self.env_json()).unwrap()).unwrap();
    }

    fn withdraw_env_json(&self) {
        std::fs::remove_file(proxy_env_file::env_json_path(&self.ol)).unwrap();
    }

    fn register(&self, pid: u32) {
        std::fs::create_dir_all(self.service_json.parent().unwrap()).unwrap();
        std::fs::write(
            &self.service_json,
            format!(
                r#"{{"url":"http://127.0.0.1:4096","pid":{pid},"version":"2.0.26","password":"fixture-pw","id":"x"}}"#
            ),
        )
        .unwrap();
    }

    /// A service Desktop spawned from its staged copy, registered and answering.
    fn foreign_service(&self, pid: u32) -> ProcId {
        let dir = self.staged_root.join("2.0.26");
        std::fs::create_dir_all(&dir).unwrap();
        let exe = dir.join(exe_name());
        std::fs::write(&exe, b"bundled opencode-cli 2.0.26").unwrap();
        let id = self.table.add(pid, &exe);
        self.register(pid);
        self.probe.answer(pid, pid);
        id
    }

    /// Spawn the daemon's own service and let it register: returns its pid.
    fn armed_own(&self, m: &mut Manager, now: Instant) -> u32 {
        assert_eq!(m.tick(now, 1_000), ServiceState::Starting);
        let pid = ledger::read(&self.ol).expect("ledger written").pid;
        self.register(pid);
        self.probe.answer(pid, pid);
        assert_eq!(m.tick(now, 2_000), ServiceState::Armed);
        pid
    }

    fn files(&self) -> Vec<PathBuf> {
        fn walk(dir: &Path, out: &mut Vec<PathBuf>) {
            for e in std::fs::read_dir(dir).unwrap().flatten() {
                let p = e.path();
                if p.is_dir() {
                    walk(&p, out);
                } else {
                    out.push(p);
                }
            }
        }
        let mut out = Vec::new();
        walk(&self.root, &mut out);
        out.sort();
        out
    }
}

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

#[test]
fn no_bundled_cli_is_absent_and_a_sandboxed_desktop_is_unsupported() {
    let fx = Fx::new();
    std::fs::remove_file(fx.resources.join(exe_name())).unwrap();
    let mut m = fx.manager();
    assert_eq!(m.tick(Instant::now(), 0), ServiceState::Absent);
    assert_eq!(fx.table.spawn_count(), 0);

    let flatpak = fx
        .root
        .join("flatpak")
        .join("app")
        .join("ai.opencode.desktop");
    std::fs::create_dir_all(&flatpak).unwrap();
    m.paths.sandboxed = vec![(flatpak, "flatpak")];
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unsupported("flatpak".into())
    );
    assert_eq!(fx.table.spawn_count(), 0);
}

#[test]
fn a_bundled_cli_without_a_valid_version_file_is_not_detected() {
    let fx = Fx::new();
    std::fs::write(
        fx.resources.join("opencode-cli.version"),
        "2.0.26; rm -rf /",
    )
    .unwrap();
    assert_eq!(detect_in(&fx.paths()), Detection::Absent);
    std::fs::write(fx.resources.join("opencode-cli.version"), " 2.0.27 \n").unwrap();
    assert_eq!(
        detect_in(&fx.paths()),
        Detection::Found(BundledCli {
            path: fx.resources.join(exe_name()),
            version: "2.0.27".into()
        })
    );
}

#[test]
fn the_staged_copy_runs_only_when_its_digest_equals_the_bundled_one() {
    let fx = Fx::new();
    let cli = BundledCli {
        path: fx.resources.join(exe_name()),
        version: "2.0.26".into(),
    };
    let mut digests = DigestCache::default();
    let (exe, _) = resolve_executable(&cli, Some(&fx.staged_root), &mut digests).unwrap();
    assert_eq!(exe, cli.path, "no staged copy: the bundled one");

    let staged = fx.staged_root.join("2.0.26").join(exe_name());
    std::fs::create_dir_all(staged.parent().unwrap()).unwrap();
    std::fs::write(&staged, b"tampered").unwrap();
    let (exe, _) = resolve_executable(&cli, Some(&fx.staged_root), &mut digests).unwrap();
    assert_eq!(exe, cli.path, "a different staged copy is never run");

    std::fs::write(&staged, b"bundled opencode-cli 2.0.26").unwrap();
    let (exe, digest) = resolve_executable(&cli, Some(&fx.staged_root), &mut digests).unwrap();
    assert_eq!(exe, staged);
    assert_eq!(digest.len(), 64);
}

#[test]
fn the_service_json_path_follows_xdg_state_home_else_local_state() {
    let home = PathBuf::from("home");
    assert_eq!(
        service_json_path(None, Some(&home)),
        Some(
            home.join(".local")
                .join("state")
                .join("opencode")
                .join("service.json")
        )
    );
    let xdg = std::env::temp_dir().join("xdg-state");
    assert_eq!(
        service_json_path(Some(xdg.clone().into_os_string()), Some(&home)),
        Some(xdg.join("opencode").join("service.json"))
    );
    assert_eq!(service_json_path(Some(OsString::new()), None), None);
}

// ---------------------------------------------------------------------------
// Start (AC-26, D-16, D-21) and AC-19
// ---------------------------------------------------------------------------

#[test]
fn with_no_live_service_it_starts_one_with_exactly_the_relay_keys_and_writes_only_its_ledger() {
    let fx = Fx::new();
    let before = fx.files();
    let mut m = fx.manager();
    assert_eq!(m.tick(Instant::now(), 1_234), ServiceState::Starting);

    let (exe, env) = fx.table.last_spawn();
    assert_eq!(exe, fx.resources.join(exe_name()));
    let added: Vec<&str> = env.add.iter().map(|(k, _)| *k).collect();
    assert_eq!(added, process::RELAY_KEYS.to_vec());
    assert!(env.add.contains(&("HTTPS_PROXY", PROXY.to_string())));
    for k in [
        "OPENLATCH_API_KEY",
        "OPENLATCH_TOKEN",
        "ALL_PROXY",
        "all_proxy",
    ] {
        assert!(env.remove.contains(&OsString::from(k)), "{k} stripped");
    }
    assert!(!env.remove.contains(&OsString::from("PATH")));

    let l = ledger::read(&fx.ol).expect("ledger");
    assert_eq!(l.version, "2.0.26");
    assert_eq!(l.proxy_url, PROXY);
    assert_eq!(l.ca_sha256, "aa".repeat(32));
    assert_eq!(l.started_at, 1_234);
    assert_eq!(l.sha256.len(), 64);
    assert_eq!(l.binary, exe);

    // AC-19 / PB-2: the ledger is the only file Leg B wrote.
    let after = fx.files();
    let new: Vec<&PathBuf> = after.iter().filter(|p| !before.contains(p)).collect();
    assert_eq!(new, vec![&ledger::ledger_path(&fx.ol)]);
}

#[test]
fn its_own_registered_service_with_current_keys_is_armed() {
    let fx = Fx::new();
    let mut m = fx.manager();
    let now = Instant::now();
    let pid = fx.armed_own(&mut m, now);
    assert!(fx.table.alive(pid));
    assert!(fx.table.signalled().is_empty());
    assert_eq!(fx.table.spawn_count(), 1);
}

/// D-19: a registration naming the daemon's own pid means its service registered, so a failed
/// `/api/info` probe of it is an unverified identity — nothing stopped, nothing started — never
/// "not registered yet", whose limit would end in a stop.
#[test]
fn its_own_registered_service_failing_the_probe_is_unverified_never_replaced() {
    let fx = Fx::new();
    let mut m = fx.manager();
    let now = Instant::now();
    assert_eq!(m.tick(now, 1_000), ServiceState::Starting);
    let pid = ledger::read(&fx.ol).unwrap().pid;
    // Registered, but its probe fails.
    fx.register(pid);
    assert_eq!(
        m.tick(now, 2_000),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert_eq!(fx.table.spawn_count(), 1, "no second service");
    assert!(fx.table.signalled().is_empty());
}

/// AC-26 / D-19: an aged service of the daemon's own, a Desktop attached, and one probe timeout
/// during a Desktop turn: nothing is stopped — neither by the identity rule nor by the
/// never-registered limit, which the attached rule also holds back.
#[test]
fn an_aged_own_service_with_desktop_attached_survives_a_failed_probe() {
    let late = u64::try_from(STARTING_LIMIT.as_millis()).unwrap() + 1_001;
    let now = Instant::now();
    // Registered, then its probe fails once.
    let fx = Fx::new();
    let mut m = fx.manager();
    let own = fx.armed_own(&mut m, now);
    fx.table.add(600, &fx.root.join("OpenCode"));
    fx.probe.silence(own);
    assert_eq!(
        m.tick(now, late),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty());
    assert!(fx.table.alive(own));
    assert_eq!(fx.table.spawn_count(), 1);
    assert!(ledger::read(&fx.ol).is_some());

    // Never registered past the limit, with Desktop attached: held back by the attached rule.
    let fx = Fx::new();
    let mut m = fx.manager();
    assert_eq!(m.tick(now, 0), ServiceState::Starting);
    let pid = ledger::read(&fx.ol).unwrap().pid;
    fx.table.add(600, &fx.root.join("OpenCode"));
    assert_eq!(
        m.tick(now, late),
        ServiceState::Unarmed(UnarmedCause::AwaitingRespawn)
    );
    assert!(fx.table.signalled().is_empty());
    assert!(fx.table.alive(pid));
    // And an unreadable table holds it back too, as an unverified identity.
    fx.table.kill(600);
    fx.table.unreadable_table.store(true, Ordering::SeqCst);
    assert_eq!(
        m.tick(now, late),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty());
}

/// D-19: the daemon's own pid still runs, but its identity cannot be read — and its probe fails.
/// The ledger is kept, nothing is signalled, and no second service is started.
#[test]
fn an_unreadable_own_identity_keeps_the_ledger_and_starts_nothing() {
    let fx = Fx::new();
    let mut m = fx.manager();
    let now = Instant::now();
    let own = fx.armed_own(&mut m, now);
    fx.table.kill(own);
    fx.table.unreadable.lock().unwrap().insert(own);
    fx.probe.silence(own);
    assert_eq!(
        m.tick(now, 3_000),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert_eq!(
        ledger::read(&fx.ol).map(|l| l.pid),
        Some(own),
        "ledger kept"
    );
    assert_eq!(fx.table.spawn_count(), 1, "no duplicate service");
    assert!(fx.table.signalled().is_empty());

    // Once the pid is confirmed gone, the ledger is stale and goes.
    fx.table.unreadable.lock().unwrap().remove(&own);
    let _ = m.tick(now, 4_000);
    assert_ne!(ledger::read(&fx.ol).map(|l| l.pid), Some(own));
}

#[test]
fn a_service_that_never_registers_is_stopped_and_counted_after_the_limit() {
    let fx = Fx::new();
    let mut m = fx.manager();
    let now = Instant::now();
    assert_eq!(m.tick(now, 0), ServiceState::Starting);
    let pid = ledger::read(&fx.ol).unwrap().pid;
    let late = u64::try_from(STARTING_LIMIT.as_millis()).unwrap() + 1;
    // Stopped, counted as an exit; the backoff holds the respawn.
    assert_eq!(
        m.tick(now, late),
        ServiceState::Unarmed(UnarmedCause::AwaitingRespawn)
    );
    assert_eq!(fx.table.signalled(), vec![pid]);
    assert!(ledger::read(&fx.ol).is_none());
    assert_eq!(m.tick(now + BACKOFF_INITIAL, late), ServiceState::Starting);
}

#[test]
fn an_elevated_daemon_never_starts_the_service() {
    let fx = Fx::new();
    fx.elevated.store(true, Ordering::SeqCst);
    let mut m = fx.manager();
    assert!(matches!(m.tick(Instant::now(), 0), ServiceState::Failed(_)));
    assert_eq!(fx.table.spawn_count(), 0);
    assert!(ledger::read(&fx.ol).is_none());
}

#[test]
fn a_failed_spawn_is_reported_failed() {
    let fx = Fx::new();
    fx.table.spawn_fails.store(true, Ordering::SeqCst);
    let mut m = fx.manager();
    let state = m.tick(Instant::now(), 0);
    assert!(
        matches!(&state, ServiceState::Failed(r) if r.contains("exec failed")),
        "{state:?}"
    );
    assert!(ledger::read(&fx.ol).is_none());
}

// ---------------------------------------------------------------------------
// Foreign service (AC-26, D-19)
// ---------------------------------------------------------------------------

#[test]
fn a_foreign_service_is_left_alone_while_desktop_a_cli_or_a_tui_is_attached() {
    for attached_exe in [
        "OpenCode",
        "opencode",
        "opencode.exe",
        "ai.opencode.desktop",
    ] {
        let fx = Fx::new();
        fx.foreign_service(500);
        fx.table.add(600, &fx.root.join(attached_exe));
        let mut m = fx.manager();
        assert_eq!(
            m.tick(Instant::now(), 0),
            ServiceState::Unarmed(UnarmedCause::ForeignAttached),
            "{attached_exe}"
        );
        assert!(
            fx.table.signalled().is_empty(),
            "{attached_exe}: nothing stopped"
        );
        assert_eq!(fx.table.spawn_count(), 0, "{attached_exe}: nothing started");
        assert!(fx.table.alive(500));
    }
}

/// D-19: an unreadable process table counts as attached (fail closed), and names its own cause:
/// `identity_unverified`, never `foreign_attached`.
#[test]
fn an_unreadable_process_table_counts_as_attached() {
    let fx = Fx::new();
    fx.foreign_service(500);
    fx.table.unreadable_table.store(true, Ordering::SeqCst);
    let mut m = fx.manager();
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty());
}

#[test]
fn a_foreign_service_with_none_attached_is_replaced() {
    let fx = Fx::new();
    fx.foreign_service(500);
    let mut m = fx.manager();
    assert_eq!(m.tick(Instant::now(), 0), ServiceState::Starting);
    assert_eq!(fx.table.signalled(), vec![500]);
    assert_eq!(fx.table.spawn_count(), 1);
    assert!(ledger::read(&fx.ol).is_some());
}

#[test]
fn a_foreign_service_is_not_stopped_when_the_daemon_could_not_start_its_own() {
    let fx = Fx::new();
    fx.foreign_service(500);
    fx.withdraw_env_json();
    let mut m = fx.manager();
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::ForeignAttached)
    );
    assert!(fx.table.signalled().is_empty());
}

#[test]
fn a_stale_registration_whose_pid_a_live_tui_now_holds_is_identity_unverified() {
    let fx = Fx::new();
    // service.json names 700, which is now a developer's `opencode` TUI; the TUI does not
    // answer `/api/info` with that pid.
    fx.table.add(700, &fx.root.join("bin").join("opencode"));
    fx.register(700);
    let mut m = fx.manager();
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty(), "nothing stopped");
    assert_eq!(fx.table.spawn_count(), 0, "nothing started");
    assert!(fx.table.alive(700));

    // A probe answering another pid is no better.
    fx.probe.answer(700, 701);
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty());
}

#[test]
fn a_confirmed_registration_whose_identity_is_unreadable_is_never_signalled() {
    let fx = Fx::new();
    fx.table.unreadable.lock().unwrap().insert(800);
    fx.register(800);
    fx.probe.answer(800, 800);
    // `/api/info` confirms 800, but the table cannot read its identity.
    let mut m = fx.manager();
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty());
    assert_eq!(fx.table.spawn_count(), 0);
}

#[test]
fn a_pid_reused_between_the_read_and_the_signal_is_never_signalled() {
    let fx = Fx::new();
    fx.foreign_service(500);
    *fx.table.reuse_on_stop.lock().unwrap() = Some(500);
    let mut m = fx.manager();
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty());
    assert_eq!(fx.table.spawn_count(), 0);
}

#[test]
fn a_registered_service_outside_the_detect_paths_is_never_signalled() {
    let fx = Fx::new();
    let elsewhere = fx.root.join("elsewhere").join(exe_name());
    std::fs::create_dir_all(elsewhere.parent().unwrap()).unwrap();
    std::fs::write(&elsewhere, b"x").unwrap();
    fx.table.add(500, &elsewhere);
    fx.register(500);
    fx.probe.answer(500, 500);
    let mut m = fx.manager();
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert!(fx.table.signalled().is_empty());
}

#[test]
fn a_non_loopback_registration_is_never_probed() {
    let fx = Fx::new();
    fx.foreign_service(500);
    std::fs::write(
        &fx.service_json,
        r#"{"url":"http://10.0.0.5:4096","pid":500,"password":"fixture-pw"}"#,
    )
    .unwrap();
    let mut m = fx.manager();
    assert_eq!(
        m.tick(Instant::now(), 0),
        ServiceState::Unarmed(UnarmedCause::IdentityUnverified)
    );
    assert_eq!(fx.probe.calls.load(Ordering::SeqCst), 0, "never probed");
    assert!(fx.table.alive(500));
}

#[test]
fn a_stale_registration_whose_pid_is_gone_is_no_live_service() {
    let fx = Fx::new();
    fx.register(900);
    let mut m = fx.manager();
    assert_eq!(m.tick(Instant::now(), 0), ServiceState::Starting);
    assert_eq!(fx.table.spawn_count(), 1);
}

#[test]
fn a_stale_ledger_is_deleted_and_never_acted_on() {
    let fx = Fx::new();
    let stale = ledger::Ledger {
        pid: 4242,
        start: "old".into(),
        binary: fx.resources.join(exe_name()),
        sha256: "00".repeat(32),
        version: "2.0.26".into(),
        started_at: 0,
        proxy_url: PROXY.into(),
        ca_sha256: "aa".repeat(32),
    };
    ledger::write(&fx.ol, &stale).unwrap();
    // The pid now names another process.
    fx.table.add(4242, &fx.root.join("bash"));
    let mut m = fx.manager();
    assert_eq!(m.tick(Instant::now(), 0), ServiceState::Starting);
    assert!(fx.table.alive(4242), "the reused pid is never signalled");
    assert_ne!(ledger::read(&fx.ol).unwrap().pid, 4242);
}

// ---------------------------------------------------------------------------
// Trust release, CA rotation, update (D-18, D-21)
// ---------------------------------------------------------------------------

#[test]
fn a_ca_rotation_across_a_daemon_restart_is_read_from_the_ledger() {
    let fx = Fx::new();
    let now = Instant::now();
    let old = {
        let mut first = fx.manager();
        fx.armed_own(&mut first, now)
    };
    // `doctor --fix` rotated the CA in place while the daemon was stopped; a new daemon starts.
    *fx.ca.lock().unwrap() = Some("bb".repeat(32));
    fx.table.add(600, &fx.root.join("OpenCode"));
    let mut second = fx.manager();
    assert_eq!(
        second.tick(now, 3_000),
        ServiceState::Unarmed(UnarmedCause::AwaitingRespawn)
    );
    assert!(fx.table.alive(old), "left alone while Desktop is attached");

    fx.table.kill(600);
    assert_eq!(second.tick(now, 4_000), ServiceState::Starting);
    assert_eq!(fx.table.signalled(), vec![old]);
    let l = ledger::read(&fx.ol).unwrap();
    assert_ne!(l.pid, old);
    assert_eq!(l.ca_sha256, "bb".repeat(32));
}

#[test]
fn withdrawing_env_json_releases_trust_and_the_respawn_waits_for_it() {
    let fx = Fx::new();
    let now = Instant::now();
    let mut m = fx.manager();
    let own = fx.armed_own(&mut m, now);

    fx.withdraw_env_json();
    assert_eq!(
        m.tick(now, 3_000),
        ServiceState::Unarmed(UnarmedCause::AwaitingRespawn)
    );
    assert_eq!(
        fx.table.signalled(),
        vec![own],
        "the stale service is stopped"
    );
    assert_eq!(fx.table.spawn_count(), 1, "no respawn without the keys");
    assert!(ledger::read(&fx.ol).is_none());

    fx.write_env_json();
    assert_eq!(m.tick(now, 4_000), ServiceState::Starting);
    assert_eq!(fx.table.spawn_count(), 2);
}

#[test]
fn a_bundled_version_change_restarts_on_the_new_binary() {
    let fx = Fx::new();
    let now = Instant::now();
    let mut m = fx.manager();
    let own = fx.armed_own(&mut m, now);

    std::fs::write(fx.resources.join("opencode-cli.version"), "2.0.27").unwrap();
    std::fs::write(
        fx.resources.join(exe_name()),
        b"bundled opencode-cli 2.0.27",
    )
    .unwrap();
    fx.table.add(600, &fx.root.join("OpenCode"));
    assert_eq!(
        m.tick(now, 3_000),
        ServiceState::Unarmed(UnarmedCause::UpdateWindow)
    );
    assert!(fx.table.alive(own));

    fx.table.kill(600);
    assert_eq!(m.tick(now, 4_000), ServiceState::Starting);
    assert_eq!(fx.table.signalled(), vec![own]);
    assert_eq!(ledger::read(&fx.ol).unwrap().version, "2.0.27");
}

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

#[test]
fn a_service_that_keeps_exiting_is_restarted_with_backoff_then_failed() {
    let fx = Fx::new();
    let mut m = fx.manager();
    let mut now = Instant::now();
    let mut backoff = BACKOFF_INITIAL;
    assert_eq!(m.tick(now, 0), ServiceState::Starting);
    // The first start is charged too: four restarts fill the hour's five starts.
    for exit in 1..=MAX_STARTS_PER_HOUR {
        let pid = ledger::read(&fx.ol).unwrap().pid;
        fx.table.kill(pid);
        let state = m.tick(now, 0);
        if exit < MAX_STARTS_PER_HOUR {
            assert_eq!(
                state,
                ServiceState::Unarmed(UnarmedCause::AwaitingRespawn),
                "exit {exit}: held by the backoff"
            );
            now += backoff;
            backoff = (backoff * 2).min(BACKOFF_MAX);
            assert_eq!(
                m.tick(now, 0),
                ServiceState::Starting,
                "exit {exit}: restarted"
            );
        } else {
            assert!(
                matches!(state, ServiceState::Failed(_)),
                "exit {exit}: {state:?}"
            );
        }
    }
    let spawned = fx.table.spawn_count();
    now += BACKOFF_MAX;
    assert!(
        matches!(m.tick(now, 0), ServiceState::Failed(_)),
        "stays failed"
    );
    assert_eq!(fx.table.spawn_count(), spawned);

    // A new bundled version lifts the bound.
    std::fs::write(fx.resources.join("opencode-cli.version"), "2.0.27").unwrap();
    assert_eq!(m.tick(now, 0), ServiceState::Starting);
    assert_eq!(fx.table.spawn_count(), spawned + 1);
}

#[test]
fn starts_older_than_an_hour_leave_the_window() {
    let mut bound = RestartBound::default();
    let t0 = Instant::now();
    for i in 0..u32::try_from(MAX_STARTS_PER_HOUR).unwrap() {
        bound.note_start(t0 + Duration::from_secs(u64::from(i)));
    }
    assert_eq!(
        bound.permit(t0 + Duration::from_secs(10), "v"),
        Permit::Exhausted
    );
    let mut bound = RestartBound::default();
    for i in 0..4u32 {
        bound.note_start(t0 + Duration::from_secs(u64::from(i)));
    }
    let later = t0 + BOUND_WINDOW + Duration::from_secs(10);
    assert_eq!(bound.permit(later, "v"), Permit::Go);
    bound.note_start(later);
    assert_eq!(bound.starts.len(), 1);
}

/// D-18 bounds starts, not exits: replacing the daemon's own service after each CA change — a
/// stop and a spawn with no exit in between — is charged too, so a sixth start in the hour fails.
#[test]
fn replacements_after_ca_changes_are_bounded_per_hour() {
    let fx = Fx::new();
    let mut m = fx.manager();
    let now = Instant::now();
    fx.armed_own(&mut m, now);
    for round in 1..MAX_STARTS_PER_HOUR {
        *fx.ca.lock().unwrap() = Some(format!("{round:064x}"));
        assert_eq!(m.tick(now, 3_000), ServiceState::Starting, "round {round}");
        let pid = ledger::read(&fx.ol).unwrap().pid;
        fx.register(pid);
        fx.probe.answer(pid, pid);
        assert_eq!(m.tick(now, 4_000), ServiceState::Armed, "round {round}");
    }
    assert_eq!(fx.table.spawn_count(), MAX_STARTS_PER_HOUR);
    *fx.ca.lock().unwrap() = Some("ff".repeat(32));
    let state = m.tick(now, 5_000);
    assert!(matches!(state, ServiceState::Failed(_)), "{state:?}");
    assert_eq!(
        fx.table.spawn_count(),
        MAX_STARTS_PER_HOUR,
        "no sixth start"
    );
}

// ---------------------------------------------------------------------------
// Uninstall (AC-18 local half, with T07)
// ---------------------------------------------------------------------------

#[test]
fn uninstall_stops_only_the_owned_service_and_removes_the_ledger() {
    let fx = Fx::new();
    let now = Instant::now();
    let mut m = fx.manager();
    let own = fx.armed_own(&mut m, now);
    let out = uninstall_with(&fx.ol, Some(&fx.service_json), &*fx.table, &*fx.probe);
    assert_eq!(out.stopped, Some(own));
    assert_eq!(out.left_running, None);
    assert!(out.ledger_removed);
    assert!(!out.opencode_attached);
    assert!(ledger::read(&fx.ol).is_none());
}

#[test]
fn uninstall_names_and_leaves_a_process_that_fails_the_ownership_check() {
    let fx = Fx::new();
    let now = Instant::now();
    let mut m = fx.manager();
    let own = fx.armed_own(&mut m, now);
    // The registration's `/api/info` no longer confirms the pid.
    fx.probe.silence(own);
    fx.table.add(600, &fx.root.join("OpenCode"));
    let out = uninstall_with(&fx.ol, Some(&fx.service_json), &*fx.table, &*fx.probe);
    assert_eq!(out.stopped, None);
    assert_eq!(out.left_running.map(|(p, _)| p), Some(own));
    assert!(fx.table.alive(own));
    assert!(out.ledger_removed);
    assert!(out.opencode_attached, "Desktop must be restarted");
}

/// An owned service whose stop fails is reported as such, never as a failed ownership check,
/// and its ledger is kept so a later uninstall or daemon can still prove it is ours.
#[test]
fn uninstall_keeps_the_ledger_of_an_owned_service_it_could_not_stop() {
    let fx = Fx::new();
    let now = Instant::now();
    let mut m = fx.manager();
    let own = fx.armed_own(&mut m, now);
    fx.table.stop_fails.store(true, Ordering::SeqCst);
    let out = uninstall_with(&fx.ol, Some(&fx.service_json), &*fx.table, &*fx.probe);
    assert_eq!(out.stopped, None);
    assert_eq!(out.left_running, None);
    assert_eq!(out.stop_failed.as_ref().map(|(p, _, _)| *p), Some(own));
    assert!(fx.table.alive(own));
    assert!(!out.ledger_removed);
    assert_eq!(
        ledger::read(&fx.ol).map(|e| e.pid),
        Some(own),
        "the ledger is kept"
    );
}

/// AC-18: a malformed or oversized ledger names no process, so nothing is signalled — and it is
/// removed, never reported removed while it stays.
#[test]
fn uninstall_removes_a_malformed_or_oversized_ledger() {
    for body in ["{not a ledger".to_string(), " ".repeat(20 * 1024)] {
        let fx = Fx::new();
        fx.foreign_service(500);
        let path = ledger::ledger_path(&fx.ol);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, &body).unwrap();
        let out = uninstall_with(&fx.ol, Some(&fx.service_json), &*fx.table, &*fx.probe);
        assert!(out.ledger_removed);
        assert!(!path.exists(), "the bad ledger is gone");
        assert!(fx.table.signalled().is_empty());
        assert!(fx.table.alive(500));
    }
}

#[test]
fn uninstall_with_no_ledger_touches_nothing() {
    let fx = Fx::new();
    fx.foreign_service(500);
    let out = uninstall_with(&fx.ol, Some(&fx.service_json), &*fx.table, &*fx.probe);
    assert_eq!(out.stopped, None);
    assert!(fx.table.signalled().is_empty());
    assert!(fx.table.alive(500));
}

// ---------------------------------------------------------------------------
// service.json and the `/api/info` probe (D-19)
// ---------------------------------------------------------------------------

#[test]
fn the_registration_is_capped_and_its_password_never_printed() {
    let fx = Fx::new();
    fx.register(42);
    let reg = read_registration(&fx.service_json).unwrap();
    assert_eq!(reg.pid, 42);
    let printed = format!("{reg:?}");
    assert!(!printed.contains("fixture-pw"), "{printed}");
    assert!(printed.contains("<redacted>"));

    std::fs::write(&fx.service_json, " ".repeat(70 * 1024)).unwrap();
    assert!(read_registration(&fx.service_json).is_none());
    std::fs::write(&fx.service_json, r#"{"url":"http://127.0.0.1:1","pid":0}"#).unwrap();
    assert!(
        read_registration(&fx.service_json).is_none(),
        "pid 0 rejected"
    );
}

#[test]
fn only_loopback_http_urls_are_probed() {
    for ok in [
        "http://127.0.0.1:4096",
        "http://127.0.0.1:4096/",
        "http://[::1]:4096",
        "http://localhost:4096",
    ] {
        assert!(is_loopback_url(ok), "{ok}");
    }
    for bad in [
        "https://127.0.0.1:4096",
        "http://10.0.0.1:4096",
        "http://user@127.0.0.1:4096",
        "http://127.0.0.1:4096/api",
        "http://127.0.0.1",
        "http://127.0.0.1:0",
        "http://127.0.0.1.evil.com:4096",
    ] {
        assert!(!is_loopback_url(bad), "{bad}");
    }
}

/// Serve one HTTP answer after reading the whole request head, and return what was read
/// (cross-platform rule: a test server reads the whole request before it answers).
fn serve_once(answer: &'static [u8]) -> (String, std::thread::JoinHandle<String>) {
    use std::io::{Read, Write};
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let handle = std::thread::spawn(move || {
        let (mut s, _) = listener.accept().unwrap();
        s.set_nonblocking(false).unwrap();
        let mut got = Vec::new();
        let mut buf = [0u8; 1024];
        while !got.windows(4).any(|w| w == b"\r\n\r\n") {
            let n = s.read(&mut buf).unwrap();
            if n == 0 {
                break;
            }
            got.extend_from_slice(&buf[..n]);
        }
        s.write_all(answer).unwrap();
        String::from_utf8_lossy(&got).into_owned()
    });
    (url, handle)
}

fn reg_at(url: &str, pid: u32) -> Registration {
    Registration {
        url: url.into(),
        pid,
        version: None,
        password: Some(ServicePassword("fixture-pw".into())),
    }
}

#[test]
fn the_probe_authenticates_as_opencode_and_reads_the_pid() {
    let (url, server) = serve_once(
        b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 34\r\n\r\n{\"pid\":4242,\"version\":\"2.0.26\"}   ",
    );
    let answer = HttpInfoProbe
        .probe(&reg_at(&url, 4242), Duration::from_secs(5))
        .unwrap();
    assert_eq!(
        answer,
        InfoAnswer {
            pid: 4242,
            version: Some("2.0.26".into())
        }
    );
    let request = server.join().unwrap();
    assert!(
        request.starts_with("GET /api/info HTTP/1.1\r\n"),
        "{request}"
    );
    // base64("opencode:fixture-pw")
    assert!(
        request.contains("Authorization: Basic b3BlbmNvZGU6Zml4dHVyZS1wdw==\r\n"),
        "{request}"
    );
}

#[test]
fn the_probe_reads_a_chunked_answer() {
    let (url, server) = serve_once(
        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\nb\r\n{\"pid\":7}  \r\n0\r\n\r\n",
    );
    let answer = HttpInfoProbe
        .probe(&reg_at(&url, 7), Duration::from_secs(5))
        .unwrap();
    assert_eq!(answer.pid, 7);
    server.join().unwrap();
}

/// A chunk size at the top of `usize` (or past it) is malformed, never an overflow or a panic —
/// the synchronous uninstall probe has no panic guard.
#[test]
fn a_maximum_size_chunk_header_is_malformed_not_a_panic() {
    for body in [
        &b"ffffffffffffffff\r\nab\r\n0\r\n\r\n"[..],
        &b"fffffffffffffffe\r\nab\r\n0\r\n\r\n"[..],
        &b"10000000000000000\r\nab\r\n0\r\n\r\n"[..],
        &b"5\r\nab\r\n0\r\n\r\n"[..],
    ] {
        assert_eq!(dechunk(body), None, "{}", String::from_utf8_lossy(body));
    }
    assert_eq!(dechunk(b"2\r\nab\r\n0\r\n\r\n"), Some(b"ab".to_vec()));
}

#[test]
fn a_failed_probe_never_carries_the_password() {
    let (url, server) = serve_once(b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n");
    let err = HttpInfoProbe
        .probe(&reg_at(&url, 1), Duration::from_secs(5))
        .unwrap_err();
    assert!(err.contains("401"), "{err}");
    assert!(!err.contains("fixture-pw"));
    server.join().unwrap();

    // Nothing listening.
    let closed = {
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        format!("http://{}", l.local_addr().unwrap())
    };
    let err = HttpInfoProbe
        .probe(&reg_at(&closed, 1), Duration::from_millis(500))
        .unwrap_err();
    assert!(!err.contains("fixture-pw"));
}

#[test]
fn the_published_state_is_readable() {
    // The only publisher in this test binary: ticks above are driven directly, not through
    // `DesktopService`.
    let rx = subscribe();
    publish(&ServiceState::Unarmed(UnarmedCause::UpdateWindow));
    assert_eq!(
        current_state(),
        ServiceState::Unarmed(UnarmedCause::UpdateWindow)
    );
    assert_eq!(
        *rx.borrow(),
        ServiceState::Unarmed(UnarmedCause::UpdateWindow)
    );
    assert_eq!(UnarmedCause::ForeignAttached.as_str(), "foreign_attached");
    assert_eq!(
        UnarmedCause::IdentityUnverified.as_str(),
        "identity_unverified"
    );
    assert_eq!(UnarmedCause::UpdateWindow.as_str(), "update_window");
    assert_eq!(UnarmedCause::AwaitingRespawn.as_str(), "awaiting_respawn");
    assert_eq!(task_spec().name, TASK_NAME);
    assert_eq!(task_spec().restart, RestartPolicy::Always);
}