basemind 0.23.1

//! CLI helpers for the agent-comms broker lifecycle: the shell statusline snapshot and the
//! `basemind comms` lifecycle subcommands (daemon / start / stop / status). Extracted from
//! `main.rs` to keep the binary root under the module-size cap; behavior is unchanged. Most items
//! are gated on the `comms` feature; `cmd_statusline` compiles unconditionally and is a no-op
//! without it.

#[cfg(all(feature = "comms", any(unix, windows)))]
use anyhow::Context;
use anyhow::Result;

/// Print a statusline. Two modes:
///
/// - `root == Some(path)` (invoked as `basemind statusline --root <path>`): render the compact
///   per-repo line for that workspace, read CHEAPLY from the `status.json` sidecar + `telemetry.jsonl`
///   — never opening the Fjall index (no [`basemind::store::Store::open`], no index recovery), so it
///   is safe to refresh every few seconds. This is the path the shell plugin delegates to when the
///   index lives in the machine-global cache (nothing in the repo to read).
/// - `root == None` (invoked as bare `basemind statusline`): the daemon hot-workspace summary
///   (unchanged). Fast and silent: a missing daemon prints nothing and exits 0. Without the `comms`
///   feature there is no daemon, so that path is a no-op.
pub(crate) fn cmd_statusline(root: Option<&std::path::Path>) -> Result<()> {
    if let Some(root) = root {
        println!("{}", render_repo_statusline(root));
        return Ok(());
    }
    #[cfg(all(feature = "comms", any(unix, windows)))]
    {
        use basemind::comms::client::CommsClient;
        use basemind::comms::ids::AgentId;
        use basemind::comms::singleton;

        let line = (|| -> Option<String> {
            let paths = singleton::resolve_paths().ok()?;
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .ok()?;
            runtime.block_on(async move {
                let agent = AgentId::parse("basemind-statusline").ok()?;
                let mut client = CommsClient::connect(&paths, agent, None, None).await.ok()?;
                let hot = client.accessed_paths().await.ok()?;
                Some(format_statusline(&hot))
            })
        })();
        if let Some(line) = line {
            println!("{line}");
        }
    }
    Ok(())
}

// ANSI palette mirroring `.claude-plugin/statusline.sh` so the delegated line matches the shell's
// aesthetic. True-color brand orange (#F97316) + 256-color accents; a single `\x1b[0m` resets each span.
const BRAND: &str = "\x1b[38;2;249;115;22m";
const CYAN: &str = "\x1b[38;5;51m";
const MAGENTA: &str = "\x1b[38;5;201m";
const LABEL: &str = "\x1b[38;5;255m";
const SEP: &str = "\x1b[38;5;240m";
const BOLD: &str = "\x1b[1m";
const RESET: &str = "\x1b[0m";
const BRAND_GLYPH: &str = "◆";

/// The `◆ basemind` brand mark, matching the shell renderer's `mark()`.
fn brand_mark() -> String {
    format!("{BRAND}{BRAND_GLYPH}{RESET} {BOLD}{BRAND}basemind{RESET}")
}

/// Render the compact per-repo statusline for `root`, reading ONLY the cheap `status.json` sidecar
/// and `telemetry.jsonl` tail — never opening the index. When the workspace has no sidecar (never
/// scanned, or an unrecognized schema), returns the same "no index" hint the shell shows so the bar
/// is never blank.
fn render_repo_statusline(root: &std::path::Path) -> String {
    use basemind::store::{read_status_sidecar, workspace_cache_dir};

    let basemind_dir = workspace_cache_dir(root);
    let Some(status) = read_status_sidecar(&basemind_dir) else {
        return format!(
            "{} {SEP}│{RESET} {LABEL}no index — run:{RESET} {BOLD}{CYAN}basemind scan{RESET}",
            brand_mark()
        );
    };

    let age = format_scan_age(status.scanned_unix);
    let (calls, saved) = telemetry_today(&basemind_dir);

    let mut out = format!(
        "{}  {BOLD}{CYAN}{}{RESET} {LABEL}files{RESET} {SEP}·{RESET} {BOLD}{CYAN}{age}{RESET}",
        brand_mark(),
        fmt_count(status.file_count as u64),
    );
    out.push_str(&format!(
        "  {SEP}│{RESET}  {BOLD}{MAGENTA}{}{RESET} {LABEL}calls{RESET} {SEP}·{RESET} {BOLD}{MAGENTA}{}{RESET} {LABEL}saved{RESET}",
        fmt_count(calls),
        fmt_count(saved),
    ));
    out
}

/// Human-readable age of a Unix-epoch-seconds scan timestamp (`Ns/Nm/Nh/Nd ago`), mirroring the
/// shell renderer's buckets. `"never"` when the timestamp is non-positive or in the future.
fn format_scan_age(scanned_unix: i64) -> String {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    let delta = now - scanned_unix;
    if scanned_unix <= 0 || delta < 0 {
        return "never".to_string();
    }
    if delta < 60 {
        format!("{delta}s ago")
    } else if delta < 3_600 {
        format!("{}m ago", delta / 60)
    } else if delta < 86_400 {
        format!("{}h ago", delta / 3_600)
    } else {
        format!("{}d ago", delta / 86_400)
    }
}

