AkurAI-Build

Latest change 3fbc0824e064f703c15581bca65fb7c2f0007d9b - fix(ec2): parse escaped quotes in catalog values, and check real deploy contracts by Ólafur Búi Ólafsson

//! Host-side verbs: `ssh`, `ship`, `status`, `processes`/`ps`, `logs`, `ports`,
//! `allow-ssh`, `health`.
//!
//! Ported verbatim from the `akurai-ec2` bash CLI. The box-side introspection
//! (systemd, ss, nginx) is inherently shell, so those payloads run unchanged on
//! the remote; everything else (registry parsing, drift analysis, health
//! fan-out) is native Rust with the same output contract.

use std::collections::{HashMap, HashSet};
use std::path::Path;
use std::process::{Command, Stdio};

use anyhow::{Context, Result, bail};
use serde_json::Value;

use super::{APP_POOL_HI, APP_POOL_LO, Ec2, say, shell_quote};

/// `status` payload — remote introspection, verbatim from the bash.
const STATUS_SCRIPT: &str = r#"echo "=== app services ==="
systemctl list-units --type=service --state=running | grep -iE 'akurai|golfset|mail|idp|bunfork' || true
echo "=== listening ports ==="
sudo ss -tlnp 2>/dev/null | grep -E 'LISTEN' | awk '{print $4}' | sort -u | tail -20
echo "=== nginx vhosts ==="
ls /etc/nginx/sites-enabled/
echo "=== disk ==="
df -h / | tail -1
"#;

