GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

self-session-live: her Self session stays open across turns; briefs are derived from her mind and told between tool calls

Browse source 
A result ends a turn and marks the session idle; it no longer closes stdin
(decide.rs: Result => CloseStdin deleted). The session ends where its prompt
cache would be evicted (ruling live-session-floor-cache-evicts: the flush is
asked at the last request's TTL less 60 s, stdin closes after the flush's own
result), at an idle rotation deadline under the run cap, mid-turn at the cap's
flush margin, and at the context ceiling. PreCompact blocks auto-compaction
and asks the same flush.

self_run/inbox.rs derives her inbox from the mind (in-force follow-ups with
owner self-run that no closed run claims), spools them, and claims by rename
so the PostToolBatch hook and the actuator's idle tell deliver each once.
mind-self-run inject is the hook. Brief closure is not in this commit: the
mind refuses a Recorded resolution of a follow-up (question admitted).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 15:03:03 +01:00
commit 9f5e83e6c1
11 changed files with 1292 additions and 324 deletions
Download patch file Download diff file Expand all files Collapse all files

26
crates/mind-body/src/bin/mind-self-run.rs
View file

@ -1,3 +1,8 @@
//! `mind-self-run inject --spool <dir> [--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 <slug> [--endpoint rudp://<ip>:<port>]`: carries
//! her one open Self run (`mind_body::self_run::carry`) inside
//! `mind-self@<instance>`. This reads the process's arguments and environment,
@ -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 <slug> [--endpoint rudp://<
#[allow(clippy::disallowed_methods, reason = "the binary reads its own arguments and environment, and hands them to the library")]
fn run() -> Result<i32> {
let args: Vec<String> = 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<i32> {
.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<i32> {
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 <dir>"))?;
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),

160
crates/mind-body/src/self_run.rs
View file

@ -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<String> {
pub fn claude_args(session: Uuid, hooks: &Hooks) -> Vec<String> {
[
"-p",
"--input-format",
@ -186,6 +207,8 @@ pub fn claude_args(session: Uuid) -> Vec<String> {
"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<bool> {
let Ok(Value::Object(event)) = serde_json::from_str::<Value>(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::<Value>(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<Box<dyn Child>>;
}
@ -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<SelfRunOutcome> {
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<dyn Child>, (run, task): (&PipelineRef, &str), (plan_quota, now_ms): (&[PlanQuota], u64), heard: &mut Heard, out: &mut dyn Write) -> Result<SelfRunOutcome> {
#[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<dyn Child>,
hooks: &Hooks,
(run, task): (&PipelineRef, &str),
(plan_quota, now_ms, run_cap_s): (&[PlanQuota], u64, u64),
heard: &mut Heard,
out: &mut dyn Write,
) -> Result<SelfRunOutcome> {
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<dyn Child>, (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<PathBuf>,
/// 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<Item = (OsString, OsString)>) -> Result<Self> {
/// `injector` is the absolute path of `mind-self-run`, which the hooks run as `inject`.
pub fn new(unit_environment: impl IntoIterator<Item = (OsString, OsString)>, injector: PathBuf) -> Result<Self> {
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<Box<dyn Child>> {
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;

453
crates/mind-body/src/self_run/decide.rs
View file

@ -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<i32>),
StdoutEof,
/// Time passed, or a line that no rule reads.
Tick,
/// What the operating system said of the signal the last action sent.
Signalled { how: Stop, delivered: bool },
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Action {
None,
WriteFlush,
CloseStdin,
/// To the child's whole process group.
Signal(Stop),
Finish(SelfRunOutcome),
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct Flush {
/// The prefix at which the flush was written: the headroom's base.
prefix: u64,
written: DateTime<Utc>,
/// A result has come since the write: the flush is the next turn.
turn_begun: bool,
/// A request of the flush's turn reached the headroom.
exhausted: bool,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct State {
seen: HashSet<String>,
pub(super) requests: u64,
pub(super) prefix: u64,
result_seen: bool,
flush: Option<Flush>,
/// The last signal attempted and when.
stopping: Option<(Stop, DateTime<Utc>)>,
/// The strongest signal delivered.
delivered: Option<Stop>,
exited: Option<(Option<i32>, DateTime<Utc>)>,
eof: bool,
finished: bool,
}
//! 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<i32>),
StdoutEof,
/// Time passed, or a line that no rule reads.
Tick,
/// What the operating system said of the signal the last action sent.
Signalled { how: Stop, delivered: bool },
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Action {
None,
WriteFlush,
/// 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<Utc>,
/// A result has come since the write: the flush is the next turn.
turn_begun: bool,
/// A request of the flush's turn reached the headroom.
exhausted: bool,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct State {
seen: HashSet<String>,
pub(super) requests: u64,
pub(super) prefix: u64,
/// 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<DateTime<Utc>>,
/// 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<Utc>, i64)>,
stdin_closed: bool,
flush: Option<Flush>,
/// The last signal attempted and when.
stopping: Option<(Stop, DateTime<Utc>)>,
/// The strongest signal delivered.
delivered: Option<Stop>,
exited: Option<(Option<i32>, DateTime<Utc>)>,
eof: bool,
finished: bool,
}
impl State {
/// A session whose start message was just written: its first turn is running.
pub(super) fn started(at: DateTime<Utc>, 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<Utc>) -> Option<i64> {
(!self.busy).then(|| self.last_result_at.map_or(0, |at| (now - at).num_seconds()))
}
}
fn outcome(state: &State, cause: Cause, code: Option<i32>) -> 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<DateTime<Utc>> {
if state.finished {
return None;
}
if let Some((_, exited)) = state.exited {
return (!state.eof).then(|| exited + Duration::seconds(EXIT_DRAIN_S));
}
match (&state.stopping, &state.flush) {
(Some((_, sent)), _) => Some(*sent + Duration::seconds(ESCALATE_AFTER_S)),
(None, Some(flush)) => Some(flush.written + Duration::seconds(FLUSH_GRACE_S)),
(None, None) => None,
}
}
pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (State, Action) {
if state.finished {
return (state, Action::None);
}
let mut action = Action::None;
match event {
Event::Request { id, prefix } => {
if state.seen.insert(id) {
state.requests += 1;
state.prefix = prefix;
if let Some(flush) = state.flush.as_mut()
&& flush.turn_begun
&& flush.prefix.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit)
{
flush.exhausted = true;
}
if state.flush.is_none() && !state.result_seen && state.exited.is_none() && prefix >= CEILING_TOKENS {
state.flush = Some(Flush { prefix, written: now, turn_begun: false, exhausted: false });
action = Action::WriteFlush;
}
}
}
Event::Result => {
if !state.result_seen {
state.result_seen = true;
if let Some(flush) = state.flush.as_mut() {
flush.turn_begun = true;
}
action = Action::CloseStdin;
}
}
Event::ChildExited(code) => {
state.exited.get_or_insert((code, now));
}
Event::StdoutEof => state.eof = true,
Event::Tick => {}
Event::Signalled { how, delivered } => {
if delivered {
state.delivered = Some(how);
}
}
}
// A child that has exited is never written to or signalled: the run is over
// once its stream has ended, or once the stream has had its drain.
if let Some((code, exited)) = state.exited {
if state.eof || now >= exited + Duration::seconds(EXIT_DRAIN_S) {
state.finished = true;
let outcome = outcome(&state, cause(&state), code);
return (state, Action::Finish(outcome));
}
return (state, Action::None);
}
if action != Action::None {
return (state, action);
}
if let Some((Stop::Kill, sent)) = state.stopping {
// SIGKILL went out and the child still has not been reaped: stop waiting for it.
if now >= sent + Duration::seconds(ESCALATE_AFTER_S) {
state.finished = true;
let cause = if state.delivered == Some(Stop::Kill) { Cause::Signalled(Stop::Kill) } else { Cause::Unstoppable };
let outcome = outcome(&state, cause, None);
return (state, Action::Finish(outcome));
}
return (state, Action::None);
}
let how = match (&state.stopping, &state.flush) {
(Some((last, sent)), _) => (now >= *sent + Duration::seconds(ESCALATE_AFTER_S)).then(|| last.next()),
(None, Some(flush)) => (flush.exhausted || now >= flush.written + Duration::seconds(FLUSH_GRACE_S)).then_some(Stop::Interrupt),
(None, None) => None,
};
match how {
Some(how) => {
state.stopping = Some((how, now));
(state, Action::Signal(how))
}
None => (state, Action::None),
}
}
/// The moment the loop must next be heard at, if any. A moment already past
/// means an action is due now.
pub(super) fn next_deadline(state: &State) -> Option<DateTime<Utc>> {
if state.finished {
return None;
}
if let Some((_, exited)) = state.exited {
return (!state.eof).then(|| exited + Duration::seconds(EXIT_DRAIN_S));
}
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<DateTime<Utc>> {
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<Utc>) -> 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<Utc>) -> 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<Utc>) -> (State, Action) {
if state.finished {
return (state, Action::None);
}
let mut action = Action::None;
match event {
Event::Request { id, prefix } => {
if state.seen.insert(id) {
state.requests += 1;
state.prefix = prefix;
if let Some(flush) = state.flush.as_mut()
&& flush.turn_begun
&& flush.prefix.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit)
{
flush.exhausted = true;
}
if 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;

275
crates/mind-body/src/self_run/decide/tests.rs
View file

@ -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<Utc> {
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<Action>) {
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<i32>),
Eof,
Soon,
Bound,
Grace,
}
const ALPHABET: [Sym; 11] = [
Sym::Low,
Sym::AtCeiling,
Sym::Huge,
Sym::Repeat,
Sym::Result,
Sym::Exit(Some(0)),
Sym::Exit(None),
Sym::Eof,
Sym::Soon,
Sym::Bound,
Sym::Grace,
];
/// Which signals the operating system lets through.
#[derive(Clone, Copy, Debug)]
enum Policy {
All,
Nothing,
NoKill,
}
/// The facts of the history, kept apart from `State`: what the invariants are asked of.
#[derive(Default)]
struct Oracle {
ids: Vec<String>,
/// 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<i64>,
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<i32>),
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<Sym>, 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);
}
}

251
crates/mind-body/src/self_run/inbox.rs Normal file
View file

@ -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
//! `<spool>/<ordinal>-<label>.json`. A claim renames entries into `<spool>/told/`
//! and returns them: the rename is the exactly-once point, so the PostToolBatch
//! hook (`inject`) and the actuator's idle tell race safely, and `told/` is the
//! record of what this session has heard. Everything here is derived or a cache:
//! the spool is lost with the unit and rebuilt from the mind.
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use eureka_pipeline::PipelineKind;
use huginn_mind::{PipelineFacts, PipelineStatusSummary};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use crate::queue::{MindPort, any_of, headers, in_force, of_kind};
/// The owner a follow-up names when it is a brief for her Self.
pub const OWNER: &str = "self-run";
/// The entries this session has claimed.
const TOLD: &str = "told";
/// The plan line as of the lease grant, which a tell ends with.
const PLAN: &str = "plan.txt";
/// The PreCompact hook's marker: the ceiling is reached by the CLI's own measure.
const PRE_COMPACT: &str = "pre-compact";
/// What the hook event is called when its input does not say.
const DEFAULT_EVENT: &str = "PostToolBatch";
/// One brief from her Persona: a follow-up of the mind, as the session hears it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Brief {
/// The follow-up's full id.
pub id: String,
pub campaign: String,
pub repo: String,
pub item: String,
/// The admission ordinal: the order briefs were given in.
pub ordinal: u64,
}
impl Brief {
fn file_name(&self) -> String {
let label = self.id.rsplit(':').next().unwrap_or(&self.id);
format!("{:020}-{label}.json", self.ordinal)
}
}
/// Where the hooks find the spool and the program they run.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Hooks {
pub spool: PathBuf,
/// The absolute path of `mind-self-run`.
pub injector: PathBuf,
}
fn quoted(path: &Path) -> String {
format!("'{}'", path.to_string_lossy().replace('\'', "'\\''"))
}
impl Hooks {
/// The `--settings` JSON: the tell between tool calls, and the block on auto-compaction.
pub fn settings(&self) -> String {
let command = format!("{} inject --spool {}", quoted(&self.injector), quoted(&self.spool));
json!({"hooks": {
"PostToolBatch": [{"hooks": [{"type": "command", "command": command}]}],
"PreCompact": [{"matcher": "auto", "hooks": [{"type": "command", "command": format!("{command} --pre-compact")}]}],
}})
.to_string()
}
}
/// The ids of the follow-ups a closed run claims, with the run each was claimed by.
fn claimed_by_closed(mind: &dyn MindPort) -> Result<HashMap<String, String>> {
let runs = headers(mind, &any_of(of_kind(PipelineKind::Run), "root", &[&mind.instance().0]))?;
let mut claimed = HashMap::new();
for run in runs {
let (PipelineStatusSummary::Resolved { .. }, PipelineFacts::Run { claims, .. }) = (&run.status, &run.facts) else { continue };
for claim in claims.iter().filter(|claim| claim.kind == PipelineKind::FollowUp) {
claimed.insert(claim.id.0.clone(), run.id.id.0.clone());
}
}
Ok(claimed)
}
/// Every in-force follow-up with owner `self-run`, oldest first.
fn open_briefs(mind: &dyn MindPort) -> Result<Vec<Brief>> {
let mut briefs: Vec<Brief> = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))?
.into_iter()
.filter_map(|summary| match summary.facts {
PipelineFacts::FollowUp { repo, owner, item, .. } if owner.0 == OWNER => {
let campaign = summary.id.id.0.split(':').next().unwrap_or("").to_string();
Some(Brief { id: summary.id.id.0, campaign, repo: repo.0, item: item.0, ordinal: summary.admission.ordinal })
}
_ => None,
})
.collect();
briefs.sort_by_key(|brief| brief.ordinal);
Ok(briefs)
}
/// Her open briefs that no closed run claims and the session does not hold yet.
pub fn derive(mind: &dyn MindPort, spool: &Path) -> Result<Vec<Brief>> {
let held: HashSet<String> = entries(spool)?.into_iter().chain(entries(&spool.join(TOLD))?).map(|brief| brief.id).collect();
let mut open: Vec<Brief> = open_briefs(mind)?.into_iter().filter(|brief| !held.contains(&brief.id)).collect();
if !open.is_empty() {
let claimed = claimed_by_closed(mind)?;
open.retain(|brief| !claimed.contains_key(&brief.id));
}
Ok(open)
}
fn entries(dir: &Path) -> Result<Vec<Brief>> {
let read = match std::fs::read_dir(dir) {
Ok(read) => read,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(error).context("the inbox could not be read"),
};
let mut names: Vec<PathBuf> = read.filter_map(|entry| entry.ok()).filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_file())).map(|entry| entry.path()).collect();
names.retain(|path| path.extension().is_some_and(|extension| extension == "json") && !path.file_name().is_some_and(|name| name.to_string_lossy().starts_with('.')));
names.sort();
names.iter().filter_map(|path| read_entry(path).transpose()).collect()
}
/// None where another claimer renamed the entry away since it was listed.
fn read_entry(path: &Path) -> Result<Option<Brief>> {
let text = match std::fs::read_to_string(path) {
Ok(text) => text,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error).context("an inbox entry could not be read"),
};
serde_json::from_str(&text).map(Some).context("an inbox entry is not a brief")
}
fn write_entry(dir: &Path, brief: &Brief) -> Result<()> {
std::fs::create_dir_all(dir).context("the inbox could not be made")?;
let name = brief.file_name();
// Written under a dot name, then renamed: a claimer never sees half an entry.
let partial = dir.join(format!(".{name}"));
std::fs::write(&partial, serde_json::to_vec(brief)?).context("an inbox entry could not be written")?;
std::fs::rename(&partial, dir.join(name)).context("an inbox entry could not be placed")
}
/// Writes each brief into the spool.
pub fn spool(spool: &Path, briefs: &[Brief]) -> Result<()> {
briefs.iter().try_for_each(|brief| write_entry(spool, brief))
}
/// Records briefs as already heard (the start message carries them).
pub fn mark_told(spool: &Path, briefs: &[Brief]) -> Result<()> {
briefs.iter().try_for_each(|brief| write_entry(&spool.join(TOLD), brief))
}
/// Whether a brief waits in the spool.
pub fn pending(spool: &Path) -> Result<bool> {
Ok(!entries(spool)?.is_empty())
}
/// Takes every waiting entry. A rename into `told/` is the claim: an entry some
/// other claimer renamed first is theirs, and is not returned here.
pub fn claim(spool: &Path) -> Result<Vec<Brief>> {
let told = spool.join(TOLD);
std::fs::create_dir_all(&told).context("the inbox could not be made")?;
let mut taken = Vec::new();
for brief in entries(spool)? {
let name = brief.file_name();
match std::fs::rename(spool.join(&name), told.join(&name)) {
Ok(()) => taken.push(brief),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error).context("an inbox entry could not be claimed"),
}
}
Ok(taken)
}
/// Starts the spool empty: a session hears only what it is told after it starts.
pub fn reset(spool: &Path) -> Result<()> {
match std::fs::remove_dir_all(spool) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error).context("a stale inbox could not be removed"),
}
std::fs::create_dir_all(spool).context("the inbox could not be made")
}
/// The plan line a tell ends with: as of the lease grant.
pub fn write_plan(spool: &Path, line: &str) -> Result<()> {
std::fs::create_dir_all(spool).context("the inbox could not be made")?;
std::fs::write(spool.join(PLAN), line).context("the plan line could not be written")
}
/// One factual block per brief, then the plan line.
pub fn render(briefs: &[Brief], plan: &str) -> String {
let mut blocks: Vec<String> = briefs
.iter()
.map(|brief| {
format!(
"A brief from your Persona: follow-up {} (campaign {}, repo {}): {}. Work you take on beyond this answer is a follow-up of your own.",
brief.id, brief.campaign, brief.repo, brief.item
)
})
.collect();
if !plan.is_empty() {
blocks.push(plan.to_string());
}
blocks.join("\n\n")
}
/// What a hook prints, and the status it exits with.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Injection {
pub stdout: String,
pub exit: i32,
}
/// The hook: data only. It claims the spool and prints what it claimed; with an
/// agent id (a subagent's call) it prints nothing; `pre_compact` writes the
/// marker and exits 2, which blocks the compaction. It reads neither the mind,
/// chat nor the lease, and opens no socket.
pub fn inject(spool: &Path, input: &str, pre_compact: bool) -> Result<Injection> {
if pre_compact {
std::fs::create_dir_all(spool).context("the inbox could not be made")?;
std::fs::write(spool.join(PRE_COMPACT), b"").context("the marker could not be written")?;
return Ok(Injection { stdout: String::new(), exit: 2 });
}
let event: Value = serde_json::from_str(input).unwrap_or(Value::Null);
if event.get("agent_id").is_some_and(|id| !id.is_null()) {
return Ok(Injection { stdout: String::new(), exit: 0 });
}
let briefs = claim(spool)?;
if briefs.is_empty() {
return Ok(Injection { stdout: String::new(), exit: 0 });
}
let plan = std::fs::read_to_string(spool.join(PLAN)).unwrap_or_default();
let name = event.get("hook_event_name").and_then(Value::as_str).unwrap_or(DEFAULT_EVENT);
let stdout = json!({"hookSpecificOutput": {"hookEventName": name, "additionalContext": render(&briefs, &plan)}}).to_string();
Ok(Injection { stdout, exit: 0 })
}
/// Whether the PreCompact hook fired since the last look; the marker is consumed.
pub fn take_pre_compact(spool: &Path) -> bool {
std::fs::remove_file(spool.join(PRE_COMPACT)).is_ok()
}
#[cfg(test)]
mod tests;