/// One telemetry row, read for its two aggregate fields only. Unknown fields are ignored by serde,
/// so this stays forward-compatible with the full `TelemetryRow` schema without coupling to it.
#[derive(serde::Deserialize)]
struct StatuslineTelemetryRow {
    ts_micros: i64,
    #[serde(default)]
    est_tokens_saved: u64,
}

/// Aggregate today's `(calls, est_tokens_saved)` from `telemetry.jsonl`, tailing the last rows and
/// counting those within the last 24h — the same "today" window the MCP telemetry summary uses.
/// Best-effort: a missing/unreadable log yields `(0, 0)`.
fn telemetry_today(basemind_dir: &std::path::Path) -> (u64, u64) {
    use std::io::{BufRead, BufReader};

    const TAIL_ROWS: usize = 2_000;
    const DAY_MICROS: i64 = 24 * 3_600 * 1_000_000;

    let now_micros = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| i64::try_from(d.as_micros()).unwrap_or(i64::MAX))
        .unwrap_or(0);
    let cutoff = now_micros.saturating_sub(DAY_MICROS);

    let Ok(file) = std::fs::File::open(basemind_dir.join("telemetry.jsonl")) else {
        return (0, 0);
    };
    let mut tail: std::collections::VecDeque<StatuslineTelemetryRow> =
        std::collections::VecDeque::with_capacity(TAIL_ROWS);
    for line in BufReader::new(file).lines().map_while(Result::ok) {
        if line.trim().is_empty() {
            continue;
        }
        if let Ok(row) = serde_json::from_str::<StatuslineTelemetryRow>(&line) {
            if tail.len() == TAIL_ROWS {
                tail.pop_front();
            }
            tail.push_back(row);
        }
    }
    let mut calls = 0u64;
    let mut saved = 0u64;
    for row in tail.iter().filter(|r| r.ts_micros >= cutoff) {
        calls += 1;
        saved = saved.saturating_add(row.est_tokens_saved);
    }
    (calls, saved)
}

/// Compact count formatting mirroring the shell renderer's `fmt_count`: plain under 1k, one-decimal
/// `k` under 10k, integer `k` under 1M, integer `M` beyond.
fn fmt_count(n: u64) -> String {
    if n < 1_000 {
        format!("{n}")
    } else if n < 10_000 {
        format!("{}.{}k", n / 1_000, (n * 10 / 1_000) % 10)
    } else if n < 1_000_000 {
        format!("{}k", n / 1_000)
    } else {
        format!("{}M", n / 1_000_000)
    }
}

/// Render the daemon's hot-workspace snapshot into one compact line (e.g. `bm: web · api +2 · 5
/// hot`). An empty set — daemon up but nothing hot — reads `bm: idle`. Names are the workspace
/// directory basenames; the list is capped so the line stays short regardless of the hot count.
#[cfg(all(feature = "comms", any(unix, windows)))]
fn format_statusline(workspaces: &[basemind::comms::workspace_pool::AccessedWorkspace]) -> String {
    if workspaces.is_empty() {
        return "bm: idle".to_string();
    }
    const MAX_NAMES: usize = 3;
    let names: Vec<&str> = workspaces
        .iter()
        .take(MAX_NAMES)
        .map(|w| w.root.file_name().and_then(|n| n.to_str()).unwrap_or("?"))
        .collect();
    let mut label = names.join(" · ");
    if workspaces.len() > MAX_NAMES {
        label.push_str(&format!(" +{}", workspaces.len() - MAX_NAMES));
    }
    format!("bm: {label} · {} hot", workspaces.len())
}

/// Dispatch a comms lifecycle subcommand. Each command drives a small current-thread tokio
/// runtime — the broker daemon itself uses a multi-thread runtime so concurrent links don't
/// serialize.
#[cfg(all(feature = "comms", any(unix, windows)))]
pub(crate) fn cmd_comms(root: &std::path::Path, action: crate::CommsLifecycleCmd, json: bool) -> Result<()> {
    match action {
        crate::CommsLifecycleCmd::Daemon => basemind::cli::comms_daemon::run(),
        crate::CommsLifecycleCmd::Start => cmd_comms_start(),
        crate::CommsLifecycleCmd::Stop => cmd_comms_lifecycle_rpc(CommsRpc::Stop, json),
        crate::CommsLifecycleCmd::Status => cmd_comms_lifecycle_rpc(CommsRpc::Status, json),
        crate::CommsLifecycleCmd::Agent(cmd) => basemind::cli::comms::run(root, json, cmd),
    }
}

#[cfg(all(feature = "comms", any(unix, windows)))]
enum CommsRpc {
    Stop,
    Status,
}

