From f3068d917abe47b3f8d9ade22cb8cc27d6aeda83 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 2 Nov 2026 11:14:05 +0100 Subject: [PATCH] flush-open-bound-at-decode: a line's open items are bounded when its reply is decoded A flush written alone shows the people and open items the session's lines showed; which of those are still open is read when a reply is decoded. A settle on a shown item that an earlier turn closed is dropped with a trace note and the rest of the reply is admitted. The flush no longer reads the mind at write time, so flush_text returns a String again. Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/persona.rs | 42 ++++++----- crates/mind-body/src/persona/tests.rs | 88 ++++++++++++++++++++++++ crates/mind-body/src/self_run.rs | 4 +- crates/mind-body/src/self_run/session.rs | 4 +- 4 files changed, 117 insertions(+), 21 deletions(-) diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index d00110c..9a577db 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -848,7 +848,7 @@ fn one_process( echoed: Vec::new(), shown: Shown::default(), echoed_refusals: 0, - seen: Vec::new(), + seen: Shown::default(), retell: None, }; drive(organs.clock, child, &mut inbox, remaining_s, &mut std::io::stdout())?; @@ -906,7 +906,7 @@ impl Conversation { /// What a batch's line showed her: the only people she may name and the only open items she may /// settle or take on in the result that answers it. -#[derive(Default)] +#[derive(Clone, Default)] struct Shown { people: Vec, open: Vec, @@ -973,8 +973,8 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> { /// The most refusals any line echoed since the last result had already had. echoed_refusals: u32, shown: Shown, - /// Every person a line this process wrote showed: the bound a flush written alone starts from. - seen: Vec, + /// Everyone and every open item the lines this process wrote showed: what a flush written alone may name and settle. + seen: Shown, /// What refused results left to be told again. retell: Option, } @@ -1003,11 +1003,7 @@ impl PersonaInbox<'_, C, S> { if let Some(through) = batch.iter().map(|delivery| delivery.seq).max() { self.told_through = self.told_through.max(through); } - for person in &shown.people { - if !self.seen.contains(person) { - self.seen.push(person.clone()); - } - } + self.seen.add(shown.clone()); self.written = Some(Written { batch, shown, refusals }); Ok(match prefix { Some(prefix) => format!("{prefix}\n\n{input}"), @@ -1060,7 +1056,7 @@ impl SessionInbox for PersonaInbox<'_, C, S> self.write(batch, None, 0) } - fn flush_text(&mut self, why: Why) -> Result { + fn flush_text(&mut self, why: Why) -> String { let cause = match why { Why::Ceiling => "its context is full", Why::Rotation => "the unit's time is nearly up", @@ -1068,14 +1064,13 @@ impl SessionInbox for PersonaInbox<'_, C, S> }; let text = format!("This session ends soon ({cause}). Put into `remember` what you must still know when it is gone, including what you were in the middle of and why, and `settle` what is finished. Speak only if leaving would strand something someone is waiting on."); if self.written.is_none() { - // A flush alone is a line like any other: it shows the people the session showed and the items open now. - let organs = self.organs; - let goals = render_responsibilities(organs.mind, organs.mind.instance(), &BTreeMap::new())?; - let shown: Vec<&str> = self.seen.iter().map(String::as_str).chain(goals.people.iter().map(String::as_str)).collect(); + // A flush alone is a line like any other: it may name the people and settle the items the session showed. + // Which of those items are still open is read when its reply is decoded, after every earlier result is admitted. + let shown: Vec<&str> = self.seen.people.iter().map(String::as_str).collect(); let people = people_she_may_name(self.config, &[], &shown); - self.written = Some(Written { batch: Vec::new(), shown: Shown { people, open: goals.open }, refusals: 0 }); + self.written = Some(Written { batch: Vec::new(), shown: Shown { people, open: self.seen.open.clone() }, refusals: 0 }); } - Ok(text) + text } fn echoed(&mut self) { @@ -1179,7 +1174,7 @@ fn act(inbox: &PersonaInbox<'_, C, S>, batch let inputs = trace_inputs(config, batch); let now = organs.clock.now(); let record = |outputs: Vec, outcome: String| TurnTrace { run: label_of(run).into(), turn, at: now, inputs: inputs.clone(), outputs, outcome }; - let reply = match decode_reply(text, &inbox.rooms, &shown.people, &shown.open) { + let mut reply = match decode_reply(text, &inbox.rooms, &shown.people, &shown.open) { Ok(reply) => reply, Err(why) => { let raw = TraceItem { provenance: "her own".into(), room: String::new(), text: text.chars().take(TRACE_RAW_MAX).collect() }; @@ -1193,7 +1188,20 @@ fn act(inbox: &PersonaInbox<'_, C, S>, batch (Some(first), Some(last)) => format!("s{first}-{last}"), _ => format!("{}-t{turn}", label_of(run)), }; + // An item shown to her and closed since (by an earlier turn of this session) is not a reason to lose the reply. + let mut closed = Vec::new(); + for settled in std::mem::take(&mut reply.settle) { + let keys = Selection { keys: Some(vec![settled.item.clone()]), ..in_force(of_kind(PipelineKind::Memory)) }; + if headers(organs.mind, &keys)?.is_empty() { + closed.push(settled.item); + } else { + reply.settle.push(settled); + } + } let mut outputs = vec![TraceItem { provenance: "her own".into(), room: String::new(), text: format!("intent: {}", reply.intent) }]; + for item in closed { + outputs.push(TraceItem { provenance: "her own".into(), room: String::new(), text: format!("settle {item} dropped: already closed") }); + } for (index, say) in reply.say.iter().enumerate() { // The gate binds the send as it bound the call: a hold placed while the model was answering stops the reply. if let Effective::Held { reason } = read(control, swarm, &instance) { diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index 5c3d476..b6cd043 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/crates/mind-body/src/persona/tests.rs @@ -2804,3 +2804,91 @@ fn a_start_that_answered_nothing_keeps_the_old_rationale() { let lines = recent_runs(&rig.mind, &Slug(INSTANCE.into())).unwrap(); assert!(lines[0].contains("The Persona unit's start ended.") && !lines[0].contains("Her last intent"), "{lines:?}"); } + +/// A rig whose store holds one open commitment of hers, and the commitment's id. +fn rig_with_open_commitment() -> (Rig, String) { + let rig = Rig::new(vec![delivery(1, "swarm", "@libby:chat", "plan")], &[reply(&[("swarm", "ok")], json!([keeps("Committed", "I integrate the fork", &["@libby:chat"])]), json!([]), json!(null)), says(&[])]); + let _ = rig.go(); + let id = memories(&rig).remove(0).0; + rig.model.replies.borrow_mut().clear(); + (rig, id) +} + +fn settling(say: &[(&str, &str)], remember: serde_json::Value, id: &str, why: &str) -> String { + reply(say, remember, json!([{ "item": id, "outcome": "done", "why": why }]), json!(null)) +} + +#[test] +fn a_flush_reply_keeps_its_memories_when_an_earlier_turn_closed_the_item_it_also_settles() { + let (rig, id) = rig_with_open_commitment(); + rig.cap.set(u64::try_from(ROTATE_MARGIN_S).unwrap()); + rig.model.replies.borrow_mut().push_back(settling(&[("swarm", "done")], json!([]), &id, "first")); + rig.model.replies.borrow_mut().push_back(settling(&[], json!([keeps("Did", "Closed the fork with Libby", &["@libby:chat"])]), &id, "again")); + rig.chat.deliveries.borrow_mut().push(delivery(2, "swarm", "@libby:chat", "status")); + assert_eq!(failure_of(rig.go()), None); + assert_eq!(last_outcome(&rig), "answered"); + assert!(memories(&rig).iter().any(|(_, memory)| memory.text.0 == "Closed the fork with Libby"), "{:?}", memories(&rig)); + assert_eq!(outcome_of(&rig, &id), Some(ResolutionOutcome::Recorded { reason: "first".into() })); + let traced = rig.trace.0.borrow(); + assert!(traced.last().unwrap().outputs.iter().any(|item| item.text.contains(&format!("settle {id} dropped"))), "{traced:?}"); +} + +#[test] +fn a_tell_reply_keeps_its_memories_when_the_turn_running_as_it_was_written_closed_the_item_it_also_settles() { + let (rig, id) = rig_with_open_commitment(); + rig.model.replies.borrow_mut().push_back(settling(&[("swarm", "done")], json!([]), &id, "first")); + rig.model.replies.borrow_mut().push_back(settling(&[("swarm", "again")], json!([keeps("Did", "Closed the fork with Libby", &["@libby:chat"])]), &id, "again")); + rig.chat.deliveries.borrow_mut().push(delivery(2, "swarm", "@libby:chat", "status")); + let (deliveries, now, pushed) = (rig.chat.deliveries.clone(), rig.clock.now.clone(), Cell::new(false)); + // The second line is written while the first line's turn runs, so it shows the item before the first result is admitted. + *rig.model.during.borrow_mut() = Some(Box::new(move || { + if !pushed.replace(true) { + deliveries.borrow_mut().push(delivery(3, "swarm", "@libby:chat", "more")); + now.set(now.get() + 10_000); + } + })); + assert_eq!(failure_of(rig.go()), None); + assert!(told(&rig).iter().any(|line| line.contains(r#"text="more""#)), "the tell was written: {:?}", told(&rig)); + assert_eq!(last_outcome(&rig), "answered"); + assert!(memories(&rig).iter().any(|(_, memory)| memory.text.0 == "Closed the fork with Libby"), "{:?}", memories(&rig)); + assert_eq!(outcome_of(&rig, &id), Some(ResolutionOutcome::Recorded { reason: "first".into() })); +} + +#[test] +fn a_flush_may_not_settle_an_item_opened_after_the_session_last_line() { + let rig = Rig::new(vec![delivery(1, "swarm", "@libby:chat", "plan")], &[reply(&[("swarm", "ok")], json!([keeps("Committed", "I integrate the fork", &["@libby:chat"])]), json!([]), json!(null)), says(&[])]); + let _ = rig.go(); + let id = memories(&rig).remove(0).0; + // A second run of the same shape: the commitment is opened by the first line's own reply, after that line was written. + let fresh = Rig::new(vec![delivery(1, "swarm", "@libby:chat", "plan")], &[]); + fresh.cap.set(u64::try_from(ROTATE_MARGIN_S).unwrap() + 15); + let opened = reply(&[("swarm", "ok")], json!([keeps("Committed", "I integrate the fork", &["@libby:chat"])]), json!([]), json!(null)); + let guess = settling(&[], json!([]), &id, "guessed"); + *fresh.model.replies.borrow_mut() = VecDeque::from([opened, guess.clone(), guess]); + let _ = fresh.go(); + assert_eq!(memories(&fresh).iter().map(|(found, _)| found.clone()).collect::>(), [id.clone()], "the same commitment, opened by the first reply"); + assert_eq!(outcome_of(&fresh, &id), None, "the flush named an item no line of the session showed"); + assert_eq!(last_outcome(&fresh), "failed: the reply is not a persona reply"); +} + +#[test] +fn a_flush_written_beside_a_tell_leaves_the_tell_to_be_answered_and_acked() { + let rig = Rig::new(vec![operator_says(1, "one")], &[says(&[("operator", "for one")]), says(&[("operator", "for two")]), says(&[("operator", "for three")])]); + rig.cap.set(u64::try_from(ROTATE_MARGIN_S).unwrap() + 15); + let (deliveries, now, count) = (rig.chat.deliveries.clone(), rig.clock.now.clone(), Cell::new(0)); + // Each of the first two lines is written as a delivery arrives and the clock passes a poll; the second line is + // outstanding when the rotation falls due, so the third write carries delivery 3 and the flush ask together. + *rig.model.during.borrow_mut() = Some(Box::new(move || { + count.set(count.get() + 1); + if count.get() <= 2 { + deliveries.borrow_mut().push(operator_says(count.get() + 1, "more")); + now.set(now.get() + 10_000); + } + })); + let Outcome::Served { failure, .. } = rig.go() else { panic!("served") }; + assert_eq!(failure, None); + let flushed: Vec = told(&rig).into_iter().filter(|line| line.contains("This session ends soon")).collect(); + assert_eq!(flushed.len(), 1, "one line carries the tell and the flush: {:?}", told(&rig)); + assert!(flushed[0].contains(r#"text="more""#), "{}", flushed[0]); + assert_eq!(rig.chat.acked.borrow().last().copied(), Some(3), "the tell beside the flush is answered and acked: {:?}", rig.chat.acked.borrow()); +} diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index d9a926b..94651c5 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -597,8 +597,8 @@ impl SessionInbox for SelfInbox<'_> { Ok(text) } - fn flush_text(&mut self, why: Why) -> Result { - Ok(flush_text(why)) + fn flush_text(&mut self, why: Why) -> String { + flush_text(why) } fn line(&mut self, line: &str) { diff --git a/crates/mind-body/src/self_run/session.rs b/crates/mind-body/src/self_run/session.rs index 5e46f42..0a34884 100644 --- a/crates/mind-body/src/self_run/session.rs +++ b/crates/mind-body/src/self_run/session.rs @@ -35,7 +35,7 @@ pub(crate) trait SessionInbox { fn poll(&mut self) -> Result; /// The text of what waits, which is thereby told. fn tell(&mut self) -> Result; - fn flush_text(&mut self, why: Why) -> Result; + fn flush_text(&mut self, why: Why) -> String; /// Every line of the child's stream, before it is read. fn line(&mut self, _line: &str) {} /// The CLI took one of our lines into a request. @@ -117,7 +117,7 @@ pub(crate) fn drive(clock: &dyn Clock, mut child: Box, inbox: &mut dy parts.push(inbox.tell()?); } if let Some(why) = *flush { - parts.push(inbox.flush_text(why)?); + parts.push(inbox.flush_text(why)); } child.write_line(&user_line(&parts.join("\n\n"))); }