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 <noreply@anthropic.com>

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<String>,
     open: Vec<String>,
@@ -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<String>,
+    /// 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<Retell>,
 }
@@ -1003,11 +1003,7 @@ impl<C: ControlSource, S: ControlSource> 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<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
         self.write(batch, None, 0)
     }
 
-    fn flush_text(&mut self, why: Why) -> Result<String> {
+    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<C: ControlSource, S: ControlSource> 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<C: ControlSource, S: ControlSource>(inbox: &PersonaInbox<'_, C, S>, batch
     let inputs = trace_inputs(config, batch);
     let now = organs.clock.now();
     let record = |outputs: Vec<TraceItem>, 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<C: ControlSource, S: ControlSource>(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) {

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::<Vec<_>>(), [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<String> = 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());
+}

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<String> {
-        Ok(flush_text(why))
+    fn flush_text(&mut self, why: Why) -> String {
+        flush_text(why)
     }
 
     fn line(&mut self, line: &str) {

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

@@ -35,7 +35,7 @@ pub(crate) trait SessionInbox {
     fn poll(&mut self) -> Result<Polled>;
     /// The text of what waits, which is thereby told.
     fn tell(&mut self) -> Result<String>;
-    fn flush_text(&mut self, why: Why) -> Result<String>;
+    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<dyn Child>, 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")));
                 }