mind-self-run tests: a loop that never ends fails the test instead of hanging it

The scripted child counts its reads and the real child is bounded at 15 s, so a mutant that stops the run from ending is a failing test, not a cargo-mutants timeout. Builds; every test passes.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Cargo.lock

@@ -1126,8 +1126,10 @@ dependencies = [
  "quote",
  "rust_decimal",
  "serde",
+ "serde_json",
  "syn 2.0.119",
  "tempfile",
+ "uuid",
 ]
 
 [[package]]
@@ -1755,6 +1757,12 @@ dependencies = [
  "serde_core",
 ]
 
+[[package]]
+name = "sha1_smol"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d"
+
 [[package]]
 name = "sha2"
 version = "0.10.9"
@@ -2097,6 +2105,7 @@ checksum = "2ef6dac1e96601b4fb3acccccff2139741fcb757cb9a36089bf5be91cfb285ce"
 dependencies = [
  "getrandom 0.4.3",
  "js-sys",
+ "sha1_smol",
  "wasm-bindgen",
 ]
 

crates/mind-body/Cargo.toml

@@ -11,11 +11,15 @@ 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
-# (`wake`, the `mind-wake` binary) is one cycle over them.
+# (`wake`, the `mind-wake` binary) is one cycle over them. The actuator
+# (`self_run`, the `mind-self-run` binary) carries her one open Self run.
 #
 # `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
 # serialises as a string, so the stored document carries no float.
+#
+# `serde_json` reads Claude Code's stream-json, a foreign owner's wire (`self_run`);
+# `uuid`'s v5 makes a Self run's session id derivable from the run's id.
 [dependencies]
 anyhow = "1"
 chrono = { version = "0.4.44", features = ["serde"] }
@@ -26,6 +30,8 @@ eureka-state = { path = "../eureka-state", default-features = false }
 huginn-mind = { path = "../huginn-mind" }
 rust_decimal = { version = "1", default-features = false, features = ["std", "serde"] }
 serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+uuid = { version = "1", default-features = false, features = ["v5"] }
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
@@ -50,3 +56,7 @@ path = "src/main.rs"
 [[bin]]
 name = "mind-wake"
 path = "src/bin/mind-wake.rs"
+
+[[bin]]
+name = "mind-self-run"
+path = "src/bin/mind-self-run.rs"

crates/mind-body/README.md

@@ -1,7 +1,14 @@
 # mind-body
 
-The Mind's body: the control store, the launch primitive and the `mind-wake`
-one-shot.
+The Mind's body: the control store, the launch primitive, the `mind-wake`
+one-shot and the `mind-self-run` actuator.
+
+`mind-self-run --instance <slug> [--endpoint rudp://<ip>:<port>]` runs inside
+`mind-self@<instance>`. It finds her one open Self run, starts `claude` over
+stream-json with a fixed argument list, asks for the flush once at the context
+ceiling and stops a child that outlives the flush (SIGINT, then SIGTERM and SIGKILL). It exits 0 when the child exited 0, 1
+otherwise, 3 when she has no single open Self run and 2 when the arguments are
+wrong or the mind cannot be reached.
 
 ## The container lane
 
@@ -25,3 +32,17 @@ Mutation testing of the waker needs the glob form, because `-f src/wake.rs`
 finds no mutants in this workspace: `cargo mutants -p mind-body -f
 "**/src/wake.rs" -f "**/src/bin/mind-wake.rs"`. The second glob reaches
 `host()`, whose mutants only the container lane kills.
+
+The actuator has its own ignored test, `the_real_bin_finds_her_run_...` in
+`tests/self_run.rs`: the real `mind-self-run` against the real daemon with a
+fake `claude` first on `PATH`. It needs the daemon built beside the binary and a
+POSIX shell, not root:
+
+```sh
+cargo build -p huginn-daemon -p mind-body --all-targets   && cargo test -p mind-body --test self_run -- --include-ignored
+```
+
+Mutation testing of the actuator, with the globs that reach the library and the
+binary: `cargo mutants -p mind-body -f "**/src/self_run.rs" -f
+"**/src/bin/mind-self-run.rs"`. The binary's mutants are killed only by the
+lane above.

crates/mind-body/src/bin/mind-self-run.rs

@@ -0,0 +1,44 @@
+//! `mind-self-run --instance <slug> [--endpoint rudp://<ip>:<port>]`: carries
+//! her one open Self run (`mind_body::self_run::carry`) inside
+//! `mind-self@<instance>`. This reads the process's arguments and environment,
+//! the things the library never reads, and prints the request lines and the
+//! outcome. It exits 0 when the child exited 0, 1 when it exited otherwise or by
+//! a signal, 3 when she has no single open Self run (no child started), and 2
+//! when the mind cannot be reached or the arguments are wrong. Without
+//! `--endpoint` the daemon is `HUGINN_ENDPOINT`.
+
+use std::time::Duration;
+
+use anyhow::Result;
+use eureka_state::HuginnClient;
+use mind_body::cli::instance_and_endpoint;
+use mind_body::launch::SystemClock;
+use mind_body::queue::HuginnMind;
+use mind_body::self_run::{ClaudeSpawner, carry};
+
+/// How long one call to the mind may take.
+const TIMEOUT: Duration = Duration::from_secs(15);
+
+const USAGE: &str = "usage: mind-self-run --instance <slug> [--endpoint rudp://<ip>:<port>]";
+
+// 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<i32> {
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    let (instance, address) = instance_and_endpoint(&args, std::env::var("HUGINN_ENDPOINT").ok(), USAGE)?;
+    let mind = HuginnMind::new(HuginnClient::new(address, instance, TIMEOUT));
+    let outcome = carry(&mind, &SystemClock, &ClaudeSpawner, &mut std::io::stdout())?;
+    println!("{outcome}");
+    Ok(outcome.exit_status())
+}
+
+fn main() {
+    match run() {
+        Ok(status) => std::process::exit(status),
+        Err(error) => {
+            eprintln!("mind-self-run: {error:#}");
+            std::process::exit(2);
+        }
+    }
+}

crates/mind-body/src/bin/mind-wake.rs

@@ -5,13 +5,12 @@
 //! 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 anyhow::Result;
 use eureka_state::HuginnClient;
+use mind_body::cli::instance_and_endpoint;
 use mind_body::control::{CONTROL_ROOT, FileSource, control_path};
 use mind_body::launch::{Ports, SystemClock, SystemdLauncher};
 use mind_body::queue::HuginnMind;
@@ -22,25 +21,6 @@ const TIMEOUT: Duration = Duration::from_secs(15);
 
 const USAGE: &str = "usage: mind-wake --instance <slug> [--endpoint rudp://<ip>:<port>]";
 
-/// 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<String>)> {
-    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())
@@ -51,13 +31,8 @@ fn host() -> String {
 #[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<String> {
     let args: Vec<String> = std::env::args().skip(1).collect();
-    let (instance, endpoint) = parse(&args)?;
+    let (instance, address) = instance_and_endpoint(&args, std::env::var("HUGINN_ENDPOINT").ok(), USAGE)?;
     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::<SocketAddr>().ok())
-        .ok_or_else(|| anyhow!("the endpoint is not rudp://<ip>:<port>"))?;
     let mind = HuginnMind::new(HuginnClient::new(address, instance, TIMEOUT));
     let launcher = SystemdLauncher::new();
     let host = host();

crates/mind-body/src/cli.rs

@@ -18,6 +18,7 @@
 //! fails if either step failed. No other command stops anything.
 
 use std::io::Write;
+use std::net::SocketAddr;
 use std::path::Path;
 use std::str::FromStr;
 
@@ -32,6 +33,36 @@ use crate::launch::unit;
 const USAGE: &str =
     "usage: mind-control [--root <dir>] --instance <slug> show | check | brake hold | brake release | dial set --heat H --base-cooldown-s S --base-run-usd U";
 
