From f624feec75f08246dbc6fc2a86c1775db794fe7d Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Sat, 10 Oct 2026 05:32:19 +0200
Subject: [PATCH] mind-body launch: dead runs closed in the opening batch, Busy
 from admission, breaker over the item's citations

open_and_launch takes a Grant by value. Step 0 reads the in-force runs of
hers of the request's turn on the instance (typed facts): a live unit or a
run inside START_GRACE_S answers Busy; a dead one is closed Recorded in the
same admission that opens the next run, so the in-force record is the one
owner of liveness. Admission's AlreadyLive maps to Busy. The alias Busy query
and the whole-mind breaker scan are deleted: the breaker counts the documents
that cite the item, starting after the newest question raised in it, and a
trip's question is keyed by the run label. SystemdLauncher holds no
instance; unit names are validated before systemctl sees them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/mind-body/src/launch.rs  | 557 +++++++++++++++++++++++++-------
 crates/mind-body/src/queue.rs   | 205 +++++++-----
 crates/mind-body/src/testkit.rs |  40 ++-
 3 files changed, 594 insertions(+), 208 deletions(-)

diff --git a/crates/mind-body/src/launch.rs b/crates/mind-body/src/launch.rs
index 922b9f4..cdd312a 100644
--- a/crates/mind-body/src/launch.rs
+++ b/crates/mind-body/src/launch.rs
@@ -1,29 +1,40 @@
 //! The one way her organs open a run and start its unit.
 //!
-//! `open_and_launch` is the whole path: Busy, queue membership and the
-//! repetition breaker are checked in that order, the run is admitted into the
-//! mind, and only after it commits is the unit started. The run is the grant
-//! (target invariant run-is-the-grant), so there is no pressure file, lease or
-//! breaker state beside it, and a unit never starts without a committed run.
-//! The primitive owns no file and stops no unit: stopping her units is the
-//! operator's brake hold, which is another cut's.
+//! `open_and_launch` is the whole path, in this order: (0) the in-force runs of
+//! hers of the request's turn are read, a live one (its unit active, or
+//! admitted within `START_GRACE_S`) answers Busy and a dead one is to be closed
+//! Recorded; (1) every cut-spec claim must be in the queue; (2) no claim may
+//! trip the repetition breaker; (3) the run is admitted, in the one batch that
+//! closes the dead holders, so liveness is decided by the in-force record alone
+//! and the unit table is only the evidence for a close; (4) the unit is started,
+//! only after that batch commits. The run is the grant (target invariant
+//! run-is-the-grant), so there is no pressure file, lease or breaker state
+//! beside it, and a unit never starts without a committed run. A run opens
+//! only against a `Grant`, which `read_effective` alone makes. The primitive
+//! owns no file and stops no unit: stopping her units is the operator's brake
+//! hold, which is another cut's.
 
 use anyhow::{Result, anyhow};
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, Duration, Utc};
+use cultnet_rs::Selection;
 use eureka_pipeline::{
-    Date, Label, PipelineDocument, PipelineKind, PipelineQuestion, PipelineRef, PipelineResolution, PipelineRun, QuestionOption,
-    ResolutionOutcome, RunOperator, RunTurn, Slug,
+    Date, Label, PipelineDocument, PipelineKind, PipelineQuestion, PipelineRef, PipelineResolution, PipelineRun, QuestionOption, ResolutionOutcome,
+    RunOperator, RunTurn, Slug, pipeline_key,
 };
-use huginn_mind::{MindRefusal, PipelineAdmissionOutcome};
-use rust_decimal::Decimal;
+use huginn_mind::{MindRefusal, PipelineAdmissionOutcome, PipelineFacts};
 
-use crate::queue::{Breaker, MindPort, any_of, breaker, headers, in_force, of_kind, queue};
+use crate::control::Grant;
+use crate::queue::{Breaker, MindPort, any_of, breaker, citing, headers, in_force, of_kind, queue};
+
+/// How long a run just admitted may have no active unit before its slot is
+/// treated as dead: `systemctl start --no-block` returns before the unit runs.
+pub const START_GRACE_S: i64 = 120;
 
 /// Starts the unit that carries a run, and says whether it is still running.
 /// Nothing here stops a unit.
 pub trait Launcher {
-    fn start(&self, turn: RunTurn, label: &Label) -> Result<()>;
-    fn alive(&self, turn: RunTurn, label: &Label) -> bool;
+    fn start(&self, instance: &Slug, turn: RunTurn, label: &Label) -> Result<()>;
+    fn alive(&self, instance: &Slug, turn: RunTurn, label: &Label) -> bool;
 }
 
 /// The time the primitive reads: run labels, dates and receipts.
