use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader};
use tokio::process::Command;
use crate::harness::types::{ExecInvocation, ExecResult, HarnessOutputMode};
use crate::state::types::{HarnessEvent, HarnessEventKind, HarnessEventStream};
#[derive(Clone, Default)]
pub struct ExecuteOptions {
pub output_mode: HarnessOutputMode,
pub on_event: Option<Arc<dyn Fn(HarnessEvent) + Send + Sync>>,
pub step_path: Option<Vec<String>>,
pub exec_ordinal: usize,
}
fn create_harness_event(
stream: HarnessEventStream,
kind: HarnessEventKind,
raw: String,
options: &ExecuteOptions,
parsed: Option<serde_json::Value>,
) -> HarnessEvent {
HarnessEvent {
sequence: 0,
exec_ordinal: options.exec_ordinal,
stream,
kind,
raw,
parsed,
step_path: options.step_path.clone(),
boundary_index: None,
timestamp: None,
}
}
async fn read_stream<R: AsyncRead + Unpin>(
reader: R,
stream: HarnessEventStream,
options: ExecuteOptions,
) -> Result<(String, Vec<HarnessEvent>, Option<String>), String> {
let mut lines = BufReader::new(reader).lines();
let mut text = String::new();
let mut events = Vec::new();
let mut invalid_stream_json_line = None;
while let Some(line) = lines
.next_line()
.await
.map_err(|e| format!("Failed to read process output: {}", e))?
{
text.push_str(&line);
text.push('\n');
if line.trim().is_empty() {
continue;
}
let event = if options.output_mode == HarnessOutputMode::StreamJson
&& stream == HarnessEventStream::Stdout
{
if invalid_stream_json_line.is_some() {
None
} else {
match serde_json::from_str::<serde_json::Value>(&line) {
Ok(parsed) if parsed.is_object() => Some(create_harness_event(
HarnessEventStream::Stdout,
HarnessEventKind::Json,
line,
&options,
Some(parsed),
)),
_ => {
invalid_stream_json_line = Some(line);
None
}
}
}
} else {
Some(create_harness_event(
stream.clone(),
HarnessEventKind::Text,
line,
&options,
None,
))
};
if let Some(event) = event {
if let Some(callback) = &options.on_event {
callback(event.clone());
}
events.push(event);
}
}
Ok((text, events, invalid_stream_json_line))
}
pub async fn execute(
invocation: &ExecInvocation,
options: ExecuteOptions,
) -> Result<ExecResult, String> {
if invocation.command.is_empty() {
return Err("Empty command".to_string());
}
let mut cmd = Command::new(&invocation.command[0]);
cmd.args(&invocation.command[1..])
.current_dir(&invocation.cwd)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped());
if let Some(ref env) = invocation.env {
for (key, value) in env {
cmd.env(key, value);
}
}
let mut child = cmd
.spawn()
.map_err(|e| format!("Failed to spawn process: {}", e))?;
let stdout = child
.stdout
.take()
.ok_or_else(|| "Failed to capture stdout".to_string())?;
let stderr = child
.stderr
.take()
.ok_or_else(|| "Failed to capture stderr".to_string())?;
let stdout_options = options.clone();
let stderr_options = options.clone();
let stdout_task = tokio::spawn(async move {
read_stream(stdout, HarnessEventStream::Stdout, stdout_options).await
});
let stderr_task = tokio::spawn(async move {
read_stream(stderr, HarnessEventStream::Stderr, stderr_options).await
});
let status = child
.wait()
.await
.map_err(|e| format!("Failed to wait for process: {}", e))?;
let (stdout, stdout_events, invalid_stream_json_line) = stdout_task
.await
.map_err(|e| format!("Failed to join stdout reader: {}", e))??;
let (stderr, stderr_events, _) = stderr_task
.await
.map_err(|e| format!("Failed to join stderr reader: {}", e))??;
if let Some(line) = invalid_stream_json_line {
return Err(format!(
"Harness emitted invalid stream-json line: {}",
line
));
}
let mut harness_events = stdout_events;
harness_events.extend(stderr_events);
Ok(ExecResult {
exit_code: status.code().unwrap_or(-1),
stdout,
stderr,
harness_events,
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Mutex};
#[tokio::test]
async fn test_execute_echo() {
let invocation = ExecInvocation {
command: vec!["echo".to_string(), "hello".to_string()],
cwd: ".".to_string(),
env: None,
};
let result = execute(&invocation, ExecuteOptions::default())
.await
.unwrap();
assert_eq!(result.exit_code, 0);
assert_eq!(result.stdout.trim(), "hello");
}
#[tokio::test]
async fn test_execute_nonexistent_command() {
let invocation = ExecInvocation {
command: vec!["nonexistent_command_xyz".to_string()],
cwd: ".".to_string(),
env: None,
};
let result = execute(&invocation, ExecuteOptions::default()).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_execute_exit_code() {
let invocation = ExecInvocation {
command: vec!["sh".to_string(), "-c".to_string(), "exit 42".to_string()],
cwd: ".".to_string(),
env: None,
};
let result = execute(&invocation, ExecuteOptions::default())
.await
.unwrap();
assert_eq!(result.exit_code, 42);
}
#[tokio::test]
async fn test_execute_empty_command() {
let invocation = ExecInvocation {
command: vec![],
cwd: ".".to_string(),
env: None,
};
let result = execute(&invocation, ExecuteOptions::default()).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Empty command"));
}
#[tokio::test]
async fn test_execute_captures_stderr() {
let invocation = ExecInvocation {
command: vec![
"sh".to_string(),
"-c".to_string(),
"echo err >&2".to_string(),
],
cwd: ".".to_string(),
env: None,
};
let result = execute(&invocation, ExecuteOptions::default())
.await
.unwrap();
assert_eq!(result.exit_code, 0);
assert_eq!(result.stderr.trim(), "err");
}
#[tokio::test]
async fn test_execute_stream_json_stdout() {
let invocation = ExecInvocation {
command: vec![
"sh".to_string(),
"-c".to_string(),
"printf '{\"type\":\"message\"}\\n'".to_string(),
],
cwd: ".".to_string(),
env: None,
};
let events = Arc::new(Mutex::new(Vec::<HarnessEvent>::new()));
let events_for_callback = events.clone();
let result = execute(
&invocation,
ExecuteOptions {
output_mode: HarnessOutputMode::StreamJson,
on_event: Some(Arc::new(move |event| {
events_for_callback.lock().unwrap().push(event);
})),
step_path: Some(vec!["main".to_string(), "exec:test".to_string()]),
exec_ordinal: 2,
},
)
.await
.unwrap();
assert_eq!(result.exit_code, 0);
assert_eq!(result.harness_events.len(), 1);
assert_eq!(result.harness_events[0].kind, HarnessEventKind::Json);
assert_eq!(result.harness_events[0].stream, HarnessEventStream::Stdout);
assert_eq!(result.harness_events[0].exec_ordinal, 2);
assert_eq!(
result.harness_events[0].step_path.as_ref().unwrap(),
&vec!["main".to_string(), "exec:test".to_string()]
);
assert_eq!(events.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn test_execute_stream_json_invalid_line_fails() {
let invocation = ExecInvocation {
command: vec![
"sh".to_string(),
"-c".to_string(),
"printf 'not-json\\n'".to_string(),
],
cwd: ".".to_string(),
env: None,
};
let result = execute(
&invocation,
ExecuteOptions {
output_mode: HarnessOutputMode::StreamJson,
..ExecuteOptions::default()
},
)
.await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("invalid stream-json line"));
}
#[tokio::test]
async fn test_execute_stream_json_invalid_line_keeps_draining_until_exit() {
let invocation = ExecInvocation {
command: vec![
"python3".to_string(),
"-c".to_string(),
"import sys; sys.stdout.write('not-json\\n'); sys.stdout.flush(); sys.stdout.write('x' * 200000); sys.stdout.flush()".to_string(),
],
cwd: ".".to_string(),
env: None,
};
let result = tokio::time::timeout(
std::time::Duration::from_secs(2),
execute(
&invocation,
ExecuteOptions {
output_mode: HarnessOutputMode::StreamJson,
..ExecuteOptions::default()
},
),
)
.await;
assert!(
result.is_ok(),
"execute timed out instead of draining output"
);
let result = result.unwrap();
assert!(result.is_err());
assert!(result.unwrap_err().contains("invalid stream-json line"));
}
}