diff --git a/crates/mind-body/src/bin/mind-self-run.rs b/crates/mind-body/src/bin/mind-self-run.rs index 52322cf..3a1e9b8 100644 --- a/crates/mind-body/src/bin/mind-self-run.rs +++ b/crates/mind-body/src/bin/mind-self-run.rs @@ -1,3 +1,8 @@ +//! `mind-self-run inject --spool [--pre-compact]` is the hook `claude` runs +//! inside the session: it prints the briefs waiting in the spool as context for +//! the next request, and with `--pre-compact` blocks an auto-compaction (exit 2). +//! It reads the hook's JSON on stdin and the spool, nothing else. +//! //! `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, @@ -13,6 +18,8 @@ //! The connector's lease socket is `CLAUDE_LEASE_SOCKET`: the token itself is //! never in this process's environment, only the descriptor the lease grants. +use std::io::Read; +use std::path::PathBuf; use std::time::Duration; use anyhow::{Result, anyhow}; @@ -21,6 +28,7 @@ use mind_body::cli::instance_and_endpoint; use mind_body::launch::SystemClock; use mind_body::lease_client::{LEASE_SOCKET_VARIABLE, SocketLender}; use mind_body::queue::HuginnMind; +use mind_body::self_run::inbox::inject; use mind_body::self_run::{ClaudeSpawner, RUN_CAP_VARIABLE, carry}; /// How long one call to the mind may take. @@ -33,6 +41,9 @@ const USAGE: &str = "usage: mind-self-run --instance [--endpoint rudp://< #[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and environment, and hands them to the library")] fn run() -> Result { let args: Vec = std::env::args().skip(1).collect(); + if args.first().map(String::as_str) == Some("inject") { + return hook(&args[1..]); + } let (instance, address) = instance_and_endpoint(&args, std::env::var("HUGINN_ENDPOINT").ok(), USAGE)?; let socket = std::env::var_os(LEASE_SOCKET_VARIABLE).ok_or_else(|| anyhow!("{LEASE_SOCKET_VARIABLE} is not set"))?; let run_cap_s: u64 = std::env::var(RUN_CAP_VARIABLE) @@ -41,11 +52,24 @@ fn run() -> Result { .filter(|seconds| *seconds > 0) .ok_or_else(|| anyhow!("{RUN_CAP_VARIABLE} is not set to the unit's RuntimeMaxSec in seconds"))?; let mind = HuginnMind::new(HuginnClient::new(address, instance, TIMEOUT)); - let outcome = carry(&mind, &SystemClock, &SocketLender::new(socket), &ClaudeSpawner::new(std::env::vars_os())?, run_cap_s, &mut std::io::stdout())?; + let outcome = carry(&mind, &SystemClock, &SocketLender::new(socket), &ClaudeSpawner::new(std::env::vars_os(), std::env::current_exe()?)?, run_cap_s, &mut std::io::stdout())?; println!("{outcome}"); Ok(outcome.exit_status()) } +/// The hook: the spool named by `--spool`, the hook's input on stdin. +fn hook(args: &[String]) -> Result { + let spool = args.iter().position(|arg| arg == "--spool").and_then(|at| args.get(at + 1)).map(PathBuf::from).ok_or_else(|| anyhow!("inject needs --spool "))?; + let mut input = String::new(); + std::io::stdin().read_to_string(&mut input)?; + let injection = inject(&spool, &input, args.iter().any(|arg| arg == "--pre-compact"))?; + if injection.exit == 2 { + eprintln!("compaction blocked: the actuator asks for the flush"); + } + print!("{}", injection.stdout); + Ok(injection.exit) +} + fn main() { match run() { Ok(status) => std::process::exit(status), diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index fe9cfcf..348fce3 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -6,13 +6,20 @@ //! 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 +//! the mind lacks and ends, it never summarises), 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. //! +//! One run is one live session (ruling `eureka-body:ruling:live-mind-sessions`). +//! A result ends a turn, not the session. What her Self is asked is derived from +//! her mind (the `inbox` module): an open follow-up with owner `self-run` is a +//! brief not yet answered. The actuator polls for briefs and tells them to the +//! session, between tool calls through the PostToolBatch hook and between turns +//! as a user line. The session ends where its cache would be evicted, near the +//! unit's cap, at the context ceiling, or after the flush each of those asks. +//! //! 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. @@ -38,6 +45,7 @@ //! foreign owner's JSON: it is parsed into `serde_json::Value` here and nothing //! parsed is stored. +use std::collections::VecDeque; use std::ffi::{OsStr, OsString}; use std::fmt; use std::io::{BufRead, BufReader, Write}; @@ -47,7 +55,7 @@ use std::sync::mpsc::{Receiver, RecvTimeoutError, channel}; use std::time::Duration as StdDuration; use anyhow::{Context, Result, anyhow, bail}; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Duration as ChronoDuration, Utc}; use cultnet_rs::Selection; use eureka_pipeline::{PipelineDocument, PipelineRef, ResolutionOutcome, RunTurn, Slug}; use huginn_mind::{PipelineAdmissionOutcome, PipelinePageItems}; @@ -61,6 +69,7 @@ use model_connector::PlanQuota; use crate::queue::MindPort; use decide::{Action, Event, State, decide, next_deadline}; +use inbox::{Brief, Hooks}; /// The measured Self ceiling: at this prefix the flush is asked for. pub const CEILING_TOKENS: u64 = 300_000; @@ -74,6 +83,16 @@ 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; +/// How long before the unit's cap an idle session is closed for rotation. +pub const ROTATE_MARGIN_S: i64 = 20 * 60; +/// How long before the unit's cap a working session is asked for its flush. +pub const FLUSH_MARGIN_S: i64 = 10 * 60; +/// How often the mind is asked for new briefs. +pub const INBOX_POLL_S: i64 = 30; +/// The prompt cache's lifetimes, and how long before one runs out the session ends. +pub const TTL_1H_S: i64 = 3600; +pub const TTL_5M_S: i64 = 300; +pub const TTL_MARGIN_S: i64 = 60; /// A backstop `claude` enforces; the ceiling is reached first. pub const MAX_TURNS: u32 = 500; /// Names the variable the Self unit sets to its own `RuntimeMaxSec`, in seconds: @@ -101,13 +120,13 @@ pub const AGENT: &str = "mind-self-run"; 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); +const POLL: StdDuration = StdDuration::from_secs(INBOX_POLL_S as u64); /// What the child is told at the start. The contract with the `eureka` /// skill: it names the skill, the instance and the run id. pub fn start_message(instance: &Slug, run: &PipelineRef, plan_quota: &[PlanQuota], now_ms: u64) -> String { format!( - "Use the eureka skill. You are Self for the instance `{}`; your run is `{}`. Admit as faculty SelfFaculty, agent {AGENT}.\n\n{}", + "Use the eureka skill. You are Self for the instance `{}`; your run is `{}`. Admit as faculty SelfFaculty, agent {AGENT}.\n\nThis session stays open after you answer: briefs from your Persona arrive while you work and between tasks. Work you take on beyond an answer is a follow-up of your own.\n\n{}", instance.0, run.id.0, render_plan(plan_quota, now_ms) @@ -160,7 +179,9 @@ pub fn session_id(run: &PipelineRef) -> Uuid { } /// The argument list of the child, and nothing else: no resume, continue or -/// fork, no model and no settings of its own belong to this actuator. It fixes +/// fork, no model and no settings of its own belong to this actuator, except the +/// hooks the actuator builds on its own argv (`--settings`, never a file in her +/// sandbox): the tell between tool calls and the block on auto-compaction. It fixes /// the permission mode: print mode starts in Manual when nothing on the command /// line sets a mode (the managed `defaultMode` was not applied in print mode, /// 2026-10-31), and `--permission-prompts none` then denies every tool that @@ -168,7 +189,7 @@ pub fn session_id(run: &PipelineRef) -> Uuid { /// user's settings load (the fresh configuration directory /// the seed fills); the working directory's project and local settings, which a /// run could have written, do not. -pub fn claude_args(session: Uuid) -> Vec { +pub fn claude_args(session: Uuid, hooks: &Hooks) -> Vec { [ "-p", "--input-format", @@ -186,6 +207,8 @@ pub fn claude_args(session: Uuid) -> Vec { "none", "--setting-sources", "user", + "--settings", + &hooks.settings(), ] .map(String::from) .to_vec() @@ -204,6 +227,18 @@ pub enum Usage { Unreadable, } +/// Whether a main-thread request wrote its prompt cache for an hour (the split +/// `cache_creation.ephemeral_1h_input_tokens`), else for five minutes. None for +/// any other line. +pub fn writes_hour_cache(line: &str) -> Option { + let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return None }; + if event.get("type").and_then(Value::as_str) != Some("assistant") || !event.get("parent_tool_use_id").is_none_or(Value::is_null) { + return None; + } + let hour = event.get("message")?.get("usage")?.get("cache_creation")?.get("ephemeral_1h_input_tokens").and_then(Value::as_u64); + Some(hour.unwrap_or(0) > 0) +} + /// 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 }; @@ -264,6 +299,8 @@ impl Stop { } pub trait Spawner { + /// The spool the child's hooks drain and the program they run. + fn hooks(&self) -> Hooks; /// Starts the child with `token` as its credential descriptor. fn spawn(&self, args: &[String], token: &TokenFd) -> Result>; } @@ -426,7 +463,21 @@ fn withdraw_misconfigured(mind: &dyn MindPort, clock: &dyn Clock, run: &Pipeline /// error, and the launch that opens the next run closes the run left open. pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write) -> Result { let Some((run, claims)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) }; - let task = task_section(mind, &claims)?; + let hooks = spawner.hooks(); + inbox::reset(&hooks.spool)?; + let mut task = task_section(mind, &claims)?; + let open = inbox::derive(mind, &hooks.spool)?; + let also: Vec<&Brief> = open.iter().filter(|brief| !claims.iter().any(|claim| claim.id.0 == brief.id)).collect(); + if !also.is_empty() { + task.push_str(" + +Also open for you:"); + for brief in also { + task.push_str(&format!(" +- follow-up `{}` (campaign `{}`, repo `{}`): {}", brief.id, brief.campaign, brief.repo, brief.item)); + } + } + inbox::mark_told(&hooks.spool, &open)?; let mut heard = Heard::default(); let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &task), &mut heard); let closed = close_if_open(mind, clock, &run, &ran, &heard); @@ -452,8 +503,9 @@ fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, s } Lent::Refused(refused) => return close_refused(mind, clock, run, refused.kind), }; - let child = spawner.spawn(&claude_args(session_id(run)), lease.token())?; - drive(mind, clock, child, (run, task), (lease.plan_quota(), started_ms), heard, out) + let hooks = spawner.hooks(); + let child = spawner.spawn(&claude_args(session_id(run), &hooks), lease.token())?; + drive(mind, clock, child, &hooks, (run, task), (lease.plan_quota(), started_ms, run_cap_s), heard, out) } /// The most of a result's text that reaches the journal. @@ -504,21 +556,41 @@ fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> { } /// The loop: facts in, one clock read per fact, actions out. -fn drive(mind: &dyn MindPort, clock: &dyn Clock, mut child: Box, (run, task): (&PipelineRef, &str), (plan_quota, now_ms): (&[PlanQuota], u64), heard: &mut Heard, out: &mut dyn Write) -> Result { +#[allow(clippy::too_many_arguments, reason = "the loop's ports and the run's facts, passed on whole")] +fn drive( + mind: &dyn MindPort, + clock: &dyn Clock, + mut child: Box, + hooks: &Hooks, + (run, task): (&PipelineRef, &str), + (plan_quota, now_ms, run_cap_s): (&[PlanQuota], u64, u64), + heard: &mut Heard, + out: &mut dyn Write, +) -> Result { let user_line = |text: &str| json!({"type": "user", "message": {"role": "user", "content": text}}).to_string(); + let plan = render_plan(plan_quota, now_ms); + inbox::write_plan(&hooks.spool, &plan)?; child.write_line(&user_line(&format!("{}{task}", start_message(mind.instance(), run, plan_quota, now_ms)))); - let mut state = State::default(); let mut now = clock.now(); + let mut state = State::started(now, run_cap_s); + let mut polled = now; + let mut ttl_1h = false; + let mut closed = false; loop { - let wait = next_deadline(&state).map_or(POLL, |at| (at - now).to_std().unwrap_or_default().min(POLL)); + let poll_due = polled + ChronoDuration::seconds(INBOX_POLL_S); + let wake = next_deadline(&state).into_iter().chain((!closed).then_some(poll_due)).min(); + let wait = wake.map_or(POLL, |at| (at - now).to_std().unwrap_or_default().min(POLL)); let fact = match child.next(wait) { Read::Line(line) => { heard.hear(&line); write_result(&line, out)?; + if let Some(hour) = writes_hour_cache(&line) { + ttl_1h = hour; + } match read_usage(&line) { Usage::Request { id, prefix } => Event::Request { id, prefix }, - Usage::Result => Event::Result, + Usage::Result => Event::Result { ttl_1h }, Usage::Other | Usage::Unreadable => Event::Tick, } } @@ -527,20 +599,51 @@ fn drive(mind: &dyn MindPort, clock: &dyn Clock, mut child: Box, (run Read::Exited(code) => Event::ChildExited(code), }; now = clock.now(); - let mut event = Some(fact); - while let Some(heard) = event.take() { + let mut events = VecDeque::from([fact]); + if !closed && now >= poll_due { + polled = now; + match inbox::derive(mind, &hooks.spool).and_then(|briefs| inbox::spool(&hooks.spool, &briefs)).and_then(|()| inbox::pending(&hooks.spool)) { + Ok(pending) => events.push_back(Event::Inbox { pending }), + Err(_) => writeln!(out, "inbox poll failed")?, + } + } + if inbox::take_pre_compact(&hooks.spool) { + events.push_back(Event::PreCompact); + } + while let Some(heard_event) = events.pop_front() { let counted = state.requests; - let (next, action) = decide(state, heard, now); + let (next, action) = decide(state, heard_event, now); state = next; if state.requests != counted { writeln!(out, "request {} prefix {}", state.requests, state.prefix)?; } - match action { + 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), + Action::WriteTells => match inbox::claim(&hooks.spool) { + Ok(briefs) if !briefs.is_empty() => { + child.write_line(&user_line(&inbox::render(&briefs, &plan))); + events.push_front(Event::Told); + } + Ok(_) => events.push_front(Event::Inbox { pending: false }), + Err(_) => { + writeln!(out, "inbox claim failed")?; + events.push_front(Event::Inbox { pending: false }); + } + }, + Action::CloseStdin => { + child.close_input(); + closed = true; + } + Action::Signal(how) => events.push_front(Event::Signalled { how: *how, delivered: child.stop(*how) }), + Action::Finish(_) => {} + } + if !matches!(action, Action::None | Action::Finish(_)) { + let idle = state.idle_s(now).map_or_else(|| "working".to_string(), |seconds| seconds.to_string()); + writeln!(out, "decision {action:?} prefix {} ttl_s {} idle_s {idle} inbox {}", state.prefix, state.ttl_s, state.inbox)?; + } + if let Action::Finish(outcome) = action { + return Ok(outcome); } } } @@ -555,21 +658,31 @@ pub struct ClaudeSpawner { unit_environment: Vec<(OsString, OsString)>, scratch: PathBuf, seed: Option, + /// The program the child's hooks run: this actuator's own binary. + injector: PathBuf, } impl ClaudeSpawner { /// From the unit's environment: it must name [`SCRATCH_VARIABLE`], and may name [`SEED_VARIABLE`]. - pub fn new(unit_environment: impl IntoIterator) -> Result { + /// `injector` is the absolute path of `mind-self-run`, which the hooks run as `inject`. + pub fn new(unit_environment: impl IntoIterator, injector: PathBuf) -> Result { let unit_environment: Vec<(OsString, OsString)> = unit_environment.into_iter().collect(); let named = |variable: &str| unit_environment.iter().find(|(name, _)| name == variable).map(|(_, value)| PathBuf::from(value)); // systemd lists several runtime directories colon-separated; the unit names one. let scratch = named(SCRATCH_VARIABLE).filter(|path| !path.as_os_str().is_empty()).ok_or_else(|| anyhow!("{SCRATCH_VARIABLE} is not set: the unit needs RuntimeDirectory="))?; let seed = named(SEED_VARIABLE).filter(|path| !path.as_os_str().is_empty()); - Ok(Self { unit_environment, scratch, seed }) + Ok(Self { unit_environment, scratch, seed, injector }) } } +/// The directory name of the spool inside the scratch directory. +const SPOOL_DIR_NAME: &str = "inbox"; + impl Spawner for ClaudeSpawner { + fn hooks(&self) -> Hooks { + Hooks { spool: self.scratch.join(SPOOL_DIR_NAME), injector: self.injector.clone() } + } + fn spawn(&self, args: &[String], token: &TokenFd) -> Result> { let config = FreshConfig::create(&self.scratch, self.seed.as_deref())?; spawn_program(Path::new("claude"), args, token, &self.unit_environment, config) @@ -778,6 +891,7 @@ impl Child for ClaudeProcess { } mod decide; +pub mod inbox; #[cfg(test)] mod tests; diff --git a/crates/mind-body/src/self_run/decide.rs b/crates/mind-body/src/self_run/decide.rs index e9f96e8..5403d3b 100644 --- a/crates/mind-body/src/self_run/decide.rs +++ b/crates/mind-body/src/self_run/decide.rs @@ -1,78 +1,119 @@ -//! 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, -} - +//! 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 is 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 action at the next event, and `next_deadline` says it is now. +//! +//! A result ends a turn, never the session: it marks the session idle. A brief +//! waiting while idle is told. +//! The session ends where its cache would be evicted (ruling +//! `eureka-body:ruling:live-session-floor-cache-evicts`): at the last request's +//! cache TTL less a margin, while no brief waits, the flush is asked, and stdin +//! closes after the flush's own result. Under the run cap it closes at an idle +//! rotation deadline; the PreCompact hook and the cap's margin ask the same +//! flush the ceiling asks. + +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, FLUSH_MARGIN_S, ROTATE_MARGIN_S, SelfRunOutcome, Stop, TTL_1H_S, TTL_5M_S, + TTL_MARGIN_S, +}; + +/// 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 }, + /// A turn ended. `ttl_1h`: the last request wrote its cache for an hour, else for five minutes. + Result { ttl_1h: bool }, + /// A poll of the inbox: whether a brief waits in the spool. + Inbox { pending: bool }, + /// A user line carrying briefs was written: a turn has begun. + Told, + /// The PreCompact hook blocked an auto-compaction. + PreCompact, + /// `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, + /// Claim the spool and write its briefs as one user line (the session is idle). + WriteTells, + 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, + /// A turn is running or queued: the start message, a tell and the flush begin one; a result ends it. + busy: bool, + last_result_at: Option>, + /// The cache lifetime the last result left, in seconds. + pub(super) ttl_s: i64, + pub(super) inbox: bool, + /// When the process started and the unit's cap in seconds; none where no cap is known. + cap: Option<(DateTime, i64)>, + stdin_closed: 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, +} + +impl State { + /// A session whose start message was just written: its first turn is running. + pub(super) fn started(at: DateTime, run_cap_s: u64) -> Self { + Self { busy: true, cap: Some((at, i64::try_from(run_cap_s).unwrap_or(i64::MAX))), ..Self::default() } + } + + /// Seconds idle since the last result, none while a turn runs. + pub(super) fn idle_s(&self, now: DateTime) -> Option { + (!self.busy).then(|| self.last_result_at.map_or(0, |at| (now - at).num_seconds())) + } +} + fn outcome(state: &State, cause: Cause, code: Option) -> SelfRunOutcome { SelfRunOutcome::Ended { cause, code, requests: state.requests, prefix: state.prefix } } @@ -86,100 +127,172 @@ fn cause(state: &State) -> Cause { } } -/// 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), - } -} +/// 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)); + } + let signals = match (&state.stopping, &state.flush) { + (Some((_, sent)), _) => return Some(*sent + Duration::seconds(ESCALATE_AFTER_S)), + (None, Some(flush)) => Some(flush.written + Duration::seconds(FLUSH_GRACE_S)), + (None, None) => None, + }; + signals.into_iter().chain(lifecycle_deadline(state)).min() +} + +/// The deadline of the next lifecycle action: the flush margin while a turn +/// runs; while idle, now for a pending tell or a finished flush, else the +/// earlier of the rotation and the cache. +fn lifecycle_deadline(state: &State) -> Option> { + if state.stdin_closed { + return None; + } + let rotation = |margin: i64| state.cap.map(|(started, cap)| started + Duration::seconds(cap - margin)); + if state.busy { + return if state.flush.is_none() { rotation(FLUSH_MARGIN_S) } else { None }; + } + let last = state.last_result_at?; + if state.flush.as_ref().is_some_and(|flush| flush.turn_begun) || state.inbox { + return Some(last); + } + [rotation(ROTATE_MARGIN_S), Some(last + Duration::seconds(state.ttl_s - TTL_MARGIN_S))].into_iter().flatten().min() +} + +/// The flush is asked once, by the ceiling, the PreCompact hook, the cap's +/// margin or the cache's end; never to a child whose input is closed or that has exited. +fn ask_flush(state: &mut State, now: DateTime) -> bool { + if state.flush.is_some() || state.stdin_closed || state.exited.is_some() { + return false; + } + // Written while idle, its turn begins at once; written mid-turn, it is the next one. + state.flush = Some(Flush { prefix: state.prefix, written: now, turn_begun: !state.busy, exhausted: false }); + state.busy = true; + true +} + +/// What the session does next on its own. Mid-turn: the flush at the cap's +/// margin. Idle: close after the flush's own result or at the rotation; tell a +/// waiting brief; ask the flush where the cache would be evicted. +fn lifecycle(state: &mut State, now: DateTime) -> Action { + if state.stdin_closed || state.exited.is_some() { + return Action::None; + } + if state.busy { + let due = state.cap.is_some_and(|(started, cap)| now >= started + Duration::seconds(cap - FLUSH_MARGIN_S)); + return if due && ask_flush(state, now) { Action::WriteFlush } else { Action::None }; + } + let last = state.last_result_at.unwrap_or(now); + let rotate = state.cap.is_some_and(|(started, cap)| now >= started + Duration::seconds(cap - ROTATE_MARGIN_S)); + if rotate || state.flush.as_ref().is_some_and(|flush| flush.turn_begun) { + state.stdin_closed = true; + return Action::CloseStdin; + } + if state.inbox { + return Action::WriteTells; + } + if now >= last + Duration::seconds(state.ttl_s - TTL_MARGIN_S) && ask_flush(state, now) { + return Action::WriteFlush; + } + Action::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 prefix >= CEILING_TOKENS && ask_flush(&mut state, now) { + action = Action::WriteFlush; + } + } + } + Event::Result { ttl_1h } => { + state.busy = false; + state.last_result_at = Some(now); + state.ttl_s = if ttl_1h { TTL_1H_S } else { TTL_5M_S }; + // The turn already running ends here; a queued flush is the next one. + if state.flush.as_mut().is_some_and(|flush| !std::mem::replace(&mut flush.turn_begun, true)) { + state.busy = true; + } + } + Event::Inbox { pending } => state.inbox = pending, + Event::Told => { + state.inbox = false; + state.busy = true; + } + Event::PreCompact => { + if ask_flush(&mut state, now) { + action = Action::WriteFlush; + } + } + 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); + } + if state.stopping.is_none() { + let next = lifecycle(&mut state, now); + if next != Action::None { + return (state, next); + } + } + 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 index 814abd2..b5ba1da 100644 --- a/crates/mind-body/src/self_run/decide/tests.rs +++ b/crates/mind-body/src/self_run/decide/tests.rs @@ -2,9 +2,21 @@ //! 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 std::collections::VecDeque; + use super::*; use crate::testkit::at; +/// Long enough that no table below reaches the cap by accident. +const CAP_S: u64 = 100_000; +const RESULT: Event = Event::Result { ttl_1h: true }; +const SHORT_RESULT: Event = Event::Result { ttl_1h: false }; + +/// A session whose first turn is running. +fn working() -> State { + State::started(t(0), CAP_S) +} + fn t(seconds: i64) -> DateTime { at(seconds * 1000) } @@ -19,7 +31,7 @@ fn feed(state: State, event: Event, seconds: i64) -> (State, Action) { /// 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 state = working(); let mut actions = Vec::new(); for (event, seconds) in events { let (next, action) = feed(state, event, seconds); @@ -52,8 +64,6 @@ fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or 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. @@ -62,10 +72,81 @@ fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or } #[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); +fn a_result_ends_a_turn_and_never_closes_the_input_while_a_brief_waits_or_the_cache_is_warm() { + // The result asks for its briefs to be closed and nothing else; the next pass has nothing to do. + assert_eq!(last(vec![(RESULT, 0)]), Action::None); + assert_eq!(last(vec![(RESULT, 0), (Event::Tick, 0)]), Action::None); + assert_eq!(last(vec![(RESULT, 0), (Event::Tick, 3000)]), Action::None); + // A waiting brief is told, however long the session has been idle, and the input stays open. + assert_eq!(last(vec![(RESULT, 0), (Event::Inbox { pending: true }, 10)]), Action::WriteTells); + assert_eq!(last(vec![(RESULT, 0), (Event::Inbox { pending: true }, 10), (Event::Told, 10), (Event::Tick, 11)]), Action::None); + let waiting = |seconds: i64| last(vec![(RESULT, 0), (Event::Inbox { pending: true }, 10), (Event::Tick, seconds)]); + assert_eq!(waiting(TTL_1H_S * 3), Action::WriteTells, "a waiting brief outranks the cache's end"); + // Mid-turn, a brief is the hook's: the actuator writes nothing, and the next result comes. + assert_eq!(last(vec![(Event::Inbox { pending: true }, 10)]), Action::None); + assert_eq!(last(vec![(Event::Inbox { pending: true }, 10), (RESULT, 20), (Event::Tick, 20)]), Action::WriteTells); +} + +#[test] +fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_closes_after_the_flushs_own_result() { + let idle_at = |result: Event, seconds: i64| last(vec![(result, 0), (Event::Tick, seconds)]); + // An hour's cache: the flush at 3540 s, not before. + assert_eq!(idle_at(RESULT, TTL_1H_S - TTL_MARGIN_S - 1), Action::None); + assert_eq!(idle_at(RESULT, TTL_1H_S - TTL_MARGIN_S), Action::WriteFlush); + // A five-minute cache: at 240 s. + assert_eq!(idle_at(SHORT_RESULT, TTL_5M_S - TTL_MARGIN_S - 1), Action::None); + assert_eq!(idle_at(SHORT_RESULT, TTL_5M_S - TTL_MARGIN_S), Action::WriteFlush); + // The deadline the loop waits for is the same moment. + let (state, _) = run(vec![(RESULT, 5)]); + assert_eq!(next_deadline(&state), Some(t(5 + TTL_1H_S - TTL_MARGIN_S))); + // The input stays open through the flush's turn and closes after its result, one pass later. + let flushed = vec![(RESULT, 0), (Event::Tick, TTL_1H_S - TTL_MARGIN_S)]; + assert_eq!(last([flushed.clone(), vec![(Event::Tick, TTL_1H_S)]].concat()), Action::None); + let ended_turn = [flushed, vec![(SHORT_RESULT, TTL_1H_S)]].concat(); + assert_eq!(last(ended_turn), Action::CloseStdin); +} + +#[test] +fn near_the_cap_an_idle_session_rotates_and_a_working_one_is_asked_for_its_flush() { + let cap = i64::try_from(CAP_S).unwrap(); + // A result so late that the rotation comes before the cache's end. + let late = cap - 1300; + let idle_at = |seconds: i64| last(vec![(RESULT, late), (Event::Tick, seconds)]); + assert_eq!(idle_at(cap - ROTATE_MARGIN_S - 1), Action::None); + assert_eq!(idle_at(cap - ROTATE_MARGIN_S), Action::CloseStdin); + // A waiting brief does not hold the process past the rotation: the next process tells it. + assert_eq!(last(vec![(RESULT, late), (Event::Inbox { pending: true }, cap - ROTATE_MARGIN_S)]), Action::CloseStdin); + // Mid-turn the flush is asked at the margin, once, and it keeps its grace and escalation. + let working_at = |seconds: i64| last(vec![(req("a", 10), 1), (Event::Tick, seconds)]); + assert_eq!(working_at(cap - FLUSH_MARGIN_S - 1), Action::None); + assert_eq!(working_at(cap - FLUSH_MARGIN_S), Action::WriteFlush); + let asked = vec![(req("a", 10), 1), (Event::Tick, cap - FLUSH_MARGIN_S)]; + assert_eq!(last([asked.clone(), vec![(Event::Tick, cap - FLUSH_MARGIN_S + 1)]].concat()), Action::None); + assert_eq!(last([asked, vec![(Event::Tick, cap - FLUSH_MARGIN_S + GRACE)]].concat()), Action::Signal(Stop::Interrupt)); + let (state, _) = run(vec![(req("a", 10), 1)]); + assert_eq!(next_deadline(&state), Some(t(cap - FLUSH_MARGIN_S))); +} + +#[test] +fn pre_compact_asks_the_same_flush_the_ceiling_asks_once() { + assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1)]), Action::WriteFlush); + assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1), (Event::PreCompact, 2)]), Action::None); + // The ceiling after it asks nothing more; the ceiling before it leaves nothing for the hook to ask. + assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1), (req("b", CEILING_TOKENS), 2)]), Action::None); + assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (Event::PreCompact, 1)]), Action::None); + // Its flush has the grace and the headroom of any other. + assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1), (Event::Tick, 1 + GRACE)]), Action::Signal(Stop::Interrupt)); + // A child that is gone is not written to. + assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (Event::PreCompact, 1)]), Action::None); +} + +#[test] +fn a_flush_is_never_asked_of_a_session_whose_input_is_closed() { + let cap = i64::try_from(CAP_S).unwrap(); + let closed = vec![(RESULT, 1), (Event::Tick, cap - ROTATE_MARGIN_S)]; + assert_eq!(last(closed.clone()), Action::CloseStdin); + assert_eq!(last([closed.clone(), vec![(Event::PreCompact, cap)]].concat()), Action::None); + assert_eq!(last([closed, vec![(req("b", CEILING_TOKENS), cap)]].concat()), Action::None); } #[test] @@ -86,25 +167,23 @@ fn the_grace_runs_from_the_flush_write_to_the_second_and_no_later_request_moves_ #[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)]); + let after_result = |request: u64| last(vec![(req("a", CEILING_TOKENS), 0), (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)]); + let high = |request: u64| last(vec![(req("a", 400_000), 0), (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); + assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", LIMIT + 1), 1), (RESULT, 2), (req("c", LIMIT - 1), 3)]), 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()); + let saturated = |rest: Vec<(Event, i64)>| last([vec![(req("a", u64::MAX), 0), (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)]); + let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 0), (RESULT, 1), (req("b", LIMIT), 2)]); assert_eq!(actions[2], Action::Signal(Stop::Interrupt)); assert_eq!(next_deadline(&state), Some(t(2 + ESC))); } @@ -112,7 +191,7 @@ fn the_headroom_counts_the_requests_after_the_first_result_after_the_flush_from_ #[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); + let (state, action) = feed(working(), 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. @@ -197,7 +276,7 @@ fn a_child_that_has_exited_is_never_signalled_and_the_run_ends_with_its_stream_o #[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)]; + let quiet = vec![(req("a", 10), 0), (RESULT, 1), (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)); @@ -207,53 +286,22 @@ fn without_a_flush_nothing_is_ever_signalled_and_a_finished_run_decides_nothing_ // ---- 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, + /// When the flush was written and the prefix it was written at. flush: Option<(i64, u64)>, flushes: u32, closes: u32, + /// The flush is the turn that runs now or has run. boundary: bool, exceeded: bool, - result_seen: bool, + /// No turn is running. + idle: bool, + inbox: bool, + closed: bool, + prefix: u64, exited: Option, eof: bool, signals: Vec<(Stop, i64)>, @@ -263,11 +311,12 @@ struct Oracle { 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()); + let was_exited = oracle.exited.is_some(); match event { Event::Request { id, prefix } => { if !oracle.ids.contains(id) { oracle.ids.push(id.clone()); + oracle.prefix = *prefix; if oracle.boundary && let Some((_, base)) = oracle.flush && base.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| *prefix >= limit) @@ -276,29 +325,46 @@ fn check_step(oracle: &mut Oracle, event: &Event, now: i64, action: &Action) { } } } - Event::Result => { - if oracle.flush.is_some() { + Event::Result { .. } => { + oracle.idle = true; + if oracle.flush.is_some() && !oracle.boundary { oracle.boundary = true; + oracle.idle = false; } - oracle.result_seen = true; } + Event::Inbox { pending } => oracle.inbox = *pending, + Event::Told => { + oracle.inbox = false; + oracle.idle = false; + } + Event::PreCompact | Event::Tick | Event::Signalled { .. } => {} 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)); + if let Event::Request { prefix, .. } = event { + assert!(*prefix >= CEILING_TOKENS, "a flush at {prefix} on a request"); + } + assert!(flush_before.is_none() && !was_exited && !oracle.closed, "flush twice, to a gone child or to a closed input"); + oracle.flush = Some((now, oracle.prefix)); + // Written while idle, its turn begins at once. + oracle.boundary = oracle.idle; + oracle.idle = false; + } + Action::WriteTells => { + assert!(oracle.idle && oracle.inbox && !oracle.closed && !was_exited, "a tell outside idle with a brief waiting"); } Action::CloseStdin => { oracle.closes += 1; - assert!(matches!(event, Event::Result) && !was_result && !was_exited); + oracle.closed = true; + let flushed = oracle.flush.is_some() && oracle.boundary; + let rotation = now >= i64::try_from(CAP_S).unwrap() - ROTATE_MARGIN_S; + assert!(oracle.idle && !was_exited && (flushed || rotation), "closed on {event:?} flushed {flushed} rotation {rotation}"); } Action::Signal(how) => { let (flushed_at, _) = flush_before.expect("a signal before the flush was written"); @@ -341,8 +407,55 @@ fn check_step(oracle: &mut Oracle, event: &Event, now: i64, action: &Action) { assert!(oracle.signals.len() <= 3); } +/// What can happen next, each at a time after the last. +#[derive(Clone, Copy, Debug)] +enum Sym { + Low, + AtCeiling, + Huge, + Repeat, + Result, + ShortResult, + Brief, + NoBrief, + PreCompact, + Exit(Option), + Eof, + Soon, + Bound, + Grace, + Cache, +} + +const ALPHABET: [Sym; 16] = [ + Sym::Low, + Sym::AtCeiling, + Sym::Huge, + Sym::Repeat, + Sym::Result, + Sym::ShortResult, + Sym::Brief, + Sym::NoBrief, + Sym::PreCompact, + Sym::Exit(Some(0)), + Sym::Exit(None), + Sym::Eof, + Sym::Soon, + Sym::Bound, + Sym::Grace, + Sym::Cache, +]; + +/// Which signals the operating system lets through. +#[derive(Clone, Copy, Debug)] +enum Policy { + All, + Nothing, + NoKill, +} + fn play(symbols: &[Sym], policy: Policy) { - let mut state = State::default(); + let mut state = working(); let mut oracle = Oracle::default(); let mut now = 0i64; let mut unique = 0u32; @@ -353,27 +466,37 @@ fn play(symbols: &[Sym], policy: Policy) { 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::Result => (RESULT, 1), + Sym::ShortResult => (SHORT_RESULT, 1), + Sym::Brief => (Event::Inbox { pending: true }, 1), + Sym::NoBrief => (Event::Inbox { pending: false }, 1), + Sym::PreCompact => (Event::PreCompact, 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), + Sym::Cache => (Event::Tick, TTL_1H_S), }; now += dt; events.push(event.clone()); - let mut next = Some(event); - while let Some(event) = next.take() { + let mut pending = VecDeque::from([event]); + while let Some(event) = pending.pop_front() { 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 }); + match action { + Action::Signal(how) => { + let delivered = match policy { + Policy::All => true, + Policy::Nothing => false, + Policy::NoKill => how != Stop::Kill, + }; + pending.push_front(Event::Signalled { how, delivered }); + } + // The loop claims the spool and writes the line: a turn begins. + Action::WriteTells => pending.push_front(Event::Told), + _ => {} } } } @@ -417,9 +540,9 @@ fn enumerate(length: usize, symbols: &mut Vec, policy: Policy) { } #[test] -fn every_order_of_up_to_five_events_keeps_the_invariants_under_every_policy_of_delivery() { +fn every_order_of_up_to_four_events_keeps_the_invariants_under_every_policy_of_delivery() { for policy in [Policy::All, Policy::Nothing, Policy::NoKill] { - for length in 0..=5 { + for length in 0..=4 { enumerate(length, &mut Vec::new(), policy); } } diff --git a/crates/mind-body/src/self_run/inbox.rs b/crates/mind-body/src/self_run/inbox.rs new file mode 100644 index 0000000..92a4381 --- /dev/null +++ b/crates/mind-body/src/self_run/inbox.rs @@ -0,0 +1,251 @@ +//! Her Self's inbox: the briefs her Persona has given her Self, derived from the +//! mind, spooled for the session, and claimed exactly once. +//! +//! The mind owns what her Self is asked: an in-force follow-up with owner +//! `self-run` is a brief not yet answered. `derive` lists those that no closed +//! run claims and that the session does not hold yet; `spool` writes each as +//! `/-