openlatch-client 0.6.19

//! The daemon side of `model_relay::readiness`: every [`READINESS_TICK`], gather each wired
//! agent's facts — its preflight verdict, its delivery read back, its traffic, and its running
//! surfaces from ONE read of the process table — and write `relay-readiness.json` when a verdict
//! changed.
//!
//! Its own task, never a step of the wiring supervisor: that loop ticks once a minute and blocks
//! on probes, while a user who just quit and reopened Cursor, or sent a first prompt, should see
//! `init` and `doctor` move within seconds.

use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use std::time::SystemTime;

use chrono::{DateTime, Utc};

use crate::config::Config;
use crate::hooks::binding::{EndpointConvention, ProxyDelivery};
use crate::hooks::cline_app::process::ExeProcess;
use crate::model_relay::preflight::{Verdict as Path_, WiringState};
use crate::model_relay::readiness::{
    self, Age, AgentReadiness, Facts, Restart, RunningSurface, SurfaceDecl, Traffic,
};

/// Each wired agent's `wired_since`, as the last pass left it, for the hook path: read on a hook
/// event with no file read ([`predating_notice`]).
static WIRED_SINCE: std::sync::RwLock<BTreeMap<String, SystemTime>> =
    std::sync::RwLock::new(BTreeMap::new());

/// What a session that predates its agent's relay wiring is told, once.
pub(crate) const PREDATING_NOTICE: &str = "OpenLatch: this session started before OpenLatch \
     connected its model calls, so they go straight to the provider and are not captured. Start a \
     new session to have them captured.";

/// The hook event a notice reaches the developer on, per agent: Claude Code shows a
/// `systemMessage` on a submitted prompt; Codex shows one only on a `PreToolUse` it allows; Cline's
/// `appendContext` rides a `PreToolUse`. Every other agent has no notice channel.
pub(crate) fn notice_event(source: &str) -> Option<crate::envelope::HookEventType> {
    use crate::envelope::HookEventType;
    match source {
        "claude-code" => Some(HookEventType::UserPromptSubmit),
        "codex-cli" | "cline" => Some(HookEventType::PreToolUse),
        _ => None,
    }
}

/// The notice for `session` of `source`, when this event is the one its agent shows notices on
/// and the session was live before the agent's current relay wiring. At most once per session.
pub(crate) fn predating_notice(
    registry: &crate::model_relay::session::SessionRegistry,
    install_id: &str,
    source: &str,
    session: &str,
    event: &crate::envelope::HookEventType,
) -> Option<&'static str> {
    if notice_event(source).as_ref() != Some(event) {
        return None;
    }
    let wired_since = WIRED_SINCE
        .read()
        .ok()
        .and_then(|m| m.get(source).copied())?;
    registry
        .take_predating_notice(install_id, source, session, wired_since)
        .then_some(PREDATING_NOTICE)
}

/// How often the verdicts are recomputed. One process-table read per pass — a PowerShell run on
/// Windows — so not faster.
const READINESS_TICK: std::time::Duration = std::time::Duration::from_secs(20);

/// The tick for the daemon's first [`STARTUP_WINDOW`].
const STARTUP_TICK: std::time::Duration = std::time::Duration::from_secs(2);

/// How long after the daemon starts the readiness pass runs every [`STARTUP_TICK`].
const STARTUP_WINDOW: std::time::Duration = std::time::Duration::from_secs(90);

