GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

mind-self-run: cut the orphan sweep (the mind admits one Self run in force; launch closes a dead one); fix splice; closure tests

Browse source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 04:35:00 +01:00
commit 3c258bd9db
3 changed files with 34 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

39
crates/mind-body/src/self_run.rs
View file

@ -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<Option<(PipelineRef, String)>> {
/// 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<Option<(PipelineRef, String)>> {
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<SelfRunOutcome>, 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<SelfRunOutcome> {
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<dyn Child>, run: &PipelineRef, spent: &mut Spent, out: &mut dyn Write) -> Result<SelfRunOutcome> {

25
crates/mind-body/src/self_run/tests.rs
View file

@ -269,13 +269,14 @@ impl Drop for FakeLease {
struct Seen<'a> {
inner: &'a TestMind,
instance: Slug,
doubled: bool,
admit_fails: bool,
admits: Cell<usize>,
}
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<PipelineSelectionPage> {
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<PipelineDocument>) -> Result<PipelineAdmissionOutcome> {
@ -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();

17
crates/mind-body/src/self_run/tests/closure.rs
View file

@ -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();