openlatch-client 0.6.0

//! `/admin/*` route group — privileged daemon endpoints, all gated by
//! the same Bearer auth as `/hooks` and `/shutdown`.
//!
//! Today this module hosts the manual auto-update RPC:
//! - `POST /admin/update` — kick off an apply pipeline. Returns 202 +
//!   a `stream_url` the CLI long-polls; the daemon owns the swap +
//!   drain + restart sequence.
//! - `GET /admin/update/status` — snapshot of the in-flight pipeline
//!   for long-poll display.
//!
//! Future privileged endpoints (severity gating override,
//! cargo-install force, etc.) belong under the same `/admin/*` prefix
//! and inherit the auth middleware via `router()` below.
//!
//! See `.local/brainstorms/auto-update/PHASE-2-manual-via-rpc.md` § 4.

use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;

use axum::{
    extract::{rejection::JsonRejection, DefaultBodyLimit, State},
    http::StatusCode,
    middleware,
    response::{IntoResponse, Response},
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};

use crate::daemon::{auth, AppState};
use crate::install_state;
use crate::telemetry::handoff::HandoffBatch;
use crate::update::{
    self, ApplyMode, ApplyResult, ApplyStage, CheckResult, UpdateStatusKind, UpdateStatusSnapshot,
};

/// Request body for `POST /admin/update`.
#[derive(Debug, Deserialize, Default)]
#[serde(default)]
pub struct AdminUpdateRequest {
    /// Optional override of the cargo-install gate. Even an
    /// authenticated caller cannot bypass minisign verification — this
    /// only opts in to `cargo install`-managed binaries getting
    /// rewritten by self-replace, which is the user's call to make.
    pub force_cargo_install: bool,
}

/// Response body for the success path of `POST /admin/update` (HTTP 202).
#[derive(Debug, Serialize)]
pub struct AdminUpdateResponse {
    pub started: bool,
    pub from: String,
    pub to: String,
    pub stream_url: &'static str,
}

/// Failure modes for `POST /admin/update`. Each variant maps to the
/// HTTP status code in the brainstorm doc § 4.
#[derive(Debug)]
pub enum AdminError {
    /// `cargo install`-managed daemon refused to rewrite itself.
    CargoInstall,
    /// Already on the latest version.
    AlreadyUpToDate { current: String },
    /// Manifest fetch / probe failed (registry unreachable).
    CheckFailed(String),
    /// A previous `POST /admin/update` is still running.
    AlreadyInProgress,
}

impl IntoResponse for AdminError {
    fn into_response(self) -> axum::response::Response {
        let (status, body) = match &self {
            Self::CargoInstall => (
                StatusCode::CONFLICT,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_UPDATE_REFUSED_CARGO_INSTALL,
                        "message": "this daemon was installed via `cargo install` — auto-update would not take effect",
                        "suggestion": "Run: cargo install --force --locked openlatch-client"
                    }
                }),
            ),
            Self::AlreadyUpToDate { current } => (
                StatusCode::CONFLICT,
                serde_json::json!({
                    "idempotent": true,
                    "current": current,
                    "message": "already on the latest version"
                }),
            ),
            Self::CheckFailed(reason) => (
                StatusCode::BAD_GATEWAY,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_CLOUD_UNREACHABLE,
                        "message": format!("update check failed: {reason}")
                    }
                }),
            ),
            Self::AlreadyInProgress => (
                StatusCode::SERVICE_UNAVAILABLE,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_DAEMON_START_FAILED,
                        "message": "another auto-update is already in progress"
                    }
                }),
            ),
        };
        (status, Json(body)).into_response()
    }
}

/// Build the `/admin/*` router with bearer auth applied to every route.
pub fn router(state: Arc<AppState>) -> Router<Arc<AppState>> {
    Router::new()
        .route("/admin/update", post(handle_admin_update))
        .route("/admin/update/status", get(handle_admin_update_status))
        .route("/admin/inventory/status", get(handle_inventory_status))
        .route("/admin/inventory/rescan", post(handle_inventory_rescan))
        .route(
            "/admin/inventory/inspect/{source_id}",
            get(handle_inventory_inspect),
        )
        .route("/admin/inventory/projects", get(handle_inventory_projects))
        .route("/admin/inventory/ack", post(handle_inventory_ack))
        .route("/admin/auth/refresh", post(handle_admin_auth_refresh))
        .route("/admin/egress/status", get(handle_egress_status))
        .route(
            crate::telemetry::handoff::PATH,
            post(handle_telemetry_handoff)
                .layer(DefaultBodyLimit::max(crate::telemetry::handoff::MAX_BYTES)),
        )
        .route_layer(middleware::from_fn_with_state(state, auth::bearer_auth))
}

