Self run: deliver a result whole, never clipped #25
5 changed files with 98 additions and 44 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 -----------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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() {
|
||||
|
|
|
|||
|
|
@ -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:?}");
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue