From 780366f59a331268c73727d1ef24cb33082e586b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 19:39:25 +0100 Subject: [PATCH] self-session-live: one stdin writer; the replay echo is the consumption point; no hooks, no spool, no CLI compaction Deletes first: the PostToolBatch and PreCompact hooks, --settings, the inject subcommand, the on-disk spool and its rename claim, close_past as a standing rule, close_told and the result-timed brief closure, Event::Told/PreCompact, Action::CloseBriefs, FLUSH_MARGIN_S. Then: drive is the only writer into her session (start, tells, flush). decide counts lines written and lines echoed (--replay-user-messages) and derives idle from them; the flush is answered at the first result after its echo, folded or queued; an error before any answer or a blocked result closes at once; rotation and the idle cache end ask the same flush with a why; the grace is clamped so the signal ladder ends before the cap. DISABLE_COMPACT=1 replaces the PreCompact block. Briefs are follow-ups owned by self-run and sourced from a Persona turn; her Self resolves each she answers, the actuator closes only what she left open at an idle end. inbox::backlog lists, and never closes, the one-time sweep. Tests: the committed 2.1.312 streams replay through read_usage and decide; the unit tables, a ladder sweep and an event-order enumeration pin the rules. Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/self_run.rs | 2 +- crates/mind-body/src/self_run/decide/tests.rs | 642 +++++++----------- crates/mind-body/src/self_run/inbox/tests.rs | 154 ++--- crates/mind-body/src/self_run/tests.rs | 513 ++++++++------ .../src/self_run/tests/process/lease_child.rs | 12 +- crates/mind-body/tests/public_surface.txt | 31 +- 6 files changed, 655 insertions(+), 699 deletions(-) diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index bcdaf67..a9fec8a 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -674,7 +674,7 @@ fn drive( while let Some(heard_event) = events.pop_front() { let counted = state.requests; // The decision's inputs, as they stood when the event came. - let (ttl_s, inbox, idle_s) = (state.ttl_s, state.inbox, 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); state = next; heard.idle_end = state.idle_end_answered(); diff --git a/crates/mind-body/src/self_run/decide/tests.rs b/crates/mind-body/src/self_run/decide/tests.rs index f8cfbd9..ec3ba3f 100644 --- a/crates/mind-body/src/self_run/decide/tests.rs +++ b/crates/mind-body/src/self_run/decide/tests.rs @@ -1,18 +1,19 @@ //! `decide` is pure, so its rules are pinned here with no child, no mind and no -//! clock: tables, one row per boundary, and an enumeration of event orders that -//! asserts the invariants against an oracle kept from the event history alone. - -use std::collections::VecDeque; +//! clock: tables, one row per boundary, a sweep that the signal ladder always +//! ends before the cap, and an enumeration of event orders that asserts the +//! write accounting from the actions alone. use super::*; use crate::testkit::at; /// Long enough that no table below reaches the cap by accident. const CAP_S: u64 = 100_000; -const RESULT: Event = Event::Result { ttl_1h: true }; -const SHORT_RESULT: Event = Event::Result { ttl_1h: false }; +const OK: Event = Event::Result { ok: true, blocked: false, ttl_1h: true }; +const OK_SHORT: Event = Event::Result { ok: true, blocked: false, ttl_1h: false }; +const FAILED: Event = Event::Result { ok: false, blocked: false, ttl_1h: true }; +const BLOCKED: Event = Event::Result { ok: false, blocked: true, ttl_1h: true }; -/// A session whose first turn is running. +/// A session whose start message was written and not yet echoed. fn working() -> State { State::started(t(0), CAP_S) } @@ -29,9 +30,8 @@ fn feed(state: State, event: Event, seconds: i64) -> (State, Action) { decide(state, event, t(seconds)) } -/// Runs the events in order, each at its own second, and keeps every action. -fn run(events: Vec<(Event, i64)>) -> (State, Vec) { - let mut state = working(); +/// Runs the events in order, each at its own second, from a state, and keeps every action. +fn run_from(mut state: State, events: Vec<(Event, i64)>) -> (State, Vec) { let mut actions = Vec::new(); for (event, seconds) in events { let (next, action) = feed(state, event, seconds); @@ -41,6 +41,10 @@ fn run(events: Vec<(Event, i64)>) -> (State, Vec) { (state, actions) } +fn run(events: Vec<(Event, i64)>) -> (State, Vec) { + run_from(working(), events) +} + fn last(events: Vec<(Event, i64)>) -> Action { run(events).1.pop().unwrap() } @@ -57,154 +61,225 @@ const GRACE: i64 = FLUSH_GRACE_S; const ESC: i64 = ESCALATE_AFTER_S; const LIMIT: u64 = CEILING_TOKENS + FLUSH_HEADROOM_TOKENS; +/// The start message echoed and answered: idle at `at`, the cache written for an hour. +fn idle_at(at: i64) -> (State, Vec) { + run(vec![(Event::Echo, 0), (req("a", 10), 1), (OK, at)]) +} + #[test] fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or_for_a_repeated_request() { assert_eq!(last(vec![(req("a", CEILING_TOKENS - 1), 0)]), Action::None); - assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0)]), Action::WriteFlush); - assert_eq!(last(vec![(req("a", CEILING_TOKENS + 1), 0)]), Action::WriteFlush); - // Once. - assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", CEILING_TOKENS + 5), 1)]), Action::None); - // A child that is gone is not written to. + 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!((state.written, state.requests), (2, 2)); assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (req("a", CEILING_TOKENS), 1)]), Action::None); - // The same message id is the same request again: it neither counts nor decides. - let (state, actions) = run(vec![(req("a", 10), 0), (req("a", CEILING_TOKENS), 1)]); - assert_eq!((actions[1].clone(), state.requests, state.prefix), (Action::None, 1, 10)); } #[test] -fn a_result_ends_a_turn_and_never_closes_the_input_while_a_brief_waits_or_the_cache_is_warm() { - // The result asks for its briefs to be closed and nothing else; the next pass has nothing to do. - assert_eq!(last(vec![(RESULT, 0)]), Action::CloseBriefs); - assert_eq!(last(vec![(RESULT, 0), (Event::Tick, 0)]), Action::None); - assert_eq!(last(vec![(RESULT, 0), (Event::Tick, 3000)]), Action::None); - // A waiting brief is told, however long the session has been idle, and the input stays open. - assert_eq!(last(vec![(RESULT, 0), (Event::Inbox { pending: true }, 10)]), Action::WriteTells); - assert_eq!(last(vec![(RESULT, 0), (Event::Inbox { pending: true }, 10), (Event::Told, 10), (Event::Tick, 11)]), Action::None); - let waiting = |seconds: i64| last(vec![(RESULT, 0), (Event::Inbox { pending: true }, 10), (Event::Tick, seconds)]); - assert_eq!(waiting(TTL_1H_S * 3), Action::WriteTells, "a waiting brief outranks the cache's end"); - // The loop is woken at once for a waiting brief, and for a finished flush, not at the cache's end. - let (state, _) = run(vec![(RESULT, 5), (Event::Inbox { pending: true }, 6)]); - assert_eq!(next_deadline(&state), Some(t(5))); - let (state, _) = run(vec![(RESULT, 5), (Event::Tick, 5 + TTL_1H_S - TTL_MARGIN_S)]); - assert_eq!(next_deadline(&state), Some(t(5 + TTL_1H_S - TTL_MARGIN_S + GRACE)), "the flush's own grace: the turn runs"); - let (state, _) = run(vec![(RESULT, 5), (Event::Tick, 5 + TTL_1H_S - TTL_MARGIN_S), (SHORT_RESULT, 6 + TTL_1H_S)]); - assert_eq!(state.idle_s(t(10 + TTL_1H_S)), Some(4)); - assert_eq!(working().idle_s(t(1)), None); - // Mid-turn, a brief is the hook's: the actuator writes nothing, and the next result comes. - assert_eq!(last(vec![(Event::Inbox { pending: true }, 10)]), Action::None); - assert_eq!(last(vec![(Event::Inbox { pending: true }, 10), (RESULT, 20), (Event::Tick, 20)]), Action::WriteTells); +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![]); + assert_eq!((state.written, state.consumed), (1, 0)); + assert_eq!(state.idle_s(t(5)), None, "the start message is not yet echoed"); + let (state, _) = run(vec![(Event::Echo, 0)]); + assert_eq!(state.idle_s(t(5)), None, "its turn is open"); + let (state, _) = run(vec![(Event::Echo, 0), (OK, 7)]); + 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!(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)]); + assert_eq!(state.idle_s(t(3)), None, "the tell written at 1 has not been echoed"); } #[test] -fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_closes_after_the_flushs_own_result() { - let idle_at = |result: Event, seconds: i64| last(vec![(result, 0), (Event::Tick, seconds)]); - // An hour's cache: the flush at 3540 s, not before. - assert_eq!(idle_at(RESULT, TTL_1H_S - TTL_MARGIN_S - 1), Action::None); - assert_eq!(idle_at(RESULT, TTL_1H_S - TTL_MARGIN_S), Action::WriteFlush); - // A five-minute cache: at 240 s. - assert_eq!(idle_at(SHORT_RESULT, TTL_5M_S - TTL_MARGIN_S - 1), Action::None); - assert_eq!(idle_at(SHORT_RESULT, TTL_5M_S - TTL_MARGIN_S), Action::WriteFlush); - // The deadline the loop waits for is the same moment. - let (state, _) = run(vec![(RESULT, 5)]); - assert_eq!(next_deadline(&state), Some(t(5 + TTL_1H_S - TTL_MARGIN_S))); - // The input stays open through the flush's turn and closes after its result, one pass later. - let flushed = vec![(RESULT, 0), (Event::Tick, TTL_1H_S - TTL_MARGIN_S)]; - assert_eq!(last([flushed.clone(), vec![(Event::Tick, TTL_1H_S)]].concat()), Action::None); - let ended_turn = [flushed, vec![(SHORT_RESULT, TTL_1H_S)]].concat(); - assert_eq!(last(ended_turn.clone()), Action::CloseBriefs); - assert_eq!(last([ended_turn, vec![(Event::Tick, TTL_1H_S)]].concat()), Action::CloseStdin); +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!(next_deadline(&state), Some(t(1 + GRACE)), "closed: only the flush's grace backstop remains, no lifecycle deadline"); } #[test] -fn near_the_cap_an_idle_session_rotates_and_a_working_one_is_asked_for_its_flush() { - let cap = i64::try_from(CAP_S).unwrap(); - // A result so late that the rotation comes before the cache's end. - let late = cap - 1300; - let idle_at = |seconds: i64| last(vec![(RESULT, late), (Event::Tick, seconds)]); - assert_eq!(idle_at(cap - ROTATE_MARGIN_S - 1), Action::None); - assert_eq!(idle_at(cap - ROTATE_MARGIN_S), Action::CloseStdin); - // A waiting brief does not hold the process past the rotation: the next process tells it. - assert_eq!(last(vec![(RESULT, late), (Event::Inbox { pending: true }, cap - ROTATE_MARGIN_S)]), Action::CloseStdin); - // Mid-turn the flush is asked at the margin, once, and it keeps its grace and escalation. - let working_at = |seconds: i64| last(vec![(req("a", 10), 1), (Event::Tick, seconds)]); - assert_eq!(working_at(cap - FLUSH_MARGIN_S - 1), Action::None); - assert_eq!(working_at(cap - FLUSH_MARGIN_S), Action::WriteFlush); - let asked = vec![(req("a", 10), 1), (Event::Tick, cap - FLUSH_MARGIN_S)]; - assert_eq!(last([asked.clone(), vec![(Event::Tick, cap - FLUSH_MARGIN_S + 1)]].concat()), Action::None); - assert_eq!(last([asked, vec![(Event::Tick, cap - FLUSH_MARGIN_S + GRACE)]].concat()), Action::Signal(Stop::Interrupt)); - let (state, _) = run(vec![(req("a", 10), 1)]); - assert_eq!(next_deadline(&state), Some(t(cap - FLUSH_MARGIN_S))); +fn a_flush_queued_behind_the_running_turn_is_answered_by_the_second_result_not_the_first() { + let (_, actions) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 1), (OK, 2), (Event::Echo, 3), (req("b", 11), 4), (OK, 5)]); + assert_eq!(actions[2], Action::None, "the first result answers the turn already running"); + assert_eq!(actions[5], Action::CloseStdin); + assert!(actions.iter().all(|action| !matches!(action, Action::Signal(_))), "never a grace interrupt"); + // The queued flush is not idleness: no second flush, no tell, while it waits. + let (state, _) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 1), (OK, 2)]); + assert_eq!(state.idle_s(t(3)), None); } #[test] -fn pre_compact_asks_the_same_flush_the_ceiling_asks_once() { - assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1)]), Action::WriteFlush); - assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1), (Event::PreCompact, 2)]), Action::None); - // The ceiling after it asks nothing more; the ceiling before it leaves nothing for the hook to ask. - assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1), (req("b", CEILING_TOKENS), 2)]), Action::None); - assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (Event::PreCompact, 1)]), Action::None); - // Its flush has the grace and the headroom of any other. - assert_eq!(last(vec![(req("a", 10), 0), (Event::PreCompact, 1), (Event::Tick, 1 + GRACE)]), Action::Signal(Stop::Interrupt)); - // A child that is gone is not written to. - assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (Event::PreCompact, 1)]), Action::None); +fn an_error_result_before_any_answer_or_a_blocked_one_closes_stdin_at_once_with_no_flush() { + for (result, why) in [(FAILED, "an error before any answer"), (BLOCKED, "blocked")] { + let (state, actions) = run(vec![(Event::Echo, 0), (result, 1)]); + assert_eq!(actions[1], Action::CloseStdin, "{why}"); + assert_eq!(state.written, 1, "{why}: no flush was written"); + } + // Blocked closes even after an answer. + let (_, actions) = run(vec![(Event::Echo, 0), (OK, 1), (Event::Echo, 2), (BLOCKED, 3)]); + assert_eq!(actions[3], Action::CloseStdin); +} + +#[test] +fn an_error_after_an_ok_answer_ends_the_turn_and_not_the_session() { + let (state, actions) = run(vec![(Event::Echo, 0), (OK, 1), (Event::Inbox { pending: true }, 2), (Event::Echo, 3), (FAILED, 4)]); + assert_eq!(actions[4], Action::None); + assert!(next_deadline(&state).is_some(), "the session waits on its cache still"); + assert_eq!(state.idle_s(t(5)), Some(1)); +} + +#[test] +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!(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); + assert_eq!(feed(state, Event::Tick, 3).1, Action::None); + // 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); + // After the flush no line is written, a brief waiting or not. + let (state, _) = run(vec![(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); +} + +#[test] +fn an_idle_session_asks_the_flush_where_the_cache_would_lapse_and_a_waiting_brief_keeps_it_open() { + let (idle, _) = idle_at(10); + 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)); + // 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))); + // A brief waiting is told, not flushed, and the lapse is not the end while a line is outstanding. + 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()); +} + +#[test] +fn near_the_cap_the_session_is_asked_for_its_flush_idle_or_working_and_closes_after_its_answer() { + let cap = 10_000_u64; + let rotate = cap as i64 - ROTATE_MARGIN_S; + let started = || State::started(t(0), cap); + // Working: the start message is still being answered. + 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)); + // 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)); + // 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]); +} + +#[test] +fn a_stdin_is_never_closed_without_a_flush_except_by_a_blocked_result_or_an_error_before_any_answer() { + let (_, actions) = run(vec![(Event::Echo, 0), (OK, 1), (Event::Echo, 2), (OK, 3), (Event::Tick, CAP_S as i64 / 2)]); + assert!(actions.iter().all(|action| *action != Action::CloseStdin)); } #[test] fn a_flush_is_never_asked_of_a_session_whose_input_is_closed() { - let cap = i64::try_from(CAP_S).unwrap(); - let closed = vec![(RESULT, 1), (Event::Tick, cap - ROTATE_MARGIN_S)]; - assert_eq!(last(closed.clone()), Action::CloseStdin); - assert_eq!(last([closed.clone(), vec![(Event::PreCompact, cap)]].concat()), Action::None); - assert_eq!(last([closed, vec![(req("b", CEILING_TOKENS), cap)]].concat()), Action::None); + let (state, _) = run(vec![(Event::Echo, 0), (FAILED, 1)]); + assert_eq!(feed(state.clone(), req("a", CEILING_TOKENS), 2).1, Action::None); + assert_eq!(feed(state, Event::Tick, CAP_S as i64).1, Action::None); } #[test] fn the_grace_runs_from_the_flush_write_to_the_second_and_no_later_request_moves_it() { - let flushed = |rest: Vec<(Event, i64)>| last([vec![(req("a", 10), 0), (req("b", CEILING_TOKENS), 100)], rest].concat()); - assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE - 1)]), Action::None); - assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE)]), Action::Signal(Stop::Interrupt)); - assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE + 5000)]), Action::Signal(Stop::Interrupt)); - // Requests after the flush, however late, do not refresh it. - assert_eq!(flushed(vec![(req("c", CEILING_TOKENS + 1), 100 + GRACE - 10), (Event::Tick, 100 + GRACE)]), Action::Signal(Stop::Interrupt)); - // Not from the first request, not from the start. - assert_eq!(last(vec![(req("a", 10), 0), (req("b", CEILING_TOKENS), 100), (Event::Tick, GRACE)]), Action::None); - // The deadline the loop waits for is the same moment. - let (state, _) = run(vec![(req("a", CEILING_TOKENS), 100)]); + let (state, _) = run(vec![(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::default()), 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)); } #[test] -fn the_headroom_counts_the_requests_after_the_first_result_after_the_flush_from_the_prefix_of_the_flush() { - let after_result = |request: u64| last(vec![(req("a", CEILING_TOKENS), 0), (RESULT, 1), (req("b", request), 2)]); - assert_eq!(after_result(LIMIT - 1), Action::None); - assert_eq!(after_result(LIMIT), Action::Signal(Stop::Interrupt)); - assert_eq!(after_result(LIMIT + 1), Action::Signal(Stop::Interrupt)); - // Measured from the flush's own prefix, not from the ceiling. - let high = |request: u64| last(vec![(req("a", 400_000), 0), (RESULT, 1), (req("b", request), 2)]); - assert_eq!((high(459_999), high(460_000)), (Action::None, Action::Signal(Stop::Interrupt))); - // The turn already running is not the flush's turn: before a result no request counts. - assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", LIMIT + 400_000), 1)]), Action::None); - // A big request before the result is forgotten by the boundary; only those after it count. - assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", LIMIT + 1), 1), (RESULT, 2), (req("c", LIMIT - 1), 3)]), Action::None); - // A prefix that saturates leaves no headroom to count: only the grace is left. - let saturated = |rest: Vec<(Event, i64)>| last([vec![(req("a", u64::MAX), 0), (RESULT, 1), (req("b", u64::MAX), 2)], rest].concat()); - assert_eq!(saturated(vec![]), Action::None); - assert_eq!(saturated(vec![(Event::Tick, GRACE)]), Action::Signal(Stop::Interrupt)); - // A request counted past the headroom stays counted: a later smaller one does not take the interrupt back. - let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 0), (RESULT, 1), (req("b", LIMIT), 2)]); - assert_eq!(actions[2], Action::Signal(Stop::Interrupt)); - assert_eq!(next_deadline(&state), Some(t(2 + ESC))); +fn the_headroom_counts_the_requests_after_the_flushs_echo_from_the_prefix_of_the_flush() { + let asked = vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 1)]; + // Queued flush: requests before its echo never count, however large. + let before_echo = [asked.clone(), vec![(req("b", u64::MAX), 2)]].concat(); + assert!(run(before_echo).1.iter().all(|action| !matches!(action, Action::Signal(_)))); + let echoed = [asked, vec![(Event::Echo, 3)]].concat(); + let after = |prefix: u64| last([echoed.clone(), vec![(req("c", prefix), 4)]].concat()); + assert_eq!(after(LIMIT - 1), Action::None); + assert_eq!(after(LIMIT), Action::Signal(Stop::Interrupt)); +} + +/// Sweep: whatever second the flush is written at, and whatever asked for it, a +/// child that ignores every signal is gone before the unit's cap. +#[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)]); + // 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(_))) { + continue; + } + let mut state = state; + let mut now = second; + let finished = loop { + let Some(at_deadline) = next_deadline(&state) else { break None }; + now = now.max((at_deadline - t(0)).num_seconds()); + let (next, action) = feed(state, Event::Tick, now); + state = next; + match action { + Action::Signal(how) => state = feed(state, Event::Signalled { how, delivered: true }, now).0, + Action::Finish(_) => break Some(now), + _ => {} + } + }; + assert!(finished.is_some_and(|end| end < cap as i64), "flush at {second}: ended at {finished:?}, cap {cap}"); + } + } + // The rotation margin leaves room: this is the arithmetic the const assertion pins. + assert!(GRACE + LADDER_S < ROTATE_MARGIN_S); +} + +#[test] +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)]); + 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); + assert_eq!(action, Action::WriteFlush(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); + assert_eq!(last(vec![(req("a", u64::MAX), 0)]), Action::WriteFlush(Why::Ceiling)); } #[test] @@ -285,7 +360,7 @@ fn a_child_that_has_exited_is_never_signalled_and_the_run_ends_with_its_stream_o #[test] fn without_a_flush_nothing_is_ever_signalled_and_a_finished_run_decides_nothing_more() { - let quiet = vec![(req("a", 10), 0), (RESULT, 1), (Event::Tick, 100_000)]; + let quiet = vec![(req("a", 10), 0), (OK, 1), (Event::Tick, 100_000)]; assert!(run(quiet).1.iter().all(|action| !matches!(action, Action::Signal(_)))); let (state, actions) = run(vec![(Event::ChildExited(Some(3)), 0), (Event::StdoutEof, 0)]); assert_eq!(actions[1], ended(Cause::Own, Some(3), 0, 0)); @@ -293,285 +368,90 @@ fn without_a_flush_nothing_is_ever_signalled_and_a_finished_run_decides_nothing_ assert_eq!(feed(state, Event::Tick, 10_000).1, Action::None); } + // ---- every order of events ------------------------------------------------------ -/// The facts of the history, kept apart from `State`: what the invariants are asked of. -#[derive(Default)] -struct Oracle { - ids: Vec, - /// When the flush was written and the prefix it was written at. - flush: Option<(i64, u64)>, - flushes: u32, - closes: u32, - /// The flush is the turn that runs now or has run. - boundary: bool, - exceeded: bool, - /// No turn is running. - idle: bool, - inbox: bool, - closed: bool, - prefix: u64, - exited: Option, - eof: bool, - signals: Vec<(Stop, i64)>, - finished: bool, -} - -fn check_step(oracle: &mut Oracle, event: &Event, now: i64, action: &Action) { - assert!(!oracle.finished || *action == Action::None, "an action after the run ended: {action:?} on {event:?}"); - let flush_before = oracle.flush; - let was_exited = oracle.exited.is_some(); - let was_idle = oracle.idle; - match event { - Event::Request { id, prefix } => { - if !oracle.ids.contains(id) { - oracle.ids.push(id.clone()); - oracle.prefix = *prefix; - if oracle.boundary - && let Some((_, base)) = oracle.flush - && base.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| *prefix >= limit) - { - oracle.exceeded = true; - } - } - } - Event::Result { .. } => { - oracle.idle = true; - if oracle.flush.is_some() && !oracle.boundary { - oracle.boundary = true; - oracle.idle = false; - } - } - Event::Inbox { pending } => oracle.inbox = *pending, - Event::Told => { - oracle.inbox = false; - oracle.idle = false; - } - Event::PreCompact | Event::Tick | Event::Signalled { .. } => {} - Event::ChildExited(_) => { - oracle.exited.get_or_insert(now); - } - Event::StdoutEof => oracle.eof = true, - } - match action { - Action::None => {} - Action::WriteFlush => { - oracle.flushes += 1; - if let Event::Request { prefix, .. } = event { - assert!(*prefix >= CEILING_TOKENS, "a flush at {prefix} on a request"); - } - assert!(flush_before.is_none() && !was_exited && !oracle.closed, "flush twice, to a gone child or to a closed input"); - oracle.flush = Some((now, oracle.prefix)); - // Written while idle, its turn begins at once. - oracle.boundary = oracle.idle; - oracle.idle = false; - } - Action::WriteTells => { - assert!(oracle.idle && oracle.inbox && !oracle.closed && !was_exited, "a tell outside idle with a brief waiting"); - } - Action::CloseBriefs => assert!(matches!(event, Event::Result { .. }), "briefs closed on {event:?}"), - Action::CloseStdin => { - oracle.closes += 1; - oracle.closed = true; - let flushed = oracle.flush.is_some() && oracle.boundary; - let rotation = now >= i64::try_from(CAP_S).unwrap() - ROTATE_MARGIN_S; - assert!(was_idle && !was_exited && (flushed || rotation), "closed on {event:?} idle {was_idle} flushed {flushed} rotation {rotation}"); - } - Action::Signal(how) => { - let (flushed_at, _) = flush_before.expect("a signal before the flush was written"); - assert!(oracle.exited.is_none(), "a signal to a child that already exited"); - match oracle.signals.last() { - None => { - assert_eq!(*how, Stop::Interrupt); - assert!(now >= flushed_at + GRACE || oracle.exceeded, "interrupt before the flush had its turn: now {now}, flush {flushed_at}, exceeded {}", oracle.exceeded); - assert!(now > flushed_at || oracle.exceeded, "interrupt in the pass that wrote the flush"); - } - Some((last, sent)) => { - assert_eq!(*how, last.next(), "escalation out of order"); - assert_ne!(*last, Stop::Kill, "a signal after the kill"); - assert!(now >= sent + ESC, "escalated after {} s, bound {ESC}", now - sent); - } - } - oracle.signals.push((*how, now)); - } - Action::Finish(outcome) => { - oracle.finished = true; - let SelfRunOutcome::Ended { cause, code, .. } = outcome else { panic!("{outcome:?}") }; - match oracle.exited { - Some(at) => assert!(oracle.eof || now >= at + EXIT_DRAIN_S, "finished before the stream or the drain"), - None => { - let (last, sent) = *oracle.signals.last().expect("finished with the child alive and no signal sent"); - assert!(last == Stop::Kill && now >= sent + ESC, "finished with the child alive"); - assert_eq!(*code, None); - assert!(matches!(cause, Cause::Signalled(Stop::Kill) | Cause::Unstoppable), "{cause:?}"); - } - } - if matches!(cause, Cause::Own) { - assert!(oracle.flush.is_none() && oracle.signals.is_empty()); - } - if matches!(cause, Cause::FlushRequested) { - assert!(oracle.flush.is_some()); - } - } - } - assert!(oracle.flushes <= 1 && oracle.closes <= 1); - assert!(oracle.signals.len() <= 3); -} - -/// What can happen next, each at a time after the last. +/// One thing that can happen next, for the enumeration. #[derive(Clone, Copy, Debug)] enum Sym { - Low, - AtCeiling, - Huge, - Repeat, - Result, - ShortResult, - Brief, - NoBrief, - PreCompact, - Exit(Option), - Eof, - Soon, - Bound, - Grace, - Cache, -} - -const ALPHABET: [Sym; 16] = [ - Sym::Low, - Sym::AtCeiling, - Sym::Huge, - Sym::Repeat, - Sym::Result, - Sym::ShortResult, - Sym::Brief, - Sym::NoBrief, - Sym::PreCompact, - Sym::Exit(Some(0)), - Sym::Exit(None), - Sym::Eof, - Sym::Soon, - Sym::Bound, - Sym::Grace, - Sym::Cache, -]; - -/// Which signals the operating system lets through. -#[derive(Clone, Copy, Debug)] -enum Policy { - All, + Request, + Ceiling, + Echo, + Ok, + Failed, + Waiting, Nothing, - NoKill, + Late, } -fn play(symbols: &[Sym], policy: Policy) { +const SYMBOLS: [Sym; 8] = [Sym::Request, Sym::Ceiling, Sym::Echo, Sym::Ok, Sym::Failed, Sym::Waiting, Sym::Nothing, Sym::Late]; + +/// Plays the symbols and asserts, from the actions alone, the write accounting +/// and the single-writer invariants. +fn play(symbols: &[Sym]) { let mut state = working(); - let mut oracle = Oracle::default(); - let mut now = 0i64; - let mut unique = 0u32; - let mut events: Vec = Vec::new(); - for symbol in symbols { - let (event, dt) = match symbol { - Sym::Low => (req_new(&mut unique, 1_000), 1), - Sym::AtCeiling => (req_new(&mut unique, CEILING_TOKENS), 1), - Sym::Huge => (req_new(&mut unique, LIMIT + 400_000), 1), - Sym::Repeat => (req("u1", CEILING_TOKENS + 7), 1), - Sym::Result => (RESULT, 1), - Sym::ShortResult => (SHORT_RESULT, 1), - Sym::Brief => (Event::Inbox { pending: true }, 1), - Sym::NoBrief => (Event::Inbox { pending: false }, 1), - Sym::PreCompact => (Event::PreCompact, 1), - Sym::Exit(code) => (Event::ChildExited(*code), 1), - Sym::Eof => (Event::StdoutEof, 1), - Sym::Soon => (Event::Tick, 1), - Sym::Bound => (Event::Tick, ESC + 1), - Sym::Grace => (Event::Tick, GRACE + 1), - Sym::Cache => (Event::Tick, TTL_1H_S), + let (mut tells, mut flushes, mut closes, mut echoes) = (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; + for sym in symbols { + now += match sym { + Sym::Late => 3_000, + _ => 10, }; - now += dt; - events.push(event.clone()); - let mut pending = VecDeque::from([event]); - while let Some(event) = pending.pop_front() { - let (after, action) = decide(state, event.clone(), t(now)); - state = after; - check_step(&mut oracle, &event, now, &action); - match action { - Action::Signal(how) => { - let delivered = match policy { - Policy::All => true, - Policy::Nothing => false, - Policy::NoKill => how != Stop::Kill, - }; - pending.push_front(Event::Signalled { how, delivered }); - } - // The loop claims the spool and writes the line: a turn begins. - Action::WriteTells => pending.push_front(Event::Told), - _ => {} + unique += 1; + let event = match sym { + Sym::Request => req(&format!("r{unique}"), 10), + Sym::Ceiling => req(&format!("r{unique}"), CEILING_TOKENS), + Sym::Echo => { + echoes += 1; + Event::Echo } - } - } - // No spinning: a deadline that has come is never answered with nothing. - if let Some(deadline) = next_deadline(&state) { - let at = deadline.max(t(now)); - let (_, action) = decide(state.clone(), Event::Tick, at); - assert_ne!(action, Action::None, "a due deadline gave no action: {symbols:?} {policy:?}"); - } - // Once the child has exited and its stream has ended, the run is over, whatever came before. - for event in [Event::ChildExited(Some(0)), Event::StdoutEof] { - now += 1; - let mut next = Some(event); - while let Some(event) = next.take() { - let (after, action) = decide(state, event.clone(), t(now)); - state = after; - check_step(&mut oracle, &event, now, &action); - if let Action::Signal(how) = action { - next = Some(Event::Signalled { how, delivered: true }); + Sym::Ok => OK, + Sym::Failed => FAILED, + Sym::Waiting => Event::Inbox { pending: true }, + Sym::Nothing => Event::Inbox { pending: false }, + Sym::Late => Event::Tick, + }; + 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::CloseStdin => { + assert!(!closed, "stdin closed twice: {symbols:?}"); + closed = true; + closes += 1; + } + 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.consumed, echoes, "every echo is counted once: {symbols:?}"); + assert!(closes <= 1 && flushes <= 1); } - assert!(oracle.finished, "no end after the exit and the end of the stream: {symbols:?} {policy:?} {events:?}"); } -fn req_new(unique: &mut u32, prefix: u64) -> Event { - *unique += 1; - req(&format!("u{unique}"), prefix) -} - -fn enumerate(length: usize, symbols: &mut Vec, policy: Policy) { +fn enumerate(length: usize, symbols: &mut Vec) { + play(symbols); if symbols.len() == length { - play(symbols, policy); return; } - for symbol in ALPHABET { - symbols.push(symbol); - enumerate(length, symbols, policy); + for sym in SYMBOLS { + symbols.push(sym); + enumerate(length, symbols); symbols.pop(); } } #[test] -fn every_order_of_up_to_four_events_keeps_the_invariants_under_every_policy_of_delivery() { - for policy in [Policy::All, Policy::Nothing, Policy::NoKill] { - for length in 0..=4 { - enumerate(length, &mut Vec::new(), policy); - } - } -} - -#[test] -fn a_long_random_order_of_events_keeps_the_invariants_and_always_ends_once_the_child_is_gone() { - // xorshift: the same orders every run. - let mut seed = 0x2545_F491_4F6C_DD1Du64; - let mut next = || { - seed ^= seed << 13; - seed ^= seed >> 7; - seed ^= seed << 17; - seed - }; - for round in 0..40_000 { - let length = 6 + (next() % 30) as usize; - let symbols: Vec = (0..length).map(|_| ALPHABET[(next() % ALPHABET.len() as u64) as usize]).collect(); - play(&symbols, [Policy::All, Policy::Nothing, Policy::NoKill][round % 3]); - } +fn every_order_of_up_to_four_events_keeps_the_write_accounting_and_the_single_flush() { + enumerate(4, &mut Vec::new()); } diff --git a/crates/mind-body/src/self_run/inbox/tests.rs b/crates/mind-body/src/self_run/inbox/tests.rs index 9f4ae85..f778c26 100644 --- a/crates/mind-body/src/self_run/inbox/tests.rs +++ b/crates/mind-body/src/self_run/inbox/tests.rs @@ -1,17 +1,13 @@ -//! The inbox over a real spool directory and a mind in memory: what is derived, -//! the exactly-once claim, and the hook that prints it. +//! 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. 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, spec_ref}; +use crate::testkit::{TestMind, at, close as closing, follow_up, id, recorded, reference, run, run_ref, spec}; -fn spool_dir() -> tempfile::TempDir { - tempfile::tempdir().unwrap() -} - -fn brief_doc(label: &str, owner: &str) -> PipelineDocument { - let PipelineDocument::FollowUp(mut document) = follow_up(label, &spec_ref("x")) else { unreachable!() }; +fn brief_doc(label: &str, owner: &str, source: &PipelineRef) -> PipelineDocument { + let PipelineDocument::FollowUp(mut document) = follow_up(label, source) else { unreachable!() }; document.owner = owner.into(); document.item = format!("Do {label}.").as_str().into(); PipelineDocument::FollowUp(document) @@ -21,116 +17,86 @@ fn brief_ref(label: &str) -> PipelineRef { reference(K::FollowUp, &id("follow_up", label)) } -fn mind_with(documents: Vec) -> TestMind { +/// A mind holding a Persona turn's run `persona`, a Self run `selfrun` and a spec. +fn mind() -> TestMind { let mind = TestMind::seeded(); mind.committed(vec![spec("x")]); - mind.committed(documents); + mind.committed(vec![run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[]), run("selfrun", RunTurn::SelfRun, RunOperator::Mind, &[])]); + // Only one Self run is live at a time: hers has ended, and stays a Self run. + mind.committed(vec![closing(run_ref("selfrun"), recorded())]); mind } -fn entry(n: u64) -> Brief { - Brief { id: format!("eureka-body:follow_up:b{n}"), campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: format!("Item {n}."), ordinal: n } +fn ids(briefs: Vec) -> Vec { + briefs.into_iter().map(|brief| brief.id).collect() } #[test] -fn the_inbox_is_the_open_self_run_briefs_that_no_closed_run_claims_and_the_session_does_not_hold() { - let mind = mind_with(vec![brief_doc("open", OWNER), brief_doc("claimed-by-old", OWNER), brief_doc("someone-elses", "Hands"), brief_doc("answered", OWNER), brief_doc("held", OWNER)]); - mind.committed(vec![run("old", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed-by-old")])]); - mind.committed(vec![closing(run_ref("old"), recorded()), closing(brief_ref("answered"), recorded())]); - let spool = spool_dir(); - mark_told(spool.path(), &[Brief { id: brief_ref("held").id.0, ..entry(9) }]).unwrap(); - let derived: Vec = derive(&mind, spool.path()).unwrap().into_iter().map(|brief| brief.id).collect(); - assert_eq!(derived, vec![brief_ref("open").id.0], "only the open one nobody answered and nobody holds"); - // A run that is still open claims without answering: its brief is still asked. - mind.committed(vec![run("live", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("open")])]); - assert_eq!(derive(&mind, spool.path()).unwrap().len(), 1); +fn a_brief_is_an_in_force_self_run_follow_up_sourced_from_a_persona_turn() { + 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), + ]); + 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"); } #[test] -fn spooled_entries_are_claimed_in_order_and_hand_back_what_was_written() { - let spool = spool_dir(); - spool_briefs(spool.path(), &[entry(2), entry(1)]); - assert!(pending(spool.path()).unwrap()); - let claimed = claim(spool.path()).unwrap(); - assert_eq!(claimed, vec![entry(1), entry(2)], "oldest first"); - assert!(!pending(spool.path()).unwrap() && claim(spool.path()).unwrap().is_empty(), "a claim empties the spool"); - assert_eq!(told(spool.path()).unwrap(), vec![entry(1), entry(2)]); - forget(spool.path(), &claimed[..1]); - assert_eq!(told(spool.path()).unwrap(), vec![entry(2)]); -} - -fn spool_briefs(dir: &Path, briefs: &[Brief]) { - spool(dir, briefs).unwrap(); +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)]); + let told: HashSet = [brief_ref("second").id.0].into(); + assert_eq!(ids(derive(&mind, &told).unwrap()), vec![brief_ref("first").id.0, brief_ref("third").id.0]); } #[test] -fn a_hook_and_the_actuator_claiming_at_once_deliver_each_entry_to_exactly_one_of_them() { - for _ in 0..20 { - let spool = spool_dir(); - let briefs: Vec = (1..=60).map(entry).collect(); - spool_briefs(spool.path(), &briefs); - let path = spool.path().to_path_buf(); - let racers: Vec<_> = (0..4) - .map(|_| { - let path = path.clone(); - std::thread::spawn(move || claim(&path).unwrap()) - }) - .collect(); - let mut seen: Vec = racers.into_iter().flat_map(|racer| racer.join().unwrap()).map(|brief| brief.ordinal).collect(); - seen.sort_unstable(); - assert_eq!(seen, (1..=60).collect::>(), "every entry once, none twice, none lost"); - } +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"))]); + 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"); } #[test] -fn the_hook_prints_data_only_and_only_for_the_main_thread() { - let spool = spool_dir(); - write_plan(spool.path(), "Plan usage: unobserved.").unwrap(); - assert_eq!(inject(spool.path(), "{}", false).unwrap(), Injection { stdout: String::new(), exit: 0 }, "an empty spool prints nothing"); - spool_briefs(spool.path(), &[entry(1)]); - // A subagent's call (the input carries an agent id) prints nothing and claims nothing. - let subagent = inject(spool.path(), "{\"hook_event_name\":\"PostToolBatch\",\"agent_id\":\"a1\"}", false).unwrap(); - assert_eq!(subagent, Injection { stdout: String::new(), exit: 0 }); - assert!(pending(spool.path()).unwrap()); - let printed = inject(spool.path(), "{\"hook_event_name\":\"PostToolBatch\"}", false).unwrap(); - assert_eq!(printed.exit, 0); - let output: Value = serde_json::from_str(&printed.stdout).unwrap(); - assert_eq!(output["hookSpecificOutput"]["hookEventName"], "PostToolBatch"); - let context = output["hookSpecificOutput"]["additionalContext"].as_str().unwrap(); - for needle in ["eureka-body:follow_up:b1", "campaign eureka-body", "repo GameCult/Huginn", "Item 1.", "Your answer closes it", "Plan usage: unobserved."] { - assert!(context.contains(needle), "{needle}: {context}"); - } - assert_eq!(inject(spool.path(), "{}", false).unwrap().stdout, "", "claimed once"); - // Unreadable input is not a reason to lose a brief or to fail the tool call. - spool_briefs(spool.path(), &[entry(2)]); - assert!(inject(spool.path(), "not json", false).unwrap().stdout.contains("Item 2.")); +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"))]); + 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")])]); + let before = mind.ids(K::Resolution).len(); + let listed = backlog(&mind).unwrap(); + assert_eq!(listed.iter().map(|(brief, run)| (brief.id.clone(), run.clone())).collect::>(), vec![(brief_ref("stale").id.0, run_ref("closed-self").id.0)], "only a Persona brief claimed by a closed run"); + assert_eq!(mind.ids(K::Resolution).len(), before, "listing admits nothing"); } #[test] -fn the_pre_compact_hook_blocks_with_exit_2_and_leaves_one_marker() { - let spool = spool_dir(); - spool_briefs(spool.path(), &[entry(1)]); - assert_eq!(inject(spool.path(), "{}", true).unwrap(), Injection { stdout: String::new(), exit: 2 }); - assert!(pending(spool.path()).unwrap(), "blocking claims nothing"); - assert!(take_pre_compact(spool.path()) && !take_pre_compact(spool.path()), "the marker is consumed once"); -} - -#[test] -fn the_settings_name_the_two_hooks_with_absolute_quoted_commands() { - let hooks = Hooks { spool: PathBuf::from("/run/mind-x-self/inbox"), injector: PathBuf::from("/opt/it's/mind-self-run") }; - let settings: Value = serde_json::from_str(&hooks.settings()).unwrap(); - let tell = settings["hooks"]["PostToolBatch"][0]["hooks"][0]["command"].as_str().unwrap(); - assert_eq!(tell, "'/opt/it'\\''s/mind-self-run' inject --spool '/run/mind-x-self/inbox'"); - let compact = &settings["hooks"]["PreCompact"][0]; - assert_eq!(compact["matcher"], "auto"); - assert_eq!(compact["hooks"][0]["command"].as_str().unwrap(), format!("{tell} --pre-compact")); +fn a_tell_names_each_brief_the_plan_and_how_she_resolves_it() { + let brief = Brief { id: "eureka-body:follow_up:b1".into(), campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: "Wire the thing.".into(), ordinal: 1 }; + let text = render(&[brief], "Plan: 5h at 40%."); + assert!(text.contains("eureka-body:follow_up:b1") && text.contains("Wire the thing."), "{text}"); + assert!(text.contains("resolve follow-up eureka-body:follow_up:b1 Recorded") && text.contains("Withdrawn"), "she is told to resolve it: {text}"); + 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_with(vec![brief_doc("a", OWNER), brief_doc("b", OWNER)]); + 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 briefs = [(Brief { id: brief_ref("a").id.0, ..entry(1) }, "The answer.".to_string()), (Brief { id: brief_ref("b").id.0, ..entry(2) }, "Ignored.".to_string())]; + 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 = mind diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 079b6c9..f2a30ff 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -59,6 +59,16 @@ fn result_line() -> String { PROBE.lines().nth(2).unwrap().to_string() } +/// The replay echo of a user line the CLI took into a request. +fn echo() -> String { + json!({"type": "user", "isReplay": true, "uuid": "00000000-0000-4000-8000-000000000001", "message": {"role": "user", "content": "a line"}}).to_string() +} + +/// A turn that ended well, with an answer. +fn ok_result() -> String { + json!({"type": "result", "subtype": "success", "is_error": false, "terminal_reason": "completed", "result": "The answer."}).to_string() +} + fn init_line() -> String { PROBE.lines().next().unwrap().to_string() } @@ -188,29 +198,15 @@ struct Fake { child: RefCell>, spawns: RefCell>>, refuse: bool, - hooks: Hooks, } impl Fake { fn with(child: Option, refuse: bool) -> Self { - Self { child: RefCell::new(child), spawns: RefCell::new(Vec::new()), refuse, hooks: test_hooks() } + Self { child: RefCell::new(child), spawns: RefCell::new(Vec::new()), refuse } } } -/// A spool of its own, so that tests running side by side share nothing. Its directory outlives -/// the test: a run's spool is read back after `carry` returns. -fn test_hooks() -> Hooks { - let dir = tempfile::tempdir().unwrap(); - let spool = dir.path().join("inbox"); - std::mem::forget(dir); - Hooks { spool, injector: PathBuf::from("/opt/mind-self-run") } -} - impl Spawner for Fake { - fn hooks(&self) -> Hooks { - self.hooks.clone() - } - fn spawn(&self, args: &[String], _token: &TokenFd) -> Result> { self.spawns.borrow_mut().push(args.to_vec()); if self.refuse { @@ -337,7 +333,6 @@ impl MindPort for Seen<'_> { } struct Ran { - hooks: Hooks, outcome: Result, out: String, events: Vec, @@ -401,15 +396,13 @@ fn carry_signalled(seen: &Seen, steps: Vec, code: Option, dies_on: Op } fn carry_scripted(seen: &Seen, steps: Vec, code: Option, dies_on: Option, refuses: &[Stop], hangs: bool) -> Ran { - carry_hooked(seen, steps, (code, dies_on, refuses, hangs), test_hooks()) + carry_one(seen, steps, (code, dies_on, refuses, hangs)) } -/// The same with the spool the child's hooks drain named by the test, so that a step can fire the hook. -fn carry_hooked(seen: &Seen, steps: Vec, (code, dies_on, refuses, hangs): (Option, Option, &[Stop], bool), hooks: Hooks) -> Ran { +fn carry_one(seen: &Seen, steps: Vec, (code, dies_on, refuses, hangs): (Option, Option, &[Stop], bool)) -> Ran { let log = Rc::new(RefCell::new(Vec::new())); let clock = Rc::new(Fixed(Cell::new(at_start()))); - let mut fake = Fake::with(Some(Scripted { steps: steps.into(), log: log.clone(), clock: clock.clone(), code, dies_on, refuses: refuses.to_vec(), hangs, exited: false, eof: false, asked: 0 }), false); - fake.hooks = hooks; + let fake = Fake::with(Some(Scripted { steps: steps.into(), log: log.clone(), clock: clock.clone(), code, dies_on, refuses: refuses.to_vec(), hangs, exited: false, eof: false, asked: 0 }), false); let lender = FakeLender::logging(log.clone(), Answer::Grant); let mut out = Vec::new(); let outcome = carry(seen, &*clock, &lender, &fake, TEST_CAP_S, &mut out); @@ -417,7 +410,7 @@ fn carry_hooked(seen: &Seen, steps: Vec, (code, dies_on, refuses, hangs): // The child's own events; the lease's are in `all`, in order with them. let events = all.iter().filter(|event| !matches!(event, Event::Lent | Event::Spawned | Event::Abandoned)).cloned().collect(); let spawns = fake.spawns.borrow().clone(); - Ran { hooks: fake.hooks(), outcome, out: String::from_utf8(out).unwrap(), events, all, spawns, admits: seen.admits.get(), asked: lender.asked.borrow().clone() } + Ran { outcome, out: String::from_utf8(out).unwrap(), events, all, spawns, admits: seen.admits.get(), asked: lender.asked.borrow().clone() } } fn at_start() -> chrono::DateTime { @@ -439,7 +432,7 @@ fn go(steps: Vec) -> Ran { type Seed = Box; fn flushes(ran: &Ran) -> usize { - ran.written().iter().filter(|text| text.as_str() == FLUSH_MESSAGE).count() + ran.written().iter().filter(|text| text.starts_with(FLUSH_MESSAGE)).count() } // ---- who is carried ----------------------------------------------------- @@ -511,8 +504,7 @@ fn a_mind_that_cannot_be_read_or_a_child_that_cannot_start_is_an_error_not_an_ou #[test] fn claude_args_is_exactly_the_fixed_list_for_a_session() { let session = Uuid::parse_str(SESSION_OF_RUN_A).unwrap(); - let hooks = test_hooks(); - let args = claude_args(session, &hooks); + let args = claude_args(session); assert_eq!( args, [ @@ -532,8 +524,7 @@ fn claude_args_is_exactly_the_fixed_list_for_a_session() { "none", "--setting-sources", "user", - "--settings", - hooks.settings().as_str(), + "--replay-user-messages", ] ); // Those belong to the run measure or to the unit's managed settings. @@ -561,7 +552,7 @@ fn the_child_is_started_for_her_run_with_the_fixed_list_whatever_the_run_record_ mind.committed(vec![self_run("selfrun-b", budget)]); let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); assert_eq!(ran.spawns.len(), 1, "{budget}"); - assert_eq!(ran.spawns[0], claude_args(Uuid::parse_str(SESSION_OF_RUN_B).unwrap(), &ran.hooks), "{budget}"); + assert_eq!(ran.spawns[0], claude_args(Uuid::parse_str(SESSION_OF_RUN_B).unwrap()), "{budget}"); } } @@ -718,7 +709,18 @@ fn read_usage_classifies_the_owners_real_records() { } } assert_eq!(seen.iter().map(|(_, prefix)| *prefix).collect::>(), sums); - assert_eq!(read_usage(&result_line()), Usage::Result); + assert_eq!(read_usage(&result_line()), Usage::Result { ok: false, blocked: false }); + assert_eq!(read_usage(&ok_result()), Usage::Result { ok: true, blocked: false }); + let blocked = json!({"type": "result", "subtype": "success", "is_error": true, "terminal_reason": "blocking_limit", "result": "Prompt is too long"}).to_string(); + assert_eq!(read_usage(&blocked), Usage::Result { ok: false, blocked: true }); + // An echo is a main-thread user line the CLI replayed, with string content; nothing else is. + assert_eq!(read_usage(&echo()), Usage::Echo); + let tool_result = json!({"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t", "content": "x"}]}}).to_string(); + assert_eq!(read_usage(&tool_result), Usage::Other, "a tool result is not one of our lines"); + let not_replayed = json!({"type": "user", "isReplay": false, "message": {"role": "user", "content": "a line"}}).to_string(); + assert_eq!(read_usage(¬_replayed), Usage::Other); + let subagent_echo = json!({"type": "user", "isReplay": true, "parent_tool_use_id": "toolu_1", "message": {"role": "user", "content": "a line"}}).to_string(); + assert_eq!(read_usage(&subagent_echo), Usage::Other, "a subagent's line is not ours"); assert_eq!(read_usage(&init_line()), Usage::Other); assert_eq!(read_usage(&subagent("m", 5)), Usage::Other, "a subagent's request is not the main thread's"); let mut no_id: Value = serde_json::from_str(&at("m", 5)).unwrap(); @@ -779,9 +781,8 @@ fn below_the_ceiling_no_flush_is_written_and_at_it_exactly_one() { #[test] fn a_closed_input_is_never_asked_for_a_flush_whatever_the_prefix_grows_to() { - let cap = i64::try_from(TEST_CAP_S).unwrap(); - // Rotation closes the idle session; a request that comes after it asks nothing. - let ran = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(200), Out(at("m2", CEILING_TOKENS + 1))]); + // An error before any answer closes the input; a request that comes after it asks nothing. + let ran = go(vec![Out(echo()), Out(result_line()), Out(at("m2", CEILING_TOKENS + 1))]); assert_eq!(ran.count(&Event::InputClosed), 1); assert_eq!(flushes(&ran), 0); assert_eq!(ran.ended().0, Cause::Own); @@ -789,15 +790,20 @@ fn a_closed_input_is_never_asked_for_a_flush_whatever_the_prefix_grows_to() { } #[test] -fn stdin_closes_once_after_the_flushs_own_result_and_nothing_is_written_after() { - // The first result ends the turn the flush was queued behind; the second is the flush's own. - let flushed = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(result_line())]); - assert_eq!(flushed.count(&Event::InputClosed), 1); - let closed_at = flushed.events.iter().position(|event| *event == Event::InputClosed).unwrap(); - assert_eq!(flushed.events[..closed_at].iter().filter(|event| matches!(event, Event::Wrote(_))).count(), 2, "start and flush before the close"); - assert!(flushed.events[closed_at..].iter().all(|event| !matches!(event, Event::Wrote(_)))); - // One result closes nothing: the flush is still to have its turn. - let one = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())]); +fn stdin_closes_once_at_the_result_after_the_flushs_echo_folded_or_queued_and_nothing_is_written_after() { + let writes_before_close = |ran: &Ran| { + assert_eq!(ran.count(&Event::InputClosed), 1, "{:?}", ran.events); + let closed_at = ran.events.iter().position(|event| *event == Event::InputClosed).unwrap(); + assert_eq!(ran.events[..closed_at].iter().filter(|event| matches!(event, Event::Wrote(_))).count(), 2, "start and flush before the close"); + assert!(ran.events[closed_at..].iter().all(|event| !matches!(event, Event::Wrote(_)))); + assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0, "never a grace interrupt"); + }; + // Folded: the flush lands in the running turn; its one result answers it. + writes_before_close(&go(vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(echo()), Out(at("m2", 11)), Out(ok_result())])); + // Queued: the first result ends the turn it was queued behind; the second is the flush's own. + writes_before_close(&go(vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(ok_result()), Out(echo()), Out(at("m2", 11)), Out(ok_result())])); + // The first result of a queued flush closes nothing: the flush is still to have its turn. + let one = go(vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(ok_result())]); assert_eq!(one.count(&Event::InputClosed), 0, "{:?}", one.events); let never = go(vec![Out(at("m1", 10))]); @@ -840,17 +846,17 @@ fn the_wait_for_the_child_never_passes_the_grace_deadline_and_never_spins_after_ #[test] fn the_headroom_counts_the_requests_after_the_first_result_that_follows_the_flush_and_one_short_of_it_does_not_interrupt() { let hard = CEILING_TOKENS + FLUSH_HEADROOM_TOKENS; - let short = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(at("m2", hard - 1))]); + let short = go(vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(ok_result()), Out(echo()), Out(at("m2", hard - 1))]); assert_eq!(short.count(&Event::Stopped(Stop::Interrupt)), 0); assert_eq!(short.ended().0, Cause::FlushRequested); - let at_hard = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(at("m2", hard)), Out(at("m3", hard + 1))], Some(130), true); + let at_hard = carry_with(&Seen::of(&with_run()), vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(ok_result()), Out(echo()), Out(at("m2", hard)), Out(at("m3", hard + 1))], Some(130), true); assert_eq!(at_hard.count(&Event::Stopped(Stop::Interrupt)), 1); assert_eq!(at_hard.ended().0, Cause::Signalled(Stop::Interrupt)); assert_eq!(FLUSH_HEADROOM_TOKENS, 60_000); - // Before that result the running turn's own requests are not the flush's turn: however far they grow, no interrupt. - let same_turn = go(vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", hard)), Out(at("m3", hard + 500_000))]); + // Before the flush's echo the running turn's own requests are not the flush's turn: however far they grow, no interrupt. + let same_turn = go(vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(at("m2", hard)), Out(at("m3", hard + 500_000))]); assert_eq!(same_turn.count(&Event::Stopped(Stop::Interrupt)), 0); } @@ -858,7 +864,7 @@ fn the_headroom_counts_the_requests_after_the_first_result_that_follows_the_flus fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_the_running_turn_cannot_use_up_the_flushs_turn() { let jump = carry_with( &Seen::of(&with_run()), - vec![Out(at("m1", 290_000)), Out(at("m2", 400_000)), Out(at("m3", 470_000)), Out(result_line()), Out(at("m4", 459_999)), Out(at("m5", 460_000)), Out(at("m6", 470_000))], + vec![Out(echo()), Out(at("m1", 290_000)), Out(at("m2", 400_000)), Out(at("m3", 470_000)), Out(ok_result()), Out(echo()), Out(at("m4", 459_999)), Out(at("m5", 460_000)), Out(at("m6", 470_000))], Some(130), true, ); @@ -873,15 +879,15 @@ fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_the_running_tu _ => None, }) .collect(); - // Reads: m1, m2 (the jump, which writes the flush), m3 (the running turn's 470k, past the headroom: no interrupt). - assert_eq!(kinds[..5], ["read", "read", "flush", "read", "read"], "{kinds:?}"); + // Reads: the start echo, m1, m2 (the jump, which writes the flush), then m3 (the running turn's 470k, past the headroom: no interrupt). + assert_eq!(kinds[..6], ["read", "read", "read", "flush", "read", "read"], "{kinds:?}"); // The headroom is measured from the prefix the flush was asked at (400k) and counts the requests after the - // first result only: 459_999 does not interrupt; 460_000 does, once, and m6 is never read. + // flush's echo only: 459_999 does not interrupt; 460_000 does, once, and m6 is never read. assert_eq!(jump.count(&Event::Stopped(Stop::Interrupt)), 1, "{kinds:?}"); assert_eq!(jump.lines(), "request 1 prefix 290000 request 2 prefix 400000 request 3 prefix 470000 -result subtype=success is_error=true: Failed to authenticate: OAuth session expired and could not be refreshed +result subtype=success is_error=false: The answer. request 4 prefix 459999 request 5 prefix 460000 "); @@ -933,10 +939,11 @@ fn a_usage_sum_that_overflows_is_at_or_over_the_ceiling_not_a_panic_or_a_wrap() } #[test] -fn a_result_before_any_request_leaves_the_input_open_and_the_run_ends_with_a_typed_outcome() { - // The shape of an API or authentication failure: the child reports an error result and no request ever came. - let ran = carry_with(&Seen::of(&with_run()), vec![Out(result_line())], Some(1), false); - assert_eq!(ran.count(&Event::InputClosed), 0); +fn an_error_result_before_any_answer_closes_the_input_at_once_and_the_run_ends_with_a_typed_outcome() { + // The shape of an API or authentication failure: an error result before any answer. A flush would fail the same way. + let ran = carry_with(&Seen::of(&with_run()), vec![Out(echo()), Out(result_line())], Some(1), false); + assert_eq!(ran.count(&Event::InputClosed), 1); + assert_eq!(flushes(&ran), 0); assert_eq!(ran.ended(), (Cause::Own, Some(1), 0, 0)); assert_eq!(ran.outcome.as_ref().unwrap().exit_status(), 1); assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0); @@ -944,8 +951,8 @@ fn a_result_before_any_request_leaves_the_input_open_and_the_run_ends_with_a_typ #[test] fn without_a_flush_nothing_is_ever_signalled_however_large_the_prefix_grows() { - let cap = i64::try_from(TEST_CAP_S).unwrap(); - let ran = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(200), Out(at("m2", 900_000)), Idle(100_000)]); + // An error before any answer closed the input: a request far over the ceiling asks for no flush and nothing is signalled. + let ran = go(vec![Out(echo()), Out(result_line()), Out(at("m2", 900_000)), Idle(100_000)]); assert_eq!(ran.events.iter().filter(|event| matches!(event, Event::Stopped(_))).count(), 0); assert_eq!(ran.ended().0, Cause::Own); } @@ -1018,15 +1025,15 @@ fn the_clock_is_read_once_per_event_so_a_step_of_the_wall_clock_cannot_split_a_p now } } - let steps: VecDeque = vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", CEILING_TOKENS + 1)), Idle(0)].into(); + let steps: VecDeque = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(at("m2", CEILING_TOKENS + 1)), Idle(0)].into(); let child = Scripted { steps, log: log.clone(), clock: Rc::new(Fixed(Cell::new(at_start()))), code: Some(0), dies_on: None, refuses: vec![], hangs: false, exited: false, eof: false, asked: 0 }; let fake = Fake::with(Some(child), false); let outcome = carry(&Seen::of(&mind), &Stepped(clock.clone()), &FakeLender::new(Answer::Grant), &fake, TEST_CAP_S, &mut Vec::new()).unwrap(); let events = log.borrow().clone(); let reads = events.iter().filter(|event| matches!(event, Event::Asked(_))).count() as u32; - // One read for the lease's expiry, one for closing the past, one before the first wait, then one per fact heard, and one for the closure: + // One read for the lease's expiry, one before the first wait, then one per fact heard, and one for the closure: // never a second read inside a pass. - assert_eq!(clock.1.get(), 4 + reads, "{events:?}"); + assert_eq!(clock.1.get(), 3 + reads, "{events:?}"); // The flush was written on the first request at time 0; the very next pass sees the stepped clock. // The flush pass itself has no interrupt, whatever the step: the interrupt is on a later event. let flush_at = events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains("context ceiling"))).unwrap(); @@ -1038,9 +1045,9 @@ fn the_clock_is_read_once_per_event_so_a_step_of_the_wall_clock_cannot_split_a_p #[test] fn a_deadline_that_came_during_a_line_is_answered_without_waiting_for_the_next_poll() { // The result arrives after the grace has run out: the input closes, and the interrupt is due at once. - let ran = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Step::After(FLUSH_GRACE_S + 5, result_line())], Some(130), Some(Stop::Interrupt), &[], true); + let ran = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Step::After(FLUSH_GRACE_S + 5, ok_result())], Some(130), Some(Stop::Interrupt), &[], true); assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 1); - assert_eq!(ran.count(&Event::InputClosed), 0, "the flush's turn has begun, not ended"); + assert_eq!(ran.count(&Event::InputClosed), 0, "the flush is queued behind the turn that just ended: its echo has not come"); let stop_at = ran.events.iter().position(|event| matches!(event, Event::Stopped(_))).unwrap(); let reads_before = ran.events[..stop_at].iter().filter(|event| matches!(event, Event::Asked(_))).count(); assert_eq!(reads_before, 2, "the interrupt is in the pass that heard the result, not a poll away: {:?}", ran.events); @@ -1244,9 +1251,6 @@ mod process { struct Spawn(PathBuf); impl Spawner for Spawn { - fn hooks(&self) -> Hooks { - test_hooks() - } fn spawn(&self, args: &[String], token: &TokenFd) -> Result> { Ok(Box::new(Prompt(start_with(&self.0, args, token)))) } @@ -1273,9 +1277,6 @@ mod process { struct Patient(PathBuf); impl Spawner for Patient { - fn hooks(&self) -> Hooks { - test_hooks() - } fn spawn(&self, args: &[String], token: &TokenFd) -> Result> { Ok(Box::new(Bounded(start_with(&self.0, args, token), std::time::Instant::now()))) } @@ -1286,15 +1287,16 @@ mod process { let dir = tempfile::tempdir().unwrap(); let lines = [at("m1", 10), at("m2", CEILING_TOKENS)].join(" "); - let result = result_line(); - // Prints two requests, waits for the flush on stdin, then the result, as a stand-in for a Self that obeys. + let (echo, result) = (echo(), ok_result()); + // Prints two requests, waits for the start and the flush on stdin, echoes both, then answers, as a stand-in for a Self that obeys. let body = format!("cat <<'EOF' {lines} EOF read -t 5 start read -t 5 flush cat <<'EOF' -{result} +{echo} +{echo} {result} EOF while read more; do :; done @@ -1305,8 +1307,10 @@ exit 0"); let outcome = carry(&Seen::of(&mind), &SystemClock, &FakeLender::new(Answer::Grant), &Spawn(program), TEST_CAP_S, &mut out).unwrap(); assert_eq!(outcome, SelfRunOutcome::Ended { cause: Cause::FlushRequested, code: Some(0), requests: 2, prefix: 300_000 }); let printed: String = String::from_utf8(out).unwrap().lines().filter(|line| !line.starts_with("decision ")).map(|line| format!("{line}\n")).collect(); - let failed = "result subtype=success is_error=true: Failed to authenticate: OAuth session expired and could not be refreshed\n"; - assert_eq!(printed, format!("request 1 prefix 10\nrequest 2 prefix 300000\n{failed}{failed}")); + assert_eq!(printed, "request 1 prefix 10 +request 2 prefix 300000 +result subtype=success is_error=false: The answer. +"); } #[test] @@ -1485,17 +1489,15 @@ fn a_run_without_claims_gets_the_identity_lines_alone() { use eureka_pipeline::PipelineResolution; -const NO_SIGNALS: &[Stop] = &[]; - fn live_mind() -> Rc { let mind = Rc::new(with_run()); - mind.committed(vec![spec("x")]); + mind.committed(vec![spec("x"), run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]); mind } -/// A brief from her Persona: a follow-up with owner `self-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 { - let PipelineDocument::FollowUp(mut document) = follow_up(label, &spec_ref("x")) else { unreachable!() }; + let PipelineDocument::FollowUp(mut document) = follow_up(label, &run_ref("persona")) else { unreachable!() }; document.item = item.into(); document.owner = inbox::OWNER.into(); PipelineDocument::FollowUp(document) @@ -1511,6 +1513,11 @@ fn admits(mind: &Rc, seconds: i64, documents: Vec) - Step::Do(seconds, Rc::new(move || mind.committed(documents.clone()))) } +/// Her Self resolves the brief herself, through her mind tool. +fn resolves(mind: &Rc, label: &str) -> Step { + admits(mind, 0, vec![close(brief_ref(label), recorded())]) +} + fn resolutions_of(mind: &TestMind, subject: &PipelineRef) -> Vec { mind.ids(K::Resolution) .iter() @@ -1521,11 +1528,6 @@ fn resolutions_of(mind: &TestMind, subject: &PipelineRef) -> Vec String { - serde_json::from_str::(&result_line()).unwrap()["result"].as_str().unwrap().to_string() -} - fn five_minute_request(id: &str, prefix: u64) -> String { let mut event: Value = serde_json::from_str(&at(id, prefix)).unwrap(); event["message"]["usage"]["cache_creation"] = json!({"ephemeral_1h_input_tokens": 0, "ephemeral_5m_input_tokens": 100}); @@ -1536,167 +1538,286 @@ fn tells(ran: &Ran) -> Vec { ran.written().into_iter().filter(|text| text.contains("A brief from your Persona")).collect() } +fn mind_run(mind: &Rc, steps: Vec) -> Ran { + carry_with(&Seen::of(mind), steps, Some(0), false) +} + #[test] fn a_result_never_ends_the_session_the_child_ends_it_or_the_cache_does() { - let ran = go(vec![Out(at("m1", 10)), Out(result_line())]); + let ran = go(vec![Out(echo()), Out(at("m1", 10)), Out(ok_result())]); assert_eq!(ran.count(&Event::InputClosed), 0, "a result is the end of a turn: {:?}", ran.events); assert_eq!(ran.ended().0, Cause::Own); } #[test] -fn a_brief_to_a_working_self_is_told_once_between_turns_and_closed_by_the_result_that_answers_it() { +fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_never_told_twice() { let mind = live_mind(); let steps = vec![ + 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.")]), - // The first turn ends: the brief was spooled while it ran, so it is told now, as the next turn. - Out(result_line()), - Idle(1), - // A later poll does not spool it again. + Out(echo()), + // Later polls do not tell it again. Idle(31), - Out(result_line()), - Idle(1), + Idle(31), + Out(ok_result()), ]; - let ran = carry_with(&Seen::of(&mind), steps, Some(0), false); + let ran = mind_run(&mind, steps); let told = tells(&ran); assert_eq!(told.len(), 1, "{:?}", ran.written()); assert!(told[0].contains(&brief_ref("b1").id.0) && told[0].contains("Survey the noise functions."), "{}", told[0]); - assert!(told[0].contains("Your answer closes 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!(ran.out.contains("decision WriteTells prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out); assert_eq!(ran.count(&Event::InputClosed), 0); - let closures = resolutions_of(&mind, &brief_ref("b1")); - assert_eq!(closures.len(), 1, "closed once"); - assert!(matches!(closures[0].outcome, ResolutionOutcome::Recorded { .. })); - assert_eq!(closures[0].rationale.0, answer_text()); - assert!(!inbox::pending(&ran.hooks.spool).unwrap(), "nothing left to tell"); + // A result is not an answer: nothing closes the brief by its timing. + assert!(resolutions_of(&mind, &brief_ref("b1")).is_empty()); } #[test] -fn a_brief_told_mid_turn_by_the_hook_is_closed_by_that_turns_result_and_never_told_twice() { +fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_until_its_echo() { let mind = live_mind(); - let hooks = test_hooks(); - let printed: Rc>> = Rc::new(RefCell::new(Vec::new())); - let fire = { - let (spool, printed) = (hooks.spool.clone(), printed.clone()); - Step::Do(0, Rc::new(move || printed.borrow_mut().push(inbox::inject(&spool, "{\"hook_event_name\":\"PostToolBatch\"}", false).unwrap().stdout))) - }; - let steps = vec![Out(at("m1", 10)), admits(&mind, 31, vec![brief("b2", "Wire the shader uniforms.")]), fire.clone(), fire, Out(result_line()), Idle(1)]; - let ran = carry_hooked(&Seen::of(&mind), steps, (Some(0), None, NO_SIGNALS, false), hooks); - let printed = printed.borrow(); - assert!(printed[0].contains(&brief_ref("b2").id.0) && printed[0].contains("Wire the shader uniforms."), "{printed:?}"); - assert_eq!(printed[1], "", "the second call finds nothing: each brief enters once"); - assert!(tells(&ran).is_empty(), "the hook told it; the actuator did not: {:?}", ran.written()); - let closures = resolutions_of(&mind, &brief_ref("b2")); - assert_eq!((closures.len(), closures[0].rationale.0.as_str()), (1, answer_text().as_str())); - // The closure is admitted under the run's own label. - let sessions: Vec = mind - .ids(K::Resolution) - .iter() - .filter_map(|resolution| mind.view(K::Resolution, resolution)) - .filter(|view| matches!(&view.document, PipelineDocument::Resolution(document) if document.subject == brief_ref("b2"))) - .map(|view| view.admission.provenance.session.0) - .collect(); - assert_eq!(sessions, vec!["selfrun-a".to_string()]); + 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 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); + assert_eq!(ran.count(&Event::InputClosed), 0); + assert!(resolutions_of(&mind, &brief_ref("b2")).is_empty()); } #[test] -fn a_brief_her_self_resolved_is_not_resolved_again_and_one_not_yet_told_stays_open_past_the_result() { +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 hooks = test_hooks(); - let claim_and_resolve = { - let (spool, mind) = (hooks.spool.clone(), mind.clone()); - Step::Do( - 0, - Rc::new(move || { - inbox::inject(&spool, "{}", false).unwrap(); - mind.committed(vec![close(brief_ref("b3"), recorded())]); - }), - ) - }; - let steps = vec![Out(at("m1", 10)), admits(&mind, 31, vec![brief("b3", "Mine.")]), claim_and_resolve, admits(&mind, 31, vec![brief("b4", "Later.")]), Out(result_line())]; - let ran = carry_hooked(&Seen::of(&mind), steps, (Some(0), None, NO_SIGNALS, false), hooks); - assert_eq!(resolutions_of(&mind, &brief_ref("b3")).len(), 1, "her own resolution stands alone"); - assert!(resolutions_of(&mind, &brief_ref("b4")).is_empty(), "told after the close: this result did not answer it"); - assert_eq!(tells(&ran).len(), 1, "b4 is told once, in the result's own pass, after the close: {:?}", ran.written()); - assert!(tells(&ran)[0].contains(&brief_ref("b4").id.0)); -} - -#[test] -fn the_past_is_closed_at_session_start_and_only_what_no_closed_run_claimed_is_told() { - let mind = TestMind::seeded(); - mind.committed(vec![spec("x")]); - mind.committed(vec![brief("p1", "Stale."), brief("p2", "Still open.")]); - mind.committed(vec![run("old", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("p1")])]); - mind.committed(vec![close(run_ref("old"), recorded())]); - mind.committed(vec![self_run("selfrun-a", "5")]); - let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); - let closures = resolutions_of(&mind, &brief_ref("p1")); - assert_eq!(closures.len(), 1); - assert!(matches!(closures[0].outcome, ResolutionOutcome::Recorded { .. })); - assert!(closures[0].rationale.0.contains(&run_ref("old").id.0), "names the run that answered it: {}", closures[0].rationale.0); + mind.committed(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("p2").id.0) && start.contains("Still open."), "{start}"); - assert!(!start.contains(&brief_ref("p1").id.0), "never told: {start}"); - // The unclaimed one was told in the start message, so the first result answers it. - assert_eq!(resolutions_of(&mind, &brief_ref("p2")).len(), 1); + assert!(start.contains("Also open for you") && start.contains(&brief_ref("p1").id.0) && start.contains("Still open."), "{start}"); + assert!(start.contains(&format!("resolve follow-up {} Recorded", brief_ref("p1").id.0)), "{start}"); + assert_eq!(ran.written().len(), 1, "never told again: {:?}", ran.written()); +} + +#[test] +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)]); + 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() { + let mind = live_mind(); + mind.committed(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_eq!(resolutions_of(&mind, &brief_ref("p2")).len(), 1, "her own resolution stands alone"); +} + +#[test] +fn a_rotation_a_ceiling_an_error_or_a_crash_leaves_the_told_briefs_open_for_the_successor() { + let cap = i64::try_from(TEST_CAP_S).unwrap(); + let ends: Vec<(&str, Vec)> = vec![ + ("rotation", vec![Out(echo()), After(cap - 1400, at("m1", 10)), Out(echo()), Out(ok_result())]), + ("ceiling", vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(echo()), Out(ok_result())]), + ("error before an answer", vec![Out(echo()), Out(result_line())]), + ("idle flush answered by an error", vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), Idle(3539), Idle(1), Out(echo()), Out(result_line())]), + ("crash", vec![Out(echo()), Out(at("m1", 10)), Exit(Some(137))]), + ]; + for (name, steps) in ends { + let mind = live_mind(); + mind.committed(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"); + } +} + +#[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 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"); + assert!(resolutions_of(&mind, &brief_ref("late")).is_empty()); } #[test] fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_brief_keeps_it_open() { - // An hour's cache: warm at 3539 s, the flush at 3540, the input closed after the flush's own result. - let hour = go(vec![Out(at("m1", 10)), Out(result_line()), Idle(3539), Idle(1), Out(result_line()), Idle(1)]); - assert_eq!(hour.written().iter().filter(|text| text.as_str() == FLUSH_MESSAGE).count(), 1); + // An hour's cache: warm at 3539 s, the flush at 3540, the input closed at the result after the flush's echo. + let hour = go(vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), Idle(3539), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]); + assert_eq!(flushes(&hour), 1); let flush_at = hour.events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains(FLUSH_MESSAGE))).unwrap(); let closed_at = hour.events.iter().position(|event| *event == Event::InputClosed).unwrap(); assert!(flush_at < closed_at && hour.count(&Event::InputClosed) == 1, "{:?}", hour.events); - assert!(hour.out.contains("decision WriteFlush prefix 10 ttl_s 3600 idle_s 3540 inbox false briefs_closed 0 -"), "{}", hour.out); + 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.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(five_minute_request("m1", 10)), Out(result_line()), Idle(239), Idle(1), Out(result_line()), 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)]); assert_eq!(flushes(&short), 1); // Not a second early. - let early = go(vec![Out(at("m1", 10)), Out(result_line()), Idle(3539)]); + let early = go(vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), Idle(3539)]); assert_eq!(flushes(&early), 0); // A brief waiting at the cache's end is told, not flushed: the session stays open. let mind = live_mind(); - let steps = vec![Out(at("m1", 10)), Out(result_line()), admits(&mind, 3539, vec![brief("b5", "One more.")]), Idle(1), Idle(1), Out(result_line()), Idle(1)]; - let ran = carry_with(&Seen::of(&mind), steps, Some(0), false); - assert_eq!((flushes(&ran), tells(&ran).len(), ran.count(&Event::InputClosed)), (0, 1, 0), "{:?}", ran.written()); + 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 ran = mind_run(&mind, steps); + assert_eq!((flushes(&ran), tells(&ran).len(), ran.count(&Event::InputClosed)), (0, 1, 0), "{:?} +{}", ran.written(), ran.out); } #[test] -fn near_the_cap_an_idle_session_closes_for_rotation_and_a_working_one_is_asked_for_its_flush() { +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 ran = mind_run(&mind, steps); + assert_eq!((tells(&ran).len(), flushes(&ran)), (1, 0), "{:?}", ran.written()); +} + +#[test] +fn near_the_cap_the_session_is_asked_for_its_flush_idle_or_working_and_closes_at_the_result_after_its_echo() { let cap = i64::try_from(TEST_CAP_S).unwrap(); - // Idle at cap - 1200 s: closed, with no flush asked. - let idle = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(199), Idle(1)]); - assert_eq!((idle.count(&Event::InputClosed), flushes(&idle)), (1, 0), "{:?}", idle.events); - let early = go(vec![After(cap - 1400, at("m1", 10)), Out(result_line()), Idle(199)]); - assert_eq!(early.count(&Event::InputClosed), 0); - // Mid-turn at cap - 600 s: the flush, once. - let working = go(vec![After(cap - 600, at("m1", 10))]); - assert_eq!(flushes(&working), 1); - let before = go(vec![After(cap - 601, at("m1", 10))]); + // Idle at cap - 1700 s: the flush at cap - 1500, not a second before. + let before = go(vec![Out(echo()), After(cap - 1700, at("m1", 10)), Out(ok_result()), Idle(199)]); assert_eq!(flushes(&before), 0); + let idle = go(vec![Out(echo()), After(cap - 1700, at("m1", 10)), Out(ok_result()), Idle(199), Idle(1), Out(echo()), Out(ok_result())]); + assert_eq!((flushes(&idle), idle.count(&Event::InputClosed)), (1, 1), "{:?}", idle.events); + assert!(idle.written().iter().any(|text| text.contains("the unit's time is nearly up")), "{:?}", idle.written()); + // Mid-turn: the flush at cap - 1500, once. + let working = go(vec![Out(echo()), After(cap - 1500, at("m1", 10))]); + assert_eq!(flushes(&working), 1); + let not_yet = go(vec![Out(echo()), After(cap - 1501, at("m1", 10))]); + assert_eq!(flushes(¬_yet), 0); } #[test] -fn a_session_starts_with_an_empty_spool_whatever_an_earlier_one_left() { - let hooks = test_hooks(); - inbox::spool(&hooks.spool, &[Brief { id: "eureka-body:follow_up:stale".into(), campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: "Stale.".into(), ordinal: 1 }]).unwrap(); - let mind = with_run(); - let ran = carry_hooked(&Seen::of(&mind), vec![Out(result_line())], (Some(0), None, NO_SIGNALS, false), hooks); - assert!(!inbox::pending(&ran.hooks.spool).unwrap(), "the stale entry is gone"); - assert_eq!(ran.written().len(), 1, "and was never told: {:?}", ran.written()); +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())]; + let ran = mind_run(&mind, steps); + let written = ran.written(); + assert_eq!(written.len(), 2, "{written:?}"); + assert!(written[1].starts_with(FLUSH_MESSAGE), "{}", written[1]); + assert_eq!(ran.spawns.len(), 1); + assert!(ran.spawns[0].iter().all(|arg| arg != "--settings"), "no hooks on argv: {:?}", ran.spawns[0]); } -#[test] -fn a_pre_compact_block_leaves_a_marker_and_the_session_asks_for_the_flush_once() { - let hooks = test_hooks(); - let fire = { - let spool = hooks.spool.clone(); - Step::Do(1, Rc::new(move || assert_eq!(inbox::inject(&spool, "{}", true).unwrap().exit, 2))) - }; - let mind = with_run(); - let ran = carry_hooked(&Seen::of(&mind), vec![Out(at("m1", 10)), fire.clone(), fire, Out(result_line())], (Some(0), None, NO_SIGNALS, false), hooks); - assert_eq!(flushes(&ran), 1, "{:?}", ran.written()); - assert!(!inbox::take_pre_compact(&ran.hooks.spool), "the marker was taken"); +/// The committed 2.1.312 streams, replayed through the real parser and the decision. +mod live_session { + use crate::self_run::decide::{Action as Do, Event as Fact, Why}; + use super::*; + + /// 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); + let mut ttl_1h = false; + let mut trace = Vec::new(); + for entry in jsonl.lines() { + let entry: Value = serde_json::from_str(entry).unwrap(); + let Some(line) = entry.get("line").and_then(Value::as_str) else { continue }; + let at = crate::testkit::at((entry["t"].as_f64().unwrap() * 1000.0) as i64); + if let Some(hour) = writes_hour_cache(line) { + ttl_1h = hour; + } + let (label, fact) = match read_usage(line) { + Usage::Request { id, prefix } => (format!("request:{prefix}"), Fact::Request { id, prefix }), + Usage::Echo => ("echo".to_string(), Fact::Echo), + Usage::Result { ok, blocked } => (format!("result:{}", if blocked { "blocked" } else if ok { "ok" } else { "error" }), Fact::Result { ok, blocked, ttl_1h }), + Usage::Other | Usage::Unreadable => ("other".to_string(), Fact::Tick), + }; + let (next, action) = decide(state, fact, at); + state = next; + trace.push((label, action)); + } + trace + } + + fn kinds(trace: &[(String, Do)]) -> Vec<&str> { + trace.iter().map(|(label, _)| label.as_str()).collect() + } + + /// The indexes at which the decision acted, and how. + fn acted(trace: &[(String, Do)]) -> Vec<(usize, Do)> { + trace.iter().enumerate().filter(|(_, (_, action))| *action != Do::None).map(|(at, (_, action))| (at, action.clone())).collect() + } + + const FOLDED: &str = include_str!("claude-2.1.312-live-session-folded.jsonl"); + const QUEUED: [&str; 2] = [include_str!("claude-2.1.312-live-session-queued.jsonl"), include_str!("claude-2.1.312-live-session-queued-scrub.jsonl")]; + const CEILING_FOLDED: &str = include_str!("claude-2.1.312-live-session-ceiling-flush-folded.jsonl"); + const CEILING_QUEUED: [&str; 2] = [include_str!("claude-2.1.312-live-session-ceiling-flush-queued.jsonl"), include_str!("claude-2.1.312-live-session-ceiling-flush-queued-scrub.jsonl")]; + const API_ERROR: [&str; 2] = [include_str!("claude-2.1.312-live-session-api-error-first.jsonl"), include_str!("claude-2.1.312-live-session-api-error-first-scrub.jsonl")]; + const BLOCKING: [&str; 2] = [include_str!("claude-2.1.312-live-session-blocking-limit.jsonl"), include_str!("claude-2.1.312-live-session-blocking-limit-scrub.jsonl")]; + const COMPACTION_OFF: [&str; 2] = [include_str!("claude-2.1.312-live-session-compaction-off-360k.jsonl"), include_str!("claude-2.1.312-live-session-compaction-off-360k-scrub.jsonl")]; + + #[test] + fn a_line_written_during_a_tool_call_is_folded_one_result_and_its_echo_before_the_request_that_carries_it() { + let trace = replay(FOLDED); + // The tool_result's own user line is no echo: it is a foreign owner's array content. + assert_eq!(kinds(&trace), ["other", "echo", "request:10", "other", "other", "other", "echo", "request:10", "result:ok"]); + assert!(acted(&trace).is_empty(), "{:?}", acted(&trace)); + } + + #[test] + fn a_line_written_while_the_last_request_is_in_flight_is_queued_two_results_and_its_echo_after_the_first() { + for stream in QUEUED { + let trace = replay(stream); + assert_eq!(kinds(&trace), ["other", "echo", "request:10", "result:ok", "other", "echo", "request:10", "result:ok"]); + assert!(acted(&trace).is_empty(), "{:?}", acted(&trace)); + } + } + + #[test] + 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"); + } + + #[test] + fn the_flush_queued_behind_the_turn_that_reached_the_ceiling_is_answered_at_its_own_second_result() { + for stream in CEILING_QUEUED { + 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"); + } + } + + #[test] + 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!(acted(&trace), vec![(3, Do::CloseStdin)]); + } + } + + #[test] + 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)]); + } + } + + #[test] + fn with_compaction_off_a_request_at_360k_is_sent_and_answered_not_compacted_or_blocked() { + 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)]); + } + } } diff --git a/crates/mind-body/src/self_run/tests/process/lease_child.rs b/crates/mind-body/src/self_run/tests/process/lease_child.rs index b5686ca..187997c 100644 --- a/crates/mind-body/src/self_run/tests/process/lease_child.rs +++ b/crates/mind-body/src/self_run/tests/process/lease_child.rs @@ -11,6 +11,7 @@ fd="$CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR" read -r -u "$fd" credential echo "fd=$fd" echo "scrub=$CLAUDE_CODE_SUBPROCESS_ENV_SCRUB" +echo "compact=$DISABLE_COMPACT" echo "argv=$*" echo "in_env=$(env | grep -c "$credential")" echo "got=$credential" @@ -26,13 +27,14 @@ fn granted_fd_reaches_child_env_var() { let dir = tempfile::tempdir().unwrap(); let program = script(&dir, READS_ITS_CREDENTIAL); let token = token_holding("tok-canary-7f3a"); - let mut child = start_with(&program, &claude_args(session_id(&run_ref("selfrun-a")), &Hooks { spool: PathBuf::from("/run/spool"), injector: PathBuf::from("/bin/x") }), &token); + let mut child = start_with(&program, &claude_args(session_id(&run_ref("selfrun-a"))), &token); child.close_input(); let (lines, code) = drain(&mut *child); assert_eq!(code, Some(0), "{lines:?}"); let number: i32 = field(&lines, "fd").parse().unwrap(); assert!(number >= 3, "never a stdio number: {number}"); assert_eq!(field(&lines, "scrub"), "1"); + assert_eq!(field(&lines, "compact"), "1", "the CLI compaction is off: the mind is the one summary"); assert_eq!(field(&lines, "got"), "tok-canary-7f3a", "the credential came through the descriptor"); assert_eq!(field(&lines, "in_env"), "0", "the token text is in no environment variable"); assert!(!field(&lines, "argv").contains("tok-canary"), "nor in the arguments"); @@ -128,11 +130,11 @@ fn each_run_gets_a_fresh_owner_only_configuration_that_what_it_writes_does_not_o #[test] fn the_spawner_needs_the_units_runtime_directory_and_reads_the_seed_from_the_units_environment() { - assert!(ClaudeSpawner::new(Vec::new(), PathBuf::from("/bin/x")).is_err(), "no scratch directory named"); - assert!(ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "".into())], PathBuf::from("/bin/x")).is_err(), "an empty one is none"); - let named = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into()), ("MIND_SELF_CONFIG_SEED".into(), "/etc/seed".into())], PathBuf::from("/bin/x")).unwrap(); + assert!(ClaudeSpawner::new(Vec::new()).is_err(), "no scratch directory named"); + assert!(ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "".into())]).is_err(), "an empty one is none"); + let named = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into()), ("MIND_SELF_CONFIG_SEED".into(), "/etc/seed".into())]).unwrap(); assert_eq!((named.scratch.as_path(), named.seed.as_deref()), (Path::new("/run/mind"), Some(Path::new("/etc/seed")))); - let bare = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into())], PathBuf::from("/bin/x")).unwrap(); + let bare = ClaudeSpawner::new([("RUNTIME_DIRECTORY".into(), "/run/mind".into())]).unwrap(); assert_eq!(bare.seed, None); } diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 7b49662..7439e66 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -207,16 +207,16 @@ mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq , PartialO mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Cause { Own , FlushRequested , Signalled (Stop) , Unstoppable , } mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Read { Line (String) , Timeout , Eof , Exited (Option < i32 >) , } mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum SelfRunOutcome { NoRun , LeaseRefused { kind : LeaseRefusalKind } , Ended { cause : Cause , code : Option < i32 > , requests : u64 , prefix : u64 } , } -mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Usage { Request { id : String , prefix : u64 } , Result , Other , Unreadable , } +mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Usage { Request { id : String , prefix : u64 } , Echo , Result { ok : bool , blocked : bool } , Other , Unreadable , } mind_body::self_run: const AGENT : & str mind_body::self_run: const CEILING_TOKENS : u64 mind_body::self_run: const ESCALATE_AFTER_S : i64 mind_body::self_run: const EXIT_DRAIN_S : i64 mind_body::self_run: const FLUSH_GRACE_S : i64 mind_body::self_run: const FLUSH_HEADROOM_TOKENS : u64 -mind_body::self_run: const FLUSH_MARGIN_S : i64 mind_body::self_run: const FLUSH_MESSAGE : & str mind_body::self_run: const INBOX_POLL_S : i64 +mind_body::self_run: const LADDER_S : i64 mind_body::self_run: const MAX_TURNS : u32 mind_body::self_run: const ROTATE_MARGIN_S : i64 mind_body::self_run: const RUN_CAP_VARIABLE : & str @@ -227,14 +227,14 @@ mind_body::self_run: const TTL_1H_S : i64 mind_body::self_run: const TTL_5M_S : i64 mind_body::self_run: const TTL_MARGIN_S : i64 mind_body::self_run: fn carry (mind : & dyn MindPort , clock : & dyn Clock , lender : & dyn Lender , spawner : & dyn Spawner , run_cap_s : u64 , out : & mut dyn Write) -> Result < SelfRunOutcome > -mind_body::self_run: fn claude_args (session : Uuid , hooks : & Hooks) -> Vec < String > +mind_body::self_run: fn claude_args (session : Uuid) -> Vec < String > mind_body::self_run: fn read_usage (line : & str) -> Usage mind_body::self_run: fn session_id (run : & PipelineRef) -> Uuid mind_body::self_run: fn start_message (instance : & Slug , run : & PipelineRef , plan_quota : & [PlanQuota] , now_ms : u64) -> String mind_body::self_run: fn writes_hour_cache (line : & str) -> Option < bool > mind_body::self_run: impl Child for ClaudeProcess mind_body::self_run: impl ClaudeSpawner -mind_body::self_run: impl ClaudeSpawner: fn new (unit_environment : impl IntoIterator < Item = (OsString , OsString) > , injector : PathBuf) -> Result < Self > +mind_body::self_run: impl ClaudeSpawner: fn new (unit_environment : impl IntoIterator < Item = (OsString , OsString) >) -> Result < Self > mind_body::self_run: impl Drop for FreshConfig mind_body::self_run: impl FreshConfig mind_body::self_run: impl Heard @@ -247,30 +247,17 @@ mind_body::self_run: impl fmt :: Display for SelfRunOutcome mind_body::self_run: impl std :: error :: Error for SeedLinkRefused mind_body::self_run: mod inbox mind_body::self_run: pub trait Child { fn write_line (& mut self , line : & str) ; fn close_input (& mut self) ; fn next (& mut self , timeout : StdDuration) -> Read ; fn stop (& mut self , how : Stop) -> bool ; } -mind_body::self_run: pub trait Spawner { fn hooks (& self) -> Hooks ; fn spawn (& self , args : & [String] , token : & TokenFd) -> Result < Box < dyn Child > > ; } +mind_body::self_run: pub trait Spawner { fn spawn (& self , args : & [String] , token : & TokenFd) -> Result < Box < dyn Child > > ; } mind_body::self_run: struct ClaudeSpawner pub-fields[] mind_body::self_run::decide: impl State mind_body::self_run::inbox: const OWNER : & str -mind_body::self_run::inbox: fn claim (spool : & Path) -> Result < Vec < Brief > > +mind_body::self_run::inbox: fn backlog (mind : & dyn MindPort) -> Result < Vec < (Brief , String) > > mind_body::self_run::inbox: fn close (mind : & dyn MindPort , now : DateTime < Utc > , session : & str , briefs : & [(Brief , String)] , reason : & str) -> Result < usize > -mind_body::self_run::inbox: fn close_past (mind : & dyn MindPort , now : DateTime < Utc > , session : & str) -> Result < usize > -mind_body::self_run::inbox: fn derive (mind : & dyn MindPort , spool : & Path) -> Result < Vec < Brief > > -mind_body::self_run::inbox: fn forget (spool : & Path , briefs : & [Brief]) -mind_body::self_run::inbox: fn inject (spool : & Path , input : & str , pre_compact : bool) -> Result < Injection > -mind_body::self_run::inbox: fn mark_told (spool : & Path , briefs : & [Brief]) -> Result < () > -mind_body::self_run::inbox: fn pending (spool : & Path) -> Result < bool > +mind_body::self_run::inbox: fn derive (mind : & dyn MindPort , told : & HashSet < String >) -> Result < Vec < Brief > > mind_body::self_run::inbox: fn render (briefs : & [Brief] , plan : & str) -> String -mind_body::self_run::inbox: fn reset (spool : & Path) -> Result < () > -mind_body::self_run::inbox: fn spool (spool : & Path , briefs : & [Brief]) -> Result < () > -mind_body::self_run::inbox: fn take_pre_compact (spool : & Path) -> bool -mind_body::self_run::inbox: fn told (spool : & Path) -> Result < Vec < Brief > > -mind_body::self_run::inbox: fn write_plan (spool : & Path , line : & str) -> Result < () > +mind_body::self_run::inbox: fn resolving_sentence (id : & str) -> String mind_body::self_run::inbox: impl Brief -mind_body::self_run::inbox: impl Hooks -mind_body::self_run::inbox: impl Hooks: fn settings (& self) -> String -mind_body::self_run::inbox: struct Brief # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[id: String, campaign: String, repo: String, item: String, ordinal: u64] -mind_body::self_run::inbox: struct Hooks # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[spool: PathBuf, injector: PathBuf] -mind_body::self_run::inbox: struct Injection # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[stdout: String, exit: i32] +mind_body::self_run::inbox: struct Brief # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[id: String, campaign: String, repo: String, item: String, ordinal: u64] mind_body::tempo: const DEFAULT_BEAT_S : u32 mind_body::tempo: const LOG_BOUND_ORDERS : f64 mind_body::tempo: fn pick (rows : & [Standing] , beat_s : u32) -> Option < usize >