openlatch-client 0.6.3

//! The Cline plugin delivery task (Cline I-5 plan 02 §3, INDEX D-08): the ONE owner of Cline.app's hub, and the
//! only writer of the two login-environment variables VS Code's Cline reads.
//!
//! It runs whether or not the model relay runs (D-08) — blocking must not depend on model-traffic capture — and
//! ONLY in a `reconcile_wiring` daemon: the embedded/test daemon links the lib without `cfg(test)`, where the process
//! table and the login-env sink are the real ones, so spawning it there would `launchctl setenv` and restart the
//! developer's real hub from `cargo test`. The relay's wiring loop no longer touches the hub: it publishes its
//! capture state through a [`CapturePublisher`] and nudges this task.
//!
//! Schedule: the first tick at spawn, then every [`TICK`], plus a tick on every nudge. Every tick reads the world
//! ([`gather`]) and hands it to [`tick_from`], which the unit tests reach with a recording sink and a fake table.
//!
//! Prime invariant (design): the plugin is advertised to a host only when that host is proven able to load it;
//! anything that goes wrong — a runtime that disappeared, a host below the floor, a broken tree, the kill switch —
//! removes OUR values in the same tick rather than leaving a host pointing at a runtime that fails every Cline run.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::{watch, Notify};

use crate::config::Config;
use crate::core::login_env::{self, Ledger, LoginEnvOutcome, Sink};
use crate::core::supervision::task::{spawn_supervised, HealthRegistry, RestartPolicy, TaskSpec};
use crate::hooks::bindings::cline::ClineBinding;
use crate::hooks::cline_app::{
    self, process::ProcessTable, ClineApp, Desired, PluginEnv, ProxyEnvVars,
};
use crate::hooks::cline_hosts::HostClass;
use crate::hooks::cline_runtime::{self, reason, HostAssessment};

use super::cline_hub::{self, HubAction, HubOutcome};

/// The periodic tick, between nudges.
pub(crate) const TICK: Duration = Duration::from_secs(60);

/// `<ol_dir>/state/cline-plugin-delivery.json`.
pub(crate) const STATE_FILE: &str = "cline-plugin-delivery.json";

/// What the relay has reported about Cline's capture. `None` = the relay has not reported yet (skip the hub step:
/// a first tick must never read "no capture" off a hub that still carries it from a crashed daemon and restart it
/// NOW); `Some(None)` = capture not live; `Some(Some(v))` = capture live with `v`.
pub(crate) type CaptureState = Option<Option<ProxyEnvVars>>;

/// The relay's side of the capture watch: it publishes, the delivery task reads.
#[derive(Clone)]
pub(crate) struct CapturePublisher {
    tx: Arc<watch::Sender<CaptureState>>,
    nudge: Arc<Notify>,
}

impl CapturePublisher {
    /// Publishes `Some(capture)` and, when that changed what the task last saw, nudges it.
    pub(crate) fn publish(&self, capture: Option<ProxyEnvVars>) {
        let next = Some(capture);
        let changed = self.tx.send_if_modified(|current| {
            if *current == next {
                false
            } else {
                *current = next;
                true
            }
        });
        if changed {
            self.nudge.notify_one();
        }
    }
}

/// The delivery task's side.
pub(crate) struct CaptureWatch {
    rx: watch::Receiver<CaptureState>,
    nudge: Arc<Notify>,
}

/// One watch, starting at `None` (not yet reported), and the nudge both sides share.
pub(crate) fn channel() -> (CapturePublisher, CaptureWatch) {
    let (tx, rx) = watch::channel(None);
    let nudge = Arc::new(Notify::new());
    (
        CapturePublisher {
            tx: Arc::new(tx),
            nudge: nudge.clone(),
        },
        CaptureWatch { rx, nudge },
    )
}

// ---------------------------------------------------------------------------
// The state file (§3 step 7)
// ---------------------------------------------------------------------------

/// One host's delivery, as doctor and plan 03 read it.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) struct HostDelivery {
    pub(crate) delivered: bool,
    /// Unix ms at which delivery started; unchanged while delivered, null while not.
    pub(crate) since: Option<i64>,
    /// `"<code>"` or `"<code>: <detail>"` (INDEX D-20); null when delivered.
    pub(crate) reason: Option<String>,
    /// The host's embedded `@cline/core` version, when readable.
    pub(crate) core: Option<String>,
    pub(crate) customer_override: bool,
}

/// `<ol_dir>/state/cline-plugin-delivery.json`. Keys are exactly `HostClass`'s serde spelling. A host that is not
/// detected is `null`.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) struct DeliveryState {
    pub(crate) hub: Option<HostDelivery>,
    pub(crate) vscode: Option<HostDelivery>,
    /// The Desired passed to `cline_hub::reconcile`; null when the hub is not managed. What doctor classifies the
    /// hub against.
    pub(crate) hub_desired: Option<Desired>,
}

pub(crate) fn state_path(ol_dir: &Path) -> PathBuf {
    ol_dir.join("state").join(STATE_FILE)
}