/// How long `daemon ensure` waits for the streamable-HTTP transport to answer after ensuring the
/// daemon is up. Generous relative to a cold daemon spawn + bind.
#[cfg(all(feature = "comms", any(unix, windows)))]
const HTTP_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

/// Dispatch a `basemind daemon` subcommand.
#[cfg(all(feature = "comms", any(unix, windows)))]
pub(crate) fn cmd_daemon(action: crate::DaemonCmd, json: bool) -> Result<()> {
    match action {
        crate::DaemonCmd::Ensure => cmd_daemon_ensure(json),
    }
}

/// Ensure the daemon is running and its streamable-HTTP MCP transport is ready, then print the base
/// URL. This is what a launcher/hook calls; it only implements the verb (no manifest wiring here).
#[cfg(all(feature = "comms", any(unix, windows)))]
fn cmd_daemon_ensure(json: bool) -> Result<()> {
    use basemind::comms::http_frontend;
    use basemind::comms::singleton;

    let paths = singleton::resolve_paths().context("resolve comms paths")?;
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime")?;

    let addr = runtime.block_on(async move {
        singleton::ensure_daemon(&paths)
            .await
            .map_err(|e| anyhow::anyhow!("ensure comms daemon: {e}"))?;
        http_frontend::await_http_ready(&paths.comms_dir, HTTP_READY_TIMEOUT)
            .await
            .context("wait for streamable-HTTP MCP transport")
    })?;

    let url = http_frontend::base_url(&addr);
    if json {
        println!("{{\"ready\":true,\"addr\":\"{addr}\",\"url\":\"{url}\"}}");
    } else {
        println!("{url}");
    }
    Ok(())
}

/// Ensure a daemon is running, spawning it detached if needed.
#[cfg(all(feature = "comms", any(unix, windows)))]
fn cmd_comms_start() -> Result<()> {
    use basemind::comms::singleton;
    let paths = singleton::resolve_paths().context("resolve comms paths")?;
    let socket_path = paths.socket_path.clone();
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime")?;
    runtime.block_on(async move {
        singleton::ensure_daemon(&paths)
            .await
            .map_err(|e| anyhow::anyhow!("ensure comms daemon: {e}"))
    })?;
    println!("comms daemon is running ({})", socket_path.display());
    Ok(())
}

/// Connect to the running daemon and issue a Stop or Status RPC.
#[cfg(all(feature = "comms", any(unix, windows)))]
fn cmd_comms_lifecycle_rpc(rpc: CommsRpc, json: bool) -> Result<()> {
    use basemind::comms::client::CommsClient;
    use basemind::comms::singleton;

    let paths = singleton::resolve_paths().context("resolve comms paths")?;
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime")?;

    runtime.block_on(async move {
        let root = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
        let agent = basemind::comms::identity::cli_agent_id(&root);
        let mut client = CommsClient::connect(&paths, agent, None, None)
            .await
            .map_err(|e| anyhow::anyhow!("connect to comms daemon: {e}"))?;
        match rpc {
            CommsRpc::Stop => {
                client.stop().await.map_err(|e| anyhow::anyhow!("stop: {e}"))?;
                if json {
                    println!("{{\"stopped\":true}}");
                } else {
                    println!("comms daemon stopping");
                }
            }
            CommsRpc::Status => {
                let status = client.status().await.map_err(|e| anyhow::anyhow!("status: {e}"))?;
                if json {
                    println!(
                        "{}",
                        serde_json::to_string(&status).map_err(|e| anyhow::anyhow!("serialize status: {e}"))?
                    );
                } else {
                    println!(
                        "pid={} version={} proto={} uptime={}s threads={} subscribers={}",
                        status.pid,
                        status.version,
                        status.proto_ver,
                        status.uptime_secs,
                        status.threads,
                        status.subscribers,
                    );
                }
            }
        }
        Ok::<(), anyhow::Error>(())
    })?;
    Ok(())
}

#[cfg(all(test, feature = "comms", any(unix, windows)))]
mod statusline_tests {
    use std::path::PathBuf;

    use basemind::comms::workspace_pool::AccessedWorkspace;

    fn ws(root: &str) -> AccessedWorkspace {
        AccessedWorkspace {
            root: PathBuf::from(root),
            key: "k".to_string(),
            idle_secs: 0,
        }
    }

    #[test]
    fn empty_hot_set_reads_idle() {
        assert_eq!(super::format_statusline(&[]), "bm: idle");
    }

    #[test]
    fn lists_workspace_basenames_and_the_hot_count() {
        let hot = [ws("/repos/web"), ws("/repos/api")];
        assert_eq!(super::format_statusline(&hot), "bm: web · api · 2 hot");
    }

    #[test]
    fn caps_the_name_list_with_an_overflow_marker() {
        let hot = [ws("/a/one"), ws("/a/two"), ws("/a/three"), ws("/a/four"), ws("/a/five")];
        assert_eq!(super::format_statusline(&hot), "bm: one · two · three +2 · 5 hot");
    }
}