@@ -61,42 +72,53 @@ impl Systemctl for RealSystemctl {
 
 /// Her units under systemd, templated by instance:
 /// `mind-persona@<instance>:<label>.service` and
-/// `mind-self@<instance>:<label>.service`.
+/// `mind-self@<instance>:<label>.service`. The launcher holds no instance: the
+/// caller names it on every call, and `open_and_launch` always names the
+/// mind's own.
 pub struct SystemdLauncher {
-    instance: Slug,
     systemctl: Box<dyn Systemctl>,
 }
 
 impl SystemdLauncher {
-    pub fn new(instance: Slug) -> Self {
-        Self::over(instance, Box::new(RealSystemctl))
+    pub fn new() -> Self {
+        Self::over(Box::new(RealSystemctl))
     }
 
-    pub(crate) fn over(instance: Slug, systemctl: Box<dyn Systemctl>) -> Self {
-        Self { instance, systemctl }
-    }
-
-    /// The unit of one run: the template by turn, then the instance and the
-    /// label. Start and alive both name it here.
-    pub(crate) fn unit(&self, turn: RunTurn, label: &Label) -> String {
-        let template = match turn {
-            RunTurn::PersonaTurn => "mind-persona",
-            RunTurn::SelfRun => "mind-self",
-        };
-        format!("{template}@{}:{}.service", self.instance.0, label.0)
+    pub(crate) fn over(systemctl: Box<dyn Systemctl>) -> Self {
+        Self { systemctl }
     }
 }
 
+impl Default for SystemdLauncher {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// The unit of one run: the template by turn, then the instance and the label,
+/// both held to the leaf's grammar first so neither can escape the name. The
+/// error names neither value. Start and alive both name the unit here.
+pub(crate) fn unit(instance: &Slug, turn: RunTurn, label: &Label) -> Result<String> {
+    instance.validate_slug().map_err(|_| anyhow!("the instance is not a unit name"))?;
+    label.validate_label().map_err(|_| anyhow!("the label is not a unit name"))?;
+    let template = match turn {
+        RunTurn::PersonaTurn => "mind-persona",
+        RunTurn::SelfRun => "mind-self",
+    };
+    Ok(format!("{template}@{}:{}.service", instance.0, label.0))
+}
+
 impl Launcher for SystemdLauncher {
-    fn start(&self, turn: RunTurn, label: &Label) -> Result<()> {
-        let reply = self.systemctl.run(&["start", "--no-block", &self.unit(turn, label)])?;
+    fn start(&self, instance: &Slug, turn: RunTurn, label: &Label) -> Result<()> {
+        let reply = self.systemctl.run(&["start", "--no-block", &unit(instance, turn, label)?])?;
         if reply.success { Ok(()) } else { Err(anyhow!("systemctl did not start the unit")) }
     }
 
-    fn alive(&self, turn: RunTurn, label: &Label) -> bool {
-        self.systemctl
-            .run(&["is-active", &self.unit(turn, label)])
-            .is_ok_and(|reply| matches!(reply.stdout.trim(), "active" | "activating" | "reloading"))
+    fn alive(&self, instance: &Slug, turn: RunTurn, label: &Label) -> bool {
+        let Ok(unit) = unit(instance, turn, label) else {
+            return false;
+        };
+        self.systemctl.run(&["is-active", &unit]).is_ok_and(|reply| matches!(reply.stdout.trim(), "active" | "activating" | "reloading"))
     }
 }
 
@@ -120,8 +142,9 @@ pub struct LaunchRequest {
     pub claims: Vec<PipelineRef>,
     /// The admitting agent's name on the receipt.
     pub agent: String,
-    /// The most one run may spend, from the dial's `read_effective`.
-    pub run_cap_usd: Decimal,
+    /// The reading of a released brake and an in-bounds dial; its cap is the
+    /// run's budget. Taken by value, so one reading opens at most one run.
+    pub grant: Grant,
 }
 
 /// Why admission did not open the run: its own refusal, or one of the two
@@ -137,17 +160,23 @@ pub enum Declined {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum LaunchOutcome {
-    Launched { run: PipelineRef },
-    NotQueued { item: PipelineRef },
-    Tripped { item: PipelineRef, question: PipelineRef },
-    Busy { run: PipelineRef },
+    Launched {
+        run: PipelineRef,
+    },
+    NotQueued {
+        item: PipelineRef,
+    },
+    Tripped {
+        item: PipelineRef,
+        question: PipelineRef,
+    },
+    /// The live holder of the request's turn, from step 0 or from admission.
+    Busy {
+        run: PipelineRef,
+    },
     Refused(Declined),
 }
 
-fn opened_run(instance: &Slug, label: &Label) -> PipelineRef {
-    PipelineRef { kind: PipelineKind::Run, id: format!("{}:run:{}", instance.0, label.0).as_str().into() }
-}
-
 fn date(now: DateTime<Utc>) -> Date {
     Date(now.format("%Y-%m-%d").to_string())
 }
@@ -157,30 +186,77 @@ fn root_of(item: &PipelineRef) -> &str {
     item.id.0.split(':').next().unwrap_or_default()
 }
 
+fn key_of(document: &PipelineDocument) -> Result<String> {
+    pipeline_key(document).map_err(|_| anyhow!("the document could not be keyed"))
+}
+
+/// Whether a run admitted at `admitted_at` has had `START_GRACE_S` to start
+/// its unit. A stamp that does not parse has not.
+fn past_grace(admitted_at: &str, now: DateTime<Utc>) -> bool {
+    DateTime::parse_from_rfc3339(admitted_at).is_ok_and(|at| now.signed_duration_since(at) >= Duration::seconds(START_GRACE_S))
+}
+
+/// The resolution that closes a dead run of hers: Recorded, not Withdrawn, so
+/// a unit that keeps dying counts as an empty run and trips the breaker. Its
+/// sequence is the run's next, counting a closure that was itself withdrawn.
+fn close_dead(mind: &dyn MindPort, run: &PipelineRef, now: DateTime<Utc>) -> Result<PipelineDocument> {
+    let closures = Selection { cites: Some(citing(run, Some("subject"))), ..of_kind(PipelineKind::Resolution) };
+    let latest = headers(mind, &closures)?
+        .iter()
+        .filter_map(|header| match header.facts {
+            PipelineFacts::Resolution { sequence, .. } => Some(sequence),
+            _ => None,
+        })
+        .max()
+        .unwrap_or(0);
+    Ok(PipelineDocument::Resolution(PipelineResolution {
+        subject: run.clone(),
+        sequence: latest + 1,
+        outcome: ResolutionOutcome::Recorded { reason: "unit not active".into() },
+        rationale: "The run's unit was not active when its turn's slot was next wanted.".into(),
+        resolved_on: date(now),
+    }))
+}
+
 pub fn open_and_launch(ports: &Ports, request: LaunchRequest) -> Result<LaunchOutcome> {
-    let LaunchRequest { turn, claims, agent, run_cap_usd } = request;
+    let LaunchRequest { turn, claims, agent, grant } = request;
     let instance = ports.mind.instance().clone();
     let now = ports.clock.now();
     let label = run_label(now);
 
-    if turn == RunTurn::SelfRun {
-        let live = in_force(any_of(any_of(any_of(of_kind(PipelineKind::Run), "root", &[&instance.0]), "turn", &["SelfRun"]), "operated_by", &["Mind"]));
-        if let Some(run) = headers(ports.mind, &live)?.into_iter().next() {
-            return Ok(LaunchOutcome::Busy { run: run.id });
+    // Step 0: a live holder of this turn's slot answers Busy; a dead one is closed
+    // in the batch that admits whatever comes next.
+    let mut batch: Vec<PipelineDocument> = Vec::new();
+    let mut ending: Vec<PipelineRef> = Vec::new();
+    for holder in headers(ports.mind, &in_force(any_of(of_kind(PipelineKind::Run), "root", &[&instance.0])))? {
+        let PipelineFacts::Run { label: held, turn: held_turn, operated_by: RunOperator::Mind, .. } = &holder.facts else {
+            continue;
+        };
+        if *held_turn != turn {
+            continue;
         }
+        if ports.launcher.alive(&instance, turn, held) || !past_grace(&holder.admission.admitted_at, now) {
+            return Ok(LaunchOutcome::Busy { run: holder.id });
+        }
+        batch.push(close_dead(ports.mind, &holder.id, now)?);
+        ending.push(holder.id);
     }
-    let open = queue(ports.mind, &instance)?;
-    if let Some(item) = claims.iter().find(|claim| claim.kind == PipelineKind::CutSpec && !open.contains(claim)) {
-        return Ok(LaunchOutcome::NotQueued { item: item.clone() });
-    }
-    for item in &claims {
-        if breaker(ports.mind, item)? == Breaker::Trip {
-            let question = breaker_question(item, now);
-            let id = PipelineRef { kind: PipelineKind::Question, id: format!("{}:question:{}", root_of(item), breaker_label(item).0).as_str().into() };
-            return Ok(match committed(ports.mind.admit(&agent, &label.0, vec![PipelineDocument::Question(question)])?) {
-                Ok(()) => LaunchOutcome::Tripped { item: item.clone(), question: id },
-                Err(declined) => LaunchOutcome::Refused(declined),
-            });
+
+    if turn == RunTurn::SelfRun {
+        let open = queue(ports.mind, &instance, &ending)?;
+        if let Some(item) = claims.iter().find(|claim| claim.kind == PipelineKind::CutSpec && !open.contains(claim)) {
+            return Ok(LaunchOutcome::NotQueued { item: item.clone() });
+        }
+        for item in &claims {
+            if breaker(ports.mind, item, &ending)? == Breaker::Trip {
+                let question = PipelineDocument::Question(breaker_question(item, &label, now));
+                let id = PipelineRef { kind: PipelineKind::Question, id: key_of(&question)?.as_str().into() };
+                batch.push(question);
+                return Ok(match committed(ports.mind.admit(&agent, &label.0, batch)?) {
+                    Ok(()) => LaunchOutcome::Tripped { item: item.clone(), question: id },
+                    Err(declined) => LaunchOutcome::Refused(declined),
+                });
+            }
         }
     }
 
@@ -191,24 +267,32 @@ pub fn open_and_launch(ports: &Ports, request: LaunchRequest) -> Result<LaunchOu
             campaigns.push(root);
         }
     }
-    let run = PipelineRun {
+    let run = PipelineDocument::Run(PipelineRun {
         instance: instance.clone(),
         label: label.clone(),
         operated_by: RunOperator::Mind,
         turn,
         host: ports.host.into(),
         started_on: date(now),
-        budget_usd: run_cap_usd.normalize().to_string().as_str().into(),
+        budget_usd: grant.run_cap_usd().normalize().to_string().as_str().into(),
         claims,
         campaigns,
-    };
-    if let Err(declined) = committed(ports.mind.admit(&agent, &label.0, vec![PipelineDocument::Run(run)])?) {
-        return Ok(LaunchOutcome::Refused(declined));
+    });
+    let opened = PipelineRef { kind: PipelineKind::Run, id: key_of(&run)?.as_str().into() };
+    batch.push(run);
+    match ports.mind.admit(&agent, &label.0, batch)? {
+        PipelineAdmissionOutcome::Refused(MindRefusal::AlreadyLive { run: holder }) => {
+            return Ok(LaunchOutcome::Busy { run: holder.to_ref() });
+        }
+        answer => {
+            if let Err(declined) = committed(answer) {
+                return Ok(LaunchOutcome::Refused(declined));
+            }
+        }
     }
-    let run = opened_run(&instance, &label);
-    if ports.launcher.start(turn, &label).is_err() {
+    if ports.launcher.start(&instance, turn, &label).is_err() {
         let withdrawal = PipelineResolution {
-            subject: run,
+            subject: opened,
             sequence: 1,
             outcome: ResolutionOutcome::Withdrawn { reason: "the unit failed to start".into() },
             rationale: "The run committed and its unit did not start, so it holds no work.".into(),
@@ -219,7 +303,7 @@ pub fn open_and_launch(ports: &Ports, request: LaunchRequest) -> Result<LaunchOu
             Err(_) => Err(anyhow!("the run could not be withdrawn after its unit failed to start")),
         };
     }
-    Ok(LaunchOutcome::Launched { run })
+    Ok(LaunchOutcome::Launched { run: opened })
 }
 
 /// Whether admission landed the batch; if not, why.
@@ -232,19 +316,21 @@ fn committed(outcome: PipelineAdmissionOutcome) -> Result<(), Declined> {
     }
 }
 
-/// `breaker-<item local>`, with the local's dots (a label has none) as dashes.
-fn breaker_label(item: &PipelineRef) -> Label {
-    let local = item.id.0.splitn(3, ':').nth(2).unwrap_or_default();
-    Label(format!("breaker-{}", local.replace('.', "-")))
+/// `breaker-<run label>`: one question per trip, since each trip is a launch
+/// with a label of its own.
+fn breaker_label(label: &Label) -> Label {
+    Label(format!("breaker-{}", label.0))
 }
 
 /// The question a tripped item raises, in the item's own campaign.
-fn breaker_question(item: &PipelineRef, now: DateTime<Utc>) -> PipelineQuestion {
+fn breaker_question(item: &PipelineRef, label: &Label, now: DateTime<Utc>) -> PipelineQuestion {
     PipelineQuestion {
         campaign: Slug(root_of(item).to_string()),
-        label: breaker_label(item),
+        label: breaker_label(label),
         title: "Repeated runs on one item admitted nothing".into(),
-        question: "The last runs on this item admitted nothing but themselves, so it launches no more. A ruling or any other new document reopens it.".into(),
+        question:
+            "The last runs on this item left nothing that cites it, so it launches no more. Answering this reopens the item and starts a fresh count."
+                .into(),
         options: vec![
             QuestionOption { label: "reopen".into(), text: "Rule on the item, which reopens it.".into() },
             QuestionOption { label: "retire".into(), text: "Resolve the item so it is no longer worked.".into() },
@@ -259,9 +345,11 @@ fn breaker_question(item: &PipelineRef, now: DateTime<Utc>) -> PipelineQuestion
 #[cfg(test)]
 mod tests {
     use std::cell::{Cell, RefCell};
+    use std::time::Duration as StdDuration;
 
     use eureka_pipeline::PipelineKind as K;
-    use huginn_mind::PipelineAdmissionOutcome as Outcome;
+    use huginn_mind::{PipelineAdmissionOutcome as Outcome, PipelineStatus};
+    use rust_decimal::Decimal;
 
     use super::*;
     use crate::testkit::*;
@@ -282,14 +370,26 @@ mod tests {
         }
 
         fn launch_through(&self, mind: &dyn MindPort, turn: RunTurn, claims: &[PipelineRef], cap: Decimal) -> LaunchOutcome {
+            self.try_launch_through(mind, turn, claims, cap).unwrap()
+        }
+
+        fn try_launch_through(&self, mind: &dyn MindPort, turn: RunTurn, claims: &[PipelineRef], cap: Decimal) -> Result<LaunchOutcome> {
             let ports = Ports { mind, launcher: &self.launcher, clock: &self.clock, host: "yggdrasil-host" };
-            open_and_launch(&ports, LaunchRequest { turn, claims: claims.to_vec(), agent: "agent-x".into(), run_cap_usd: cap }).unwrap()
+            let grant = Grant::for_test(StdDuration::from_secs(60), cap);
+            open_and_launch(&ports, LaunchRequest { turn, claims: claims.to_vec(), agent: "agent-x".into(), grant })
         }
 
         fn at(&self, milliseconds: i64) {
             self.clock.0.set(at(milliseconds));
         }
 
+        /// The clock `seconds` after the run with this id was admitted.
+        fn age(&self, run: &str, seconds: i64) {
+            let admitted = self.mind.view(K::Run, run).unwrap().admission.admitted_at;
+            let admitted = DateTime::parse_from_rfc3339(&admitted).unwrap().with_timezone(&Utc);
+            self.clock.0.set(admitted + Duration::seconds(seconds));
+        }
+
         fn started(&self) -> usize {
             self.launcher.started.borrow().len()
         }
@@ -297,6 +397,16 @@ mod tests {
         fn runs(&self) -> usize {
             self.mind.ids(K::Run).len()
         }
+
+        /// A run of hers in force, with a unit that is running.
+        fn seed(&self, label: &str, turn: RunTurn, claims: &[PipelineRef]) {
+            self.mind.committed(vec![run(label, turn, RunOperator::Mind, claims)]);
+            self.launcher.revive(label);
+        }
+
+        fn in_force(&self, run: &str) -> bool {
+            self.mind.view(K::Run, &run_ref(run).id.0).unwrap().status == PipelineStatus::InForce
+        }
     }
 
     fn five() -> Decimal {
@@ -308,6 +418,12 @@ mod tests {
         run
     }
 
+    fn the_closure(rig: &Rig, run: &str, sequence: u32) -> (PipelineResolution, String) {
+        let view = rig.mind.view(K::Resolution, &format!("{INSTANCE}:resolution:run.{run}.n{sequence}")).expect("the closure");
+        let PipelineDocument::Resolution(closure) = view.document else { panic!("not a resolution") };
+        (closure, view.admission.receipt_id)
+    }
+
     #[test]
     fn a_queued_spec_opens_a_run_that_records_its_launch_and_then_its_unit_starts() {
         let rig = Rig::new();
@@ -322,11 +438,19 @@ mod tests {
         assert_eq!(held.campaigns, vec![Slug(CAMPAIGN.into())]);
         assert_eq!(view.admission.provenance.agent.0, "agent-x");
         assert_eq!(view.admission.provenance.session.0, "mind-20261009T131500000Z");
-        assert_eq!(*rig.launcher.started.borrow(), vec![(RunTurn::SelfRun, label("mind-20261009T131500000Z"))]);
+        assert_eq!(*rig.launcher.started.borrow(), vec![(Slug(INSTANCE.into()), RunTurn::SelfRun, label("mind-20261009T131500000Z"))]);
     }
 
     #[test]
-    fn the_runs_budget_is_the_cap_passed_in() {
+    fn the_unit_started_names_the_minds_instance() {
+        let rig = Rig::new();
+        rig.launch(RunTurn::PersonaTurn, &[], five());
+        let started = rig.launcher.started.borrow();
+        assert_eq!(started[0].0, *rig.mind.instance());
+    }
+
+    #[test]
+    fn the_runs_budget_is_the_grants_cap_at_two_caps() {
         let rig = Rig::new();
         for (n, (cap, want)) in [(Decimal::new(5, 0), "5"), (Decimal::new(750, 2), "7.5"), (Decimal::new(125, 1), "12.5")].into_iter().enumerate() {
             rig.at(n as i64);
@@ -335,23 +459,20 @@ mod tests {
             assert_eq!(the_run(&view).budget_usd.0, want);
             assert_eq!(the_run(&view).turn, RunTurn::PersonaTurn);
             assert!(the_run(&view).claims.is_empty() && the_run(&view).campaigns.is_empty());
-            // One live Persona turn of hers at a time (admission): end this one.
+            // One live Persona turn of hers at a time: end this one.
             rig.mind.committed(vec![close(run, recorded())]);
         }
         assert_eq!(rig.started(), 3, "each Persona turn ended before the next opened, so none was refused");
     }
 
     #[test]
-    fn a_second_live_persona_turn_is_refused_by_admission_and_starts_nothing() {
+    fn a_live_persona_turn_is_busy_to_a_persona_launch_and_does_not_block_a_self_run() {
         let rig = Rig::new();
         let LaunchOutcome::Launched { run } = rig.launch(RunTurn::PersonaTurn, &[], five()) else { panic!("not launched") };
         rig.at(1);
-        let second = rig.launch(RunTurn::PersonaTurn, &[], five());
-        let LaunchOutcome::Refused(Declined::Admission(MindRefusal::AlreadyLive { run: holder })) = second else {
-            panic!("not refused as live: {second:?}")
-        };
-        assert_eq!(holder.to_ref(), run);
-        assert_eq!((rig.runs(), rig.started()), (1, 1));
+        assert_eq!(rig.launch(RunTurn::PersonaTurn, &[], five()), LaunchOutcome::Busy { run });
+        assert_eq!((rig.runs(), rig.started()), (1, 1), "Busy admits and starts nothing");
+        assert!(matches!(rig.launch(RunTurn::SelfRun, &[], five()), LaunchOutcome::Launched { .. }));
     }
 
     #[test]
@@ -364,7 +485,8 @@ mod tests {
     fn two_launches_in_one_millisecond_open_one_run() {
         let rig = Rig::new();
         assert!(matches!(rig.launch(RunTurn::PersonaTurn, &[], five()), LaunchOutcome::Launched { .. }));
-        let second = rig.launch(RunTurn::PersonaTurn, &[], Decimal::new(6, 0));
+        // A Self run holds another slot, so only the key (instance and label) collides.
+        let second = rig.launch(RunTurn::SelfRun, &[], Decimal::new(6, 0));
         assert!(matches!(second, LaunchOutcome::Refused(_)), "{second:?}");
         assert_eq!((rig.runs(), rig.started()), (1, 1));
     }
@@ -397,10 +519,8 @@ mod tests {
         assert_eq!(outcome, LaunchOutcome::Refused(Declined::StartFailed));
         let opened = format!("{INSTANCE}:run:mind-20261009T131500000Z");
         let status = |id: &str| rig.mind.view(K::Run, id).unwrap().status;
-        assert!(
-            matches!(status(&opened), huginn_mind::PipelineStatus::Resolved { record, .. } if matches!(record.outcome, ResolutionOutcome::Withdrawn { .. }))
-        );
-        assert_eq!(status(&run_ref("other").id.0), huginn_mind::PipelineStatus::InForce);
+        assert!(matches!(status(&opened), PipelineStatus::Resolved { record, .. } if matches!(record.outcome, ResolutionOutcome::Withdrawn { .. })));
+        assert_eq!(status(&run_ref("other").id.0), PipelineStatus::InForce);
         // The spec is free again.
         rig.launcher.fail.set(false);
         rig.at(1);
@@ -431,7 +551,7 @@ mod tests {
         rig.mind.committed(vec![run("op", RunTurn::SelfRun, RunOperator::Operator, &[])]);
         rig.mind.committed(vec![run("gone", RunTurn::SelfRun, RunOperator::Mind, &[])]);
         rig.mind.committed(vec![close(run_ref("gone"), withdrawn())]);
-        rig.mind.committed(vec![run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]);
+        rig.seed("persona", RunTurn::PersonaTurn, &[]);
         rig.at(1);
         let LaunchOutcome::Launched { run: first } = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()) else {
             panic!("blocked by a run that is not hers")
@@ -446,6 +566,142 @@ mod tests {
         assert!(matches!(rig.launch(RunTurn::SelfRun, &[spec_ref("y")], five()), LaunchOutcome::Launched { .. }));
     }
 
+    #[test]
+    fn an_alive_run_is_busy_however_old_and_is_not_closed() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![spec("x")]);
+        rig.seed("old", RunTurn::SelfRun, &[spec_ref("x")]);
+        rig.age(&run_ref("old").id.0, 24 * 3600);
+        assert_eq!(rig.launch(RunTurn::SelfRun, &[], five()), LaunchOutcome::Busy { run: run_ref("old") });
+        assert!(rig.in_force("old"));
+        assert_eq!((rig.runs(), rig.started()), (1, 0));
+    }
+
+    #[test]
+    fn a_claimless_alive_run_is_busy() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![spec("x")]);
+        rig.seed("bare", RunTurn::SelfRun, &[]);
+        rig.age(&run_ref("bare").id.0, 3600);
+        assert_eq!(rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()), LaunchOutcome::Busy { run: run_ref("bare") });
+        assert!(rig.in_force("bare"));
+    }
+
+    #[test]
+    fn a_dead_run_is_closed_past_the_grace_in_the_batch_that_opens_the_next() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![spec("x")]);
+        // Her run on x, admitted, and its unit never ran.
+        rig.mind.committed(vec![run("dead", RunTurn::SelfRun, RunOperator::Mind, &[spec_ref("x")])]);
+        let dead = run_ref("dead").id.0;
+        rig.age(&dead, START_GRACE_S - 1);
+        assert_eq!(
+            rig.launch(RunTurn::SelfRun, &[], five()),
+            LaunchOutcome::Busy { run: run_ref("dead") },
+            "inside the grace the unit may still be starting"
+        );
+        assert!(rig.in_force("dead"));
+        assert_eq!((rig.runs(), rig.started()), (1, 0));
+
+        rig.age(&dead, START_GRACE_S);
+        let LaunchOutcome::Launched { run: next } = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()) else {
+            panic!("the dead run's claim stayed held")
+        };
+        assert!(!rig.in_force("dead"));
+        let (closure, closed_in) = the_closure(&rig, "dead", 1);
+        assert_eq!(closure.outcome, ResolutionOutcome::Recorded { reason: "unit not active".into() });
+        assert_eq!(closure.subject, run_ref("dead"));
+        assert_eq!(
+            closed_in,
+            rig.mind.view(K::Run, &next.id.0).unwrap().admission.receipt_id,
+            "one admission closes the dead run and opens the next"
+        );
+        assert_eq!(rig.started(), 1);
+    }
+
+    #[test]
+    fn a_dead_runs_closure_takes_the_next_sequence_of_its_subject() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![run("dead", RunTurn::SelfRun, RunOperator::Mind, &[])]);
+        rig.mind.committed(vec![close(run_ref("dead"), recorded())]);
+        // Withdrawing that closure puts the run back in force.
+        let first = reference(K::Resolution, &format!("{INSTANCE}:resolution:run.dead.n1"));
+        rig.mind.committed(vec![close(first, withdrawn())]);
+        assert!(rig.in_force("dead"));
+        rig.age(&run_ref("dead").id.0, 600);
+        assert!(matches!(rig.launch(RunTurn::SelfRun, &[], five()), LaunchOutcome::Launched { .. }));
+        assert!(!rig.in_force("dead"));
+        assert_eq!(the_closure(&rig, "dead", 2).0.sequence, 2);
+    }
+
+    #[test]
+    fn only_a_dead_holder_of_the_requests_turn_is_closed() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![run("op", RunTurn::SelfRun, RunOperator::Operator, &[])]);
+        rig.mind.committed(vec![run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]);
+        rig.mind.committed(vec![run("dead", RunTurn::SelfRun, RunOperator::Mind, &[])]);
+        rig.age(&run_ref("dead").id.0, 3600);
+        assert!(matches!(rig.launch(RunTurn::SelfRun, &[], five()), LaunchOutcome::Launched { .. }));
+        assert!(!rig.in_force("dead"));
+        assert!(rig.in_force("op") && rig.in_force("persona"), "an Operator run and a run of the other turn are never closed by this launch");
+    }
+
+    #[test]
+    fn a_run_whose_admit_answer_was_lost_is_closed_by_the_next_launch_of_its_turn() {
+        struct LosesAnswer<'a>(&'a TestMind);
+        impl MindPort for LosesAnswer<'_> {
+            fn instance(&self) -> &Slug {
+                self.0.instance()
+            }
+
+            fn query(&self, selection: &Selection) -> Result<huginn_mind::PipelineSelectionPage> {
+                self.0.query(selection)
+            }
+
+            fn admit(&self, agent: &str, session: &str, documents: Vec<PipelineDocument>) -> Result<Outcome> {
+                self.0.admit(agent, session, documents)?;
+                Err(anyhow!("the answer was lost"))
+            }
+        }
+        let rig = Rig::new();
+        assert!(rig.try_launch_through(&LosesAnswer(&rig.mind), RunTurn::SelfRun, &[], five()).is_err());
+        assert_eq!((rig.runs(), rig.started()), (1, 0), "committed, and no unit was started");
+        let lost = rig.mind.ids(K::Run).remove(0);
+        rig.age(&lost, START_GRACE_S);
+        assert!(matches!(rig.launch(RunTurn::SelfRun, &[], five()), LaunchOutcome::Launched { .. }));
+        assert_eq!((rig.runs(), rig.started()), (2, 1));
+        let lost_label = lost.rsplit(':').next().unwrap().to_string();
+        assert!(!rig.in_force(&lost_label));
+    }
+
+    #[test]
+    fn a_port_that_admits_a_rival_live_run_between_step_0_and_the_admission_is_busy_and_starts_nothing() {
+        struct Racing<'a> {
+            inner: &'a TestMind,
+            raced: Cell<bool>,
+        }
+        impl MindPort for Racing<'_> {
+            fn instance(&self) -> &Slug {
+                self.inner.instance()
+            }
+
+            fn query(&self, selection: &Selection) -> Result<huginn_mind::PipelineSelectionPage> {
+                self.inner.query(selection)
+            }
+
+            fn admit(&self, agent: &str, session: &str, documents: Vec<PipelineDocument>) -> Result<Outcome> {
+                if !self.raced.replace(true) {
+                    self.inner.committed(vec![run("rival", RunTurn::SelfRun, RunOperator::Mind, &[])]);
+                }
+                self.inner.admit(agent, session, documents)
+            }
+        }
+        let rig = Rig::new();
+        let racing = Racing { inner: &rig.mind, raced: Cell::new(false) };
+        assert_eq!(rig.launch_through(&racing, RunTurn::SelfRun, &[], five()), LaunchOutcome::Busy { run: run_ref("rival") });
+        assert_eq!((rig.runs(), rig.started()), (1, 0));
+    }
+
     #[test]
     fn a_tripped_item_raises_one_question_launches_nothing_and_a_ruling_reopens_it() {
         let rig = Rig::new();
@@ -453,7 +709,7 @@ mod tests {
         history(&rig.mind, &spec_ref("x"), &[false, false, false]);
         let before = rig.runs();
         let outcome = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five());
-        let question = id("question", "breaker-cut-x-r1");
+        let question = id("question", "breaker-mind-20261009T131500000Z");
         assert_eq!(outcome, LaunchOutcome::Tripped { item: spec_ref("x"), question: reference(K::Question, &question) });
         assert_eq!((rig.runs(), rig.started()), (before, 0));
         assert_eq!(rig.mind.ids(K::Question), vec![question.clone()]);
@@ -463,12 +719,65 @@ mod tests {
         assert_eq!(view.admission.provenance.session.0, "mind-20261009T131500000Z");
         // Open, the question blocks the item.
         assert_eq!(rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()), LaunchOutcome::NotQueued { item: spec_ref("x") });
-        // The ruling that answers it is a document after the last run.
+        // The ruling that answers it opens the item, and the count starts afresh.
         rig.mind.committed(vec![ruling("reopen-x", Some((&question, "reopen")))]);
         rig.at(1);
         assert!(matches!(rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()), LaunchOutcome::Launched { .. }));
     }
 
+    #[test]
+    fn launching_a_tripped_item_with_another_queued_trips_and_launches_nothing() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![spec("x"), spec("y")]);
+        history_as(&rig.mind, &spec_ref("x"), "x", &[false, false, false]);
+        let outcome = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five());
+        assert!(matches!(outcome, LaunchOutcome::Tripped { .. }), "{outcome:?}");
+        assert_eq!(rig.started(), 0);
+        assert!(matches!(rig.launch(RunTurn::SelfRun, &[spec_ref("y")], five()), LaunchOutcome::Launched { .. }), "the other item still launches");
+    }
+
+    #[test]
+    fn a_second_trip_raises_a_second_question_with_its_own_key() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![spec("x")]);
+        history(&rig.mind, &spec_ref("x"), &[false, false, false]);
+        let LaunchOutcome::Tripped { question: first, .. } = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()) else { panic!("no first trip") };
+        rig.mind.committed(vec![ruling("reopen", Some((&first.id.0, "reopen")))]);
+        history_as(&rig.mind, &spec_ref("x"), "g", &[false, false, false]);
+        rig.at(1);
+        let LaunchOutcome::Tripped { question: second, .. } = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()) else {
+            panic!("no second trip")
+        };
+        assert_ne!(first, second);
+        assert_eq!(rig.mind.ids(K::Question).len(), 2);
+        assert_eq!(rig.started(), 0);
+    }
+
+    #[test]
+    fn nine_empty_runs_of_fifty_before_the_answer_do_not_trip_after_it() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![spec("x")]);
+        let item = spec_ref("x");
+        history(&rig.mind, &item, &[false, false, true, false, false, true, false, false, true, false, false, true, false]);
+        rig.mind.committed(vec![question_in("fork", &item)]);
+        rig.mind.committed(vec![ruling("answer", Some((&id("question", "fork"), "A")))]);
+        history_as(&rig.mind, &item, "g", &[false]);
+        assert!(matches!(rig.launch(RunTurn::SelfRun, &[item], five()), LaunchOutcome::Launched { .. }));
+    }
+
+    #[test]
+    fn a_dead_run_is_an_ended_empty_run_for_the_breaker_and_its_claim_is_free() {
+        let rig = Rig::new();
+        rig.mind.committed(vec![spec("x")]);
+        history(&rig.mind, &spec_ref("x"), &[false, false]);
+        rig.mind.committed(vec![run("dead", RunTurn::SelfRun, RunOperator::Mind, &[spec_ref("x")])]);
+        rig.age(&run_ref("dead").id.0, 600);
+        let outcome = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five());
+        assert!(matches!(outcome, LaunchOutcome::Tripped { .. }), "{outcome:?}");
+        assert!(!rig.in_force("dead"), "the trip's batch closes the dead run too");
+        assert_eq!((rig.runs(), rig.started()), (3, 0));
+    }
+
     struct Spy {
         calls: RefCell<Vec<Vec<String>>>,
         success: Cell<bool>,
@@ -482,45 +791,67 @@ mod tests {
         }
     }
 
-    fn spied(instance: &str) -> (SystemdLauncher, &'static Spy) {
+    fn spied() -> (SystemdLauncher, &'static Spy) {
         let spy: &'static Spy = Box::leak(Box::new(Spy { calls: RefCell::default(), success: Cell::new(true), stdout: RefCell::default() }));
-        (SystemdLauncher::over(Slug(instance.into()), Box::new(spy)), spy)
+        (SystemdLauncher::over(Box::new(spy)), spy)
+    }
+
+    fn eureka() -> Slug {
+        Slug("eureka".into())
     }
 
     #[test]
     fn units_are_templated_by_instance_and_turn() {
-        let (launcher, spy) = spied("eureka");
+        let (launcher, spy) = spied();
         let l = label("mind-20261009T131500123Z");
-        assert_eq!(launcher.unit(RunTurn::PersonaTurn, &l), "mind-persona@eureka:mind-20261009T131500123Z.service");
-        assert_eq!(launcher.unit(RunTurn::SelfRun, &l), "mind-self@eureka:mind-20261009T131500123Z.service");
-        launcher.start(RunTurn::SelfRun, &l).unwrap();
-        launcher.start(RunTurn::PersonaTurn, &l).unwrap();
+        assert_eq!(unit(&eureka(), RunTurn::PersonaTurn, &l).unwrap(), "mind-persona@eureka:mind-20261009T131500123Z.service");
+        assert_eq!(unit(&eureka(), RunTurn::SelfRun, &l).unwrap(), "mind-self@eureka:mind-20261009T131500123Z.service");
+        launcher.start(&eureka(), RunTurn::SelfRun, &l).unwrap();
+        launcher.start(&Slug("eureka.test".into()), RunTurn::PersonaTurn, &l).unwrap();
         assert_eq!(
             *spy.calls.borrow(),
             vec![
                 vec!["start", "--no-block", "mind-self@eureka:mind-20261009T131500123Z.service"],
-                vec!["start", "--no-block", "mind-persona@eureka:mind-20261009T131500123Z.service"],
+                vec!["start", "--no-block", "mind-persona@eureka.test:mind-20261009T131500123Z.service"],
             ]
         );
     }
 
     #[test]
     fn a_unit_is_alive_while_active_activating_or_reloading() {
-        let (launcher, spy) = spied("eureka");
+        let (launcher, spy) = spied();
         let l = label("a");
-        for (said, alive) in [("active\n", true), ("activating\n", true), ("reloading\n", true), ("inactive\n", false), ("failed\n", false), ("", false)] {
+        for (said, alive) in
+            [("active\n", true), ("activating\n", true), ("reloading\n", true), ("inactive\n", false), ("failed\n", false), ("", false)]
+        {
             *spy.stdout.borrow_mut() = said.to_string();
-            assert_eq!(launcher.alive(RunTurn::SelfRun, &l), alive, "{said:?}");
+            assert_eq!(launcher.alive(&eureka(), RunTurn::SelfRun, &l), alive, "{said:?}");
         }
         assert_eq!(spy.calls.borrow()[0], vec!["is-active", "mind-self@eureka:a.service"]);
     }
 
     #[test]
     fn a_failed_start_names_neither_instance_nor_label() {
-        let (launcher, spy) = spied("canary-instance-7f3a91");
+        let (launcher, spy) = spied();
         spy.success.set(false);
-        let error = launcher.start(RunTurn::SelfRun, &label("canary-label-2c8d04")).unwrap_err();
+        let error = launcher.start(&Slug("canary-instance-7f3a91".into()), RunTurn::SelfRun, &label("canary-label-2c8d04")).unwrap_err();
         let shown = format!("{error:#}");
         assert!(!shown.contains("canary"), "{shown}");
     }
+
+    #[test]
+    fn hostile_instances_and_labels_make_start_fail_and_alive_false_with_no_systemctl_call() {
+        let (launcher, spy) = spied();
+        *spy.stdout.borrow_mut() = "active\n".into();
+        for hostile in ["", ".", "..", "a/b", "a@b", "a:b", "a b", "a\nb", "eureka/../x\n", "x\n--now", "canary/9"] {
+            let as_instance = Slug(hostile.into());
+            let error = launcher.start(&as_instance, RunTurn::SelfRun, &label("ok")).unwrap_err();
+            assert!(!format!("{error:#}").contains("canary"), "{error:#}");
+            assert!(!launcher.alive(&as_instance, RunTurn::SelfRun, &label("ok")), "instance {hostile:?}");
+            let error = launcher.start(&eureka(), RunTurn::PersonaTurn, &label(hostile)).unwrap_err();
+            assert!(!format!("{error:#}").contains("canary"), "{error:#}");
+            assert!(!launcher.alive(&eureka(), RunTurn::PersonaTurn, &label(hostile)), "label {hostile:?}");
+        }
+        assert!(spy.calls.borrow().is_empty(), "no hostile name reached systemctl: {:?}", spy.calls.borrow());
+    }
 }
diff --git a/crates/mind-body/src/queue.rs b/crates/mind-body/src/queue.rs
index 891d553..6772c76 100644
--- a/crates/mind-body/src/queue.rs
+++ b/crates/mind-body/src/queue.rs
@@ -8,7 +8,7 @@
 
 use anyhow::{Result, anyhow, bail};
 use cultnet_rs::{Citation, FieldPredicate, Incoming, RecordRef, Selection};
-use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, Slug};
+use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, RunOperator, RunTurn, Slug};
 use eureka_state::{ClientError, HuginnClient};
 use huginn_mind::{
     Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineDocumentSummary, PipelineFacts,
@@ -113,41 +113,30 @@ pub(crate) fn in_force(selection: Selection) -> Selection {
     any_of(selection, "in_force", &["true"])
 }
 
-/// Every page of a selection, in the mind's order, handed to `visit` until it
-/// answers `false`. The one paging loop.
-fn scan(mind: &dyn MindPort, selection: &Selection, mut visit: impl FnMut(PipelineDocumentSummary) -> bool) -> Result<()> {
+/// Every page of a selection, in the mind's order. The one paging loop.
+pub(crate) fn headers(mind: &dyn MindPort, selection: &Selection) -> Result<Vec<PipelineDocumentSummary>> {
+    let mut all = Vec::new();
     let mut asking = selection.clone();
     loop {
         let page = mind.query(&asking)?;
         let PipelinePageItems::Headers(items) = page.items else { bail!("the mind answered documents to a header selection") };
-        for item in items {
-            if !visit(item) {
-                return Ok(());
-            }
-        }
+        all.extend(items);
         match page.next {
             Some(cursor) => asking.cursor = Some(cursor),
-            None => return Ok(()),
+            None => return Ok(all),
         }
     }
 }
 
-pub(crate) fn headers(mind: &dyn MindPort, selection: &Selection) -> Result<Vec<PipelineDocumentSummary>> {
-    let mut all = Vec::new();
-    scan(mind, selection, |header| {
-        all.push(header);
-        true
-    })?;
-    Ok(all)
-}
-
-/// The claims of the runs of `instance` that are still in force: the work
-/// someone holds.
-pub(crate) fn live_claims(mind: &dyn MindPort, instance: &Slug) -> Result<Vec<PipelineRef>> {
+/// The claims of the runs of `instance` that are still in force and not in
+/// `ending`: the work someone holds.
+pub(crate) fn live_claims(mind: &dyn MindPort, instance: &Slug, ending: &[PipelineRef]) -> Result<Vec<PipelineRef>> {
     let runs = in_force(any_of(of_kind(PipelineKind::Run), "root", &[&instance.0]));
     let mut claims = Vec::new();
     for header in headers(mind, &runs)? {
-        if let PipelineFacts::Run { claims: held, .. } = header.facts {
+        if let PipelineFacts::Run { claims: held, .. } = header.facts
+            && !ending.contains(&header.id)
+        {
             claims.extend(held);
         }
     }
@@ -155,8 +144,10 @@ pub(crate) fn live_claims(mind: &dyn MindPort, instance: &Slug) -> Result<Vec<Pi
 }
 
 /// What may be worked, oldest first: the cut specs in force that no report
-/// answers, that no open question blocks, and that no live run claims.
-pub fn queue(mind: &dyn MindPort, instance: &Slug) -> Result<Vec<PipelineRef>> {
+/// answers, that no open question blocks, and that no live run claims. A run in
+/// `ending` is one the caller closes in the same admission that opens the next,
+/// so it claims nothing here.
+pub fn queue(mind: &dyn MindPort, instance: &Slug, ending: &[PipelineRef]) -> Result<Vec<PipelineRef>> {
     let unreported = Selection { cited: Some(Incoming { role: "cut_spec".into(), exists: false }), ..in_force(of_kind(PipelineKind::CutSpec)) };
     let mut blocked: Vec<PipelineRef> = Vec::new();
     for header in headers(mind, &in_force(of_kind(PipelineKind::Question)))? {
@@ -164,12 +155,8 @@ pub fn queue(mind: &dyn MindPort, instance: &Slug) -> Result<Vec<PipelineRef>> {
             blocked.push(item);
         }
     }
-    let claimed = live_claims(mind, instance)?;
-    Ok(headers(mind, &unreported)?
-        .into_iter()
-        .map(|header| header.id)
-        .filter(|spec| !blocked.contains(spec) && !claimed.contains(spec))
-        .collect())
+    let claimed = live_claims(mind, instance, ending)?;
+    Ok(headers(mind, &unreported)?.into_iter().map(|header| header.id).filter(|spec| !blocked.contains(spec) && !claimed.contains(spec)).collect())
 }
 
 /// Whether an item may still be launched on.
@@ -179,44 +166,46 @@ pub enum Breaker {
     Trip,
 }
 
-/// The repetition breaker, derived from the item's Self runs and from what
-/// else the mind admitted.
+/// The repetition breaker, derived from the documents that cite the item.
 ///
-/// A run is a Self run of the item that claims it. A run that was withdrawn
-/// never ran and is no run here. An ended run is empty when no document other
-/// than a run of this item, or the resolution that closes one, was admitted
-/// from its own admission up to the next run's (or to now). Trip when the last
-/// `CONSECUTIVE` ended runs are empty, or `EMPTY_SHARE` of the last `WINDOW`.
-/// A document admitted after the last run, such as a ruling, makes that run
-/// non-empty, so reopening an item resets it.
-pub fn breaker(mind: &dyn MindPort, item: &PipelineRef) -> Result<Breaker> {
-    let claiming = Citation { target: RecordRef::new(item.kind.type_id(), item.id.0.clone()), role: Some("claims".into()) };
-    let runs = headers(mind, &Selection { cites: Some(claiming), ..any_of(of_kind(PipelineKind::Run), "turn", &["SelfRun"]) })?;
-    let withdrawn = |run: &PipelineDocumentSummary| {
-        matches!(run.status, PipelineStatusSummary::Resolved { outcome: ResolutionOutcome::Withdrawn { .. }, .. })
-    };
-    // (admission ordinal, ended) of each run that ran, oldest first.
-    let ran: Vec<(u64, bool)> = runs
+/// One role-less citation selection over every schema gives the set: the runs
+/// that claim the item, and whatever else points at it (a follow-up sourced
+/// from it, a report, a resolution, a question raised in it). The newest
+/// question raised in the item is the boundary: only runs admitted after it
+/// count, so answering a trip starts a fresh count. A counted run is a Self run
+/// of hers (typed `turn` and `operated_by`) that was not withdrawn; a
+/// withdrawn run never ran and is no run here. A counted run is ended when
+/// resolved or listed in `ending`, and is empty when no document of the set
+/// other than a run was admitted from its own admission up to the next counted
+/// run's (or to now). Trip when the last `CONSECUTIVE` ended runs are empty, or
+/// `EMPTY_SHARE` of the last `WINDOW`. What else the mind admitted, such as a
+/// Persona turn's run or an unrelated ruling, cites nothing here and counts for
+/// nothing.
+pub fn breaker(mind: &dyn MindPort, item: &PipelineRef, ending: &[PipelineRef]) -> Result<Breaker> {
+    let citing = Citation { target: RecordRef::new(item.kind.type_id(), item.id.0.clone()), role: None };
+    let mut set = headers(mind, &Selection { cites: Some(citing), ..Selection::default() })?;
+    set.sort_by_key(|header| header.admission.ordinal);
+    let boundary = set
         .iter()
-        .filter(|run| !withdrawn(run))
-        .map(|run| (run.admission.ordinal, matches!(run.status, PipelineStatusSummary::Resolved { .. })))
+        .filter(|header| matches!(&header.facts, PipelineFacts::Question { raised_in: Some(raised), .. } if raised == item))
+        .map(|header| header.admission.ordinal)
+        .max();
+    let after = |header: &&PipelineDocumentSummary| boundary.is_none_or(|boundary| header.admission.ordinal > boundary);
+    let withdrawn =
+        |run: &PipelineDocumentSummary| matches!(run.status, PipelineStatusSummary::Resolved { outcome: ResolutionOutcome::Withdrawn { .. }, .. });
+    // (admission ordinal, ended) of each counted run, oldest first.
+    let ran: Vec<(u64, bool)> = set
+        .iter()
+        .filter(after)
+        .filter(|header| {
+            matches!(header.facts, PipelineFacts::Run { turn: RunTurn::SelfRun, operated_by: RunOperator::Mind, .. }) && !withdrawn(header)
+        })
+        .map(|run| (run.admission.ordinal, matches!(run.status, PipelineStatusSummary::Resolved { .. }) || ending.contains(&run.id)))
         .collect();
     let ended: Vec<usize> = (0..ran.len()).filter(|index| ran[*index].1).collect();
     let considered: Vec<usize> = ended.iter().rev().take(WINDOW).copied().collect();
-    let Some(oldest) = considered.last().map(|index| ran[*index].0) else { return Ok(Breaker::Clear) };
-
-    let run_ids: Vec<&PipelineRef> = runs.iter().map(|run| &run.id).collect();
-    let mut admitted = Vec::new();
-    scan(mind, &Selection { descending: true, ..Selection::default() }, |header| {
-        if header.admission.ordinal < oldest {
-            return false;
-        }
-        let closes_a_run = matches!(&header.facts, PipelineFacts::Resolution { subject, .. } if run_ids.contains(&subject));
-        if !run_ids.contains(&&header.id) && !closes_a_run {
-            admitted.push(header.admission.ordinal);
-        }
-        true
-    })?;
+    let admitted: Vec<u64> =
+        set.iter().filter(|header| !matches!(header.facts, PipelineFacts::Run { .. })).map(|header| header.admission.ordinal).collect();
 
     // Newest first, like `considered`.
     let empty: Vec<bool> = considered
@@ -255,7 +244,7 @@ mod tests {
 
     fn tripped(runs: &[bool]) -> bool {
         let (mind, item) = worked(runs);
-        breaker(&mind, &item).unwrap() == Breaker::Trip
+        breaker(&mind, &item, &[]).unwrap() == Breaker::Trip
     }
 
     #[test]
@@ -292,6 +281,55 @@ mod tests {
         assert!(!tripped(&runs), "the tenth empty run is the 51st newest");
     }
 
+    #[test]
+    fn three_empty_runs_trip_after_any_history_of_non_empty_ones() {
+        for before in [0usize, 1, 12, 20] {
+            let mut runs = vec![N; before];
+            runs.extend([E, E, E]);
+            assert!(tripped(&runs), "{before} non-empty runs then three empty ones");
+            assert!(!tripped(&[vec![N; before], vec![E, E]].concat()), "two empty ones never trip");
+        }
+    }
+
+    #[test]
+    fn only_documents_that_cite_the_item_make_a_run_non_empty() {
+        let (mind, item) = worked(&[]);
+        mind.committed(vec![spec("other")]);
+        for n in 0..3 {
+            // A Persona turn's run, a ruling and a follow-up of another item all land in the window.
+            mind.committed(vec![run(&format!("p{n}"), RunTurn::PersonaTurn, RunOperator::Mind, &[])]);
+            mind.committed(vec![ruling(&format!("r{n}"), None), follow_up(&format!("o{n}"), &spec_ref("other"))]);
+            ended_run(&mind, &item, "h", n, false);
+            mind.committed(vec![close(run_ref(&format!("p{n}")), recorded())]);
+        }
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip);
+    }
+
+    #[test]
+    fn a_run_in_ending_is_an_ended_empty_run() {
+        let (mind, item) = worked(&[E, E]);
+        mind.committed(vec![run("dying", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item))]);
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear, "open, it is not yet ended");
+        assert_eq!(breaker(&mind, &item, &[run_ref("dying")]).unwrap(), Breaker::Trip);
+        let instance = Slug(INSTANCE.into());
+        assert!(queue(&mind, &instance, &[]).unwrap().iter().all(|spec| *spec != item), "it claims the item");
+        assert!(queue(&mind, &instance, &[run_ref("dying")]).unwrap().contains(&item), "an ending run claims nothing");
+    }
+
+    #[test]
+    fn the_newest_question_in_the_item_starts_a_fresh_count() {
+        // Nine empty of the last fifty, then a question about the item is raised and answered.
+        let (mind, item) = worked(&[E, E, N, E, E, N, E, E, N, E, E, N, E]);
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear);
+        mind.committed(vec![question_in("fork", &item)]);
+        mind.committed(vec![ruling("answer", Some((&id("question", "fork"), "A")))]);
+        ended_run(&mind, &item, "g", 0, false);
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear, "the tenth empty run, but nine of them are before the question");
+        ended_run(&mind, &item, "g", 1, false);
+        ended_run(&mind, &item, "g", 2, false);
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip, "three empty runs after it");
+    }
+
     #[test]
     fn a_withdrawn_run_never_ran_and_is_no_boundary() {
         let withdrawn_between = |ruling_after: bool| {
@@ -299,11 +337,11 @@ mod tests {
             mind.committed(vec![run("w", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item))]);
             mind.committed(vec![close(run_ref("w"), withdrawn())]);
             if ruling_after {
-                mind.committed(vec![ruling("after", None)]);
+                mind.committed(vec![follow_up("after", &item)]);
             }
             ended_run(&mind, &item, "h", 1, false);
             ended_run(&mind, &item, "h", 2, false);
-            breaker(&mind, &item).unwrap()
+            breaker(&mind, &item, &[]).unwrap()
         };
         assert_eq!(withdrawn_between(false), Breaker::Trip, "the withdrawn run is neither counted nor does it break the streak");
         assert_eq!(withdrawn_between(true), Breaker::Clear, "a document after the withdrawn run still lands in the first run's window");
@@ -313,13 +351,13 @@ mod tests {
     fn a_runs_window_includes_its_own_batch_and_stops_before_the_next_runs() {
         let (mind, item) = worked(&[E, E]);
         // The third run opens with a ruling in its batch; it is still open.
-        mind.committed(vec![run("h2", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item)), ruling("with-run", None)]);
+        mind.committed(vec![run("h2", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item)), follow_up("with-run", &item)]);
         mind.committed(vec![close(run_ref("h2"), recorded())]);
-        assert_eq!(breaker(&mind, &item).unwrap(), Breaker::Clear, "the ruling in its own batch is in its window");
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear, "the ruling in its own batch is in its window");
 
         let (mind, item) = worked(&[E, E, E]);
-        mind.committed(vec![run("live", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item)), ruling("with-next", None)]);
-        assert_eq!(breaker(&mind, &item).unwrap(), Breaker::Trip, "the next run's batch is not the previous run's window");
+        mind.committed(vec![run("live", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item)), follow_up("with-next", &item)]);
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip, "the next run's batch is not the previous run's window");
     }
 
     /// The adapter over the wire vocabulary, answered by a real mind.
@@ -329,11 +367,14 @@ mod tests {
         mind.committed(vec![spec("x")]);
         let port = HuginnMind::over(Slug(INSTANCE.into()), Box::new(|request| Ok(mind.answer(request))));
         assert_eq!(port.instance().0, INSTANCE);
-        assert_eq!(queue(&port, port.instance()).unwrap(), vec![spec_ref("x")]);
+        assert_eq!(queue(&port, port.instance(), &[]).unwrap(), vec![spec_ref("x")]);
         let outcome = port.admit("agent-x", "session-y", vec![run("p", RunTurn::PersonaTurn, RunOperator::Mind, &[])]).unwrap();
         assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
         let facts = mind.view(PipelineKind::Run, &run_ref("p").id.0).unwrap().admission.provenance;
-        assert_eq!((facts.faculty, facts.agent.0.as_str(), facts.session.0.as_str(), facts.tool.0.as_str()), (Faculty::SelfFaculty, "agent-x", "session-y", "mind-body"));
+        assert_eq!(
+            (facts.faculty, facts.agent.0.as_str(), facts.session.0.as_str(), facts.tool.0.as_str()),
+            (Faculty::SelfFaculty, "agent-x", "session-y", "mind-body")
+        );
         let refused = port.query(&any_of(of_kind(PipelineKind::Ruling), "turn", &["canary-turn-5d2e"])).unwrap_err();
         assert!(format!("{refused:#}").contains("refused the query"));
         assert!(!format!("{refused:#}").contains("canary"), "{refused:#}");
@@ -345,18 +386,18 @@ mod tests {
         mind.committed(vec![spec("x"), spec("y")]);
         history_as(&mind, &spec_ref("x"), "x", &[E, E]);
         history_as(&mind, &spec_ref("y"), "y", &[E, E, E]);
-        assert_eq!(breaker(&mind, &spec_ref("x")).unwrap(), Breaker::Clear);
-        assert_eq!(breaker(&mind, &spec_ref("y")).unwrap(), Breaker::Trip);
+        assert_eq!(breaker(&mind, &spec_ref("x"), &[]).unwrap(), Breaker::Clear);
+        assert_eq!(breaker(&mind, &spec_ref("y"), &[]).unwrap(), Breaker::Trip);
     }
 
     #[test]
     fn the_oldest_counted_runs_own_batch_is_in_its_window() {
         let (mind, item) = worked(&[]);
-        mind.committed(vec![run("h0", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item)), ruling("with-oldest", None)]);
+        mind.committed(vec![run("h0", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item)), follow_up("with-oldest", &item)]);
         mind.committed(vec![close(run_ref("h0"), recorded())]);
         ended_run(&mind, &item, "h", 1, false);
         ended_run(&mind, &item, "h", 2, false);
