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

heartbeat: a standing vote stands until she replaces it; a winner that cannot start passes the beat on

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 07:48:27 +01:00
commit 79a64c1256
3 changed files with 100 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

71
crates/mind-body/src/heartbeat.rs
View file

@ -334,10 +334,6 @@ impl fmt::Display for Outcome {
}
}
struct Turn {
admitted: DateTime<Utc>,
}
fn parse_time(text: &str) -> Result<DateTime<Utc>> {
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<Vec<Turn>> {
/// 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<Option<DateTime<Utc>>> {
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<Vec<Turn>> {
"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<u16>,
}
/// 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<Vec<Pulse>> {
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<MindRead> {
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<Outcome> {
}
})
.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<Standing> = 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) {

54
crates/mind-body/src/heartbeat/tests.rs
View file

@ -208,9 +208,14 @@ impl Rig {
/// `who` admits a pulse, `ago` seconds before now.
fn pulse(&self, who: usize, ago: i64, beat: Option<u32>, urgency: Option<u16>) {
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<u32>, urgency: Option<u16>) {
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();

4
crates/mind-body/src/testkit.rs
View file

@ -45,6 +45,8 @@ impl Clock for Fixed {
pub(crate) struct Recording {
pub(crate) started: RefCell<Vec<(Slug, RunTurn)>>,
pub(crate) fail: Cell<bool>,
/// One named instance whose unit never starts.
pub(crate) fail_for: RefCell<Option<Slug>>,
running: RefCell<Vec<(Slug, RunTurn)>>,
}
@ -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));