openlatch-client 0.6.3

//! The per-user LOGIN environment: what a Dock-, Finder-, Explorer- or desktop-launched process inherits, which no
//! shell profile reaches. PRD D-20 admits it for EXACTLY two names; this module refuses any other ([`ALLOWED`]).
//!
//! | OS      | apply                                                         | clear                                  | read                          |
//! | ------- | ------------------------------------------------------------- | -------------------------------------- | ----------------------------- |
//! | macOS   | `launchctl setenv K V`                                        | `launchctl unsetenv K`                 | `launchctl getenv K`          |
//! | Windows | HKCU\Environment REG_SZ K=V + `SendMessageTimeoutW(HWND_BROADCAST, WM_SETTINGCHANGE, 0, "Environment", SMTO_ABORTIFHUNG, 5000)` | `RegDeleteKeyValueW` + broadcast | `RegGetValueW` |
//! | Linux   | `~/.config/environment.d/60-openlatch-cline.conf` (K=V lines) + `systemctl --user set-environment K=V` + `dbus-update-activation-environment --systemd K=V` | remove our lines/file + `systemctl --user unset-environment K` + dbus `K=` | `systemctl --user show-environment` (fallback: our file) |
//!
//! Reaches processes launched AFTER the call; a running app keeps its environment (VS Code must restart — D-10).
//! MACHINE-GLOBAL: callers check `owns_wiring_for && supervision::owns_machine_supervision()`; an isolated instance
//! never calls in. External commands are best-effort within a 5 s budget; a failure lands in the [`Report`].
//! Linux refuses a value environment.d(5) cannot quote (doctor: `probe-failed: unsupported path characters`).
//! Only the daemon's delivery task and `uninstall` call [`os_sink`]; unit tests pass a recording sink.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

/// Compiled on every OS: [`value_is_deliverable`] uses its pure value check.
mod linux;
#[cfg(target_os = "macos")]
mod macos;
#[cfg(windows)]
mod windows;

/// This OS's implementation: the one place the platform is chosen.
#[cfg(not(any(target_os = "macos", windows)))]
use linux::session as imp;
#[cfg(target_os = "macos")]
use macos as imp;
#[cfg(windows)]
use windows as imp;

/// The only two names this module ever reads, writes or clears (PRD D-20).
pub const ALLOWED: [&str; 2] = ["CLINE_WRAPPER_PATH", "CLINE_JS_RUNTIME_PATH"];

/// The budget of every external command a sink runs (Windows makes none).
#[cfg(not(windows))]
pub(crate) const COMMAND_BUDGET: std::time::Duration = std::time::Duration::from_secs(5);

/// The ledger's file name under `<ol_dir>/state/`.
pub const LEDGER_FILE: &str = "cline-login-env.json";

/// One OS's login environment. `read` answers `Ok(None)` for an unset name.
pub trait Sink: Send + Sync {
    fn read(&self, name: &str) -> Result<Option<String>, String>;
    fn apply(&self, name: &str, value: &str) -> Result<(), String>;
    fn clear(&self, name: &str) -> Result<(), String>;
}

/// The refusal every entry point returns for a name outside [`ALLOWED`].
fn refused(name: &str) -> String {
    format!("{name} is not one of the login-environment names OpenLatch may touch")
}

fn allowed(name: &str) -> Result<&'static str, String> {
    ALLOWED
        .iter()
        .find(|n| **n == name)
        .copied()
        .ok_or_else(|| refused(name))
}

/// [`Sink::read`] behind the allow-list: a name outside [`ALLOWED`] never reaches the sink.
pub fn read(sink: &dyn Sink, name: &str) -> Result<Option<String>, String> {
    sink.read(allowed(name)?)
}

/// [`Sink::apply`] behind the allow-list and the value check ([`value_is_deliverable`]).
pub fn apply(sink: &dyn Sink, name: &str, value: &str) -> Result<(), String> {
    let name = allowed(name)?;
    if !value_is_deliverable(value) {
        return Err(format!(
            "{name}: the value holds characters this OS's login environment cannot carry"
        ));
    }
    sink.apply(name, value)
}

/// [`Sink::clear`] behind the allow-list.
pub fn clear(sink: &dyn Sink, name: &str) -> Result<(), String> {
    sink.clear(allowed(name)?)
}

/// Whether `value` can be written verbatim into this OS's login environment. Linux refuses what
/// environment.d(5) cannot quote; macOS and Windows carry any UTF-8 string.
pub fn value_is_deliverable(value: &str) -> bool {
    !cfg!(target_os = "linux") || linux::validate_value(value).is_ok()
}