125
crates/mind-body/src/self_run/inbox/tests.rs Normal file
View file

@ -0,0 +1,125 @@
//! The inbox over a real spool directory and a mind in memory: what is derived,
//! the exactly-once claim, and the hook that prints it.
use eureka_pipeline::{PipelineDocument, PipelineKind as K, PipelineRef, RunOperator, RunTurn};
use super::*;
use crate::testkit::{TestMind, close as closing, follow_up, id, recorded, reference, run, run_ref, spec, spec_ref, withdrawn};
fn spool_dir() -> tempfile::TempDir {
tempfile::tempdir().unwrap()
}
fn brief_doc(label: &str, owner: &str) -> PipelineDocument {
let PipelineDocument::FollowUp(mut document) = follow_up(label, &spec_ref("x")) else { unreachable!() };
document.owner = owner.into();
document.item = format!("Do {label}.").as_str().into();
PipelineDocument::FollowUp(document)
}
fn brief_ref(label: &str) -> PipelineRef {
reference(K::FollowUp, &id("follow_up", label))
}
fn mind_with(documents: Vec<PipelineDocument>) -> TestMind {
let mind = TestMind::seeded();
mind.committed(vec![spec("x")]);
mind.committed(documents);
mind
}
fn entry(n: u64) -> Brief {
Brief { id: format!("eureka-body:follow_up:b{n}"), campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: format!("Item {n}."), ordinal: n }
}
#[test]
fn the_inbox_is_the_open_self_run_briefs_that_no_closed_run_claims_and_the_session_does_not_hold() {
let mind = mind_with(vec![brief_doc("open", OWNER), brief_doc("claimed-by-old", OWNER), brief_doc("someone-elses", "Hands"), brief_doc("answered", OWNER), brief_doc("held", OWNER)]);
mind.committed(vec![run("old", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed-by-old")])]);
mind.committed(vec![closing(run_ref("old"), recorded()), closing(brief_ref("answered"), withdrawn())]);
let spool = spool_dir();
mark_told(spool.path(), &[Brief { id: brief_ref("held").id.0, ..entry(9) }]).unwrap();
let derived: Vec<String> = derive(&mind, spool.path()).unwrap().into_iter().map(|brief| brief.id).collect();
assert_eq!(derived, vec![brief_ref("open").id.0], "only the open one nobody answered and nobody holds");
// A run that is still open claims without answering: its brief is still asked.
mind.committed(vec![run("live", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("open")])]);
assert_eq!(derive(&mind, spool.path()).unwrap().len(), 1);
}
#[test]
fn spooled_entries_are_claimed_in_order_and_hand_back_what_was_written() {
let spool = spool_dir();
spool_briefs(spool.path(), &[entry(2), entry(1)]);
assert!(pending(spool.path()).unwrap());
let claimed = claim(spool.path()).unwrap();
assert_eq!(claimed, vec![entry(1), entry(2)], "oldest first");
assert!(!pending(spool.path()).unwrap() && claim(spool.path()).unwrap().is_empty(), "a claim empties the spool");
assert_eq!(entries(&spool.path().join("told")).unwrap(), vec![entry(1), entry(2)], "the claimed entries are the session's record");
}
fn spool_briefs(dir: &Path, briefs: &[Brief]) {
spool(dir, briefs).unwrap();
}
#[test]
fn a_hook_and_the_actuator_claiming_at_once_deliver_each_entry_to_exactly_one_of_them() {
for _ in 0..20 {
let spool = spool_dir();
let briefs: Vec<Brief> = (1..=60).map(entry).collect();
spool_briefs(spool.path(), &briefs);
let path = spool.path().to_path_buf();
let racers: Vec<_> = (0..4)
.map(|_| {
let path = path.clone();
std::thread::spawn(move || claim(&path).unwrap())
})
.collect();
let mut seen: Vec<u64> = racers.into_iter().flat_map(|racer| racer.join().unwrap()).map(|brief| brief.ordinal).collect();
seen.sort_unstable();
assert_eq!(seen, (1..=60).collect::<Vec<u64>>(), "every entry once, none twice, none lost");
}
}
#[test]
fn the_hook_prints_data_only_and_only_for_the_main_thread() {
let spool = spool_dir();
write_plan(spool.path(), "Plan usage: unobserved.").unwrap();
assert_eq!(inject(spool.path(), "{}", false).unwrap(), Injection { stdout: String::new(), exit: 0 }, "an empty spool prints nothing");
spool_briefs(spool.path(), &[entry(1)]);
// A subagent's call (the input carries an agent id) prints nothing and claims nothing.
let subagent = inject(spool.path(), "{\"hook_event_name\":\"PostToolBatch\",\"agent_id\":\"a1\"}", false).unwrap();
assert_eq!(subagent, Injection { stdout: String::new(), exit: 0 });
assert!(pending(spool.path()).unwrap());
let printed = inject(spool.path(), "{\"hook_event_name\":\"PostToolBatch\"}", false).unwrap();
assert_eq!(printed.exit, 0);
let output: Value = serde_json::from_str(&printed.stdout).unwrap();
assert_eq!(output["hookSpecificOutput"]["hookEventName"], "PostToolBatch");
let context = output["hookSpecificOutput"]["additionalContext"].as_str().unwrap();
for needle in ["eureka-body:follow_up:b1", "campaign eureka-body", "repo GameCult/Huginn", "Item 1.", "Plan usage: unobserved."] {
assert!(context.contains(needle), "{needle}: {context}");
}
assert_eq!(inject(spool.path(), "{}", false).unwrap().stdout, "", "claimed once");
// Unreadable input is not a reason to lose a brief or to fail the tool call.
spool_briefs(spool.path(), &[entry(2)]);
assert!(inject(spool.path(), "not json", false).unwrap().stdout.contains("Item 2."));
}
#[test]
fn the_pre_compact_hook_blocks_with_exit_2_and_leaves_one_marker() {
let spool = spool_dir();
spool_briefs(spool.path(), &[entry(1)]);
assert_eq!(inject(spool.path(), "{}", true).unwrap(), Injection { stdout: String::new(), exit: 2 });
assert!(pending(spool.path()).unwrap(), "blocking claims nothing");
assert!(take_pre_compact(spool.path()) && !take_pre_compact(spool.path()), "the marker is consumed once");
}
#[test]
fn the_settings_name_the_two_hooks_with_absolute_quoted_commands() {
let hooks = Hooks { spool: PathBuf::from("/run/mind-x-self/inbox"), injector: PathBuf::from("/opt/it's/mind-self-run") };
let settings: Value = serde_json::from_str(&hooks.settings()).unwrap();
let tell = settings["hooks"]["PostToolBatch"][0]["hooks"][0]["command"].as_str().unwrap();
assert_eq!(tell, "'/opt/it'\\''s/mind-self-run' inject --spool '/run/mind-x-self/inbox'");
let compact = &settings["hooks"]["PreCompact"][0];
assert_eq!(compact["matcher"], "auto");
assert_eq!(compact["hooks"][0]["command"].as_str().unwrap(), format!("{tell} --pre-compact"));
}

276
crates/mind-body/src/self_run/tests.rs
View file

@ -95,9 +95,11 @@ enum Step {
Exit(Option<i32>),
/// Its stdout ends; the child is not reaped yet.
Eof,
/// The clock moves, this runs (the mind changes, a tool call's hook fires), and the read times out.
Do(i64, Rc<dyn Fn()>),
}
use Step::{Eof, Exit, Idle, Line as Out};
use Step::{After, Eof, Exit, Idle, Line as Out};
struct Scripted {
steps: VecDeque<Step>,
@ -143,6 +145,11 @@ impl Child for Scripted {
self.exited = true;
Read::Exited(code)
}
Some(Step::Do(seconds, act)) => {
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds));
act();
Read::Timeout
}
Some(Step::Eof) => {
self.eof = true;
Read::Eof
@ -181,9 +188,27 @@ struct Fake {
child: RefCell<Option<Scripted>>,
spawns: RefCell<Vec<Vec<String>>>,
refuse: bool,
hooks: Hooks,
}
impl Fake {
fn with(child: Option<Scripted>, refuse: bool) -> Self {
Self { child: RefCell::new(child), spawns: RefCell::new(Vec::new()), refuse, hooks: test_hooks() }
}
}
/// A spool of its own under the system temporary directory, so that tests running side by side share nothing.
fn test_hooks() -> Hooks {
static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
Hooks { spool: std::env::temp_dir().join(format!("mind-self-run-spool-{}-{n}", std::process::id())), injector: PathBuf::from("/opt/mind-self-run") }
}
impl Spawner for Fake {
fn hooks(&self) -> Hooks {
self.hooks.clone()
}
fn spawn(&self, args: &[String], _token: &TokenFd) -> Result<Box<dyn Child>> {
self.spawns.borrow_mut().push(args.to_vec());
if self.refuse {
@ -310,6 +335,7 @@ impl MindPort for Seen<'_> {
}
struct Ran {
hooks: Hooks,
outcome: Result<SelfRunOutcome>,
out: String,
events: Vec<Event>,
@ -343,6 +369,11 @@ impl Ran {
.collect()
}
/// What the unit's stdout holds apart from the decisions the actuator journals.
fn lines(&self) -> String {
self.out.lines().filter(|line| !line.starts_with("decision ")).map(|line| format!("{line}\n")).collect()
}
fn count(&self, wanted: &Event) -> usize {
self.events.iter().filter(|event| *event == wanted).count()
}
@ -368,13 +399,15 @@ fn carry_signalled(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on: Op
}
fn carry_scripted(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on: Option<Stop>, refuses: &[Stop], hangs: bool) -> Ran {
carry_hooked(seen, steps, (code, dies_on, refuses, hangs), test_hooks())
}
/// The same with the spool the child's hooks drain named by the test, so that a step can fire the hook.
fn carry_hooked(seen: &Seen, steps: Vec<Step>, (code, dies_on, refuses, hangs): (Option<i32>, Option<Stop>, &[Stop], bool), hooks: Hooks) -> Ran {
let log = Rc::new(RefCell::new(Vec::new()));
let clock = Rc::new(Fixed(Cell::new(at_start())));
let fake = Fake {
child: RefCell::new(Some(Scripted { steps: steps.into(), log: log.clone(), clock: clock.clone(), code, dies_on, refuses: refuses.to_vec(), hangs, exited: false, eof: false, asked: 0 })),
spawns: RefCell::new(Vec::new()),
refuse: false,
};
let mut fake = Fake::with(Some(Scripted { steps: steps.into(), log: log.clone(), clock: clock.clone(), code, dies_on, refuses: refuses.to_vec(), hangs, exited: false, eof: false, asked: 0 }), false);
fake.hooks = hooks;
let lender = FakeLender::logging(log.clone(), Answer::Grant);
let mut out = Vec::new();
let outcome = carry(seen, &*clock, &lender, &fake, TEST_CAP_S, &mut out);
@ -382,7 +415,7 @@ fn carry_scripted(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on: Opt
// The child's own events; the lease's are in `all`, in order with them.
let events = all.iter().filter(|event| !matches!(event, Event::Lent | Event::Spawned | Event::Abandoned)).cloned().collect();
let spawns = fake.spawns.borrow().clone();
Ran { outcome, out: String::from_utf8(out).unwrap(), events, all, spawns, admits: seen.admits.get(), asked: lender.asked.borrow().clone() }
Ran { hooks: fake.hooks(), outcome, out: String::from_utf8(out).unwrap(), events, all, spawns, admits: seen.admits.get(), asked: lender.asked.borrow().clone() }
}
fn at_start() -> chrono::DateTime<chrono::Utc> {
@ -449,7 +482,7 @@ fn without_exactly_one_open_self_run_of_hers_it_is_no_run_and_no_child_is_spawne
fn a_mind_that_cannot_be_read_or_a_child_that_cannot_start_is_an_error_not_an_outcome() {
let mind = with_run();
let clock = Rc::new(Fixed(Cell::new(at_start())));
let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true };
let refusing = Fake::with(None, true);
assert!(carry(&Seen::of(&mind), &*clock, &FakeLender::new(Answer::Grant), &refusing, TEST_CAP_S, &mut Vec::new()).is_err());
assert_eq!(refusing.spawns.borrow().len(), 1, "it tried exactly once");
@ -466,7 +499,7 @@ fn a_mind_that_cannot_be_read_or_a_child_that_cannot_start_is_an_error_not_an_ou
unreachable!()
}
}
let never = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: false };
let never = Fake::with(None, false);
assert!(carry(&Down, &*clock, &FakeLender::new(Answer::Grant), &never, TEST_CAP_S, &mut Vec::new()).is_err());
assert!(never.spawns.borrow().is_empty());
}
@ -476,7 +509,8 @@ fn a_mind_that_cannot_be_read_or_a_child_that_cannot_start_is_an_error_not_an_ou
#[test]
fn claude_args_is_exactly_the_fixed_list_for_a_session() {
let session = Uuid::parse_str(SESSION_OF_RUN_A).unwrap();
let args = claude_args(session);
let hooks = test_hooks();
let args = claude_args(session, &hooks);
assert_eq!(
args,
[
@ -496,10 +530,12 @@ fn claude_args_is_exactly_the_fixed_list_for_a_session() {
"none",
"--setting-sources",
"user",
"--settings",
hooks.settings().as_str(),
]
);
// Those belong to the run measure or to the unit's managed settings.
for forbidden in ["--resume", "--continue", "--fork-session", "--model", "--dangerously-skip-permissions", "--settings", "--system-prompt", "--max-budget"] {
for forbidden in ["--resume", "--continue", "--fork-session", "--model", "--dangerously-skip-permissions", "--system-prompt", "--max-budget"] {
assert!(!args.iter().any(|arg| arg.starts_with(forbidden) || arg.contains("dangerously")), "{forbidden}");
}
// Print mode starts in Manual without a mode, and prompts none then denies.
@ -523,7 +559,7 @@ fn the_child_is_started_for_her_run_with_the_fixed_list_whatever_the_run_record_
mind.committed(vec![self_run("selfrun-b", budget)]);
let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false);
assert_eq!(ran.spawns.len(), 1, "{budget}");
assert_eq!(ran.spawns[0], claude_args(Uuid::parse_str(SESSION_OF_RUN_B).unwrap()), "{budget}");
assert_eq!(ran.spawns[0], claude_args(Uuid::parse_str(SESSION_OF_RUN_B).unwrap(), &ran.hooks), "{budget}");
}
}
@ -740,26 +776,27 @@ fn below_the_ceiling_no_flush_is_written_and_at_it_exactly_one() {
}
#[test]
fn a_result_before_the_ceiling_means_no_flush_ever() {
let ran = go(vec![Out(at("m1", 10)), Out(result_line()), Out(at("m2", CEILING_TOKENS + 1))]);
fn a_closed_input_is_never_asked_for_a_flush_whatever_the_prefix_grows_to() {
let cap = i64::try_from(TEST_CAP_S).unwrap();
// Rotation closes the idle session; a request that comes after it asks nothing.
let ran = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(200), Out(at("m2", CEILING_TOKENS + 1))]);
assert_eq!(ran.count(&Event::InputClosed), 1);
assert_eq!(flushes(&ran), 0);
assert_eq!(ran.ended().0, Cause::Own);
assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0);
}
#[test]
fn stdin_closes_once_at_the_first_result_flush_or_no_flush_and_nothing_is_written_after() {
let plain = go(vec![Out(at("m1", 10)), Out(result_line()), Out(result_line())]);
assert_eq!(plain.count(&Event::InputClosed), 1);
let closed_at = plain.events.iter().position(|event| *event == Event::InputClosed).unwrap();
assert!(plain.events[closed_at..].iter().all(|event| !matches!(event, Event::Wrote(_))), "no input after the close");
assert!(plain.events[..closed_at].iter().any(|event| matches!(event, Event::Wrote(_))), "the start went before");
let flushed = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())]);
fn stdin_closes_once_after_the_flushs_own_result_and_nothing_is_written_after() {
// The first result ends the turn the flush was queued behind; the second is the flush's own.
let flushed = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(result_line())]);
assert_eq!(flushed.count(&Event::InputClosed), 1);
let closed_at = flushed.events.iter().position(|event| *event == Event::InputClosed).unwrap();
assert_eq!(flushed.events[..closed_at].iter().filter(|event| matches!(event, Event::Wrote(_))).count(), 2, "start and flush before the close");
assert!(flushed.events[closed_at..].iter().all(|event| !matches!(event, Event::Wrote(_))));
// One result closes nothing: the flush is still to have its turn.
let one = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())]);
assert_eq!(one.count(&Event::InputClosed), 0, "{:?}", one.events);
let never = go(vec![Out(at("m1", 10))]);
assert_eq!(never.count(&Event::InputClosed), 0, "no result, no close");
@ -794,8 +831,8 @@ fn after_the_flush_the_grace_interrupts_exactly_once_and_not_a_second_early() {
#[test]
fn the_wait_for_the_child_never_passes_the_grace_deadline_and_never_spins_after_the_interrupt() {
let ran = go(vec![Out(at("m1", CEILING_TOKENS)), Idle(840), Idle(59), Idle(1)]);
// 60 until the grace is near, the one second left, then the escalation's own wait, then the drain after the child's exit.
assert_eq!(ran.asked(), [60, 60, 60, 1, 60, 2], "{:?}", ran.asked());
// The inbox poll's 30 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(), [30, 30, 30, 1, 29, 2], "{:?}", ran.asked());
}
#[test]
@ -839,7 +876,7 @@ fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_the_running_tu
// The headroom is measured from the prefix the flush was asked at (400k) and counts the requests after the
// first result only: 459_999 does not interrupt; 460_000 does, once, and m6 is never read.
assert_eq!(jump.count(&Event::Stopped(Stop::Interrupt)), 1, "{kinds:?}");
assert_eq!(jump.out, "request 1 prefix 290000
assert_eq!(jump.lines(), "request 1 prefix 290000
request 2 prefix 400000
request 3 prefix 470000
result subtype=success is_error=true: Failed to authenticate: OAuth session expired and could not be refreshed
@ -894,10 +931,10 @@ fn a_usage_sum_that_overflows_is_at_or_over_the_ceiling_not_a_panic_or_a_wrap()
}
#[test]
fn a_result_before_any_request_closes_the_input_once_and_the_run_ends_with_a_typed_outcome() {
fn a_result_before_any_request_leaves_the_input_open_and_the_run_ends_with_a_typed_outcome() {
// The shape of an API or authentication failure: the child reports an error result and no request ever came.
let ran = carry_with(&Seen::of(&with_run()), vec![Out(result_line())], Some(1), false);
assert_eq!(ran.count(&Event::InputClosed), 1);
assert_eq!(ran.count(&Event::InputClosed), 0);
assert_eq!(ran.ended(), (Cause::Own, Some(1), 0, 0));
assert_eq!(ran.outcome.as_ref().unwrap().exit_status(), 1);
assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0);
@ -905,7 +942,8 @@ fn a_result_before_any_request_closes_the_input_once_and_the_run_ends_with_a_typ
#[test]
fn without_a_flush_nothing_is_ever_signalled_however_large_the_prefix_grows() {
let ran = go(vec![Out(at("m1", 10)), Out(result_line()), Out(at("m2", 900_000)), Idle(100_000)]);
let cap = i64::try_from(TEST_CAP_S).unwrap();
let ran = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(200), Out(at("m2", 900_000)), Idle(100_000)]);
assert_eq!(ran.events.iter().filter(|event| matches!(event, Event::Stopped(_))).count(), 0);
assert_eq!(ran.ended().0, Cause::Own);
}
@ -980,7 +1018,7 @@ fn the_clock_is_read_once_per_event_so_a_step_of_the_wall_clock_cannot_split_a_p
}
let steps: VecDeque<Step> = vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", CEILING_TOKENS + 1)), Idle(0)].into();
let child = Scripted { steps, log: log.clone(), clock: Rc::new(Fixed(Cell::new(at_start()))), code: Some(0), dies_on: None, refuses: vec![], hangs: false, exited: false, eof: false, asked: 0 };
let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false };
let fake = Fake::with(Some(child), false);
let outcome = carry(&Seen::of(&mind), &Stepped(clock.clone()), &FakeLender::new(Answer::Grant), &fake, TEST_CAP_S, &mut Vec::new()).unwrap();
let events = log.borrow().clone();
let reads = events.iter().filter(|event| matches!(event, Event::Asked(_))).count() as u32;
@ -1000,9 +1038,10 @@ fn a_deadline_that_came_during_a_line_is_answered_without_waiting_for_the_next_p
// 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:?}");
assert_eq!(ran.count(&Event::InputClosed), 0, "the flush's turn has begun, not ended");
let stop_at = ran.events.iter().position(|event| matches!(event, Event::Stopped(_))).unwrap();
let reads_before = ran.events[..stop_at].iter().filter(|event| matches!(event, Event::Asked(_))).count();
assert_eq!(reads_before, 2, "the interrupt is in the pass that heard the result, not a poll away: {:?}", ran.events);
}
#[test]
@ -1057,7 +1096,7 @@ fn unreadable_and_foreign_lines_are_skipped_and_never_printed() {
Out(at("m1", 10)),
Out(result_line()),
]);
assert_eq!(ran.out, "request 1 prefix 10
assert_eq!(ran.lines(), "request 1 prefix 10
result subtype=success is_error=true: Failed to authenticate: OAuth session expired and could not be refreshed
");
let outcome = ran.outcome.as_ref().unwrap().to_string();
@ -1203,6 +1242,9 @@ mod process {
struct Spawn(PathBuf);
impl Spawner for Spawn {
fn hooks(&self) -> Hooks {
test_hooks()
}
fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
Ok(Box::new(Prompt(start_with(&self.0, args, token))))
}
@ -1229,13 +1271,16 @@ mod process {
struct Patient(PathBuf);
impl Spawner for Patient {
fn hooks(&self) -> Hooks {
test_hooks()
}
fn spawn(&self, args: &[String], token: &TokenFd) -> Result<Box<dyn Child>> {
Ok(Box::new(Bounded(start_with(&self.0, args, token), std::time::Instant::now())))
}
}
#[test]
fn a_whole_run_through_the_real_child_flushes_at_the_ceiling_and_ends_on_the_result() {
fn a_whole_run_through_the_real_child_flushes_at_the_ceiling_and_ends_on_the_flushs_own_result() {
let dir = tempfile::tempdir().unwrap();
let lines = [at("m1", 10), at("m2", CEILING_TOKENS)].join("
");
@ -1248,6 +1293,7 @@ read -t 5 start
read -t 5 flush
cat <<'EOF'
{result}
{result}
EOF
while read more; do :; done
exit 0");
@ -1256,19 +1302,17 @@ exit 0");
let mut out = Vec::new();
let outcome = carry(&Seen::of(&mind), &SystemClock, &FakeLender::new(Answer::Grant), &Spawn(program), TEST_CAP_S, &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
result subtype=success is_error=true: Failed to authenticate: OAuth session expired and could not be refreshed
");
let printed: String = String::from_utf8(out).unwrap().lines().filter(|line| !line.starts_with("decision ")).map(|line| format!("{line}\n")).collect();
let failed = "result subtype=success is_error=true: Failed to authenticate: OAuth session expired and could not be refreshed\n";
assert_eq!(printed, format!("request 1 prefix 10\nrequest 2 prefix 300000\n{failed}{failed}"));
}
#[test]
fn a_failure_result_before_any_request_closes_stdin_and_ends_the_run_promptly_through_the_real_child() {
fn a_child_that_ends_after_a_failure_result_ends_the_run_promptly_through_the_real_child() {
let dir = tempfile::tempdir().unwrap();
let result = result_line();
// An authentication or API failure: the error result is the first thing it prints, then it
// waits for its input to end, as `claude -p` over stream-json does, and exits 1.
let program = script(&dir, &format!("read -t 5 start\ncat <<'EOF'\n{result}\nEOF\nwhile read more; do :; done\nexit 1"));
// An authentication or API failure: the error result is the first thing it prints, then it exits 1.
let program = script(&dir, &format!("read -t 5 start\ncat <<'EOF'\n{result}\nEOF\nexit 1"));
let started = std::time::Instant::now();
let mut out = Vec::new();
let outcome = carry(&Seen::of(&with_run()), &SystemClock, &FakeLender::new(Answer::Grant), &Spawn(program), TEST_CAP_S, &mut out).unwrap();
@ -1434,3 +1478,151 @@ fn a_run_without_claims_gets_the_identity_lines_alone() {
let ran = go(vec![Out(result_line())]);
assert_eq!(ran.written()[0], start_message(&Slug("yggdrasil".into()), &run_ref("selfrun-a"), &[plan_window()], 1_793_306_000_000));
}
// ---- the live session -----------------------------------------------------
const NO_SIGNALS: &[Stop] = &[];
fn live_mind() -> Rc<TestMind> {
let mind = Rc::new(with_run());
mind.committed(vec![spec("x")]);
mind
}
/// A brief from her Persona: a follow-up with owner `self-run`.
fn brief(label: &str, item: &str) -> PipelineDocument {
let PipelineDocument::FollowUp(mut document) = follow_up(label, &spec_ref("x")) else { unreachable!() };
document.item = item.into();
document.owner = inbox::OWNER.into();
PipelineDocument::FollowUp(document)
}
fn brief_ref(label: &str) -> PipelineRef {
reference(K::FollowUp, &id("follow_up", label))
}
/// The mind gains these documents, `seconds` into the run.
fn admits(mind: &Rc<TestMind>, seconds: i64, documents: Vec<PipelineDocument>) -> Step {
let mind = mind.clone();
Step::Do(seconds, Rc::new(move || mind.committed(documents.clone())))
}
fn five_minute_request(id: &str, prefix: u64) -> String {
let mut event: Value = serde_json::from_str(&at(id, prefix)).unwrap();
event["message"]["usage"]["cache_creation"] = json!({"ephemeral_1h_input_tokens": 0, "ephemeral_5m_input_tokens": 100});
event.to_string()
}
fn tells(ran: &Ran) -> Vec<String> {
ran.written().into_iter().filter(|text| text.contains("A brief from your Persona")).collect()
}
#[test]
fn a_result_never_ends_the_session_the_child_ends_it_or_the_cache_does() {
let ran = go(vec![Out(at("m1", 10)), Out(result_line())]);
assert_eq!(ran.count(&Event::InputClosed), 0, "a result is the end of a turn: {:?}", ran.events);
assert_eq!(ran.ended().0, Cause::Own);
}
#[test]
fn a_brief_to_a_working_self_is_told_once_between_turns() {
let mind = live_mind();
let steps = vec![
Out(at("m1", 10)),
admits(&mind, 31, vec![brief("b1", "Survey the noise functions.")]),
// The first turn ends: the brief was spooled while it ran, so it is told now, as the next turn.
Out(result_line()),
Idle(1),
// A later poll does not spool it again.
Idle(31),
Out(result_line()),
Idle(1),
];
let ran = carry_with(&Seen::of(&mind), steps, Some(0), false);
let told = tells(&ran);
assert_eq!(told.len(), 1, "{:?}", ran.written());
assert!(told[0].contains(&brief_ref("b1").id.0) && told[0].contains("Survey the noise functions."), "{}", told[0]);
assert!(told[0].contains("Plan usage"), "the plan line ends a tell: {}", told[0]);
assert_eq!(ran.count(&Event::InputClosed), 0);
assert!(!inbox::pending(&ran.hooks.spool).unwrap(), "nothing left to tell");
}
#[test]
fn a_brief_told_mid_turn_by_the_hook_is_never_told_again_by_the_actuator() {
let mind = live_mind();
let hooks = test_hooks();
let printed: Rc<RefCell<Vec<String>>> = Rc::new(RefCell::new(Vec::new()));
let fire = {
let (spool, printed) = (hooks.spool.clone(), printed.clone());
Step::Do(0, Rc::new(move || printed.borrow_mut().push(inbox::inject(&spool, "{\"hook_event_name\":\"PostToolBatch\"}", false).unwrap().stdout)))
};
let steps = vec![Out(at("m1", 10)), admits(&mind, 31, vec![brief("b2", "Wire the shader uniforms.")]), fire.clone(), fire, Out(result_line()), Idle(1), Idle(31)];
let ran = carry_hooked(&Seen::of(&mind), steps, (Some(0), None, NO_SIGNALS, false), hooks);
let printed = printed.borrow();
assert!(printed[0].contains(&brief_ref("b2").id.0) && printed[0].contains("Wire the shader uniforms."), "{printed:?}");
assert_eq!(printed[1], "", "the second call finds nothing: each brief enters once");
assert!(tells(&ran).is_empty(), "the hook told it; the actuator did not: {:?}", ran.written());
}
#[test]
fn what_a_closed_run_claimed_is_never_told_and_what_nobody_claimed_is() {
let mind = TestMind::seeded();
mind.committed(vec![spec("x")]);
mind.committed(vec![brief("p1", "Stale."), brief("p2", "Still open.")]);
mind.committed(vec![run("old", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("p1")])]);
mind.committed(vec![close(run_ref("old"), recorded())]);
mind.committed(vec![self_run("selfrun-a", "5")]);
let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false);
let start = &ran.written()[0];
assert!(start.contains("Also open for you") && start.contains(&brief_ref("p2").id.0) && start.contains("Still open."), "{start}");
assert!(!start.contains(&brief_ref("p1").id.0), "never told: {start}");
}
#[test]
fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_brief_keeps_it_open() {
// An hour's cache: warm at 3539 s, the flush at 3540, the input closed after the flush's own result.
let hour = go(vec![Out(at("m1", 10)), Out(result_line()), Idle(3539), Idle(1), Out(result_line()), Idle(1)]);
assert_eq!(hour.written().iter().filter(|text| text.as_str() == FLUSH_MESSAGE).count(), 1);
let flush_at = hour.events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains(FLUSH_MESSAGE))).unwrap();
let closed_at = hour.events.iter().position(|event| *event == Event::InputClosed).unwrap();
assert!(flush_at < closed_at && hour.count(&Event::InputClosed) == 1, "{:?}", hour.events);
// Five minutes' cache: the flush at 240 s.
let short = go(vec![Out(five_minute_request("m1", 10)), Out(result_line()), Idle(239), Idle(1), Out(result_line()), Idle(1)]);
assert_eq!(flushes(&short), 1);
// Not a second early.
let early = go(vec![Out(at("m1", 10)), Out(result_line()), Idle(3539)]);
assert_eq!(flushes(&early), 0);
// A brief waiting at the cache's end is told, not flushed: the session stays open.
let mind = live_mind();
let steps = vec![Out(at("m1", 10)), Out(result_line()), admits(&mind, 3539, vec![brief("b5", "One more.")]), Idle(1), Idle(1), Out(result_line()), Idle(1)];
let ran = carry_with(&Seen::of(&mind), steps, Some(0), false);
assert_eq!((flushes(&ran), tells(&ran).len(), ran.count(&Event::InputClosed)), (0, 1, 0), "{:?}", ran.written());
}
#[test]
fn near_the_cap_an_idle_session_closes_for_rotation_and_a_working_one_is_asked_for_its_flush() {
let cap = i64::try_from(TEST_CAP_S).unwrap();
// Idle at cap - 1200 s: closed, with no flush asked.
let idle = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(199), Idle(1)]);
assert_eq!((idle.count(&Event::InputClosed), flushes(&idle)), (1, 0), "{:?}", idle.events);
let early = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(199)]);
assert_eq!(early.count(&Event::InputClosed), 0);
// Mid-turn at cap - 600 s: the flush, once.
let working = go(vec![After(cap - 600, at("m1", 10))]);
assert_eq!(flushes(&working), 1);
let before = go(vec![After(cap - 601, at("m1", 10))]);
assert_eq!(flushes(&before), 0);
}
#[test]
fn a_pre_compact_block_leaves_a_marker_and_the_session_asks_for_the_flush_once() {
let hooks = test_hooks();
let fire = {
let spool = hooks.spool.clone();
Step::Do(1, Rc::new(move || assert_eq!(inbox::inject(&spool, "{}", true).unwrap().exit, 2)))
};
let mind = with_run();
let ran = carry_hooked(&Seen::of(&mind), vec![Out(at("m1", 10)), fire.clone(), fire, Out(result_line())], (Some(0), None, NO_SIGNALS, false), hooks);
assert_eq!(flushes(&ran), 1, "{:?}", ran.written());
assert!(!inbox::take_pre_compact(&ran.hooks.spool), "the marker was taken");
}

6
crates/mind-body/src/self_run/tests/closure.rs
View file

@ -84,7 +84,7 @@ fn a_child_stopped_by_the_actuator_resolves_the_run_with_how_it_ended() {
fn a_child_that_cannot_start_resolves_the_run_and_the_error_stands() {
let mind = with_run();
let clock = Rc::new(Fixed(Cell::new(at_start())));
let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true };
let refusing = Fake::with(None, true);
let error = carry(&Seen::of(&mind), &*clock, &FakeLender::new(Answer::Grant), &refusing, TEST_CAP_S, &mut Vec::new()).unwrap_err();
assert_eq!(error.to_string(), "refused");
assert!(reason_of(&mind, "selfrun-a").contains("the actuator failed"));
@ -105,7 +105,7 @@ fn an_error_after_the_run_opened_resolves_it() {
let log = Rc::new(RefCell::new(Vec::new()));
let clock = Rc::new(Fixed(Cell::new(at_start())));
let child = Scripted { steps: vec![Out(at("m1", 10))].into(), log: log.clone(), clock: clock.clone(), code: Some(0), dies_on: None, refuses: vec![], hangs: true, exited: false, eof: false, asked: 0 };
let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false };
let fake = Fake::with(Some(child), false);
assert!(carry(&Seen::of(&mind), &*clock, &FakeLender::logging(log, Answer::Grant), &fake, TEST_CAP_S, &mut Broken).is_err());
assert!(reason_of(&mind, "selfrun-a").contains("the actuator failed"));
}
@ -133,7 +133,7 @@ fn a_resolve_that_fails_is_logged_and_tried_once() {
// The same when the run's own error is what is being reported: that error stands.
let clock = Rc::new(Fixed(Cell::new(at_start())));
let refusing = Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: true };
let refusing = Fake::with(None, true);
let again = Seen { admit_fails: true, ..Seen::of(&mind) };
let error = carry(&again, &*clock, &FakeLender::new(Answer::Grant), &refusing, TEST_CAP_S, &mut Vec::new()).unwrap_err();
assert_eq!(error.to_string(), "refused");

2
crates/mind-body/src/self_run/tests/lease.rs
View file

@ -7,7 +7,7 @@ use eureka_pipeline::ResolutionOutcome;
use super::*;
fn never_spawns() -> Fake {
Fake { child: RefCell::new(None), spawns: RefCell::new(Vec::new()), refuse: false }
Fake::with(None, false)
}
#[test]

10
crates/mind-body/src/self_run/tests/process/lease_child.rs
View file

@ -26,7 +26,7 @@ fn granted_fd_reaches_child_env_var() {
let dir = tempfile::tempdir().unwrap();
let program = script(&dir, READS_ITS_CREDENTIAL);
let token = token_holding("tok-canary-7f3a");
let mut child = start_with(&program, &claude_args(session_id(&run_ref("selfrun-a"))), &token);
let mut child = start_with(&program, &claude_args(session_id(&run_ref("selfrun-a")), &Hooks { spool: PathBuf::from("/run/spool"), injector: PathBuf::from("/bin/x") }), &token);
child.close_input();
let (lines, code) = drain(&mut *child);
assert_eq!(code, Some(0), "{lines:?}");
@ -128,11 +128,11 @@ fn each_run_gets_a_fresh_owner_only_configuration_that_what_it_writes_does_not_o
#[test]
fn the_spawner_needs_the_units_runtime_directory_and_reads_the_seed_from_the_units_environment() {
assert!(ClaudeSpawner::new(Vec::new()).is_err(), "no scratch directory named");
assert!(ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "".into())]).is_err(), "an empty one is none");
let named = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into()), ("MIND_SELF_CONFIG_SEED".into(), "/etc/seed".into())]).unwrap();
assert!(ClaudeSpawner::new(Vec::new(), PathBuf::from("/bin/x")).is_err(), "no scratch directory named");
assert!(ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "".into())], PathBuf::from("/bin/x")).is_err(), "an empty one is none");
let named = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into()), ("MIND_SELF_CONFIG_SEED".into(), "/etc/seed".into())], PathBuf::from("/bin/x")).unwrap();
assert_eq!((named.scratch.as_path(), named.seed.as_deref()), (Path::new("/run/mind"), Some(Path::new("/etc/seed"))));
let bare = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into())]).unwrap();
let bare = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into())], PathBuf::from("/bin/x")).unwrap();
assert_eq!(bare.seed, None);
}

