diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index a9fec8a..65fe12a 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -14,14 +14,17 @@ //! //! One run is one live session (ruling `eureka-body:ruling:live-mind-sessions`). //! A result ends a turn, not the session. What her Self is asked is derived from -//! her mind (the `inbox` module): an open follow-up with owner `self-run` sourced -//! from a Persona turn is a brief. `drive` is the one writer into her session: the -//! start message, every tell and the flush are stdin lines, a tell written whenever -//! a brief waits and the flush has not been written, mid-turn or idle. The CLI -//! echoes each line it takes into a request (`--replay-user-messages`); the echo is -//! the consumption point, folded or queued, and idle is derived from the lines -//! written and the lines echoed. Her Self resolves each brief she answers; the -//! actuator closes only what she left open at an idle end. Compaction is off in +//! her mind (the `inbox` module): an open follow-up with owner `self-run` admitted +//! by her Persona turn is a brief. `drive` is the one writer into her +//! session, and keeps one write outstanding (ruling +//! `eureka-body:ruling:self-session-one-outstanding-write`): the start message, +//! then one line at a time holding the briefs that wait and the flush ask, each +//! sent only after the previous line's echo, mid-turn or idle. The CLI echoes each +//! line it takes into a request (`--replay-user-messages`) and joins lines queued +//! behind one request into a single echo, so the echo is the consumption point +//! only while no second line is in flight. Her Self resolves each brief she +//! answers; the actuator closes none, and a brief she left open is told to her +//! next session. Compaction is off in //! the CLI (`DISABLE_COMPACT=1`): the organ's own ceiling, flush, end and rehydrate //! is the only path. The session ends where its cache would be evicted, near the //! unit's cap, at the context ceiling, or after the flush each of those asks. @@ -79,6 +82,12 @@ use inbox::Brief; /// The measured Self ceiling: at this prefix the flush is asked for. pub const CEILING_TOKENS: u64 = 300_000; +/// The context window the CLI reports for the model a Self runs (`modelUsage.contextWindow` in every result of the +/// `claude-2.1.312-live-session-*` fixtures: the default model, `claude-opus-5-5`, 1M). The seed sets no model, so the CLI's default is the model. +/// The window under the real subscription is read from a result's `contextWindow` and not otherwise observed here. +pub const SELF_CONTEXT_WINDOW_TOKENS: u64 = 1_000_000; +/// What the CLI keeps free below its window before it refuses a prompt (`blocking_limit`): the blocking-limit fixture sends a 990k request and refuses the next. +pub const CLI_BLOCKING_BUFFER_TOKENS: u64 = 10_000; /// What the prefix may grow by beyond the prefix the flush was asked at before /// the actuator interrupts. Relative to the flush, never an absolute limit. pub const FLUSH_HEADROOM_TOKENS: u64 = 60_000; @@ -94,6 +103,9 @@ pub const ROTATE_MARGIN_S: i64 = 25 * 60; /// The signal ladder's length: three escalations, the stream drain and a minute of slack. The grace interrupt is never later than the cap less this. pub const LADDER_S: i64 = 3 * ESCALATE_AFTER_S + EXIT_DRAIN_S + 60; const _: () = assert!(FLUSH_GRACE_S + LADDER_S < ROTATE_MARGIN_S, "the flush and the whole ladder must finish before the unit's cap"); +/// The ceiling, and the headroom its flush's own turn may grow by, sit below the CLI's blocking limit with 600k to spare at a 1M window; a window under +/// 400k has no room for the flush's turn, and the assertion refuses to compile until the ceiling is cut to fit the window. +const _: () = assert!(CEILING_TOKENS + FLUSH_HEADROOM_TOKENS + 40_000 < SELF_CONTEXT_WINDOW_TOKENS - CLI_BLOCKING_BUFFER_TOKENS, "the ceiling and its flush headroom must fit under the CLI's blocking limit"); /// How often the mind is asked for new briefs. pub const INBOX_POLL_S: i64 = 30; /// The prompt cache's lifetimes, and how long before one runs out the session ends. @@ -234,6 +246,9 @@ pub fn claude_args(session: Uuid) -> Vec { .to_vec() } +/// The model name on the CLI's own assistant lines, which no request produced. +const SYNTHETIC_MODEL: &str = ""; + /// One line of the child's stream, as far as the actuator reads it. #[derive(Clone, Debug, PartialEq, Eq)] pub enum Usage { @@ -279,6 +294,10 @@ pub fn read_usage(line: &str) -> Usage { } Some("assistant") if event.get("parent_tool_use_id").is_none_or(Value::is_null) => { let message = event.get("message"); + // The CLI's own lines after an API error or a refused slash command carry a uuid id and no usage: no request was made. + if message.and_then(|message| message.get("model")).and_then(Value::as_str) == Some(SYNTHETIC_MODEL) { + return Usage::Other; + } let Some(id) = message.and_then(|message| message.get("id")).and_then(Value::as_str) else { return Usage::Other }; let usage = message.and_then(|message| message.get("usage")); let field = |name: &str| usage.and_then(|usage| usage.get(name)).and_then(Value::as_u64).unwrap_or(0); @@ -414,21 +433,13 @@ fn her_run(mind: &dyn MindPort) -> Result) struct Heard { /// The last result heard, ok or not: its head line and its text cut to `RATIONALE_RESULT_LIMIT`. concluded: Option<(String, String)>, - /// The text of the last ok result: what an idle end records as her answer. - answer: Option, - /// The idle flush was answered by an ok result. - idle_end: bool, } impl Heard { fn hear(&mut self, line: &str) { let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return }; if event.get("type").and_then(Value::as_str) == Some("result") { - let text = cut_text(result_text(&event), RATIONALE_RESULT_LIMIT); - if !event.get("is_error").and_then(Value::as_bool).unwrap_or(false) { - self.answer = Some(text.clone()); - } - self.concluded = Some((result_head(&event), text)); + self.concluded = Some((result_head(&event), cut_text(result_text(&event), RATIONALE_RESULT_LIMIT))); } } } @@ -521,10 +532,6 @@ Also open for you:"); } let mut heard = Heard::default(); let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &task), &mut told, &mut heard); - // An idle end answered by an ok result: what she left open is closed, saying so. After a rotation, a ceiling, an error or a crash the told briefs stay open for the successor. - if heard.idle_end && ran.is_ok() { - close_left_open(mind, clock, &run, &told, &heard, out); - } let closed = close_if_open(mind, clock, &run, &ran, &heard); if closed.is_err() { let _ = writeln!(out, "the run could not be closed: the next launch closes it"); @@ -599,20 +606,6 @@ fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> { writeln!(out, "{}", describe_result(&event, RESULT_TEXT_LIMIT)) } -/// What a brief's closing resolution says when her last answer had no text. -const NO_ANSWER_TEXT: &str = "The turn ended with no answer text."; - -/// At an idle end: resolves Recorded the told briefs she left in force, the last ok result's text as the rationale. One try; a failure is logged and the briefs stay open for the successor. -fn close_left_open(mind: &dyn MindPort, clock: &dyn Clock, run: &PipelineRef, told: &[Brief], heard: &Heard, out: &mut dyn Write) { - let rationale = heard.answer.as_deref().filter(|text| !text.is_empty()).unwrap_or(NO_ANSWER_TEXT); - let label = session_label(run); - let reason = format!("told to Self run {label}; the session went idle and ended without her resolving it"); - let answered: Vec<(Brief, String)> = told.iter().map(|brief| (brief.clone(), rationale.to_string())).collect(); - if inbox::close(mind, clock.now(), label, &answered, &reason).is_err() { - let _ = writeln!(out, "briefs could not be closed: they stay open for the next session"); - } -} - /// The loop: facts in, one clock read per fact, actions out. The one writer into her session. #[allow(clippy::too_many_arguments, reason = "the loop's ports and the run's facts, passed on whole")] fn drive( @@ -677,16 +670,21 @@ fn drive( let (ttl_s, inbox, idle_s) = (state.ttl_s, state.inbox || matches!(&heard_event, Event::Inbox { pending: true }), state.idle_s(now)); let (next, action) = decide(state, heard_event, now); state = next; - heard.idle_end = state.idle_end_answered(); if state.requests != counted { writeln!(out, "request {} prefix {}", state.requests, state.prefix)?; } match &action { Action::None => {} - Action::WriteFlush(why) => child.write_line(&user_line(&flush_text(*why))), - Action::WriteTells => { - child.write_line(&user_line(&inbox::render(&waiting, &plan))); - told.append(&mut waiting); + Action::Write { tells, flush } => { + let mut parts = Vec::new(); + if *tells { + parts.push(inbox::render(&waiting, &plan)); + told.append(&mut waiting); + } + parts.extend(flush.map(flush_text)); + child.write_line(&user_line(&parts.join(" + +"))); } Action::CloseStdin => { child.close_input(); diff --git a/crates/mind-body/src/self_run/decide.rs b/crates/mind-body/src/self_run/decide.rs index fcc6786..968641d 100644 --- a/crates/mind-body/src/self_run/decide.rs +++ b/crates/mind-body/src/self_run/decide.rs @@ -7,25 +7,29 @@ //! into events, reads the clock once per event, and executes the action. A //! timing rule is pinned here, by tables and by an enumeration of event orders. //! -//! Delivery is counted, not guessed: every stdin line the actuator writes (the -//! start message, a tell, the flush) increments `written`; every replay echo of -//! a line increments `consumed`. The echo is the point the CLI takes a line into -//! a request, folded into the running turn or queued for the next, so one rule -//! serves both. Idle is derived: no turn open and `written == consumed`. +//! Delivery is counted, not guessed, and at most one write is outstanding (ruling +//! `eureka-body:ruling:self-session-one-outstanding-write`): the CLI joins the +//! lines queued behind a request into one echo, so a second line is never +//! written before the first one's echo. Every stdin line the actuator writes (the +//! start message, a write of tells and the flush) increments `written`; every +//! replay echo increments `consumed`. The echo is the point the CLI takes a line +//! into a request, folded into the running turn or queued for the next. What +//! waits (the briefs, the flush ask) is coalesced into the next single write. +//! Idle is derived: no turn open and no write outstanding. //! //! The rules, in this order, while stdin is open and no signal is out: //! (1) a result that is blocked, or an error before any answer, closes stdin: no //! cache is worth keeping and a flush would fail the same way; (2) the first -//! result after the flush's echo closes stdin, folded or queued alike; (3) a -//! brief waiting and no flush written: tell it, idle or mid-turn; (4) the cap -//! less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (5) idle -//! with no brief waiting, at the last result's cache lifetime less a margin: ask -//! the flush (idle). The ceiling asks the same flush from a request. Compaction -//! is off in the CLI (`DISABLE_COMPACT=1`), so the flush is the only way a -//! session ends short of its cache. +//! result after the flush's echo closes stdin, folded or queued alike; (3) the +//! cap less `ROTATE_MARGIN_S`: ask the flush (rotation), idle or mid-turn; (4) +//! idle with no brief waiting, at the last result's cache lifetime less a margin: +//! ask the flush (idle); (5) with nothing outstanding, write what waits: the +//! briefs and the flush ask if there is one, as one line. The ceiling asks the +//! same flush from a request. Compaction is off in the CLI (`DISABLE_COMPACT=1`), +//! so the flush is the only way a session ends short of its cache. //! //! The headroom counts the requests after the flush's echo, from the prefix the -//! flush was written at. The grace interrupt is at the earlier of the grace +//! flush was asked at. The grace interrupt is at the earlier of the grace //! after the flush write and the cap less `LADDER_S`, so the signal ladder //! finishes before the unit's cap (ruling //! `eureka-body:ruling:flush-always-gets-a-turn`). At most one action comes out @@ -74,9 +78,8 @@ pub(super) enum Event { #[derive(Clone, Debug, PartialEq, Eq)] pub(super) enum Action { None, - WriteFlush(Why), - /// Write the waiting briefs as one user line. - WriteTells, + /// The one write: the waiting briefs and the flush ask, coalesced into one user line. + Write { tells: bool, flush: Option }, CloseStdin, /// To the child's whole process group. Signal(Stop), @@ -85,11 +88,11 @@ pub(super) enum Action { #[derive(Clone, Debug, PartialEq, Eq)] struct Flush { - /// The flush's line number: its echo is the `seq`-th. - seq: u32, - /// The prefix at which the flush was written: the headroom's base. + /// The flush's line number once written (its echo is the `seq`-th); none while it waits for the outstanding write's echo. + seq: Option, + /// The prefix at which the flush was asked: the headroom's base. prefix: u64, - written: DateTime, + asked: DateTime, /// Its echo has come: the CLI took it into a request. consumed: bool, /// A request after its echo reached the headroom. @@ -116,8 +119,6 @@ pub(super) struct State { cap: Option<(DateTime, i64)>, stdin_closed: bool, flush: Option, - /// An ok result answered the flush. - flush_answered_ok: bool, /// The last signal attempted and when. stopping: Option<(Stop, DateTime)>, /// The strongest signal delivered. @@ -125,8 +126,6 @@ pub(super) struct State { exited: Option<(Option, DateTime)>, eof: bool, finished: bool, - /// The clock at the last decision: where a due action says "now". - at: Option>, } impl State { @@ -135,19 +134,19 @@ impl State { Self { written: 1, cap: Some((at, i64::try_from(run_cap_s).unwrap_or(i64::MAX))), ..Self::default() } } + /// A line written whose echo has not come. + fn outstanding(&self) -> bool { + self.written > self.consumed + } + fn idle(&self) -> bool { - !self.turn_open && self.written == self.consumed + !self.turn_open && !self.outstanding() } /// Seconds idle since the last result, none while a line or a turn is outstanding. pub(super) fn idle_s(&self, now: DateTime) -> Option { self.idle().then(|| self.last_result_at.map_or(0, |at| (now - at).num_seconds())) } - - /// The session went idle, its cache would have lapsed, and its idle flush was answered by an ok result. - pub(super) fn idle_end_answered(&self) -> bool { - self.flush_answered_ok && self.flush.as_ref().is_some_and(|flush| flush.why == Why::Idle) - } } fn outcome(state: &State, cause: Cause, code: Option) -> SelfRunOutcome { @@ -166,28 +165,21 @@ fn cause(state: &State) -> Cause { /// When the flush's grace runs out: the grace after the write, never later than /// the point the signal ladder needs to finish before the unit's cap. fn grace_deadline(state: &State, flush: &Flush) -> DateTime { - let grace = flush.written + Duration::seconds(FLUSH_GRACE_S); + let grace = flush.asked + Duration::seconds(FLUSH_GRACE_S); match state.cap { Some((started, cap)) => grace.min(started + Duration::seconds(cap.saturating_sub(LADDER_S))), None => grace, } } -/// What the session does next on its own, and when. -enum Due { - Tells, - Flush(Why), -} - -fn due(state: &State) -> Option<(DateTime, Due)> { +/// When the flush is next asked on its own, and why: the cap's margin, or the +/// cache's end while idle with no brief waiting. +fn due(state: &State) -> Option<(DateTime, Why)> { if state.stdin_closed || state.exited.is_some() || state.flush.is_some() { return None; } - if state.inbox { - return state.at.map(|at| (at, Due::Tells)); - } - let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Due::Flush(Why::Rotation))); - let idle = state.idle().then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Due::Flush(Why::Idle))); + let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Why::Rotation)); + let idle = (state.idle() && !state.inbox).then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Why::Idle)); [rotation, idle].into_iter().flatten().min_by_key(|(at, _)| *at) } @@ -209,34 +201,48 @@ pub(super) fn next_deadline(state: &State) -> Option> { } /// The flush is asked once, by the ceiling, the cap's margin or the cache's end; -/// never to a child whose input is closed or that has exited. +/// never to a child whose input is closed or that has exited. Asking writes +/// nothing: `send` writes it when no write is outstanding. fn ask_flush(state: &mut State, now: DateTime, why: Why) -> bool { if state.flush.is_some() || state.stdin_closed || state.exited.is_some() { return false; } - state.written += 1; - state.flush = Some(Flush { seq: state.written, prefix: state.prefix, written: now, consumed: false, exhausted: false, why }); + state.flush = Some(Flush { seq: None, prefix: state.prefix, asked: now, consumed: false, exhausted: false, why }); true } -/// The lifecycle rules 3 to 5: tell a waiting brief, or ask the flush where it is due. +/// The lifecycle rules 3 to 5: ask the flush where it is due, then write what waits. fn lifecycle(state: &mut State, now: DateTime) -> Action { - match due(state) { - Some((at, Due::Tells)) if at <= now => { - state.inbox = false; - state.written += 1; - Action::WriteTells - } - Some((at, Due::Flush(why))) if at <= now && ask_flush(state, now, why) => Action::WriteFlush(why), - _ => Action::None, + if let Some((at, why)) = due(state) + && at <= now + { + ask_flush(state, now, why); } + send(state) +} + +/// With nothing outstanding, the one write: the waiting briefs (until the flush +/// is written) and the flush, if it has not been written, as one line. +fn send(state: &mut State) -> Action { + if state.stdin_closed || state.exited.is_some() || state.outstanding() || state.flush.as_ref().is_some_and(|flush| flush.seq.is_some()) { + return Action::None; + } + let flush = state.flush.as_ref().map(|flush| flush.why); + if !state.inbox && flush.is_none() { + return Action::None; + } + state.written += 1; + let seq = state.written; + if let Some(flush) = state.flush.as_mut() { + flush.seq = Some(seq); + } + Action::Write { tells: std::mem::take(&mut state.inbox), flush } } pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (State, Action) { if state.finished { return (state, Action::None); } - state.at = Some(now); let mut action = Action::None; match event { Event::Request { id, prefix } => { @@ -249,8 +255,8 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta { flush.exhausted = true; } - if prefix >= CEILING_TOKENS && ask_flush(&mut state, now, Why::Ceiling) { - action = Action::WriteFlush(Why::Ceiling); + if prefix >= CEILING_TOKENS { + ask_flush(&mut state, now, Why::Ceiling); } } } @@ -259,7 +265,7 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta state.turn_open = true; let consumed = state.consumed; if let Some(flush) = state.flush.as_mut() - && consumed >= flush.seq + && flush.seq.is_some_and(|seq| consumed >= seq) { flush.consumed = true; } @@ -271,7 +277,6 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta let no_answer = !ok && !state.answered_once; state.answered_once |= ok; let flush_answered = state.flush.as_ref().is_some_and(|flush| flush.consumed); - state.flush_answered_ok |= flush_answered && ok; if (blocked || no_answer || flush_answered) && !state.stdin_closed && state.exited.is_none() { state.stdin_closed = true; action = Action::CloseStdin; diff --git a/crates/mind-body/src/self_run/decide/tests.rs b/crates/mind-body/src/self_run/decide/tests.rs index c3160aa..8da3a5b 100644 --- a/crates/mind-body/src/self_run/decide/tests.rs +++ b/crates/mind-body/src/self_run/decide/tests.rs @@ -49,6 +49,15 @@ fn last(events: Vec<(Event, i64)>) -> Action { run(events).1.pop().unwrap() } +/// The one write with only the waiting briefs, and with only the flush. +fn tells() -> Action { + Action::Write { tells: true, flush: None } +} + +fn flush(why: Why) -> Action { + Action::Write { tells: false, flush: Some(why) } +} + fn delivered(how: Stop) -> Event { Event::Signalled { how, delivered: true } } @@ -68,13 +77,33 @@ fn idle_at(at: i64) -> (State, Vec) { #[test] fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or_for_a_repeated_request() { - assert_eq!(last(vec![(req("a", CEILING_TOKENS - 1), 0)]), Action::None); - let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 0), (req("a", CEILING_TOKENS), 1), (req("b", CEILING_TOKENS + 1), 2)]); - assert_eq!(actions, vec![Action::WriteFlush(Why::Ceiling), Action::None, Action::None]); + assert_eq!(last(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS - 1), 0)]), Action::None); + let (state, actions) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 0), (req("a", CEILING_TOKENS), 1), (req("b", CEILING_TOKENS + 1), 2)]); + assert_eq!(actions, vec![Action::None, flush(Why::Ceiling), Action::None, Action::None]); assert_eq!((state.written, state.requests), (2, 2)); assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (req("a", CEILING_TOKENS), 1)]), Action::None); } +#[test] +fn what_waits_is_one_write_sent_only_after_the_outstanding_writes_echo_the_briefs_and_the_flush_ask_together() { + // The start message is not echoed: the ceiling asks the flush and a brief waits, and nothing is written. + let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 1), (Event::Inbox { pending: true }, 2), (Event::Tick, 3)]); + assert_eq!(actions, vec![Action::None, Action::None, Action::None]); + assert_eq!(state.written, 1); + // The echo frees the writer: one write carries both, and the flush is asked only once. + let (state, action) = feed(state, Event::Echo, 4); + assert_eq!(action, Action::Write { tells: true, flush: Some(Why::Ceiling) }); + assert_eq!((state.written, state.consumed), (2, 1)); + assert_eq!(next_deadline(&state), Some(t(1 + GRACE)), "the grace runs from the ask"); + // Briefs alone: the first poll writes, the second finds that write outstanding and waits for its echo. + let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1), (Event::Inbox { pending: true }, 2)]); + assert_eq!(actions, vec![Action::None, tells(), Action::None]); + assert_eq!(state.written, 2); + let (state, action) = feed(state, Event::Echo, 3); + assert_eq!(action, tells(), "the second poll's brief is written at the echo"); + assert_eq!(state.written, 3); +} + #[test] fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every_line_was_echoed_and_no_turn_runs() { let (state, _) = run(vec![]); @@ -86,7 +115,7 @@ fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every assert_eq!(state.idle_s(t(10)), Some(3)); // A tell written after the result is outstanding until its echo: not idle. let (state, actions) = run(vec![(Event::Echo, 0), (OK, 7), (Event::Inbox { pending: true }, 8)]); - assert_eq!(actions[2], Action::WriteTells); + assert_eq!(actions[2], tells()); assert_eq!(state.idle_s(t(9)), None); // The result of a turn with a queued line outstanding is not idleness. let (state, _) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1), (OK, 2)]); @@ -96,7 +125,7 @@ fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every #[test] fn a_flush_folded_into_the_running_turn_is_answered_by_the_result_after_its_echo() { let (state, actions) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 1), (Event::Echo, 2), (OK, 3)]); - assert_eq!(actions, vec![Action::None, Action::WriteFlush(Why::Ceiling), Action::None, Action::CloseStdin]); + assert_eq!(actions, vec![Action::None, flush(Why::Ceiling), Action::None, Action::CloseStdin]); assert_eq!(next_deadline(&state), Some(t(1 + GRACE)), "closed: only the flush's grace backstop remains, no lifecycle deadline"); } @@ -135,7 +164,7 @@ fn an_error_after_an_ok_answer_ends_the_turn_and_not_the_session() { fn a_brief_waiting_is_told_mid_turn_or_idle_once_and_never_after_the_flush() { // Mid-turn: written at once. let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1)]); - assert_eq!(actions[1], Action::WriteTells); + assert_eq!(actions[1], tells()); assert_eq!(state.written, 2); // The same poll answered again with nothing new is no second tell. assert_eq!(feed(state.clone(), Event::Inbox { pending: false }, 2).1, Action::None); @@ -143,9 +172,9 @@ fn a_brief_waiting_is_told_mid_turn_or_idle_once_and_never_after_the_flush() { // Idle: the same. let (_, actions) = idle_at(5); assert!(actions.iter().all(|action| *action == Action::None)); - assert_eq!(feed(idle_at(5).0, Event::Inbox { pending: true }, 6).1, Action::WriteTells); + assert_eq!(feed(idle_at(5).0, Event::Inbox { pending: true }, 6).1, tells()); // After the flush no line is written, a brief waiting or not. - let (state, _) = run(vec![(req("a", CEILING_TOKENS), 0)]); + let (state, _) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 0)]); assert_eq!(feed(state.clone(), Event::Inbox { pending: true }, 1).1, Action::None); assert_eq!(feed(state, Event::Tick, 2).1, Action::None); } @@ -156,7 +185,7 @@ fn an_idle_session_asks_the_flush_where_the_cache_would_lapse_and_a_waiting_brie let lapse = 10 + TTL_1H_S - TTL_MARGIN_S; assert_eq!(next_deadline(&idle), Some(t(lapse))); assert_eq!(feed(idle.clone(), Event::Tick, lapse - 1).1, Action::None); - assert_eq!(feed(idle, Event::Tick, lapse).1, Action::WriteFlush(Why::Idle)); + assert_eq!(feed(idle, Event::Tick, lapse).1, flush(Why::Idle)); // A five-minute cache lapses sooner. let (short, _) = run(vec![(Event::Echo, 0), (req("a", 10), 1), (OK_SHORT, 10)]); assert_eq!(next_deadline(&short), Some(t(10 + TTL_5M_S - TTL_MARGIN_S))); @@ -164,20 +193,8 @@ fn an_idle_session_asks_the_flush_where_the_cache_would_lapse_and_a_waiting_brie let (waiting, _) = run_from(idle_at(10).0, vec![(Event::Inbox { pending: true }, 11)]); assert_eq!(feed(waiting, Event::Tick, lapse + 100).1, Action::None); // The idle flush closes after its own answer. - let (state, actions) = run_from(idle_at(10).0, vec![(Event::Tick, lapse), (Event::Echo, lapse + 1), (req("b", 11), lapse + 1), (OK, lapse + 2)]); - assert_eq!(actions, vec![Action::WriteFlush(Why::Idle), Action::None, Action::None, Action::CloseStdin]); - assert!(state.idle_end_answered()); -} - -#[test] -fn an_idle_end_answered_by_an_error_is_not_an_answered_idle_end() { - let lapse = 10 + TTL_1H_S - TTL_MARGIN_S; - let (state, actions) = run_from(idle_at(10).0, vec![(Event::Tick, lapse), (Event::Echo, lapse + 1), (FAILED, lapse + 2)]); - assert_eq!(actions[2], Action::CloseStdin); - assert!(!state.idle_end_answered()); - // A rotation or ceiling flush is never an idle end. - let (state, _) = run(vec![(req("a", CEILING_TOKENS), 0), (Event::Echo, 1), (Event::Echo, 2), (OK, 3)]); - assert!(!state.idle_end_answered()); + let (_, actions) = run_from(idle_at(10).0, vec![(Event::Tick, lapse), (Event::Echo, lapse + 1), (req("b", 11), lapse + 1), (OK, lapse + 2)]); + assert_eq!(actions, vec![flush(Why::Idle), Action::None, Action::None, Action::CloseStdin]); } #[test] @@ -189,13 +206,13 @@ fn near_the_cap_the_session_is_asked_for_its_flush_idle_or_working_and_closes_af let (state, _) = run_from(started(), vec![(Event::Echo, 0)]); assert_eq!(next_deadline(&state), Some(t(rotate))); assert_eq!(feed(state.clone(), Event::Tick, rotate - 1).1, Action::None); - assert_eq!(feed(state, Event::Tick, rotate).1, Action::WriteFlush(Why::Rotation)); + assert_eq!(feed(state, Event::Tick, rotate).1, flush(Why::Rotation)); // Idle, its cache still warm. let (idle, _) = run_from(started(), vec![(Event::Echo, 0), (OK, rotate - 100)]); - assert_eq!(feed(idle.clone(), Event::Tick, rotate).1, Action::WriteFlush(Why::Rotation)); + assert_eq!(feed(idle.clone(), Event::Tick, rotate).1, flush(Why::Rotation)); // And the close follows the answer, never a bare close. let (_, actions) = run_from(idle, vec![(Event::Tick, rotate), (Event::Echo, rotate + 1), (OK, rotate + 2)]); - assert_eq!(actions, vec![Action::WriteFlush(Why::Rotation), Action::None, Action::CloseStdin]); + assert_eq!(actions, vec![flush(Why::Rotation), Action::None, Action::CloseStdin]); } #[test] @@ -212,8 +229,8 @@ fn a_flush_is_never_asked_of_a_session_whose_input_is_closed() { } #[test] -fn the_grace_runs_from_the_flush_write_to_the_second_and_no_later_request_moves_it() { - let (state, _) = run(vec![(req("a", CEILING_TOKENS), 100), (req("b", CEILING_TOKENS + 1), 500)]); +fn the_grace_runs_from_the_flush_ask_to_the_second_and_no_later_request_moves_it() { + let (state, _) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 100), (req("b", CEILING_TOKENS + 1), 500)]); assert_eq!(next_deadline(&state), Some(t(100 + GRACE))); assert_eq!(feed(state.clone(), Event::Tick, 100 + GRACE - 1).1, Action::None); assert_eq!(feed(state, Event::Tick, 100 + GRACE).1, Action::Signal(Stop::Interrupt)); @@ -236,12 +253,12 @@ fn the_headroom_counts_the_requests_after_the_flushs_echo_from_the_prefix_of_the #[test] fn the_ladder_ends_before_the_cap_whenever_the_flush_was_written() { let cap = 5_400_u64; - let ceiling = |second: i64| run_from(State::started(t(0), cap), vec![(req("a", CEILING_TOKENS), second)]); - let rotation = |second: i64| run_from(State::started(t(0), cap), vec![(Event::Tick, second)]); + let ceiling = |second: i64| run_from(State::started(t(0), cap), vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), second)]); + let rotation = |second: i64| run_from(State::started(t(0), cap), vec![(Event::Echo, 0), (Event::Tick, second)]); // A flush written inside the last LADDER_S cannot finish its ladder: the unit's own cap is the backstop there. for second in (0..cap as i64 - LADDER_S).step_by(7) { for (state, actions) in [ceiling(second), rotation(second)] { - if !matches!(actions.last(), Some(Action::WriteFlush(_))) { + if !matches!(actions.last(), Some(Action::Write { flush: Some(_), .. })) { continue; } let mut state = state; @@ -266,18 +283,19 @@ fn the_ladder_ends_before_the_cap_whenever_the_flush_was_written() { fn a_flush_written_late_has_its_grace_clamped_so_the_ladder_still_ends_under_the_cap() { let cap = 5_400_i64; let late = cap - 100; - let (state, _) = run_from(State::started(t(0), cap as u64), vec![(req("a", CEILING_TOKENS), late)]); + let (state, _) = run_from(State::started(t(0), cap as u64), vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), late)]); assert_eq!(next_deadline(&state), Some(t(cap - LADDER_S)), "the interrupt is not the full grace after the write"); } #[test] fn the_flush_pass_never_signals_whatever_the_clock_does() { // Two events apart by more than the grace: the write is its own pass, the signal comes in the next. - let (state, action) = feed(working(), req("a", CEILING_TOKENS), 0); - assert_eq!(action, Action::WriteFlush(Why::Ceiling)); + let (state, _) = feed(working(), Event::Echo, 0); + let (state, action) = feed(state, req("a", CEILING_TOKENS), 0); + assert_eq!(action, flush(Why::Ceiling)); assert_eq!(feed(state, Event::Tick, 10 * GRACE).1, Action::Signal(Stop::Interrupt)); // A request that both crosses the ceiling and is far past the headroom writes the flush and nothing else. - assert_eq!(last(vec![(req("a", u64::MAX), 0)]), Action::WriteFlush(Why::Ceiling)); + assert_eq!(last(vec![(Event::Echo, 0), (req("a", u64::MAX), 0)]), flush(Why::Ceiling)); } #[test] @@ -388,7 +406,7 @@ const SYMBOLS: [Sym; 8] = [Sym::Request, Sym::Ceiling, Sym::Echo, Sym::Ok, Sym:: /// and the single-writer invariants. fn play(symbols: &[Sym]) { let mut state = working(); - let (mut tells, mut flushes, mut closes, mut echoes) = (0_u32, 0_u32, 0_u32, 0_u32); + let (mut writes, mut tells, mut flushes, mut closes, mut echoes) = (0_u32, 0_u32, 0_u32, 0_u32, 0_u32); let (mut now, mut unique) = (0_i64, 0_u32); let mut flushed = false; let mut closed = false; @@ -414,14 +432,14 @@ fn play(symbols: &[Sym]) { let (next, action) = feed(state, event, now); state = next; match action { - Action::WriteTells => { - assert!(!flushed && !closed, "a tell after the flush or the close: {symbols:?}"); - tells += 1; - } - Action::WriteFlush(_) => { - assert!(!flushed && !closed, "a second flush, or one after the close: {symbols:?}"); - flushed = true; - flushes += 1; + Action::Write { tells: told, flush } => { + assert!(!flushed && !closed, "a write after the flush or the close: {symbols:?}"); + writes += 1; + tells += u32::from(told); + if flush.is_some() { + flushed = true; + flushes += 1; + } } Action::CloseStdin => { assert!(!closed, "stdin closed twice: {symbols:?}"); @@ -431,7 +449,9 @@ fn play(symbols: &[Sym]) { Action::Signal(_) => assert!(flushed, "a signal with no flush: {symbols:?}"), Action::None | Action::Finish(_) => {} } - assert_eq!(state.written, 1 + tells + flushes, "every write is counted once: {symbols:?}"); + assert_eq!(state.written, 1 + writes, "every write is counted once: {symbols:?}"); + assert!(state.written <= state.consumed + 1, "two writes outstanding: {symbols:?}"); + assert!(tells <= writes && flushes <= writes); assert_eq!(state.consumed, echoes, "every echo is counted once: {symbols:?}"); assert!(closes <= 1 && flushes <= 1); } diff --git a/crates/mind-body/src/self_run/inbox.rs b/crates/mind-body/src/self_run/inbox.rs index d571395..61bad3b 100644 --- a/crates/mind-body/src/self_run/inbox.rs +++ b/crates/mind-body/src/self_run/inbox.rs @@ -2,23 +2,22 @@ //! mind and told to the live session by the actuator's one stdin writer. //! //! The mind owns what her Self is asked: an in-force follow-up with owner -//! `self-run` whose source is a Persona turn's run is a brief. A follow-up her -//! Self files for herself, or one whose source the mind does not hold, is not. -//! `derive` lists the briefs the session has not been told (the told set is the -//! actuator's process memory, a cache; the mind is the truth). Her Self resolves -//! each brief she answers; `close` resolves the ones she left open at an idle -//! end. `backlog` lists, and never closes, the in-force briefs a closed run -//! claimed: the one-time sweep at deploy. +//! `self-run` that her Persona turn admitted is a brief. Who admitted it is the +//! receipt's provenance, which the mind records and a writer cannot choose; the +//! follow-up's `source` is the writer's word and decides nothing. A follow-up her +//! Self files for herself is not a brief, whatever it cites. `derive` lists the +//! briefs the session has not been told (the told set is the actuator's process +//! memory, a cache; the mind is the truth). Her Self resolves each brief she +//! answers and the actuator closes none. `backlog` lists, and never closes, the +//! in-force briefs a closed run claimed: the one-time sweep at deploy. use std::collections::{HashMap, HashSet}; -use anyhow::{Result, anyhow}; -use chrono::{DateTime, Utc}; -use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, RunTurn}; -use huginn_mind::{PipelineAdmissionOutcome, PipelineFacts, PipelineStatusSummary}; +use anyhow::Result; +use eureka_pipeline::PipelineKind; +use huginn_mind::{PipelineFacts, PipelineStatusSummary}; -use super::AGENT; -use crate::launch::closure; +use crate::persona; use crate::queue::{MindPort, any_of, headers, in_force, of_kind}; /// The owner a follow-up names when it is a brief for her Self. @@ -36,12 +35,6 @@ pub struct Brief { pub ordinal: u64, } -impl Brief { - fn follow_up(&self) -> PipelineRef { - PipelineRef { kind: PipelineKind::FollowUp, id: self.id.as_str().into() } - } -} - /// The ids of the follow-ups a closed run claims, with the run each was claimed by. fn claimed_by_closed(mind: &dyn MindPort) -> Result> { let runs = headers(mind, &any_of(of_kind(PipelineKind::Run), "root", &[&mind.instance().0]))?; @@ -55,29 +48,19 @@ fn claimed_by_closed(mind: &dyn MindPort) -> Result> { Ok(claimed) } -/// Every in-force brief, oldest first: owner `self-run`, sourced from a run of -/// the mind whose turn is a Persona turn. +/// Every in-force brief, oldest first: owner `self-run`, admitted by her Persona +/// turn (the receipt's agent), whatever source the writer named. fn open_briefs(mind: &dyn MindPort) -> Result> { - let mut candidates: Vec<(Brief, String)> = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))? + let mut briefs: Vec = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))? .into_iter() .filter_map(|summary| match summary.facts { - PipelineFacts::FollowUp { repo, owner, item, source, .. } if owner.0 == OWNER => { + PipelineFacts::FollowUp { repo, owner, item, .. } if owner.0 == OWNER && summary.admission.provenance.agent.0 == persona::AGENT => { let campaign = summary.id.id.0.split(':').next().unwrap_or("").to_string(); - Some((Brief { id: summary.id.id.0, campaign, repo: repo.0, item: item.0, ordinal: summary.admission.ordinal }, source.id.0)) + Some(Brief { id: summary.id.id.0, campaign, repo: repo.0, item: item.0, ordinal: summary.admission.ordinal }) } _ => None, }) .collect(); - if candidates.is_empty() { - return Ok(Vec::new()); - } - let persona: HashSet = headers(mind, &any_of(of_kind(PipelineKind::Run), "root", &[&mind.instance().0]))? - .into_iter() - .filter(|run| matches!(run.facts, PipelineFacts::Run { turn: RunTurn::PersonaTurn, .. })) - .map(|run| run.id.id.0) - .collect(); - candidates.retain(|(_, source)| persona.contains(source)); - let mut briefs: Vec = candidates.into_iter().map(|(brief, _)| brief).collect(); briefs.sort_by_key(|brief| brief.ordinal); Ok(briefs) } @@ -120,23 +103,5 @@ pub fn resolving_sentence(id: &str) -> String { format!("When you answer this brief, resolve follow-up {id} Recorded with your answer as rationale; if you decline it, resolve it Withdrawn with why.") } -/// Resolves Recorded each of `briefs` that is still in force, in one admission: -/// a brief her Self resolved herself is left as it is. Returns how many. -pub fn close(mind: &dyn MindPort, now: DateTime, session: &str, briefs: &[(Brief, String)], reason: &str) -> Result { - let standing: HashSet = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))?.into_iter().map(|summary| summary.id.id.0).collect(); - let mut documents: Vec = Vec::new(); - for (brief, rationale) in briefs.iter().filter(|(brief, _)| standing.contains(&brief.id)) { - documents.push(closure(mind, &brief.follow_up(), now, ResolutionOutcome::Recorded { reason: reason.into() }, rationale)?); - } - let count = documents.len(); - if count == 0 { - return Ok(0); - } - match mind.admit(AGENT, session, documents)? { - PipelineAdmissionOutcome::Committed { .. } => Ok(count), - _ => Err(anyhow!("the briefs could not be closed")), - } -} - #[cfg(test)] mod tests; diff --git a/crates/mind-body/src/self_run/inbox/tests.rs b/crates/mind-body/src/self_run/inbox/tests.rs index f778c26..a2a161f 100644 --- a/crates/mind-body/src/self_run/inbox/tests.rs +++ b/crates/mind-body/src/self_run/inbox/tests.rs @@ -1,10 +1,12 @@ -//! The inbox over a mind in memory: what is a brief, what is told, what the -//! backlog lists, and what the idle-end close resolves. +//! The inbox over a mind in memory: what is a brief, what is told and what the +//! backlog lists. use eureka_pipeline::{PipelineDocument, PipelineKind as K, RunOperator, RunTurn}; use super::*; -use crate::testkit::{TestMind, at, close as closing, follow_up, id, recorded, reference, run, run_ref, spec}; +use crate::persona::AGENT as PERSONA; +use crate::self_run::AGENT as SELF_RUN; +use crate::testkit::{TestMind, close as closing, follow_up, id, recorded, reference, run, run_ref, spec}; fn brief_doc(label: &str, owner: &str, source: &PipelineRef) -> PipelineDocument { let PipelineDocument::FollowUp(mut document) = follow_up(label, source) else { unreachable!() }; @@ -13,6 +15,12 @@ fn brief_doc(label: &str, owner: &str, source: &PipelineRef) -> PipelineDocument PipelineDocument::FollowUp(document) } +/// The mind gains these documents under `agent`'s receipt. +fn admitted_by(mind: &TestMind, agent: &str, documents: Vec) { + let outcome = mind.admit(agent, "session-a", documents).unwrap(); + assert!(matches!(outcome, huginn_mind::PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}"); +} + fn brief_ref(label: &str) -> PipelineRef { reference(K::FollowUp, &id("follow_up", label)) } @@ -32,27 +40,37 @@ fn ids(briefs: Vec) -> Vec { } #[test] -fn a_brief_is_an_in_force_self_run_follow_up_sourced_from_a_persona_turn() { +fn a_brief_is_an_in_force_self_run_follow_up_her_persona_turn_admitted_whatever_source_it_names() { let mind = mind(); let persona = run_ref("persona"); - mind.committed(vec![ - brief_doc("from-persona", OWNER, &persona), - brief_doc("from-her-self", OWNER, &run_ref("selfrun")), - brief_doc("from-no-run", OWNER, &crate::testkit::spec_ref("x")), - brief_doc("someone-elses", "Hands", &persona), - brief_doc("answered", OWNER, &persona), - ]); + admitted_by( + &mind, + PERSONA, + vec![ + brief_doc("from-persona", OWNER, &persona), + brief_doc("persona-names-a-spec", OWNER, &crate::testkit::spec_ref("x")), + brief_doc("someone-elses", "Hands", &persona), + brief_doc("answered", OWNER, &persona), + ], + ); mind.committed(vec![closing(brief_ref("answered"), recorded())]); - assert_eq!(ids(derive(&mind, &HashSet::new()).unwrap()), vec![brief_ref("from-persona").id.0], "a follow-up her Self files for herself is never a brief"); + // Her Self names the Persona turn's run as the source, and so does a hand: the receipt says who admitted them. + admitted_by(&mind, SELF_RUN, vec![brief_doc("self-cites-persona", OWNER, &persona)]); + mind.committed(vec![brief_doc("hand-cites-persona", OWNER, &persona)]); + assert_eq!( + ids(derive(&mind, &HashSet::new()).unwrap()), + vec![brief_ref("from-persona").id.0, brief_ref("persona-names-a-spec").id.0], + "the admitting agent decides, never the source the writer chose" + ); } #[test] fn derive_leaves_out_what_the_session_was_told_and_orders_by_admission() { let mind = mind(); let persona = run_ref("persona"); - mind.committed(vec![brief_doc("first", OWNER, &persona)]); - mind.committed(vec![brief_doc("second", OWNER, &persona)]); - mind.committed(vec![brief_doc("third", OWNER, &persona)]); + admitted_by(&mind, PERSONA, vec![brief_doc("first", OWNER, &persona)]); + admitted_by(&mind, PERSONA, vec![brief_doc("second", OWNER, &persona)]); + admitted_by(&mind, PERSONA, vec![brief_doc("third", OWNER, &persona)]); let told: HashSet = [brief_ref("second").id.0].into(); assert_eq!(ids(derive(&mind, &told).unwrap()), vec![brief_ref("first").id.0, brief_ref("third").id.0]); } @@ -60,7 +78,7 @@ fn derive_leaves_out_what_the_session_was_told_and_orders_by_admission() { #[test] fn a_brief_a_closed_run_claimed_is_still_a_brief_and_no_standing_rule_closes_it() { let mind = mind(); - mind.committed(vec![brief_doc("old", OWNER, &run_ref("persona"))]); + admitted_by(&mind, PERSONA, vec![brief_doc("old", OWNER, &run_ref("persona"))]); mind.committed(vec![run("old-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("old")])]); mind.committed(vec![closing(run_ref("old-self"), recorded())]); assert_eq!(ids(derive(&mind, &HashSet::new()).unwrap()), vec![brief_ref("old").id.0], "derive does not filter by past claims"); @@ -70,7 +88,8 @@ fn a_brief_a_closed_run_claimed_is_still_a_brief_and_no_standing_rule_closes_it( fn the_backlog_lists_in_force_persona_briefs_claimed_by_closed_runs_and_admits_nothing() { let mind = mind(); let persona = run_ref("persona"); - mind.committed(vec![brief_doc("stale", OWNER, &persona), brief_doc("fresh", OWNER, &persona), brief_doc("claimed-live", OWNER, &persona), brief_doc("hers", OWNER, &run_ref("selfrun"))]); + admitted_by(&mind, PERSONA, vec![brief_doc("stale", OWNER, &persona), brief_doc("fresh", OWNER, &persona), brief_doc("claimed-live", OWNER, &persona)]); + admitted_by(&mind, SELF_RUN, vec![brief_doc("hers", OWNER, &run_ref("selfrun"))]); mind.committed(vec![run("closed-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("stale"), brief_ref("hers")])]); mind.committed(vec![closing(run_ref("closed-self"), recorded())]); mind.committed(vec![run("live-self", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed-live")])]); @@ -89,24 +108,3 @@ fn a_tell_names_each_brief_the_plan_and_how_she_resolves_it() { assert!(text.ends_with("Plan: 5h at 40%."), "{text}"); assert!(!render(&[], "").contains("brief"), "no brief, no plan, nothing"); } - -#[test] -fn closing_resolves_only_the_briefs_still_in_force_in_one_admission() { - let mind = mind(); - mind.committed(vec![brief_doc("a", OWNER, &run_ref("persona")), brief_doc("b", OWNER, &run_ref("persona"))]); - mind.committed(vec![closing(brief_ref("b"), recorded())]); - let entry = |label: &str, n: u64| Brief { id: brief_ref(label).id.0, campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: format!("Item {n}."), ordinal: n }; - let briefs = [(entry("a", 1), "The answer.".to_string()), (entry("b", 2), "Ignored.".to_string())]; - let now = at(0); - assert_eq!(close(&mind, now, "selfrun-a", &briefs, "answered").unwrap(), 1); - let rationales: Vec = mind - .ids(K::Resolution) - .iter() - .filter_map(|resolution| match mind.view(K::Resolution, resolution)?.document { - PipelineDocument::Resolution(document) if document.subject == brief_ref("a") => Some(document.rationale.0), - _ => None, - }) - .collect(); - assert_eq!(rationales, vec!["The answer.".to_string()]); - assert_eq!(close(&mind, now, "selfrun-a", &briefs[..1], "answered").unwrap(), 0, "already closed"); -} diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 9a0ea54..8bc145d 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -1495,6 +1495,18 @@ fn live_mind() -> Rc { mind } +/// The mind gains these documents as her Persona turn's admission: the provenance that makes a self-run follow-up a brief. +fn persona_commit(mind: &TestMind, documents: Vec) { + let outcome = mind.admit(crate::persona::AGENT, "persona-a", documents).unwrap(); + assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}"); +} + +/// Her Persona admits these briefs, `seconds` into the run. +fn briefed(mind: &Rc, seconds: i64, documents: Vec) -> Step { + let mind = mind.clone(); + Step::Do(seconds, Rc::new(move || persona_commit(&mind, documents.clone()))) +} + /// A brief from her Persona: a follow-up with owner `self-run`, sourced from the Persona turn's run. fn brief(label: &str, item: &str) -> PipelineDocument { let PipelineDocument::FollowUp(mut document) = follow_up(label, &run_ref("persona")) else { unreachable!() }; @@ -1556,7 +1568,7 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_ Out(echo()), Out(at("m1", 10)), // The brief arrives while the first turn runs: the next poll writes it at once, mid-turn. - admits(&mind, 31, vec![brief("b1", "Survey the noise functions.")]), + briefed(&mind, 31, vec![brief("b1", "Survey the noise functions.")]), Out(echo()), // Later polls do not tell it again. Idle(31), @@ -1569,7 +1581,7 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_ assert!(told[0].contains(&brief_ref("b1").id.0) && told[0].contains("Survey the noise functions."), "{}", told[0]); assert!(told[0].contains("resolve follow-up") && told[0].contains("Recorded"), "she is told how to close it: {}", told[0]); assert_eq!(ran.written().len(), 2, "the start message and one tell: {:?}", ran.written()); - assert!(ran.out.contains("decision WriteTells prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out); + assert!(ran.out.contains("decision Write { tells: true, flush: None } prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out); assert_eq!(ran.count(&Event::InputClosed), 0); // A result is not an answer: nothing closes the brief by its timing. assert!(resolutions_of(&mind, &brief_ref("b1")).is_empty()); @@ -1578,7 +1590,7 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_ #[test] fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_until_its_echo() { let mind = live_mind(); - let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), admits(&mind, 31, vec![brief("b2", "Wire the shader uniforms.")]), Out(echo()), Out(ok_result())]; + let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), briefed(&mind, 31, vec![brief("b2", "Wire the shader uniforms.")]), Out(echo()), Out(ok_result())]; let ran = mind_run(&mind, steps); assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written()); assert!(ran.out.contains("idle_s 31 inbox true written 2 consumed 1\n"), "idle when it was written: {}", ran.out); @@ -1589,7 +1601,7 @@ fn a_brief_derived_while_the_session_is_idle_is_written_and_the_session_is_busy_ #[test] fn a_brief_open_at_start_is_in_the_start_message_with_how_to_resolve_it_and_is_not_told_again() { let mind = live_mind(); - mind.committed(vec![brief("p1", "Still open.")]); + persona_commit(&mind, vec![brief("p1", "Still open.")]); let ran = mind_run(&mind, vec![Out(echo()), Idle(31), Idle(31)]); let start = &ran.written()[0]; assert!(start.contains("Also open for you") && start.contains(&brief_ref("p1").id.0) && start.contains("Still open."), "{start}"); @@ -1602,24 +1614,23 @@ fn a_follow_up_her_self_filed_for_herself_is_never_told() { let mind = live_mind(); let PipelineDocument::FollowUp(mut hers) = follow_up("hers", &run_ref("selfrun-a")) else { unreachable!() }; hers.owner = inbox::OWNER.into(); - mind.committed(vec![PipelineDocument::FollowUp(hers)]); + // Naming the Persona turn's run as the source does not make it hers to be told: the receipt records who admitted it. + let PipelineDocument::FollowUp(mut citing) = follow_up("citing", &run_ref("persona")) else { unreachable!() }; + citing.owner = inbox::OWNER.into(); + mind.committed(vec![PipelineDocument::FollowUp(hers), PipelineDocument::FollowUp(citing)]); let ran = mind_run(&mind, vec![Out(echo()), Idle(31)]); assert_eq!(ran.written().len(), 1, "{:?}", ran.written()); assert!(!ran.written()[0].contains("Also open"), "{}", ran.written()[0]); } #[test] -fn an_idle_end_closes_the_briefs_she_left_open_saying_so_and_leaves_the_ones_she_resolved() { +fn an_idle_end_closes_no_brief_the_ones_she_left_open_stay_open_and_her_resolutions_stand() { let mind = live_mind(); - mind.committed(vec![brief("p1", "Left open."), brief("p2", "Answered by her.")]); + persona_commit(&mind, vec![brief("p1", "Left open."), brief("p2", "Answered by her.")]); let steps = vec![Out(echo()), Out(at("m1", 10)), resolves(&mind, "p2"), Out(ok_result()), Idle(3539), Idle(1), Out(echo()), Out(ok_result())]; let ran = mind_run(&mind, steps); assert_eq!(ran.count(&Event::InputClosed), 1, "{:?}", ran.events); - let left = resolutions_of(&mind, &brief_ref("p1")); - assert_eq!(left.len(), 1); - let ResolutionOutcome::Recorded { reason } = &left[0].outcome else { panic!("{:?}", left[0].outcome) }; - assert!(reason.0.contains("selfrun-a") && reason.0.contains("went idle") && reason.0.contains("without her resolving it"), "{}", reason.0); - assert_eq!(left[0].rationale.0, "The answer.", "the last ok result's text"); + assert!(resolutions_of(&mind, &brief_ref("p1")).is_empty(), "the actuator resolved nothing she left open"); assert_eq!(resolutions_of(&mind, &brief_ref("p2")).len(), 1, "her own resolution stands alone"); } @@ -1635,7 +1646,7 @@ fn a_rotation_a_ceiling_an_error_or_a_crash_leaves_the_told_briefs_open_for_the_ ]; for (name, steps) in ends { let mind = live_mind(); - mind.committed(vec![brief("p1", "Left open.")]); + persona_commit(&mind, vec![brief("p1", "Left open.")]); let ran = mind_run(&mind, steps); assert!(ran.outcome.is_ok(), "{name}"); assert!(resolutions_of(&mind, &brief_ref("p1")).is_empty(), "{name}: the brief stays open"); @@ -1645,7 +1656,7 @@ fn a_rotation_a_ceiling_an_error_or_a_crash_leaves_the_told_briefs_open_for_the_ #[test] fn a_brief_admitted_after_the_flush_is_written_stays_in_force_and_untold() { let mind = live_mind(); - let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), admits(&mind, 31, vec![brief("late", "Too late.")]), Idle(31), Out(echo()), Out(ok_result())]; + let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), briefed(&mind, 31, vec![brief("late", "Too late.")]), Idle(31), Out(echo()), Out(ok_result())]; let ran = mind_run(&mind, steps); assert!(tells(&ran).is_empty(), "{:?}", ran.written()); assert_eq!(ran.written().len(), 2, "the start and the flush; no line after the flush"); @@ -1660,7 +1671,7 @@ fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_ let flush_at = hour.events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains(FLUSH_MESSAGE))).unwrap(); let closed_at = hour.events.iter().position(|event| *event == Event::InputClosed).unwrap(); assert!(flush_at < closed_at && hour.count(&Event::InputClosed) == 1, "{:?}", hour.events); - assert!(hour.out.contains("decision WriteFlush(Idle) prefix 10 ttl_s 3600 idle_s 3540 inbox false written 2 consumed 1\n"), "{}", hour.out); + assert!(hour.out.contains("decision Write { tells: false, flush: Some(Idle) } prefix 10 ttl_s 3600 idle_s 3540 inbox false written 2 consumed 1\n"), "{}", hour.out); assert!(hour.written().iter().any(|text| text.contains("nothing is waiting and its cache would lapse") && text.contains("Resolve each brief you answered")), "{:?}", hour.written()); // Five minutes' cache: the flush at 240 s. let short = go(vec![Out(echo()), Out(five_minute_request("m1", 10)), Out(ok_result()), Idle(239), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]); @@ -1670,7 +1681,7 @@ fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_ assert_eq!(flushes(&early), 0); // A brief waiting at the cache's end is told, not flushed: the session stays open. let mind = live_mind(); - let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), admits(&mind, 3539, vec![brief("b5", "One more.")]), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]; + let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), briefed(&mind, 3539, vec![brief("b5", "One more.")]), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]; let ran = mind_run(&mind, steps); assert_eq!((flushes(&ran), tells(&ran).len(), ran.count(&Event::InputClosed)), (0, 1, 0), "{:?} {}", ran.written(), ran.out); @@ -1679,7 +1690,7 @@ fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_ #[test] fn a_tell_that_is_not_yet_echoed_keeps_the_session_busy_so_the_cache_end_does_not_flush_over_it() { let mind = live_mind(); - let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), admits(&mind, 31, vec![brief("b6", "Pending.")]), Idle(3600)]; + let steps = vec![Out(echo()), Out(at("m1", 10)), Out(ok_result()), briefed(&mind, 31, vec![brief("b6", "Pending.")]), Idle(3600)]; let ran = mind_run(&mind, steps); assert_eq!((tells(&ran).len(), flushes(&ran)), (1, 0), "{:?}", ran.written()); } @@ -1703,8 +1714,8 @@ fn near_the_cap_the_session_is_asked_for_its_flush_idle_or_working_and_closes_at #[test] fn no_stdin_line_is_written_after_the_flush_and_nothing_but_the_writer_writes() { let mind = live_mind(); - mind.committed(vec![brief("b7", "Early.")]); - let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), admits(&mind, 31, vec![brief("b8", "After the flush.")]), Idle(31), Idle(31), Out(echo()), Out(at("m2", 11)), Out(ok_result())]; + persona_commit(&mind, vec![brief("b7", "Early.")]); + let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), briefed(&mind, 31, vec![brief("b8", "After the flush.")]), Idle(31), Idle(31), Out(echo()), Out(at("m2", 11)), Out(ok_result())]; let ran = mind_run(&mind, steps); let written = ran.written(); assert_eq!(written.len(), 2, "{written:?}"); @@ -1718,6 +1729,10 @@ mod live_session { use crate::self_run::decide::{Action as Do, Event as Fact, Why}; use super::*; + fn flush(why: Why) -> Do { + Do::Write { tells: false, flush: Some(why) } + } + /// One recorded line as the loop would hear it: a label for the stream, the decision's action. fn replay(jsonl: &str) -> Vec<(String, Do)> { let mut state = State::started(crate::testkit::at(0), 100_000); @@ -1781,7 +1796,7 @@ mod live_session { fn the_flush_folded_into_a_running_turn_is_answered_at_its_one_result() { let trace = replay(CEILING_FOLDED); assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "other", "other", "other", "echo", "request:10", "result:ok"]); - assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (8, Do::CloseStdin)], "never a grace interrupt"); + assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (8, Do::CloseStdin)], "never a grace interrupt"); } #[test] @@ -1790,7 +1805,7 @@ mod live_session { let trace = replay(stream); assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "result:ok", "other", "echo", "request:10", "result:ok"]); // The first result ends the turn the flush was queued behind: it closes nothing. - assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (7, Do::CloseStdin)], "never a grace interrupt"); + assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)], "never a grace interrupt"); } } @@ -1798,7 +1813,7 @@ mod live_session { fn an_api_error_before_any_answer_closes_the_session_at_once_with_no_flush() { for stream in API_ERROR { let trace = replay(stream); - assert_eq!(kinds(&trace), ["other", "echo", "request:0", "result:error"]); + assert_eq!(kinds(&trace), ["other", "echo", "other", "result:error"]); assert_eq!(acted(&trace), vec![(3, Do::CloseStdin)]); } } @@ -1807,8 +1822,8 @@ mod live_session { fn a_context_over_the_blocking_limit_is_a_blocked_result_that_closes_the_session() { for stream in BLOCKING { let trace = replay(stream); - assert_eq!(kinds(&trace), ["other", "echo", "request:990000", "result:ok", "other", "echo", "request:0", "result:blocked"]); - assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (7, Do::CloseStdin)]); + assert_eq!(kinds(&trace), ["other", "echo", "request:990000", "result:ok", "other", "echo", "other", "result:blocked"]); + assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)]); } } @@ -1817,7 +1832,7 @@ mod live_session { for stream in COMPACTION_OFF { let trace = replay(stream); assert_eq!(kinds(&trace), ["other", "echo", "request:360000", "result:ok", "other", "echo", "request:10", "result:ok"]); - assert_eq!(acted(&trace), vec![(2, Do::WriteFlush(Why::Ceiling)), (7, Do::CloseStdin)]); + assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)]); } } } @@ -1826,7 +1841,7 @@ mod live_session { fn a_brief_the_run_claims_is_in_the_task_once_and_not_again_under_also_open() { let mind = TestMind::seeded(); mind.committed(vec![spec("x"), run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]); - mind.committed(vec![brief("claimed", "Do the claimed thing.")]); + persona_commit(&mind, vec![brief("claimed", "Do the claimed thing.")]); mind.committed(vec![run("selfrun-a", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("claimed")])]); let ran = carry_with(&Seen::of(&mind), vec![Out(echo())], Some(0), false); let start = &ran.written()[0]; diff --git a/docs/eureka-body-map.md b/docs/eureka-body-map.md index 65199de..d353bc6 100644 --- a/docs/eureka-body-map.md +++ b/docs/eureka-body-map.md @@ -807,11 +807,16 @@ in-force `operated_by Mind` Self run of its instance, which admission keeps uniq - Not its decision: continuation (fresh only until `self-run-measure`), the brake (the unit's gate), the stop (the hold), the budget (the run's `budget_usd`, from the dial at launch), the environment, token and permissions (the unit and managed settings, gamecult-ops). -- Delivery is counted: `--replay-user-messages` makes the CLI echo each stdin line right before - the request that carries it, folded into the running turn or queued as the next, so the echo is - the consumption point and idle is `lines written == lines echoed` with no turn open. The flush - is answered at the first result after its echo, folded or queued alike; the grace interrupt - (earlier of 15 min after the write and the cap less the ladder) is the backstop. Compaction is +- Delivery is counted, with one write outstanding (ruling `self-session-one-outstanding-write`): + `--replay-user-messages` makes the CLI echo each stdin line right before the request that + carries it, folded into the running turn or queued as the next, but it joins lines queued + behind one request into a single echo. So `drive` writes the next line only after the previous + line's echo, coalescing what waits (the briefs and the flush ask) into that one line, and idle + is no write outstanding with no turn open. The flush is answered at the first result after its + echo; the grace interrupt (earlier of 15 min after the ask and the cap less the ladder) is the + backstop. A brief is an in-force `self-run` follow-up her Persona turn admitted (the receipt's + agent, not the source the writer named); the actuator closes none, and a brief left open is + told to the next session. Compaction is off in the CLI (`DISABLE_COMPACT=1` in the child environment). - With `CLAUDE_CODE_SUBPROCESS_ENV_SCRUB` set the CLI forces the permission mode to default: `--permission-mode bypassPermissions` has no effect there and only the seed's allow list decides