diff --git a/src/harness/agy.rs b/src/harness/agy.rs index 1b8193a7d..519f2fe29 100644 --- a/src/harness/agy.rs +++ b/src/harness/agy.rs @@ -4,7 +4,7 @@ pub(crate) mod stream; pub use catalog::Catalog; use super::adapter::{Adapter, Invocation, SkillLoading, without}; -use super::interpretation::{Interpretation, Retained}; +use super::interpretation::Interpretation; use super::settings::{Terminal, ask_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; use crate::progress; @@ -142,11 +142,7 @@ impl Adapter for Agy { super::instructions::link_fallback(worktree, "GEMINI.md", &["AGENTS.md"]) } fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation { - Interpretation::new( - self.name(), - Box::new(stream::AgyProgress::default()), - Retained::None, - ) + Interpretation::new(self.name(), Box::new(stream::AgyProgress::default())) } } diff --git a/src/harness/claude.rs b/src/harness/claude.rs index 664224545..3d41f5b92 100644 --- a/src/harness/claude.rs +++ b/src/harness/claude.rs @@ -6,7 +6,7 @@ use std::process::Command; use anyhow::{Context, Result, bail}; use super::adapter::{Adapter, Invocation, SkillLoading}; -use super::interpretation::{Interpretation, Retained, Stream}; +use super::interpretation::{Interpretation, Stream}; use super::settings::{Terminal, ask_setting}; use super::{Choice, Harness, ModelAndEffort, Settings, said}; use crate::progress; @@ -67,7 +67,7 @@ impl Adapter for Claude { ask_claude(outside, current).map(Some) } fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation { - Interpretation::new(self.name(), Box::new(Stream::claude()), Retained::None) + Interpretation::new(self.name(), Box::new(Stream::claude())) } } diff --git a/src/harness/codex.rs b/src/harness/codex.rs index 7d8be32f3..347d95a59 100644 --- a/src/harness/codex.rs +++ b/src/harness/codex.rs @@ -9,7 +9,7 @@ use std::path::Path; use anyhow::Result; use super::adapter::{Adapter, Invocation, SkillLoading}; -use super::interpretation::{Interpretation, Retained}; +use super::interpretation::Interpretation; use super::settings::{Terminal, ask_checked_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; use crate::progress; @@ -132,7 +132,6 @@ impl Adapter for Codex { Interpretation::new( self.name(), Box::new(stream::CodexProgress::in_worktree(worktree)), - Retained::None, ) } } diff --git a/src/harness/grok.rs b/src/harness/grok.rs index 7f52d32ae..f6d6fa300 100644 --- a/src/harness/grok.rs +++ b/src/harness/grok.rs @@ -8,7 +8,7 @@ use std::path::Path; use anyhow::Result; use super::adapter::{Adapter, Invocation, SkillLoading, without}; -use super::interpretation::{Interpretation, Retained, Stream}; +use super::interpretation::{Interpretation, Stream}; use super::settings::{Terminal, ask_checked_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; @@ -139,7 +139,7 @@ impl Adapter for Grok { Ok(Some(ModelAndEffort { model, effort })) } fn interpretation(&self, _worktree: &Path, _prompt: &str) -> Interpretation { - Interpretation::new(self.name(), Box::new(Stream::grok()), Retained::None) + Interpretation::new(self.name(), Box::new(Stream::grok())) } } diff --git a/src/harness/interpretation.rs b/src/harness/interpretation.rs index b289679ac..938bb170b 100644 --- a/src/harness/interpretation.rs +++ b/src/harness/interpretation.rs @@ -1,11 +1,10 @@ //! 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 super::said; use crate::interrupt; mod stream; @@ -20,7 +19,7 @@ use security::Security; pub struct Interpretation { cli: &'static str, decoder: Box, - retained: Retained, + retained: Option>, security: Option, models: models::Models, } @@ -33,12 +32,25 @@ pub struct Completion { #[derive(Default)] pub struct Report { + /// Ordered progress supplied after retained completion. + pub completion_progress: Vec, pub warnings: Vec, pub summary: Option, /// Models reported by session output or retained records. pub models: Vec, } +impl Report { + /// Retained adapters opt into completion Model progress after reconciliation. + pub(super) fn set_model_progress(&mut self) { + self.completion_progress = self + .models + .iter() + .map(|model| format!("Model: {model}")) + .collect(); + } +} + #[derive(Debug)] pub struct Ended { pub session_id: Option, @@ -101,46 +113,10 @@ pub(super) enum ModelScope { AttributionOnly, } -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; - } - facts.report.models = log.models; - } - } - 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; - facts.report.models = export.models; - if export.failure.is_some() { - facts.outcome = TurnOutcome::Failed; - facts.diagnostic = facts.diagnostic.take().or(export.failure); - } - } - } - } - Ok(()) - } +/// Optional retained reconciliation belongs to the supplying Harness adapter. +/// Owned implementations follow Interpretation into the transcript reader. +pub(super) trait Retained: Send { + fn reconcile(self: Box, facts: &mut Facts) -> Result<()>; } enum Failure { @@ -185,16 +161,21 @@ impl Failure { } impl Interpretation { - pub(super) fn new(cli: &'static str, decoder: Box, retained: Retained) -> Self { + pub(super) fn new(cli: &'static str, decoder: Box) -> Self { Self { cli, decoder, - retained, + retained: None, security: None, models: models::Models::default(), } } + pub(super) fn with_retained(mut self, retained: Box) -> Self { + self.retained = Some(retained); + self + } + /// Apply refusal and Model reporting rules independently of the session's log label. pub fn for_security(mut self, requested_model: Option<&str>) -> Self { self.security = Some(Security::new(requested_model)); @@ -262,7 +243,9 @@ impl Interpretation { if refusal.is_some() { facts.outcome = TurnOutcome::Failed; } - let recovered = self.retained.reconcile(&mut facts); + let recovered = self + .retained + .map_or(Ok(()), |retained| retained.reconcile(&mut facts)); if let Err(error) = interrupt::check() { return (None, Err(Failure::Execution(error))); } diff --git a/src/harness/interpretation_tests.rs b/src/harness/interpretation_tests.rs index 55e9c0cbd..ffbe4b0c5 100644 --- a/src/harness/interpretation_tests.rs +++ b/src/harness/interpretation_tests.rs @@ -1,18 +1,6 @@ use super::*; use std::os::unix::process::ExitStatusExt; -#[test] -fn agy_security_progress_names_the_resolved_model() { - let mut interpretation = stream(Harness::Agy, "").for_security(Some("gemini-3.8-flash")); - let lines = interpretation.condense( - r#"{"event":"init","conversation_id":"s1","init":{"model":"gemini-3.8-flash-high"}}"#, - ); - assert!( - lines.contains(&"Model: gemini-3.8-flash-high".to_string()), - "{lines:?}" - ); -} - #[test] fn a_security_session_can_discuss_the_cyber_marker_without_being_refused() { let mut interpretation = stream(Harness::Claude, "").for_security(None); @@ -347,16 +335,12 @@ pub(crate) fn recording_interpretation( root: &Path, retained_harness: Harness, ) -> interpretation::Interpretation { - let retained = match retained_harness { - Harness::Muse => interpretation::Retained::Muse(Some(root.to_path_buf())), - Harness::OpenCode => interpretation::Retained::OpenCode(root.to_path_buf()), + let decoder = Box::new(interpretation::Stream::claude()); + match retained_harness { + Harness::Muse => muse::log::recording_interpretation(root, decoder), + Harness::OpenCode => opencode::export::recording_interpretation(root, decoder), _ => panic!("recording fixtures support only retained Harnesses"), - }; - interpretation::Interpretation::new( - "claude", - Box::new(interpretation::Stream::claude()), - retained, - ) + } } #[test] @@ -671,3 +655,180 @@ fn retained_harnesses_do_not_invent_live_model_or_safeguard_evidence() { assert!(completion.report.unwrap().models.is_empty()); } } + +#[test] +fn muse_completion_reports_retained_model_order_and_multiplicity() { + with_muse_data( + "muse_completion_reports_retained_model_order_and_multiplicity", + |root| { + muse_log(root, &[ + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"first"}}}"#, + r#"{"stream":{"id":"child"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"child"}}}"#, + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"first"}}}"#, + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"second"}}}"#, + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"first"}}}"#, + ].join("\n")); + let report = finish(muse_reply()).report.unwrap(); + assert_eq!(report.models, ["first", "second", "first"]); + assert_eq!( + report.completion_progress, + ["Model: first", "Model: second", "Model: first"] + ); + }, + ); +} + +fn with_opencode_export(test: &str, run: impl FnOnce(&Path)) { + const ROOT: &str = "THIRDSHIFT_TEST_OPENCODE_EXPORT"; + if let Some(root) = std::env::var_os(ROOT) { + run(Path::new(&root)); + return; + } + let root = tempfile::tempdir().unwrap(); + crate::test_support::write_executable( + &root.path().join("opencode"), + "#!/bin/sh\ncat \"$THIRDSHIFT_TEST_OPENCODE_EXPORT/export.json\"\n", + ); + let path = std::env::join_paths( + std::iter::once(root.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", &format!("harness::interpretation_tests::{test}")]) + .env(ROOT, root.path()) + .env("PATH", path) + .output() + .unwrap(); + assert!(output.status.success(), "{output:?}"); +} + +#[test] +fn opencode_completion_reports_retained_model_order_and_multiplicity() { + with_opencode_export( + "opencode_completion_reports_retained_model_order_and_multiplicity", + |root| { + std::fs::write( + root.join("export.json"), + r#"{"info":{"outcome":"succeeded"},"messages":[ + {"type":"assistant","model":{"providerID":"provider","id":"first"},"content":[]}, + {"type":"user","model":{"id":"user"}}, + {"type":"assistant","model":{"providerID":"provider","id":"first"},"content":[]}, + {"type":"assistant","model":{"providerID":"provider","id":"second"},"content":[]}, + {"type":"assistant","model":{"providerID":"provider","id":"first"},"content":[]} + ]}"#, + ) + .unwrap(); + let mut interpretation = Harness::OpenCode.adapter().interpretation(root, ""); + interpretation + .condense(r#"{"type":"text","sessionID":"s1","part":{"text":"stream reply"}}"#); + let completion = finish(interpretation); + assert!(completion.outcome.is_ok()); + let report = completion.report.unwrap(); + assert_eq!( + report.models, + ["provider/first", "provider/second", "provider/first"] + ); + assert_eq!( + report.completion_progress, + [ + "Model: provider/first", + "Model: provider/second", + "Model: provider/first" + ] + ); + }, + ); +} + +#[test] +fn stream_only_harnesses_leave_completion_progress_empty_after_live_model_progress() { + for (harness, event) in [ + ( + Harness::Claude, + r#"{"type":"assistant","message":{"model":"resolved-model","content":[]}}"#, + ), + ( + Harness::Grok, + r#"{"type":"assistant","message":{"model":"resolved-model","content":[]}}"#, + ), + ( + Harness::Agy, + r#"{"event":"init","conversation_id":"s1","init":{"model":"gemini-3.8-flash-high"}}"#, + ), + ( + Harness::Codex, + r#"{"type":"thread.started","thread_id":"s1"}"#, + ), + ] { + let mut interpretation = stream(harness, "").for_security(Some("requested-model")); + let live = interpretation.condense(event); + assert!( + live.iter().any(|line| line.starts_with("Model:")), + "{harness:?}: {live:?}" + ); + let completion = finish(interpretation); + assert!(completion.outcome.is_ok()); + assert!( + completion.report.unwrap().completion_progress.is_empty(), + "{harness:?}" + ); + } +} + +#[test] +fn retained_completion_reports_stream_fallback_then_authoritative_empty_models_and_usage() { + with_opencode_export( + "retained_completion_reports_stream_fallback_then_authoritative_empty_models_and_usage", + |root| { + for harness in [Harness::Muse, Harness::OpenCode] { + let file = if harness == Harness::Muse { + let dir = root.join("sessions/s1"); + std::fs::create_dir_all(&dir).unwrap(); + dir.join("session.jsonl") + } else { + root.join("export.json") + }; + let empty = if harness == Harness::Muse { + "" + } else { + r#"{"info":{"outcome":"succeeded"},"messages":[]}"# + }; + for retained in [None, Some("not JSON"), Some(empty)] { + if let Some(retained) = retained { + std::fs::write(&file, retained).unwrap(); + } + let mut interpretation = recording_interpretation(root, harness); + for event in [ + r#"{"type":"system","subtype":"init","session_id":"s1"}"#, + r#"{"type":"assistant","message":{"model":"stream-model","content":[]}}"#, + r#"{"type":"result","subtype":"success","result":"stream reply","num_turns":2,"total_cost_usd":0.25}"#, + ] { + interpretation.condense(event); + } + let completion = finish(interpretation); + let ended = completion.outcome.unwrap(); + let report = completion.report.unwrap(); + let readable = retained == Some(empty); + assert_eq!( + ended.final_message.as_deref(), + if readable && harness == Harness::OpenCode { + None + } else { + Some("stream reply") + } + ); + if readable { + assert!(report.models.is_empty()); + assert!(report.completion_progress.is_empty()); + assert_eq!(report.summary, None); + } else { + assert_eq!(report.models, ["stream-model"]); + assert_eq!(report.completion_progress, ["Model: stream-model"]); + assert_eq!(report.summary.as_deref(), Some("2 turns, $0.25")); + } + } + } + }, + ); +} diff --git a/src/harness/muse.rs b/src/harness/muse.rs index 600a9a1a8..099f84249 100644 --- a/src/harness/muse.rs +++ b/src/harness/muse.rs @@ -10,7 +10,7 @@ use std::process::Command; use anyhow::{Context, Result}; use super::adapter::{Adapter, Invocation, SkillLoading, without}; -use super::interpretation::{Interpretation, Retained}; +use super::interpretation::Interpretation; use super::settings::{Terminal, ask_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; @@ -124,8 +124,8 @@ impl Adapter for Muse { Interpretation::new( self.name(), Box::new(stream::MuseProgress::for_prompt(prompt)), - Retained::Muse(data_dir()), ) + .with_retained(Box::new(log::Completion::new(data_dir()))) } } @@ -168,7 +168,6 @@ pub fn check_model_and_effort(chosen: &ModelAndEffort) -> Result Interpretation::new( Muse.name(), Box::new(stream::MuseProgress::for_prompt("Reply with OK.")), - Retained::None, ) .check_output( &output, diff --git a/src/harness/muse/log.rs b/src/harness/muse/log.rs index 7d997fcb7..538c3bf23 100644 --- a/src/harness/muse/log.rs +++ b/src/harness/muse/log.rs @@ -1,5 +1,6 @@ //! Read Muse's retained session records for the last reply and token usage. +use crate::harness::interpretation::{Facts, Retained}; use anyhow::Result; use serde_json::Value; use std::fs; @@ -104,3 +105,42 @@ pub(in crate::harness) fn read(root: &Path, id: &str) -> Option { let path = find_log(&root.join("sessions"), id, 3)?; SessionLog::parse(&fs::read_to_string(path).ok()?, id).ok() } + +/// Muse's readable records own usage and Models, but text only when present. +pub(super) struct Completion { + root: Option, +} + +impl Completion { + pub(super) fn new(root: Option) -> Self { + Self { root } + } +} + +impl Retained for Completion { + fn reconcile(self: Box, facts: &mut Facts) -> Result<()> { + if let Some(log) = self + .root + .as_deref() + .zip(facts.ended.session_id.as_deref()) + .and_then(|(root, id)| read(root, id)) + { + facts.report.summary = log.summary(); + if log.message.is_some() { + facts.ended.final_message = log.message; + } + facts.report.models = log.models; + } + facts.report.set_model_progress(); + Ok(()) + } +} + +#[cfg(test)] +pub(in crate::harness) fn recording_interpretation( + root: &Path, + decoder: Box, +) -> crate::harness::interpretation::Interpretation { + crate::harness::interpretation::Interpretation::new("claude", decoder) + .with_retained(Box::new(Completion::new(Some(root.to_path_buf())))) +} diff --git a/src/harness/opencode.rs b/src/harness/opencode.rs index b485c95a1..f834ba83a 100644 --- a/src/harness/opencode.rs +++ b/src/harness/opencode.rs @@ -3,7 +3,7 @@ pub(super) mod export; mod stream; use super::adapter::{Adapter, Invocation, SkillLoading, without}; -use super::interpretation::{Interpretation, Retained}; +use super::interpretation::Interpretation; use super::settings::{Terminal, ask_setting}; use super::{Choice, Harness, ModelAndEffort, Settings}; use anyhow::{Context, Result, bail}; @@ -107,8 +107,8 @@ impl Adapter for OpenCode { Interpretation::new( self.name(), Box::new(stream::OpenCodeProgress::for_prompt(prompt)), - Retained::OpenCode(worktree.to_path_buf()), ) + .with_retained(Box::new(export::Completion::new(worktree.to_path_buf()))) } fn link_instruction_fallback(&self, worktree: &Path) -> Result<()> { super::instructions::link_fallback(worktree, "AGENTS.md", &[]) diff --git a/src/harness/opencode/export.rs b/src/harness/opencode/export.rs index 50d76ad6b..f2bfbd3d8 100644 --- a/src/harness/opencode/export.rs +++ b/src/harness/opencode/export.rs @@ -2,9 +2,10 @@ //! survive dropped stream events. Session totals also count the title call. use super::OpenCode; use crate::harness::Adapter; +use crate::harness::interpretation::{Facts, Retained, TurnOutcome}; use anyhow::{Result, anyhow}; use serde_json::Value; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::process::Command; pub(in crate::harness) struct SessionExport { @@ -124,3 +125,41 @@ pub(in crate::harness) fn read(worktree: &Path, id: &str) -> Result Ok(None), } } + +/// A readable standalone export owns completion, including empty fields. +pub(super) struct Completion { + worktree: PathBuf, +} + +impl Completion { + pub(super) fn new(worktree: PathBuf) -> Self { + Self { worktree } + } +} + +impl Retained for Completion { + fn reconcile(self: Box, facts: &mut Facts) -> Result<()> { + if let Some(id) = facts.ended.session_id.as_deref() + && let Some(export) = read(&self.worktree, id)? + { + facts.ended.final_message = export.message; + facts.report.summary = export.summary; + facts.report.models = export.models; + if export.failure.is_some() { + facts.outcome = TurnOutcome::Failed; + facts.diagnostic = facts.diagnostic.take().or(export.failure); + } + } + facts.report.set_model_progress(); + Ok(()) + } +} + +#[cfg(test)] +pub(in crate::harness) fn recording_interpretation( + root: &Path, + decoder: Box, +) -> crate::harness::interpretation::Interpretation { + crate::harness::interpretation::Interpretation::new("claude", decoder) + .with_retained(Box::new(Completion::new(root.to_path_buf()))) +} diff --git a/src/session/recording.rs b/src/session/recording.rs index d98984fa3..c03cabba4 100644 --- a/src/session/recording.rs +++ b/src/session/recording.rs @@ -107,15 +107,8 @@ impl Recording { let completion = executed.state.finish(executed.execution); if let Some(report) = completion.report { self.keep(kind, log, choice, report.models.clone()); - // Stream models already have security progress lines. Retained records - // need theirs here, as before. - if matches!( - choice.harness, - crate::harness::Harness::Muse | crate::harness::Harness::OpenCode - ) { - for model in report.models { - progress::step(format_args!("{kind}: Model: {model}")); - } + for line in report.completion_progress { + progress::step(format_args!("{kind}: {line}")); } for warning in report.warnings { progress::step(format_args!("{kind}: {warning}")); diff --git a/src/session/recording/tests.rs b/src/session/recording/tests.rs index 8e597868c..c7f22a893 100644 --- a/src/session/recording/tests.rs +++ b/src/session/recording/tests.rs @@ -479,3 +479,85 @@ fn startup_failure_keeps_its_cause_without_writing_evidence() { ); assert!(fixture.recording.notification_lines().is_empty()); } +#[test] +fn completion_progress_comes_from_interpretation_without_duplicate_live_lines() { + const CHILD: &str = "THIRDSHIFT_RECORDING_PROGRESS"; + if std::env::var_os(CHILD).is_some() { + let fixture = Fixture::new(); + let dir = fixture.root.path().join("sessions/s1"); + fs::create_dir_all(&dir).unwrap(); + fs::write(dir.join("session.jsonl"), [ + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"first"}}}"#, + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"second"}}}"#, + r#"{"stream":{"id":"s1"},"payload":{"kind":"run","event":{"kind":"model_completed","model":"first"}}}"#, + ].join("\n")).unwrap(); + // Choice controls execution and metadata; Interpretation supplies reporting. + fixture + .recording + .run( + "retained", + &Choice::default(), + Command::new("/bin/sh").args(["-c", &format!("echo '{INIT}'; echo '{MODEL}'")]), + None, + &fixture.log("retained"), + crate::harness::interpretation_tests::recording_interpretation( + fixture.root.path(), + Harness::Muse, + ), + ) + .unwrap(); + fs::write(dir.join("session.jsonl"), "").unwrap(); + fixture + .recording + .run( + "empty", + &Choice::default(), + Command::new("/bin/sh").args(["-c", &format!("echo '{INIT}'; echo '{MODEL}'")]), + None, + &fixture.log("empty"), + crate::harness::interpretation_tests::recording_interpretation( + fixture.root.path(), + Harness::Muse, + ), + ) + .unwrap(); + fixture + .recording + .run( + "live", + &Choice::default(), + Command::new("/bin/sh").args(["-c", &format!("echo '{MODEL}'; echo '{MODEL}'")]), + None, + &fixture.log("live"), + Harness::Claude + .adapter() + .interpretation(fixture.root.path(), "") + .for_security(None), + ) + .unwrap(); + return; + } + let output = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "session::recording::tests::completion_progress_comes_from_interpretation_without_duplicate_live_lines", "--nocapture"]) + .env(CHILD, "1").output().unwrap(); + assert!(output.status.success(), "{output:?}"); + let stderr = String::from_utf8_lossy(&output.stderr); + let models: Vec<_> = stderr + .lines() + .filter_map(|line| match progress::ChildLine::of(line) { + progress::ChildLine::Own(message) if message.contains(": Model:") => Some(message), + _ => None, + }) + .collect(); + assert_eq!( + models, + [ + "retained: Model: first", + "retained: Model: second", + "retained: Model: first", + "live: Model: reported-model" + ], + "{stderr}" + ); + assert_eq!(stderr.matches("session ended after").count(), 3, "{stderr}"); +}