/// How to remove both names by hand on this OS, for when OpenLatch could not.
pub fn manual_clear_hint() -> String {
    imp::MANUAL_CLEAR.to_string()
}

/// Runs `program` within [`COMMAND_BUDGET`]: its stdout on success, else why it failed.
#[cfg(not(windows))]
pub(super) fn run_checked(program: &str, args: &[&str]) -> Result<String, String> {
    use crate::model_relay::trust_store::{Ran, Runner, SystemRunner};

    let args: Vec<String> = args.iter().map(|a| (*a).to_string()).collect();
    match SystemRunner.run(program, &args, COMMAND_BUDGET) {
        Ran::Exited {
            success: true,
            stdout,
            ..
        } => Ok(stdout),
        Ran::Exited { stderr, .. } => Err(format!("{program} failed: {}", stderr.trim())),
        Ran::Missing => Err(format!("{program} is not installed")),
        Ran::TimedOut => Err(format!(
            "{program} did not answer within {} s",
            COMMAND_BUDGET.as_secs()
        )),
        Ran::Failed(e) => Err(format!("{program}: {e}")),
    }
}

/// The OS sink the daemon and `uninstall` hand down. Reached only through the allow-listed entry points.
struct OsSink;

impl Sink for OsSink {
    fn read(&self, name: &str) -> Result<Option<String>, String> {
        imp::read(name)
    }

    fn apply(&self, name: &str, value: &str) -> Result<(), String> {
        imp::apply(name, value)
    }

    fn clear(&self, name: &str) -> Result<(), String> {
        imp::clear(name)
    }
}

static OS_SINK: OsSink = OsSink;

/// The real OS sink. Under `cfg(test)` it PANICS, so a unit test can never touch the login env (the same guard as
/// `cline_app::process::table()`).
pub fn os_sink() -> &'static dyn Sink {
    if cfg!(test) {
        panic!("os_sink reached from a test — pass a RecordingSink");
    }
    &OS_SINK
}

/// What one run of external commands did: every step that succeeded, and every one that failed with why.
#[derive(Debug, Default)]
pub struct Report {
    pub done: Vec<&'static str>,
    pub failed: Vec<(&'static str, String)>,
}

// ---------------------------------------------------------------------------
// §2 · The ownership ledger (D-14)
// ---------------------------------------------------------------------------

/// One name's ledger entry: the value OpenLatch wrote, and — only across an in-flight update — the previous value it
/// wrote.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct LedgerEntry {
    pub value: String,
    #[serde(default)]
    pub prev: Option<String>,
}

/// `<ol_dir>/state/cline-login-env.json` = `{ "<NAME>": {"value": "<v>", "prev": null}, … }`, written owner-only.
/// A value not in the ledger is NEVER adopted, even when it equals the one wanted: equality does not prove we wrote
/// it, and adopting would let uninstall or the kill switch delete a customer's setting (D-14).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Ledger {
    path: PathBuf,
    entries: BTreeMap<String, LedgerEntry>,
}

impl Ledger {
    /// `<ol_dir>/state/cline-login-env.json`.
    pub fn path_in(ol_dir: &Path) -> PathBuf {
        ol_dir.join("state").join(LEDGER_FILE)
    }

