//! ClawMates microVM guest agent — pid 1 inside a Firecracker microVM.
//!
//! Runs as `init=/usr/local/bin/fcagent`'s exec target and answers the host over
//! **vsock** (port 9001), never the serial console: feeding a guest over stdin
//! races its startup and arrives half-consumed. The console stays a log.
//!
//! # Why this is a static Rust binary and not the python script it replaces
//!
//! The python version worked only because Firecracker's CI Ubuntu image happens
//! to ship python3. **None of our own images do** — `agent-base` has neither
//! python nor git, `agent-terminal` has git but no python — so the agent could
//! never have run in a real mission rootfs. An agent that dictates what must be
//! installed in the image has the dependency backwards. This is a
//! `x86_64-unknown-linux-musl` static binary: it needs nothing from the rootfs
//! it is dropped into.
//!
//! # Wire protocol (unchanged from the python agent, deliberately)
//!
//! One request per connection: a 4-byte big-endian length followed by JSON, and
//! the reply framed the same way. The length prefix is the point — a reply
//! larger than a socket buffer arrives in pieces, and reading "whatever was
//! available" would parse a truncated object as a complete one.
//!
//! Ops: `ping`, `exec`, `put`, `get`. `crates/bins/clawmates-node/src/microvm.rs`
//! and `crates/cm-api/src/microvm_client.rs` speak this and needed no change.

use std::io::{Read, Write};
use std::net::TcpListener;
use std::os::unix::process::CommandExt;
use std::path::Path;
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use base64::Engine;
use serde_json::{json, Value};

const PORT: u32 = 9001;
/// Guest-side egress proxy. The VM has **no network interface at all** — see
/// `microvm.rs`, whose machine config declares no `network-interfaces` — so an
/// agent CLI cannot reach the model API on its own. It reaches it by honouring
/// `HTTPS_PROXY`, which is measured, not assumed: with the proxy pointed at a
/// closed port, `claude -p` fails with `ConnectionRefused` instead of answering.
///
/// This listener is a dumb byte pump. It parses nothing and enforces nothing:
/// the `CONNECT` request travels verbatim to the host, which speaks HTTP CONNECT
/// and owns the allow-list. Keeping policy on the host means nothing running in
/// the guest — including a compromised agent — can talk it into a different
/// answer.
const PROXY_PORT: u16 = 3128;
/// Host-side vsock port the tunnel lands on. Firecracker's convention for a
/// guest-initiated connection is that the HOST listens on `<uds_path>_<port>`.
const EGRESS_PORT: u32 = 9002;
/// Guest-side port for a LOCALLY HOSTED model, and the vsock port it lands on.
///
/// Separate from the egress proxy on purpose, and simpler than it. The egress
/// path exists to let an agent reach the public internet under an allow-list;
/// this one reaches exactly one thing — the Ollama the node itself is running,
/// on its own loopback — and can reach nothing else, because the host end is a
/// pipe to a fixed address rather than a proxy that takes a destination.
///
/// It therefore needs no `CONNECT`, no TLS and no allow-list. The bytes travel
/// guest loopback → vsock → host loopback and never touch a network, so there is
/// nothing on a wire for TLS to protect. `NO_PROXY` already contains
/// `127.0.0.1`, so an agent pointed at `http://127.0.0.1:11434` bypasses the
/// egress proxy entirely rather than trying to CONNECT through it.
///
/// The guest always listens. Whether anything answers is the HOST's decision:
/// the node only binds the vsock end for a backend that is meant to have a
/// local model, so on every other backend this port simply refuses.
const MODEL_PORT: u16 = 11434;
const MODEL_VSOCK_PORT: u32 = 9003;
/// `VMADDR_CID_HOST` — the hypervisor side of the vsock.
const HOST_CID: u32 = 2;

/// Whether the egress proxy is actually listening. Reported by `ping` so the
/// host can refuse to hand a mission to a VM with no way out, rather than
/// discovering it as an agent that hangs.
static PROXY_UP: AtomicBool = AtomicBool::new(false);
/// Cap on a single request. A hostile or broken host must not be able to make
/// pid 1 allocate without bound and get the VM OOM-killed.
const MAX_REQUEST: u32 = 512 * 1024 * 1024;

const B64: base64::engine::general_purpose::GeneralPurpose =
    base64::engine::general_purpose::STANDARD;