/// Missing or unreadable → None.
pub(crate) fn read_state(ol_dir: &Path) -> Option<DeliveryState> {
    let raw = std::fs::read_to_string(state_path(ol_dir)).ok()?;
    serde_json::from_str(&raw).ok()
}

/// Whether the delivery task recorded the hub as delivered.
pub(crate) fn hub_delivered(ol_dir: &Path) -> bool {
    read_state(ol_dir)
        .and_then(|state| state.hub)
        .is_some_and(|hub| hub.delivered)
}

/// What doctor classifies the hub against (§4.3): the last `hub_desired`, or the empty Desired when there is none.
pub(crate) fn read_hub_desired(ol_dir: &Path) -> Desired {
    read_state(ol_dir)
        .and_then(|s| s.hub_desired)
        .unwrap_or_default()
}

/// Whether the state file now holds `state`.
fn write_state(ol_dir: &Path, state: &DeliveryState) -> bool {
    match crate::fs_secure::write_json_state(&state_path(ol_dir), state) {
        Ok(()) => true,
        Err(e) => {
            tracing::warn!(
                code = crate::error::ERR_CLINE_PLUGIN_NOT_LOADED,
                error = %e,
                "could not write the Cline plugin delivery state"
            );
            false
        }
    }
}

// ---------------------------------------------------------------------------
// One tick
// ---------------------------------------------------------------------------

/// Everything a tick takes from the world, read once by [`gather`] so [`tick_from`] is pure over its inputs.
#[derive(Debug, Clone)]
pub(crate) struct TickInputs {
    /// A Cline agent is detected. Without one: release the login env and stop (§3 step 0).
    pub(crate) binding: bool,
    /// This instance owns the machine-global login env: `owns_wiring_for && owns_machine_supervision()` (with no
    /// binding, the binding-free `owns_agent_wiring() && owns_machine_supervision()` `release_hub` uses).
    pub(crate) owns: bool,
    /// `cline_hub::manages_hub` (unchanged gate); with no binding, `manages_hub_without_binding` (`release_hub`'s).
    pub(crate) hub_managed: bool,
    /// `cline_app::visible_app()`: `Some(None)` = not detectable on this platform (Linux, A-8). With no binding,
    /// `Some(cline_app::detect())` — what `release_hub` releases against.
    pub(crate) app: Option<Option<ClineApp>>,
    /// Steps 3-4, per detected host.
    pub(crate) assessments: Vec<HostAssessment>,
    pub(crate) capture: CaptureState,
}

/// Where the hub step reaches the process table.
pub(crate) struct HubSeam<'a> {
    pub(crate) table: &'a dyn ProcessTable,
    pub(crate) port: u16,
    pub(crate) recheck: Duration,
    pub(crate) poll: Duration,
}

/// What one tick did, for the log and the tests.
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct TickReport {
    pub(crate) state: DeliveryState,
    /// None when the hub step did not run (not managed, skipped, no app, or a release).
    pub(crate) hub_action: Option<HubAction>,
    /// What the hub step (or, with no binding, the release) achieved; None when neither ran.
    pub(crate) hub_outcome: Option<HubOutcome>,
    /// None when the login env was not touched (an isolated instance).
    pub(crate) login_env: Option<LoginEnvOutcome>,
}

/// [`tick_from`] seeded from the state file, as a fresh task's first tick is.
#[cfg(test)]
pub(crate) fn tick_with(
    sink: &dyn Sink,
    hub: &HubSeam<'_>,
    ol_dir: &Path,
    inputs: TickInputs,
    now_ms: i64,
) -> TickReport {
    tick_from(sink, hub, ol_dir, inputs, now_ms, &mut read_state(ol_dir))
}