+/// The arguments every organ binary shares: `--instance <slug>` and an optional
+/// `--endpoint rudp://<ip>:<port>`, which `env_endpoint` (the binary's reading of
+/// `HUGINN_ENDPOINT`) replaces when absent. A flag given twice, an unknown
+/// argument, a name that is not a slug or an endpoint that is not
+/// `rudp://<ip>:<port>` is refused with `usage` or a fixed message, and no
+/// value is echoed.
+pub fn instance_and_endpoint(args: &[String], env_endpoint: Option<String>, usage: &str) -> Result<(Slug, SocketAddr)> {
+    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}");
+        }
+    }
+    let instance = Slug(instance.ok_or_else(|| anyhow!("{usage}"))?);
+    instance.validate_slug().map_err(|_| anyhow!("the instance is not a slug"))?;
+    let endpoint = endpoint.or(env_endpoint).ok_or_else(|| anyhow!("no daemon endpoint: --endpoint or HUGINN_ENDPOINT"))?;
+    let address = endpoint
+        .strip_prefix("rudp://")
+        .and_then(|address| address.parse::<SocketAddr>().ok())
+        .ok_or_else(|| anyhow!("the endpoint is not rudp://<ip>:<port>"))?;
+    Ok((instance, address))
+}
+
 /// Who made the change: `SUDO_USER`, else `USER`, else `unknown`. An empty
 /// variable counts as unset.
 pub fn operator_name(sudo_user: Option<String>, user: Option<String>) -> String {

crates/mind-body/src/launch.rs

@@ -273,8 +273,9 @@ fn close_dead(mind: &dyn MindPort, run: &PipelineRef, now: DateTime<Utc>) -> Res
 
 /// The in-force runs of hers (operated by the mind) of `turn` on `instance`,
 /// read from typed facts: the holders of that turn's slot. The one reader of
-/// the slot, for step 0 and for checking a holder admission names.
-fn holders(mind: &dyn MindPort, instance: &Slug, turn: RunTurn) -> Result<Vec<PipelineDocumentSummary>> {
+/// the slot, for step 0 and for checking a holder admission names. `self_run`
+/// reads her open Self run through it too.
+pub(crate) fn holders(mind: &dyn MindPort, instance: &Slug, turn: RunTurn) -> Result<Vec<PipelineDocumentSummary>> {
     let runs = headers(mind, &in_force(any_of(of_kind(PipelineKind::Run), "root", &[&instance.0])))?;
     Ok(runs
         .into_iter()

crates/mind-body/src/lib.rs

@@ -10,6 +10,10 @@
 //! 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.
+//!
+//! `self_run` carries one Self run inside `mind-self@<instance>`: the
+//! `mind-self-run` binary runs `claude` over stream-json for her one open Self
+//! run, asks for the flush at the context ceiling and stops a child that outlives it.
 
 // 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
@@ -23,6 +27,7 @@ pub mod cli;
 pub mod control;
 pub mod launch;
 pub mod queue;
+pub mod self_run;
 pub mod wake;
 
 #[cfg(test)]

crates/mind-body/src/self_run.rs

@@ -0,0 +1,374 @@
+//! The Self-run actuator: one `claude` child carries her one open Self run.
+//!
+//! `carry` finds her run (the one in-force Self run operated by the mind, read
+//! through `launch::holders`: the run is the grant), starts `claude` with the
+//! fixed argument list, writes the start message and watches the child's
+//! stream-json output. It counts the main thread's requests, asks for the flush
+//! once when the context prefix reaches the ceiling (ruling
+//! `eureka-body:ruling:self-lifecycle-flush-rehydrate`: a Self run admits what
+//! the mind lacks and ends, it never summarises), closes the child's input at
+//! the first result, and stops a child that outlives its flush: an interrupt
+//! after the grace, or once the requests of the flush's own turn have grown the
+//! prefix by the headroom beyond the prefix the flush was asked at (ruling
+//! `eureka-body:ruling:flush-always-gets-a-turn`), then SIGTERM and SIGKILL if
+//! the child ignores that. Signals go to the child's process group.
+//!
+//! Every timing decision is one pure function, `decide` (the `decide` module,
+//! ruling `eureka-body:ruling:self-run-decider`). `carry` is the loop that turns
+//! the operating system's facts into events and executes the actions.
+//!
+//! The actuator admits nothing, 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 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.
+
+use std::fmt;
+use std::io::{BufRead, BufReader, Write};
+use std::path::Path;
+use std::process::{ChildStdin, Command, Stdio};
+use std::sync::mpsc::{Receiver, RecvTimeoutError, channel};
+use std::time::Duration as StdDuration;
+
+use anyhow::{Context, Result, anyhow, bail};
+use cultnet_rs::Selection;
+use eureka_pipeline::{PipelineDocument, PipelineRef, RunTurn, Slug};
+use huginn_mind::PipelinePageItems;
+use serde_json::{Value, json};
+use uuid::Uuid;
+
+use crate::launch::{Clock, holders};
+use crate::queue::MindPort;
+
+use decide::{Action, Event, State, decide, next_deadline};
+
+/// The measured Self ceiling: at this prefix the flush is asked for.
+pub const CEILING_TOKENS: u64 = 300_000;
+/// What the prefix may grow by beyond the prefix the flush was asked at before
+/// the actuator interrupts. Relative to the flush, never an absolute limit.
+pub const FLUSH_HEADROOM_TOKENS: u64 = 60_000;
+/// How long the child may take over the flush before it is interrupted.
+pub const FLUSH_GRACE_S: i64 = 15 * 60;
+/// How long a child may ignore a signal before the next, stronger one.
+pub const ESCALATE_AFTER_S: i64 = 120;
+/// How long a stream may stay open after the child has exited (a descendant
+/// holding its stdout) before the run is over without it.
+pub const EXIT_DRAIN_S: i64 = 2;
+/// A backstop `claude` enforces; the ceiling is reached first.
+pub const MAX_TURNS: u32 = 500;
+/// The name the start message tells her to admit as.
+pub const AGENT: &str = "mind-self-run";
+/// The namespace of the session id: the run id is the name in it.
+pub const SESSION_NAMESPACE: Uuid = Uuid::NAMESPACE_OID;
+
+/// The longest the loop waits on the child when no deadline is nearer.
+const POLL: StdDuration = StdDuration::from_secs(60);
+
+/// What the child is told at the start. The contract with the `mind-self`
+/// skill: it names the skill, the instance and the run id.
+pub fn start_message(instance: &Slug, run: &PipelineRef) -> String {
+    format!(
+        "Use the mind-self skill. You are Self for the instance `{}`; your run is `{}`. Admit as faculty SelfFaculty, agent {AGENT}.",
+        instance.0, run.id.0
+    )
+}
+
+/// What the child is told once, at the context ceiling.
+pub const FLUSH_MESSAGE: &str = "The context ceiling is reached. Admit to the mind what it lacks, close your run, and end.";
+
+/// The session id of a run: UUIDv5 over the run's id. Derived, never stored.
+pub fn session_id(run: &PipelineRef) -> Uuid {
+    Uuid::new_v5(&SESSION_NAMESPACE, run.id.0.as_bytes())
+}
+
+/// The argument list of the child, and nothing else: no resume, continue or
+/// fork, no model, no permission mode and no settings belong to this actuator.
+/// `budget_usd` is the run's, as stored.
+pub fn claude_args(session: Uuid, budget_usd: &str) -> Vec<String> {
+    [
+        "-p",
+        "--input-format",
+        "stream-json",
+        "--output-format",
+        "stream-json",
+        "--verbose",
+        "--session-id",
+        &session.to_string(),
+        "--max-budget-usd",
+        budget_usd,
+        "--max-turns",
+        &MAX_TURNS.to_string(),
+        "--permission-prompts",
+        "none",
+    ]
+    .map(String::from)
+    .to_vec()
+}
+
+/// One line of the child's stream, as far as the actuator reads it.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Usage {
+    /// A request of the main thread: the message's id and the context it
+    /// read (input, cache read and cache write tokens).
+    Request { id: String, prefix: u64 },
+    Result,
+    /// Valid JSON that is none of the above, or a request of a subagent.
+    Other,
+    /// Not a JSON object.
+    Unreadable,
+}
+
+/// Reads one stream line.
+pub fn read_usage(line: &str) -> Usage {
+    let Ok(Value::Object(event)) = serde_json::from_str::<Value>(line) else { return Usage::Unreadable };
+    match event.get("type").and_then(Value::as_str) {
+        Some("result") => Usage::Result,
+        Some("assistant") if event.get("parent_tool_use_id").is_none_or(Value::is_null) => {
+            let message = event.get("message");
+            let Some(id) = message.and_then(|message| message.get("id")).and_then(Value::as_str) else { return Usage::Other };
+            let usage = message.and_then(|message| message.get("usage"));
+            let field = |name: &str| usage.and_then(|usage| usage.get(name)).and_then(Value::as_u64).unwrap_or(0);
+            // Saturating: a sum that overflows is at or over the ceiling, never a panic or a wrap.
+            let prefix = field("input_tokens").saturating_add(field("cache_read_input_tokens")).saturating_add(field("cache_creation_input_tokens"));
+            Usage::Request { id: id.to_string(), prefix }
+        }
+        _ => Usage::Other,
+    }
+}
+
+/// What the child gave in one wait: the operating system's facts about it.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Read {
+    Line(String),
+    Timeout,
+    /// Its stdout ended.
+    Eof,
+    /// It was reaped: its exit code, none when a signal ended it.
+    Exited(Option<i32>),
+}
+
+/// The running `claude`. A test hands in a scripted one.
+pub trait Child {
+    /// Writes one line to its input. A dead child's stream closes, which is
+    /// how `carry` learns of it, so a failed write is not an error here.
+    fn write_line(&mut self, line: &str);
+    /// Closes its input; later writes go nowhere.
+    fn close_input(&mut self);
+    fn next(&mut self, timeout: StdDuration) -> Read;
+    /// Sends the signal to its whole process group; whether it was delivered.
+    fn stop(&mut self, how: Stop) -> bool;
+}
+
+/// The signals the actuator sends, weakest first.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub enum Stop {
+    /// SIGINT, as an operator's Ctrl-C.
+    Interrupt,
+    Terminate,
+    Kill,
+}
+
+impl Stop {
+    fn next(self) -> Self {
+        match self {
+            Self::Interrupt => Self::Terminate,
+            Self::Terminate | Self::Kill => Self::Kill,
+        }
+    }
+}
+
+pub trait Spawner {
+    fn spawn(&self, args: &[String]) -> Result<Box<dyn Child>>;
+}
+
+/// Why a run ended.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Cause {
+    /// The child ended with no flush asked for and no interrupt.
+    Own,
+    /// The ceiling was reached and the flush asked for; the child ended with no
+    /// signal. Whether it ran the flush is for the mind to say, not this label.
+    FlushRequested,
+    /// A signal was delivered, and this was the strongest one delivered.
+    Signalled(Stop),
+    /// Not even SIGKILL could be delivered: the child was not waited for.
+    Unstoppable,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum SelfRunOutcome {
+    /// She has not exactly one open Self run: nothing was started.
+    NoRun,
+    Ended { cause: Cause, code: Option<i32>, requests: u64, prefix: u64 },
+}
+
+impl SelfRunOutcome {
+    /// 0 when the child exited 0, 3 when no child was started, else 1 (a
+    /// child killed by a signal has no code).
+    pub fn exit_status(&self) -> i32 {
+        match self {
+            Self::NoRun => 3,
+            Self::Ended { code: Some(0), .. } => 0,
+            Self::Ended { .. } => 1,
+        }
+    }
+}
+
+impl fmt::Display for SelfRunOutcome {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::NoRun => write!(f, "no run: she has no single open Self run"),
+            Self::Ended { cause, code, requests, prefix } => {
+                let cause = match cause {
+                    Cause::Own => "own",
+                    Cause::FlushRequested => "flush-requested",
+                    Cause::Signalled(Stop::Interrupt) => "interrupted",
+                    Cause::Signalled(Stop::Terminate) => "terminated",
+                    Cause::Signalled(Stop::Kill) => "killed",
+                    Cause::Unstoppable => "unstoppable",
+                };
+                match code {
+                    Some(code) => write!(f, "ended: {cause} code {code} requests {requests} prefix {prefix}"),
+                    None => write!(f, "ended: {cause} signal requests {requests} prefix {prefix}"),
+                }
+            }
+        }
+    }
+}
+
+/// 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<Option<(PipelineRef, String)>> {
+    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))),
+        _ => Err(anyhow!("the run's document could not be read")),
+    }
+}
+
+/// Carries her one open Self run: one `claude` child from the start message to
+/// its end. The request lines go to `out`. A mind that cannot be reached or a
+/// child that cannot be started is an error; every other end is an outcome.
+/// This is only the loop: facts in, one clock read per fact, actions out.
+pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, spawner: &dyn Spawner, out: &mut dyn Write) -> Result<SelfRunOutcome> {
+    let Some((run, budget)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) };
+    let mut child = spawner.spawn(&claude_args(session_id(&run), &budget))?;
+    let user_line = |text: &str| json!({"type": "user", "message": {"role": "user", "content": text}}).to_string();
+    child.write_line(&user_line(&start_message(mind.instance(), &run)));
+
+    let mut state = State::default();
+    let mut now = clock.now();
+    loop {
+        let wait = next_deadline(&state).map_or(POLL, |at| (at - now).to_std().unwrap_or_default().min(POLL));
+        let fact = match child.next(wait) {
+            Read::Line(line) => match read_usage(&line) {
+                Usage::Request { id, prefix } => Event::Request { id, prefix },
+                Usage::Result => Event::Result,
+                Usage::Other | Usage::Unreadable => Event::Tick,
+            },
+            Read::Timeout => Event::Tick,
+            Read::Eof => Event::StdoutEof,
+            Read::Exited(code) => Event::ChildExited(code),
+        };
+        now = clock.now();
+        let mut event = Some(fact);
+        while let Some(heard) = event.take() {
+            let counted = state.requests;
+            let (next, action) = decide(state, heard, now);
+            state = next;
+            if state.requests != counted {
+                writeln!(out, "request {} prefix {}", state.requests, state.prefix)?;
+            }
+            match action {
+                Action::None => {}
+                Action::WriteFlush => child.write_line(&user_line(FLUSH_MESSAGE)),
+                Action::CloseStdin => child.close_input(),
+                Action::Signal(how) => event = Some(Event::Signalled { how, delivered: child.stop(how) }),
+                Action::Finish(outcome) => return Ok(outcome),
+            }
+        }
+    }
+}
+
+/// The real spawner: the `claude` on the unit's `PATH`, with the unit's
+/// environment (the library names no variable), its stderr in the journal.
+pub struct ClaudeSpawner;
+
+impl Spawner for ClaudeSpawner {
+    fn spawn(&self, args: &[String]) -> Result<Box<dyn Child>> {
+        spawn_program(Path::new("claude"), args)
+    }
+}
+
+fn spawn_program(program: &Path, args: &[String]) -> Result<Box<dyn Child>> {
+    let mut command = Command::new(program);
+    command.args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit());
+    // Its own process group, so that a signal reaches what the child started too.
+    #[cfg(unix)]
+    std::os::unix::process::CommandExt::process_group(&mut command, 0);
+    let mut process = command.spawn().context("claude could not be started")?;
+    let group = process.id();
+    let stdin = process.stdin.take();
+    let stdout = process.stdout.take().ok_or_else(|| anyhow!("claude has no stdout"))?;
+    let (sender, facts) = channel();
+    let lines = sender.clone();
+    std::thread::spawn(move || {
+        for line in BufReader::new(stdout).split(b'\n') {
+            let Ok(bytes) = line else { break };
+            if lines.send(Read::Line(String::from_utf8_lossy(&bytes).into_owned())).is_err() {
+                return;
+            }
+        }
+        let _ = lines.send(Read::Eof);
+    });
+    // The one waitpid: the child's end is a fact of its own, not inferred from its stream.
+    std::thread::spawn(move || {
+        let _ = sender.send(Read::Exited(process.wait().ok().and_then(|status| status.code())));
+    });
+    Ok(Box::new(ClaudeProcess { stdin, facts, group }))
+}
+
+struct ClaudeProcess {
+    stdin: Option<ChildStdin>,
+    facts: Receiver<Read>,
+    /// The child's pid, which is its process group's id.
+    group: u32,
+}
+
+impl Child for ClaudeProcess {
+    fn write_line(&mut self, line: &str) {
+        if let Some(stdin) = self.stdin.as_mut() {
+            let _ = stdin.write_all(line.as_bytes()).and_then(|()| stdin.write_all(b"\n")).and_then(|()| stdin.flush());
+        }
+    }
+
+    fn close_input(&mut self) {
+        self.stdin = None;
+    }
+
+    fn next(&mut self, timeout: StdDuration) -> Read {
+        match self.facts.recv_timeout(timeout) {
+            Ok(fact) => fact,
+            Err(RecvTimeoutError::Timeout) => Read::Timeout,
+            Err(RecvTimeoutError::Disconnected) => Read::Eof,
+        }
+    }
+
+    fn stop(&mut self, how: Stop) -> bool {
+        // The `kill` program, not a libc call: the library names no FFI.
+        let signal = match how {
+            Stop::Interrupt => "INT",
+            Stop::Terminate => "TERM",
+            Stop::Kill => "KILL",
+        };
+        Command::new("kill").args(["-s", signal, "--", &format!("-{}", self.group)]).stderr(Stdio::null()).status().is_ok_and(|status| status.success())
+    }
+}
+
+mod decide;
+
+#[cfg(test)]
+mod tests;

