Merge pull request 'Self run: deliver a result whole, never clipped' (#25) from eureka/self-run-full-result into main

crates/mind-body/src/self_run.rs

@@ -795,17 +795,19 @@ impl<'a> SelfInbox<'a> {
         Self { mind, start, plan: render_plan(plan_quota, now_ms), told, waiting: Vec::new(), heard, run: run.clone(), carried, echoed: Vec::new(), serving, last_answers, flush_written: false, flushing: false, replies: 0, unadmitted: Vec::new() }
     }
 
-    /// Admits each reply the mind has not taken, in order; one try each. A refusal or an unreachable mind keeps the reply and never ends the session.
+    /// Admits each reply the mind has not taken, in order; one try each. A refusal or an unreachable mind keeps that reply and every one behind it
+    /// (the admission ordinal is the order the Persona hears them in, so none may overtake a refused one) and never ends the session.
     fn admit_unadmitted(&mut self) {
         let label = self.run.id.0.rsplit(':').next().unwrap_or(&self.run.id.0).to_string();
-        let mut kept = Vec::new();
-        for reply in std::mem::take(&mut self.unadmitted) {
+        let mut queue = std::mem::take(&mut self.unadmitted).into_iter();
+        for reply in queue.by_ref() {
             let taken = matches!(self.mind.admit(AGENT, &label, vec![PipelineDocument::Reply(reply.clone())]), Ok(PipelineAdmissionOutcome::Committed { .. } | PipelineAdmissionOutcome::AlreadyAdmitted { .. }));
             if !taken {
-                kept.push(reply);
+                self.unadmitted.push(reply);
+                break;
             }
         }
-        self.unadmitted = kept;
+        self.unadmitted.extend(queue);
     }
 
     /// One result of the main thread: with words, one reply answering the briefs the CLI has taken since the last reply (the first 64, in tell
@@ -816,21 +818,25 @@ impl<'a> SelfInbox<'a> {
         let flush = std::mem::take(&mut self.flushing);
         // The flush result is the session's note to its successor, in the journal: it is no reply.
         if ok && !flush && !text.trim().is_empty() {
-            self.replies += 1;
-            let run_label = self.run.id.0.rsplit(':').next().unwrap_or(&self.run.id.0);
+            let run_label = self.run.id.0.rsplit(':').next().unwrap_or(&self.run.id.0).to_string();
             let taken = std::mem::take(&mut self.echoed);
             let answered: Vec<String> = if taken.is_empty() { self.last_answers.clone() } else { taken.into_iter().take(64).collect() };
-            let answers = answered.iter().map(|id| PipelineRef { kind: PipelineKind::FollowUp, id: id.as_str().into() }).collect();
+            let answers: Vec<PipelineRef> = answered.iter().map(|id| PipelineRef { kind: PipelineKind::FollowUp, id: id.as_str().into() }).collect();
             if !answered.is_empty() {
                 self.last_answers = answered;
             }
-            self.unadmitted.push(PipelineReply {
-                instance: self.mind.instance().clone(),
-                label: format!("{run_label}-r{}", self.replies).as_str().into(),
-                run: self.run.clone(),
-                answers,
-                text: cut_text(text, REPLY_TEXT_LIMIT).as_str().into(),
-            });
+            // A reply holds a bounded text, so a long result is delivered whole as consecutive replies, in order,
+            // each carrying the answers. A piece of only whitespace (the tail after a cut) is no reply.
+            for chunk in chunks(text, REPLY_TEXT_LIMIT).filter(|chunk| !chunk.trim().is_empty()) {
+                self.replies += 1;
+                self.unadmitted.push(PipelineReply {
+                    instance: self.mind.instance().clone(),
+                    label: format!("{run_label}-r{}", self.replies).as_str().into(),
+                    run: self.run.clone(),
+                    answers: answers.clone(),
+                    text: chunk.into(),
+                });
+            }
         }
         self.admit_unadmitted();
     }
@@ -892,10 +898,7 @@ impl SessionInbox for SelfInbox<'_> {
     }
 }
 
-/// The most of a result's text that reaches the journal.
-const RESULT_TEXT_LIMIT: usize = 4096;
-/// The most of a result's text that a reply holds
-/// (a `Para` holds 4000 bytes; the cut marker rides inside the rest).
+/// The most of a result's text that one reply holds (a `Para` holds 4000 bytes).
 const REPLY_TEXT_LIMIT: usize = 3800;
 
 /// A result event's subtype and error flag when it has them, as one line.