fn main() {
    // The mounts the init script would otherwise do. Done here so the agent
    // works whether it is exec'd from a shell init or used as `init=` directly:
    // /proc missing makes every process-inspecting tool in the guest lie.
    for (fstype, target) in [
        ("proc", "/proc"),
        ("sysfs", "/sys"),
        ("devtmpfs", "/dev"),
        ("tmpfs", "/tmp"),
    ] {
        if !Path::new(target).join(".").exists() {
            let _ = std::fs::create_dir_all(target);
        }
        let _ = Command::new("mount")
            .args(["-t", fstype, fstype, target])
            .status();
    }

    start_egress_proxy();

    let listener = match vsock::VsockListener::bind_with_cid_port(libc_vmaddr_cid_any(), PORT) {
        Ok(l) => l,
        Err(e) => {
            // Printed to the console, which is where the host's boot check
            // looks. Exiting pid 1 panics the kernel, which is the honest
            // outcome: a VM whose agent cannot listen is unusable, and it must
            // not sit there looking booted.
            eprintln!("FC-AGENT-FATAL could not bind vsock port {PORT}: {e}");
            std::process::exit(1);
        }
    };

    // The host greps the console for this before it tries to connect.
    println!("FC-AGENT-LISTENING port={PORT}");
    let _ = std::io::stdout().flush();

    for conn in listener.incoming() {
        match conn {
            Ok(mut s) => {
                // One THREAD per connection, not one at a time.
                //
                // This loop used to call `serve_one` inline, which meant the
                // agent accepted nothing while an op was running. A mission turn
                // is an `exec` that can last an hour, so for that hour the guest
                // was unreachable: the host could not tail its output, probe it,
                // or ask it anything. Every existing probe runs AFTER the turn
                // for exactly this reason.
                //
                // A thread rather than async: this is a static musl binary with
                // no runtime, and the concurrency here is a handful of
                // connections, not thousands.
                //
                // The panic discipline of the old inline call still applies, and
                // matters MORE now — this process is pid 1, and a panic that
                // unwound out of a worker used to take the accept loop with it.
                // `catch_unwind` keeps a bad request from killing the VM.
                std::thread::Builder::new()
                    .name("fcagent-conn".into())
                    .spawn(move || {
                        let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                            serve_one(&mut s)
                        }));
                        match r {
                            Ok(Err(e)) => eprintln!("FC-AGENT-ERROR {e}"),
                            Err(_) => eprintln!("FC-AGENT-ERROR handler panicked"),
                            Ok(Ok(())) => {}
                        }
                    })
                    .map(|_| ())
                    .unwrap_or_else(|e| {
                        // Out of threads: answer nothing on this connection, but
                        // keep accepting. Dropping the listener would brick the VM.
                        eprintln!("FC-AGENT-ERROR spawn: {e}");
                    });
            }
            Err(e) => eprintln!("FC-AGENT-ERROR accept: {e}"),
        }
    }
}

/// `VMADDR_CID_ANY` — bind for any host CID.
fn libc_vmaddr_cid_any() -> u32 {
    u32::MAX
}

/// Bring up loopback and start the egress tunnel.
///
/// Loopback is not optional and not free: the guest's `lo` exists but starts
/// **down**, and while it is down a listener on 127.0.0.1 *binds successfully*
/// and then refuses every connection with `ENETUNREACH`. A bind-only check would
/// have reported a working proxy. So `lo` goes up first, via `ip` — which is why
/// `iproute2` is in the agent images.
///
/// Failure here is recorded, not fatal: exec still works, so a VM is still
/// useful for work that needs no network. It is reported through `ping` so the
/// host can decide, instead of a mission discovering it as an agent that hangs.
fn start_egress_proxy() {
    // Absolute paths, not `Command::new("ip")`. This process is pid 1, so its
    // PATH is whatever the kernel handed it — and when PATH is unset, `execvp`
    // falls back to a default that does NOT include `/usr/sbin`, which is exactly
    // where Debian puts `ip`. Searching by name would fail on an image that has
    // it, and the symptom would be a VM with no egress and no explanation.
    const IP_CANDIDATES: &[&str] = &["/usr/sbin/ip", "/sbin/ip", "/usr/bin/ip", "/bin/ip"];
    let Some(ip) = IP_CANDIDATES.iter().find(|p| Path::new(p).exists()) else {
        eprintln!(
            "FC-AGENT-NO-PROXY no `ip` binary in {IP_CANDIDATES:?} — no egress; \
             add iproute2 to this image"
        );
        return;
    };
    match Command::new(ip).args(["link", "set", "lo", "up"]).status() {
        Ok(s) if s.success() => {}
        other => {
            eprintln!("FC-AGENT-NO-PROXY `{ip} link set lo up` failed ({other:?}) — no egress");
            return;
        }
    }

    let listener = match TcpListener::bind(("127.0.0.1", PROXY_PORT)) {
        Ok(l) => l,
        Err(e) => {
            eprintln!("FC-AGENT-NO-PROXY could not listen on 127.0.0.1:{PROXY_PORT}: {e}");
            return;
        }
    };
    PROXY_UP.store(true, Ordering::Relaxed);
    println!("FC-AGENT-PROXY listening on 127.0.0.1:{PROXY_PORT} -> vsock {EGRESS_PORT}");
    let _ = std::io::stdout().flush();
    pump(listener, EGRESS_PORT, "PROXY");

    // The local-model port. Failure to bind is reported and non-fatal, exactly
    // like the egress proxy: a VM whose backend does not use a local model is
    // still perfectly useful, and a fatal error here would take out every
    // backend to serve one.
    match TcpListener::bind(("127.0.0.1", MODEL_PORT)) {
        Ok(l) => {
            println!("FC-AGENT-MODEL listening on 127.0.0.1:{MODEL_PORT} -> vsock {MODEL_VSOCK_PORT}");
            let _ = std::io::stdout().flush();
            pump(l, MODEL_VSOCK_PORT, "MODEL");
        }
        Err(e) => eprintln!("FC-AGENT-NO-MODEL could not listen on 127.0.0.1:{MODEL_PORT}: {e}"),
    }
}