-        assert_eq!(breaker(&mind, &item).unwrap(), Breaker::Clear);
+        assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear);
     }
 
     #[test]
@@ -365,18 +406,18 @@ mod tests {
         mind.committed(vec![spec("a"), spec("b"), spec("c"), spec("d"), spec("e")]);
         let instance = Slug(INSTANCE.into());
         let all: Vec<PipelineRef> = ["a", "b", "c", "d", "e"].iter().map(|cut| spec_ref(cut)).collect();
-        assert_eq!(queue(&mind, &instance).unwrap(), all);
+        assert_eq!(queue(&mind, &instance, &[]).unwrap(), all);
 
         mind.committed(vec![report("b")]);
         mind.committed(vec![question_in("fork", &spec_ref("c"))]);
         mind.committed(vec![run("held", RunTurn::SelfRun, RunOperator::Operator, &[spec_ref("d")])]);
-        assert_eq!(queue(&mind, &instance).unwrap(), vec![spec_ref("a"), spec_ref("e")]);
+        assert_eq!(queue(&mind, &instance, &[]).unwrap(), vec![spec_ref("a"), spec_ref("e")]);
 
         mind.committed(vec![close(run_ref("held"), recorded())]);
-        assert_eq!(queue(&mind, &instance).unwrap(), vec![spec_ref("a"), spec_ref("d"), spec_ref("e")], "a closed run claims nothing");
+        assert_eq!(queue(&mind, &instance, &[]).unwrap(), vec![spec_ref("a"), spec_ref("d"), spec_ref("e")], "a closed run claims nothing");
         let other = Slug("another-instance".into());
         mind.committed(vec![run("again", RunTurn::SelfRun, RunOperator::Operator, &[spec_ref("a")])]);
-        assert_eq!(queue(&mind, &other).unwrap().len(), 3, "claims are read for the instance asked");
+        assert_eq!(queue(&mind, &other, &[]).unwrap().len(), 3, "claims are read for the instance asked");
     }
 
     /// A daemon that is not there: the failure names no input, endpoint or