/// Plan 02 §3 steps 0-8 over `inputs`. `previous` is the state file's content (None = no file); the file is
/// written, and `previous` updated, only when the state changed (never at step 0).
fn tick_from(
    sink: &dyn Sink,
    hub: &HubSeam<'_>,
    ol_dir: &Path,
    inputs: TickInputs,
    now_ms: i64,
    previous: &mut Option<DeliveryState>,
) -> TickReport {
    // 0. No Cline agent: our login values must not outlive it (step 8), and neither does anything of ours on the
    //    hub — released with the empty Desired under `release_hub`'s binding-free gate, this tick.
    if !inputs.binding {
        let login_env = inputs.owns.then(|| {
            let mut ledger = Ledger::load(ol_dir);
            login_env::reconcile_login_env(sink, &mut ledger, None)
        });
        let hub_outcome = inputs.hub_managed.then(|| {
            cline_hub::release_with(
                hub.table,
                inputs.app.as_ref().and_then(Option::as_ref),
                hub.port,
                hub.recheck,
                hub.poll,
            )
        });
        return TickReport {
            state: DeliveryState::default(),
            hub_action: None,
            hub_outcome,
            login_env,
        };
    }

    let empty = DeliveryState::default();
    let before = previous.as_ref().unwrap_or(&empty);
    let find = |class: HostClass| inputs.assessments.iter().find(|a| a.host.class == class);

    // 5. The hub.
    let mut hub_action = None;
    let mut hub_outcome = None;
    let mut hub_desired = None;
    let hub_entry = match (&inputs.app, find(HostClass::Hub)) {
        // A-8: Cline.app cannot be seen on this platform — a row for doctor to render.
        (Some(None), _) => Some(HostDelivery {
            reason: Some(reason::NOT_DETECTED_ON_PLATFORM.to_string()),
            ..HostDelivery::default()
        }),
        (_, None) => None,
        (_, Some(a)) if !inputs.hub_managed => Some(entry(a, Some(reason::NOT_MANAGED.into()))),
        // The relay has not reported yet: the hub is not touched this tick.
        (_, Some(a)) if inputs.capture.is_none() => {
            hub_desired = before.hub_desired.clone();
            Some(before.hub.clone().unwrap_or_else(|| {
                entry(
                    a,
                    Some(format!(
                        "{}: waiting for the model relay's first report",
                        reason::PENDING_RESTART
                    )),
                )
            }))
        }
        (app, Some(a)) => {
            let desired = Desired {
                capture: inputs.capture.clone().flatten(),
                plugin: a
                    .deliverable()
                    .then(|| a.wrapper.clone())
                    .flatten()
                    .map(|runtime| PluginEnv {
                        wrapper_path: crate::hooks::cline_bootstrap::wrapper_path(ol_dir),
                        runtime,
                    }),
            };
            let outcome = match app {
                Some(Some(app)) => {
                    let (action, outcome) = cline_hub::reconcile(
                        hub.table,
                        app,
                        hub.port,
                        &desired,
                        hub.recheck,
                        hub.poll,
                    );
                    hub_action = Some(action);
                    Some(outcome)
                }
                _ => None,
            };
            // Delivered only when the hub now carries the desired env; a host not deliverable keeps its own reason.
            let why = a.reason.clone().or_else(|| match &outcome {
                Some(o) if o.carries_desired() => None,
                Some(HubOutcome::DeferredBusy | HubOutcome::IdleUnknown) => {
                    Some(reason::PENDING_RESTART.to_string())
                }
                Some(HubOutcome::Failed(detail)) => {
                    Some(format!("{}: hub: {detail}", reason::PROBE_FAILED))
                }
                Some(_) => Some(format!(
                    "{}: hub: the port is held by a process OpenLatch does not manage",
                    reason::NOT_MANAGED
                )),
                None => Some(format!(
                    "{}: hub: Cline.app not found",
                    reason::PROBE_FAILED
                )),
            });
            hub_outcome = outcome;
            hub_desired = Some(desired);
            Some(entry(a, why))
        }
    };

    // 6. VS Code, through the machine-global login env: only the owning instance ever reads it.
    let vscode = find(HostClass::VsCode);
    let (vscode_entry, login_env) = if !inputs.owns {
        (
            vscode.map(|a| entry(a, Some(reason::NOT_MANAGED.into()))),
            None,
        )
    } else {
        let wrapper_path = crate::hooks::cline_bootstrap::wrapper_path(ol_dir);
        let want = vscode
            .filter(|a| a.deliverable())
            .and_then(|a| a.wrapper.as_ref())
            .and_then(|runtime| Some((wrapper_path.to_str()?, runtime.to_str()?)));
        let mut ledger = Ledger::load(ol_dir);
        let outcome = login_env::reconcile_login_env(sink, &mut ledger, want);
        let entry = vscode.map(|a| {
            let mut e = entry(
                a,
                match &outcome {
                    LoginEnvOutcome::Delivered | LoginEnvOutcome::Released => a.reason.clone(),
                    LoginEnvOutcome::CustomerOverride { names } => Some(format!(
                        "{}: {}",
                        reason::CUSTOMER_OVERRIDE,
                        names.join(", ")
                    )),
                    LoginEnvOutcome::Failed { detail } => {
                        Some(format!("{}: login-env: {detail}", reason::PROBE_FAILED))
                    }
                },
            );
            e.customer_override = matches!(outcome, LoginEnvOutcome::CustomerOverride { .. });
            // A host whose values did not both read back is not delivered, whatever it was assessed as.
            if e.reason.is_none() && outcome != LoginEnvOutcome::Delivered {
                e.reason = Some(format!("{}: login-env", reason::PROBE_FAILED));
            }
            e
        });
        (entry, Some(outcome))
    };

    // 7. `delivered` = deliverable and nothing above said otherwise; `since` survives while delivered.
    let stamp = |e: Option<HostDelivery>, before: &Option<HostDelivery>| {
        e.map(|mut e| {
            e.delivered = e.reason.is_none();
            e.since = if e.delivered {
                before
                    .as_ref()
                    .filter(|b| b.delivered)
                    .and_then(|b| b.since)
                    .or(Some(now_ms))
            } else {
                None
            };
            e
        })
    };
    let state = DeliveryState {
        hub: stamp(hub_entry, &before.hub),
        vscode: stamp(vscode_entry, &before.vscode),
        hub_desired,
    };
    if previous.as_ref() != Some(&state) && write_state(ol_dir, &state) {
        *previous = Some(state.clone());
    }
    TickReport {
        state,
        hub_action,
        hub_outcome,
        login_env,
    }
}

