From 79a64c125640096d3aae7f8d353eb69b074d25bc Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 07:48:27 +0100 Subject: [PATCH] heartbeat: a standing vote stands until she replaces it; a winner that cannot start passes the beat on A Mind's vote is her newest pulse that carries one, read back through the pages of her pulses rather than the newest eight. The pick passes over a winner whose launch is refused to the next eligible Mind, and her last turn is her newest turn that ran, however many withdrawn launches follow it. Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/heartbeat.rs | 71 +++++++++++++++---------- crates/mind-body/src/heartbeat/tests.rs | 54 ++++++++++++++++++- crates/mind-body/src/testkit.rs | 4 +- 3 files changed, 100 insertions(+), 29 deletions(-) diff --git a/crates/mind-body/src/heartbeat.rs b/crates/mind-body/src/heartbeat.rs index 6a60c77..fe39b03 100644 --- a/crates/mind-body/src/heartbeat.rs +++ b/crates/mind-body/src/heartbeat.rs @@ -334,10 +334,6 @@ impl fmt::Display for Outcome { } } -struct Turn { - admitted: DateTime, -} - fn parse_time(text: &str) -> Result> { DateTime::parse_from_rfc3339(text) .map(|at| at.with_timezone(&Utc)) @@ -372,8 +368,10 @@ fn latest_resolution( Ok(ending.map(|(_, at, outcome)| (at, outcome))) } -/// The instance's latest Persona runs, newest first, with how each ended. -fn recent_turns(mind: &dyn MindPort, instance: &Slug) -> Result> { +/// When her newest Persona turn that actually ran was admitted. A turn withdrawn +/// unrun never happened: it neither resets her wait nor counts as her last turn, +/// so the walk goes back past any number of them. +fn last_turn(mind: &dyn MindPort, instance: &Slug) -> Result>> { let turn = format!("{:?}", RunTurn::PersonaTurn); let operator = format!("{:?}", RunOperator::Mind); let runs = any_of( @@ -385,26 +383,30 @@ fn recent_turns(mind: &dyn MindPort, instance: &Slug) -> Result> { "operated_by", &[&operator], ); - let selection = Selection { + let mut asking = Selection { descending: true, limit: Some(RECENT), ..runs }; - let PipelinePageItems::Headers(runs) = mind.query(&selection)?.items else { - bail!("the mind answered documents to a header selection") - }; - let mut turns = Vec::new(); - for run in runs { - if !matches!(run.facts, PipelineFacts::Run { .. }) { - continue; + loop { + let page = mind.query(&asking)?; + let PipelinePageItems::Headers(runs) = page.items else { + bail!("the mind answered documents to a header selection") + }; + for run in runs { + if !matches!(run.facts, PipelineFacts::Run { .. }) { + continue; + } + if let Some((_, ResolutionOutcome::Withdrawn { .. })) = latest_resolution(mind, &run.id)? { + continue; + } + return Ok(Some(parse_time(&run.admission.admitted_at)?)); } - // A turn withdrawn unrun never happened: it neither resets her wait nor counts as her last turn. - if let Some((_, ResolutionOutcome::Withdrawn { .. })) = latest_resolution(mind, &run.id)? { - continue; + match page.next { + Some(cursor) => asking.cursor = Some(cursor), + None => return Ok(None), } - turns.push(Turn { admitted: parse_time(&run.admission.admitted_at)? }); } - Ok(turns) } /// Whether `text` holds `name` as a whole word, ignoring case: the characters on @@ -466,14 +468,24 @@ struct Pulse { urgency_permille: Option, } -/// Her latest pulses, newest first by the time the mind admitted them. +/// Her pulses, newest first by the time the mind admitted them, read back only as +/// far as her standing vote and her newest declared urgency: a vote stands until +/// she replaces it, however many pulses of other kinds follow. fn recent_pulses(mind: &dyn MindPort, instance: &Slug) -> Result> { - let selection = Selection { descending: true, limit: Some(RECENT), ..any_of(of_kind(PipelineKind::Pulse), "root", &[&instance.0]) }; - let PipelinePageItems::Headers(found) = mind.query(&selection)?.items else { bail!("the mind answered documents to a header selection") }; + let mut asking = Selection { descending: true, limit: Some(RECENT), ..any_of(of_kind(PipelineKind::Pulse), "root", &[&instance.0]) }; let mut pulses = Vec::new(); - for header in found { - if let PipelineFacts::Pulse { tempo_vote_beat_s, urgency_permille, .. } = header.facts { - pulses.push(Pulse { admitted: parse_time(&header.admission.admitted_at)?, beat_s: tempo_vote_beat_s, urgency_permille }); + loop { + let page = mind.query(&asking)?; + let PipelinePageItems::Headers(found) = page.items else { bail!("the mind answered documents to a header selection") }; + for header in found { + if let PipelineFacts::Pulse { tempo_vote_beat_s, urgency_permille, .. } = header.facts { + pulses.push(Pulse { admitted: parse_time(&header.admission.admitted_at)?, beat_s: tempo_vote_beat_s, urgency_permille }); + } + } + let read_enough = pulses.iter().any(|pulse| pulse.beat_s.is_some()) && pulses.iter().any(|pulse| pulse.urgency_permille.is_some()); + match page.next { + Some(cursor) if !read_enough => asking.cursor = Some(cursor), + _ => break, } } pulses.sort_by(|a, b| b.admitted.cmp(&a.admitted)); @@ -508,7 +520,7 @@ struct MindRead { fn read_mind(mind: &dyn MindPort, slug: &Slug) -> Result { Ok(MindRead { busy: !holders(mind, slug, RunTurn::PersonaTurn)?.is_empty(), - last_turn: recent_turns(mind, slug)?.first().map(|turn| turn.admitted), + last_turn: last_turn(mind, slug)?, pulses: recent_pulses(mind, slug)?, self_end: last_self_end(mind, slug)?, }) @@ -674,17 +686,22 @@ pub fn poll(organs: &Organs, config: &Config) -> Result { } }) .collect(); - if due && let Some(winner) = tempo::pick(&standings, beat_s) { + // A winner whose unit cannot start loses the beat to the next of the pick, so + // one failing Mind never starves the others; a launch ends the beat. + let mut passed: Vec = standings.clone(); + while due && let Some(winner) = tempo::pick(&passed, beat_s) { match launch(organs, &rows[winner])? { Launched::Run(label) => { state.last_beat_at = now; rows[winner].busy = true; launches.push(label); launched.push(rows[winner].slug().clone()); + break; } Launched::Busy => rows[winner].busy = true, Launched::Refused(why) => rows[winner].note = why, } + passed[winner].eligible = false; } if stored.as_ref() != Some(&state) { diff --git a/crates/mind-body/src/heartbeat/tests.rs b/crates/mind-body/src/heartbeat/tests.rs index 32844fb..3737ae5 100644 --- a/crates/mind-body/src/heartbeat/tests.rs +++ b/crates/mind-body/src/heartbeat/tests.rs @@ -208,9 +208,14 @@ impl Rig { /// `who` admits a pulse, `ago` seconds before now. fn pulse(&self, who: usize, ago: i64, beat: Option, urgency: Option) { - self.minds[who].admit_at(self.now() - Duration::seconds(ago)); let label = format!("p{}", self.labels.get()); self.labels.set(self.labels.get() + 1); + self.pulse_labelled(who, ago, &label, beat, urgency); + } + + fn pulse_labelled(&self, who: usize, ago: i64, label: &str, beat: Option, urgency: Option) { + self.minds[who].admit_at(self.now() - Duration::seconds(ago)); + let label = label.to_string(); self.minds[who].committed(vec![eureka_pipeline::PipelineDocument::Pulse(eureka_pipeline::PipelinePulse { instance: Slug(NAMES[who].into()), label: eureka_pipeline::Label(label), @@ -309,6 +314,53 @@ fn a_minds_newest_vote_replaces_her_old_one_and_a_pulse_with_no_vote_keeps_it() assert_eq!(trace.instances[ANN].vote_beat_s, Some(7200)); } +#[test] +fn a_standing_vote_stands_until_she_replaces_it_whatever_else_she_writes() { + let rig = Rig::new(); + rig.turn(ANN, 3000); + rig.turn(BOB, 3000); + rig.pulse(ANN, 2000, Some(3600), None); + for n in 1..=20 { + rig.pulse(ANN, 2000 - n * 10, None, Some(1)); + } + rig.pulse(BOB, 100, Some(86_400), None); + rig.poll(); + let trace = rig.store.traces.borrow().last().cloned().unwrap(); + assert_eq!(trace.instances[ANN].vote_beat_s, Some(3600)); + assert_eq!(trace.instances[ANN].urgency_permille, 1, "the newest declared urgency is read too"); + assert_eq!(trace.tempo_s, 3600); +} + +#[test] +fn her_newest_vote_is_the_latest_admitted_not_the_highest_label() { + let rig = Rig::new(); + rig.turn(ANN, 3000); + rig.turn(BOB, 3000); + rig.pulse_labelled(ANN, 300, "2026-10-31-t9", Some(600), None); + rig.pulse_labelled(ANN, 200, "2026-10-31-t10", Some(7200), None); + rig.pulse_labelled(ANN, 100, "2026-10-31-t11", None, Some(5)); + rig.poll(); + let trace = rig.store.traces.borrow().last().cloned().unwrap(); + assert_eq!(trace.instances[ANN].vote_beat_s, Some(7200)); +} + +#[test] +fn a_mind_whose_launch_keeps_failing_neither_starves_the_beat_nor_reads_as_never_acted() { + let rig = Rig::new(); + rig.turn(ANN, 5000); + rig.turn(BOB, 1000); + *rig.launcher.fail_for.borrow_mut() = Some(Slug("ann".into())); + for beat in 1..=12 { + rig.advance(1000); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![Slug("bob".into())] }, "beat {beat}"); + rig.end_run(BOB); + let trace = rig.store.traces.borrow().last().cloned().unwrap(); + assert!(trace.instances[ANN].note.contains("launch"), "beat {beat}: {:?}", trace.instances[ANN]); + assert!(trace.instances[ANN].idle_s >= 5000, "beat {beat}: she acted before; withdrawn launches do not erase that: {:?}", trace.instances[ANN]); + } + assert_eq!(rig.launched().iter().filter(|slug| slug.0 == "bob").count(), 12); +} + #[test] fn no_votes_is_the_default_beat() { let rig = Rig::new(); diff --git a/crates/mind-body/src/testkit.rs b/crates/mind-body/src/testkit.rs index fb5c70e..5017f7f 100644 --- a/crates/mind-body/src/testkit.rs +++ b/crates/mind-body/src/testkit.rs @@ -45,6 +45,8 @@ impl Clock for Fixed { pub(crate) struct Recording { pub(crate) started: RefCell>, pub(crate) fail: Cell, + /// One named instance whose unit never starts. + pub(crate) fail_for: RefCell>, running: RefCell>, } @@ -65,7 +67,7 @@ impl Recording { impl Launcher for Recording { fn start(&self, instance: &Slug, turn: RunTurn) -> Result<()> { - if self.fail.get() { + if self.fail.get() || self.fail_for.borrow().as_ref() == Some(instance) { return Err(anyhow!("refused")); } self.started.borrow_mut().push((instance.clone(), turn));