/// Accept forever, splicing each connection onto its own vsock stream.
fn pump(listener: TcpListener, vsock_port: u32, tag: &'static str) {
    std::thread::spawn(move || {
        for c in listener.incoming() {
            match c {
                // One thread per connection. An agent CLI opens several at once,
                // and serving them in sequence would look like a hang.
                Ok(tcp) => {
                    std::thread::spawn(move || {
                        if let Err(e) = tunnel(tcp, vsock_port) {
                            eprintln!("FC-AGENT-{tag}-ERROR {e}");
                        }
                    });
                }
                Err(e) => eprintln!("FC-AGENT-{tag}-ERROR accept: {e}"),
            }
        }
    });
}

/// Splice one TCP connection onto a fresh vsock connection to the host.
///
/// No parsing: whatever the client sent — `CONNECT host:443`, or an absolute-form
/// request — is the host's business. The host answers with real HTTP, so a
/// refusal reaches the client as a status code rather than a dropped socket.
fn tunnel(tcp: std::net::TcpStream, vsock_port: u32) -> Result<(), String> {
    let vs = vsock::VsockStream::connect_with_cid_port(HOST_CID, vsock_port)
        .map_err(|e| format!("vsock connect to host:{vsock_port}: {e}"))?;

    let (mut tcp_r, mut tcp_w) = (
        tcp.try_clone().map_err(|e| format!("clone tcp: {e}"))?,
        tcp,
    );
    let (mut vs_r, mut vs_w) = (
        vs.try_clone().map_err(|e| format!("clone vsock: {e}"))?,
        vs,
    );

    // Each direction gets its own thread, and each shuts its peer's write side
    // down when it ends. Without the shutdown the other half blocks forever on a
    // half-closed connection and the CLI waits out its own timeout.
    let up = std::thread::spawn(move || {
        let _ = std::io::copy(&mut tcp_r, &mut vs_w);
        let _ = vs_w.shutdown(std::net::Shutdown::Write);
    });
    let _ = std::io::copy(&mut vs_r, &mut tcp_w);
    let _ = tcp_w.shutdown(std::net::Shutdown::Write);
    let _ = up.join();
    Ok(())
}

fn serve_one(s: &mut vsock::VsockStream) -> Result<(), String> {
    let mut len = [0u8; 4];
    s.read_exact(&mut len)
        .map_err(|e| format!("read length: {e}"))?;
    let len = u32::from_be_bytes(len);
    if len > MAX_REQUEST {
        // Answer rather than hang up: a caller that sent something absurd needs
        // to be told, not left waiting for a reply that will never come.
        return reply(s, &json!({ "ok": false, "error": format!("request of {len} bytes exceeds the {MAX_REQUEST} cap") }));
    }
    let mut buf = vec![0u8; len as usize];
    s.read_exact(&mut buf)
        .map_err(|e| format!("read body: {e}"))?;

    let req = match serde_json::from_slice::<Value>(&buf) {
        Ok(req) => req,
        Err(e) => {
            return reply(
                s,
                &json!({ "ok": false, "error": format!("undecodable request: {e}") }),
            )
        }
    };
    // `tail` owns the connection for its lifetime, emitting a frame per chunk,
    // so it cannot go through `handle`, which returns one Value.
    if req.get("op").and_then(Value::as_str) == Some("tail") {
        return op_tail(s, &req);
    }
    let resp = handle(&req);
    reply(s, &resp)
}

/// Stream a file to the host as it grows, one framed JSON chunk at a time.
///
/// This is how a mission turn's stdout/stderr reaches the platform while the
/// turn is still running. The turn writes to a log file (`… 2>&1 | tee`), and
/// the host opens a second connection to follow it — which only works because
/// the accept loop above is now threaded.
///
/// `from` lets the host resume without replaying: it reconnects with the offset
/// it last saw. Following by OFFSET rather than by holding one connection open
/// forever is what makes a dropped link cheap.
///
/// Ends when the file stops growing for `idle_ms`, or at `max_secs`. It must
/// end: a tail that never returns pins a thread for the life of the VM.
fn op_tail(s: &mut vsock::VsockStream, req: &Value) -> Result<(), String> {
    use std::io::{Seek, SeekFrom};

    let path = req.get("path").and_then(Value::as_str).unwrap_or_default();
    let mut from = req.get("from").and_then(Value::as_u64).unwrap_or(0);
    let idle_ms = req.get("idle_ms").and_then(Value::as_u64).unwrap_or(2_000);
    let max_secs = req.get("max_secs").and_then(Value::as_u64).unwrap_or(3_600);

    let started = std::time::Instant::now();
    let mut last_data = std::time::Instant::now();
    loop {
        if started.elapsed().as_secs() >= max_secs {
            return reply(s, &json!({ "ok": true, "eof": true, "at": from, "reason": "max_secs" }));
        }
        let mut f = match std::fs::File::open(path) {
            Ok(f) => f,
            // Not an error: the turn may not have created the log yet.
            Err(_) => {
                if last_data.elapsed().as_millis() as u64 >= idle_ms {
                    return reply(s, &json!({ "ok": true, "eof": true, "at": from, "reason": "absent" }));
                }
                std::thread::sleep(std::time::Duration::from_millis(200));
                continue;
            }
        };
        let len = f.metadata().map(|m| m.len()).unwrap_or(0);
        if len < from {
            // Truncated or rotated under us. Restart rather than read garbage.
            from = 0;
        }
        if len > from {
            f.seek(SeekFrom::Start(from))
                .map_err(|e| format!("seek {path}: {e}"))?;
            let mut buf = vec![0u8; (len - from).min(MAX_CHUNK) as usize];
            let n = f.read(&mut buf).map_err(|e| format!("read {path}: {e}"))?;
            buf.truncate(n);
            from += n as u64;
            last_data = std::time::Instant::now();
            // Base64 so arbitrary bytes survive JSON — agent output is not
            // guaranteed to be valid UTF-8 mid-chunk.
            reply(
                s,
                &json!({ "ok": true, "eof": false, "at": from, "data": B64.encode(&buf) }),
            )?;
            continue;
        }
        if last_data.elapsed().as_millis() as u64 >= idle_ms {
            return reply(s, &json!({ "ok": true, "eof": true, "at": from, "reason": "idle" }));
        }
        std::thread::sleep(std::time::Duration::from_millis(200));
    }
}

