diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index ea79e1e..caf6b0d 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -891,7 +891,9 @@ fn one_process( return Ok(()); }; let mut inbox = PersonaInbox::new(organs, control, swarm, config, run, progress, first, started_ms, render_plan(lease.plan_quota(), started_ms)); - drive(organs.clock, child, &mut inbox, remaining_s, &mut std::io::stdout())?; + let driven = drive(organs.clock, child, &mut inbox, remaining_s, &mut std::io::stdout()); + inbox.settle_replies(); + driven?; Ok(()) } @@ -1093,6 +1095,9 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> { waiting_replies: Vec, /// The replies the CLI took in since the last result: that result is what answers them. echoed_replies: Vec, + /// Replies she answered whose resolution the mind has not yet shown took, with the follow-up her answer started: retried at each + /// result and at close, never told again. A later process tells only what the mind still holds in force. + answered_replies: Vec<(HeardReply, Option)>, /// When her replies are next read; 0 reads them at the first chance, the session's first line. reply_read_due_ms: u64, /// The notice lines for changes in her repos, to go with the next tell. Only a live session fills them, so a notice never starts one. @@ -1141,6 +1146,7 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { told_replies: BTreeSet::new(), waiting_replies: Vec::new(), echoed_replies: Vec::new(), + answered_replies: Vec::new(), reply_read_due_ms: 0, notices: Vec::new(), repos: Vec::new(), @@ -1215,7 +1221,8 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { } /// 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 her answer, by someone else, or by a resolution whose answer was lost) leaves for good: it stays in `told_replies`, so no read returns it. + /// (resolved by someone else, or by a resolution whose answer was lost) leaves for good: it stays in `told_replies`, so no read returns it. + /// A read that fails leaves the reply in force: told twice is better than lost unheard. fn untell(&mut self, replies: Vec) { for reply in replies { if matches!(reply_in_force(self.organs.mind, &reply.id), Ok(false)) { @@ -1227,6 +1234,18 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { self.waiting_replies.sort_by_key(|reply| reply.ordinal); } + /// Resolves each reply she answered, one by one so that one refusal holds no other hostage: Answered by the follow-up her answer's + /// `start_self_run` admitted, else Recorded as heard. Whether it took is read from the mind, not from the admit's answer: a reply + /// still in force stays here to be retried at the next result and at close; one out of force (resolved, or resolved elsewhere) is done. + fn settle_replies(&mut self) { + for (reply, by) in std::mem::take(&mut self.answered_replies) { + let _ = resolve_reply(self.organs, self.run, &reply, by.as_ref()); + if !matches!(reply_in_force(self.organs.mind, &reply.id), Ok(false)) { + self.answered_replies.push((reply, by)); + } + } + } + /// Her in-force replies this process has not told become waiting replies, at most once in `REPLY_READ_EVERY_S`: one selection of her /// replies in force, then the documents and the briefs' repos of the new ones. A read that fails waits for the next and costs /// her neither the reply nor her chat, which needs no mind. @@ -1425,8 +1444,9 @@ 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. - hear_replies(self.organs, self.run, &heard, started.as_ref().map(|(_, follow_up)| follow_up)); - self.untell(heard); + let by = started.as_ref().map(|(_, follow_up)| follow_up.clone()); + self.answered_replies.extend(heard.into_iter().map(|reply| (reply, by.clone()))); + self.settle_replies(); // 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 @@ -1837,16 +1857,7 @@ fn unheard_replies(mind: &dyn MindPort, told: &BTreeSet, waiting: &[Hear Ok(found) } -/// 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 resolved reply is out of force, so no later process tells it. Whether it took is -/// read from the mind (`reply_in_force`, in `untell`), not from this answer: a reply whose resolution failed stays in force and is told -/// again, one that someone else resolved, or whose answer was lost, is not; and one failure holds no other reply hostage. -fn hear_replies(organs: &Organs, run: &PipelineRef, heard: &[HeardReply], follow_up: Option<&PipelineRef>) { - for reply in heard { - let _ = resolve_reply(organs, run, reply, follow_up); - } -} - +/// One heard reply resolved: Answered by `follow_up`, else Recorded as heard. 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 { diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index b73b55b..604a702 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/crates/mind-body/src/persona/tests.rs @@ -4279,11 +4279,14 @@ struct Timeline<'a> { mind: &'a TestMind, doing: Rc>>>, reads: Rc>, + /// How many of the next keyed in-force reads of a reply fail, and how many of the next resolutions of a reply are refused by the port. + keyed_fail: Rc>, + admit_fail: Rc>, } impl<'a> Timeline<'a> { fn over(mind: &'a TestMind) -> Self { - Self { mind, doing: Rc::default(), reads: Rc::default() } + Self { mind, doing: Rc::default(), reads: Rc::default(), keyed_fail: Rc::default(), admit_fail: Rc::default() } } } @@ -4299,6 +4302,11 @@ impl MindPort for Timeline<'_> { if *selection == in_force(any_of(of_kind(K::Reply), "root", &[INSTANCE])) { self.reads.set(self.reads.get() + 1); } + let keyed = selection.keys.is_some() && cultnet_rs::Selection { keys: None, ..selection.clone() } == in_force(of_kind(K::Reply)); + if keyed && self.keyed_fail.get() > 0 { + self.keyed_fail.set(self.keyed_fail.get() - 1); + anyhow::bail!("the mind did not answer"); + } self.mind.query(selection) } @@ -4307,6 +4315,11 @@ impl MindPort for Timeline<'_> { } fn admit(&self, agent: &str, session: &str, documents: Vec) -> Result { + let resolves_a_reply = documents.iter().any(|document| matches!(document, PipelineDocument::Resolution(resolution) if resolution.subject.kind == K::Reply)); + if resolves_a_reply && self.admit_fail.get() > 0 { + self.admit_fail.set(self.admit_fail.get() - 1); + anyhow::bail!("the mind did not take it"); + } self.mind.admit(agent, session, documents) } } @@ -4323,11 +4336,17 @@ fn all_notices(lines: &[String]) -> Vec<&str> { /// next read, each past the next reply read and with a message, so a line carries what that read found; then two more quiet messages. /// Answers the lines. fn a_session_then(rig: &Rig, acts: Vec>) -> Vec { + a_session_with(rig, acts, |_| {}) +} + +/// `a_session_then` over a mind `flaws` has first set to fail. +fn a_session_with(rig: &Rig, acts: Vec>, flaws: impl FnOnce(&Timeline)) -> Vec { rig.model.replies.borrow_mut().extend(std::iter::repeat_n(says(&[]), acts.len() + 4)); rig.clock.step.set(1); rig.model.linger.set(true); rig.open_the_run(); let timeline = Timeline::over(&rig.mind); + flaws(&timeline); let (now, linger, deliveries, echoes) = (rig.clock.now.clone(), rig.model.linger.clone(), rig.chat.deliveries.clone(), Cell::new(1u64)); let doing_then = timeline.doing.clone(); let acts = RefCell::new(VecDeque::from(acts)); @@ -4373,6 +4392,87 @@ fn a_reply_admitted_while_her_session_lives_reaches_it_once_on_the_next_tell() { assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), heard_outcome()); } +/// How many resolutions `subject` has in the mind. +fn resolutions_of(rig: &Rig, subject: &str) -> usize { + rig.mind + .ids(K::Resolution) + .into_iter() + .filter(|id| matches!(rig.mind.view(K::Resolution, id).unwrap().document, PipelineDocument::Resolution(resolution) if resolution.subject.id.0 == subject)) + .count() +} + +/// A reply she answered is told once and resolved once however the resolution goes: a keyed read that fails after it took, or a resolution +/// the port refuses, is retried at the next result without telling her the reply again. +#[test] +fn an_answered_reply_is_told_once_and_resolved_once_when_the_read_or_the_admit_fails() { + for flaw in ["keyed read", "admit"] { + let rig = Rig::new(vec![operator_says(1, "hello")], &[]); + let mut act = her_self_asked(); + act.push(self_reply("self-r-r1", "The shaders are reviewed.")); + let lines = a_session_with(&rig, vec![act], |timeline| if flaw == "admit" { timeline.admit_fail.set(1) } else { timeline.keyed_fail.set(1) }); + assert_eq!(all_replies(&lines).len(), 1, "{flaw}: told once: {lines:?}"); + assert!(lines.len() >= 3, "{flaw}: lines followed her answer: {lines:?}"); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), heard_outcome(), "{flaw}"); + assert_eq!(resolutions_of(&rig, &reply_id("self-r-r1")), 1, "{flaw}: resolved once"); + } +} + +/// A resolution the port refuses at her last result is retried when the process closes. +#[test] +fn a_resolution_refused_at_her_last_result_is_retried_at_close() { + let rig = Rig::new(vec![operator_says(1, "hello")], &[says(&[])]); + rig.mind.committed(her_self_asked()); + rig.mind.committed(vec![self_reply("self-r-r1", "Done.")]); + rig.open_the_run(); + let timeline = Timeline::over(&rig.mind); + timeline.admit_fail.set(1); + assert_eq!(failure_of(rig.start_over(&timeline, &config())), None); + assert_eq!(all_replies(&told(&rig)).len(), 1); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), heard_outcome()); + assert_eq!(resolutions_of(&rig, &reply_id("self-r-r1")), 1); +} + +/// A reply told on a line her refused result did not answer: another writer resolves it meanwhile and it is not told again, nor does +/// the retell hear it; or the mind cannot say whether it is in force and it is told again, once, until her answer resolves it. +#[test] +fn a_refused_line_retells_a_reply_the_mind_holds_in_force_or_cannot_say_and_not_one_resolved_elsewhere() { + for flaw in ["resolved elsewhere", "unreadable"] { + let rig = Rig::new(vec![operator_says(1, "a")], &["not a reply at all".to_string(), says(&[])]); + rig.mind.committed(her_self_asked()); + rig.mind.committed(vec![self_reply("self-r-r1", "Done.")]); + rig.open_the_run(); + let timeline = Timeline::over(&rig.mind); + if flaw == "unreadable" { + timeline.keyed_fail.set(1); + } else { + let doing_then = timeline.doing.clone(); + let once = Cell::new(false); + *rig.model.before_result.borrow_mut() = Some(Box::new(move || { + if once.replace(true) { + return; + } + 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()), + })]); + })); + } + assert_eq!(failure_of(rig.start_over(&timeline, &config())), None, "{flaw}"); + let lines = told(&rig); + if flaw == "unreadable" { + assert_eq!(all_replies(&lines).len(), 2, "{flaw}: told on the refused line and again on the retell: {lines:?}"); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), heard_outcome(), "{flaw}"); + } else { + assert_eq!(all_replies(&lines).len(), 1, "{flaw}: told once: {lines:?}"); + assert_eq!(outcome_of(&rig, &reply_id("self-r-r1")), Some(ResolutionOutcome::Recorded { reason: "elsewhere".into() }), "{flaw}"); + } + assert_eq!(resolutions_of(&rig, &reply_id("self-r-r1")), 1, "{flaw}"); + } +} + /// However often chat is polled, her replies are read once at the start and then once in `REPLY_READ_EVERY_S`. #[test] fn her_replies_are_read_once_an_interval_not_once_a_poll() {