Preserve Session completion replies after background notifications

site/how-it-works/index.html

@@ -100,6 +100,7 @@ <h1>How thirdshift works</h1>
       <summary class="sec-head"><p class="eyebrow">03 / What the agent does</p><h2 id="session-title">Let the agent implement and review</h2></summary>
       <div class="guide-content" id="session">
       <p>thirdshift starts a headless session in the selected Harness, in the Worktree. The <strong>Session prompt</strong> supplies the issue, Base branch, Issue branch and review instructions. <strong>Factory skills</strong> are linked where that Harness discovers them. The opening session carries the change from implementation through review to a committed branch and pull request.</p>
+      <p>On Claude and Grok, a Session's completion line survives later background-task notifications. A later reply carrying a completion line takes precedence.</p>
       <figure class="guide-diagram grid-bg">
         <figcaption>Inside the opening agent session</figcaption>
         <ol class="guide-flow">

src/harness/interpretation/stream.rs

@@ -2,7 +2,8 @@
 //!
 //! A result ends a turn, not necessarily the session: background sub-agents
 //! can resume it. Only the first init sets shared session fields, and only
-//! completion supplies the last result and any work killed since that result.
+//! completion supplies the last reply holding a completion line (otherwise
+//! the last result) and any work killed since the latest result.
 
 use std::collections::{BTreeMap, HashMap};
 
@@ -13,6 +14,7 @@ use super::{
     TurnOutcome,
 };
 use crate::progress::{bash, shorten};
