trusty-mpm 1.7.2

//! HTTP transport and JSON projections for one daemon's health endpoints.
//!
//! Why: the background poller needs a small, testable transport that yields a
//! projected [`PanelData`] or a clean error string, plus the per-daemon list
//! endpoints that feed the collections pane. Keeping the projections pure lets
//! them be unit-tested without a live daemon.
//! What: the [`HealthClient`] impl (poll / fetch / count / list / stop methods),
//! the [`client_for`] constructor, and the `project_*` helpers that turn raw
//! JSON payloads into typed counts and [`CollectionRow`]s.
//! Test: `health_client_stores_base_url`, `poll_unreachable_daemon_is_offline`,
//! `project_memory_counts_reads_status_fields`, `project_palace_rows_reads_palaces`,
//! `project_log_tail_reads_fields`, `project_edge_kinds_sorts_desc`.

use std::time::Duration;

use serde_json::{Value, json};

use crate::tui::health::format::format_relative_time;
use crate::tui::health::types::{
    CollectionRow, Daemon, HealthClient, HealthWire, PanelData, PanelState,
};

/// Per-request timeout for a daemon health probe.
///
/// Why: a hung daemon must not stall the poll task; a short timeout turns an
/// unresponsive daemon into a clean "offline" state on the next tick.
/// What: three seconds, comfortably above a healthy local round-trip.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(3);

impl HealthClient {
    /// Build a client targeting `base` for the given `daemon`.
    ///
    /// Why: the health screen is pointed at a daemon address from a CLI flag or
    /// the documented default.
    /// What: stores the base URL and a pooled `reqwest::Client` whose request
    /// timeout bounds a hung daemon.
    /// Test: `health_client_stores_base_url`.
    pub fn new(base: impl Into<String>, daemon: Daemon) -> Self {
        let http = reqwest::Client::builder()
            .timeout(REQUEST_TIMEOUT)
            .build()
            .unwrap_or_default();
        Self {
            base: base.into(),
            daemon,
            http,
        }
    }

    /// The address this client targets — a base URL for trusty-search, a
    /// socket path for trusty-memory (#6286).
    ///
    /// Why: the offline panel renders the daemon address it failed to reach.
    /// Test: `health_client_stores_base_url`.
    pub fn base_url(&self) -> &str {
        &self.base
    }

    /// Poll the daemon and project the result into a [`PanelState`].
    ///
    /// Why: the background task wants one infallible call per tick that always
    /// yields a renderable state — `Online` on success, `Offline` on any
    /// transport or decode failure.
    /// What: GETs `/health`, then the daemon-specific list endpoints for the
    /// key counts, folding everything into [`PanelData`]. Any error along the
    /// way becomes `Offline` carrying the error string.
    /// Test: live behaviour is covered by the daemon suites; the offline path
    /// is exercised by `poll_unreachable_daemon_is_offline`.
    /// One RPC call on trusty-memory's socket (#6286).
    ///
    /// Why: this client's `base` is a socket path when `daemon` is
    /// `Daemon::Memory`, and every memory leg below goes through here rather
    /// than through `self.http`, which has nothing left to talk to.
    async fn memory_call(&self, method: &str, params: serde_json::Value) -> anyhow::Result<Value> {
        trusty_common::memory_rpc::call_memory_tool_at_with_timeout(
            std::path::Path::new(&self.base),
            method,
            params,
            REQUEST_TIMEOUT,
        )
        .await
    }

    pub async fn poll(&self) -> PanelState {
        match self.fetch().await {
            Ok(data) => PanelState::Online(data),
            Err(e) => PanelState::Offline {
                last_error: e.to_string(),
            },
        }
    }