crates/mind-body/src/self_run/claude-2.1.268-assistant-usage-records.jsonl

@@ -0,0 +1,12 @@
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNLSfEPL5Th8GSs7vkj", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 39033, "cache_read_input_tokens": 45175, "output_tokens": 73, "output_tokens_details": {"thinking_tokens": 22}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 39033, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 73, "cache_read_input_tokens": 45175, "cache_creation_input_tokens": 39033, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 39033}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "thinking"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNLSfEPL5Th8GSs7vkj", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 39033, "cache_read_input_tokens": 45175, "output_tokens": 73, "output_tokens_details": {"thinking_tokens": 22}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 39033, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 73, "cache_read_input_tokens": 45175, "cache_creation_input_tokens": 39033, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 39033}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNLcAxtUhNghfqBmEjA", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 16541, "cache_read_input_tokens": 84208, "output_tokens": 136, "output_tokens_details": {"thinking_tokens": 0}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 16541, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 136, "cache_read_input_tokens": 84208, "cache_creation_input_tokens": 16541, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 16541}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNLjffqfsnx2t6FFAh9", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 16423, "cache_read_input_tokens": 100749, "output_tokens": 282, "output_tokens_details": {"thinking_tokens": 0}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 16423, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 282, "cache_read_input_tokens": 100749, "cache_creation_input_tokens": 16423, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 16423}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNLjffqfsnx2t6FFAh9", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 16423, "cache_read_input_tokens": 100749, "output_tokens": 282, "output_tokens_details": {"thinking_tokens": 0}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 16423, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 282, "cache_read_input_tokens": 100749, "cache_creation_input_tokens": 16423, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 16423}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNMiXFAsUftWZLhShSJ", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 1162, "cache_read_input_tokens": 117172, "output_tokens": 303, "output_tokens_details": {"thinking_tokens": 26}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 1162, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 303, "cache_read_input_tokens": 117172, "cache_creation_input_tokens": 1162, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 1162}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "thinking"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNMiXFAsUftWZLhShSJ", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 1162, "cache_read_input_tokens": 117172, "output_tokens": 303, "output_tokens_details": {"thinking_tokens": 26}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 1162, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 303, "cache_read_input_tokens": 117172, "cache_creation_input_tokens": 1162, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 1162}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNN8DmkfNuiMD3mFFg1", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 8410, "cache_read_input_tokens": 118334, "output_tokens": 299, "output_tokens_details": {"thinking_tokens": 166}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 8410, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 299, "cache_read_input_tokens": 118334, "cache_creation_input_tokens": 8410, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 8410}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "thinking"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNN8DmkfNuiMD3mFFg1", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 8410, "cache_read_input_tokens": 118334, "output_tokens": 299, "output_tokens_details": {"thinking_tokens": 166}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 8410, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 299, "cache_read_input_tokens": 118334, "cache_creation_input_tokens": 8410, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 8410}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNNRZ46gcJVp99ra6zp", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 504, "cache_read_input_tokens": 126744, "output_tokens": 137, "output_tokens_details": {"thinking_tokens": 0}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 504, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 137, "cache_read_input_tokens": 126744, "cache_creation_input_tokens": 504, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 504}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "text"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNNRZ46gcJVp99ra6zp", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 504, "cache_read_input_tokens": 126744, "output_tokens": 137, "output_tokens_details": {"thinking_tokens": 0}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 504, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 137, "cache_read_input_tokens": 126744, "cache_creation_input_tokens": 504, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 504}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}
+{"type": "assistant", "parent_tool_use_id": null, "message": {"id": "msg_011CgHNNaiDHie99GoJBYBDt", "model": "claude-opus-5-5", "stop_reason": "tool_use", "usage": {"input_tokens": 2, "cache_creation_input_tokens": 2603, "cache_read_input_tokens": 127248, "output_tokens": 553, "output_tokens_details": {"thinking_tokens": 0}, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 2603, "ephemeral_5m_input_tokens": 0}, "inference_geo": "global", "iterations": [{"input_tokens": 2, "output_tokens": 553, "cache_read_input_tokens": 127248, "cache_creation_input_tokens": 2603, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 2603}, "type": "message"}], "speed": "standard", "fallback_credit": null}, "content": [{"type": "tool_use"}]}}

crates/mind-body/src/self_run/claude-2.1.268-stream-json-probe.jsonl

