diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 2c457ad..a919156 100644 --- a/crates/mind-body/src/self_run.rs +++ b/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 = 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 = 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) -> &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 { + 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, 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 { + 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 diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index fd77512..1510503 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/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 ----------------------------------------------------------- diff --git a/crates/mind-body/src/self_run/tests/closure.rs b/crates/mind-body/src/self_run/tests/closure.rs index 5fcbbc2..b028682 100644 --- a/crates/mind-body/src/self_run/tests/closure.rs +++ b/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] diff --git a/crates/mind-body/src/self_run/tests/conversation.rs b/crates/mind-body/src/self_run/tests/conversation.rs index 2c7d8b1..948b8c6 100644 --- a/crates/mind-body/src/self_run/tests/conversation.rs +++ b/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() { diff --git a/crates/mind-body/src/self_run/tests/reply.rs b/crates/mind-body/src/self_run/tests/reply.rs index 0a9244a..0a4e37c 100644 --- a/crates/mind-body/src/self_run/tests/reply.rs +++ b/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 = (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:?}"); +}