+use crate::prompt;
 mod events;
 
 pub struct Stream {
@@ -21,6 +23,7 @@ pub struct Stream {
     /// Cumulative totals survive later results without a complete pair.
     turns_and_cost: Option<(u64, f64)>,
     final_message: Option<String>,
+    completion_message: Option<String>,
     completed: bool,
     dialect: Dialect,
 }
@@ -118,6 +121,31 @@ fn message_text(event: &Value) -> Option<String> {
     (!text.is_empty()).then(|| text.join("\n"))
 }
 
+/// Recognize handoffs here, leaving their payload validation to each reader.
+/// Keep the whole reply: Architecture reviews also carry a strength line.
+fn has_completion_line(message: &str) -> bool {
+    let line = message.trim_end().lines().last().unwrap_or("").trim();
+    matches!(
+        line,
+        prompt::BUG_REPRODUCED_LINE
+            | prompt::AUDIT_COMPLETE_LINE
+            | prompt::AUDIT_INCOMPLETE_LINE
+            | prompt::CLOSE_CHECK_PASSES_LINE
+    ) || [
+        prompt::BUG_NOT_REPRODUCED_LINE,
+        prompt::PLAN_LINE,
+        prompt::IDEA_LINE,
+        prompt::ALREADY_FILED_LINE,
+        prompt::CLOSE_CHECK_FAILS_PREFIX,
+        prompt::SECURITY_FIX_LINE,
+        prompt::SECURITY_FIX_SPEC_LINE,
+        "Security review: ",
+        "Security reproduction: ",
+    ]
+    .iter()
+    .any(|prefix| line.starts_with(prefix))
+}
+
 impl Dialect {
     /// Extract live facts before root-result and initialization guards. Grok
     /// shares assistant framing without Claude's usage or safeguard rules.
@@ -187,6 +215,7 @@ impl Stream {
             init: None,
             turns_and_cost: None,
             final_message: None,
+            completion_message: None,
             completed: false,
             dialect,
         }
@@ -210,6 +239,12 @@ impl Stream {
         }
     }
 
+    fn remember_completion(&mut self, message: &str) {
+        if has_completion_line(message) {
+            self.completion_message = Some(message.to_string());
+        }
+    }
+
     fn tool_use(&self, name: &str, input: &Value) -> String {
         let grok = matches!(self.dialect, Dialect::Grok(_));
         let name = if grok && matches!(name, "bash" | "run_terminal_command") {
@@ -263,6 +298,11 @@ impl Stream {
             Some("assistant") => {
                 if event["parent_tool_use_id"].is_null() {
                     self.completed = false;
+                    if !assistant_error(event)
+                        && let Some(message) = message_text(event)
+                    {
+                        self.remember_completion(&message);
+                    }
                 }
                 if let Dialect::Claude(claude) = &mut self.dialect {
                     claude.assistant(event);
@@ -289,6 +329,14 @@ impl Stream {
                 {
                     return Vec::new();
                 }
+                // A child result may replace the latest text, but it is not a
+                // root handoff. Remembering it would let that text outlive a
+                // later root reply that carries no completion line.
+                if event["parent_tool_use_id"].is_null()
+                    && let Some(message) = event["result"].as_str()
+                {
+                    self.remember_completion(message);
+                }
                 self.final_message = event["result"].as_str().map(String::from);
                 self.completed = true;
                 if let (Some(turns), Some(cost)) = (
@@ -348,7 +396,13 @@ impl Decoder for Stream {
         let mut ended = Ended {
             session_id: self.init.and_then(|init| init.session_id),
             killed: Vec::new(),
-            final_message: self.final_message,
+            // A handoff can precede the sole result, but cannot supply a
+            // missing terminal result itself.
+            final_message: if self.completed {
+                self.completion_message.or(self.final_message)
+            } else {
+                self.final_message
+            },
             completed: self.completed,
         };
         let (failed, diagnostic) = match self.dialect {

src/harness/interpretation_tests.rs

@@ -239,6 +239,243 @@ fn claude_completion_keeps_the_last_result_and_the_last_complete_usage() {
     );
 }
 
+#[test]
+fn assistant_completion_line_survives_a_late_result() {
+    for harness in [Harness::Grok, Harness::Claude] {
+        for earlier in [
+            "PR ready.\nBug reproduced: fixed",
+            "Bug not reproduced: https://github.com/acme/widgets/issues/7#issuecomment-42",
+            "Security audit: complete",
+        ] {
+            let later = "The earlier background failure was already handled.";
+            let mut interpretation = stream(harness, "");
+            for event in [
+                serde_json::json!({
+                    "type": "assistant",
+                    "message": {
+                        "id": "earlier", "role": "assistant",
+                        "content": [{"type": "text", "text": earlier}]
+                    }
+                }),
+                serde_json::json!({
+                    "type": "system", "subtype": "task_notification",
+                    "task_id": "test", "status": "failed",
+                    "summary": "The earlier background test failed."
+                }),
+                serde_json::json!({
+                    "type": "assistant",
+                    "message": {
+                        "id": "later", "role": "assistant",
+                        "content": [{"type": "text", "text": later}]
+                    }
+                }),
+                serde_json::json!({
+                    "type": "result", "subtype": "success", "result": later
+                }),
+            ] {
+                interpretation.condense(&event.to_string());
+            }
+            let ended = finish(interpretation).outcome.unwrap();
+            assert_eq!(
+                ended.require_final_message("fixture").unwrap(),
+                earlier,
+                "{harness:?}"
+            );
+        }
+    }
+}
+
+#[test]
+fn assistant_handoffs_keep_reply_order_and_require_a_root_result() {
+    for harness in [Harness::Grok, Harness::Claude] {
+        let mut interpretation = stream(harness, "");
+        interpretation.condense(r#"{"type":"assistant","message":{"content":[{"type":"text","text":"Bug reproduced: fixed"}]}}"#);
+        assert_eq!(finish(interpretation).outcome.unwrap().final_message, None);
+
+        let mut interpretation = stream(harness, "");
+        for event in [
+            r#"{"type":"result","subtype":"success","result":"Bug reproduced: fixed"}"#,
+            r#"{"type":"assistant","message":{"content":[{"type":"text","text":"Security audit: incomplete"}]}}"#,
+            r#"{"type":"assistant","parent_tool_use_id":"child","message":{"content":[{"type":"text","text":"Security audit: complete"}]}}"#,
+            r#"{"type":"result","subtype":"success","result":"Background task handled."}"#,
+        ] {
+            interpretation.decode(event);
+        }
+        assert_eq!(
+            finish(interpretation)
+                .outcome
+                .unwrap()
+                .final_message
+                .as_deref(),
+            Some("Security audit: incomplete"),
+            "{harness:?}"
+        );
+    }
+}
+
+#[test]
+fn completion_line_survives_background_notification() {
+    let messages: Vec<_> = [Harness::Grok, Harness::Claude]
+        .into_iter()
+        .map(|harness| {
+            let completion = after_background_notification(
+                harness,
+                "PR ready.\nBug reproduced: fixed",
+                "That background failure was already handled. The PR is unchanged.",
+            );
+            completion.outcome.unwrap().final_message
+        })
+        .collect();
+    assert_eq!(
+        messages,
+        vec![Some("PR ready.\nBug reproduced: fixed".to_string()); 2]
+    );
+}
+
+#[test]
+fn completion_handoffs_survive_late_replies_and_later_handoffs_win() {
+    for harness in [Harness::Grok, Harness::Claude] {
+        for message in [
+            "Bug not reproduced: https://github.com/acme/widgets/issues/7#issuecomment-42",
+            "Recommendation strength: Strong\nArchitecture review plan: https://github.com/acme/widgets/issues/8",
+            "Architecture review idea: https://github.com/acme/widgets/issues/9",
+            "Architecture review already filed: https://github.com/acme/widgets/issues/10",
+            "Security audit: complete",
+            "Security audit: incomplete",
+            "Security close check: passes",
+            "Security close check: fails: test still fails",
+            "Security fix Ticket: https://github.com/acme/widgets/issues/11",
+            "Security fix Spec: https://github.com/acme/widgets/issues/12",
+            "Security reproduction: reproduced high single",
+            "Security reproduction: not reproduced",
+        ] {
+            for later in ["Background task handled.", ""] {
+                let ended = after_background_notification(harness, message, later)
+                    .outcome
+                    .unwrap();
+                assert_eq!(
+                    ended.require_final_message("fixture").unwrap(),
+                    message,
+                    "{harness:?}"
+                );
+            }
+            let ended = after_background_notification(harness, "Bug reproduced: fixed", message)
+                .outcome
+                .unwrap();
+            assert_eq!(ended.final_message.as_deref(), Some(message), "{harness:?}");
+        }
+    }
+}
+
+#[test]
+fn security_review_reads_its_completion_line_after_a_background_notification() {
+    let message = concat!(
+        "Reviewed.\n",
+        r#"Security review: {"unaddressed_count":1,"findings":["Cross-tenant read"],"pre_existing_count":0}"#
+    );
+    for harness in [Harness::Grok, Harness::Claude] {
+        let ended = after_background_notification(harness, message, "Background task handled.")
+            .outcome
+            .unwrap();
+        let review =
+            crate::security::review::Review::from_final_message(ended.final_message.as_deref())
+                .unwrap();
+        assert_eq!(review.findings, ["Cross-tenant read"]);
+    }
+}
+
+#[test]
+fn only_root_result_handoffs_count_and_later_failures_still_fail() {
+    let mut interpretation = stream(Harness::Claude, "");
+    for event in [
+        r#"{"type":"result","subtype":"success","result":"Bug reproduced: fixed"}"#,
+        r#"{"type":"result","subtype":"success","parent_tool_use_id":"child","result":"Security audit: incomplete"}"#,
+    ] {
+        interpretation.decode(event);
+    }
+    assert_eq!(
+        finish(interpretation)
+            .outcome
+            .unwrap()
+            .final_message
+            .as_deref(),
+        Some("Bug reproduced: fixed")
+    );
+    for harness in [Harness::Grok, Harness::Claude] {
+        let mut interpretation = stream(harness, "");
+        for event in [
+            r#"{"type":"result","subtype":"success","result":"Bug reproduced: fixed"}"#,
+            r#"{"type":"result","subtype":"error","is_error":true,"result":"Late turn failed"}"#,
+        ] {
+            interpretation.decode(event);
+        }
+        assert!(finish(interpretation).outcome.is_err(), "{harness:?}");
+    }
+}
+
+/// A delegated result is not a root handoff. Its completion line must not
+/// replace the root session's security review when a later root reply has none.
+#[test]
+fn grok_child_result_cannot_clear_a_root_security_review() {
+    let root = concat!(
+        "Reviewed.\n",
+        r#"Security review: {"unaddressed_count":1,"findings":["Cross-tenant read"],"pre_existing_count":0}"#
+    );
+    let child = r#"Security review: {"unaddressed_count":0,"findings":[],"pre_existing_count":0}"#;
+    for harness in [Harness::Grok, Harness::Claude] {
+        let mut interpretation = stream(harness, "");
+        for event in [
+            serde_json::json!({"type": "result", "subtype": "success", "result": root}),
+            serde_json::json!({
+                "type": "result",
+                "subtype": "success",
+                "parent_tool_use_id": "child",
+                "result": child
+            }),
+            serde_json::json!({
+                "type": "result",
+                "subtype": "success",
+                "result": "Background task handled."
+            }),
+        ] {
+            interpretation.condense(&event.to_string());
+        }
+        let message = finish(interpretation)
+            .outcome
+            .unwrap()
+            .final_message
+            .unwrap_or_default();
+        assert_ne!(message, child, "{harness:?}");
+        let review = crate::security::review::Review::from_final_message(Some(&message));
+        assert!(
+            review
+                .as_ref()
+                .ok()
+                .is_none_or(|review| !review.findings.is_empty()),
+            "{harness:?} accepted a cleared review from {message:?}"
+        );
+    }
+}
+
+fn after_background_notification(
+    harness: Harness,
+    earlier: &str,
+    later: &str,
+) -> interpretation::Completion {
+    let mut interpretation = stream(harness, "");
+    for event in [
+        serde_json::json!({"type": "result", "subtype": "success", "result": earlier}),
+        serde_json::json!({
+            "type": "system", "subtype": "task_notification", "task_id": "test",
+            "status": "failed", "summary": "The earlier background test failed."
+        }),
+        serde_json::json!({"type": "result", "subtype": "success", "result": later}),
+    ] {
+        interpretation.condense(&event.to_string());
+    }
+    finish(interpretation)
+}
+
 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.