openlatch-client 0.6.16

//! Wiring an agent that appears after the daemon started.
//!
//! A customer who installs an agent's desktop app after OpenLatch expects it
//! enforced without running `init` again. Startup detection cannot see it, and
//! the reconciler deliberately never heals an agent it has no hook rows for —
//! so this pass looks again on a timer, and wires an agent that is **new**:
//! detected, installable, and absent from
//! [`crate::hooks::decided_agents`]. One `init --agent` excluded, or one whose
//! hooks were removed by hand, is in that record and is never touched.
//!
//! A wired arrival joins the shared bindings (its ask/hold caps reach the
//! verdict path) and the reconciler's targets (its file is watched and healed
//! from then on). Its model relay wiring is the wiring supervisor's, which
//! re-detects on its own tick.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use tokio::sync::mpsc;

use crate::daemon::reconciler::request::ReconcileRequest;
use crate::daemon::reconciler::AgentTarget;
use crate::daemon::SharedBindings;
use crate::hooks::decided_agents::Decided;
use crate::hooks::DetectedAgent;

/// How often detection runs — the config monitor's inventory cadence, so a new
/// agent is wired about when it is first inventoried.
pub const TICK: Duration = Duration::from_secs(60);

/// How long a failed install waits before it is tried again.
const RETRY_AFTER: Duration = Duration::from_secs(600);

/// The arrivals among `detected`: installable, and not yet decided.
pub(crate) fn arrivals(detected: Vec<DetectedAgent>, decided: &Decided) -> Vec<DetectedAgent> {
    detected
        .into_iter()
        .filter(|a| a.installable() && !decided.contains(a.agent_type()))
        .collect()
}

/// Installs one agent's hooks. A seam so a test does not write a real
/// agent's config or stage a binary.
pub(crate) type Installer = Box<dyn Fn(&DetectedAgent) -> Result<(), crate::error::OlError> + Send>;

/// The production installer: stage the hook binary (already there after any
/// `init`), then [`crate::hooks::install_hooks`] — which, for Codex, also asks
/// Codex to trust the hook.
pub(crate) fn hook_installer(openlatch_dir: PathBuf, port: u16, token: String) -> Installer {
    Box::new(move |agent| {
        crate::hooks::staging::stage_hook_binary(&openlatch_dir)?;
        crate::hooks::install_hooks(&*agent.binding, port, &token).map(|_| ())
    })
}

/// State kept across passes — and across a supervised restart, behind the
/// caller's lock.
pub(crate) struct ArrivalWiring {
    openlatch_dir: PathBuf,
    install: Installer,
    bindings: SharedBindings,
    reconcile_tx: mpsc::Sender<ReconcileRequest>,
    /// Arms the reconciler's watcher on an arrival's files; `None` in a test.
    watch: Option<ArmWatcher>,
    /// Held for the daemon's lifetime: dropping one stops its watcher.
    watchers: Vec<WatchGuard>,
    retry_at: HashMap<&'static str, Instant>,
}

/// A live filesystem watch, kept only to be held.
pub(crate) type WatchGuard = Box<dyn std::any::Any + Send>;

/// Arms a watch on paths, feeding the reconciler's channel.
type ArmWatcher = fn(&[PathBuf], mpsc::Sender<ReconcileRequest>) -> Option<WatchGuard>;

/// The production watcher arm, the same call the daemon makes at start.
pub(crate) fn arm_watcher(
    paths: &[PathBuf],
    tx: mpsc::Sender<ReconcileRequest>,
) -> Option<WatchGuard> {
    match crate::daemon::watcher::spawn_watcher(paths, tx) {
        Ok(w) => Some(Box::new(w)),
        Err(e) => {
            tracing::warn!(error = %e, "filesystem watcher failed — falling back to poll-only");
            None
        }
    }
}

impl ArrivalWiring {
    pub(crate) fn new(
        openlatch_dir: PathBuf,
        install: Installer,
        bindings: SharedBindings,
        reconcile_tx: mpsc::Sender<ReconcileRequest>,
    ) -> Self {
        Self {
            openlatch_dir,
            install,
            bindings,
            reconcile_tx,
            watch: Some(arm_watcher),
            watchers: Vec::new(),
            retry_at: HashMap::new(),
        }
    }