/// `POST /admin/telemetry/events` — a CLI process's last telemetry events,
/// handed over on its way out (`telemetry::handoff`) to ride this daemon's next
/// batch as they were captured.
///
/// `204` whether or not this daemon's own telemetry is on: the CLI learns
/// nothing it could act on, and a daemon with telemetry off drops the events
/// without a trace (I4, I10). Anything but a capped batch of telemetry events
/// is refused whole — `413` over the byte cap, `400` otherwise — and relays
/// nothing.
async fn handle_telemetry_handoff(body: Result<Json<HandoffBatch>, JsonRejection>) -> Response {
    let refuse = |status: StatusCode, message: &str| {
        (
            status,
            Json(serde_json::json!({
                "error": {
                    "code": crate::error::ERR_TELEMETRY_HANDOFF_REJECTED,
                    "message": message,
                }
            })),
        )
            .into_response()
    };
    let batch = match body {
        Ok(Json(batch)) => batch,
        Err(rejection) if rejection.status() == StatusCode::PAYLOAD_TOO_LARGE => {
            return refuse(
                StatusCode::PAYLOAD_TOO_LARGE,
                "the hand-off is larger than MAX_BYTES",
            );
        }
        Err(_) => {
            return refuse(
                StatusCode::BAD_REQUEST,
                "the hand-off is not a batch of telemetry events",
            )
        }
    };
    if let Err(reason) = batch.validate() {
        return refuse(StatusCode::BAD_REQUEST, reason);
    }
    if let Some(handle) = crate::telemetry::global() {
        for event in batch.into_queued() {
            handle.relay(event);
        }
    }
    StatusCode::NO_CONTENT.into_response()
}

/// `GET /admin/egress/status` — the runtime egress view, in the frozen schema.
///
/// **This is the only surface that carries the proxy URL, and that is the whole
/// reason it lives under `/admin/*`.** A corporate proxy's address is internal
/// topology; `/metrics` is unauthenticated and therefore gets the gauges and the
/// status word and nothing else. The URL served here is masked regardless — the
/// state module never stores an unmasked one — so even an authenticated reader
/// cannot lift a password out of it.
///
/// Auth is inherited from `router`'s `route_layer`, exactly like every other
/// `/admin/*` route.
async fn handle_egress_status(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
    let egress = &state.egress;
    let snapshot = egress.snapshot();

    let mut body = serde_json::Map::new();
    body.insert("status".into(), snapshot.status.as_str().into());
    body.insert("proxy_in_use".into(), snapshot.proxy_in_use.into());
    // Absent, not null, when the route is direct — per the frozen table.
    if let Some(url) = snapshot.proxy_url_masked.as_deref() {
        body.insert("proxy_url".into(), url.into());
    }
    body.insert(
        "source".into(),
        match snapshot.source {
            Some(source) => source.as_str().into(),
            None => serde_json::Value::Null,
        },
    );
    body.insert("auth_scheme".into(), snapshot.auth_scheme.as_str().into());
    body.insert("ca_source".into(), snapshot.ca_source.as_str().into());
    body.insert(
        "tls_intercepted".into(),
        match snapshot.tls_intercepted {
            Some(v) => v.into(),
            None => serde_json::Value::Null,
        },
    );
    body.insert(
        "last_ok_at".into(),
        match egress.last_ok_at().and_then(iso8601) {
            Some(ts) => ts.into(),
            None => serde_json::Value::Null,
        },
    );
    body.insert(
        "last_error".into(),
        match egress.last_error() {
            Some(e) => serde_json::json!({ "code": e.code, "message": e.message }),
            None => serde_json::Value::Null,
        },
    );
    body.insert(
        "consecutive_failures".into(),
        egress.consecutive_failures().into(),
    );
    body.insert("probing".into(), egress.is_probing().into());
    // TWO ADDITIVE FIELDS, beyond the frozen table. Both are flagged for
    // ratification rather than folded in silently:
    //
    // - `tls_issuer` — the doctor's TLS line has to be able to name the observed
    //   issuer, and the frozen `tls_intercepted` boolean cannot carry a name.
    // - `warnings` — `status: "degraded"` is produced by configuration warnings,
    //   and the CLI output contract forbids a warning without an actionable
    //   line. Re-deriving them CLI-side would answer from the CLI's environment
    //   rather than the daemon's, which is the exact confusion this endpoint
    //   exists to end.
    body.insert(
        "tls_issuer".into(),
        match snapshot.tls_issuer.as_deref() {
            Some(issuer) => issuer.into(),
            None => serde_json::Value::Null,
        },
    );
    body.insert("warnings".into(), egress.warnings().into());

    Json(serde_json::Value::Object(body))
}