32
crates/mind-body/tests/public_surface.txt
View file

@ -214,20 +214,27 @@ 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_MARGIN_S : i64
mind_body::self_run: const FLUSH_MESSAGE : & str
mind_body::self_run: const INBOX_POLL_S : i64
mind_body::self_run: const MAX_TURNS : u32
mind_body::self_run: const ROTATE_MARGIN_S : i64
mind_body::self_run: const RUN_CAP_VARIABLE : & str
mind_body::self_run: const SCRATCH_VARIABLE : & str
mind_body::self_run: const SEED_VARIABLE : & str
mind_body::self_run: const SESSION_NAMESPACE : Uuid
mind_body::self_run: const TTL_1H_S : i64
mind_body::self_run: const TTL_5M_S : i64
mind_body::self_run: const TTL_MARGIN_S : i64
mind_body::self_run: fn carry (mind : & dyn MindPort , clock : & dyn Clock , lender : & dyn Lender , spawner : & dyn Spawner , run_cap_s : u64 , out : & mut dyn Write) -> Result < SelfRunOutcome >
mind_body::self_run: fn claude_args (session : Uuid) -> Vec < String >
mind_body::self_run: fn claude_args (session : Uuid , hooks : & Hooks) -> Vec < String >
mind_body::self_run: fn read_usage (line : & str) -> Usage
mind_body::self_run: fn session_id (run : & PipelineRef) -> Uuid
mind_body::self_run: fn start_message (instance : & Slug , run : & PipelineRef , plan_quota : & [PlanQuota] , now_ms : u64) -> String
mind_body::self_run: fn writes_hour_cache (line : & str) -> Option < bool >
mind_body::self_run: impl Child for ClaudeProcess
mind_body::self_run: impl ClaudeSpawner
mind_body::self_run: impl ClaudeSpawner: fn new (unit_environment : impl IntoIterator < Item = (OsString , OsString) >) -> Result < Self >
mind_body::self_run: impl ClaudeSpawner: fn new (unit_environment : impl IntoIterator < Item = (OsString , OsString) > , injector : PathBuf) -> Result < Self >
mind_body::self_run: impl Drop for FreshConfig
mind_body::self_run: impl FreshConfig
mind_body::self_run: impl Heard
@ -238,9 +245,28 @@ mind_body::self_run: impl Stop
mind_body::self_run: impl fmt :: Display for SeedLinkRefused
mind_body::self_run: impl fmt :: Display for SelfRunOutcome
mind_body::self_run: impl std :: error :: Error for SeedLinkRefused
mind_body::self_run: mod inbox
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] , token : & TokenFd) -> Result < Box < dyn Child > > ; }
mind_body::self_run: pub trait Spawner { fn hooks (& self) -> Hooks ; fn spawn (& self , args : & [String] , token : & TokenFd) -> Result < Box < dyn Child > > ; }
mind_body::self_run: struct ClaudeSpawner pub-fields[]
mind_body::self_run::decide: impl State
mind_body::self_run::inbox: const OWNER : & str
mind_body::self_run::inbox: fn claim (spool : & Path) -> Result < Vec < Brief > >
mind_body::self_run::inbox: fn derive (mind : & dyn MindPort , spool : & Path) -> Result < Vec < Brief > >
mind_body::self_run::inbox: fn inject (spool : & Path , input : & str , pre_compact : bool) -> Result < Injection >
mind_body::self_run::inbox: fn mark_told (spool : & Path , briefs : & [Brief]) -> Result < () >
mind_body::self_run::inbox: fn pending (spool : & Path) -> Result < bool >
mind_body::self_run::inbox: fn render (briefs : & [Brief] , plan : & str) -> String
mind_body::self_run::inbox: fn reset (spool : & Path) -> Result < () >
mind_body::self_run::inbox: fn spool (spool : & Path , briefs : & [Brief]) -> Result < () >
mind_body::self_run::inbox: fn take_pre_compact (spool : & Path) -> bool
mind_body::self_run::inbox: fn write_plan (spool : & Path , line : & str) -> Result < () >
mind_body::self_run::inbox: impl Brief
mind_body::self_run::inbox: impl Hooks
mind_body::self_run::inbox: impl Hooks: fn settings (& self) -> String
mind_body::self_run::inbox: struct Brief # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[id: String, campaign: String, repo: String, item: String, ordinal: u64]
mind_body::self_run::inbox: struct Hooks # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[spool: PathBuf, injector: PathBuf]
mind_body::self_run::inbox: struct Injection # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[stdout: String, exit: i32]
mind_body::tempo: const DEFAULT_BEAT_S : u32
mind_body::tempo: const LOG_BOUND_ORDERS : f64
mind_body::tempo: fn pick (rows : & [Standing] , beat_s : u32) -> Option < usize >