mind-body: the waker, one cycle that reads the brake once, waits out the cadence since the newest Persona turn, and hands its Grant to open_and_launch
wake::cycle reads the brake and dial through the one derivation behind read_effective and the unit gate, returns Held before any query, admission or unit call, finds the newest Persona turn of hers (any status) with one descending single-row query, and launches through open_and_launch. The mind-wake binary prints the outcome. control_path no longer echoes the instance it refuses. Does not build the public_surface snapshot yet (the next commit updates it). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
94ced33419
commit
937293d351
7 changed files with 561 additions and 2 deletions
|
|
@ -11,7 +11,7 @@ publish.workspace = true
|
|||
# binary) and the one fail-closed reader every acting organ calls
|
||||
# (`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.
|
||||
# (`wake`, the `mind-wake` binary) is one cycle over them.
|
||||
#
|
||||
# `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
|
||||
|
|
@ -46,3 +46,7 @@ path = "src/lib.rs"
|
|||
[[bin]]
|
||||
name = "mind-control"
|
||||
path = "src/main.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "mind-wake"
|
||||
path = "src/bin/mind-wake.rs"
|
||||
|
|
|
|||
76
crates/mind-body/src/bin/mind-wake.rs
Normal file
76
crates/mind-body/src/bin/mind-wake.rs
Normal file
|
|
@ -0,0 +1,76 @@
|
|||
//! `mind-wake --instance <slug> [--endpoint rudp://<ip>:<port>]`: one waker
|
||||
//! cycle for one Mind. All behaviour is `mind_body::wake::cycle`; this reads the
|
||||
//! process's arguments, environment and clock, the things the library never
|
||||
//! reads, and prints the outcome. It exits 0 for every outcome and 2 when the
|
||||
//! mind cannot be reached or the arguments are wrong. Without `--endpoint` the
|
||||
//! daemon is `HUGINN_ENDPOINT`.
|
||||
|
||||
use std::net::SocketAddr;
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Result, anyhow, bail};
|
||||
use eureka_pipeline::Slug;
|
||||
use eureka_state::HuginnClient;
|
||||
use mind_body::control::{CONTROL_ROOT, FileSource, control_path};
|
||||
use mind_body::launch::{Ports, SystemClock, SystemdLauncher};
|
||||
use mind_body::queue::HuginnMind;
|
||||
use mind_body::wake::cycle;
|
||||
|
||||
/// How long one call to the mind may take.
|
||||
const TIMEOUT: Duration = Duration::from_secs(15);
|
||||
|
||||
const USAGE: &str = "usage: mind-wake --instance <slug> [--endpoint rudp://<ip>:<port>]";
|
||||
|
||||
/// The instance and the endpoint as given. A flag given twice or an unknown
|
||||
/// argument is refused without printing it.
|
||||
fn parse(args: &[String]) -> Result<(Slug, Option<String>)> {
|
||||
let (mut instance, mut endpoint) = (None, None);
|
||||
let mut rest = args.iter();
|
||||
while let Some(flag) = rest.next() {
|
||||
let slot = match flag.as_str() {
|
||||
"--instance" => &mut instance,
|
||||
"--endpoint" => &mut endpoint,
|
||||
_ => bail!("{USAGE}"),
|
||||
};
|
||||
let value = rest.next().ok_or_else(|| anyhow!("{USAGE}"))?;
|
||||
if slot.replace(value.clone()).is_some() {
|
||||
bail!("{USAGE}");
|
||||
}
|
||||
}
|
||||
Ok((Slug(instance.ok_or_else(|| anyhow!("{USAGE}"))?), endpoint))
|
||||
}
|
||||
|
||||
/// The host a run records: the kernel's name for this machine.
|
||||
fn host() -> String {
|
||||
std::fs::read_to_string("/proc/sys/kernel/hostname").map(|name| name.trim().to_string()).ok().filter(|name| !name.is_empty()).unwrap_or_else(|| "unknown".into())
|
||||
}
|
||||
|
||||
// The binary is the one place that reads the process: its arguments and the
|
||||
// daemon's address. The library reads neither (clippy.toml, `#![forbid]` in lib.rs).
|
||||
#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and the daemon's address, and hands them to the library")]
|
||||
fn run() -> Result<String> {
|
||||
let args: Vec<String> = std::env::args().skip(1).collect();
|
||||
let (instance, endpoint) = parse(&args)?;
|
||||
let source = FileSource::new(control_path(Path::new(CONTROL_ROOT), &instance)?);
|
||||
let endpoint = endpoint.or_else(|| std::env::var("HUGINN_ENDPOINT").ok()).ok_or_else(|| anyhow!("no daemon endpoint: --endpoint or HUGINN_ENDPOINT"))?;
|
||||
let address = endpoint
|
||||
.strip_prefix("rudp://")
|
||||
.and_then(|address| address.parse::<SocketAddr>().ok())
|
||||
.ok_or_else(|| anyhow!("the endpoint is not rudp://<ip>:<port>"))?;
|
||||
let mind = HuginnMind::new(HuginnClient::new(address, instance, TIMEOUT));
|
||||
let launcher = SystemdLauncher::new();
|
||||
let host = host();
|
||||
let ports = Ports { mind: &mind, launcher: &launcher, clock: &SystemClock, host: &host };
|
||||
Ok(cycle(&ports, &source)?.to_string())
|
||||
}
|
||||
|
||||
fn main() {
|
||||
match run() {
|
||||
Ok(outcome) => println!("{outcome}"),
|
||||
Err(error) => {
|
||||
eprintln!("mind-wake: {error:#}");
|
||||
std::process::exit(2);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -48,7 +48,7 @@ pub const DIAL_KEY: &str = "dial";
|
|||
/// `<root>/<instance>/control.cc`, after the leaf's slug grammar has passed, so
|
||||
/// an instance name cannot walk out of the root.
|
||||
pub fn control_path(root: &Path, instance: &Slug) -> Result<PathBuf> {
|
||||
instance.validate_slug().map_err(|refusal| anyhow::anyhow!("instance {:?} is not a slug: {refusal:?}", instance.0))?;
|
||||
instance.validate_slug().map_err(|_| anyhow::anyhow!("the instance is not a slug"))?;
|
||||
Ok(root.join(&instance.0).join("control.cc"))
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,8 @@
|
|||
//! `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.
|
||||
//!
|
||||
//! `wake` is the waker's one cycle over them: the `mind-wake` binary.
|
||||
|
||||
// The control root is a constant, so the library reads no part of the process
|
||||
// environment (clippy.toml lists the readers); `forbid` keeps an `allow` from
|
||||
|
|
@ -21,6 +23,7 @@ pub mod cli;
|
|||
pub mod control;
|
||||
pub mod launch;
|
||||
pub mod queue;
|
||||
pub mod wake;
|
||||
|
||||
#[cfg(test)]
|
||||
mod testkit;
|
||||
|
|
|
|||
117
crates/mind-body/src/wake.rs
Normal file
117
crates/mind-body/src/wake.rs
Normal file
|
|
@ -0,0 +1,117 @@
|
|||
//! The waker: one cycle, no memory between cycles.
|
||||
//!
|
||||
//! A cycle is three steps in a fixed order. (1) The brake and the dial are read
|
||||
//! once, through `control::derive`, the derivation behind `read_effective` and
|
||||
//! the unit gate; held ends the cycle before the mind, the clock or a unit is
|
||||
//! touched. (2) The newest Persona turn of hers is found, whatever became of
|
||||
//! it, and her cadence (the Grant's, from the dial) is waited out since its
|
||||
//! admission. (3) `open_and_launch` is called once for a Persona turn with no
|
||||
//! claims and the Grant, and its answer is the cycle's.
|
||||
//!
|
||||
//! The waker owns no file, no timer state and no cursor: the run in the mind is
|
||||
//! the grant (target invariant run-is-the-grant), and whether she is awake is
|
||||
//! decided by admission and answered by the launch primitive. The timer that
|
||||
//! calls a cycle is a fixed poll; it decides nothing. The waker starts no Self
|
||||
//! run and ends no run or unit.
|
||||
|
||||
use std::fmt;
|
||||
|
||||
use anyhow::{Result, anyhow, bail};
|
||||
use chrono::{DateTime, Duration, Utc};
|
||||
use cultnet_rs::Selection;
|
||||
use eureka_pipeline::{PipelineKind, PipelineRef, RunOperator, RunTurn, Slug};
|
||||
use huginn_mind::PipelinePageItems;
|
||||
|
||||
use crate::control::{ControlSource, Effective, HeldReason, derive};
|
||||
use crate::launch::{Declined, LaunchOutcome, LaunchRequest, Ports, open_and_launch};
|
||||
use crate::queue::{MindPort, any_of, of_kind};
|
||||
|
||||
/// The admitting agent's name on the receipt of every run the waker opens.
|
||||
pub const AGENT: &str = "mind-wake";
|
||||
|
||||
/// What one cycle did. Every outcome is an answer; only a mind that cannot be
|
||||
/// reached is an error.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum WakeOutcome {
|
||||
/// The brake or the dial did not allow a launch; nothing else was touched.
|
||||
Held { reason: HeldReason },
|
||||
/// Her cadence since the newest Persona turn has not passed; `until` is when it has.
|
||||
NotYet { until: DateTime<Utc> },
|
||||
/// A Persona turn of hers is live.
|
||||
Busy { run: PipelineRef },
|
||||
Launched { run: PipelineRef },
|
||||
Refused(Declined),
|
||||
}
|
||||
|
||||
impl fmt::Display for WakeOutcome {
|
||||
/// One line, with ids and times the mind gave and never a value a refusal
|
||||
/// carries.
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::Held { reason } => write!(f, "held: {reason:?}"),
|
||||
Self::NotYet { until } => write!(f, "not yet: until {}", until.to_rfc3339()),
|
||||
Self::Busy { run } => write!(f, "busy: {}", run.id.0),
|
||||
Self::Launched { run } => write!(f, "launched: {}", run.id.0),
|
||||
Self::Refused(declined) => write!(f, "refused: {}", declined_code(declined)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn declined_code(declined: &Declined) -> &'static str {
|
||||
match declined {
|
||||
Declined::Admission(_) => "admission",
|
||||
Declined::Conflict => "conflict",
|
||||
Declined::AlreadyAdmitted => "already_admitted",
|
||||
Declined::StartFailed => "start_failed",
|
||||
Declined::GrantForOtherInstance => "grant_for_other_instance",
|
||||
Declined::PersonaRunTakesNoClaims => "persona_run_takes_claims",
|
||||
}
|
||||
}
|
||||
|
||||
/// When the newest Persona turn of hers, of any status, was admitted. The
|
||||
/// spellings of the two filters are the types' own, and one page of one row is
|
||||
/// asked for: an idle wake is a query.
|
||||
fn newest_persona_turn(mind: &dyn MindPort, instance: &Slug) -> Result<Option<DateTime<Utc>>> {
|
||||
let turn = format!("{:?}", RunTurn::PersonaTurn);
|
||||
let operator = format!("{:?}", RunOperator::Mind);
|
||||
let runs = any_of(of_kind(PipelineKind::Run), "root", &[&instance.0]);
|
||||
let selection = Selection { descending: true, limit: Some(1), ..any_of(any_of(runs, "turn", &[&turn]), "operated_by", &[&operator]) };
|
||||
let PipelinePageItems::Headers(headers) = mind.query(&selection)?.items else {
|
||||
bail!("the mind answered documents to a header selection");
|
||||
};
|
||||
headers
|
||||
.first()
|
||||
.map(|newest| {
|
||||
DateTime::parse_from_rfc3339(&newest.admission.admitted_at)
|
||||
.map(|at| at.with_timezone(&Utc))
|
||||
.map_err(|_| anyhow!("the newest Persona turn's admission time is not a timestamp"))
|
||||
})
|
||||
.transpose()
|
||||
}
|
||||
|
||||
/// One cycle. `control` is the instance's control store; the reading is taken
|
||||
/// once and its Grant is the one handed to the launch.
|
||||
pub fn cycle(ports: &Ports, control: &impl ControlSource) -> Result<WakeOutcome> {
|
||||
let instance = ports.mind.instance().clone();
|
||||
let grant = match derive(control, &instance) {
|
||||
Effective::Held { reason } => return Ok(WakeOutcome::Held { reason }),
|
||||
Effective::Released(grant) => grant,
|
||||
};
|
||||
if let Some(newest) = newest_persona_turn(ports.mind, &instance)? {
|
||||
let cadence = Duration::from_std(grant.cadence()).map_err(|_| anyhow!("the dial's cadence is out of range"))?;
|
||||
let until = newest + cadence;
|
||||
if until > ports.clock.now() {
|
||||
return Ok(WakeOutcome::NotYet { until });
|
||||
}
|
||||
}
|
||||
let request = LaunchRequest { turn: RunTurn::PersonaTurn, claims: vec![], agent: AGENT.into(), grant };
|
||||
match open_and_launch(ports, request)? {
|
||||
LaunchOutcome::Launched { run } => Ok(WakeOutcome::Launched { run }),
|
||||
LaunchOutcome::Busy { run } => Ok(WakeOutcome::Busy { run }),
|
||||
LaunchOutcome::Refused(declined) => Ok(WakeOutcome::Refused(declined)),
|
||||
LaunchOutcome::NotQueued { .. } | LaunchOutcome::Tripped { .. } => bail!("the launch answered a Persona turn with an answer only a Self run has"),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
308
crates/mind-body/src/wake/tests.rs
Normal file
308
crates/mind-body/src/wake/tests.rs
Normal file
|
|
@ -0,0 +1,308 @@
|
|||
//! The waker's tests: a real mind over a temporary directory, the real control
|
||||
//! store written by the operator's writer and read by the real derivation, a
|
||||
//! recording launcher and a settable clock. A wait is observed where it is
|
||||
//! decided, as the `until` the cycle returns, never after a sleep.
|
||||
|
||||
use std::cell::Cell;
|
||||
use std::str::FromStr;
|
||||
|
||||
use anyhow::{Result, anyhow};
|
||||
use chrono::{DateTime, Duration, Utc};
|
||||
use cultnet_rs::Selection;
|
||||
use eureka_pipeline::{PipelineDocument, PipelineKind as K, ResolutionOutcome, RunOperator, RunTurn, Slug};
|
||||
use huginn_mind::{MindRefusal, PipelineAdmissionOutcome, PipelineSelectionPage};
|
||||
use rust_decimal::Decimal;
|
||||
use tempfile::TempDir;
|
||||
|
||||
use super::*;
|
||||
use crate::control::{BurnRate, ControlWriter, FileSource};
|
||||
use crate::launch::Launcher;
|
||||
use crate::testkit::*;
|
||||
|
||||
/// Every call the units port receives, counted over a recording launcher.
|
||||
struct Units {
|
||||
rec: Recording,
|
||||
calls: Cell<usize>,
|
||||
}
|
||||
|
||||
impl Launcher for Units {
|
||||
fn start(&self, instance: &Slug, turn: RunTurn) -> Result<()> {
|
||||
self.calls.set(self.calls.get() + 1);
|
||||
self.rec.start(instance, turn)
|
||||
}
|
||||
|
||||
fn alive(&self, instance: &Slug, turn: RunTurn) -> bool {
|
||||
self.calls.set(self.calls.get() + 1);
|
||||
self.rec.alive(instance, turn)
|
||||
}
|
||||
}
|
||||
|
||||
/// A mind that counts the queries and admissions it is asked for.
|
||||
struct Counting<'a> {
|
||||
inner: &'a TestMind,
|
||||
queries: Cell<usize>,
|
||||
admits: Cell<usize>,
|
||||
}
|
||||
|
||||
impl MindPort for Counting<'_> {
|
||||
fn instance(&self) -> &Slug {
|
||||
self.inner.instance()
|
||||
}
|
||||
|
||||
fn query(&self, selection: &Selection) -> Result<PipelineSelectionPage> {
|
||||
self.queries.set(self.queries.get() + 1);
|
||||
self.inner.query(selection)
|
||||
}
|
||||
|
||||
fn admit(&self, agent: &str, session: &str, documents: Vec<PipelineDocument>) -> Result<PipelineAdmissionOutcome> {
|
||||
self.admits.set(self.admits.get() + 1);
|
||||
self.inner.admit(agent, session, documents)
|
||||
}
|
||||
}
|
||||
|
||||
/// A mind that cannot be reached.
|
||||
struct Unreachable(Slug);
|
||||
|
||||
impl MindPort for Unreachable {
|
||||
fn instance(&self) -> &Slug {
|
||||
&self.0
|
||||
}
|
||||
|
||||
fn query(&self, _selection: &Selection) -> Result<PipelineSelectionPage> {
|
||||
Err(anyhow!("the mind could not be asked"))
|
||||
}
|
||||
|
||||
fn admit(&self, _agent: &str, _session: &str, _documents: Vec<PipelineDocument>) -> Result<PipelineAdmissionOutcome> {
|
||||
Err(anyhow!("the mind could not be asked"))
|
||||
}
|
||||
}
|
||||
|
||||
struct Rig {
|
||||
mind: TestMind,
|
||||
launcher: Units,
|
||||
clock: Fixed,
|
||||
dir: TempDir,
|
||||
}
|
||||
|
||||
fn dec(text: &str) -> Decimal {
|
||||
Decimal::from_str(text).unwrap()
|
||||
}
|
||||
|
||||
fn seconds(whole: i64, nanos: i64) -> Duration {
|
||||
Duration::seconds(whole) + Duration::nanoseconds(nanos)
|
||||
}
|
||||
|
||||
impl Rig {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
mind: TestMind::seeded(),
|
||||
launcher: Units { rec: Recording::default(), calls: Cell::new(0) },
|
||||
clock: Fixed(Cell::new(at(0))),
|
||||
dir: TempDir::new().unwrap(),
|
||||
}
|
||||
}
|
||||
|
||||
fn store(&self) -> std::path::PathBuf {
|
||||
self.dir.path().join(INSTANCE).join("control.cc")
|
||||
}
|
||||
|
||||
/// The brake released and the dial set, through the operator's writer.
|
||||
fn release(&self, heat: &str, cooldown_s: u32, run_usd: &str) {
|
||||
let mut writer = ControlWriter::open(self.store()).unwrap();
|
||||
writer.set_brake(true, at(0), "op").unwrap();
|
||||
writer.set_dial(BurnRate { heat: dec(heat), base_cooldown_s: cooldown_s, base_run_usd: dec(run_usd), set_at: at(0), set_by: "op".into() }).unwrap();
|
||||
}
|
||||
|
||||
fn cycle(&self) -> WakeOutcome {
|
||||
self.cycle_through(&self.mind).unwrap()
|
||||
}
|
||||
|
||||
fn cycle_through(&self, mind: &dyn MindPort) -> Result<WakeOutcome> {
|
||||
let ports = Ports { mind, launcher: &self.launcher, clock: &self.clock, host: "yggdrasil-host" };
|
||||
super::cycle(&ports, &FileSource::new(self.store()))
|
||||
}
|
||||
|
||||
/// A Persona turn of hers, opened and then ended as given.
|
||||
fn persona(&self, label: &str, ending: ResolutionOutcome) {
|
||||
self.mind.committed(vec![run(label, RunTurn::PersonaTurn, RunOperator::Mind, &[])]);
|
||||
self.mind.committed(vec![close(run_ref(label), ending)]);
|
||||
}
|
||||
|
||||
fn admitted(&self, label: &str) -> DateTime<Utc> {
|
||||
let stamp = self.mind.view(K::Run, &run_ref(label).id.0).unwrap().admission.admitted_at;
|
||||
DateTime::parse_from_rfc3339(&stamp).unwrap().with_timezone(&Utc)
|
||||
}
|
||||
|
||||
fn runs(&self) -> usize {
|
||||
self.mind.ids(K::Run).len()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn with_the_brake_held_or_the_dial_absent_a_cycle_queries_nothing_admits_nothing_and_calls_no_unit() {
|
||||
let cases: [(&str, HeldReason, fn(&Rig)); 4] = [
|
||||
("no store", HeldReason::BrakeAbsent, |_| {}),
|
||||
("brake held", HeldReason::BrakeHeld, |rig| {
|
||||
rig.release("1", 600, "5");
|
||||
ControlWriter::open(rig.store()).unwrap().set_brake(false, at(1), "op").unwrap();
|
||||
}),
|
||||
("dial absent", HeldReason::DialAbsent, |rig| {
|
||||
ControlWriter::open(rig.store()).unwrap().set_brake(true, at(0), "op").unwrap();
|
||||
}),
|
||||
("store undecodable", HeldReason::Undecodable, |rig| {
|
||||
rig.release("1", 600, "5");
|
||||
std::fs::write(rig.store(), b"junk").unwrap();
|
||||
}),
|
||||
];
|
||||
for (name, reason, prepare) in cases {
|
||||
// No Persona turn yet, so a released brake would launch one.
|
||||
let rig = Rig::new();
|
||||
prepare(&rig);
|
||||
let counting = Counting { inner: &rig.mind, queries: Cell::new(0), admits: Cell::new(0) };
|
||||
let outcome = rig.cycle_through(&counting).unwrap();
|
||||
assert_eq!(outcome, WakeOutcome::Held { reason }, "{name}");
|
||||
assert_eq!((counting.queries.get(), counting.admits.get(), rig.launcher.calls.get(), rig.runs()), (0, 0, 0, 0), "{name}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_wait_is_the_grants_cadence_after_the_newest_persona_turn_and_ends_exactly_there() {
|
||||
// (heat, base cooldown, the cadence base_cooldown / heat): whole, halved,
|
||||
// and one that carries nanoseconds, so a clamp, offset or scale shows.
|
||||
let heats = [("1", 600, seconds(600, 0)), ("2", 600, seconds(300, 0)), ("1.5", 1000, seconds(666, 666_666_667)), ("0.05", 3600, seconds(72_000, 0))];
|
||||
for (heat, cooldown, cadence) in heats {
|
||||
let rig = Rig::new();
|
||||
rig.release(heat, cooldown, "5");
|
||||
rig.persona("p1", recorded());
|
||||
let until = rig.admitted("p1") + cadence;
|
||||
|
||||
rig.clock.0.set(until - Duration::milliseconds(1));
|
||||
assert_eq!(rig.cycle(), WakeOutcome::NotYet { until }, "heat {heat}, a millisecond early");
|
||||
assert_eq!((rig.launcher.calls.get(), rig.runs()), (0, 1), "heat {heat}: waiting touches no unit and admits nothing");
|
||||
|
||||
rig.clock.0.set(until);
|
||||
assert!(matches!(rig.cycle(), WakeOutcome::Launched { .. }), "heat {heat}, at the boundary");
|
||||
assert_eq!(rig.runs(), 2, "heat {heat}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_same_age_waits_under_one_dial_and_launches_under_another() {
|
||||
let age = |heat: &str| {
|
||||
let rig = Rig::new();
|
||||
rig.release(heat, 600, "5");
|
||||
rig.persona("p1", recorded());
|
||||
rig.clock.0.set(rig.admitted("p1") + Duration::seconds(400));
|
||||
rig.cycle()
|
||||
};
|
||||
assert!(matches!(age("1"), WakeOutcome::NotYet { .. }), "600 s cadence at 400 s");
|
||||
assert!(matches!(age("2"), WakeOutcome::Launched { .. }), "300 s cadence at 400 s");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_newer_self_run_or_operator_turn_does_not_move_the_boundary() {
|
||||
let rig = Rig::new();
|
||||
rig.release("1", 600, "5");
|
||||
rig.persona("p1", recorded());
|
||||
let until = rig.admitted("p1") + Duration::seconds(600);
|
||||
rig.mind.committed(vec![run("s1", RunTurn::SelfRun, RunOperator::Mind, &[])]);
|
||||
rig.mind.committed(vec![run("op1", RunTurn::PersonaTurn, RunOperator::Operator, &[])]);
|
||||
assert!(rig.admitted("s1") > rig.admitted("p1") && rig.admitted("op1") > rig.admitted("p1"));
|
||||
|
||||
rig.clock.0.set(until - Duration::milliseconds(1));
|
||||
assert_eq!(rig.cycle(), WakeOutcome::NotYet { until });
|
||||
rig.clock.0.set(until);
|
||||
assert!(matches!(rig.cycle(), WakeOutcome::Launched { .. }), "the boundary is still her Persona turn's");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_ended_persona_turn_sets_the_boundary_whatever_ended_it_and_the_newest_wins() {
|
||||
for (name, ending) in [("recorded", recorded()), ("withdrawn", withdrawn())] {
|
||||
let rig = Rig::new();
|
||||
rig.release("1", 600, "5");
|
||||
rig.persona("p1", recorded());
|
||||
rig.persona("p2", ending);
|
||||
assert!(rig.admitted("p2") > rig.admitted("p1"), "{name}");
|
||||
let until = rig.admitted("p2") + Duration::seconds(600);
|
||||
rig.clock.0.set(until - Duration::milliseconds(1));
|
||||
assert_eq!(rig.cycle(), WakeOutcome::NotYet { until }, "{name}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_live_persona_turn_past_the_cadence_answers_busy_and_the_cycle_admits_nothing_itself() {
|
||||
let rig = Rig::new();
|
||||
rig.release("1", 600, "5");
|
||||
rig.mind.committed(vec![run("p1", RunTurn::PersonaTurn, RunOperator::Mind, &[])]);
|
||||
rig.launcher.rec.revive(RunTurn::PersonaTurn);
|
||||
rig.clock.0.set(rig.admitted("p1") + Duration::seconds(601));
|
||||
let counting = Counting { inner: &rig.mind, queries: Cell::new(0), admits: Cell::new(0) };
|
||||
assert_eq!(rig.cycle_through(&counting).unwrap(), WakeOutcome::Busy { run: run_ref("p1") });
|
||||
assert_eq!(counting.admits.get(), 0);
|
||||
assert!(rig.launcher.rec.started.borrow().is_empty());
|
||||
assert_eq!(rig.runs(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_first_released_cycle_launches_a_persona_run_with_no_claims_and_the_grants_cap() {
|
||||
let rig = Rig::new();
|
||||
// Decoys: a Self run of hers and an operator's Persona turn are not her Persona turns.
|
||||
rig.mind.committed(vec![run("s1", RunTurn::SelfRun, RunOperator::Mind, &[])]);
|
||||
rig.mind.committed(vec![run("op1", RunTurn::PersonaTurn, RunOperator::Operator, &[])]);
|
||||
rig.release("2", 600, "3.25");
|
||||
let WakeOutcome::Launched { run: opened } = rig.cycle() else { panic!("not launched") };
|
||||
let view = rig.mind.view(K::Run, &opened.id.0).unwrap();
|
||||
let PipelineDocument::Run(held) = &view.document else { panic!("not a run") };
|
||||
assert_eq!((held.operated_by, held.turn), (RunOperator::Mind, RunTurn::PersonaTurn));
|
||||
assert!(held.claims.is_empty() && held.campaigns.is_empty());
|
||||
assert_eq!(held.budget_usd.0, "6.5", "base_run_usd x heat");
|
||||
assert_eq!(held.host.0, "yggdrasil-host");
|
||||
assert_eq!(view.admission.provenance.agent.0, AGENT);
|
||||
assert_eq!(*rig.launcher.rec.started.borrow(), vec![(Slug(INSTANCE.into()), RunTurn::PersonaTurn)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn two_consecutive_cycles_launch_once_and_the_second_waits_or_answers_busy() {
|
||||
let rig = Rig::new();
|
||||
rig.release("1", 600, "5");
|
||||
let WakeOutcome::Launched { run: first } = rig.cycle() else { panic!("not launched") };
|
||||
let stamp = rig.mind.view(K::Run, &first.id.0).unwrap().admission.admitted_at;
|
||||
let until = DateTime::parse_from_rfc3339(&stamp).unwrap().with_timezone(&Utc) + Duration::seconds(600);
|
||||
assert_eq!(rig.cycle(), WakeOutcome::NotYet { until }, "straight after: within the cadence");
|
||||
rig.clock.0.set(until);
|
||||
assert_eq!(rig.cycle(), WakeOutcome::Busy { run: first }, "past the cadence with her turn still live");
|
||||
assert_eq!((rig.runs(), rig.launcher.rec.started.borrow().len()), (1, 1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_unit_that_does_not_start_and_an_admission_that_is_refused_are_refusals_not_errors() {
|
||||
let rig = Rig::new();
|
||||
rig.release("1", 600, "5");
|
||||
rig.launcher.rec.fail.set(true);
|
||||
assert_eq!(rig.cycle(), WakeOutcome::Refused(Declined::StartFailed));
|
||||
|
||||
let rig = Rig::new();
|
||||
rig.release("1", 600, "5");
|
||||
let refusing = Answering { inner: &rig.mind, outcome: PipelineAdmissionOutcome::Conflict { identities: vec![] } };
|
||||
assert_eq!(rig.cycle_through(&refusing).unwrap(), WakeOutcome::Refused(Declined::Conflict));
|
||||
assert!(rig.launcher.rec.started.borrow().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_mind_that_cannot_be_reached_is_an_error_and_starts_nothing() {
|
||||
let rig = Rig::new();
|
||||
rig.release("1", 600, "5");
|
||||
assert!(rig.cycle_through(&Unreachable(Slug(INSTANCE.into()))).is_err());
|
||||
assert_eq!(rig.launcher.calls.get(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_outcome_prints_ids_and_codes_and_never_a_refusals_values() {
|
||||
let canary = "canary-8c1f";
|
||||
let refused = Declined::Admission(MindRefusal::RepoNotInCampaign { repo: canary.into() });
|
||||
assert_eq!(WakeOutcome::Refused(refused).to_string(), "refused: admission");
|
||||
assert_eq!(WakeOutcome::Held { reason: HeldReason::BrakeHeld }.to_string(), "held: BrakeHeld");
|
||||
assert_eq!(WakeOutcome::Launched { run: run_ref("p1") }.to_string(), format!("launched: {INSTANCE}:run:p1"));
|
||||
assert_eq!(WakeOutcome::Busy { run: run_ref("p1") }.to_string(), format!("busy: {INSTANCE}:run:p1"));
|
||||
assert_eq!(WakeOutcome::NotYet { until: at(0) }.to_string(), "not yet: until 2026-10-09T13:15:00+00:00");
|
||||
}
|
||||
51
crates/mind-body/tests/wake.rs
Normal file
51
crates/mind-body/tests/wake.rs
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
//! The `mind-wake` binary as a process: its arguments and its exit codes. The
|
||||
//! brake's root is a constant, so a process test can only ever see the absent
|
||||
//! store of an instance no host has; the cycle's behaviour is tested in the
|
||||
//! library against a real mind and a real store.
|
||||
|
||||
use std::process::{Command, Output};
|
||||
|
||||
const NO_SUCH_INSTANCE: &str = "wake-test-no-such-instance";
|
||||
|
||||
fn wake(args: &[&str]) -> Output {
|
||||
Command::new(env!("CARGO_BIN_EXE_mind-wake")).args(args).env_remove("HUGINN_ENDPOINT").output().unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn without_an_instance_or_with_a_stray_argument_it_exits_2_with_the_usage_and_echoes_nothing() {
|
||||
let canary = "canary-argument-3f9a";
|
||||
for args in [vec![], vec!["--endpoint", "rudp://127.0.0.1:9"], vec!["--instance", NO_SUCH_INSTANCE, canary], vec!["--instance", NO_SUCH_INSTANCE, "--instance", canary]] {
|
||||
let output = wake(&args);
|
||||
assert_eq!(output.status.code(), Some(2), "{args:?}");
|
||||
let stderr = String::from_utf8_lossy(&output.stderr);
|
||||
assert!(stderr.contains("usage: mind-wake"), "{args:?}: {stderr}");
|
||||
assert!(!stderr.contains(canary) && output.stdout.is_empty(), "{args:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_instance_that_is_not_a_slug_is_exit_2_and_is_not_echoed() {
|
||||
let canary = "Canary/../Instance-5e2b";
|
||||
let output = wake(&["--instance", canary, "--endpoint", "rudp://127.0.0.1:9"]);
|
||||
assert_eq!(output.status.code(), Some(2));
|
||||
assert!(!String::from_utf8_lossy(&output.stderr).contains("anary"), "{}", String::from_utf8_lossy(&output.stderr));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_held_brake_ends_the_cycle_with_exit_0_before_the_mind_is_contacted() {
|
||||
// The endpoint answers no one (port 9 on loopback), so a cycle that
|
||||
// reached the mind would exit 2; the absent store holds the cycle first.
|
||||
let output = wake(&["--instance", NO_SUCH_INSTANCE, "--endpoint", "rudp://127.0.0.1:9"]);
|
||||
assert_eq!(output.status.code(), Some(0), "{}", String::from_utf8_lossy(&output.stderr));
|
||||
assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "held: BrakeAbsent");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_malformed_endpoint_or_none_is_exit_2_and_the_endpoint_is_not_echoed() {
|
||||
let canary = "canary-endpoint-77c1";
|
||||
let bad = wake(&["--instance", NO_SUCH_INSTANCE, "--endpoint", canary]);
|
||||
assert_eq!(bad.status.code(), Some(2));
|
||||
assert!(!String::from_utf8_lossy(&bad.stderr).contains(canary));
|
||||
let none = wake(&["--instance", NO_SUCH_INSTANCE]);
|
||||
assert_eq!(none.status.code(), Some(2));
|
||||
}
|
||||
Loading…
Reference in a new issue