openlatch-client 0.5.4

//! `/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::State,
    http::StatusCode,
    middleware,
    response::IntoResponse,
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};

use crate::daemon::{auth, AppState};
use crate::install_state;
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),
    /// `min_supported_client` declared by the new release is greater
    /// than the running daemon's version — apply is refused.
    PreconditionFailed {
        latest: String,
        min_supported: 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::PreconditionFailed {
                latest,
                min_supported,
            } => (
                StatusCode::PRECONDITION_FAILED,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_UPDATE_VERIFY_FAILED,
                        "message": format!(
                            "release {latest} requires client >= {min_supported}; manual `npm install -g @openlatch/client@{latest}` required"
                        ),
                        "latest": latest,
                        "min_supported": min_supported,
                    }
                }),
            ),
            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_layer(middleware::from_fn_with_state(state, auth::bearer_auth))
}

/// `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/412/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, min_supported, _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,
            min_supported,
            tarball_url,
            tarball_integrity,
            ..
        } => (
            latest,
            severity,
            min_supported,
            tarball_url,
            tarball_integrity,
        ),
    };

    if let Some(ref min) = min_supported {
        if !update::version_at_least(&current_version, min) {
            crate::telemetry::capture_global(
                crate::telemetry::Event::update_blocked_by_min_supported(
                    &current_version,
                    &latest,
                    min,
                ),
            );
            return Err(AdminError::PreconditionFailed {
                latest,
                min_supported: min.clone(),
            });
        }
    }

    // 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.
    {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        *snap = UpdateStatusSnapshot {
            status: UpdateStatusKind::InProgress,
            stage: Some(ApplyStage::Check),
            from: Some(current_version.clone()),
            to: Some(latest.clone()),
            started_at: Some(install_state::now_rfc3339()),
            ended_at: None,
            error: None,
        };
    }

    // 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,
        };
        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.
async fn handle_admin_update_status(
    State(state): State<Arc<AppState>>,
) -> Json<UpdateStatusSnapshot> {
    let snap = state
        .update_status
        .lock()
        .expect("status mutex poisoned")
        .clone();
    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| {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        snap.stage = Some(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 mark_completed = |duration_ms: u64| {
        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());
        snap.error = None;
        // Duration is in telemetry; the snapshot doesn't carry it.
        let _ = duration_ms;
    };

    stamp_stage(ApplyStage::Check);
    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.
            tracing::info!(target: "update", current = %current, "concurrent check found us up-to-date — releasing lock");
            mark_completed(started.elapsed().as_millis() as u64);
            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;
    }

    // Stage: drain. Trigger graceful shutdown of axum, then race a 5s
    // budget against the listener actually closing. The 1s force-close
    // window mimics the brainstorm doc § 4.
    stamp_stage(ApplyStage::Drain);
    tracing::info!(target: "update", "draining axum prior to restart");
    state.admin_shutdown_request.notify_waiters();
    let drain_deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    tokio::time::sleep_until(drain_deadline).await;
    tokio::time::sleep(Duration::from_secs(1)).await;

    // Telemetry: completion is captured *before* the exec call —
    // execv on Unix never returns, and we want the event to land while
    // the runtime is still alive. Failure of restart_into_new_binary
    // is reported via mark_failed below for the rare case it returns.
    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,
    ));
    mark_completed(duration_ms);

    // Stage: restart. On success this never returns on Unix
    // (execv reuses the PID); on Windows the daemon spawns a detached
    // child and exits the current process. If it does return (rare
    // failure mode like ENOEXEC), we mark_failed so the status surface
    // catches it and leave the daemon running on the OLD binary —
    // the new binary is live on disk and a manual restart picks it up.
    stamp_stage(ApplyStage::Restart);
    if let Err(e) = update::restart_into_new_binary() {
        mark_failed(ApplyStage::Restart, e);
        release_lock(&state);
    }
}

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(),
            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())),
            admin_shutdown_request: Arc::new(tokio::sync::Notify::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"
        );
    }
}