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

self-session-live: one stdin write outstanding, coalesced; an idle end closes no brief; a brief is decided by admission provenance; synthetic assistant lines are no request; decide's dead Tells guard and clock deleted; window margin asserted

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 20:42:34 +01:00
commit 9ad80a1b72
7 changed files with 272 additions and 266 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -14,14 +14,17 @@
//! //!
//! One run is one live session (ruling `eureka-body:ruling:live-mind-sessions`). //! 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 //! 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` sourced //! her mind (the `inbox` module): an open follow-up with owner `self-run` admitted
//! from a Persona turn is a brief. `drive` is the one writer into her session: the //! by her Persona turn is a brief. `drive` is the one writer into her
//! start message, every tell and the flush are stdin lines, a tell written whenever //! session, and keeps one write outstanding (ruling
//! a brief waits and the flush has not been written, mid-turn or idle. The CLI //! `eureka-body:ruling:self-session-one-outstanding-write`): the start message,
//! echoes each line it takes into a request (`--replay-user-messages`); the echo is //! then one line at a time holding the briefs that wait and the flush ask, each
//! the consumption point, folded or queued, and idle is derived from the lines //! sent only after the previous line's echo, mid-turn or idle. The CLI echoes each
//! written and the lines echoed. Her Self resolves each brief she answers; the //! line it takes into a request (`--replay-user-messages`) and joins lines queued
//! actuator closes only what she left open at an idle end. Compaction is off in //! behind one request into a single echo, so the echo is the consumption point
//! only while no second line is in flight. Her Self resolves each brief she
//! answers; the actuator closes none, and a brief she left open is told to her
//! next session. Compaction is off in
//! the CLI (`DISABLE_COMPACT=1`): the organ's own ceiling, flush, end and rehydrate //! the CLI (`DISABLE_COMPACT=1`): the organ's own ceiling, flush, end and rehydrate
//! is the only path. The session ends where its cache would be evicted, near the //! is the only path. 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. //! unit's cap, at the context ceiling, or after the flush each of those asks.
@ -79,6 +82,12 @@ use inbox::Brief;
/// The measured Self ceiling: at this prefix the flush is asked for. /// The measured Self ceiling: at this prefix the flush is asked for.
pub const CEILING_TOKENS: u64 = 300_000; pub const CEILING_TOKENS: u64 = 300_000;
/// The context window the CLI reports for the model a Self runs (`modelUsage.contextWindow` in every result of the
/// `claude-2.1.312-live-session-*` fixtures: the default model, `claude-opus-5-5`, 1M). The seed sets no model, so the CLI's default is the model.
/// The window under the real subscription is read from a result's `contextWindow` and not otherwise observed here.
pub const SELF_CONTEXT_WINDOW_TOKENS: u64 = 1_000_000;
/// What the CLI keeps free below its window before it refuses a prompt (`blocking_limit`): the blocking-limit fixture sends a 990k request and refuses the next.
pub const CLI_BLOCKING_BUFFER_TOKENS: u64 = 10_000;
/// What the prefix may grow by beyond the prefix the flush was asked at before /// What the prefix may grow by beyond the prefix the flush was asked at before
/// the actuator interrupts. Relative to the flush, never an absolute limit. /// the actuator interrupts. Relative to the flush, never an absolute limit.
pub const FLUSH_HEADROOM_TOKENS: u64 = 60_000; pub const FLUSH_HEADROOM_TOKENS: u64 = 60_000;
@ -94,6 +103,9 @@ pub const ROTATE_MARGIN_S: i64 = 25 * 60;
/// The signal ladder's length: three escalations, the stream drain and a minute of slack. The grace interrupt is never later than the cap less this. /// The signal ladder's length: three escalations, the stream drain and a minute of slack. The grace interrupt is never later than the cap less this.
pub const LADDER_S: i64 = 3 * ESCALATE_AFTER_S + EXIT_DRAIN_S + 60; pub const LADDER_S: i64 = 3 * ESCALATE_AFTER_S + EXIT_DRAIN_S + 60;
const _: () = assert!(FLUSH_GRACE_S + LADDER_S < ROTATE_MARGIN_S, "the flush and the whole ladder must finish before the unit's cap"); const _: () = assert!(FLUSH_GRACE_S + LADDER_S < ROTATE_MARGIN_S, "the flush and the whole ladder must finish before the unit's cap");
/// The ceiling, and the headroom its flush's own turn may grow by, sit below the CLI's blocking limit with 600k to spare at a 1M window; a window under
/// 400k has no room for the flush's turn, and the assertion refuses to compile until the ceiling is cut to fit the window.
const _: () = assert!(CEILING_TOKENS + FLUSH_HEADROOM_TOKENS + 40_000 < SELF_CONTEXT_WINDOW_TOKENS - CLI_BLOCKING_BUFFER_TOKENS, "the ceiling and its flush headroom must fit under the CLI's blocking limit");
/// How often the mind is asked for new briefs. /// How often the mind is asked for new briefs.
pub const INBOX_POLL_S: i64 = 30; pub const INBOX_POLL_S: i64 = 30;
/// The prompt cache's lifetimes, and how long before one runs out the session ends. /// The prompt cache's lifetimes, and how long before one runs out the session ends.
@ -234,6 +246,9 @@ pub fn claude_args(session: Uuid) -> Vec<String> {
.to_vec() .to_vec()
} }
/// The model name on the CLI's own assistant lines, which no request produced.
const SYNTHETIC_MODEL: &str = "<synthetic>";
/// One line of the child's stream, as far as the actuator reads it. /// One line of the child's stream, as far as the actuator reads it.
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub enum Usage { pub enum Usage {
@ -279,6 +294,10 @@ pub fn read_usage(line: &str) -> Usage {
} }
Some("assistant") if event.get("parent_tool_use_id").is_none_or(Value::is_null) => { Some("assistant") if event.get("parent_tool_use_id").is_none_or(Value::is_null) => {
let message = event.get("message"); let message = event.get("message");
// The CLI's own lines after an API error or a refused slash command carry a uuid id and no usage: no request was made.
if message.and_then(|message| message.get("model")).and_then(Value::as_str) == Some(SYNTHETIC_MODEL) {
return Usage::Other;
}
let Some(id) = message.and_then(|message| message.get("id")).and_then(Value::as_str) else { return Usage::Other }; let Some(id) = message.and_then(|message| message.get("id")).and_then(Value::as_str) else { return Usage::Other };
let usage = message.and_then(|message| message.get("usage")); let usage = message.and_then(|message| message.get("usage"));
let field = |name: &str| usage.and_then(|usage| usage.get(name)).and_then(Value::as_u64).unwrap_or(0); let field = |name: &str| usage.and_then(|usage| usage.get(name)).and_then(Value::as_u64).unwrap_or(0);
@ -414,21 +433,13 @@ fn her_run(mind: &dyn MindPort) -> Result<Option<(PipelineRef, Vec<PipelineRef>)
struct Heard { struct Heard {
/// The last result heard, ok or not: its head line and its text cut to `RATIONALE_RESULT_LIMIT`. /// The last result heard, ok or not: its head line and its text cut to `RATIONALE_RESULT_LIMIT`.
concluded: Option<(String, String)>, concluded: Option<(String, String)>,
/// The text of the last ok result: what an idle end records as her answer.
answer: Option<String>,
/// The idle flush was answered by an ok result.
idle_end: bool,
} }
impl Heard { impl Heard {
fn hear(&mut self, line: &str) { fn hear(&mut self, line: &str) {
let Ok(Value::Object(event)) = serde_json::from_str::<Value>(line) else { return }; let Ok(Value::Object(event)) = serde_json::from_str::<Value>(line) else { return };
if event.get("type").and_then(Value::as_str) == Some("result") { if event.get("type").and_then(Value::as_str) == Some("result") {
let text = cut_text(result_text(&event), RATIONALE_RESULT_LIMIT); self.concluded = Some((result_head(&event), cut_text(result_text(&event), RATIONALE_RESULT_LIMIT)));
if !event.get("is_error").and_then(Value::as_bool).unwrap_or(false) {
self.answer = Some(text.clone());
}
self.concluded = Some((result_head(&event), text));
} }
} }
} }
@ -521,10 +532,6 @@ Also open for you:");
} }
let mut heard = Heard::default(); let mut heard = Heard::default();
let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &task), &mut told, &mut heard); let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &task), &mut told, &mut heard);
// An idle end answered by an ok result: what she left open is closed, saying so. After a rotation, a ceiling, an error or a crash the told briefs stay open for the successor.
if heard.idle_end && ran.is_ok() {
close_left_open(mind, clock, &run, &told, &heard, out);
}
let closed = close_if_open(mind, clock, &run, &ran, &heard); let closed = close_if_open(mind, clock, &run, &ran, &heard);
if closed.is_err() { if closed.is_err() {
let _ = writeln!(out, "the run could not be closed: the next launch closes it"); let _ = writeln!(out, "the run could not be closed: the next launch closes it");
@ -599,20 +606,6 @@ fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> {
writeln!(out, "{}", describe_result(&event, RESULT_TEXT_LIMIT)) writeln!(out, "{}", describe_result(&event, RESULT_TEXT_LIMIT))
} }
/// What a brief's closing resolution says when her last answer had no text.
const NO_ANSWER_TEXT: &str = "The turn ended with no answer text.";
/// At an idle end: resolves Recorded the told briefs she left in force, the last ok result's text as the rationale. One try; a failure is logged and the briefs stay open for the successor.
fn close_left_open(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef, told: &[Brief], heard: &Heard, out: &mut dyn Write) {
let rationale = heard.answer.as_deref().filter(|text| !text.is_empty()).unwrap_or(NO_ANSWER_TEXT);
let label = session_label(run);
let reason = format!("told to Self run {label}; the session went idle and ended without her resolving it");
let answered: Vec<(Brief, String)> = told.iter().map(|brief| (brief.clone(), rationale.to_string())).collect();
if inbox::close(mind, clock.now(), label, &answered, &reason).is_err() {
let _ = writeln!(out, "briefs could not be closed: they stay open for the next session");
}
}
/// The loop: facts in, one clock read per fact, actions out. The one writer into her session. /// The loop: facts in, one clock read per fact, actions out. The one writer into her session.
#[allow(clippy::too_many_arguments, reason = "the loop's ports and the run's facts, passed on whole")] #[allow(clippy::too_many_arguments, reason = "the loop's ports and the run's facts, passed on whole")]
fn drive( fn drive(
@ -677,16 +670,21 @@ fn drive(
let (ttl_s, inbox, idle_s) = (state.ttl_s, state.inbox || matches!(&heard_event, Event::Inbox { pending: true }), state.idle_s(now)); let (ttl_s, inbox, idle_s) = (state.ttl_s, state.inbox || matches!(&heard_event, Event::Inbox { pending: true }), state.idle_s(now));
let (next, action) = decide(state, heard_event, now); let (next, action) = decide(state, heard_event, now);
state = next; state = next;
heard.idle_end = state.idle_end_answered();
if state.requests != counted { if state.requests != counted {
writeln!(out, "request {} prefix {}", state.requests, state.prefix)?; writeln!(out, "request {} prefix {}", state.requests, state.prefix)?;
} }
match &action { match &action {
Action::None => {} Action::None => {}
Action::WriteFlush(why) => child.write_line(&user_line(&flush_text(*why))), Action::Write { tells, flush } => {
Action::WriteTells => { let mut parts = Vec::new();
child.write_line(&user_line(&inbox::render(&waiting, &plan))); if *tells {
told.append(&mut waiting); parts.push(inbox::render(&waiting, &plan));
told.append(&mut waiting);
}
parts.extend(flush.map(flush_text));
child.write_line(&user_line(&parts.join("
")));
} }
Action::CloseStdin => { Action::CloseStdin => {
child.close_input(); child.close_input();

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

@ -7,25 +7,29 @@
//! into events, reads the clock once per event, and executes the action. A //! 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. //! timing rule is pinned here, by tables and by an enumeration of event orders.
//! //!
//! Delivery is counted, not guessed: every stdin line the actuator writes (the //! Delivery is counted, not guessed, and at most one write is outstanding (ruling
//! start message, a tell, the flush) increments `written`; every replay echo of //! `eureka-body:ruling:self-session-one-outstanding-write`): the CLI joins the
//! a line increments `consumed`. The echo is the point the CLI takes a line into //! lines queued behind a request into one echo, so a second line is never
//! a request, folded into the running turn or queued for the next, so one rule //! written before the first one's echo. Every stdin line the actuator writes (the
//! serves both. Idle is derived: no turn open and `written == consumed`. //! start message, a write of tells and the flush) increments `written`; every
//! replay echo increments `consumed`. The echo is the point the CLI takes a line
//! into a request, folded into the running turn or queued for the next. What
//! waits (the briefs, the flush ask) is coalesced into the next single write.
//! Idle is derived: no turn open and no write outstanding.
//! //!
//! The rules, in this order, while stdin is open and no signal is out: //! The rules, in this order, while stdin is open and no signal is out:
//! (1) a result that is blocked, or an error before any answer, closes stdin: no //! (1) a result that is blocked, or an error before any answer, closes stdin: no
//! cache is worth keeping and a flush would fail the same way; (2) the first //! cache is worth keeping and a flush would fail the same way; (2) the first
//! result after the flush's echo closes stdin, folded or queued alike; (3) a //! result after the flush's echo closes stdin, folded or queued alike; (3) the
//! brief waiting and no flush written: tell it, idle or mid-turn; (4) the cap //! cap less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (4)
//! less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (5) idle //! idle with no brief waiting, at the last result's cache lifetime less a margin:
//! with no brief waiting, at the last result's cache lifetime less a margin: ask //! ask the flush (idle); (5) with nothing outstanding, write what waits: the
//! the flush (idle). The ceiling asks the same flush from a request. Compaction //! briefs and the flush ask if there is one, as one line. The ceiling asks the
//! is off in the CLI (`DISABLE_COMPACT=1`), so the flush is the only way a //! same flush from a request. Compaction is off in the CLI (`DISABLE_COMPACT=1`),
//! session ends short of its cache. //! so the flush is the only way a session ends short of its cache.
//! //!
//! The headroom counts the requests after the flush's echo, from the prefix the //! The headroom counts the requests after the flush's echo, from the prefix the
//! flush was written at. The grace interrupt is at the earlier of the grace //! flush was asked at. The grace interrupt is at the earlier of the grace
//! after the flush write and the cap less `LADDER_S`, so the signal ladder //! after the flush write and the cap less `LADDER_S`, so the signal ladder
//! finishes before the unit's cap (ruling //! finishes before the unit's cap (ruling
//! `eureka-body:ruling:flush-always-gets-a-turn`). At most one action comes out //! `eureka-body:ruling:flush-always-gets-a-turn`). At most one action comes out
@ -74,9 +78,8 @@ pub(super) enum Event {
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Action { pub(super) enum Action {
None, None,
WriteFlush(Why), /// The one write: the waiting briefs and the flush ask, coalesced into one user line.
/// Write the waiting briefs as one user line. Write { tells: bool, flush: Option<Why> },
WriteTells,
CloseStdin, CloseStdin,
/// To the child's whole process group. /// To the child's whole process group.
Signal(Stop), Signal(Stop),
@ -85,11 +88,11 @@ pub(super) enum Action {
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
struct Flush { struct Flush {
/// The flush's line number: its echo is the `seq`-th. /// The flush's line number once written (its echo is the `seq`-th); none while it waits for the outstanding write's echo.
seq: u32, seq: Option<u32>,
/// The prefix at which the flush was written: the headroom's base. /// The prefix at which the flush was asked: the headroom's base.
prefix: u64, prefix: u64,
written: DateTime<Utc>, asked: DateTime<Utc>,
/// Its echo has come: the CLI took it into a request. /// Its echo has come: the CLI took it into a request.
consumed: bool, consumed: bool,
/// A request after its echo reached the headroom. /// A request after its echo reached the headroom.
@ -116,8 +119,6 @@ pub(super) struct State {
cap: Option<(DateTime<Utc>, i64)>, cap: Option<(DateTime<Utc>, i64)>,
stdin_closed: bool, stdin_closed: bool,
flush: Option<Flush>, flush: Option<Flush>,
/// An ok result answered the flush.
flush_answered_ok: bool,
/// The last signal attempted and when. /// The last signal attempted and when.
stopping: Option<(Stop, DateTime<Utc>)>, stopping: Option<(Stop, DateTime<Utc>)>,
/// The strongest signal delivered. /// The strongest signal delivered.
@ -125,8 +126,6 @@ pub(super) struct State {
exited: Option<(Option<i32>, DateTime<Utc>)>, exited: Option<(Option<i32>, DateTime<Utc>)>,
eof: bool, eof: bool,
finished: bool, finished: bool,
/// The clock at the last decision: where a due action says "now".
at: Option<DateTime<Utc>>,
} }
impl State { impl State {
@ -135,19 +134,19 @@ impl State {
Self { written: 1, cap: Some((at, i64::try_from(run_cap_s).unwrap_or(i64::MAX))), ..Self::default() } Self { written: 1, cap: Some((at, i64::try_from(run_cap_s).unwrap_or(i64::MAX))), ..Self::default() }
} }
/// A line written whose echo has not come.
fn outstanding(&self) -> bool {
self.written > self.consumed
}
fn idle(&self) -> bool { fn idle(&self) -> bool {
!self.turn_open && self.written == self.consumed !self.turn_open && !self.outstanding()
} }
/// Seconds idle since the last result, none while a line or a turn is outstanding. /// Seconds idle since the last result, none while a line or a turn is outstanding.
pub(super) fn idle_s(&self, now: DateTime<Utc>) -> Option<i64> { pub(super) fn idle_s(&self, now: DateTime<Utc>) -> Option<i64> {
self.idle().then(|| self.last_result_at.map_or(0, |at| (now - at).num_seconds())) self.idle().then(|| self.last_result_at.map_or(0, |at| (now - at).num_seconds()))
} }
/// The session went idle, its cache would have lapsed, and its idle flush was answered by an ok result.
pub(super) fn idle_end_answered(&self) -> bool {
self.flush_answered_ok && self.flush.as_ref().is_some_and(|flush| flush.why == Why::Idle)
}
} }
fn outcome(state: &State, cause: Cause, code: Option<i32>) -> SelfRunOutcome { fn outcome(state: &State, cause: Cause, code: Option<i32>) -> SelfRunOutcome {
@ -166,28 +165,21 @@ fn cause(state: &State) -> Cause {
/// When the flush's grace runs out: the grace after the write, never later than /// When the flush's grace runs out: the grace after the write, never later than
/// the point the signal ladder needs to finish before the unit's cap. /// the point the signal ladder needs to finish before the unit's cap.
fn grace_deadline(state: &State, flush: &Flush) -> DateTime<Utc> { fn grace_deadline(state: &State, flush: &Flush) -> DateTime<Utc> {
let grace = flush.written + Duration::seconds(FLUSH_GRACE_S); let grace = flush.asked + Duration::seconds(FLUSH_GRACE_S);
match state.cap { match state.cap {
Some((started, cap)) => grace.min(started + Duration::seconds(cap.saturating_sub(LADDER_S))), Some((started, cap)) => grace.min(started + Duration::seconds(cap.saturating_sub(LADDER_S))),
None => grace, None => grace,
} }
} }
/// What the session does next on its own, and when. /// When the flush is next asked on its own, and why: the cap's margin, or the
enum Due { /// cache's end while idle with no brief waiting.
Tells, fn due(state: &State) -> Option<(DateTime<Utc>, Why)> {
Flush(Why),
}
fn due(state: &State) -> Option<(DateTime<Utc>, Due)> {
if state.stdin_closed || state.exited.is_some() || state.flush.is_some() { if state.stdin_closed || state.exited.is_some() || state.flush.is_some() {
return None; return None;
} }
if state.inbox { let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Why::Rotation));
return state.at.map(|at| (at, Due::Tells)); let idle = (state.idle() && !state.inbox).then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Why::Idle));
}
let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Due::Flush(Why::Rotation)));
let idle = state.idle().then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Due::Flush(Why::Idle)));
[rotation, idle].into_iter().flatten().min_by_key(|(at, _)| *at) [rotation, idle].into_iter().flatten().min_by_key(|(at, _)| *at)
} }
@ -209,34 +201,48 @@ pub(super) fn next_deadline(state: &State) -> Option<DateTime<Utc>> {
} }
/// The flush is asked once, by the ceiling, the cap's margin or the cache's end; /// The flush is asked once, by the ceiling, the cap's margin or the cache's end;
/// never to a child whose input is closed or that has exited. /// never to a child whose input is closed or that has exited. Asking writes
/// nothing: `send` writes it when no write is outstanding.
fn ask_flush(state: &mut State, now: DateTime<Utc>, why: Why) -> bool { fn ask_flush(state: &mut State, now: DateTime<Utc>, why: Why) -> bool {
if state.flush.is_some() || state.stdin_closed || state.exited.is_some() { if state.flush.is_some() || state.stdin_closed || state.exited.is_some() {
return false; return false;
} }
state.written += 1; state.flush = Some(Flush { seq: None, prefix: state.prefix, asked: now, consumed: false, exhausted: false, why });
state.flush = Some(Flush { seq: state.written, prefix: state.prefix, written: now, consumed: false, exhausted: false, why });
true true
} }
/// The lifecycle rules 3 to 5: tell a waiting brief, or ask the flush where it is due. /// The lifecycle rules 3 to 5: ask the flush where it is due, then write what waits.
fn lifecycle(state: &mut State, now: DateTime<Utc>) -> Action { fn lifecycle(state: &mut State, now: DateTime<Utc>) -> Action {
match due(state) { if let Some((at, why)) = due(state)
Some((at, Due::Tells)) if at <= now => { && at <= now
state.inbox = false; {
state.written += 1; ask_flush(state, now, why);
Action::WriteTells
}
Some((at, Due::Flush(why))) if at <= now && ask_flush(state, now, why) => Action::WriteFlush(why),
_ => Action::None,
} }
send(state)
}
/// With nothing outstanding, the one write: the waiting briefs (until the flush
/// is written) and the flush, if it has not been written, as one line.
fn send(state: &mut State) -> Action {
if state.stdin_closed || state.exited.is_some() || state.outstanding() || state.flush.as_ref().is_some_and(|flush| flush.seq.is_some()) {
return Action::None;
}
let flush = state.flush.as_ref().map(|flush| flush.why);
if !state.inbox && flush.is_none() {
return Action::None;
}
state.written += 1;
let seq = state.written;
if let Some(flush) = state.flush.as_mut() {
flush.seq = Some(seq);
}
Action::Write { tells: std::mem::take(&mut state.inbox), flush }
} }
pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (State, Action) { pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (State, Action) {
if state.finished { if state.finished {
return (state, Action::None); return (state, Action::None);
} }
state.at = Some(now);
let mut action = Action::None; let mut action = Action::None;
match event { match event {
Event::Request { id, prefix } => { Event::Request { id, prefix } => {
@ -249,8 +255,8 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (Sta
{ {
flush.exhausted = true; flush.exhausted = true;
} }
if prefix >= CEILING_TOKENS && ask_flush(&mut state, now, Why::Ceiling) { if prefix >= CEILING_TOKENS {
action = Action::WriteFlush(Why::Ceiling); ask_flush(&mut state, now, Why::Ceiling);
} }
} }
} }
@ -259,7 +265,7 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (Sta
state.turn_open = true; state.turn_open = true;
let consumed = state.consumed; let consumed = state.consumed;
if let Some(flush) = state.flush.as_mut() if let Some(flush) = state.flush.as_mut()
&& consumed >= flush.seq && flush.seq.is_some_and(|seq| consumed >= seq)
{ {
flush.consumed = true; flush.consumed = true;
} }
@ -271,7 +277,6 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (Sta
let no_answer = !ok && !state.answered_once; let no_answer = !ok && !state.answered_once;
state.answered_once |= ok; state.answered_once |= ok;
let flush_answered = state.flush.as_ref().is_some_and(|flush| flush.consumed); let flush_answered = state.flush.as_ref().is_some_and(|flush| flush.consumed);
state.flush_answered_ok |= flush_answered && ok;
if (blocked || no_answer || flush_answered) && !state.stdin_closed && state.exited.is_none() { if (blocked || no_answer || flush_answered) && !state.stdin_closed && state.exited.is_none() {
state.stdin_closed = true; state.stdin_closed = true;
action = Action::CloseStdin; action = Action::CloseStdin;

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

@ -49,6 +49,15 @@ fn last(events: Vec<(Event, i64)>) -> Action {
run(events).1.pop().unwrap() run(events).1.pop().unwrap()
} }
/// The one write with only the waiting briefs, and with only the flush.
fn tells() -> Action {
Action::Write { tells: true, flush: None }
}
fn flush(why: Why) -> Action {
Action::Write { tells: false, flush: Some(why) }
}
fn delivered(how: Stop) -> Event { fn delivered(how: Stop) -> Event {
Event::Signalled { how, delivered: true } Event::Signalled { how, delivered: true }
} }
@ -68,13 +77,33 @@ fn idle_at(at: i64) -> (State, Vec<Action>) {
#[test] #[test]
fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or_for_a_repeated_request() { fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or_for_a_repeated_request() {
assert_eq!(last(vec![(req("a", CEILING_TOKENS - 1), 0)]), Action::None); assert_eq!(last(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS - 1), 0)]), Action::None);
let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 0), (req("a", CEILING_TOKENS), 1), (req("b", CEILING_TOKENS + 1), 2)]); let (state, actions) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 0), (req("a", CEILING_TOKENS), 1), (req("b", CEILING_TOKENS + 1), 2)]);
assert_eq!(actions, vec![Action::WriteFlush(Why::Ceiling), Action::None, Action::None]); assert_eq!(actions, vec![Action::None, flush(Why::Ceiling), Action::None, Action::None]);
assert_eq!((state.written, state.requests), (2, 2)); assert_eq!((state.written, state.requests), (2, 2));
assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (req("a", CEILING_TOKENS), 1)]), Action::None); assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (req("a", CEILING_TOKENS), 1)]), Action::None);
} }
#[test]
fn what_waits_is_one_write_sent_only_after_the_outstanding_writes_echo_the_briefs_and_the_flush_ask_together() {
// The start message is not echoed: the ceiling asks the flush and a brief waits, and nothing is written.
let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 1), (Event::Inbox { pending: true }, 2), (Event::Tick, 3)]);
assert_eq!(actions, vec![Action::None, Action::None, Action::None]);
assert_eq!(state.written, 1);
// The echo frees the writer: one write carries both, and the flush is asked only once.
let (state, action) = feed(state, Event::Echo, 4);
assert_eq!(action, Action::Write { tells: true, flush: Some(Why::Ceiling) });
assert_eq!((state.written, state.consumed), (2, 1));
assert_eq!(next_deadline(&state), Some(t(1 + GRACE)), "the grace runs from the ask");
// Briefs alone: the first poll writes, the second finds that write outstanding and waits for its echo.
let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1), (Event::Inbox { pending: true }, 2)]);
assert_eq!(actions, vec![Action::None, tells(), Action::None]);
assert_eq!(state.written, 2);
let (state, action) = feed(state, Event::Echo, 3);
assert_eq!(action, tells(), "the second poll's brief is written at the echo");
assert_eq!(state.written, 3);
}
#[test] #[test]
fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every_line_was_echoed_and_no_turn_runs() { fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every_line_was_echoed_and_no_turn_runs() {
let (state, _) = run(vec![]); let (state, _) = run(vec![]);
@ -86,7 +115,7 @@ fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every
assert_eq!(state.idle_s(t(10)), Some(3)); assert_eq!(state.idle_s(t(10)), Some(3));
// A tell written after the result is outstanding until its echo: not idle. // A tell written after the result is outstanding until its echo: not idle.
let (state, actions) = run(vec![(Event::Echo, 0), (OK, 7), (Event::Inbox { pending: true }, 8)]); let (state, actions) = run(vec![(Event::Echo, 0), (OK, 7), (Event::Inbox { pending: true }, 8)]);
assert_eq!(actions[2], Action::WriteTells); assert_eq!(actions[2], tells());
assert_eq!(state.idle_s(t(9)), None); assert_eq!(state.idle_s(t(9)), None);
// The result of a turn with a queued line outstanding is not idleness. // The result of a turn with a queued line outstanding is not idleness.
let (state, _) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1), (OK, 2)]); let (state, _) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1), (OK, 2)]);
@ -96,7 +125,7 @@ fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every
#[test] #[test]
fn a_flush_folded_into_the_running_turn_is_answered_by_the_result_after_its_echo() { fn a_flush_folded_into_the_running_turn_is_answered_by_the_result_after_its_echo() {
let (state, actions) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 1), (Event::Echo, 2), (OK, 3)]); let (state, actions) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 1), (Event::Echo, 2), (OK, 3)]);
assert_eq!(actions, vec![Action::None, Action::WriteFlush(Why::Ceiling), Action::None, Action::CloseStdin]); assert_eq!(actions, vec![Action::None, flush(Why::Ceiling), Action::None, Action::CloseStdin]);
assert_eq!(next_deadline(&state), Some(t(1 + GRACE)), "closed: only the flush's grace backstop remains, no lifecycle deadline"); assert_eq!(next_deadline(&state), Some(t(1 + GRACE)), "closed: only the flush's grace backstop remains, no lifecycle deadline");
} }
@ -135,7 +164,7 @@ fn an_error_after_an_ok_answer_ends_the_turn_and_not_the_session() {
fn a_brief_waiting_is_told_mid_turn_or_idle_once_and_never_after_the_flush() { fn a_brief_waiting_is_told_mid_turn_or_idle_once_and_never_after_the_flush() {
// Mid-turn: written at once. // Mid-turn: written at once.
let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1)]); let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1)]);
assert_eq!(actions[1], Action::WriteTells); assert_eq!(actions[1], tells());
assert_eq!(state.written, 2); assert_eq!(state.written, 2);
// The same poll answered again with nothing new is no second tell. // The same poll answered again with nothing new is no second tell.
assert_eq!(feed(state.clone(), Event::Inbox { pending: false }, 2).1, Action::None); assert_eq!(feed(state.clone(), Event::Inbox { pending: false }, 2).1, Action::None);
@ -143,9 +172,9 @@ fn a_brief_waiting_is_told_mid_turn_or_idle_once_and_never_after_the_flush() {
// Idle: the same. // Idle: the same.
let (_, actions) = idle_at(5); let (_, actions) = idle_at(5);
assert!(actions.iter().all(|action| *action == Action::None)); assert!(actions.iter().all(|action| *action == Action::None));
assert_eq!(feed(idle_at(5).0, Event::Inbox { pending: true }, 6).1, Action::WriteTells); assert_eq!(feed(idle_at(5).0, Event::Inbox { pending: true }, 6).1, tells());
// After the flush no line is written, a brief waiting or not. // After the flush no line is written, a brief waiting or not.
let (state, _) = run(vec![(req("a", CEILING_TOKENS), 0)]); let (state, _) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 0)]);
assert_eq!(feed(state.clone(), Event::Inbox { pending: true }, 1).1, Action::None); assert_eq!(feed(state.clone(), Event::Inbox { pending: true }, 1).1, Action::None);
assert_eq!(feed(state, Event::Tick, 2).1, Action::None); assert_eq!(feed(state, Event::Tick, 2).1, Action::None);
} }
@ -156,7 +185,7 @@ fn an_idle_session_asks_the_flush_where_the_cache_would_lapse_and_a_waiting_brie
let lapse = 10 + TTL_1H_S - TTL_MARGIN_S; let lapse = 10 + TTL_1H_S - TTL_MARGIN_S;
assert_eq!(next_deadline(&idle), Some(t(lapse))); assert_eq!(next_deadline(&idle), Some(t(lapse)));
assert_eq!(feed(idle.clone(), Event::Tick, lapse - 1).1, Action::None); assert_eq!(feed(idle.clone(), Event::Tick, lapse - 1).1, Action::None);
assert_eq!(feed(idle, Event::Tick, lapse).1, Action::WriteFlush(Why::Idle)); assert_eq!(feed(idle, Event::Tick, lapse).1, flush(Why::Idle));
// A five-minute cache lapses sooner. // A five-minute cache lapses sooner.
let (short, _) = run(vec![(Event::Echo, 0), (req("a", 10), 1), (OK_SHORT, 10)]); let (short, _) = run(vec![(Event::Echo, 0), (req("a", 10), 1), (OK_SHORT, 10)]);
assert_eq!(next_deadline(&short), Some(t(10 + TTL_5M_S - TTL_MARGIN_S))); assert_eq!(next_deadline(&short), Some(t(10 + TTL_5M_S - TTL_MARGIN_S)));
@ -164,20 +193,8 @@ fn an_idle_session_asks_the_flush_where_the_cache_would_lapse_and_a_waiting_brie
let (waiting, _) = run_from(idle_at(10).0, vec![(Event::Inbox { pending: true }, 11)]); let (waiting, _) = run_from(idle_at(10).0, vec![(Event::Inbox { pending: true }, 11)]);
assert_eq!(feed(waiting, Event::Tick, lapse + 100).1, Action::None); assert_eq!(feed(waiting, Event::Tick, lapse + 100).1, Action::None);
// The idle flush closes after its own answer. // The idle flush closes after its own answer.
let (state, actions) = run_from(idle_at(10).0, vec![(Event::Tick, lapse), (Event::Echo, lapse + 1), (req("b", 11), lapse + 1), (OK, lapse + 2)]); let (_, actions) = run_from(idle_at(10).0, vec![(Event::Tick, lapse), (Event::Echo, lapse + 1), (req("b", 11), lapse + 1), (OK, lapse + 2)]);
assert_eq!(actions, vec![Action::WriteFlush(Why::Idle), Action::None, Action::None, Action::CloseStdin]); assert_eq!(actions, vec![flush(Why::Idle), Action::None, Action::None, Action::CloseStdin]);
assert!(state.idle_end_answered());
}
#[test]
fn an_idle_end_answered_by_an_error_is_not_an_answered_idle_end() {
let lapse = 10 + TTL_1H_S - TTL_MARGIN_S;
let (state, actions) = run_from(idle_at(10).0, vec![(Event::Tick, lapse), (Event::Echo, lapse + 1), (FAILED, lapse + 2)]);
assert_eq!(actions[2], Action::CloseStdin);
assert!(!state.idle_end_answered());
// A rotation or ceiling flush is never an idle end.
let (state, _) = run(vec![(req("a", CEILING_TOKENS), 0), (Event::Echo, 1), (Event::Echo, 2), (OK, 3)]);
assert!(!state.idle_end_answered());
} }
#[test] #[test]
@ -189,13 +206,13 @@ fn near_the_cap_the_session_is_asked_for_its_flush_idle_or_working_and_closes_af
let (state, _) = run_from(started(), vec![(Event::Echo, 0)]); let (state, _) = run_from(started(), vec![(Event::Echo, 0)]);
assert_eq!(next_deadline(&state), Some(t(rotate))); assert_eq!(next_deadline(&state), Some(t(rotate)));
assert_eq!(feed(state.clone(), Event::Tick, rotate - 1).1, Action::None); assert_eq!(feed(state.clone(), Event::Tick, rotate - 1).1, Action::None);
assert_eq!(feed(state, Event::Tick, rotate).1, Action::WriteFlush(Why::Rotation)); assert_eq!(feed(state, Event::Tick, rotate).1, flush(Why::Rotation));
// Idle, its cache still warm. // Idle, its cache still warm.
let (idle, _) = run_from(started(), vec![(Event::Echo, 0), (OK, rotate - 100)]); let (idle, _) = run_from(started(), vec![(Event::Echo, 0), (OK, rotate - 100)]);
assert_eq!(feed(idle.clone(), Event::Tick, rotate).1, Action::WriteFlush(Why::Rotation)); assert_eq!(feed(idle.clone(), Event::Tick, rotate).1, flush(Why::Rotation));
// And the close follows the answer, never a bare close. // And the close follows the answer, never a bare close.
let (_, actions) = run_from(idle, vec![(Event::Tick, rotate), (Event::Echo, rotate + 1), (OK, rotate + 2)]); let (_, actions) = run_from(idle, vec![(Event::Tick, rotate), (Event::Echo, rotate + 1), (OK, rotate + 2)]);
assert_eq!(actions, vec![Action::WriteFlush(Why::Rotation), Action::None, Action::CloseStdin]); assert_eq!(actions, vec![flush(Why::Rotation), Action::None, Action::CloseStdin]);
} }
#[test] #[test]
@ -212,8 +229,8 @@ fn a_flush_is_never_asked_of_a_session_whose_input_is_closed() {
} }
#[test] #[test]
fn the_grace_runs_from_the_flush_write_to_the_second_and_no_later_request_moves_it() { fn the_grace_runs_from_the_flush_ask_to_the_second_and_no_later_request_moves_it() {
let (state, _) = run(vec![(req("a", CEILING_TOKENS), 100), (req("b", CEILING_TOKENS + 1), 500)]); let (state, _) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 100), (req("b", CEILING_TOKENS + 1), 500)]);
assert_eq!(next_deadline(&state), Some(t(100 + GRACE))); assert_eq!(next_deadline(&state), Some(t(100 + GRACE)));
assert_eq!(feed(state.clone(), Event::Tick, 100 + GRACE - 1).1, Action::None); assert_eq!(feed(state.clone(), Event::Tick, 100 + GRACE - 1).1, Action::None);
assert_eq!(feed(state, Event::Tick, 100 + GRACE).1, Action::Signal(Stop::Interrupt)); assert_eq!(feed(state, Event::Tick, 100 + GRACE).1, Action::Signal(Stop::Interrupt));
@ -236,12 +253,12 @@ fn the_headroom_counts_the_requests_after_the_flushs_echo_from_the_prefix_of_the
#[test] #[test]
fn the_ladder_ends_before_the_cap_whenever_the_flush_was_written() { fn the_ladder_ends_before_the_cap_whenever_the_flush_was_written() {
let cap = 5_400_u64; let cap = 5_400_u64;
let ceiling = |second: i64| run_from(State::started(t(0), cap), vec![(req("a", CEILING_TOKENS), second)]); let ceiling = |second: i64| run_from(State::started(t(0), cap), vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), second)]);
let rotation = |second: i64| run_from(State::started(t(0), cap), vec![(Event::Tick, second)]); let rotation = |second: i64| run_from(State::started(t(0), cap), vec![(Event::Echo, 0), (Event::Tick, second)]);
// A flush written inside the last LADDER_S cannot finish its ladder: the unit's own cap is the backstop there. // A flush written inside the last LADDER_S cannot finish its ladder: the unit's own cap is the backstop there.
for second in (0..cap as i64 - LADDER_S).step_by(7) { for second in (0..cap as i64 - LADDER_S).step_by(7) {
for (state, actions) in [ceiling(second), rotation(second)] { for (state, actions) in [ceiling(second), rotation(second)] {
if !matches!(actions.last(), Some(Action::WriteFlush(_))) { if !matches!(actions.last(), Some(Action::Write { flush: Some(_), .. })) {
continue; continue;
} }
let mut state = state; let mut state = state;
@ -266,18 +283,19 @@ fn the_ladder_ends_before_the_cap_whenever_the_flush_was_written() {
fn a_flush_written_late_has_its_grace_clamped_so_the_ladder_still_ends_under_the_cap() { fn a_flush_written_late_has_its_grace_clamped_so_the_ladder_still_ends_under_the_cap() {
let cap = 5_400_i64; let cap = 5_400_i64;
let late = cap - 100; let late = cap - 100;
let (state, _) = run_from(State::started(t(0), cap as u64), vec![(req("a", CEILING_TOKENS), late)]); let (state, _) = run_from(State::started(t(0), cap as u64), vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), late)]);
assert_eq!(next_deadline(&state), Some(t(cap - LADDER_S)), "the interrupt is not the full grace after the write"); assert_eq!(next_deadline(&state), Some(t(cap - LADDER_S)), "the interrupt is not the full grace after the write");
} }
#[test] #[test]
fn the_flush_pass_never_signals_whatever_the_clock_does() { 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. // Two events apart by more than the grace: the write is its own pass, the signal comes in the next.
let (state, action) = feed(working(), req("a", CEILING_TOKENS), 0); let (state, _) = feed(working(), Event::Echo, 0);
assert_eq!(action, Action::WriteFlush(Why::Ceiling)); let (state, action) = feed(state, req("a", CEILING_TOKENS), 0);
assert_eq!(action, flush(Why::Ceiling));
assert_eq!(feed(state, Event::Tick, 10 * GRACE).1, Action::Signal(Stop::Interrupt)); 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. // A request that both crosses the ceiling and is far past the headroom writes the flush and nothing else.
assert_eq!(last(vec![(req("a", u64::MAX), 0)]), Action::WriteFlush(Why::Ceiling)); assert_eq!(last(vec![(Event::Echo, 0), (req("a", u64::MAX), 0)]), flush(Why::Ceiling));
} }
#[test] #[test]
@ -388,7 +406,7 @@ const SYMBOLS: [Sym; 8] = [Sym::Request, Sym::Ceiling, Sym::Echo, Sym::Ok, Sym::
/// and the single-writer invariants. /// and the single-writer invariants.
fn play(symbols: &[Sym]) { fn play(symbols: &[Sym]) {
let mut state = working(); let mut state = working();
let (mut tells, mut flushes, mut closes, mut echoes) = (0_u32, 0_u32, 0_u32, 0_u32); let (mut writes, mut tells, mut flushes, mut closes, mut echoes) = (0_u32, 0_u32, 0_u32, 0_u32, 0_u32);
let (mut now, mut unique) = (0_i64, 0_u32); let (mut now, mut unique) = (0_i64, 0_u32);
let mut flushed = false; let mut flushed = false;
let mut closed = false; let mut closed = false;
@ -414,14 +432,14 @@ fn play(symbols: &[Sym]) {
let (next, action) = feed(state, event, now); let (next, action) = feed(state, event, now);
state = next; state = next;
match action { match action {
Action::WriteTells => { Action::Write { tells: told, flush } => {
assert!(!flushed && !closed, "a tell after the flush or the close: {symbols:?}"); assert!(!flushed && !closed, "a write after the flush or the close: {symbols:?}");
tells += 1; writes += 1;
} tells += u32::from(told);
Action::WriteFlush(_) => { if flush.is_some() {
assert!(!flushed && !closed, "a second flush, or one after the close: {symbols:?}"); flushed = true;
flushed = true; flushes += 1;
flushes += 1; }
} }
Action::CloseStdin => { Action::CloseStdin => {
assert!(!closed, "stdin closed twice: {symbols:?}"); assert!(!closed, "stdin closed twice: {symbols:?}");
@ -431,7 +449,9 @@ fn play(symbols: &[Sym]) {
Action::Signal(_) => assert!(flushed, "a signal with no flush: {symbols:?}"), Action::Signal(_) => assert!(flushed, "a signal with no flush: {symbols:?}"),
Action::None | Action::Finish(_) => {} Action::None | Action::Finish(_) => {}
} }
assert_eq!(state.written, 1 + tells + flushes, "every write is counted once: {symbols:?}"); assert_eq!(state.written, 1 + writes, "every write is counted once: {symbols:?}");
assert!(state.written <= state.consumed + 1, "two writes outstanding: {symbols:?}");
assert!(tells <= writes && flushes <= writes);
assert_eq!(state.consumed, echoes, "every echo is counted once: {symbols:?}"); assert_eq!(state.consumed, echoes, "every echo is counted once: {symbols:?}");
assert!(closes <= 1 && flushes <= 1); assert!(closes <= 1 && flushes <= 1);
} }

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

@ -2,23 +2,22 @@
//! mind and told to the live session by the actuator's one stdin writer. //! mind and told to the live session by the actuator's one stdin writer.
//! //!
//! The mind owns what her Self is asked: an in-force follow-up with owner //! The mind owns what her Self is asked: an in-force follow-up with owner
//! `self-run` whose source is a Persona turn's run is a brief. A follow-up her //! `self-run` that her Persona turn admitted is a brief. Who admitted it is the
//! Self files for herself, or one whose source the mind does not hold, is not. //! receipt's provenance, which the mind records and a writer cannot choose; the
//! `derive` lists the briefs the session has not been told (the told set is the //! follow-up's `source` is the writer's word and decides nothing. A follow-up her
//! actuator's process memory, a cache; the mind is the truth). Her Self resolves //! Self files for herself is not a brief, whatever it cites. `derive` lists the
//! each brief she answers; `close` resolves the ones she left open at an idle //! briefs the session has not been told (the told set is the actuator's process
//! end. `backlog` lists, and never closes, the in-force briefs a closed run //! memory, a cache; the mind is the truth). Her Self resolves each brief she
//! claimed: the one-time sweep at deploy. //! answers and the actuator closes none. `backlog` lists, and never closes, the
//! in-force briefs a closed run claimed: the one-time sweep at deploy.
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use anyhow::{Result, anyhow}; use anyhow::Result;
use chrono::{DateTime, Utc}; use eureka_pipeline::PipelineKind;
use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, RunTurn}; use huginn_mind::{PipelineFacts, PipelineStatusSummary};
use huginn_mind::{PipelineAdmissionOutcome, PipelineFacts, PipelineStatusSummary};
use super::AGENT; use crate::persona;
use crate::launch::closure;
use crate::queue::{MindPort, any_of, headers, in_force, of_kind}; 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. /// The owner a follow-up names when it is a brief for her Self.
@ -36,12 +35,6 @@ pub struct Brief {
pub ordinal: u64, pub ordinal: u64,
} }
impl Brief {
fn follow_up(&self) -> PipelineRef {
PipelineRef { kind: PipelineKind::FollowUp, id: self.id.as_str().into() }
}
}
/// The ids of the follow-ups a closed run claims, with the run each was claimed by. /// 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>> { 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 runs = headers(mind, &any_of(of_kind(PipelineKind::Run), "root", &[&mind.instance().0]))?;
@ -55,29 +48,19 @@ fn claimed_by_closed(mind: &dyn MindPort) -> Result<HashMap<String, String>> {
Ok(claimed) Ok(claimed)
} }
/// Every in-force brief, oldest first: owner `self-run`, sourced from a run of /// Every in-force brief, oldest first: owner `self-run`, admitted by her Persona
/// the mind whose turn is a Persona turn. /// turn (the receipt's agent), whatever source the writer named.
fn open_briefs(mind: &dyn MindPort) -> Result<Vec<Brief>> { fn open_briefs(mind: &dyn MindPort) -> Result<Vec<Brief>> {
let mut candidates: Vec<(Brief, String)> = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))? let mut briefs: Vec<Brief> = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))?
.into_iter() .into_iter()
.filter_map(|summary| match summary.facts { .filter_map(|summary| match summary.facts {
PipelineFacts::FollowUp { repo, owner, item, source, .. } if owner.0 == OWNER => { PipelineFacts::FollowUp { repo, owner, item, .. } if owner.0 == OWNER && summary.admission.provenance.agent.0 == persona::AGENT => {
let campaign = summary.id.id.0.split(':').next().unwrap_or("").to_string(); 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 }, source.id.0)) Some(Brief { id: summary.id.id.0, campaign, repo: repo.0, item: item.0, ordinal: summary.admission.ordinal })
} }
_ => None, _ => None,
}) })
.collect(); .collect();
if candidates.is_empty() {
return Ok(Vec::new());
}
let persona: HashSet<String> = headers(mind, &any_of(of_kind(PipelineKind::Run), "root", &[&mind.instance().0]))?
.into_iter()
.filter(|run| matches!(run.facts, PipelineFacts::Run { turn: RunTurn::PersonaTurn, .. }))
.map(|run| run.id.id.0)
.collect();
candidates.retain(|(_, source)| persona.contains(source));
let mut briefs: Vec<Brief> = candidates.into_iter().map(|(brief, _)| brief).collect();
briefs.sort_by_key(|brief| brief.ordinal); briefs.sort_by_key(|brief| brief.ordinal);
Ok(briefs) Ok(briefs)
} }
@ -120,23 +103,5 @@ pub fn resolving_sentence(id: &str) -> String {
format!("When you answer this brief, resolve follow-up {id} Recorded with your answer as rationale; if you decline it, resolve it Withdrawn with why.") format!("When you answer this brief, resolve follow-up {id} Recorded with your answer as rationale; if you decline it, resolve it Withdrawn with why.")
} }
/// Resolves Recorded each of `briefs` that is still in force, in one admission:
/// a brief her Self resolved herself is left as it is. Returns how many.
pub fn close(mind: &dyn MindPort, now: DateTime<Utc>, session: &str, briefs: &[(Brief, String)], reason: &str) -> Result<usize> {
let standing: HashSet<String> = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))?.into_iter().map(|summary| summary.id.id.0).collect();
let mut documents: Vec<PipelineDocument> = Vec::new();
for (brief, rationale) in briefs.iter().filter(|(brief, _)| standing.contains(&brief.id)) {
documents.push(closure(mind, &brief.follow_up(), now, ResolutionOutcome::Recorded { reason: reason.into() }, rationale)?);
}
let count = documents.len();
if count == 0 {
return Ok(0);
}
match mind.admit(AGENT, session, documents)? {
PipelineAdmissionOutcome::Committed { .. } => Ok(count),
_ => Err(anyhow!("the briefs could not be closed")),
}
}
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;

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