    /// Loads the ledger; a missing or unparseable file is an empty ledger (a lost ledger shows as customer-override
    /// until the owner clears the value — never as ours).
    pub fn load(ol_dir: &Path) -> Ledger {
        let path = Self::path_in(ol_dir);
        let entries = std::fs::read_to_string(&path)
            .ok()
            .and_then(|raw| serde_json::from_str::<BTreeMap<String, LedgerEntry>>(&raw).ok())
            .map(|m| {
                m.into_iter()
                    .filter(|(k, _)| ALLOWED.contains(&k.as_str()))
                    .collect()
            })
            .unwrap_or_default();
        Ledger { path, entries }
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    pub fn get(&self, name: &str) -> Option<&LedgerEntry> {
        self.entries.get(name)
    }

    /// `c ∈ ours`: the ledger's value, or the previous value of an in-flight update.
    pub fn is_ours(&self, name: &str, current: &str) -> bool {
        self.entries
            .get(name)
            .is_some_and(|e| e.value == current || e.prev.as_deref() == Some(current))
    }

    fn set(&mut self, name: &str, value: &str, prev: Option<String>) {
        self.entries.insert(
            name.to_string(),
            LedgerEntry {
                value: value.to_string(),
                prev,
            },
        );
    }

    /// Whether an entry was removed.
    fn remove(&mut self, name: &str) -> bool {
        self.entries.remove(name).is_some()
    }

    /// Writes the ledger (owner-only). An empty ledger is written as `{}` rather than deleted: only uninstall deletes
    /// the file.
    pub fn persist(&self) -> Result<(), String> {
        if let Some(parent) = self.path.parent() {
            std::fs::create_dir_all(parent).map_err(|e| format!("ledger directory: {e}"))?;
        }
        let body =
            serde_json::to_string_pretty(&self.entries).map_err(|e| format!("ledger: {e}"))?;
        crate::fs_secure::write_owner_only(&self.path, &body).map_err(|e| format!("ledger: {e}"))
    }

    /// Deletes the ledger file (uninstall). A missing file is fine.
    pub fn delete_file(&self) -> std::io::Result<()> {
        match std::fs::remove_file(&self.path) {
            Err(e) if e.kind() != std::io::ErrorKind::NotFound => Err(e),
            _ => Ok(()),
        }
    }
}

/// What [`reconcile_login_env`] left in place.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LoginEnvOutcome {
    /// Both names hold exactly the wanted values, both are ours, and both read back.
    Delivered,
    /// Nothing is wanted: every value of ours is gone; a customer's value is untouched.
    Released,
    /// At least one name holds a value OpenLatch did not write. Neither name was applied (a half-ours pair would
    /// point a customer's bootstrap at our runtime or the reverse); ours, if any, were cleared.
    CustomerOverride { names: Vec<&'static str> },
    /// A read, apply, persist or read-back failed. Every value of ours in the pair was cleared (fail-closed).
    Failed { detail: String },
}

/// Reconciles both [`ALLOWED`] names with `want` (`None` = release ours). A value the ledger does not claim is never
/// touched.
/// WRITE-AHEAD: the ledger is persisted before every apply, so a crash between the two leaves a value still ours.
/// PAIR FAIL-CLOSED: `Delivered` only once both are applied, read back and the ledger confirmed; any failure clears
/// every value of ours in the pair. The pair moves together: one customer value holds the other back.
pub fn reconcile_login_env(
    sink: &dyn Sink,
    ledger: &mut Ledger,
    want: Option<(&str, &str)>,
) -> LoginEnvOutcome {
    let mut report = Report::default();
    let outcome = reconcile_inner(sink, ledger, want, &mut report);
    if !report.failed.is_empty() {
        tracing::warn!(
            code = crate::error::ERR_CLINE_PLUGIN_NOT_LOADED,
            done = ?report.done,
            failed = ?report.failed,
            "the Cline login environment could not be fully reconciled"
        );
    } else if !report.done.is_empty() {
        tracing::info!(done = ?report.done, "Cline login environment reconciled");
    }
    outcome
}

fn reconcile_inner(
    sink: &dyn Sink,
    ledger: &mut Ledger,
    want: Option<(&str, &str)>,
    report: &mut Report,
) -> LoginEnvOutcome {
    // Read both first: nothing is decided on half a picture.
    let mut slots = Vec::with_capacity(ALLOWED.len());
    for (name, want) in ALLOWED
        .into_iter()
        .zip([want.map(|w| w.0), want.map(|w| w.1)])
    {
        match read(sink, name) {
            Ok(current) => slots.push(Slot {
                name,
                want,
                current,
            }),
            Err(e) => {
                report.failed.push((name, format!("read: {e}")));
                return LoginEnvOutcome::Failed {
                    detail: format!("read {name}: {e}"),
                };
            }
        }
    }

    if want.is_none() {
        let failures: Vec<String> = slots
            .iter()
            .filter_map(|s| release(sink, ledger, s, report).err())
            .collect();
        return if failures.is_empty() {
            LoginEnvOutcome::Released
        } else {
            LoginEnvOutcome::Failed {
                detail: failures.join("; "),
            }
        };
    }

    // The pair moves together: one customer value holds the other back.
    let theirs: Vec<&'static str> = slots
        .iter()
        .filter(|s| {
            s.current
                .as_deref()
                .is_some_and(|c| !ledger.is_ours(s.name, c))
        })
        .map(|s| s.name)
        .collect();
    if !theirs.is_empty() {
        for s in &slots {
            let _ = release(sink, ledger, s, report);
        }
        return LoginEnvOutcome::CustomerOverride { names: theirs };
    }

    // Apply: every current value is ours or absent.
    let mut failure: Option<String> = None;
    for s in &slots {
        let (name, Some(v)) = (s.name, s.want) else {
            continue;
        };
        let prev = match s.current.as_deref() {
            Some(c) if c == v => continue,
            c => c.map(str::to_string),
        };
        ledger.set(name, v, prev);
        if let Err(e) = ledger.persist() {
            report.failed.push((name, e.clone()));
            failure = Some(format!("{name}: {e}"));
            break;
        }
        match apply(sink, name, v) {
            Ok(()) => report.done.push(name),
            Err(e) => {
                // The write-ahead entry stays: the sink may hold `v`, `prev` or nothing, and the fail-closed
                // cleanup below clears whichever of ours it finds before dropping the entry.
                report.failed.push((name, format!("apply: {e}")));
                failure = Some(format!("apply {name}: {e}"));
                break;
            }
        }
    }

    // Read both back.
    if failure.is_none() {
        for s in &slots {
            let (name, Some(v)) = (s.name, s.want) else {
                continue;
            };
            match read(sink, name) {
                Ok(Some(got)) if got == v => {}
                Ok(got) => {
                    report
                        .failed
                        .push((name, format!("read-back: {got:?} is not {v:?}")));
                    failure = Some(format!("{name} did not read back"));
                    break;
                }
                Err(e) => {
                    report.failed.push((name, format!("read-back: {e}")));
                    failure = Some(format!("read back {name}: {e}"));
                    break;
                }
            }
        }
    }

    // Confirmed: the in-flight `prev` is dropped. Delivered only once that is on disk — a ledger that still says
    // `prev` after a failed persist would claim a value OpenLatch no longer holds.
    if failure.is_none() {
        let mut changed = false;
        for s in &slots {
            let (name, Some(v)) = (s.name, s.want) else {
                continue;
            };
            if ledger.get(name).is_some_and(|e| e.prev.is_some()) {
                ledger.set(name, v, None);
                changed = true;
            }
        }
        if changed {
            if let Err(e) = ledger.persist() {
                report.failed.push(("ledger", e.clone()));
                failure = Some(format!("ledger: {e}"));
            }
        }
    }

    match failure {
        Some(detail) => {
            fail_closed(sink, ledger, &slots, report);
            LoginEnvOutcome::Failed { detail }
        }
        None => LoginEnvOutcome::Delivered,
    }
}

/// Nothing of ours stays behind for a pair that is not whole: for every name with a ledger entry, the sink's
/// CURRENT value is cleared when it is ours (the entry's `value` or `prev`), and the entry is dropped. A customer's
/// value is never cleared; an unreadable one is cleared (it was ours or absent at this run's first read).
fn fail_closed(sink: &dyn Sink, ledger: &mut Ledger, slots: &[Slot<'_>], report: &mut Report) {
    for Slot { name, .. } in slots {
        if ledger.get(name).is_none() {
            continue;
        }
        match read(sink, name) {
            Ok(Some(c)) if !ledger.is_ours(name, &c) => {
                ledger.remove(name);
                let _ = ledger.persist();
            }
            Ok(None) => {
                ledger.remove(name);
                let _ = ledger.persist();
            }
            Ok(Some(_)) | Err(_) => {
                let _ = clear_ours(sink, ledger, name, report);
            }
        }
    }
}

/// One name as this run found it.
struct Slot<'a> {
    name: &'static str,
    want: Option<&'a str>,
    current: Option<String>,
}

/// Releases one name: ours is cleared, a customer's is left alone, a stale entry for an absent value is dropped.
fn release(
    sink: &dyn Sink,
    ledger: &mut Ledger,
    slot: &Slot<'_>,
    report: &mut Report,
) -> Result<(), String> {
    match slot.current.as_deref() {
        Some(c) if ledger.is_ours(slot.name, c) => clear_ours(sink, ledger, slot.name, report),
        Some(_) => Ok(()),
        None if ledger.remove(slot.name) => ledger
            .persist()
            .inspect_err(|e| report.failed.push((slot.name, e.clone()))),
        None => Ok(()),
    }
}

/// Clear first, then drop the entry and persist.
fn clear_ours(
    sink: &dyn Sink,
    ledger: &mut Ledger,
    name: &'static str,
    report: &mut Report,
) -> Result<(), String> {
    if let Err(e) = clear(sink, name) {
        report.failed.push((name, format!("clear: {e}")));
        return Err(format!("clear {name}: {e}"));
    }
    report.done.push(name);
    ledger.remove(name);
    ledger
        .persist()
        .inspect_err(|e| report.failed.push((name, e.clone())))
}

// ---------------------------------------------------------------------------
// The test sink
// ---------------------------------------------------------------------------

/// A sink over an in-memory map that records every call, with failures a test can inject. The only sink a unit test
/// may pass: [`os_sink`] panics under `cfg(test)`.
#[cfg(test)]
pub(crate) mod test_support {
    use std::collections::BTreeMap;
    use std::sync::Mutex;

