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

mind-body launch: dead runs closed in the opening batch, Busy from admission, breaker over the item's citations

Browse source 
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>
This commit is contained in:
Meta Vens 2026-10-10 05:32:19 +02:00
commit f624feec75
3 changed files with 592 additions and 206 deletions
Download patch file Download diff file Expand all files Collapse all files

557
crates/mind-body/src/launch.rs
View file

@ -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());
}
}

205
crates/mind-body/src/queue.rs
View file

@ -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}");

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

@ -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);
}