@ -1,10 +1,12 @@
//! The inbox over a mind in memory: what is a brief, what is told, what the //! The inbox over a mind in memory: what is a brief, what is told and what the
//! backlog lists, and what the idle-end close resolves. //! backlog lists.
use eureka_pipeline::{PipelineDocument, PipelineKind as K, RunOperator, RunTurn}; use eureka_pipeline::{PipelineDocument, PipelineKind as K, RunOperator, RunTurn};
use super::*; use super::*;
use crate::testkit::{TestMind, at, close as closing, follow_up, id, recorded, reference, run, run_ref, spec}; use crate::persona::AGENT as PERSONA;
use crate::self_run::AGENT as SELF_RUN;
use crate::testkit::{TestMind, close as closing, follow_up, id, recorded, reference, run, run_ref, spec};
fn brief_doc(label: &str, owner: &str, source: &PipelineRef) -> PipelineDocument { fn brief_doc(label: &str, owner: &str, source: &PipelineRef) -> PipelineDocument {
let PipelineDocument::FollowUp(mut document) = follow_up(label, source) else { unreachable!() }; let PipelineDocument::FollowUp(mut document) = follow_up(label, source) else { unreachable!() };
@ -13,6 +15,12 @@ fn brief_doc(label: &str, owner: &str, source: &PipelineRef) -> PipelineDocument
PipelineDocument::FollowUp(document) PipelineDocument::FollowUp(document)
} }
/// The mind gains these documents under `agent`'s receipt.
fn admitted_by(mind: &TestMind, agent: &str, documents: Vec<PipelineDocument>) {
let outcome = mind.admit(agent, "session-a", documents).unwrap();
assert!(matches!(outcome, huginn_mind::PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
}
fn brief_ref(label: &str) -> PipelineRef { fn brief_ref(label: &str) -> PipelineRef {
reference(K::FollowUp, &id("follow_up", label)) reference(K::FollowUp, &id("follow_up", label))
} }
@ -32,27 +40,37 @@ fn ids(briefs: Vec<Brief>) -> Vec<String> {
} }
#[test] #[test]
fn a_brief_is_an_in_force_self_run_follow_up_sourced_from_a_persona_turn() { fn a_brief_is_an_in_force_self_run_follow_up_her_persona_turn_admitted_whatever_source_it_names() {
let mind = mind(); let mind = mind();
let persona = run_ref("persona"); let persona = run_ref("persona");
mind.committed(vec![ admitted_by(
brief_doc("from-persona", OWNER, &persona), &mind,
brief_doc("from-her-self", OWNER, &run_ref("selfrun")), PERSONA,
brief_doc("from-no-run", OWNER, &crate::testkit::spec_ref("x")), vec![
brief_doc("someone-elses", "Hands", &persona), brief_doc("from-persona", OWNER, &persona),
brief_doc("answered", OWNER, &persona), brief_doc("persona-names-a-spec", OWNER, &crate::testkit::spec_ref("x")),
]); brief_doc("someone-elses", "Hands", &persona),
brief_doc("answered", OWNER, &persona),
],
);
mind.committed(vec![closing(brief_ref("answered"), recorded())]); mind.committed(vec![closing(brief_ref("answered"), recorded())]);
assert_eq!(ids(derive(&mind, &HashSet::new()).unwrap()), vec![brief_ref("from-persona").id.0], "a follow-up her Self files for herself is never a brief"); // Her Self names the Persona turn's run as the source, and so does a hand: the receipt says who admitted them.
admitted_by(&mind, SELF_RUN, vec![brief_doc("self-cites-persona", OWNER, &persona)]);
mind.committed(vec![brief_doc("hand-cites-persona", OWNER, &persona)]);
assert_eq!(
ids(derive(&mind, &HashSet::new()).unwrap()),
vec![brief_ref("from-persona").id.0, brief_ref("persona-names-a-spec").id.0],
"the admitting agent decides, never the source the writer chose"
);
} }
#[test] #[test]
fn derive_leaves_out_what_the_session_was_told_and_orders_by_admission() { fn derive_leaves_out_what_the_session_was_told_and_orders_by_admission() {
let mind = mind(); let mind = mind();
let persona = run_ref("persona"); let persona = run_ref("persona");
mind.committed(vec![brief_doc("first", OWNER, &persona)]); admitted_by(&mind, PERSONA, vec![brief_doc("first", OWNER, &persona)]);
mind.committed(vec![brief_doc("second", OWNER, &persona)]); admitted_by(&mind, PERSONA, vec![brief_doc("second", OWNER, &persona)]);
mind.committed(vec![brief_doc("third", OWNER, &persona)]); admitted_by(&mind, PERSONA, vec![brief_doc("third", OWNER, &persona)]);
let told: HashSet<String> = [brief_ref("second").id.0].into(); let told: HashSet<String> = [brief_ref("second").id.0].into();
assert_eq!(ids(derive(&mind, &told).unwrap()), vec![brief_ref("first").id.0, brief_ref("third").id.0]); assert_eq!(ids(derive(&mind, &told).unwrap()), vec![brief_ref("first").id.0, brief_ref("third").id.0]);
} }
@ -60,7 +78,7 @@ fn derive_leaves_out_what_the_session_was_told_and_orders_by_admission() {
#[test] #[test]
fn a_brief_a_closed_run_claimed_is_still_a_brief_and_no_standing_rule_closes_it() { fn a_brief_a_closed_run_claimed_is_still_a_brief_and_no_standing_rule_closes_it() {
let mind = mind(); let mind = mind();
mind.committed(vec![brief_doc("old", OWNER, &run_ref("persona"))]); admitted_by(&mind, PERSONA, vec![brief_doc("old", OWNER, &run_ref("persona"))]);
mind.committed(vec![run("old-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("old")])]); mind.committed(vec![run("old-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("old")])]);
mind.committed(vec![closing(run_ref("old-self"), recorded())]); mind.committed(vec![closing(run_ref("old-self"), recorded())]);
assert_eq!(ids(derive(&mind, &HashSet::new()).unwrap()), vec![brief_ref("old").id.0], "derive does not filter by past claims"); assert_eq!(ids(derive(&mind, &HashSet::new()).unwrap()), vec![brief_ref("old").id.0], "derive does not filter by past claims");
@ -70,7 +88,8 @@ fn a_brief_a_closed_run_claimed_is_still_a_brief_and_no_standing_rule_closes_it(
fn the_backlog_lists_in_force_persona_briefs_claimed_by_closed_runs_and_admits_nothing() { fn the_backlog_lists_in_force_persona_briefs_claimed_by_closed_runs_and_admits_nothing() {
let mind = mind(); let mind = mind();
let persona = run_ref("persona"); let persona = run_ref("persona");
mind.committed(vec![brief_doc("stale", OWNER, &persona), brief_doc("fresh", OWNER, &persona), brief_doc("claimed-live", OWNER, &persona), brief_doc("hers", OWNER, &run_ref("selfrun"))]); admitted_by(&mind, PERSONA, vec![brief_doc("stale", OWNER, &persona), brief_doc("fresh", OWNER, &persona), brief_doc("claimed-live", OWNER, &persona)]);
admitted_by(&mind, SELF_RUN, vec![brief_doc("hers", OWNER, &run_ref("selfrun"))]);
mind.committed(vec![run("closed-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("stale"), brief_ref("hers")])]); mind.committed(vec![run("closed-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("stale"), brief_ref("hers")])]);
mind.committed(vec![closing(run_ref("closed-self"), recorded())]); mind.committed(vec![closing(run_ref("closed-self"), recorded())]);
mind.committed(vec![run("live-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed-live")])]); mind.committed(vec![run("live-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed-live")])]);
@ -89,24 +108,3 @@ fn a_tell_names_each_brief_the_plan_and_how_she_resolves_it() {
assert!(text.ends_with("Plan: 5h at 40%."), "{text}"); assert!(text.ends_with("Plan: 5h at 40%."), "{text}");
assert!(!render(&[], "").contains("brief"), "no brief, no plan, nothing"); assert!(!render(&[], "").contains("brief"), "no brief, no plan, nothing");
} }
#[test]
fn closing_resolves_only_the_briefs_still_in_force_in_one_admission() {
let mind = mind();
mind.committed(vec![brief_doc("a", OWNER, &run_ref("persona")), brief_doc("b", OWNER, &run_ref("persona"))]);
mind.committed(vec![closing(brief_ref("b"), recorded())]);
let entry = |label: &str, n: u64| Brief { id: brief_ref(label).id.0, campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: format!("Item {n}."), ordinal: n };
let briefs = [(entry("a", 1), "The answer.".to_string()), (entry("b", 2), "Ignored.".to_string())];
let now = at(0);
assert_eq!(close(&mind, now, "selfrun-a", &briefs, "answered").unwrap(), 1);
let rationales: Vec<String> = mind
.ids(K::Resolution)
.iter()
.filter_map(|resolution| match mind.view(K::Resolution, resolution)?.document {
PipelineDocument::Resolution(document) if document.subject == brief_ref("a") => Some(document.rationale.0),
_ => None,
})
.collect();
assert_eq!(rationales, vec!["The answer.".to_string()]);
assert_eq!(close(&mind, now, "selfrun-a", &briefs[..1], "answered").unwrap(), 0, "already closed");
}

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