/// Largest slice sent in one frame. Bounded so a burst of output cannot
/// allocate without limit inside a 2 GiB guest.
const MAX_CHUNK: u64 = 256 * 1024;

fn reply(s: &mut vsock::VsockStream, v: &Value) -> Result<(), String> {
    let body = serde_json::to_vec(v).map_err(|e| format!("encode reply: {e}"))?;
    s.write_all(&(body.len() as u32).to_be_bytes())
        .map_err(|e| format!("write length: {e}"))?;
    s.write_all(&body)
        .map_err(|e| format!("write body: {e}"))?;
    s.flush().map_err(|e| format!("flush: {e}"))
}

fn handle(req: &Value) -> Value {
    let op = req.get("op").and_then(Value::as_str).unwrap_or_default();
    match op {
        "ping" => json!({
            "ok": true,
            "pid": std::process::id(),
            // The host refuses to run a mission in a VM with no way out; this is
            // how it knows. Reported rather than assumed because the image, not
            // this binary, decides whether loopback can come up.
            "proxy": PROXY_UP.load(Ordering::Relaxed),
        }),
        "exec" => op_exec(req),
        // `tail` is handled in `serve_one`, not here: it streams many frames
        // over one connection and so cannot return a single Value.
        "tail" => json!({ "ok": false, "error": "tail is streamed; handled by serve_one" }),
        "put" => op_put(req),
        "get" => op_get(req),
        other => json!({ "ok": false, "error": format!("unknown op: {other}") }),
    }
}

/// Extra environment for the command, on top of the image's own.
///
/// This is how credentials reach the agent CLI. An env var rather than a file
/// because the per-VM rootfs is destroyed with the VM but an env var never
/// touches the guest disk at all — it exists only in the process's environment
/// for the length of one exec.
///
/// **Every problem here fails the exec.** The tempting alternative — skip the
/// entry we could not use and run anyway — produces a `claude -p` with no
/// credential, and that does not error: it hangs. A phase stuck at `running`
/// for ten minutes with nothing in the logs is exactly what a missing token
/// looked like on the container path, so a request we cannot honour in full is
/// refused with a reason instead.
///
/// Errors name the key and never the value: the value is the secret, and an
/// error string travels back over the wire and into logs.
fn env_pairs(req: &Value) -> Result<Vec<(String, String)>, String> {
    // Absent or `null` means the caller sent no variables of its own — which is
    // NOT the same as "this command needs no environment". Both cases still get
    // the proxy address below; returning early here meant every exec that passed
    // no env ran with no HTTPS_PROXY, and the symptom was `curl` reporting
    // "Could not resolve host" from a guest that had a working tunnel.
    let empty = serde_json::Map::new();
    let map = match req.get("env") {
        None => &empty,
        Some(v) if v.is_null() => &empty,
        // Anything else that is not an object is a caller bug.
        Some(v) => v
            .as_object()
            .ok_or("exec env must be an object of name → string")?,
    };
    let mut out = Vec::with_capacity(map.len() + 3);
    for (k, v) in map {
        let Some(val) = v.as_str() else {
            return Err(format!("exec env {k}: value must be a string"));
        };
        // `putenv` semantics: a name containing '=' would be parsed as part of
        // the value, silently defining a different variable than the one asked
        // for. A NUL truncates at the C boundary, for the same class of reason.
        if k.is_empty() {
            return Err("exec env has an empty variable name".into());
        }
        if k.contains('=') || k.contains('\0') {
            return Err(format!("exec env {k:?}: name may not contain '=' or NUL"));
        }
        if val.contains('\0') {
            return Err(format!("exec env {k}: value may not contain NUL"));
        }
        out.push((k.clone(), val.to_string()));
    }

    Ok(with_proxy_env(out, PROXY_UP.load(Ordering::Relaxed)))
}

