openlatch-client 0.5.4

//! The daemon's half of per-slot relay wiring: serve every recorded provider
//! slot, then wire, re-apply, follow or hand back each one the agent's files
//! show, and prove the editor uses it.
//!
//! [`crate::hooks::cline_providers::decide`] says what to do with a slot; this
//! does it, in the one order that keeps every failure on the agent's original
//! value:
//!
//! 1. **Serve first.** Every live record is bound before anything is written,
//!    so an editor that read a relay URL at its start never dials a port nobody
//!    serves — not after a daemon restart, not after a crash.
//! 2. **Record before write.** A slot's record (with the prior) exists before
//!    the value that names its port is written, so a crash in between leaves a
//!    record and never an unrecorded relay URL.
//! 3. **Probe before write.** A round trip through the slot's own endpoint to
//!    its own origin; a provider that cannot be reached is not pointed at.
//! 4. **Compare before rename.** A write lands only on the value the pass
//!    observed; an editor save in between wins and the next pass decides again.
//!
//! Nothing here restores on a daemon stop: see
//! [`crate::hooks::provider_endpoints`] for why only `uninstall` and a relay
//! switched off hand the slots back.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex};

use crate::config::Config;
use crate::core::cloud::tamper::{
    FieldDelta, TamperEvent, DETECTION_RELAY_URL_REVERTED, DETECTION_RELAY_WIRING_MISCONFIGURED,
    DETECTION_RELAY_WIRING_PENDING, RELAY_WIRING_HOOK_EVENT,
};
use crate::core::fs_secure::{fingerprint, FileFingerprint};
use crate::error::{
    ERR_MODEL_RELAY_ENDPOINT_PORTS, ERR_MODEL_RELAY_PREFLIGHT_FAILED, ERR_MODEL_RELAY_STATE_FILE,
};
use crate::hooks::atomic::RewriteOutcome;
use crate::hooks::cline_providers::{
    decide, relay_value, DecideCtx, Decision, SlotId, SlotObservation,
};
use crate::hooks::model_relay_endpoints::{
    self as records, EndpointRecord, Proof, ReleasedBy, SlotState, SlotValue,
};
use crate::hooks::provider_endpoints::{names_port, now_unix, ProviderEndpoints};
use crate::model_relay::endpoints::{
    endpoint_port_block, normalize_origin, EndpointListeners, EndpointSpec, Origin, RelayPorts,
};
use crate::model_relay::preflight::{self, EndpointVerdict, WiringState};
use crate::model_relay::session::SessionRegistry;
use crate::model_relay::wire_format::WireFormat;

use super::reconciler::TamperSinks;

/// The shortest gap between two re-applies of one slot.
///
/// A running editor saves its whole in-memory settings on every change, putting
/// the old URL back each time. Re-applying at once on every save would turn
/// every settings click into two writes of a file the editor is also writing;
/// ten seconds lets a burst of saves settle into one re-apply.
const REAPPLY_MIN_GAP: std::time::Duration = std::time::Duration::from_secs(10);

/// The gap once a slot is contested.
const CONTESTED_GAP: std::time::Duration = std::time::Duration::from_secs(60);

/// How many re-applies inside [`CONTESTED_WINDOW`] make a slot contested.
const CONTESTED_REVERTS: usize = 20;

/// The window [`CONTESTED_REVERTS`] is counted over.
const CONTESTED_WINDOW: std::time::Duration = std::time::Duration::from_secs(600);

/// When a slot may be re-applied again: at most once per [`REAPPLY_MIN_GAP`],
/// and once per [`CONTESTED_GAP`] after [`CONTESTED_REVERTS`] reverts inside
/// [`CONTESTED_WINDOW`] — something is fighting the slot, and matching it write
/// for write only churns the developer's file.
#[derive(Debug, Default)]
pub(crate) struct ReapplyGate {
    recent: std::collections::VecDeque<std::time::Instant>,
}

impl ReapplyGate {
    /// Whether a re-apply may happen at `now`.
    pub(crate) fn allows(&self, now: std::time::Instant) -> bool {
        let gap = if self.contested(now) {
            CONTESTED_GAP
        } else {
            REAPPLY_MIN_GAP
        };
        self.recent
            .back()
            .is_none_or(|last| now.saturating_duration_since(*last) >= gap)
    }

    /// Record a re-apply at `now`.
    pub(crate) fn note(&mut self, now: std::time::Instant) {
        self.recent.push_back(now);
        while self
            .recent
            .front()
            .is_some_and(|t| now.saturating_duration_since(*t) > CONTESTED_WINDOW)
        {
            self.recent.pop_front();
        }
    }

    /// Whether the slot has been reverted often enough to be contested.
    pub(crate) fn contested(&self, now: std::time::Instant) -> bool {
        self.recent
            .iter()
            .filter(|t| now.saturating_duration_since(**t) <= CONTESTED_WINDOW)
            .count()
            >= CONTESTED_REVERTS
    }
}

