GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

persona: once she answered a line, its heard replies never return to waiting; resolution is retried silently

Browse source 
A heard reply whose resolution did not take (admit failed, or the keyed read errs) went back through untell and was told again. The answered
replies now wait in answered_replies, retried at each result and at close (settle_replies), one by one so no refusal holds another hostage;
untell serves only the lines she did not answer. Tests pin the retry, the close retry, the reply resolved elsewhere on a refused line, and the
unreadable in-force read on that path (told again, kept as retell).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-06 22:55:22 +00:00 • committed by Meta Vens
commit 26303f045b
2 changed files with 126 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

39
crates/mind-body/src/persona.rs
View file

@ -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<HeardReply>,
/// The replies the CLI took in since the last result: that result is what answers them.
echoed_replies: Vec<HeardReply>,
/// 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<PipelineRef>)>,
/// 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<HeardReply>) {
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<C: ControlSource, S: ControlSource> 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<String>, 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 {

102
crates/mind-body/src/persona/tests.rs
View file

@ -4279,11 +4279,14 @@ struct Timeline<'a> {
mind: &'a TestMind,
doing: Rc<RefCell<VecDeque<Vec<PipelineDocument>>>>,
reads: Rc<Cell<u32>>,
/// 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<Cell<u32>>,
admit_fail: Rc<Cell<u32>>,
}
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<PipelineDocument>) -> Result<AdmissionOutcome> {
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<PipelineDocument>>) -> Vec<String> {
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<Vec<PipelineDocument>>, flaws: impl FnOnce(&Timeline)) -> Vec<String> {
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() {