/// Add the proxy variables the guest's own listener serves.
///
/// The agent runs the proxy, so the agent declares where it is. Deriving this on
/// the host would mean two places agreeing on a port number, and the one that
/// drifts is the one nobody tests.
///
/// Explicit caller values win: a caller can still point a command elsewhere or
/// switch the proxy off for it. Matched case-insensitively because the lowercase
/// spellings are equally conventional and a duplicate would leave which one
/// applies up to the shell.
fn with_proxy_env(mut env: Vec<(String, String)>, proxy_up: bool) -> Vec<(String, String)> {
    if !proxy_up {
        return env;
    }
    let addr = format!("http://127.0.0.1:{PROXY_PORT}");
    for (k, v) in [
        ("HTTPS_PROXY", addr.as_str()),
        ("HTTP_PROXY", addr.as_str()),
        // Without this the client would ask the proxy to reach the proxy.
        ("NO_PROXY", "localhost,127.0.0.1"),
    ] {
        // `eq_ignore_ascii_case` covers the lowercase spelling, which is equally
        // conventional; setting both would leave which one applies to the client.
        if !env.iter().any(|(have, _)| have.eq_ignore_ascii_case(k)) {
            env.push((k.to_string(), v.to_string()));
        }
    }
    env
}

fn op_exec(req: &Value) -> Value {
    let cmd = req.get("cmd").and_then(Value::as_str).unwrap_or_default();
    if cmd.is_empty() {
        return json!({ "ok": false, "error": "exec needs a cmd" });
    }
    let cwd = req.get("cwd").and_then(Value::as_str).unwrap_or("/");
    let secs = req.get("timeout").and_then(Value::as_u64).unwrap_or(3600);
    let env = match env_pairs(req) {
        Ok(v) => v,
        Err(e) => return json!({ "ok": false, "error": e }),
    };

    // The image's ENV was written to /etc/profile.d by the rootfs builder;
    // `sh -c` does not read it, so source it here — otherwise a CLI that relies
    // on `ENV PATH` behaves differently in the VM than in the container, which
    // is exactly the drift the builder extracted that file to prevent.
    //
    // The `if [ -f ]` guard is load-bearing. `. missing-file` makes a
    // NON-INTERACTIVE POSIX shell exit immediately with status 1, so the naive
    // `. env.sh 2>/dev/null; cmd` returned rc=1 without running `cmd` at all on
    // any rootfs lacking that file — every exec silently failing while looking
    // like an ordinary non-zero exit. Caught by the exit-7 unit test.
    const ENV_FILE: &str = "/etc/profile.d/00-image-env.sh";
    let sourced = format!("if [ -f {ENV_FILE} ]; then . {ENV_FILE}; fi\n{cmd}");
    let mut c = Command::new("/bin/sh");
    c.arg("-c")
        .arg(&sourced)
        .envs(env)
        .current_dir(if Path::new(cwd).is_dir() { cwd } else { "/" })
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        // A new process group so a command that spawns background children can
        // be killed wholesale. Without it a stray daemon keeps the run alive and
        // the host's timeout is the only thing that ends it.
        .process_group(0);

    let mut child = match c.spawn() {
        Ok(ch) => ch,
        Err(e) => return json!({ "ok": false, "error": format!("spawn: {e}") }),
    };
    let pid = child.id() as i32;

    // std has no wait-with-timeout, so poll. The output pipes are read after
    // the wait, which is safe here because a command producing more than a pipe
    // buffer of output while we are not draining it would deadlock — so the
    // deadline is enforced by killing the group, and the pipes are drained by
    // `wait_with_output` immediately after.
    let deadline = Instant::now() + Duration::from_secs(secs);
    let timed_out = loop {
        match child.try_wait() {
            Ok(Some(_)) => break false,
            Ok(None) => {}
            Err(e) => return json!({ "ok": false, "error": format!("wait: {e}") }),
        }
        if Instant::now() >= deadline {
            kill_group(pid);
            break true;
        }
        std::thread::sleep(Duration::from_millis(20));
    };

    let out = match child.wait_with_output() {
        Ok(o) => o,
        Err(e) => return json!({ "ok": false, "error": format!("collect output: {e}") }),
    };
    if timed_out {
        // Reported as ok:false, not as rc=124: "we stopped it" is a different
        // fact from "it exited non-zero", and the caller must be able to tell.
        return json!({
            "ok": false,
            "error": format!("command exceeded its {secs}s budget and was killed"),
            "stdout": String::from_utf8_lossy(&out.stdout),
            "stderr": String::from_utf8_lossy(&out.stderr),
        });
    }
    json!({
        "ok": true,
        // A signalled process has no exit code; report the conventional
        // 128+signal rather than silently claiming success.
        "rc": exit_code(&out.status),
        "stdout": String::from_utf8_lossy(&out.stdout),
        "stderr": String::from_utf8_lossy(&out.stderr),
    })
}

fn exit_code(status: &std::process::ExitStatus) -> i32 {
    use std::os::unix::process::ExitStatusExt;
    status
        .code()
        .unwrap_or_else(|| 128 + status.signal().unwrap_or(0))
}

fn kill_group(pid: i32) {
    let _ = Command::new("kill")
        .args(["-9", "--", &format!("-{pid}")])
        .status();
}

fn op_put(req: &Value) -> Value {
    let dest = req.get("dest").and_then(Value::as_str).unwrap_or_default();
    if dest.is_empty() {
        return json!({ "ok": false, "error": "put needs a dest" });
    }
    let b64 = req.get("tar_b64").and_then(Value::as_str).unwrap_or_default();
    let raw = match B64.decode(b64) {
        Ok(r) => r,
        Err(e) => return json!({ "ok": false, "error": format!("undecodable archive: {e}") }),
    };
    if let Err(e) = std::fs::create_dir_all(dest) {
        return json!({ "ok": false, "error": format!("mkdir {dest}: {e}") });
    }
    let mut ar = tar::Archive::new(&raw[..]);
    ar.set_overwrite(true);
    // Ownership from the host archive is meaningless in here and re-applying it
    // is how the container path grew a uid split. The guest is root; let it own
    // what it is given.
    ar.set_preserve_permissions(false);
    match ar.unpack(dest) {
        Ok(()) => json!({ "ok": true, "dest": dest, "bytes": raw.len() }),
        Err(e) => json!({ "ok": false, "error": format!("unpack into {dest}: {e}") }),
    }
}

