maestro/crates/maestro-acp/tests/spawn.rs

//! `maestro-acp` against the fake adapter in `tests/fake_acp_agent.py`, plus
//! ignored smoke tests against each maintained adapter.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use maestro_acp::schema::{ContentBlock, SessionUpdate};
use maestro_acp::{AcpAdapter, AcpError, SpawnConfig, ADAPTER_BIN, INSTALL_COMMAND};
use tempfile::TempDir;

fn fake_agent() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fake_acp_agent.py")
}

/// A temp `bin/` holding the fake agent under the adapter's name, plus a
/// log file the agent appends to. `PATH` for the spawn is just that dir.
struct Fake {
    dir: TempDir,
}

impl Fake {
    fn new() -> Self {
        let dir = TempDir::new().unwrap();
        let bin = dir.path().join("bin");
        std::fs::create_dir_all(&bin).unwrap();
        std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap();
        std::fs::copy(fake_agent(), bin.join(AcpAdapter::Codex.binary())).unwrap();
        Self { dir }
    }

    fn log_path(&self) -> PathBuf {
        self.dir.path().join("agent.log")
    }

    fn log(&self) -> String {
        std::fs::read_to_string(self.log_path()).unwrap_or_default()
    }

    /// The fake's `bin/` first, so it shadows any real adapter, then the
    /// process `PATH` so the script's `python3` still resolves.
    fn env(&self) -> HashMap<String, String> {
        let path = format!(
            "{}:{}",
            self.dir.path().join("bin").display(),
            std::env::var("PATH").unwrap_or_default()
        );
        HashMap::from([
            ("PATH".to_string(), path),
            (
                "FAKE_ACP_LOG".to_string(),
                self.log_path().to_string_lossy().into_owned(),
            ),
        ])
    }

    fn config(&self, cwd: &Path) -> SpawnConfig {
        let mut config = SpawnConfig::new(cwd, self.env());
        config.handshake_timeout = Duration::from_secs(10);
        config
    }

    async fn wait_for_log(&self, needle: &str) {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            if self.log().contains(needle) {
                return;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        panic!("log never contained {needle:?}:\n{}", self.log());
    }

    fn pid(&self) -> u32 {
        self.log()
            .lines()
            .find_map(|line| line.strip_prefix("start pid="))
            .and_then(|rest| rest.split(' ').next())
            .and_then(|pid| pid.parse().ok())
            .expect("pid in log")
    }

    async fn wait_for_exit(&self) {
        let pid = self.pid();
        let deadline = Instant::now() + Duration::from_secs(5);
        while pid_alive(pid) && Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(!pid_alive(pid), "child {pid} still alive");
    }
}

fn pid_alive(pid: u32) -> bool {
    Path::new(&format!("/proc/{pid}")).exists()
        && std::fs::read_to_string(format!("/proc/{pid}/stat"))
            .map(|s| !s.contains(") Z "))
            .unwrap_or(false)
}

#[tokio::test]
async fn spawn_runs_adapter_in_cwd_with_env_and_restrictive_caps() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake.config(cwd.path());
    config
        .env
        .insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
    config
        .env
        .insert("MAESTRO_SESSION_ID".into(), "sess-123".into());

    let session = maestro_acp::spawn(config).await.expect("spawn");
    assert!(session.session_id().starts_with("fake-"));
    assert!(pid_alive(session.pid()));

    let log = fake.log();
    let cwd_canon = cwd.path().canonicalize().unwrap();
    assert!(
        log.contains(&format!("cwd={}", cwd_canon.display())),
        "child cwd missing from log:\n{log}"
    );
    assert!(log.contains("env MAESTRO_SESSION_ID=sess-123"), "{log}");
    assert!(log.contains("recv initialize"), "{log}");
    assert!(log.contains("recv session/new"), "{log}");
    let caps = log
        .lines()
        .find_map(|l| l.strip_prefix("caps "))
        .expect("caps line");
    let caps: serde_json::Value = serde_json::from_str(caps).unwrap();
    assert_eq!(caps["fs"]["readTextFile"], false);
    assert_eq!(caps["fs"]["writeTextFile"], false);
    assert_eq!(caps["terminal"], false);

