From 55dd620113f41b365855f84551d8149f048b47e3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 04:32:06 +0100 Subject: [PATCH 1/4] mind-self-run: every exit of carry resolves her run Recorded once; a run left open by a dead process is closed at the next start Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/self_run.rs | 133 ++++++++++++----- crates/mind-body/src/self_run/tests.rs | 42 ++---- .../mind-body/src/self_run/tests/closure.rs | 136 ++++++++++++++++++ crates/mind-body/src/self_run/tests/lease.rs | 7 +- 4 files changed, 252 insertions(+), 66 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..4fc21fc 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -28,9 +28,13 @@ //! 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), and a run found open at +//! the start besides the one carried is an orphan of a process that ended +//! without closing it: one Self unit per Mind runs at a time (the systemd +//! template instance), so the older runs are closed first. 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. @@ -284,11 +288,22 @@ 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. -fn her_run(mind: &dyn MindPort) -> Result> { +/// Why a run found open beside the carried one was closed. +const ORPHANED: &str = "orphaned: the run's process ended without closing it"; + +/// Her open Self run and its budget, read through the launch primitive's holder +/// reader. The newest in force is the one this unit carries; any older one is an +/// orphan (its process ended without closing it, and one Self unit per Mind runs +/// at a time), closed here before anything is claimed. None when she has none. +fn her_run(mind: &dyn MindPort, clock: &dyn Clock, out: &mut dyn Write) -> Result> { let mut held = holders(mind, mind.instance(), RunTurn::SelfRun)?; - let (Some(run), None) = (held.pop(), held.pop()) else { return Ok(None) }; + held.sort_by(|a, b| (&a.admission.admitted_at, &a.id.id.0).cmp(&(&b.admission.admitted_at, &b.id.id.0))); + held.dedup_by(|a, b| a.id == b.id); + let Some(run) = held.pop() else { return Ok(None) }; + for orphan in held { + admit_closure(mind, clock, &orphan.id, ResolutionOutcome::Recorded { reason: ORPHANED.into() }, "An earlier Self run's process ended without closing it; the next start closed it.")?; + let _ = writeln!(out, "closed an orphaned run"); + } let one = Selection { keys: Some(vec![run.id.id.0.clone()]), projection: "document".into(), ..Selection::default() }; let PipelinePageItems::Documents(views) = mind.query(&one)?.items else { bail!("the mind answered headers to a document selection") }; match views.into_iter().next().map(|view| view.document) { @@ -310,6 +325,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 +337,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 +405,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 to the next start's orphan sweep. +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 +454,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 next start closes the run as an orphan. 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 Some((run, budget)) = her_run(mind, clock, out)? 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 start closes it as an orphan"); + } + 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 +494,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 +511,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 +525,23 @@ 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. diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 4a8c635..77f1f21 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -265,17 +265,17 @@ 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(), admit_fails: false, admits: Cell::new(0) } } } @@ -285,19 +285,15 @@ impl MindPort for Seen<'_> { } fn query(&self, selection: &Selection) -> Result { - let mut page = self.inner.query(selection)?; - if self.doubled - && let PipelinePageItems::Headers(items) = &mut page.items - { - let again = items.clone(); - items.extend(again); - } - Ok(page) + self.inner.query(selection) } - 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) } } @@ -406,7 +402,7 @@ fn flushes(ran: &Ran) -> usize { // ---- who is carried ----------------------------------------------------- #[test] -fn without_exactly_one_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawned() { +fn without_an_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawned() { let cases: Vec<(&str, Seed)> = vec![ ("no run at all", Box::new(|_| {})), ("a Self run the operator operates", Box::new(|mind| mind.committed(vec![run("op", RunTurn::SelfRun, RunOperator::Operator, &[])]))), @@ -432,13 +428,6 @@ fn without_exactly_one_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawne let mut other = Seen::of(&mind); other.instance = Slug("another".into()); assert_eq!(carry_with(&other, vec![], Some(0), false).outcome.unwrap(), SelfRunOutcome::NoRun); - - // Two open runs are not one: nothing is spawned for either. - let mut two = Seen::of(&mind); - two.doubled = true; - let ran = carry_with(&two, vec![], Some(0), false); - assert_eq!(ran.outcome.unwrap(), SelfRunOutcome::NoRun); - assert!(ran.spawns.is_empty()); } #[test] @@ -1043,16 +1032,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 +1299,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..01edeba --- /dev/null +++ b/crates/mind-body/src/self_run/tests/closure.rs @@ -0,0 +1,136 @@ +//! Every exit of `carry` resolves her run Recorded exactly once, and a run left +//! open by a process that died is closed at the next start. + +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_run_still_open_at_the_next_start_is_an_orphan_and_is_closed_before_the_newer_one_is_carried() { + let mind = TestMind::seeded(); + mind.committed(vec![self_run("selfrun-old", "5")]); + mind.committed(vec![self_run("selfrun-a", "5")]); + let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); + assert!(ran.outcome.is_ok()); + assert_eq!(ran.spawns.len(), 1); + assert!(ran.spawns[0].contains(&SESSION_OF_RUN_A.to_string()), "the newer run is the one carried"); + assert_eq!(reason_of(&mind, "selfrun-old"), "orphaned: the run's process ended without closing it"); + assert!(reason_of(&mind, "selfrun-a").contains("subtype=success")); + assert_eq!(mind.ids(K::Resolution).len(), 2); +} + +#[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] From 551a0340bfb43af1e23906e0ae052bc981d96688 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 04:32:26 +0100 Subject: [PATCH 2/4] mind-self-run: the child sees EUREKA_INSTANCE and HUGINN_ENDPOINT, the huginn CLI's identity Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/self_run.rs | 4 +++- crates/mind-body/src/self_run/tests/process/lease_child.rs | 4 ++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 4fc21fc..9c2f441 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -723,9 +723,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/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 3c258bd9db28bc87ca7c32438e26588da869434c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 04:35:00 +0100 Subject: [PATCH 3/4] mind-self-run: cut the orphan sweep (the mind admits one Self run in force; launch closes a dead one); fix splice; closure tests Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/self_run.rs | 39 +++++++------------ crates/mind-body/src/self_run/tests.rs | 25 +++++++++--- .../mind-body/src/self_run/tests/closure.rs | 17 +------- 3 files changed, 34 insertions(+), 47 deletions(-) diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 9c2f441..e1ec554 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -30,11 +30,9 @@ //! //! 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), and a run found open at -//! the start besides the one carried is an orphan of a process that ended -//! without closing it: one Self unit per Mind runs at a time (the systemd -//! template instance), so the older runs are closed first. 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 +//! 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. @@ -288,23 +286,13 @@ impl fmt::Display for SelfRunOutcome { } } -/// Why a run found open beside the carried one was closed. -const ORPHANED: &str = "orphaned: the run's process ended without closing it"; - -/// Her open Self run and its budget, read through the launch primitive's holder -/// reader. The newest in force is the one this unit carries; any older one is an -/// orphan (its process ended without closing it, and one Self unit per Mind runs -/// at a time), closed here before anything is claimed. None when she has none. -fn her_run(mind: &dyn MindPort, clock: &dyn Clock, out: &mut dyn Write) -> Result> { +/// Her one open Self run and its budget, read through the launch primitive's +/// 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)?; - held.sort_by(|a, b| (&a.admission.admitted_at, &a.id.id.0).cmp(&(&b.admission.admitted_at, &b.id.id.0))); - held.dedup_by(|a, b| a.id == b.id); - let Some(run) = held.pop() else { return Ok(None) }; - for orphan in held { - admit_closure(mind, clock, &orphan.id, ResolutionOutcome::Recorded { reason: ORPHANED.into() }, "An earlier Self run's process ended without closing it; the next start closed it.")?; - let _ = writeln!(out, "closed an orphaned run"); - } - let one = Selection { keys: Some(vec![run.id.id.0.clone()]), projection: "document".into(), ..Selection::default() }; + let (Some(run), None) = (held.pop(), held.pop()) else { return Ok(None) }; let PipelinePageItems::Documents(views) = mind.query(&one)?.items else { bail!("the mind answered headers to a document selection") }; match views.into_iter().next().map(|view| view.document) { Some(PipelineDocument::Run(document)) => Ok(Some((run.id, document.budget_usd.0))), @@ -421,7 +409,7 @@ fn admit_closure(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef, outc /// 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 to the next start's orphan sweep. +/// 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(()); @@ -459,14 +447,14 @@ fn withdraw_misconfigured(mind: &dyn MindPort, clock: &dyn Clock, run: &Pipeline /// 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 next start closes the run as an orphan. +/// 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, clock, out)? else { return Ok(SelfRunOutcome::NoRun) }; + 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 start closes it as an orphan"); + let _ = writeln!(out, "the run could not be closed: the next launch closes it"); } match (ran, closed) { (Err(error), _) => Err(error), @@ -542,7 +530,6 @@ fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> { } writeln!(out, "{}", describe_result(&event, RESULT_TEXT_LIMIT)) } -} /// The loop: facts in, one clock read per fact, actions out. fn drive(mind: &dyn MindPort, clock: &dyn Clock, mut child: Box, run: &PipelineRef, spent: &mut Spent, out: &mut dyn Write) -> Result { diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 77f1f21..3a40abc 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -269,13 +269,14 @@ impl Drop for FakeLease { 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(), admit_fails: false, admits: Cell::new(0) } + Self { inner, instance: inner.instance().clone(), doubled: false, admit_fails: false, admits: Cell::new(0) } } } @@ -285,7 +286,14 @@ impl MindPort for Seen<'_> { } fn query(&self, selection: &Selection) -> Result { - self.inner.query(selection) + let mut page = self.inner.query(selection)?; + if self.doubled + && let PipelinePageItems::Headers(items) = &mut page.items + { + let again = items.clone(); + items.extend(again); + } + Ok(page) } fn admit(&self, agent: &str, session: &str, documents: Vec) -> Result { @@ -402,7 +410,7 @@ fn flushes(ran: &Ran) -> usize { // ---- who is carried ----------------------------------------------------- #[test] -fn without_an_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawned() { +fn without_exactly_one_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawned() { let cases: Vec<(&str, Seed)> = vec![ ("no run at all", Box::new(|_| {})), ("a Self run the operator operates", Box::new(|mind| mind.committed(vec![run("op", RunTurn::SelfRun, RunOperator::Operator, &[])]))), @@ -428,6 +436,13 @@ fn without_an_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawned() { let mut other = Seen::of(&mind); other.instance = Slug("another".into()); assert_eq!(carry_with(&other, vec![], Some(0), false).outcome.unwrap(), SelfRunOutcome::NoRun); + + // Two open runs are not one: nothing is spawned for either. + let mut two = Seen::of(&mind); + two.doubled = true; + let ran = carry_with(&two, vec![], Some(0), false); + assert_eq!(ran.outcome.unwrap(), SelfRunOutcome::NoRun); + assert!(ran.spawns.is_empty()); } #[test] @@ -956,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(); diff --git a/crates/mind-body/src/self_run/tests/closure.rs b/crates/mind-body/src/self_run/tests/closure.rs index 01edeba..7e7270a 100644 --- a/crates/mind-body/src/self_run/tests/closure.rs +++ b/crates/mind-body/src/self_run/tests/closure.rs @@ -1,5 +1,4 @@ -//! Every exit of `carry` resolves her run Recorded exactly once, and a run left -//! open by a process that died is closed at the next start. +//! Every exit of `carry` resolves her run Recorded exactly once. use eureka_pipeline::ResolutionOutcome; @@ -101,20 +100,6 @@ fn a_run_the_child_closed_itself_is_not_closed_again() { assert_eq!(seen.admits.get(), 0); } -#[test] -fn a_run_still_open_at_the_next_start_is_an_orphan_and_is_closed_before_the_newer_one_is_carried() { - let mind = TestMind::seeded(); - mind.committed(vec![self_run("selfrun-old", "5")]); - mind.committed(vec![self_run("selfrun-a", "5")]); - let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); - assert!(ran.outcome.is_ok()); - assert_eq!(ran.spawns.len(), 1); - assert!(ran.spawns[0].contains(&SESSION_OF_RUN_A.to_string()), "the newer run is the one carried"); - assert_eq!(reason_of(&mind, "selfrun-old"), "orphaned: the run's process ended without closing it"); - assert!(reason_of(&mind, "selfrun-a").contains("subtype=success")); - assert_eq!(mind.ids(K::Resolution).len(), 2); -} - #[test] fn a_resolve_that_fails_is_logged_and_tried_once() { let mind = with_run(); From 93f93c12a9f981020d316040499a0a87213d2127 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 04:35:46 +0100 Subject: [PATCH 4/4] mind-self-run: restore the run document selection Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/self_run.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index e1ec554..ca743b3 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -293,6 +293,7 @@ impl fmt::Display for SelfRunOutcome { 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) }; + let one = Selection { keys: Some(vec![run.id.id.0.clone()]), projection: "document".into(), ..Selection::default() }; let PipelinePageItems::Documents(views) = mind.query(&one)?.items else { bail!("the mind answered headers to a document selection") }; match views.into_iter().next().map(|view| view.document) { Some(PipelineDocument::Run(document)) => Ok(Some((run.id, document.budget_usd.0))),