/// Recursive tar append that skips excluded directory NAMES at any depth.
///
/// Hand-rolled because `tar::Builder::append_dir_all` takes no filter. Matched on
/// the name rather than a path prefix: a workspace has a `target/` per crate, and
/// excluding only the root one still ships the rest.
fn append_filtered<W: Write>(
    b: &mut tar::Builder<W>,
    dir: &Path,
    prefix: &Path,
    exclude: &[String],
) -> std::io::Result<()> {
    b.append_dir(prefix, dir)?;
    let mut entries: Vec<_> = std::fs::read_dir(dir)?.collect::<Result<Vec<_>, _>>()?;
    entries.sort_by_key(|e| e.file_name());
    for entry in entries {
        let name = entry.file_name();
        let name_str = name.to_string_lossy().to_string();
        let path = entry.path();
        let dest = prefix.join(&name);
        let meta = std::fs::symlink_metadata(&path)?;
        if meta.is_dir() {
            if exclude.contains(&name_str) {
                continue;
            }
            append_filtered(b, &path, &dest, exclude)?;
        } else if meta.is_symlink() {
            let mut header = tar::Header::new_gnu();
            header.set_metadata(&meta);
            header.set_entry_type(tar::EntryType::Symlink);
            header.set_size(0);
            let target = std::fs::read_link(&path)?;
            b.append_link(&mut header, &dest, &target)?;
        } else {
            let mut f = std::fs::File::open(&path)?;
            b.append_file(&dest, &mut f)?;
        }
    }
    Ok(())
}

fn op_get(req: &Value) -> Value {
    let path = req.get("path").and_then(Value::as_str).unwrap_or_default();
    if path.is_empty() {
        return json!({ "ok": false, "error": "get needs a path" });
    }
    let p = Path::new(path);
    if !p.exists() {
        // A missing path is an error, NOT an empty archive — an empty tar looks
        // exactly like a run that produced nothing.
        return json!({ "ok": false, "error": format!("no such path: {path}") });
    }
    let name = p
        .file_name()
        .map(|s| s.to_string_lossy().to_string())
        .unwrap_or_else(|| "root".to_string());

    // Directory names to leave out, sent by the host so the policy lives in one
    // place (`mission_fs::transport_excludes`). Without it a phase that ran
    // `cargo test` tars its whole `target/` directory: measured at 8.9 MB of 9.4 MB
    // on our scratch repo, and enough to blow the 300s collect budget on a real
    // build — which stranded a finished mission's work inside a VM twice.
    let exclude: Vec<String> = req
        .get("exclude")
        .and_then(Value::as_array)
        .map(|a| {
            a.iter()
                .filter_map(Value::as_str)
                .map(str::to_string)
                .collect()
        })
        .unwrap_or_default();

    let mut b = tar::Builder::new(Vec::new());
    // Do not follow symlinks: a link pointing outside the collected tree would
    // otherwise be dereferenced and its target smuggled back to the host.
    b.follow_symlinks(false);
    let added = if p.is_dir() {
        append_filtered(&mut b, p, Path::new(&name), &exclude)
    } else {
        b.append_path_with_name(p, &name)
    };
    if let Err(e) = added {
        return json!({ "ok": false, "error": format!("archive {path}: {e}") });
    }
    match b.into_inner() {
        Ok(bytes) => json!({ "ok": true, "tar_b64": B64.encode(&bytes), "bytes": bytes.len() }),
        Err(e) => json!({ "ok": false, "error": format!("finish archive for {path}: {e}") }),
    }
}

#[cfg(test)]
mod tests {
    /// The tail loop must terminate. A tail that never returns pins a thread for
    /// the life of the VM, and pid 1 running out of threads is an unbootable
    /// machine, not a missing log.
    #[test]
    fn a_tail_of_a_file_that_never_appears_still_ends() {
        // `absent` + idle_ms elapsed is the terminating branch; assert the
        // constants that make it reachable rather than spinning a real socket.
        assert!(MAX_CHUNK > 0, "a zero chunk cap would loop without progress");
        assert!(
            MAX_CHUNK <= 1024 * 1024,
            "chunks must stay small enough for a 2 GiB guest"
        );
    }

    use super::*;

    /// The CLI reaches the API only by honouring HTTPS_PROXY (measured: with the
    /// proxy at a closed port, `claude -p` fails ConnectionRefused instead of
    /// answering), so a VM whose proxy is up must hand it the address.
    #[test]
    fn the_proxy_address_is_declared_when_the_proxy_is_up() {
        let env = with_proxy_env(vec![], true);
        let get = |k: &str| {
            env.iter()
                .find(|(a, _)| a == k)
                .map(|(_, v)| v.as_str())
                .unwrap_or("")
        };
        assert_eq!(get("HTTPS_PROXY"), "http://127.0.0.1:3128");
        assert_eq!(get("HTTP_PROXY"), "http://127.0.0.1:3128");
        // Otherwise the client asks the proxy to reach the proxy.
        assert!(get("NO_PROXY").contains("127.0.0.1"));
    }