    session
        .terminate(Duration::from_millis(100), Duration::from_secs(2))
        .await;
}

#[tokio::test]
async fn missing_adapter_names_the_install_command() {
    let empty = TempDir::new().unwrap();
    let cwd = TempDir::new().unwrap();
    let env = HashMap::from([(
        "PATH".to_string(),
        empty.path().to_string_lossy().into_owned(),
    )]);
    let err = maestro_acp::spawn(SpawnConfig::new(cwd.path(), env))
        .await
        .expect_err("no adapter on PATH");
    assert!(matches!(
        err,
        AcpError::AdapterMissing {
            adapter: AcpAdapter::ClaudeCode
        }
    ));
    let msg = err.to_string();
    assert!(msg.contains(ADAPTER_BIN), "{msg}");
    assert!(msg.contains(INSTALL_COMMAND), "{msg}");
}

#[test]
fn codex_adapter_has_its_own_binary_and_install_guidance() {
    assert_eq!(AcpAdapter::Codex.binary(), "codex-acp");
    assert!(AcpAdapter::Codex
        .install_command()
        .contains("@agentclientprotocol/codex-acp"));
}

#[tokio::test]
async fn spawn_selects_the_codex_adapter() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let session = maestro_acp::spawn(fake.config(cwd.path()).adapter(AcpAdapter::Codex))
        .await
        .expect("spawn Codex adapter");
    assert!(fake.log().contains("recv session/new"));
    session
        .terminate(Duration::from_millis(100), Duration::from_secs(2))
        .await;
}

#[tokio::test]
async fn session_updates_reach_the_taken_receiver() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let session = maestro_acp::spawn(fake.config(cwd.path()))
        .await
        .expect("spawn");

    let mut updates = session.take_updates().expect("first take");
    assert!(session.take_updates().is_none(), "handed out once");
    let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
        .await
        .expect("update within 5s")
        .expect("channel open");
    let maestro_acp::AcpUpdate::Notification(first) = first else {
        panic!("expected Notification");
    };
    assert_eq!(first.session_id.0.as_ref(), session.session_id());
    match first.update {
        SessionUpdate::AgentMessageChunk(chunk) => match chunk.content {
            ContentBlock::Text(t) => assert_eq!(t.text, "hello from fake"),
            other => panic!("unexpected content {other:?}"),
        },
        other => panic!("unexpected update {other:?}"),
    }

    session
        .terminate(Duration::from_millis(100), Duration::from_secs(2))
        .await;
}

#[tokio::test]
async fn advertised_settings_are_normalized_and_mutable() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake.config(cwd.path());
    config.env.insert("FAKE_ACP_SETTINGS".into(), "1".into());
    let session = maestro_acp::spawn(config).await.expect("spawn");
    let mut updates = session.take_updates().expect("updates");

    let initial = loop {
        let update = tokio::time::timeout(Duration::from_secs(5), updates.recv())
            .await
            .expect("settings update")
            .expect("channel open");
        let maestro_acp::AcpUpdate::Notification(n) = update else {
            continue;
        };
        if let SessionUpdate::ConfigOptionUpdate(options) = n.update {
            break options.config_options;
        }
    };
    assert_eq!(initial.len(), 3);
    assert!(initial
        .iter()
        .any(|option| option.id.0.as_ref() == "effort"));
    assert!(initial
        .iter()
        .any(|option| option.id.0.as_ref() == "__maestro_legacy_model"));
    assert!(initial
        .iter()
        .any(|option| option.id.0.as_ref() == "__maestro_legacy_mode"));
    let model = initial
        .iter()
        .find(|option| option.id.0.as_ref() == "__maestro_legacy_model")
        .expect("legacy model option");
    let model = serde_json::to_value(model).unwrap();
    assert_eq!(model["options"][0]["name"], "Opus");
    assert!(model["options"]
        .as_array()
        .unwrap()
        .iter()
        .any(|option| option["value"] == "fable"));

    let changed = session
        .set_config(
            "__maestro_legacy_model".into(),
            maestro_acp::schema::SessionConfigOptionValue::value_id("sonnet"),
        )
        .expect("queue model change");
    changed
        .await
        .expect("response delivered")
        .expect("model changed");
    fake.wait_for_log("recv session/set_model").await;

    session
        .terminate(Duration::from_millis(100), Duration::from_secs(2))
        .await;
}

