From b4477991249a29c6d26820f3c72b25fd260da2d4 Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 6 Oct 2026 17:57:42 -0400 Subject: [PATCH 1/3] Deepen completed Harness interpretation (#457) --- src/harness.rs | 3 + src/harness/adapter.rs | 15 +- src/harness/agy.rs | 11 +- src/harness/agy/stream.rs | 99 ++++--- src/harness/claude.rs | 11 +- src/harness/claude/stream.rs | 391 +++++++++++++++++++++++++--- src/harness/codex.rs | 11 +- src/harness/codex/stream.rs | 128 +++++---- src/harness/grok.rs | 17 +- src/harness/grok/stream.rs | 101 ++++--- src/harness/interpretation.rs | 207 +++++++++++++++ src/harness/interpretation_tests.rs | 215 +++++++++++++++ src/harness/muse.rs | 43 ++- src/harness/muse/log.rs | 108 +------- src/harness/muse/stream.rs | 70 ++--- src/harness/opencode.rs | 45 ++-- src/harness/opencode/export.rs | 138 ++-------- src/harness/opencode/stream.rs | 89 ++++--- src/progress.rs | 357 ------------------------- src/session.rs | 107 +++----- tests/background_work.rs | 22 ++ tests/opencode_sessions.rs | 105 ++++++++ tests/process_tree.rs | 12 + 23 files changed, 1341 insertions(+), 964 deletions(-) create mode 100644 src/harness/interpretation.rs create mode 100644 src/harness/interpretation_tests.rs diff --git a/src/harness.rs b/src/harness.rs index 970efb784..69de9d129 100644 --- a/src/harness.rs +++ b/src/harness.rs @@ -15,6 +15,9 @@ pub(crate) mod claude; pub(crate) mod codex; pub(crate) mod grok; mod instructions; +pub(crate) mod interpretation; +#[cfg(test)] +pub(crate) mod interpretation_tests; pub(crate) mod muse; pub(crate) mod opencode; pub(crate) mod process; diff --git a/src/harness/adapter.rs b/src/harness/adapter.rs index c9046cc31..7bcea19fb 100644 --- a/src/harness/adapter.rs +++ b/src/harness/adapter.rs @@ -5,9 +5,9 @@ use std::process::Child; use anyhow::Result; +use super::interpretation::Interpretation; use super::settings::Terminal; use super::{Choice, Harness, ModelAndEffort, Settings, agy, claude, codex, grok, muse, opencode}; -use crate::progress::Stream; /// One Harness's session protocol and its Model and Effort rules. Callers /// select it once, then use the same adapter for a session and its Resume. @@ -29,7 +29,7 @@ pub trait Adapter: Sync { outside: &mut dyn Terminal, current: &ModelAndEffort, ) -> Result>; - fn stream(&self, worktree: &Path, prompt: &str) -> Box; + fn interpretation(&self, worktree: &Path, prompt: &str) -> Interpretation; /// Stop the session and its descendants, including commands that made /// their own process group or session, with this Harness's stop signals. @@ -47,17 +47,6 @@ pub trait Adapter: Sync { fn link_instruction_fallback(&self, _worktree: &Path) -> Result<()> { Ok(()) } - - /// Optionally reconcile final text, usage and outcome with the CLI's - /// own session log or export after exit, before reporting the ending. - /// Missing or unreadable data falls back to the stream; interruption fails. - fn read_after_exit( - &self, - _worktree: &Path, - stream: Box, - ) -> Result> { - Ok(stream) - } } /// A session's arguments and optional prompt on stdin. `None` keeps stdin diff --git a/src/harness/agy.rs b/src/harness/agy.rs index a55eb6335..9a4c0507f 100644 --- a/src/harness/agy.rs +++ b/src/harness/agy.rs @@ -4,9 +4,10 @@ pub(crate) mod stream; pub use catalog::Catalog; use super::adapter::{Adapter, Invocation, SkillLoading}; +use super::interpretation::{Interpretation, Retained}; use super::settings::{Terminal, ask_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; -use crate::progress::{self, Stream}; +use crate::progress; use anyhow::Result; use std::path::Path; @@ -110,8 +111,12 @@ impl Adapter for Agy { fn link_instruction_fallback(&self, worktree: &Path) -> Result<()> { super::instructions::link_fallback(worktree, "GEMINI.md", &["AGENTS.md"]) } - fn stream(&self, _worktree: &Path, _prompt: &str) -> Box { - Box::new(stream::AgyProgress::default()) + fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation { + Interpretation::new( + self.name(), + Box::new(stream::AgyProgress::default()), + Retained::None, + ) } } diff --git a/src/harness/agy/stream.rs b/src/harness/agy/stream.rs index df6ee94d0..eee6337f8 100644 --- a/src/harness/agy/stream.rs +++ b/src/harness/agy/stream.rs @@ -1,6 +1,7 @@ //! Antigravity CLI's event envelopes. Final text, failure and cumulative //! usage come only from the final result, never concatenated text deltas. -use crate::progress::{Stream, bash, shorten}; +use crate::harness::interpretation::{Decoder, Ended, Facts, Report, TurnOutcome}; +use crate::progress::{bash, shorten}; use serde_json::Value; use std::collections::HashSet; @@ -11,7 +12,7 @@ pub struct AgyProgress { result: Option, } -impl Stream for AgyProgress { +impl Decoder for AgyProgress { fn condense(&mut self, raw: &str) -> Vec { let Ok(event) = serde_json::from_str::(raw) else { return Vec::new(); @@ -67,6 +68,31 @@ impl Stream for AgyProgress { _ => Vec::new(), } } + fn complete(self: Box) -> Facts { + let summary = self.summary(); + let result = self.result.as_ref(); + Facts { + report: Report { + warnings: Vec::new(), + summary, + }, + ended: Ended { + session_id: self.session_id.clone(), + // ACTIVE describes a tool step, not killed background work. + killed: Vec::new(), + final_message: result + .and_then(|result| result["response"].as_str()) + .map(String::from), + }, + outcome: TurnOutcome::from_failed( + result.is_some_and(|result| result["status"] == "ERROR"), + ), + diagnostic: self.error().map(String::from), + } + } +} + +impl AgyProgress { fn summary(&self) -> Option { let result = self.result.as_ref()?; let usage = &result["usage"]; @@ -80,22 +106,6 @@ impl Stream for AgyProgress { usage["total_tokens"].as_u64()? )) } - fn final_message(&self) -> Option<&str> { - self.result.as_ref()?["response"].as_str() - } - fn session_id(&self) -> Option<&str> { - self.session_id.as_deref() - } - fn killed_background_work(&self) -> Vec<&str> { - // ACTIVE describes a tool step, not background work killed at exit. - // The recorded protocol has no explicit killed-work signal. - Vec::new() - } - fn failed(&self) -> bool { - self.result - .as_ref() - .is_some_and(|result| result["status"] == "ERROR") - } fn error(&self) -> Option<&str> { self.result.as_ref()?["error"] .as_str() @@ -105,7 +115,10 @@ impl Stream for AgyProgress { #[cfg(test)] mod tests { - use super::*; + use crate::harness::{ + Harness, + interpretation_tests::{finish, stream}, + }; const INIT: &str = r#"{"event":"init","conversation_id":"recorded-conversation","init":{"cwd":"/work/widgets","permission_mode":"always-proceed"}}"#; const TOOL: &str = r#"{"event":"step_update","step_update":{"step_index":2,"state":"ACTIVE","step_type":"tool","tool_name":"run_command","tool_info":{"name":"run_command","parameters":{"CommandLine":"cargo test"}}}}"#; @@ -113,7 +126,7 @@ mod tests { #[test] fn recorded_events_give_progress_id_final_reply_and_cumulative_usage() { - let mut stream = AgyProgress::default(); + let mut stream = stream(Harness::Agy, ""); assert_eq!(stream.condense(INIT), ["session started"]); assert!(stream.condense(INIT).is_empty()); assert_eq!(stream.condense(TOOL), ["$ cargo test"]); @@ -121,45 +134,54 @@ mod tests { stream.condense(&TOOL.replace("ACTIVE", "DONE")); stream.condense(r#"{"event":"step_update","step_update":{"step_index":3,"step_type":"agent_response","text_delta":"Earlier reply."}}"#); stream.condense(RESULT); - assert_eq!(stream.session_id(), Some("recorded-conversation")); - assert_eq!(stream.final_message(), Some("Final reply only.")); + let stream = finish(stream); assert_eq!( - stream.summary().as_deref(), + stream.outcome.as_ref().unwrap().session_id.as_deref(), + Some("recorded-conversation") + ); + assert_eq!( + stream.outcome.as_ref().unwrap().final_message.as_deref(), + Some("Final reply only.") + ); + assert_eq!( + stream.report.as_ref().unwrap().summary.clone().as_deref(), Some( "2 turns, 12518 input tokens (200 cached), 29 output tokens (28 thinking), 12547 total tokens" ) ); - assert!(!stream.failed()); - assert_eq!(stream.error(), None); - assert!(stream.killed_background_work().is_empty()); + assert!(stream.outcome.is_ok()); + + assert!(stream.outcome.as_ref().unwrap().killed_work().is_empty()); } #[test] fn the_last_result_replaces_reply_usage_and_outcome() { - let mut stream = AgyProgress::default(); + let mut stream = stream(Harness::Agy, ""); stream.condense(RESULT); let error = r#"{"event":"result","result":{"status":"ERROR","error":"interrupted","response":""}}"#; assert_eq!(stream.condense(error), ["error: interrupted"]); - assert!(stream.failed()); - assert_eq!(stream.error(), Some("interrupted")); - assert_eq!(stream.final_message(), Some("")); - assert_eq!(stream.summary(), None); + let stream = finish(stream); + assert!(stream.outcome.is_err()); + assert_eq!( + stream.outcome.as_ref().unwrap_err().to_string(), + "agy's turn failed: interrupted" + ); + + assert_eq!(stream.report.as_ref().unwrap().summary.clone(), None); } #[test] fn an_active_tool_step_does_not_prove_background_work_was_killed() { - let mut stream = AgyProgress::default(); + let mut stream = stream(Harness::Agy, ""); stream.condense(TOOL); stream.condense(RESULT); - assert!(stream.killed_background_work().is_empty()); - stream.condense(r#"{"event":"result","result":{"status":"ERROR","error":"interrupted"}}"#); - assert!(stream.killed_background_work().is_empty()); + assert!(finish(stream).outcome.unwrap().killed.is_empty()); } #[test] fn unknown_and_malformed_lines_are_ignored() { - let mut stream = AgyProgress::default(); + let mut stream = stream(Harness::Agy, ""); for line in [ "not JSON", "{}", @@ -168,7 +190,8 @@ mod tests { ] { assert!(stream.condense(line).is_empty()); } - assert_eq!(stream.session_id(), None); - assert_eq!(stream.summary(), None); + let stream = finish(stream); + assert_eq!(stream.outcome.as_ref().unwrap().session_id.as_deref(), None); + assert_eq!(stream.report.as_ref().unwrap().summary.clone(), None); } } diff --git a/src/harness/claude.rs b/src/harness/claude.rs index 23e41abf5..4e7b93d07 100644 --- a/src/harness/claude.rs +++ b/src/harness/claude.rs @@ -8,9 +8,10 @@ use std::process::Command; use anyhow::{Context, Result, bail}; use super::adapter::{Adapter, Invocation, SkillLoading}; +use super::interpretation::{Interpretation, Retained}; use super::settings::{Terminal, ask_setting}; use super::{Choice, Harness, ModelAndEffort, Settings, said}; -use crate::progress::{self, Stream}; +use crate::progress; pub struct Claude; pub const REGISTRATION: (Harness, &dyn Adapter) = (Harness::Claude, &Claude); @@ -57,8 +58,12 @@ impl Adapter for Claude { ) -> Result> { ask_claude(outside, current).map(Some) } - fn stream(&self, _worktree: &Path, _prompt: &str) -> Box { - Box::new(stream::ClaudeProgress::default()) + fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation { + Interpretation::new( + self.name(), + Box::new(stream::ClaudeProgress::default()), + Retained::None, + ) } } diff --git a/src/harness/claude/stream.rs b/src/harness/claude/stream.rs index 61d836ff0..2ab7ddeb3 100644 --- a/src/harness/claude/stream.rs +++ b/src/harness/claude/stream.rs @@ -1,10 +1,11 @@ //! Claude Code's stream-json session protocol. +use crate::harness::interpretation::{Decoder, Ended, Facts, Report, TurnOutcome}; use std::collections::HashMap; use serde_json::Value; -use crate::progress::{Stream, bash, shorten}; +use crate::progress::{bash, shorten}; /// Condenses one Claude session's stream-json, a line at a time. Unknown and /// malformed lines are skipped, never an error. @@ -13,12 +14,12 @@ use crate::progress::{Stream, bash, shorten}; /// background sub-agents resumes, with another `system`/`init` event, and ends /// each resumption with another `result`. So only the first `init` gives a /// line, `result` events give none, and [`ClaudeProgress::summary`] reports the last -/// one once the process has exited. +/// one when the interpretation is finished. /// /// A background task still running when the session ends its turn for good is /// killed as the process exits, so a task killed after the last `result` is /// work the session may have been waiting on, or work it had given up on and -/// left running: see [`ClaudeProgress::killed_background_work`]. +/// left running: the completed interpretation records that evidence. #[derive(Default)] pub struct ClaudeProgress { started: bool, @@ -38,7 +39,7 @@ pub struct ClaudeProgress { final_message: Option, } -impl Stream for ClaudeProgress { +impl Decoder for ClaudeProgress { fn condense(&mut self, raw: &str) -> Vec { let Ok(event) = serde_json::from_str::(raw) else { return Vec::new(); @@ -79,6 +80,29 @@ impl Stream for ClaudeProgress { } } + fn complete(self: Box) -> Facts { + let summary = self.summary(); + Facts { + report: Report { + warnings: Vec::new(), + summary, + }, + ended: Ended { + session_id: self.session_id, + killed: self + .killed + .into_iter() + .map(|(_, description)| description) + .collect(), + final_message: self.final_message, + }, + outcome: TurnOutcome::from_failed(self.failed), + diagnostic: None, + } + } +} + +impl ClaudeProgress { /// Turns and cost from the last `result` event that reported them. On a /// claude.ai subscription the cost is the API-price equivalent, not a /// charge, and says so. @@ -92,32 +116,6 @@ impl Stream for ClaudeProgress { Some(format!("{turns} turns, ${cost:.2}{basis}")) } - /// The text of the last `result` event. None if that `result` had no - /// text, as one that is an error may not. - fn final_message(&self) -> Option<&str> { - self.final_message.as_deref() - } - - /// The session's id, from its first `init` event. - fn session_id(&self) -> Option<&str> { - self.session_id.as_deref() - } - - /// Descriptions of the background tasks killed after the last `result`. - /// The stream doesn't say whether the session was waiting on them, or - /// had abandoned them. None if that `result` was an error. - fn killed_background_work(&self) -> Vec<&str> { - if self.failed { - return Vec::new(); - } - self.killed - .iter() - .map(|(_, description)| description.as_str()) - .collect() - } -} - -impl ClaudeProgress { /// Track a background task's description and whether it was killed. fn track_task(&mut self, event: &Value) { let Some(id) = event["task_id"].as_str() else { @@ -171,3 +169,336 @@ impl ClaudeProgress { .to_string() } } + +#[cfg(test)] +mod tests { + use crate::harness::{ + Harness, + interpretation::Completion, + interpretation_tests::{finish, stream}, + }; + use serde_json::{Value, json}; + fn lines(events: &[Value]) -> (Completion, Vec>) { + let mut progress = stream(Harness::Claude, ""); + let lines = events + .iter() + .map(|event| progress.condense(&event.to_string())) + .collect(); + (finish(progress), lines) + } + + fn tool_use(name: &str, input: Value) -> Value { + json!({ + "type": "assistant", + "message": { "content": [{ "type": "tool_use", "name": name, "input": input }] } + }) + } + + fn init(cwd: &str) -> Value { + json!({ "type": "system", "subtype": "init", "cwd": cwd }) + } + + fn line_for(event: Value) -> Option { + let mut lines = lines(&[init("/work/widgets-issue-7"), event]) + .1 + .pop() + .unwrap(); + assert!(lines.len() <= 1, "several lines: {lines:?}"); + lines.pop() + } + + fn result(turns: u64, cost: f64) -> Value { + json!({ "type": "result", "subtype": "success", "num_turns": turns, "total_cost_usd": cost }) + } + + #[test] + fn only_the_first_init_starts_the_session() { + let (_, lines) = lines(&[init("/a"), init("/a")]); + assert_eq!(lines, [vec!["session started".to_string()], vec![]]); + } + + #[test] + fn a_skill_is_named() { + assert_eq!( + line_for(tool_use( + "Skill", + json!({ "skill": "thirdshift-tdd", "args": "x" }) + )), + Some("skill thirdshift-tdd".to_string()) + ); + } + + #[test] + fn commits_and_pushes_are_named() { + let line = |command: &str| line_for(tool_use("Bash", json!({ "command": command }))); + assert_eq!(line("git commit -m 'x'"), Some("commit".to_string())); + assert_eq!(line("git push -u origin issue-7"), Some("push".to_string())); + assert_eq!( + line("git add -A && git commit -m x && git push"), + Some("commit and push".to_string()) + ); + } + + #[test] + fn mentions_of_commit_or_push_are_not_commits_or_pushes() { + let line = |command: &str| line_for(tool_use("Bash", json!({ "command": command }))); + assert_eq!( + line("grep 'git push' README.md"), + Some("$ grep 'git push' README.md".to_string()) + ); + assert_eq!( + line("git commit-tree HEAD^{tree}"), + Some("$ git commit-tree HEAD^{tree}".to_string()) + ); + } + + #[test] + fn other_commands_show_their_first_line() { + assert_eq!( + line_for(tool_use( + "Bash", + json!({ "command": "cargo test\necho done" }) + )), + Some("$ cargo test".to_string()) + ); + } + + #[test] + fn file_tools_show_the_path_relative_to_the_session() { + assert_eq!( + line_for(tool_use( + "Edit", + json!({ "file_path": "/work/widgets-issue-7/src/run.rs" }) + )), + Some("Edit src/run.rs".to_string()) + ); + assert_eq!( + line_for(tool_use("Read", json!({ "file_path": "/etc/hosts" }))), + Some("Read /etc/hosts".to_string()) + ); + } + + #[test] + fn other_tools_show_their_most_telling_input() { + assert_eq!( + line_for(tool_use("Grep", json!({ "pattern": "fn main" }))), + Some("Grep fn main".to_string()) + ); + assert_eq!( + line_for(tool_use( + "Agent", + json!({ "description": "Standards review" }) + )), + Some("Agent Standards review".to_string()) + ); + assert_eq!( + line_for(tool_use("TodoWrite", json!({ "todos": [] }))), + Some("TodoWrite".to_string()) + ); + } + + #[test] + fn long_details_are_cut_short() { + let line = line_for(tool_use("Bash", json!({ "command": "é".repeat(150) }))).unwrap(); + assert_eq!(line, format!("$ {}…", "é".repeat(100))); + } + + #[test] + fn several_tool_uses_in_one_message_get_a_line_each() { + let event = json!({ + "type": "assistant", + "message": { "content": [ + { "type": "text", "text": "Reading both." }, + { "type": "tool_use", "name": "Read", "input": { "file_path": "a.rs" } }, + { "type": "tool_use", "name": "Read", "input": { "file_path": "b.rs" } } + ] } + }); + let (_, lines) = lines(&[event]); + assert_eq!( + lines, + [vec!["Read a.rs".to_string(), "Read b.rs".to_string()]] + ); + } + + #[test] + fn text_tool_results_and_results_give_no_line() { + let (_, lines) = lines(&[ + json!({ "type": "assistant", "message": { "content": [{ "type": "text", "text": "hi" }] } }), + json!({ "type": "user", "message": { "content": [{ "type": "tool_result" }] } }), + json!({ "type": "result", "num_turns": 3, "total_cost_usd": 0.1 }), + ]); + assert_eq!(lines, [vec![], vec![], vec![]] as [Vec; 3]); + } + + #[test] + fn the_summary_is_the_last_result_with_totals() { + let (progress, _) = lines(&[ + result(10, 0.5), + result(34, 1.8249), + json!({ "type": "result", "subtype": "success" }), + ]); + assert_eq!( + progress.report.as_ref().unwrap().summary.clone().as_deref(), + Some("34 turns, $1.82") + ); + } + + #[test] + fn the_final_message_is_the_text_of_the_last_result() { + let said = |text: &str| json!({ "type": "result", "subtype": "success", "result": text }); + let (progress, _) = lines(&[said("Waiting on the tests."), said("Done.")]); + assert_eq!( + progress.outcome.as_ref().unwrap().final_message.as_deref(), + Some("Done.") + ); + + let (progress, _) = lines(&[said("Done."), result(3, 0.1)]); + assert_eq!( + progress.outcome.as_ref().unwrap().final_message.as_deref(), + None + ); + } + + #[test] + fn on_a_subscription_the_cost_is_marked_as_at_api_prices() { + let (progress, _) = lines(&[ + json!({ "type": "system", "subtype": "init", "apiKeySource": "none" }), + result(34, 1.82), + ]); + assert_eq!( + progress.report.as_ref().unwrap().summary.clone().as_deref(), + Some("34 turns, $1.82 at API prices") + ); + } + + #[test] + fn no_result_means_no_summary() { + assert_eq!(lines(&[]).0.report.unwrap().summary, None); + } + + fn task_started(id: &str, description: &str) -> Value { + json!({ "type": "system", "subtype": "task_started", "task_id": id, "description": description }) + } + + fn task_updated(id: &str, status: &str) -> Value { + json!({ "type": "system", "subtype": "task_updated", "task_id": id, "patch": { "status": status } }) + } + + fn task_notification(id: &str, status: &str, summary: &str) -> Value { + json!({ "type": "system", "subtype": "task_notification", "task_id": id, "status": status, "summary": summary }) + } + + #[test] + fn a_task_killed_after_the_last_result_is_killed_background_work() { + let (progress, _) = lines(&[ + json!({ "type": "system", "subtype": "init", "session_id": "s-1" }), + task_started("b1", "./mvnw test -Dtest='GamesPageTest'"), + result(12, 0.4), + task_updated("b1", "killed"), + ]); + assert_eq!( + progress.outcome.as_ref().unwrap().session_id.as_deref(), + Some("s-1") + ); + assert_eq!( + progress.outcome.as_ref().unwrap().killed_work(), + ["./mvnw test -Dtest='GamesPageTest'"] + ); + } + + #[test] + fn a_stopped_task_notification_is_killed_background_work_once() { + let (progress, _) = lines(&[ + task_started("b1", "cargo test"), + result(12, 0.4), + task_updated("b1", "killed"), + task_notification("b1", "stopped", "cargo test"), + task_notification("b2", "stopped", "npm run build"), + ]); + assert_eq!( + progress.outcome.as_ref().unwrap().killed_work(), + ["cargo test", "npm run build"] + ); + } + + #[test] + fn no_killed_tasks_means_no_killed_background_work() { + let (progress, _) = lines(&[ + task_started("b1", "cargo test"), + task_notification("b1", "completed", "cargo test"), + result(12, 0.4), + ]); + assert!(progress.outcome.as_ref().unwrap().killed_work().is_empty()); + } + + #[test] + fn background_sub_agents_that_resumed_the_session_are_not_killed_background_work() { + let (progress, _) = lines(&[ + task_started("a1", "Standards review"), + result(12, 0.4), + task_updated("a1", "completed"), + task_notification("a1", "completed", "Standards review"), + init("/work/widgets-issue-7"), + result(20, 0.9), + ]); + assert!(progress.outcome.as_ref().unwrap().killed_work().is_empty()); + } + + #[test] + fn a_task_killed_before_a_later_result_is_not_killed_background_work() { + let (progress, _) = lines(&[ + task_started("b1", "cargo watch"), + task_updated("b1", "killed"), + result(12, 0.4), + init("/work/widgets-issue-7"), + task_notification("b1", "stopped", "cargo watch"), + result(20, 0.9), + ]); + assert!(progress.outcome.as_ref().unwrap().killed_work().is_empty()); + } + + #[test] + fn tasks_killed_after_an_error_result_are_not_killed_background_work() { + for last in [ + json!({ "type": "result", "subtype": "error_max_turns", "is_error": true }), + json!({ "type": "result", "subtype": "success", "is_error": true }), + ] { + let (progress, _) = lines(&[ + task_started("b1", "cargo test"), + last, + task_updated("b1", "killed"), + ]); + assert!(progress.outcome.is_err()); + } + } + + #[test] + fn an_init_without_a_session_id_gives_none() { + let (progress, _) = lines(&[init("/a")]); + assert_eq!( + progress.outcome.as_ref().unwrap().session_id.as_deref(), + None + ); + } + + #[test] + fn unknown_and_malformed_lines_are_skipped() { + let mut progress = stream(Harness::Claude, ""); + for raw in [ + "", + "not json", + "[1, 2]", + "42", + r#"{"type": "assistant""#, + r#"{"no_type": true}"#, + r#"{"type": 7}"#, + r#"{"type": "stream_event", "event": {}}"#, + r#"{"type": "assistant", "message": {"content": "oops"}}"#, + r#"{"type": "assistant", "message": {"content": [{"type": "tool_use"}]}}"#, + r#"{"type": "result", "num_turns": "many", "total_cost_usd": null}"#, + ] { + assert_eq!(progress.condense(raw), Vec::::new(), "{raw}"); + } + assert_eq!(finish(progress).report.unwrap().summary, None); + } +} diff --git a/src/harness/codex.rs b/src/harness/codex.rs index 962d2d1a0..551b4140d 100644 --- a/src/harness/codex.rs +++ b/src/harness/codex.rs @@ -9,9 +9,10 @@ use std::path::Path; use anyhow::Result; use super::adapter::{Adapter, Invocation, SkillLoading}; +use super::interpretation::{Interpretation, Retained}; use super::settings::{Terminal, ask_checked_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; -use crate::progress::{self, Stream}; +use crate::progress; pub struct Codex; pub const REGISTRATION: (Harness, &dyn Adapter) = (Harness::Codex, &Codex); @@ -61,8 +62,12 @@ impl Adapter for Codex { } } } - fn stream(&self, worktree: &Path, _prompt: &str) -> Box { - Box::new(stream::CodexProgress::in_worktree(worktree)) + fn interpretation(&self, worktree: &Path, _prompt: &str) -> Interpretation { + Interpretation::new( + self.name(), + Box::new(stream::CodexProgress::in_worktree(worktree)), + Retained::None, + ) } } diff --git a/src/harness/codex/stream.rs b/src/harness/codex/stream.rs index 69dddc205..d253b1300 100644 --- a/src/harness/codex/stream.rs +++ b/src/harness/codex/stream.rs @@ -5,19 +5,20 @@ //! error the session failed with. See //! `docs/research/codex-headless-harness.md`. +use crate::harness::interpretation::{Decoder, Ended, Facts, Report, TurnOutcome}; use std::collections::HashSet; use std::path::Path; use serde_json::Value; -use crate::progress::{Stream, bash, shorten}; +use crate::progress::{bash, shorten}; /// Condenses one Codex session's JSONL stream, a line at a time. /// /// `codex exec` runs one turn per process. A command or sub-agent call the /// agent left running shows as a `command_execution` or `collab_tool_call` /// item still `in_progress` when the turn completes, and it is killed as -/// Codex exits: see [`Stream::killed_background_work`]. +/// Codex exits: the completed interpretation records that evidence. #[derive(Default)] pub struct CodexProgress { /// The worktree the session runs in, which Codex's stream doesn't name. @@ -134,7 +135,7 @@ impl CodexProgress { } } -impl Stream for CodexProgress { +impl Decoder for CodexProgress { fn condense(&mut self, raw: &str) -> Vec { let Ok(event) = serde_json::from_str::(raw) else { return Vec::new(); @@ -179,6 +180,26 @@ impl Stream for CodexProgress { } } + fn complete(self: Box) -> Facts { + let summary = self.summary(); + let outcome = TurnOutcome::from_failed(self.failure.is_some()); + Facts { + report: Report { + warnings: Vec::new(), + summary, + }, + ended: Ended { + session_id: self.session_id, + killed: self.killed, + final_message: self.final_message, + }, + outcome, + diagnostic: self.failure.or(self.last_error), + } + } +} + +impl CodexProgress { /// The token totals from `turn.completed`: Codex reports no turns and no /// cost. fn summary(&self) -> Option { @@ -191,33 +212,6 @@ impl Stream for CodexProgress { "{input} input tokens ({cached} cached), {output} output tokens" )) } - - /// The text of the last `agent_message` item. - fn final_message(&self) -> Option<&str> { - self.final_message.as_deref() - } - - /// The thread id from `thread.started`, which `codex exec resume` takes. - fn session_id(&self) -> Option<&str> { - self.session_id.as_deref() - } - - /// The commands and sub-agent calls still running when the turn - /// completed. - fn killed_background_work(&self) -> Vec<&str> { - if self.failure.is_some() { - return Vec::new(); - } - self.killed.iter().map(String::as_str).collect() - } - - fn failed(&self) -> bool { - self.failure.is_some() - } - - fn error(&self) -> Option<&str> { - self.failure.as_deref().or(self.last_error.as_deref()) - } } /// The sub-agent call `item` as killed background work is described: by its @@ -265,15 +259,20 @@ fn unwrapped(command: &str) -> String { #[cfg(test)] mod tests { use super::*; + use crate::harness::{ + Harness, + interpretation::Completion, + interpretation_tests::{finish, stream}, + }; use serde_json::json; - fn lines(events: &[Value]) -> (CodexProgress, Vec>) { - let mut progress = CodexProgress::in_worktree(Path::new("/work/widgets-issue-7")); + fn lines(events: &[Value]) -> (Completion, Vec>) { + let mut progress = stream(Harness::Codex, ""); let lines = events .iter() .map(|event| progress.condense(&event.to_string())) .collect(); - (progress, lines) + (finish(progress), lines) } fn item(kind: &str, id: &str, item: Value) -> Value { @@ -304,7 +303,10 @@ mod tests { ]); assert_eq!(lines, [vec!["session started".to_string()], vec![]]); - assert_eq!(progress.session_id(), Some("019a-thread")); + assert_eq!( + progress.outcome.as_ref().unwrap().session_id.as_deref(), + Some("019a-thread") + ); } #[test] @@ -417,7 +419,10 @@ mod tests { }; let (progress, _) = lines(&[said("item_1", "Running tests."), said("item_5", "Done.")]); - assert_eq!(progress.final_message(), Some("Done.")); + assert_eq!( + progress.outcome.as_ref().unwrap().final_message.as_deref(), + Some("Done.") + ); } #[test] @@ -425,10 +430,10 @@ mod tests { let (progress, _) = lines(&[completed(12000, 345)]); assert_eq!( - progress.summary().as_deref(), + progress.report.as_ref().unwrap().summary.clone().as_deref(), Some("12000 input tokens (100 cached), 345 output tokens") ); - assert_eq!(lines(&[]).0.summary(), None); + assert_eq!(lines(&[]).0.report.unwrap().summary, None); } #[test] @@ -445,8 +450,11 @@ mod tests { completed(10, 2), ]); - assert_eq!(progress.killed_background_work(), ["sleep 188"]); - assert!(!progress.failed()); + assert_eq!( + progress.outcome.as_ref().unwrap().killed_work(), + ["sleep 188"] + ); + assert!(progress.outcome.is_ok()); } #[test] @@ -457,7 +465,10 @@ mod tests { completed(10, 2), ]); - assert_eq!(progress.killed_background_work(), ["npm run dev"]); + assert_eq!( + progress.outcome.as_ref().unwrap().killed_work(), + ["npm run dev"] + ); } #[test] @@ -484,7 +495,7 @@ mod tests { ]); assert_eq!( - progress.killed_background_work(), + progress.outcome.as_ref().unwrap().killed_work(), ["agent spawn_agent: Review the diff", "agent spawn_agent"] ); } @@ -497,28 +508,35 @@ mod tests { json!({ "type": "turn.failed", "error": { "message": "The 'GPT-6.1-Sol' model is not supported." } }), ]); - assert!(progress.failed()); + assert!(progress.outcome.is_err()); assert_eq!( - progress.error(), - Some("The 'GPT-6.1-Sol' model is not supported.") + progress.outcome.as_ref().unwrap_err().to_string(), + "codex's turn failed: The 'GPT-6.1-Sol' model is not supported." ); - assert!(progress.killed_background_work().is_empty()); } #[test] - fn an_error_event_alone_does_not_fail_the_session_but_is_its_last_error() { - let (progress, _) = lines(&[ - json!({ "type": "error", "message": "Reconnecting... 1/5" }), - completed(10, 2), - ]); - - assert!(!progress.failed()); - assert_eq!(progress.error(), Some("Reconnecting... 1/5")); + fn a_retried_diagnostic_does_not_fail_a_turn_but_explains_a_nonzero_exit() { + use std::os::unix::process::ExitStatusExt; + for code in [0, 3] { + let mut progress = stream(Harness::Codex, ""); + progress.condense(r#"{"type":"error","message":"Reconnecting... 1/5"}"#); + progress.condense(&completed(10, 2).to_string()); + let completion = progress.finish(Ok(std::process::ExitStatus::from_raw(code << 8))); + if code == 0 { + assert!(completion.outcome.is_ok()); + } else { + assert_eq!( + completion.outcome.unwrap_err().to_string(), + "codex exited 3: Reconnecting... 1/5" + ); + } + } } #[test] fn unknown_and_malformed_lines_are_skipped() { - let mut progress = CodexProgress::default(); + let mut progress = stream(Harness::Codex, ""); for raw in [ "", "not json", @@ -531,7 +549,7 @@ mod tests { let lines = progress.condense(raw); assert!(lines.len() <= 1, "{raw}: {lines:?}"); } - assert_eq!(progress.summary(), None); + assert_eq!(finish(progress).report.unwrap().summary, None); } #[test] diff --git a/src/harness/grok.rs b/src/harness/grok.rs index 9f42c54e9..0ffe56ce6 100644 --- a/src/harness/grok.rs +++ b/src/harness/grok.rs @@ -9,9 +9,9 @@ use std::path::Path; use anyhow::Result; use super::adapter::{Adapter, Invocation, SkillLoading}; +use super::interpretation::{Interpretation, Retained}; use super::settings::{Terminal, ask_checked_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; -use crate::progress::Stream; pub struct Grok; pub const REGISTRATION: (Harness, &dyn Adapter) = (Harness::Grok, &Grok); @@ -116,8 +116,12 @@ impl Adapter for Grok { )?; Ok(Some(ModelAndEffort { model, effort })) } - fn stream(&self, _worktree: &Path, _prompt: &str) -> Box { - Box::new(stream::GrokProgress::default()) + fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation { + Interpretation::new( + self.name(), + Box::new(stream::GrokProgress::default()), + Retained::None, + ) } } @@ -128,7 +132,7 @@ mod tests { #[test] fn a_resume_uses_a_streamed_id_with_the_same_flags_null_stdin_and_fixed_environment() { - let mut progress = stream::GrokProgress::default(); + let mut progress = Grok.interpretation(Path::new("/repo"), ""); progress .condense(r#"{"type":"system","subtype":"init","session_id":"abc123","cwd":"/repo"}"#); let choice = Choice { @@ -138,7 +142,10 @@ mod tests { chosen_by: ChosenBy::Command, }; let adapter = choice.harness.adapter(); - let resume = adapter.session(&choice, progress.session_id(), "Continue."); + let ended = crate::harness::interpretation_tests::finish(progress) + .outcome + .unwrap(); + let resume = adapter.session(&choice, ended.session_id.as_deref(), "Continue."); assert_eq!( resume.args, [ diff --git a/src/harness/grok/stream.rs b/src/harness/grok/stream.rs index ca3ba0c28..a1586f3c9 100644 --- a/src/harness/grok/stream.rs +++ b/src/harness/grok/stream.rs @@ -1,9 +1,10 @@ //! Grok's Messages stream shares Claude's framing, with different tool names. +use crate::harness::interpretation::{Decoder, Ended, Facts, Report, TurnOutcome}; use serde_json::Value; use crate::harness::claude::stream::ClaudeProgress; -use crate::progress::{Stream, shorten}; +use crate::progress::shorten; #[derive(Default)] pub struct GrokProgress { @@ -15,7 +16,7 @@ pub struct GrokProgress { cost_known: bool, } -impl Stream for GrokProgress { +impl Decoder for GrokProgress { fn condense(&mut self, raw: &str) -> Vec { let Ok(mut event) = serde_json::from_str::(raw) else { return Vec::new(); @@ -94,54 +95,70 @@ impl Stream for GrokProgress { } lines } - fn summary(&self) -> Option { - if self.cost_known { - self.claude.summary().or_else(|| self.usage.clone()) + fn complete(self: Box) -> Facts { + let claude = Box::new(self.claude).complete(); + let summary = if self.cost_known { + claude.report.summary.or(self.usage) } else { - self.usage.clone() + self.usage + }; + Facts { + report: Report { + warnings: Vec::new(), + summary, + }, + ended: Ended { + session_id: claude.ended.session_id.or(self.session_id), + // No Messages frame documents killed work at exit. + killed: Vec::new(), + final_message: claude.ended.final_message, + }, + outcome: TurnOutcome::from_failed(self.failed), + diagnostic: self.error, } } - fn final_message(&self) -> Option<&str> { - self.claude.final_message() - } - fn session_id(&self) -> Option<&str> { - self.claude.session_id().or(self.session_id.as_deref()) - } - fn killed_background_work(&self) -> Vec<&str> { - // Grok documents no Messages frame proving work was killed at exit. - Vec::new() - } - fn failed(&self) -> bool { - self.failed - } - fn error(&self) -> Option<&str> { - self.error.as_deref() - } } #[cfg(test)] mod tests { - use super::*; + use crate::harness::{ + Harness, + interpretation_tests::{finish, stream}, + }; #[test] fn messages_frames_reuse_claude_progress_and_keep_groks_id_final_text_and_spend() { - let mut stream = GrokProgress::default(); + let mut stream = stream(Harness::Grok, ""); assert_eq!(stream.condense(r#"{"type":"system","subtype":"init","session_id":"abc123","apiKeySource":"oauth","cwd":"/repo"}"#), ["session started"]); assert_eq!(stream.condense(r#"{"type":"assistant","message":{"content":[{"type":"tool_use","name":"read_file","input":{"path":"/repo/src/main.rs"}},{"type":"tool_use","name":"bash","input":{"command":"cargo test"}}]}}"#), ["read_file src/main.rs", "$ cargo test"]); stream.condense(r#"{"type":"result","subtype":"success","is_error":false,"num_turns":7,"result":"Here's a summary...","total_cost_usd":0.0127,"usage":{"input_tokens":812,"output_tokens":210,"cache_read_input_tokens":0,"cache_creation_input_tokens":0},"session_id":"abc123"}"#); - assert_eq!(stream.session_id(), Some("abc123")); - assert_eq!(stream.final_message(), Some("Here's a summary...")); - assert_eq!(stream.summary().as_deref(), Some("7 turns, $0.01")); - assert!(!stream.failed()); + let stream = finish(stream); + assert_eq!( + stream.outcome.as_ref().unwrap().session_id.as_deref(), + Some("abc123") + ); + assert_eq!( + stream.outcome.as_ref().unwrap().final_message.as_deref(), + Some("Here's a summary...") + ); + assert_eq!( + stream.report.as_ref().unwrap().summary.clone().as_deref(), + Some("7 turns, $0.01") + ); + assert!(stream.outcome.is_ok()); } #[test] fn usage_without_a_complete_cost_and_a_result_without_init_are_still_read() { - let mut stream = GrokProgress::default(); + let mut stream = stream(Harness::Grok, ""); stream.condense(r#"{"type":"result","subtype":"success","result":"done","num_turns":1,"total_cost_usd":0,"usage":{"input_tokens":812,"output_tokens":210,"cache_read_input_tokens":45,"cache_creation_input_tokens":12},"session_id":"abc123"}"#); - assert_eq!(stream.session_id(), Some("abc123")); + let stream = finish(stream); assert_eq!( - stream.summary().as_deref(), + stream.outcome.as_ref().unwrap().session_id.as_deref(), + Some("abc123") + ); + assert_eq!( + stream.report.as_ref().unwrap().summary.clone().as_deref(), Some( "812 input tokens, 45 cache read tokens, 12 cache creation tokens, 210 output tokens" ) @@ -160,16 +177,20 @@ mod tests { "quota exceeded", ), ] { - let mut stream = GrokProgress::default(); + let mut stream = stream(Harness::Grok, ""); assert_eq!(stream.condense(raw), [format!("error: {message}")]); - assert!(stream.failed()); - assert_eq!(stream.error(), Some(message)); + let stream = finish(stream); + assert!(stream.outcome.is_err()); + assert_eq!( + stream.outcome.as_ref().unwrap_err().to_string(), + format!("grok's turn failed: {message}") + ); } } #[test] fn malformed_and_unrelated_lines_are_ignored_and_error_frames_report_the_failure() { - let mut stream = GrokProgress::default(); + let mut stream = stream(Harness::Grok, ""); for line in [ "not json", "null", @@ -180,10 +201,12 @@ mod tests { assert!(stream.condense(line).is_empty()); } stream.condense(r#"{"type":"error","message":"backend unavailable"}"#); - assert!(stream.failed()); - assert_eq!(stream.error(), Some("backend unavailable")); stream.condense(r#"{"type":"result","subtype":"error_during_execution","is_error":true,"errors":["quota exceeded"]}"#); - assert!(stream.failed()); - assert_eq!(stream.error(), Some("quota exceeded")); + let stream = finish(stream); + assert!(stream.outcome.is_err()); + assert_eq!( + stream.outcome.as_ref().unwrap_err().to_string(), + "grok's turn failed: quota exceeded" + ); } } diff --git a/src/harness/interpretation.rs b/src/harness/interpretation.rs new file mode 100644 index 000000000..cbcdffe6b --- /dev/null +++ b/src/harness/interpretation.rs @@ -0,0 +1,207 @@ +//! Interpret a Harness session once, including optional retained records. + +use std::path::PathBuf; +use std::process::{ExitStatus, Output}; + +use anyhow::{Result, anyhow}; + +use super::{muse, opencode, said}; +use crate::interrupt; + +/// Live interpretation has only line condensation. Consume it after the +/// process owner has stopped or waited for the child and joined its workers. +pub struct Interpretation { + cli: &'static str, + decoder: Box, + retained: Retained, +} + +/// Reporting survives ordinary failures; only success supplies Resume facts. +pub struct Completion { + pub report: Option, + pub outcome: Result, +} + +pub struct Report { + pub warnings: Vec, + pub summary: Option, +} + +#[derive(Debug)] +pub struct Ended { + pub session_id: Option, + pub killed: Vec, + pub final_message: Option, +} + +impl Ended { + pub fn killed_work(&self) -> Vec<&str> { + self.killed.iter().map(String::as_str).collect() + } +} + +/// Each real protocol must supply its outcome explicitly. Streams with no +/// terminal outcome remain accepted, as before, unless execution failed. +pub(super) enum TurnOutcome { + NotFailed, + Failed, +} + +impl TurnOutcome { + pub(super) fn from_failed(failed: bool) -> Self { + if failed { + Self::Failed + } else { + Self::NotFailed + } + } +} + +pub(super) struct Facts { + pub report: Report, + pub ended: Ended, + pub outcome: TurnOutcome, + /// May be a retried diagnostic even when the turn did not fail. + pub diagnostic: Option, +} + +pub(super) trait Decoder: Send { + fn condense(&mut self, raw: &str) -> Vec; + fn complete(self: Box) -> Facts; +} + +pub(super) enum Retained { + None, + Muse(Option), + OpenCode(PathBuf), +} + +impl Retained { + fn reconcile(self, facts: &mut Facts) -> Result<()> { + match self { + Self::None => {} + Self::Muse(root) => { + if let Some(log) = root + .as_deref() + .zip(facts.ended.session_id.as_deref()) + .and_then(|(root, id)| muse::log::read(root, id)) + { + facts.report.summary = log.summary(); + if log.message.is_some() { + facts.ended.final_message = log.message; + } + } + } + Self::OpenCode(worktree) => { + if let Some(id) = facts.ended.session_id.as_deref() + && let Some(export) = opencode::export::read(&worktree, id)? + { + // A readable export is authoritative, even without text. + facts.ended.final_message = export.message; + facts.report.summary = export.summary; + if export.failure.is_some() { + facts.outcome = TurnOutcome::Failed; + facts.diagnostic = facts.diagnostic.take().or(export.failure); + } + } + } + } + Ok(()) + } +} + +enum Failure { + Execution(anyhow::Error), + Rejected { + status: ExitStatus, + diagnostic: Option, + }, +} + +impl Failure { + fn session_error(self, cli: &str) -> anyhow::Error { + match self { + Self::Execution(error) => error, + Self::Rejected { status, diagnostic } => { + let ended = if status.success() { + "'s turn failed".to_string() + } else { + format!( + " exited {}", + status + .code() + .map_or("by signal".to_string(), |code| code.to_string()) + ) + }; + match diagnostic { + Some(error) => anyhow!("{cli}{ended}: {error}"), + None => anyhow!("{cli}{ended}"), + } + } + } + } +} + +impl Interpretation { + pub(super) fn new(cli: &'static str, decoder: Box, retained: Retained) -> Self { + Self { + cli, + decoder, + retained, + } + } + + /// Unknown or malformed lines produce no progress, never an error. + pub fn condense(&mut self, raw: &str) -> Vec { + self.decoder.condense(raw) + } + + pub fn finish(self, execution: Result) -> Completion { + let cli = self.cli; + let (report, outcome) = self.complete(execution); + Completion { + report, + outcome: outcome.map_err(|failure| failure.session_error(cli)), + } + } + + /// Captured checks use exactly the same validation and retained policy, + /// but keep their own refusal context and raw-output fallback. + pub(super) fn check_output(mut self, output: &Output) -> Result<()> { + for line in String::from_utf8_lossy(&output.stdout).lines() { + self.condense(line); + } + let (_, outcome) = self.complete(Ok(output.status)); + outcome.map(drop).map_err(|failure| match failure { + Failure::Execution(error) => error, + Failure::Rejected { diagnostic, .. } => { + anyhow!(diagnostic.unwrap_or_else(|| said(output))) + } + }) + } + + fn complete(self, execution: Result) -> (Option, Result) { + if let Err(error) = interrupt::check() { + return (None, Err(Failure::Execution(error))); + } + let mut facts = self.decoder.complete(); + let recovered = self.retained.reconcile(&mut facts); + if let Err(error) = interrupt::check() { + return (None, Err(Failure::Execution(error))); + } + let outcome = execution + .and_then(|status| recovered.map(|()| status)) + .map_err(Failure::Execution) + .and_then(|status| { + if !status.success() || matches!(facts.outcome, TurnOutcome::Failed) { + Err(Failure::Rejected { + status, + diagnostic: facts.diagnostic, + }) + } else { + Ok(facts.ended) + } + }); + (Some(facts.report), outcome) + } +} diff --git a/src/harness/interpretation_tests.rs b/src/harness/interpretation_tests.rs new file mode 100644 index 000000000..9e8b3082c --- /dev/null +++ b/src/harness/interpretation_tests.rs @@ -0,0 +1,215 @@ +use super::*; +use std::os::unix::process::ExitStatusExt; + +#[test] +fn claude_completion_keeps_the_last_result_and_the_last_complete_usage() { + let mut interpretation = Harness::Claude + .adapter() + .interpretation(Path::new("/work"), ""); + for line in [ + r#"{"type":"system","subtype":"init","session_id":"s1","apiKeySource":"none"}"#, + r#"{"type":"result","subtype":"error_max_turns","is_error":true,"num_turns":34,"total_cost_usd":1.8249}"#, + r#"{"type":"result","subtype":"success","result":"Done."}"#, + ] { + interpretation.condense(line); + } + let completion = interpretation.finish(Ok(std::process::ExitStatus::from_raw(0))); + let ended = completion.outcome.unwrap(); + assert_eq!(ended.session_id.as_deref(), Some("s1")); + assert_eq!(ended.final_message.as_deref(), Some("Done.")); + assert_eq!( + completion.report.unwrap().summary.as_deref(), + Some("34 turns, $1.82 at API prices") + ); +} + +pub(crate) fn stream(harness: Harness, prompt: &str) -> interpretation::Interpretation { + // A missing working directory makes optional OpenCode acquisition fail + // locally, so in-process fallback tests never invoke a machine's CLI. + let absent = tempfile::tempdir().unwrap().path().join("absent"); + let worktree = if harness == Harness::OpenCode { + absent.as_path() + } else { + Path::new("/work/widgets-issue-7") + }; + harness.adapter().interpretation(worktree, prompt) +} + +pub(crate) fn finish(stream: interpretation::Interpretation) -> interpretation::Completion { + stream.finish(Ok(std::process::ExitStatus::from_raw(0))) +} + +#[test] +fn every_harness_accepts_missing_final_data_without_inventing_usage_or_a_session_id() { + for harness in Harness::ALL { + let completion = finish(stream(harness, "")); + let ended = completion.outcome.unwrap(); + assert_eq!(ended.session_id, None, "{harness:?}"); + assert_eq!(ended.final_message, None, "{harness:?}"); + assert!(ended.killed.is_empty(), "{harness:?}"); + assert_eq!(completion.report.unwrap().summary, None, "{harness:?}"); + } +} + +#[test] +fn claude_error_subtypes_and_error_flags_fail_without_diagnostic_text() { + for line in [ + r#"{"type":"result","subtype":"error_max_turns"}"#, + r#"{"type":"result","subtype":"success","is_error":true}"#, + ] { + let mut interpretation = stream(Harness::Claude, ""); + interpretation.condense(line); + let completion = finish(interpretation); + assert_eq!( + completion.outcome.unwrap_err().to_string(), + "claude's turn failed" + ); + assert!(completion.report.is_some()); + } +} + +// Muse's data root is process configuration. Isolate it in a subprocess so +// filesystem tests use the adapter without mutating the parallel test runner. +fn with_muse_data(test: &str, run: impl FnOnce(&Path)) { + if let Some(root) = std::env::var_os("THIRDSHIFT_TEST_MUSE_DATA") { + run(Path::new(&root)); + return; + } + let root = tempfile::tempdir().unwrap(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", &format!("harness::interpretation_tests::{test}")]) + .env("THIRDSHIFT_TEST_MUSE_DATA", root.path()) + .env("XDG_DATA_HOME", root.path()) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +fn muse_log(root: &Path, text: &str) { + let dir = root.join("muse/sessions/2026/10/06/s1"); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("session.jsonl"), text).unwrap(); +} + +fn muse_reply() -> interpretation::Interpretation { + let mut interpretation = stream(Harness::Muse, ""); + for line in [ + r#"{"stream":{"kind":"session","id":"s1"},"payload_type":"run.output.delta","payload":{"text":"stream reply"}}"#, + r#"{"stream":{"kind":"session","id":"s1"},"payload_type":"run.terminal.completed","payload":{}}"#, + ] { + interpretation.condense(line); + } + interpretation +} + +#[test] +fn muse_completion_recovers_the_last_reply_and_parent_usage_including_retained_transactions() { + with_muse_data( + "muse_completion_recovers_the_last_reply_and_parent_usage_including_retained_transactions", + |root| { + use serde_json::json; + let record = + |event| json!({"stream":{"id":"s1"},"payload":{"kind":"run","event":event}}); + let message = + record(json!({"kind":"assistant_message_committed","text":"retained reply"})); + let lines = [ + record(json!({"kind":"assistant_message_committed","text":"earlier reply"})), + record(json!({"kind":"model_completed","usage":{"input_tokens":26000,"cached_tokens":12000,"output_tokens":40}})), + json!({"stream":{"id":"child"},"payload":{"kind":"run","event":{"kind":"model_completed","usage":{"input_tokens":999999}}}}), + record(json!({"kind":"model_completed","usage":{"input_tokens":24000,"cached_tokens":20000,"output_tokens":30}})), + json!({"children":[{"record_json":message.to_string()}]}), + ].map(|line| line.to_string()).join("\n"); + muse_log(root, &lines); + let completion = finish(muse_reply()); + let ended = completion.outcome.unwrap(); + assert_eq!(ended.final_message.as_deref(), Some("retained reply")); + assert_eq!(ended.session_id.as_deref(), Some("s1")); + assert_eq!( + completion.report.unwrap().summary.as_deref(), + Some("50000 input tokens (32000 cached), 70 output tokens") + ); + }, + ); +} + +#[test] +fn muse_retained_records_without_text_preserve_the_stream_reply() { + with_muse_data( + "muse_retained_records_without_text_preserve_the_stream_reply", + |root| { + muse_log( + root, + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","usage":{"input_tokens":42,"output_tokens":7}}}}"#, + ); + let completion = finish(muse_reply()); + assert_eq!( + completion.outcome.unwrap().final_message.as_deref(), + Some("stream reply") + ); + assert_eq!( + completion.report.unwrap().summary.as_deref(), + Some("42 input tokens (0 cached), 7 output tokens") + ); + }, + ); +} + +#[test] +fn unreadable_muse_records_fall_back_without_usage() { + with_muse_data("unreadable_muse_records_fall_back_without_usage", |root| { + for text in [ + None, + Some("not JSON"), + Some("{"), + Some(r#"{"children":[{"record_json":"bad"}]}"#), + ] { + if let Some(text) = text { + muse_log(root, text); + } + let completion = finish(muse_reply()); + assert_eq!( + completion.outcome.unwrap().final_message.as_deref(), + Some("stream reply") + ); + assert_eq!(completion.report.unwrap().summary, None); + } + }); +} + +#[test] +fn a_transport_failure_preserves_recovered_usage_and_warnings_ahead_of_turn_failure() { + with_muse_data( + "a_transport_failure_preserves_recovered_usage_and_warnings_ahead_of_turn_failure", + |root| { + muse_log( + root, + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","usage":{"input_tokens":42,"output_tokens":7}}}}"#, + ); + let mut interpretation = Harness::Muse + .adapter() + .interpretation(Path::new("/work"), "/thirdshift-implement issue-url"); + interpretation.condense(r#"{"stream":{"kind":"session","id":"s1"},"payload_type":"run.terminal.failed","payload":{"reason":"quota exceeded"}}"#); + let completion = interpretation.finish(Err(anyhow::anyhow!( + "could not write the session log: Broken pipe" + ))); + assert_eq!( + completion.outcome.unwrap_err().to_string(), + "could not write the session log: Broken pipe" + ); + let report = completion.report.unwrap(); + assert_eq!( + report.warnings, + ["warning: the session never loaded thirdshift-implement with its skill tool"] + ); + assert_eq!( + report.summary.as_deref(), + Some("42 input tokens (0 cached), 7 output tokens") + ); + }, + ); +} diff --git a/src/harness/muse.rs b/src/harness/muse.rs index bc4c0f1bd..9b1325950 100644 --- a/src/harness/muse.rs +++ b/src/harness/muse.rs @@ -1,18 +1,18 @@ //! Muse Code: unattended sessions, skill-tool loading and the read after exit. mod catalog; -mod log; +pub(super) mod log; mod stream; use std::path::{Path, PathBuf}; use std::process::Command; -use anyhow::{Context, Result, bail}; +use anyhow::{Context, Result}; use super::adapter::{Adapter, Invocation, SkillLoading}; +use super::interpretation::{Interpretation, Retained}; use super::settings::{Terminal, ask_setting}; -use super::{Choice, Harness, ModelAndEffort, Settings, said}; -use crate::progress::Stream; +use super::{Choice, Harness, ModelAndEffort, Settings}; pub struct Muse; pub const REGISTRATION: (Harness, &dyn Adapter) = (Harness::Muse, &Muse); @@ -96,15 +96,12 @@ impl Adapter for Muse { } } } - fn stream(&self, _worktree: &Path, prompt: &str) -> Box { - Box::new(stream::MuseProgress::for_prompt(prompt)) - } - fn read_after_exit( - &self, - _worktree: &Path, - stream: Box, - ) -> Result> { - Ok(log::read_after_exit(stream)) + fn interpretation(&self, _worktree: &Path, prompt: &str) -> Interpretation { + Interpretation::new( + self.name(), + Box::new(stream::MuseProgress::for_prompt(prompt)), + Retained::Muse(data_dir()), + ) } } @@ -144,19 +141,13 @@ pub fn check_model_and_effort(chosen: &ModelAndEffort) -> Result None, ) .context("could not run muse to check the Model")?; - let mut stream = Muse.stream(Path::new("."), "Reply with OK."); - for line in String::from_utf8_lossy(&output.stdout).lines() { - stream.condense(line); - } - if !output.status.success() || stream.failed() { - bail!( - "muse refused a test call on the Model {model}: {}", - stream - .error() - .map(String::from) - .unwrap_or_else(|| said(&output)) - ); - } + Interpretation::new( + Muse.name(), + Box::new(stream::MuseProgress::for_prompt("Reply with OK.")), + Retained::None, + ) + .check_output(&output) + .with_context(|| format!("muse refused a test call on the Model {model}"))?; } } Ok(settled) diff --git a/src/harness/muse/log.rs b/src/harness/muse/log.rs index 7c328ea84..46c003228 100644 --- a/src/harness/muse/log.rs +++ b/src/harness/muse/log.rs @@ -1,14 +1,13 @@ //! Read Muse's retained session records for the last reply and token usage. -use crate::progress::Stream; use anyhow::Result; use serde_json::Value; use std::fs; use std::path::{Path, PathBuf}; #[derive(Default)] -pub struct SessionLog { - message: Option, +pub(in crate::harness) struct SessionLog { + pub(in crate::harness) message: Option, tokens: Option, } @@ -66,7 +65,7 @@ impl SessionLog { _ => {} } } - fn summary(&self) -> Option { + pub(in crate::harness) fn summary(&self) -> Option { self.tokens.as_ref().map(|tokens| { format!( "{} input tokens ({} cached), {} output tokens", @@ -92,101 +91,10 @@ fn find_log(root: &Path, id: &str, depth: usize) -> Option { .find_map(|entry| find_log(&entry.path(), id, depth - 1)) } -pub fn read_after_exit(stream: Box) -> Box { - let log = stream - .session_id() - .filter(|id| !id.contains(['/', '\\']) && *id != "..") - .and_then(|id| { - let dir = super::data_dir()?.join("sessions"); - let path = find_log(&dir, id, 3)?; - SessionLog::parse(&fs::read_to_string(path).ok()?, id).ok() - }); - Box::new(AfterExit { stream, log }) -} - -struct AfterExit { - stream: Box, - log: Option, -} - -impl Stream for AfterExit { - fn condense(&mut self, raw: &str) -> Vec { - self.stream.condense(raw) - } - fn summary(&self) -> Option { - self.log.as_ref().and_then(SessionLog::summary) - } - fn final_message(&self) -> Option<&str> { - self.log - .as_ref() - .and_then(|log| log.message.as_deref()) - .or_else(|| self.stream.final_message()) - } - fn session_id(&self) -> Option<&str> { - self.stream.session_id() - } - fn killed_background_work(&self) -> Vec<&str> { - self.stream.killed_background_work() - } - fn failed(&self) -> bool { - self.stream.failed() - } - fn error(&self) -> Option<&str> { - self.stream.error() - } - fn warnings(&self) -> Vec { - self.stream.warnings() - } -} - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - // Retained records observed in Muse 1.4.2, trimmed to the fields read. - fn record(event: Value) -> Value { - json!({"stream":{"kind":"session","id":"s1"},"payload_type":"runtime.session", - "payload":{"kind":"run","event":event}}) - } - - #[test] - fn the_last_reply_is_kept_and_usage_is_summed_without_child_session_usage() { - let lines = [ - record(json!({"kind":"assistant_message_committed","text":"first reply"})), - record(json!({"kind":"model_completed","usage":{"input_tokens":26000,"cached_tokens":12000,"output_tokens":40}})), - json!({"stream":{"id":"child"},"payload":{"kind":"run","event":{"kind":"model_completed","usage":{"input_tokens":999999}}}}), - record(json!({"kind":"model_completed","usage":{"input_tokens":24000,"cached_tokens":20000,"output_tokens":30}})), - record(json!({"kind":"assistant_message_committed","text":"Architecture review idea: https://github.com/acme/widgets/issues/8"})), - ].map(|line| line.to_string()).join("\n"); - let log = SessionLog::parse(&lines, "s1").unwrap(); - assert_eq!( - log.message.as_deref(), - Some("Architecture review idea: https://github.com/acme/widgets/issues/8") - ); - assert_eq!( - log.summary().as_deref(), - Some("50000 input tokens (32000 cached), 70 output tokens") - ); - } - - #[test] - fn retained_permission_transactions_are_unwrapped() { - let message = record(json!({"kind":"assistant_message_committed","text":"last reply"})); - let frame = json!({"retained_frame":"session_permission_transaction","children":[{"record_json":message.to_string()}]}); - let log = SessionLog::parse(&frame.to_string(), "s1").unwrap(); - assert_eq!(log.message.as_deref(), Some("last reply")); - assert_eq!(log.summary(), None); - } - - #[test] - fn a_truncated_or_non_json_log_is_unreadable() { - for text in [ - "not JSON", - "{", - "{\"children\":[{\"record_json\":\"bad\"}]}", - ] { - assert!(SessionLog::parse(text, "s1").is_err()); - } +pub(in crate::harness) fn read(root: &Path, id: &str) -> Option { + if id.contains(['/', '\\']) || id == ".." { + return None; } + let path = find_log(&root.join("sessions"), id, 3)?; + SessionLog::parse(&fs::read_to_string(path).ok()?, id).ok() } diff --git a/src/harness/muse/stream.rs b/src/harness/muse/stream.rs index cd2938004..4ed46f95a 100644 --- a/src/harness/muse/stream.rs +++ b/src/harness/muse/stream.rs @@ -1,7 +1,8 @@ //! Muse's JSONL event envelopes. Usage is read from its session log after exit. +use crate::harness::interpretation::{Decoder, Ended, Facts, Report, TurnOutcome}; use crate::harness::skill_load::SkillLoad; -use crate::progress::{Stream, bash, shorten}; +use crate::progress::{bash, shorten}; use serde_json::Value; #[derive(Default)] @@ -54,7 +55,7 @@ impl MuseProgress { } } -impl Stream for MuseProgress { +impl Decoder for MuseProgress { fn condense(&mut self, raw: &str) -> Vec { let Ok(event) = serde_json::from_str::(raw) else { return Vec::new(); @@ -128,36 +129,30 @@ impl Stream for MuseProgress { _ => Vec::new(), } } - fn summary(&self) -> Option { - None - } - fn final_message(&self) -> Option<&str> { - self.final_message.as_deref() - } - fn session_id(&self) -> Option<&str> { - self.session_id.as_deref() - } - fn killed_background_work(&self) -> Vec<&str> { - if self.failed() { - Vec::new() - } else { - self.killed.iter().map(String::as_str).collect() + fn complete(self: Box) -> Facts { + let outcome = TurnOutcome::from_failed(self.failure.is_some()); + Facts { + report: Report { + warnings: self.skill_load.warnings(), + summary: None, + }, + ended: Ended { + session_id: self.session_id, + killed: self.killed, + final_message: self.final_message, + }, + outcome, + diagnostic: self.failure, } } - fn failed(&self) -> bool { - self.failure.is_some() - } - fn error(&self) -> Option<&str> { - self.failure.as_deref() - } - fn warnings(&self) -> Vec { - self.skill_load.warnings() - } } #[cfg(test)] mod tests { - use super::*; + use crate::harness::{ + Harness, + interpretation_tests::{finish, stream}, + }; use serde_json::json; #[test] @@ -169,17 +164,22 @@ mod tests { r#"{"stream":{"kind":"session","id":"s1"},"payload_type":"run.output.delta","payload":{"text":"OK MANGO"}}"#, r#"{"stream":{"kind":"session","id":"s1"},"payload_type":"run.terminal.completed","payload":{"text":"MANGOOK MANGO"}}"#, ]; - let mut stream = MuseProgress::default(); + let mut stream = stream(Harness::Muse, ""); for line in lines { stream.condense(line); } - assert_eq!(stream.final_message(), Some("OK MANGO")); - assert_eq!(stream.summary(), None); + let stream = finish(stream); + assert_eq!( + stream.outcome.as_ref().unwrap().final_message.as_deref(), + Some("OK MANGO") + ); + assert_eq!(stream.report.as_ref().unwrap().summary.clone(), None); } #[test] fn a_successful_read_skill_result_loads_only_the_named_skill() { - let mut stream = MuseProgress::for_prompt( + let mut stream = stream( + Harness::Muse, "/thirdshift-implement https://github.com/acme/widgets/issues/7", ); let result = |name: &str| { @@ -189,12 +189,16 @@ mod tests { }}).to_string() }; stream.condense(&result("thirdshift-tdd")); - assert_eq!(stream.warnings().len(), 1); + assert_eq!( stream.condense(&result("thirdshift-implement")), ["skill thirdshift-implement"] ); - assert!(stream.warnings().is_empty()); - assert_eq!(stream.session_id(), Some("s1")); + let stream = finish(stream); + assert!(stream.report.as_ref().unwrap().warnings.clone().is_empty()); + assert_eq!( + stream.outcome.as_ref().unwrap().session_id.as_deref(), + Some("s1") + ); } } diff --git a/src/harness/opencode.rs b/src/harness/opencode.rs index e8dbe5832..065cc4b4c 100644 --- a/src/harness/opencode.rs +++ b/src/harness/opencode.rs @@ -1,11 +1,11 @@ //! OpenCode: private standalone sessions, stdin prompts and an authoritative export. -mod export; +pub(super) mod export; mod stream; use super::adapter::{Adapter, Invocation, SkillLoading}; +use super::interpretation::{Interpretation, Retained}; use super::settings::{Terminal, ask_setting}; -use super::{Choice, Harness, ModelAndEffort, Settings, said}; -use crate::progress::Stream; +use super::{Choice, Harness, ModelAndEffort, Settings}; use anyhow::{Context, Result, bail}; use std::path::Path; use std::process::Command; @@ -93,15 +93,16 @@ impl Adapter for OpenCode { } } } - fn stream(&self, _worktree: &Path, prompt: &str) -> Box { - Box::new(stream::OpenCodeProgress::for_prompt(prompt)) + fn interpretation(&self, worktree: &Path, prompt: &str) -> Interpretation { + Interpretation::new( + self.name(), + Box::new(stream::OpenCodeProgress::for_prompt(prompt)), + Retained::OpenCode(worktree.to_path_buf()), + ) } fn link_instruction_fallback(&self, worktree: &Path) -> Result<()> { super::instructions::link_fallback(worktree, "AGENTS.md", &[]) } - fn read_after_exit(&self, worktree: &Path, stream: Box) -> Result> { - export::read_after_exit(worktree, stream) - } } /// Setup and Commands make the same minimal call: there is no free catalog @@ -124,20 +125,10 @@ pub fn check_model_and_effort(chosen: &ModelAndEffort) -> Result<()> { invocation.stdin.as_deref(), ) .context("could not run opencode to check the Model and Effort")?; - let mut stream = OpenCode.stream(Path::new("."), "Reply with OK."); - for line in String::from_utf8_lossy(&output.stdout).lines() { - stream.condense(line); - } - let stream = OpenCode.read_after_exit(Path::new("."), stream)?; - if !output.status.success() || stream.failed() { - bail!( - "opencode refused the Model or Effort in its test call: {}", - stream - .error() - .map(String::from) - .unwrap_or_else(|| said(&output)) - ); - } + OpenCode + .interpretation(Path::new("."), "Reply with OK.") + .check_output(&output) + .context("opencode refused the Model or Effort in its test call")?; Ok(()) } @@ -159,9 +150,13 @@ mod tests { chosen_by: super::super::ChosenBy::Command, }; let prompt = "/thirdshift-implement issue-url"; - let mut stream = OpenCode.stream(Path::new("."), prompt); + let mut stream = + OpenCode.interpretation(Path::new("/missing-thirdshift-test-worktree"), prompt); stream.condense(r#"{"type":"step_start","sessionID":"ses_eedb9657fffec0RAa52IqAvyM1","part":{"type":"step-start"}}"#); - let invocation = OpenCode.session(&choice, stream.session_id(), prompt); + let ended = crate::harness::interpretation_tests::finish(stream) + .outcome + .unwrap(); + let invocation = OpenCode.session(&choice, ended.session_id.as_deref(), prompt); assert_eq!( invocation.args, [ @@ -184,6 +179,6 @@ mod tests { OpenCode.environment(), [("OPENCODE_DISABLE_AUTOUPDATE", "1")] ); - assert!(stream.killed_background_work().is_empty()); + assert!(ended.killed.is_empty()); } } diff --git a/src/harness/opencode/export.rs b/src/harness/opencode/export.rs index f6df4b9a4..f3628df11 100644 --- a/src/harness/opencode/export.rs +++ b/src/harness/opencode/export.rs @@ -2,16 +2,15 @@ //! survive dropped stream events. Session totals also count the title call. use super::OpenCode; use crate::harness::Adapter; -use crate::progress::Stream; use anyhow::{Result, anyhow}; use serde_json::Value; use std::path::Path; use std::process::Command; -pub struct SessionExport { - message: Option, - summary: Option, - failure: Option, +pub(in crate::harness) struct SessionExport { + pub(in crate::harness) message: Option, + pub(in crate::harness) summary: Option, + pub(in crate::harness) failure: Option, } /// OpenCode records uncached input and visible output separately from cache @@ -98,122 +97,19 @@ impl SessionExport { } } -pub fn read_after_exit(worktree: &Path, stream: Box) -> Result> { - let export = if let Some(id) = stream.session_id() { - let output = super::super::process::output( - &OpenCode, - Command::new(OpenCode.name()) - .args(["session", "export", "--standalone", id]) - .current_dir(worktree), - None, - ); - match output { - Ok(output) if output.status.success() => { - SessionExport::parse(&String::from_utf8_lossy(&output.stdout)).ok() - } - Err(error) if crate::interrupt::requested() => return Err(error), - _ => None, - } - } else { - None - }; - Ok(Box::new(AfterExit { stream, export })) -} - -struct AfterExit { - stream: Box, - export: Option, -} -impl Stream for AfterExit { - fn condense(&mut self, raw: &str) -> Vec { - self.stream.condense(raw) - } - fn summary(&self) -> Option { - self.export - .as_ref() - .and_then(|export| export.summary.clone()) - } - fn final_message(&self) -> Option<&str> { - match &self.export { - Some(export) => export.message.as_deref(), - None => self.stream.final_message(), +pub(in crate::harness) fn read(worktree: &Path, id: &str) -> Result> { + let output = super::super::process::output( + &OpenCode, + Command::new(OpenCode.name()) + .args(["session", "export", "--standalone", id]) + .current_dir(worktree), + None, + ); + match output { + Ok(output) if output.status.success() => { + Ok(SessionExport::parse(&String::from_utf8_lossy(&output.stdout)).ok()) } - } - fn session_id(&self) -> Option<&str> { - self.stream.session_id() - } - fn killed_background_work(&self) -> Vec<&str> { - if self.failed() { - Vec::new() - } else { - self.stream.killed_background_work() - } - } - fn failed(&self) -> bool { - self.stream.failed() - || self - .export - .as_ref() - .is_some_and(|export| export.failure.is_some()) - } - fn error(&self) -> Option<&str> { - self.stream.error().or_else(|| { - self.export - .as_ref() - .and_then(|export| export.failure.as_deref()) - }) - } - fn warnings(&self) -> Vec { - self.stream.warnings() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn the_last_assistant_text_and_summed_usage_exclude_the_title_call_and_user() { - // OpenCode 2.0.24's exported message format, with irrelevant fields removed. - let export = SessionExport::parse(r#"{ - "info":{"outcome":"succeeded","tokens":{"input":999999},"cost":9}, - "messages":[ - {"type":"user","text":"prompt","tokens":{"input":999999}}, - {"type":"assistant","tokens":{"input":31,"output":37,"reasoning":158,"cache":{"read":7840,"write":20}},"cost":0.004,"content":[{"type":"text","text":"earlier reply"}]}, - {"type":"assistant","tokens":{"input":39,"output":11,"reasoning":42,"cache":{"read":496,"write":7}},"cost":0.002,"content":[{"type":"text","text":"first part"},{"type":"reasoning","text":"private reasoning"},{"type":"text","text":"Architecture review idea: https://github.com/acme/widgets/issues/8"}]}, - {"type":"idle","outcome":"succeeded"} - ] - }"#).unwrap(); - assert_eq!( - export.message.as_deref(), - Some("Architecture review idea: https://github.com/acme/widgets/issues/8") - ); - assert_eq!( - export.summary.as_deref(), - Some( - "8433 input tokens (8336 cache read, 27 cache write), 248 output tokens (200 reasoning), $0.0060" - ) - ); - assert!(export.failure.is_none()); - } - - #[test] - fn a_failed_export_reports_the_assistants_error_when_the_session_has_none() { - let export = SessionExport::parse(r#"{"info":{"outcome":"failed"},"messages":[{"type":"assistant","content":[],"error":{"type":"provider.no-route","message":"Variant unavailable"}}]}"#).unwrap(); - assert_eq!(export.failure.as_deref(), Some("Variant unavailable")); - } - - #[test] - fn incomplete_or_malformed_exports_are_unreadable() { - for text in ["{", "{}", r#"{"info":{"outcome":"succeeded"}}"#] { - assert!(SessionExport::parse(text).is_err()); - } - } - - #[test] - fn an_assistant_without_text_does_not_reuse_an_earlier_reply() { - let export = SessionExport::parse(r#"{"info":{"outcome":"succeeded"},"messages":[{"type":"assistant","content":[{"type":"text","text":"earlier reply"}]},{"type":"assistant","content":[{"type":"tool","name":"bash"}]}]}"#).unwrap(); - assert!(export.message.is_none()); - assert!(export.summary.is_none()); + Err(error) if crate::interrupt::requested() => Err(error), + _ => Ok(None), } } diff --git a/src/harness/opencode/stream.rs b/src/harness/opencode/stream.rs index 58d767315..9c9f0e4f3 100644 --- a/src/harness/opencode/stream.rs +++ b/src/harness/opencode/stream.rs @@ -1,6 +1,7 @@ //! OpenCode's JSONL progress; final usage is deliberately left to session export. +use crate::harness::interpretation::{Decoder, Ended, Facts, Report, TurnOutcome}; use crate::harness::skill_load::SkillLoad; -use crate::progress::{Stream, bash, shorten}; +use crate::progress::{bash, shorten}; use serde_json::Value; #[derive(Default)] @@ -21,7 +22,7 @@ impl OpenCodeProgress { } } -impl Stream for OpenCodeProgress { +impl Decoder for OpenCodeProgress { fn condense(&mut self, raw: &str) -> Vec { let Ok(event) = serde_json::from_str::(raw) else { return Vec::new(); @@ -78,38 +79,35 @@ impl Stream for OpenCodeProgress { _ => Vec::new(), } } - fn summary(&self) -> Option { - None - } - fn final_message(&self) -> Option<&str> { - self.message.as_deref() - } - fn session_id(&self) -> Option<&str> { - self.session_id.as_deref() - } - fn killed_background_work(&self) -> Vec<&str> { - // OpenCode's stream has no documented event proving background work - // was killed at session end. Tool cancellation alone is not evidence. - Vec::new() - } - fn failed(&self) -> bool { - self.failure.is_some() - } - fn error(&self) -> Option<&str> { - self.failure.as_deref() - } - fn warnings(&self) -> Vec { - self.skill_load.warnings() + fn complete(self: Box) -> Facts { + let outcome = TurnOutcome::from_failed(self.failure.is_some()); + Facts { + report: Report { + warnings: self.skill_load.warnings(), + summary: None, + }, + ended: Ended { + session_id: self.session_id, + // Tool cancellation alone is not killed-work evidence. + killed: Vec::new(), + final_message: self.message, + }, + outcome, + diagnostic: self.failure, + } } } #[cfg(test)] mod tests { - use super::*; + use crate::harness::{ + Harness, + interpretation_tests::{finish, stream}, + }; #[test] fn a_dropped_last_step_keeps_the_last_text_without_trusting_stream_usage() { - let mut stream = OpenCodeProgress::for_prompt("/thirdshift-implement issue-url"); + let mut stream = stream(Harness::OpenCode, "/thirdshift-implement issue-url"); for line in [ r#"{"type":"step_start","sessionID":"s1","part":{}}"#, r#"{"type":"tool_use","sessionID":"s1","part":{"tool":"skill","state":{"status":"completed","input":{"id":"thirdshift-implement"}}}}"#, @@ -120,36 +118,53 @@ mod tests { ] { stream.condense(line); } - assert_eq!(stream.final_message(), Some("last reply")); - assert_eq!(stream.session_id(), Some("s1")); - assert!(stream.warnings().is_empty()); - assert!(stream.summary().is_none()); + let stream = finish(stream); + assert_eq!( + stream.outcome.as_ref().unwrap().final_message.as_deref(), + Some("last reply") + ); + assert_eq!( + stream.outcome.as_ref().unwrap().session_id.as_deref(), + Some("s1") + ); + assert!(stream.report.as_ref().unwrap().warnings.clone().is_empty()); + assert!(stream.report.as_ref().unwrap().summary.clone().is_none()); } #[test] fn a_wrong_or_failed_skill_load_still_warns() { - let mut stream = OpenCodeProgress::for_prompt("/thirdshift-implement issue-url"); + let mut stream = stream(Harness::OpenCode, "/thirdshift-implement issue-url"); for line in [ r#"{"type":"tool_use","part":{"tool":"skill","state":{"status":"completed","input":{"id":"thirdshift-tdd"}}}}"#, r#"{"type":"tool_use","part":{"tool":"skill","state":{"status":"error","input":{"id":"thirdshift-implement"}}}}"#, ] { stream.condense(line); } + let stream = finish(stream); assert_eq!( - stream.warnings(), + stream.report.as_ref().unwrap().warnings.clone(), ["warning: the session never loaded thirdshift-implement with its skill tool"] ); - assert!(stream.session_id().is_none()); + assert!( + stream + .outcome + .as_ref() + .unwrap() + .session_id + .as_deref() + .is_none() + ); } #[test] fn a_recorded_no_route_error_fails_and_names_the_model_or_effort_problem() { - let mut stream = OpenCodeProgress::default(); + let mut stream = stream(Harness::OpenCode, ""); stream.condense(r#"{"type":"error","sessionID":"s1","error":{"type":"provider.no-route","message":"Model unavailable: thirdshift-invalid-provider/invalid"}}"#); - assert!(stream.failed()); + let stream = finish(stream); + assert!(stream.outcome.is_err()); assert_eq!( - stream.error(), - Some("provider.no-route: Model unavailable: thirdshift-invalid-provider/invalid") + stream.outcome.as_ref().unwrap_err().to_string(), + "opencode's turn failed: provider.no-route: Model unavailable: thirdshift-invalid-provider/invalid" ); } } diff --git a/src/progress.rs b/src/progress.rs index c670d841f..8e1271d65 100644 --- a/src/progress.rs +++ b/src/progress.rs @@ -8,11 +8,7 @@ use std::fmt::Display; use chrono::Local; -#[cfg(test)] -use crate::harness::claude::stream::ClaudeProgress; use crate::logs; -#[cfg(test)] -use serde_json::Value; /// The longest detail a session line shows before it is cut short. const MAX_DETAIL: usize = 100; @@ -110,49 +106,6 @@ fn split_stamp(line: &str) -> Option<(&str, &str)> { is_stamp.then_some((time, message)) } -/// What a session's stream shows, read a line at a time: the progress lines -/// it condenses to, and once it has ended, how the session went. Each -/// Harness streams its own events, parsed by its adapter. -pub trait Stream: Send { - /// The stderr lines, without the `thirdshift: ` prefix, for one line of - /// the stream: one per notable event in it, often none. Unknown and - /// malformed lines give none, never an error. - fn condense(&mut self, raw: &str) -> Vec; - - /// Warnings reported once after exit, without failing or retrying the session. - fn warnings(&self) -> Vec { - Vec::new() - } - - /// What the session-ended line says the session took, if the stream - /// told. - fn summary(&self) -> Option; - - /// The session's final message: what the agent said as it ended its - /// last turn, if anything. - fn final_message(&self) -> Option<&str>; - - /// The session's id, which a Resume continues. - fn session_id(&self) -> Option<&str>; - - /// Descriptions of the background work killed as the session ended: the - /// work still running when it ended its last turn, which it may have - /// been waiting on. None if that turn failed: the session ended for - /// another reason. - fn killed_background_work(&self) -> Vec<&str>; - - /// Whether the stream says the session failed, whatever its exit code. - fn failed(&self) -> bool { - false - } - - /// The error the stream gave for the session, for a failure's cause: - /// what failed its turn, else the last error it reported, if any. - fn error(&self) -> Option<&str> { - None - } -} - /// Commits and pushes by name; any other command by its first line. pub(crate) fn bash(command: &str) -> String { let actions: Vec<&str> = ["commit", "push"] @@ -184,295 +137,6 @@ pub(crate) fn shorten(text: &str) -> String { #[cfg(test)] mod tests { use super::*; - use serde_json::json; - - fn lines(events: &[Value]) -> (ClaudeProgress, Vec>) { - let mut progress = ClaudeProgress::default(); - let lines = events - .iter() - .map(|event| progress.condense(&event.to_string())) - .collect(); - (progress, lines) - } - - fn tool_use(name: &str, input: Value) -> Value { - json!({ - "type": "assistant", - "message": { "content": [{ "type": "tool_use", "name": name, "input": input }] } - }) - } - - fn init(cwd: &str) -> Value { - json!({ "type": "system", "subtype": "init", "cwd": cwd }) - } - - fn line_for(event: Value) -> Option { - let mut lines = lines(&[init("/work/widgets-issue-7"), event]) - .1 - .pop() - .unwrap(); - assert!(lines.len() <= 1, "several lines: {lines:?}"); - lines.pop() - } - - fn result(turns: u64, cost: f64) -> Value { - json!({ "type": "result", "subtype": "success", "num_turns": turns, "total_cost_usd": cost }) - } - - #[test] - fn only_the_first_init_starts_the_session() { - let (_, lines) = lines(&[init("/a"), init("/a")]); - assert_eq!(lines, [vec!["session started".to_string()], vec![]]); - } - - #[test] - fn a_skill_is_named() { - assert_eq!( - line_for(tool_use( - "Skill", - json!({ "skill": "thirdshift-tdd", "args": "x" }) - )), - Some("skill thirdshift-tdd".to_string()) - ); - } - - #[test] - fn commits_and_pushes_are_named() { - let line = |command: &str| line_for(tool_use("Bash", json!({ "command": command }))); - assert_eq!(line("git commit -m 'x'"), Some("commit".to_string())); - assert_eq!(line("git push -u origin issue-7"), Some("push".to_string())); - assert_eq!( - line("git add -A && git commit -m x && git push"), - Some("commit and push".to_string()) - ); - } - - #[test] - fn mentions_of_commit_or_push_are_not_commits_or_pushes() { - let line = |command: &str| line_for(tool_use("Bash", json!({ "command": command }))); - assert_eq!( - line("grep 'git push' README.md"), - Some("$ grep 'git push' README.md".to_string()) - ); - assert_eq!( - line("git commit-tree HEAD^{tree}"), - Some("$ git commit-tree HEAD^{tree}".to_string()) - ); - } - - #[test] - fn other_commands_show_their_first_line() { - assert_eq!( - line_for(tool_use( - "Bash", - json!({ "command": "cargo test\necho done" }) - )), - Some("$ cargo test".to_string()) - ); - } - - #[test] - fn file_tools_show_the_path_relative_to_the_session() { - assert_eq!( - line_for(tool_use( - "Edit", - json!({ "file_path": "/work/widgets-issue-7/src/run.rs" }) - )), - Some("Edit src/run.rs".to_string()) - ); - assert_eq!( - line_for(tool_use("Read", json!({ "file_path": "/etc/hosts" }))), - Some("Read /etc/hosts".to_string()) - ); - } - - #[test] - fn other_tools_show_their_most_telling_input() { - assert_eq!( - line_for(tool_use("Grep", json!({ "pattern": "fn main" }))), - Some("Grep fn main".to_string()) - ); - assert_eq!( - line_for(tool_use( - "Agent", - json!({ "description": "Standards review" }) - )), - Some("Agent Standards review".to_string()) - ); - assert_eq!( - line_for(tool_use("TodoWrite", json!({ "todos": [] }))), - Some("TodoWrite".to_string()) - ); - } - - #[test] - fn long_details_are_cut_short() { - let line = line_for(tool_use("Bash", json!({ "command": "é".repeat(150) }))).unwrap(); - assert_eq!(line, format!("$ {}…", "é".repeat(100))); - } - - #[test] - fn several_tool_uses_in_one_message_get_a_line_each() { - let event = json!({ - "type": "assistant", - "message": { "content": [ - { "type": "text", "text": "Reading both." }, - { "type": "tool_use", "name": "Read", "input": { "file_path": "a.rs" } }, - { "type": "tool_use", "name": "Read", "input": { "file_path": "b.rs" } } - ] } - }); - let (_, lines) = lines(&[event]); - assert_eq!( - lines, - [vec!["Read a.rs".to_string(), "Read b.rs".to_string()]] - ); - } - - #[test] - fn text_tool_results_and_results_give_no_line() { - let (_, lines) = lines(&[ - json!({ "type": "assistant", "message": { "content": [{ "type": "text", "text": "hi" }] } }), - json!({ "type": "user", "message": { "content": [{ "type": "tool_result" }] } }), - json!({ "type": "result", "num_turns": 3, "total_cost_usd": 0.1 }), - ]); - assert_eq!(lines, [vec![], vec![], vec![]] as [Vec; 3]); - } - - #[test] - fn the_summary_is_the_last_result_with_totals() { - let (progress, _) = lines(&[ - result(10, 0.5), - result(34, 1.8249), - json!({ "type": "result", "subtype": "success" }), - ]); - assert_eq!(progress.summary().as_deref(), Some("34 turns, $1.82")); - } - - #[test] - fn the_final_message_is_the_text_of_the_last_result() { - let said = |text: &str| json!({ "type": "result", "subtype": "success", "result": text }); - let (progress, _) = lines(&[said("Waiting on the tests."), said("Done.")]); - assert_eq!(progress.final_message(), Some("Done.")); - - let (progress, _) = lines(&[said("Done."), result(3, 0.1)]); - assert_eq!(progress.final_message(), None); - } - - #[test] - fn on_a_subscription_the_cost_is_marked_as_at_api_prices() { - let (progress, _) = lines(&[ - json!({ "type": "system", "subtype": "init", "apiKeySource": "none" }), - result(34, 1.82), - ]); - assert_eq!( - progress.summary().as_deref(), - Some("34 turns, $1.82 at API prices") - ); - } - - #[test] - fn no_result_means_no_summary() { - assert_eq!(ClaudeProgress::default().summary(), None); - } - - fn task_started(id: &str, description: &str) -> Value { - json!({ "type": "system", "subtype": "task_started", "task_id": id, "description": description }) - } - - fn task_updated(id: &str, status: &str) -> Value { - json!({ "type": "system", "subtype": "task_updated", "task_id": id, "patch": { "status": status } }) - } - - fn task_notification(id: &str, status: &str, summary: &str) -> Value { - json!({ "type": "system", "subtype": "task_notification", "task_id": id, "status": status, "summary": summary }) - } - - #[test] - fn a_task_killed_after_the_last_result_is_killed_background_work() { - let (progress, _) = lines(&[ - json!({ "type": "system", "subtype": "init", "session_id": "s-1" }), - task_started("b1", "./mvnw test -Dtest='GamesPageTest'"), - result(12, 0.4), - task_updated("b1", "killed"), - ]); - assert_eq!(progress.session_id(), Some("s-1")); - assert_eq!( - progress.killed_background_work(), - ["./mvnw test -Dtest='GamesPageTest'"] - ); - } - - #[test] - fn a_stopped_task_notification_is_killed_background_work_once() { - let (progress, _) = lines(&[ - task_started("b1", "cargo test"), - result(12, 0.4), - task_updated("b1", "killed"), - task_notification("b1", "stopped", "cargo test"), - task_notification("b2", "stopped", "npm run build"), - ]); - assert_eq!( - progress.killed_background_work(), - ["cargo test", "npm run build"] - ); - } - - #[test] - fn no_killed_tasks_means_no_killed_background_work() { - let (progress, _) = lines(&[ - task_started("b1", "cargo test"), - task_notification("b1", "completed", "cargo test"), - result(12, 0.4), - ]); - assert!(progress.killed_background_work().is_empty()); - } - - #[test] - fn background_sub_agents_that_resumed_the_session_are_not_killed_background_work() { - let (progress, _) = lines(&[ - task_started("a1", "Standards review"), - result(12, 0.4), - task_updated("a1", "completed"), - task_notification("a1", "completed", "Standards review"), - init("/work/widgets-issue-7"), - result(20, 0.9), - ]); - assert!(progress.killed_background_work().is_empty()); - } - - #[test] - fn a_task_killed_before_a_later_result_is_not_killed_background_work() { - let (progress, _) = lines(&[ - task_started("b1", "cargo watch"), - task_updated("b1", "killed"), - result(12, 0.4), - init("/work/widgets-issue-7"), - task_notification("b1", "stopped", "cargo watch"), - result(20, 0.9), - ]); - assert!(progress.killed_background_work().is_empty()); - } - - #[test] - fn tasks_killed_after_an_error_result_are_not_killed_background_work() { - for last in [ - json!({ "type": "result", "subtype": "error_max_turns", "is_error": true }), - json!({ "type": "result", "subtype": "success", "is_error": true }), - ] { - let (progress, _) = lines(&[ - task_started("b1", "cargo test"), - last, - task_updated("b1", "killed"), - ]); - assert!(progress.killed_background_work().is_empty()); - } - } - - #[test] - fn an_init_without_a_session_id_gives_none() { - let (progress, _) = lines(&[init("/a")]); - assert_eq!(progress.session_id(), None); - } #[test] fn a_stamped_line_splits_into_its_time_and_message() { @@ -551,25 +215,4 @@ mod tests { ); } } - - #[test] - fn unknown_and_malformed_lines_are_skipped() { - let mut progress = ClaudeProgress::default(); - for raw in [ - "", - "not json", - "[1, 2]", - "42", - r#"{"type": "assistant""#, - r#"{"no_type": true}"#, - r#"{"type": 7}"#, - r#"{"type": "stream_event", "event": {}}"#, - r#"{"type": "assistant", "message": {"content": "oops"}}"#, - r#"{"type": "assistant", "message": {"content": [{"type": "tool_use"}]}}"#, - r#"{"type": "result", "num_turns": "many", "total_cost_usd": null}"#, - ] { - assert_eq!(progress.condense(raw), Vec::::new(), "{raw}"); - } - assert_eq!(progress.summary(), None); - } } diff --git a/src/session.rs b/src/session.rs index 9899d8f95..3cf60e500 100644 --- a/src/session.rs +++ b/src/session.rs @@ -8,17 +8,20 @@ use std::process::Command; use std::sync::mpsc; use std::time::{Duration, Instant}; -use anyhow::{Context, Result, bail}; +#[cfg(test)] +use anyhow::bail; +use anyhow::{Context, Result}; #[cfg(test)] pub(crate) use crate::harness::claude::claude_args; #[cfg(test)] pub(crate) use crate::harness::codex::{codex_args, codex_prompt}; +use crate::harness::interpretation::{Ended, Interpretation}; use crate::harness::{Adapter, Choice, Invocation, process}; use crate::interrupt; use crate::issue::{IssueUrl, Repo}; use crate::logs; -use crate::progress::{self, Stream}; +use crate::progress; use crate::prompt; use crate::skills; @@ -202,23 +205,6 @@ trait Outside { fn step(&mut self, line: String); } -/// What a session ended with, as its stream showed it. -struct Ended { - /// Its id, which a Resume continues. - session_id: Option, - /// Descriptions of the background work killed as it ended. - killed: Vec, - /// What the agent said as it ended its last turn, if anything. - final_message: Option, -} - -impl Ended { - /// The descriptions of the killed background work, borrowed. - fn killed_work(&self) -> Vec<&str> { - self.killed.iter().map(String::as_str).collect() - } -} - /// Sessions run by the Command's `harness`'s CLI in `worktree`, where it /// finds the Factory skills, and progress lines printed on stderr. struct OnMachine { @@ -236,17 +222,15 @@ impl Outside for OnMachine { log: &Path, ) -> Result { let invocation = self.adapter.session(&self.harness, resume, prompt); - let stream = self.adapter.stream(&self.worktree, prompt); - let stream = run(kind, self.adapter, &self.worktree, invocation, log, stream)?; - Ok(Ended { - session_id: stream.session_id().map(String::from), - killed: stream - .killed_background_work() - .into_iter() - .map(String::from) - .collect(), - final_message: stream.final_message().map(String::from), - }) + let interpretation = self.adapter.interpretation(&self.worktree, prompt); + run( + kind, + self.adapter, + &self.worktree, + invocation, + log, + interpretation, + ) } fn step(&mut self, line: String) { @@ -279,8 +263,8 @@ fn run( worktree: &Path, invocation: Invocation, log: &Path, - mut stream: Box, -) -> Result> { + mut stream: Interpretation, +) -> Result { let cli = adapter.name(); if let Some(dir) = log.parent() { fs::create_dir_all(dir).with_context(|| format!("could not create {}", dir.display()))?; @@ -299,13 +283,7 @@ fn run( .args(&invocation.args) .current_dir(worktree), invocation.stdin.as_deref(), - move |output| match follow( - &kind_owned, - output, - &mut log_file, - &log_owned, - stream.as_mut(), - ) { + move |output| match follow(&kind_owned, output, &mut log_file, &log_owned, &mut stream) { Ok(()) => Ok(stream), Err(error) => { let _ = failed_stream.send(stream); @@ -326,43 +304,20 @@ fn run( } }; - let stream = match adapter.read_after_exit(worktree, stream) { - Ok(stream) => stream, - Err(error) => { - if interrupt::requested() { - return Err(error); - } - return Err(status.err().unwrap_or(error)); + let completion = stream.finish(status); + if let Some(report) = completion.report { + for warning in report.warnings { + progress::step(format_args!("{kind}: {warning}")); } - }; - for warning in stream.warnings() { - progress::step(format_args!("{kind}: {warning}")); - } - let elapsed = minutes_and_seconds(started.elapsed()); - let summary = stream - .summary() - .map_or(String::new(), |summary| format!(": {summary}")); - progress::step(format_args!( - "{kind}: session ended after {elapsed}{summary}" - )); - let status = status?; - if status.success() && !stream.failed() { - return Ok(stream); - } - let ended = if status.success() { - "'s turn failed".to_string() - } else { - format!( - " exited {}", - status - .code() - .map_or("by signal".to_string(), |code| code.to_string()) - ) - }; - match stream.error() { - Some(error) => bail!("{cli}{ended}: {error}"), - None => bail!("{cli}{ended}"), - } + let elapsed = minutes_and_seconds(started.elapsed()); + let summary = report + .summary + .map_or(String::new(), |summary| format!(": {summary}")); + progress::step(format_args!( + "{kind}: session ended after {elapsed}{summary}" + )); + } + completion.outcome } /// Copy every line of `stream` to `log_file` and print the progress lines it @@ -372,7 +327,7 @@ fn follow( stream: impl Read, log_file: &mut File, log: &Path, - progress: &mut dyn Stream, + progress: &mut Interpretation, ) -> Result<()> { let mut stream = BufReader::new(stream); let mut line = Vec::new(); diff --git a/tests/background_work.rs b/tests/background_work.rs index 04f0387e6..0244ec130 100644 --- a/tests/background_work.rs +++ b/tests/background_work.rs @@ -133,3 +133,25 @@ fn a_session_that_ends_in_an_error_with_killed_background_work_is_not_resumed_or result.stderr ); } + +#[test] +fn a_zero_exit_failed_claude_turn_leaves_the_pr_draft_without_a_resume() { + let scenario = Scenario::new(); + scenario.agent_does(&format!( + "{AGENT_COMMITS_AND_OPENS_PR}{}\n\ + echo '{{\"type\":\"result\",\"subtype\":\"error_max_turns\",\"is_error\":true}}' >> \"$FAKE_CLAUDE_AFTER_RESULT\"\n", + leaves_running(TESTS) + )); + + let result = scenario.run(&[&scenario.issue_url(7)]); + + assert_eq!(result.code, Some(1), "stderr: {}", result.stderr); + assert!( + result.stderr.contains("claude's turn failed"), + "stderr: {}", + result.stderr + ); + assert_eq!(scenario.claude_calls().len(), 1); + assert_eq!(scenario.gh_state()["prs"][0]["isDraft"], true); + scenario.assert_cleaned_up("issue-7"); +} diff --git a/tests/opencode_sessions.rs b/tests/opencode_sessions.rs index 0e4f5dddc..5a9922afa 100644 --- a/tests/opencode_sessions.rs +++ b/tests/opencode_sessions.rs @@ -16,6 +16,111 @@ fn records(scenario: &Scenario, extension: &str) -> Vec { .unwrap_or_default() } +/// Replace the fake CLI's persisted export at its external acquisition seam. +fn export_script(scenario: &Scenario, text: &str) -> std::path::PathBuf { + let script = scenario.path("replace-export.sh"); + std::fs::write( + &script, + format!( + "cat > '{}' <<'EXPORT'\n{text}\nEXPORT\n", + scenario + .path("opencode-calls.fake-opencode-1.export.json") + .display() + ), + ) + .unwrap(); + script +} + +fn architecture_review(scenario: &Scenario) { + scenario.repo_has_labels(&["needs-triage", "ready-for-agent", "architecture"]); + scenario.agent_does_in_session( + 1, + r#" +url=$(gh issue create --title 'An idea' --body 'The idea' --label needs-triage) +printf 'Architecture review idea: %s\n' "$url" > "$FAKE_CLAUDE_FINAL_MESSAGE" +"#, + ); +} + +#[test] +fn a_readable_export_owns_final_text_and_assistant_usage_without_counting_the_title_call() { + let scenario = Scenario::new(); + architecture_review(&scenario); + let script = export_script( + &scenario, + r#"{ + "info":{"outcome":"succeeded","tokens":{"input":999999},"cost":9}, + "messages":[ + {"type":"user","text":"prompt","tokens":{"input":999999}}, + {"type":"assistant","tokens":{"input":31,"output":37,"reasoning":158,"cache":{"read":7840,"write":20}},"cost":0.004,"content":[{"type":"text","text":"earlier reply"}]}, + {"type":"assistant","tokens":{"input":39,"output":11,"reasoning":42,"cache":{"read":496,"write":7}},"cost":0.002,"content":[{"type":"text","text":"first part"},{"type":"reasoning","text":"private reasoning"},{"type":"text","text":"Architecture review already filed: https://github.com/acme/widgets/issues/8"}]}, + {"type":"idle","outcome":"succeeded"} + ] + }"#, + ); + let result = scenario.run_with_env( + &["architect", "harness", "opencode"], + &[("FAKE_OPENCODE_EXPORT_SCRIPT", script.to_str().unwrap())], + ); + assert_eq!(result.code, Some(0), "{}", result.stderr); + assert!( + result.stderr.contains("so it filed nothing"), + "{}", + result.stderr + ); + assert!(result.stderr.contains("8433 input tokens (8336 cache read, 27 cache write), 248 output tokens (200 reasoning), $0.0060"), "{}", result.stderr); +} + +#[test] +fn readable_exports_without_final_assistant_text_do_not_reuse_the_stream_reply() { + for text in [ + r#"{"info":{"outcome":"succeeded"},"messages":[]}"#, + r#"{"info":{"outcome":"succeeded"},"messages":[{"type":"assistant","content":[{"type":"text","text":"Architecture review idea: https://github.com/acme/widgets/issues/8"}]},{"type":"assistant","content":[{"type":"tool","name":"bash"}]}]}"#, + ] { + let scenario = Scenario::new(); + architecture_review(&scenario); + let script = export_script(&scenario, text); + let result = scenario.run_with_env( + &["architect", "harness", "opencode"], + &[("FAKE_OPENCODE_EXPORT_SCRIPT", script.to_str().unwrap())], + ); + assert_eq!(result.code, Some(1), "{}", result.stderr); + assert!( + result.stderr.contains("ended without the final line"), + "{}", + result.stderr + ); + assert!(!result.stderr.contains("input tokens"), "{}", result.stderr); + } +} + +#[test] +fn an_exported_failure_uses_the_assistant_diagnostic_unless_the_stream_reported_one() { + for streamed in [None, Some("stream failure")] { + let scenario = Scenario::new(); + scenario.agent_does_for(7, &streamed.map(|error| format!(r#"echo '{{"type":"error","sessionID":"fake-opencode-1","error":{{"message":"{error}"}}}}'"#)).unwrap_or_else(|| "true".into())); + let script = export_script( + &scenario, + r#"{"info":{"outcome":"failed"},"messages":[{"type":"assistant","content":[],"error":{"type":"provider.no-route","message":"Variant unavailable"}}]}"#, + ); + let result = scenario.run_with_env( + &["harness", "opencode", &scenario.issue_url(7)], + &[("FAKE_OPENCODE_EXPORT_SCRIPT", script.to_str().unwrap())], + ); + assert_eq!(result.code, Some(1), "{}", result.stderr); + assert!( + result.stderr.contains(&format!( + "opencode's turn failed: {}", + streamed.unwrap_or("Variant unavailable") + )), + "{}", + result.stderr + ); + assert_eq!(records(&scenario, "json").len(), 1); + } +} + #[test] fn opencode_runs_standalone_with_stdin_skills_and_usage_despite_a_dropped_last_step() { let scenario = Scenario::new(); diff --git a/tests/process_tree.rs b/tests/process_tree.rs index c5e7998a4..021329028 100644 --- a/tests/process_tree.rs +++ b/tests/process_tree.rs @@ -275,6 +275,18 @@ touch "{started}" assert!(scenario.issue_labels(7).is_empty()); assert_eq!(scenario.entries("work"), ["widgets"]); } + if matches!(during, During::Export) { + assert!( + !result.stderr.contains("session ended after"), + "{}", + result.stderr + ); + assert!( + !result.stderr.contains("warning: the session never loaded"), + "{}", + result.stderr + ); + } let limit = if matches!(during, During::CheckAfterExit) { Duration::from_secs(2) } else { From 37013fdd5b347dd825aba270b7b35ed28d3eab5a Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 6 Oct 2026 17:59:36 -0400 Subject: [PATCH 2/3] Share interpretation test setup after Standards review --- src/harness/claude/stream.rs | 161 ++++++++++--------- src/harness/codex/stream.rs | 235 +++++++++++++++------------- src/harness/interpretation_tests.rs | 12 ++ 3 files changed, 229 insertions(+), 179 deletions(-) diff --git a/src/harness/claude/stream.rs b/src/harness/claude/stream.rs index 2ab7ddeb3..650fb79de 100644 --- a/src/harness/claude/stream.rs +++ b/src/harness/claude/stream.rs @@ -174,18 +174,9 @@ impl ClaudeProgress { mod tests { use crate::harness::{ Harness, - interpretation::Completion, - interpretation_tests::{finish, stream}, + interpretation_tests::{finish, lines, stream}, }; use serde_json::{Value, json}; - fn lines(events: &[Value]) -> (Completion, Vec>) { - let mut progress = stream(Harness::Claude, ""); - let lines = events - .iter() - .map(|event| progress.condense(&event.to_string())) - .collect(); - (finish(progress), lines) - } fn tool_use(name: &str, input: Value) -> Value { json!({ @@ -199,7 +190,7 @@ mod tests { } fn line_for(event: Value) -> Option { - let mut lines = lines(&[init("/work/widgets-issue-7"), event]) + let mut lines = lines(Harness::Claude, &[init("/work/widgets-issue-7"), event]) .1 .pop() .unwrap(); @@ -213,7 +204,7 @@ mod tests { #[test] fn only_the_first_init_starts_the_session() { - let (_, lines) = lines(&[init("/a"), init("/a")]); + let (_, lines) = lines(Harness::Claude, &[init("/a"), init("/a")]); assert_eq!(lines, [vec!["session started".to_string()], vec![]]); } @@ -313,7 +304,7 @@ mod tests { { "type": "tool_use", "name": "Read", "input": { "file_path": "b.rs" } } ] } }); - let (_, lines) = lines(&[event]); + let (_, lines) = lines(Harness::Claude, &[event]); assert_eq!( lines, [vec!["Read a.rs".to_string(), "Read b.rs".to_string()]] @@ -322,21 +313,27 @@ mod tests { #[test] fn text_tool_results_and_results_give_no_line() { - let (_, lines) = lines(&[ - json!({ "type": "assistant", "message": { "content": [{ "type": "text", "text": "hi" }] } }), - json!({ "type": "user", "message": { "content": [{ "type": "tool_result" }] } }), - json!({ "type": "result", "num_turns": 3, "total_cost_usd": 0.1 }), - ]); + let (_, lines) = lines( + Harness::Claude, + &[ + json!({ "type": "assistant", "message": { "content": [{ "type": "text", "text": "hi" }] } }), + json!({ "type": "user", "message": { "content": [{ "type": "tool_result" }] } }), + json!({ "type": "result", "num_turns": 3, "total_cost_usd": 0.1 }), + ], + ); assert_eq!(lines, [vec![], vec![], vec![]] as [Vec; 3]); } #[test] fn the_summary_is_the_last_result_with_totals() { - let (progress, _) = lines(&[ - result(10, 0.5), - result(34, 1.8249), - json!({ "type": "result", "subtype": "success" }), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + result(10, 0.5), + result(34, 1.8249), + json!({ "type": "result", "subtype": "success" }), + ], + ); assert_eq!( progress.report.as_ref().unwrap().summary.clone().as_deref(), Some("34 turns, $1.82") @@ -346,13 +343,16 @@ mod tests { #[test] fn the_final_message_is_the_text_of_the_last_result() { let said = |text: &str| json!({ "type": "result", "subtype": "success", "result": text }); - let (progress, _) = lines(&[said("Waiting on the tests."), said("Done.")]); + let (progress, _) = lines( + Harness::Claude, + &[said("Waiting on the tests."), said("Done.")], + ); assert_eq!( progress.outcome.as_ref().unwrap().final_message.as_deref(), Some("Done.") ); - let (progress, _) = lines(&[said("Done."), result(3, 0.1)]); + let (progress, _) = lines(Harness::Claude, &[said("Done."), result(3, 0.1)]); assert_eq!( progress.outcome.as_ref().unwrap().final_message.as_deref(), None @@ -361,10 +361,13 @@ mod tests { #[test] fn on_a_subscription_the_cost_is_marked_as_at_api_prices() { - let (progress, _) = lines(&[ - json!({ "type": "system", "subtype": "init", "apiKeySource": "none" }), - result(34, 1.82), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + json!({ "type": "system", "subtype": "init", "apiKeySource": "none" }), + result(34, 1.82), + ], + ); assert_eq!( progress.report.as_ref().unwrap().summary.clone().as_deref(), Some("34 turns, $1.82 at API prices") @@ -373,7 +376,7 @@ mod tests { #[test] fn no_result_means_no_summary() { - assert_eq!(lines(&[]).0.report.unwrap().summary, None); + assert_eq!(lines(Harness::Claude, &[]).0.report.unwrap().summary, None); } fn task_started(id: &str, description: &str) -> Value { @@ -390,12 +393,15 @@ mod tests { #[test] fn a_task_killed_after_the_last_result_is_killed_background_work() { - let (progress, _) = lines(&[ - json!({ "type": "system", "subtype": "init", "session_id": "s-1" }), - task_started("b1", "./mvnw test -Dtest='GamesPageTest'"), - result(12, 0.4), - task_updated("b1", "killed"), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + json!({ "type": "system", "subtype": "init", "session_id": "s-1" }), + task_started("b1", "./mvnw test -Dtest='GamesPageTest'"), + result(12, 0.4), + task_updated("b1", "killed"), + ], + ); assert_eq!( progress.outcome.as_ref().unwrap().session_id.as_deref(), Some("s-1") @@ -408,13 +414,16 @@ mod tests { #[test] fn a_stopped_task_notification_is_killed_background_work_once() { - let (progress, _) = lines(&[ - task_started("b1", "cargo test"), - result(12, 0.4), - task_updated("b1", "killed"), - task_notification("b1", "stopped", "cargo test"), - task_notification("b2", "stopped", "npm run build"), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + task_started("b1", "cargo test"), + result(12, 0.4), + task_updated("b1", "killed"), + task_notification("b1", "stopped", "cargo test"), + task_notification("b2", "stopped", "npm run build"), + ], + ); assert_eq!( progress.outcome.as_ref().unwrap().killed_work(), ["cargo test", "npm run build"] @@ -423,37 +432,46 @@ mod tests { #[test] fn no_killed_tasks_means_no_killed_background_work() { - let (progress, _) = lines(&[ - task_started("b1", "cargo test"), - task_notification("b1", "completed", "cargo test"), - result(12, 0.4), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + task_started("b1", "cargo test"), + task_notification("b1", "completed", "cargo test"), + result(12, 0.4), + ], + ); assert!(progress.outcome.as_ref().unwrap().killed_work().is_empty()); } #[test] fn background_sub_agents_that_resumed_the_session_are_not_killed_background_work() { - let (progress, _) = lines(&[ - task_started("a1", "Standards review"), - result(12, 0.4), - task_updated("a1", "completed"), - task_notification("a1", "completed", "Standards review"), - init("/work/widgets-issue-7"), - result(20, 0.9), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + task_started("a1", "Standards review"), + result(12, 0.4), + task_updated("a1", "completed"), + task_notification("a1", "completed", "Standards review"), + init("/work/widgets-issue-7"), + result(20, 0.9), + ], + ); assert!(progress.outcome.as_ref().unwrap().killed_work().is_empty()); } #[test] fn a_task_killed_before_a_later_result_is_not_killed_background_work() { - let (progress, _) = lines(&[ - task_started("b1", "cargo watch"), - task_updated("b1", "killed"), - result(12, 0.4), - init("/work/widgets-issue-7"), - task_notification("b1", "stopped", "cargo watch"), - result(20, 0.9), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + task_started("b1", "cargo watch"), + task_updated("b1", "killed"), + result(12, 0.4), + init("/work/widgets-issue-7"), + task_notification("b1", "stopped", "cargo watch"), + result(20, 0.9), + ], + ); assert!(progress.outcome.as_ref().unwrap().killed_work().is_empty()); } @@ -463,18 +481,21 @@ mod tests { json!({ "type": "result", "subtype": "error_max_turns", "is_error": true }), json!({ "type": "result", "subtype": "success", "is_error": true }), ] { - let (progress, _) = lines(&[ - task_started("b1", "cargo test"), - last, - task_updated("b1", "killed"), - ]); + let (progress, _) = lines( + Harness::Claude, + &[ + task_started("b1", "cargo test"), + last, + task_updated("b1", "killed"), + ], + ); assert!(progress.outcome.is_err()); } } #[test] fn an_init_without_a_session_id_gives_none() { - let (progress, _) = lines(&[init("/a")]); + let (progress, _) = lines(Harness::Claude, &[init("/a")]); assert_eq!( progress.outcome.as_ref().unwrap().session_id.as_deref(), None diff --git a/src/harness/codex/stream.rs b/src/harness/codex/stream.rs index d253b1300..9d64f017d 100644 --- a/src/harness/codex/stream.rs +++ b/src/harness/codex/stream.rs @@ -261,20 +261,10 @@ mod tests { use super::*; use crate::harness::{ Harness, - interpretation::Completion, - interpretation_tests::{finish, stream}, + interpretation_tests::{finish, lines, stream}, }; use serde_json::json; - fn lines(events: &[Value]) -> (Completion, Vec>) { - let mut progress = stream(Harness::Codex, ""); - let lines = events - .iter() - .map(|event| progress.condense(&event.to_string())) - .collect(); - (finish(progress), lines) - } - fn item(kind: &str, id: &str, item: Value) -> Value { let mut item = item; item["id"] = json!(id); @@ -297,10 +287,13 @@ mod tests { #[test] fn the_first_thread_started_starts_the_session_and_gives_its_id() { - let (progress, lines) = lines(&[ - json!({ "type": "thread.started", "thread_id": "019a-thread" }), - json!({ "type": "turn.started" }), - ]); + let (progress, lines) = lines( + Harness::Codex, + &[ + json!({ "type": "thread.started", "thread_id": "019a-thread" }), + json!({ "type": "turn.started" }), + ], + ); assert_eq!(lines, [vec!["session started".to_string()], vec![]]); assert_eq!( @@ -311,27 +304,30 @@ mod tests { #[test] fn a_command_gets_one_line_without_its_shell_naming_commits_and_pushes() { - let (_, lines) = lines(&[ - command( - "item.started", - "item_1", - "/bin/bash -lc 'cargo test'", - "in_progress", - ), - command( - "item.completed", - "item_1", - "/bin/bash -lc 'cargo test'", - "completed", - ), - command( - "item.started", - "item_2", - "bash -lc 'git add -A && git commit -m '\\''x'\\'' && git push'", - "in_progress", - ), - command("item.completed", "item_3", "git push", "completed"), - ]); + let (_, lines) = lines( + Harness::Codex, + &[ + command( + "item.started", + "item_1", + "/bin/bash -lc 'cargo test'", + "in_progress", + ), + command( + "item.completed", + "item_1", + "/bin/bash -lc 'cargo test'", + "completed", + ), + command( + "item.started", + "item_2", + "bash -lc 'git add -A && git commit -m '\\''x'\\'' && git push'", + "in_progress", + ), + command("item.completed", "item_3", "git push", "completed"), + ], + ); assert_eq!( lines, @@ -346,32 +342,35 @@ mod tests { #[test] fn file_changes_tool_calls_and_searches_get_a_line_each() { - let (_, lines) = lines(&[ - item( - "item.completed", - "item_1", - json!({ "type": "file_change", "status": "completed", "changes": [ + let (_, lines) = lines( + Harness::Codex, + &[ + item( + "item.completed", + "item_1", + json!({ "type": "file_change", "status": "completed", "changes": [ { "path": "/work/widgets-issue-7/src/run.rs", "kind": "update" }, { "path": "/work/widgets-issue-7/NOTES.md", "kind": "add" }, { "path": "/etc/hosts", "kind": "delete" } ] }), - ), - item( - "item.started", - "item_2", - json!({ "type": "mcp_tool_call", "server": "github", "tool": "get_issue" }), - ), - item( - "item.started", - "item_3", - json!({ "type": "web_search", "query": "codex exec json" }), - ), - item( - "item.started", - "item_4", - json!({ "type": "collab_tool_call", "tool": "spawn_agent" }), - ), - ]); + ), + item( + "item.started", + "item_2", + json!({ "type": "mcp_tool_call", "server": "github", "tool": "get_issue" }), + ), + item( + "item.started", + "item_3", + json!({ "type": "web_search", "query": "codex exec json" }), + ), + item( + "item.started", + "item_4", + json!({ "type": "collab_tool_call", "tool": "spawn_agent" }), + ), + ], + ); assert_eq!( lines, @@ -390,20 +389,23 @@ mod tests { #[test] fn messages_reasoning_and_turn_events_give_no_line() { - let (_, lines) = lines(&[ - item( - "item.completed", - "item_1", - json!({ "type": "reasoning", "text": "Thinking." }), - ), - item( - "item.completed", - "item_2", - json!({ "type": "agent_message", "text": "Done." }), - ), - json!({ "type": "turn.started" }), - completed(10, 2), - ]); + let (_, lines) = lines( + Harness::Codex, + &[ + item( + "item.completed", + "item_1", + json!({ "type": "reasoning", "text": "Thinking." }), + ), + item( + "item.completed", + "item_2", + json!({ "type": "agent_message", "text": "Done." }), + ), + json!({ "type": "turn.started" }), + completed(10, 2), + ], + ); assert_eq!(lines, [vec![], vec![], vec![], vec![]] as [Vec; 4]); } @@ -417,7 +419,10 @@ mod tests { json!({ "type": "agent_message", "text": text }), ) }; - let (progress, _) = lines(&[said("item_1", "Running tests."), said("item_5", "Done.")]); + let (progress, _) = lines( + Harness::Codex, + &[said("item_1", "Running tests."), said("item_5", "Done.")], + ); assert_eq!( progress.outcome.as_ref().unwrap().final_message.as_deref(), @@ -427,28 +432,31 @@ mod tests { #[test] fn the_summary_is_the_token_totals() { - let (progress, _) = lines(&[completed(12000, 345)]); + let (progress, _) = lines(Harness::Codex, &[completed(12000, 345)]); assert_eq!( progress.report.as_ref().unwrap().summary.clone().as_deref(), Some("12000 input tokens (100 cached), 345 output tokens") ); - assert_eq!(lines(&[]).0.report.unwrap().summary, None); + assert_eq!(lines(Harness::Codex, &[]).0.report.unwrap().summary, None); } #[test] fn a_command_still_running_when_the_turn_completes_is_killed_background_work() { - let (progress, _) = lines(&[ - command( - "item.started", - "item_1", - "/bin/bash -lc 'sleep 188'", - "in_progress", - ), - command("item.started", "item_2", "cargo build", "in_progress"), - command("item.completed", "item_2", "cargo build", "completed"), - completed(10, 2), - ]); + let (progress, _) = lines( + Harness::Codex, + &[ + command( + "item.started", + "item_1", + "/bin/bash -lc 'sleep 188'", + "in_progress", + ), + command("item.started", "item_2", "cargo build", "in_progress"), + command("item.completed", "item_2", "cargo build", "completed"), + completed(10, 2), + ], + ); assert_eq!( progress.outcome.as_ref().unwrap().killed_work(), @@ -459,11 +467,14 @@ mod tests { #[test] fn a_command_reconciled_as_in_progress_at_turn_end_is_killed_background_work() { - let (progress, _) = lines(&[ - command("item.started", "item_1", "npm run dev", "in_progress"), - command("item.completed", "item_1", "npm run dev", "in_progress"), - completed(10, 2), - ]); + let (progress, _) = lines( + Harness::Codex, + &[ + command("item.started", "item_1", "npm run dev", "in_progress"), + command("item.completed", "item_1", "npm run dev", "in_progress"), + completed(10, 2), + ], + ); assert_eq!( progress.outcome.as_ref().unwrap().killed_work(), @@ -480,19 +491,22 @@ mod tests { json!({ "type": "collab_tool_call", "tool": tool, "prompt": prompt, "status": status }), ) }; - let (progress, _) = lines(&[ - call( - "item.started", - "item_1", - "spawn_agent", - "Review the diff\nThen report.", - "in_progress", - ), - call("item.started", "item_2", "wait", "", "in_progress"), - call("item.completed", "item_2", "wait", "", "completed"), - call("item.started", "item_3", "spawn_agent", "", "in_progress"), - completed(10, 2), - ]); + let (progress, _) = lines( + Harness::Codex, + &[ + call( + "item.started", + "item_1", + "spawn_agent", + "Review the diff\nThen report.", + "in_progress", + ), + call("item.started", "item_2", "wait", "", "in_progress"), + call("item.completed", "item_2", "wait", "", "completed"), + call("item.started", "item_3", "spawn_agent", "", "in_progress"), + completed(10, 2), + ], + ); assert_eq!( progress.outcome.as_ref().unwrap().killed_work(), @@ -502,11 +516,14 @@ mod tests { #[test] fn a_failed_turn_fails_with_its_error_and_kills_no_background_work() { - let (progress, _) = lines(&[ - command("item.started", "item_1", "sleep 188", "in_progress"), - json!({ "type": "error", "message": "Reconnecting... 1/5" }), - json!({ "type": "turn.failed", "error": { "message": "The 'GPT-6.1-Sol' model is not supported." } }), - ]); + let (progress, _) = lines( + Harness::Codex, + &[ + command("item.started", "item_1", "sleep 188", "in_progress"), + json!({ "type": "error", "message": "Reconnecting... 1/5" }), + json!({ "type": "turn.failed", "error": { "message": "The 'GPT-6.1-Sol' model is not supported." } }), + ], + ); assert!(progress.outcome.is_err()); assert_eq!( diff --git a/src/harness/interpretation_tests.rs b/src/harness/interpretation_tests.rs index 9e8b3082c..fef3d9f46 100644 --- a/src/harness/interpretation_tests.rs +++ b/src/harness/interpretation_tests.rs @@ -39,6 +39,18 @@ pub(crate) fn finish(stream: interpretation::Interpretation) -> interpretation:: stream.finish(Ok(std::process::ExitStatus::from_raw(0))) } +pub(crate) fn lines( + harness: Harness, + events: &[serde_json::Value], +) -> (interpretation::Completion, Vec>) { + let mut interpretation = stream(harness, ""); + let lines = events + .iter() + .map(|event| interpretation.condense(&event.to_string())) + .collect(); + (finish(interpretation), lines) +} + #[test] fn every_harness_accepts_missing_final_data_without_inventing_usage_or_a_session_id() { for harness in Harness::ALL { From e66a89b94da98fcfb8e903c473405931570ceb2d Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 6 Oct 2026 18:05:04 -0400 Subject: [PATCH 3/3] Preserve check interruption wording after Spec review --- src/harness/interpretation.rs | 4 +-- src/harness/interpretation_tests.rs | 45 +++++++++++++++++++++++++++++ src/harness/muse.rs | 6 ++-- src/harness/opencode.rs | 6 ++-- tests/fakes/opencode.rs | 9 ++++-- tests/process_tree.rs | 28 ++++++++++++++++-- 6 files changed, 86 insertions(+), 12 deletions(-) diff --git a/src/harness/interpretation.rs b/src/harness/interpretation.rs index cbcdffe6b..5b4bc1d2d 100644 --- a/src/harness/interpretation.rs +++ b/src/harness/interpretation.rs @@ -167,7 +167,7 @@ impl Interpretation { /// Captured checks use exactly the same validation and retained policy, /// but keep their own refusal context and raw-output fallback. - pub(super) fn check_output(mut self, output: &Output) -> Result<()> { + pub(super) fn check_output(mut self, output: &Output, refusal: &str) -> Result<()> { for line in String::from_utf8_lossy(&output.stdout).lines() { self.condense(line); } @@ -175,7 +175,7 @@ impl Interpretation { outcome.map(drop).map_err(|failure| match failure { Failure::Execution(error) => error, Failure::Rejected { diagnostic, .. } => { - anyhow!(diagnostic.unwrap_or_else(|| said(output))) + anyhow!("{refusal}: {}", diagnostic.unwrap_or_else(|| said(output))) } }) } diff --git a/src/harness/interpretation_tests.rs b/src/harness/interpretation_tests.rs index fef3d9f46..a8aebd596 100644 --- a/src/harness/interpretation_tests.rs +++ b/src/harness/interpretation_tests.rs @@ -225,3 +225,48 @@ fn a_transport_failure_preserves_recovered_usage_and_warnings_ahead_of_turn_fail }, ); } + +#[test] +fn an_interrupted_opencode_check_export_returns_interrupted_without_refusal_context() { + use std::os::unix::fs::PermissionsExt; + if std::env::var_os("THIRDSHIFT_TEST_CHECK_EXPORT_INTERRUPT").is_some() { + crate::interrupt::install().unwrap(); + let error = opencode::check_model_and_effort(&ModelAndEffort::default()).unwrap_err(); + assert_eq!(format!("{error:#}"), "interrupted"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let cli = dir.path().join("opencode"); + // A local CLI records a session, then interrupts its exact owner during + // export. The process owner must stop and join it before the check returns. + std::fs::write( + &cli, + r#"#!/bin/bash +if test "$1" = run; then + cat >/dev/null + echo '{"type":"text","sessionID":"s1","part":{"text":"OK"}}' +else + kill -INT "$PPID" + sleep 30 +fi +"#, + ) + .unwrap(); + std::fs::set_permissions(&cli, std::fs::Permissions::from_mode(0o755)).unwrap(); + let path = std::env::join_paths( + std::iter::once(dir.path().to_path_buf()) + .chain(std::env::split_paths(&std::env::var_os("PATH").unwrap())), + ) + .unwrap(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "harness::interpretation_tests::an_interrupted_opencode_check_export_returns_interrupted_without_refusal_context"]) + .env("THIRDSHIFT_TEST_CHECK_EXPORT_INTERRUPT", "1") + .env("PATH", path) + .output().unwrap(); + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} diff --git a/src/harness/muse.rs b/src/harness/muse.rs index 9b1325950..6e93839ae 100644 --- a/src/harness/muse.rs +++ b/src/harness/muse.rs @@ -146,8 +146,10 @@ pub fn check_model_and_effort(chosen: &ModelAndEffort) -> Result Box::new(stream::MuseProgress::for_prompt("Reply with OK.")), Retained::None, ) - .check_output(&output) - .with_context(|| format!("muse refused a test call on the Model {model}"))?; + .check_output( + &output, + &format!("muse refused a test call on the Model {model}"), + )?; } } Ok(settled) diff --git a/src/harness/opencode.rs b/src/harness/opencode.rs index 065cc4b4c..8f5c3ea8e 100644 --- a/src/harness/opencode.rs +++ b/src/harness/opencode.rs @@ -127,8 +127,10 @@ pub fn check_model_and_effort(chosen: &ModelAndEffort) -> Result<()> { .context("could not run opencode to check the Model and Effort")?; OpenCode .interpretation(Path::new("."), "Reply with OK.") - .check_output(&output) - .context("opencode refused the Model or Effort in its test call")?; + .check_output( + &output, + "opencode refused the Model or Effort in its test call", + )?; Ok(()) } diff --git a/tests/fakes/opencode.rs b/tests/fakes/opencode.rs index c9df60425..a6e511e80 100644 --- a/tests/fakes/opencode.rs +++ b/tests/fakes/opencode.rs @@ -28,9 +28,12 @@ pub fn main(argv: Vec) { println!("truncated {{"); crate::exit(0); } - if argv.last().is_some_and(|id| id != "check-session") - && let Ok(script) = std::env::var("FAKE_OPENCODE_EXPORT_SCRIPT") - { + let export_script = if argv.last().is_some_and(|id| id == "check-session") { + "FAKE_OPENCODE_CHECK_EXPORT_SCRIPT" + } else { + "FAKE_OPENCODE_EXPORT_SCRIPT" + }; + if let Ok(script) = std::env::var(export_script) { outlast_interrupts(); let status = Command::new("bash").arg("-e").arg(script).status().unwrap(); if !status.success() { diff --git a/tests/process_tree.rs b/tests/process_tree.rs index 021329028..7d5f623d8 100644 --- a/tests/process_tree.rs +++ b/tests/process_tree.rs @@ -89,6 +89,7 @@ enum During { Session, Check, CheckAfterExit, + CheckExport, Export, } @@ -189,6 +190,11 @@ fn interrupting_an_opencode_export_stops_its_whole_process_tree() { assert_session_command_stopped("opencode", During::Export); } +#[test] +fn interrupting_an_opencode_check_export_stops_its_tree_without_a_model_refusal() { + assert_session_command_stopped("opencode", During::CheckExport); +} + #[test] fn interrupting_a_grok_run_stops_a_command_in_its_own_session() { assert_session_command_stopped("grok", During::Session); @@ -202,7 +208,10 @@ fn interrupting_a_muse_run_stops_a_command_in_its_own_session() { fn assert_session_command_stopped(harness: &str, during: During) { let scenario = Scenario::new(); let pid = scenario.path("detached-pid"); - let checking = matches!(during, During::Check | During::CheckAfterExit); + let checking = matches!( + during, + During::Check | During::CheckAfterExit | During::CheckExport + ); let ending = if matches!(during, During::Session) { "while :; do sleep 0.1; done" } else { @@ -251,9 +260,14 @@ touch "{started}" } env } - During::Export => { + During::Export | During::CheckExport => { scenario.agent_does("true"); - vec![("FAKE_OPENCODE_EXPORT_SCRIPT", script.to_str().unwrap())] + let name = if matches!(during, During::CheckExport) { + "FAKE_OPENCODE_CHECK_EXPORT_SCRIPT" + } else { + "FAKE_OPENCODE_EXPORT_SCRIPT" + }; + vec![(name, script.to_str().unwrap())] } }; let url = scenario.issue_url(7); @@ -275,6 +289,14 @@ touch "{started}" assert!(scenario.issue_labels(7).is_empty()); assert_eq!(scenario.entries("work"), ["widgets"]); } + if matches!(during, During::CheckExport) { + assert!( + result.stderr.ends_with("thirdshift: interrupted\n"), + "{}", + result.stderr + ); + assert!(!result.stderr.contains("refused"), "{}", result.stderr); + } if matches!(during, During::Export) { assert!( !result.stderr.contains("session ended after"),