/// Epoch seconds as ISO 8601 UTC. `None` for a timestamp no calendar can render,
/// which on a sane clock cannot happen.
fn iso8601(secs: i64) -> Option<String> {
    chrono::DateTime::from_timestamp(secs, 0).map(|dt| dt.format("%Y-%m-%dT%H:%M:%SZ").to_string())
}

/// `POST /admin/auth/refresh` — clear the cloud worker's `auth_error`
/// latch after a successful `openlatch system auth login`. Returns 200 with the
/// new state so the CLI can confirm the daemon picked up the signal.
///
/// This is the WR-01 escape hatch the credential-poll loop used to do
/// unconditionally on every poll, which caused a 60s flap against
/// permanently-revoked keys. Now the CLI flips the flag exactly once
/// per login, the worker mirrors the change on its next poll, and a
/// stable-revoked key stops re-trying.
///
/// **Also invalidates the credential provider (issue #306).** `auth login`
/// runs in a different process from the daemon, so it cannot reach the
/// daemon's in-process keyring memo the way `store`/`delete` do when called
/// from the same process — see `core::auth::keyring::READ_MEMO`. Without
/// this, clearing the flag alone would let the worker believe it is healthy
/// again while its next POST still carries the exact key that was just
/// replaced, re-latching on the very next attempt. Unconditional, not gated
/// on `was_auth_error`: a quiet key rotation while forwarding was healthy
/// needs the same next-read refresh.
async fn handle_admin_auth_refresh(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
    if let Some(handle) = crate::telemetry::global() {
        crate::telemetry::identity::refresh(handle, &crate::config::openlatch_dir());
    }
    if let Some(provider) = state.credential_provider.as_ref() {
        provider.invalidate();
    }
    let was_auth_error = state
        .cloud_state
        .as_ref()
        .is_some_and(|cs| cs.clear_auth_error());
    if was_auth_error {
        let dir = crate::config::openlatch_dir();
        if let Err(e) = crate::cloud::worker::persist_cloud_state(&dir, false) {
            tracing::warn!(error = %e, "admin auth/refresh: failed to persist cloud_state.json");
        }
        tracing::info!("admin auth/refresh: cleared auth_error after CLI login");
    }
    Json(serde_json::json!({
        "auth_error": false,
        "cleared": was_auth_error,
    }))
}

