Launch primitive: queue, breaker, open_and_launch and the run aliases
Adds the turn and operated_by aliases on the run kind, and in mind-body the Launcher port with SystemdLauncher, the work queue, the repetition breaker and open_and_launch. Not yet built: this commit may not compile. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
f6b3ec5e78
commit
ea9c324663
7 changed files with 1291 additions and 4 deletions
|
|
@ -1298,6 +1298,41 @@ mod tests {
|
|||
)), "{facts:?}");
|
||||
}
|
||||
|
||||
/// The run aliases read the run's own fields, so a launcher can ask for the
|
||||
/// live Self runs of hers; their domains are the leaf's closed sets.
|
||||
#[test]
|
||||
fn the_turn_and_operated_by_aliases_select_runs_by_their_fields() {
|
||||
let mut mind = seeded();
|
||||
let variant = |label: &str, turn: RunTurn, operated_by: RunOperator| {
|
||||
let D::Run(mut value) = run(label, &[]) else { panic!() };
|
||||
(value.turn, value.operated_by) = (turn, operated_by);
|
||||
D::Run(value)
|
||||
};
|
||||
committed(admit(
|
||||
&mut mind,
|
||||
vec![
|
||||
variant("self-mind", RunTurn::SelfRun, RunOperator::Mind),
|
||||
variant("self-op", RunTurn::SelfRun, RunOperator::Operator),
|
||||
variant("persona-mind", RunTurn::PersonaTurn, RunOperator::Mind),
|
||||
variant("persona-op", RunTurn::PersonaTurn, RunOperator::Operator),
|
||||
],
|
||||
));
|
||||
let asking = |turn: &[&str], operated_by: &[&str]| {
|
||||
let selection = with(with(of_kinds(&[K::Run]), any_of("turn", turn)), any_of("operated_by", operated_by));
|
||||
ids(&mind.query(&selection).unwrap())
|
||||
};
|
||||
let run_id = |label: &str| format!("{INSTANCE}:run:{label}");
|
||||
assert_eq!(asking(&["PersonaTurn"], &["Mind"]), vec![run_id("persona-mind")]);
|
||||
assert_eq!(asking(&["SelfRun"], &["Operator"]), vec![run_id("self-op")]);
|
||||
assert_eq!(asking(&["SelfRun", "PersonaTurn"], &["Mind"]).len(), 2);
|
||||
let outside = with(of_kinds(&[K::Run]), any_of("turn", &["Waker"]));
|
||||
assert!(invalid("fields[0].values", "Waker")(&refused(&mind, &outside)));
|
||||
let outside = with(of_kinds(&[K::Run]), any_of("operated_by", &["Root"]));
|
||||
assert!(invalid("fields[0].values", "Root")(&refused(&mind, &outside)));
|
||||
let other_kind = with(of_kinds(&[K::Ruling]), any_of("turn", &["SelfRun"]));
|
||||
assert!(matches!(mind.query(&other_kind), Err(MindRefusal::SelectionInvalid { .. })));
|
||||
}
|
||||
|
||||
/// The substrate's own refusals cross as themselves: an undeclared alias
|
||||
/// is a `SelectionInvalid` from the substrate's door, before the organ's.
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ use std::collections::BTreeMap;
|
|||
use cultnet_rs::{Row, RowSet, Selection};
|
||||
use eureka_pipeline::{
|
||||
FindingConfidence, FindingOrigin, FindingSeverity, Label, OrgRepo, PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome,
|
||||
RulingAuthority, Slug,
|
||||
RulingAuthority, RunOperator, RunTurn, Slug,
|
||||
};
|
||||
|
||||
use crate::docs::{CitationRole, Held, kind_of_id, kind_of_type};
|
||||
|
|
@ -40,6 +40,10 @@ pub(crate) enum Alias {
|
|||
Authority,
|
||||
ClaimOutcome,
|
||||
Outcome,
|
||||
/// A run's `turn`, read from its fields.
|
||||
Turn,
|
||||
/// A run's `operated_by`, read from its fields.
|
||||
OperatedBy,
|
||||
}
|
||||
|
||||
const FACULTIES: [&str; 7] = ["SelfFaculty", "Imagination", "Hands", "Soul", "Life", "Eyes", "Operator"];
|
||||
|
|
@ -47,7 +51,7 @@ const OUTCOMES: [&str; 6] = ["Superseded", "Answered", "Fixed", "Deferred", "Rec
|
|||
const BOOLEANS: [&str; 2] = ["true", "false"];
|
||||
|
||||
impl Alias {
|
||||
const ALL: [Alias; 11] = [
|
||||
const ALL: [Alias; 13] = [
|
||||
Self::Root,
|
||||
Self::InForce,
|
||||
Self::Faculty,
|
||||
|
|
@ -59,6 +63,8 @@ impl Alias {
|
|||
Self::Authority,
|
||||
Self::ClaimOutcome,
|
||||
Self::Outcome,
|
||||
Self::Turn,
|
||||
Self::OperatedBy,
|
||||
];
|
||||
|
||||
fn name(self) -> &'static str {
|
||||
|
|
@ -74,6 +80,8 @@ impl Alias {
|
|||
Self::Authority => "authority",
|
||||
Self::ClaimOutcome => "claim_outcome",
|
||||
Self::Outcome => "outcome",
|
||||
Self::Turn => "turn",
|
||||
Self::OperatedBy => "operated_by",
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -93,6 +101,7 @@ impl Alias {
|
|||
Self::Authority => kind == K::Ruling,
|
||||
Self::ClaimOutcome => kind == K::Verdict,
|
||||
Self::Outcome => kind == K::Resolution,
|
||||
Self::Turn | Self::OperatedBy => kind == K::Run,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -107,6 +116,8 @@ impl Alias {
|
|||
Self::Authority => Some(RulingAuthority::NAMES),
|
||||
Self::ClaimOutcome => Some(eureka_pipeline::ClaimOutcome::NAMES),
|
||||
Self::Outcome => Some(&OUTCOMES),
|
||||
Self::Turn => Some(RunTurn::NAMES),
|
||||
Self::OperatedBy => Some(RunOperator::NAMES),
|
||||
Self::Root | Self::Repo | Self::Cut => None,
|
||||
}
|
||||
}
|
||||
|
|
@ -208,6 +219,10 @@ impl SelectionRow {
|
|||
put(Alias::ClaimOutcome, verdict.claims.iter().map(|claim| format!("{:?}", claim.outcome)).collect())
|
||||
}
|
||||
D::Resolution(resolution) => put(Alias::Outcome, vec![outcome_name(&resolution.outcome).to_string()]),
|
||||
D::Run(run) => {
|
||||
put(Alias::Turn, vec![format!("{:?}", run.turn)]);
|
||||
put(Alias::OperatedBy, vec![format!("{:?}", run.operated_by)]);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
Self { kind: held.kind, key: held.key.clone(), ordinal: view.admission.ordinal, values, refs, view }
|
||||
|
|
@ -367,7 +382,7 @@ mod tests {
|
|||
(K::Instance, &[]),
|
||||
(K::Stewardship, &["repo"]),
|
||||
(K::HandOff, &["repo"]),
|
||||
(K::Run, &[]),
|
||||
(K::Run, &["turn", "operated_by"]),
|
||||
];
|
||||
for (kind, extra) in table {
|
||||
let want = [&["root", "in_force", "faculty"][..], extra].concat();
|
||||
|
|
|
|||
|
|
@ -9,7 +9,9 @@ publish.workspace = true
|
|||
# operator's brake and burn-rate dial today: a root-owned per-instance CultCache
|
||||
# store (`control`), the operator's CLI over it (`cli`, the `mind-control`
|
||||
# binary) and the one fail-closed reader every acting organ calls
|
||||
# (`control::read_effective`). The waker joins this crate in cut mind-wake.
|
||||
# (`control::read_effective`). It also holds the launch primitive (`launch`,
|
||||
# `queue`): the one way her organs open a run and start its unit. The waker
|
||||
# joins this crate in cut mind-wake.
|
||||
#
|
||||
# `rust_decimal` is the dial's number type: heat, cadence and the run cap are
|
||||
# money-adjacent and the tests pin exact products, which a float would blur. It
|
||||
|
|
@ -18,7 +20,10 @@ publish.workspace = true
|
|||
anyhow = "1"
|
||||
chrono = { version = "0.4.44", features = ["serde"] }
|
||||
cultcache-rs.workspace = true
|
||||
cultnet-rs.workspace = true
|
||||
eureka-pipeline = { path = "../eureka-pipeline" }
|
||||
eureka-state = { path = "../eureka-state" }
|
||||
huginn-mind = { path = "../huginn-mind" }
|
||||
rust_decimal = { version = "1", default-features = false, features = ["std", "serde"] }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
|
||||
|
|
|
|||
511
crates/mind-body/src/launch.rs
Normal file
511
crates/mind-body/src/launch.rs
Normal file
|
|
@ -0,0 +1,511 @@
|
|||
//! 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.
|
||||
|
||||
use anyhow::{Result, anyhow};
|
||||
use chrono::{DateTime, Utc};
|
||||
use eureka_pipeline::{
|
||||
Date, Label, PipelineDocument, PipelineKind, PipelineQuestion, PipelineRef, PipelineResolution, PipelineRun, QuestionOption,
|
||||
ResolutionOutcome, RunOperator, RunTurn, Slug,
|
||||
};
|
||||
use huginn_mind::{MindRefusal, PipelineAdmissionOutcome};
|
||||
use rust_decimal::Decimal;
|
||||
|
||||
use crate::queue::{Breaker, MindPort, any_of, breaker, headers, in_force, of_kind, queue};
|
||||
|
||||
/// 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;
|
||||
}
|
||||
|
||||
/// The time the primitive reads: run labels, dates and receipts.
|
||||
pub trait Clock {
|
||||
fn now(&self) -> DateTime<Utc>;
|
||||
}
|
||||
|
||||
pub struct SystemClock;
|
||||
|
||||
impl Clock for SystemClock {
|
||||
fn now(&self) -> DateTime<Utc> {
|
||||
Utc::now()
|
||||
}
|
||||
}
|
||||
|
||||
/// What `systemctl` answered: whether it succeeded, and what it printed.
|
||||
pub struct SystemctlReply {
|
||||
pub success: bool,
|
||||
pub stdout: String,
|
||||
}
|
||||
|
||||
/// The one door to `systemctl`, so tests never reach a real unit.
|
||||
pub trait Systemctl {
|
||||
fn run(&self, args: &[&str]) -> Result<SystemctlReply>;
|
||||
}
|
||||
|
||||
struct RealSystemctl;
|
||||
|
||||
impl Systemctl for RealSystemctl {
|
||||
fn run(&self, args: &[&str]) -> Result<SystemctlReply> {
|
||||
let output = std::process::Command::new("systemctl").args(args).output().map_err(|_| anyhow!("systemctl could not be run"))?;
|
||||
Ok(SystemctlReply { success: output.status.success(), stdout: String::from_utf8_lossy(&output.stdout).into_owned() })
|
||||
}
|
||||
}
|
||||
|
||||
/// Her units under systemd, templated by instance:
|
||||
/// `mind-persona@<instance>:<label>.service` and
|
||||
/// `mind-self@<instance>:<label>.service`.
|
||||
pub struct SystemdLauncher {
|
||||
instance: Slug,
|
||||
systemctl: Box<dyn Systemctl>,
|
||||
}
|
||||
|
||||
impl SystemdLauncher {
|
||||
pub fn new(instance: Slug) -> Self {
|
||||
Self::over(instance, 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)
|
||||
}
|
||||
}
|
||||
|
||||
impl Launcher for SystemdLauncher {
|
||||
fn start(&self, turn: RunTurn, label: &Label) -> Result<()> {
|
||||
let reply = self.systemctl.run(&["start", "--no-block", &self.unit(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"))
|
||||
}
|
||||
}
|
||||
|
||||
/// A run's label from the instant it opens: UTC, to the millisecond. Two
|
||||
/// launches in one millisecond collide on the run's key, admission answers
|
||||
/// Conflict, and nothing launches.
|
||||
pub fn run_label(now: DateTime<Utc>) -> Label {
|
||||
Label(now.format("mind-%Y%m%dT%H%M%S%3fZ").to_string())
|
||||
}
|
||||
|
||||
pub struct Ports<'a> {
|
||||
pub mind: &'a dyn MindPort,
|
||||
pub launcher: &'a dyn Launcher,
|
||||
pub clock: &'a dyn Clock,
|
||||
/// The `host` the run records.
|
||||
pub host: &'a str,
|
||||
}
|
||||
|
||||
pub struct LaunchRequest {
|
||||
pub turn: RunTurn,
|
||||
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,
|
||||
}
|
||||
|
||||
/// Why admission did not open the run: its own refusal, or one of the two
|
||||
/// answers that mean the run's key is taken.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum Declined {
|
||||
Admission(MindRefusal),
|
||||
Conflict,
|
||||
AlreadyAdmitted,
|
||||
/// The run committed, its unit did not start, and the run was withdrawn.
|
||||
StartFailed,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum LaunchOutcome {
|
||||
Launched { run: PipelineRef },
|
||||
NotQueued { item: PipelineRef },
|
||||
Tripped { item: PipelineRef, question: PipelineRef },
|
||||
Busy { run: PipelineRef },
|
||||
Refused(Declined),
|
||||
}
|
||||
|
||||
fn run_ref(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())
|
||||
}
|
||||
|
||||
/// The root of a full pipeline id: the campaign slug of a campaign's document.
|
||||
fn root_of(item: &PipelineRef) -> &str {
|
||||
item.id.0.split(':').next().unwrap_or_default()
|
||||
}
|
||||
|
||||
pub fn open_and_launch(ports: &Ports, request: LaunchRequest) -> Result<LaunchOutcome> {
|
||||
let LaunchRequest { turn, claims, agent, run_cap_usd } = 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 });
|
||||
}
|
||||
}
|
||||
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),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
let mut campaigns: Vec<Slug> = Vec::new();
|
||||
for claim in &claims {
|
||||
let root = Slug(root_of(claim).to_string());
|
||||
if !campaigns.contains(&root) {
|
||||
campaigns.push(root);
|
||||
}
|
||||
}
|
||||
let 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(),
|
||||
claims,
|
||||
campaigns,
|
||||
};
|
||||
if let Err(declined) = committed(ports.mind.admit(&agent, &label.0, vec![PipelineDocument::Run(run)])?) {
|
||||
return Ok(LaunchOutcome::Refused(declined));
|
||||
}
|
||||
let run = run_ref(&instance, &label);
|
||||
if ports.launcher.start(turn, &label).is_err() {
|
||||
let withdrawal = PipelineResolution {
|
||||
subject: run,
|
||||
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(),
|
||||
resolved_on: date(now),
|
||||
};
|
||||
return match committed(ports.mind.admit(&agent, &label.0, vec![PipelineDocument::Resolution(withdrawal)])?) {
|
||||
Ok(()) => Ok(LaunchOutcome::Refused(Declined::StartFailed)),
|
||||
Err(_) => Err(anyhow!("the run could not be withdrawn after its unit failed to start")),
|
||||
};
|
||||
}
|
||||
Ok(LaunchOutcome::Launched { run })
|
||||
}
|
||||
|
||||
/// Whether admission landed the batch; if not, why.
|
||||
fn committed(outcome: PipelineAdmissionOutcome) -> Result<(), Declined> {
|
||||
match outcome {
|
||||
PipelineAdmissionOutcome::Committed { .. } => Ok(()),
|
||||
PipelineAdmissionOutcome::Refused(refusal) => Err(Declined::Admission(refusal)),
|
||||
PipelineAdmissionOutcome::Conflict { .. } => Err(Declined::Conflict),
|
||||
PipelineAdmissionOutcome::AlreadyAdmitted { .. } => Err(Declined::AlreadyAdmitted),
|
||||
}
|
||||
}
|
||||
|
||||
/// `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('.', "-")))
|
||||
}
|
||||
|
||||
/// The question a tripped item raises, in the item's own campaign.
|
||||
fn breaker_question(item: &PipelineRef, now: DateTime<Utc>) -> PipelineQuestion {
|
||||
PipelineQuestion {
|
||||
campaign: Slug(root_of(item).to_string()),
|
||||
label: breaker_label(item),
|
||||
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(),
|
||||
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() },
|
||||
],
|
||||
recommended: "reopen".into(),
|
||||
depends: vec![],
|
||||
raised_in: Some(item.clone()),
|
||||
asked_on: date(now),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::cell::{Cell, RefCell};
|
||||
|
||||
use eureka_pipeline::PipelineKind as K;
|
||||
use huginn_mind::PipelineAdmissionOutcome as Outcome;
|
||||
|
||||
use super::*;
|
||||
use crate::testkit::*;
|
||||
|
||||
struct Rig {
|
||||
mind: TestMind,
|
||||
launcher: Recording,
|
||||
clock: Fixed,
|
||||
}
|
||||
|
||||
impl Rig {
|
||||
fn new() -> Self {
|
||||
Self { mind: TestMind::seeded(), launcher: Recording::default(), clock: Fixed(Cell::new(at(0))) }
|
||||
}
|
||||
|
||||
fn launch(&self, turn: RunTurn, claims: &[PipelineRef], cap: Decimal) -> LaunchOutcome {
|
||||
self.launch_through(&self.mind, turn, claims, cap)
|
||||
}
|
||||
|
||||
fn launch_through(&self, mind: &dyn MindPort, turn: RunTurn, claims: &[PipelineRef], cap: Decimal) -> 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()
|
||||
}
|
||||
|
||||
fn at(&self, milliseconds: i64) {
|
||||
self.clock.0.set(at(milliseconds));
|
||||
}
|
||||
|
||||
fn started(&self) -> usize {
|
||||
self.launcher.started.borrow().len()
|
||||
}
|
||||
|
||||
fn runs(&self) -> usize {
|
||||
self.mind.ids(K::Run).len()
|
||||
}
|
||||
}
|
||||
|
||||
fn five() -> Decimal {
|
||||
Decimal::new(5, 0)
|
||||
}
|
||||
|
||||
fn the_run(view: &huginn_mind::PipelineDocumentView) -> &PipelineRun {
|
||||
let PipelineDocument::Run(run) = &view.document else { panic!("not a run") };
|
||||
run
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_queued_spec_opens_a_run_that_records_its_launch_and_then_its_unit_starts() {
|
||||
let rig = Rig::new();
|
||||
rig.mind.committed(vec![spec("x")]);
|
||||
let LaunchOutcome::Launched { run } = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()) else { panic!("not launched") };
|
||||
assert_eq!(run.id.0, "yggdrasil:run:mind-20261009T131500000Z");
|
||||
let view = rig.mind.view(K::Run, &run.id.0).unwrap();
|
||||
let held = the_run(&view);
|
||||
assert_eq!((held.operated_by, held.turn), (RunOperator::Mind, RunTurn::SelfRun));
|
||||
assert_eq!((held.host.0.as_str(), held.started_on.0.as_str(), held.budget_usd.0.as_str()), ("yggdrasil-host", "2026-10-09", "5"));
|
||||
assert_eq!(held.claims, vec![spec_ref("x")]);
|
||||
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"))]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_runs_budget_is_the_cap_passed_in() {
|
||||
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);
|
||||
let LaunchOutcome::Launched { run } = rig.launch(RunTurn::PersonaTurn, &[], cap) else { panic!("not launched") };
|
||||
let view = rig.mind.view(K::Run, &run.id.0).unwrap();
|
||||
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());
|
||||
}
|
||||
assert_eq!(rig.started(), 3, "a Persona turn is never Busy");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_labels_are_utc_to_the_millisecond() {
|
||||
assert_eq!(run_label(at(123)).0, "mind-20261009T131500123Z");
|
||||
assert_ne!(run_label(at(123)), run_label(at(124)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
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));
|
||||
assert!(matches!(second, LaunchOutcome::Refused(_)), "{second:?}");
|
||||
assert_eq!((rig.runs(), rig.started()), (1, 1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_refused_run_never_starts_a_unit() {
|
||||
let rig = Rig::new();
|
||||
rig.mind.committed(vec![spec("x"), follow_up("later", &spec_ref("x"))]);
|
||||
let later = reference(K::FollowUp, &id("follow_up", "later"));
|
||||
rig.mind.committed(vec![run("op", RunTurn::SelfRun, RunOperator::Operator, std::slice::from_ref(&later))]);
|
||||
let outcome = rig.launch(RunTurn::SelfRun, std::slice::from_ref(&later), five());
|
||||
assert!(matches!(outcome, LaunchOutcome::Refused(Declined::Admission(MindRefusal::AlreadyClaimed { .. }))), "{outcome:?}");
|
||||
for (answer, declined) in [
|
||||
(Outcome::Conflict { identities: vec![] }, Declined::Conflict),
|
||||
(Outcome::AlreadyAdmitted { receipt_id: "r".into() }, Declined::AlreadyAdmitted),
|
||||
] {
|
||||
let mind = Answering { inner: &rig.mind, outcome: answer };
|
||||
assert_eq!(rig.launch_through(&mind, RunTurn::SelfRun, &[spec_ref("x")], five()), LaunchOutcome::Refused(declined));
|
||||
}
|
||||
assert_eq!((rig.runs(), rig.started()), (1, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_start_failure_withdraws_the_run_it_opened_and_only_that_run() {
|
||||
let rig = Rig::new();
|
||||
rig.mind.committed(vec![spec("x"), spec("y")]);
|
||||
rig.mind.committed(vec![run("other", RunTurn::SelfRun, RunOperator::Operator, &[spec_ref("y")])]);
|
||||
rig.launcher.fail.set(true);
|
||||
let outcome = rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five());
|
||||
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);
|
||||
// The spec is free again.
|
||||
rig.launcher.fail.set(false);
|
||||
rig.at(1);
|
||||
assert!(matches!(rig.launch(RunTurn::SelfRun, &[spec_ref("x")], five()), LaunchOutcome::Launched { .. }));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_spec_that_is_reported_blocked_or_claimed_is_not_queued_and_admits_nothing() {
|
||||
let rig = Rig::new();
|
||||
rig.mind.committed(vec![spec("done"), spec("blocked"), spec("held"), spec("free")]);
|
||||
rig.mind.committed(vec![report("done")]);
|
||||
rig.mind.committed(vec![question_in("fork", &spec_ref("blocked"))]);
|
||||
rig.mind.committed(vec![run("op", RunTurn::SelfRun, RunOperator::Operator, &[spec_ref("held")])]);
|
||||
for item in ["done", "blocked", "held"] {
|
||||
let outcome = rig.launch(RunTurn::SelfRun, &[spec_ref(item)], five());
|
||||
assert_eq!(outcome, LaunchOutcome::NotQueued { item: spec_ref(item) });
|
||||
}
|
||||
let mixed = rig.launch(RunTurn::SelfRun, &[spec_ref("free"), spec_ref("done")], five());
|
||||
assert_eq!(mixed, LaunchOutcome::NotQueued { item: spec_ref("done") }, "one unqueued claim refuses the whole request");
|
||||
assert_eq!((rig.runs(), rig.started()), (1, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_second_live_self_run_of_hers_is_busy_and_other_runs_do_not_count() {
|
||||
let rig = Rig::new();
|
||||
rig.mind.committed(vec![spec("x"), spec("y")]);
|
||||
// None of these is a live Self run of hers.
|
||||
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.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")
|
||||
};
|
||||
let (runs, started) = (rig.runs(), rig.started());
|
||||
rig.at(2);
|
||||
assert_eq!(rig.launch(RunTurn::SelfRun, &[spec_ref("y")], five()), LaunchOutcome::Busy { run: first.clone() });
|
||||
assert_eq!((rig.runs(), rig.started()), (runs, started), "Busy admits and starts nothing");
|
||||
// Ending her run frees the slot.
|
||||
rig.mind.committed(vec![close(first, recorded())]);
|
||||
rig.at(3);
|
||||
assert!(matches!(rig.launch(RunTurn::SelfRun, &[spec_ref("y")], five()), LaunchOutcome::Launched { .. }));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_tripped_item_raises_one_question_launches_nothing_and_a_ruling_reopens_it() {
|
||||
let rig = Rig::new();
|
||||
rig.mind.committed(vec![spec("x")]);
|
||||
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");
|
||||
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()]);
|
||||
let view = rig.mind.view(K::Question, &question).unwrap();
|
||||
let PipelineDocument::Question(asked) = &view.document else { panic!("not a question") };
|
||||
assert_eq!((asked.raised_in.clone(), asked.campaign.0.as_str()), (Some(spec_ref("x")), CAMPAIGN));
|
||||
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.
|
||||
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 { .. }));
|
||||
}
|
||||
|
||||
struct Spy {
|
||||
calls: RefCell<Vec<Vec<String>>>,
|
||||
success: Cell<bool>,
|
||||
stdout: RefCell<String>,
|
||||
}
|
||||
|
||||
impl Systemctl for &'static Spy {
|
||||
fn run(&self, args: &[&str]) -> Result<SystemctlReply> {
|
||||
self.calls.borrow_mut().push(args.iter().map(|arg| arg.to_string()).collect());
|
||||
Ok(SystemctlReply { success: self.success.get(), stdout: self.stdout.borrow().clone() })
|
||||
}
|
||||
}
|
||||
|
||||
fn spied(instance: &str) -> (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)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn units_are_templated_by_instance_and_turn() {
|
||||
let (launcher, spy) = spied("eureka");
|
||||
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!(
|
||||
*spy.calls.borrow(),
|
||||
vec![
|
||||
vec!["start", "--no-block", "mind-self@eureka:mind-20261009T131500123Z.service"],
|
||||
vec!["start", "--no-block", "mind-persona@eureka:mind-20261009T131500123Z.service"],
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_unit_is_alive_while_active_activating_or_reloading() {
|
||||
let (launcher, spy) = spied("eureka");
|
||||
let l = label("a");
|
||||
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!(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");
|
||||
spy.success.set(false);
|
||||
let error = launcher.start(RunTurn::SelfRun, &label("canary-label-2c8d04")).unwrap_err();
|
||||
let shown = format!("{error:#}");
|
||||
assert!(!shown.contains("canary"), "{shown}");
|
||||
}
|
||||
}
|
||||
|
|
@ -4,6 +4,15 @@
|
|||
//! the operator's hand on it. The brake and the dial are the operator's consent
|
||||
//! (ruling `eureka-body:ruling:brake-dial-root-store-now`), not her body: the
|
||||
//! store is root-owned and no organ of hers writes it.
|
||||
//!
|
||||
//! `launch` and `queue` are the launch primitive: the work queue, the
|
||||
//! repetition breaker, the one-live-Self-run rule, and the single path that
|
||||
//! opens a run in the mind and only then starts its unit.
|
||||
|
||||
pub mod cli;
|
||||
pub mod control;
|
||||
pub mod launch;
|
||||
pub mod queue;
|
||||
|
||||
#[cfg(test)]
|
||||
mod testkit;
|
||||
|
|
|
|||
344
crates/mind-body/src/queue.rs
Normal file
344
crates/mind-body/src/queue.rs
Normal file
|
|
@ -0,0 +1,344 @@
|
|||
//! The Persona's view of open work, and the guard on it: what may be worked
|
||||
//! (`queue`) and whether an item has stopped paying for its runs (`breaker`).
|
||||
//!
|
||||
//! Nothing here keeps state. Both answers are derived from the mind on every
|
||||
//! call, through `MindPort`: the mind is the only store, so a breaker that
|
||||
//! needed a file of its own would be a second owner of what the runs already
|
||||
//! record (target invariant run-is-the-grant).
|
||||
|
||||
use anyhow::{Result, anyhow, bail};
|
||||
use cultnet_rs::{Citation, FieldPredicate, Incoming, RecordRef, Selection};
|
||||
use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, Slug};
|
||||
use eureka_state::HuginnClient;
|
||||
use huginn_mind::{
|
||||
Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineDocumentSummary, PipelineFacts,
|
||||
PipelinePageItems, PipelineProvenance, PipelineSelectionPage, PipelineStatusSummary,
|
||||
};
|
||||
|
||||
/// The last `CONSECUTIVE` ended runs over an item, all empty, trip its breaker.
|
||||
pub const CONSECUTIVE: usize = 3;
|
||||
/// So do `EMPTY_SHARE` empty ones among the last `WINDOW` (Codex auto-review's
|
||||
/// per-turn breaker, `docs/research/autonomous-harness-prior-art.md` section 0).
|
||||
pub const WINDOW: usize = 50;
|
||||
pub const EMPTY_SHARE: usize = 10;
|
||||
|
||||
/// The two things her launch organs ask of the mind: a typed selection, and
|
||||
/// an admission as herself. The impl for a daemon is `HuginnMind`; a test
|
||||
/// hands in a mind opened over a temporary directory.
|
||||
pub trait MindPort {
|
||||
/// The instance this port acts for.
|
||||
fn instance(&self) -> &Slug;
|
||||
/// One page of the selection. A refusal is an error here and names no
|
||||
/// value.
|
||||
fn query(&self, selection: &Selection) -> Result<PipelineSelectionPage>;
|
||||
/// Admits `documents` as one batch, attributed to `agent` in `session`.
|
||||
fn admit(&self, agent: &str, session: &str, documents: Vec<PipelineDocument>) -> Result<PipelineAdmissionOutcome>;
|
||||
}
|
||||
|
||||
/// The daemon's mind, reached over the `eureka-state` client.
|
||||
pub struct HuginnMind {
|
||||
client: HuginnClient,
|
||||
}
|
||||
|
||||
impl HuginnMind {
|
||||
pub fn new(client: HuginnClient) -> Self {
|
||||
Self { client }
|
||||
}
|
||||
}
|
||||
|
||||
impl MindPort for HuginnMind {
|
||||
fn instance(&self) -> &Slug {
|
||||
self.client.instance()
|
||||
}
|
||||
|
||||
fn query(&self, selection: &Selection) -> Result<PipelineSelectionPage> {
|
||||
let request = HuginnMindRequest::Query { instance: self.instance().clone(), selection: selection.clone(), semantic: None };
|
||||
match self.client.call(request).map_err(|error| anyhow!("the mind could not be asked: {}", client_error_code(&error)))? {
|
||||
HuginnMindResponse::Query(page) => Ok(page),
|
||||
HuginnMindResponse::Refused(_) => bail!("the mind refused the query"),
|
||||
_ => bail!("the mind answered a query with something else"),
|
||||
}
|
||||
}
|
||||
|
||||
fn admit(&self, agent: &str, session: &str, documents: Vec<PipelineDocument>) -> Result<PipelineAdmissionOutcome> {
|
||||
let batch = PipelineAdmissionBatch {
|
||||
instance: self.instance().clone(),
|
||||
provenance: PipelineProvenance { faculty: Faculty::SelfFaculty, agent: agent.into(), session: session.into(), tool: "mind-body".into() },
|
||||
documents,
|
||||
};
|
||||
match self.client.call(HuginnMindRequest::Admit(batch)).map_err(|error| anyhow!("the mind could not be asked: {}", client_error_code(&error)))? {
|
||||
HuginnMindResponse::Admit(outcome) => Ok(outcome),
|
||||
_ => bail!("the mind answered an admission with something else"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The variant of a client failure, without its detail (an endpoint and a
|
||||
/// transport message are not ours to print).
|
||||
fn client_error_code(error: &eureka_state::ClientError) -> &'static str {
|
||||
match error {
|
||||
eureka_state::ClientError::Unavailable { .. } => "unavailable",
|
||||
eureka_state::ClientError::Rejected { .. } => "rejected",
|
||||
eureka_state::ClientError::TooLarge { .. } => "too_large",
|
||||
eureka_state::ClientError::Unencodable { .. } => "unencodable",
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn of_kind(kind: PipelineKind) -> Selection {
|
||||
Selection { schemas: Some(vec![kind.type_id().to_string()]), ..Selection::default() }
|
||||
}
|
||||
|
||||
pub(crate) fn any_of(mut selection: Selection, index: &str, values: &[&str]) -> Selection {
|
||||
selection.fields.get_or_insert_with(Vec::new).push(FieldPredicate {
|
||||
index: index.into(),
|
||||
op: "any_of".into(),
|
||||
values: Some(values.iter().map(|value| value.to_string()).collect()),
|
||||
number: None,
|
||||
});
|
||||
selection
|
||||
}
|
||||
|
||||
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<()> {
|
||||
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(());
|
||||
}
|
||||
}
|
||||
match page.next {
|
||||
Some(cursor) => asking.cursor = Some(cursor),
|
||||
None => return Ok(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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>> {
|
||||
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 {
|
||||
claims.extend(held);
|
||||
}
|
||||
}
|
||||
Ok(claims)
|
||||
}
|
||||
|
||||
/// 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>> {
|
||||
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)))? {
|
||||
if let PipelineFacts::Question { raised_in: Some(item), .. } = header.facts {
|
||||
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())
|
||||
}
|
||||
|
||||
/// Whether an item may still be launched on.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum Breaker {
|
||||
Clear,
|
||||
Trip,
|
||||
}
|
||||
|
||||
/// The repetition breaker, derived from the item's Self runs and from what
|
||||
/// else the mind admitted.
|
||||
///
|
||||
/// 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
|
||||
.iter()
|
||||
.filter(|run| !withdrawn(run))
|
||||
.map(|run| (run.admission.ordinal, matches!(run.status, PipelineStatusSummary::Resolved { .. })))
|
||||
.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
|
||||
})?;
|
||||
|
||||
// Newest first, like `considered`.
|
||||
let empty: Vec<bool> = considered
|
||||
.iter()
|
||||
.map(|index| {
|
||||
let end = ran.get(index + 1).map_or(u64::MAX, |next| next.0);
|
||||
!admitted.iter().any(|ordinal| *ordinal >= ran[*index].0 && *ordinal < end)
|
||||
})
|
||||
.collect();
|
||||
let consecutive = empty.len() >= CONSECUTIVE && empty[..CONSECUTIVE].iter().all(|empty| *empty);
|
||||
let share = empty.iter().filter(|empty| **empty).count() >= EMPTY_SHARE;
|
||||
Ok(if consecutive || share { Breaker::Trip } else { Breaker::Clear })
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::net::SocketAddr;
|
||||
use std::time::Duration;
|
||||
|
||||
use super::*;
|
||||
use crate::testkit::*;
|
||||
|
||||
/// A mind holding the spec `x` and a history of ended runs on it.
|
||||
fn worked(runs: &[bool]) -> (TestMind, PipelineRef) {
|
||||
let mind = TestMind::seeded();
|
||||
mind.committed(vec![spec("x")]);
|
||||
let item = spec_ref("x");
|
||||
history(&mind, &item, runs);
|
||||
(mind, item)
|
||||
}
|
||||
|
||||
const E: bool = false;
|
||||
const N: bool = true;
|
||||
|
||||
fn tripped(runs: &[bool]) -> bool {
|
||||
let (mind, item) = worked(runs);
|
||||
breaker(&mind, &item).unwrap() == Breaker::Trip
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn three_consecutive_empty_runs_trip_and_two_do_not() {
|
||||
assert!(!tripped(&[]));
|
||||
assert!(!tripped(&[E, E]));
|
||||
assert!(tripped(&[E, E, E]));
|
||||
assert!(tripped(&[N, E, E, E]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_document_in_a_runs_window_makes_that_run_not_empty() {
|
||||
assert!(!tripped(&[E, E, N, E, E]), "one run with a document in its window breaks the streak");
|
||||
assert!(tripped(&[E, E, N, E, E, E]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ten_empty_of_the_last_fifty_trip_with_gaps_and_nine_do_not() {
|
||||
let ten: Vec<bool> = [E, E, N].repeat(5);
|
||||
assert!(tripped(&ten), "10 empty, never 3 in a row");
|
||||
let mut nine: Vec<bool> = [E, E, N].repeat(4);
|
||||
nine.push(E);
|
||||
assert!(!tripped(&nine), "9 empty, never 3 in a row");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn only_the_last_fifty_ended_runs_are_counted() {
|
||||
let mut runs = vec![E];
|
||||
for _ in 0..9 {
|
||||
runs.extend([E, N]);
|
||||
}
|
||||
runs.extend([N].repeat(32));
|
||||
assert_eq!(runs.len(), 51);
|
||||
assert!(!tripped(&runs), "the tenth empty run is the 51st newest");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_withdrawn_run_never_ran_and_is_no_boundary() {
|
||||
let withdrawn_between = |ruling_after: bool| {
|
||||
let (mind, item) = worked(&[E]);
|
||||
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)]);
|
||||
}
|
||||
ended_run(&mind, &item, 1, false);
|
||||
ended_run(&mind, &item, 2, false);
|
||||
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");
|
||||
}
|
||||
|
||||
#[test]
|
||||
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![close(run_ref("h2"), recorded())]);
|
||||
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");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_queue_is_the_unreported_unblocked_unclaimed_specs_oldest_first() {
|
||||
let mind = TestMind::seeded();
|
||||
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);
|
||||
|
||||
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")]);
|
||||
|
||||
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");
|
||||
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");
|
||||
}
|
||||
|
||||
/// A daemon that is not there: the failure names no input, endpoint or
|
||||
/// instance.
|
||||
#[test]
|
||||
fn an_unreachable_mind_names_no_input() {
|
||||
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 shown = format!("{error:#}");
|
||||
assert!(shown.contains("unavailable"), "{shown}");
|
||||
assert!(!shown.contains(canary) && !shown.contains("127.0.0.1"), "{shown}");
|
||||
}
|
||||
}
|
||||
368
crates/mind-body/src/testkit.rs
Normal file
368
crates/mind-body/src/testkit.rs
Normal file
|
|
@ -0,0 +1,368 @@
|
|||
//! Test support: a real mind over a temporary directory behind `MindPort`, a
|
||||
//! recording launcher and a settable clock. Every test document is built here,
|
||||
//! so a field the leaf adds to a kind is one edit.
|
||||
|
||||
use std::cell::{Cell, RefCell};
|
||||
|
||||
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,
|
||||
PipelineTarget, QuestionOption, ResolutionOutcome, RulingAuthority, RunOperator, RunTurn, Sha, Slug, StructuralDelta, TargetInvariant,
|
||||
};
|
||||
use huginn_mind::{
|
||||
Faculty, Mind, OwnedRedbMessagePackBackingStore, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineDocumentView, PipelineProvenance,
|
||||
PipelineSelectionPage,
|
||||
};
|
||||
use tempfile::TempDir;
|
||||
|
||||
use crate::launch::{Clock, Launcher};
|
||||
use crate::queue::MindPort;
|
||||
|
||||
pub(crate) const INSTANCE: &str = "yggdrasil";
|
||||
pub(crate) const CAMPAIGN: &str = "eureka-body";
|
||||
pub(crate) const REPO: &str = "GameCult/Huginn";
|
||||
|
||||
pub(crate) fn at(milliseconds: i64) -> DateTime<Utc> {
|
||||
Utc.with_ymd_and_hms(2026, 10, 9, 13, 15, 0).unwrap() + Duration::milliseconds(milliseconds)
|
||||
}
|
||||
|
||||
pub(crate) struct Fixed(pub(crate) Cell<DateTime<Utc>>);
|
||||
|
||||
impl Clock for Fixed {
|
||||
fn now(&self) -> DateTime<Utc> {
|
||||
self.0.get()
|
||||
}
|
||||
}
|
||||
|
||||
/// A launcher that records every start, and fails them on command.
|
||||
#[derive(Default)]
|
||||
pub(crate) struct Recording {
|
||||
pub(crate) started: RefCell<Vec<(RunTurn, Label)>>,
|
||||
pub(crate) fail: Cell<bool>,
|
||||
}
|
||||
|
||||
impl Launcher for Recording {
|
||||
fn start(&self, turn: RunTurn, label: &Label) -> Result<()> {
|
||||
if self.fail.get() {
|
||||
return Err(anyhow!("refused"));
|
||||
}
|
||||
self.started.borrow_mut().push((turn, label.clone()));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn alive(&self, _turn: RunTurn, _label: &Label) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct TestMind {
|
||||
_dir: TempDir,
|
||||
instance: Slug,
|
||||
mind: RefCell<Mind<OwnedRedbMessagePackBackingStore>>,
|
||||
now: Cell<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
impl TestMind {
|
||||
/// A mind holding its identity, the repo's stewardship, the campaign and a
|
||||
/// first target.
|
||||
pub(crate) fn seeded() -> Self {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let mind = Mind::open(dir.path(), &Slug(INSTANCE.into())).unwrap();
|
||||
let this = Self { _dir: dir, instance: Slug(INSTANCE.into()), mind: RefCell::new(mind), now: Cell::new(at(0)) };
|
||||
this.put(vec![instance(), stewardship(), campaign(), target()]);
|
||||
this
|
||||
}
|
||||
|
||||
/// Admits as Hands, one second after the last admission.
|
||||
pub(crate) fn put(&self, documents: Vec<PipelineDocument>) -> PipelineAdmissionOutcome {
|
||||
self.now.set(self.now.get() + Duration::seconds(1));
|
||||
let batch = PipelineAdmissionBatch {
|
||||
instance: Slug(INSTANCE.into()),
|
||||
provenance: PipelineProvenance { faculty: Faculty::Hands, agent: "test".into(), session: "test".into(), tool: "test".into() },
|
||||
documents,
|
||||
};
|
||||
self.mind.borrow_mut().admit(batch, self.now.get())
|
||||
}
|
||||
|
||||
pub(crate) fn committed(&self, documents: Vec<PipelineDocument>) {
|
||||
let outcome = self.put(documents);
|
||||
assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
|
||||
}
|
||||
|
||||
pub(crate) fn view(&self, kind: PipelineKind, id: &str) -> Option<PipelineDocumentView> {
|
||||
self.mind.borrow().view(&PipelineRef { kind, id: id.into() }).unwrap()
|
||||
}
|
||||
|
||||
/// Every document of a kind, as ids.
|
||||
pub(crate) fn ids(&self, kind: PipelineKind) -> Vec<String> {
|
||||
let selection = Selection { schemas: Some(vec![kind.type_id().to_string()]), ..Selection::default() };
|
||||
let page = self.query(&selection).unwrap();
|
||||
match page.items {
|
||||
huginn_mind::PipelinePageItems::Headers(headers) => headers.into_iter().map(|header| header.id.id.0).collect(),
|
||||
huginn_mind::PipelinePageItems::Documents(views) => views.into_iter().map(|view| view.id.id.0).collect(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl MindPort for TestMind {
|
||||
fn instance(&self) -> &Slug {
|
||||
&self.instance
|
||||
}
|
||||
|
||||
fn query(&self, selection: &Selection) -> Result<PipelineSelectionPage> {
|
||||
self.mind.borrow().query(selection).map_err(|_| anyhow!("refused"))
|
||||
}
|
||||
|
||||
fn admit(&self, agent: &str, session: &str, documents: Vec<PipelineDocument>) -> Result<PipelineAdmissionOutcome> {
|
||||
self.now.set(self.now.get() + Duration::seconds(1));
|
||||
let batch = PipelineAdmissionBatch {
|
||||
instance: Slug(INSTANCE.into()),
|
||||
provenance: PipelineProvenance { faculty: Faculty::SelfFaculty, agent: agent.into(), session: session.into(), tool: "mind-body".into() },
|
||||
documents,
|
||||
};
|
||||
Ok(self.mind.borrow_mut().admit(batch, self.now.get()))
|
||||
}
|
||||
}
|
||||
|
||||
/// A mind whose admissions answer as told, over a real mind's reads.
|
||||
pub(crate) struct Answering<'a> {
|
||||
pub(crate) inner: &'a TestMind,
|
||||
pub(crate) outcome: PipelineAdmissionOutcome,
|
||||
}
|
||||
|
||||
impl MindPort for Answering<'_> {
|
||||
fn instance(&self) -> &Slug {
|
||||
self.inner.instance()
|
||||
}
|
||||
|
||||
fn query(&self, selection: &Selection) -> Result<PipelineSelectionPage> {
|
||||
self.inner.query(selection)
|
||||
}
|
||||
|
||||
fn admit(&self, _agent: &str, _session: &str, _documents: Vec<PipelineDocument>) -> Result<PipelineAdmissionOutcome> {
|
||||
Ok(self.outcome.clone())
|
||||
}
|
||||
}
|
||||
|
||||
fn doc_ref() -> DocRef {
|
||||
DocRef { path: "docs/eureka-body-target.md".into(), start_line: 1, end_line: 9, commit: Sha("df7507b".into()) }
|
||||
}
|
||||
|
||||
fn instance() -> PipelineDocument {
|
||||
PipelineDocument::Instance(PipelineInstance {
|
||||
instance: Slug(INSTANCE.into()),
|
||||
display_name: "test mind".into(),
|
||||
created_at: Date("2026-10-09".into()),
|
||||
host: "test".into(),
|
||||
})
|
||||
}
|
||||
|
||||
fn stewardship() -> PipelineDocument {
|
||||
PipelineDocument::Stewardship(PipelineStewardship {
|
||||
instance: Slug(INSTANCE.into()),
|
||||
repo: REPO.into(),
|
||||
sequence: 1,
|
||||
assigned_on: Date("2026-10-09".into()),
|
||||
note: "assigned".into(),
|
||||
})
|
||||
}
|
||||
|
||||
fn campaign() -> PipelineDocument {
|
||||
PipelineDocument::Campaign(PipelineCampaign {
|
||||
slug: Slug(CAMPAIGN.into()),
|
||||
title: "Eureka body".into(),
|
||||
repos: vec![REPO.into()],
|
||||
working_branch: "main".into(),
|
||||
target_doc: doc_ref(),
|
||||
})
|
||||
}
|
||||
|
||||
fn target() -> PipelineDocument {
|
||||
PipelineDocument::Target(PipelineTarget {
|
||||
campaign: Slug(CAMPAIGN.into()),
|
||||
revision: 1,
|
||||
invariants: vec![TargetInvariant { label: "run-is-the-grant".into(), statement: "A run is the grant.".into() }],
|
||||
not_in_scope: vec![],
|
||||
canonical_implementations: vec![],
|
||||
doc: doc_ref(),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn id(kind: &str, local: &str) -> String {
|
||||
format!("{CAMPAIGN}:{kind}:{local}")
|
||||
}
|
||||
|
||||
pub(crate) fn reference(kind: PipelineKind, id: &str) -> PipelineRef {
|
||||
PipelineRef { kind, id: id.into() }
|
||||
}
|
||||
|
||||
pub(crate) fn spec_ref(cut: &str) -> PipelineRef {
|
||||
reference(PipelineKind::CutSpec, &id("cut_spec", &format!("cut-{cut}.r1")))
|
||||
}
|
||||
|
||||
pub(crate) fn spec(cut: &str) -> PipelineDocument {
|
||||
PipelineDocument::CutSpec(PipelineCutSpec {
|
||||
campaign: Slug(CAMPAIGN.into()),
|
||||
cut: cut.into(),
|
||||
revision: 1,
|
||||
title: "A cut".into(),
|
||||
repo: REPO.into(),
|
||||
branch: "eureka-body/x".into(),
|
||||
base: Sha("df7507b".into()),
|
||||
depends_on: vec![],
|
||||
first: vec![],
|
||||
deletes: vec![],
|
||||
keeps_moves: vec![],
|
||||
adds: vec![],
|
||||
file_changes: vec![],
|
||||
authority_map: None,
|
||||
verification: CutVerification { builds: vec![], tests: vec![], negative: vec![], operator: vec![] },
|
||||
estimate: delta(),
|
||||
rulings: vec![],
|
||||
questions: vec![],
|
||||
})
|
||||
}
|
||||
|
||||
fn delta() -> StructuralDelta {
|
||||
StructuralDelta {
|
||||
lines_added: 0,
|
||||
lines_removed: 0,
|
||||
dependencies_added: vec![],
|
||||
dependencies_removed: vec![],
|
||||
formats_added: vec![],
|
||||
formats_removed: vec![],
|
||||
targets_added: vec![],
|
||||
targets_removed: vec![],
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn report(cut: &str) -> PipelineDocument {
|
||||
PipelineDocument::CutReport(PipelineCutReport {
|
||||
campaign: Slug(CAMPAIGN.into()),
|
||||
cut_spec: id("cut_spec", &format!("cut-{cut}.r1")).as_str().into(),
|
||||
attempt: 1,
|
||||
repo: REPO.into(),
|
||||
branch: "eureka-body/x".into(),
|
||||
commits: vec![eureka_pipeline::ReportCommit { sha: Sha("df7507b".into()), subject: "Land the cut".into(), builds: true }],
|
||||
range: CommitRange { base: Sha("df7507b".into()), head: Sha("df7507b".into()) },
|
||||
verification: vec![],
|
||||
mutations: vec![],
|
||||
deviations: vec![],
|
||||
forks: vec![],
|
||||
structural_delta: delta(),
|
||||
landed_names: vec![],
|
||||
undone: vec![],
|
||||
promises: vec![],
|
||||
})
|
||||
}
|
||||
|
||||
/// A question raised in `item`, which blocks it.
|
||||
pub(crate) fn question_in(label: &str, item: &PipelineRef) -> PipelineDocument {
|
||||
PipelineDocument::Question(PipelineQuestion {
|
||||
campaign: Slug(CAMPAIGN.into()),
|
||||
label: label.into(),
|
||||
title: "A fork".into(),
|
||||
question: "Which?".into(),
|
||||
options: ["A", "B"].iter().map(|name| QuestionOption { label: (*name).into(), text: "an option".into() }).collect(),
|
||||
recommended: "A".into(),
|
||||
depends: vec![],
|
||||
raised_in: Some(item.clone()),
|
||||
asked_on: Date("2026-10-09".into()),
|
||||
})
|
||||
}
|
||||
|
||||
/// A standing ruling: the plain "something else was admitted" document. With
|
||||
/// `answers` (a question's id and one of its option labels), the ruling that
|
||||
/// answers it.
|
||||
pub(crate) fn ruling(label: &str, answers: Option<(&str, &str)>) -> PipelineDocument {
|
||||
PipelineDocument::Ruling(PipelineRuling {
|
||||
campaign: Slug(CAMPAIGN.into()),
|
||||
label: label.into(),
|
||||
title: "A ruling".into(),
|
||||
answers: answers.map(|(question, _)| question.into()),
|
||||
choice: answers.map(|(_, choice)| choice.into()),
|
||||
ruling: "So.".into(),
|
||||
operator_quote: None,
|
||||
ruled_on: Date("2026-10-09".into()),
|
||||
precedents: vec![],
|
||||
authority: RulingAuthority::Standing,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn run_ref(label: &str) -> PipelineRef {
|
||||
reference(PipelineKind::Run, &format!("{INSTANCE}:run:{label}"))
|
||||
}
|
||||
|
||||
pub(crate) fn run(label: &str, turn: RunTurn, operated_by: RunOperator, claims: &[PipelineRef]) -> PipelineDocument {
|
||||
PipelineDocument::Run(PipelineRun {
|
||||
instance: Slug(INSTANCE.into()),
|
||||
label: label.into(),
|
||||
operated_by,
|
||||
turn,
|
||||
host: "test".into(),
|
||||
started_on: Date("2026-10-09".into()),
|
||||
budget_usd: "5".into(),
|
||||
claims: claims.to_vec(),
|
||||
campaigns: vec![Slug(CAMPAIGN.into())],
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn close(subject: PipelineRef, outcome: ResolutionOutcome) -> PipelineDocument {
|
||||
PipelineDocument::Resolution(PipelineResolution {
|
||||
subject,
|
||||
sequence: 1,
|
||||
outcome,
|
||||
rationale: "Closed.".into(),
|
||||
resolved_on: Date("2026-10-09".into()),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn recorded() -> ResolutionOutcome {
|
||||
ResolutionOutcome::Recorded { reason: "finished".into() }
|
||||
}
|
||||
|
||||
pub(crate) fn withdrawn() -> ResolutionOutcome {
|
||||
ResolutionOutcome::Withdrawn { reason: "never ran".into() }
|
||||
}
|
||||
|
||||
/// 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).
|
||||
pub(crate) fn ended_run(mind: &TestMind, item: &PipelineRef, n: usize, leaves_something: bool) {
|
||||
let label = format!("h{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));
|
||||
}
|
||||
mind.committed(closing);
|
||||
}
|
||||
|
||||
/// A history of ended runs on `item`, oldest first; `true` leaves a document.
|
||||
pub(crate) fn history(mind: &TestMind, item: &PipelineRef, runs: &[bool]) {
|
||||
for (n, leaves_something) in runs.iter().enumerate() {
|
||||
ended_run(mind, item, n, *leaves_something);
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn label(value: &str) -> Label {
|
||||
value.into()
|
||||
}
|
||||
|
||||
/// A follow-up sourced from `source`: work a run may claim that the queue does
|
||||
/// not list.
|
||||
pub(crate) fn follow_up(label: &str, source: &PipelineRef) -> PipelineDocument {
|
||||
PipelineDocument::FollowUp(PipelineFollowUp {
|
||||
campaign: Slug(CAMPAIGN.into()),
|
||||
label: label.into(),
|
||||
source: source.clone(),
|
||||
repo: REPO.into(),
|
||||
locations: vec![],
|
||||
item: "Later.".into(),
|
||||
why_it_can_wait: "It can.".into(),
|
||||
owner: "Hands".into(),
|
||||
})
|
||||
}
|
||||
Loading…
Reference in a new issue