diff --git a/Cargo.lock b/Cargo.lock index 04c05d1..64af826 100644 --- a/Cargo.lock +++ b/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", ] diff --git a/crates/mind-body/Cargo.toml b/crates/mind-body/Cargo.toml index 14da03e..efdcc4b 100644 --- a/crates/mind-body/Cargo.toml +++ b/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" diff --git a/crates/mind-body/README.md b/crates/mind-body/README.md index 5838632..7241223 100644 --- a/crates/mind-body/README.md +++ b/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 [--endpoint rudp://:]` runs inside +`mind-self@`. 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. diff --git a/crates/mind-body/src/bin/mind-self-run.rs b/crates/mind-body/src/bin/mind-self-run.rs new file mode 100644 index 0000000..04bc20d --- /dev/null +++ b/crates/mind-body/src/bin/mind-self-run.rs @@ -0,0 +1,44 @@ +//! `mind-self-run --instance [--endpoint rudp://:]`: carries +//! her one open Self run (`mind_body::self_run::carry`) inside +//! `mind-self@`. 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 [--endpoint rudp://:]"; + +// 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 { + let args: Vec = 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); + } + } +} diff --git a/crates/mind-body/src/bin/mind-wake.rs b/crates/mind-body/src/bin/mind-wake.rs index 761c086..56385b7 100644 --- a/crates/mind-body/src/bin/mind-wake.rs +++ b/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 [--endpoint rudp://:]"; -/// 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)> { - 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 { let args: Vec = 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::().ok()) - .ok_or_else(|| anyhow!("the endpoint is not rudp://:"))?; let mind = HuginnMind::new(HuginnClient::new(address, instance, TIMEOUT)); let launcher = SystemdLauncher::new(); let host = host(); diff --git a/crates/mind-body/src/cli.rs b/crates/mind-body/src/cli.rs index edd59aa..81d531e 100644 --- a/crates/mind-body/src/cli.rs +++ b/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 ] --instance 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 ` and an optional +/// `--endpoint rudp://:`, 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://:` is refused with `usage` or a fixed message, and no +/// value is echoed. +pub fn instance_and_endpoint(args: &[String], env_endpoint: Option, 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::().ok()) + .ok_or_else(|| anyhow!("the endpoint is not rudp://:"))?; + 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, user: Option) -> String { diff --git a/crates/mind-body/src/launch.rs b/crates/mind-body/src/launch.rs index e135c2c..9b7d45b 100644 --- a/crates/mind-body/src/launch.rs +++ b/crates/mind-body/src/launch.rs @@ -273,8 +273,9 @@ fn close_dead(mind: &dyn MindPort, run: &PipelineRef, now: DateTime) -> 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> { +/// 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> { let runs = headers(mind, &in_force(any_of(of_kind(PipelineKind::Run), "root", &[&instance.0])))?; Ok(runs .into_iter() diff --git a/crates/mind-body/src/lib.rs b/crates/mind-body/src/lib.rs index 4a86f31..19f33a6 100644 --- a/crates/mind-body/src/lib.rs +++ b/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@`: 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)] diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs new file mode 100644 index 0000000..220c262 --- /dev/null +++ b/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 { + [ + "-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::(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), +} + +/// 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>; +} + +/// 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, 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> { + 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 { + 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> { + spawn_program(Path::new("claude"), args) + } +} + +fn spawn_program(program: &Path, args: &[String]) -> Result> { + 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, + facts: Receiver, + /// 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; diff --git a/crates/mind-body/src/self_run/claude-2.1.268-assistant-usage-records.jsonl b/crates/mind-body/src/self_run/claude-2.1.268-assistant-usage-records.jsonl new file mode 100644 index 0000000..fc0a08d --- /dev/null +++ b/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"}]}} diff --git a/crates/mind-body/src/self_run/claude-2.1.268-stream-json-probe.jsonl b/crates/mind-body/src/self_run/claude-2.1.268-stream-json-probe.jsonl new file mode 100644 index 0000000..81e249c --- /dev/null +++ b/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":"","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} diff --git a/crates/mind-body/src/self_run/decide.rs b/crates/mind-body/src/self_run/decide.rs new file mode 100644 index 0000000..e9f96e8 --- /dev/null +++ b/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), + 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, + /// 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, + pub(super) requests: u64, + pub(super) prefix: u64, + result_seen: bool, + flush: Option, + /// The last signal attempted and when. + stopping: Option<(Stop, DateTime)>, + /// The strongest signal delivered. + delivered: Option, + exited: Option<(Option, DateTime)>, + eof: bool, + finished: bool, +} + +fn outcome(state: &State, cause: Cause, code: Option) -> 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> { + 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) -> (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; diff --git a/crates/mind-body/src/self_run/decide/tests.rs b/crates/mind-body/src/self_run/decide/tests.rs new file mode 100644 index 0000000..814abd2 --- /dev/null +++ b/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 { + 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) { + 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, 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), + 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, + flush: Option<(i64, u64)>, + flushes: u32, + closes: u32, + boundary: bool, + exceeded: bool, + result_seen: bool, + exited: Option, + 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 = 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, 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 = (0..length).map(|_| ALPHABET[(next() % ALPHABET.len() as u64) as usize]).collect(); + play(&symbols, [Policy::All, Policy::Nothing, Policy::NoKill][round % 3]); + } +} diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs new file mode 100644 index 0000000..5399d34 --- /dev/null +++ b/crates/mind-body/src/self_run/tests.rs @@ -0,0 +1,1087 @@ +//! The actuator's tests: a real mind over a temporary directory holds her run, +//! a scripted `Child` plays the stream and a settable clock plays time. The +//! stream lines are built from Claude Code 2.1.268's own records (the two +//! fixture files beside this one), varying only ids and usage numbers. The real +//! `ClaudeProcess` is driven through a fake `claude` script on Unix. + +use std::cell::{Cell, RefCell}; +use std::collections::VecDeque; +use std::rc::Rc; + +use chrono::Duration as ChronoDuration; +use eureka_pipeline::{PipelineDocument, PipelineKind as K, PipelineRef, RunOperator, RunTurn, Slug}; +use huginn_mind::{PipelineAdmissionOutcome, PipelineSelectionPage}; + +use super::*; +use crate::testkit::*; + +const RECORDS: &str = include_str!("claude-2.1.268-assistant-usage-records.jsonl"); +const PROBE: &str = include_str!("claude-2.1.268-stream-json-probe.jsonl"); + +// Pinned against Python's uuid.uuid5(uuid.NAMESPACE_OID, ), not against the code under test. +const SESSION_OF_MIND_RUN: &str = "00cb82a4-c0ac-5ad4-83e2-8e685589fb67"; +const SESSION_OF_RUN_A: &str = "d6db8da1-224c-5430-91b5-17aa4df8ae7a"; +const SESSION_OF_RUN_B: &str = "546bec88-f7a9-5fa0-bf07-0fada54f071b"; + +fn foreign_run(id: &str) -> PipelineRef { + PipelineRef { kind: K::Run, id: id.into() } +} + +/// The first real assistant record with its id and usage replaced. +fn request(id: &str, input: u64, read: u64, write: u64) -> String { + let mut event: Value = serde_json::from_str(RECORDS.lines().next().unwrap()).unwrap(); + event["message"]["id"] = json!(id); + event["message"]["usage"]["input_tokens"] = json!(input); + event["message"]["usage"]["cache_read_input_tokens"] = json!(read); + event["message"]["usage"]["cache_creation_input_tokens"] = json!(write); + event.to_string() +} + +/// A request whose whole prefix is `prefix`, under a message id of its own. +fn at(id: &str, prefix: u64) -> String { + request(id, 0, prefix, 0) +} + +fn subagent(id: &str, prefix: u64) -> String { + let mut event: Value = serde_json::from_str(&at(id, prefix)).unwrap(); + event["parent_tool_use_id"] = json!("toolu_0001"); + event.to_string() +} + +fn result_line() -> String { + PROBE.lines().nth(2).unwrap().to_string() +} + +fn init_line() -> String { + PROBE.lines().next().unwrap().to_string() +} + +fn self_run(label: &str, budget: &str) -> PipelineDocument { + let PipelineDocument::Run(mut document) = run(label, RunTurn::SelfRun, RunOperator::Mind, &[]) else { unreachable!() }; + document.budget_usd = budget.into(); + PipelineDocument::Run(document) +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum Event { + Wrote(String), + InputClosed, + /// A signal was attempted, delivered or not. + Stopped(Stop), + Asked(StdDuration), +} + +#[derive(Clone)] +enum Step { + Line(String), + /// Time passes with no output: the clock moves and the read times out. + Idle(i64), + /// The clock moves, then this line arrives. + After(i64, String), + /// The child is reaped with this code; its stream stays open. + Exit(Option), + /// Its stdout ends; the child is not reaped yet. + Eof, +} + +use Step::{Eof, Exit, Idle, Line as Out}; + +struct Scripted { + steps: VecDeque, + log: Rc>>, + clock: Rc, + code: Option, + /// The weakest signal that ends it; weaker ones are ignored. + dies_on: Option, + /// Signals that cannot be delivered (no `kill` program, say). + refuses: Vec, + /// When the steps run out it neither exits nor closes its stream: time just passes. + hangs: bool, + exited: bool, + eof: bool, + /// How many times it was asked: a loop that never ends is a failure, not a hang. + asked: u32, +} + +impl Child for Scripted { + fn write_line(&mut self, line: &str) { + self.log.borrow_mut().push(Event::Wrote(line.to_string())); + } + + fn close_input(&mut self) { + self.log.borrow_mut().push(Event::InputClosed); + } + + fn next(&mut self, timeout: StdDuration) -> Read { + self.log.borrow_mut().push(Event::Asked(timeout)); + self.asked += 1; + assert!(self.asked < 500, "the loop never ended"); + match self.steps.pop_front() { + Some(Step::Line(line)) => Read::Line(line), + Some(Step::Idle(seconds)) => { + self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds)); + Read::Timeout + } + Some(Step::After(seconds, line)) => { + self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds)); + Read::Line(line) + } + Some(Step::Exit(code)) => { + self.exited = true; + Read::Exited(code) + } + Some(Step::Eof) => { + self.eof = true; + Read::Eof + } + // The stream is over: the child is reaped, then its stdout ends. + None if self.hangs => { + self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(timeout.as_secs() as i64)); + Read::Timeout + } + None if !self.exited => { + self.exited = true; + Read::Exited(self.code) + } + None if !self.eof => { + self.eof = true; + Read::Eof + } + None => panic!("the loop kept waiting on a child that ended"), + } + } + + fn stop(&mut self, how: Stop) -> bool { + self.log.borrow_mut().push(Event::Stopped(how)); + if self.refuses.contains(&how) { + return false; + } + if self.dies_on.is_some_and(|weakest| how >= weakest) { + self.steps.clear(); + self.hangs = false; + } + true + } +} + +struct Fake { + child: RefCell>, + spawns: RefCell>>, + refuse: bool, +} + +impl Spawner for Fake { + fn spawn(&self, args: &[String]) -> Result> { + self.spawns.borrow_mut().push(args.to_vec()); + if self.refuse { + return Err(anyhow!("refused")); + } + Ok(Box::new(self.child.borrow_mut().take().expect("spawned twice"))) + } +} + +/// The mind as the actuator sees it, over a `TestMind`: another instance's name, +/// doubled run headers, and a count of admissions it must never receive. +struct Seen<'a> { + inner: &'a TestMind, + instance: Slug, + doubled: bool, + admits: Cell, +} + +impl<'a> Seen<'a> { + fn of(inner: &'a TestMind) -> Self { + Self { inner, instance: inner.instance().clone(), doubled: false, admits: Cell::new(0) } + } +} + +impl MindPort for Seen<'_> { + fn instance(&self) -> &Slug { + &self.instance + } + + fn query(&self, selection: &Selection) -> Result { + let mut page = self.inner.query(selection)?; + if self.doubled + && let PipelinePageItems::Headers(items) = &mut page.items + { + let again = items.clone(); + items.extend(again); + } + Ok(page) + } + + fn admit(&self, _agent: &str, _session: &str, _documents: Vec) -> Result { + self.admits.set(self.admits.get() + 1); + Err(anyhow!("the actuator admits nothing")) + } +} + +struct Ran { + outcome: Result, + out: String, + events: Vec, + spawns: Vec>, + admits: usize, +} + +impl Ran { + fn ended(&self) -> (Cause, Option, u64, u64) { + match self.outcome.as_ref().unwrap() { + SelfRunOutcome::Ended { cause, code, requests, prefix } => (*cause, *code, *requests, *prefix), + other => panic!("not ended: {other:?}"), + } + } + + /// The text of every user line written to the child, in order. + fn written(&self) -> Vec { + self.events + .iter() + .filter_map(|event| match event { + Event::Wrote(line) => { + let line: Value = serde_json::from_str(line).unwrap(); + assert_eq!((line["type"].as_str(), line["message"]["role"].as_str()), (Some("user"), Some("user")), "{line}"); + Some(line["message"]["content"].as_str().unwrap().to_string()) + } + _ => None, + }) + .collect() + } + + fn count(&self, wanted: &Event) -> usize { + self.events.iter().filter(|event| *event == wanted).count() + } + + fn asked(&self) -> Vec { + self.events + .iter() + .filter_map(|event| match event { + Event::Asked(timeout) => Some(timeout.as_secs()), + _ => None, + }) + .collect() + } +} + +/// `dies_on_interrupt`: the child ends at the first signal, else it ignores all. +fn carry_with(seen: &Seen, steps: Vec, code: Option, dies_on_interrupt: bool) -> Ran { + carry_signalled(seen, steps, code, dies_on_interrupt.then_some(Stop::Interrupt), &[]) +} + +fn carry_signalled(seen: &Seen, steps: Vec, code: Option, dies_on: Option, refuses: &[Stop]) -> Ran { + carry_scripted(seen, steps, code, dies_on, refuses, false) +} + +fn carry_scripted(seen: &Seen, steps: Vec, code: Option, dies_on: Option, refuses: &[Stop], hangs: bool) -> Ran { + let log = Rc::new(RefCell::new(Vec::new())); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let fake = Fake { + child: RefCell::new(Some(Scripted { steps: steps.into(), log: log.clone(), clock: clock.clone(), code, dies_on, refuses: refuses.to_vec(), hangs, exited: false, eof: false, asked: 0 })), + spawns: RefCell::new(Vec::new()), + refuse: false, + }; + let mut out = Vec::new(); + let outcome = carry(seen, &*clock, &fake, &mut out); + let events = log.borrow().clone(); + let spawns = fake.spawns.borrow().clone(); + Ran { outcome, out: String::from_utf8(out).unwrap(), events, spawns, admits: seen.admits.get() } +} + +fn at_start() -> chrono::DateTime { + crate::testkit::at(0) +} + +/// A mind holding her one open Self run `selfrun-a` with budget 5. +fn with_run() -> TestMind { + let mind = TestMind::seeded(); + mind.committed(vec![self_run("selfrun-a", "5")]); + mind +} + +fn go(steps: Vec) -> Ran { + let mind = with_run(); + carry_with(&Seen::of(&mind), steps, Some(0), false) +} + +type Seed = Box; + +fn flushes(ran: &Ran) -> usize { + ran.written().iter().filter(|text| text.as_str() == FLUSH_MESSAGE).count() +} + +// ---- who is carried ----------------------------------------------------- + +#[test] +fn without_exactly_one_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawned() { + let cases: Vec<(&str, Seed)> = vec![ + ("no run at all", Box::new(|_| {})), + ("a Self run the operator operates", Box::new(|mind| mind.committed(vec![run("op", RunTurn::SelfRun, RunOperator::Operator, &[])]))), + ("her Persona turn", Box::new(|mind| mind.committed(vec![run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]))), + ( + "a Self run of hers that is closed", + Box::new(|mind| { + mind.committed(vec![self_run("selfrun-a", "5")]); + mind.committed(vec![close(run_ref("selfrun-a"), recorded())]); + }), + ), + ]; + for (name, seed) in cases { + let mind = TestMind::seeded(); + seed(&mind); + let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); + assert_eq!(ran.outcome.unwrap(), SelfRunOutcome::NoRun, "{name}"); + assert!(ran.spawns.is_empty() && ran.events.is_empty() && ran.out.is_empty(), "{name}"); + } + + // Another instance's mind holds none of this instance's runs. + let mind = with_run(); + let mut other = Seen::of(&mind); + other.instance = Slug("another".into()); + assert_eq!(carry_with(&other, vec![], Some(0), false).outcome.unwrap(), SelfRunOutcome::NoRun); + + // Two open runs are not one: nothing is spawned for either. + let mut two = Seen::of(&mind); + two.doubled = true; + let ran = carry_with(&two, vec![], Some(0), false); + assert_eq!(ran.outcome.unwrap(), SelfRunOutcome::NoRun); + assert!(ran.spawns.is_empty()); +} + +#[test] +fn a_mind_that_cannot_be_read_or_a_child_that_cannot_start_is_an_error_not_an_outcome() { + let mind = with_run(); + let clock = Rc::new(Fixed(Cell::new(at_start()))); + let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true }; + assert!(carry(&Seen::of(&mind), &*clock, &refusing, &mut Vec::new()).is_err()); + assert_eq!(refusing.spawns.borrow().len(), 1, "it tried exactly once"); + + struct Down; + impl MindPort for Down { + fn instance(&self) -> &Slug { + static INSTANCE: std::sync::OnceLock = std::sync::OnceLock::new(); + INSTANCE.get_or_init(|| Slug("yggdrasil".into())) + } + fn query(&self, _selection: &Selection) -> Result { + Err(anyhow!("down")) + } + fn admit(&self, _agent: &str, _session: &str, _documents: Vec) -> Result { + unreachable!() + } + } + let never = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: false }; + assert!(carry(&Down, &*clock, &never, &mut Vec::new()).is_err()); + assert!(never.spawns.borrow().is_empty()); +} + +// ---- what is started ------------------------------------------------------ + +#[test] +fn claude_args_is_exactly_the_fixed_list_for_a_session_and_a_budget() { + let session = Uuid::parse_str(SESSION_OF_RUN_A).unwrap(); + let args = claude_args(session, "12.25"); + assert_eq!( + args, + [ + "-p", + "--input-format", + "stream-json", + "--output-format", + "stream-json", + "--verbose", + "--session-id", + SESSION_OF_RUN_A, + "--max-budget-usd", + "12.25", + "--max-turns", + "500", + "--permission-prompts", + "none", + ] + ); + // Those belong to the run measure or to the unit's managed settings. + for forbidden in ["--resume", "--continue", "--fork-session", "--model", "--permission-mode", "--dangerously-skip-permissions", "--settings", "--system-prompt"] { + assert!(!args.iter().any(|arg| arg.starts_with(forbidden) || arg.contains("dangerously") || arg == "bypassPermissions"), "{forbidden}"); + } + assert_eq!(MAX_TURNS, 500); +} + +#[test] +fn the_session_id_is_uuid_v5_of_the_run_id_and_differs_per_run() { + assert_eq!(session_id(&foreign_run("eureka:run:mind-20261022T120000000Z")).to_string(), SESSION_OF_MIND_RUN); + assert_eq!(session_id(&run_ref("selfrun-a")).to_string(), SESSION_OF_RUN_A); + assert_eq!(session_id(&run_ref("selfrun-b")).to_string(), SESSION_OF_RUN_B); +} + +#[test] +fn the_child_is_started_for_her_run_with_the_runs_own_budget_verbatim() { + // Trailing zeros are significant: "10" is ten dollars, not one. + for budget in ["0.5", "5", "12.25", "10", "100"] { + let mind = TestMind::seeded(); + mind.committed(vec![self_run("selfrun-b", budget)]); + let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); + assert_eq!(ran.spawns.len(), 1, "{budget}"); + assert_eq!(ran.spawns[0], claude_args(Uuid::parse_str(SESSION_OF_RUN_B).unwrap(), budget), "{budget}"); + } +} + +#[test] +fn the_start_message_is_the_first_thing_written_and_names_the_skill_the_instance_and_the_run() { + let ran = go(vec![Out(result_line())]); + assert!(matches!(&ran.events[0], Event::Wrote(_)), "the start precedes the first read: {:?}", ran.events[0]); + let written = ran.written(); + assert_eq!(written[0], start_message(&Slug("yggdrasil".into()), &run_ref("selfrun-a"))); + for needle in ["mind-self", "yggdrasil", "yggdrasil:run:selfrun-a", AGENT] { + assert!(written[0].contains(needle), "{needle}: {}", written[0]); + } + assert_eq!(written.len(), 1, "no flush without the ceiling"); +} + +// ---- what is read ----------------------------------------------------------- + +#[test] +fn read_usage_sums_input_cache_read_and_cache_write_and_reads_absent_fields_as_zero() { + assert_eq!(read_usage(&request("m1", 1, 20, 300)), Usage::Request { id: "m1".into(), prefix: 321 }); + // Each term alone, so dropping or doubling any one is seen. + assert_eq!(read_usage(&request("m", 7, 0, 0)), Usage::Request { id: "m".into(), prefix: 7 }); + assert_eq!(read_usage(&request("m", 0, 7, 0)), Usage::Request { id: "m".into(), prefix: 7 }); + assert_eq!(read_usage(&request("m", 0, 0, 7)), Usage::Request { id: "m".into(), prefix: 7 }); + for absent in ["input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens"] { + let mut event: Value = serde_json::from_str(&request("m", 1, 20, 300)).unwrap(); + event["message"]["usage"].as_object_mut().unwrap().remove(absent); + let want = 321 - [("input_tokens", 1), ("cache_read_input_tokens", 20), ("cache_creation_input_tokens", 300)].iter().find(|(name, _)| *name == absent).unwrap().1; + assert_eq!(read_usage(&event.to_string()), Usage::Request { id: "m".into(), prefix: want }, "{absent} absent"); + } + let mut no_usage: Value = serde_json::from_str(&request("m", 1, 1, 1)).unwrap(); + no_usage["message"].as_object_mut().unwrap().remove("usage"); + assert_eq!(read_usage(&no_usage.to_string()), Usage::Request { id: "m".into(), prefix: 0 }); +} + +#[test] +fn read_usage_classifies_the_owners_real_records() { + let sums = [84210, 100751, 117174, 118336, 126746, 127250, 129853]; + let mut seen = Vec::new(); + for line in RECORDS.lines() { + let Usage::Request { id, prefix } = read_usage(line) else { panic!("an assistant record of the main thread is a request: {line}") }; + if !seen.iter().any(|(known, _)| *known == id) { + seen.push((id, prefix)); + } + } + assert_eq!(seen.iter().map(|(_, prefix)| *prefix).collect::>(), sums); + assert_eq!(read_usage(&result_line()), Usage::Result); + assert_eq!(read_usage(&init_line()), Usage::Other); + assert_eq!(read_usage(&subagent("m", 5)), Usage::Other, "a subagent's request is not the main thread's"); + let mut no_id: Value = serde_json::from_str(&at("m", 5)).unwrap(); + no_id["message"].as_object_mut().unwrap().remove("id"); + assert_eq!(read_usage(&no_id.to_string()), Usage::Other); + let mut null_parent: Value = serde_json::from_str(&at("m", 5)).unwrap(); + null_parent["parent_tool_use_id"] = Value::Null; + assert_eq!(read_usage(&null_parent.to_string()), Usage::Request { id: "m".into(), prefix: 5 }); + let mut no_parent: Value = serde_json::from_str(&at("m", 5)).unwrap(); + no_parent.as_object_mut().unwrap().remove("parent_tool_use_id"); + assert_eq!(read_usage(&no_parent.to_string()), Usage::Request { id: "m".into(), prefix: 5 }); + for unreadable in ["", "not json", "[1,2]", "\"assistant\"", "12", "null", "{\"type\":"] { + assert_eq!(read_usage(unreadable), Usage::Unreadable, "{unreadable:?}"); + } +} + +#[test] +fn records_that_share_a_message_id_count_once_and_subagent_records_never_move_the_prefix() { + let mut steps: Vec = RECORDS.lines().map(|line| Out(line.to_string())).collect(); + steps.insert(3, Out(subagent("msg_sub", 999_999))); + let ran = go(steps); + assert_eq!( + ran.out, + "request 1 prefix 84210\nrequest 2 prefix 100751\nrequest 3 prefix 117174\nrequest 4 prefix 118336\nrequest 5 prefix 126746\nrequest 6 prefix 127250\nrequest 7 prefix 129853\n" + ); + let (cause, _, requests, prefix) = ran.ended(); + assert_eq!((cause, requests, prefix), (Cause::Own, 7, 129_853), "the newest prefix, 129853, not a subagent's 999999"); + assert_eq!(flushes(&ran), 0); +} + +#[test] +fn the_prefix_is_the_newest_requests_not_the_largest() { + let ran = go(vec![Out(at("m1", 200_000)), Out(at("m2", 150_000))]); + assert_eq!(ran.ended().3, 150_000); +} + +// ---- the flush -------------------------------------------------------------- + +#[test] +fn below_the_ceiling_no_flush_is_written_and_at_it_exactly_one() { + let below = go(vec![Out(at("m1", CEILING_TOKENS - 1))]); + assert_eq!(flushes(&below), 0); + assert_eq!(below.ended().0, Cause::Own); + + let at_it = go(vec![Out(at("m1", CEILING_TOKENS - 1)), Out(at("m2", CEILING_TOKENS)), Out(at("m3", CEILING_TOKENS + 10_000))]); + assert_eq!(flushes(&at_it), 1, "{:?}", at_it.written()); + assert_eq!(at_it.written().len(), 2, "the start and the flush"); + assert_eq!(at_it.ended().0, Cause::FlushRequested); + assert_eq!(CEILING_TOKENS, 300_000); +} + +#[test] +fn a_result_before_the_ceiling_means_no_flush_ever() { + let ran = go(vec![Out(at("m1", 10)), Out(result_line()), Out(at("m2", CEILING_TOKENS + 1))]); + assert_eq!(flushes(&ran), 0); + assert_eq!(ran.ended().0, Cause::Own); + assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0); +} + +#[test] +fn stdin_closes_once_at_the_first_result_flush_or_no_flush_and_nothing_is_written_after() { + let plain = go(vec![Out(at("m1", 10)), Out(result_line()), Out(result_line())]); + assert_eq!(plain.count(&Event::InputClosed), 1); + let closed_at = plain.events.iter().position(|event| *event == Event::InputClosed).unwrap(); + assert!(plain.events[closed_at..].iter().all(|event| !matches!(event, Event::Wrote(_))), "no input after the close"); + assert!(plain.events[..closed_at].iter().any(|event| matches!(event, Event::Wrote(_))), "the start went before"); + + let flushed = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())]); + assert_eq!(flushed.count(&Event::InputClosed), 1); + let closed_at = flushed.events.iter().position(|event| *event == Event::InputClosed).unwrap(); + assert_eq!(flushed.events[..closed_at].iter().filter(|event| matches!(event, Event::Wrote(_))).count(), 2, "start and flush before the close"); + assert!(flushed.events[closed_at..].iter().all(|event| !matches!(event, Event::Wrote(_)))); + + let never = go(vec![Out(at("m1", 10))]); + assert_eq!(never.count(&Event::InputClosed), 0, "no result, no close"); +} + +// ---- the interrupt ------------------------------------------------------------ + +#[test] +fn after_the_flush_the_grace_interrupts_exactly_once_and_not_a_second_early() { + let grace = FLUSH_GRACE_S; + // Just short of the grace: no interrupt, and the child ended on its own. + let early = go(vec![Out(at("m1", CEILING_TOKENS)), Idle(grace - 1)]); + assert_eq!(early.count(&Event::Stopped(Stop::Interrupt)), 0); + assert_eq!(early.ended().0, Cause::FlushRequested); + + // At the grace, and long past it: one interrupt. + let on_time = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(grace), Idle(grace), Idle(grace)], Some(130), true); + assert_eq!(on_time.count(&Event::Stopped(Stop::Interrupt)), 1); + assert_eq!(on_time.ended().0, Cause::Signalled(Stop::Interrupt)); + + // Exactly at the grace is the grace: the interrupt comes then, not on the next wait. + let exact = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(grace)], Some(130), true); + assert_eq!(exact.ended().0, Cause::Signalled(Stop::Interrupt)); + + // A child that ignores it is interrupted once and still waited for. + let stubborn = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(grace), Idle(grace), Idle(grace)], Some(0), false); + assert_eq!(stubborn.count(&Event::Stopped(Stop::Interrupt)), 1); + assert_eq!(stubborn.ended().1, Some(0), "its own end is waited for"); + assert_eq!(FLUSH_GRACE_S, 900); +} + +#[test] +fn the_wait_for_the_child_never_passes_the_grace_deadline_and_never_spins_after_the_interrupt() { + let ran = go(vec![Out(at("m1", CEILING_TOKENS)), Idle(840), Idle(59), Idle(1)]); + // 60 until the grace is near, the one second left, then the escalation's own wait, then the drain after the child's exit. + assert_eq!(ran.asked(), [60, 60, 60, 1, 60, 2], "{:?}", ran.asked()); +} + +#[test] +fn the_headroom_counts_the_requests_after_the_first_result_that_follows_the_flush_and_one_short_of_it_does_not_interrupt() { + let hard = CEILING_TOKENS + FLUSH_HEADROOM_TOKENS; + let short = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(at("m2", hard - 1))]); + assert_eq!(short.count(&Event::Stopped(Stop::Interrupt)), 0); + assert_eq!(short.ended().0, Cause::FlushRequested); + + let at_hard = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(at("m2", hard)), Out(at("m3", hard + 1))], Some(130), true); + assert_eq!(at_hard.count(&Event::Stopped(Stop::Interrupt)), 1); + assert_eq!(at_hard.ended().0, Cause::Signalled(Stop::Interrupt)); + assert_eq!(FLUSH_HEADROOM_TOKENS, 60_000); + + // Before that result the running turn's own requests are not the flush's turn: however far they grow, no interrupt. + let same_turn = go(vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", hard)), Out(at("m3", hard + 500_000))]); + assert_eq!(same_turn.count(&Event::Stopped(Stop::Interrupt)), 0); +} + +#[test] +fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_the_running_turn_cannot_use_up_the_flushs_turn() { + let jump = carry_with( + &Seen::of(&with_run()), + vec![Out(at("m1", 290_000)), Out(at("m2", 400_000)), Out(at("m3", 470_000)), Out(result_line()), Out(at("m4", 459_999)), Out(at("m5", 460_000)), Out(at("m6", 470_000))], + Some(130), + true, + ); + let kinds: Vec<&str> = jump + .events + .iter() + .filter_map(|event| match event { + Event::Wrote(line) if line.contains("context ceiling") => Some("flush"), + Event::Stopped(_) => Some("interrupt"), + Event::InputClosed => Some("closed"), + Event::Asked(_) => Some("read"), + _ => None, + }) + .collect(); + // Reads: m1, m2 (the jump, which writes the flush), m3 (the running turn's 470k, past the headroom: no interrupt). + assert_eq!(kinds[..5], ["read", "read", "flush", "read", "read"], "{kinds:?}"); + // The headroom is measured from the prefix the flush was asked at (400k) and counts the requests after the + // first result only: 459_999 does not interrupt; 460_000 does, once, and m6 is never read. + assert_eq!(jump.count(&Event::Stopped(Stop::Interrupt)), 1, "{kinds:?}"); + assert_eq!(jump.out, "request 1 prefix 290000\nrequest 2 prefix 400000\nrequest 3 prefix 470000\nrequest 4 prefix 459999\nrequest 5 prefix 460000\n"); + assert_eq!(jump.ended().0, Cause::Signalled(Stop::Interrupt)); + assert_eq!(flushes(&jump), 1); +} + +#[test] +fn a_child_that_ends_before_the_grace_is_never_interrupted() { + let ran = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())]); + assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0); + assert_eq!(ran.ended().0, Cause::FlushRequested); +} + +#[test] +fn the_grace_is_the_full_grace_from_the_moment_the_flush_was_asked_whichever_request_that_was() { + let grace = FLUSH_GRACE_S; + // Five requests, time passing before the ceiling, then the flush on the sixth. + let before: Vec = vec![Out(at("m1", 10)), Idle(500), Out(at("m2", 20)), Out(at("m3", 30)), Out(at("m4", 40)), Out(at("m5", 50)), Out(at("m6", CEILING_TOKENS))]; + let early = carry_with(&Seen::of(&with_run()), [before.clone(), vec![Idle(grace - 1)]].concat(), Some(0), true); + assert_eq!((flushes(&early), early.count(&Event::Stopped(Stop::Interrupt))), (1, 0), "one second short of the grace, flush on request 6"); + let on_time = carry_with(&Seen::of(&with_run()), [before, vec![Idle(grace - 1), Idle(1)]].concat(), Some(0), true); + assert_eq!(on_time.count(&Event::Stopped(Stop::Interrupt)), 1, "the grace is the same after six requests as after one"); +} + +#[test] +fn the_message_ids_of_a_long_run_are_all_remembered() { + let ids: Vec = (1..=12).map(|n| format!("m{n}")).collect(); + let mut steps: Vec = ids.iter().map(|id| Out(at(id, 1_000))).collect(); + // Every earlier request seen again, the oldest and the newest among them: none counts twice. + steps.extend(ids.iter().map(|id| Out(at(id, 2_000)))); + let ran = go(steps); + let (_, _, requests, prefix) = ran.ended(); + assert_eq!((requests, prefix), (12, 1_000)); + assert_eq!(ran.out.lines().count(), 12); +} + +#[test] +fn a_usage_sum_that_overflows_is_at_or_over_the_ceiling_not_a_panic_or_a_wrap() { + let line = request("m", u64::MAX, 1, 0); + assert_eq!(read_usage(&line), Usage::Request { id: "m".into(), prefix: u64::MAX }); + assert_eq!(read_usage(&request("m", u64::MAX, u64::MAX, u64::MAX)), Usage::Request { id: "m".into(), prefix: u64::MAX }); + assert_eq!(read_usage(&request("m", 1, u64::MAX, 1)), Usage::Request { id: "m".into(), prefix: u64::MAX }); + // Through the actuator: the flush is asked for, and the headroom beyond a saturated prefix does not overflow + // either, and the pass that wrote the flush does not interrupt. + let ran = go(vec![Out(request("m1", u64::MAX, 7, 0)), Out(request("m2", 0, 0, 0))]); + assert_eq!(flushes(&ran), 1); + assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0); +} + +#[test] +fn a_result_before_any_request_closes_the_input_once_and_the_run_ends_with_a_typed_outcome() { + // The shape of an API or authentication failure: the child reports an error result and no request ever came. + let ran = carry_with(&Seen::of(&with_run()), vec![Out(result_line())], Some(1), false); + assert_eq!(ran.count(&Event::InputClosed), 1); + assert_eq!(ran.ended(), (Cause::Own, Some(1), 0, 0)); + assert_eq!(ran.outcome.as_ref().unwrap().exit_status(), 1); + assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0); +} + +#[test] +fn without_a_flush_nothing_is_ever_signalled_however_large_the_prefix_grows() { + let ran = go(vec![Out(at("m1", 10)), Out(result_line()), Out(at("m2", 900_000)), Idle(100_000)]); + assert_eq!(ran.events.iter().filter(|event| matches!(event, Event::Stopped(_))).count(), 0); + assert_eq!(ran.ended().0, Cause::Own); +} + +#[test] +fn a_child_that_ignores_a_signal_gets_the_next_one_after_the_bound_and_the_outcome_names_the_one_that_ended_it() { + let hung = |dies_on: Option, refuses: &[Stop], rest: Vec| { + carry_signalled(&Seen::of(&with_run()), [vec![Out(at("m1", CEILING_TOKENS)), Idle(FLUSH_GRACE_S)], rest].concat(), Some(0), dies_on, refuses) + }; + let stops = |ran: &Ran| -> Vec { ran.events.iter().filter(|event| matches!(event, Event::Stopped(_))).cloned().collect() }; + let after = ESCALATE_AFTER_S; + + // Ignores SIGINT and SIGTERM: each next signal waits out the bound, SIGKILL ends it. + let killed = hung(Some(Stop::Kill), &[], vec![Idle(after - 1), Idle(1), Idle(after - 1), Idle(1), Idle(after)]); + assert_eq!(stops(&killed), [Event::Stopped(Stop::Interrupt), Event::Stopped(Stop::Terminate), Event::Stopped(Stop::Kill)]); + assert_eq!(killed.ended().0, Cause::Signalled(Stop::Kill)); + + // One second short of the bound there is no second signal. + let waiting = hung(None, &[], vec![Idle(after - 1)]); + assert_eq!(stops(&waiting), [Event::Stopped(Stop::Interrupt)]); + + // SIGTERM ends it: the outcome says terminated, and nothing stronger is sent. + let terminated = hung(Some(Stop::Terminate), &[], vec![Idle(after), Idle(after)]); + assert_eq!(stops(&terminated), [Event::Stopped(Stop::Interrupt), Event::Stopped(Stop::Terminate)]); + assert_eq!(terminated.outcome.as_ref().unwrap().to_string(), "ended: terminated code 0 requests 1 prefix 300000"); +} + +#[test] +fn a_signal_that_could_not_be_delivered_is_never_reported_as_sent() { + let hung = |refuses: &[Stop], rest: Vec| { + carry_signalled(&Seen::of(&with_run()), [vec![Out(at("m1", CEILING_TOKENS)), Idle(FLUSH_GRACE_S)], rest].concat(), Some(0), None, refuses) + }; + // SIGINT is refused and the child ends by itself afterwards: the flush was requested, nothing interrupted it. + let refused = hung(&[Stop::Interrupt], vec![]); + assert_eq!(refused.count(&Event::Stopped(Stop::Interrupt)), 1, "it was attempted"); + assert_eq!(refused.ended().0, Cause::FlushRequested); + + // SIGINT delivered but ignored, SIGTERM refused, the child ends on its own: the strongest delivered is SIGINT. + let partly = hung(&[Stop::Terminate], vec![Idle(ESCALATE_AFTER_S)]); + assert_eq!(partly.ended().0, Cause::Signalled(Stop::Interrupt)); + + // Not even SIGKILL is delivered and the child never ends: after the bound the run is over without it, and the label says so. + let stuck = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(FLUSH_GRACE_S)], Some(0), None, &[Stop::Interrupt, Stop::Terminate, Stop::Kill], true); + assert_eq!(stuck.ended(), (Cause::Unstoppable, None, 1, CEILING_TOKENS)); + assert_eq!(stuck.events.iter().filter(|event| matches!(event, Event::Stopped(_))).count(), 3); + assert_eq!(stuck.outcome.unwrap().exit_status(), 1); +} + +/// A clock that counts its reads. +struct Counting(Fixed, Cell); + +impl Clock for Counting { + fn now(&self) -> chrono::DateTime { + self.1.set(self.1.get() + 1); + self.0.now() + } +} + +#[test] +fn the_clock_is_read_once_per_event_so_a_step_of_the_wall_clock_cannot_split_a_pass() { + let mind = with_run(); + let log = Rc::new(RefCell::new(Vec::new())); + let clock = Rc::new(Counting(Fixed(Cell::new(at_start())), Cell::new(0))); + // The child's reads move the clock by far more than the grace, as a stepped wall clock would. + struct Stepped(Rc); + impl Clock for Stepped { + fn now(&self) -> chrono::DateTime { + let now = self.0.now(); + self.0.0.0.set(now + ChronoDuration::seconds(FLUSH_GRACE_S + 1)); + now + } + } + let steps: VecDeque = vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", CEILING_TOKENS + 1)), Idle(0)].into(); + let child = Scripted { steps, log: log.clone(), clock: Rc::new(Fixed(Cell::new(at_start()))), code: Some(0), dies_on: None, refuses: vec![], hangs: false, exited: false, eof: false, asked: 0 }; + let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false }; + let outcome = carry(&Seen::of(&mind), &Stepped(clock.clone()), &fake, &mut Vec::new()).unwrap(); + let events = log.borrow().clone(); + let reads = events.iter().filter(|event| matches!(event, Event::Asked(_))).count() as u32; + // One read before the first wait, then one per fact heard: never a second read inside a pass. + assert_eq!(clock.1.get(), 1 + reads, "{events:?}"); + // The flush was written on the first request at time 0; the very next pass sees the stepped clock. + // The flush pass itself has no interrupt, whatever the step: the interrupt is on a later event. + let flush_at = events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains("context ceiling"))).unwrap(); + let stop_at = events.iter().position(|event| matches!(event, Event::Stopped(_))).unwrap(); + assert!(events[flush_at..stop_at].iter().any(|event| matches!(event, Event::Asked(_))), "a read between the flush and the interrupt: {events:?}"); + assert_eq!(outcome.exit_status(), 0); +} + +#[test] +fn a_deadline_that_came_during_a_line_is_answered_without_waiting_for_the_next_poll() { + // The result arrives after the grace has run out: the input closes, and the interrupt is due at once. + let ran = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Step::After(FLUSH_GRACE_S + 5, result_line())], Some(130), Some(Stop::Interrupt), &[], true); + assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 1); + assert_eq!(ran.count(&Event::InputClosed), 1); + let asks = ran.asked(); + assert_eq!(asks[2], 0, "the wait after the result: the deadline is past, not a poll away: {asks:?}"); +} + +#[test] +fn a_child_that_exited_is_never_signalled_however_long_its_stream_stays_open() { + // Flush asked for, the child exits, then the grace and more pass with its stream still open. + let ran = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Exit(Some(0)), Idle(1)], Some(0), None, &[], true); + assert_eq!(ran.events.iter().filter(|event| matches!(event, Event::Stopped(_))).count(), 0); + assert_eq!(ran.ended(), (Cause::FlushRequested, Some(0), 1, CEILING_TOKENS)); + // The stream ending first and the exit after: the same end. + let eof_first = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", 10)), Eof], Some(0), false); + assert_eq!(eof_first.ended(), (Cause::Own, Some(0), 1, 10)); +} + +// ---- how it ends -------------------------------------------------------------- + +#[test] +fn the_outcome_names_how_it_ended_and_the_exit_status_follows_the_childs_code() { + let own = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", 10)), Out(result_line())], Some(0), false); + assert_eq!(own.outcome.as_ref().unwrap(), &SelfRunOutcome::Ended { cause: Cause::Own, code: Some(0), requests: 1, prefix: 10 }); + assert_eq!(own.outcome.as_ref().unwrap().to_string(), "ended: own code 0 requests 1 prefix 10"); + assert_eq!(own.outcome.unwrap().exit_status(), 0); + + let flushed = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())], Some(0), false); + assert_eq!(flushed.outcome.as_ref().unwrap().to_string(), "ended: flush-requested code 0 requests 1 prefix 300000"); + assert_eq!(flushed.outcome.unwrap().exit_status(), 0); + + let nonzero = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", 10))], Some(2), false); + assert_eq!(nonzero.outcome.as_ref().unwrap().to_string(), "ended: own code 2 requests 1 prefix 10"); + assert_eq!(nonzero.outcome.unwrap().exit_status(), 1); + + let signalled = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(FLUSH_GRACE_S)], None, true); + assert_eq!(signalled.outcome.as_ref().unwrap().to_string(), "ended: interrupted signal requests 1 prefix 300000"); + assert_eq!(signalled.outcome.unwrap().exit_status(), 1, "a child killed by a signal has no code"); + + assert_eq!(SelfRunOutcome::NoRun.exit_status(), 3); + assert_eq!(SelfRunOutcome::NoRun.to_string(), "no run: she has no single open Self run"); +} + +#[test] +fn unreadable_and_foreign_lines_are_skipped_and_never_printed() { + let canary = "canary-stream-line-8c41"; + let mut foreign: Value = serde_json::from_str(&init_line()).unwrap(); + foreign["cwd"] = json!(canary); + foreign["type"] = json!(canary); + let ran = go(vec![ + Out(format!("not json {canary}")), + Out(foreign.to_string()), + Out(format!("[\"{canary}\"]")), + Out(at("m1", 10)), + Out(result_line()), + ]); + assert_eq!(ran.out, "request 1 prefix 10\n"); + let outcome = ran.outcome.as_ref().unwrap().to_string(); + assert!(!ran.out.contains(canary) && !outcome.contains(canary)); + assert!(!ran.written().iter().any(|text| text.contains(canary)), "nothing from the stream is written back"); +} + +#[test] +fn the_actuator_admits_nothing_to_the_mind_through_a_whole_flushed_run() { + let mind = with_run(); + let seen = Seen::of(&mind); + let ran = carry_with(&seen, vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())], Some(0), false); + assert_eq!(ran.admits, 0); + assert!(ran.outcome.is_ok()); + // And the run is exactly as she left it: still in force until her flush, or the next launch, closes it. + assert_eq!(mind.ids(K::Resolution), Vec::::new()); +} + +// ---- the real child ----------------------------------------------------------- + +#[cfg(unix)] +mod process { + use std::os::unix::fs::PermissionsExt; + use std::path::PathBuf; + + use super::*; + use crate::launch::SystemClock; + + /// A fake `claude`: the shell script `body`, in a directory that outlives the child. + fn script(dir: &tempfile::TempDir, body: &str) -> PathBuf { + let path = dir.path().join("claude"); + std::fs::write(&path, format!("#!/bin/bash\n{body}\n")).unwrap(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap(); + path + } + + /// Spawns it, retrying the one error a freshly written executable can give + /// while another test's fork still holds its write handle. + fn start(program: &Path, args: &[String]) -> Box { + for _ in 0..50 { + match spawn_program(program, args) { + Ok(child) => return child, + Err(error) if format!("{error:#}").contains("busy") => std::thread::sleep(StdDuration::from_millis(100)), + Err(error) => panic!("{error:#}"), + } + } + panic!("the script stayed busy"); + } + + /// Everything the child gives until it has been reaped and its stdout has ended. + fn drain(child: &mut dyn Child) -> (Vec, Option) { + let (mut lines, mut code, mut eof) = (Vec::new(), None, false); + loop { + match child.next(StdDuration::from_secs(20)) { + Read::Line(line) => lines.push(line), + Read::Eof => eof = true, + Read::Exited(status) => code = Some(status), + Read::Timeout => panic!("the fake claude hung"), + } + if let (true, Some(status)) = (eof, code) { + return (lines, status); + } + } + } + + /// Whether a process of this id still exists. + fn alive(pid: &str) -> bool { + Command::new("kill").args(["-0", pid]).stderr(Stdio::null()).status().unwrap().success() + } + + #[test] + fn the_child_gets_its_arguments_and_lines_in_order_and_ends_when_its_input_closes() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, "echo \"args: $*\"\nwhile read line; do echo \"got: $line\"; done\necho bye\nexit 7"); + let mut child = start(&program, &["-p".to_string(), "--max-turns".to_string(), "500".to_string()]); + child.write_line("first"); + child.write_line("second"); + child.close_input(); + assert_eq!(drain(&mut *child), (["args: -p --max-turns 500", "got: first", "got: second", "bye"].map(String::from).to_vec(), Some(7))); + } + + #[test] + fn a_silent_child_times_out_instead_of_blocking_and_a_signalled_child_has_no_code() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, "kill -9 $$"); + let mut child = start(&program, &[]); + assert_eq!(drain(&mut *child), (Vec::new(), None)); + + let quiet = script(&dir, "exec sleep 5"); + let mut child = start(&quiet, &[]); + assert_eq!(child.next(StdDuration::from_millis(100)), Read::Timeout); + child.stop(Stop::Interrupt); + assert_eq!(drain(&mut *child), (Vec::new(), None), "SIGINT ended the unprepared child"); + } + + #[test] + fn interrupt_is_sigint_the_childs_own_handler_can_catch() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, "trap 'echo interrupted; exit 130' INT\necho ready\nwhile :; do sleep 0.05; done"); + let mut child = start(&program, &[]); + assert_eq!(child.next(StdDuration::from_secs(20)), Read::Line("ready".into())); + child.stop(Stop::Interrupt); + assert_eq!(drain(&mut *child), (vec!["interrupted".to_string()], Some(130))); + } + + #[test] + fn a_program_that_cannot_start_is_an_error_that_names_no_argument() { + let canary = "canary-argument-6b0d"; + let error = spawn_program(Path::new("/nonexistent/claude"), &[canary.to_string()]).err().unwrap(); + assert!(!format!("{error:#}").contains(canary) && !format!("{error:#}").contains("nonexistent"), "{error:#}"); + } + + /// A read that times out here means the stream stopped, never a wait worth 60 s: + /// fail at once instead of hanging the suite. + struct Prompt(Box); + impl Child for Prompt { + fn write_line(&mut self, line: &str) { + self.0.write_line(line) + } + fn close_input(&mut self) { + self.0.close_input() + } + fn next(&mut self, timeout: StdDuration) -> Read { + match self.0.next(StdDuration::from_secs(20).min(timeout)) { + Read::Timeout => panic!("the fake claude went quiet"), + read => read, + } + } + fn stop(&mut self, how: Stop) -> bool { + self.0.stop(how) + } + } + + struct Spawn(PathBuf); + impl Spawner for Spawn { + fn spawn(&self, args: &[String]) -> Result> { + Ok(Box::new(Prompt(start(&self.0, args)))) + } + } + + /// The real child as it is: a read may time out, which is how the drain after an exit is waited. + /// A run that has not ended by 15 s is a failure, not a hang. + struct Bounded(Box, std::time::Instant); + impl Child for Bounded { + fn write_line(&mut self, line: &str) { + self.0.write_line(line) + } + fn close_input(&mut self) { + self.0.close_input() + } + fn next(&mut self, timeout: StdDuration) -> Read { + assert!(self.1.elapsed() < StdDuration::from_secs(15), "the run did not end"); + self.0.next(StdDuration::from_secs(1).min(timeout)) + } + fn stop(&mut self, how: Stop) -> bool { + self.0.stop(how) + } + } + + struct Patient(PathBuf); + impl Spawner for Patient { + fn spawn(&self, args: &[String]) -> Result> { + Ok(Box::new(Bounded(start(&self.0, args), std::time::Instant::now()))) + } + } + + #[test] + fn a_whole_run_through_the_real_child_flushes_at_the_ceiling_and_ends_on_the_result() { + let dir = tempfile::tempdir().unwrap(); + let lines = [at("m1", 10), at("m2", CEILING_TOKENS)].join("\n"); + let result = result_line(); + // Prints two requests, waits for the flush on stdin, then the result, as a stand-in for a Self that obeys. + let body = format!("cat <<'EOF'\n{lines}\nEOF\nread -t 5 start\nread -t 5 flush\ncat <<'EOF'\n{result}\nEOF\nwhile read more; do :; done\nexit 0"); + let program = script(&dir, &body); + let mind = with_run(); + let mut out = Vec::new(); + let outcome = carry(&Seen::of(&mind), &SystemClock, &Spawn(program), &mut out).unwrap(); + assert_eq!(outcome, SelfRunOutcome::Ended { cause: Cause::FlushRequested, code: Some(0), requests: 2, prefix: 300_000 }); + assert_eq!(String::from_utf8(out).unwrap(), "request 1 prefix 10\nrequest 2 prefix 300000\n"); + } + + #[test] + fn a_failure_result_before_any_request_closes_stdin_and_ends_the_run_promptly_through_the_real_child() { + let dir = tempfile::tempdir().unwrap(); + let result = result_line(); + // An authentication or API failure: the error result is the first thing it prints, then it + // waits for its input to end, as `claude -p` over stream-json does, and exits 1. + let program = script(&dir, &format!("read -t 5 start\ncat <<'EOF'\n{result}\nEOF\nwhile read more; do :; done\nexit 1")); + let started = std::time::Instant::now(); + let mut out = Vec::new(); + let outcome = carry(&Seen::of(&with_run()), &SystemClock, &Spawn(program), &mut out).unwrap(); + assert_eq!(outcome, SelfRunOutcome::Ended { cause: Cause::Own, code: Some(1), requests: 0, prefix: 0 }); + assert_eq!(outcome.exit_status(), 1); + assert!(out.is_empty()); + assert!(started.elapsed() < StdDuration::from_secs(20), "it ended promptly"); + } + + /// The Yggdrasil probe shape: the child prints its result and exits, and a background + /// process it started still holds its stdout, so the stream never ends. The run is over + /// when the child is, and the actuator signals nothing: the child is gone. + #[test] + fn a_child_that_exits_while_a_descendant_holds_its_stdout_ends_the_run_without_a_signal() { + for flushed in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let pids = dir.path().join("holder.pid"); + let (result, ceiling) = (result_line(), at("m1", CEILING_TOKENS)); + let before = if flushed { format!("cat <<'EOF'\n{ceiling}\nEOF\nread -t 5 start\nread -t 5 flush\n") } else { "read -t 5 start\n".to_string() }; + let program = script(&dir, &format!("{before}cat <<'EOF'\n{result}\nEOF\nsleep 30 &\necho $! > {}\nexit 0", pids.display())); + let started = std::time::Instant::now(); + let mut out = Vec::new(); + let outcome = carry(&Seen::of(&with_run()), &SystemClock, &Patient(program), &mut out).unwrap(); + let holder = std::fs::read_to_string(&pids).unwrap().trim().to_string(); + let cause = if flushed { Cause::FlushRequested } else { Cause::Own }; + let (requests, prefix) = if flushed { (1, CEILING_TOKENS) } else { (0, 0) }; + assert_eq!(outcome, SelfRunOutcome::Ended { cause, code: Some(0), requests, prefix }, "flushed: {flushed}"); + assert!(alive(&holder), "the actuator signals no one after the child is gone"); + assert!(started.elapsed() < StdDuration::from_secs(20), "it ended on the child's exit: {:?}", started.elapsed()); + Command::new("kill").arg(&holder).status().unwrap(); + } + } + + #[test] + fn each_signal_reaches_a_child_that_ignores_the_weaker_ones_and_one_for_a_gone_child_is_not_delivered() { + let dir = tempfile::tempdir().unwrap(); + let program = script(&dir, "trap '' INT TERM\necho ready\nwhile :; do sleep 0.05; done"); + let mut child = start(&program, &[]); + assert_eq!(child.next(StdDuration::from_secs(20)), Read::Line("ready".into())); + assert!(child.stop(Stop::Interrupt) && child.stop(Stop::Terminate), "delivered, and ignored"); + assert_eq!(child.next(StdDuration::from_millis(300)), Read::Timeout, "still running"); + assert!(child.stop(Stop::Kill)); + assert_eq!(drain(&mut *child), (Vec::new(), None), "SIGKILL leaves no code"); + + let quick = script(&dir, "exit 0"); + let mut gone = start(&quick, &[]); + assert_eq!(drain(&mut *gone), (Vec::new(), Some(0))); + assert!(!gone.stop(Stop::Interrupt), "no such process"); + } + + #[test] + fn a_signal_reaches_the_whole_process_group_the_child_started() { + // Not SIGINT: a background job of a non-interactive shell ignores it, which is the shell's rule, not ours. + for how in [Stop::Terminate, Stop::Kill] { + let dir = tempfile::tempdir().unwrap(); + // The child starts a descendant of its own and waits for it. + let program = script(&dir, "sleep 300 &\necho $!\nwait"); + let mut child = start(&program, &[]); + let Read::Line(holder) = child.next(StdDuration::from_secs(20)) else { panic!("no pid") }; + assert!(alive(&holder)); + assert!(child.stop(how), "{how:?} delivered to the group"); + drain(&mut *child); + let mut waited = 0; + while alive(&holder) && waited < 100 { + std::thread::sleep(StdDuration::from_millis(50)); + waited += 1; + } + assert!(!alive(&holder), "{how:?} reached the descendant too"); + } + } +} diff --git a/crates/mind-body/tests/cli.rs b/crates/mind-body/tests/cli.rs index 159efc0..fdc2105 100644 --- a/crates/mind-body/tests/cli.rs +++ b/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 "; + + fn parse(args: &[&str], env: Option<&str>) -> anyhow::Result<(eureka_pipeline::Slug, std::net::SocketAddr)> { + let args: Vec = 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://:"), + (vec!["--instance", "a", "--endpoint", "http://127.0.0.1:9"], None, "the endpoint is not rudp://:"), + (vec!["--instance", "a", "--endpoint", "rudp://not-an-address"], None, "the endpoint is not rudp://:"), + (vec!["--instance", "a"], Some(canary), "the endpoint is not rudp://:"), + ]; + for (args, env, message) in cases { + let error = parse(&args, env).unwrap_err(); + assert_eq!(format!("{error:#}"), message, "{args:?} {env:?}"); + } + } +} diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 405b5e6..354f7ee 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/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 > diff --git a/crates/mind-body/tests/self_run.rs b/crates/mind-body/tests/self_run.rs new file mode 100644 index 0000000..9c6d1b2 --- /dev/null +++ b/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 { + 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) -> (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"); + } +}