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: 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`).
//! 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
//! from a Persona turn is a brief. `drive` is the one writer into her session: the
//! start message, every tell and the flush are stdin lines, a tell written whenever
//! a brief waits and the flush has not been written, mid-turn or idle. The CLI
//! echoes each line it takes into a request (`--replay-user-messages`); the echo is
//! the consumption point, folded or queued, and idle is derived from the lines
//! written and the lines echoed. Her Self resolves each brief she answers; the
//! actuator closes only what she left open at an idle end. Compaction is off in
//! her mind (the `inbox` module): an open follow-up with owner `self-run` admitted
//! by her Persona turn is a brief. `drive` is the one writer into her
//! session, and keeps one write outstanding (ruling
//! `eureka-body:ruling:self-session-one-outstanding-write`): the start message,
//! then one line at a time holding the briefs that wait and the flush ask, each
//! sent only after the previous line's echo, mid-turn or idle. The CLI echoes each
//! line it takes into a request (`--replay-user-messages`) and joins lines queued
//! 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
//! 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.
@ -79,6 +82,12 @@ use inbox::Brief;
/// The measured Self ceiling: at this prefix the flush is asked for.
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
/// the actuator interrupts. Relative to the flush, never an absolute limit.
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.
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");
/// 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.
pub const INBOX_POLL_S: i64 = 30;
/// 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()
}
/// 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.
#[derive(Clone, Debug, PartialEq, Eq)]
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) => {
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 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);
@ -414,21 +433,13 @@ fn her_run(mind: &dyn MindPort) -> Result<Option<(PipelineRef, Vec<PipelineRef>)
struct Heard {
/// The last result heard, ok or not: its head line and its text cut to `RATIONALE_RESULT_LIMIT`.
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 {
fn hear(&mut self, line: &str) {
let Ok(Value::Object(event)) = serde_json::from_str::<Value>(line) else { return };
if event.get("type").and_then(Value::as_str) == Some("result") {
let text = 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));
self.concluded = Some((result_head(&event), cut_text(result_text(&event), RATIONALE_RESULT_LIMIT)));
}
}
}
@ -521,10 +532,6 @@ Also open for you:");
}
let mut heard = Heard::default();
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);
if closed.is_err() {
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))
}
/// 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.
#[allow(clippy::too_many_arguments, reason = "the loop's ports and the run's facts, passed on whole")]
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 (next, action) = decide(state, heard_event, now);
state = next;
heard.idle_end = state.idle_end_answered();
if state.requests != counted {
writeln!(out, "request {} prefix {}", state.requests, state.prefix)?;
}
match &action {
Action::None => {}
Action::WriteFlush(why) => child.write_line(&user_line(&flush_text(*why))),
Action::WriteTells => {
child.write_line(&user_line(&inbox::render(&waiting, &plan)));
told.append(&mut waiting);
Action::Write { tells, flush } => {
let mut parts = Vec::new();
if *tells {
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 => {
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
//! 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
//! start message, a tell, the flush) increments `written`; every replay echo of
//! a line 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, so one rule
//! serves both. Idle is derived: no turn open and `written == consumed`.
//! Delivery is counted, not guessed, and at most one write is outstanding (ruling
//! `eureka-body:ruling:self-session-one-outstanding-write`): the CLI joins the
//! lines queued behind a request into one echo, so a second line is never
//! written before the first one's echo. Every stdin line the actuator writes (the
//! 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:
//! (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
//! result after the flush's echo closes stdin, folded or queued alike; (3) a
//! brief waiting and no flush written: tell it, idle or mid-turn; (4) the cap
//! less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (5) idle
//! with no brief waiting, at the last result's cache lifetime less a margin: ask
//! the flush (idle). The ceiling asks the same flush from a request. Compaction
//! is off in the CLI (`DISABLE_COMPACT=1`), so the flush is the only way a
//! session ends short of its cache.
//! result after the flush's echo closes stdin, folded or queued alike; (3) the
//! cap less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (4)
//! idle with no brief waiting, at the last result's cache lifetime less a margin:
//! ask the flush (idle); (5) with nothing outstanding, write what waits: the
//! briefs and the flush ask if there is one, as one line. The ceiling asks the
//! same flush from a request. Compaction is off in the CLI (`DISABLE_COMPACT=1`),
//! 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
//! 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
//! finishes before the unit's cap (ruling
//! `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)]
pub(super) enum Action {
None,
WriteFlush(Why),
/// Write the waiting briefs as one user line.
WriteTells,
/// The one write: the waiting briefs and the flush ask, coalesced into one user line.
Write { tells: bool, flush: Option<Why> },
CloseStdin,
/// To the child's whole process group.
Signal(Stop),
@ -85,11 +88,11 @@ pub(super) enum Action {
#[derive(Clone, Debug, PartialEq, Eq)]
struct Flush {
/// The flush's line number: its echo is the `seq`-th.
seq: u32,
/// The prefix at which the flush was written: the headroom's base.
/// The flush's line number once written (its echo is the `seq`-th); none while it waits for the outstanding write's echo.
seq: Option<u32>,
/// The prefix at which the flush was asked: the headroom's base.
prefix: u64,
written: DateTime<Utc>,
asked: DateTime<Utc>,
/// Its echo has come: the CLI took it into a request.
consumed: bool,
/// A request after its echo reached the headroom.
@ -116,8 +119,6 @@ pub(super) struct State {
cap: Option<(DateTime<Utc>, i64)>,
stdin_closed: bool,
flush: Option<Flush>,
/// An ok result answered the flush.
flush_answered_ok: bool,
/// The last signal attempted and when.
stopping: Option<(Stop, DateTime<Utc>)>,
/// The strongest signal delivered.
@ -125,8 +126,6 @@ pub(super) struct State {
exited: Option<(Option<i32>, DateTime<Utc>)>,
eof: bool,
finished: bool,
/// The clock at the last decision: where a due action says "now".
at: Option<DateTime<Utc>>,
}
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() }
}
/// A line written whose echo has not come.
fn outstanding(&self) -> bool {
self.written > self.consumed
}
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.
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()))
}
/// 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 {
@ -166,28 +165,21 @@ fn cause(state: &State) -> Cause {
/// 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.
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 {
Some((started, cap)) => grace.min(started + Duration::seconds(cap.saturating_sub(LADDER_S))),
None => grace,
}
}
/// What the session does next on its own, and when.
enum Due {
Tells,
Flush(Why),
}
fn due(state: &State) -> Option<(DateTime<Utc>, Due)> {
/// When the flush is next asked on its own, and why: the cap's margin, or the
/// cache's end while idle with no brief waiting.
fn due(state: &State) -> Option<(DateTime<Utc>, Why)> {
if state.stdin_closed || state.exited.is_some() || state.flush.is_some() {
return None;
}
if state.inbox {
return state.at.map(|at| (at, Due::Tells));
}
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)));
let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Why::Rotation));
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));
[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;
/// 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 {
if state.flush.is_some() || state.stdin_closed || state.exited.is_some() {
return false;
}
state.written += 1;
state.flush = Some(Flush { seq: state.written, prefix: state.prefix, written: now, consumed: false, exhausted: false, why });
state.flush = Some(Flush { seq: None, prefix: state.prefix, asked: now, consumed: false, exhausted: false, why });
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 {
match due(state) {
Some((at, Due::Tells)) if at <= now => {
state.inbox = false;
state.written += 1;
Action::WriteTells
}
Some((at, Due::Flush(why))) if at <= now && ask_flush(state, now, why) => Action::WriteFlush(why),
_ => Action::None,
if let Some((at, why)) = due(state)
&& at <= now
{
ask_flush(state, now, why);
}
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) {
if state.finished {
return (state, Action::None);
}
state.at = Some(now);
let mut action = Action::None;
match event {
Event::Request { id, prefix } => {
@ -249,8 +255,8 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (Sta
{
flush.exhausted = true;
}
if prefix >= CEILING_TOKENS && ask_flush(&mut state, now, Why::Ceiling) {
action = Action::WriteFlush(Why::Ceiling);
if prefix >= CEILING_TOKENS {
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;
let consumed = state.consumed;
if let Some(flush) = state.flush.as_mut()
&& consumed >= flush.seq
&& flush.seq.is_some_and(|seq| consumed >= seq)
{
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;
state.answered_once |= ok;
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() {
state.stdin_closed = true;
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()
}
/// 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 {
Event::Signalled { how, delivered: true }
}
@ -68,13 +77,33 @@ fn idle_at(at: i64) -> (State, Vec<Action>) {
#[test]
fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or_for_a_repeated_request() {
assert_eq!(last(vec![(req("a", CEILING_TOKENS - 1), 0)]), Action::None);
let (state, actions) = run(vec![(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!(last(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS - 1), 0)]), Action::None);
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::None, flush(Why::Ceiling), Action::None, Action::None]);
assert_eq!((state.written, state.requests), (2, 2));
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]
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![]);
@ -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));
// 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)]);
assert_eq!(actions[2], Action::WriteTells);
assert_eq!(actions[2], tells());
assert_eq!(state.idle_s(t(9)), None);
// 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)]);
@ -96,7 +125,7 @@ fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every
#[test]
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)]);
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");
}
@ -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() {
// Mid-turn: written at once.
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);
// 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);
@ -143,9 +172,9 @@ fn a_brief_waiting_is_told_mid_turn_or_idle_once_and_never_after_the_flush() {
// Idle: the same.
let (_, actions) = idle_at(5);
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.
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, 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;
assert_eq!(next_deadline(&idle), Some(t(lapse)));
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.
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)));
@ -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)]);
assert_eq!(feed(waiting, Event::Tick, lapse + 100).1, Action::None);
// 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)]);
assert_eq!(actions, vec![Action::WriteFlush(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());
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![flush(Why::Idle), Action::None, Action::None, Action::CloseStdin]);
}
#[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)]);
assert_eq!(next_deadline(&state), Some(t(rotate)));
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.
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.
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]
@ -212,8 +229,8 @@ fn a_flush_is_never_asked_of_a_session_whose_input_is_closed() {
}
#[test]
fn the_grace_runs_from_the_flush_write_to_the_second_and_no_later_request_moves_it() {
let (state, _) = run(vec![(req("a", CEILING_TOKENS), 100), (req("b", CEILING_TOKENS + 1), 500)]);
fn the_grace_runs_from_the_flush_ask_to_the_second_and_no_later_request_moves_it() {
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!(feed(state.clone(), Event::Tick, 100 + GRACE - 1).1, Action::None);
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]
fn the_ladder_ends_before_the_cap_whenever_the_flush_was_written() {
let cap = 5_400_u64;
let ceiling = |second: i64| run_from(State::started(t(0), cap), vec![(req("a", CEILING_TOKENS), second)]);
let rotation = |second: i64| run_from(State::started(t(0), cap), vec![(Event::Tick, 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::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.
for second in (0..cap as i64 - LADDER_S).step_by(7) {
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;
}
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() {
let cap = 5_400_i64;
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");
}
#[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(working(), req("a", CEILING_TOKENS), 0);
assert_eq!(action, Action::WriteFlush(Why::Ceiling));
let (state, _) = feed(working(), Event::Echo, 0);
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));
// 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]
@ -388,7 +406,7 @@ const SYMBOLS: [Sym; 8] = [Sym::Request, Sym::Ceiling, Sym::Echo, Sym::Ok, Sym::
/// and the single-writer invariants.
fn play(symbols: &[Sym]) {
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 flushed = false;
let mut closed = false;
@ -414,14 +432,14 @@ fn play(symbols: &[Sym]) {
let (next, action) = feed(state, event, now);
state = next;
match action {
Action::WriteTells => {
assert!(!flushed && !closed, "a tell after the flush or the close: {symbols:?}");
tells += 1;
}
Action::WriteFlush(_) => {
assert!(!flushed && !closed, "a second flush, or one after the close: {symbols:?}");
flushed = true;
flushes += 1;
Action::Write { tells: told, flush } => {
assert!(!flushed && !closed, "a write after the flush or the close: {symbols:?}");
writes += 1;
tells += u32::from(told);
if flush.is_some() {
flushed = true;
flushes += 1;
}
}
Action::CloseStdin => {
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::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!(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.
//!
//! 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 files for herself, or one whose source the mind does not hold, is not.
//! `derive` lists the briefs the session has not been told (the told set is the
//! actuator's process memory, a cache; the mind is the truth). Her Self resolves
//! each brief she answers; `close` resolves the ones she left open at an idle
//! end. `backlog` lists, and never closes, the in-force briefs a closed run
//! claimed: the one-time sweep at deploy.
//! `self-run` that her Persona turn admitted is a brief. Who admitted it is the
//! receipt's provenance, which the mind records and a writer cannot choose; the
//! follow-up's `source` is the writer's word and decides nothing. A follow-up her
//! Self files for herself is not a brief, whatever it cites. `derive` lists the
//! briefs the session has not been told (the told set is the actuator's process
//! memory, a cache; the mind is the truth). Her Self resolves each brief she
//! 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 anyhow::{Result, anyhow};
use chrono::{DateTime, Utc};
use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, RunTurn};
use huginn_mind::{PipelineAdmissionOutcome, PipelineFacts, PipelineStatusSummary};
use anyhow::Result;
use eureka_pipeline::PipelineKind;
use huginn_mind::{PipelineFacts, PipelineStatusSummary};
use super::AGENT;
use crate::launch::closure;
use crate::persona;
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.
@ -36,12 +35,6 @@ pub struct Brief {
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.
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]))?;
@ -55,29 +48,19 @@ fn claimed_by_closed(mind: &dyn MindPort) -> Result<HashMap<String, String>> {
Ok(claimed)
}
/// Every in-force brief, oldest first: owner `self-run`, sourced from a run of
/// the mind whose turn is a Persona turn.
/// Every in-force brief, oldest first: owner `self-run`, admitted by her Persona
/// turn (the receipt's agent), whatever source the writer named.
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()
.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();
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,
})
.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);
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.")
}
/// 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)]
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
//! backlog lists, and what the idle-end close resolves.
//! The inbox over a mind in memory: what is a brief, what is told and what the
//! backlog lists.
use eureka_pipeline::{PipelineDocument, PipelineKind as K, RunOperator, RunTurn};
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 {
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)
}
/// 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 {
reference(K::FollowUp, &id("follow_up", label))
}
@ -32,27 +40,37 @@ fn ids(briefs: Vec<Brief>) -> Vec<String> {
}
#[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 persona = run_ref("persona");
mind.committed(vec![
brief_doc("from-persona", OWNER, &persona),
brief_doc("from-her-self", OWNER, &run_ref("selfrun")),
brief_doc("from-no-run", OWNER, &crate::testkit::spec_ref("x")),
brief_doc("someone-elses", "Hands", &persona),
brief_doc("answered", OWNER, &persona),
]);
admitted_by(
&mind,
PERSONA,
vec![
brief_doc("from-persona", 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())]);
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]
fn derive_leaves_out_what_the_session_was_told_and_orders_by_admission() {
let mind = mind();
let persona = run_ref("persona");
mind.committed(vec![brief_doc("first", OWNER, &persona)]);
mind.committed(vec![brief_doc("second", OWNER, &persona)]);
mind.committed(vec![brief_doc("third", OWNER, &persona)]);
admitted_by(&mind, PERSONA, vec![brief_doc("first", OWNER, &persona)]);
admitted_by(&mind, PERSONA, vec![brief_doc("second", OWNER, &persona)]);
admitted_by(&mind, PERSONA, vec![brief_doc("third", OWNER, &persona)]);
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]);
}
@ -60,7 +78,7 @@ fn derive_leaves_out_what_the_session_was_told_and_orders_by_admission() {
#[test]
fn a_brief_a_closed_run_claimed_is_still_a_brief_and_no_standing_rule_closes_it() {
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![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");
@ -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() {
let mind = mind();
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![closing(run_ref("closed-self"), recorded())]);
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!(!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
}
/// 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.
fn brief(label: &str, item: &str) -> PipelineDocument {
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(at("m1", 10)),
// 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()),
// Later polls do not tell it again.
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("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!(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);
// A result is not an answer: nothing closes the brief by its timing.
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]
fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_until_its_echo() {
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);
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);
@ -1589,7 +1601,7 @@ fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_
#[test]
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();
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 start = &ran.written()[0];
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 PipelineDocument::FollowUp(mut hers) = follow_up("hers", &run_ref("selfrun-a")) else { unreachable!() };
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)]);
assert_eq!(ran.written().len(), 1, "{:?}", ran.written());
assert!(!ran.written()[0].contains("Also open"), "{}", ran.written()[0]);
}
#[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();
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 ran = mind_run(&mind, steps);
assert_eq!(ran.count(&Event::InputClosed), 1, "{:?}", ran.events);
let left = resolutions_of(&mind, &brief_ref("p1"));
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!(resolutions_of(&mind, &brief_ref("p1")).is_empty(), "the actuator resolved nothing she left open");
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 {
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);
assert!(ran.outcome.is_ok(), "{name}");
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]
fn a_brief_admitted_after_the_flush_is_written_stays_in_force_and_untold() {
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);
assert!(tells(&ran).is_empty(), "{:?}", ran.written());
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 closed_at = hour.events.iter().position(|event| *event == Event::InputClosed).unwrap();
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());
// 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)]);
@ -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);
// A brief waiting at the cache's end is told, not flushed: the session stays open.
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);
assert_eq!((flushes(&ran), tells(&ran).len(), ran.count(&Event::InputClosed)), (0, 1, 0), "{:?}
{}", 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]
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 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);
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]
fn no_stdin_line_is_written_after_the_flush_and_nothing_but_the_writer_writes() {
let mind = live_mind();
mind.committed(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())];
persona_commit(&mind, vec![brief("b7", "Early.")]);
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 written = ran.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 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.
fn replay(jsonl: &str) -> Vec<(String, Do)> {
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() {
let trace = replay(CEILING_FOLDED);
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]
@ -1790,7 +1805,7 @@ mod live_session {
let trace = replay(stream);
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.
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() {
for stream in API_ERROR {
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)]);
}
}
@ -1807,8 +1822,8 @@ mod live_session {
fn a_context_over_the_blocking_limit_is_a_blocked_result_that_closes_the_session() {
for stream in BLOCKING {
let trace = replay(stream);
assert_eq!(kinds(&trace), ["other", "echo", "request:990000", "result:ok", "other", "echo", "request:0", "result:blocked"]);
assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (7, Do::CloseStdin)]);
assert_eq!(kinds(&trace), ["other", "echo", "request:990000", "result:ok", "other", "echo", "other", "result:blocked"]);
assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)]);
}
}
@ -1817,7 +1832,7 @@ mod live_session {
for stream in COMPACTION_OFF {
let trace = replay(stream);
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() {
let mind = TestMind::seeded();
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")])]);
let ran = carry_with(&Seen::of(&mind), vec![Out(echo())], Some(0), false);
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
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).
- Delivery is counted: `--replay-user-messages` makes the CLI echo each stdin line right before
the request that carries it, folded into the running turn or queued as the next, so the echo is
the consumption point and idle is `lines written == lines echoed` with no turn open. The flush
is answered at the first result after its echo, folded or queued alike; the grace interrupt
(earlier of 15 min after the write and the cap less the ladder) is the backstop. Compaction is
- Delivery is counted, with one write outstanding (ruling `self-session-one-outstanding-write`):
`--replay-user-messages` makes the CLI echo each stdin line right before the request that
carries it, folded into the running turn or queued as the next, but it joins lines queued
behind one request into a single echo. So `drive` writes the next line only after the previous
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).
- 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