Deepen completed Harness session interpretation

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;

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<Option<ModelAndEffort>>;
-    fn stream(&self, worktree: &Path, prompt: &str) -> Box<dyn Stream>;
+    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<dyn Stream>,
-    ) -> Result<Box<dyn Stream>> {
-        Ok(stream)
-    }
 }
 
 /// A session's arguments and optional prompt on stdin. `None` keeps stdin

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<dyn Stream> {
-        Box::new(stream::AgyProgress::default())
+    fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation {
+        Interpretation::new(
+            self.name(),
+            Box::new(stream::AgyProgress::default()),
+            Retained::None,
+        )
     }
 }
 

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<Value>,
 }
 
-impl Stream for AgyProgress {
+impl Decoder for AgyProgress {
     fn condense(&mut self, raw: &str) -> Vec<String> {
         let Ok(event) = serde_json::from_str::<Value>(raw) else {
             return Vec::new();
@@ -67,6 +68,31 @@ impl Stream for AgyProgress {
             _ => Vec::new(),
         }
     }
+    fn complete(self: Box<Self>) -> 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<String> {
         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,61 +115,73 @@ 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"}}}}"#;
     const RESULT: &str = r#"{"event":"result","result":{"conversation_id":"recorded-conversation","status":"SUCCESS","response":"Final reply only.","num_turns":2,"usage":{"input_tokens":12518,"output_tokens":29,"thinking_tokens":28,"cache_read_tokens":200,"total_tokens":12547}}}"#;
 
     #[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"]);
         assert!(stream.condense(TOOL).is_empty());
         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);
     }
 }

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<Option<ModelAndEffort>> {
         ask_claude(outside, current).map(Some)
     }
-    fn stream(&self, _worktree: &Path, _prompt: &str) -> Box<dyn Stream> {
-        Box::new(stream::ClaudeProgress::default())
+    fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation {
+        Interpretation::new(
+            self.name(),
+            Box::new(stream::ClaudeProgress::default()),
+            Retained::None,
+        )
     }
 }