diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 58e05f8..220c262 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -8,10 +8,14 @@ //! `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 prefix has grown by the headroom beyond the -//! prefix the flush was asked at (ruling +//! 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. +//! 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 @@ -19,16 +23,14 @@ //! foreign owner's JSON: it is parsed into `serde_json::Value` here and nothing //! parsed is stored. -use std::collections::HashSet; use std::fmt; use std::io::{BufRead, BufReader, Write}; use std::path::Path; -use std::process::{Child as OsChild, ChildStdin, Command, Stdio}; +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 chrono::{DateTime, Duration, Utc}; use cultnet_rs::Selection; use eureka_pipeline::{PipelineDocument, PipelineRef, RunTurn, Slug}; use huginn_mind::PipelinePageItems; @@ -38,6 +40,8 @@ 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 @@ -47,6 +51,9 @@ pub const FLUSH_HEADROOM_TOKENS: u64 = 60_000; 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. @@ -129,13 +136,15 @@ pub fn read_usage(line: &str) -> Usage { } } -/// What the child's stdout gave in one wait. +/// 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, - /// The stream ended: the child is gone or going. - Closed, + /// 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. @@ -145,11 +154,9 @@ pub trait Child { fn write_line(&mut self, line: &str); /// Closes its input; later writes go nowhere. fn close_input(&mut self); - fn next_line(&mut self, timeout: StdDuration) -> Read; - /// Sends it the signal; whether the signal was delivered. + 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; - /// Its exit code, none when a signal ended it. - fn wait(&mut self) -> Option; } /// The signals the actuator sends, weakest first. @@ -245,76 +252,45 @@ fn her_run(mind: &dyn MindPort) -> Result> { /// 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 seen, mut requests, mut prefix) = (HashSet::new(), 0u64, 0u64); - let mut result_seen = false; - // The prefix and the grace deadline of the flush, once asked for. - let mut flush: Option<(u64, DateTime)> = None; - // The last signal attempted and when, and the strongest one delivered. - let mut stopping: Option<(Stop, DateTime)> = None; - let mut delivered: Option = None; + let mut state = State::default(); + let mut now = clock.now(); loop { - // The one deadline the wait must not pass: the next escalation once a signal went out, else the grace. - let deadline = match (stopping, flush) { - (Some((_, sent)), _) => Some(sent + Duration::seconds(ESCALATE_AFTER_S)), - (None, Some((_, grace_ends))) => Some(grace_ends), - (None, None) => None, - }; - let wait = deadline.map_or(POLL, |ends| (ends - clock.now()).to_std().unwrap_or_default().min(POLL)); - match child.next_line(wait) { + 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) { - // A message id seen before is the same request again (F24). - Usage::Request { id, prefix: newest } if seen.insert(id.clone()) => { - requests += 1; - prefix = newest; - writeln!(out, "request {requests} prefix {prefix}")?; - } - Usage::Result if !result_seen => { - result_seen = true; - child.close_input(); - } - _ => {} + Usage::Request { id, prefix } => Event::Request { id, prefix }, + Usage::Result => Event::Result, + Usage::Other | Usage::Unreadable => Event::Tick, }, - Read::Timeout => {} - Read::Closed => break, - } - if flush.is_none() && !result_seen && prefix >= CEILING_TOKENS { - child.write_line(&user_line(FLUSH_MESSAGE)); - flush = Some((prefix, clock.now() + Duration::seconds(FLUSH_GRACE_S))); - } - // The backstop is relative to the flush point, so it cannot fire in the pass that wrote the flush. - let due = match (stopping, flush) { - (Some((_, sent)), _) => clock.now() >= sent + Duration::seconds(ESCALATE_AFTER_S), - (None, Some((at, grace_ends))) => clock.now() >= grace_ends || at.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit), - (None, None) => false, + Read::Timeout => Event::Tick, + Read::Eof => Event::StdoutEof, + Read::Exited(code) => Event::ChildExited(code), }; - if due { - let how = stopping.map_or(Stop::Interrupt, |(last, _)| last.next()); - if child.stop(how) { - delivered = Some(how); + 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)?; } - stopping = Some((how, clock.now())); - if how == Stop::Kill { - // A grandchild can hold the stream open past a dead child: do not wait for its end. - break; + 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), } } } - if matches!(stopping, Some((Stop::Kill, _))) && delivered != Some(Stop::Kill) { - return Ok(SelfRunOutcome::Ended { cause: Cause::Unstoppable, code: None, requests, prefix }); - } - let code = child.wait(); - let cause = match (delivered, flush) { - (Some(how), _) => Cause::Signalled(how), - (None, Some(_)) => Cause::FlushRequested, - (None, None) => Cause::Own, - }; - Ok(SelfRunOutcome::Ended { cause, code, requests, prefix }) } /// The real spawner: the `claude` on the unit's `PATH`, with the unit's @@ -328,25 +304,38 @@ impl Spawner for ClaudeSpawner { } fn spawn_program(program: &Path, args: &[String]) -> Result> { - let mut process = Command::new(program).args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit()).spawn().context("claude could not be started")?; + 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, lines) = channel(); + 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 { return }; - if sender.send(String::from_utf8_lossy(&bytes).into_owned()).is_err() { + 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); }); - Ok(Box::new(ClaudeProcess { process, stdin, lines })) + // 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 { - process: OsChild, stdin: Option, - lines: Receiver, + facts: Receiver, + /// The child's pid, which is its process group's id. + group: u32, } impl Child for ClaudeProcess { @@ -360,28 +349,26 @@ impl Child for ClaudeProcess { self.stdin = None; } - fn next_line(&mut self, timeout: StdDuration) -> Read { - match self.lines.recv_timeout(timeout) { - Ok(line) => Read::Line(line), + 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::Closed, + 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 => return self.process.kill().is_ok(), + Stop::Interrupt => "INT", + Stop::Terminate => "TERM", + Stop::Kill => "KILL", }; - Command::new("kill").args([signal, &self.process.id().to_string()]).status().is_ok_and(|status| status.success()) - } - - fn wait(&mut self) -> Option { - self.process.wait().ok().and_then(|status| status.code()) + 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/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 index 0159c33..4592fa3 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -69,7 +69,6 @@ enum Event { /// A signal was attempted, delivered or not. Stopped(Stop), Asked(StdDuration), - Waited, } #[derive(Clone)] @@ -77,9 +76,15 @@ 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::{Idle, Line as Out}; +use Step::{Eof, Exit, Idle, Line as Out}; struct Scripted { steps: VecDeque, @@ -90,6 +95,10 @@ struct Scripted { 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, } impl Child for Scripted { @@ -101,7 +110,7 @@ impl Child for Scripted { self.log.borrow_mut().push(Event::InputClosed); } - fn next_line(&mut self, timeout: StdDuration) -> Read { + fn next(&mut self, timeout: StdDuration) -> Read { self.log.borrow_mut().push(Event::Asked(timeout)); match self.steps.pop_front() { Some(Step::Line(line)) => Read::Line(line), @@ -109,7 +118,32 @@ impl Child for Scripted { self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds)); Read::Timeout } - None => Read::Closed, + 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"), } } @@ -120,14 +154,10 @@ impl Child for Scripted { } if self.dies_on.is_some_and(|weakest| how >= weakest) { self.steps.clear(); + self.hangs = false; } true } - - fn wait(&mut self) -> Option { - self.log.borrow_mut().push(Event::Waited); - self.code - } } struct Fake { @@ -235,10 +265,14 @@ fn carry_with(seen: &Seen, steps: Vec, code: Option, dies_on_interrup } 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() })), + 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 })), spawns: RefCell::new(Vec::new()), refuse: false, }; @@ -529,50 +563,59 @@ fn after_the_flush_the_grace_interrupts_exactly_once_and_not_a_second_early() { // 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.count(&Event::Waited), 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)]); - assert_eq!(ran.asked(), [60, 60, 60, 1, 60], "{:?}", ran.asked()); + // 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 a_prefix_at_the_ceiling_plus_the_allowance_interrupts_at_once_and_one_short_of_it_does_not() { +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(at("m2", hard - 1))]); + 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(at("m2", hard)), Out(at("m3", hard + 1))], Some(130), true); + 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_gives_it_a_turn_before_any_interrupt() { - let jump = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", 290_000)), Out(at("m2", 400_000)), Out(at("m3", 430_000)), Out(at("m4", 459_999)), Out(at("m5", 460_000)), Out(at("m6", 470_000))], Some(130), true); +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), then a read after it before anything else. - assert_eq!(kinds[..4], ["read", "read", "flush", "read"], "{kinds:?}"); - // The headroom is measured from the prefix the flush was asked at (400k), not from the ceiling: - // 430k is past 300k + 60k and does not interrupt; 459_999 does not; 460_000 does, once. + // 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!(kinds.iter().filter(|kind| **kind == "read").count(), 6, "interrupted on m5, the sixth read ends the stream: {kinds:?}"); - assert_eq!(jump.out, "request 1 prefix 290000\nrequest 2 prefix 400000\nrequest 3 prefix 430000\nrequest 4 prefix 459999\nrequest 5 prefix 460000\n"); + 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); } @@ -581,7 +624,7 @@ fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_gives_it_a_tur 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.count(&Event::Waited), 1); + assert_eq!(ran.ended().0, Cause::FlushRequested); } #[test] @@ -649,7 +692,6 @@ fn a_child_that_ignores_a_signal_gets_the_next_one_after_the_bound_and_the_outco 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)); - assert_eq!(killed.count(&Event::Waited), 1); // One second short of the bound there is no second signal. let waiting = hung(None, &[], vec![Idle(after - 1)]); @@ -675,13 +717,74 @@ fn a_signal_that_could_not_be_delivered_is_never_reported_as_sent() { let partly = hung(&[Stop::Terminate], vec![Idle(ESCALATE_AFTER_S)]); assert_eq!(partly.ended().0, Cause::Signalled(Stop::Interrupt)); - // Not even SIGKILL is delivered: the run ends at once, the child is not waited for, and the label says so. - let stuck = hung(&[Stop::Interrupt, Stop::Terminate, Stop::Kill], vec![Idle(ESCALATE_AFTER_S), Idle(ESCALATE_AFTER_S), Idle(ESCALATE_AFTER_S)]); + // 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.count(&Event::Waited), 0); + 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 }; + 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] @@ -768,17 +871,27 @@ mod process { panic!("the script stayed busy"); } - fn drain(child: &mut dyn Child) -> Vec { - let mut lines = Vec::new(); + /// 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_line(StdDuration::from_secs(20)) { + match child.next(StdDuration::from_secs(20)) { Read::Line(line) => lines.push(line), - Read::Closed => return lines, + 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(); @@ -787,8 +900,7 @@ mod process { 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"]); - assert_eq!(child.wait(), Some(7)); + assert_eq!(drain(&mut *child), (["args: -p --max-turns 500", "got: first", "got: second", "bye"].map(String::from).to_vec(), Some(7))); } #[test] @@ -796,15 +908,13 @@ mod process { let dir = tempfile::tempdir().unwrap(); let program = script(&dir, "kill -9 $$"); let mut child = start(&program, &[]); - assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Closed); - assert_eq!(child.wait(), None); + assert_eq!(drain(&mut *child), (Vec::new(), None)); let quiet = script(&dir, "exec sleep 5"); let mut child = start(&quiet, &[]); - assert_eq!(child.next_line(StdDuration::from_millis(100)), Read::Timeout); + assert_eq!(child.next(StdDuration::from_millis(100)), Read::Timeout); child.stop(Stop::Interrupt); - assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Closed); - assert_eq!(child.wait(), None, "SIGINT ended the unprepared child"); + assert_eq!(drain(&mut *child), (Vec::new(), None), "SIGINT ended the unprepared child"); } #[test] @@ -812,10 +922,9 @@ mod process { 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_line(StdDuration::from_secs(20)), Read::Line("ready".into())); + assert_eq!(child.next(StdDuration::from_secs(20)), Read::Line("ready".into())); child.stop(Stop::Interrupt); - assert_eq!(drain(&mut *child), ["interrupted"]); - assert_eq!(child.wait(), Some(130)); + assert_eq!(drain(&mut *child), (vec!["interrupted".to_string()], Some(130))); } #[test] @@ -835,8 +944,8 @@ mod process { fn close_input(&mut self) { self.0.close_input() } - fn next_line(&mut self, timeout: StdDuration) -> Read { - match self.0.next_line(StdDuration::from_secs(20).min(timeout)) { + 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, } @@ -844,9 +953,6 @@ mod process { fn stop(&mut self, how: Stop) -> bool { self.0.stop(how) } - fn wait(&mut self) -> Option { - self.0.wait() - } } struct Spawn(PathBuf); @@ -856,31 +962,27 @@ mod process { } } + /// The real child as it is: a read may time out, which is how the drain after an exit is waited. + struct Patient(PathBuf); + impl Spawner for Patient { + fn spawn(&self, args: &[String]) -> Result> { + Ok(start(&self.0, args)) + } + } + #[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(" -"); + 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' -{lines} -EOF -read -t 5 start -read -t 5 flush -cat <<'EOF' -{result} -EOF -while read more; do :; done -exit 0"); + 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 -request 2 prefix 300000 -"); + assert_eq!(String::from_utf8(out).unwrap(), "request 1 prefix 10\nrequest 2 prefix 300000\n"); } #[test] @@ -889,12 +991,7 @@ request 2 prefix 300000 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 -cat <<'EOF' -{result} -EOF -while read more; do :; done -exit 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(); @@ -904,24 +1001,65 @@ exit 1")); 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 -echo ready -while :; do sleep 0.05; done"); + let program = script(&dir, "trap '' INT TERM\necho ready\nwhile :; do sleep 0.05; done"); let mut child = start(&program, &[]); - assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Line("ready".into())); + 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_line(StdDuration::from_millis(300)), Read::Timeout, "still running"); + assert_eq!(child.next(StdDuration::from_millis(300)), Read::Timeout, "still running"); assert!(child.stop(Stop::Kill)); - assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Closed); - assert_eq!(child.wait(), None, "SIGKILL leaves no code"); + 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!(gone.next_line(StdDuration::from_secs(20)), Read::Closed); - assert_eq!(gone.wait(), Some(0)); + 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/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 4514e39..354f7ee 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -75,12 +75,13 @@ mind_body::queue: pub trait MindPort { fn instance (& self) -> & Slug ; fn query 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 , Closed , } +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 @@ -97,7 +98,7 @@ 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_line (& mut self , timeout : StdDuration) -> Read ; fn stop (& mut self , how : Stop) -> bool ; fn wait (& mut self) -> Option < i32 > ; } +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) , }