    /// One pass over `detected`. Returns the agent types it wired.
    ///
    /// Blocking: installs write files, and Codex's trust grant starts Codex.
    /// The caller runs it off the async runtime.
    pub(crate) fn pass(&mut self, detected: Vec<DetectedAgent>, now: Instant) -> Vec<&'static str> {
        let decided = match crate::hooks::decided_agents::load(&self.openlatch_dir) {
            Ok(Some(decided)) => decided,
            // A host from before the record: everything it has now was there
            // before this build and is decided as it stands — an agent left
            // unwired stays a `doctor` finding, never a surprise install.
            Ok(None) => {
                if let Err(e) = crate::hooks::decided_agents::record(
                    &self.openlatch_dir,
                    detected.iter().map(DetectedAgent::agent_type),
                ) {
                    tracing::warn!(error = %e.message, "could not seed the decided-agents record");
                }
                return Vec::new();
            }
            // Unreadable is not "nothing decided": wiring on it could wire an
            // agent somebody excluded.
            Err(e) => {
                tracing::warn!(error = %e.message, "decided-agents record unreadable — no agent is wired late");
                return Vec::new();
            }
        };

        let mut wired = Vec::new();
        for agent in arrivals(detected, &decided) {
            let kind = agent.agent_type();
            if self.retry_at.get(kind).is_some_and(|at| now < *at) {
                continue;
            }
            if let Err(e) = (self.install)(&agent) {
                tracing::warn!(
                    agent = kind,
                    code = e.code,
                    error = %e.message,
                    "a newly installed agent could not be wired — retrying later"
                );
                self.retry_at.insert(kind, now + RETRY_AFTER);
                continue;
            }
            self.retry_at.remove(kind);
            if let Err(e) = crate::hooks::decided_agents::record(&self.openlatch_dir, [kind]) {
                // Wired either way; the next pass finds its hooks already in
                // place and re-records.
                tracing::warn!(agent = kind, error = %e.message, "wired, but not recorded as decided");
            }
            tracing::info!(
                agent = kind,
                hooks = %agent.settings_path().display(),
                "wired an agent installed after the daemon started"
            );

            let target = AgentTarget::for_agent(&agent);
            if let Some(watch) = self.watch {
                let mut paths = vec![target.settings_path.clone()];
                paths.extend(target.binding.tamper_watch_paths());
                self.watchers
                    .extend(watch(&paths, self.reconcile_tx.clone()));
            }
            if self
                .reconcile_tx
                .blocking_send(ReconcileRequest::AddTarget(target))
                .is_err()
            {
                tracing::warn!(
                    agent = kind,
                    "reconciler gone — the new agent is not watched"
                );
            }
            self.bindings.add(agent);
            wired.push(kind);
        }
        wired
    }
}

