//! Binary contracts for atomic init, environment flags, signals, and the piped console. use sausage::{Host, world::World}; use serde_json::{Value, json}; use std::{ path::{Path, PathBuf}, process::Stdio, time::Duration, }; use tokio::{ io::AsyncWriteExt, process::{Child, Command}, }; use super::common::{CACHE, probe}; fn binary() -> Command { let mut command = Command::new(env!("CARGO_BIN_EXE_sausage")); command .arg("--cache-dir") .arg(CACHE) .env_remove("SAUSAGE_HEADLESS") .kill_on_drop(true); command } fn world() -> (tempfile::TempDir, PathBuf) { let folder = tempfile::tempdir().unwrap(); let world = World::create(&folder.path().join("world")).unwrap(); (folder, world.root) } async fn status(child: &mut Child) -> std::process::ExitStatus { tokio::time::timeout(Duration::from_secs(15), child.wait()) .await .expect("CLI did not exit") .unwrap() } async fn journal_exists(world: &Path) { tokio::time::timeout(Duration::from_secs(15), async { while !world.join("state/core/journal.sqlite").exists() { tokio::task::yield_now().await; } }) .await .unwrap(); } fn signal(child: &Child, signal: nix::sys::signal::Signal) { nix::sys::signal::kill( nix::unistd::Pid::from_raw(child.id().unwrap() as i32), signal, ) .unwrap(); } fn install_probe(world: &Path) { std::fs::create_dir_all(world.join("plugins/probe")).unwrap(); std::fs::copy(probe(), world.join("plugins/probe/plugin.wasm")).unwrap(); } /// A failed agent build leaves no world. Bare sausage retries and publishes a complete world. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn init_builds_agent() { let temporary = tempfile::tempdir().unwrap(); let world = temporary.path().join("world"); let configured = || { let mut command = binary(); command .current_dir(temporary.path()) .env("SAUSAGE_API", "anthropic-messages") .env("SAUSAGE_BASE_URL", "http://127.0.0.1:1/") .env("SAUSAGE_MODEL", "init-test") .env("SAUSAGE_API_TOKEN", "fake-token") .stdin(Stdio::null()); command }; let failed = configured() .arg("init") .env("RUSTC", "/nonexistent-sausage-rustc") .output() .await .unwrap(); assert!(!failed.status.success()); assert!(!world.exists()); let output = configured().output().await.unwrap(); assert!( output.status.success(), "stdout: {}\nstderr: {}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); assert!( String::from_utf8_lossy(&output.stdout).contains("first agent build takes a few minutes") ); let config: Value = serde_json::from_slice(&std::fs::read(world.join("config/agent.json")).unwrap()).unwrap(); assert_eq!( config, json!({"api":"anthropic-messages","base_url":"http://127.0.0.1:1","model":"init-test","token":"fake-token"}) ); assert!(String::from_utf8_lossy(&output.stdout).contains("sausage run")); assert_eq!( serde_json::from_slice::(&std::fs::read(world.join("config/boot.json")).unwrap()) .unwrap(), json!([]) ); for file in [ "plugins/cargo/sdk/Cargo.toml", "plugins/cargo/sdk/src/lib.rs", "plugins/agent/source/Cargo.toml", "plugins/agent/source/src/main.rs", "plugins/agent/plugin.wasm", ] { assert!(world.join(file).is_file(), "missing {file}"); } assert_eq!( std::fs::read(world.join("plugins/agent/source/Cargo.lock")).unwrap(), include_bytes!("../../../Cargo.lock") ); assert!(!world.join("state/cargo/target").exists()); assert!( Path::new(CACHE) .join("cargo-target/wasm32-wasip2/release") .is_dir() ); let host = Host::start_with_cache(&world, Path::new(CACHE)) .await .unwrap(); let doc = host.call("agent.rpc.discover", json!({})).await.unwrap(); assert_eq!(doc["x-sausage-grace"], 600); assert!( doc["methods"] .as_array() .unwrap() .iter() .any(|method| method["name"] == "chat") ); host.stop().await; } /// Both headless forms ignore EOF. False and empty values preserve the plain console. #[tokio::test] async fn headless_flag_and_environment() { for value in [None, Some("1"), Some("true")] { let (_folder, world) = world(); let mut command = binary(); command.arg("run").arg(&world).stdin(Stdio::null()); if let Some(value) = value { command.env("SAUSAGE_HEADLESS", value); } else { command.arg("--headless"); } let mut child = command.spawn().unwrap(); journal_exists(&world).await; assert!( tokio::time::timeout(Duration::from_millis(300), child.wait()) .await .is_err() ); signal(&child, nix::sys::signal::Signal::SIGTERM); assert_eq!(status(&mut child).await.code(), Some(0)); } for value in ["0", ""] { let (_folder, world) = world(); let mut child = binary() .arg("run") .arg(world) .env("SAUSAGE_HEADLESS", value) .stdin(Stdio::null()) .spawn() .unwrap(); assert_eq!(status(&mut child).await.code(), Some(0)); } } /// Signals interrupt an unanswered boot service instead of waiting forever. #[tokio::test] async fn signals_work_during_boot() { for sig in [ nix::sys::signal::Signal::SIGTERM, nix::sys::signal::Signal::SIGINT, ] { let (_folder, world) = world(); std::fs::write( world.join("config/boot.json"), json!([ {"method":"core.lock","params":{"path":"boot"}}, {"method":"core.lock","params":{"path":"boot"}} ]) .to_string(), ) .unwrap(); let mut child = binary() .arg("run") .arg(&world) .arg("--headless") .stdin(Stdio::null()) .spawn() .unwrap(); journal_exists(&world).await; tokio::time::timeout(Duration::from_secs(15), async { loop { let db = rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap(); if let Ok(2) = db.query_row( "SELECT count(*) FROM journal WHERE line='service start core.lock'", [], |row| row.get::<_, i64>(0), ) { break; } tokio::task::yield_now().await; } }) .await .unwrap(); signal(&child, sig); assert_eq!(status(&mut child).await.code(), Some(0)); } } /// Piped input supports commands, quit, and the restart exit code. #[tokio::test] async fn piped_console_and_exit_codes() { for (line, expected) in [("help\n", 0), ("quit\n", 0), ("core.exit\n", 75)] { let (_folder, world) = world(); let mut child = binary() .arg("run") .arg(world) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .spawn() .unwrap(); child .stdin .take() .unwrap() .write_all(line.as_bytes()) .await .unwrap(); let output = tokio::time::timeout(Duration::from_secs(15), child.wait_with_output()) .await .unwrap() .unwrap(); assert_eq!(output.status.code(), Some(expected)); if expected == 0 { assert!(!output.stdout.is_empty()); } } } /// A console input error still stops live actors and flushes the journal. #[tokio::test] async fn console_error_stops_host() { let (_folder, world) = world(); install_probe(&world); std::fs::write( world.join("config/boot.json"), json!([{"method":"probe.serve","params":{}}]).to_string(), ) .unwrap(); let mut child = binary() .arg("run") .arg(&world) .stdin(Stdio::piped()) .stderr(Stdio::piped()) .spawn() .unwrap(); child .stdin .take() .unwrap() .write_all(&[255, 10]) .await .unwrap(); assert_eq!(status(&mut child).await.code(), Some(1)); let db = rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap(); let stopped: i64 = db .query_row( "SELECT count(*) FROM journal WHERE plugin='probe' AND line='stop'", [], |row| row.get(0), ) .unwrap(); assert!(stopped > 0); } /// SIGINT stops only the wait, so ps can still find the live call. #[tokio::test] async fn interrupt_leaves_call_running() { let (_folder, world) = world(); install_probe(&world); let mut child = binary() .arg("run") .arg(&world) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .spawn() .unwrap(); let mut input = child.stdin.take().unwrap(); input.write_all(b"probe.busy\n").await.unwrap(); let pid = tokio::time::timeout(Duration::from_secs(20), async { loop { if world.join("state/core/journal.sqlite").exists() { let db = rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap(); if let Ok(pid) = db.query_row( "SELECT pid FROM journal WHERE plugin='probe' AND line='start busy {}'", [], |r| r.get::<_, String>(0), ) { break pid; } } tokio::task::yield_now().await; } }) .await .unwrap(); signal(&child, nix::sys::signal::Signal::SIGINT); input .write_all(format!("ps\nkill {pid}\nquit\n").as_bytes()) .await .unwrap(); drop(input); let output = tokio::time::timeout(Duration::from_secs(20), child.wait_with_output()) .await .unwrap() .unwrap(); let text = String::from_utf8(output.stdout).unwrap(); assert_eq!(output.status.code(), Some(0), "{text}"); assert!( text.contains("stopped waiting; the call continues, see ps"), "{text}" ); assert!( text.contains(&format!("{pid} probe busy console running")), "{text}" ); } /// Piped calls wait for activation, and EOF or quit preserves the last result. #[tokio::test] async fn piped_console_runs_dependent_commands_in_order() { for ending in ["", "quit\n"] { let (_folder, world) = world(); std::fs::copy(probe(), world.join("candidate.wasm")).unwrap(); std::fs::create_dir_all(world.join("state/probe")).unwrap(); std::fs::write(world.join("state/probe/discover.json"), r#"{"methods":[]}"#).unwrap(); let activate = r#"core.activate {"plugin":"notes","wasm":"candidate.wasm","force":true}"#; let call = r#"notes.echo {"ordered":true}"#; let mut child = binary() .arg("run") .arg(&world) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn() .unwrap(); child .stdin .take() .unwrap() .write_all(format!("{activate}\n{call}\n{ending}").as_bytes()) .await .unwrap(); let output = tokio::time::timeout(Duration::from_secs(20), child.wait_with_output()) .await .unwrap() .unwrap(); let stdout = String::from_utf8(output.stdout).unwrap(); assert_eq!( output.status.code(), Some(0), "{}", String::from_utf8_lossy(&output.stderr) ); assert!(!stdout.contains("error:"), "{stdout}"); assert!(stdout.contains(r#""ordered": true"#), "{stdout}"); assert!(!stdout.contains("> core.activate"), "{stdout}"); if !ending.is_empty() { assert!( stdout.find(r#""ordered": true"#).unwrap() < stdout.find("Goodbye.").unwrap(), "{stdout}" ); } } }