newt-eval 0.8.0

//! Subprocess runner that drives a `newt worker` over ACP stdio JSON-RPC.
//!
//! For each [`TestCase`] the runner:
//!
//! 1. Copies the case's `workspace/` fixture into a fresh tempdir,
//!    `git init`s it, and commits the baseline so post-turn diffs are
//!    meaningful.
//! 2. Copies the same fixture into a second `baseline/` tempdir for the
//!    evaluators that need a pre-state to compare against.
//! 3. Spawns `newt worker` as a child process with `OLLAMA_HOST` set
//!    (mock URL in mock mode, unset in live mode so the worker uses its
//!    default endpoint search).
//! 4. Drives the ACP protocol over stdin/stdout: `initialize` →
//!    `new_session` → optional `set_session_model` → `prompt`.
//! 5. Returns the [`TaskReply`] plus the post-worker workspace path for
//!    evaluators to inspect.
//!
//! Each case is **completely isolated** — fresh tempdirs, fresh git init,
//! fresh subprocess. No state crosses between cases.

use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::time::Duration;

use newt_acp_worker::TaskReply;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, Command};

use crate::cases::TestCase;

/// What the runner returns after one full case execution.
///
/// `workspace` and `baseline` are tempdir paths kept alive by the embedded
/// `_*_guard` fields until the outcome is dropped.
#[derive(Debug)]
pub struct RunOutcome {
    pub case: TestCase,
    pub reply: TaskReply,
    pub workspace: PathBuf,
    pub baseline: PathBuf,
    /// The spec's content id and visibility, recorded before the worker ran.
    pub pre_run: crate::grade::PreRun,
    /// Kept alive so the temp dirs aren't pruned out from under the
    /// evaluators. Dropped when the outcome is dropped.
    pub _workspace_guard: tempfile::TempDir,
    pub _baseline_guard: tempfile::TempDir,
}

/// How to run the worker. `mock_endpoint` is set in mock mode (the
/// wiremock URL); `None` in live mode (the worker discovers Ollama via
/// `OLLAMA_HOST` or its default endpoint list).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunnerConfig {
    pub worker_bin: PathBuf,
    pub mock_endpoint: Option<String>,
    pub model_override: Option<String>,
    /// Hard upper bound on a single case. 60s is roomy for mock and
    /// short enough that a hung live model doesn't stall CI forever.
    pub timeout: Duration,
    /// When true, spawn the worker with `--coder` so the newt-coder
    /// plugin handles prompts (whole-file emit + diff normalization).
    /// Closes failure mode T0b for local Ollama models that can't
    /// fabricate valid hunk headers.
    pub coder_mode: bool,
}

impl RunnerConfig {
    pub fn new(worker_bin: impl Into<PathBuf>) -> Self {
        Self {
            worker_bin: worker_bin.into(),
            mock_endpoint: None,
            model_override: None,
            timeout: Duration::from_secs(60),
            coder_mode: false,
        }
    }

    pub fn with_mock_endpoint(mut self, url: impl Into<String>) -> Self {
        self.mock_endpoint = Some(url.into());
        self
    }

    pub fn with_model(mut self, model: impl Into<String>) -> Self {
        self.model_override = Some(model.into());
        self
    }

    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    pub fn with_coder_mode(mut self, on: bool) -> Self {
        self.coder_mode = on;
        self
    }
}

