diff --git a/crates/mind-body/src/heartbeat.rs b/crates/mind-body/src/heartbeat.rs index 2d46291..a897478 100644 --- a/crates/mind-body/src/heartbeat.rs +++ b/crates/mind-body/src/heartbeat.rs @@ -9,8 +9,13 @@ //! waited, bent by the urgency she declared for herself, a member addressing her //! and (a later cut) topics she subscribed to. Two things launch a Persona off //! the beat: the operator tagging or naming her, once per message, and her own -//! Self run having ended since her last turn. Nothing is metered here: no -//! meter and no model call. +//! Self run having ended since her last turn. A beat's launch for a Mind whose +//! Self holds briefs (`inbox::derive`) is a Self run claiming them, in place of +//! her Persona turn: a session that ended with a brief open, or a brief admitted +//! after its flush, is never an orphan, and a Self start takes the beat like any +//! launch. A brief with an open question raised in it (the repetition breaker's +//! trip) is not claimed until that question is answered. Nothing is metered here: +//! no meter and no model call. //! //! The heartbeat owns `heartbeat.cc` and is its only writer; Persona units read //! it. The brakes stay the operator's and are read on every poll. Busy and idle @@ -21,6 +26,7 @@ //! A held swarm brake ends the poll before the mind or the mouth is touched. use std::fmt; +use std::collections::HashSet; use std::net::SocketAddr; use std::path::PathBuf; @@ -34,7 +40,8 @@ use huginn_mind::{PipelineDocumentSummary, PipelineFacts, PipelinePageItems}; use crate::chat_client::{Chat, ChatDelivery}; use crate::control::{ControlSource, Effective, FileSource, HeldReason, SwarmEffective, read_effective, read_envelope, set_mode_world_readable}; use crate::launch::{Clock, LaunchOutcome, LaunchRequest, Launcher, Ports, holders, open_and_launch}; -use crate::queue::{MindPort, any_of, citing, headers, of_kind}; +use crate::queue::{MindPort, any_of, citing, headers, in_force, of_kind}; +use crate::self_run::inbox; use crate::tempo::{self, Standing}; /// The admitting agent's name on every run the heartbeat opens. @@ -548,12 +555,13 @@ enum Launched { Refused(String), } -/// The one way the heartbeat starts a Persona turn: the same brake read and the -/// same opener for a beat, an operator mention and a Self run that ended. -fn launch(organs: &Organs, row: &Row) -> Result { +/// The one way the heartbeat starts a run: the same brake read and the same +/// opener for a Persona turn (a beat, an operator mention, a Self run that ended) +/// and for the Self run a beat gives to the briefs her Self holds. +fn launch(organs: &Organs, row: &Row, turn: RunTurn, claims: Vec) -> Result { let Effective::Released(grant) = organs.control.effective(row.slug()) else { return Ok(Launched::Refused("the brake is held".into())) }; let ports = Ports { mind: row.seat.mind, launcher: organs.launcher, clock: organs.clock, host: organs.host }; - let request = LaunchRequest { turn: RunTurn::PersonaTurn, claims: vec![], agent: AGENT.into(), grant }; + let request = LaunchRequest { turn, claims, agent: AGENT.into(), grant }; Ok(match open_and_launch(&ports, request) { Ok(LaunchOutcome::Launched { run }) => Launched::Run(label_of(&run)), Ok(LaunchOutcome::Busy { .. }) => Launched::Busy, @@ -562,6 +570,33 @@ fn launch(organs: &Organs, row: &Row) -> Result { }) } +/// The most briefs one Self run claims. +const BRIEFS_PER_SELF_RUN: usize = 64; + +/// The briefs a Self run would claim for this Mind: those in force that no live +/// Self run holds and that no open question blocks. A question raised in a brief +/// is the repetition breaker's trip; it stays blocking until it is answered or +/// withdrawn, so a trip is a stop and never a fresh count. +fn claimable(row: &Row) -> Result> { + let mind = row.seat.mind; + if !holders(mind, row.slug(), RunTurn::SelfRun)?.is_empty() { + return Ok(Vec::new()); + } + let blocked: Vec = headers(mind, &in_force(of_kind(PipelineKind::Question)))? + .into_iter() + .filter_map(|header| match header.facts { + PipelineFacts::Question { raised_in: Some(item), .. } => Some(item), + _ => None, + }) + .collect(); + Ok(inbox::derive(mind, &HashSet::new())? + .into_iter() + .map(|brief| PipelineRef { kind: PipelineKind::FollowUp, id: brief.id.as_str().into() }) + .filter(|claim| !blocked.contains(claim)) + .take(BRIEFS_PER_SELF_RUN) + .collect()) +} + /// One poll. pub fn poll(organs: &Organs, config: &Config) -> Result { if let SwarmEffective::Held(reason) = organs.control.envelope() { @@ -626,7 +661,7 @@ pub fn poll(organs: &Organs, config: &Config) -> Result { if mentioned.is_none() && !self_ended { continue; } - match launch(organs, row)? { + match launch(organs, row, RunTurn::PersonaTurn, vec![])? { Launched::Run(label) => { if let Some(key) = mentioned { state.handled_mentions.push(key); @@ -664,15 +699,21 @@ pub fn poll(organs: &Organs, config: &Config) -> Result { // 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])? { + // Her Self holds briefs: the beat is a Self run for them; otherwise a Persona turn. + let claims = claimable(&rows[winner]).unwrap_or_else(|error| { + rows[winner].note = format!("briefs: {error}"); + Vec::new() + }); + let turn = if claims.is_empty() { RunTurn::PersonaTurn } else { RunTurn::SelfRun }; + match launch(organs, &rows[winner], turn, claims)? { Launched::Run(label) => { state.last_beat_at = now; - rows[winner].busy = true; + rows[winner].busy = turn == RunTurn::PersonaTurn; launches.push(label); launched.push(rows[winner].slug().clone()); break; } - Launched::Busy => rows[winner].busy = true, + Launched::Busy => rows[winner].busy = turn == RunTurn::PersonaTurn, Launched::Refused(why) => rows[winner].note = why, } passed[winner].eligible = false; diff --git a/crates/mind-body/src/heartbeat/tests.rs b/crates/mind-body/src/heartbeat/tests.rs index 5e10aab..4203827 100644 --- a/crates/mind-body/src/heartbeat/tests.rs +++ b/crates/mind-body/src/heartbeat/tests.rs @@ -780,3 +780,119 @@ fn every_newest_read_asks_for_exactly_one_header() { let queries = rig.calls[ANN].get(); assert!(queries >= 3, "the poll did read her mind ({queries} queries)"); } + +// ---- the beat gives a Self run to a Mind whose Self holds briefs ---------------------- + +use eureka_pipeline::PipelineDocument; + +/// A self-run follow-up for `who`, admitted under `agent`'s receipt. +fn brief_by(rig: &Rig, who: usize, agent: &str, label: &str, source: &PipelineRef) { + let PipelineDocument::FollowUp(mut document) = crate::testkit::follow_up(label, source) else { unreachable!() }; + document.owner = "self-run".into(); + let outcome = rig.minds[who].admit(agent, "session-a", vec![PipelineDocument::FollowUp(document)]).unwrap(); + assert!(matches!(outcome, huginn_mind::PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}"); +} + +/// Two Minds that have both acted, ann longest ago, and a Persona turn of hers to cite. +fn rig_with_a_persona_turn() -> (Box, PipelineRef) { + let rig = Rig::new(); + let label = rig.turn(ANN, 3000); + rig.turn(BOB, 1000); + let source = reference(K::Run, &format!("ann:run:{label}")); + (rig, source) +} + +fn self_runs(rig: &Rig, who: usize) -> Vec { + rig.minds[who] + .ids(K::Run) + .iter() + .filter_map(|id| match rig.minds[who].view(K::Run, id)?.document { + PipelineDocument::Run(run) if run.turn == RunTurn::SelfRun => Some(run), + _ => None, + }) + .collect() +} + +#[test] +fn the_beat_won_by_a_mind_whose_self_holds_briefs_is_one_self_run_claiming_them_and_the_outcome_says_so() { + let (rig, source) = rig_with_a_persona_turn(); + brief_by(&rig, ANN, crate::persona::AGENT, "first", &source); + brief_by(&rig, ANN, crate::persona::AGENT, "second", &source); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![Slug("ann".into())] }, "the launch is in the outcome, not only the trace"); + let runs = self_runs(&rig, ANN); + assert_eq!(runs.len(), 1, "one Self run for both briefs"); + let claimed: Vec = runs[0].claims.iter().map(|claim| claim.id.0.clone()).collect(); + assert_eq!(claimed, vec!["eureka-body:follow_up:first".to_string(), "eureka-body:follow_up:second".to_string()]); + assert_eq!(rig.launcher.started.borrow().iter().filter(|(_, turn)| *turn == RunTurn::SelfRun).count(), 1); + assert!(self_runs(&rig, BOB).is_empty(), "bob holds no brief"); + assert_eq!(rig.store.traces.borrow().last().unwrap().launches.len(), 1); +} + +#[test] +fn a_self_run_waits_for_the_beat_like_any_launch() { + let (rig, source) = rig_with_a_persona_turn(); + rig.poll(); + brief_by(&rig, ANN, crate::persona::AGENT, "waiting", &source); + rig.end_run(ANN); + rig.advance(60); + assert_eq!(rig.poll(), Outcome::Polled { launched: vec![] }, "the beat has not passed: no Self run starts"); + assert!(self_runs(&rig, ANN).is_empty()); + rig.advance(900); + let Outcome::Polled { launched } = rig.poll() else { panic!("held") }; + assert_eq!(launched.len(), 1, "one launch for the beat"); +} + +#[test] +fn a_live_self_run_holds_the_briefs_and_the_beat_is_a_persona_turn() { + let (rig, source) = rig_with_a_persona_turn(); + brief_by(&rig, ANN, crate::persona::AGENT, "taken", &source); + rig.poll(); + assert_eq!(self_runs(&rig, ANN).len(), 1); + rig.advance(1000); + rig.poll(); + assert_eq!(self_runs(&rig, ANN).len(), 1, "a live Self run is never given a second"); +} + +#[test] +fn a_follow_up_her_persona_did_not_admit_launches_no_self_run_whatever_source_it_names() { + let (rig, source) = rig_with_a_persona_turn(); + brief_by(&rig, ANN, crate::self_run::AGENT, "hers", &source); + brief_by(&rig, ANN, "test", "by-hand", &source); + rig.poll(); + assert!(self_runs(&rig, ANN).is_empty()); +} + +#[test] +fn a_braked_mind_launches_no_self_run() { + let (rig, source) = rig_with_a_persona_turn(); + brief_by(&rig, ANN, crate::persona::AGENT, "waiting", &source); + rig.control.braked.borrow_mut().push("ann".into()); + rig.poll(); + assert!(self_runs(&rig, ANN).is_empty()); +} + +#[test] +fn a_brief_that_keeps_failing_trips_once_and_a_trip_is_a_stop_until_the_question_is_answered() { + let (rig, source) = rig_with_a_persona_turn(); + brief_by(&rig, ANN, crate::persona::AGENT, "stuck", &source); + let item = reference(K::FollowUp, "eureka-body:follow_up:stuck"); + // Three Self runs have ended on it, each leaving nothing. + for n in 0..3 { + let label = format!("failed-{n}"); + rig.minds[ANN].committed(vec![run_in("ann", &label, RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item))]); + rig.minds[ANN].committed(vec![close(reference(K::Run, &format!("ann:run:{label}")), recorded())]); + } + let before = self_runs(&rig, ANN).len(); + // Beat after beat, whoever launches: no new Self run on the brief, and the breaker asks once. + for _ in 0..8 { + rig.advance(1000); + rig.poll(); + for who in [ANN, BOB] { + if rig.minds[who].ids(K::Run).last().is_some_and(|id| rig.minds[who].view(K::Run, id).is_some_and(|view| matches!(view.status, huginn_mind::PipelineStatus::InForce))) { + rig.end_run(who); + } + } + } + assert_eq!(self_runs(&rig, ANN).len(), before, "no Self run after the trip"); + assert_eq!(rig.minds[ANN].ids(K::Question).len(), 1, "the breaker asked once; an open question is not asked again nor a fresh count"); +}