@@ -386,7 +427,7 @@ mod tests {
         let canary = "canary-instance-7f3a91";
         let endpoint: SocketAddr = "127.0.0.1:9".parse().unwrap();
         let port = HuginnMind::new(HuginnClient::new(endpoint, Slug(canary.into()), Duration::from_millis(300)));
-        let error = queue(&port, &Slug(canary.into())).unwrap_err();
+        let error = queue(&port, &Slug(canary.into()), &[]).unwrap_err();
         let shown = format!("{error:#}");
         assert!(shown.contains("unavailable"), "{shown}");
         assert!(!shown.contains(canary) && !shown.contains("127.0.0.1"), "{shown}");
diff --git a/crates/mind-body/src/testkit.rs b/crates/mind-body/src/testkit.rs
index a8d71e3..8c3580b 100644
--- a/crates/mind-body/src/testkit.rs
+++ b/crates/mind-body/src/testkit.rs
@@ -8,13 +8,13 @@ use anyhow::{Result, anyhow};
 use chrono::{DateTime, Duration, TimeZone, Utc};
 use cultnet_rs::Selection;
 use eureka_pipeline::{
-    CommitRange, CutVerification, Date, DocRef, Label, PipelineCampaign, PipelineCutReport, PipelineCutSpec, PipelineDocument,
-    PipelineFollowUp, PipelineInstance, PipelineKind, PipelineQuestion, PipelineRef, PipelineResolution, PipelineRuling, PipelineRun, PipelineStewardship,
+    CommitRange, CutVerification, Date, DocRef, Label, PipelineCampaign, PipelineCutReport, PipelineCutSpec, PipelineDocument, PipelineFollowUp,
+    PipelineInstance, PipelineKind, PipelineQuestion, PipelineRef, PipelineResolution, PipelineRuling, PipelineRun, PipelineStewardship,
     PipelineTarget, QuestionOption, ResolutionOutcome, RulingAuthority, RunOperator, RunTurn, Sha, Slug, StructuralDelta, TargetInvariant,
 };
 use huginn_mind::{
-    Faculty, HuginnMindRequest, HuginnMindResponse, Mind, OwnedRedbMessagePackBackingStore, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineDocumentView, PipelineProvenance,
-    PipelineSelectionPage,
+    Faculty, HuginnMindRequest, HuginnMindResponse, Mind, OwnedRedbMessagePackBackingStore, PipelineAdmissionBatch, PipelineAdmissionOutcome,
+    PipelineDocumentView, PipelineProvenance, PipelineSelectionPage,
 };
 use tempfile::TempDir;
 
@@ -37,24 +37,38 @@ impl Clock for Fixed {
     }
 }
 
