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