From 937293d351217ab9bb3a46035bd00fbcb2604c71 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 12 Oct 2026 08:16:10 +0200 Subject: [PATCH] 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 --- crates/mind-body/Cargo.toml | 6 +- crates/mind-body/src/bin/mind-wake.rs | 76 +++++++ crates/mind-body/src/control.rs | 2 +- crates/mind-body/src/lib.rs | 3 + crates/mind-body/src/wake.rs | 117 ++++++++++ crates/mind-body/src/wake/tests.rs | 308 ++++++++++++++++++++++++++ crates/mind-body/tests/wake.rs | 51 +++++ 7 files changed, 561 insertions(+), 2 deletions(-) create mode 100644 crates/mind-body/src/bin/mind-wake.rs create mode 100644 crates/mind-body/src/wake.rs create mode 100644 crates/mind-body/src/wake/tests.rs create mode 100644 crates/mind-body/tests/wake.rs diff --git a/crates/mind-body/Cargo.toml b/crates/mind-body/Cargo.toml index 3825871..14da03e 100644 --- a/crates/mind-body/Cargo.toml +++ b/crates/mind-body/Cargo.toml @@ -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" diff --git a/crates/mind-body/src/bin/mind-wake.rs b/crates/mind-body/src/bin/mind-wake.rs new file mode 100644 index 0000000..761c086 --- /dev/null +++ b/crates/mind-body/src/bin/mind-wake.rs @@ -0,0 +1,76 @@ +//! `mind-wake --instance [--endpoint rudp://:]`: 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 [--endpoint rudp://:]"; + +/// 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)> { + 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 { + let args: Vec = 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::().ok()) + .ok_or_else(|| anyhow!("the endpoint is not rudp://:"))?; + 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); + } + } +} diff --git a/crates/mind-body/src/control.rs b/crates/mind-body/src/control.rs index 0792c44..9d0cac0 100644 --- a/crates/mind-body/src/control.rs +++ b/crates/mind-body/src/control.rs @@ -48,7 +48,7 @@ pub const DIAL_KEY: &str = "dial"; /// `//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 { - 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")) } diff --git a/crates/mind-body/src/lib.rs b/crates/mind-body/src/lib.rs index 598c6f0..4a86f31 100644 --- a/crates/mind-body/src/lib.rs +++ b/crates/mind-body/src/lib.rs @@ -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; diff --git a/crates/mind-body/src/wake.rs b/crates/mind-body/src/wake.rs new file mode 100644 index 0000000..40ceb91 --- /dev/null +++ b/crates/mind-body/src/wake.rs @@ -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 }, + /// 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>> { + 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 { + 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; diff --git a/crates/mind-body/src/wake/tests.rs b/crates/mind-body/src/wake/tests.rs new file mode 100644 index 0000000..9dc2aa3 --- /dev/null +++ b/crates/mind-body/src/wake/tests.rs @@ -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, +} + +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, + admits: Cell, +} + +impl MindPort for Counting<'_> { + fn instance(&self) -> &Slug { + self.inner.instance() + } + + fn query(&self, selection: &Selection) -> Result { + self.queries.set(self.queries.get() + 1); + self.inner.query(selection) + } + + fn admit(&self, agent: &str, session: &str, documents: Vec) -> Result { + 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 { + Err(anyhow!("the mind could not be asked")) + } + + fn admit(&self, _agent: &str, _session: &str, _documents: Vec) -> Result { + 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 { + 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 { + 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"); +} diff --git a/crates/mind-body/tests/wake.rs b/crates/mind-body/tests/wake.rs new file mode 100644 index 0000000..cbd2ddc --- /dev/null +++ b/crates/mind-body/tests/wake.rs @@ -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)); +}