/// Per-slot wiring for every agent that carries provider slots.
pub(crate) struct EndpointWiring {
    listeners: Arc<EndpointListeners>,
    wiring: Arc<WiringState>,
    ports: RelayPorts,
    started_at: u64,
    sinks: TamperSinks,
    /// Each file as it stood right after this process last wrote it.
    ///
    /// Per FILE, never per slot: two slots share a file, and wiring the second
    /// changes the file under the first. A later fingerprint that differs while a
    /// slot's value is still ours means someone ELSE saved a file carrying our
    /// value — the editor, which can only do that once it has loaded it.
    written: Mutex<BTreeMap<std::path::PathBuf, FileFingerprint>>,
    /// Ports another process held when this pass tried to bind them. Skipped by
    /// allocation for the rest of this process's life, so a squatted port is
    /// not retried — and the slot not re-written and restored — every tick.
    squatted: Mutex<BTreeSet<u16>>,
    /// Whether this process has reclaimed a retired wiring convention's leftovers.
    reclaimed: std::sync::atomic::AtomicBool,
    /// Per-slot re-apply pacing; see [`ReapplyGate`].
    reapplies: Mutex<BTreeMap<String, ReapplyGate>>,
    /// The hook side's record of which provider each agent says it is using.
    sightings: Arc<SessionRegistry>,
}

impl EndpointWiring {
    pub(crate) fn new(
        listeners: Arc<EndpointListeners>,
        wiring: Arc<WiringState>,
        ports: RelayPorts,
        sinks: TamperSinks,
        sightings: Arc<SessionRegistry>,
    ) -> Self {
        Self {
            listeners,
            wiring,
            ports,
            started_at: now_unix(),
            sinks,
            written: Mutex::new(BTreeMap::new()),
            // The daemon's own port can sit inside the block on an instance with
            // a hand-picked port; it is never a slot's.
            squatted: Mutex::new(BTreeSet::from([ports.daemon])),
            reclaimed: std::sync::atomic::AtomicBool::new(false),
            reapplies: Mutex::new(BTreeMap::new()),
            sightings,
        }
    }