/// A host's entry before `delivered`/`since` are stamped.
fn entry(a: &HostAssessment, reason: Option<String>) -> HostDelivery {
    HostDelivery {
        delivered: false,
        since: None,
        reason,
        core: a.host.core_version.as_ref().map(ToString::to_string),
        customer_override: false,
    }
}

/// Reads the world for one tick (blocking: host detection, the tree, the cached probe).
fn gather(config: &Config, ol_dir: &Path, capture: CaptureState) -> TickInputs {
    let owns_machine = crate::supervision::owns_machine_supervision();
    let cline = crate::hooks::detect_agents()
        .into_iter()
        .find(|a| a.agent_type() == ClineBinding::AGENT_TYPE);
    match cline {
        None => {
            let hub_managed = cline_hub::manages_hub_without_binding(config);
            TickInputs {
                binding: false,
                owns: config.model_relay.owns_agent_wiring() && owns_machine,
                hub_managed,
                app: hub_managed.then(cline_app::detect),
                assessments: Vec::new(),
                capture,
            }
        }
        Some(a) => {
            // Detected once: the hub step and host detection see the same Cline.app.
            let detected = cline_app::detect();
            TickInputs {
                binding: true,
                owns: super::owns_wiring_for(config, &*a.binding) && owns_machine,
                hub_managed: cline_hub::manages_hub(config, &*a.binding),
                app: cline_app::visible_app_from(detected.clone()),
                assessments: cline_runtime::assess_hosts_with(
                    ol_dir,
                    config.cline.plugin_delivery,
                    detected,
                ),
                capture,
            }
        }
    }
}

/// Serialises a tick against the teardown's release: a tick in flight finishes before `release_on_stop` runs, and
/// none starts after it.
static TICK_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
static STOPPED: AtomicBool = AtomicBool::new(false);

fn tick(
    config: &Config,
    ol_dir: &Path,
    capture: CaptureState,
    sink: &dyn Sink,
    previous: &mut Option<DeliveryState>,
) {
    let _held = TICK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    if STOPPED.load(Ordering::SeqCst) {
        return;
    }
    let inputs = gather(config, ol_dir, capture);
    let table = cline_app::process::table();
    let seam = HubSeam {
        table: &*table,
        port: cline_app::hub_port(),
        recheck: cline_app::IDLE_RECHECK,
        poll: cline_app::HUB_STARTUP_POLL,
    };
    let report = tick_from(sink, &seam, ol_dir, inputs, now_ms(), previous);
    tracing::debug!(
        hub = ?report.state.hub,
        vscode = ?report.state.vscode,
        hub_action = ?report.hub_action,
        hub_outcome = ?report.hub_outcome,
        login_env = ?report.login_env,
        "Cline plugin delivery tick"
    );
}

fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
}

/// The graceful stop's release (plan 02 §4.4): no tick runs after it, and one in flight finishes first. Then the
/// hub is restarted without capture and without plugin (`release_hub`). Blocking.
pub(crate) fn release_on_stop(config: &Config) {
    STOPPED.store(true, Ordering::SeqCst);
    let _held = TICK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    cline_hub::release_hub(config);
}

/// One run of the task: tick at once, then on every nudge or [`TICK`], until shutdown. A fresh future per call (the
/// supervised-factory rule): everything it holds is a clone.
async fn run(
    config: Arc<Config>,
    mut capture: watch::Receiver<CaptureState>,
    nudge: Arc<Notify>,
    sink: &'static dyn Sink,
    mut shutdown: watch::Receiver<bool>,
) {
    let ol_dir = crate::config::openlatch_dir();
    // The last state written, seeded from the file once per run; behind a lock so a panicking tick keeps it.
    let previous = Arc::new(std::sync::Mutex::new(read_state(&ol_dir)));
    loop {
        if *shutdown.borrow() {
            return;
        }
        let state = capture.borrow_and_update().clone();
        let cfg = config.clone();
        let dir = ol_dir.clone();
        let prev = previous.clone();
        let ticked = tokio::task::spawn_blocking(move || {
            let mut prev = prev.lock().unwrap_or_else(|e| e.into_inner());
            tick(&cfg, &dir, state, sink, &mut prev);
        });
        if let Err(e) = ticked.await {
            tracing::warn!(error = %e, "Cline plugin delivery tick did not complete");
        }
        tokio::select! {
            _ = shutdown.wait_for(|stop| *stop) => return,
            _ = tokio::time::sleep(TICK) => {}
            _ = nudge.notified() => {}
        }
    }
}

