diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index f866714..88be021 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -1214,9 +1214,13 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { (text, taken) } - /// Replies that were told and not answered are told again: the line failed, was stale or was refused. + /// Replies that were told and not answered are told again: the line failed, was stale or was refused. A reply no longer in force + /// (resolved by someone, or by a resolution whose answer was lost) leaves for good: it stays in `told_replies`, so no read returns it. fn untell(&mut self, replies: Vec) { for reply in replies { + if matches!(reply_in_force(self.organs.mind, &reply.id), Ok(false)) { + continue; + } self.told_replies.remove(&reply.id); self.waiting_replies.push(reply); } @@ -1421,9 +1425,8 @@ impl SessionInbox for PersonaInbox<'_, C, S> match step { Step::Answered(started, intent, opened) => { // The result answers the replies its lines carried: they are heard now, and only now. - if hear_replies(self.organs, self.run, &heard, started.as_ref().map(|(_, follow_up)| follow_up)).is_err() { - self.untell(heard); - } + let unresolved = hear_replies(self.organs, self.run, heard, started.as_ref().map(|(_, follow_up)| follow_up)); + self.untell(unresolved); // What she admitted this turn is hers to settle or take on for the rest of the session, though no line showed its id. self.seen.add(Shown { people: Vec::new(), open: opened }); // The one ack rule: through the highest answered seq, never past a delivery still told @@ -1834,33 +1837,38 @@ fn unheard_replies(mind: &dyn MindPort, told: &BTreeSet, waiting: &[Hear Ok(found) } -/// The result that answers a line resolves the replies the line carried, in one batch: Answered by the follow-up her answer's `start_self_run` -/// admitted, else Recorded as heard. A reply resolved is out of force, so no later process tells it. -fn hear_replies(organs: &Organs, run: &PipelineRef, heard: &[HeardReply], follow_up: Option<&PipelineRef>) -> Result<()> { - if heard.is_empty() { - return Ok(()); - } - let mut resolutions = Vec::new(); - for reply in heard { - let subject = PipelineRef { kind: PipelineKind::Reply, id: reply.id.as_str().into() }; - let outcome = match follow_up { - Some(by) => ResolutionOutcome::Answered { by: by.clone() }, - None => ResolutionOutcome::Recorded { reason: "heard".into() }, - }; - resolutions.push(PipelineDocument::Resolution(PipelineResolution { - sequence: next_sequence(organs.mind, &subject)?, - subject, - outcome, - rationale: "Her Persona heard this reply of her Self in a turn and answered it.".into(), - resolved_on: date(organs.clock.now()), - })); - } - match organs.mind.admit(AGENT, label_of(run), resolutions)? { +/// The result that answers a line resolves each reply the line carried, one by one: Answered by the follow-up her answer's +/// `start_self_run` admitted, else Recorded as heard. A reply resolved is out of force, so no later process tells it. A reply whose +/// resolution fails does not hold the others hostage: it is returned, to be told again if it is still in force. +fn hear_replies(organs: &Organs, run: &PipelineRef, heard: Vec, follow_up: Option<&PipelineRef>) -> Vec { + heard.into_iter().filter(|reply| resolve_reply(organs, run, reply, follow_up).is_err()).collect() +} + +fn resolve_reply(organs: &Organs, run: &PipelineRef, reply: &HeardReply, follow_up: Option<&PipelineRef>) -> Result<()> { + let subject = PipelineRef { kind: PipelineKind::Reply, id: reply.id.as_str().into() }; + let outcome = match follow_up { + Some(by) => ResolutionOutcome::Answered { by: by.clone() }, + None => ResolutionOutcome::Recorded { reason: "heard".into() }, + }; + let resolution = PipelineDocument::Resolution(PipelineResolution { + sequence: next_sequence(organs.mind, &subject)?, + subject, + outcome, + rationale: "Her Persona heard this reply of her Self in a turn and answered it.".into(), + resolved_on: date(organs.clock.now()), + }); + match organs.mind.admit(AGENT, label_of(run), vec![resolution])? { PipelineAdmissionOutcome::Committed { .. } | PipelineAdmissionOutcome::AlreadyAdmitted { .. } => Ok(()), - _ => bail!("the replies could not be resolved"), + _ => bail!("the reply could not be resolved"), } } +/// Whether the reply is still in force in her mind. +fn reply_in_force(mind: &dyn MindPort, id: &str) -> Result { + let one = Selection { keys: Some(vec![id.to_string()]), ..in_force(of_kind(PipelineKind::Reply)) }; + Ok(!headers(mind, &one)?.is_empty()) +} + /// The brief becomes a follow-up of the Persona turn's run, and a Self run /// claims it. A Self run already live is her Self run: the follow-up waits in /// the queue for it. Answers that run and the follow-up it admitted. diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index 80145e6..17ae31c 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/crates/mind-body/src/persona/tests.rs @@ -177,6 +177,8 @@ struct FakeSession { refuse: Cell, /// The child stays while replies are left, though the chat holds nothing: a line with no delivery can still come. linger: Rc>, + /// How many reads a written line's echo is held back for: the driver sees timeouts, and polls, before it. + stall: Rc>, } impl Spawner for FakeSession { @@ -198,6 +200,7 @@ impl Spawner for FakeSession { instructions, schema: after("--json-schema"), linger: self.linger.clone(), + stall: self.stall.clone(), queue: VecDeque::new(), closed: false, answered: false, @@ -217,6 +220,7 @@ struct Replying { instructions: String, schema: String, linger: Rc>, + stall: Rc>, queue: VecDeque, closed: bool, /// A canned reply was given; once they are spent and the queue is empty the child ends. @@ -255,6 +259,10 @@ impl Child for Replying { fn next(&mut self, _timeout: StdDuration) -> Read { self.asked += 1; assert!(self.asked < 3000, "the loop never ended"); + if !self.queue.is_empty() && self.stall.get() > 0 { + self.stall.set(self.stall.get() - 1); + return Read::Timeout; + } if let Some(read) = self.queue.pop_front() { if matches!(&read, Read::Line(line) if line.contains("isReplay")) && let Some(after_echo) = &*self.after_echo.borrow() @@ -405,6 +413,7 @@ impl Rig { pending, refuse: Cell::new(false), linger: Rc::default(), + stall: Rc::default(), }, lender: FakeLender::new(Answer::Grant), persona: FakePersona(Cell::new(false)), @@ -4468,6 +4477,97 @@ fn a_refused_result_leaves_its_replies_in_force_and_they_are_told_again() { assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), None, "nothing is heard that was not answered"); } +/// A reply another writer resolved after it was told is out of force: its resolution is refused, it is not told again, and it does not +/// keep the reply told beside it from being heard. +#[test] +fn a_reply_put_out_of_force_after_it_was_told_leaves_for_good_and_holds_no_other_reply_hostage() { + let rig = Rig::new(vec![operator_says(1, "hello")], &[says(&[]), says(&[]), says(&[]), says(&[])]); + rig.mind.committed(her_self_asked()); + rig.mind.committed(vec![self_reply("self-r-r1", "One."), self_reply("self-r-r2", "Two.")]); + rig.clock.step.set(1); + rig.model.linger.set(true); + rig.open_the_run(); + let timeline = Timeline::over(&rig.mind); + let (now, linger, deliveries, echoes) = (rig.clock.now.clone(), rig.model.linger.clone(), rig.chat.deliveries.clone(), Cell::new(0u64)); + let doing_then = timeline.doing.clone(); + *rig.model.after_echo.borrow_mut() = Some(Box::new(move || { + echoes.set(echoes.get() + 1); + now.set(now.get() + PAST_THE_REPLY_READ_MS); + if echoes.get() == 1 { + // Told on the first line; someone else resolves the first before her answer is heard. + doing_then.borrow_mut().push_back(vec![PipelineDocument::Resolution(eureka_pipeline::PipelineResolution { + subject: reference(K::Reply, &reply_id("self-r-r1")), + sequence: 1, + outcome: ResolutionOutcome::Recorded { reason: "elsewhere".into() }, + rationale: "Resolved by another writer.".into(), + resolved_on: Date("2026-10-09".into()), + })]); + } + if echoes.get() <= 3 { + deliveries.borrow_mut().push(operator_says(echoes.get() + 1, "and on")); + } else { + linger.set(false); + } + })); + assert_eq!(failure_of(rig.start_over(&timeline, &config())), None); + let lines = told(&rig); + let told_of = |label: &str| all_replies(&lines).iter().filter(|line| line.contains(&reply_id(label))).count(); + assert_eq!((told_of("self-r-r1"), told_of("self-r-r2")), (1, 1), "each told once: {lines:?}"); + assert!(lines.len() >= 3, "later lines followed: {lines:?}"); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), Some(ResolutionOutcome::Recorded { reason: "elsewhere".into() })); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r2")), heard_outcome(), "the reply beside it was heard"); +} + +/// A reply that reaches a session with no chat pending is read at the next interval, told on the line it makes, and resolved. +#[test] +fn a_reply_reaching_an_idle_session_with_no_chat_pending_is_told_once_and_resolved() { + let rig = Rig::new(vec![operator_says(1, "hello")], &[says(&[]), says(&[])]); + rig.mind.committed(her_self_asked()); + rig.clock.step.set(1); + rig.model.linger.set(true); + rig.open_the_run(); + let timeline = Timeline::over(&rig.mind); + let (now, linger, results) = (rig.clock.now.clone(), rig.model.linger.clone(), Cell::new(0u64)); + let doing_then = timeline.doing.clone(); + *rig.model.before_result.borrow_mut() = Some(Box::new(move || { + results.set(results.get() + 1); + now.set(now.get() + PAST_THE_REPLY_READ_MS); + if results.get() == 1 { + // Her first line is answered and acked; with nothing left in the chat, the reply arrives while she is idle, and the poll after that result reads it. + doing_then.borrow_mut().push_back(vec![self_reply("self-r-r1", "The shaders are reviewed.")]); + } else { + linger.set(false); + } + })); + assert_eq!(failure_of(rig.start_over(&timeline, &config())), None); + let lines = told(&rig); + assert_eq!(all_replies(&lines).len(), 1, "told once: {lines:?}"); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), heard_outcome()); +} + +/// A reply read while a written line is not yet echoed waits, and a later read of the same interval does not wait it twice. +#[test] +fn a_reply_waiting_across_a_read_while_a_line_is_not_echoed_is_told_once() { + let rig = Rig::new(vec![operator_says(1, "hello")], &[says(&[]), says(&[])]); + rig.mind.committed(her_self_asked()); + rig.clock.step.set(PAST_THE_REPLY_READ_MS); + rig.open_the_run(); + let timeline = Timeline::over(&rig.mind); + let doing_then = timeline.doing.clone(); + // The first line is written; her reply arrives, and the echo is held past three reads of her replies. + rig.model.stall.set(3); + let arrived = Cell::new(false); + *rig.model.during.borrow_mut() = Some(Box::new(move || { + if !arrived.replace(true) { + doing_then.borrow_mut().push_back(vec![self_reply("self-r-r1", "Which shader first?")]); + } + })); + assert_eq!(failure_of(rig.start_over(&timeline, &config())), None); + let lines = told(&rig); + assert_eq!(all_replies(&lines).len(), 1, "told once: {lines:?}"); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), heard_outcome()); +} + /// Nothing a reply's text holds passes for a message, a notice or a field of another reply. #[test] fn nothing_a_replys_text_holds_passes_for_a_message_a_notice_or_a_reply() {