    use super::Sink;

    #[derive(Debug, Clone, PartialEq, Eq)]
    pub(crate) enum Call {
        Read(String),
        Apply(String, String),
        Clear(String),
    }

    #[derive(Default)]
    pub(crate) struct RecordingSink {
        pub(crate) values: Mutex<BTreeMap<String, String>>,
        pub(crate) calls: Mutex<Vec<Call>>,
        /// Names whose `apply` fails.
        pub(crate) fail_apply: Mutex<Vec<String>>,
        /// Names whose `apply` "succeeds" without storing (a read-back mismatch).
        pub(crate) drop_apply: Mutex<Vec<String>>,
        /// Names whose `read` fails.
        pub(crate) fail_read: Mutex<Vec<String>>,
        /// Names whose `clear` fails (the value stays).
        pub(crate) fail_clear: Mutex<Vec<String>>,
        /// Called at each `apply`, before it is recorded (e.g. to snapshot the ledger file).
        #[allow(clippy::type_complexity)]
        pub(crate) on_apply: Mutex<Option<Box<dyn Fn(&str, &str) + Send>>>,
    }

    fn lock<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
        m.lock().unwrap_or_else(|e| e.into_inner())
    }

    impl RecordingSink {
        pub(crate) fn with(values: &[(&str, &str)]) -> Self {
            let s = Self::default();
            for (k, v) in values {
                lock(&s.values).insert((*k).to_string(), (*v).to_string());
            }
            s
        }