/// `processes` remote script — unit/state/pid/mem/uptime/ports, verbatim.
const PROCESS_SCRIPT: &str = r#"pat="$1"
# pid -> listening port(s), so each unit can show what it's actually serving.
declare -A pid_ports
while read -r line; do
  laddr=$(echo "$line" | awk '{print $4}'); port=${laddr##*:}
  case "$port" in (*[!0-9]*|"") continue;; esac
  pid=$(echo "$line" | grep -oE 'pid=[0-9]+' | head -1 | grep -oE '[0-9]+')
  [ -n "$pid" ] && pid_ports[$pid]+="$port "
done < <(sudo ss -ltnpH 2>/dev/null)

printf '%-24s %-9s %-7s %-8s %-19s %s\n' UNIT STATE PID MEM SINCE PORTS
n=0
for unit in $(systemctl list-units --type=service --state=running --no-legend --plain 2>/dev/null | awk '{print $1}' | grep -iE "$pat" | sort -u); do
  pid=$(systemctl show "$unit" -p MainPID --value 2>/dev/null)
  sub=$(systemctl show "$unit" -p SubState --value 2>/dev/null)
  memraw=$(systemctl show "$unit" -p MemoryCurrent --value 2>/dev/null)
  since=$(systemctl show "$unit" -p ActiveEnterTimestamp --value 2>/dev/null | awk '{print $2" "$3}')
  mem="-"
  case "$memraw" in (''|'[not set]'|18446744073709551615) ;; (*) mem=$(numfmt --to=iec "$memraw" 2>/dev/null || echo "$memraw");; esac
  ports="${pid_ports[$pid]:-—}"
  printf '%-24s %-9s %-7s %-8s %-19s %s\n' "${unit%.service}" "$sub" "$pid" "$mem" "$since" "$ports"
  n=$((n+1))
done
echo
echo "total running: $n"
"#;

/// `ports` live-state fetch — port->unit (via cgroup) and vhost->port, verbatim.
const RFETCH_SCRIPT: &str = r####"echo "### PORTS"
sudo ss -ltnpH 2>/dev/null | while read -r line; do
  laddr=$(echo "$line" | awk '{print $4}'); port=${laddr##*:}
  case "$port" in (*[!0-9]*|"") continue;; esac
  pid=$(echo "$line" | grep -oE 'pid=[0-9]+' | head -1 | grep -oE '[0-9]+'); unit=""
  [ -n "$pid" ] && unit=$(grep -aoE '[A-Za-z0-9_.@-]+\.service' /proc/$pid/cgroup 2>/dev/null | head -1)
  printf '%s|%s\n' "$port" "${unit%.service}"
done | sort -t'|' -k1,1n -u
echo "### VHOSTS"
for f in /etc/nginx/sites-enabled/*; do
  n=$(basename "$f")
  # Resolve the ROOT "location / " upstream (trailing space avoids matching /mcp etc).
  # Handle both direct (proxy_pass http://127.0.0.1:PORT) and named upstreams
  # (proxy_pass http://name; upstream name { server 127.0.0.1:PORT; }).
  tgt=$(grep -A4 'location / ' "$f" 2>/dev/null | grep -oE 'proxy_pass http://[^;]+' | head -1 | sed 's#proxy_pass http://##')
  case "$tgt" in
    127.0.0.1:*) p=${tgt##*:} ;;
    "")          p=static ;;
    *) p=$(grep -E "upstream[[:space:]]+${tgt}[[:space:]]*\{" -A3 "$f" 2>/dev/null | grep -oE 'server[[:space:]]+127\.0\.0\.1:[0-9]+' | grep -oE '[0-9]+$' | head -1)
       [ -z "$p" ] && p=static ;;
  esac
  printf '%s|%s\n' "$n" "${p:-static}"
done
"####;

/// `processes` default fleet match (egrep alternation).
const DEFAULT_PROCESS_PATTERN: &str = "akurai|golfset|idp|mail|router|drive|fulltrui|platform|notes|passvault|crm|vpn|monitor|bunfork";

/// `akurai-ec2 ssh '<cmd>'` — run a command on the box, inheriting stdio.
pub fn ssh(ec2: &Ec2, args: &[String]) -> Result<()> {
    // ssh joins its post-host argv with spaces into the remote command; joining
    // here reproduces that exactly (the self-heal lives in Ec2's ssh wrapper).
    let command = args.join(" ");
    let code = ec2.ssh_streaming_status(&command)?;
    if code != 0 {
        bail!("ssh {}: command failed (exit {code})", ec2.ssh_host);
    }
    Ok(())
}

/// `akurai-ec2 ship <local> <remote>` — scp a file up (creates the parent dir).
pub fn ship(ec2: &Ec2, local: &Path, remote: &str) -> Result<()> {
    ec2.ship(local, remote)?;
    say(&format!("shipped {} → {remote}", local.display()));
    Ok(())
}

/// `akurai-ec2 status` — services, ports, nginx, disk.
pub fn status(ec2: &Ec2) -> Result<()> {
    ec2.ssh_stdin_streaming("bash -s", STATUS_SCRIPT.as_bytes())
}

/// `akurai-ec2 processes [pattern]` / `ps` — the running AkurAI fleet.
pub fn processes(ec2: &Ec2, pattern: Option<&str>) -> Result<()> {
    let pat = pattern.unwrap_or(DEFAULT_PROCESS_PATTERN);
    say(&format!(
        "AkurAI processes on {} (match: {pat})",
        ec2.ssh_host
    ));
    let command = format!("bash -s -- {}", shell_quote(pat));
    ec2.ssh_stdin_streaming(&command, PROCESS_SCRIPT.as_bytes())
}

/// `akurai-ec2 logs <service>` — journalctl -u <service> -n 80.
pub fn logs(ec2: &Ec2, service: &str) -> Result<()> {
    // Quoted, unlike the bash: a service name is a unit identifier, never shell
    // code. Behaviour for every valid name is identical to the original.
    ec2.ssh_streaming(&format!(
        "journalctl -u {} -n 80 --no-pager",
        shell_quote(service)
    ))
}

/// `akurai-ec2 ports [list|free [n]|check <port>|drift]`.
pub fn ports(ec2: &Ec2, args: &[String]) -> Result<()> {
    if !ec2.ports_registry.exists() {
        bail!(
            "registry not found: {} (see the _VMPORTS skill)",
            ec2.ports_registry.display()
        );
    }
    let sub = args.first().map(String::as_str).unwrap_or("list");
    let rest = &args[args.len().min(1)..];

    let live_text = ec2.ssh(RFETCH_SCRIPT)?;
    let live = parse_live(&live_text);
    let registry_text = std::fs::read_to_string(&ec2.ports_registry)
        .with_context(|| format!("reading {}", ec2.ports_registry.display()))?;
    let reg = parse_registry(&registry_text);

    match sub {
        "list" => list_ports(&reg, &live),
        "free" => free_ports_cmd(&reg, &live, rest),
        "check" => check_port(&reg, &live, rest),
        "drift" => drift(&reg, &live),
        other => {
            bail!("usage: akurai-ec2 ports [list|free [n]|check <port>|drift] (got {other:?})")
        }
    }
}

#[derive(Default)]
struct Registry {
    /// Numeric registry ports, in file order.
    ports: Vec<u16>,
    unit: HashMap<u16, String>,
    domain: HashMap<u16, String>,
    pool: HashMap<u16, String>,
    note: HashMap<u16, String>,
}

fn parse_registry(text: &str) -> Registry {
    let mut reg = Registry::default();
    for line in text.lines() {
        let mut fields = line.splitn(6, '|');
        let port_raw = fields.next().unwrap_or("").trim();
        let unit = fields.next().unwrap_or("").trim().to_string();
        let domain = fields.next().unwrap_or("").trim().to_string();
        let _repo = fields.next().unwrap_or("").trim();
        let pool = fields.next().unwrap_or("").trim().to_string();
        // Bash `IFS='|' read` folds any further separators into the last field.
        let note = fields.next().unwrap_or("").trim().to_string();
        if port_raw.is_empty() || port_raw.starts_with('#') {
            continue;
        }
        let Ok(p) = port_raw.parse::<u16>() else {
            continue;
        };
        reg.ports.push(p);
        reg.unit.insert(p, unit);
        reg.domain.insert(p, domain);
        reg.pool.insert(p, pool);
        reg.note.insert(p, note);
    }
    reg
}

#[derive(Default)]
struct LiveState {
    /// port -> unit (empty unit = bound by something unlabelled).
    unit: HashMap<u16, String>,
    /// vhost name -> port string ("static" when no upstream).
    vhost: HashMap<String, String>,
}

fn parse_live(text: &str) -> LiveState {
    let mut state = LiveState::default();
    let mut section = "";
    for line in text.lines() {
        match line {
            "### PORTS" => {
                section = "p";
                continue;
            }
            "### VHOSTS" => {
                section = "v";
                continue;
            }
            "" => continue,
            _ => {}
        }
        let Some((key, value)) = line.split_once('|') else {
            continue;
        };
        if section == "p" {
            if let Ok(p) = key.parse::<u16>() {
                state.unit.insert(p, value.to_string());
            }
        } else if section == "v" {
            state.vhost.insert(key.to_string(), value.to_string());
        }
    }
    state
}

/// Taken if claimed in the registry OR bound live.
fn is_taken(reg: &Registry, live: &LiveState, p: u16) -> bool {
    reg.unit.contains_key(&p) || live.unit.contains_key(&p)
}

fn list_ports(reg: &Registry, live: &LiveState) -> Result<()> {
    println!(
        "\x1b[1m{:<6} {:<20} {:<7} {:<30} LIVE\x1b[0m",
        "PORT", "UNIT", "POOL", "DOMAIN"
    );
    let mut sorted = reg.ports.clone();
    sorted.sort_unstable();
    for p in sorted {
        let want = reg.unit.get(&p).map(String::as_str).unwrap_or("");
        let got = live.unit.get(&p).map(String::as_str).unwrap_or("");
        let pool = reg.pool.get(&p).map(String::as_str).unwrap_or("");
        let domain = reg.domain.get(&p).map(String::as_str).unwrap_or("");
        // System services carry a descriptive label, not an exact unit name —
        // any live binding counts as bound for them.
        let mark = if want == got || (pool == "system" && !got.is_empty()) {
            "\x1b[0;32m●\x1b[0m bound".to_string()
        } else if !got.is_empty() {
            format!("\x1b[0;31m✗\x1b[0m {got}")
        } else {
            "\x1b[0;33m○\x1b[0m not bound".to_string()
        };
        println!("{p:<6} {want:<20} {pool:<7} {domain:<30} {mark}");
    }
    // Live app-pool ports with no registry row = unregistered.
    let mut live_keys: Vec<u16> = live.unit.keys().copied().collect();
    live_keys.sort_unstable();
    let unregistered: Vec<String> = live_keys
        .into_iter()
        .filter(|p| (APP_POOL_LO..=APP_POOL_HI).contains(p))
        .filter(|p| !reg.unit.contains_key(p))
        .map(|p| format!("  {p} -> {}", live.unit[&p]))
        .collect();
    if !unregistered.is_empty() {
        println!("\n\x1b[1;31munregistered live ports (app pool):\x1b[0m");
        for line in unregistered {
            println!("{line}");
        }
    }
    Ok(())
}

fn free_ports_cmd(reg: &Registry, live: &LiveState, rest: &[String]) -> Result<()> {
    let n = match rest.first() {
        None => 1,
        Some(raw) => raw
            .parse::<u32>()
            .context("ports free: count must be an integer")?,
    };
    let found: Vec<u16> = free_ports(reg, live).into_iter().take(n as usize).collect();
    for port in &found {
        println!("{port}");
    }
    if found.is_empty() {
        bail!("no free ports in {APP_POOL_LO}-{APP_POOL_HI}");
    }
    Ok(())
}

fn check_port(reg: &Registry, live: &LiveState, rest: &[String]) -> Result<()> {
    let Some(raw) = rest.first() else {
        bail!("usage: akurai-ec2 ports check <port>");
    };
    say(&format!("port {raw}"));
    // Non-numeric ports never match anything (bash associative-array lookup).
    let p = raw.parse::<u16>().ok();
    if let Some(p) = p {
        let unit = reg.unit.get(&p).map(String::as_str).unwrap_or("");
        if !unit.is_empty() {
            let pool = reg.pool.get(&p).map(String::as_str).unwrap_or("");
            let domain = reg.domain.get(&p).map(String::as_str).unwrap_or("");
            let note = reg.note.get(&p).map(String::as_str).unwrap_or("");
            println!("  registry: {unit} ({pool}) {domain}  — {note}");
        } else {
            println!("  registry: (unclaimed)");
        }
        if let Some(live_unit) = live.unit.get(&p) {
            println!("  live:     bound by {live_unit}");
        } else {
            println!("  live:     not bound");
        }
        if is_taken(reg, live, p) {
            println!("\x1b[1;31m  => TAKEN\x1b[0m");
        } else {
            println!("\x1b[1;32m  => FREE\x1b[0m");
        }
    } else {
        println!("  registry: (unclaimed)");
        println!("  live:     not bound");
        println!("\x1b[1;32m  => FREE\x1b[0m");
    }
    Ok(())
}

fn drift(reg: &Registry, live: &LiveState) -> Result<()> {
    let mut issues = 0usize;
    let mut flagged_domains: HashSet<&str> = HashSet::new();

    // (a) Registry port expected unit U, but live differs / missing.
    let mut sorted = reg.ports.clone();
    sorted.sort_unstable();
    for p in sorted {
        if reg.pool.get(&p).map(String::as_str) == Some("system") {
            continue;
        }
        let want = reg.unit.get(&p).map(String::as_str).unwrap_or("");
        if want.is_empty() {
            continue;
        }
        let got = live.unit.get(&p).map(String::as_str).unwrap_or("");
        let note = reg.note.get(&p).map(String::as_str).unwrap_or("");
        if got.is_empty() {
            println!("\x1b[0;33m○ {want:<18} NOT BOUND on its port {p}\x1b[0m  ({note})");
            issues += 1;
        } else if got != want {
            println!("\x1b[0;31m✗ port {p}: expected {want}, live is {got}\x1b[0m");
            issues += 1;
        }
    }

    // (b) nginx vhost points at a port other than what the registry intends.
    // A port may carry comma-separated domain aliases; a "host/path" entry is
    // path-routed (e.g. notes /mcp) and is not its own vhost — skip it.
    let mut domains: Vec<u16> = reg.domain.keys().copied().collect();
    domains.sort_unstable();
    for p in domains {
        let raw = reg.domain.get(&p).map(String::as_str).unwrap_or("");
        if raw == "-" {
            continue;
        }
        let unit = reg.unit.get(&p).map(String::as_str).unwrap_or("");
        for d in raw.split(',') {
            if d.contains('/') {
                continue;
            }
            let Some(lv) = live.vhost.get(d) else {
                continue;
            };
            let flagged = flagged_domains.contains(d);
            if lv == "static" {
                println!(
                    "\x1b[0;33m○ vhost {d} is static — registry routes it to {p} ({unit})\x1b[0m"
                );
                flagged_domains.insert(d);
                issues += 1;
            } else if lv != &p.to_string() {
                println!("\x1b[0;31m✗ vhost {d} -> {lv}, registry intends {p} ({unit})\x1b[0m");
                flagged_domains.insert(d);
                issues += 1;
            } else if flagged {
                // Already reported; keep the set for phase (d).
            }
        }
    }

    // (c) Live app-pool listeners with no registry row.
    let mut live_keys: Vec<u16> = live.unit.keys().copied().collect();
    live_keys.sort_unstable();
    for &p in &live_keys {
        if !(APP_POOL_LO..=APP_POOL_HI).contains(&p) {
            continue;
        }
        if !reg.unit.contains_key(&p) {
            println!(
                "\x1b[0;31m✗ port {p} bound by {} — not in registry\x1b[0m",
                live.unit[&p]
            );
            issues += 1;
        }
    }

    // (d) Live vhost serving an app-pool port that belongs to a different app.
    let mut vhosts: Vec<(&String, &String)> = live.vhost.iter().collect();
    vhosts.sort_by(|a, b| a.0.cmp(b.0));
    for (d, lv) in vhosts {
        if flagged_domains.contains(d.as_str()) {
            continue; // already reported in (b)
        }
        let Ok(lv_port) = lv.parse::<u16>() else {
            continue;
        };
        if !(APP_POOL_LO..=APP_POOL_HI).contains(&lv_port) {
            continue;
        }
        let raw = reg.domain.get(&lv_port).map(String::as_str).unwrap_or("");
        if raw.is_empty() || raw == "-" {
            continue;
        }
        let owner = raw.split(',').any(|alias| alias == d);
        if !owner {
            let unit = reg.unit.get(&lv_port).map(String::as_str).unwrap_or("");
            println!("\x1b[0;31m✗ vhost {d} -> {lv}, but {lv} is {unit} ({raw})\x1b[0m");
            issues += 1;
        }
    }

    if issues == 0 {
        say("no drift — registry matches the box");
        Ok(())
    } else {
        bail!("{issues} drift issue(s) — see above");
    }
}

/// `akurai-ec2 allow-ssh [ip]` — authorize an IP on the SG, then verify SSH.
pub fn allow_ssh(ec2: &Ec2, ip: Option<&str>) -> Result<()> {
    ec2.ensure_my_ip(ip).context("could not authorize SSH")?;
    say(&format!("Verifying SSH to {}…", ec2.ssh_host));
    let ok = Command::new("ssh")
        .arg("-F")
        .arg(&ec2.ssh_config)
        .arg("-o")
        .arg("ConnectTimeout=15")
        .arg("-o")
        .arg("BatchMode=yes")
        .arg(&ec2.ssh_host)
        .arg("echo ok")
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|status| status.success())
        .unwrap_or(false);
    if ok {
        say(&format!("SSH OK ✓ ({})", ec2.ssh_host));
        Ok(())
    } else {
        bail!(
            "authorized, but SSH to {} still fails (route/key/DNS?) — investigate",
            ec2.ssh_host
        );
    }
}

/// `akurai-ec2 health [--all]` — fan-out health check across AKURAI_APPS.toml.
pub fn health(_ec2: &Ec2, _all: bool) -> Result<()> {
    let home = std::env::var("HOME").context("HOME is not set")?;
    let apps_toml = Path::new(&home)
        .join("Projects")
        .join("AkurAI-Framework")
        .join("AKURAI_APPS.toml");
    if !apps_toml.exists() {
        bail!("AKURAI_APPS.toml not found at {}", apps_toml.display());
    }
    println!("=== Fleet Health Check ===");
    let text = std::fs::read_to_string(&apps_toml)
        .with_context(|| format!("reading {}", apps_toml.display()))?;
    for app in parse_apps_toml(&text) {
        probe_app(&app);
    }
    Ok(())
}

/// One app's fleet-audit finding: exact hosted commit (when resolvable),
/// pipeline conformance violations, and evidence still missing. Pipeline
/// conformance (static SDP-002..006/010 pattern checks via
/// `config::parse`) is reported distinctly from persisted deployment
/// success (SDP-006/007/009/010 runtime evidence), which this command
/// cannot itself observe — declaring a rule satisfied in `.akurai.yml`
/// text is never accepted as proof a production deploy actually ran.
#[derive(serde::Serialize)]
struct FleetAuditEntry {
    app: String,
    hosted_commit: Option<String>,
    violations: Vec<String>,
    missing_evidence: Vec<&'static str>,
}

/// `akurai-ec2 fleet-audit [--json]` — walk every active app in
/// AKURAI_APPS.toml, resolve its hosted commit, and validate its
/// `.akurai.yml` production job(s) against the suite deployment policy
/// (docs/suite-deployment-policy.md). Retired apps are skipped.
pub fn fleet_audit(json: bool) -> Result<()> {
    let home = std::env::var("HOME").context("HOME is not set")?;
    let apps_toml = Path::new(&home)
        .join("Projects")
        .join("AkurAI-Framework")
        .join("AKURAI_APPS.toml");
    if !apps_toml.exists() {
        bail!("AKURAI_APPS.toml not found at {}", apps_toml.display());
    }
    let text = std::fs::read_to_string(&apps_toml)
        .with_context(|| format!("reading {}", apps_toml.display()))?;
    let mut entries = Vec::new();
    for app in parse_apps_toml(&text) {
        if app.retired || app.name.is_empty() {
            continue;
        }
        entries.push(audit_app(&app));
    }
    if json {
        println!("{}", serde_json::to_string_pretty(&entries)?);
    } else {
        println!("=== Fleet Deployment Policy Audit ===");
        for entry in &entries {
            let commit = entry.hosted_commit.as_deref().unwrap_or("unresolved");
            if entry.violations.is_empty() {
                println!("  [\x1b[32mOK\x1b[0m] {:<20} {commit}", entry.app);
            } else {
                println!("  [\x1b[31mFAIL\x1b[0m] {:<20} {commit}", entry.app);
                for violation in &entry.violations {
                    println!("           - {violation}");
                }
            }
            if !entry.missing_evidence.is_empty() {
                println!(
                    "           missing runtime evidence: {}",
                    entry.missing_evidence.join(", ")
                );
            }
        }
    }
    let failed = entries.iter().any(|entry| !entry.violations.is_empty());
    if failed {
        bail!("fleet audit found non-conforming pipelines");
    }
    Ok(())
}

fn audit_app(app: &AppCfg) -> FleetAuditEntry {
    let mut violations = Vec::new();
    let checkout = Path::new(&app.checkout);
    let hosted_commit = if app.checkout.is_empty() {
        None
    } else {
        Command::new("git")
            .arg("-C")
            .arg(checkout)
            .args(["rev-parse", "HEAD"])
            .output()
            .ok()
            .filter(|out| out.status.success())
            .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_owned())
    };
    let pipeline_path = checkout.join(".akurai.yml");
    match std::fs::read_to_string(&pipeline_path) {
        Ok(source) => {
            if let Err(error) = crate::config::parse(&source) {
                violations.push(format!("{error:#}"));
            }
        }
        Err(error) => violations.push(format!(
            "no .akurai.yml at {}: {error}",
            pipeline_path.display()
        )),
    }
    // The catalog `deploy` field is manual-fallback tooling, not something
    // Build enforcement reads, but a raw direct-deploy command invites
    // bypassing Build by hand — flag it as a policy violation too. Match the
    // real tool-call sequence rather than the bare `akurai-build:` label: the
    // label is prose that proves nothing, and a field naming the actual
    // repo_sync -> run_queue -> run_promote calls with their argument keys is
    // something a caller can execute.
    if !app.deploy.is_empty() && !is_executable_deploy_contract(&app.deploy) {
        violations.push(format!(
            "AKURAI_APPS.toml deploy field bypasses AkurAI Build: {}",
            app.deploy
        ));
    }
    // Runtime evidence (SDP-006/007/009/010: host-agent handoff actually
    // executed, migration backup verified, run reached `succeeded`, health
    // check passed) lives in AkurAI Build's persisted run/job-log rows,
    // which this filesystem-only audit cannot read; a fleet audit is never
    // a substitute for checking `akurai_run_show`/`akurai_runs` before
    // calling an app deployed.
    let mut missing_evidence = vec!["SDP-009 persisted succeeded run"];
    if !app.db_snapshot.is_empty() {
        missing_evidence.push("SDP-007 migration backup+recovery");
    }
    missing_evidence.push("SDP-006 host-agent handoff execution");
    missing_evidence.push("SDP-010 post-deploy health check result");
    FleetAuditEntry {
        app: app.name.clone(),
        hosted_commit,
        violations,
        missing_evidence,
    }
}

#[derive(Default, Debug)]
struct AppCfg {
    name: String,
    retired: bool,
    health_type: String,
    dns_server: String,
    dns_name: String,
    dns_expect: String,
    domain: String,
    health_path: String,
    repo: String,
    checkout: String,
    deploy: String,
    db_snapshot: String,
}

/// Minimal TOML-subset reader for AKURAI_APPS.toml: `[[app]]` sections with
/// `key = value` lines (quoted strings or bare words). Enough for the fields
/// the health check reads; no TOML crate is in the dependency budget.
fn parse_apps_toml(text: &str) -> Vec<AppCfg> {
    let mut apps = Vec::new();
    let mut current: Option<AppCfg> = None;
    for line in text.lines() {
        let line = line.trim();
        if line == "[[app]]" {
            if let Some(cfg) = current.take() {
                apps.push(cfg);
            }
            current = Some(AppCfg::default());
            continue;
        }
        if line.is_empty() || line.starts_with('#') || line.starts_with('[') {
            continue;
        }
        let Some((key, value)) = line.split_once('=') else {
            continue;
        };
        let Some(cfg) = current.as_mut() else {
            continue;
        };
        match key.trim() {
            "name" => cfg.name = toml_value(value),
            "retired" => cfg.retired = toml_value(value) == "true",
            "health_type" => cfg.health_type = toml_value(value),
            "dns_server" => cfg.dns_server = toml_value(value),
            "dns_name" => cfg.dns_name = toml_value(value),
            "dns_expect" => cfg.dns_expect = toml_value(value),
            "domain" => cfg.domain = toml_value(value),
            "health_path" => cfg.health_path = toml_value(value),
            "repo" => cfg.repo = toml_value(value),
            "checkout" => cfg.checkout = toml_value(value),
            "deploy" => cfg.deploy = toml_value(value),
            "db_snapshot" => cfg.db_snapshot = toml_value(value),
            _ => {}
        }
    }
    if let Some(cfg) = current.take() {
        apps.push(cfg);
    }
    apps
}

/// A catalog `deploy` field is only worth trusting if it names the real MCP
/// tool-call sequence with its argument keys -- `akurai_repo_sync` with
/// `name=`/`source=`, then `akurai_run_queue` with `repository=`, then
/// `akurai_run_promote` with `id=`/`environment=`. Prose, a bare
/// `akurai-build:` label, or a raw `ec2 release` command does not qualify,
/// however plausible it reads.
fn is_executable_deploy_contract(deploy: &str) -> bool {
    const REQUIRED_TOKENS: &[&str] = &[
        "akurai_repo_sync",
        "name=",
        "source=",
        "akurai_run_queue",
        "repository=",
        "akurai_run_promote",
        "id=",
        "environment=",
    ];
    REQUIRED_TOKENS.iter().all(|token| deploy.contains(token))
}

fn toml_value(raw: &str) -> String {
    let raw = raw.trim();
    if let Some(stripped) = raw.strip_prefix('"') {
        // Quoted string: up to the closing quote (trailing comment stays out).
        // `find('"')` is not good enough -- TOML basic strings escape an inner
        // quote as \" , and stopping at the first raw quote truncates the value
        // there. A `deploy` field carrying tool-call arguments
        // (name=\"AkurAI-Platform\", ...) came back as `...repo_sync(name=\`,
        // silently dropping everything a caller wanted to read.
        let mut value = String::with_capacity(stripped.len());
        let mut characters = stripped.chars();
        while let Some(character) = characters.next() {
            match character {
                '"' => return value,
                '\\' => match characters.next() {
                    // Only the escapes this catalog actually uses; anything
                    // else keeps both characters so nothing is silently eaten.
                    Some('"') => value.push('"'),
                    Some('\\') => value.push('\\'),
                    Some(other) => {
                        value.push('\\');
                        value.push(other);
                    }
                    None => value.push('\\'),
                },
                other => value.push(other),
            }
        }
        // Unterminated string: return what we have, as before.
        return value;
    }
    // Bare word / boolean / integer.
    raw.split([' ', '#']).next().unwrap_or("").to_string()
}

fn probe_app(app: &AppCfg) {
    let name = if app.name.is_empty() { "?" } else { &app.name };
    // DNS-only services expose no HTTP endpoint; probe a TXT health record.
    if app.health_type == "dns" {
        let server = &app.dns_server;
        let qname = &app.dns_name;
        let expect = if app.dns_expect.is_empty() {
            "ok"
        } else {
            &app.dns_expect
        };
        match Command::new("dig")
            .arg(format!("@{server}"))
            .arg(qname)
            .arg("TXT")
            .arg("+short")
            .arg("+time=3")
            .arg("+tries=1")
            .output()
        {
            Ok(out) if out.status.success() => {
                let stdout = String::from_utf8_lossy(&out.stdout);
                if stdout.contains(expect) {
                    println!("  [\x1b[32mOK\x1b[0m] {name:<30} dns:{qname:<25} status={expect}");
                } else {
                    println!(
                        "  [\x1b[31mFAIL\x1b[0m] {name:<30} dig {qname} @{server} (want TXT {expect})"
                    );
                }
            }
            Ok(_) => println!(
                "  [\x1b[31mFAIL\x1b[0m] {name:<30} dig {qname} @{server} (want TXT {expect})"
            ),
            Err(error) => println!("  [\x1b[31mERR\x1b[0m]  {name:<30} {error}"),
        }
        return;
    }
    let domain = &app.domain;
    if domain.is_empty() {
        return;
    }
    let health_url = if app.health_path == "/health" {
        format!("https://{domain}/health")
    } else {
        format!("https://{domain}/api/health")
    };
    match Command::new("curl")
        .args(["-fsS", "--max-time", "5", &health_url])
        .output()
    {
        Ok(out) if out.status.success() => {
            let stdout = String::from_utf8_lossy(&out.stdout);
            let (label, status) = match serde_json::from_str::<Value>(&stdout) {
                Ok(data) => {
                    let label = data
                        .get("app")
                        .or_else(|| data.get("service"))
                        .or_else(|| data.get("framework"))
                        .map(json_to_string)
                        .unwrap_or_else(|| name.to_string());
                    let status = data
                        .get("status")
                        .map(json_to_string)
                        .unwrap_or_else(|| "ok".to_string());
                    (label, status)
                }
                Err(_) => (name.to_string(), "ok".to_string()),
            };
            println!("  [\x1b[32mOK\x1b[0m] {name:<30} {label:<25} status={status}");
        }
        Ok(_) => println!("  [\x1b[31mFAIL\x1b[0m] {name:<30} {health_url}"),
        Err(error) => println!("  [\x1b[31mERR\x1b[0m]  {name:<30} {error}"),
    }
}

fn json_to_string(value: &Value) -> String {
    match value {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

/// Free ports helper factored out for testing (bash `ports free [n]` core).
fn free_ports(reg: &Registry, live: &LiveState) -> Vec<u16> {
    (APP_POOL_LO..=APP_POOL_HI)
        .filter(|p| !is_taken(reg, live, *p))
        .collect()
}

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

    #[test]
    fn toml_value_keeps_escaped_quotes_instead_of_truncating() {
        // The regression: `find('"')` stopped at the first quote of an inner
        // \" escape, so a deploy field carrying tool-call arguments was cut
        // down to `...repo_sync(name=\` and every token after it vanished.
        let raw = r#""mcp__akurai_build__akurai_repo_sync(name=\"AkurAI-Platform\", source=\"/p\") -> done""#;
        let value = toml_value(raw);
        assert!(
            value.contains("source=") && value.ends_with("done"),
            "escaped quotes must not truncate the value: {value:?}"
        );
        assert_eq!(value.matches('"').count(), 4, "escapes unescape to quotes");
    }

    #[test]
    fn toml_value_still_stops_at_the_real_closing_quote() {
        assert_eq!(
            toml_value(r#""plain value"   # trailing comment"#),
            "plain value"
        );
        assert_eq!(toml_value("bare_word # comment"), "bare_word");
        assert_eq!(toml_value(r#""unterminated"#), "unterminated");
        // A lone backslash must not swallow the terminator or panic.
        assert_eq!(
            toml_value(r#""ends with backslash \"#),
            "ends with backslash \\"
        );
    }

    #[test]
    fn executable_deploy_contract_accepts_real_calls_and_rejects_prose() {
        let real = "mcp__akurai_build__akurai_repo_sync(name=\"A\", source=\"/p\") -> \
                    mcp__akurai_build__akurai_run_queue(repository=\"A\", commit=<sha>) -> \
                    mcp__akurai_build__akurai_run_promote(id=<run>, environment=\"production\")";
        assert!(is_executable_deploy_contract(real));
        assert!(!is_executable_deploy_contract(
            "akurai-build: akurai_repo_sync a -> akurai_run_queue(commit=<sha>) -> akurai_run_promote(id, environment=production)"
        ));
        assert!(!is_executable_deploy_contract(
            "akurai-ec2 release akurai-platform"
        ));
        assert!(!is_executable_deploy_contract(""));
    }

    #[test]
    fn live_catalog_deploy_fields_survive_parsing_and_are_executable() {
        // Reads the real AKURAI_APPS.toml rather than a fixture, because the
        // defect only appeared against the real file: the fixture-shaped
        // values had no escaped quotes. Skips when the sibling checkout is
        // absent (CI containers), and says so rather than passing silently.
        let Ok(home) = std::env::var("HOME") else {
            eprintln!("skipped: HOME unset");
            return;
        };
        let path = Path::new(&home).join("Projects/AkurAI-Framework/AKURAI_APPS.toml");
        let Ok(text) = std::fs::read_to_string(&path) else {
            eprintln!("skipped: {} not present", path.display());
            return;
        };
        let apps: Vec<AppCfg> = parse_apps_toml(&text)
            .into_iter()
            .filter(|app| !app.retired && !app.name.is_empty() && !app.deploy.is_empty())
            .collect();
        assert!(!apps.is_empty(), "catalog parsed to zero deployable apps");
        for app in &apps {
            assert!(
                !app.deploy.ends_with('\\'),
                "app {} deploy value was truncated at an escape: {}",
                app.name,
                app.deploy
            );
        }
    }

    #[test]
    fn registry_parsing_skips_comments_and_garbage() {
        let text = "# port|unit|domain|repo|pool|note\n8094|mail|mail.olibuijr.com|akurai-mail|app|primary mail\n\n|bad|row\n8199|system-thing|-|repo|system|note with|extra pipes\n";
        let reg = parse_registry(text);
        assert_eq!(reg.ports, vec![8094, 8199]);
        assert_eq!(reg.unit[&8094], "mail");
        assert_eq!(reg.pool[&8199], "system");
        assert_eq!(reg.note[&8199], "note with|extra pipes");
    }

    #[test]
    fn live_parsing_sections() {
        let live = parse_live(
            "### PORTS\n8080|web\n8094|mail\n### VHOSTS\nmail.olibuijr.com|8094\nstatic.site|static\n",
        );
        assert_eq!(live.unit[&8080], "web");
        assert_eq!(live.vhost["mail.olibuijr.com"], "8094");
        assert_eq!(live.vhost["static.site"], "static");
    }

    #[test]
    fn free_ports_skips_registered_and_live() {
        let reg = parse_registry("8094|mail|mail.olibuijr.com|repo|app|note\n");
        let live = parse_live("### PORTS\n8095|other\n### VHOSTS\n");
        assert_eq!(
            free_ports(&reg, &live)
                .into_iter()
                .take(3)
                .collect::<Vec<_>>(),
            vec![8096, 8097, 8098]
        );
        assert!(is_taken(&reg, &live, 8094));
        assert!(is_taken(&reg, &live, 8095));
        assert!(!is_taken(&reg, &live, 8096));
    }
    #[test]
    fn apps_toml_minimal_parse() {
        let text = "[[app]]\nname = \"mail\"\ndomain = \"mail.olibuijr.com\"\nhealth_path = \"/health\"\n\n[[app]]\nname = \"dns-only\"\nretired = false\nhealth_type = \"dns\"\ndns_server = \"ns1\"\ndns_name = \"_health\"\ndns_expect = \"ok\"\n";
        let apps = parse_apps_toml(text);
        assert_eq!(apps.len(), 2);
        assert_eq!(apps[0].name, "mail");
        assert_eq!(apps[1].dns_name, "_health");
    }

    #[test]
    fn free_ports_list_matches_pool_bounds() {
        assert_eq!(APP_POOL_LO, 8094);
        assert_eq!(APP_POOL_HI, 8199);
    }
}