@@ -0,0 +1,3 @@
+{"type":"system","subtype":"init","cwd":"C:\\Users\\Meta\\AppData\\Local\\Temp\\claude\\F--Projects-CultLib\\21443208-468a-45eb-ab1d-a731c1c79772\\scratchpad\\imag-self-run\\p1","session_id":"7a1c2d3e-4b5a-5c6d-8e7f-0a1b2c3d4e5f","tools":["Task","Bash","CronCreate","CronDelete","CronList","DesignSync","Edit","EnterWorktree","ExitWorktree","Glob","Grep","ListAgents","Monitor","NotebookEdit","PowerShell","PushNotification","Read","ReportFindings","ScheduleWakeup","SendMessage","Skill","TaskCreate","TaskGet","TaskList","TaskOutput","TaskStop","TaskUpdate","ToolSearch","WebFetch","WebSearch","Workflow","Write"],"mcp_servers":[],"model":"claude-haiku-4-5-20251001","permissionMode":"default","slash_commands":["deep-research","design","design-sync","dataviz","update-config","verify","debug","code-review","simplify","batch","fewer-permission-prompts","doctor","loop","claude-api","workflow-authoring","run","run-skill-generator","plugin-authoring","advisor","agents","auto-mode-setup","autocompact","clear","color","compact","config","context","effort","fast","heapdump","init","mcp","import","model","__remote-workflow","workflow-launch-exec","plugin-types","reload-plugins","reload-skills","rename","security-review","usage","insights","recap","skill-doctor","goal","design-consent","design-revoke","list-agents","team-onboarding"],"terminal_slash_commands":["doctor","color","reload-plugins"],"apiKeySource":"none","claude_code_version":"2.1.268","output_style":"Eureka Self","agents":["claude","Explore","general-purpose","Plan"],"skills":["deep-research","design","design-sync","dataviz","update-config","verify","debug","code-review","simplify","batch","fewer-permission-prompts","doctor","loop","claude-api","workflow-authoring","run","run-skill-generator","plugin-authoring"],"plugins":[{"name":"plugin-authoring","path":"builtin","source":"plugin-authoring@builtin"}],"capabilities":["interrupt_receipt_v1","interrupt_cancel_queued_v1","msg_lifecycle_v1"],"analytics_disabled":false,"product_feedback_disabled":false,"uuid":"b8f04cea-6d04-476d-84e7-bd71a903691d","messaging_socket_path":"\\\\.\\pipe\\LOCAL\\cc-msg-08f60b361528ebff88fceca2e74f13fd","fast_mode_state":"off","fast_mode_disabled_reason":"sdk_opt_in_required","powershell_path":"C:\\WINDOWS\\System32\\WindowsPowerShell\\v1.0\\powershell.exe"}
+{"type":"assistant","message":{"diagnostics":null,"id":"3af6018f-a66e-4e87-a97a-8a534e41ac02","container":null,"model":"<synthetic>","role":"assistant","stop_details":null,"stop_reason":"stop_sequence","stop_sequence":"","type":"message","usage":{"output_tokens_details":null,"input_tokens":0,"output_tokens":0,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"server_tool_use":{"web_search_requests":0,"web_fetch_requests":0},"service_tier":null,"cache_creation":{"ephemeral_1h_input_tokens":0,"ephemeral_5m_input_tokens":0},"inference_geo":null,"iterations":null,"speed":null},"content":[{"type":"text","text":"Failed to authenticate: OAuth session expired and could not be refreshed"}],"context_management":null},"parent_tool_use_id":null,"session_id":"7a1c2d3e-4b5a-5c6d-8e7f-0a1b2c3d4e5f","uuid":"39600172-ebb4-4b39-9a7e-160b49591050","timestamp":"2026-10-22T15:23:04.474Z","error":"authentication_failed","is_api_error_message":true}
+{"duration_api_ms":0,"stop_reason":"stop_sequence","session_id":"7a1c2d3e-4b5a-5c6d-8e7f-0a1b2c3d4e5f","total_cost_usd":0,"usage":{"output_tokens_details":{"thinking_tokens":0},"input_tokens":0,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":0,"server_tool_use":{"web_search_requests":0,"web_fetch_requests":0},"service_tier":"standard","cache_creation":{"ephemeral_1h_input_tokens":0,"ephemeral_5m_input_tokens":0},"inference_geo":"","iterations":[],"speed":"standard"},"modelUsage":{},"permission_denials":[],"terminal_reason":"api_error","fast_mode_state":"off","fast_mode_disabled_reason":"sdk_opt_in_required","subagent_stats":{"spawned":0,"requested":{"background":0,"foreground":0,"unset":0},"started_in_background":0,"max_depth":0,"spawned_by_subagents":0,"completed":0,"failed":0,"killed":{"parent":0,"user":0,"system":0},"refused":{"depth_limit":0,"concurrency_limit":0,"budget":0},"by_type":{}},"is_error":true,"num_turns":1,"subtype":"success","api_error_status":null,"result":"Failed to authenticate: OAuth session expired and could not be refreshed","type":"result","duration_ms":222,"uuid":"41e16004-c8b3-4218-af94-69aa2fc52190","queued_turn_count":0,"result_index":0}

crates/mind-body/src/self_run/decide.rs

@@ -0,0 +1,185 @@
+//! The Self-run actuator's lifecycle decision, and nothing else.
+//!
+//! `decide(state, event, now) -> (state, action)` is the one owner of when the
+//! flush is asked for, when the stdin closes, when a signal goes out and when
+//! the run is over (ruling `eureka-body:ruling:self-run-decider`). It performs
+//! no I/O and reads no clock: the loop in `carry` turns operating-system facts
+//! into events, reads the clock once per event, and executes the action. A
+//! timing rule is pinned here, by tables and by an enumeration of event orders.
+//!
+//! The flush's turn begins at the first result after the flush was written (the
+//! turn already running ends there; the queued flush is the next one). The
+//! headroom counts only requests after that boundary, measured from the prefix
+//! the flush was written at, and the grace from the flush write (ruling
+//! `eureka-body:ruling:flush-always-gets-a-turn`, refined by the decider ruling).
+//! At most one action comes out of an event; a deadline that is already due
+//! yields its signal at the next event, and `next_deadline` says it is now.
+
+use std::collections::HashSet;
+
+use chrono::{DateTime, Duration, Utc};
+
+use super::{CEILING_TOKENS, Cause, ESCALATE_AFTER_S, EXIT_DRAIN_S, FLUSH_GRACE_S, FLUSH_HEADROOM_TOKENS, SelfRunOutcome, Stop};
+
+/// What the loop saw, already turned into a fact the decision can use.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) enum Event {
+    /// A request of the main thread: its message id and the context it read.
+    Request { id: String, prefix: u64 },
+    Result,
+    /// `waitpid` returned: the child's exit code, none when a signal ended it.
+    ChildExited(Option<i32>),
+    StdoutEof,
+    /// Time passed, or a line that no rule reads.
+    Tick,
+    /// What the operating system said of the signal the last action sent.
+    Signalled { how: Stop, delivered: bool },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) enum Action {
+    None,
+    WriteFlush,
+    CloseStdin,
+    /// To the child's whole process group.
+    Signal(Stop),
+    Finish(SelfRunOutcome),
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct Flush {
+    /// The prefix at which the flush was written: the headroom's base.
+    prefix: u64,
+    written: DateTime<Utc>,
+    /// A result has come since the write: the flush is the next turn.
+    turn_begun: bool,
+    /// A request of the flush's turn reached the headroom.
+    exhausted: bool,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub(super) struct State {
+    seen: HashSet<String>,
+    pub(super) requests: u64,
+    pub(super) prefix: u64,
+    result_seen: bool,
+    flush: Option<Flush>,
+    /// The last signal attempted and when.
+    stopping: Option<(Stop, DateTime<Utc>)>,
+    /// The strongest signal delivered.
+    delivered: Option<Stop>,
+    exited: Option<(Option<i32>, DateTime<Utc>)>,
+    eof: bool,
+    finished: bool,
+}
+
+fn outcome(state: &State, cause: Cause, code: Option<i32>) -> SelfRunOutcome {
+    SelfRunOutcome::Ended { cause, code, requests: state.requests, prefix: state.prefix }
+}
+
+/// How the run ended, by what was done to the child: the strongest signal delivered, else the flush, else nothing.
+fn cause(state: &State) -> Cause {
+    match (state.delivered, &state.flush) {
+        (Some(how), _) => Cause::Signalled(how),
+        (None, Some(_)) => Cause::FlushRequested,
+        (None, None) => Cause::Own,
+    }
+}
+
+/// The moment the loop must next be heard at, if any. A moment already past
+/// means an action is due now.
+pub(super) fn next_deadline(state: &State) -> Option<DateTime<Utc>> {
+    if state.finished {
+        return None;
+    }
+    if let Some((_, exited)) = state.exited {
+        return (!state.eof).then(|| exited + Duration::seconds(EXIT_DRAIN_S));
+    }
+    match (&state.stopping, &state.flush) {
+        (Some((_, sent)), _) => Some(*sent + Duration::seconds(ESCALATE_AFTER_S)),
+        (None, Some(flush)) => Some(flush.written + Duration::seconds(FLUSH_GRACE_S)),
+        (None, None) => None,
+    }
+}
+
+pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (State, Action) {
+    if state.finished {
+        return (state, Action::None);
+    }
+    let mut action = Action::None;
+    match event {
+        Event::Request { id, prefix } => {
+            if state.seen.insert(id) {
+                state.requests += 1;
+                state.prefix = prefix;
+                if let Some(flush) = state.flush.as_mut()
+                    && flush.turn_begun
+                    && flush.prefix.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit)
+                {
+                    flush.exhausted = true;
+                }
+                if state.flush.is_none() && !state.result_seen && state.exited.is_none() && prefix >= CEILING_TOKENS {
+                    state.flush = Some(Flush { prefix, written: now, turn_begun: false, exhausted: false });
+                    action = Action::WriteFlush;
+                }
+            }
+        }
+        Event::Result => {
+            if !state.result_seen {
+                state.result_seen = true;
+                if let Some(flush) = state.flush.as_mut() {
+                    flush.turn_begun = true;
+                }
+                action = Action::CloseStdin;
+            }
+        }
+        Event::ChildExited(code) => {
+            state.exited.get_or_insert((code, now));
+        }
+        Event::StdoutEof => state.eof = true,
+        Event::Tick => {}
+        Event::Signalled { how, delivered } => {
+            if delivered {
+                state.delivered = Some(how);
+            }
+        }
+    }
+
+    // A child that has exited is never written to or signalled: the run is over
+    // once its stream has ended, or once the stream has had its drain.
+    if let Some((code, exited)) = state.exited {
+        if state.eof || now >= exited + Duration::seconds(EXIT_DRAIN_S) {
+            state.finished = true;
+            let outcome = outcome(&state, cause(&state), code);
+            return (state, Action::Finish(outcome));
+        }
+        return (state, Action::None);
+    }
+    if action != Action::None {
+        return (state, action);
+    }
+    if let Some((Stop::Kill, sent)) = state.stopping {
+        // SIGKILL went out and the child still has not been reaped: stop waiting for it.
+        if now >= sent + Duration::seconds(ESCALATE_AFTER_S) {
+            state.finished = true;
+            let cause = if state.delivered == Some(Stop::Kill) { Cause::Signalled(Stop::Kill) } else { Cause::Unstoppable };
+            let outcome = outcome(&state, cause, None);
+            return (state, Action::Finish(outcome));
+        }
+        return (state, Action::None);
+    }
+    let how = match (&state.stopping, &state.flush) {
+        (Some((last, sent)), _) => (now >= *sent + Duration::seconds(ESCALATE_AFTER_S)).then(|| last.next()),
+        (None, Some(flush)) => (flush.exhausted || now >= flush.written + Duration::seconds(FLUSH_GRACE_S)).then_some(Stop::Interrupt),
+        (None, None) => None,
+    };
+    match how {
+        Some(how) => {
+            state.stopping = Some((how, now));
+            (state, Action::Signal(how))
+        }
+        None => (state, Action::None),
+    }
+}
+#[cfg(test)]
+mod tests;

crates/mind-body/src/self_run/decide/tests.rs

@@ -0,0 +1,443 @@
+//! `decide` is pure, so its rules are pinned here with no child, no mind and no
+//! clock: tables, one row per boundary, and an enumeration of event orders that
+//! asserts the invariants against an oracle kept from the event history alone.
+
+use super::*;
+use crate::testkit::at;
+
+fn t(seconds: i64) -> DateTime<Utc> {
+    at(seconds * 1000)
+}
+
+fn req(id: &str, prefix: u64) -> Event {
+    Event::Request { id: id.to_string(), prefix }
+}
+
+fn feed(state: State, event: Event, seconds: i64) -> (State, Action) {
+    decide(state, event, t(seconds))
+}
+
+/// Runs the events in order, each at its own second, and keeps every action.
+fn run(events: Vec<(Event, i64)>) -> (State, Vec<Action>) {
+    let mut state = State::default();
+    let mut actions = Vec::new();
+    for (event, seconds) in events {
+        let (next, action) = feed(state, event, seconds);
+        state = next;
+        actions.push(action);
+    }
+    (state, actions)
+}
+
+fn last(events: Vec<(Event, i64)>) -> Action {
+    run(events).1.pop().unwrap()
+}
+
+fn delivered(how: Stop) -> Event {
+    Event::Signalled { how, delivered: true }
+}
+
+fn ended(cause: Cause, code: Option<i32>, requests: u64, prefix: u64) -> Action {
+    Action::Finish(SelfRunOutcome::Ended { cause, code, requests, prefix })
+}
+
+const GRACE: i64 = FLUSH_GRACE_S;
+const ESC: i64 = ESCALATE_AFTER_S;
+const LIMIT: u64 = CEILING_TOKENS + FLUSH_HEADROOM_TOKENS;
+
+#[test]
+fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or_for_a_repeated_request() {
+    assert_eq!(last(vec![(req("a", CEILING_TOKENS - 1), 0)]), Action::None);
+    assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0)]), Action::WriteFlush);
+    assert_eq!(last(vec![(req("a", CEILING_TOKENS + 1), 0)]), Action::WriteFlush);
+    // Once.
+    assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", CEILING_TOKENS + 5), 1)]), Action::None);
+    // A result first: the child is finishing, there is nothing to flush.
+    assert_eq!(last(vec![(Event::Result, 0), (req("a", CEILING_TOKENS), 1)]), Action::None);
+    // A child that is gone is not written to.
+    assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (req("a", CEILING_TOKENS), 1)]), Action::None);
+    // The same message id is the same request again: it neither counts nor decides.
+    let (state, actions) = run(vec![(req("a", 10), 0), (req("a", CEILING_TOKENS), 1)]);
+    assert_eq!((actions[1].clone(), state.requests, state.prefix), (Action::None, 1, 10));
+}
+
+#[test]
+fn the_input_closes_once_at_the_first_result() {
+    assert_eq!(last(vec![(Event::Result, 0)]), Action::CloseStdin);
+    assert_eq!(last(vec![(Event::Result, 0), (Event::Result, 1)]), Action::None);
+    assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (Event::Result, 1)]), Action::CloseStdin);
+}
+
+#[test]
+fn the_grace_runs_from_the_flush_write_to_the_second_and_no_later_request_moves_it() {
+    let flushed = |rest: Vec<(Event, i64)>| last([vec![(req("a", 10), 0), (req("b", CEILING_TOKENS), 100)], rest].concat());
+    assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE - 1)]), Action::None);
+    assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE)]), Action::Signal(Stop::Interrupt));
+    assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE + 5000)]), Action::Signal(Stop::Interrupt));
+    // Requests after the flush, however late, do not refresh it.
+    assert_eq!(flushed(vec![(req("c", CEILING_TOKENS + 1), 100 + GRACE - 10), (Event::Tick, 100 + GRACE)]), Action::Signal(Stop::Interrupt));
+    // Not from the first request, not from the start.
+    assert_eq!(last(vec![(req("a", 10), 0), (req("b", CEILING_TOKENS), 100), (Event::Tick, GRACE)]), Action::None);
+    // The deadline the loop waits for is the same moment.
+    let (state, _) = run(vec![(req("a", CEILING_TOKENS), 100)]);
+    assert_eq!(next_deadline(&state), Some(t(100 + GRACE)));
+    assert_eq!(next_deadline(&State::default()), None);
+}
+
+#[test]
+fn the_headroom_counts_the_requests_after_the_first_result_after_the_flush_from_the_prefix_of_the_flush() {
+    let after_result = |request: u64| last(vec![(req("a", CEILING_TOKENS), 0), (Event::Result, 1), (req("b", request), 2)]);
+    assert_eq!(after_result(LIMIT - 1), Action::None);
+    assert_eq!(after_result(LIMIT), Action::Signal(Stop::Interrupt));
+    assert_eq!(after_result(LIMIT + 1), Action::Signal(Stop::Interrupt));
+    // Measured from the flush's own prefix, not from the ceiling.
+    let high = |request: u64| last(vec![(req("a", 400_000), 0), (Event::Result, 1), (req("b", request), 2)]);
+    assert_eq!((high(459_999), high(460_000)), (Action::None, Action::Signal(Stop::Interrupt)));
+    // The turn already running is not the flush's turn: before a result no request counts.
+    assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", LIMIT + 400_000), 1)]), Action::None);
+    // A big request before the result is forgotten by the boundary; only those after it count.
+    assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", LIMIT + 1), 1), (Event::Result, 2), (req("c", LIMIT - 1), 3)]), Action::None);
+    // The result before the flush is not the boundary: the flush cannot be written then.
+    assert_eq!(last(vec![(Event::Result, 0), (req("a", CEILING_TOKENS), 1), (req("b", 900_000), 2)]), Action::None);
+    // A prefix that saturates leaves no headroom to count: only the grace is left.
+    let saturated = |rest: Vec<(Event, i64)>| last([vec![(req("a", u64::MAX), 0), (Event::Result, 1), (req("b", u64::MAX), 2)], rest].concat());
+    assert_eq!(saturated(vec![]), Action::None);
+    assert_eq!(saturated(vec![(Event::Tick, GRACE)]), Action::Signal(Stop::Interrupt));
+    // A request counted past the headroom stays counted: a later smaller one does not take the interrupt back.
+    let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 0), (Event::Result, 1), (req("b", LIMIT), 2)]);
+    assert_eq!(actions[2], Action::Signal(Stop::Interrupt));
+    assert_eq!(next_deadline(&state), Some(t(2 + ESC)));
+}
+
+#[test]
+fn the_flush_pass_never_signals_whatever_the_clock_does() {
+    // Two events apart by more than the grace: the write is its own pass, the signal comes in the next.
+    let (state, action) = feed(State::default(), req("a", CEILING_TOKENS), 0);
+    assert_eq!(action, Action::WriteFlush);
+    assert_eq!(feed(state, Event::Tick, 10 * GRACE).1, Action::Signal(Stop::Interrupt));
+    // A request that both crosses the ceiling and is far past the headroom writes the flush and nothing else.
+    assert_eq!(last(vec![(req("a", u64::MAX), 0)]), Action::WriteFlush);
+}
+
+#[test]
+fn each_signal_follows_the_last_by_the_bound_and_the_run_ends_when_the_kill_has_had_its_time() {
+    let interrupted = vec![(req("a", CEILING_TOKENS), 0), (Event::Tick, GRACE), (delivered(Stop::Interrupt), GRACE)];
+    let then = |rest: Vec<(Event, i64)>| last([interrupted.clone(), rest].concat());
+    assert_eq!(then(vec![(Event::Tick, GRACE + ESC - 1)]), Action::None);
+    assert_eq!(then(vec![(Event::Tick, GRACE + ESC)]), Action::Signal(Stop::Terminate));
+    // The bound runs from the last signal, not from the first or from the flush.
+    let late_term = [interrupted.clone(), vec![(Event::Tick, GRACE + ESC + 30), (delivered(Stop::Terminate), GRACE + ESC + 30)]].concat();
+    let kill_at = |seconds: i64| last([late_term.clone(), vec![(Event::Tick, seconds)]].concat());
+    assert_eq!(kill_at(GRACE + 2 * ESC + 29), Action::None);
+    assert_eq!(kill_at(GRACE + 2 * ESC + 30), Action::Signal(Stop::Kill));
+    // After the kill: the child may still be reaped until the bound, then the run ends without it.
+    let killed = [late_term, vec![(Event::Tick, GRACE + 2 * ESC + 30), (delivered(Stop::Kill), GRACE + 2 * ESC + 30)]].concat();
+    let after = |seconds: i64| last([killed.clone(), vec![(Event::Tick, seconds)]].concat());
+    let end = GRACE + 3 * ESC + 30;
+    assert_eq!(after(end - 1), Action::None);
+    assert_eq!(after(end), ended(Cause::Signalled(Stop::Kill), None, 1, CEILING_TOKENS));
+    // A kill that was not delivered, with the child still there at the bound: Unstoppable, whatever else got through.
+    let refused_kill = [killed[..killed.len() - 1].to_vec(), vec![(Event::Signalled { how: Stop::Kill, delivered: false }, GRACE + 2 * ESC + 30)]].concat();
+    assert_eq!(last([refused_kill, vec![(Event::Tick, end)]].concat()), ended(Cause::Unstoppable, None, 1, CEILING_TOKENS));
+}
+
+#[test]
+fn a_kill_that_was_never_delivered_ends_the_run_as_unstoppable_when_nothing_was_delivered_at_all() {
+    let all_refused = |how: Stop| Event::Signalled { how, delivered: false };
+    let (state, _) = run(vec![
+        (req("a", CEILING_TOKENS), 0),
+        (Event::Tick, GRACE),
+        (all_refused(Stop::Interrupt), GRACE),
+        (Event::Tick, GRACE + ESC),
+        (all_refused(Stop::Terminate), GRACE + ESC),
+        (Event::Tick, GRACE + 2 * ESC),
+        (all_refused(Stop::Kill), GRACE + 2 * ESC),
+    ]);
+    assert_eq!(next_deadline(&state), Some(t(GRACE + 3 * ESC)));
+    assert_eq!(feed(state.clone(), Event::Tick, GRACE + 3 * ESC - 1).1, Action::None);
+    assert_eq!(feed(state, Event::Tick, GRACE + 3 * ESC).1, ended(Cause::Unstoppable, None, 1, CEILING_TOKENS));
+}
+
+#[test]
+fn a_child_that_ends_while_the_kill_was_in_flight_ends_the_run_by_its_own_cause_not_unstoppable() {
+    // The kill was refused because the child was already gone; the reap is on its way.
+    let (state, _) = run(vec![
+        (req("a", CEILING_TOKENS), 0),
+        (Event::Tick, GRACE),
+        (delivered(Stop::Interrupt), GRACE),
+        (Event::Tick, GRACE + ESC),
+        (Event::Tick, GRACE + 2 * ESC),
+        (Event::Signalled { how: Stop::Kill, delivered: false }, GRACE + 2 * ESC),
+        (Event::ChildExited(Some(0)), GRACE + 2 * ESC + 1),
+    ]);
+    assert_eq!(feed(state, Event::StdoutEof, GRACE + 2 * ESC + 1).1, ended(Cause::Signalled(Stop::Interrupt), Some(0), 1, CEILING_TOKENS));
+}
+
+#[test]
+fn a_child_that_has_exited_is_never_signalled_and_the_run_ends_with_its_stream_or_after_the_drain() {
+    // A flush asked for, the grace long past, the child already gone: no signal, only the end.
+    let gone = vec![(req("a", CEILING_TOKENS), 0), (Event::ChildExited(Some(0)), 10)];
+    let tick = |seconds: i64| last([gone.clone(), vec![(Event::Tick, seconds)]].concat());
+    assert_eq!(tick(11), Action::None);
+    assert_eq!(tick(10 + EXIT_DRAIN_S - 1), Action::None);
+    assert_eq!(tick(10 + EXIT_DRAIN_S), ended(Cause::FlushRequested, Some(0), 1, CEILING_TOKENS));
+    assert_eq!(tick(10 + GRACE * 4), ended(Cause::FlushRequested, Some(0), 1, CEILING_TOKENS));
+    // With the stream ended too, it ends at once, in either order.
+    assert_eq!(last([gone.clone(), vec![(Event::StdoutEof, 11)]].concat()), ended(Cause::FlushRequested, Some(0), 1, CEILING_TOKENS));
+    assert_eq!(last(vec![(Event::StdoutEof, 0), (Event::ChildExited(None), 1)]), ended(Cause::Own, None, 0, 0));
+    // The end of the stream alone is not the end of the child.
+    assert_eq!(last(vec![(Event::StdoutEof, 0), (Event::Tick, 5)]), Action::None);
+    // No flush, a signal's cause named by what was delivered, a code from the exit.
+    let (state, _) = run(vec![(req("a", CEILING_TOKENS), 0), (Event::Tick, GRACE), (delivered(Stop::Interrupt), GRACE), (Event::ChildExited(Some(130)), GRACE + 1)]);
+    assert_eq!(feed(state, Event::StdoutEof, GRACE + 1).1, ended(Cause::Signalled(Stop::Interrupt), Some(130), 1, CEILING_TOKENS));
+    // The deadline of a child that has gone is the end of its drain.
+    let (state, _) = run(vec![(Event::ChildExited(Some(0)), 7)]);
+    assert_eq!(next_deadline(&state), Some(t(7 + EXIT_DRAIN_S)));
+}
+
+#[test]
+fn without_a_flush_nothing_is_ever_signalled_and_a_finished_run_decides_nothing_more() {
+    let quiet = vec![(req("a", 10), 0), (Event::Result, 1), (req("b", 900_000), 2), (Event::Tick, 100_000)];
+    assert!(run(quiet).1.iter().all(|action| !matches!(action, Action::Signal(_))));
+    let (state, actions) = run(vec![(Event::ChildExited(Some(3)), 0), (Event::StdoutEof, 0)]);
+    assert_eq!(actions[1], ended(Cause::Own, Some(3), 0, 0));
+    assert_eq!(next_deadline(&state), None);
+    assert_eq!(feed(state, Event::Tick, 10_000).1, Action::None);
+}
+
+// ---- every order of events ------------------------------------------------------
+
+/// What can happen next, each at a time after the last.
+#[derive(Clone, Copy, Debug)]
+enum Sym {
+    Low,
+    AtCeiling,
+    Huge,
+    Repeat,
+    Result,
+    Exit(Option<i32>),
+    Eof,
+    Soon,
+    Bound,
+    Grace,
+}
+
+const ALPHABET: [Sym; 11] = [
+    Sym::Low,
+    Sym::AtCeiling,
+    Sym::Huge,
+    Sym::Repeat,
+    Sym::Result,
+    Sym::Exit(Some(0)),
+    Sym::Exit(None),
+    Sym::Eof,
+    Sym::Soon,
+    Sym::Bound,
+    Sym::Grace,
+];
+
+/// Which signals the operating system lets through.
+#[derive(Clone, Copy, Debug)]
+enum Policy {
+    All,
+    Nothing,
+    NoKill,
+}
+
+/// The facts of the history, kept apart from `State`: what the invariants are asked of.
+#[derive(Default)]
+struct Oracle {
+    ids: Vec<String>,
+    flush: Option<(i64, u64)>,
+    flushes: u32,
+    closes: u32,
+    boundary: bool,
+    exceeded: bool,
+    result_seen: bool,
+    exited: Option<i64>,
+    eof: bool,
+    signals: Vec<(Stop, i64)>,
+    finished: bool,
+}
+
+fn check_step(oracle: &mut Oracle, event: &Event, now: i64, action: &Action) {
+    assert!(!oracle.finished || *action == Action::None, "an action after the run ended: {action:?} on {event:?}");
+    let flush_before = oracle.flush;
+    let (was_result, was_exited) = (oracle.result_seen, oracle.exited.is_some());
+    match event {
+        Event::Request { id, prefix } => {
+            if !oracle.ids.contains(id) {
+                oracle.ids.push(id.clone());
+                if oracle.boundary
+                    && let Some((_, base)) = oracle.flush
+                    && base.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| *prefix >= limit)
+                {
+                    oracle.exceeded = true;
+                }
+            }
+        }
+        Event::Result => {
+            if oracle.flush.is_some() {
+                oracle.boundary = true;
+            }
+            oracle.result_seen = true;
+        }
+        Event::ChildExited(_) => {
+            oracle.exited.get_or_insert(now);
+        }
+        Event::StdoutEof => oracle.eof = true,
+        Event::Tick | Event::Signalled { .. } => {}
+    }
+    match action {
+        Action::None => {}
+        Action::WriteFlush => {
+            oracle.flushes += 1;
+            let Event::Request { prefix, .. } = event else { panic!("a flush on {event:?}") };
+            assert!(*prefix >= CEILING_TOKENS && !was_result && !was_exited && flush_before.is_none(), "flush at {prefix} after {was_result} {was_exited}");
+            oracle.flush = Some((now, *prefix));
+        }
+        Action::CloseStdin => {
+            oracle.closes += 1;
+            assert!(matches!(event, Event::Result) && !was_result && !was_exited);
+        }
+        Action::Signal(how) => {
+            let (flushed_at, _) = flush_before.expect("a signal before the flush was written");
+            assert!(oracle.exited.is_none(), "a signal to a child that already exited");
+            match oracle.signals.last() {
+                None => {
+                    assert_eq!(*how, Stop::Interrupt);
+                    assert!(now >= flushed_at + GRACE || oracle.exceeded, "interrupt before the flush had its turn: now {now}, flush {flushed_at}, exceeded {}", oracle.exceeded);
+                    assert!(now > flushed_at || oracle.exceeded, "interrupt in the pass that wrote the flush");
+                }
+                Some((last, sent)) => {
+                    assert_eq!(*how, last.next(), "escalation out of order");
+                    assert_ne!(*last, Stop::Kill, "a signal after the kill");
+                    assert!(now >= sent + ESC, "escalated after {} s, bound {ESC}", now - sent);
+                }
+            }
+            oracle.signals.push((*how, now));
+        }
+        Action::Finish(outcome) => {
+            oracle.finished = true;
+            let SelfRunOutcome::Ended { cause, code, .. } = outcome else { panic!("{outcome:?}") };
+            match oracle.exited {
+                Some(at) => assert!(oracle.eof || now >= at + EXIT_DRAIN_S, "finished before the stream or the drain"),
+                None => {
+                    let (last, sent) = *oracle.signals.last().expect("finished with the child alive and no signal sent");
+                    assert!(last == Stop::Kill && now >= sent + ESC, "finished with the child alive");
+                    assert_eq!(*code, None);
+                    assert!(matches!(cause, Cause::Signalled(Stop::Kill) | Cause::Unstoppable), "{cause:?}");
+                }
+            }
+            if matches!(cause, Cause::Own) {
+                assert!(oracle.flush.is_none() && oracle.signals.is_empty());
+            }
+            if matches!(cause, Cause::FlushRequested) {
+                assert!(oracle.flush.is_some());
+            }
+        }
+    }
+    assert!(oracle.flushes <= 1 && oracle.closes <= 1);
+    assert!(oracle.signals.len() <= 3);
+}
+
+fn play(symbols: &[Sym], policy: Policy) {
+    let mut state = State::default();
+    let mut oracle = Oracle::default();
+    let mut now = 0i64;
+    let mut unique = 0u32;
+    let mut events: Vec<Event> = Vec::new();
+    for symbol in symbols {
+        let (event, dt) = match symbol {
+            Sym::Low => (req_new(&mut unique, 1_000), 1),
+            Sym::AtCeiling => (req_new(&mut unique, CEILING_TOKENS), 1),
+            Sym::Huge => (req_new(&mut unique, LIMIT + 400_000), 1),
+            Sym::Repeat => (req("u1", CEILING_TOKENS + 7), 1),
+            Sym::Result => (Event::Result, 1),
+            Sym::Exit(code) => (Event::ChildExited(*code), 1),
+            Sym::Eof => (Event::StdoutEof, 1),
+            Sym::Soon => (Event::Tick, 1),
+            Sym::Bound => (Event::Tick, ESC + 1),
+            Sym::Grace => (Event::Tick, GRACE + 1),
+        };
+        now += dt;
+        events.push(event.clone());
+        let mut next = Some(event);
+        while let Some(event) = next.take() {
+            let (after, action) = decide(state, event.clone(), t(now));
+            state = after;
+            check_step(&mut oracle, &event, now, &action);
+            if let Action::Signal(how) = action {
+                let delivered = match policy {
+                    Policy::All => true,
+                    Policy::Nothing => false,
+                    Policy::NoKill => how != Stop::Kill,
+                };
+                next = Some(Event::Signalled { how, delivered });
+            }
+        }
+    }
+    // No spinning: a deadline that has come is never answered with nothing.
+    if let Some(deadline) = next_deadline(&state) {
+        let at = deadline.max(t(now));
+        let (_, action) = decide(state.clone(), Event::Tick, at);
+        assert_ne!(action, Action::None, "a due deadline gave no action: {symbols:?} {policy:?}");
+    }
+    // Once the child has exited and its stream has ended, the run is over, whatever came before.
+    for event in [Event::ChildExited(Some(0)), Event::StdoutEof] {
+        now += 1;
+        let mut next = Some(event);
+        while let Some(event) = next.take() {
+            let (after, action) = decide(state, event.clone(), t(now));
+            state = after;
+            check_step(&mut oracle, &event, now, &action);
+            if let Action::Signal(how) = action {
+                next = Some(Event::Signalled { how, delivered: true });
+            }
+        }
+    }
+    assert!(oracle.finished, "no end after the exit and the end of the stream: {symbols:?} {policy:?} {events:?}");
+}
+
+fn req_new(unique: &mut u32, prefix: u64) -> Event {
+    *unique += 1;
+    req(&format!("u{unique}"), prefix)
+}
+
+fn enumerate(length: usize, symbols: &mut Vec<Sym>, policy: Policy) {
+    if symbols.len() == length {
+        play(symbols, policy);
+        return;
+    }
+    for symbol in ALPHABET {
+        symbols.push(symbol);
+        enumerate(length, symbols, policy);
+        symbols.pop();
+    }
+}
+
+#[test]
+fn every_order_of_up_to_five_events_keeps_the_invariants_under_every_policy_of_delivery() {
+    for policy in [Policy::All, Policy::Nothing, Policy::NoKill] {
+        for length in 0..=5 {
+            enumerate(length, &mut Vec::new(), policy);
+        }
+    }
+}
+
+#[test]
+fn a_long_random_order_of_events_keeps_the_invariants_and_always_ends_once_the_child_is_gone() {
+    // xorshift: the same orders every run.
+    let mut seed = 0x2545_F491_4F6C_DD1Du64;
+    let mut next = || {
+        seed ^= seed << 13;
+        seed ^= seed >> 7;
+        seed ^= seed << 17;
+        seed
+    };
+    for round in 0..40_000 {
+        let length = 6 + (next() % 30) as usize;
+        let symbols: Vec<Sym> = (0..length).map(|_| ALPHABET[(next() % ALPHABET.len() as u64) as usize]).collect();
+        play(&symbols, [Policy::All, Policy::Nothing, Policy::NoKill][round % 3]);
+    }
+}