#[tokio::test]
async fn terminate_sends_cancel_then_sigterm() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let session = maestro_acp::spawn(fake.config(cwd.path()).adapter(AcpAdapter::Codex))
        .await
        .expect("spawn");
    let pid = session.pid();

    session
        .terminate(Duration::from_millis(200), Duration::from_secs(2))
        .await;
    assert!(session.has_exited());
    fake.wait_for_log("signal SIGTERM").await;

    let log = fake.log();
    let cancel = log.find("recv session/cancel").expect("cancel logged");
    let term = log.find("signal SIGTERM").expect("SIGTERM logged");
    assert!(cancel < term, "cancel must precede SIGTERM:\n{log}");
    assert!(!pid_alive(pid), "child {pid} still alive");
}

#[tokio::test]
async fn terminate_is_satisfied_by_an_exit_on_cancel() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake.config(cwd.path());
    config
        .env
        .insert("FAKE_ACP_EXIT_ON_CANCEL".into(), "1".into());
    let session = maestro_acp::spawn(config).await.expect("spawn");

    session
        .terminate(Duration::from_secs(2), Duration::from_secs(2))
        .await;
    assert!(session.has_exited());
    let log = fake.log();
    assert!(log.contains("recv session/cancel"), "{log}");
    assert!(!log.contains("signal SIGTERM"), "cancel was enough:\n{log}");
}

#[tokio::test]
async fn handshake_error_fails_the_spawn_and_reaps_the_child() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake.config(cwd.path()).adapter(AcpAdapter::Codex);
    config.env.insert("FAKE_ACP_INIT_FAIL".into(), "1".into());

    let err = maestro_acp::spawn(config).await.expect_err("init fails");
    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
    assert!(err.to_string().contains("fake init failure"), "{err}");
    // Closing the connection gives the fake EOF on stdin; the SIGTERM that
    // follows may find it already gone. Either way it must not survive.
    let pid: u32 = fake
        .log()
        .lines()
        .find_map(|l| l.strip_prefix("start pid="))
        .and_then(|rest| rest.split(' ').next())
        .and_then(|p| p.parse().ok())
        .expect("pid in log");
    let deadline = Instant::now() + Duration::from_secs(5);
    while pid_alive(pid) && Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(!pid_alive(pid), "child {pid} outlived the failed spawn");
}

#[tokio::test]
async fn early_exit_is_observed_after_a_successful_handshake() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake.config(cwd.path());
    config
        .env
        .insert("FAKE_ACP_EXIT_AFTER_NEW".into(), "1".into());
    let session = maestro_acp::spawn(config).await.expect("spawn");

    let status = tokio::time::timeout(Duration::from_secs(5), session.wait())
        .await
        .expect("exit within 5s")
        .expect("status");
    assert!(status.success(), "{status}");
    assert!(session.has_exited());
}

#[tokio::test]
async fn child_exit_before_handshake_reports_exited_early() {
    // A "adapter" that dies before answering anything.
    let dir = TempDir::new().unwrap();
    let bin = dir.path().join("bin");
    std::fs::create_dir_all(&bin).unwrap();
    let script = bin.join(ADAPTER_BIN);
    std::fs::write(&script, "#!/bin/sh\necho 'not logged in' >&2\nexit 3\n").unwrap();
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
    }
    let cwd = TempDir::new().unwrap();
    let env = HashMap::from([("PATH".to_string(), bin.to_string_lossy().into_owned())]);
    let mut config = SpawnConfig::new(cwd.path(), env);
    config.handshake_timeout = Duration::from_secs(10);

    let err = maestro_acp::spawn(config).await.expect_err("child died");
    match &err {
        AcpError::ExitedEarly { status, stderr, .. } => {
            assert!(status.contains('3'), "{status}");
            assert!(stderr.contains("not logged in"), "{stderr}");
        }
        other => panic!("unexpected {other:?}"),
    }
}