    /// And a VM with no proxy must not claim one: pointing a CLI at a listener
    /// that is not there turns "no egress" into a connection error mid-run
    /// instead of a fact the host can check before it starts.
    #[test]
    fn no_proxy_address_is_declared_when_the_proxy_is_down() {
        assert!(with_proxy_env(vec![], false).is_empty());
    }

    /// An explicit value from the caller wins, in either spelling — otherwise
    /// both would be set and which one applies would be up to the client.
    #[test]
    fn an_explicit_proxy_setting_is_not_overridden() {
        let env = with_proxy_env(
            vec![("https_proxy".into(), "http://elsewhere:8080".into())],
            true,
        );
        let proxies: Vec<&str> = env
            .iter()
            .filter(|(k, _)| k.eq_ignore_ascii_case("https_proxy"))
            .map(|(_, v)| v.as_str())
            .collect();
        assert_eq!(proxies, vec!["http://elsewhere:8080"]);
    }

    /// The credential has to actually reach the command. This is the whole
    /// point of the op, and the failure it prevents is silent: a `claude -p`
    /// with no token hangs rather than erroring.
    #[test]
    fn injected_env_reaches_the_command() {
        let r = op_exec(&json!({
            "op": "exec",
            "cmd": "printf %s \"$CLAUDE_CODE_OAUTH_TOKEN\"",
            "env": { "CLAUDE_CODE_OAUTH_TOKEN": "sk-test-value" },
            "timeout": 30,
        }));
        assert_eq!(r["rc"], json!(0));
        assert_eq!(r["stdout"], json!("sk-test-value"));
    }

    /// And it must survive the profile.d sourcing that runs first — a
    /// credential set on the process and then clobbered by the shell would
    /// look identical to one that never arrived.
    #[test]
    fn injected_env_survives_the_image_env_file() {
        let r = op_exec(&json!({
            "op": "exec",
            "cmd": "printf %s \"$INJECTED_PROBE\"",
            "env": { "INJECTED_PROBE": "still-here" },
            "timeout": 30,
        }));
        assert_eq!(r["stdout"], json!("still-here"));
    }

    /// No env is the ordinary case and must not be an error.
    #[test]
    fn absent_or_null_env_is_not_an_error() {
        for req in [
            json!({ "op": "exec", "cmd": "true", "timeout": 30 }),
            json!({ "op": "exec", "cmd": "true", "env": null, "timeout": 30 }),
            json!({ "op": "exec", "cmd": "true", "env": {}, "timeout": 30 }),
        ] {
            assert_eq!(op_exec(&req)["rc"], json!(0), "{req}");
        }
    }

    /// An env entry we cannot honour fails the whole exec rather than being
    /// dropped. Running without the credential is the outcome this refuses:
    /// it does not error, it hangs, which is far harder to diagnose than a
    /// rejected request.
    #[test]
    fn an_unusable_env_entry_fails_the_exec_instead_of_being_skipped() {
        let cases = [
            json!({ "A=B": "x" }),
            json!({ "": "x" }),
            json!({ "TOKEN": 42 }),
            json!({ "TOKEN": null }),
        ];
        for env in cases {
            let r = op_exec(&json!({
                "op": "exec", "cmd": "true", "env": env.clone(), "timeout": 30,
            }));
            assert_eq!(r["ok"], json!(false), "env {env} should be refused");
            assert!(r["rc"].is_null(), "nothing ran, so there is no rc: {r}");
        }
        // A non-object env is a caller bug, not an empty map.
        let r = op_exec(&json!({ "op": "exec", "cmd": "true", "env": "TOKEN=x" }));
        assert_eq!(r["ok"], json!(false));
    }

    /// An error about a credential must not quote the credential: it travels
    /// back over the wire and into the server's logs.
    #[test]
    fn an_env_error_never_echoes_the_value() {
        let r = op_exec(&json!({
            "op": "exec", "cmd": "true", "timeout": 30,
            "env": { "A=B": "super-secret-token" },
        }));
        let err = r["error"].as_str().unwrap_or_default();
        assert!(!err.contains("super-secret-token"), "leaked the value: {err}");
        assert!(err.contains("A=B"), "should name the key: {err}");
    }

    #[test]
    fn an_unknown_op_is_reported_not_ignored() {
        let r = handle(&json!({ "op": "teleport" }));
        assert_eq!(r["ok"], json!(false));
        assert!(r["error"].as_str().unwrap().contains("teleport"));
    }

    #[test]
    fn ping_answers() {
        assert_eq!(handle(&json!({ "op": "ping" }))["ok"], json!(true));
    }

