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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
7ffaf3a5f9
commit
780366f59a
6 changed files with 653 additions and 697 deletions
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<Action>) {
|
||||
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<Action>) {
|
||||
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<Action>) {
|
|||
(state, actions)
|
||||
}
|
||||
|
||||
fn run(events: Vec<(Event, i64)>) -> (State, Vec<Action>) {
|
||||
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<Action>) {
|
||||
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<String>,
|
||||
/// When the flush was written and the prefix it was written at.
|
||||
flush: Option<(i64, u64)>,
|
||||
flushes: u32,
|
||||
closes: u32,
|
||||
/// The flush is the turn that runs now or has run.
|
||||
boundary: bool,
|
||||
exceeded: bool,
|
||||
/// No turn is running.
|
||||
idle: bool,
|
||||
inbox: bool,
|
||||
closed: bool,
|
||||
prefix: u64,
|
||||
exited: Option<i64>,
|
||||
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<i32>),
|
||||
Eof,
|
||||
Soon,
|
||||
Bound,
|
||||
Grace,
|
||||
Cache,
|
||||
}
|
||||
|
||||
const ALPHABET: [Sym; 16] = [
|
||||
Sym::Low,
|
||||
Sym::AtCeiling,
|
||||
Sym::Huge,
|
||||
Sym::Repeat,
|
||||
Sym::Result,
|
||||
Sym::ShortResult,
|
||||
Sym::Brief,
|
||||
Sym::NoBrief,
|
||||
Sym::PreCompact,
|
||||
Sym::Exit(Some(0)),
|
||||
Sym::Exit(None),
|
||||
Sym::Eof,
|
||||
Sym::Soon,
|
||||
Sym::Bound,
|
||||
Sym::Grace,
|
||||
Sym::Cache,
|
||||
];
|
||||
|
||||
/// Which signals the operating system lets through.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum Policy {
|
||||
All,
|
||||
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<Event> = 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<Sym>, policy: Policy) {
|
||||
fn enumerate(length: usize, symbols: &mut Vec<Sym>) {
|
||||
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<Sym> = (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());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<PipelineDocument>) -> 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<Brief>) -> Vec<String> {
|
||||
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<String> = derive(&mind, spool.path()).unwrap().into_iter().map(|brief| brief.id).collect();
|
||||
assert_eq!(derived, vec![brief_ref("open").id.0], "only the open one nobody answered and nobody holds");
|
||||
// A run that is still open claims without answering: its brief is still asked.
|
||||
mind.committed(vec![run("live", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("open")])]);
|
||||
assert_eq!(derive(&mind, spool.path()).unwrap().len(), 1);
|
||||
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<String> = [brief_ref("second").id.0].into();
|
||||
assert_eq!(ids(derive(&mind, &told).unwrap()), vec![brief_ref("first").id.0, brief_ref("third").id.0]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_hook_and_the_actuator_claiming_at_once_deliver_each_entry_to_exactly_one_of_them() {
|
||||
for _ in 0..20 {
|
||||
let spool = spool_dir();
|
||||
let briefs: Vec<Brief> = (1..=60).map(entry).collect();
|
||||
spool_briefs(spool.path(), &briefs);
|
||||
let path = spool.path().to_path_buf();
|
||||
let racers: Vec<_> = (0..4)
|
||||
.map(|_| {
|
||||
let path = path.clone();
|
||||
std::thread::spawn(move || claim(&path).unwrap())
|
||||
})
|
||||
.collect();
|
||||
let mut seen: Vec<u64> = racers.into_iter().flat_map(|racer| racer.join().unwrap()).map(|brief| brief.ordinal).collect();
|
||||
seen.sort_unstable();
|
||||
assert_eq!(seen, (1..=60).collect::<Vec<u64>>(), "every entry once, none twice, none lost");
|
||||
}
|
||||
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<_>>(), 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<String> = mind
|
||||
|
|
|
|||
|
|
@ -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<Option<Scripted>>,
|
||||
spawns: RefCell<Vec<Vec<String>>>,
|
||||
refuse: bool,
|
||||
hooks: Hooks,
|
||||
}
|
||||
|
||||
impl Fake {
|
||||
fn with(child: Option<Scripted>, refuse: bool) -> Self {
|
||||
Self { child: RefCell::new(child), spawns: RefCell::new(Vec::new()), refuse, hooks: test_hooks() }
|
||||
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<Box<dyn Child>> {
|
||||
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<SelfRunOutcome>,
|
||||
out: String,
|
||||
events: Vec<Event>,
|
||||
|
|
@ -401,15 +396,13 @@ fn carry_signalled(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on: Op
|
|||
}
|
||||
|
||||
fn carry_scripted(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on: Option<Stop>, refuses: &[Stop], hangs: bool) -> Ran {
|
||||
carry_hooked(seen, steps, (code, dies_on, refuses, hangs), test_hooks())
|
||||
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<Step>, (code, dies_on, refuses, hangs): (Option<i32>, Option<Stop>, &[Stop], bool), hooks: Hooks) -> Ran {
|
||||
fn carry_one(seen: &Seen, steps: Vec<Step>, (code, dies_on, refuses, hangs): (Option<i32>, Option<Stop>, &[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<Step>, (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<chrono::Utc> {
|
||||
|
|
@ -439,7 +432,7 @@ fn go(steps: Vec<Step>) -> Ran {
|
|||
type Seed = Box<dyn Fn(&TestMind)>;
|
||||
|
||||
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::<Vec<_>>(), 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<Step> = vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", CEILING_TOKENS + 1)), Idle(0)].into();
|
||||
let steps: VecDeque<Step> = 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<Box<dyn Child>> {
|
||||
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<Box<dyn Child>> {
|
||||
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<TestMind> {
|
||||
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<TestMind>, seconds: i64, documents: Vec<PipelineDocument>) -
|
|||
Step::Do(seconds, Rc::new(move || mind.committed(documents.clone())))
|
||||
}
|
||||
|
||||
/// Her Self resolves the brief herself, through her mind tool.
|
||||
fn resolves(mind: &Rc<TestMind>, label: &str) -> Step {
|
||||
admits(mind, 0, vec![close(brief_ref(label), recorded())])
|
||||
}
|
||||
|
||||
fn resolutions_of(mind: &TestMind, subject: &PipelineRef) -> Vec<PipelineResolution> {
|
||||
mind.ids(K::Resolution)
|
||||
.iter()
|
||||
|
|
@ -1521,11 +1528,6 @@ fn resolutions_of(mind: &TestMind, subject: &PipelineRef) -> Vec<PipelineResolut
|
|||
.collect()
|
||||
}
|
||||
|
||||
/// The text of the probe's result line: what a brief's closure carries.
|
||||
fn answer_text() -> String {
|
||||
serde_json::from_str::<Value>(&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<String> {
|
|||
ran.written().into_iter().filter(|text| text.contains("A brief from your Persona")).collect()
|
||||
}
|
||||
|
||||
fn mind_run(mind: &Rc<TestMind>, steps: Vec<Step>) -> 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<RefCell<Vec<String>>> = Rc::new(RefCell::new(Vec::new()));
|
||||
let fire = {
|
||||
let (spool, printed) = (hooks.spool.clone(), printed.clone());
|
||||
Step::Do(0, Rc::new(move || printed.borrow_mut().push(inbox::inject(&spool, "{\"hook_event_name\":\"PostToolBatch\"}", false).unwrap().stdout)))
|
||||
};
|
||||
let steps = vec![Out(at("m1", 10)), admits(&mind, 31, vec![brief("b2", "Wire the shader uniforms.")]), fire.clone(), fire, Out(result_line()), Idle(1)];
|
||||
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<String> = 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<Step>)> = 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)]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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 >
|
||||
|
|
|
|||
Loading…
Reference in a new issue