    /// Fetch and project the panel payload, returning a `Result` for `?`.
    ///
    /// Why: keeps [`Self::poll`]'s error-to-`Offline` mapping in one place
    /// while the happy path stays terse with `?`.
    /// What: reads the daemon's health and its list surface; for search the
    /// counts are index count + summed chunk counts, for memory they come from
    /// `memory.status`. Since #6286 the two use different transports — search
    /// is still HTTP, memory is the socket.
    /// Test: covered indirectly by `poll`; the count projections are unit-tested
    /// via `project_search_counts` / `project_memory_counts`.
    async fn fetch(&self) -> anyhow::Result<PanelData> {
        let health: HealthWire = match self.daemon {
            Daemon::Search => {
                self.http
                    .get(format!("{}/health", self.base))
                    .send()
                    .await?
                    .error_for_status()?
                    .json()
                    .await?
            }
            Daemon::Memory => {
                serde_json::from_value(self.memory_call("memory.health", json!({})).await?)?
            }
        };

        let (count_a, count_b, count_c, count_d) = match self.daemon {
            Daemon::Search => self.search_counts().await,
            Daemon::Memory => self.memory_counts().await,
        };

        Ok(PanelData {
            version: health.version,
            rss_mb: health.rss_mb,
            cpu_pct: health.cpu_pct,
            uptime_secs: health.uptime_secs,
            disk_bytes: health.disk_bytes,
            count_a,
            count_b,
            count_c,
            count_d,
        })
    }

    /// Resolve the search key counts: `(indexes, total_chunks, 0, 0)`.
    ///
    /// Why: the search panel shows index count and summed chunk count; a
    /// failure to enumerate indexes degrades to zeroes rather than failing the
    /// whole poll, since the resource block already rendered.
    /// What: GETs `/indexes`, then `/indexes/:id/status` per index, summing
    /// `chunk_count`. Any error yields all zeroes.
    /// Test: the JSON projection is unit-tested via `project_search_counts`.
    async fn search_counts(&self) -> (u64, u64, u64, u64) {
        let Ok(list) = self.get_json(format!("{}/indexes", self.base)).await else {
            return (0, 0, 0, 0);
        };
        let ids = list
            .get("indexes")
            .and_then(|v| v.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|v| v.as_str().map(str::to_string))
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();
        let mut total_chunks = 0u64;
        for id in &ids {
            if let Ok(status) = self
                .get_json(format!("{}/indexes/{id}/status", self.base))
                .await
            {
                total_chunks = total_chunks.saturating_add(
                    status
                        .get("chunk_count")
                        .and_then(|v| v.as_u64())
                        .unwrap_or(0),
                );
            }
        }
        (ids.len() as u64, total_chunks, 0, 0)
    }

    /// Resolve the memory key counts from `memory.status`.
    ///
    /// Why: the memory panel shows palaces, vectors, drawers, and KG triples;
    /// one method returns all four.
    /// What: calls `memory.status` and projects `palace_count`,
    /// `total_vectors`, `total_drawers`, `total_kg_triples`. Any error yields
    /// all zeroes.
    /// Test: the JSON projection is unit-tested via `project_memory_counts`.
    async fn memory_counts(&self) -> (u64, u64, u64, u64) {
        match self.memory_call("memory.status", json!({})).await {
            Ok(status) => project_memory_counts(&status),
            Err(_) => (0, 0, 0, 0),
        }
    }

    /// GET `url` and decode the response body as JSON.
    ///
    /// Why: the count probes share the same GET-and-decode shape.
    /// What: GETs `url`, maps a non-2xx response to an error, and decodes the
    /// body into a [`serde_json::Value`].
    /// Test: covered indirectly by `search_counts` / `memory_counts`.
    async fn get_json(&self, url: String) -> anyhow::Result<serde_json::Value> {
        Ok(self
            .http
            .get(url)
            .send()
            .await?
            .error_for_status()?
            .json()
            .await?)
    }