    fn written(&self) -> std::sync::MutexGuard<'_, BTreeMap<std::path::PathBuf, FileFingerprint>> {
        self.written.lock().unwrap_or_else(|e| e.into_inner())
    }

    fn squatted(&self) -> BTreeSet<u16> {
        self.squatted
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    /// The one place a port is chosen for `key`, so a first pick and a
    /// same-pass re-pick after a bind failure obey the same rule:
    /// [`records::allocate_port`]'s own tiering (this key's existing port,
    /// then the first port nothing has ever named, then the least recently
    /// released tombstone) filtered to exclude whatever is squatted right
    /// now, and only when that yields nothing, a last-resort scan for any
    /// port free of both squatting and another key's live claim.
    fn pick_port(
        &self,
        key: &str,
        block: &std::ops::RangeInclusive<u16>,
        usable: &[(String, EndpointRecord)],
    ) -> Option<u16> {
        let squatted = self.squatted();
        records::allocate_port(key, block.clone(), usable)
            .filter(|p| !squatted.contains(p))
            .or_else(|| {
                block.clone().find(|p| {
                    !squatted.contains(p)
                        && !usable.iter().any(|(_, r)| r.port == *p && r.is_live())
                })
            })
    }

    /// One pass over every agent this daemon owns the wiring for. Returns
    /// whether any slot failed, for the supervisor's shared backoff.
    pub(crate) async fn reconcile(
        &self,
        config: &Config,
        agents: &[crate::hooks::DetectedAgent],
    ) -> bool {
        let mut any_failed = false;
        for agent in agents {
            let Some(endpoints) = agent.binding.provider_endpoints() else {
                continue;
            };
            if !super::owns_wiring_for(config, &*agent.binding) {
                continue;
            }
            any_failed |= self.reconcile_agent(endpoints).await;
        }
        any_failed
    }

    /// Every file a wiring pass reads, for the agents this daemon owns the
    /// wiring for — what the supervisor's file trigger watches.
    pub(crate) fn watch_files(
        &self,
        config: &Config,
        agents: &[crate::hooks::DetectedAgent],
    ) -> Vec<std::path::PathBuf> {
        let mut files: Vec<std::path::PathBuf> = agents
            .iter()
            .filter(|agent| super::owns_wiring_for(config, &*agent.binding))
            .filter_map(|agent| agent.binding.provider_endpoints())
            .flat_map(|endpoints| endpoints.watch_files())
            .collect();
        files.sort();
        files.dedup();
        files
    }

    async fn reconcile_agent(&self, endpoints: &'static dyn ProviderEndpoints) -> bool {
        if !self
            .reclaimed
            .swap(true, std::sync::atomic::Ordering::Relaxed)
        {
            endpoints.reclaim_retired(self.ports.main);
        }
        let prefix = endpoints.record_prefix();
        let recorded = match records::endpoint_records(prefix) {
            Ok(recorded) => recorded,
            Err(e) => {
                tracing::warn!(
                    agent = endpoints.agent_type(),
                    code = %e.code,
                    error = %e.message,
                    "endpoint records unreadable — provider slots are left as they are"
                );
                return true;
            }
        };

        // 1. Serve every slot an editor may be dialling.
        for (key, rec) in &recorded {
            if rec.state == SlotState::Released && rec.released_by == Some(ReleasedBy::Teardown) {
                continue;
            }
            self.serve(endpoints, key, rec).await;
        }

        // 2. Decide each slot the files show.
        let keys: BTreeSet<String> = recorded.iter().map(|(k, _)| k.clone()).collect();
        let observation = endpoints.observe(&keys);
        for problem in &observation.problems {
            tracing::warn!(
                ?problem,
                "a Cline state file cannot be read safely — its slots are left alone"
            );
        }
        let ctx = DecideCtx {
            ports: self.ports,
            started_at: self.started_at,
        };
        // Updated as the pass records slots, so two slots wired in one pass can
        // never be handed the same port.
        let mut recorded = recorded;
        let mut any_failed = false;
        for obs in &observation.slots {
            let key = obs.slot.record_key();
            let rec = recorded
                .iter()
                .find(|(k, _)| *k == key)
                .map(|(_, r)| r.clone());
            let rec = rec.as_ref();
            let failed = match decide(obs, rec, &ctx) {
                Decision::Keep => {
                    if let Some(rec) = rec {
                        self.keep(endpoints, &key, rec, obs);
                    }
                    false
                }
                Decision::Skip => false,
                Decision::Uncovered(_) => {
                    // Reported by the detector from the same decision; nothing
                    // for the daemon to add.
                    self.wiring.set_endpoint_verdict(&key, None);
                    false
                }
                Decision::Release { restore } => {
                    if let Some(rec) = rec {
                        self.release(endpoints, &key, rec, obs, restore).await;
                    }
                    false
                }
                Decision::Reapply => match rec {
                    Some(rec) => self.reapply(endpoints, &key, rec, obs),
                    None => false,
                },
                Decision::Wire {
                    path,
                    origin,
                    prior,
                }
                | Decision::NewPrior {
                    path,
                    origin,
                    prior,
                } => {
                    self.wire(
                        endpoints,
                        &key,
                        rec,
                        &mut recorded,
                        obs,
                        &path,
                        origin,
                        prior,
                    )
                    .await
                }
            };
            any_failed |= failed;
        }
        any_failed
    }

    /// Bind `rec`'s endpoint if it is not already served.
    async fn serve(&self, endpoints: &dyn ProviderEndpoints, key: &str, rec: &EndpointRecord) {
        if self.squatted().contains(&rec.port) {
            // Known bad for this process's life already. Both paths that add
            // a port here (this method's own failure branch below, and
            // `wire`'s) release the record that named it in the same call, so
            // a record still naming a squatted port is never live — there is
            // nothing to actually serve, and retrying a bind already known to
            // fail is pure noise on every tick until the daemon restarts.
            self.wiring.set_endpoint_verdict(
                key,
                Some(EndpointVerdict {
                    code: ERR_MODEL_RELAY_ENDPOINT_PORTS,
                    detail: format!("loopback port {} is held by another process", rec.port),
                }),
            );
            return;
        }
        let Some(origin) = parse_origin(&rec.origin, &self.ports) else {
            return;
        };
        let spec = EndpointSpec {
            key: key.to_string(),
            agent: endpoints.agent_type(),
            family: rec.family.as_deref().and_then(family_of),
            port: rec.port,
            origin,
        };
        if let Err(e) = self.listeners.ensure(spec).await {
            self.wiring.set_endpoint_verdict(
                key,
                Some(EndpointVerdict {
                    code: ERR_MODEL_RELAY_ENDPOINT_PORTS,
                    detail: e.message.clone(),
                }),
            );
            self.squatted
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .insert(rec.port);
            // An editor pointed at a port something else holds sends that
            // provider's traffic — and its key — to that process. Put the
            // agent's own value back; the next pass wires it on another port.
            if rec.is_live() {
                let Some(slot) = SlotId::from_record_key(key) else {
                    return;
                };
                let _ = records::update_endpoint(key, |r| {
                    r.state = SlotState::Released;
                    r.released_by = Some(ReleasedBy::Wiring);
                    r.changed_at = now_unix();
                });
                let port = rec.port;
                if let Ok(RewriteOutcome::Written) =
                    endpoints.write_slot(&rec.file, &slot, &rec.prior, &|cur| names_port(cur, port))
                {
                    self.remember_write(&rec.file);
                }
            }
        }
    }

    /// A slot whose file still holds our value: prove the editor uses it, and
    /// say so when an editor that holds the URL routes around it.
    fn keep(
        &self,
        endpoints: &dyn ProviderEndpoints,
        key: &str,
        rec: &EndpointRecord,
        obs: &SlotObservation,
    ) {
        self.wiring.set_endpoint_verdict(key, None);
        if rec.state == SlotState::Pending {
            let _ = records::update_endpoint(key, |r| r.state = SlotState::Wired);
        }
        let now = now_unix();

        if self
            .listeners
            .get(key)
            .is_some_and(|ep| rec.served_since_written(ep.last_request_unix()))
        {
            if rec.proven_by != Some(Proof::Traffic) || rec.misconfigured_event.is_some() {
                self.prove(endpoints, key, obs, Proof::Traffic, now);
            }
            return;
        }

        if rec.proven_at.is_none() {
            // Where the editor may not route by the file it saved, only a
            // request proves the slot.
            if obs.traffic_only().is_some() {
                return;
            }
            let saved_by_editor = {
                let written = self.written();
                match (written.get(&obs.file), fingerprint(&obs.file)) {
                    (Some(ours), Ok(current)) => *ours != current,
                    // No write of ours this process saw: a file that names our
                    // value after a restart proves nothing about who saved it.
                    _ => false,
                }
            };
            if saved_by_editor {
                self.prove(endpoints, key, obs, Proof::EditorSave, now);
            }
            return;
        }

        // The editor holds our URL and has never sent a request through it.
        // If its own hooks say it is using the provider, something replaces
        // the URL in its memory.
        if rec.proven_by != Some(Proof::EditorSave)
            || rec.misconfigured_event.is_some()
            || !obs.hooks_name_provider()
        {
            return;
        }
        let loaded_at = rec.proven_at.unwrap_or(now);
        let sightings: Vec<u64> = obs
            .provider_ids()
            .iter()
            .flat_map(|id| {
                self.sightings
                    .provider_sightings(endpoints.agent_type(), id)
            })
            .collect();
        let Some(seen) = misconfigured_since(loaded_at, &sightings, now) else {
            return;
        };
        let event = self.event(
            endpoints,
            key,
            &obs.file,
            obs,
            DETECTION_RELAY_WIRING_MISCONFIGURED,
        );
        self.sinks.publish_detected(&event);
        let _ = records::update_endpoint(key, |r| {
            r.misconfigured_event = Some(event.tamper.event_id.clone());
        });
        tracing::warn!(
            agent = endpoints.agent_type(),
            key,
            loaded_at,
            seen,
            "the editor uses this provider and loaded its relay URL, yet no request reached the endpoint — something overrides it"
        );
    }

    /// Record `proof` for a slot and heal the events it closes: the pending
    /// event on any proof, the misconfigured one only on a request.
    fn prove(
        &self,
        endpoints: &dyn ProviderEndpoints,
        key: &str,
        obs: &SlotObservation,
        proof: Proof,
        now: u64,
    ) {
        let updated = records::update_endpoint(key, |r| {
            r.proven_at.get_or_insert(now);
            r.proven_by = Some(proof);
        });
        tracing::info!(
            agent = endpoints.agent_type(),
            key,
            ?proof,
            "provider slot proven: the editor uses its relay endpoint"
        );
        let Ok(Some(rec)) = updated else {
            return;
        };
        if let Some(id) = rec.pending_event.as_deref() {
            self.heal(
                endpoints,
                key,
                &rec,
                obs,
                DETECTION_RELAY_WIRING_PENDING,
                id,
            );
            let _ = records::update_endpoint(key, |r| r.pending_event = None);
        }
        if proof == Proof::Traffic {
            if let Some(id) = rec.misconfigured_event.as_deref() {
                self.heal(
                    endpoints,
                    key,
                    &rec,
                    obs,
                    DETECTION_RELAY_WIRING_MISCONFIGURED,
                    id,
                );
                let _ = records::update_endpoint(key, |r| r.misconfigured_event = None);
            }
        }
    }

    /// Publish the heal of the detection `id`, linked to it.
    fn heal(
        &self,
        endpoints: &dyn ProviderEndpoints,
        key: &str,
        rec: &EndpointRecord,
        obs: &SlotObservation,
        method: &str,
        id: &str,
    ) {
        let detected = self
            .event(endpoints, key, &rec.file, obs, method)
            .with_event_id(id);
        self.sinks.publish_healed(&TamperEvent::new_healed(
            &detected,
            "succeeded",
            1,
            "closed",
        ));
    }

    async fn release(
        &self,
        endpoints: &dyn ProviderEndpoints,
        key: &str,
        rec: &EndpointRecord,
        obs: &SlotObservation,
        restore: Option<SlotValue>,
    ) {
        // A provider entry the developer deleted: the next bundle re-reads
        // `providers.json` for every new task, so no running editor still holds
        // this port, and the listener goes with the entry.
        if !obs.entry_present {
            self.listeners.release(key).await;
        }
        let _ = records::update_endpoint(key, |r| {
            r.state = SlotState::Released;
            r.released_by = Some(ReleasedBy::Wiring);
            r.changed_at = now_unix();
        });
        if let Some(value) = restore {
            let port = rec.port;
            match endpoints.write_slot(&obs.file, &obs.slot, &value, &|cur| names_port(cur, port)) {
                Ok(RewriteOutcome::Written) => self.remember_write(&obs.file),
                Ok(_) => {}
                Err(e) => {
                    tracing::warn!(key, code = %e.code, error = %e.message, "could not hand a provider slot back")
                }
            }
        }
        self.wiring.set_endpoint_verdict(key, None);
        tracing::info!(
            agent = endpoints.agent_type(),
            key,
            "provider slot handed back"
        );
    }

    fn reapply(
        &self,
        endpoints: &dyn ProviderEndpoints,
        key: &str,
        rec: &EndpointRecord,
        obs: &SlotObservation,
    ) -> bool {
        let Some(ours) = rec.last_written.clone() else {
            return false;
        };
        let now = std::time::Instant::now();
        {
            let mut gates = self.reapplies.lock().unwrap_or_else(|e| e.into_inner());
            let gate = gates.entry(key.to_string()).or_default();
            if !gate.allows(now) {
                // Paced, not dropped: the next trigger or tick past the gap
                // re-applies.
                return false;
            }
            let was_contested = gate.contested(now);
            gate.note(now);
            let contested = gate.contested(now);
            if let Some(endpoint) = self.listeners.get(key) {
                endpoint.set_contested(contested);
            }
            if contested && !was_contested {
                tracing::warn!(
                    agent = endpoints.agent_type(),
                    key,
                    "a provider slot keeps being reverted — re-applying it at most once a minute"
                );
            }
        }
        let observed = obs.value.clone();
        let detected = (rec.state == SlotState::Wired)
            .then(|| self.event(endpoints, key, &obs.file, obs, DETECTION_RELAY_URL_REVERTED));
        if let Some(event) = &detected {
            self.sinks.publish_detected(event);
        }
        // Stamped before the write: a request that lands right after it counts.
        let written_at = now_unix();
        match endpoints.write_slot(&obs.file, &obs.slot, &SlotValue::Text(ours), &|cur| {
            *cur == observed
        }) {
            Ok(RewriteOutcome::Written) => {
                self.remember_write(&obs.file);
                // The editor that saved the old value holds it in memory until
                // it restarts: whatever proved the slot no longer does.
                let _ = records::update_endpoint(key, |r| {
                    r.state = SlotState::Wired;
                    r.changed_at = written_at;
                    r.proven_at = None;
                    r.proven_by = None;
                });
                if let Some(event) = &detected {
                    self.sinks.publish_healed(&TamperEvent::new_healed(
                        event,
                        "succeeded",
                        1,
                        "closed",
                    ));
                }
                tracing::info!(
                    agent = endpoints.agent_type(),
                    key,
                    "relay URL re-applied to a reverted provider slot"
                );
                false
            }
            Ok(_) => false,
            Err(e) => {
                self.state_file_verdict(key, &e);
                if let Some(event) = &detected {
                    self.sinks
                        .publish_healed(&TamperEvent::new_healed(event, "failed", 1, "closed"));
                }
                true
            }
        }
    }

    #[allow(clippy::too_many_arguments)]
    async fn wire(
        &self,
        endpoints: &dyn ProviderEndpoints,
        key: &str,
        rec: Option<&EndpointRecord>,
        recorded: &mut Vec<(String, EndpointRecord)>,
        obs: &SlotObservation,
        path: &str,
        origin: Origin,
        prior: SlotValue,
    ) -> bool {
        let block = endpoint_port_block(self.ports.main);
        let squatted = self.squatted();
        let usable: Vec<(String, EndpointRecord)> = recorded
            .iter()
            .filter(|(k, r)| k != key || !squatted.contains(&r.port))
            .cloned()
            .collect();
        let Some(port) = self.pick_port(key, &block, &usable) else {
            self.wiring.set_endpoint_verdict(
                key,
                Some(EndpointVerdict {
                    code: ERR_MODEL_RELAY_ENDPOINT_PORTS,
                    detail: format!(
                        "every endpoint port in {}–{} is in use",
                        block.start(),
                        block.end()
                    ),
                }),
            );
            return true;
        };
        let family = obs.row.and_then(|r| r.family);
        let mut value = relay_value(port, path);

        // Record BEFORE the value that names the port exists anywhere.
        let record = EndpointRecord {
            port,
            origin: origin.to_string(),
            prior,
            last_written: Some(value.clone()),
            file: obs.file.clone(),
            state: SlotState::Pending,
            released_by: None,
            changed_at: now_unix(),
            proven_at: None,
            family: family.map(|f| f.as_str().to_string()),
            proven_by: None,
            pending_event: rec.and_then(|r| r.pending_event.clone()),
            misconfigured_event: rec.and_then(|r| r.misconfigured_event.clone()),
        };
        if let Err(e) = records::put_endpoint(key, record.clone()) {
            self.state_file_verdict(key, &e);
            return true;
        }
        match recorded.iter_mut().find(|(k, _)| k == key) {
            Some((_, existing)) => *existing = record.clone(),
            None => recorded.push((key.to_string(), record.clone())),
        }

        let mut port = port;
        let spec = EndpointSpec {
            key: key.to_string(),
            agent: endpoints.agent_type(),
            family,
            port,
            origin: origin.clone(),
        };
        if let Err(first_err) = self.listeners.ensure(spec).await {
            self.squatted
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .insert(port);
            // The pinned port can be lost to an unrelated ephemeral bind in the
            // instant between choosing it and binding it — a millisecond
            // collision, not a real occupant. Re-pick once, in this same pass,
            // rather than leaving the slot unwired until the next reconcile —
            // #377: with a long enough tick that "next reconcile" is minutes
            // away for a race nothing recorded here caused. Through the same
            // `pick_port` the first attempt used, so an unused port still
            // wins over a released tombstone here too.
            let Some(retry_port) = self.pick_port(key, &block, &usable) else {
                self.abandon(key, ERR_MODEL_RELAY_ENDPOINT_PORTS, first_err.message);
                return true;
            };
            port = retry_port;
            value = relay_value(port, path);
            let mut record = record;
            record.port = port;
            record.last_written = Some(value.clone());
            if let Err(e) = records::put_endpoint(key, record.clone()) {
                self.state_file_verdict(key, &e);
                return true;
            }
            if let Some((_, existing)) = recorded.iter_mut().find(|(k, _)| k == key) {
                *existing = record;
            }
            let spec = EndpointSpec {
                key: key.to_string(),
                agent: endpoints.agent_type(),
                family,
                port,
                origin: origin.clone(),
            };
            if let Err(e) = self.listeners.ensure(spec).await {
                self.squatted
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .insert(port);
                self.abandon(key, ERR_MODEL_RELAY_ENDPOINT_PORTS, e.message);
                return true;
            }
        }

        let probe_format = family.unwrap_or(WireFormat::Unknown);
        if let Err(why) = preflight::probe(
            port,
            probe_format,
            origin.as_url().as_str(),
            preflight::PREFLIGHT_TIMEOUT,
        )
        .await
        {
            self.abandon(key, ERR_MODEL_RELAY_PREFLIGHT_FAILED, why);
            return true;
        }

        let observed = obs.value.clone();
        // Stamped before the write: a request that lands right after it counts.
        let written_at = now_unix();
        match endpoints.write_slot(&obs.file, &obs.slot, &SlotValue::Text(value), &|cur| {
            *cur == observed
        }) {
            Ok(RewriteOutcome::Written) => {
                self.remember_write(&obs.file);
                self.wiring.set_endpoint_verdict(key, None);
                let event = self.event(
                    endpoints,
                    key,
                    &obs.file,
                    obs,
                    DETECTION_RELAY_WIRING_PENDING,
                );
                self.sinks.publish_detected(&event);
                let _ = records::update_endpoint(key, |r| {
                    r.state = SlotState::Wired;
                    r.changed_at = written_at;
                    r.pending_event = Some(event.tamper.event_id.clone());
                });
                tracing::info!(
                    agent = endpoints.agent_type(),
                    key,
                    port,
                    origin = %origin,
                    "provider slot wired to its relay endpoint — the editor picks it up at its next start"
                );
                false
            }
            // Someone saved between our read and our write: the record stays
            // Pending and the next pass decides, but the status now says so —
            // a pass that keeps landing here must not look silent to a caller
            // with no short retry of its own (#377).
            Ok(RewriteOutcome::Contended) => {
                self.wiring.set_endpoint_verdict(
                    key,
                    Some(EndpointVerdict {
                        code: ERR_MODEL_RELAY_STATE_FILE,
                        detail:
                            "a save landed between the read and the write; retried on the next pass"
                                .into(),
                    }),
                );
                false
            }
            // The value is already ours: nothing to report.
            Ok(RewriteOutcome::Unchanged) => false,
            Ok(RewriteOutcome::Absent) => {
                self.abandon(
                    key,
                    ERR_MODEL_RELAY_STATE_FILE,
                    "the settings file disappeared".into(),
                );
                false
            }
            Err(e) => {
                self.abandon(key, e.code, e.message);
                true
            }
        }
    }

    /// Mark a slot this pass could not wire as released, with a verdict. The
    /// file was not written, so there is nothing to restore.
    fn abandon(&self, key: &str, code: &'static str, detail: String) {
        let _ = records::update_endpoint(key, |r| {
            r.state = SlotState::Released;
            r.released_by = Some(ReleasedBy::Wiring);
            r.changed_at = now_unix();
        });
        tracing::warn!(key, code, detail = %detail, "provider slot not wired");
        self.wiring
            .set_endpoint_verdict(key, Some(EndpointVerdict { code, detail }));
    }

    fn state_file_verdict(&self, key: &str, e: &crate::error::OlError) {
        self.wiring.set_endpoint_verdict(
            key,
            Some(EndpointVerdict {
                code: ERR_MODEL_RELAY_STATE_FILE,
                detail: e.message.clone(),
            }),
        );
    }

    fn remember_write(&self, file: &std::path::Path) {
        if let Ok(now) = fingerprint(file) {
            self.written().insert(file.to_path_buf(), now);
        }
    }

    fn event(
        &self,
        endpoints: &dyn ProviderEndpoints,
        key: &str,
        file: &std::path::Path,
        obs: &SlotObservation,
        method: &str,
    ) -> TamperEvent {
        let field = match &obs.slot {
            SlotId::GlobalState { key, .. } => key.clone(),
            SlotId::ProvidersJson { id } => format!("providers.{id}.settings.baseUrl"),
        };
        TamperEvent::new(
            key.to_string(),
            endpoints.agent_type().to_string(),
            crate::core::hook_state::hash_settings_path(file),
            RELAY_WIRING_HOOK_EVENT.to_string(),
            method.to_string(),
        )
        .with_field_deltas(vec![FieldDelta {
            field,
            change: "modified".to_string(),
        }])
    }
}