@ -1495,6 +1495,18 @@ fn live_mind() -> Rc<TestMind> {
mind mind
} }
/// The mind gains these documents as her Persona turn's admission: the provenance that makes a self-run follow-up a brief.
fn persona_commit(mind: &TestMind, documents: Vec<PipelineDocument>) {
let outcome = mind.admit(crate::persona::AGENT, "persona-a", documents).unwrap();
assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
}
/// Her Persona admits these briefs, `seconds` into the run.
fn briefed(mind: &Rc<TestMind>, seconds: i64, documents: Vec<PipelineDocument>) -> Step {
let mind = mind.clone();
Step::Do(seconds, Rc::new(move || persona_commit(&mind, documents.clone())))
}
/// A brief from her Persona: a follow-up with owner `self-run`, sourced from the Persona turn's run. /// A brief from her Persona: a follow-up with owner `self-run`, sourced from the Persona turn's run.
fn brief(label: &str, item: &str) -> PipelineDocument { fn brief(label: &str, item: &str) -> PipelineDocument {
let PipelineDocument::FollowUp(mut document) = follow_up(label, &run_ref("persona")) else { unreachable!() }; let PipelineDocument::FollowUp(mut document) = follow_up(label, &run_ref("persona")) else { unreachable!() };
@ -1556,7 +1568,7 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_
Out(echo()), Out(echo()),
Out(at("m1", 10)), Out(at("m1", 10)),
// The brief arrives while the first turn runs: the next poll writes it at once, mid-turn. // The brief arrives while the first turn runs: the next poll writes it at once, mid-turn.
admits(&mind, 31, vec![brief("b1", "Survey the noise functions.")]), briefed(&mind, 31, vec![brief("b1", "Survey the noise functions.")]),
Out(echo()), Out(echo()),
// Later polls do not tell it again. // Later polls do not tell it again.
Idle(31), Idle(31),
@ -1569,7 +1581,7 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_
assert!(told[0].contains(&brief_ref("b1").id.0) && told[0].contains("Survey the noise functions."), "{}", told[0]); assert!(told[0].contains(&brief_ref("b1").id.0) && told[0].contains("Survey the noise functions."), "{}", told[0]);
assert!(told[0].contains("resolve follow-up") && told[0].contains("Recorded"), "she is told how to close it: {}", told[0]); assert!(told[0].contains("resolve follow-up") && told[0].contains("Recorded"), "she is told how to close it: {}", told[0]);
assert_eq!(ran.written().len(), 2, "the start message and one tell: {:?}", ran.written()); assert_eq!(ran.written().len(), 2, "the start message and one tell: {:?}", ran.written());
assert!(ran.out.contains("decision WriteTells prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out); assert!(ran.out.contains("decision Write { tells: true, flush: None } prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out);
assert_eq!(ran.count(&Event::InputClosed), 0); assert_eq!(ran.count(&Event::InputClosed), 0);
// A result is not an answer: nothing closes the brief by its timing. // A result is not an answer: nothing closes the brief by its timing.
assert!(resolutions_of(&mind, &brief_ref("b1")).is_empty()); assert!(resolutions_of(&mind, &brief_ref("b1")).is_empty());
@ -1578,7 +1590,7 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_
#[test] #[test]
fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_until_its_echo() { fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_until_its_echo() {
let mind = live_mind(); let mind = live_mind();
let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), admits(&mind, 31, vec![brief("b2", "Wire the shader uniforms.")]), Out(echo()), Out(ok_result())]; let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), briefed(&mind, 31, vec![brief("b2", "Wire the shader uniforms.")]), Out(echo()), Out(ok_result())];
let ran = mind_run(&mind, steps); let ran = mind_run(&mind, steps);
assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written()); assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written());
assert!(ran.out.contains("idle_s 31 inbox true written 2 consumed 1\n"), "idle when it was written: {}", ran.out); assert!(ran.out.contains("idle_s 31 inbox true written 2 consumed 1\n"), "idle when it was written: {}", ran.out);
@ -1589,7 +1601,7 @@ fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_
#[test] #[test]
fn a_brief_open_at_start_is_in_the_start_message_with_how_to_resolve_it_and_is_not_told_again() { fn a_brief_open_at_start_is_in_the_start_message_with_how_to_resolve_it_and_is_not_told_again() {
let mind = live_mind(); let mind = live_mind();
mind.committed(vec![brief("p1", "Still open.")]); persona_commit(&mind, vec![brief("p1", "Still open.")]);
let ran = mind_run(&mind, vec![Out(echo()), Idle(31), Idle(31)]); let ran = mind_run(&mind, vec![Out(echo()), Idle(31), Idle(31)]);
let start = &ran.written()[0]; let start = &ran.written()[0];
assert!(start.contains("Also open for you") && start.contains(&brief_ref("p1").id.0) && start.contains("Still open."), "{start}"); assert!(start.contains("Also open for you") && start.contains(&brief_ref("p1").id.0) && start.contains("Still open."), "{start}");
@ -1602,24 +1614,23 @@ fn a_follow_up_her_self_filed_for_herself_is_never_told() {
let mind = live_mind(); let mind = live_mind();
let PipelineDocument::FollowUp(mut hers) = follow_up("hers", &run_ref("selfrun-a")) else { unreachable!() }; let PipelineDocument::FollowUp(mut hers) = follow_up("hers", &run_ref("selfrun-a")) else { unreachable!() };
hers.owner = inbox::OWNER.into(); hers.owner = inbox::OWNER.into();
mind.committed(vec![PipelineDocument::FollowUp(hers)]); // Naming the Persona turn's run as the source does not make it hers to be told: the receipt records who admitted it.
let PipelineDocument::FollowUp(mut citing) = follow_up("citing", &run_ref("persona")) else { unreachable!() };
citing.owner = inbox::OWNER.into();
mind.committed(vec![PipelineDocument::FollowUp(hers), PipelineDocument::FollowUp(citing)]);
let ran = mind_run(&mind, vec![Out(echo()), Idle(31)]); let ran = mind_run(&mind, vec![Out(echo()), Idle(31)]);
assert_eq!(ran.written().len(), 1, "{:?}", ran.written()); assert_eq!(ran.written().len(), 1, "{:?}", ran.written());
assert!(!ran.written()[0].contains("Also open"), "{}", ran.written()[0]); assert!(!ran.written()[0].contains("Also open"), "{}", ran.written()[0]);
} }
#[test] #[test]
fn an_idle_end_closes_the_briefs_she_left_open_saying_so_and_leaves_the_ones_she_resolved() { fn an_idle_end_closes_no_brief_the_ones_she_left_open_stay_open_and_her_resolutions_stand() {
let mind = live_mind(); let mind = live_mind();
mind.committed(vec![brief("p1", "Left open."), brief("p2", "Answered by her.")]); persona_commit(&mind, vec![brief("p1", "Left open."), brief("p2", "Answered by her.")]);
let steps = vec![Out(echo()), Out(at("m1", 10)), resolves(&mind, "p2"), Out(ok_result()), Idle(3539), Idle(1), Out(echo()), Out(ok_result())]; let steps = vec![Out(echo()), Out(at("m1", 10)), resolves(&mind, "p2"), Out(ok_result()), Idle(3539), Idle(1), Out(echo()), Out(ok_result())];
let ran = mind_run(&mind, steps); let ran = mind_run(&mind, steps);
assert_eq!(ran.count(&Event::InputClosed), 1, "{:?}", ran.events); assert_eq!(ran.count(&Event::InputClosed), 1, "{:?}", ran.events);
let left = resolutions_of(&mind, &brief_ref("p1")); assert!(resolutions_of(&mind, &brief_ref("p1")).is_empty(), "the actuator resolved nothing she left open");
assert_eq!(left.len(), 1);
let ResolutionOutcome::Recorded { reason } = &left[0].outcome else { panic!("{:?}", left[0].outcome) };
assert!(reason.0.contains("selfrun-a") && reason.0.contains("went idle") && reason.0.contains("without her resolving it"), "{}", reason.0);
assert_eq!(left[0].rationale.0, "The answer.", "the last ok result's text");
assert_eq!(resolutions_of(&mind, &brief_ref("p2")).len(), 1, "her own resolution stands alone"); assert_eq!(resolutions_of(&mind, &brief_ref("p2")).len(), 1, "her own resolution stands alone");
} }
@ -1635,7 +1646,7 @@ fn a_rotation_a_ceiling_an_error_or_a_crash_leaves_the_told_briefs_open_for_the_
]; ];
for (name, steps) in ends { for (name, steps) in ends {
let mind = live_mind(); let mind = live_mind();
mind.committed(vec![brief("p1", "Left open.")]); persona_commit(&mind, vec![brief("p1", "Left open.")]);
let ran = mind_run(&mind, steps); let ran = mind_run(&mind, steps);
assert!(ran.outcome.is_ok(), "{name}"); assert!(ran.outcome.is_ok(), "{name}");
assert!(resolutions_of(&mind, &brief_ref("p1")).is_empty(), "{name}: the brief stays open"); assert!(resolutions_of(&mind, &brief_ref("p1")).is_empty(), "{name}: the brief stays open");
@ -1645,7 +1656,7 @@ fn a_rotation_a_ceiling_an_error_or_a_crash_leaves_the_told_briefs_open_for_the_
#[test] #[test]
fn a_brief_admitted_after_the_flush_is_written_stays_in_force_and_untold() { fn a_brief_admitted_after_the_flush_is_written_stays_in_force_and_untold() {
let mind = live_mind(); let mind = live_mind();
let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), admits(&mind, 31, vec![brief("late", "Too late.")]), Idle(31), Out(echo()), Out(ok_result())]; let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), briefed(&mind, 31, vec![brief("late", "Too late.")]), Idle(31), Out(echo()), Out(ok_result())];
let ran = mind_run(&mind, steps); let ran = mind_run(&mind, steps);
assert!(tells(&ran).is_empty(), "{:?}", ran.written()); assert!(tells(&ran).is_empty(), "{:?}", ran.written());
assert_eq!(ran.written().len(), 2, "the start and the flush; no line after the flush"); assert_eq!(ran.written().len(), 2, "the start and the flush; no line after the flush");
@ -1660,7 +1671,7 @@ fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_
let flush_at = hour.events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains(FLUSH_MESSAGE))).unwrap(); 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(); let closed_at = hour.events.iter().position(|event| *event == Event::InputClosed).unwrap();
assert!(flush_at < closed_at && hour.count(&Event::InputClosed) == 1, "{:?}", hour.events); assert!(flush_at < closed_at && hour.count(&Event::InputClosed) == 1, "{:?}", hour.events);
assert!(hour.out.contains("decision WriteFlush(Idle) prefix 10 ttl_s 3600 idle_s 3540 inbox false written 2 consumed 1\n"), "{}", hour.out); assert!(hour.out.contains("decision Write { tells: false, flush: Some(Idle) } prefix 10 ttl_s 3600 idle_s 3540 inbox false written 2 consumed 1\n"), "{}", hour.out);
assert!(hour.written().iter().any(|text| text.contains("nothing is waiting and its cache would lapse") && text.contains("Resolve each brief you answered")), "{:?}", hour.written()); assert!(hour.written().iter().any(|text| text.contains("nothing is waiting and its cache would lapse") && text.contains("Resolve each brief you answered")), "{:?}", hour.written());
// Five minutes' cache: the flush at 240 s. // Five minutes' cache: the flush at 240 s.
let short = go(vec![Out(echo()), Out(five_minute_request("m1", 10)), Out(ok_result()), Idle(239), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]); let short = go(vec![Out(echo()), Out(five_minute_request("m1", 10)), Out(ok_result()), Idle(239), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]);
@ -1670,7 +1681,7 @@ fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_
assert_eq!(flushes(&early), 0); assert_eq!(flushes(&early), 0);
// A brief waiting at the cache's end is told, not flushed: the session stays open. // A brief waiting at the cache's end is told, not flushed: the session stays open.
let mind = live_mind(); let mind = live_mind();
let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), admits(&mind, 3539, vec![brief("b5", "One more.")]), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]; let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), briefed(&mind, 3539, vec![brief("b5", "One more.")]), Idle(1), Out(echo()), Out(ok_result()), Idle(1)];
let ran = mind_run(&mind, steps); let ran = mind_run(&mind, steps);
assert_eq!((flushes(&ran), tells(&ran).len(), ran.count(&Event::InputClosed)), (0, 1, 0), "{:?} assert_eq!((flushes(&ran), tells(&ran).len(), ran.count(&Event::InputClosed)), (0, 1, 0), "{:?}
{}", ran.written(), ran.out); {}", ran.written(), ran.out);
@ -1679,7 +1690,7 @@ fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_
#[test] #[test]
fn a_tell_that_is_not_yet_echoed_keeps_the_session_busy_so_the_cache_end_does_not_flush_over_it() { fn a_tell_that_is_not_yet_echoed_keeps_the_session_busy_so_the_cache_end_does_not_flush_over_it() {
let mind = live_mind(); let mind = live_mind();
let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), admits(&mind, 31, vec![brief("b6", "Pending.")]), Idle(3600)]; let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), briefed(&mind, 31, vec![brief("b6", "Pending.")]), Idle(3600)];
let ran = mind_run(&mind, steps); let ran = mind_run(&mind, steps);
assert_eq!((tells(&ran).len(), flushes(&ran)), (1, 0), "{:?}", ran.written()); assert_eq!((tells(&ran).len(), flushes(&ran)), (1, 0), "{:?}", ran.written());
} }
@ -1703,8 +1714,8 @@ fn near_the_cap_the_session_is_asked_for_its_flush_idle_or_working_and_closes_at
#[test] #[test]
fn no_stdin_line_is_written_after_the_flush_and_nothing_but_the_writer_writes() { fn no_stdin_line_is_written_after_the_flush_and_nothing_but_the_writer_writes() {
let mind = live_mind(); let mind = live_mind();
mind.committed(vec![brief("b7", "Early.")]); persona_commit(&mind, vec![brief("b7", "Early.")]);
let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), admits(&mind, 31, vec![brief("b8", "After the flush.")]), Idle(31), Idle(31), Out(echo()), Out(at("m2", 11)), Out(ok_result())]; let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), briefed(&mind, 31, vec![brief("b8", "After the flush.")]), Idle(31), Idle(31), Out(echo()), Out(at("m2", 11)), Out(ok_result())];
let ran = mind_run(&mind, steps); let ran = mind_run(&mind, steps);
let written = ran.written(); let written = ran.written();
assert_eq!(written.len(), 2, "{written:?}"); assert_eq!(written.len(), 2, "{written:?}");
@ -1718,6 +1729,10 @@ mod live_session {
use crate::self_run::decide::{Action as Do, Event as Fact, Why}; use crate::self_run::decide::{Action as Do, Event as Fact, Why};
use super::*; use super::*;
fn flush(why: Why) -> Do {
Do::Write { tells: false, flush: Some(why) }
}
/// One recorded line as the loop would hear it: a label for the stream, the decision's action. /// One recorded line as the loop would hear it: a label for the stream, the decision's action.
fn replay(jsonl: &str) -> Vec<(String, Do)> { fn replay(jsonl: &str) -> Vec<(String, Do)> {
let mut state = State::started(crate::testkit::at(0), 100_000); let mut state = State::started(crate::testkit::at(0), 100_000);
@ -1781,7 +1796,7 @@ mod live_session {
fn the_flush_folded_into_a_running_turn_is_answered_at_its_one_result() { fn the_flush_folded_into_a_running_turn_is_answered_at_its_one_result() {
let trace = replay(CEILING_FOLDED); let trace = replay(CEILING_FOLDED);
assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "other", "other", "other", "echo", "request:10", "result:ok"]); assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "other", "other", "other", "echo", "request:10", "result:ok"]);
assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (8, Do::CloseStdin)], "never a grace interrupt"); assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (8, Do::CloseStdin)], "never a grace interrupt");
} }
#[test] #[test]
@ -1790,7 +1805,7 @@ mod live_session {
let trace = replay(stream); let trace = replay(stream);
assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "result:ok", "other", "echo", "request:10", "result:ok"]); assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "result:ok", "other", "echo", "request:10", "result:ok"]);
// The first result ends the turn the flush was queued behind: it closes nothing. // The first result ends the turn the flush was queued behind: it closes nothing.
assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (7, Do::CloseStdin)], "never a grace interrupt"); assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)], "never a grace interrupt");
} }
} }
@ -1798,7 +1813,7 @@ mod live_session {
fn an_api_error_before_any_answer_closes_the_session_at_once_with_no_flush() { fn an_api_error_before_any_answer_closes_the_session_at_once_with_no_flush() {
for stream in API_ERROR { for stream in API_ERROR {
let trace = replay(stream); let trace = replay(stream);
assert_eq!(kinds(&trace), ["other", "echo", "request:0", "result:error"]); assert_eq!(kinds(&trace), ["other", "echo", "other", "result:error"]);
assert_eq!(acted(&trace), vec![(3, Do::CloseStdin)]); assert_eq!(acted(&trace), vec![(3, Do::CloseStdin)]);
} }
} }
@ -1807,8 +1822,8 @@ mod live_session {
fn a_context_over_the_blocking_limit_is_a_blocked_result_that_closes_the_session() { fn a_context_over_the_blocking_limit_is_a_blocked_result_that_closes_the_session() {
for stream in BLOCKING { for stream in BLOCKING {
let trace = replay(stream); let trace = replay(stream);
assert_eq!(kinds(&trace), ["other", "echo", "request:990000", "result:ok", "other", "echo", "request:0", "result:blocked"]); assert_eq!(kinds(&trace), ["other", "echo", "request:990000", "result:ok", "other", "echo", "other", "result:blocked"]);
assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (7, Do::CloseStdin)]); assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)]);
} }
} }
@ -1817,7 +1832,7 @@ mod live_session {
for stream in COMPACTION_OFF { for stream in COMPACTION_OFF {
let trace = replay(stream); let trace = replay(stream);
assert_eq!(kinds(&trace), ["other", "echo", "request:360000", "result:ok", "other", "echo", "request:10", "result:ok"]); assert_eq!(kinds(&trace), ["other", "echo", "request:360000", "result:ok", "other", "echo", "request:10", "result:ok"]);
assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (7, Do::CloseStdin)]); assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)]);
} }
} }
} }
@ -1826,7 +1841,7 @@ mod live_session {
fn a_brief_the_run_claims_is_in_the_task_once_and_not_again_under_also_open() { fn a_brief_the_run_claims_is_in_the_task_once_and_not_again_under_also_open() {
let mind = TestMind::seeded(); let mind = TestMind::seeded();
mind.committed(vec![spec("x"), run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]); mind.committed(vec![spec("x"), run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]);
mind.committed(vec![brief("claimed", "Do the claimed thing.")]); persona_commit(&mind, vec![brief("claimed", "Do the claimed thing.")]);
mind.committed(vec![run("selfrun-a", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed")])]); mind.committed(vec![run("selfrun-a", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed")])]);
let ran = carry_with(&Seen::of(&mind), vec![Out(echo())], Some(0), false); let ran = carry_with(&Seen::of(&mind), vec![Out(echo())], Some(0), false);
let start = &ran.written()[0]; let start = &ran.written()[0];

15
docs/eureka-body-map.md
View file

@ -807,11 +807,16 @@ in-force `operated_by Mind` Self run of its instance, which admission keeps uniq
- Not its decision: continuation (fresh only until `self-run-measure`), the brake (the unit's - Not its decision: continuation (fresh only until `self-run-measure`), the brake (the unit's
gate), the stop (the hold), the budget (the run's `budget_usd`, from the dial at launch), gate), the stop (the hold), the budget (the run's `budget_usd`, from the dial at launch),
the environment, token and permissions (the unit and managed settings, gamecult-ops). the environment, token and permissions (the unit and managed settings, gamecult-ops).
- Delivery is counted: `--replay-user-messages` makes the CLI echo each stdin line right before - Delivery is counted, with one write outstanding (ruling `self-session-one-outstanding-write`):
the request that carries it, folded into the running turn or queued as the next, so the echo is `--replay-user-messages` makes the CLI echo each stdin line right before the request that
the consumption point and idle is `lines written == lines echoed` with no turn open. The flush carries it, folded into the running turn or queued as the next, but it joins lines queued
is answered at the first result after its echo, folded or queued alike; the grace interrupt behind one request into a single echo. So `drive` writes the next line only after the previous
(earlier of 15 min after the write and the cap less the ladder) is the backstop. Compaction is line's echo, coalescing what waits (the briefs and the flush ask) into that one line, and idle
is no write outstanding with no turn open. The flush is answered at the first result after its
echo; the grace interrupt (earlier of 15 min after the ask and the cap less the ladder) is the
backstop. A brief is an in-force `self-run` follow-up her Persona turn admitted (the receipt's
agent, not the source the writer named); the actuator closes none, and a brief left open is
told to the next session. Compaction is
off in the CLI (`DISABLE_COMPACT=1` in the child environment). off in the CLI (`DISABLE_COMPACT=1` in the child environment).
- With `CLAUDE_CODE_SUBPROCESS_ENV_SCRUB` set the CLI forces the permission mode to default: - With `CLAUDE_CODE_SUBPROCESS_ENV_SCRUB` set the CLI forces the permission mode to default:
`--permission-mode bypassPermissions` has no effect there and only the seed's allow list decides `--permission-mode bypassPermissions` has no effect there and only the seed's allow list decides