crates/mind-body/tests/cli.rs

@@ -375,3 +375,49 @@ fn the_binary_hold_fails_with_status_2_when_systemctl_does() {
     assert!(stderr.contains("units were not stopped") && stderr.contains("exit status 5"), "{stderr}");
     assert!(!String::from_utf8_lossy(&output.stdout).contains("stopped"));
 }
+
+// ---- the arguments every organ binary shares ----
+
+mod shared_arguments {
+    use mind_body::cli::instance_and_endpoint;
+
+    const USAGE: &str = "usage: test-organ --instance <slug>";
+
+    fn parse(args: &[&str], env: Option<&str>) -> anyhow::Result<(eureka_pipeline::Slug, std::net::SocketAddr)> {
+        let args: Vec<String> = args.iter().map(|a| a.to_string()).collect();
+        instance_and_endpoint(&args, env.map(str::to_string), USAGE)
+    }
+
+    #[test]
+    fn the_flag_names_the_endpoint_and_wins_over_the_environment_which_is_the_fallback() {
+        let (instance, address) = parse(&["--instance", "yggdrasil", "--endpoint", "rudp://10.77.0.1:17872"], Some("rudp://127.0.0.1:1")).unwrap();
+        assert_eq!((instance.0.as_str(), address.to_string().as_str()), ("yggdrasil", "10.77.0.1:17872"));
+        let (_, address) = parse(&["--endpoint", "rudp://10.77.0.1:17872", "--instance", "yggdrasil"], None).unwrap();
+        assert_eq!(address.to_string(), "10.77.0.1:17872", "flag order does not matter");
+        let (_, address) = parse(&["--instance", "yggdrasil"], Some("rudp://127.0.0.1:9")).unwrap();
+        assert_eq!(address.to_string(), "127.0.0.1:9");
+    }
+
+    #[test]
+    fn every_refusal_is_fixed_text_that_echoes_no_value() {
+        let canary = "canary-value-91e7";
+        let cases: Vec<(Vec<&str>, Option<&str>, &str)> = vec![
+            (vec![], Some("rudp://127.0.0.1:9"), USAGE),
+            (vec!["--instance"], None, USAGE),
+            (vec!["--instance", "a", "--instance", canary], Some("rudp://127.0.0.1:9"), USAGE),
+            (vec!["--instance", "a", "--endpoint", "rudp://127.0.0.1:9", "--endpoint", canary], None, USAGE),
+            (vec!["--instance", "a", canary], Some("rudp://127.0.0.1:9"), USAGE),
+            (vec!["--endpoint", "rudp://127.0.0.1:9"], None, USAGE),
+            (vec!["--instance", "Canary/../x"], Some("rudp://127.0.0.1:9"), "the instance is not a slug"),
+            (vec!["--instance", "a"], None, "no daemon endpoint: --endpoint or HUGINN_ENDPOINT"),
+            (vec!["--instance", "a", "--endpoint", canary], None, "the endpoint is not rudp://<ip>:<port>"),
+            (vec!["--instance", "a", "--endpoint", "http://127.0.0.1:9"], None, "the endpoint is not rudp://<ip>:<port>"),
+            (vec!["--instance", "a", "--endpoint", "rudp://not-an-address"], None, "the endpoint is not rudp://<ip>:<port>"),
+            (vec!["--instance", "a"], Some(canary), "the endpoint is not rudp://<ip>:<port>"),
+        ];
+        for (args, env, message) in cases {
+            let error = parse(&args, env).unwrap_err();
+            assert_eq!(format!("{error:#}"), message, "{args:?} {env:?}");
+        }
+    }
+}

