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

persona: a reply out of force leaves for good; replies resolve one by one; idle and waiting-duplicate paths pinned

Browse source 
untell checks a reply is still in force before it waits again, and hear_replies resolves each reply on its own, so one refused
resolution neither retells a reply nobody can answer nor holds the rest of its batch. Tests: out-of-force retell, a reply reaching an
idle session with no chat pending, and a reply waiting across a read while a line is not echoed (a stalled child).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-06 22:57:42 +01:00
commit 8ab5959d46
2 changed files with 135 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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<HeardReply>) {
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<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.
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<String>, 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<HeardReply>, follow_up: Option<&PipelineRef>) -> Vec<HeardReply> {
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<bool> {
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.

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

@ -177,6 +177,8 @@ struct FakeSession {
refuse: Cell<bool>,
/// The child stays while replies are left, though the chat holds nothing: a line with no delivery can still come.
linger: Rc<Cell<bool>>,
/// How many reads a written line's echo is held back for: the driver sees timeouts, and polls, before it.
stall: Rc<Cell<u32>>,
}
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<Cell<bool>>,
stall: Rc<Cell<u32>>,
queue: VecDeque<Read>,
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() {