/// Run one case end-to-end. See module docs for the protocol.
pub async fn run_case(case: &TestCase, config: &RunnerConfig) -> anyhow::Result<RunOutcome> {
    if !config.worker_bin.exists() {
        anyhow::bail!(
            "newt worker binary not found at {} — run `cargo build --bin newt`",
            config.worker_bin.display()
        );
    }

    let workspace_guard = tempfile::tempdir()?;
    let baseline_guard = tempfile::tempdir()?;
    let workspace = workspace_guard.path().to_path_buf();
    let baseline = baseline_guard.path().to_path_buf();

    copy_fixture(&case.workspace_fixture(), &workspace)?;
    copy_fixture(&case.workspace_fixture(), &baseline)?;
    init_baseline_git(&workspace)?;
    let pre_run = crate::grade::pre_run(case, &workspace)?;

    // `_home_guard` isolates the worker's `HOME`; it must outlive the child.
    let (mut child, _home_guard) = spawn_worker(config)?;

    let result = tokio::time::timeout(config.timeout, async {
        let reply = drive_acp(
            &mut child,
            &workspace,
            &case.prompt,
            config.model_override.as_deref(),
        )
        .await?;
        // drive_acp closes stdin. Let EOF finish the worker and its exit-time
        // writes under the SAME case deadline before grading its reply.
        let status = child.wait().await?;
        anyhow::ensure!(status.success(), "worker exited with {status}");
        Ok(reply)
    })
    .await;

    // A failed or timed-out conversation still kills and reaps its child.
    // Successful workers are already reaped; killing after a reply used to
    // race their finalization (including instrumentation profile writes).
    if !matches!(result, Ok(Ok(_))) {
        let _ = child.start_kill();
        let _ = child.wait().await;
    }

    let reply = match result {
        Ok(Ok(reply)) => reply,
        Ok(Err(e)) => return Err(e),
        Err(_) => anyhow::bail!("worker timed out after {:?}", config.timeout),
    };

    Ok(RunOutcome {
        case: case.clone(),
        reply,
        workspace,
        baseline,
        pre_run,
        _workspace_guard: workspace_guard,
        _baseline_guard: baseline_guard,
    })
}

/// Recursively copy `src` into `dst` (which must exist).
pub(crate) fn copy_fixture(src: &Path, dst: &Path) -> anyhow::Result<()> {
    let mut opts = fs_extra::dir::CopyOptions::new();
    opts.content_only = true;
    opts.overwrite = true;
    fs_extra::dir::copy(src, dst, &opts)
        .map_err(|e| anyhow::anyhow!("copy fixture {} -> {}: {e}", src.display(), dst.display()))?;
    Ok(())
}

/// `git init` + commit everything as baseline so the worker's
/// `git diff --no-color` capture step has something to compare against.
///
/// `GIT_DIR` / `GIT_WORK_TREE` / friends are stripped from the child
/// env so this works when invoked from inside a git hook (e.g. the
/// pre-push hook that calls `cargo test`). Without this, `git init`
/// targets the hook's repo instead of `workspace`.
pub(crate) fn init_baseline_git(workspace: &Path) -> anyhow::Result<()> {
    let run = |args: &[&str]| -> anyhow::Result<()> {
        let output = std::process::Command::new("git")
            .args(args)
            .current_dir(workspace)
            .env_remove("GIT_DIR")
            .env_remove("GIT_WORK_TREE")
            .env_remove("GIT_INDEX_FILE")
            .env_remove("GIT_COMMON_DIR")
            .env_remove("GIT_PREFIX")
            .output()?;
        if !output.status.success() {
            anyhow::bail!(
                "git {} failed: {}",
                args.join(" "),
                String::from_utf8_lossy(&output.stderr)
            );
        }
        Ok(())
    };
    run(&["init", "-q", "-b", "main"])?;
    run(&["config", "user.email", "eval@newt-eval"])?;
    run(&["config", "user.name", "newt-eval"])?;
    run(&["add", "-A"])?;
    run(&["commit", "-q", "-m", "baseline", "--allow-empty"])?;
    Ok(())
}