/// Spawns the supervised `cline-plugin-delivery` task. `daemon/mod.rs` calls it only when `reconcile_wiring`, and
/// passes `login_env::os_sink()`.
pub(crate) fn spawn(
    health: &Arc<HealthRegistry>,
    config: Arc<Config>,
    watch: CaptureWatch,
    sink: &'static dyn Sink,
    shutdown: watch::Receiver<bool>,
) -> tokio::task::JoinHandle<()> {
    let task_shutdown = shutdown.clone();
    spawn_supervised(
        health,
        TaskSpec::new(
            super::subsystem::CLINE_PLUGIN_DELIVERY,
            RestartPolicy::Always,
        ),
        shutdown,
        move || {
            run(
                config.clone(),
                watch.rx.clone(),
                watch.nudge.clone(),
                sink,
                task_shutdown.clone(),
            )
        },
    )
}

#[cfg(test)]
mod tests {
    use std::path::{Path, PathBuf};
    use std::time::Duration;

    use super::*;
    use crate::core::login_env::test_support::{Call, RecordingSink};
    use crate::core::login_env::ALLOWED;
    use crate::hooks::cline_app::process::test_support::FakeTable;
    use crate::hooks::cline_app::test_fixture::{
        clean_env, hub_fx, listen, record_exists, seed_record_for, HubFx,
    };
    use crate::hooks::cline_hosts::Host;
    use crate::hooks::cline_runtime::ProbeResult;

    const RELAY: &str = "http://127.0.0.1:7600";

    fn host(class: HostClass, runtime: PathBuf) -> Host {
        Host {
            class,
            runtime: runtime.clone(),
            runtime_var: if class == HostClass::Hub {
                "BUN_BE_BUN"
            } else {
                "ELECTRON_RUN_AS_NODE"
            },
            core_version: Some(semver::Version::new(0, 0, 86)),
            source: runtime,
        }
    }

    /// A host assessed deliverable, its wrapper at the real wrapper path (no file is needed: nothing execs it).
    fn deliverable(ol: &Path, class: HostClass, runtime: PathBuf) -> HostAssessment {
        HostAssessment {
            host: host(class, runtime),
            wrapper: Some(cline_runtime::wrapper_for(ol, class)),
            reason: None,
        }
    }

    fn with_reason(mut a: HostAssessment, reason: &str) -> HostAssessment {
        a.reason = Some(reason.to_string());
        a
    }

    fn both_deliverable(fx: &HubFx, ol: &Path) -> Vec<HostAssessment> {
        vec![
            deliverable(ol, HostClass::Hub, fx.app.sidecar.clone()),
            deliverable(ol, HostClass::VsCode, fx.root.path().join("Code")),
        ]
    }

    fn inputs(fx: &HubFx, assessments: Vec<HostAssessment>, capture: CaptureState) -> TickInputs {
        TickInputs {
            binding: true,
            owns: true,
            hub_managed: true,
            app: Some(Some(fx.app.clone())),
            assessments,
            capture,
        }
    }

    fn run(
        sink: &RecordingSink,
        t: &FakeTable,
        fx: &HubFx,
        ol: &Path,
        i: TickInputs,
    ) -> TickReport {
        let seam = HubSeam {
            table: t,
            port: fx.port,
            recheck: Duration::ZERO,
            poll: Duration::ZERO,
        };
        tick_with(sink, &seam, ol, i, 1_000)
    }

    /// The plugin env the hub gets for this OL dir.
    fn plugin(ol: &Path) -> PluginEnv {
        PluginEnv {
            wrapper_path: crate::hooks::cline_bootstrap::wrapper_path(ol),
            runtime: cline_runtime::wrapper_for(ol, HostClass::Hub),
        }
    }

    /// A fake whose next spawn listens as `pid` (a hub-mode process on the fixture's sidecar).
    fn starts_as(fx: &HubFx, t: &FakeTable, pid: u32) {
        *t.on_spawn_listen.lock().unwrap() = Some(pid);
        *t.next_pid.lock().unwrap() = pid;
        t.procs
            .lock()
            .unwrap()
            .insert(pid, fx.hub_info(Some(clean_env())));
    }

    fn login_values(sink: &RecordingSink) -> [Option<String>; 2] {
        [sink.value(ALLOWED[0]), sink.value(ALLOWED[1])]
    }

    fn code(reason: Option<&String>) -> Option<&str> {
        reason.map(|r| r.split(':').next().unwrap_or(r))
    }