/// How long after a hook names a provider its endpoint may stay silent. The
/// hook comes at the start of a task or around a tool call, and the model
/// request it belongs to follows within seconds.
const MISCONFIGURED_AFTER_SECS: u64 = 120;

/// The sighting that shows an editor routing around a URL it loaded at
/// `loaded_at`, on an endpoint that has served nothing: a hook naming the
/// provider after the load, and long enough ago that its request would have
/// arrived.
fn misconfigured_since(loaded_at: u64, sightings: &[u64], now: u64) -> Option<u64> {
    sightings
        .iter()
        .copied()
        .filter(|seen| *seen >= loaded_at && now.saturating_sub(*seen) >= MISCONFIGURED_AFTER_SECS)
        .min()
}

fn parse_origin(raw: &str, ports: &RelayPorts) -> Option<Origin> {
    normalize_origin(&reqwest::Url::parse(raw).ok()?, ports).ok()
}

fn family_of(name: &str) -> Option<WireFormat> {
    WireFormat::ALL.into_iter().find(|f| f.as_str() == name)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Two ports from the ephemeral range, for tests that must never touch a
    /// well-known one (7600, say) a real `openlatch` install on the test host
    /// could actually be using — the unit-test equivalent of
    /// `tests/model_relay_cline_endpoints.rs::free_port_with_block`.
    fn ephemeral_ports() -> RelayPorts {
        let free = || {
            std::net::TcpListener::bind(("127.0.0.1", 0))
                .and_then(|l| l.local_addr())
                .expect("ephemeral port")
                .port()
        };
        RelayPorts {
            daemon: free(),
            main: free(),
        }
    }
    use std::time::{Duration, Instant};

    #[test]
    fn a_reapply_is_paced_and_a_contested_slot_slows_down() {
        let mut gate = ReapplyGate::default();
        let start = Instant::now();
        assert!(gate.allows(start));
        gate.note(start);
        assert!(
            !gate.allows(start + Duration::from_secs(5)),
            "within the gap"
        );
        assert!(gate.allows(start + Duration::from_secs(10)));

        // Twenty reverts inside ten minutes: contested, and the gap widens.
        let mut now = start;
        for _ in 1..CONTESTED_REVERTS {
            now += Duration::from_secs(10);
            gate.note(now);
        }
        assert!(gate.contested(now));
        assert!(!gate.allows(now + Duration::from_secs(30)));
        assert!(gate.allows(now + Duration::from_secs(60)));

        // Quiet for longer than the window: no longer contested.
        let later = now + CONTESTED_WINDOW + Duration::from_secs(1);
        assert!(!gate.contested(later));
    }

    /// The whole path through `keep`: an editor that saved our value, a hook
    /// naming the provider long enough ago, no request on the endpoint — one
    /// `relay_wiring_misconfigured` event, recorded, and not repeated.
    #[test]
    fn keep_reports_an_editor_routing_around_its_loaded_url_once() {
        use crate::hooks::cline_providers::{state_lanes_from, ClineProviderEndpoints};
        use crate::hooks::model_relay_endpoints::Proof;

        let _lock = crate::config::OPENLATCH_DIR_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let dir = tempfile::tempdir().expect("tempdir");
        let _env = crate::hooks::cline::EnvOverride::apply([(
            "OPENLATCH_DIR",
            Some(dir.path().join("openlatch").into_os_string()),
        )]);
        let ports = RelayPorts {
            daemon: 7500,
            main: 7600,
        };
        let gs = dir.path().join("data").join("globalState.json");
        std::fs::create_dir_all(gs.parent().expect("parent")).expect("mkdir");
        std::fs::write(
            &gs,
            r#"{"actModeApiProvider":"gemini","geminiBaseUrl":"http://127.0.0.1:7601"}"#,
        )
        .expect("write");
        let endpoints: &'static ClineProviderEndpoints = Box::leak(Box::new(
            ClineProviderEndpoints::at(state_lanes_from(Some(gs.clone()), Some(gs.clone())), None),
        ));
        let key = "cline:gs:shared:geminiBaseUrl";
        let now = now_unix();
        let record = EndpointRecord {
            port: 7601,
            origin: "https://generativelanguage.googleapis.com/".into(),
            prior: SlotValue::Absent,
            last_written: Some("http://127.0.0.1:7601".into()),
            file: gs.clone(),
            state: SlotState::Wired,
            released_by: None,
            changed_at: now - 1_000,
            proven_at: Some(now - 1_000),
            family: None,
            pending_event: None,
            proven_by: Some(Proof::EditorSave),
            misconfigured_event: None,
        };
        records::put_endpoint(key, record).expect("put");

        let (logger, mut events) = crate::logging::tamper_log::TamperLogger::channel();
        let registry = Arc::new(SessionRegistry::default());
        let factory: crate::model_relay::endpoints::StateFactory =
            Arc::new(|_| -> crate::model_relay::ModelRelayState { unreachable!("not served") });
        let wiring = EndpointWiring::new(
            Arc::new(EndpointListeners::new(factory)),
            Arc::new(WiringState::default()),
            ports,
            TamperSinks {
                logger: Some(logger),
                cloud_tx: None,
                agent_id: String::new(),
                client_version: String::new(),
            },
            registry.clone(),
        );
        let keep = || {
            let rec = records::endpoint_records("cline:")
                .expect("records")
                .into_iter()
                .find(|(k, _)| k == key)
                .expect("record")
                .1;
            let observation = endpoints.observe(&BTreeSet::from([key.to_string()]));
            let obs = observation
                .slots
                .iter()
                .find(|s| s.slot.record_key() == key)
                .expect("slot");
            wiring.keep(endpoints, key, &rec, obs);
        };

        // In use only before the editor loaded the URL.
        registry.note_provider("cline", "gemini", now - 2_000);
        keep();
        assert!(events.try_recv().is_err());

        // In use long enough ago, and nothing arrived.
        registry.note_provider("cline", "gemini", now - 600);
        keep();
        let event = events.try_recv().expect("a misconfigured event");
        assert_eq!(
            event.tamper.detection_method,
            DETECTION_RELAY_WIRING_MISCONFIGURED
        );
        let recorded = records::endpoint_records("cline:").expect("records");
        assert_eq!(
            recorded[0].1.misconfigured_event.as_deref(),
            Some(event.tamper.event_id.as_str())
        );

        keep();
        assert!(events.try_recv().is_err(), "reported once");
    }

    /// #377: a fresh slot's first-choice port (main + 1, see [`records::allocate_port`])
    /// can be lost to an unrelated ephemeral bind between the pick and the real
    /// bind — a millisecond collision on a busy host, not a real occupant. The
    /// pass that hits it must not sit unwired until the next reconcile: with a
    /// long enough tick, "the next reconcile" never comes for the span of a
    /// test, and in production it is still a needless 60 seconds of an
    /// unwired provider.
    ///
    /// A plain `#[test]` driving its own runtime, not `#[tokio::test]`: the
    /// env lock below has to stay held across the `reconcile_agent` call —
    /// `OPENLATCH_DIR` is process-wide and a concurrent test repointing it
    /// mid-pass would corrupt this one — and holding a `std::sync::Mutex`
    /// guard across an `.await` inside an `async fn` is exactly what
    /// `clippy::await_holding_lock` exists to catch.
    #[test]
    fn a_squatted_port_is_re_picked_within_the_same_pass() {
        use crate::hooks::cline_providers::{state_lanes_from, ClineProviderEndpoints};

        let _lock = crate::config::OPENLATCH_DIR_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let dir = tempfile::tempdir().expect("tempdir");
        let _env = crate::hooks::cline::EnvOverride::apply([(
            "OPENLATCH_DIR",
            Some(dir.path().join("openlatch").into_os_string()),
        )]);
        let ports = ephemeral_ports();
        let block = endpoint_port_block(ports.main);
        // Squat the ONLY port a brand-new slot can allocate (main + 1, the
        // first free port `allocate_port` finds in an empty block) before the
        // daemon ever gets a chance to bind it.
        let squat_port = *block.start();
        let squatter = std::net::TcpListener::bind(("127.0.0.1", squat_port)).expect("squat");

        let gs = dir.path().join("data").join("globalState.json");
        std::fs::create_dir_all(gs.parent().expect("parent")).expect("mkdir");
        // Port 9 (discard) refuses loopback connections instantly, so preflight
        // fails fast; the retry-port assertion below does not depend on it
        // succeeding.
        std::fs::write(
            &gs,
            r#"{"actModeApiProvider":"ollama","ollamaBaseUrl":"http://127.0.0.1:9"}"#,
        )
        .expect("write");
        let endpoints: &'static ClineProviderEndpoints = Box::leak(Box::new(
            ClineProviderEndpoints::at(state_lanes_from(Some(gs.clone()), Some(gs.clone())), None),
        ));

        let factory: crate::model_relay::endpoints::StateFactory = Arc::new(|_| {
            crate::model_relay::ModelRelayState::new(
                reqwest::Url::parse("http://127.0.0.1:9").expect("url"),
                0,
                1,
                &[],
            )
        });
        let (logger, _events) = crate::logging::tamper_log::TamperLogger::channel();
        let wiring = EndpointWiring::new(
            Arc::new(EndpointListeners::new(factory)),
            Arc::new(WiringState::default()),
            ports,
            TamperSinks {
                logger: Some(logger),
                cloud_tx: None,
                agent_id: String::new(),
                client_version: String::new(),
            },
            Arc::new(SessionRegistry::default()),
        );

        tokio::runtime::Runtime::new()
            .expect("runtime")
            .block_on(wiring.reconcile_agent(endpoints));
        drop(squatter);

        let recorded = records::endpoint_records("cline:").expect("records");
        let (_, rec) = recorded
            .iter()
            .find(|(k, _)| k == "cline:gs:shared:ollamaBaseUrl")
            .expect("a record for the ollama slot");
        assert_ne!(
            rec.port, squat_port,
            "the squatted port must never be the one recorded as wired"
        );
        assert!(
            block.contains(&rec.port),
            "the re-picked port must still be in this slot's own endpoint block, got {}",
            rec.port
        );
    }

    /// The other half of the same defect (#377): a record `wire` or `serve`
    /// already gave up on — released, naming a port this process marked
    /// squatted — must not have that exact bind retried by `serve` on a later
    /// pass. Both paths that squat a port release the record naming it in the
    /// same call, so a record still naming a squatted port is never live;
    /// `serve` used to bind it anyway, once per tick, forever, until restart.
    ///
    /// The port is left genuinely free here: a bind succeeding would be the
    /// proof that `serve` ignored `squatted` and tried regardless.
    #[test]
    fn a_squatted_port_is_never_retried_by_serve() {
        let _lock = crate::config::OPENLATCH_DIR_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let dir = tempfile::tempdir().expect("tempdir");
        let _env = crate::hooks::cline::EnvOverride::apply([(
            "OPENLATCH_DIR",
            Some(dir.path().join("openlatch").into_os_string()),
        )]);
        let ports = ephemeral_ports();
        let port = *endpoint_port_block(ports.main).start();
        let key = "cline:gs:shared:ollamaBaseUrl";
        let record = EndpointRecord {
            port,
            origin: "http://127.0.0.1:9".into(),
            prior: SlotValue::Absent,
            last_written: Some(format!("http://127.0.0.1:{port}")),
            file: dir.path().join("data").join("globalState.json"),
            state: SlotState::Released,
            released_by: Some(ReleasedBy::Wiring),
            changed_at: now_unix(),
            proven_at: None,
            family: None,
            proven_by: None,
            pending_event: None,
            misconfigured_event: None,
        };
        records::put_endpoint(key, record.clone()).expect("put");

        let factory: crate::model_relay::endpoints::StateFactory =
            Arc::new(|_| -> crate::model_relay::ModelRelayState { unreachable!("never served") });
        let (logger, _events) = crate::logging::tamper_log::TamperLogger::channel();
        let wiring = EndpointWiring::new(
            Arc::new(EndpointListeners::new(factory)),
            Arc::new(WiringState::default()),
            ports,
            TamperSinks {
                logger: Some(logger),
                cloud_tx: None,
                agent_id: String::new(),
                client_version: String::new(),
            },
            Arc::new(SessionRegistry::default()),
        );
        wiring
            .squatted
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(port);

        tokio::runtime::Runtime::new()
            .expect("runtime")
            .block_on(wiring.serve(
                &crate::hooks::cline_providers::ClineProviderEndpoints::RESOLVED,
                key,
                &record,
            ));

        assert!(
            wiring.listeners.get(key).is_none(),
            "serve() must not bind a port it already knows is squatted"
        );
    }

    #[test]
    fn a_provider_in_use_with_a_silent_endpoint_is_misconfigured_only_after_the_window() {
        let loaded_at = 10_000;
        // Used before the editor loaded the URL: says nothing about the URL.
        assert_eq!(misconfigured_since(loaded_at, &[9_000], 20_000), None);
        // Used after, but its request may still be on its way.
        assert_eq!(
            misconfigured_since(loaded_at, &[10_050], 10_050 + MISCONFIGURED_AFTER_SECS - 1),
            None
        );
        // Used after, and long enough ago.
        assert_eq!(
            misconfigured_since(loaded_at, &[10_900, 10_050, 9_000], 11_000),
            Some(10_050)
        );
        assert_eq!(misconfigured_since(loaded_at, &[], 99_999), None);
    }
}