@@ -914,19 +917,26 @@ fn result_text(event: &serde_json::Map<String, Value>) -> &str {
     event.get("result").and_then(Value::as_str).unwrap_or("")
 }
 
-/// `text` cut at a character boundary to `limit` bytes, marked when cut.
-fn cut_text(text: &str, limit: usize) -> String {
-    let mut end = text.len().min(limit);
-    while !text.is_char_boundary(end) {
-        end -= 1;
-    }
-    let marker = if end < text.len() { "\n[truncated]" } else { "" };
-    format!("{}{marker}", &text[..end])
+/// `text` in consecutive pieces of at most `limit` bytes, each cut at a character boundary; their concatenation is `text`.
+fn chunks(text: &str, limit: usize) -> impl Iterator<Item = &str> {
+    let mut rest = text;
+    std::iter::from_fn(move || {
+        if rest.is_empty() {
+            return None;
+        }
+        let mut end = rest.len().min(limit);
+        while !rest.is_char_boundary(end) {
+            end -= 1;
+        }
+        let (piece, tail) = rest.split_at(end);
+        rest = tail;
+        Some(piece)
+    })
 }
 
-/// A result event as one line: its head, then its text cut to `limit` bytes.
-fn describe_result(event: &serde_json::Map<String, Value>, limit: usize) -> String {
-    format!("{}: {}", result_head(event), cut_text(result_text(event), limit))
+/// A result event as one line: its head, then its whole text.
+fn describe_result(event: &serde_json::Map<String, Value>) -> String {
+    format!("{}: {}", result_head(event), result_text(event))
 }
 
 /// What `claude` concluded, written to the unit's stdout so it reaches the
@@ -936,7 +946,7 @@ fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> {
     if event.get("type").and_then(Value::as_str) != Some("result") {
         return Ok(());
     }
-    writeln!(out, "{}", describe_result(&event, RESULT_TEXT_LIMIT))
+    writeln!(out, "{}", describe_result(&event))
 }
 
 /// The real spawner: the `claude` on the unit's `PATH`, its stderr in the

crates/mind-body/src/self_run/tests.rs

@@ -475,14 +475,13 @@ fn a_result_events_text_reaches_the_journal_with_its_subtype_and_error_flag() {
 }
 
 #[test]
-fn a_long_result_is_cut_at_the_limit_on_a_character_boundary_and_marked() {
-    // One ASCII byte first, so the limit falls in the middle of a two-byte character.
-    let text = format!("a{}", "é".repeat(RESULT_TEXT_LIMIT));
+fn a_long_result_reaches_the_journal_whole() {
+    let text = format!("a{}", "é".repeat(5000));
     let line = json!({"type": "result", "result": text}).to_string();
     let ran = go(vec![Out(line)]);
     let written = ran.out.lines().find(|l| l.starts_with("result")).unwrap();
-    assert_eq!(written, format!("result: a{}", "é".repeat(RESULT_TEXT_LIMIT / 2 - 1)));
-    assert!(ran.out.contains("\n[truncated]\n"), "{}", ran.out);
+    assert_eq!(written, format!("result: {text}"));
+    assert!(!ran.out.contains("[truncated]"), "{}", ran.out);
 }
 
 // ---- what is read -----------------------------------------------------------

crates/mind-body/src/self_run/tests/closure.rs

@@ -56,16 +56,22 @@ fn the_closure_holds_no_result_text_however_long_it_is() {
 }
 
 #[test]
-fn a_result_over_the_bound_is_cut_in_its_reply_at_a_character_boundary_marked_and_still_admitted() {
+fn a_result_over_the_bound_is_delivered_whole_in_order_across_replies_each_within_it() {
     let mind = with_run();
-    // Two bytes a character after a one-byte head, so the bound falls inside a character.
-    let text = format!("x{}", "é".repeat(3000));
+    // Two bytes a character after a one-byte head, so a bound falls inside a character.
+    let text = format!("x{}", "é".repeat(9000));
     carry_with(&Seen::of(&mind), vec![Out(result_with(&text))], Some(0), false).outcome.unwrap();
-    let PipelineDocument::Reply(reply) = mind.view(K::Reply, &format!("{INSTANCE}:reply:selfrun-a-r1")).expect("the reply was admitted").document else { panic!("a reply") };
-    let kept = reply.text.0.strip_suffix("\n[truncated]").expect("a cut result is marked");
-    assert_eq!(kept.len(), REPLY_TEXT_LIMIT - 1);
-    assert!(text.starts_with(kept));
-    assert!(reply.text.0.len() <= 4000);
+    let mut whole = String::new();
+    let mut n = 1;
+    while let Some(view) = mind.view(K::Reply, &format!("{INSTANCE}:reply:selfrun-a-r{n}")) {
+        let PipelineDocument::Reply(reply) = view.document else { panic!("a reply") };
+        assert!(reply.text.0.len() <= REPLY_TEXT_LIMIT && !reply.text.0.is_empty());
+        whole.push_str(&reply.text.0);
+        n += 1;
+    }
+    assert_eq!(n - 1, 5, "{} bytes in pieces of at most {REPLY_TEXT_LIMIT}", text.len());
+    assert_eq!(whole, text);
+    assert_eq!(mind.ids(K::Reply).len(), 5);
 }
 
 #[test]

crates/mind-body/src/self_run/tests/conversation.rs

@@ -580,6 +580,33 @@ fn every_result_of_a_span_answers_the_spans_briefs() {
     assert!(answered(&mind, 1).is_empty() && answered(&mind, 2).is_empty());
 }
 
+/// A long result is several replies; each answers the briefs, and a piece of only whitespace is no reply.
+#[test]
+fn every_piece_of_a_long_result_answers_the_briefs_and_a_whitespace_tail_is_no_piece() {
+    let mind = live_mind();
+    persona_commit(&mind, vec![libby("l1")]);
+    let long = "z".repeat(REPLY_TEXT_LIMIT * 2 + 1);
+    let ran = mind_run(&mind, vec![Out(echo()), Out(saying(&long))]);
+    assert!(ran.outcome.is_ok());
+    for n in 1..=3 {
+        assert_eq!(answered(&mind, n), vec![id_of("l1")], "piece {n}");
+    }
+    assert!(reply_of(&mind, 4).is_none());
+
+    let mind = live_mind();
+    persona_commit(&mind, vec![libby("l1")]);
+    let text = format!("{}\n \n", "y".repeat(REPLY_TEXT_LIMIT));
+    let ran = mind_run(&mind, vec![Out(echo()), Out(saying(&text))]);
+    assert!(ran.outcome.is_ok());
+    assert_eq!(answered(&mind, 1), vec![id_of("l1")]);
+    assert!(reply_of(&mind, 2).is_none(), "the whitespace tail is no reply");
+    // Nor is it ever offered to the mind (an admission refused as empty would be retried for the rest of the run).
+    let mind = live_mind();
+    persona_commit(&mind, vec![libby("l1")]);
+    let bare = mind_run(&mind, vec![Out(echo()), Out(saying(&"y".repeat(REPLY_TEXT_LIMIT)))]);
+    assert_eq!(ran.admits, bare.admits, "the tail cost an admission");
+}
+
 /// The same conversation's new brief is written at once, whatever runs; another conversation's waits for the turn to end.
 #[test]
 fn the_served_conversations_briefs_drop_in_mid_turn_and_another_conversations_wait() {

crates/mind-body/src/self_run/tests/reply.rs

@@ -106,3 +106,15 @@ fn a_reply_the_mind_did_not_take_is_tried_once_more_when_the_run_ends() {
     assert!(ran.outcome.is_ok());
     assert_eq!(text_of(&mind, 1), "last words");
 }
+
+#[test]
+fn a_refused_piece_of_a_long_result_holds_back_the_pieces_behind_it() {
+    let mind = live_mind();
+    let seen = Seen::of(&mind);
+    seen.fail_admits.set(1);                                   // the first piece is refused
+    let long = "w".repeat(REPLY_TEXT_LIMIT * 2 + 1);
+    let ran = carry_with(&seen, vec![Out(echo()), Out(at("m1", 10)), Out(saying(&long))], Some(0), false);
+    assert!(ran.outcome.is_ok());
+    let ordinals: Vec<u64> = (1..=3).map(|n| mind.view(K::Reply, &format!("{INSTANCE}:reply:selfrun-a-r{n}")).unwrap_or_else(|| panic!("piece {n} lost")).admission.ordinal).collect();
+    assert!(ordinals.windows(2).all(|w| w[0] < w[1]), "the pieces reached the mind out of order: {ordinals:?}");
+}