    /// §3 step 5, tri-state watch: `None` (the relay is spawned but has not reported) never touches the hub — not
    /// even one whose record carries capture — while the login env is still reconciled on that tick.
    #[test]
    fn unpublished_capture_never_restarts_the_hub() {
        let fx = hub_fx();
        let ol = crate::config::openlatch_dir();
        let t = fx.fake();
        seed_record_for(
            &fx,
            500,
            1,
            "0.0.34",
            &crate::hooks::cline_app::test_fixture::captured(RELAY),
        );
        listen(&t, 500, fx.hub_info(Some(clean_env())));
        *t.clients.lock().unwrap() = Ok(vec![500, 4242]);
        let sink = RecordingSink::default();

        let report = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, both_deliverable(&fx, &ol), None),
        );

        assert_eq!(
            report.hub_action, None,
            "no hub action on an unreported tick"
        );
        assert_eq!(report.hub_outcome, None);
        assert!(t.terminated.lock().unwrap().is_empty());
        assert!(t.spawns.lock().unwrap().is_empty());
        assert!(record_exists(), "the record is untouched");
        assert_eq!(report.login_env, Some(LoginEnvOutcome::Delivered));
        let wrapper = crate::hooks::cline_bootstrap::wrapper_path(&ol);
        let runtime = cline_runtime::wrapper_for(&ol, HostClass::VsCode);
        assert_eq!(
            login_values(&sink),
            [
                Some(wrapper.display().to_string()),
                Some(runtime.display().to_string())
            ]
        );
        let hub = report.state.hub.as_ref().expect("a hub row");
        assert!(!hub.delivered);
        assert_eq!(code(hub.reason.as_ref()), Some(reason::PENDING_RESTART));
        assert!(
            report
                .state
                .vscode
                .as_ref()
                .expect("a vscode row")
                .delivered
        );
        assert_eq!(read_state(&ol), Some(report.state));
    }

    /// R1-25: the relay reports "off" (`Some(None)`) while the hub record carries capture → RestartNow (even busy),
    /// and the respawn still carries the plugin env.
    #[test]
    fn relay_off_takes_capture_off_now_keeping_plugin() {
        let fx = hub_fx();
        let ol = crate::config::openlatch_dir();
        let t = fx.fake();
        seed_record_for(
            &fx,
            500,
            1,
            "0.0.34",
            &Desired {
                capture: crate::hooks::cline_app::test_fixture::captured(RELAY).capture,
                plugin: Some(plugin(&ol)),
            },
        );
        listen(&t, 500, fx.hub_info(Some(clean_env())));
        *t.clients.lock().unwrap() = Ok(vec![500, 4242]);
        starts_as(&fx, &t, 900);
        let sink = RecordingSink::default();

        let report = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, both_deliverable(&fx, &ol), Some(None)),
        );

        assert_eq!(report.hub_action, Some(HubAction::RestartNow));
        assert_eq!(report.hub_outcome, Some(HubOutcome::Restarted));
        assert_eq!(*t.terminated.lock().unwrap(), vec![500]);
        let spawns = t.spawns.lock().unwrap().clone();
        assert_eq!(spawns.len(), 1);
        let want = Desired {
            capture: None,
            plugin: Some(plugin(&ol)),
        };
        assert_eq!(spawns[0].3, cline_app::hub_env(fx.port, &want));
        let names: Vec<&str> = spawns[0].3.iter().map(|(k, _)| k.as_str()).collect();
        assert!(ALLOWED.iter().all(|n| names.contains(n)), "{names:?}");
        assert!(
            !names.iter().any(|n| n.eq_ignore_ascii_case("HTTPS_PROXY")),
            "{names:?}"
        );
        assert_eq!(report.state.hub_desired, Some(want));
        let hub = report.state.hub.as_ref().expect("a hub row");
        assert!(hub.delivered);
        assert_eq!(hub.since, Some(1_000));
    }

    /// §5: the kill switch (every host `delivery-off`) clears our login values and takes the plugin off the hub
    /// only when idle — never NOW — keeping the live capture.
    #[test]
    fn delivery_off_clears_ours_and_restarts_hub_if_idle() {
        let fx = hub_fx();
        let ol = crate::config::openlatch_dir();
        let t = fx.fake();
        let capture = crate::hooks::cline_app::ProxyEnvVars::for_relay(7600);
        let sink = RecordingSink::default();

        // Delivered first: the hub started with capture + plugin, the login values ours.
        starts_as(&fx, &t, 900);
        let first = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, both_deliverable(&fx, &ol), Some(Some(capture.clone()))),
        );
        assert_eq!(first.hub_action, Some(HubAction::Start));
        assert_eq!(first.hub_outcome, Some(HubOutcome::Started));
        assert_eq!(first.login_env, Some(LoginEnvOutcome::Delivered));
        assert!(login_values(&sink).iter().all(Option::is_some));

        // A BUSY hub: the plugin-only change waits.
        *t.clients.lock().unwrap() = Ok(vec![900, 4242]);
        let off = || {
            both_deliverable(&fx, &ol)
                .into_iter()
                .map(|a| with_reason(a, reason::DELIVERY_OFF))
                .collect::<Vec<_>>()
        };
        let busy = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, off(), Some(Some(capture.clone()))),
        );
        assert_eq!(busy.hub_action, Some(HubAction::RestartIfIdle));
        assert_eq!(busy.hub_outcome, Some(HubOutcome::DeferredBusy));
        assert!(t.terminated.lock().unwrap().is_empty(), "never Now");
        assert_eq!(busy.login_env, Some(LoginEnvOutcome::Released));
        assert_eq!(
            login_values(&sink),
            [None, None],
            "our login values are cleared"
        );
        assert!(sink
            .calls()
            .iter()
            .any(|c| matches!(c, Call::Clear(n) if n == ALLOWED[0])));

        // Idle: restarted with the capture and without the plugin.
        *t.clients.lock().unwrap() = Ok(Vec::new());
        starts_as(&fx, &t, 901);
        let idle = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, off(), Some(Some(capture.clone()))),
        );
        assert_eq!(idle.hub_action, Some(HubAction::RestartIfIdle));
        assert_eq!(idle.hub_outcome, Some(HubOutcome::Restarted));
        assert_eq!(*t.terminated.lock().unwrap(), vec![900]);
        let spawns = t.spawns.lock().unwrap().clone();
        assert_eq!(spawns.len(), 2);
        let want = Desired {
            capture: Some(capture.clone()),
            plugin: None,
        };
        assert_eq!(
            spawns[1].3,
            cline_app::hub_env(fx.port, &want),
            "capture untouched"
        );
        assert_eq!(idle.state.hub_desired, Some(want));
        for row in [idle.state.hub.as_ref(), idle.state.vscode.as_ref()] {
            let row = row.expect("a row");
            assert!(!row.delivered);
            assert_eq!(row.reason.as_deref(), Some(reason::DELIVERY_OFF));
        }
        assert!(Ledger::load(&ol).is_empty());
    }

    /// §3 step 8: a runtime binary that disappeared makes the host not deliverable, and the SAME tick removes our
    /// values — the login env cleared, the hub restarted without the plugin.
    #[test]
    fn delivery_tick_removes_values_when_runtime_disappears() {
        let fx = hub_fx();
        let ol = crate::config::openlatch_dir();
        let t = fx.fake();
        let sink = RecordingSink::default();
        // The runtimes step 4 checks. The hub's is its own file here, so the fixture's app (whose sidecar
        // `detect` needs) stays visible and the hub step still runs on the tick that sees the runtime gone.
        let code_bin = fx.root.path().join("Code");
        let hub_bin = fx.root.path().join("hub-runtime");
        std::fs::write(&code_bin, "").expect("a fake VS Code runtime");
        std::fs::write(&hub_bin, "").expect("a fake hub runtime");

        // Delivered: relay off, the hub started with the plugin only.
        starts_as(&fx, &t, 900);
        let first = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(
                &fx,
                vec![
                    deliverable(&ol, HostClass::Hub, hub_bin.clone()),
                    deliverable(&ol, HostClass::VsCode, code_bin.clone()),
                ],
                Some(None),
            ),
        );
        assert_eq!(first.hub_action, Some(HubAction::Start));
        assert_eq!(first.login_env, Some(LoginEnvOutcome::Delivered));
        assert!(first.state.hub.as_ref().unwrap().delivered);
        assert!(first.state.vscode.as_ref().unwrap().delivered);

        // Both runtimes are gone: assessed as the real step 4 does.
        std::fs::remove_file(&code_bin).expect("remove the VS Code runtime");
        std::fs::remove_file(&hub_bin).expect("remove the hub runtime");
        let probe_ran = std::sync::atomic::AtomicBool::new(false);
        let probe = |_: &Path, _: &Host| {
            probe_ran.store(true, Ordering::SeqCst);
            ProbeResult {
                ok: true,
                stage: None,
                error: None,
            }
        };
        let assessments: Vec<HostAssessment> = [
            host(HostClass::Hub, hub_bin.clone()),
            host(HostClass::VsCode, code_bin.clone()),
        ]
        .iter()
        .map(|h| cline_runtime::assess_with(&ol, h, true, &Ok(()), &probe))
        .collect();
        assert!(
            !probe_ran.load(Ordering::SeqCst),
            "no probe past a missing runtime"
        );
        for a in &assessments {
            assert_eq!(a.reason.as_deref(), Some(reason::RUNTIME_MISSING), "{a:?}");
        }

        let second = run(&sink, &t, &fx, &ol, inputs(&fx, assessments, Some(None)));
        assert_eq!(second.login_env, Some(LoginEnvOutcome::Released));
        assert_eq!(login_values(&sink), [None, None]);
        assert!(Ledger::load(&ol).is_empty());
        // The hub carried only our plugin; idle → stopped, and nothing is respawned for an empty Desired.
        assert_eq!(second.hub_action, Some(HubAction::RestartIfIdle));
        assert_eq!(second.hub_outcome, Some(HubOutcome::Restarted));
        assert_eq!(*t.terminated.lock().unwrap(), vec![900]);
        assert_eq!(t.spawns.lock().unwrap().len(), 1, "only the first start");
        assert!(!record_exists());
        assert_eq!(second.state.hub_desired, Some(Desired::default()));
        for row in [second.state.hub.as_ref(), second.state.vscode.as_ref()] {
            let row = row.expect("a row");
            assert!(!row.delivered);
            assert_eq!(row.since, None);
            assert_eq!(row.reason.as_deref(), Some(reason::RUNTIME_MISSING));
        }
    }

    /// Review fix 5: the hub row is delivered only once the hub CARRIES the desired env. A restart deferred on a busy
    /// hub is `pending-restart` with no `since`; the idle tick that restarts it starts `since`.
    #[test]
    fn hub_is_delivered_only_once_it_carries_the_env() {
        let fx = hub_fx();
        let ol = crate::config::openlatch_dir();
        let t = fx.fake();
        // Cline.app's own hub, nothing of ours on it, two clients attached.
        listen(&t, 500, fx.hub_info(Some(clean_env())));
        *t.clients.lock().unwrap() = Ok(vec![500, 4242]);
        let sink = RecordingSink::default();

        let busy = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, both_deliverable(&fx, &ol), Some(None)),
        );
        assert_eq!(busy.hub_action, Some(HubAction::RestartIfIdle));
        assert_eq!(busy.hub_outcome, Some(HubOutcome::DeferredBusy));
        let hub = busy.state.hub.as_ref().expect("a hub row");
        assert!(!hub.delivered, "a deferred restart is not a delivery");
        assert_eq!(hub.since, None);
        assert_eq!(hub.reason.as_deref(), Some(reason::PENDING_RESTART));
        assert_eq!(read_state(&ol), Some(busy.state.clone()));

        // Idle: restarted with the plugin; delivered from now on.
        *t.clients.lock().unwrap() = Ok(Vec::new());
        starts_as(&fx, &t, 900);
        let idle = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, both_deliverable(&fx, &ol), Some(None)),
        );
        assert_eq!(idle.hub_outcome, Some(HubOutcome::Restarted));
        let hub = idle.state.hub.as_ref().expect("a hub row");
        assert!(hub.delivered);
        assert_eq!(hub.reason, None);
        assert_eq!(hub.since, Some(1_000));
    }

    /// Review fix 4: the Cline binding disappears while our hub carries the plugin — the same tick clears our login
    /// values AND releases the hub with the empty Desired (no plugin, no capture), under `release_hub`'s gate.
    #[test]
    fn binding_gone_releases_the_hub_this_tick() {
        let fx = hub_fx();
        let ol = crate::config::openlatch_dir();
        let t = fx.fake();
        let sink = RecordingSink::default();

        // Delivered first, with a binding.
        starts_as(&fx, &t, 900);
        let first = run(
            &sink,
            &t,
            &fx,
            &ol,
            inputs(&fx, both_deliverable(&fx, &ol), Some(None)),
        );
        assert_eq!(first.hub_outcome, Some(HubOutcome::Started));
        assert!(login_values(&sink).iter().all(Option::is_some));

        // The binding is gone; a client is still attached, so the release respawns a CLEAN hub.
        *t.clients.lock().unwrap() = Ok(vec![900, 4242]);
        let mut gone = inputs(&fx, Vec::new(), Some(None));
        gone.binding = false;
        let report = run(&sink, &t, &fx, &ol, gone);

        assert_eq!(report.login_env, Some(LoginEnvOutcome::Released));
        assert_eq!(login_values(&sink), [None, None]);
        assert_eq!(report.hub_outcome, Some(HubOutcome::Restarted));
        assert_eq!(*t.terminated.lock().unwrap(), vec![900]);
        let spawns = t.spawns.lock().unwrap().clone();
        assert_eq!(spawns.len(), 2, "the first start, then the clean respawn");
        assert_eq!(
            spawns[1].3,
            cline_app::hub_env(fx.port, &Desired::default()),
            "released with the empty Desired"
        );
        assert!(!record_exists());

        // Not managed (the binding-free gate says no): the hub is never reached.
        let t = fx.fake();
        let mut gone = inputs(&fx, Vec::new(), Some(None));
        gone.binding = false;
        gone.hub_managed = false;
        let report = run(&sink, &t, &fx, &ol, gone);
        assert_eq!(report.hub_outcome, None);
        assert!(t.spawns.lock().unwrap().is_empty() && t.terminated.lock().unwrap().is_empty());
    }

    /// §3 step 6: an instance that does not own the machine never reaches the login env — no read, no apply, no
    /// clear — with or without a Cline binding, whatever it would deliver.
    #[test]
    fn isolated_instance_never_touches_login_env() {
        let fx = hub_fx();
        let ol = crate::config::openlatch_dir();
        assert!(!crate::supervision::owns_machine_supervision());
        // A ledger and live values that WOULD be cleared or updated by an owning instance.
        let sink =
            RecordingSink::with(&[(ALLOWED[0], "/old/wrapper"), (ALLOWED[1], "/old/runtime")]);
        let t = fx.fake();

        for (binding, capture) in [(true, Some(None)), (true, None), (false, Some(None))] {
            let mut i = inputs(&fx, both_deliverable(&fx, &ol), capture);
            i.binding = binding;
            i.owns = false;
            i.hub_managed = false;
            let report = run(&sink, &t, &fx, &ol, i);
            assert_eq!(report.login_env, None, "binding={binding}");
            if binding {
                let vscode = report.state.vscode.as_ref().expect("a vscode row");
                assert!(!vscode.delivered);
                assert_eq!(code(vscode.reason.as_ref()), Some(reason::NOT_MANAGED));
            }
        }
        assert_eq!(sink.calls(), Vec::<Call>::new(), "zero calls on the sink");
        assert!(t.spawns.lock().unwrap().is_empty());
        assert!(!Ledger::path_in(&ol).exists());
    }
}