/// Response body for `GET /admin/inventory/status`.
#[derive(Debug, Serialize)]
pub struct InventoryStatusResponse {
    pub enabled: bool,
    pub state: crate::daemon::config_monitor::ConfigMonitorState,
    pub cache_size: usize,
    pub manifest_loaded: bool,
    pub pending_alerts: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

async fn handle_inventory_status(
    State(state): State<Arc<AppState>>,
) -> Json<InventoryStatusResponse> {
    let monitor = state.config_monitor.snapshot();
    Json(InventoryStatusResponse {
        enabled: state.config.inventory_monitor.enabled,
        state: monitor.state,
        cache_size: state.content_hash_cache.len(),
        manifest_loaded: monitor.manifest_loaded,
        pending_alerts: state.pending_alerts.pending_count(),
        error: monitor.error,
    })
}

/// Request body for `POST /admin/inventory/rescan`.
#[derive(Debug, Deserialize, Default)]
#[serde(default)]
pub struct InventoryRescanRequest {
    pub path: Option<std::path::PathBuf>,
}

async fn handle_inventory_rescan(
    State(state): State<Arc<AppState>>,
    body: Option<Json<InventoryRescanRequest>>,
) -> StatusCode {
    let path_filter = body.and_then(|Json(b)| b.path);
    if let Some(tx) = state.config_monitor.request_tx() {
        if tx
            .send(crate::daemon::config_monitor::ConfigChangeRequest::ManualRescan { path_filter })
            .await
            .is_err()
        {
            return StatusCode::SERVICE_UNAVAILABLE;
        }
        StatusCode::ACCEPTED
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    }
}

/// `POST /admin/update` — kick off the apply pipeline.
///
/// Returns 202 immediately if the pipeline started; the long-running
/// work runs in a spawned task. The CLI polls `GET
/// /admin/update/status` for progress and treats a connection-drop
/// mid-poll as the expected restart-in-progress signal.
async fn handle_admin_update(
    State(state): State<Arc<AppState>>,
    body: Option<Json<AdminUpdateRequest>>,
) -> Result<axum::response::Response, AdminError> {
    let req = body.map(|Json(b)| b).unwrap_or_default();
    let current_version = env!("CARGO_PKG_VERSION").to_string();

    // 1. Cargo-install gate. Done first so we never hit the network
    //    for a binary we won't update.
    if !req.force_cargo_install
        && matches!(
            install_state::detect_install_method(),
            install_state::InstallMethod::CargoInstall
        )
    {
        return Err(AdminError::CargoInstall);
    }

    // 2. Probe the manifest synchronously — this is cheap and lets us
    //    answer 409/502 before committing the in-progress lock.
    let registry_origin = state.config.update.registry_origin.clone();
    let download_timeout = Duration::from_secs(state.config.update.download_timeout_secs.max(1));
    let check = update::check(&current_version, &registry_origin, &state.config.egress).await;
    let (latest, severity, _tarball_url, _tarball_integrity) = match check {
        CheckResult::UpToDate { current } => {
            return Err(AdminError::AlreadyUpToDate { current });
        }
        CheckResult::Failed { reason } => return Err(AdminError::CheckFailed(reason)),
        CheckResult::Available {
            latest,
            severity,
            tarball_url,
            tarball_integrity,
            ..
        } => (latest, severity, tarball_url, tarball_integrity),
    };

    // 3. Reserve the single-update slot. AcqRel is the standard pattern
    //    for compare-and-swap: Acquire on the prior load, Release on
    //    the new value so other threads see all our subsequent writes.
    if state
        .update_in_progress
        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
        .is_err()
    {
        return Err(AdminError::AlreadyInProgress);
    }

    // 4. Stamp the status snapshot to InProgress so the long-poll
    //    starts seeing real data immediately. We re-stamp at every
    //    stage from inside the spawned task.
    state.update_progress.begin();
    *state.update_status.lock().expect("status mutex poisoned") =
        UpdateStatusSnapshot::in_progress(&current_version, &latest);

    // 5. Spawn the apply pipeline. The task owns: download → verify →
    //    extract → sanity → swap → write_sentinel → drain → restart.
    let state_for_task = state.clone();
    let egress = state.config.egress.clone();
    let from = current_version.clone();
    let to = latest.clone();
    let response_to = to.clone();
    tokio::spawn(async move {
        let opts = update::ApplyOptions {
            current_version: from.clone(),
            registry_origin,
            download_timeout,
            force_cargo_install: req.force_cargo_install,
            mode: ApplyMode::Rpc,
            egress,
            progress: state_for_task.update_progress.clone(),
        };
        run_apply_in_daemon(state_for_task, opts, severity).await;
    });

    Ok((
        StatusCode::ACCEPTED,
        Json(AdminUpdateResponse {
            started: true,
            from: current_version,
            to: response_to,
            stream_url: "/admin/update/status",
        }),
    )
        .into_response())
}

/// `GET /admin/update/status` — long-poll surface for the CLI.
///
/// Stage and byte counts are read live from the shared progress, so a
/// download reports its bytes as they land rather than once per stage.
async fn handle_admin_update_status(
    State(state): State<Arc<AppState>>,
) -> Json<UpdateStatusSnapshot> {
    let mut snap = state
        .update_status
        .lock()
        .expect("status mutex poisoned")
        .clone();
    let progress = &state.update_progress;
    if snap.status == UpdateStatusKind::InProgress {
        snap.stage = progress.stage().or(snap.stage);
    }
    snap.bytes_done = progress.bytes_done();
    snap.bytes_total = progress.bytes_total();
    Json(snap)
}

/// Spawned-task body that drives the apply pipeline inside the daemon
/// process. Owns the drain + restart sequence.
///
/// Reachable to the auto-update worker so the same code path runs
/// whether the apply was triggered manually via the RPC or by the
/// background worker. The caller is responsible for taking the
/// `update_in_progress` lock; this function releases it on every
/// failure path.
pub(crate) async fn run_apply_in_daemon(
    state: Arc<AppState>,
    opts: update::ApplyOptions,
    severity: update::Severity,
) {
    let started = std::time::Instant::now();
    let started_at = install_state::now_rfc3339();

    let stamp_stage = |stage: ApplyStage| opts.progress.set_stage(stage);
    let mark_failed = |stage: ApplyStage, reason: String| {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        snap.status = UpdateStatusKind::Failed;
        snap.stage = Some(stage);
        snap.error = Some(reason);
        snap.ended_at = Some(install_state::now_rfc3339());
    };

    let artefacts = match update::prepare_swap_artefacts(&opts).await {
        Ok(a) => a,
        Err(ApplyResult::UpToDate { current }) => {
            // Race window: between the synchronous check above and now,
            // we somehow ended up still on the latest. Treat as a no-op
            // success: nothing was swapped, so there is no restart to wait on.
            tracing::info!(target: "update", current = %current, "concurrent check found us up-to-date — releasing lock");
            {
                let mut snap = state.update_status.lock().expect("status mutex poisoned");
                snap.status = UpdateStatusKind::Completed;
                snap.stage = None;
                snap.ended_at = Some(install_state::now_rfc3339());
            }
            release_lock(&state);
            return;
        }
        Err(ApplyResult::RefusedCargoInstall { suggestion }) => {
            mark_failed(ApplyStage::Check, suggestion);
            release_lock(&state);
            return;
        }
        Err(ApplyResult::Failed { stage, reason }) => {
            mark_failed(stage, reason);
            release_lock(&state);
            return;
        }
        Err(ApplyResult::Applied { .. }) => unreachable!("prepare can't return Applied"),
    };

    // Stage: swap.
    stamp_stage(ApplyStage::Swap);
    let hook_path = match update::locate_hook_binary() {
        Ok(p) => p,
        Err(e) => {
            mark_failed(ApplyStage::Swap, format!("locate hook: {e}"));
            release_lock(&state);
            return;
        }
    };
    let swap_handle =
        match update::perform_swap(&artefacts.staging_exe, &artefacts.staging_hook, &hook_path) {
            Ok(h) => h,
            Err(e) => {
                mark_failed(ApplyStage::Swap, e.to_string());
                release_lock(&state);
                return;
            }
        };

    // Stage: write sentinel BEFORE we let axum drop connections so the
    // new daemon can pick up the breadcrumb if anything between here
    // and post-restart healthz fails.
    //
    // **A sentinel-write failure is fatal post-swap.** `should_rollback`
    // on the next start short-circuits to `false` when the sentinel is
    // missing — so without it, the supervisor-restart-loop safety net
    // is silently disabled. Re-execing into a binary we can no longer
    // safely roll back from is unacceptable. Best-effort undo the swap
    // (the `.bak` siblings are still on disk) and bail; the user
    // continues running the working OLD binary and gets a loud,
    // actionable error.
    let sentinel = update::UpdateSentinel {
        from: artefacts.from.clone(),
        to: artefacts.to.clone(),
        applied_at: started_at.clone(),
    };
    if let Err(e) = update::write_sentinel(&sentinel) {
        let rollback = update::rollback_from_bak();
        tracing::error!(
            target: "update",
            error = %e,
            rollback_error = ?rollback.as_ref().err(),
            "sentinel write failed post-swap; rolled back the swap to keep the safety net intact",
        );
        mark_failed(
            ApplyStage::Swap,
            format!("sentinel write failed post-swap: {e}"),
        );
        release_lock(&state);
        return;
    }

    // Telemetry: completion is captured before the drain, while this task
    // is still guaranteed to run; the end of `serve_with_listener` flushes it.
    //
    // The status surface does NOT say `completed` here: this process has
    // not restarted yet. It stays in progress at `restart`, and the new
    // daemon completes it once its post-restart health probe passes
    // (see `UpdateStatusSnapshot::restarted`).
    stamp_stage(ApplyStage::Restart);
    let duration_ms = started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64;
    crate::telemetry::capture_global(crate::telemetry::Event::update_completed(
        &artefacts.from,
        &artefacts.to,
        severity.as_str(),
        opts.mode.as_str(),
        true,
        Some(duration_ms),
        false,
    ));

    // Stage: drain, then restart. The hand-over is NOT this task's to run:
    // the drain ends `serve_with_listener`, and with it the runtime this task
    // lives on, so anything awaited here after the notify never happens. The
    // request names the swapped binary for the foreground runner, which hands
    // over once the drain has finished and the log has flushed.
    tracing::info!(target: "update", "draining axum prior to restart");
    let _ = state.restart_into.set(swap_handle.current_exe);
    state.admin_shutdown_request.notify_waiters();
}

fn release_lock(state: &AppState) {
    state.update_in_progress.store(false, Ordering::Release);
}

// ---------------------------------------------------------------------------
// Inventory P2 admin endpoints — inspect / projects / ignore / unignore / ack
// ---------------------------------------------------------------------------

#[derive(Debug, Serialize)]
pub struct InventoryInspectResponse {
    /// `{agent}:{kind}:{path_hash}` echoed back for stable client display.
    pub source_id: String,
    /// `Some` when the source is currently in the cache; `None` otherwise.
    pub cache_entry: Option<InventoryInspectEntry>,
    /// Pending alerts that match this source_id (delivered or undelivered).
    pub alerts: Vec<crate::daemon::config_monitor::PendingAlert>,
}

#[derive(Debug, Serialize)]
pub struct InventoryInspectEntry {
    pub agent: String,
    pub kind: String,
    pub content_hash: String,
    pub path_hash: String,
    pub path: String,
}

async fn handle_inventory_inspect(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(source_id): axum::extract::Path<String>,
) -> Json<InventoryInspectResponse> {
    let cache_entry = state
        .content_hash_cache
        .snapshot()
        .into_iter()
        .find_map(|e| {
            let path_hash_hex = hex::encode(e.path_hash);
            let id = format!("{}:{}:{}", e.agent, e.kind, path_hash_hex);
            if id == source_id {
                Some(InventoryInspectEntry {
                    agent: e.agent,
                    kind: e.kind,
                    content_hash: hex::encode(e.content_hash),
                    path_hash: path_hash_hex,
                    path: e.path.display().to_string(),
                })
            } else {
                None
            }
        });
    let alerts: Vec<_> = state
        .pending_alerts
        .snapshot()
        .into_iter()
        .filter(|a| a.source_id == source_id)
        .collect();
    Json(InventoryInspectResponse {
        source_id,
        cache_entry,
        alerts,
    })
}

#[derive(Debug, Serialize)]
pub struct InventoryProjectsResponse {
    pub projects: Vec<String>,
}

async fn handle_inventory_projects(
    State(state): State<Arc<AppState>>,
) -> Json<InventoryProjectsResponse> {
    // Project roots are inferred from cwd hashes seen so far; the cache
    // doesn't index them directly, so we surface the unique parent
    // directories of every cached entry as a heuristic. This is enough
    // for `openlatch system inventory projects` to show the user where config
    // is actively being watched without us tracking a separate set.
    let mut roots: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    for e in state.content_hash_cache.snapshot() {
        if let Some(parent) = e.path.parent() {
            roots.insert(parent.display().to_string());
        }
    }
    Json(InventoryProjectsResponse {
        projects: roots.into_iter().collect(),
    })
}

#[derive(Debug, Deserialize, Default)]
#[serde(default)]
pub struct InventoryAckRequest {
    pub alert_id: Option<String>,
}

async fn handle_inventory_ack(
    State(state): State<Arc<AppState>>,
    body: Option<Json<InventoryAckRequest>>,
) -> Json<serde_json::Value> {
    let req = body.map(|Json(b)| b).unwrap_or_default();
    let removed = state.pending_alerts.ack(req.alert_id.as_deref());
    Json(serde_json::json!({
        "acknowledged": removed,
        "alert_id": req.alert_id,
    }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cloud::CredentialProvider;
    use secrecy::{ExposeSecret, SecretString};
    use std::sync::atomic::AtomicU64;

    /// A `CredentialProvider` double that memoizes its first read exactly
    /// like the real keychain memo (`core::auth::keyring::READ_MEMO`), so a
    /// test can tell whether a caller's `invalidate()` actually cleared the
    /// cache rather than the provider happening to already be fresh.
    struct MemoizingTestProvider {
        /// What the "keychain" holds right now.
        underlying: std::sync::Mutex<String>,
        /// The cached answer, if any.
        cached: std::sync::Mutex<Option<String>>,
        invalidations: AtomicU64,
    }

    impl MemoizingTestProvider {
        fn new(key: &str) -> Arc<Self> {
            Arc::new(Self {
                underlying: std::sync::Mutex::new(key.to_string()),
                cached: std::sync::Mutex::new(None),
                invalidations: AtomicU64::new(0),
            })
        }

        /// Simulate `auth login` writing a new key to the keychain out of
        /// band — this process's cache does not see it yet.
        fn set_underlying(&self, key: &str) {
            *self.underlying.lock().unwrap() = key.to_string();
        }

        fn invalidations(&self) -> u64 {
            self.invalidations.load(Ordering::Relaxed)
        }
    }

    impl CredentialProvider for MemoizingTestProvider {
        fn retrieve(&self) -> Option<SecretString> {
            let mut cached = self.cached.lock().unwrap();
            if cached.is_none() {
                *cached = Some(self.underlying.lock().unwrap().clone());
            }
            cached.clone().map(SecretString::from)
        }

        fn invalidate(&self) {
            self.invalidations.fetch_add(1, Ordering::Relaxed);
            *self.cached.lock().unwrap() = None;
        }
    }

    /// A minimal `AppState` for route-level admin tests. Every field besides
    /// `credential_provider` is a cheap, disabled/empty stand-in — nothing
    /// here reads them.
    fn test_app_state(credential_provider: Option<Arc<dyn CredentialProvider>>) -> Arc<AppState> {
        let (event_logger, _event_rx) = crate::logging::EventLogger::channel();
        Arc::new(AppState {
            config: Arc::new(crate::core::config::Config::defaults()),
            token: "test-token".to_string(),
            dedup: crate::daemon::dedup::DedupStore::new(),
            cline_prompts: Default::default(),
            event_logger,
            privacy_filter: crate::privacy::PrivacyFilter::new(&[]),
            event_counter: std::sync::atomic::AtomicU64::new(0),
            shutdown_tx: tokio::sync::Mutex::new(None),
            started_at: std::time::Instant::now(),
            available_update: std::sync::Mutex::new(None),
            cloud_tx: None,
            cloud_state: None,
            credential_provider,
            local_ipv4: None,
            local_ipv6: None,
            public_ipv4: None,
            public_ipv6: None,
            tamper_logger: None,
            outbox: None,
            update_in_progress: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            update_status: Arc::new(std::sync::Mutex::new(UpdateStatusSnapshot::idle())),
            update_progress: Arc::default(),
            admin_shutdown_request: Arc::new(tokio::sync::Notify::new()),
            restart_into: std::sync::OnceLock::new(),
            last_hook_at_unix_secs: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            hooks_in_flight: Arc::new(std::sync::atomic::AtomicU32::new(0)),
            content_hash_cache: Arc::new(crate::daemon::config_monitor::ContentHashCache::new(16)),
            config_monitor: crate::daemon::config_monitor::ConfigMonitorRuntime::disabled(),
            pending_alerts: Arc::new(crate::daemon::config_monitor::PendingAlerts::new()),
            policy: None,
            registry: Arc::new(crate::model_relay::session::SessionRegistry::default()),
            health: Arc::new(crate::core::supervision::task::HealthRegistry::new()),
            egress: crate::egress::EgressState::new(&crate::egress::EgressConfig::direct()),
        })
    }

    /// Issue #306, contract 2, at the actual call site: `POST
    /// /admin/auth/refresh` is the handler `auth login` notifies, and it must
    /// invalidate the SAME credential provider the worker polls so the next
    /// `retrieve()` observes a login that happened in another process,
    /// instead of the process-local memoized answer. Exactly one
    /// invalidation per refresh — the route is not a polling loop.
    #[tokio::test]
    async fn admin_auth_refresh_invalidates_the_provider_so_the_next_retrieve_sees_the_login() {
        let provider = MemoizingTestProvider::new("old-key");
        let state = test_app_state(Some(provider.clone()));

        // First read memoizes "old-key" — standing in for the daemon's
        // process reading the credential once, exactly as the cloud worker
        // does at startup.
        let first = state
            .credential_provider
            .as_ref()
            .expect("provider seeded")
            .retrieve()
            .expect("key present");
        assert_eq!(first.expose_secret(), "old-key");

        // `auth login` runs in a different process and writes "new-key" to
        // the real keychain, then POSTs /admin/auth/refresh.
        provider.set_underlying("new-key");
        let _ = handle_admin_auth_refresh(State(state.clone())).await;

        assert_eq!(
            provider.invalidations(),
            1,
            "the refresh route must invalidate the provider exactly once"
        );

        let refreshed = state
            .credential_provider
            .as_ref()
            .expect("provider seeded")
            .retrieve()
            .expect("key present");
        assert_eq!(
            refreshed.expose_secret(),
            "new-key",
            "the next retrieve must observe the login, not the memoized old key"
        );
    }

    /// Serve the real `/admin/*` router — bearer auth, body cap and all — on a
    /// loopback port, the way the daemon mounts it.
    async fn serve_admin() -> String {
        let state = test_app_state(None);
        let app = router(state.clone()).with_state(state);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("addr");
        tokio::spawn(async move {
            let _ = axum::serve(listener, app).await;
        });
        format!("http://{addr}{}", crate::telemetry::handoff::PATH)
    }

    fn handoff_event(name: &str) -> serde_json::Value {
        serde_json::json!({
            "event": name,
            "properties": {
                "distinct_id": "agt_cli",
                "agent_id": "agt_cli",
                "command": "status",
            }
        })
    }

    async fn post(url: &str, token: &str, body: Vec<u8>) -> reqwest::Response {
        crate::egress::client()
            .post(url)
            .bearer_auth(token)
            .header("content-type", "application/json")
            .body(body)
            .send()
            .await
            .expect("request")
    }

    #[tokio::test]
    async fn telemetry_handoff_accepts_a_batch_of_telemetry_events() {
        let url = serve_admin().await;
        let body = serde_json::json!({ "events": [handoff_event("command_invoked")] });
        let resp = post(&url, "test-token", body.to_string().into_bytes()).await;
        assert_eq!(resp.status(), StatusCode::NO_CONTENT);
    }

    #[tokio::test]
    async fn telemetry_handoff_is_behind_the_admin_token() {
        let url = serve_admin().await;
        let body = serde_json::json!({ "events": [handoff_event("command_invoked")] });
        let resp = post(&url, "wrong-token", body.to_string().into_bytes()).await;
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn telemetry_handoff_refuses_anything_but_telemetry_events() {
        let url = serve_admin().await;
        let mut missing_identity = handoff_event("command_invoked");
        missing_identity["properties"]
            .as_object_mut()
            .expect("object")
            .remove("distinct_id");
        for body in [
            serde_json::json!({ "events": [missing_identity] }),
            serde_json::json!({ "events": [handoff_event("Not An Event")] }),
            serde_json::json!({ "events": [] }),
            serde_json::json!({ "events": [handoff_event("command_invoked")], "api_key": "phc_x" }),
            serde_json::json!(["command_invoked"]),
        ] {
            let resp = post(&url, "test-token", body.to_string().into_bytes()).await;
            assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "body: {body}");
            let error: serde_json::Value = resp.json().await.expect("json");
            assert_eq!(
                error["error"]["code"],
                crate::error::ERR_TELEMETRY_HANDOFF_REJECTED
            );
        }
    }

    #[tokio::test]
    async fn telemetry_handoff_refuses_a_body_over_the_byte_cap() {
        let url = serve_admin().await;
        let mut event = handoff_event("command_invoked");
        event["properties"]["padding"] =
            serde_json::json!("x".repeat(crate::telemetry::handoff::MAX_BYTES));
        let body = serde_json::json!({ "events": [event] });
        let resp = post(&url, "test-token", body.to_string().into_bytes()).await;
        assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
    }

    async fn update_status_json(state: &Arc<AppState>) -> serde_json::Value {
        let Json(snap) = handle_admin_update_status(State(state.clone())).await;
        serde_json::to_value(snap).expect("serialise")
    }

    #[tokio::test]
    async fn update_status_omits_byte_counts_until_the_download_reports_them() {
        let state = test_app_state(None);
        state.update_progress.begin();
        *state.update_status.lock().unwrap() = UpdateStatusSnapshot::in_progress("0.1.0", "0.2.0");

        let body = update_status_json(&state).await;
        assert_eq!(body["status"], "in_progress");
        assert_eq!(body["stage"], "check");
        assert!(body.get("bytes_done").is_none(), "{body}");
        assert!(body.get("bytes_total").is_none(), "{body}");
    }

    #[tokio::test]
    async fn update_status_reads_stage_and_bytes_live_from_the_pipeline() {
        let state = test_app_state(None);
        state.update_progress.begin();
        *state.update_status.lock().unwrap() = UpdateStatusSnapshot::in_progress("0.1.0", "0.2.0");

        // Serve 1 KiB of tarball through the real download path.
        let payload = vec![1_u8; 1024];
        let mut server = mockito::Server::new_async().await;
        let _m = server
            .mock("GET", "/t.tgz")
            .with_body(&payload)
            .create_async()
            .await;
        let sri = {
            use base64::Engine;
            use sha2::Digest;
            format!(
                "sha512-{}",
                base64::engine::general_purpose::STANDARD.encode(sha2::Sha512::digest(&payload))
            )
        };
        state.update_progress.set_stage(ApplyStage::Download);
        update::download_tarball(
            &format!("{}/t.tgz", server.url()),
            &sri,
            Duration::from_secs(5),
            &crate::egress::EgressConfig::direct(),
            &state.update_progress,
        )
        .await
        .expect("download");

        let body = update_status_json(&state).await;
        assert_eq!(body["stage"], "download");
        assert_eq!(body["bytes_done"], 1024);
        assert_eq!(body["bytes_total"], 1024);
    }

    /// The daemon that swapped itself has not restarted yet, so nothing in
    /// its apply task past the swap may say `completed`: the new daemon
    /// does, once its health probe passes.
    #[test]
    fn run_apply_in_daemon_never_reports_completed_after_the_swap() {
        let source = include_str!("admin.rs");
        let body = source
            .split("pub(crate) async fn run_apply_in_daemon")
            .nth(1)
            .and_then(|rest| rest.split("\nfn release_lock").next())
            .expect("run_apply_in_daemon body");
        let after_swap = body
            .split("update::perform_swap(")
            .nth(1)
            .expect("swap call");
        assert!(!after_swap.contains("UpdateStatusKind::Completed"));
        assert!(after_swap.contains("stamp_stage(ApplyStage::Restart)"));
        // The task the drain tears down must not own the hand-over: it only
        // names the swapped binary, before the notify that starts the drain.
        let request = after_swap.find("restart_into.set(").expect("request");
        let drain = after_swap.find("notify_waiters()").expect("drain");
        assert!(request < drain, "the request must precede the drain");
    }
}