crates/mind-body/tests/public_surface.txt

@@ -2,7 +2,9 @@ mind_body: mod cli
 mind_body: mod control
 mind_body: mod launch
 mind_body: mod queue
+mind_body: mod self_run
 mind_body: mod wake
+mind_body::cli: fn instance_and_endpoint (args : & [String] , env_endpoint : Option < String > , usage : & str) -> Result < (Slug , SocketAddr) >
 mind_body::cli: fn operator_name (sudo_user : Option < String > , user : Option < String >) -> String
 mind_body::cli: fn run (args : & [String] , root : & Path , set_by : & str , now : DateTime < Utc > , out : & mut impl Write , stop_units : & mut dyn FnMut (& Slug) -> Result < () > ,) -> Result < () >
 mind_body::control: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum HeldReason { BrakeAbsent , BrakeHeld , DialAbsent , DialOutOfBounds , Undecodable , }
@@ -71,6 +73,34 @@ mind_body::queue: impl MindPort for HuginnMind < '_ >
 mind_body::queue: impl< 'a > HuginnMind < 'a >
 mind_body::queue: pub trait MindPort { fn instance (& self) -> & Slug ; fn query (& self , selection : & Selection) -> Result < PipelineSelectionPage > ; fn admit (& self , agent : & str , session : & str , documents : Vec < PipelineDocument >) -> Result < PipelineAdmissionOutcome > ; }
 mind_body::queue: struct HuginnMind< 'a >  pub-fields[]
+mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq , PartialOrd , Ord)] pub enum Stop { Interrupt , Terminate , Kill , }
+mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Cause { Own , FlushRequested , Signalled (Stop) , Unstoppable , }
+mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Read { Line (String) , Timeout , Eof , Exited (Option < i32 >) , }
+mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum SelfRunOutcome { NoRun , Ended { cause : Cause , code : Option < i32 > , requests : u64 , prefix : u64 } , }
+mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Usage { Request { id : String , prefix : u64 } , Result , Other , Unreadable , }
+mind_body::self_run: const AGENT : & str
+mind_body::self_run: const CEILING_TOKENS : u64
+mind_body::self_run: const ESCALATE_AFTER_S : i64
+mind_body::self_run: const EXIT_DRAIN_S : i64
+mind_body::self_run: const FLUSH_GRACE_S : i64
+mind_body::self_run: const FLUSH_HEADROOM_TOKENS : u64
+mind_body::self_run: const FLUSH_MESSAGE : & str
+mind_body::self_run: const MAX_TURNS : u32
+mind_body::self_run: const SESSION_NAMESPACE : Uuid
+mind_body::self_run: fn carry (mind : & dyn MindPort , clock : & dyn Clock , spawner : & dyn Spawner , out : & mut dyn Write) -> Result < SelfRunOutcome >
+mind_body::self_run: fn claude_args (session : Uuid , budget_usd : & str) -> Vec < String >
+mind_body::self_run: fn read_usage (line : & str) -> Usage
+mind_body::self_run: fn session_id (run : & PipelineRef) -> Uuid
+mind_body::self_run: fn start_message (instance : & Slug , run : & PipelineRef) -> String
+mind_body::self_run: impl Child for ClaudeProcess
+mind_body::self_run: impl SelfRunOutcome
+mind_body::self_run: impl SelfRunOutcome: fn exit_status (& self) -> i32
+mind_body::self_run: impl Spawner for ClaudeSpawner
+mind_body::self_run: impl Stop
+mind_body::self_run: impl fmt :: Display for SelfRunOutcome
+mind_body::self_run: pub trait Child { fn write_line (& mut self , line : & str) ; fn close_input (& mut self) ; fn next (& mut self , timeout : StdDuration) -> Read ; fn stop (& mut self , how : Stop) -> bool ; }
+mind_body::self_run: pub trait Spawner { fn spawn (& self , args : & [String]) -> Result < Box < dyn Child > > ; }
+mind_body::self_run: struct ClaudeSpawner  pub-fields[]
 mind_body::wake: # [derive (Clone , Debug , PartialEq , Eq)] pub enum WakeOutcome { Held { reason : HeldReason } , NotYet { until : DateTime < Utc > } , Busy { run : PipelineRef } , Launched { run : PipelineRef } , Refused (Declined) , }
 mind_body::wake: const AGENT : & str
 mind_body::wake: fn cycle (ports : & Ports , control : & impl ControlSource) -> Result < WakeOutcome >

crates/mind-body/tests/self_run.rs

