From 21fb9b2b04c3e28b04453d766f2bb927fc503deb Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 05:02:20 +0100 Subject: [PATCH 001/265] mind-self-run: every exit closes her Run once with the result's excerpt; the child sees EUREKA_INSTANCE and HUGINN_ENDPOINT for the huginn CLI Squash of eureka-body/self-run-closes-run 55dd620..93f93c1 (Soul go; orphan sweep cut: the mind admits one Self run in force and launch's close_dead closes a dead holder). Operator: 'Look into the stale runs issue' and 'why has this xenos filth not been replaced with nice clean CLI yet'. Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/self_run.rs | 117 ++++++++++++----- crates/mind-body/src/self_run/tests.rs | 27 ++-- .../mind-body/src/self_run/tests/closure.rs | 121 ++++++++++++++++++ crates/mind-body/src/self_run/tests/lease.rs | 7 +- .../src/self_run/tests/process/lease_child.rs | 4 +- 5 files changed, 226 insertions(+), 50 deletions(-) create mode 100644 crates/mind-body/src/self_run/tests/closure.rs diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 3d65da2..ca743b3 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -28,9 +28,11 @@ //! stream reported; a run that ends without settling (killed by the brake, a //! crash) is billed its reservation by the connector. //! -//! The actuator admits only the closure of a run whose lease was refused. It -//! reads neither the brake nor the environment and stops no unit: the gate and -//! the brake hold own those. It never echoes an argument, an endpoint, a +//! The actuator admits only closures of Self runs, nothing else. Every exit of +//! `carry` that leaves her run open resolves it Recorded exactly once (the +//! child's result, or that it gave none, in the reason). It reads neither the +//! brake nor the environment and stops no unit: the gate and the brake hold own those. +//! It never echoes an argument, an endpoint, a //! connector's refusal text or a stream line. Stream-json is Claude Code's wire, a //! foreign owner's JSON: it is parsed into `serde_json::Value` here and nothing //! parsed is stored. @@ -285,7 +287,9 @@ impl fmt::Display for SelfRunOutcome { } /// Her one open Self run and its budget, read through the launch primitive's -/// holder reader. Anything but exactly one run is none. +/// holder reader. Anything but exactly one run is none: the mind admits at most +/// one Self run of hers in force, and the launch that opens the next closes a +/// dead one. fn her_run(mind: &dyn MindPort) -> Result> { let mut held = holders(mind, mind.instance(), RunTurn::SelfRun)?; let (Some(run), None) = (held.pop(), held.pop()) else { return Ok(None) }; @@ -310,6 +314,8 @@ struct Spent { reported: Option>, /// The message ids heard when `reported` was stated: its total already holds them. counted: BTreeSet, + /// The last result heard, as the reason of the run's closure: bounded. + concluded: Option, } /// The model of a message that names none; the connector prices it as it sees fit. @@ -320,7 +326,10 @@ impl Spent { let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return }; match event.get("type").and_then(Value::as_str) { Some("assistant") => {} - Some("result") => return self.hear_result(&event), + Some("result") => { + self.concluded = Some(describe_result(&event, REASON_EXCERPT_LIMIT)); + return self.hear_result(&event); + } _ => return, } let message = event.get("message"); @@ -385,25 +394,44 @@ fn unix_ms(at: DateTime) -> u64 { /// run that cannot be paid for counts as an empty run and trips the breaker. fn close_refused(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef, kind: LeaseRefusalKind) -> Result { let reason = format!("the connector refused the credential lease: {kind:?}"); - let document = closure(mind, run, clock.now(), ResolutionOutcome::Recorded { reason: reason.as_str().into() }, "The Self run's credential lease was refused, so no claude was started.")?; + admit_closure(mind, clock, run, ResolutionOutcome::Recorded { reason: reason.as_str().into() }, "The Self run's credential lease was refused, so no claude was started.")?; + Ok(SelfRunOutcome::LeaseRefused { kind }) +} + +/// Admits the closing resolution of `run`: the one writer of a closure. +fn admit_closure(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef, outcome: ResolutionOutcome, rationale: &str) -> Result<()> { + let document = closure(mind, run, clock.now(), outcome, rationale)?; let label = run.id.0.rsplit(':').next().unwrap_or(&run.id.0); match mind.admit(AGENT, label, vec![document])? { - PipelineAdmissionOutcome::Committed { .. } => Ok(SelfRunOutcome::LeaseRefused { kind }), - _ => Err(anyhow!("the run whose lease was refused could not be closed")), + PipelineAdmissionOutcome::Committed { .. } => Ok(()), + _ => Err(anyhow!("the run could not be closed")), } } +/// Resolves `run` Recorded if it is still in force: the child may have closed it +/// itself in its flush, and a refused lease was closed already. One try; a mind +/// that cannot be reached leaves it open: the launch that opens the next run closes it. +fn close_if_open(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef, ran: &Result, spent: &Spent) -> Result<()> { + if !holders(mind, mind.instance(), RunTurn::SelfRun)?.iter().any(|held| held.id == *run) { + return Ok(()); + } + let how = match ran { + Ok(outcome) => outcome.to_string(), + Err(_) => "the actuator failed after the run opened".to_string(), + }; + let reason = match &spent.concluded { + Some(concluded) => format!("{how}; {concluded}"), + None => format!("{how}; claude gave no result"), + }; + admit_closure(mind, clock, run, ResolutionOutcome::Recorded { reason: reason.as_str().into() }, "The Self run's carrier closed the run its claude left open.") +} + /// Closes a run whose lease was refused for its length: Withdrawn, because no /// claude ever ran and a misconfiguration is not an empty run; the breaker /// counts nothing and a launch that finds the unit dead has no open run to close. fn withdraw_misconfigured(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef) -> Result<()> { let outcome = ResolutionOutcome::Withdrawn { reason: "the connector refused the lease as too long".into() }; - let document = closure(mind, run, clock.now(), outcome, "The unit's runtime cap exceeds the connector's lease length, so no claude was started.")?; - let label = run.id.0.rsplit(':').next().unwrap_or(&run.id.0); - match mind.admit(AGENT, label, vec![document])? { - PipelineAdmissionOutcome::Committed { .. } => Ok(()), - _ => Err(anyhow!("the run whose lease was too long could not be withdrawn")), - } + admit_closure(mind, clock, run, outcome, "The unit's runtime cap exceeds the connector's lease length, so no claude was started.") } /// Carries her one open Self run: one `claude` child from the lease to its @@ -415,18 +443,39 @@ fn withdraw_misconfigured(mind: &dyn MindPort, clock: &dyn Clock, run: &Pipeline /// `run_cap_s` is the unit's `RuntimeMaxSec`: a connector that finds the lease /// too long is misconfigured, which is an error naming the mismatch and not a /// refusal of the run: the run is withdrawn, so the breaker counts nothing. +/// +/// Whatever happens after the run is found, it is resolved Recorded exactly once +/// before `carry` returns: the child's result (or that it gave none) in the +/// reason. A resolve that fails is logged and tried no more: the outcome (or the +/// original error) stands, a failed close of a run that ended normally is an +/// error, and the launch that opens the next run closes the run left open. pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write) -> Result { let Some((run, budget)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) }; + let mut spent = Spent::default(); + let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &budget), &mut spent); + let closed = close_if_open(mind, clock, &run, &ran, &spent); + if closed.is_err() { + let _ = writeln!(out, "the run could not be closed: the next launch closes it"); + } + match (ran, closed) { + (Err(error), _) => Err(error), + (Ok(outcome), Ok(())) => Ok(outcome), + (Ok(_), Err(error)) => Err(error.context("the run's process ended but its run could not be closed")), + } +} + +#[allow(clippy::too_many_arguments, reason = "carry's ports, passed on whole")] +fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write, (run, budget): (&PipelineRef, &str), spent: &mut Spent) -> Result { let request = LeaseRequest { expires_at_unix_ms: unix_ms(clock.now()).saturating_add(run_cap_s.saturating_mul(1000)) }; let lease = match lender.lend(&request)? { Lent::Granted(lease) => lease, Lent::Refused(refused) if refused.kind == LeaseRefusalKind::Expired => { - withdraw_misconfigured(mind, clock, &run)?; + withdraw_misconfigured(mind, clock, run)?; bail!("the connector refused a {run_cap_s} s lease as too long: raise its --lease-max-ms to at least the unit's RuntimeMaxSec ({RUN_CAP_VARIABLE}) plus slack"); } - Lent::Refused(refused) => return close_refused(mind, clock, &run, refused.kind), + Lent::Refused(refused) => return close_refused(mind, clock, run, refused.kind), }; - let child = match spawner.spawn(&claude_args(session_id(&run), &budget), lease.token()) { + let child = match spawner.spawn(&claude_args(session_id(run), budget), lease.token()) { Ok(child) => child, Err(error) => { // No child ever held the token; the empty report is the connector's cue to bill the reservation. @@ -434,8 +483,7 @@ pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawne return Err(error); } }; - let mut spent = Spent::default(); - let outcome = drive(mind, clock, child, &run, &mut spent, out)?; + let outcome = drive(mind, clock, child, run, spent, out)?; // A child that could not even be killed may still be spending: that lease is // dropped, and the connector bills the reservation. if !matches!(outcome, SelfRunOutcome::Ended { cause: Cause::Unstoppable, .. }) { @@ -452,15 +500,12 @@ pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawne /// The most of a result's text that reaches the journal. const RESULT_TEXT_LIMIT: usize = 4096; +/// The most of a result's text that reaches the closing resolution's reason. +const REASON_EXCERPT_LIMIT: usize = 800; -/// What `claude` concluded: a result event's text, with its subtype and error -/// flag when it has them, written to the unit's stdout so it reaches the journal -/// before the transcript goes with the runtime directory. Bounded, marked when cut. -fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> { - let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return Ok(()) }; - if event.get("type").and_then(Value::as_str) != Some("result") { - return Ok(()); - } +/// A result event as one line: its subtype and error flag when it has them, then +/// its text cut at a character boundary to `limit` bytes, marked when cut. +fn describe_result(event: &serde_json::Map, limit: usize) -> String { let mut head = String::from("result"); if let Some(subtype) = event.get("subtype").and_then(Value::as_str) { head.push_str(&format!(" subtype={subtype}")); @@ -469,12 +514,22 @@ fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> { head.push_str(&format!(" is_error={is_error}")); } let text = event.get("result").and_then(Value::as_str).unwrap_or(""); - let mut end = text.len().min(RESULT_TEXT_LIMIT); + let mut end = text.len().min(limit); while !text.is_char_boundary(end) { end -= 1; } let marker = if end < text.len() { "\n[truncated]" } else { "" }; - writeln!(out, "{head}: {}{marker}", &text[..end]) + format!("{head}: {}{marker}", &text[..end]) +} + +/// What `claude` concluded, written to the unit's stdout so it reaches the +/// journal before the transcript goes with the runtime directory. +fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> { + let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return Ok(()) }; + if event.get("type").and_then(Value::as_str) != Some("result") { + return Ok(()); + } + writeln!(out, "{}", describe_result(&event, RESULT_TEXT_LIMIT)) } /// The loop: facts in, one clock read per fact, actions out. @@ -656,9 +711,11 @@ fn copy_level(root: &Path, from: &Path, to: &Path) -> std::io::Result<()> { /// from `ANTHROPIC_BASE_URL`, so the child gets only what a shell needs to find /// its way; nothing credential-shaped is on the list. Its `HOME` and /// `CLAUDE_CONFIG_DIR` are its fresh configuration directory, set by the spawn. +/// `EUREKA_INSTANCE` and `HUGINN_ENDPOINT` are the `huginn` CLI's whole identity +/// (ruling `eureka-body:ruling:cli-over-mcp-agent-surface`): a name and an address, no secret. fn child_may_see(name: &OsStr) -> bool { let name = name.to_string_lossy(); - matches!(name.as_ref(), "PATH" | "LANG" | "TERM") || name.starts_with("LC_") + matches!(name.as_ref(), "PATH" | "LANG" | "TERM" | "EUREKA_INSTANCE" | "HUGINN_ENDPOINT") || name.starts_with("LC_") } /// The unit's environment as the child sees it: the allow-list, and nothing else. diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 4a8c635..3a40abc 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -265,17 +265,18 @@ impl Drop for FakeLease { } /// The mind as the actuator sees it, over a `TestMind`: another instance's name, -/// doubled run headers, and a count of admissions it must never receive. +/// admissions counted, and optionally refused as a mind that cannot be reached would. struct Seen<'a> { inner: &'a TestMind, instance: Slug, doubled: bool, + admit_fails: bool, admits: Cell, } impl<'a> Seen<'a> { fn of(inner: &'a TestMind) -> Self { - Self { inner, instance: inner.instance().clone(), doubled: false, admits: Cell::new(0) } + Self { inner, instance: inner.instance().clone(), doubled: false, admit_fails: false, admits: Cell::new(0) } } } @@ -295,9 +296,12 @@ impl MindPort for Seen<'_> { Ok(page) } - fn admit(&self, _agent: &str, _session: &str, _documents: Vec) -> Result { + fn admit(&self, agent: &str, session: &str, documents: Vec) -> Result { self.admits.set(self.admits.get() + 1); - Err(anyhow!("the actuator admits nothing")) + if self.admit_fails { + return Err(anyhow!("down")); + } + self.inner.admit(agent, session, documents) } } @@ -967,9 +971,9 @@ fn the_clock_is_read_once_per_event_so_a_step_of_the_wall_clock_cannot_split_a_p let outcome = carry(&Seen::of(&mind), &Stepped(clock.clone()), &FakeLender::new(Answer::Grant), &fake, TEST_CAP_S, &mut Vec::new()).unwrap(); let events = log.borrow().clone(); let reads = events.iter().filter(|event| matches!(event, Event::Asked(_))).count() as u32; - // One read for the lease's expiry, one before the first wait, then one per fact heard: + // One read for the lease's expiry, one before the first wait, then one per fact heard, and one for the closure: // never a second read inside a pass. - assert_eq!(clock.1.get(), 2 + reads, "{events:?}"); + assert_eq!(clock.1.get(), 3 + reads, "{events:?}"); // The flush was written on the first request at time 0; the very next pass sees the stepped clock. // The flush pass itself has no interrupt, whatever the step: the interrupt is on a later event. let flush_at = events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains("context ceiling"))).unwrap(); @@ -1043,16 +1047,6 @@ fn unreadable_and_foreign_lines_are_skipped_and_never_printed() { assert!(!ran.written().iter().any(|text| text.contains(canary)), "nothing from the stream is written back"); } -#[test] -fn the_actuator_admits_nothing_to_the_mind_through_a_whole_flushed_run() { - let mind = with_run(); - let seen = Seen::of(&mind); - let ran = carry_with(&seen, vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())], Some(0), false); - assert_eq!(ran.admits, 0); - assert!(ran.outcome.is_ok()); - // And the run is exactly as she left it: still in force until her flush, or the next launch, closes it. - assert_eq!(mind.ids(K::Resolution), Vec::::new()); -} // ---- the real child ----------------------------------------------------------- @@ -1320,3 +1314,4 @@ mod process { } mod lease; +mod closure; diff --git a/crates/mind-body/src/self_run/tests/closure.rs b/crates/mind-body/src/self_run/tests/closure.rs new file mode 100644 index 0000000..7e7270a --- /dev/null +++ b/crates/mind-body/src/self_run/tests/closure.rs @@ -0,0 +1,121 @@ +//! Every exit of `carry` resolves her run Recorded exactly once. + +use eureka_pipeline::ResolutionOutcome; + +use super::*; + +/// The reason of the run's one closure; fails if it was closed other than once. +fn reason_of(mind: &TestMind, label: &str) -> String { + let id = format!("{INSTANCE}:resolution:run.{label}.n1"); + let PipelineDocument::Resolution(closure) = mind.view(K::Resolution, &id).unwrap_or_else(|| panic!("{label} was not closed")).document else { panic!("not a resolution") }; + assert!(mind.view(K::Resolution, &format!("{INSTANCE}:resolution:run.{label}.n2")).is_none(), "{label} was closed twice"); + match closure.outcome { + ResolutionOutcome::Recorded { reason } => reason.0, + other => panic!("not Recorded: {other:?}"), + } +} + +fn result_with(text: &str) -> String { + let mut event: Value = serde_json::from_str(&result_line()).unwrap(); + event["result"] = json!(text); + event.to_string() +} + +#[test] +fn a_normal_exit_resolves_the_run_once_with_the_results_subtype_error_flag_and_text() { + let mind = with_run(); + let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); + assert!(ran.outcome.is_ok()); + assert_eq!(ran.admits, 1); + let reason = reason_of(&mind, "selfrun-a"); + assert!(reason.contains("subtype=success is_error=true") && reason.contains("Failed to authenticate: OAuth session expired"), "{reason}"); + assert_eq!(mind.ids(K::Resolution).len(), 1); +} + +#[test] +fn the_excerpt_is_bounded_and_cut_at_a_character_boundary() { + let mind = with_run(); + // Two bytes a character, so the limit falls inside one when it is odd; here it is even, so add a one-byte head. + let text = format!("x{}", "é".repeat(3000)); + carry_with(&Seen::of(&mind), vec![Out(result_with(&text))], Some(0), false).outcome.unwrap(); + let reason = reason_of(&mind, "selfrun-a"); + let excerpt = reason.split_once(": ").unwrap().1; + assert!(excerpt.contains("[truncated]") && excerpt.len() <= REASON_EXCERPT_LIMIT + "\n[truncated]".len() + 120, "{}", reason.len()); + assert!(excerpt.matches('é').count() <= 400); +} + +#[test] +fn a_child_that_ends_with_no_result_resolves_the_run_and_says_so() { + let mind = with_run(); + let ran = carry_with(&Seen::of(&mind), vec![Out(at("m1", 10))], Some(2), false); + assert_eq!(ran.ended().1, Some(2)); + let reason = reason_of(&mind, "selfrun-a"); + assert!(reason.contains("claude gave no result") && reason.contains("code 2"), "{reason}"); +} + +#[test] +fn a_child_stopped_by_the_actuator_resolves_the_run_with_how_it_ended() { + let mind = with_run(); + carry_with(&Seen::of(&mind), vec![Out(at("m1", CEILING_TOKENS)), Idle(FLUSH_GRACE_S)], None, true).outcome.unwrap(); + assert!(reason_of(&mind, "selfrun-a").contains("interrupted")); +} + +#[test] +fn a_child_that_cannot_start_resolves_the_run_and_the_error_stands() { + let mind = with_run(); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true }; + let error = carry(&Seen::of(&mind), &*clock, &FakeLender::new(Answer::Grant), &refusing, TEST_CAP_S, &mut Vec::new()).unwrap_err(); + assert_eq!(error.to_string(), "refused"); + assert!(reason_of(&mind, "selfrun-a").contains("the actuator failed")); +} + +#[test] +fn an_error_after_the_run_opened_resolves_it() { + struct Broken; + impl Write for Broken { + fn write(&mut self, _: &[u8]) -> std::io::Result { + Err(std::io::Error::other("closed")) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let mind = with_run(); + let log = Rc::new(RefCell::new(Vec::new())); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let child = Scripted { steps: vec![Out(at("m1", 10))].into(), log: log.clone(), clock: clock.clone(), code: Some(0), dies_on: None, refuses: vec![], hangs: true, exited: false, eof: false, asked: 0 }; + let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false }; + assert!(carry(&Seen::of(&mind), &*clock, &FakeLender::logging(log, Answer::Grant), &fake, TEST_CAP_S, &mut Broken).is_err()); + assert!(reason_of(&mind, "selfrun-a").contains("the actuator failed")); +} + +#[test] +fn a_run_the_child_closed_itself_is_not_closed_again() { + let mind = with_run(); + mind.committed(vec![close(run_ref("selfrun-a"), recorded())]); + // Closed, so it is not hers to carry or to close: nothing is admitted. + let seen = Seen::of(&mind); + assert_eq!(carry_with(&seen, vec![Out(result_line())], Some(0), false).outcome.unwrap(), SelfRunOutcome::NoRun); + assert_eq!(seen.admits.get(), 0); +} + +#[test] +fn a_resolve_that_fails_is_logged_and_tried_once() { + let mind = with_run(); + let mut seen = Seen::of(&mind); + seen.admit_fails = true; + let ran = carry_with(&seen, vec![Out(result_line())], Some(0), false); + assert!(format!("{:#}", ran.outcome.unwrap_err()).contains("could not be closed")); + assert_eq!(ran.admits, 1, "one try, no loop"); + assert!(ran.out.contains("the run could not be closed")); + assert!(mind.ids(K::Resolution).is_empty()); + + // The same when the run's own error is what is being reported: that error stands. + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true }; + let again = Seen { admit_fails: true, ..Seen::of(&mind) }; + let error = carry(&again, &*clock, &FakeLender::new(Answer::Grant), &refusing, TEST_CAP_S, &mut Vec::new()).unwrap_err(); + assert_eq!(error.to_string(), "refused"); + assert_eq!(again.admits.get(), 1); +} diff --git a/crates/mind-body/src/self_run/tests/lease.rs b/crates/mind-body/src/self_run/tests/lease.rs index 68fa60f..7675821 100644 --- a/crates/mind-body/src/self_run/tests/lease.rs +++ b/crates/mind-body/src/self_run/tests/lease.rs @@ -86,14 +86,17 @@ fn a_lease_refused_for_its_length_is_a_configuration_error_that_starts_nothing_a } #[test] -fn a_connector_that_cannot_be_reached_is_an_error_that_starts_nothing_and_leaves_the_run_open() { +fn a_connector_that_cannot_be_reached_is_an_error_that_starts_nothing_and_closes_the_run_recorded() { let mind = with_run(); let lender = FakeLender::new(Answer::Unreachable); let never = never_spawns(); let clock = Rc::new(Fixed(Cell::new(at_start()))); assert!(carry(&mind, &*clock, &lender, &never, TEST_CAP_S, &mut Vec::new()).is_err()); assert!(never.spawns.borrow().is_empty()); - assert_eq!(mind.ids(K::Resolution), Vec::::new(), "an unreachable connector is no refusal: the run stays open"); + // No refusal, so no refusal kind: the actuator failed after the run opened, and the run does not stay in force. + let closure = mind.view(K::Resolution, &format!("{INSTANCE}:resolution:run.selfrun-a.n1")).expect("the run is closed"); + let PipelineDocument::Resolution(closure) = closure.document else { panic!("not a resolution") }; + assert!(matches!(&closure.outcome, ResolutionOutcome::Recorded { reason } if reason.0.contains("the actuator failed")), "{:?}", closure.outcome); } #[test] diff --git a/crates/mind-body/src/self_run/tests/process/lease_child.rs b/crates/mind-body/src/self_run/tests/process/lease_child.rs index acd6d93..a64c422 100644 --- a/crates/mind-body/src/self_run/tests/process/lease_child.rs +++ b/crates/mind-body/src/self_run/tests/process/lease_child.rs @@ -70,9 +70,9 @@ echo done"); #[test] fn only_the_allow_listed_names_pass_from_the_units_environment() { let unit: Vec<(std::ffi::OsString, std::ffi::OsString)> = - ["PATH", "HOME", "LANG", "LC_ALL", "TERM", "CLAUDE_CONFIG_DIR", "ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_BASE_URL", "LCX", "XDG_DATA_HOME"].map(|name| (name.into(), "v".into())).to_vec(); + ["PATH", "HOME", "LANG", "LC_ALL", "TERM", "EUREKA_INSTANCE", "HUGINN_ENDPOINT", "CLAUDE_CONFIG_DIR", "ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_BASE_URL", "HUGINN_TOKEN", "EUREKA_INSTANCE_X", "LCX", "XDG_DATA_HOME"].map(|name| (name.into(), "v".into())).to_vec(); let passed: Vec = child_environment(&unit).into_iter().map(|(name, _)| name.to_string_lossy().into_owned()).collect(); - assert_eq!(passed, ["PATH", "LANG", "LC_ALL", "TERM"], "home and the configuration directory are the spawn's, never the unit's"); + assert_eq!(passed, ["PATH", "LANG", "LC_ALL", "TERM", "EUREKA_INSTANCE", "HUGINN_ENDPOINT"], "the CLI's two names pass; home, the configuration directory, credentials and look-alikes do not"); } /// Reports where the child's configuration is, its mode, and what is in it; leaves a file and a From a4f4d046651e5940e593e318bd5900056f8b2eef Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 05:15:15 +0100 Subject: [PATCH 002/265] eureka-pipeline: the Pulse kind (a Mind's heartbeat vote and declared urgency), instance-rooted; huginn-mind admits it in her own mind only Co-Authored-By: Claude Opus 5.5 --- crates/eureka-pipeline/src/lib.rs | 48 ++++++++++++++++++++++++++--- crates/huginn-mind/src/admission.rs | 3 +- crates/huginn-mind/src/docs.rs | 5 +-- crates/huginn-mind/src/index.rs | 2 +- crates/huginn-mind/src/mind.rs | 2 +- crates/huginn-mind/src/query.rs | 6 ++++ crates/huginn-mind/src/rows.rs | 14 +++++---- 7 files changed, 65 insertions(+), 15 deletions(-) diff --git a/crates/eureka-pipeline/src/lib.rs b/crates/eureka-pipeline/src/lib.rs index 9288d4c..93c4cbf 100644 --- a/crates/eureka-pipeline/src/lib.rs +++ b/crates/eureka-pipeline/src/lib.rs @@ -112,7 +112,7 @@ macro_rules! unbounded_scalars { )*}; } -unbounded_scalars!(u32, u64, bool); +unbounded_scalars!(u16, u32, u64, bool); fn bound(field: &str, limit: usize, actual: usize) -> Result<(), PipelineRefusal> { if actual > limit { @@ -572,6 +572,31 @@ fn run_is_well_formed(run: &PipelineRun, at: &str) -> Result<(), PipelineRefusal } } +/// The longest and shortest beat a Mind may vote for, in seconds. +pub const TEMPO_VOTE_MIN_S: u32 = 60; +pub const TEMPO_VOTE_MAX_S: u32 = 86_400; +/// Urgency is in thousandths: 0 declares none, 1000 the most. +pub const URGENCY_MAX_PERMILLE: u16 = 1000; + +/// A pulse votes a tempo or declares an urgency, in bounds, or says nothing it +/// could be read for: an empty pulse is refused. +fn pulse_is_well_formed(pulse: &PipelinePulse, at: &str) -> Result<(), PipelineRefusal> { + if let Some(beat) = pulse.tempo_vote_beat_s + && !(TEMPO_VOTE_MIN_S..=TEMPO_VOTE_MAX_S).contains(&beat) + { + return Err(format_error(&format!("{at}.tempo_vote_beat_s"), &beat.to_string())); + } + if let Some(urgency) = pulse.urgency_permille + && urgency > URGENCY_MAX_PERMILLE + { + return Err(format_error(&format!("{at}.urgency_permille"), &urgency.to_string())); + } + if pulse.tempo_vote_beat_s.is_none() && pulse.urgency_permille.is_none() { + return Err(format_error(&format!("{at}.tempo_vote_beat_s"), "a pulse that votes and declares nothing")); + } + Ok(()) +} + unit_enums! { EvidenceKind { Command, Test, Mutation, Probe, SourceRead, Capture } ClaimOutcome { Holds, Falsified, Unproven } @@ -710,6 +735,14 @@ value_types! { instance: Slug, label: Label, operated_by: RunOperator, turn: RunTurn, host: Short, started_on: Date, budget_usd: Short, claims: Vec[64], campaigns: Vec[8], } => run_is_well_formed + /// One Mind's heartbeat vote, in her own mind: the beat she wants (the swarm's + /// tempo is the lower median of every Mind's newest vote) and the urgency she + /// declares for herself until her next Persona turn. Either may be absent, not + /// both. Instance-rooted: the key is `:pulse: