heartbeat: a successor trip is the Mind's stop, one question per cause, not one per brief

A Self run claims up to 64 briefs and each empty run counts against all of
them, so one cause tripped the briefs in turn, one question per brief per beat.
claimable now returns nothing while a question is open in any follow-up of the
Mind; the answer reopens the briefs with a fresh count.

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

crates/mind-body/src/heartbeat.rs

@@ -574,25 +574,25 @@ fn launch(organs: &Organs, row: &Row, turn: RunTurn, claims: Vec<PipelineRef>) -
 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.
+/// Self run holds. None while a question is open in a brief of hers. A Self run
+/// claims up to `BRIEFS_PER_SELF_RUN` briefs and every run it ends empty counts
+/// against all of them, so one cause trips them together; a trip is the Mind's
+/// stop, one question for the cause, not one per brief. It lasts until the
+/// operator answers or withdraws that question, and the count starts afresh.
 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)))?
+    let tripped = 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();
+        .any(|header| matches!(header.facts, PipelineFacts::Question { raised_in: Some(item), .. } if item.kind == PipelineKind::FollowUp));
+    if tripped {
+        return Ok(Vec::new());
+    }
     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())
 }

crates/mind-body/src/heartbeat/tests.rs

@@ -906,3 +906,60 @@ fn a_brief_that_keeps_failing_trips_once_and_a_trip_is_a_stop_until_the_question
     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");
 }
+
+/// Three Self runs that each claimed every one of `items` and ended leaving nothing.
+fn fail_bundle(rig: &Rig, items: &[PipelineRef]) {
+    for n in 0..3 {
+        let label = format!("failed-{n}");
+        rig.minds[ANN].committed(vec![run_in("ann", &label, RunTurn::SelfRun, RunOperator::Mind, items)]);
+        rig.minds[ANN].committed(vec![close(reference(K::Run, &format!("ann:run:{label}")), recorded())]);
+    }
+}
+
+/// Beats on which every live run ends at once; the Self runs and questions the Mind has after them.
+fn beats(rig: &Rig, count: usize) {
+    for _ in 0..count {
+        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);
+            }
+        }
+    }
+}
+
+#[test]
+fn one_cause_failing_a_bundle_of_briefs_is_one_question_not_one_per_brief() {
+    let (rig, source) = rig_with_a_persona_turn();
+    let labels = ["a", "b", "c", "d", "e"];
+    let items: Vec<PipelineRef> = labels.iter().map(|label| reference(K::FollowUp, &format!("eureka-body:follow_up:{label}"))).collect();
+    for label in labels {
+        brief_by(&rig, ANN, crate::persona::AGENT, label, &source);
+    }
+    fail_bundle(&rig, &items);
+    let before = self_runs(&rig, ANN).len();
+    beats(&rig, 10);
+    assert_eq!(rig.minds[ANN].ids(K::Question).len(), 1, "five briefs, one cause, one question on the desk");
+    assert_eq!(self_runs(&rig, ANN).len(), before, "and no Self run launches while it is open");
+    assert!(rig.minds[BOB].ids(K::Question).is_empty(), "another Mind is not stopped");
+}
+
+#[test]
+fn a_trip_stops_the_minds_successor_for_a_fresh_brief_too_and_the_answer_resumes_it() {
+    let (rig, source) = rig_with_a_persona_turn();
+    brief_by(&rig, ANN, crate::persona::AGENT, "stuck", &source);
+    fail_bundle(&rig, &[reference(K::FollowUp, "eureka-body:follow_up:stuck")]);
+    beats(&rig, 2);
+    let question = rig.minds[ANN].ids(K::Question).pop().expect("the trip asked");
+    let stopped = self_runs(&rig, ANN).len();
+    brief_by(&rig, ANN, crate::persona::AGENT, "fresh", &source);
+    beats(&rig, 4);
+    assert_eq!(self_runs(&rig, ANN).len(), stopped, "a brief that arrives while the question is open waits for the answer");
+    rig.minds[ANN].committed(vec![crate::testkit::ruling("answer", Some((&question, "reopen")))]);
+    beats(&rig, 1);
+    let runs = self_runs(&rig, ANN);
+    assert_eq!(runs.len(), stopped + 1, "answered, the successor launches again");
+    let claimed: Vec<String> = runs.last().unwrap().claims.iter().map(|claim| claim.id.0.clone()).collect();
+    assert_eq!(claimed.len(), 2, "both briefs, the stuck one reopened: {claimed:?}");
+}