#[tokio::test]
async fn drop_terminates_and_reaps_the_child() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake.config(cwd.path());
    config.env.insert("FAKE_ACP_IGNORE_EOF".into(), "1".into());
    let session = maestro_acp::spawn(config).await.expect("spawn");

    drop(session);

    fake.wait_for_log("signal SIGTERM").await;
    fake.wait_for_exit().await;
}

fn chunk_text(update: &maestro_acp::AcpUpdate) -> String {
    let maestro_acp::AcpUpdate::Notification(n) = update else {
        panic!("expected Notification, got {update:?}");
    };
    match &n.update {
        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
            ContentBlock::Text(t) => t.text.clone(),
            other => panic!("unexpected content {other:?}"),
        },
        other => panic!("unexpected update {other:?}"),
    }
}

#[tokio::test]
async fn load_session_resumes_the_stored_id_and_drops_the_replay() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake
        .config(cwd.path())
        .adapter(AcpAdapter::Codex)
        .load_session("fake-old");
    config.env.insert("FAKE_ACP_LOAD_REPLAY".into(), "3".into());

    let session = maestro_acp::spawn(config).await.expect("spawn");
    assert_eq!(session.session_id(), "fake-old");
    let log = fake.log();
    assert!(
        log.contains("recv session/load sessionId=fake-old"),
        "{log}"
    );
    assert!(!log.contains("recv session/new"), "{log}");

    // The three replayed chunks never reach the receiver: the event log
    // already holds them. The first notification after the response does.
    let mut updates = session.take_updates().expect("updates");
    let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
        .await
        .expect("update within 5s")
        .expect("channel open");
    assert_eq!(chunk_text(&first), "hello after load");
    // Nothing queued behind it either.
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert!(updates.try_recv().is_err(), "replay leaked through");

    session
        .terminate(Duration::from_millis(100), Duration::from_secs(2))
        .await;
}

#[tokio::test]
async fn load_session_requires_the_load_capability() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake
        .config(cwd.path())
        .adapter(AcpAdapter::Codex)
        .load_session("fake-old");
    config.env.insert("FAKE_ACP_NO_LOAD".into(), "1".into());

    let err = maestro_acp::spawn(config)
        .await
        .expect_err("no loadSession");
    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
    assert!(err.to_string().contains("loadSession"), "{err}");
    assert!(!fake.log().contains("recv session/load"), "{}", fake.log());
}

#[tokio::test]
async fn load_session_error_fails_the_spawn() {
    let fake = Fake::new();
    let cwd = TempDir::new().unwrap();
    let mut config = fake
        .config(cwd.path())
        .adapter(AcpAdapter::Codex)
        .load_session("fake-old");
    config.env.insert("FAKE_ACP_LOAD_FAIL".into(), "1".into());

    let err = maestro_acp::spawn(config).await.expect_err("load fails");
    assert!(matches!(err, AcpError::Handshake(_)), "{err}");
    assert!(err.to_string().contains("fake load failure"), "{err}");
}

/// End to end against the installed `claude-code-acp`. Needs Claude auth,
/// takes seconds, so it is opt-in: `cargo test -p maestro-acp -- --ignored`.
#[tokio::test]
#[ignore]
async fn real_adapter_handshake() {
    let cwd = TempDir::new().unwrap();
    let session = maestro_acp::spawn(SpawnConfig::new(cwd.path(), HashMap::new()))
        .await
        .expect("real claude-code-acp handshake");
    assert!(!session.session_id().is_empty());
    assert!(pid_alive(session.pid()));
    session
        .terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
        .await;
    assert!(session.has_exited());
}

/// End to end against the maintained Codex adapter. It needs a locally
/// authenticated Codex CLI, so CI runs it only when explicitly requested.
#[tokio::test]
#[ignore]
async fn real_codex_adapter_handshake() {
    let cwd = TempDir::new().unwrap();
    let session =
        maestro_acp::spawn(SpawnConfig::new(cwd.path(), HashMap::new()).adapter(AcpAdapter::Codex))
            .await
            .expect("real codex-acp handshake");
    assert!(!session.session_id().is_empty());
    assert!(pid_alive(session.pid()));
    session
        .terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
        .await;
    assert!(session.has_exited());
}