        pub(crate) fn value(&self, name: &str) -> Option<String> {
            lock(&self.values).get(name).cloned()
        }

        pub(crate) fn calls(&self) -> Vec<Call> {
            lock(&self.calls).clone()
        }
    }

    impl Sink for RecordingSink {
        fn read(&self, name: &str) -> Result<Option<String>, String> {
            lock(&self.calls).push(Call::Read(name.to_string()));
            if lock(&self.fail_read).iter().any(|n| n == name) {
                return Err("read failed (injected)".to_string());
            }
            Ok(lock(&self.values).get(name).cloned())
        }

        fn apply(&self, name: &str, value: &str) -> Result<(), String> {
            if let Some(f) = lock(&self.on_apply).as_ref() {
                f(name, value);
            }
            lock(&self.calls).push(Call::Apply(name.to_string(), value.to_string()));
            if lock(&self.fail_apply).iter().any(|n| n == name) {
                return Err("apply failed (injected)".to_string());
            }
            if !lock(&self.drop_apply).iter().any(|n| n == name) {
                lock(&self.values).insert(name.to_string(), value.to_string());
            }
            Ok(())
        }

        fn clear(&self, name: &str) -> Result<(), String> {
            lock(&self.calls).push(Call::Clear(name.to_string()));
            if lock(&self.fail_clear).iter().any(|n| n == name) {
                return Err("clear failed (injected)".to_string());
            }
            lock(&self.values).remove(name);
            Ok(())
        }
    }
}

#[cfg(test)]
mod tests {
    use super::test_support::{Call, RecordingSink};
    use super::*;

    /// A name outside the allow-list never reaches the sink, on any entry point.
    #[test]
    fn allow_list_refuses_other_names() {
        let sink = RecordingSink::with(&[("PATH", "/usr/bin")]);
        assert!(apply(&sink, "PATH", "/tmp/evil").is_err());
        assert!(clear(&sink, "PATH").is_err());
        assert!(read(&sink, "PATH").is_err());
        assert_eq!(sink.calls(), Vec::<Call>::new(), "the sink was untouched");
        assert_eq!(sink.value("PATH").as_deref(), Some("/usr/bin"));

        // Control: an allowed name does reach it.
        assert_eq!(read(&sink, ALLOWED[0]), Ok(None));
        assert_eq!(sink.calls(), vec![Call::Read(ALLOWED[0].to_string())]);
    }

    const W: &str = ALLOWED[0];
    const R: &str = ALLOWED[1];

    /// A ledger under a fresh temp OL dir, with `entries` persisted: `(name, value, prev)`.
    fn ledger_with(dir: &Path, entries: &[(&str, &str, Option<&str>)]) -> Ledger {
        let mut ledger = Ledger::load(dir);
        for (name, value, prev) in entries {
            ledger.set(name, value, prev.map(str::to_string));
        }
        ledger.persist().expect("persist the seeded ledger");
        ledger
    }

    fn applies(sink: &RecordingSink) -> Vec<Call> {
        sink.calls()
            .into_iter()
            .filter(|c| matches!(c, Call::Apply(..)))
            .collect()
    }

    fn clears(sink: &RecordingSink) -> Vec<Call> {
        sink.calls()
            .into_iter()
            .filter(|c| matches!(c, Call::Clear(..)))
            .collect()
    }