    /// Fetch the most recent `n` log lines from the daemon.
    ///
    /// Why: the Logs tab (`[2]`) tails the daemon's in-memory log ring via
    /// the daemon's in-memory log ring (issue #35). The two daemons no longer
    /// share one endpoint: search still serves `GET /logs/tail?n=…`, memory
    /// answers `memory.logs_tail` (#6286).
    /// What: reads the daemon's log page and projects `lines` + `total`. A
    /// daemon that cannot answer yields `Ok((vec![], 0))` rather than an error,
    /// so the tab degrades to a placeholder cleanly.
    /// Test: live behaviour is covered by the daemon suites; the projection
    /// is unit-tested via `project_log_tail`.
    pub async fn logs_tail(&self, n: u32) -> anyhow::Result<(Vec<String>, u64)> {
        let raw = match self.daemon {
            Daemon::Memory => self
                .memory_call("memory.logs_tail", json!({ "n": n }))
                .await
                .unwrap_or(Value::Null),
            Daemon::Search => {
                let url = format!("{}/logs/tail?n={n}", self.base);
                match self.http.get(url).send().await {
                    Ok(resp) if resp.status().is_success() => {
                        resp.json::<Value>().await.unwrap_or(Value::Null)
                    }
                    _ => Value::Null,
                }
            }
        };
        Ok(project_log_tail(&raw))
    }