-/// A launcher that records every start, and fails them on command.
+/// A launcher that records every start as (instance, turn, label), fails them
+/// on command, and answers `alive` from a set of labels: `start` fills it,
+/// `kill` empties it for a label and `revive` fills it for a seeded run.
 #[derive(Default)]
 pub(crate) struct Recording {
-    pub(crate) started: RefCell<Vec<(RunTurn, Label)>>,
+    pub(crate) started: RefCell<Vec<(Slug, RunTurn, Label)>>,
     pub(crate) fail: Cell<bool>,
+    running: RefCell<Vec<Label>>,
+}
+
+impl Recording {
+    pub(crate) fn kill(&self, run: &str) {
+        self.running.borrow_mut().retain(|label| label.0 != run);
+    }
+
+    pub(crate) fn revive(&self, run: &str) {
+        self.running.borrow_mut().push(label(run));
+    }
 }
 
 impl Launcher for Recording {
-    fn start(&self, turn: RunTurn, label: &Label) -> Result<()> {
+    fn start(&self, instance: &Slug, turn: RunTurn, label: &Label) -> Result<()> {
         if self.fail.get() {
             return Err(anyhow!("refused"));
         }
-        self.started.borrow_mut().push((turn, label.clone()));
+        self.started.borrow_mut().push((instance.clone(), turn, label.clone()));
+        self.running.borrow_mut().push(label.clone());
         Ok(())
     }
 
-    fn alive(&self, _turn: RunTurn, _label: &Label) -> bool {
-        false
+    fn alive(&self, _instance: &Slug, _turn: RunTurn, label: &Label) -> bool {
+        self.running.borrow().contains(label)
     }
 }
 
@@ -341,14 +355,14 @@ pub(crate) fn withdrawn() -> ResolutionOutcome {
 }
 
 /// Her Self run on `item`, ended, as `n` in a history: opened, closed, and
-/// followed by a ruling when `leaves_something` (the one document that makes it
-/// non-empty).
+/// followed by a follow-up sourced from `item` when `leaves_something` (the
+/// document that cites the item, which makes the run non-empty).
 pub(crate) fn ended_run(mind: &TestMind, item: &PipelineRef, prefix: &str, n: usize, leaves_something: bool) {
     let label = format!("{prefix}{n}");
     mind.committed(vec![run(&label, RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(item))]);
     let mut closing = vec![close(run_ref(&label), recorded())];
     if leaves_something {
-        closing.push(ruling(&format!("left{n}"), None));
+        closing.push(follow_up(&format!("{prefix}left{n}"), item));
     }
     mind.committed(closing);
 }