    /// §2's seven rows, each for the pair, plus "the pair moves together". Every row also reloads the
    /// ledger from disk, so what is asserted is what the next tick (or uninstall) would see.
    #[test]
    fn customer_value_wins_matrix() {
        let (wv, rv) = (
            "/ol/cline-plugin-bootstrap/wrapper",
            "/ol/bin/cline-js-runtime-vscode",
        );
        let want = Some((wv, rv));

        // 1. (None, Some(v)) → recorded, applied, delivered.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::default();
        let mut ledger = Ledger::load(dir.path());
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::Delivered
        );
        assert_eq!(sink.value(W).as_deref(), Some(wv));
        assert_eq!(sink.value(R).as_deref(), Some(rv));
        let on_disk = Ledger::load(dir.path());
        assert!(on_disk.is_ours(W, wv) && on_disk.is_ours(R, rv));
        assert_eq!(on_disk.get(W).unwrap().prev, None);

        // 2. (Some(c), Some(v)), c ∈ ours, c == v → nothing written.
        let sink = RecordingSink::with(&[(W, wv), (R, rv)]);
        let mut ledger = ledger_with(dir.path(), &[(W, wv, None), (R, rv, None)]);
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::Delivered
        );
        assert!(applies(&sink).is_empty() && clears(&sink).is_empty());

        // 3. (Some(c), Some(v)), c ∈ ours, c != v → updated; `prev` dropped after the read-back.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::with(&[(W, "/old/wrapper"), (R, "/old/runtime")]);
        let mut ledger = ledger_with(
            dir.path(),
            &[(W, "/old/wrapper", None), (R, "/old/runtime", None)],
        );
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::Delivered
        );
        assert_eq!(sink.value(W).as_deref(), Some(wv));
        assert_eq!(sink.value(R).as_deref(), Some(rv));
        let on_disk = Ledger::load(dir.path());
        assert_eq!(on_disk.get(W).unwrap().value, wv);
        assert_eq!(on_disk.get(W).unwrap().prev, None);
        assert!(!on_disk.is_ours(W, "/old/wrapper"));

        // 4. (Some(theirs), Some(v)) → CustomerOverride: never touched, never recorded — even when the customer's
        //    value EQUALS the wanted one (equality is not proof we wrote it; D-14, no adoption).
        for theirs in ["/customer/wrapper", wv] {
            let dir = tempfile::tempdir().unwrap();
            let sink = RecordingSink::with(&[(W, theirs), (R, "/customer/runtime")]);
            let mut ledger = Ledger::load(dir.path());
            assert_eq!(
                reconcile_login_env(&sink, &mut ledger, want),
                LoginEnvOutcome::CustomerOverride { names: vec![W, R] },
                "{theirs}"
            );
            assert!(applies(&sink).is_empty() && clears(&sink).is_empty());
            assert_eq!(sink.value(W).as_deref(), Some(theirs));
            assert!(Ledger::load(dir.path()).is_empty(), "never recorded");
        }

        // 5. (Some(c), None), c ∈ ours → cleared, dropped from the ledger. `prev` counts as ours too.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::with(&[(W, wv), (R, "/old/runtime")]);
        let mut ledger = ledger_with(dir.path(), &[(W, wv, None), (R, rv, Some("/old/runtime"))]);
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, None),
            LoginEnvOutcome::Released
        );
        assert_eq!(sink.value(W), None);
        assert_eq!(sink.value(R), None);
        assert!(Ledger::load(dir.path()).is_empty());

        // 6. (Some(theirs), None) → nothing, even with a ledger entry naming another value.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::with(&[(W, "/customer/wrapper"), (R, "/customer/runtime")]);
        let mut ledger = ledger_with(dir.path(), &[(W, wv, None)]);
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, None),
            LoginEnvOutcome::Released
        );
        assert!(clears(&sink).is_empty() && applies(&sink).is_empty());
        assert_eq!(sink.value(W).as_deref(), Some("/customer/wrapper"));
        assert_eq!(sink.value(R).as_deref(), Some("/customer/runtime"));

        // 7. (None, None) → the stale ledger entry goes.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::default();
        let mut ledger = ledger_with(dir.path(), &[(W, wv, None), (R, rv, None)]);
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, None),
            LoginEnvOutcome::Released
        );
        assert!(clears(&sink).is_empty() && applies(&sink).is_empty());
        assert!(Ledger::load(dir.path()).is_empty());

        // The pair moves together (a): one customer value holds the other back; ours on the other is cleared.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::with(&[(W, wv), (R, "/customer/runtime")]);
        let mut ledger = ledger_with(dir.path(), &[(W, wv, None)]);
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::CustomerOverride { names: vec![R] }
        );
        assert!(applies(&sink).is_empty());
        assert_eq!(clears(&sink), vec![Call::Clear(W.to_string())]);
        assert_eq!(sink.value(W), None);
        assert_eq!(sink.value(R).as_deref(), Some("/customer/runtime"));
        assert!(Ledger::load(dir.path()).is_empty());

        // The pair moves together (b): the other name absent → still not applied.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::with(&[(W, "/customer/wrapper")]);
        let mut ledger = Ledger::load(dir.path());
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::CustomerOverride { names: vec![W] }
        );
        assert!(applies(&sink).is_empty() && clears(&sink).is_empty());
        assert_eq!(sink.value(R), None);
        assert!(Ledger::load(dir.path()).is_empty());

        // Fail closed: a read-back mismatch on one name clears BOTH of ours and reports Failed.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::default();
        sink.drop_apply.lock().unwrap().push(R.to_string());
        let mut ledger = Ledger::load(dir.path());
        assert!(matches!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::Failed { .. }
        ));
        assert_eq!(sink.value(W), None);
        assert_eq!(sink.value(R), None);
        assert!(Ledger::load(dir.path()).is_empty());
    }

    /// An owned v1→v2 update whose apply FAILS keeps the write-ahead `{v2, prev: v1}` entry until the fail-closed
    /// cleanup has cleared v1 — v1 is never stranded in the login environment with no ledger entry claiming it.
    #[test]
    fn update_failure_keeps_ownership_of_prev() {
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::with(&[(W, "/ol/wrapper-v1"), (R, "/ol/runtime-v1")]);
        // Apply fails WITHOUT storing: the sink still holds v1 when the cleanup reads it.
        sink.fail_apply.lock().unwrap().push(R.to_string());
        let mut ledger = ledger_with(
            dir.path(),
            &[(W, "/ol/wrapper-v1", None), (R, "/ol/runtime-v1", None)],
        );
        assert!(matches!(
            reconcile_login_env(
                &sink,
                &mut ledger,
                Some(("/ol/wrapper-v2", "/ol/runtime-v2"))
            ),
            LoginEnvOutcome::Failed { .. }
        ));
        assert_eq!(sink.value(W), None, "W (applied v2) is cleared");
        assert_eq!(
            sink.value(R),
            None,
            "R's v1 (the entry's prev) is cleared, not stranded"
        );
        assert_eq!(
            clears(&sink),
            vec![Call::Clear(W.to_string()), Call::Clear(R.to_string())]
        );
        assert!(Ledger::load(dir.path()).is_empty());

        // Control: the existing no-`prev` apply failure still leaves no entry and nothing applied behind.
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::default();
        sink.fail_apply.lock().unwrap().push(R.to_string());
        let mut ledger = Ledger::load(dir.path());
        assert!(matches!(
            reconcile_login_env(
                &sink,
                &mut ledger,
                Some(("/ol/wrapper-v2", "/ol/runtime-v2"))
            ),
            LoginEnvOutcome::Failed { .. }
        ));
        assert_eq!(sink.value(W), None);
        assert_eq!(sink.value(R), None);
        assert!(Ledger::load(dir.path()).is_empty());
    }

    /// A ledger persist that fails after the read-back (dropping `prev`) is a failure, not a delivery: the pair is
    /// cleared fail-closed.
    #[test]
    fn prev_drop_persist_failure_is_failed() {
        let dir = tempfile::tempdir().unwrap();
        let ol = dir.path().to_path_buf();
        let sink = RecordingSink::with(&[(W, "/ol/wrapper-v1"), (R, "/ol/runtime-v1")]);
        let mut ledger = ledger_with(
            &ol,
            &[(W, "/ol/wrapper-v1", None), (R, "/ol/runtime-v1", None)],
        );
        // At the LAST apply (its write-ahead persist is done), make `state/` a file: every later persist fails.
        {
            let ol = ol.clone();
            *sink.on_apply.lock().unwrap() = Some(Box::new(move |name, _| {
                if name == R {
                    let state = ol.join("state");
                    std::fs::remove_dir_all(&state).unwrap();
                    std::fs::write(&state, "not a directory").unwrap();
                }
            }));
        }
        let outcome = reconcile_login_env(
            &sink,
            &mut ledger,
            Some(("/ol/wrapper-v2", "/ol/runtime-v2")),
        );
        assert!(
            matches!(&outcome, LoginEnvOutcome::Failed { detail } if detail.starts_with("ledger")),
            "{outcome:?}"
        );
        assert_eq!(sink.value(W), None, "cleared fail-closed");
        assert_eq!(sink.value(R), None, "cleared fail-closed");
    }

    /// WRITE-AHEAD (§2): the ledger holds `v` on disk before `apply` is recorded, and both crash windows — after the
    /// persist, before or after the apply — leave a value the next run recognises as ours.
    #[test]
    fn ledger_is_written_before_apply() {
        let dir = tempfile::tempdir().unwrap();
        let ol = dir.path().to_path_buf();
        let want = Some(("/ol/wrapper-v1", "/ol/runtime-v1"));

        // The ledger file, as it stands when each apply is called.
        type Seen = Vec<(String, Option<LedgerEntry>)>;
        let seen: std::sync::Arc<std::sync::Mutex<Seen>> = std::sync::Arc::default();
        let sink = RecordingSink::default();
        {
            let seen = seen.clone();
            let ol = ol.clone();
            *sink.on_apply.lock().unwrap() = Some(Box::new(move |name, _value| {
                let entry = Ledger::load(&ol).get(name).cloned();
                seen.lock().unwrap().push((name.to_string(), entry));
            }));
        }
        let mut ledger = Ledger::load(&ol);
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::Delivered
        );
        let seen = seen.lock().unwrap().clone();
        assert_eq!(seen.len(), 2, "both names were applied: {seen:?}");
        for (name, entry) in &seen {
            let v = if name == W {
                "/ol/wrapper-v1"
            } else {
                "/ol/runtime-v1"
            };
            assert_eq!(
                entry.as_ref().map(|e| e.value.as_str()),
                Some(v),
                "{name}: the ledger held v before apply"
            );
        }

        // Crash after persist + apply (ledger v, sink v): ours — updated to v2, then cleared by want=None.
        let crashed = RecordingSink::with(&[(W, "/ol/wrapper-v1"), (R, "/ol/runtime-v1")]);
        let mut ledger = Ledger::load(&ol);
        let v2 = Some(("/ol/wrapper-v2", "/ol/runtime-v2"));
        assert_eq!(
            reconcile_login_env(&crashed, &mut ledger, v2),
            LoginEnvOutcome::Delivered
        );
        assert_eq!(crashed.value(W).as_deref(), Some("/ol/wrapper-v2"));
        let mut ledger = Ledger::load(&ol);
        assert_eq!(
            reconcile_login_env(&crashed, &mut ledger, None),
            LoginEnvOutcome::Released
        );
        assert_eq!(crashed.value(W), None);
        assert_eq!(crashed.value(R), None);

        // Crash after persist, before apply (ledger v, sink empty): the entry is ours; nothing blocks the next run.
        let dir = tempfile::tempdir().unwrap();
        let mut ledger = ledger_with(
            dir.path(),
            &[(W, "/ol/wrapper-v1", None), (R, "/ol/runtime-v1", None)],
        );
        let empty = RecordingSink::default();
        assert_eq!(
            reconcile_login_env(&empty, &mut ledger, want),
            LoginEnvOutcome::Delivered
        );

        // UPDATE crash (ledger {value: v2, prev: v1}, sink still v1): v1 is ours, not CustomerOverride; updated to
        // v2, and `prev` dropped after the read-back; want=None then clears it.
        let dir = tempfile::tempdir().unwrap();
        let mut ledger = ledger_with(
            dir.path(),
            &[
                (W, "/ol/wrapper-v2", Some("/ol/wrapper-v1")),
                (R, "/ol/runtime-v2", Some("/ol/runtime-v1")),
            ],
        );
        let sink = RecordingSink::with(&[(W, "/ol/wrapper-v1"), (R, "/ol/runtime-v1")]);
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, v2),
            LoginEnvOutcome::Delivered
        );
        assert_eq!(sink.value(W).as_deref(), Some("/ol/wrapper-v2"));
        assert_eq!(sink.value(R).as_deref(), Some("/ol/runtime-v2"));
        let on_disk = Ledger::load(dir.path());
        assert_eq!(on_disk.get(W).unwrap().prev, None);
        assert_eq!(on_disk.get(R).unwrap().prev, None);
        let mut ledger = on_disk;
        assert_eq!(
            reconcile_login_env(&sink, &mut ledger, None),
            LoginEnvOutcome::Released
        );
        assert_eq!(sink.value(W), None);
        assert_eq!(sink.value(R), None);
        assert!(Ledger::load(dir.path()).is_empty());

        // An apply failure drops the entry again (no permanent claim on a value never written).
        let dir = tempfile::tempdir().unwrap();
        let sink = RecordingSink::default();
        sink.fail_apply.lock().unwrap().push(W.to_string());
        let mut ledger = Ledger::load(dir.path());
        assert!(matches!(
            reconcile_login_env(&sink, &mut ledger, want),
            LoginEnvOutcome::Failed { .. }
        ));
        assert!(Ledger::load(dir.path()).is_empty());
    }
}