Line data Source code
1 : //! Runs a harness (Claude Code, Codex, opencode, ...) headless and reads its event stream
2 :
3 : use crate::agent::config::{HarnessAdapter, HarnessConfig};
4 : use serde_json::Value;
5 : use std::collections::BTreeMap;
6 : use std::ffi::{OsStr, OsString};
7 : use std::future::Future;
8 : use std::path::{Path, PathBuf};
9 : use std::pin::Pin;
10 : use std::process::Stdio;
11 : use std::sync::Arc;
12 : use std::time::Duration;
13 : use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
14 : use tokio::sync::{mpsc, Notify};
15 :
16 : /// What a harness reported while it ran
17 : #[derive(Clone, Debug, PartialEq, Eq)]
18 : pub enum HarnessEvent {
19 : /// A status line, e.g. the command the agent runs
20 : Status(String),
21 : /// The id of the harness session, to resume it with the next request
22 : Session(String),
23 : /// A line that the harness wrote to stdout, for the log
24 : Stdout(String),
25 : /// A line that the harness wrote to stderr, for the log
26 : Stderr(String),
27 : /// What happened to the process, for the log
28 : Process(String),
29 : }
30 :
31 : /// The outcome of a harness run
32 : #[derive(Clone, Debug, PartialEq, Eq)]
33 : pub struct HarnessOutcome {
34 : /// The final message of the agent
35 : pub message: String,
36 : /// The error of a failed run, None if it succeeded
37 : pub error: Option<String>,
38 : }
39 :
40 : impl HarnessOutcome {
41 4 : pub fn failed(error: impl Into<String>) -> Self {
42 4 : HarnessOutcome { message: String::new(), error: Some(error.into()) }
43 4 : }
44 : }
45 :
46 : /// Reads the output of a harness line by line
47 : pub trait HarnessReader: Send {
48 : fn line(&mut self, line: &str, emit: &mut dyn FnMut(HarnessEvent));
49 : /// The final message once the process exited, the error of a failed run
50 : fn finish(&self, exit_code: Option<i32>, stderr: &str) -> HarnessOutcome;
51 : }
52 :
53 : /// The model and the effort that a request picked among the ones of its harness, None for the harness's default
54 : #[derive(Clone, Debug, Default, PartialEq, Eq)]
55 : pub struct HarnessPicks {
56 : pub model: Option<String>,
57 : pub effort: Option<String>,
58 : }
59 :
60 : /// The arguments of a harness run
61 29 : pub fn build_harness_args(
62 29 : config: &HarnessConfig,
63 29 : prompt: &str,
64 29 : harness_session: Option<&str>,
65 29 : picks: &HarnessPicks,
66 29 : ) -> Vec<String> {
67 29 : let mut args = Vec::new();
68 310 : for arg in &config.command {
69 310 : match arg.as_str() {
70 310 : "{resume}" => {
71 24 : if let Some(session) = harness_session {
72 12 : args.extend(config.resume_args.iter().map(|a| a.replace("{harnessSession}", session)));
73 18 : }
74 : }
75 286 : "{model}" => {
76 24 : if let Some(model) = &picks.model {
77 8 : args.extend(config.model.args.iter().map(|a| a.replace("{model}", model)));
78 20 : }
79 : }
80 262 : "{effort}" => {
81 19 : if let Some(effort) = &picks.effort {
82 8 : args.extend(config.effort.args.iter().map(|a| a.replace("{effort}", effort)));
83 15 : }
84 : }
85 243 : "{prompt}" => args.push(prompt.to_string()),
86 219 : _ => args.push(arg.clone()),
87 : }
88 : }
89 29 : args
90 29 : }
91 :
92 : /// The first line of a text, cut to a status line
93 23 : pub fn status_line(text: &str) -> String {
94 23 : let line = text.trim().lines().next().unwrap_or_default().trim();
95 23 : if line.chars().count() > 160 {
96 1 : line.chars().take(159).chain(['…']).collect()
97 : } else {
98 22 : line.to_string()
99 : }
100 23 : }
101 :
102 34 : fn parse_event(line: &str) -> Option<Value> {
103 34 : serde_json::from_str(line).ok().filter(Value::is_object)
104 34 : }
105 :
106 : /// The message of an error that a harness reports, Codex passes the JSON body of a failed API call as message
107 3 : fn error_message(message: &str) -> String {
108 3 : let nested = parse_event(message).and_then(|body| {
109 2 : ["/error/message", "/message"].iter().find_map(|p| body.pointer(p).and_then(Value::as_str).map(str::to_string))
110 2 : });
111 3 : nested.unwrap_or_else(|| message.to_string())
112 3 : }
113 :
114 5 : fn exit_error(exit_code: Option<i32>, stderr: &str) -> String {
115 5 : let detail = status_line(stderr);
116 5 : let status = match exit_code {
117 3 : None => "was killed".to_string(),
118 2 : Some(code) => format!("exited with code {code}"),
119 : };
120 5 : if detail.is_empty() {
121 3 : format!("The harness {status}")
122 : } else {
123 2 : format!("The harness {status}: {detail}")
124 : }
125 5 : }
126 :
127 : /// The outcome of a run that reported its own error or exited
128 16 : fn outcome(message: String, error: &Option<String>, exit_code: Option<i32>, stderr: &str) -> HarnessOutcome {
129 16 : let error = match (error, exit_code) {
130 6 : (Some(error), _) => Some(error.clone()),
131 5 : (None, Some(0)) => None,
132 5 : (None, _) => Some(exit_error(exit_code, stderr)),
133 : };
134 16 : HarnessOutcome { message, error }
135 16 : }
136 :
137 : /// Reads `codex exec --json`: thread.started, item.started / item.completed, turn.completed, turn.failed, error
138 : #[derive(Default)]
139 : struct CodexReader {
140 : message: String,
141 : error: Option<String>,
142 : }
143 :
144 : impl HarnessReader for CodexReader {
145 10 : fn line(&mut self, line: &str, emit: &mut dyn FnMut(HarnessEvent)) {
146 10 : let Some(event) = parse_event(line) else { return };
147 9 : match event["type"].as_str().unwrap_or_default() {
148 9 : "thread.started" => {
149 1 : if let Some(id) = event["thread_id"].as_str() {
150 1 : emit(HarnessEvent::Session(id.into()));
151 1 : }
152 : }
153 8 : kind @ ("item.started" | "item.completed") => {
154 3 : let item = &event["item"];
155 3 : match (item["type"].as_str().unwrap_or_default(), kind) {
156 3 : ("command_execution", "item.started") => {
157 1 : if let Some(command) = item["command"].as_str() {
158 1 : emit(HarnessEvent::Status(status_line(command)));
159 1 : }
160 : }
161 1 : ("agent_message", "item.completed") => {
162 1 : if let Some(text) = item["text"].as_str() {
163 1 : self.message = text.into();
164 1 : emit(HarnessEvent::Status(status_line(text)));
165 1 : }
166 : }
167 0 : ("reasoning", _) => {
168 0 : if let Some(text) = item["text"].as_str() {
169 0 : emit(HarnessEvent::Status(status_line(text)));
170 0 : }
171 : }
172 1 : _ => {}
173 : }
174 : }
175 5 : "turn.failed" => {
176 2 : self.error =
177 2 : Some(error_message(event.pointer("/error/message").and_then(Value::as_str).unwrap_or("The turn failed")));
178 2 : }
179 3 : "error" => self.error = Some(error_message(event["message"].as_str().unwrap_or("The harness failed"))),
180 2 : _ => {}
181 : }
182 10 : }
183 :
184 5 : fn finish(&self, exit_code: Option<i32>, stderr: &str) -> HarnessOutcome {
185 5 : outcome(self.message.clone(), &self.error, exit_code, stderr)
186 5 : }
187 : }
188 :
189 : /// Reads `claude -p --output-format stream-json --verbose`: system init, assistant messages, result
190 : #[derive(Default)]
191 : struct ClaudeReader {
192 : message: String,
193 : error: Option<String>,
194 : /// The subtype of a failed result without a message, the error comes from stderr then
195 : failed: Option<String>,
196 : }
197 :
198 : impl HarnessReader for ClaudeReader {
199 7 : fn line(&mut self, line: &str, emit: &mut dyn FnMut(HarnessEvent)) {
200 7 : let Some(event) = parse_event(line) else { return };
201 7 : match event["type"].as_str().unwrap_or_default() {
202 : // Hook events carry a session id before the conversation exists, a run that fails before init leaves
203 : // no conversation to resume
204 7 : "system" if event["subtype"] == "init" => {
205 1 : if let Some(id) = event["session_id"].as_str() {
206 1 : emit(HarnessEvent::Session(id.into()));
207 1 : }
208 : }
209 4 : "assistant" => {
210 2 : for block in event.pointer("/message/content").and_then(Value::as_array).into_iter().flatten() {
211 2 : match block["type"].as_str().unwrap_or_default() {
212 2 : "text" => {
213 1 : if let Some(text) = block["text"].as_str() {
214 1 : emit(HarnessEvent::Status(status_line(text)));
215 1 : }
216 : }
217 1 : "tool_use" => {
218 1 : let command = block.pointer("/input/command").and_then(Value::as_str);
219 1 : let text = command.or(block["name"].as_str()).unwrap_or_default();
220 1 : emit(HarnessEvent::Status(status_line(text)));
221 1 : }
222 0 : _ => {}
223 : }
224 : }
225 : }
226 3 : "result" => {
227 3 : let result = event["result"].as_str();
228 3 : if let Some(result) = result {
229 1 : self.message = result.into();
230 2 : }
231 3 : let subtype = &event["subtype"];
232 3 : if event["is_error"] == true || (!subtype.is_null() && subtype != "success") {
233 0 : match result {
234 0 : Some(result) if !result.is_empty() => self.error = Some(result.into()),
235 2 : _ => self.failed = Some(subtype.as_str().unwrap_or("an error").into()),
236 : }
237 1 : }
238 : }
239 2 : _ => {}
240 : }
241 7 : }
242 :
243 3 : fn finish(&self, exit_code: Option<i32>, stderr: &str) -> HarnessOutcome {
244 : // E.g. "No conversation found with session ID: ..." on stderr for error_during_execution
245 3 : let error = self.error.clone().or_else(|| {
246 3 : let detail = status_line(stderr);
247 3 : self.failed.as_ref().map(|subtype| {
248 2 : if detail.is_empty() {
249 1 : format!("The run ended with {subtype}")
250 : } else {
251 1 : detail
252 : }
253 2 : })
254 3 : });
255 3 : outcome(self.message.clone(), &error, exit_code, stderr)
256 3 : }
257 : }
258 :
259 : /// Reads `opencode run --format json`: step_start, text, tool_use, step_finish, error.
260 : /// Every event carries the sessionID. The final message is the text of the last step without tool calls: a step
261 : /// may report its text after its tool calls.
262 : #[derive(Default)]
263 : struct OpencodeReader {
264 : texts: Vec<String>,
265 : /// The current step called a tool
266 : step_used_tools: bool,
267 : error: Option<String>,
268 : }
269 :
270 : impl HarnessReader for OpencodeReader {
271 14 : fn line(&mut self, line: &str, emit: &mut dyn FnMut(HarnessEvent)) {
272 14 : let Some(event) = parse_event(line) else { return };
273 14 : if let Some(id) = event["sessionID"].as_str().or(event.pointer("/part/sessionID").and_then(Value::as_str)) {
274 14 : emit(HarnessEvent::Session(id.into()));
275 14 : }
276 14 : match event["type"].as_str().unwrap_or_default() {
277 14 : "step_start" => {
278 3 : self.texts.clear();
279 3 : self.step_used_tools = false;
280 3 : }
281 11 : "step_finish" => {
282 4 : if self.step_used_tools || event.pointer("/part/reason").and_then(Value::as_str) == Some("tool-calls") {
283 2 : self.texts.clear();
284 2 : }
285 4 : self.step_used_tools = false;
286 : }
287 7 : "text" => {
288 4 : if let Some(text) = event.pointer("/part/text").and_then(Value::as_str).filter(|t| !t.trim().is_empty())
289 4 : {
290 4 : self.texts.push(text.into());
291 4 : emit(HarnessEvent::Status(status_line(text)));
292 4 : }
293 : }
294 3 : "tool_use" => {
295 2 : self.texts.clear();
296 2 : self.step_used_tools = true;
297 2 : let command = event.pointer("/part/state/input/command").and_then(Value::as_str);
298 2 : let text = command.or(event.pointer("/part/tool").and_then(Value::as_str)).unwrap_or_default();
299 2 : emit(HarnessEvent::Status(status_line(text)));
300 2 : }
301 1 : "error" => {
302 1 : let candidates = ["/message", "/error/data/message", "/error/message", "/error/name"];
303 2 : let error = candidates.iter().filter_map(|p| event.pointer(p)).find(|v| !v.is_null());
304 1 : self.error = Some(match error {
305 1 : Some(Value::String(s)) => s.clone(),
306 0 : Some(other) => other.to_string(),
307 0 : None => "The harness failed".into(),
308 : });
309 : }
310 0 : _ => {}
311 : }
312 14 : }
313 :
314 3 : fn finish(&self, exit_code: Option<i32>, stderr: &str) -> HarnessOutcome {
315 3 : outcome(self.texts.join("\n").trim().to_string(), &self.error, exit_code, stderr)
316 3 : }
317 : }
318 :
319 : /// Reads plain text, the output is the final message
320 : #[derive(Default)]
321 : struct TextReader {
322 : lines: Vec<String>,
323 : }
324 :
325 : impl HarnessReader for TextReader {
326 4 : fn line(&mut self, line: &str, emit: &mut dyn FnMut(HarnessEvent)) {
327 4 : self.lines.push(line.into());
328 4 : if !line.trim().is_empty() {
329 3 : emit(HarnessEvent::Status(status_line(line)));
330 3 : }
331 4 : }
332 :
333 5 : fn finish(&self, exit_code: Option<i32>, stderr: &str) -> HarnessOutcome {
334 5 : outcome(self.lines.join("\n").trim().to_string(), &None, exit_code, stderr)
335 5 : }
336 : }
337 :
338 16 : pub fn create_harness_reader(adapter: HarnessAdapter) -> Box<dyn HarnessReader> {
339 16 : match adapter {
340 5 : HarnessAdapter::Codex => Box::<CodexReader>::default(),
341 3 : HarnessAdapter::Claude => Box::<ClaudeReader>::default(),
342 3 : HarnessAdapter::Opencode => Box::<OpencodeReader>::default(),
343 5 : HarnessAdapter::Text => Box::<TextReader>::default(),
344 : }
345 16 : }
346 :
347 : /// Stops a harness run, may be called more than once
348 : pub type Kill = Arc<dyn Fn() + Send + Sync>;
349 :
350 : /// A running harness process
351 : pub struct HarnessProcess {
352 : /// Resolves once the process exited and its output is read
353 : pub outcome: Pin<Box<dyn Future<Output = HarnessOutcome> + Send>>,
354 : /// Kill the process and its children
355 : pub kill: Kill,
356 : }
357 :
358 : /// What a harness run gets
359 : pub struct HarnessSpawnArgs {
360 : pub config: HarnessConfig,
361 : pub prompt: String,
362 : pub harness_session: Option<String>,
363 : pub picks: HarnessPicks,
364 : pub cwd: PathBuf,
365 : pub env: BTreeMap<OsString, OsString>,
366 : pub events: mpsc::UnboundedSender<HarnessEvent>,
367 : }
368 :
369 : /// The stderr of a harness is kept for error messages, up to this many bytes
370 : const STDERR_LIMIT: usize = 16 * 1024;
371 : /// How long a killed harness may take to exit before it is killed for good
372 : pub const KILL_GRACE: Duration = Duration::from_secs(3);
373 :
374 : /// Start a harness run.
375 : /// The process gets its own process group, so killing the run kills the commands the agent started as well.
376 4 : pub fn spawn_harness(args: HarnessSpawnArgs) -> HarnessProcess {
377 4 : let kill = Arc::new(Notify::new());
378 4 : let run = tokio::spawn(run_harness(args, kill.clone()));
379 : HarnessProcess {
380 4 : outcome: Box::pin(async move {
381 4 : run.await.unwrap_or_else(|e| HarnessOutcome::failed(format!("The harness run failed: {e}")))
382 4 : }),
383 4 : kill: Arc::new(move || kill.notify_one()),
384 : }
385 4 : }
386 :
387 4 : async fn run_harness(args: HarnessSpawnArgs, kill: Arc<Notify>) -> HarnessOutcome {
388 4 : let argv = build_harness_args(&args.config, &args.prompt, args.harness_session.as_deref(), &args.picks);
389 4 : let Some((program, rest)) = argv.split_first() else {
390 0 : return HarnessOutcome::failed("The harness command is empty");
391 : };
392 4 : let mut command = tokio::process::Command::new(program);
393 4 : command
394 4 : .args(rest)
395 4 : .current_dir(&args.cwd)
396 4 : .env_clear()
397 4 : .envs(&args.env)
398 4 : .stdin(Stdio::null())
399 4 : .stdout(Stdio::piped())
400 4 : .stderr(Stdio::piped());
401 : #[cfg(unix)]
402 4 : command.process_group(0);
403 : #[cfg(windows)]
404 : command.creation_flags(0x08000000); // CREATE_NO_WINDOW
405 4 : let mut child = match command.spawn() {
406 3 : Ok(child) => child,
407 1 : Err(e) => return HarnessOutcome::failed(format!("Cannot start {program}: {e}")),
408 : };
409 3 : let pid = child.id();
410 3 : let events = args.events;
411 8 : let process = |text: String| {
412 8 : let _ = events.send(HarnessEvent::Process(text));
413 8 : };
414 3 : process(format!("Started {program} with pid {}", pid.map_or_else(|| "unknown".into(), |p| p.to_string())));
415 3 : let mut stderr = child.stderr.take().expect("stderr is piped");
416 3 : let stderr = tokio::spawn({
417 3 : let events = events.clone();
418 3 : async move {
419 3 : let mut kept = Vec::new();
420 3 : let mut line = Vec::new();
421 3 : let mut chunk = [0; 4096];
422 3 : let send_line = |line: &[u8]| {
423 1 : let text = String::from_utf8_lossy(line);
424 1 : let _ = events.send(HarnessEvent::Stderr(text.trim_end_matches(['\n', '\r']).to_string()));
425 1 : };
426 4 : while let Ok(n @ 1..) = stderr.read(&mut chunk).await {
427 1 : kept.extend_from_slice(&chunk[..n]);
428 1 : kept.drain(..kept.len().saturating_sub(STDERR_LIMIT));
429 1 : line.extend_from_slice(&chunk[..n]);
430 5 : while let Some(end) = line.iter().position(|b| *b == b'\n') {
431 1 : send_line(&line[..end]);
432 1 : line.drain(..=end);
433 1 : }
434 : }
435 3 : if !line.is_empty() {
436 0 : send_line(&line);
437 3 : }
438 3 : String::from_utf8_lossy(&kept).into_owned()
439 3 : }
440 : });
441 3 : let mut stdout = BufReader::new(child.stdout.take().expect("stdout is piped"));
442 3 : let mut reader = create_harness_reader(args.config.adapter);
443 3 : let stdout = tokio::spawn({
444 3 : let events = events.clone();
445 3 : async move {
446 3 : let mut line = Vec::new();
447 4 : while let Ok(1..) = stdout.read_until(b'\n', &mut line).await {
448 1 : let text = String::from_utf8_lossy(&line);
449 1 : let text = text.trim_end_matches(['\n', '\r']);
450 1 : let _ = events.send(HarnessEvent::Stdout(text.to_string()));
451 1 : reader.line(text, &mut |event| {
452 1 : let _ = events.send(event);
453 1 : });
454 1 : line.clear();
455 : }
456 3 : reader
457 3 : }
458 : });
459 3 : let status = tokio::select! {
460 3 : status = child.wait() => status,
461 3 : _ = kill.notified() => {
462 2 : process("Stopping the process group with SIGTERM".into());
463 2 : signal_process_group(pid, false);
464 2 : match tokio::time::timeout(KILL_GRACE, child.wait()).await {
465 2 : Ok(status) => status,
466 : Err(_) => {
467 0 : process(format!("Still running after {}s, killing the process group", KILL_GRACE.as_secs()));
468 0 : signal_process_group(pid, true);
469 0 : child.wait().await
470 : }
471 : }
472 : }
473 : };
474 3 : match &status {
475 3 : Ok(status) => process(format!("Exited: {status}")),
476 0 : Err(e) => process(format!("Cannot wait for the process: {e}")),
477 : }
478 3 : let exit_code = status.ok().and_then(|s| s.code());
479 3 : let stderr = stderr.await.unwrap_or_default();
480 3 : match stdout.await {
481 3 : Ok(reader) => reader.finish(exit_code, &stderr),
482 0 : Err(e) => HarnessOutcome::failed(format!("Cannot read the output of {program}: {e}")),
483 : }
484 4 : }
485 :
486 : /// Stop a process group, SIGTERM first and SIGKILL with `force`
487 : #[cfg(unix)]
488 2 : fn signal_process_group(pid: Option<u32>, force: bool) {
489 2 : let Some(pid) = pid.and_then(|p| i32::try_from(p).ok()) else { return };
490 : // SAFETY: kill has no memory effects, a group that is gone already reports ESRCH
491 : unsafe {
492 2 : libc::kill(-pid, if force { libc::SIGKILL } else { libc::SIGTERM });
493 : }
494 2 : }
495 :
496 : /// Stop a process and everything it started
497 : #[cfg(windows)]
498 : fn signal_process_group(pid: Option<u32>, _force: bool) {
499 : let Some(pid) = pid else { return };
500 : let _ = std::process::Command::new("taskkill").args(["/pid", &pid.to_string(), "/T", "/F"]).spawn();
501 : }
502 :
503 : /// Resolve the PATH of the user's login shell.
504 : /// Apps started at login or from Finder get launchd's minimal PATH without Homebrew, nvm or ~/.local/bin.
505 0 : pub async fn resolve_login_shell_path() -> Option<OsString> {
506 0 : let fallback = std::env::var_os("PATH");
507 0 : let shell = match std::env::var_os("SHELL") {
508 0 : Some(shell) if !shell.is_empty() && cfg!(unix) => shell,
509 0 : _ => return fallback,
510 : };
511 : // Interactive shells may print greetings, markers delimit the PATH
512 : const MARKER: &str = "__DASHQL_PATH__";
513 0 : let mut command = tokio::process::Command::new(shell);
514 0 : command
515 0 : .args(["-ilc", &format!("printf '{MARKER}%s{MARKER}' \"$PATH\"")])
516 0 : .stdin(Stdio::null())
517 0 : .stderr(Stdio::null())
518 0 : .kill_on_drop(true);
519 0 : match tokio::time::timeout(Duration::from_secs(5), command.output()).await {
520 0 : Ok(Ok(output)) if output.status.success() => {
521 0 : let stdout = String::from_utf8_lossy(&output.stdout);
522 0 : let path = stdout.split(MARKER).nth(1).map(str::trim).unwrap_or_default();
523 0 : if path.is_empty() {
524 0 : fallback
525 : } else {
526 0 : Some(path.into())
527 : }
528 : }
529 0 : _ => fallback,
530 : }
531 0 : }
532 :
533 14 : fn is_executable(file: &Path) -> bool {
534 14 : let Ok(metadata) = std::fs::metadata(file) else { return false };
535 : #[cfg(unix)]
536 3 : let executable = std::os::unix::fs::PermissionsExt::mode(&metadata.permissions()) & 0o111 != 0;
537 : #[cfg(not(unix))]
538 : let executable = true;
539 3 : metadata.is_file() && executable
540 14 : }
541 :
542 : /// The file that a program name resolves to on a PATH, None if there is none
543 13 : pub fn resolve_program(program: &str, path: Option<&OsStr>) -> Option<PathBuf> {
544 13 : let extensions: Vec<String> = if cfg!(windows) {
545 0 : let pathext = std::env::var("PATHEXT").unwrap_or_else(|_| ".EXE;.CMD;.BAT".into());
546 0 : std::iter::once(String::new()).chain(pathext.split(';').map(str::to_string)).collect()
547 : } else {
548 13 : vec![String::new()]
549 : };
550 14 : let find = |base: PathBuf| {
551 14 : extensions.iter().map(|ext| PathBuf::from(format!("{}{ext}", base.display()))).find(|f| is_executable(f))
552 14 : };
553 13 : if Path::new(program).is_absolute() || program.contains('/') {
554 1 : return find(PathBuf::from(program));
555 12 : }
556 13 : std::env::split_paths(path?).filter(|dir| !dir.as_os_str().is_empty()).find_map(|dir| find(dir.join(program)))
557 13 : }
558 :
559 : #[cfg(test)]
560 : mod tests {
561 : use super::*;
562 : use serde_json::json;
563 :
564 9 : fn harness(name: &str) -> HarnessConfig {
565 9 : let home = crate::agent::config::temp_dir("dashql-harness");
566 9 : let config = crate::agent::config::load_config(&home, &|_| None).unwrap();
567 9 : config.harness(name).unwrap().clone()
568 9 : }
569 :
570 : struct Read {
571 : statuses: Vec<String>,
572 : sessions: Vec<String>,
573 : outcome: HarnessOutcome,
574 : }
575 :
576 13 : fn read(adapter: HarnessAdapter, lines: &[Value], exit_code: Option<i32>, stderr: &str) -> Read {
577 13 : let mut reader = create_harness_reader(adapter);
578 13 : let (mut statuses, mut sessions) = (Vec::new(), Vec::new());
579 34 : for line in lines {
580 34 : let text = line.as_str().map_or_else(|| line.to_string(), str::to_string);
581 34 : reader.line(&text, &mut |event| match event {
582 12 : HarnessEvent::Status(s) => statuses.push(s),
583 16 : HarnessEvent::Session(s) => sessions.push(s),
584 0 : HarnessEvent::Stdout(_) | HarnessEvent::Stderr(_) | HarnessEvent::Process(_) => {}
585 28 : });
586 : }
587 13 : Read { statuses, sessions, outcome: reader.finish(exit_code, stderr) }
588 13 : }
589 :
590 5 : fn ok(message: &str) -> HarnessOutcome {
591 5 : HarnessOutcome { message: message.into(), error: None }
592 5 : }
593 :
594 : #[test]
595 1 : fn passes_the_prompt_as_one_argument() {
596 1 : assert_eq!(
597 1 : build_harness_args(&harness("codex"), "say \"hi\"; rm -rf /", None, &HarnessPicks::default()),
598 : [
599 : "codex",
600 : "exec",
601 : "--json",
602 : "--skip-git-repo-check",
603 : "--sandbox",
604 : "workspace-write",
605 : "-c",
606 : "sandbox_workspace_write.network_access=true",
607 : "--",
608 : "say \"hi\"; rm -rf /"
609 : ]
610 : );
611 : // A prompt that looks like an option stays the prompt
612 1 : assert_eq!(
613 1 : build_harness_args(&harness("opencode"), "--help", None, &HarnessPicks::default()),
614 : ["opencode", "run", "--format", "json", "--", "--help"]
615 : );
616 1 : }
617 :
618 : #[test]
619 1 : fn resumes_a_harness_session() {
620 1 : assert_eq!(
621 1 : build_harness_args(&harness("codex"), "next", Some("t-1"), &HarnessPicks::default()),
622 : [
623 : "codex",
624 : "exec",
625 : "--json",
626 : "--skip-git-repo-check",
627 : "--sandbox",
628 : "workspace-write",
629 : "-c",
630 : "sandbox_workspace_write.network_access=true",
631 : "resume",
632 : "t-1",
633 : "--",
634 : "next"
635 : ]
636 : );
637 1 : assert_eq!(
638 1 : build_harness_args(&harness("claude"), "next", Some("c-1"), &HarnessPicks::default()),
639 : [
640 : "claude",
641 : "-p",
642 : "--output-format",
643 : "stream-json",
644 : "--verbose",
645 : "--allowedTools",
646 : "Bash(dashql agent:*)",
647 : "Edit(./input/*)",
648 : "--disallowedTools",
649 : "Bash(dashql agent serve:*)",
650 : "--resume",
651 : "c-1",
652 : "--",
653 : "next"
654 : ]
655 : );
656 1 : assert_eq!(
657 1 : build_harness_args(&harness("opencode"), "next", Some("ses_1"), &HarnessPicks::default()),
658 : ["opencode", "run", "--format", "json", "--session", "ses_1", "--", "next"]
659 : );
660 1 : }
661 :
662 : #[test]
663 1 : fn passes_the_picked_model_and_effort() {
664 1 : let picks = HarnessPicks { model: Some("opus".into()), effort: Some("high".into()) };
665 1 : let args = build_harness_args(&harness("claude"), "next", Some("c-1"), &picks);
666 1 : assert_eq!(args[10..], ["--model", "opus", "--effort", "high", "--resume", "c-1", "--", "next"]);
667 1 : assert_eq!(
668 1 : build_harness_args(&harness("codex"), "next", Some("t-1"), &picks),
669 : [
670 : "codex",
671 : "exec",
672 : "--json",
673 : "--skip-git-repo-check",
674 : "--sandbox",
675 : "workspace-write",
676 : "-c",
677 : "sandbox_workspace_write.network_access=true",
678 : "-m",
679 : "opus",
680 : "-c",
681 : "model_reasoning_effort=high",
682 : "resume",
683 : "t-1",
684 : "--",
685 : "next"
686 : ]
687 : );
688 : // A harness without efforts drops the effort
689 1 : assert_eq!(
690 1 : build_harness_args(&harness("opencode"), "next", None, &picks),
691 : ["opencode", "run", "--format", "json", "--model", "opus", "--", "next"]
692 : );
693 1 : let effort = HarnessPicks { model: None, effort: Some("low".into()) };
694 1 : let args = build_harness_args(&harness("claude"), "next", None, &effort);
695 1 : assert_eq!(args[10..], ["--effort", "low", "--", "next"]);
696 1 : }
697 :
698 : #[test]
699 1 : fn reads_codex_events() {
700 1 : let read = read(
701 1 : HarnessAdapter::Codex,
702 1 : &[
703 1 : json!({ "type": "thread.started", "thread_id": "t-1" }),
704 1 : json!({ "type": "turn.started" }),
705 1 : json!({ "type": "item.started", "item": { "id": "1", "type": "command_execution", "command": "dashql agent catalog" } }),
706 1 : json!({ "type": "item.completed", "item": { "id": "1", "type": "command_execution", "command": "dashql agent catalog" } }),
707 1 : json!("not json"),
708 1 : json!({ "type": "item.completed", "item": { "id": "2", "type": "agent_message", "text": "Wrote q1.\nIt counts orders." } }),
709 1 : json!({ "type": "turn.completed", "usage": {} }),
710 1 : ],
711 1 : Some(0),
712 1 : "",
713 : );
714 1 : assert_eq!(read.sessions, ["t-1"]);
715 1 : assert_eq!(read.statuses, ["dashql agent catalog", "Wrote q1."]);
716 1 : assert_eq!(read.outcome, ok("Wrote q1.\nIt counts orders."));
717 1 : }
718 :
719 : #[test]
720 1 : fn reads_failed_codex_turns() {
721 1 : let failed = json!({ "type": "turn.failed", "error": { "message": "rate limited" } });
722 1 : assert_eq!(read(HarnessAdapter::Codex, &[failed], Some(1), "").outcome.error.unwrap(), "rate limited");
723 : // Codex passes the body of a failed API call as message
724 1 : let api = r#"{"type":"error","status":400,"error":{"type":"invalid_request_error","message":"The model is not supported"}}"#;
725 1 : let failed = json!({ "type": "turn.failed", "error": { "message": api } });
726 1 : assert_eq!(read(HarnessAdapter::Codex, &[failed], Some(1), "").outcome.error.unwrap(), "The model is not supported");
727 1 : let error = json!({ "type": "error", "message": api });
728 1 : assert_eq!(read(HarnessAdapter::Codex, &[error], Some(1), "").outcome.error.unwrap(), "The model is not supported");
729 1 : assert_eq!(
730 1 : read(HarnessAdapter::Codex, &[], Some(2), "error: unknown flag\n").outcome.error.unwrap(),
731 : "The harness exited with code 2: error: unknown flag"
732 : );
733 1 : }
734 :
735 : #[test]
736 1 : fn reads_claude_events() {
737 1 : let read = read(
738 1 : HarnessAdapter::Claude,
739 1 : &[
740 1 : // Hooks run before the conversation exists
741 1 : json!({ "type": "system", "subtype": "hook_started", "session_id": "c-0" }),
742 1 : json!({ "type": "system", "subtype": "init", "session_id": "c-1" }),
743 1 : json!({ "type": "assistant", "session_id": "c-1", "message": { "content": [
744 1 : { "type": "text", "text": "Looking at the schema" },
745 1 : { "type": "tool_use", "name": "Bash", "input": { "command": "dashql agent catalog" } },
746 1 : ] } }),
747 1 : json!({ "type": "result", "subtype": "success", "is_error": false, "result": "Done", "session_id": "c-1" }),
748 1 : ],
749 1 : Some(0),
750 1 : "",
751 : );
752 1 : assert_eq!(read.sessions, ["c-1"]);
753 1 : assert_eq!(read.statuses, ["Looking at the schema", "dashql agent catalog"]);
754 1 : assert_eq!(read.outcome, ok("Done"));
755 1 : }
756 :
757 : #[test]
758 1 : fn reads_failed_claude_runs() {
759 1 : let result = json!({ "type": "result", "subtype": "error_max_turns", "is_error": true });
760 1 : assert_eq!(
761 1 : read(HarnessAdapter::Claude, &[result], Some(1), "").outcome.error.unwrap(),
762 : "The run ended with error_max_turns"
763 : );
764 : // A run that fails before init has no session, stderr says why
765 1 : let hook = json!({ "type": "system", "subtype": "hook_response", "session_id": "c-0" });
766 1 : let result = json!({ "type": "result", "subtype": "error_during_execution", "is_error": true, "session_id": "c-0" });
767 1 : let read = read(HarnessAdapter::Claude, &[hook, result], Some(1), "No conversation found with session ID: c-0\n");
768 1 : assert!(read.sessions.is_empty());
769 1 : assert_eq!(read.outcome.error.unwrap(), "No conversation found with session ID: c-0");
770 1 : }
771 :
772 : #[test]
773 1 : fn reads_opencode_events() {
774 1 : let session = "ses_494719016ffe85dkDMj0FPRbHK";
775 1 : let read = read(
776 1 : HarnessAdapter::Opencode,
777 1 : &[
778 1 : json!({ "type": "step_start", "sessionID": session, "part": { "type": "step-start" } }),
779 1 : json!({ "type": "text", "sessionID": session, "part": { "type": "text", "text": "Looking at the schema" } }),
780 1 : json!({ "type": "tool_use", "sessionID": session, "part": { "type": "tool", "tool": "bash", "state": { "status": "completed", "input": { "command": "dashql agent catalog" } } } }),
781 1 : json!({ "type": "step_finish", "sessionID": session, "part": { "type": "step-finish", "reason": "tool-calls" } }),
782 1 : json!({ "type": "text", "sessionID": session, "part": { "type": "text", "text": "There are 42 orders." } }),
783 1 : json!({ "type": "step_finish", "sessionID": session, "part": { "type": "step-finish", "reason": "stop" } }),
784 1 : ],
785 1 : Some(0),
786 1 : "",
787 : );
788 1 : assert_eq!(read.sessions[0], session);
789 1 : assert_eq!(read.statuses, ["Looking at the schema", "dashql agent catalog", "There are 42 orders."]);
790 1 : assert_eq!(read.outcome, ok("There are 42 orders."));
791 1 : }
792 :
793 : #[test]
794 1 : fn reads_opencode_steps_that_report_text_after_tool_calls() {
795 7 : let event = |kind: &str, part: Value| json!({ "type": kind, "sessionID": "ses_1", "part": part });
796 2 : let text = |text: &str| event("text", json!({ "type": "text", "text": text }));
797 1 : let tool = event("tool_use", json!({ "type": "tool", "tool": "bash", "state": { "input": { "command": "dashql agent run q1" } } }));
798 1 : let read = read(
799 1 : HarnessAdapter::Opencode,
800 1 : &[
801 1 : event("step_start", json!({ "type": "step-start" })),
802 1 : tool,
803 1 : text("Let me run the script."),
804 1 : event("step_finish", json!({ "type": "step-finish", "reason": "tool-calls" })),
805 1 : event("step_start", json!({ "type": "step-start" })),
806 1 : text("There are 6 orders."),
807 1 : event("step_finish", json!({ "type": "step-finish", "reason": "stop" })),
808 1 : ],
809 1 : Some(0),
810 1 : "",
811 : );
812 1 : assert_eq!(read.outcome, ok("There are 6 orders."));
813 1 : }
814 :
815 : #[test]
816 1 : fn reads_opencode_errors() {
817 1 : let error = json!({ "type": "error", "sessionID": "ses_1", "error": { "name": "APIError", "data": { "message": "Rate limit exceeded" } } });
818 1 : assert_eq!(read(HarnessAdapter::Opencode, &[error], Some(1), "").outcome.error.unwrap(), "Rate limit exceeded");
819 1 : }
820 :
821 : #[test]
822 1 : fn reads_plain_text() {
823 1 : assert_eq!(read(HarnessAdapter::Text, &[json!("a"), json!(""), json!("b")], Some(0), "").outcome, ok("a\n\nb"));
824 1 : assert_eq!(read(HarnessAdapter::Text, &[], None, "").outcome.error.unwrap(), "The harness was killed");
825 1 : }
826 :
827 : #[test]
828 1 : fn cuts_status_lines() {
829 1 : assert_eq!(status_line(" first\nsecond"), "first");
830 1 : assert_eq!(status_line(&"x".repeat(200)).chars().count(), 160);
831 1 : }
832 :
833 : #[cfg(unix)]
834 : #[test]
835 1 : fn finds_executables_on_the_path() {
836 : use std::os::unix::fs::PermissionsExt;
837 1 : let dir = crate::agent::config::temp_dir("dashql-path");
838 1 : let file = dir.join("claude");
839 1 : std::fs::write(&file, "#!/bin/sh\n").unwrap();
840 1 : std::fs::set_permissions(&file, std::fs::Permissions::from_mode(0o755)).unwrap();
841 1 : std::fs::write(dir.join("codex"), "not executable").unwrap();
842 1 : let path = std::env::join_paths([Path::new("/nonexistent"), &dir]).unwrap();
843 1 : assert_eq!(resolve_program("claude", Some(&path)), Some(file.clone()));
844 1 : assert_eq!(resolve_program("codex", Some(dir.as_os_str())), None);
845 1 : assert_eq!(resolve_program(file.to_str().unwrap(), None), Some(file));
846 1 : assert_eq!(resolve_program("opencode", Some(dir.as_os_str())), None);
847 1 : }
848 :
849 : #[cfg(unix)]
850 : #[tokio::test]
851 1 : async fn runs_a_harness_and_kills_it() {
852 1 : let cwd = crate::agent::config::temp_dir("dashql-run");
853 1 : let env = std::env::vars_os().collect::<BTreeMap<_, _>>();
854 1 : let config = |command: &[&str]| HarnessConfig {
855 7 : command: command.iter().map(|s| s.to_string()).collect(),
856 3 : resume_args: Vec::new(),
857 3 : model: Default::default(),
858 3 : effort: Default::default(),
859 3 : terminal_command: None,
860 3 : adapter: HarnessAdapter::Text,
861 3 : };
862 1 : let (events, mut received) = mpsc::unbounded_channel();
863 1 : let process = spawn_harness(HarnessSpawnArgs {
864 1 : config: config(&["sh", "-c", "echo hello; echo oops >&2; exit 3"]),
865 1 : prompt: String::new(),
866 1 : harness_session: None,
867 1 : picks: HarnessPicks::default(),
868 1 : cwd: cwd.clone(),
869 1 : env: env.clone(),
870 1 : events: events.clone(),
871 1 : });
872 1 : let outcome = process.outcome.await;
873 1 : assert_eq!(outcome.message, "hello");
874 1 : assert_eq!(outcome.error.unwrap(), "The harness exited with code 3: oops");
875 1 : let mut first_run = Vec::new();
876 6 : while let Ok(event) = received.try_recv() {
877 5 : first_run.push(event);
878 5 : }
879 1 : assert!(first_run.contains(&HarnessEvent::Status("hello".into())), "{first_run:?}");
880 1 : assert!(first_run.contains(&HarnessEvent::Stdout("hello".into())), "{first_run:?}");
881 1 : assert!(first_run.contains(&HarnessEvent::Stderr("oops".into())), "{first_run:?}");
882 :
883 1 : let process = spawn_harness(HarnessSpawnArgs {
884 1 : config: config(&["sh", "-c", "sleep 30 & wait"]),
885 1 : prompt: String::new(),
886 1 : harness_session: None,
887 1 : picks: HarnessPicks::default(),
888 1 : cwd,
889 1 : env,
890 1 : events,
891 1 : });
892 1 : (process.kill)();
893 1 : let outcome = tokio::time::timeout(Duration::from_secs(10), process.outcome).await.unwrap();
894 1 : assert_eq!(outcome.error.unwrap(), "The harness was killed");
895 :
896 1 : let missing = spawn_harness(HarnessSpawnArgs {
897 1 : config: config(&["/nonexistent/harness"]),
898 1 : prompt: String::new(),
899 1 : harness_session: None,
900 1 : picks: HarnessPicks::default(),
901 1 : cwd: PathBuf::from("/"),
902 1 : env: BTreeMap::new(),
903 1 : events: mpsc::unbounded_channel().0,
904 1 : });
905 1 : assert!(missing.outcome.await.error.unwrap().starts_with("Cannot start /nonexistent/harness"));
906 1 : }
907 : }
|