/// Build the `newt worker` command (env + args) without spawning it.
/// Split from [`spawn_worker`] so the eval harness's environment contract
/// — notably the ephemeral default — is unit-testable.
fn worker_command(config: &RunnerConfig) -> Command {
    let mut cmd = Command::new(&config.worker_bin);
    cmd.arg("worker")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true);

    if config.coder_mode {
        cmd.arg("--coder");
    }

    if let Some(url) = &config.mock_endpoint {
        cmd.env("OLLAMA_HOST", url);
    }

    // The worker honors NEWT_DEFAULT_MODEL when picking the initial
    // backend model. Setting it AND firing the ACP set_session_model
    // both belt-and-suspenders the model choice — the env wins at
    // backend-construction time, the ACP call is a no-op today but
    // documents intent at the wire layer.
    if let Some(model) = &config.model_override {
        cmd.env("NEWT_DEFAULT_MODEL", model);
    }

    // Eval runs are ALWAYS ephemeral (Step 17.7; the conversation-context
    // decision doc's own requirement): a graded run must never create or
    // resume conversation rows, or ambient journals would leak into the
    // grade and eval runs would pollute real workspaces — the eval gate
    // stays honest. Today's worker has no conversation store at all, so
    // this is defense-in-depth: if persistence ever reaches the ACP
    // worker, it must honor NEWT_EPHEMERAL, and this line already opts
    // every eval run out.
    cmd.env("NEWT_EPHEMERAL", "1");

    // The worker's tracing_subscriber writes to STDOUT, which is also
    // the ACP wire — any log line would corrupt JSON-RPC parsing. So
    // we silence the worker entirely. (Live-mode users who want logs
    // can run the worker by hand outside the eval harness.)
    cmd.env("RUST_LOG", "off");

    cmd
}

/// Spawn `newt worker` in an ISOLATED `HOME` so the eval never reads a
/// developer's real `~/.newt/config.toml` — which would leak that machine's
/// configured backend/model into the graded run. (The pre-#1526 worker made
/// `$OLLAMA_HOST`'s mere presence force local discovery, which incidentally
/// masked the dev config; #1526 made `$OLLAMA_HOST` a mere endpoint hint that no
/// longer erases a configured backend, so the eval must isolate `HOME` itself.)
/// Returns the tempdir guard the caller MUST keep alive until the worker exits.
fn spawn_worker(config: &RunnerConfig) -> anyhow::Result<(Child, tempfile::TempDir)> {
    let home = tempfile::tempdir()?;
    std::fs::create_dir_all(home.path().join(".newt"))?;
    std::fs::write(home.path().join(".newt/config.toml"), "")?;
    let mut cmd = worker_command(config);
    cmd.env("HOME", home.path())
        .env_remove("NEWT_CONFIG")
        .env_remove("NEWT_CONFIG_DIR")
        // Config resolution ALSO walks up from the worker's cwd for a
        // project-local `.newt/config.toml`; if cwd stays under the developer's
        // home, that walk finds `~/.newt/config.toml` regardless of `HOME`. Root
        // the worker in the isolated tempdir so the walk-up finds nothing. The
        // worker edits the workspace via the absolute ACP `workspace_path`, not
        // cwd, so this does not affect its file operations.
        .current_dir(home.path());
    let child = cmd
        .spawn()
        .map_err(|e| anyhow::anyhow!("spawn {}: {e}", config.worker_bin.display()))?;
    Ok((child, home))
}

/// Send the ACP request sequence and return the parsed [`TaskReply`].
async fn drive_acp(
    child: &mut Child,
    workspace: &Path,
    prompt: &str,
    model_override: Option<&str>,
) -> anyhow::Result<TaskReply> {
    let mut stdin = child
        .stdin
        .take()
        .ok_or_else(|| anyhow::anyhow!("worker stdin already taken"))?;
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| anyhow::anyhow!("worker stdout already taken"))?;
    let mut stdout = BufReader::new(stdout).lines();

    // 1. initialize
    send_line(
        &mut stdin,
        &json_rpc("initialize", 1, serde_json::json!({})),
    )
    .await?;
    let _init = read_response(&mut stdout, 1).await?;

    // 2. new_session
    let new_session = json_rpc(
        "new_session",
        2,
        serde_json::json!({ "workspace_path": workspace.to_string_lossy() }),
    );
    send_line(&mut stdin, &new_session).await?;
    let session_resp = read_response(&mut stdout, 2).await?;
    let session_id = session_resp["result"]["session_id"]
        .as_str()
        .ok_or_else(|| anyhow::anyhow!("new_session response missing session_id"))?
        .to_string();

    // 3. set_session_model (optional)
    if let Some(model) = model_override {
        let req = json_rpc(
            "set_session_model",
            3,
            serde_json::json!({ "session_id": session_id, "model": model }),
        );
        send_line(&mut stdin, &req).await?;
        let _ = read_response(&mut stdout, 3).await?;
    }

    // 4. prompt
    let prompt_req = json_rpc(
        "prompt",
        4,
        serde_json::json!({ "session_id": session_id, "prompt": prompt }),
    );
    send_line(&mut stdin, &prompt_req).await?;
    let prompt_resp = read_response(&mut stdout, 4).await?;

    let reply: TaskReply = serde_json::from_value(prompt_resp["result"].clone())
        .map_err(|e| anyhow::anyhow!("decode TaskReply: {e}"))?;

    // Close stdin so the worker's `lines()` loop exits cleanly.
    drop(stdin);

    Ok(reply)
}