    /// Fetch the search daemon's index list with chunk counts.
    ///
    /// Why: the Collections list (left panel for the search service) wants a
    /// per-index name + chunk count so the operator can see at a glance
    /// which corpora are loaded.
    /// What: GETs `/indexes`, then `GET /indexes/:id/status` per index,
    /// projecting `(id, chunk_count)` into [`CollectionRow`]s. Any error
    /// yields an empty list rather than failing.
    /// Test: live behaviour is covered by the daemon suites; the projection
    /// is unit-tested via `project_index_rows`.
    pub async fn search_collections(&self) -> Vec<CollectionRow> {
        let Ok(list) = self.get_json(format!("{}/indexes", self.base)).await else {
            return Vec::new();
        };
        let ids: Vec<String> = list
            .get("indexes")
            .and_then(|v| v.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|v| v.as_str().map(str::to_string))
                    .collect()
            })
            .unwrap_or_default();
        let mut rows = Vec::with_capacity(ids.len());
        for id in ids {
            // Status: chunk count + last_indexed + disk bytes + context embedding.
            let status = self
                .get_json(format!("{}/indexes/{id}/status", self.base))
                .await
                .ok();
            let count = status
                .as_ref()
                .and_then(|v| v.get("chunk_count").and_then(|c| c.as_u64()))
                .unwrap_or(0);
            let last_indexed = status
                .as_ref()
                .and_then(|v| v.get("last_indexed").and_then(|c| c.as_str()))
                .map(str::to_string);
            let disk_bytes = status
                .as_ref()
                .and_then(|v| v.get("disk_bytes").and_then(|c| c.as_u64()))
                .unwrap_or(0);
            let has_context_embedding = status
                .as_ref()
                .and_then(|v| v.get("has_context_embedding").and_then(|c| c.as_bool()))
                .unwrap_or(false);

            // Graph stats: nodes, edges, edge kind histogram. Errors → zeroes.
            let graph = self
                .get_json(format!("{}/indexes/{id}/graph/stats", self.base))
                .await
                .ok();
            let node_count = graph
                .as_ref()
                .and_then(|v| v.get("node_count").and_then(|c| c.as_u64()))
                .unwrap_or(0);
            let edge_count = graph
                .as_ref()
                .and_then(|v| v.get("edge_count").and_then(|c| c.as_u64()))
                .unwrap_or(0);
            let edge_kinds = graph.as_ref().map(project_edge_kinds).unwrap_or_default();

            // Communities: only the top-level summary fields are needed.
            let communities = self
                .get_json(format!("{}/indexes/{id}/communities", self.base))
                .await
                .ok();
            let community_count = communities
                .as_ref()
                .and_then(|v| v.get("community_count").and_then(|c| c.as_u64()))
                .unwrap_or(0);
            let modularity = communities
                .as_ref()
                .and_then(|v| v.get("modularity").and_then(|c| c.as_f64()))
                .unwrap_or(0.0);

            let note = format_relative_time(last_indexed.as_deref());
            rows.push(CollectionRow {
                id,
                count,
                note,
                ok: true,
                last_indexed,
                node_count,
                edge_count,
                edge_kinds,
                community_count,
                modularity,
                disk_bytes,
                has_context_embedding,
                ..Default::default()
            });
        }
        rows
    }

    /// Fetch the memory daemon's palace roster for the Collections panel.
    ///
    /// Why: the Collections list (left panel for the memory service) needs the
    /// per-palace name, vector count, and KG triple count so the operator can
    /// see at a glance which palaces hold the most memory and which carry a
    /// knowledge graph. `memory.status` only exposes aggregate totals.
    ///
    /// What (#6286): one `memory.palaces_list` call. This fanned out
    /// `memory.palace_get` per id, because nothing on the socket answered a
    /// roster WITH counts — `palace_list` answers bare ids and the retired
    /// `GET /api/v1/palaces` reported peek-based placeholder zeros for any
    /// unresident palace (#4682). The fan-out also dropped a palace whose call
    /// failed, silently; `memory.palaces_list` carries a per-palace error
    /// instead, and [`project_palace_roster`] renders it.
    ///
    /// Why `counts: false` (#7125): this is a health probe on a 5-second
    /// timer, and counting is what opens every palace on disk — ~94 cold opens
    /// per tick on the estate that prompted the issue, each one left in the
    /// daemon's 64-slot LRU. A palace the daemon already has open still reports
    /// its real counts; one it does not comes back flagged `cached: false`, and
    /// [`project_palace_rows`] renders that as unknown rather than as empty.
    ///
    /// A whole-call failure still yields an empty list — the panel's other
    /// rows are worth showing, and the daemon's own health line already says
    /// it is unreachable.
    /// Test: `memory_collections_asks_for_a_roster_without_counts`; the
    /// projection is unit-tested via `project_palace_roster`.
    pub async fn memory_collections(&self) -> Vec<CollectionRow> {
        // #7125: a health poll must not open every palace on disk to count it.
        let Ok(listed) = self
            .memory_call("memory.palaces_list", json!({ "counts": false }))
            .await
        else {
            return Vec::new();
        };
        project_palace_roster(&listed)
    }

    /// Request a graceful shutdown of the daemon via its `admin/stop` endpoint.
    ///
    /// Why: the `[X]` key stops the focused daemon without the operator
    /// resolving a PID; both daemons expose an unauthenticated stop route.
    /// What: `POST /admin/stop` for search, `memory.admin_stop` for memory
    /// (#6286). A refusal on either is an error.
    /// Test: live behaviour is covered by the daemon suites; the dashboard
    /// records the outcome string in `last_action`.
    pub async fn stop(&self) -> anyhow::Result<()> {
        match self.daemon {
            Daemon::Search => {
                self.http
                    .post(format!("{}/admin/stop", self.base))
                    .json(&json!({}))
                    .send()
                    .await?
                    .error_for_status()?;
            }
            Daemon::Memory => {
                self.memory_call("memory.admin_stop", json!({})).await?;
            }
        }
        Ok(())
    }
}

/// Build a [`HealthClient`] for the given daemon at the given base URL.
///
/// Why: the background poller and the `[S]`/`[X]` key handlers all need a
/// client; centralising construction keeps the daemon→client mapping in one
/// place.
/// What: returns a [`HealthClient`] tagged with `daemon`.
/// Test: covered by `health_client_stores_base_url`.
pub fn client_for(daemon: Daemon, base_url: &str) -> HealthClient {
    HealthClient::new(base_url, daemon)
}

/// Project a `/api/v1/status` payload into `(palaces, vectors, drawers, kg)`.
///
/// Why: centralising the projection keeps [`HealthClient::memory_counts`]
/// testable without a live daemon and resilient to absent optional fields.
/// What: reads `palace_count`, `total_vectors`, `total_drawers`, and
/// `total_kg_triples`, defaulting any absent field to zero.
/// Test: `project_memory_counts`.
pub(crate) fn project_memory_counts(status: &serde_json::Value) -> (u64, u64, u64, u64) {
    let u = |key: &str| status.get(key).and_then(|v| v.as_u64()).unwrap_or(0);
    (
        u("palace_count"),
        u("total_vectors"),
        u("total_drawers"),
        u("total_kg_triples"),
    )
}