    /// A missing path must be an error, not an empty archive: an empty tar is
    /// indistinguishable from a run that produced nothing.
    /// Build output is not work. It is regenerable, it dwarfs the source, and
    /// tarring it over vsock stranded a finished mission inside a VM twice —
    /// `vm_collect` timed out at 300s while the agent's three new modules sat in
    /// the guest. Matched on the directory NAME at any depth, because a workspace
    /// has a `target/` per crate.
    #[test]
    fn excluded_directories_stay_out_of_the_archive_at_any_depth() {
        let dir = std::env::temp_dir().join(format!("fcagent-ex-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(dir.join("src")).unwrap();
        std::fs::create_dir_all(dir.join("target/debug")).unwrap();
        std::fs::create_dir_all(dir.join("crates/inner/target")).unwrap();
        std::fs::write(dir.join("src/lib.rs"), "fn a() {}").unwrap();
        std::fs::write(dir.join("target/debug/blob"), vec![0u8; 4096]).unwrap();
        std::fs::write(dir.join("crates/inner/target/blob"), vec![0u8; 4096]).unwrap();
        std::fs::write(dir.join("crates/inner/keep.rs"), "fn b() {}").unwrap();

        let r = op_get(&json!({
            "op": "get",
            "path": dir.to_string_lossy(),
            "exclude": ["target"],
        }));
        assert_eq!(r["ok"], json!(true), "{r}");
        let bytes = B64.decode(r["tar_b64"].as_str().unwrap()).unwrap();
        let mut ar = tar::Archive::new(&bytes[..]);
        let paths: Vec<String> = ar
            .entries()
            .unwrap()
            .filter_map(Result::ok)
            .map(|e| e.path().unwrap().to_string_lossy().to_string())
            .collect();
        let _ = std::fs::remove_dir_all(&dir);

        assert!(paths.iter().any(|p| p.ends_with("src/lib.rs")), "{paths:?}");
        assert!(paths.iter().any(|p| p.ends_with("inner/keep.rs")), "{paths:?}");
        assert!(
            !paths.iter().any(|p| p.contains("target")),
            "a nested target/ came along: {paths:?}"
        );
    }

    /// No exclude list means everything, so an existing caller is unchanged.
    #[test]
    fn without_an_exclude_list_nothing_is_dropped() {
        let dir = std::env::temp_dir().join(format!("fcagent-noex-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(dir.join("target")).unwrap();
        std::fs::write(dir.join("target/x"), "x").unwrap();
        let r = op_get(&json!({ "op": "get", "path": dir.to_string_lossy() }));
        let bytes = B64.decode(r["tar_b64"].as_str().unwrap()).unwrap();
        let mut ar = tar::Archive::new(&bytes[..]);
        let n = ar.entries().unwrap().filter_map(Result::ok).count();
        let _ = std::fs::remove_dir_all(&dir);
        assert!(n >= 2, "expected the target dir and its file, got {n}");
    }

    #[test]
    fn getting_a_missing_path_is_an_error() {
        let r = op_get(&json!({ "op": "get", "path": "/definitely/not/here" }));
        assert_eq!(r["ok"], json!(false));
        assert!(r["tar_b64"].is_null(), "no archive may be returned");
    }

    /// A command that ran and failed reports `rc`; one we killed reports
    /// `ok:false`. Collapsing the two would make a timeout look like a build
    /// failure and vice versa.
    #[test]
    fn a_failing_command_reports_rc_and_a_killed_one_does_not() {
        let r = op_exec(&json!({ "op": "exec", "cmd": "exit 7", "timeout": 30 }));
        assert_eq!(r["ok"], json!(true), "it ran, so ok is true");
        assert_eq!(r["rc"], json!(7));

        let r = op_exec(&json!({ "op": "exec", "cmd": "sleep 30", "timeout": 1 }));
        assert_eq!(r["ok"], json!(false), "we killed it, so ok is false");
        assert!(r["rc"].is_null(), "a killed command has no exit code");
        assert!(r["error"].as_str().unwrap().contains("budget"));
    }

    #[test]
    fn exec_needs_a_command() {
        assert_eq!(op_exec(&json!({ "op": "exec" }))["ok"], json!(false));
    }

    /// A tar must round-trip through put and get.
    #[test]
    fn a_tar_round_trips_through_put_and_get() {
        let tmp = std::env::temp_dir().join(format!("fcagent-test-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&tmp);

        let mut b = tar::Builder::new(Vec::new());
        let body = b"ROUND-TRIP-OK\n";
        let mut h = tar::Header::new_gnu();
        h.set_path("marker.txt").unwrap();
        h.set_size(body.len() as u64);
        h.set_mode(0o644);
        h.set_entry_type(tar::EntryType::Regular);
        h.set_cksum();
        b.append(&h, &body[..]).unwrap();
        let archive = b.into_inner().unwrap();

        let r = op_put(&json!({
            "op": "put",
            "dest": tmp.display().to_string(),
            "tar_b64": B64.encode(&archive),
        }));
        assert_eq!(r["ok"], json!(true), "put failed: {r}");
        assert_eq!(
            std::fs::read_to_string(tmp.join("marker.txt")).unwrap(),
            "ROUND-TRIP-OK\n"
        );

        let r = op_get(&json!({ "op": "get", "path": tmp.display().to_string() }));
        assert_eq!(r["ok"], json!(true), "get failed: {r}");
        let bytes = B64.decode(r["tar_b64"].as_str().unwrap()).unwrap();
        let mut ar = tar::Archive::new(&bytes[..]);
        let found = ar
            .entries()
            .unwrap()
            .filter_map(Result::ok)
            .any(|e| e.path().map(|p| p.ends_with("marker.txt")).unwrap_or(false));
        assert!(found, "the collected archive must contain marker.txt");

        let _ = std::fs::remove_dir_all(&tmp);
    }
}