/// Build a JSON-RPC request value.
fn json_rpc(method: &str, id: i64, params: Value) -> Value {
    serde_json::json!({
        "jsonrpc": "2.0",
        "id": id,
        "method": method,
        "params": params,
    })
}

/// Serialize `value` as a single newline-terminated line and write it.
async fn send_line<W: tokio::io::AsyncWrite + Unpin>(
    writer: &mut W,
    value: &Value,
) -> anyhow::Result<()> {
    let mut s = serde_json::to_string(value)?;
    s.push('\n');
    writer.write_all(s.as_bytes()).await?;
    writer.flush().await?;
    Ok(())
}

/// Read one JSON-RPC response from `lines` and check its `id`.
async fn read_response<R: AsyncBufRead + Unpin>(
    lines: &mut tokio::io::Lines<R>,
    expected_id: i64,
) -> anyhow::Result<Value> {
    let line = lines
        .next_line()
        .await?
        .ok_or_else(|| anyhow::anyhow!("worker closed stdout before responding"))?;
    let v: Value = serde_json::from_str(&line)
        .map_err(|e| anyhow::anyhow!("parse worker response '{line}': {e}"))?;
    if let Some(err) = v.get("error") {
        if !err.is_null() {
            anyhow::bail!("worker returned JSON-RPC error: {err}");
        }
    }
    if v.get("id").and_then(Value::as_i64) != Some(expected_id) {
        anyhow::bail!(
            "worker response id mismatch: expected {expected_id}, got {:?}",
            v.get("id")
        );
    }
    Ok(v)
}

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

    #[test]
    fn config_builders_compose() {
        let cfg = RunnerConfig::new("/tmp/newt")
            .with_mock_endpoint("http://127.0.0.1:8080")
            .with_model("llama3.1:8b")
            .with_timeout(Duration::from_secs(5));
        assert_eq!(cfg.worker_bin, PathBuf::from("/tmp/newt"));
        assert_eq!(cfg.mock_endpoint.as_deref(), Some("http://127.0.0.1:8080"));
        assert_eq!(cfg.model_override.as_deref(), Some("llama3.1:8b"));
        assert_eq!(cfg.timeout, Duration::from_secs(5));
    }

    #[test]
    fn worker_runs_ephemeral_by_default() {
        // 17.7: every eval-spawned worker carries NEWT_EPHEMERAL=1 — no
        // conversation rows from graded runs, ever (the eval gate stays
        // honest). Asserted on the built command so the contract holds
        // without spawning anything.
        let cfg = RunnerConfig::new("/tmp/newt");
        let cmd = worker_command(&cfg);
        let envs: Vec<_> = cmd.as_std().get_envs().collect();
        assert!(
            envs.iter()
                .any(|(k, v)| *k == "NEWT_EPHEMERAL" && *v == Some("1".as_ref())),
            "eval worker must default ephemeral, got env: {envs:?}"
        );
    }

    #[test]
    fn json_rpc_builds_request() {
        let req = json_rpc("ping", 7, serde_json::json!({"k": "v"}));
        assert_eq!(req["jsonrpc"], "2.0");
        assert_eq!(req["id"], 7);
        assert_eq!(req["method"], "ping");
        assert_eq!(req["params"]["k"], "v");
    }

    #[tokio::test]
    async fn run_case_errors_when_worker_bin_missing() {
        let tmp = tempfile::tempdir().unwrap();
        let case_dir = tmp.path().join("case");
        std::fs::create_dir_all(case_dir.join("workspace")).unwrap();
        std::fs::write(
            case_dir.join("case.toml"),
            r#"
name = "x"
description = ""
language = "rust"
prompt = ""
evaluators = []

[mock_response]
content = ""
"#,
        )
        .unwrap();
        let case = TestCase::load_dir(&case_dir).unwrap();
        let cfg = RunnerConfig::new("/nonexistent/newt");
        let err = run_case(&case, &cfg).await.unwrap_err();
        assert!(err.to_string().contains("not found"));
    }

    #[cfg(unix)]
    fn finalizing_worker(body: &str) -> (tempfile::TempDir, TestCase, RunnerConfig) {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir(dir.path().join("workspace")).unwrap();
        std::fs::write(dir.path().join("workspace/input.txt"), "fixture").unwrap();
        std::fs::write(
            dir.path().join("case.toml"),
            "name = 'shutdown'\ndescription = ''\nlanguage = 'text'\nprompt = ''\nevaluators = []\n[mock_response]\ncontent = ''\n",
        )
        .unwrap();
        let worker = dir.path().join("worker");
        let script = format!(
            r#"#!/bin/sh
fixture=${{0%/*}}
printf '%s' "$$" > "$fixture/pid"
IFS= read -r request || exit 90
printf '%s\n' '{{"jsonrpc":"2.0","id":1,"result":{{}}}}'
IFS= read -r request || exit 91
printf '%s\n' '{{"jsonrpc":"2.0","id":2,"result":{{"session_id":"fixture"}}}}'
IFS= read -r request || exit 92
printf '%s\n' '{{"jsonrpc":"2.0","id":4,"result":{{"model_id":"fixture","content":"done","diff":"","empty_diff":true,"diff_applied":false}}}}'
IFS= read -r request && exit 93
printf '%s' eof > "$fixture/eof"
{body}
"#
        );
        std::fs::write(&worker, script).unwrap();
        std::fs::set_permissions(&worker, std::fs::Permissions::from_mode(0o700)).unwrap();
        let case = TestCase::load_dir(dir.path()).unwrap();
        let config = RunnerConfig::new(worker).with_timeout(Duration::from_secs(5));
        (dir, case, config)
    }

    /// Grounds the ACP EOF cleanup contract in a real child: a successful reply
    /// does not authorize killing a worker while it finalizes its output.
    #[cfg(unix)]
    #[tokio::test]
    async fn worker_shutdown_waits_for_eof_finalization() {
        let (dir, case, config) = finalizing_worker(
            "while [ ! -f \"$fixture/release\" ]; do :; done\nprintf '%s' finalized > \"$fixture/finalized\"",
        );
        let run = run_case(&case, &config);
        tokio::pin!(run);
        tokio::select! {
            result = &mut run => panic!("runner returned before releasing worker finalization: {result:?}"),
            () = wait_for_worker_eof(dir.path()) => {}
        }
        assert!(
            tokio::time::timeout(Duration::from_millis(20), &mut run)
                .await
                .is_err(),
            "runner must wait for the child's finalization barrier"
        );
        std::fs::write(dir.path().join("release"), "release").unwrap();
        let outcome = run.await.expect("clean worker exit");
        assert_eq!(outcome.reply.content, "done");
        assert_eq!(
            std::fs::read_to_string(dir.path().join("finalized")).unwrap(),
            "finalized"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn worker_shutdown_rejects_nonzero_exit_after_reply() {
        let (_dir, case, config) = finalizing_worker("exit 17");
        let error = run_case(&case, &config).await.unwrap_err();
        assert!(error.to_string().contains("worker exited"), "{error:#}");
        assert!(error.to_string().contains("17"), "{error:#}");
    }

    /// A valid reply cannot make an uncooperative child outlive the case's
    /// existing deadline; waiting still kills and reaps that real process.
    #[cfg(unix)]
    #[tokio::test]
    async fn worker_shutdown_timeout_kills_and_reaps() {
        let (dir, case, config) = finalizing_worker("while :; do :; done");
        let run = run_case(&case, &config);
        tokio::pin!(run);
        tokio::select! {
            result = &mut run => panic!("runner returned before worker reached shutdown: {result:?}"),
            () = wait_for_worker_eof(dir.path()) => {}
        }
        // Expire the existing case deadline only after real ACP consumption
        // and EOF. Resume before waiting on OS reaping so virtual auto-advance
        // cannot race SIGCHLD; protocol startup has its normal fixture budget.
        tokio::time::pause();
        tokio::time::advance(config.timeout).await;
        tokio::time::resume();
        let error = tokio::time::timeout(Duration::from_secs(5), run)
            .await
            .expect("cleanup must stay bounded")
            .unwrap_err();
        assert!(error.to_string().contains("timed out"), "{error:#}");
        assert!(dir.path().join("eof").exists(), "worker reached shutdown");
        assert_worker_reaped(dir.path());
    }

    #[cfg(unix)]
    async fn wait_for_worker_eof(dir: &Path) {
        while !dir.join("eof").exists() {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn worker_shutdown_preserves_protocol_error_and_reaps() {
        let (dir, case, config) = finalizing_worker("while :; do :; done");
        let script = std::fs::read_to_string(&config.worker_bin).unwrap();
        let script = script.replace("\"id\":4,\"result\":", "\"id\":4,\"error\":");
        std::fs::write(&config.worker_bin, script).unwrap();
        let error = run_case(&case, &config).await.unwrap_err();
        assert!(
            error.to_string().contains("worker returned JSON-RPC error"),
            "{error:#}"
        );
        assert_worker_reaped(dir.path());
    }

    #[cfg(unix)]
    fn assert_worker_reaped(dir: &Path) {
        let pid = std::fs::read_to_string(dir.join("pid")).unwrap();
        let status = std::process::Command::new("kill")
            .args(["-0", &pid])
            .stderr(Stdio::null())
            .status()
            .unwrap();
        assert!(!status.success(), "worker must be reaped");
    }

    #[test]
    fn copy_fixture_clones_tree() {
        let src = tempfile::tempdir().unwrap();
        std::fs::write(src.path().join("a.txt"), "hi").unwrap();
        std::fs::create_dir_all(src.path().join("sub")).unwrap();
        std::fs::write(src.path().join("sub/b.txt"), "bye").unwrap();

        let dst = tempfile::tempdir().unwrap();
        copy_fixture(src.path(), dst.path()).unwrap();

        assert_eq!(
            std::fs::read_to_string(dst.path().join("a.txt")).unwrap(),
            "hi"
        );
        assert_eq!(
            std::fs::read_to_string(dst.path().join("sub/b.txt")).unwrap(),
            "bye"
        );
    }

    #[test]
    fn init_baseline_git_creates_repo() {
        let tmp = tempfile::tempdir().unwrap();
        std::fs::write(tmp.path().join("hello.txt"), "hi").unwrap();
        init_baseline_git(tmp.path()).unwrap();
        // .git directory exists
        assert!(tmp.path().join(".git").exists());
        // and a clean working tree (env_remove to stay scoped to `tmp`
        // when running from inside a git hook).
        let status = std::process::Command::new("git")
            .args(["status", "--porcelain"])
            .current_dir(tmp.path())
            .env_remove("GIT_DIR")
            .env_remove("GIT_WORK_TREE")
            .env_remove("GIT_INDEX_FILE")
            .env_remove("GIT_COMMON_DIR")
            .env_remove("GIT_PREFIX")
            .output()
            .unwrap();
        assert!(status.stdout.is_empty(), "expected clean working tree");
    }
}