/// Project a `/logs/tail` response into `(lines, total)`.
///
/// Why: keeps the wire-shape parsing in one testable function so the client
/// stays terse and an older daemon's quirky payload cannot crash the TUI.
/// What: reads `lines` (array of strings, defaulting to `[]`) and `total`
/// (u64, defaulting to `lines.len()`).
/// Test: `project_log_tail_reads_fields`.
pub(crate) fn project_log_tail(body: &serde_json::Value) -> (Vec<String>, u64) {
    let lines: Vec<String> = body
        .get("lines")
        .and_then(|v| v.as_array())
        .map(|a| {
            a.iter()
                .filter_map(|v| v.as_str().map(str::to_string))
                .collect()
        })
        .unwrap_or_default();
    let total = body
        .get("total")
        .and_then(|v| v.as_u64())
        .unwrap_or(lines.len() as u64);
    (lines, total)
}

/// Project a `memory.palaces_list` answer into left-panel rows (#6286).
///
/// Why: the roster's rows come in two shapes — a palace the daemon read, and a
/// palace it could not. The second must be RENDERED rather than skipped: the
/// fan-out this replaced dropped it at the call site, so the panel could list
/// fewer palaces than the daemon has with nothing saying which or why.
/// What: a readable row goes through [`project_palace_rows`], keeping its
/// empty-palace filter. A row carrying `error` becomes a row with `ok: false`
/// and the daemon's reason as its note, exempt from that filter — a palace
/// whose counts could not be read must not be mistaken for one holding
/// nothing.
/// Test: `project_palace_roster_reads_counts`,
/// `project_palace_roster_renders_a_failed_row`.
pub(crate) fn project_palace_roster(answer: &serde_json::Value) -> Vec<CollectionRow> {
    let Some(rows) = answer.get("palaces").and_then(|v| v.as_array()) else {
        return Vec::new();
    };
    let mut out = Vec::with_capacity(rows.len());
    for row in rows {
        if let Some(palace) = row.get("palace").filter(|v| v.is_object()) {
            out.extend(project_palace_rows(&serde_json::Value::Array(vec![
                palace.clone(),
            ])));
            continue;
        }
        let Some(error) = row.get("error").and_then(|v| v.as_str()) else {
            continue;
        };
        let id = row
            .get("id")
            .and_then(|v| v.as_str())
            .unwrap_or("?")
            .to_string();
        out.push(CollectionRow {
            id,
            note: format!("counts unavailable: {error}"),
            ok: false,
            // #7125: its zeros are unknown for the same reason an unopened
            // palace's are — render `?v`, not the `--v` that reads as empty.
            counts_unknown: true,
            ..Default::default()
        });
    }
    out
}
/// Project a memory daemon `/api/v1/palaces` payload into palace rows.
///
/// Why: centralising the projection keeps the renderer terse and lets a unit
/// test assert the shape without a live daemon. The wire format is a JSON
/// array of `PalaceInfo` objects with per-palace `vector_count`,
/// `drawer_count`, and `kg_triple_count`. `/api/v1/status` exposes only
/// aggregate totals, so this is the only source of per-palace counts. Empty
/// palaces (no vectors AND no KG triples) are filtered out so the left pane
/// only lists palaces that actually hold memory — an empty palace is
/// indistinguishable from a placeholder and just adds visual noise.
/// What: reads the top-level array, projecting each entry's `name` (falling
/// back to `id`), `vector_count`, and `kg_triple_count` (any absent field
/// defaults to zero). Rows where both counts are zero are dropped. A
/// non-array payload yields an empty list.
///
/// The empty-palace filter reads a ZERO, and a zero only means empty when the
/// daemon actually counted (#4637). Since #7125 the poller asks for
/// `counts: false`, so a palace the daemon has not opened answers `cached:
/// false` with placeholder zeros — that row is flagged
/// [`CollectionRow::counts_unknown`] and exempted from the filter, or the panel
/// would go blank on a cold daemon and call every palace empty.
/// Test: `project_palace_rows_reads_palaces`,
/// `project_palace_rows_filters_empty`,
/// `project_palace_rows_keep_uncached_rows_as_unknown`.
pub(crate) fn project_palace_rows(list: &serde_json::Value) -> Vec<CollectionRow> {
    let Some(arr) = list.as_array() else {
        return Vec::new();
    };
    arr.iter()
        .filter_map(|p| {
            let id = p
                .get("name")
                .or_else(|| p.get("id"))
                .and_then(|v| v.as_str())
                .unwrap_or("?")
                .to_string();
            let count = p.get("vector_count").and_then(|v| v.as_u64()).unwrap_or(0);
            let kg_count = p
                .get("kg_triple_count")
                .and_then(|v| v.as_u64())
                .unwrap_or(0);
            let drawer_count = p.get("drawer_count").and_then(|v| v.as_u64()).unwrap_or(0);
            // #4811: prefer the truthful `room_count` (ADR-0027 T8); an older
            // daemon only sends `wing_count`, which held this same number.
            let room_count = p
                .get("room_count")
                .or_else(|| p.get("wing_count"))
                .and_then(|v| v.as_u64())
                .unwrap_or(0);
            let last_write_at = p
                .get("last_write_at")
                .and_then(|v| v.as_str())
                .map(str::to_string);
            let node_count = p.get("node_count").and_then(|v| v.as_u64()).unwrap_or(0);
            let edge_count = p.get("edge_count").and_then(|v| v.as_u64()).unwrap_or(0);
            let community_count = p
                .get("community_count")
                .and_then(|v| v.as_u64())
                .unwrap_or(0);
            let is_compacting = p
                .get("is_compacting")
                .and_then(|v| v.as_bool())
                .unwrap_or(false);
            // #7125: `cached: false` means the daemon peeked instead of
            // opening, so these zeros are UNKNOWN (#4637), never empty.
            let counts_unknown = p.get("cached").and_then(|v| v.as_bool()) == Some(false);
            // Skip palaces with no vectors and no graph triples: they hold
            // nothing the operator can act on and clutter the left pane.
            if !counts_unknown && count == 0 && kg_count == 0 {
                return None;
            }
            Some(CollectionRow {
                id,
                count,
                kg_count,
                drawer_count,
                room_count,
                last_write_at,
                node_count,
                edge_count,
                community_count,
                is_compacting,
                counts_unknown,
                // Note left empty: the row shows vector + graph counts inline
                // (e.g. `12v 34g`), so a trailing badge would be redundant.
                note: String::new(),
                ok: true,
                ..Default::default()
            })
        })
        .collect()
}

/// Project a `/indexes/:id/graph/stats` payload's `edge_kinds` map into a
/// vec sorted by count descending.
///
/// Why: the INDEX tab renders one row per edge kind, ordered so the heaviest
/// relationship appears at the top; keeping the projection pure makes it
/// testable without a live daemon.
/// What: reads the `edge_kinds` object from `stats`, collects `(name, count)`
/// pairs, and sorts by count descending (ties broken by name ascending).
/// A missing object yields an empty vec.
/// Test: `project_edge_kinds_sorts_desc`.
pub(crate) fn project_edge_kinds(stats: &serde_json::Value) -> Vec<(String, u64)> {
    let Some(map) = stats.get("edge_kinds").and_then(|v| v.as_object()) else {
        return Vec::new();
    };
    let mut pairs: Vec<(String, u64)> = map
        .iter()
        .map(|(k, v)| (k.clone(), v.as_u64().unwrap_or(0)))
        .collect();
    pairs.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
    pairs
}