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

persona: an event answered once is not told again under a new seq

Browse source 
A mouth that restarts keeps its seq counter but loses its pending, so its unacked events come back from the homeserver under fresh,
higher seqs. The client keyed everything on seq and told them twice. Progress now remembers the answered event ids; a delivery whose event
was answered is dropped from the poll and acked through (leading run only).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Eureka 2026-11-11 16:57:01 +00:00
commit 5f7314c668
1 changed files with 25 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -858,6 +858,9 @@ struct Progress {
self_run: Option<PipelineRef>,
/// The batch refused and not yet answered: a session restart does not reset its count or what she is told.
refused: Option<Refused>,
/// The event ids of every delivery answered in this start. A mouth that restarts re-delivers its unacked events under fresh seqs
/// (its counter survives, its pending does not), so a seq says nothing of whether an event was answered; the event id does.
answered_events: BTreeSet<String>,
}
/// A refused batch as the start remembers it. The mouth still holds its deliveries, so this needs only their lowest seq.
@ -944,6 +947,13 @@ pub fn carry(organs: &Organs, control: &impl ControlSource, swarm: &impl Control
let table = addressing.iter().find(|mind| mind.matrix_id() == config.own_sender).cloned().unwrap_or_else(|| own_addressee(config));
let (fresh_names, fresh_repos) = stewardships_of(organs.mind).unwrap_or_default();
let fresh = own_addressee(config).with_stewardships(fresh_names, fresh_repos);
if let Some(through) = drop_answered(&progress.answered_events, &mut pending).filter(|through| *through > progress.acked) {
if organs.chat.ack(through).is_err() {
progress.failure = Some(Failure::Ack);
break;
}
progress.acked = through;
}
if let Some(through) = quiet_through(&[&table, &fresh], &pending, 0) {
if organs.chat.ack(through).is_err() {
progress.failure = Some(Failure::Ack);
@ -1696,6 +1706,11 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
self.note_repo_changes();
let mut pending = self.organs.chat.pending(self.progress.acked)?.deliveries;
pending.sort_by_key(|delivery| delivery.seq);
if let Some(through) = drop_answered(&self.progress.answered_events, &mut pending).filter(|through| *through > self.progress.acked)
&& self.organs.chat.ack(through).is_ok()
{
self.progress.acked = through;
}
// Whom a delivery is for is the heartbeat's table as it stands, her own row first: it knows her projects' names before her
// first line has read them.
if !pending.is_empty() {
@ -1839,6 +1854,7 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
// Her final answer lands after she took in every line, so a revision still waiting is answered by it and its note, that a send was not sent, would be false.
let spent = if final_reply { self.revision.take().map(|revision| revision.batch) } else { None };
self.answered_through = self.answered_through.max(batch.iter().chain(spent.iter().flatten()).map(|delivery| delivery.seq).max().unwrap_or(0));
self.progress.answered_events.extend(batch.iter().chain(spent.iter().flatten()).map(|delivery| delivery.event_id.clone()));
if self.progress.refused.as_ref().is_some_and(|refused| batch.iter().any(|delivery| delivery.seq == refused.lowest)) {
self.progress.refused = None;
}
@ -2122,6 +2138,15 @@ fn quiet_through(own: &[&Addressee], pending: &[ChatDelivery], told_through: u64
run.last().map(|last| last.seq)
}
/// Takes out of `pending` every delivery whose event was already answered (a mouth that restarted numbered it again) and returns the seq an
/// ack may pass for them: the end of the leading run of such deliveries, which no unanswered one lies below. The rest are passed by the next
/// ack, which is above them.
fn drop_answered(answered: &BTreeSet<String>, pending: &mut Vec<ChatDelivery>) -> Option<u64> {
let through = pending.iter().take_while(|delivery| answered.contains(&delivery.event_id)).last().map(|delivery| delivery.seq);
pending.retain(|delivery| !answered.contains(&delivery.event_id));
through
}
/// How long after a poll first sees `delivery` a live session is told it: none in a direct message, otherwise her rank times the
/// step. The Minds the message is for (tagged, named, or named by a project, by `address::addresses` over the swarm's addressing)
/// rank first, by a hash of the Mind and the event, and the rest follow them, ranked the same way: the first addressed Mind hears