#![cfg(feature = "integration-real-tests")]
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use meerkat_core::ops_lifecycle::OpsLifecycleRegistry;
use meerkat_core::types::SessionId;
use meerkat_runtime::RuntimeOpsLifecycleRegistry;
use meerkat_tools::builtin::BuiltinTool;
use meerkat_tools::builtin::shell::{
JobId, JobManager, JobStatus, ShellConfig, ShellError, ShellOutput, ShellTool, ShellToolSet,
};
#[cfg(unix)]
use nix::errno::Errno;
#[cfg(unix)]
use nix::sys::signal::kill;
#[cfg(unix)]
use nix::unistd::Pid;
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use tempfile::TempDir;
fn create_sh_config(temp_dir: &TempDir) -> ShellConfig {
ShellConfig {
enabled: true,
default_timeout_secs: 30,
restrict_to_project: true,
shell: "sh".to_string(),
shell_path: None,
project_root: temp_dir.path().to_path_buf(),
max_completed_jobs: 100,
completed_job_ttl_secs: 300,
max_concurrent_processes: 0, security_mode: Default::default(), security_patterns: vec![],
env_vars: std::collections::HashMap::new(),
}
}
fn create_bound_job_manager(temp_dir: &TempDir) -> JobManager {
let registry: Arc<dyn OpsLifecycleRegistry> = Arc::new(RuntimeOpsLifecycleRegistry::new());
JobManager::new(create_sh_config(temp_dir)).bind_canonical_async_ops(SessionId::new(), registry)
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_sync_execute() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": "echo 'Hello from E2E test'"
}))
.await;
assert!(result.is_ok(), "Shell command should succeed");
let output: ShellOutput = serde_json::from_value(result.unwrap().into_json().unwrap()).unwrap();
assert!(!output.timed_out, "Command should not time out");
assert_eq!(output.exit_code, Some(0), "Exit code should be 0");
assert!(
output.stdout.contains("Hello from E2E test"),
"Output should contain expected text: {}",
output.stdout
);
assert!(output.duration_secs >= 0.0, "Duration should be positive");
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_sync_timeout() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": "sleep 10",
"timeout_secs": 1
}))
.await;
assert!(
result.is_ok(),
"Timeout should return Ok with timed_out flag"
);
let output: ShellOutput = serde_json::from_value(result.unwrap().into_json().unwrap()).unwrap();
assert!(output.timed_out, "Command should have timed out");
assert!(
output.duration_secs >= 1.0,
"Duration should be at least 1 second"
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_exit_code() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": "exit 42"
}))
.await;
assert!(result.is_ok(), "Command should complete");
let output: ShellOutput = serde_json::from_value(result.unwrap().into_json().unwrap()).unwrap();
assert!(!output.timed_out, "Should not timeout");
assert_eq!(
output.exit_code,
Some(42),
"Exit code should be 42, got {:?}",
output.exit_code
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_stderr() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": "echo 'error message' 1>&2"
}))
.await;
assert!(result.is_ok(), "Command should complete");
let output: ShellOutput = serde_json::from_value(result.unwrap().into_json().unwrap()).unwrap();
assert!(!output.timed_out, "Should not timeout");
assert!(
output.stderr.contains("error message"),
"Stderr should contain error message: {}",
output.stderr
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_background_spawn() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let job_id = job_manager
.spawn_job("sleep 5", None, 60)
.await
.expect("Should spawn job");
assert!(
job_id.0.starts_with("job_"),
"Job ID should start with 'job_': {}",
job_id.0
);
let job = job_manager
.get_status(&job_id)
.await
.expect("Job should exist");
assert!(
matches!(job.status, JobStatus::Running { .. }),
"Job should be running: {:?}",
job.status
);
let _ = job_manager.cancel_job(&job_id).await;
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_background_completion() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let job_id = job_manager
.spawn_job("echo 'done'", None, 30)
.await
.expect("Should spawn job");
tokio::time::timeout(Duration::from_secs(10), async {
loop {
let status = job_manager.get_status(&job_id).await;
if status
.as_ref()
.is_some_and(|j| matches!(j.status, JobStatus::Completed { .. }))
{
break;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
})
.await
.expect("Job should complete within timeout");
let job = job_manager
.get_status(&job_id)
.await
.expect("Job should exist");
assert!(
matches!(job.status, JobStatus::Completed { .. }),
"Job should be completed"
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_background_cancel() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let job_id = job_manager
.spawn_job("sleep 60", None, 120)
.await
.expect("Should spawn job");
tokio::time::sleep(Duration::from_millis(500)).await;
let job = job_manager.get_status(&job_id).await.unwrap();
assert!(
matches!(job.status, JobStatus::Running { .. }),
"Job should be running"
);
job_manager
.cancel_job(&job_id)
.await
.expect("Cancel should succeed");
let job = job_manager.get_status(&job_id).await.unwrap();
assert!(
matches!(job.status, JobStatus::Cancelled { .. }),
"Job should be cancelled: {:?}",
job.status
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_multiple_jobs() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let id1 = job_manager
.spawn_job("sleep 30", None, 60)
.await
.expect("Should spawn job 1");
let id2 = job_manager
.spawn_job("sleep 30", None, 60)
.await
.expect("Should spawn job 2");
let id3 = job_manager
.spawn_job("sleep 30", None, 60)
.await
.expect("Should spawn job 3");
let jobs = job_manager.list_jobs().await;
assert_eq!(jobs.len(), 3, "Should have 3 jobs");
let ids: Vec<_> = jobs.iter().map(|j| &j.id).collect();
assert!(ids.contains(&&id1), "Should contain job 1");
assert!(ids.contains(&&id2), "Should contain job 2");
assert!(ids.contains(&&id3), "Should contain job 3");
let _ = job_manager.cancel_job(&id1).await;
let _ = job_manager.cancel_job(&id2).await;
let _ = job_manager.cancel_job(&id3).await;
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_not_installed() {
let temp_dir = TempDir::new().unwrap();
let config = ShellConfig {
enabled: true,
shell: "definitely_not_a_real_shell_xyz123".to_string(),
shell_path: Some(std::path::PathBuf::from("/nonexistent/path/to/shell")),
project_root: temp_dir.path().to_path_buf(),
restrict_to_project: false,
default_timeout_secs: 30,
..Default::default()
};
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": "echo test"
}))
.await;
#[cfg(unix)]
{
if let Err(err) = &result {
let err_msg = err.to_string();
assert!(
err_msg.contains("not installed")
|| err_msg.contains("Shell not installed")
|| err_msg.contains("not found"),
"Error should mention shell not found: {err_msg}"
);
}
}
#[cfg(not(unix))]
{
assert!(result.is_err(), "Should fail when shell not installed");
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("not installed") || err_msg.contains("Shell not installed"),
"Error should mention shell not installed: {}",
err_msg
);
}
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_invalid_workdir() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": "echo test",
"working_dir": "../../../etc"
}))
.await;
assert!(
result.is_err(),
"Should fail when working dir escapes project"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("outside project root")
|| err_msg.contains("escape")
|| err_msg.contains("not found")
|| err_msg.contains("Working directory"),
"Error should mention working directory issue: {err_msg}"
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_job_not_found() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let job_manager = JobManager::new(config);
let fake_id = JobId::from_string("job_nonexistent123");
let status = job_manager.get_status(&fake_id).await;
assert!(status.is_none(), "Non-existent job should return None");
let result = job_manager.cancel_job(&fake_id).await;
assert!(result.is_err(), "Cancelling non-existent job should fail");
if let Err(ShellError::JobNotFound(job_id)) = result {
assert_eq!(job_id, "job_nonexistent123");
} else {
panic!("Expected JobNotFound error, got {result:?}");
}
}
#[test]
fn test_shell_tool_name() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
assert_eq!(tool.name(), "shell");
}
#[test]
fn test_shell_tool_schema() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let def = tool.def();
assert_eq!(def.name, "shell");
assert!(!def.description.is_empty());
let schema = &def.input_schema;
assert_eq!(schema["type"], "object");
assert!(schema["properties"]["command"].is_object());
assert!(schema["properties"]["working_dir"].is_object());
assert!(schema["properties"]["timeout_secs"].is_object());
assert!(schema["properties"]["background"].is_object());
let required = schema["required"].as_array().unwrap();
assert!(required.contains(&json!("command")));
}
#[test]
fn test_shell_tool_set() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool_set = ShellToolSet::new(config);
let tools = tool_set.tools();
assert_eq!(tools.len(), 4, "Should have 4 shell tools");
let names: Vec<_> = tools.iter().map(|t| t.name()).collect();
assert!(names.contains(&"shell"), "Should have shell tool");
assert!(
names.contains(&"shell_job_status"),
"Should have shell_job_status tool"
);
assert!(names.contains(&"shell_jobs"), "Should have shell_jobs tool");
assert!(
names.contains(&"shell_job_cancel"),
"Should have shell_job_cancel tool"
);
}
#[test]
fn test_shell_disabled_by_default() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
assert!(
!tool.default_enabled(),
"Shell tool should be disabled by default for security"
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_shell_basic_sh_execution() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": "echo hello"
}))
.await;
assert!(result.is_ok(), "Basic sh execution should succeed");
let output: ShellOutput = serde_json::from_value(result.unwrap().into_json().unwrap()).unwrap();
assert!(!output.timed_out);
assert_eq!(output.exit_code, Some(0));
assert!(output.stdout.contains("hello"));
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_job_manager_basic_sh() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let job_id = job_manager
.spawn_job("echo done", None, 30)
.await
.expect("Should spawn job");
let job = job_manager.get_status(&job_id).await;
assert!(job.is_some(), "Job should exist");
tokio::time::sleep(Duration::from_millis(500)).await;
let job = job_manager.get_status(&job_id).await.unwrap();
assert!(
matches!(job.status, JobStatus::Completed { .. }),
"Job should be completed: {:?}",
job.status
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_job_status_after_completion_sh() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let job_id = job_manager
.spawn_job("echo test", None, 30)
.await
.expect("Should spawn job");
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let status = job_manager.get_status(&job_id).await;
if status
.as_ref()
.is_some_and(|j| matches!(j.status, JobStatus::Completed { .. }))
{
break;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
})
.await
.expect("Job should complete within timeout");
let job = job_manager
.get_status(&job_id)
.await
.expect("Job should exist");
assert!(
matches!(job.status, JobStatus::Completed { .. }),
"Job should be completed"
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_regression_async_execution_nonblocking() {
use std::time::Instant;
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let start = Instant::now();
let mut job_ids = Vec::new();
for _ in 0..5 {
let id = job_manager
.spawn_job("sleep 10", None, 60)
.await
.expect("Should spawn job");
job_ids.push(id);
}
let elapsed = start.elapsed();
assert!(
elapsed.as_millis() < 1000,
"Spawning 5 jobs should be nearly instant, took {elapsed:?}"
);
for id in &job_ids {
let job = job_manager.get_status(id).await.expect("Job should exist");
assert!(
matches!(job.status, JobStatus::Running { .. }),
"Job {} should be running: {:?}",
id,
job.status
);
}
for id in &job_ids {
let _ = job_manager.cancel_job(id).await;
}
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_regression_timeout_enforced() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let job_id = job_manager
.spawn_job("sleep 10", None, 1)
.await
.expect("Should spawn job");
tokio::time::sleep(Duration::from_secs(3)).await;
let job = job_manager
.get_status(&job_id)
.await
.expect("Job should exist");
assert!(
matches!(job.status, JobStatus::TimedOut { .. }),
"Job should have timed out, got {:?}",
job.status
);
if let JobStatus::TimedOut { duration_secs, .. } = &job.status {
assert!(
*duration_secs >= 1.0 && *duration_secs < 3.0,
"Duration should be close to timeout: {duration_secs}"
);
}
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
#[cfg(unix)]
async fn integration_real_regression_kill_terminates_process() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let job_id = job_manager
.spawn_job("echo $$ > pid.txt; while true; do sleep 1; done", None, 120)
.await
.expect("Should spawn job");
let pid_path = temp_dir.path().join("pid.txt");
let pid: i32 = {
let mut parsed = None;
for _ in 0..20 {
if let Ok(contents) = std::fs::read_to_string(&pid_path)
&& let Ok(value) = contents.trim().parse::<i32>()
{
parsed = Some(value);
break;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
parsed.expect("PID file should be written by the shell process")
};
let pid = Pid::from_raw(pid);
assert!(
kill(pid, None).is_ok(),
"Shell process should exist before cancellation"
);
tokio::time::sleep(Duration::from_millis(200)).await;
let job = job_manager.get_status(&job_id).await.unwrap();
assert!(
matches!(job.status, JobStatus::Running { .. }),
"Job should be running before cancel"
);
job_manager
.cancel_job(&job_id)
.await
.expect("Cancel should succeed");
let mut terminated = false;
for _ in 0..30 {
match kill(pid, None) {
Ok(()) => tokio::time::sleep(Duration::from_millis(100)).await,
Err(Errno::ESRCH) => {
terminated = true;
break;
}
Err(err) => panic!("Unexpected kill check error: {err}"),
}
}
assert!(
terminated,
"Shell process should be terminated after cancel"
);
let job = job_manager.get_status(&job_id).await.unwrap();
assert!(
matches!(job.status, JobStatus::Cancelled { .. }),
"Job should be cancelled, got {:?}",
job.status
);
tokio::time::sleep(Duration::from_millis(500)).await;
let job = job_manager.get_status(&job_id).await.unwrap();
assert!(
matches!(job.status, JobStatus::Cancelled { .. }),
"Job should still be cancelled"
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_regression_non_utf8_output() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": r"printf '\xff\xfe'"
}))
.await;
assert!(
result.is_ok(),
"Non-UTF-8 output should be handled gracefully: {result:?}"
);
let output: ShellOutput = serde_json::from_value(result.unwrap().into_json().unwrap()).unwrap();
assert!(!output.timed_out, "Should not timeout");
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_regression_truncation_keeps_tail() {
let temp_dir = TempDir::new().unwrap();
let config = create_sh_config(&temp_dir);
let tool = ShellTool::new(config);
let result = tool
.call(json!({
"command": r#"
echo "START_MARKER"
i=1
while [ "$i" -le 10000 ]; do
echo "Line $i: padding to make this longer"
i=$((i+1))
done
echo "END_MARKER"
"#
}))
.await;
assert!(result.is_ok(), "Long output command should succeed");
let output: ShellOutput = serde_json::from_value(result.unwrap().into_json().unwrap()).unwrap();
assert!(
output.stdout.contains("END_MARKER"),
"Output should contain END_MARKER (tail preserved)"
);
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_regression_concurrent_job_spawning() {
let temp_dir = TempDir::new().unwrap();
let job_manager = Arc::new(create_bound_job_manager(&temp_dir));
let mut handles = Vec::new();
for i in 0..20 {
let mgr = Arc::clone(&job_manager);
let cmd = format!("echo job{i}");
handles.push(tokio::spawn(
async move { mgr.spawn_job(&cmd, None, 30).await },
));
}
let mut job_ids = Vec::new();
for handle in handles {
let result = handle.await.unwrap();
assert!(result.is_ok(), "All spawns should succeed");
job_ids.push(result.unwrap());
}
let unique_count = {
let mut ids: Vec<_> = job_ids.iter().map(|id| &id.0).collect();
ids.sort();
ids.dedup();
ids.len()
};
assert_eq!(
unique_count, 20,
"All 20 jobs should have unique IDs, got {unique_count}"
);
tokio::time::sleep(Duration::from_millis(500)).await;
for job_id in &job_ids {
let job = job_manager
.get_status(job_id)
.await
.expect("Job should exist");
assert!(
matches!(job.status, JobStatus::Completed { .. }),
"Job {} should be completed, got {:?}",
job_id,
job.status
);
}
}
#[tokio::test]
#[ignore = "lane:e2e-system"]
async fn integration_real_regression_job_cleanup_prevents_leak() {
let temp_dir = TempDir::new().unwrap();
let job_manager = create_bound_job_manager(&temp_dir);
let mut all_ids = Vec::new();
for i in 0..50 {
let cmd = format!("echo job{i}");
let id = job_manager
.spawn_job(&cmd, None, 30)
.await
.expect("Should spawn job");
all_ids.push(id);
}
tokio::time::sleep(Duration::from_secs(1)).await;
let jobs = job_manager.list_jobs().await;
assert!(
!jobs.is_empty(),
"Should have some jobs (at least those not cleaned up yet)"
);
for job_id in all_ids.iter().take(10) {
let job = job_manager.get_status(job_id).await;
if let Some(j) = job {
assert!(
matches!(j.status, JobStatus::Completed { .. }),
"If job exists, it should be completed"
);
}
}
}