/// The task: a pass now, then every [`TICK`], until shutdown. Parked, never
/// returned, once the install is marked uninstalled — read fresh each tick, as
/// `uninstall` writes it while this daemon may still be up.
pub(crate) async fn run(
    state: Arc<Mutex<ArrivalWiring>>,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    let mut tick = tokio::time::interval(TICK);
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            _ = tick.tick() => {}
            _ = shutdown.wait_for(|stop| *stop) => return,
        }
        let uninstalled = crate::config::Config::load(None, None, false)
            .map(|c| c.supervision.marked_uninstalled())
            .unwrap_or(true);
        if uninstalled {
            continue;
        }
        let state = state.clone();
        let _ = tokio::task::spawn_blocking(move || {
            let detected = crate::hooks::detect_agents();
            let mut guard = match state.lock() {
                Ok(guard) => guard,
                Err(poisoned) => poisoned.into_inner(),
            };
            guard.pass(detected, Instant::now());
        })
        .await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hooks::binding::test_support::FakeBinding;

    fn agent(kind: &'static str, installable: bool, dir: &std::path::Path) -> DetectedAgent {
        DetectedAgent {
            kind: crate::hooks::AgentKind::ClaudeCode,
            binding: Arc::new(FakeBinding {
                agent_type: kind,
                installable,
                config_dir: dir.join(kind),
                ..FakeBinding::default()
            }),
        }
    }

    struct Harness {
        wiring: ArrivalWiring,
        installed: Arc<Mutex<Vec<&'static str>>>,
        rx: mpsc::Receiver<ReconcileRequest>,
        bindings: SharedBindings,
        _dir: tempfile::TempDir,
    }

    fn harness(fail: &'static [&'static str]) -> Harness {
        let dir = tempfile::tempdir().expect("temp dir");
        let installed = Arc::new(Mutex::new(Vec::new()));
        let seen = installed.clone();
        let install: Installer = Box::new(move |a: &DetectedAgent| {
            if fail.contains(&a.agent_type()) {
                return Err(crate::error::OlError::new("OL-1404", "refused"));
            }
            seen.lock().expect("lock").push(a.agent_type());
            Ok(())
        });
        let (tx, rx) = mpsc::channel(8);
        let bindings = SharedBindings::default();
        let mut wiring =
            ArrivalWiring::new(dir.path().to_path_buf(), install, bindings.clone(), tx);
        wiring.watch = None;
        Harness {
            wiring,
            installed,
            rx,
            bindings,
            _dir: dir,
        }
    }

    #[test]
    fn a_host_without_a_record_is_seeded_and_nothing_is_wired() {
        let mut h = harness(&[]);
        let dir = h.wiring.openlatch_dir.clone();
        let wired = h
            .wiring
            .pass(vec![agent("claude-code", true, &dir)], Instant::now());
        assert!(wired.is_empty());
        assert!(h.installed.lock().expect("lock").is_empty());
        let decided = crate::hooks::decided_agents::load(&dir)
            .expect("load")
            .expect("seeded");
        assert!(decided.contains("claude-code"));
    }

    #[test]
    fn only_an_undecided_installable_agent_is_wired() {
        let mut h = harness(&[]);
        let dir = h.wiring.openlatch_dir.clone();
        // `init` saw Claude Code and Cursor (one of them excluded by --agent).
        crate::hooks::decided_agents::record(&dir, ["claude-code", "cursor"]).expect("record");
        let detected = vec![
            agent("claude-code", true, &dir),
            agent("cursor", true, &dir),
            agent("codex-cli", true, &dir),
            agent("cline", false, &dir),
        ];
        let wired = h.wiring.pass(detected, Instant::now());

        assert_eq!(wired, ["codex-cli"]);
        assert_eq!(*h.installed.lock().expect("lock"), ["codex-cli"]);
        let decided = crate::hooks::decided_agents::load(&dir)
            .expect("load")
            .expect("present");
        assert!(decided.contains("codex-cli"));
        assert!(
            !decided.contains("cline"),
            "a non-installable agent is not decided"
        );
        match h.rx.try_recv() {
            Ok(ReconcileRequest::AddTarget(t)) => assert_eq!(t.binding.agent_type(), "codex-cli"),
            other => panic!("expected AddTarget, got {other:?}"),
        }
        let types: Vec<_> = h
            .bindings
            .snapshot()
            .iter()
            .map(DetectedAgent::agent_type)
            .collect();
        assert_eq!(types, ["codex-cli"]);

        // Decided now: the next pass leaves it alone.
        let again = h
            .wiring
            .pass(vec![agent("codex-cli", true, &dir)], Instant::now());
        assert!(again.is_empty());
    }

    #[test]
    fn a_failed_install_is_not_recorded_and_retries_only_after_the_backoff() {
        let mut h = harness(&["codex-cli"]);
        let dir = h.wiring.openlatch_dir.clone();
        crate::hooks::decided_agents::record(&dir, ["claude-code"]).expect("record");
        let start = Instant::now();
        assert!(h
            .wiring
            .pass(vec![agent("codex-cli", true, &dir)], start)
            .is_empty());
        let decided = crate::hooks::decided_agents::load(&dir)
            .expect("load")
            .expect("present");
        assert!(!decided.contains("codex-cli"));
        assert!(h
            .wiring
            .retry_at
            .get("codex-cli")
            .is_some_and(|at| *at > start + TICK));
        assert!(
            h.rx.try_recv().is_err(),
            "nothing to watch for an unwired agent"
        );
    }

    #[test]
    fn an_unreadable_record_wires_nothing() {
        let mut h = harness(&[]);
        let dir = h.wiring.openlatch_dir.clone();
        std::fs::write(crate::hooks::decided_agents::record_path(&dir), "{").expect("seed");
        assert!(h
            .wiring
            .pass(vec![agent("codex-cli", true, &dir)], Instant::now())
            .is_empty());
        assert!(h.installed.lock().expect("lock").is_empty());
    }
}