/// Spawn the readiness task. Stops with the daemon.
pub(super) fn spawn(
    config: Arc<Config>,
    port: u16,
    wiring: Arc<WiringState>,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    let ol_dir = crate::config::openlatch_dir();
    // Read before the wiring supervisor's first write, which is what makes it a fact about the
    // install as it was when this daemon started.
    let untracked = untracked_upgrade(&ol_dir);
    tokio::spawn(async move {
        let mut ledger = readiness::read_traffic(&ol_dir);
        let started = std::time::Instant::now();
        let mut delay = std::time::Duration::ZERO;
        loop {
            tokio::select! {
                _ = tokio::time::sleep(delay) => {}
                _ = shutdown.changed() => return,
            }
            // Fast while the daemon is young: `init` is waiting on the first verdicts, and the
            // preflight lands a few seconds after the start.
            delay = if started.elapsed() < STARTUP_WINDOW {
                STARTUP_TICK
            } else {
                READINESS_TICK
            };
            let (config, wiring, dir, prior) = (
                config.clone(),
                wiring.clone(),
                ol_dir.clone(),
                ledger.clone(),
            );
            // The process table and every settings file are blocking reads.
            let pass = tokio::task::spawn_blocking(move || {
                let mut ledger = prior;
                let agents = pass(&config, port, &wiring, &dir, &mut ledger, untracked);
                readiness::write_if_changed(&dir, agents);
                ledger
            })
            .await;
            match pass {
                Ok(next) => {
                    if let Ok(mut m) = WIRED_SINCE.write() {
                        *m = next
                            .iter()
                            .map(|(a, t)| (a.clone(), SystemTime::from(t.wired_since)))
                            .collect();
                    }
                    if next != ledger {
                        if let Err(e) = readiness::write_traffic(&ol_dir, &next) {
                            tracing::warn!(error = %e, "could not write the relay traffic ledger");
                        }
                        ledger = next;
                    }
                }
                Err(e) => tracing::warn!(error = %e, "relay readiness pass failed"),
            }
        }
    });
}

/// One pass: every agent the wiring supervisor holds a verdict for → its readiness. Updates
/// `ledger` in place: a wiring first seen gets its `wired_since`, a first call since then its
/// `captured_at`, and a wiring taken back (a failed preflight) loses its record.
fn pass(
    config: &Config,
    port: u16,
    wiring: &WiringState,
    ol_dir: &Path,
    ledger: &mut BTreeMap<String, Traffic>,
    untracked: bool,
) -> BTreeMap<String, AgentReadiness> {
    let now = SystemTime::now();
    let traffic = wiring.traffic();
    let agents: Vec<_> = crate::hooks::detect_agents()
        .into_iter()
        .filter(|a| a.binding.model_relay_wiring().is_some())
        .collect();

    // ONE read of the process table for every declared surface of every agent.
    let decls: Vec<(&'static str, SurfaceDecl)> = agents
        .iter()
        .filter(|a| wiring.is_wired(a.agent_type()))
        .flat_map(|a| {
            let agent = a.agent_type();
            a.binding
                .relay_surfaces()
                .into_iter()
                .map(move |d| (agent, d))
        })
        .collect();
    let table = list_processes(&decls);

    let mut out = BTreeMap::new();
    for a in &agents {
        let agent = a.agent_type();
        let path = wiring.verdict(agent);
        let (wired_since, captured_at) = match &path {
            // A wiring taken back loses its record; the failure is judged from `path` alone.
            Path_::Failed(_) => {
                ledger.remove(agent);
                (DateTime::<Utc>::from(now), None)
            }
            Path_::Ok if wiring.is_wired(agent) => {
                let record = record_for(ledger, agent, || {
                    first_wiring_time(untracked, wiring.wired_at(agent).unwrap_or(now))
                });
                if record.captured_at.is_none() {
                    record.captured_at = traffic
                        .get(agent)
                        .copied()
                        .map(DateTime::<Utc>::from)
                        .filter(|t| *t >= record.wired_since);
                }
                (record.wired_since, record.captured_at)
            }
            // Not answered yet, or proven and not yet written: no verdict, `init` keeps waiting.
            _ => continue,
        };
        let mut running = table
            .as_ref()
            .map(|rows| surfaces_of(agent, &decls, rows))
            .map_err(Clone::clone);
        if let Ok(r) = running.as_mut() {
            r.extend(hub_surfaces(config, &*a.binding));
        }
        let facts = Facts {
            path,
            delivery: delivery(&*a.binding, port, ol_dir),
            wired_since: wired_since.into(),
            captured_at: captured_at.map(Into::into),
            running,
        };
        if let Some(verdict) = readiness::judge(&facts) {
            out.insert(
                agent.to_string(),
                AgentReadiness {
                    verdict,
                    wired_since,
                },
            );
        }
    }
    out
}

/// The traffic record of a wired agent whose preflight passed, created on first sight with the
/// time it went wired in this process. A record from before a daemon restart is kept: the agent's
/// processes started since then read the same wiring.
fn record_for<'a>(
    ledger: &'a mut BTreeMap<String, Traffic>,
    agent: &str,
    wired_at: impl FnOnce() -> SystemTime,
) -> &'a mut Traffic {
    ledger.entry(agent.to_string()).or_insert_with(|| Traffic {
        captured_at: None,
        wired_since: DateTime::<Utc>::from(wired_at()),
    })
}

