heartbeat: the beat gives a Self run to a Mind whose Self holds briefs; an open breaker question blocks its brief

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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<Launched> {
+/// 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<PipelineRef>) -> Result<Launched> {
     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<Launched> {
     })
 }
 
+/// 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<Vec<PipelineRef>> {
+    let mind = row.seat.mind;
+    if !holders(mind, row.slug(), RunTurn::SelfRun)?.is_empty() {
+        return Ok(Vec::new());
+    }
+    let blocked: Vec<PipelineRef> = 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<Outcome> {
     if let SwarmEffective::Held(reason) = organs.control.envelope() {
@@ -626,7 +661,7 @@ pub fn poll(organs: &Organs, config: &Config) -> Result<Outcome> {
         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<Outcome> {
     // 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])? {
+        // 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;

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<Rig>, 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<eureka_pipeline::PipelineRun> {
+    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<String> = 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");
+}