@@ -0,0 +1,281 @@
+//! The `mind-self-run` binary as a process: its arguments, its exit codes and
+//! what it never echoes. The carrying itself is tested in the library against a
+//! real mind and a scripted child. One test, `#[ignore]`d outside a container,
+//! puts the real binary, the real daemon and a fake `claude` first on `PATH`
+//! together, which is where the real child's argument list and stream can be
+//! seen end to end.
+
+// The test hands the binary its environment, and reads the one variable the
+// harness gives it. That is test plumbing: it reaches no library code.
+#![allow(clippy::disallowed_methods, reason = "the suite sets the binary's environment")]
+
+use std::process::{Command, Output};
+
+const NO_SUCH_INSTANCE: &str = "self-run-test-no-such-instance";
+
+fn self_run(args: &[&str]) -> Output {
+    Command::new(env!("CARGO_BIN_EXE_mind-self-run")).args(args).env_remove("HUGINN_ENDPOINT").output().unwrap()
+}
+
+fn self_run_with_env_endpoint(args: &[&str], endpoint: &str) -> Output {
+    Command::new(env!("CARGO_BIN_EXE_mind-self-run")).args(args).env("HUGINN_ENDPOINT", endpoint).output().unwrap()
+}
+
+fn all_output(output: &Output) -> String {
+    format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr))
+}
+
+#[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],
+        vec![canary],
+        vec!["--resume", canary],
+    ] {
+        let output = self_run(&args);
+        assert_eq!(output.status.code(), Some(2), "{args:?}");
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        assert!(stderr.contains("usage: mind-self-run"), "{args:?}: {stderr}");
+        assert!(!all_output(&output).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 = self_run(&["--instance", canary, "--endpoint", "rudp://127.0.0.1:9"]);
+    assert_eq!(output.status.code(), Some(2));
+    assert!(!all_output(&output).contains("anary"), "{}", all_output(&output));
+    assert!(output.stdout.is_empty());
+}
+
+#[test]
+fn a_malformed_endpoint_or_none_is_exit_2_and_the_endpoint_is_not_echoed() {
+    let canary = "canary-endpoint-77c1";
+    let bad = self_run(&["--instance", NO_SUCH_INSTANCE, "--endpoint", canary]);
+    assert_eq!(bad.status.code(), Some(2));
+    assert!(!all_output(&bad).contains(canary) && bad.stdout.is_empty());
+    let none = self_run(&["--instance", NO_SUCH_INSTANCE]);
+    assert_eq!(none.status.code(), Some(2));
+    assert!(none.stdout.is_empty());
+
+    let from_env = self_run_with_env_endpoint(&["--instance", NO_SUCH_INSTANCE], "canary-env-endpoint-4d02");
+    assert_eq!(from_env.status.code(), Some(2));
+    assert!(!all_output(&from_env).contains("canary-env-endpoint-4d02") && from_env.stdout.is_empty());
+}
+
+#[test]
+fn a_mind_that_cannot_be_reached_is_exit_2_before_any_child_and_names_no_value() {
+    // Port 9 on loopback answers no one; PATH is empty, so a child could not start either.
+    let output = Command::new(env!("CARGO_BIN_EXE_mind-self-run"))
+        .args(["--instance", NO_SUCH_INSTANCE, "--endpoint", "rudp://127.0.0.1:9"])
+        .env("PATH", "")
+        .output()
+        .unwrap();
+    assert_eq!(output.status.code(), Some(2), "{}", all_output(&output));
+    assert!(output.stdout.is_empty());
+    assert!(!all_output(&output).contains("127.0.0.1"), "{}", all_output(&output));
+}
+
+#[cfg(unix)]
+mod lane {
+    use std::net::UdpSocket;
+    use std::path::Path;
+    use std::process::{Child, Command, Output, Stdio};
+    use std::time::Duration;
+
+    use chrono::Utc;
+    use eureka_pipeline::{
+        Date, DocRef, Label, PipelineCampaign, PipelineDocument, PipelineInstance, PipelineRun, PipelineStewardship, PipelineTarget, RunOperator, RunTurn, Sha,
+        Slug, TargetInvariant,
+    };
+    use huginn_mind::{Faculty, Mind, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineProvenance};
+    use mind_body::self_run::FLUSH_MESSAGE;
+
+    use super::all_output;
+    use tempfile::TempDir;
+
+    const INSTANCE: &str = "yggdrasil";
+    const CAMPAIGN: &str = "eureka-body";
+    const REPO: &str = "GameCult/Huginn";
+    // Pinned against Python's uuid.uuid5(uuid.NAMESPACE_OID, "yggdrasil:run:selfrun-lane").
+    const LANE_SESSION: &str = "6bbfe181-08ee-5c29-8e32-0ebe48ec4b33";
+
+    fn doc_ref() -> DocRef {
+        DocRef { path: "docs/eureka-body-target.md".into(), start_line: 1, end_line: 9, commit: Sha("df7507b".into()) }
+    }
+
+    /// The identity, stewardship, campaign and target a run cites, and her one
+    /// open Self run with budget 7.25.
+    fn seed() -> Vec<PipelineDocument> {
+        let date = || Date("2026-10-22".into());
+        vec![
+            PipelineDocument::Instance(PipelineInstance { instance: Slug(INSTANCE.into()), display_name: "test mind".into(), created_at: date(), host: "test".into() }),
+            PipelineDocument::Stewardship(PipelineStewardship { instance: Slug(INSTANCE.into()), repo: REPO.into(), sequence: 1, assigned_on: date(), note: "assigned".into() }),
+            PipelineDocument::Campaign(PipelineCampaign {
+                slug: Slug(CAMPAIGN.into()),
+                title: "Eureka body".into(),
+                repos: vec![REPO.into()],
+                working_branch: "main".into(),
+                target_doc: doc_ref(),
+            }),
+            PipelineDocument::Target(PipelineTarget {
+                campaign: Slug(CAMPAIGN.into()),
+                revision: 1,
+                invariants: vec![TargetInvariant { label: "run-is-the-grant".into(), statement: "A run is the grant.".into() }],
+                not_in_scope: vec![],
+                canonical_implementations: vec![],
+                doc: doc_ref(),
+            }),
+            PipelineDocument::Run(PipelineRun {
+                instance: Slug(INSTANCE.into()),
+                label: Label("selfrun-lane".into()),
+                operated_by: RunOperator::Mind,
+                turn: RunTurn::SelfRun,
+                host: "test".into(),
+                started_on: date(),
+                budget_usd: "7.25".into(),
+                claims: vec![],
+                campaigns: vec![Slug(CAMPAIGN.into())],
+            }),
+        ]
+    }
+
+    /// The daemon, ended with the test whether it passes or not.
+    struct Daemon(Child);
+
+    impl Drop for Daemon {
+        fn drop(&mut self) {
+            let _ = self.0.kill();
+            let _ = self.0.wait();
+        }
+    }
+
+    /// The daemon over a store seeded with `documents`, and the endpoint it listens on.
+    /// It needs a built `huginn-daemon` beside the binary (`cargo build -p huginn-daemon`).
+    fn serve(documents: Vec<PipelineDocument>) -> (Daemon, TempDir, String) {
+        let slug = Slug(INSTANCE.into());
+        let state = TempDir::new().unwrap();
+        {
+            let mut mind = Mind::open(state.path(), &slug).unwrap();
+            let batch = PipelineAdmissionBatch {
+                instance: slug.clone(),
+                provenance: PipelineProvenance { faculty: Faculty::Hands, agent: "test".into(), session: "test".into(), tool: "test".into() },
+                documents,
+            };
+            assert!(matches!(mind.admit(batch, Utc::now()), PipelineAdmissionOutcome::Committed { .. }));
+        }
+        let port = UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
+        let exe = Path::new(env!("CARGO_BIN_EXE_mind-self-run"));
+        let daemon_exe = exe.with_file_name("huginn-daemon");
+        assert!(daemon_exe.exists(), "build huginn-daemon first: {}", daemon_exe.display());
+        let daemon = Daemon(
+            Command::new(&daemon_exe)
+                .args(["--state-root", state.path().to_str().unwrap(), "--instance", INSTANCE, "--bind", &format!("127.0.0.1:{port}")])
+                .args(["--qdrant-url", "http://127.0.0.1:1", "--ollama-url", "http://127.0.0.1:1", "--embedding-model", "x"])
+                .stdout(Stdio::null())
+                .stderr(Stdio::null())
+                .spawn()
+                .unwrap(),
+        );
+        (daemon, state, format!("rudp://127.0.0.1:{port}"))
+    }
+
+    /// The binary with `bin` first on `PATH`, until the daemon answers (exit 2 before that
+    /// has started nothing). `SELF_RUN_ENV_CANARY` is a variable of the unit's environment.
+    fn ask(endpoint: &str, bin: &Path) -> Output {
+        let exe = Path::new(env!("CARGO_BIN_EXE_mind-self-run"));
+        let path = format!("{}:/usr/bin:/bin", bin.display());
+        for _ in 0..40 {
+            let attempt = Command::new(exe)
+                .args(["--instance", INSTANCE, "--endpoint", endpoint])
+                .env("PATH", &path)
+                .env("SELF_RUN_ENV_CANARY", ENV_CANARY)
+                .output()
+                .unwrap();
+            if attempt.status.code() != Some(2) {
+                return attempt;
+            }
+            std::thread::sleep(Duration::from_millis(500));
+        }
+        panic!("the daemon never answered");
+    }
+
+    const ENV_CANARY: &str = "canary-unit-environment-b94e";
+
+    /// Writes an executable `claude` into `bin` and returns the path of the record it keeps.
+    fn fake_claude(bin: &Path, body: &str) -> std::path::PathBuf {
+        use std::os::unix::fs::PermissionsExt;
+        let record = bin.join("record");
+        let script = format!("#!/bin/bash\necho \"$@\" > {record}\n{body}\n", record = record.display());
+        let claude = bin.join("claude");
+        std::fs::write(&claude, script).unwrap();
+        std::fs::set_permissions(&claude, std::fs::Permissions::from_mode(0o755)).unwrap();
+        record
+    }
+
+    /// The real binary against the real daemon, with a fake `claude` first on `PATH`.
+    /// The fake prints two requests (the second reaches the ceiling), waits for the start
+    /// and the flush on stdin, records its argument list, what it read and the unit
+    /// variable it inherited, prints a result and exits 0. The bin finds her run, starts
+    /// the child for it, asks for the flush and exits 0 without echoing the inherited
+    /// variable. It runs in the container lane only; `cargo test` and `cargo mutants`
+    /// skip it. Run it with the README's lane command.
+    #[test]
+    #[ignore = "container only: a built huginn-daemon beside the binary, and a POSIX shell"]
+    fn the_real_bin_finds_her_run_starts_claude_for_it_asks_for_the_flush_and_exits_0() {
+        let (daemon, _state, endpoint) = serve(seed());
+        let bin = TempDir::new().unwrap();
+        let request = |id: &str, tokens: u64| {
+            format!(
+                r#"{{"type":"assistant","parent_tool_use_id":null,"message":{{"id":"{id}","usage":{{"input_tokens":1,"cache_read_input_tokens":{tokens},"cache_creation_input_tokens":0}}}}}}"#
+            )
+        };
+        let body = format!(
+            "echo journal-line >&2
+echo '{first}'\necho '{second}'\nread -t 5 start\nread -t 5 flush\necho \"$start\" >> {record}\necho \"$flush\" >> {record}\necho \"$SELF_RUN_ENV_CANARY\" >> {record}\necho '{{\"type\":\"result\"}}'\nwhile read more; do :; done\nexit 0",
+            first = request("msg_1", 100),
+            second = request("msg_2", 299_999),
+            record = bin.path().join("record").display(),
+        );
+        let record = fake_claude(bin.path(), &body);
+        let output = ask(&endpoint, bin.path());
+        drop(daemon);
+
+        let stdout = String::from_utf8_lossy(&output.stdout).to_string();
+        assert_eq!(output.status.code(), Some(0), "{stdout}\n{}", String::from_utf8_lossy(&output.stderr));
+        assert_eq!(stdout, "request 1 prefix 101\nrequest 2 prefix 300000\nended: flush-requested code 0 requests 2 prefix 300000\n");
+        assert!(!all_output(&output).contains(ENV_CANARY), "the unit's environment is the child's, not the bin's output");
+        assert!(String::from_utf8_lossy(&output.stderr).contains("journal-line"), "the child's stderr is the unit's journal");
+
+        let recorded = std::fs::read_to_string(&record).unwrap();
+        let mut lines = recorded.lines();
+        let argv = lines.next().unwrap();
+        assert!(argv.contains(&format!("--session-id {LANE_SESSION}")) && argv.contains("--max-budget-usd 7.25"), "{argv}");
+        assert!(argv.starts_with("-p --input-format stream-json"), "{argv}");
+        assert!(lines.next().unwrap().contains("yggdrasil:run:selfrun-lane"), "the start message names her run");
+        let flush = lines.next().unwrap();
+        assert!(flush.contains(FLUSH_MESSAGE) && flush.contains("\"type\":\"user\""), "the flush went to the child's stdin: {flush}");
+        assert_eq!(lines.next().unwrap(), ENV_CANARY, "the child inherits the unit's environment");
+    }
+
+    /// With no Self run of hers in the mind the bin starts no child and exits 3.
+    #[test]
+    #[ignore = "container only: a built huginn-daemon beside the binary, and a POSIX shell"]
+    fn the_real_bin_with_no_self_run_of_hers_exits_3_and_never_starts_claude() {
+        let mut documents = seed();
+        documents.pop();
+        let (daemon, _state, endpoint) = serve(documents);
+        let bin = TempDir::new().unwrap();
+        let record = fake_claude(bin.path(), "exit 0");
+        let output = ask(&endpoint, bin.path());
+        drop(daemon);
+        assert_eq!(output.status.code(), Some(3), "{}", all_output(&output));
+        assert_eq!(String::from_utf8_lossy(&output.stdout), "no run: she has no single open Self run\n");
+        assert!(!record.exists(), "claude was started");
+    }
+}