/// Whether this daemon starts on an install that wired agents before any readiness was
/// recorded — an upgrade to this build. Its open sessions and apps read the wiring an earlier
/// daemon wrote, and nothing dates it, so they are taken as current rather than asked to restart.
/// A first install has no endpoint ledger yet; a later start has the traffic ledger.
fn untracked_upgrade(ol_dir: &Path) -> bool {
    !readiness::traffic_path(ol_dir).exists() && ol_dir.join("model-relay-endpoints.json").exists()
}

/// When an agent's wiring counts from, on first sight: the moment this daemon wired it, or — on an
/// [`untracked_upgrade`] — before any process now running, since an earlier daemon wired it at an
/// unknown time.
fn first_wiring_time(untracked: bool, wired_at: SystemTime) -> SystemTime {
    if untracked {
        SystemTime::UNIX_EPOCH
    } else {
        wired_at
    }
}

/// Every running process under any declared root. `Err` = the table could not be read.
fn list_processes(decls: &[(&'static str, SurfaceDecl)]) -> Result<Vec<ExeProcess>, String> {
    let roots: Vec<std::path::PathBuf> = decls
        .iter()
        .flat_map(|(_, d)| d.roots.iter().cloned())
        .collect();
    crate::hooks::cline_app::process::processes_under(&roots)
}

/// `agent`'s running surfaces among `rows`.
fn surfaces_of(
    agent: &str,
    decls: &[(&'static str, SurfaceDecl)],
    rows: &[ExeProcess],
) -> Vec<RunningSurface> {
    let key = crate::core::path_compat::dedup_key;
    decls
        .iter()
        .filter(|(a, _)| *a == agent)
        .flat_map(|(_, d)| {
            let roots: Vec<_> = d.roots.iter().map(|r| key(r)).collect();
            rows.iter()
                .filter(move |p| roots.iter().any(|r| key(&p.exe).starts_with(r)))
                .filter(|p| (d.counts)(p))
                .filter_map(move |p| {
                    p.started.map(|t| RunningSurface {
                        name: d.name,
                        age: Age::Started(t),
                        picks_up_live: d.picks_up_live,
                        restart: d.restart,
                    })
                })
        })
        .collect()
}

/// Cline.app's hub, from the finding the delivery task's own classification gives: a hub running
/// without capture predates it and OpenLatch restarts it on a yes; a Cline.app that re-spawned its
/// own hub is quit and reopened by the user.
fn hub_surfaces(
    config: &Config,
    binding: &dyn crate::hooks::binding::AgentBinding,
) -> Vec<RunningSurface> {
    use crate::hooks::cline_app::HubFinding;

    let agent = binding.agent_type();
    if agent != crate::hooks::bindings::cline::ClineBinding::AGENT_TYPE
        || !super::cline_hub::manages_hub(config, binding)
    {
        return Vec::new();
    }
    let finding = crate::hooks::cline_app::hub_finding(
        crate::hooks::proxy_env_file::live_proxy_url(agent).is_some(),
        &super::cline_delivery::read_hub_desired(&crate::config::openlatch_dir()),
    );
    let (name, restart) = match finding {
        Some(HubFinding::WaitingForRestart { .. } | HubFinding::OlderApp { .. }) => {
            ("Cline.app hub", Restart::ByOpenLatch)
        }
        Some(HubFinding::AppKeptItsOwn { .. }) => ("Cline.app", Restart::QuitAndReopen),
        _ => return Vec::new(),
    };
    vec![RunningSurface {
        name,
        age: Age::Predates,
        picks_up_live: false,
        restart,
    }]
}

/// The delivery read back, and the agent's own runtime trusting the relay. A base-URL agent's
/// delivery is re-proven on disk by the wiring supervisor on every pass (`hooks::wiring_on_disk`)
/// and needs no TLS trust — its loopback is plain HTTP — so only an intercepted agent has anything
/// to check here.
fn delivery(
    binding: &dyn crate::hooks::binding::AgentBinding,
    port: u16,
    ol_dir: &Path,
) -> Result<(), (String, String)> {
    let Some(EndpointConvention::ProxyEnv { delivery, .. }) =
        binding.model_relay_wiring().map(|w| w.endpoint)
    else {
        return Ok(());
    };
    let agent = binding.agent_type();
    let name = binding.display_name();
    for d in delivery {
        match d {
            ProxyDelivery::SettingsKey { file, key, .. } => {
                let Some(file) = file() else { continue };
                use crate::hooks::proxy_settings::{observe_key, Observed};
                match observe_key(&file, key, agent) {
                    Ok(Observed::Foreign(_)) => {
                        return Err((
                            format!(
                                "`{key}` in {} names another proxy, so {name} goes direct",
                                file.display()
                            ),
                            format!(
                                "OpenLatch never overwrites a proxy you set. Remove `{key}` to \
                                 route {name} through OpenLatch."
                            ),
                        ))
                    }
                    Ok(Observed::Unparseable(_)) => {
                        return Err((
                            format!(
                                "{} cannot be read safely, so `{key}` is not set",
                                file.display()
                            ),
                            "Fix the file's syntax, then run `openlatch doctor --fix`.".into(),
                        ))
                    }
                    _ => {}
                }
            }
            ProxyDelivery::EnvFile { .. } => {
                let ours = format!("http://127.0.0.1:{port}");
                if crate::hooks::proxy_env_file::live_proxy_url(agent).as_deref() != Some(&ours) {
                    return Err((
                        format!("{name}'s launcher environment is not in place"),
                        "Run `openlatch doctor --fix`.".into(),
                    ));
                }
                // The wrapper names `ca.pem` in `NODE_EXTRA_CA_CERTS`: a Node or Bun runtime
                // trusts the relay only if that file holds a certificate.
                let ca = crate::model_relay::ca::ca_dir(ol_dir);
                if crate::model_relay::ca::inspect(&ca).is_none() {
                    return Err((
                        format!(
                            "{} holds no certificate, so {name} would refuse the relay",
                            crate::model_relay::ca::ca_pem_path(&ca).display()
                        ),
                        "Run `openlatch doctor --fix`.".into(),
                    ));
                }
            }
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::PathBuf;
    use std::time::Duration;

    fn t(secs: u64) -> SystemTime {
        SystemTime::UNIX_EPOCH + Duration::from_secs(1_800_000_000 + secs)
    }

    fn decl(name: &'static str, root: &str, restart: Restart) -> SurfaceDecl {
        SurfaceDecl {
            name,
            roots: vec![PathBuf::from(root)],
            counts: |_| true,
            picks_up_live: false,
            restart,
        }
    }

    fn row(exe: &str, started: u64) -> ExeProcess {
        ExeProcess {
            exe: PathBuf::from(exe),
            args: vec![exe.to_string()],
            started: Some(t(started)),
        }
    }

    #[test]
    fn surfaces_are_assigned_to_their_own_agent_only() {
        let decls = vec![
            (
                "cursor",
                decl("Cursor", "/Applications/Cursor.app", Restart::QuitAndReopen),
            ),
            (
                "claude-code",
                decl(
                    "Claude Code session",
                    "/h/.local/share/claude",
                    Restart::NewSession,
                ),
            ),
        ];
        let rows = vec![
            row("/Applications/Cursor.app/Contents/MacOS/Cursor", 10),
            row("/h/.local/share/claude/versions/2.1.0", 20),
            row("/usr/bin/vim", 30),
        ];
        let cursor = surfaces_of("cursor", &decls, &rows);
        assert_eq!(cursor.len(), 1);
        assert_eq!(cursor[0].name, "Cursor");
        let claude = surfaces_of("claude-code", &decls, &rows);
        assert_eq!(claude.len(), 1);
        assert_eq!(claude[0].age, Age::Started(t(20)));
        assert!(surfaces_of("codex-cli", &decls, &rows).is_empty());
    }

    #[test]
    fn a_process_with_no_start_time_is_never_judged() {
        let decls = vec![("cursor", decl("Cursor", "/A", Restart::QuitAndReopen))];
        let rows = vec![ExeProcess {
            exe: PathBuf::from("/A/x"),
            args: vec![],
            started: None,
        }];
        assert!(surfaces_of("cursor", &decls, &rows).is_empty());
    }

    #[test]
    fn a_new_wiring_gets_a_record_and_an_old_one_keeps_it() {
        let mut ledger = BTreeMap::new();
        let first = record_for(&mut ledger, "cursor", || t(100)).wired_since;
        assert_eq!(first, DateTime::<Utc>::from(t(100)));
        // A daemon restart re-wires at a later time; the record from before is kept.
        let again = record_for(&mut ledger, "cursor", || t(500)).wired_since;
        assert_eq!(again, DateTime::<Utc>::from(t(100)));
    }

    #[test]
    fn a_session_live_before_the_wiring_is_told_once_on_its_notice_event() {
        use crate::envelope::HookEventType;
        let reg = crate::model_relay::session::SessionRegistry::default();
        reg.upsert("agt", "agt", "claude-code", "old");
        std::thread::sleep(Duration::from_millis(5));
        WIRED_SINCE
            .write()
            .unwrap()
            .insert("claude-code".into(), SystemTime::now());
        std::thread::sleep(Duration::from_millis(5));
        reg.upsert("agt", "agt", "claude-code", "new");
        let prompt = HookEventType::UserPromptSubmit;
        assert_eq!(
            predating_notice(
                &reg,
                "agt",
                "claude-code",
                "old",
                &HookEventType::PreToolUse
            ),
            None,
            "Claude Code is told on a prompt, never on a tool call"
        );
        assert_eq!(
            predating_notice(&reg, "agt", "claude-code", "old", &prompt),
            Some(PREDATING_NOTICE)
        );
        assert_eq!(
            predating_notice(&reg, "agt", "claude-code", "old", &prompt),
            None,
            "once per session"
        );
        assert_eq!(
            predating_notice(&reg, "agt", "claude-code", "new", &prompt),
            None
        );
        assert_eq!(
            predating_notice(&reg, "agt", "cursor", "old", &prompt),
            None,
            "no notice channel"
        );
    }

    #[test]
    fn an_upgrade_counts_running_processes_as_current_and_a_first_install_does_not() {
        let dir = tempfile::tempdir().unwrap();
        assert!(
            !untracked_upgrade(dir.path()),
            "first install: no endpoint ledger yet"
        );
        std::fs::write(dir.path().join("model-relay-endpoints.json"), "{}").unwrap();
        assert!(
            untracked_upgrade(dir.path()),
            "wired before, never tracked: an upgrade"
        );
        readiness::write_traffic(dir.path(), &BTreeMap::new()).unwrap();
        assert!(
            !untracked_upgrade(dir.path()),
            "tracked since: the ledger dates it"
        );
        assert_eq!(first_wiring_time(true, t(5)), SystemTime::UNIX_EPOCH);
        assert_eq!(first_wiring_time(false, t(5)), t(5));
    }

    #[test]
    fn a_base_url_agent_has_no_delivery_to_check() {
        let b = crate::hooks::bindings::claude_code::ClaudeCodeBinding {
            claude_dir: PathBuf::from("/nonexistent/.claude"),
            settings_path: PathBuf::from("/nonexistent/.claude/settings.json"),
        };
        assert_eq!(delivery(&b, 7600, Path::new("/nonexistent")), Ok(()));
    }
}