diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index dfe6d50..4ba5d7d 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -1416,7 +1416,7 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { /// What a poll answers: with nothing of its own to tell (not even an untold message of this conversation, whose stagger /// has not run) and another conversation's reply waiting, the session ends at its idle and the next process serves that reply. fn polled(&self, pending: bool, own_untold: bool) -> Polled { - Polled { pending, other: false, yield_now: !pending && !own_untold && self.replies_elsewhere } + Polled { pending, other: false, other_first: false, yield_now: !pending && !own_untold && self.replies_elsewhere } } fn held(&self) -> Option { @@ -1448,7 +1448,7 @@ impl SessionInbox for PersonaInbox<'_, C, S> fn poll(&mut self) -> Result { if let Some(reason) = self.held() { self.progress.failure.get_or_insert(Failure::Held(reason)); - return Ok(Polled { pending: false, other: false, yield_now: true }); + return Ok(Polled { pending: false, other: false, other_first: false, yield_now: true }); } self.read_replies(); self.note_repo_changes(); @@ -1461,7 +1461,7 @@ impl SessionInbox for PersonaInbox<'_, C, S> let revision_ready = self.revision_ready(); let Some(oldest) = pending.first() else { return Ok(self.polled(self.retell.is_some() || revision_ready || !self.notices.is_empty() || !self.waiting_replies.is_empty(), false)) }; if Conversation::of(oldest) != self.conversation { - return Ok(Polled { pending: false, other: false, yield_now: true }); + return Ok(Polled { pending: false, other: false, other_first: false, yield_now: true }); } let untold: Vec = pending.into_iter().filter(|delivery| delivery.seq > self.told_through).take_while(|delivery| Conversation::of(delivery) == self.conversation).collect(); // A delivery's stagger runs from the poll that first saw it, so a stream of newer messages moves nobody's clock. The line diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 8dafe0e..62ee9fb 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -259,25 +259,33 @@ pub fn claude_args(session: Uuid) -> Vec { /// What a user line starts with when the CLI replays a task notification to the model. const NOTIFICATION_TAG: &str = ""; +/// The names a task-stopping call carries on the stream: the tool's own and its two aliases, each of which the CLI runs as the same stop and the +/// stream shows as the model named it (real claude 2.1.312, fake API). +const STOP_TOOLS: [&str; 3] = ["TaskStop", "KillShell", "KillBash"]; + /// 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 { - /// A request of the main thread: the message's id and the context it - /// read (input, cache read and cache write tokens). - Request { id: String, prefix: u64 }, + /// A request of the main thread: the message's id, the context it + /// read (input, cache read and cache write tokens) and the tasks a stop call (`TaskStop` or its aliases) on its line names (`task_id`, else `shell_id`). + Request { id: String, prefix: u64, stops: Vec }, /// A user line of the main thread that the CLI took into a request (a replay echo): ours, never a tool result and never the CLI's /// own replay of a task notification. Echo, /// The CLI began a turn: a `system` init line. TurnBegan, - /// A backgrounded task started, by the `tool_use_id` of the call that started it. - TaskStarted { id: String }, - /// A task notification, by the `tool_use_id` of the call that started the task. - TaskNotified { id: String }, - /// The CLI replayed a task notification as a user line, taking it into a turn; the `tool_use_id` is the task's. + /// The CLI's level of background tasks (`background_tasks_changed`): `live` the `task_id` of every entry, `main` those of entries with no + /// parent task that nobody awaits. + BackgroundTasks { live: Vec, main: Vec }, + /// A task of the main session went to the background, by `task_id`: a `task_started` with `is_backgrounded` true and no parent task or + /// awaiter, or a `task_updated` whose patch backgrounds it. + TaskBackgrounded { id: String }, + /// A task notification, by `task_id`; `stopped`: its status is `stopped`. + TaskNotified { id: String, stopped: bool }, + /// The CLI replayed a task notification as a user line, taking it into a turn; the `` is the task's. NotificationTaken { id: String }, /// A turn ended. `ok`: not an error. `blocked`: the CLI refused the context (`blocking_limit`). Result { ok: bool, blocked: bool }, @@ -307,13 +315,23 @@ pub fn read_usage(line: &str) -> Usage { ok: !event.get("is_error").and_then(Value::as_bool).unwrap_or(false), blocked: event.get("terminal_reason").and_then(Value::as_str) == Some("blocking_limit"), }, - // Task lines are read for liveness only: they name a task by its `tool_use_id` and never a conversation. + // Task lines are read for liveness only: they name a task by its `task_id` and never a conversation. Some("system") => { - let id = || event.get("tool_use_id").and_then(Value::as_str).map(String::from); + let task_id = |object: &serde_json::Map| object.get("task_id").and_then(Value::as_str).map(String::from); + // A task of the main session that nobody awaits: no parent task, `awaited` not true. + let main = |object: &serde_json::Map| object.get("parent_task_id").is_none_or(Value::is_null) && object.get("awaited").and_then(Value::as_bool) != Some(true); match event.get("subtype").and_then(Value::as_str) { Some("init") => Usage::TurnBegan, - Some("task_started") if event.get("is_backgrounded").and_then(Value::as_bool) == Some(true) => id().map_or(Usage::Other, |id| Usage::TaskStarted { id }), - Some("task_notification") => id().map_or(Usage::Other, |id| Usage::TaskNotified { id }), + Some("background_tasks_changed") => { + let entries: Vec<&serde_json::Map> = event.get("tasks").and_then(Value::as_array).into_iter().flatten().filter_map(Value::as_object).collect(); + Usage::BackgroundTasks { + live: entries.iter().copied().filter_map(task_id).collect(), + main: entries.iter().copied().filter(|entry| main(entry)).filter_map(task_id).collect(), + } + } + Some("task_started") if event.get("is_backgrounded").and_then(Value::as_bool) == Some(true) && main(&event) => task_id(&event).map_or(Usage::Other, |id| Usage::TaskBackgrounded { id }), + Some("task_updated") if event.get("patch").and_then(|patch| patch.get("is_backgrounded")).and_then(Value::as_bool) == Some(true) => task_id(&event).map_or(Usage::Other, |id| Usage::TaskBackgrounded { id }), + Some("task_notification") => task_id(&event).map_or(Usage::Other, |id| Usage::TaskNotified { id, stopped: event.get("status").and_then(Value::as_str) == Some("stopped") }), _ => Usage::Other, } } @@ -321,8 +339,8 @@ pub fn read_usage(line: &str) -> Usage { match event.get("message").and_then(|message| message.get("content")).and_then(Value::as_str) { // The CLI replays a notification it takes into a turn: it is the CLI's line, never our echo. Some(content) if content.starts_with(NOTIFICATION_TAG) => content - .split_once("") - .and_then(|(_, rest)| rest.split_once("")) + .split_once("") + .and_then(|(_, rest)| rest.split_once("")) .map_or(Usage::Other, |(id, _)| Usage::NotificationTaken { id: id.to_string() }), Some(_) => Usage::Echo, None => Usage::Other, @@ -339,7 +357,16 @@ pub fn read_usage(line: &str) -> Usage { let field = |name: &str| usage.and_then(|usage| usage.get(name)).and_then(Value::as_u64).unwrap_or(0); // Saturating: a sum that overflows is at or over the ceiling, never a panic or a wrap. let prefix = field("input_tokens").saturating_add(field("cache_read_input_tokens")).saturating_add(field("cache_creation_input_tokens")); - Usage::Request { id: id.to_string(), prefix } + // The tasks a `TaskStop` call on this line stops: the CLI's kill tells the model nothing, so the decision must know it was ours. + let stops = message + .and_then(|message| message.get("content")) + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use") && block.get("name").and_then(Value::as_str).is_some_and(|name| STOP_TOOLS.contains(&name))) + .filter_map(|block| block.get("input")?.as_object()?.get("task_id").or_else(|| block.get("input")?.get("shell_id"))?.as_str().map(String::from)) + .collect(); + Usage::Request { id: id.to_string(), prefix, stops } } _ => Usage::Other, } @@ -715,7 +742,8 @@ impl SessionInbox for SelfInbox<'_> { self.waiting = inbox::derive(self.mind, &held)?; let pending = self.waiting.iter().any(|brief| brief.conversation == self.serving); let other = self.waiting.iter().any(|brief| brief.conversation != self.serving); - Ok(Polled { pending, other, yield_now: false }) + let other_first = self.waiting.first().is_some_and(|brief| brief.conversation != self.serving); + Ok(Polled { pending, other, other_first, yield_now: false }) } /// The served conversation's waiting briefs; at a switch, those of the oldest waiting brief's conversation, which then is served and starts diff --git a/crates/mind-body/src/self_run/decide.rs b/crates/mind-body/src/self_run/decide.rs index 96f1fce..fad5dba 100644 --- a/crates/mind-body/src/self_run/decide.rs +++ b/crates/mind-body/src/self_run/decide.rs @@ -17,10 +17,13 @@ //! waits (the briefs, the flush ask) is coalesced into the next single write. //! Idleness is decided here and nowhere else (ruling //! `eureka-body:ruling:self-serves-one-conversation-at-a-time`): no turn open, -//! no write outstanding, no backgrounded task running and no task notification -//! owed the turn the CLI will give it. The switch of the conversation a session -//! serves, the cache-end flush and the yield all read that one definition. The -//! task lines carry a `tool_use_id` for liveness only: they name a task that +//! no write outstanding, the CLI's own level of background tasks empty and no +//! task notification owed the turn the CLI will give it. The level is a copy of +//! the last `background_tasks_changed`, replaced whole and never edited by an +//! edge line; a notification owes a turn by the CLI's own rule (the task was +//! backgrounded and the session did not stop it with `TaskStop`). The switch of +//! the conversation a session serves, the cache-end flush and the yield all read +//! that one definition. Task lines are keyed by `task_id` and name a task that //! runs or owes its turn, never a conversation or an answer. //! //! The rules, in this order, while stdin is open and no signal is out: @@ -64,21 +67,23 @@ pub(crate) enum Why { /// What the loop saw, already turned into a fact the decision can use. #[derive(Clone, Debug, PartialEq, Eq)] pub(super) enum Event { - /// A request of the main thread: its message id and the context it read. - Request { id: String, prefix: u64 }, + /// A request of the main thread: its message id, the context it read and the tasks a `TaskStop` call on its line names. + Request { id: String, prefix: u64, stops: Vec }, /// The CLI took one of our stdin lines into a request (its replay echo). Echo, /// A turn ended. `ok`: not an error. `blocked`: the CLI refused the context. `ttl_1h`: the last request wrote its cache for an hour, else five minutes. Result { ok: bool, blocked: bool, ttl_1h: bool }, - /// A poll of the mind: whether a brief of the conversation being served waits (`pending`), and whether a brief of another conversation does (`other`). - Inbox { pending: bool, other: bool }, + /// A poll of the mind: whether a brief of the conversation being served waits (`pending`), whether a brief of another conversation does (`other`), and whether the oldest waiting brief is another conversation's (`other_first`). + Inbox { pending: bool, other: bool, other_first: bool }, /// The CLI began a turn (its `system` init line). TurnBegan, - /// A background task started (`task_started`, `is_backgrounded` true), by its `tool_use_id`. - TaskStarted { id: String }, - /// A task notification came (`task_notification`), by its `tool_use_id`. - TaskNotified { id: String }, - /// The CLI replayed a task notification as a user line: it took that notification into a turn. Never one of our echoes. + /// The CLI's level of background tasks (`background_tasks_changed`): `live` every task in it, `main` those of the main session that nobody awaits. + BackgroundTasks { live: Vec, main: Vec }, + /// A task of the main session went to the background (`task_started` backgrounded, or a `task_updated` that backgrounded it), by `task_id`. + TaskBackgrounded { id: String }, + /// A task notification came (`task_notification`), by `task_id`; `stopped`: its status says it was stopped. + TaskNotified { id: String, stopped: bool }, + /// The CLI replayed a task notification as a user line: it took that notification into a turn, by `task_id`. Never one of our echoes. NotificationTaken { id: String }, /// `waitpid` returned: the child's exit code, none when a signal ended it. ChildExited(Option), @@ -146,12 +151,17 @@ pub(super) struct State { /// Turns opened so far, and who began the open (or last) one. turns: u32, turn_by: TurnBy, - /// Backgrounded tasks started and not yet notified, by `tool_use_id`: liveness only. - tasks: BTreeSet, - /// Notifications of those tasks that have not yet been given their turn. + /// The CLI's last level of background tasks, by `task_id`: replaced whole at each `background_tasks_changed`, never edited by an edge line. + running: BTreeSet, + /// Tasks the main session ever backgrounded (the CLI's own SDK consumer's rule), by `task_id`: a monotone cache of the stream. + backgrounded: BTreeSet, + /// Tasks a `TaskStop` call of ours named: the CLI's stop tells the model nothing, so such a notification owes no turn. + stopped_by_us: BTreeSet, + /// Notifications of backgrounded tasks that have not yet been given their turn, by `task_id`. owed: BTreeMap, - /// A brief of another conversation than the one served waits. + /// A brief of another conversation than the one served waits, and whether the oldest waiting brief is another conversation's. other: bool, + other_first: bool, /// An ok result has come: the session holds an answer worth its cache. answered_once: bool, last_result_at: Option>, @@ -184,9 +194,9 @@ impl State { self.written > self.consumed } - /// The one definition of idleness: no turn open, no write outstanding, no backgrounded task running and no notification owed a turn. + /// The one definition of idleness: no turn open, no write outstanding, the CLI's level of background tasks empty and no notification owed a turn. fn idle(&self) -> bool { - !self.turn_open && !self.outstanding() && self.tasks.is_empty() && self.owed.is_empty() + !self.turn_open && !self.outstanding() && self.running.is_empty() && self.owed.is_empty() } /// A turn opens at the first turn line, echo or request while none is open; the first echo (ours) or request (the CLI's) says who began it. @@ -293,8 +303,9 @@ fn send(state: &mut State) -> Action { return Action::None; } let flush = state.flush.as_ref().map(|flush| flush.why); - // The served conversation's briefs go whenever nothing is outstanding; another conversation's only at idle, with no flush asked: the switch. - let switch = !state.inbox && state.other && flush.is_none() && state.idle(); + // The served conversation's briefs go whenever nothing is outstanding; another conversation's only at idle, with no flush asked, and then + // the oldest waiting brief first: the switch. + let switch = state.other && flush.is_none() && state.idle() && (!state.inbox || state.other_first); if !state.inbox && !switch && flush.is_none() { return Action::None; } @@ -304,7 +315,7 @@ fn send(state: &mut State) -> Action { flush.seq = Some(seq); } if switch { - state.other = false; + (state.inbox, state.other, state.other_first) = (false, false, false); } Action::Write { tells: std::mem::take(&mut state.inbox) || switch, switch, flush } } @@ -316,12 +327,17 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta let mut action = Action::None; match event { Event::TurnBegan => state.open_turn(None), - Event::TaskStarted { id } => { - state.tasks.insert(id); + Event::BackgroundTasks { live, main } => { + state.running = live.into_iter().collect(); + state.backgrounded.extend(main); } - // A task that was not started backgrounded (a foreground or unknown one) owes nothing: there is nothing to wait on. - Event::TaskNotified { id } => { - if state.tasks.remove(&id) { + Event::TaskBackgrounded { id } => { + state.backgrounded.insert(id); + } + // The CLI's rule: a backgrounded task's notification is given a turn, except that of a stop the model asked for itself. A foreground, + // subagent-owned, awaited or unknown task owes nothing; a stop the CLI made on its own (a deadline, memory pressure) owes its turn. + Event::TaskNotified { id, stopped } => { + if state.backgrounded.contains(&id) && !(stopped && state.stopped_by_us.contains(&id)) { let since = state.turns; state.owed.insert(id, Owed { since, taken: false }); } @@ -331,8 +347,9 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta owed.taken = true; } } - Event::Request { id, prefix } => { + Event::Request { id, prefix, stops } => { state.open_turn(Some(TurnBy::Cli)); + state.stopped_by_us.extend(stops); if state.seen.insert(id) { state.requests += 1; state.prefix = prefix; @@ -373,9 +390,10 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta action = Action::CloseStdin; } } - Event::Inbox { pending, other } => { + Event::Inbox { pending, other, other_first } => { state.inbox = pending; state.other = other; + state.other_first = other_first; } Event::Yield => state.yielding = true, Event::ChildExited(code) => { diff --git a/crates/mind-body/src/self_run/decide/tests.rs b/crates/mind-body/src/self_run/decide/tests.rs index 8647155..5199f0d 100644 --- a/crates/mind-body/src/self_run/decide/tests.rs +++ b/crates/mind-body/src/self_run/decide/tests.rs @@ -23,7 +23,7 @@ fn t(seconds: i64) -> DateTime { } fn req(id: &str, prefix: u64) -> Event { - Event::Request { id: id.to_string(), prefix } + Event::Request { id: id.to_string(), prefix, stops: vec![] } } fn feed(state: State, event: Event, seconds: i64) -> (State, Action) { @@ -87,7 +87,7 @@ fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or #[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, other: false }, 2), (Event::Tick, 3)]); + let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 1), (Event::Inbox { pending: true, other: false, other_first: false }, 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. @@ -96,7 +96,7 @@ fn what_waits_is_one_write_sent_only_after_the_outstanding_writes_echo_the_brief 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, other: false }, 1), (Event::Inbox { pending: true, other: false }, 2)]); + let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true, other: false, other_first: false }, 1), (Event::Inbox { pending: true, other: false, other_first: false }, 2)]); assert_eq!(actions, vec![Action::None, tells(), Action::None]); assert_eq!(state.written, 2); let (state, action) = feed(state, Event::Echo, 3); @@ -114,11 +114,11 @@ fn the_start_message_is_one_line_written_and_the_session_is_idle_only_when_every 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, other: false }, 8)]); + let (state, actions) = run(vec![(Event::Echo, 0), (OK, 7), (Event::Inbox { pending: true, other: false, other_first: false }, 8)]); 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, other: false }, 1), (OK, 2)]); + let (state, _) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true, other: false, other_first: false }, 1), (OK, 2)]); assert_eq!(state.idle_s(t(3)), None, "the tell written at 1 has not been echoed"); } @@ -154,7 +154,7 @@ fn an_error_result_before_any_answer_or_a_blocked_one_closes_stdin_at_once_with_ #[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, other: false }, 2), (Event::Echo, 3), (FAILED, 4)]); + let (state, actions) = run(vec![(Event::Echo, 0), (OK, 1), (Event::Inbox { pending: true, other: false, other_first: false }, 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)); @@ -163,19 +163,19 @@ fn an_error_after_an_ok_answer_ends_the_turn_and_not_the_session() { #[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, other: false }, 1)]); + let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true, other: false, other_first: false }, 1)]); 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, other: false }, 2).1, Action::None); + assert_eq!(feed(state.clone(), Event::Inbox { pending: false, other: false, other_first: 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, other: false }, 6).1, tells()); + assert_eq!(feed(idle_at(5).0, Event::Inbox { pending: true, other: false, other_first: false }, 6).1, tells()); // After the flush no line is written, a brief waiting or not. let (state, _) = run(vec![(Event::Echo, 0), (req("a", CEILING_TOKENS), 0)]); - assert_eq!(feed(state.clone(), Event::Inbox { pending: true, other: false }, 1).1, Action::None); + assert_eq!(feed(state.clone(), Event::Inbox { pending: true, other: false, other_first: false }, 1).1, Action::None); assert_eq!(feed(state, Event::Tick, 2).1, Action::None); } @@ -190,7 +190,7 @@ fn an_idle_session_asks_the_flush_where_the_cache_would_lapse_and_a_waiting_brie 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, other: false }, 11)]); + let (waiting, _) = run_from(idle_at(10).0, vec![(Event::Inbox { pending: true, other: false, other_first: false }, 11)]); assert_eq!(feed(waiting, Event::Tick, lapse + 100).1, Action::None); // The idle flush closes after its own answer. let (_, actions) = run_from(idle_at(10).0, vec![(Event::Tick, lapse), (Event::Echo, lapse + 1), (req("b", 11), lapse + 1), (OK, lapse + 2)]); @@ -426,8 +426,8 @@ fn play(symbols: &[Sym]) { } Sym::Ok => OK, Sym::Failed => FAILED, - Sym::Waiting => Event::Inbox { pending: true, other: false }, - Sym::Nothing => Event::Inbox { pending: false, other: false }, + Sym::Waiting => Event::Inbox { pending: true, other: false, other_first: false }, + Sym::Nothing => Event::Inbox { pending: false, other: false, other_first: false }, Sym::Other => other_waits(), Sym::Late => Event::Tick, }; @@ -503,7 +503,7 @@ fn nothing_is_written_to_a_session_after_it_was_told_to_yield() { let (state, _) = idle_at(5); let (state, closed) = feed(state, Event::Yield, 5); assert_eq!(closed, Action::CloseStdin); - assert_eq!(feed(state, Event::Inbox { pending: true, other: false }, 6).1, Action::None, "closed: text that waits is told to the next session"); + assert_eq!(feed(state, Event::Inbox { pending: true, other: false, other_first: false }, 6).1, Action::None, "closed: text that waits is told to the next session"); } #[test] @@ -514,12 +514,27 @@ fn a_flush_already_written_ends_the_session_by_itself_and_a_yield_does_not_cut_i assert_eq!(action, Action::None, "the flush's own answer closes stdin"); } -fn started(id: &str) -> Event { - Event::TaskStarted { id: id.to_string() } +/// The CLI's level after a background task of the main session came in: it is in the level and the session backgrounded it. +fn level(ids: &[&str]) -> Event { + let live: Vec = ids.iter().map(|id| id.to_string()).collect(); + Event::BackgroundTasks { main: live.clone(), live } +} + +fn backgrounded(id: &str) -> Event { + Event::TaskBackgrounded { id: id.to_string() } } fn notified(id: &str) -> Event { - Event::TaskNotified { id: id.to_string() } + Event::TaskNotified { id: id.to_string(), stopped: false } +} + +fn stopped(id: &str) -> Event { + Event::TaskNotified { id: id.to_string(), stopped: true } +} + +/// A request whose line names a `TaskStop` call on these tasks. +fn stops_of(id: &str, tasks: &[&str]) -> Event { + Event::Request { id: id.to_string(), prefix: 10, stops: tasks.iter().map(|task| task.to_string()).collect() } } fn taken(id: &str) -> Event { @@ -531,7 +546,7 @@ fn switches() -> Action { } fn other_waits() -> Event { - Event::Inbox { pending: false, other: true } + Event::Inbox { pending: false, other: true, other_first: false } } /// The start message echoed and answered: a Self turn of ours is over, at second 2. @@ -555,27 +570,37 @@ fn a_turn_opens_at_the_first_turn_line_echo_or_request_and_the_first_echo_or_req assert_eq!(by([answered_start(), vec![(Event::TurnBegan, 3), (Event::Echo, 3)]].concat()), (2, TurnBy::Us)); } +/// A task of the main session backgrounded in a turn: the CLI's level holds it from the second line on. +fn task_in_level(id: &str, second: i64) -> Vec<(Event, i64)> { + vec![(backgrounded(id), second), (level(&[id]), second)] +} + +/// Its notification and the empty level the CLI says after it. +fn task_ends(id: &str, second: i64) -> Vec<(Event, i64)> { + vec![(notified(id), second), (level(&[]), second)] +} + #[test] -fn a_session_is_idle_only_with_no_turn_no_write_no_running_task_and_no_notification_owed_its_turn() { +fn a_session_is_idle_only_with_no_turn_no_write_an_empty_level_and_no_notification_owed_its_turn() { let idle = |events: Vec<(Event, i64)>| run(events).0.idle_s(t(1_000)).is_some(); + let turn = |tail: Vec<(Event, i64)>| [vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1)], tail].concat(); assert!(!idle(vec![(Event::TurnBegan, 0)]), "a turn is open from its init line"); assert!(idle(answered_start())); - // A backgrounded task running keeps it busy after the result of the turn that started it. - assert!(!idle([answered_start(), vec![(started("t"), 2)]].concat())); - let with_task = |rest: Vec<(Event, i64)>| [vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (started("t"), 1), (OK, 2)], rest].concat(); + // The level holds the session after the result of the turn that started the task. + assert!(!idle([answered_start(), vec![(level(&["t"]), 2)]].concat()), "a task in the level, whatever started it"); + let with_task = |rest: Vec<(Event, i64)>| turn([task_in_level("t", 1), vec![(OK, 2)], rest].concat()); assert!(!idle(with_task(vec![])), "the task still runs"); - // Its notification is owed a turn: not idle, until the CLI's own turn ends. - assert!(!idle(with_task(vec![(notified("t"), 3)])), "owed its turn"); - assert!(!idle(with_task(vec![(notified("t"), 3), (Event::TurnBegan, 3), (req("b", 10), 3)])), "the notification turn runs"); - assert!(idle(with_task(vec![(notified("t"), 3), (Event::TurnBegan, 3), (req("b", 10), 3), (OK, 4)])), "the CLI's turn ended"); - // A notification of a task never started in the background owes nothing, and neither does a repeat of one already given its turn. + // Its notification is owed a turn even when the level has already emptied: not idle until the CLI's own turn ends. + assert!(!idle(with_task(task_ends("t", 3))), "owed its turn, the level empty"); + assert!(!idle(with_task([task_ends("t", 3), vec![(Event::TurnBegan, 3), (req("b", 10), 3)]].concat())), "the notification turn runs"); + assert!(idle(with_task([task_ends("t", 3), vec![(Event::TurnBegan, 3), (req("b", 10), 3), (OK, 4)]].concat())), "the CLI's turn ended"); + // A notification of a task never backgrounded (a foreground or unknown one) owes nothing. assert!(idle([answered_start(), vec![(notified("unseen"), 3)]].concat())); - assert!(idle(with_task(vec![(notified("t"), 3), (Event::TurnBegan, 3), (req("b", 10), 3), (OK, 4), (notified("t"), 5)]))); // A notification that arrives inside the turn that started the task is not settled by that turn's result. - let inside = vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (started("t"), 1), (notified("t"), 1), (OK, 2)]; + let inside = turn([task_in_level("t", 1), task_ends("t", 1), vec![(OK, 2)]].concat()); assert!(!idle(inside.clone()), "the result of the starting turn leaves the notification owed"); // A notification that arrives inside a turn the CLI began is not settled by that turn's own result: its own turn is still to come. - let nested = vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (started("t"), 1), (started("u"), 1), (notified("t"), 1), (OK, 2), (Event::TurnBegan, 3), (req("b", 10), 3), (notified("u"), 3), (OK, 4)]; + let nested = turn([task_in_level("t", 1), task_in_level("u", 1), task_ends("t", 1), vec![(OK, 2), (Event::TurnBegan, 3), (req("b", 10), 3), (notified("u"), 3), (level(&[]), 3), (OK, 4)]].concat()); assert!(!idle(nested.clone()), "the first notification is settled by the turn the CLI began, the second arrived inside it"); assert!(idle([nested, vec![(Event::TurnBegan, 5), (req("c", 10), 5), (OK, 6)]].concat()), "the CLI's next turn settles it"); // A turn our line began settles none it did not replay: the hold lasts to the rotation flush. @@ -585,33 +610,69 @@ fn a_session_is_idle_only_with_no_turn_no_write_no_running_task_and_no_notificat assert!(idle([inside, vec![(Event::TurnBegan, 3), (req("b", 10), 3), (OK, 4)]].concat()), "a turn the CLI began after it came"); } +#[test] +fn the_level_is_replaced_whole_and_an_edge_line_never_edits_it() { + let running = |events: Vec<(Event, i64)>| run(events).0.running.into_iter().collect::>(); + assert_eq!(running(vec![(level(&["t", "u"]), 1)]), ["t", "u"]); + assert_eq!(running(vec![(level(&["t", "u"]), 1), (level(&["u"]), 2)]), ["u"], "a second level drops the ids absent from it"); + assert!(running(vec![(level(&["t"]), 1), (level(&[]), 2)]).is_empty(), "an empty level empties it"); + // The edges: a start, an update and a notification say nothing of what runs. + assert!(running(vec![(backgrounded("t"), 1), (notified("t"), 2), (stopped("t"), 3)]).is_empty(), "nothing enters the level by an edge"); + assert_eq!(running(vec![(level(&["t"]), 1), (backgrounded("u"), 2), (notified("t"), 3), (stopped("u"), 3)]), ["t"], "nothing leaves it by an edge"); + // A subagent's or an awaited task is in the level (it holds idleness) but is not the main session's: it is never backgrounded. + let (state, _) = run(vec![(Event::BackgroundTasks { live: vec!["t".into(), "s".into()], main: vec!["t".into()] }, 1)]); + assert_eq!(state.running.len(), 2); + assert_eq!(state.backgrounded.into_iter().collect::>(), ["t"]); +} + +#[test] +fn a_notification_owes_a_turn_when_the_task_was_backgrounded_unless_the_session_stopped_it() { + let owes = |events: Vec<(Event, i64)>| !run([answered_start(), events].concat()).0.owed.is_empty(); + assert!(owes(vec![(backgrounded("t"), 3), (notified("t"), 4)]), "a backgrounded task's notification owes its turn"); + assert!(owes(vec![(level(&["t"]), 3), (notified("t"), 4)]), "the level's main entry is backgrounded"); + assert!(!owes(vec![(notified("t"), 4)]), "a task never backgrounded (foreground, or unknown) owes nothing"); + assert!(!owes(vec![(Event::BackgroundTasks { live: vec!["s".into()], main: vec![] }, 3), (notified("s"), 4)]), "a subagent's or an awaited task owes nothing"); + // A stop of ours: the model was told nothing but the call's result, so no turn follows. + assert!(!owes(vec![(backgrounded("t"), 3), (stops_of("b", &["t"]), 3), (stopped("t"), 4)]), "a stop the session asked for owes nothing"); + // A stop the CLI made on its own, a stop of another task, and a task that finished before our stop took hold, owe their turns. + assert!(owes(vec![(backgrounded("t"), 3), (stopped("t"), 4)]), "a stop the session did not ask for owes its turn"); + assert!(owes(vec![(backgrounded("t"), 3), (stops_of("b", &["other"]), 3), (stopped("t"), 4)])); + assert!(owes(vec![(backgrounded("t"), 3), (stops_of("b", &["t"]), 3), (notified("t"), 4)]), "a task that completed is told in its own turn: only a status of stopped is ours"); + // The stop is read from any line of the turn, including a request line already seen. + let (state, _) = run(vec![(req("a", 10), 1), (stops_of("a", &["t"]), 1)]); + assert!(state.stopped_by_us.contains("t")); +} + #[test] fn another_conversations_brief_is_written_only_at_idle_with_the_switch_and_never_with_the_flush() { let after = |events: Vec<(Event, i64)>| last([events, vec![(other_waits(), 5)]].concat()); assert_eq!(after(answered_start()), switches()); assert_eq!(after(vec![(Event::TurnBegan, 0), (Event::Echo, 0)]), Action::None, "a turn runs"); - assert_eq!(after(vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (started("t"), 1), (OK, 2)]), Action::None, "a task runs"); - assert_eq!(after(vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (started("t"), 1), (notified("t"), 1), (OK, 2)]), Action::None, "a notification is owed"); + assert_eq!(after(vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (level(&["t"]), 1), (OK, 2)]), Action::None, "a task is in the level"); + assert_eq!(after(vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (backgrounded("t"), 1), (level(&["t"]), 1), (notified("t"), 1), (level(&[]), 1), (OK, 2)]), Action::None, "a notification is owed"); assert_eq!(after(vec![(Event::TurnBegan, 0)]), Action::None); assert_eq!(last(vec![(Event::TurnBegan, 0), (other_waits(), 1)]), Action::None, "the start message is not echoed"); - // The served conversation's brief goes at any time with nothing outstanding, and wins over a waiting other: no switch. + // The served conversation's brief goes at any time with nothing outstanding while a turn runs: no switch, even when the other is older. + let both = |other_first: bool| Event::Inbox { pending: true, other: true, other_first }; let (state, _) = run(vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1)]); - assert_eq!(feed(state, Event::Inbox { pending: true, other: true }, 2).1, tells(), "mid-turn: the served conversation's brief, not a switch"); - assert_eq!(last([answered_start(), vec![(Event::Inbox { pending: true, other: true }, 5)]].concat()), tells()); + assert_eq!(feed(state, both(true), 2).1, tells(), "mid-turn: the served conversation's brief, not a switch"); + assert_eq!(last([answered_start(), vec![(both(false), 5)]].concat()), tells(), "at idle with the served brief the oldest: no switch"); + // At idle the oldest waiting brief goes first: another conversation's older brief is the switch, and the served one's waits for the next poll. + let (state, actions) = run_from(run(answered_start()).0, vec![(both(true), 5)]); + assert_eq!(actions, vec![switches()]); + assert!(!state.other && !state.inbox && !state.other_first && state.written == 2, "a switch clears what the poll said: the next poll restates it"); // At idle with the flush asked, the write is the flush alone. let rotation = i64::try_from(CAP_S).unwrap() - ROTATE_MARGIN_S; assert_eq!(last([answered_start(), vec![(other_waits(), rotation)]].concat()), flush(Why::Rotation), "the rotation flush never switches"); - // A switch takes the other waiting brief with it: the next poll finds it told. - let (state, _) = run([answered_start(), vec![(other_waits(), 5)]].concat()); - assert!(!state.other && state.written == 2); + assert_eq!(last([answered_start(), vec![(both(true), rotation)]].concat()), Action::Write { tells: true, switch: false, flush: Some(Why::Rotation) }, "the flush carries the served conversation's brief, never the other's: no switch"); } #[test] -fn the_cache_end_flush_waits_for_a_running_background_task_and_its_timer_runs_from_the_notification_turns_result() { +fn the_cache_end_flush_waits_for_a_task_in_the_level_and_its_timer_runs_from_the_notification_turns_result() { let ttl = TTL_1H_S - TTL_MARGIN_S; - let running = vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (started("t"), 1), (OK, 2)]; + let running = [vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1)], task_in_level("t", 1), vec![(OK, 2)]].concat(); assert_eq!(last([running.clone(), vec![(Event::Tick, 2 + ttl + 100)]].concat()), Action::None, "a task runs: the cache end does not close the session"); - let notified_turn = vec![(notified("t"), 3), (Event::TurnBegan, 3), (req("b", 10), 3), (OK, 4000)]; + let notified_turn = [task_ends("t", 3), vec![(Event::TurnBegan, 3), (req("b", 10), 3), (OK, 4000)]].concat(); let (state, _) = run([running.clone(), notified_turn.clone()].concat()); assert_eq!(next_deadline(&state), Some(t(4000 + ttl).min(t(i64::try_from(CAP_S).unwrap() - ROTATE_MARGIN_S)))); assert_eq!(last([running.clone(), notified_turn.clone(), vec![(Event::Tick, 2 + ttl + 100)]].concat()), Action::None, "the timer runs from the notification turn's result, not the first"); diff --git a/crates/mind-body/src/self_run/probe/driver.py b/crates/mind-body/src/self_run/probe/driver.py index 624f324..127629a 100644 --- a/crates/mind-body/src/self_run/probe/driver.py +++ b/crates/mind-body/src/self_run/probe/driver.py @@ -1,7 +1,7 @@ """Drives the real claude CLI (stream-json, --replay-user-messages) against fake.py through one scenario and records its stdout and what was written to its stdin. - PROBE_SCENARIO bg-agent | bg-late | bg-monitor (FAKE_PLAN agent,text | bg:8,text | monitor,text) | folded | queued | three-queued | serialized | ceiling-flush-folded | + PROBE_SCENARIO timed (SENDS, RUN_S; any FAKE_PLAN) | bg-agent | bg-late | bg-monitor (FAKE_PLAN agent,text | bg:8,text | monitor,text) | folded | queued | three-queued | serialized | ceiling-flush-folded | ceiling-flush-queued | api-error-first | blocking-limit | compaction-off-360k (three-queued writes a second and a third line while the first request is in flight: the CLI joins them into one echo. serialized is the actuator's policy @@ -225,6 +225,12 @@ elif SCENARIO == "bg-other": wait(lambda: results() >= 2, 30) wait(lambda: results() >= 3, 15) time.sleep(3) +elif SCENARIO == "timed": + # SENDS: a json list of [at_seconds, text]; RUN_S: seconds the session runs before stdin closes. + for at, text in json.loads(os.environ["SENDS"]): + time.sleep(max(0, T0 + at - time.time())) + send(text) + time.sleep(max(0, T0 + float(os.environ.get("RUN_S", "40")) - time.time())) elif SCENARIO in ("bg-agent", "bg-late", "bg-monitor"): send("PLAN turn A.") wait(lambda: results() >= 1) diff --git a/crates/mind-body/src/self_run/probe/fake.py b/crates/mind-body/src/self_run/probe/fake.py index c906c2f..4d7eaf3 100644 --- a/crates/mind-body/src/self_run/probe/fake.py +++ b/crates/mind-body/src/self_run/probe/fake.py @@ -9,11 +9,15 @@ usage. A marker in the newest user text picks the behaviour: MID360 report 360000 input tokens BIGCONTEXT report 990000 input tokens (the CLI blocking limit) +FAKE_PLAN is a comma list, one step per main-thread request: text | bg: | fg: | agent | monitor | hold: (text after s seconds) | +t: (a tool call from FAKE_TOOLS, a json {key: [tool name, input]}; "$ID" in the input is replaced by the last task id a tool result named). + Holds no credential: the CLI is pointed at it with a fake key. """ import json import http.server import os +import re import threading import time @@ -101,6 +105,14 @@ class H(http.server.BaseHTTPRequestHandler): name, inp = (("Agent", {"description": "probe agent", "prompt": "SUBTASK say done", "subagent_type": "general-purpose", "run_in_background": True}) if kind == "agent" else ("Monitor", {"description": "probe monitor", "command": "sleep 6; echo monitor-done", "timeout_ms": 30000, "persistent": False})) self.stream(n, req, {"type": "tool_use", "id": f"toolu_{n:04d}", "name": name, "input": {}}, {"type": "input_json_delta", "partial_json": json.dumps(inp)}, "tool_use", 10) return + if kind == "t": + name, inp = json.loads(os.environ["FAKE_TOOLS"])[arg] + text = json.dumps(inp) + if "$ID" in text: + ids = re.findall(r"with ID: ([A-Za-z0-9]+)", json.dumps(msgs)) + text = text.replace("$ID", ids[-1] if ids else "none") + self.stream(n, req, {"type": "tool_use", "id": f"toolu_{n:04d}", "name": name, "input": {}}, {"type": "input_json_delta", "partial_json": text}, "tool_use", 10) + return if kind == "hold": time.sleep(float(arg)) kind = "text" diff --git a/crates/mind-body/src/self_run/session.rs b/crates/mind-body/src/self_run/session.rs index 0947019..56c7171 100644 --- a/crates/mind-body/src/self_run/session.rs +++ b/crates/mind-body/src/self_run/session.rs @@ -24,6 +24,8 @@ pub(crate) struct Polled { pub(crate) pending: bool, /// A brief of another conversation waits: told only when the session is idle (the switch). pub(crate) other: bool, + /// The oldest waiting brief is another conversation's: at idle it goes before the served conversation's own. + pub(crate) other_first: bool, /// The session should end at its next idle: what is oldest and unanswered is not its business. pub(crate) yield_now: bool, } @@ -72,14 +74,15 @@ pub(crate) fn drive(clock: &dyn Clock, mut child: Box, inbox: &mut dy ttl_1h = hour; } match read_usage(&line) { - Usage::Request { id, prefix } => Event::Request { id, prefix }, + Usage::Request { id, prefix, stops } => Event::Request { id, prefix, stops }, Usage::Echo => { inbox.echoed(); Event::Echo } Usage::TurnBegan => Event::TurnBegan, - Usage::TaskStarted { id } => Event::TaskStarted { id }, - Usage::TaskNotified { id } => Event::TaskNotified { id }, + Usage::BackgroundTasks { live, main } => Event::BackgroundTasks { live, main }, + Usage::TaskBackgrounded { id } => Event::TaskBackgrounded { id }, + Usage::TaskNotified { id, stopped } => Event::TaskNotified { id, stopped }, Usage::NotificationTaken { id } => Event::NotificationTaken { id }, Usage::Result { ok, blocked } => { let (ok, blocked) = inbox.result(&line, ok, blocked)?; @@ -98,7 +101,7 @@ pub(crate) fn drive(clock: &dyn Clock, mut child: Box, inbox: &mut dy polled = now; match inbox.poll() { Ok(found) => { - events.push_back(Event::Inbox { pending: found.pending, other: found.other }); + events.push_back(Event::Inbox { pending: found.pending, other: found.other, other_first: found.other_first }); if found.yield_now { events.push_back(Event::Yield); } diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 462d6be..b83ebef 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -483,20 +483,20 @@ fn a_long_result_is_cut_at_the_limit_on_a_character_boundary_and_marked() { #[test] fn read_usage_sums_input_cache_read_and_cache_write_and_reads_absent_fields_as_zero() { - assert_eq!(read_usage(&request("m1", 1, 20, 300)), Usage::Request { id: "m1".into(), prefix: 321 }); + assert_eq!(read_usage(&request("m1", 1, 20, 300)), Usage::Request { id: "m1".into(), prefix: 321, stops: vec![] }); // Each term alone, so dropping or doubling any one is seen. - assert_eq!(read_usage(&request("m", 7, 0, 0)), Usage::Request { id: "m".into(), prefix: 7 }); - assert_eq!(read_usage(&request("m", 0, 7, 0)), Usage::Request { id: "m".into(), prefix: 7 }); - assert_eq!(read_usage(&request("m", 0, 0, 7)), Usage::Request { id: "m".into(), prefix: 7 }); + assert_eq!(read_usage(&request("m", 7, 0, 0)), Usage::Request { id: "m".into(), prefix: 7, stops: vec![] }); + assert_eq!(read_usage(&request("m", 0, 7, 0)), Usage::Request { id: "m".into(), prefix: 7, stops: vec![] }); + assert_eq!(read_usage(&request("m", 0, 0, 7)), Usage::Request { id: "m".into(), prefix: 7, stops: vec![] }); for absent in ["input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens"] { let mut event: Value = serde_json::from_str(&request("m", 1, 20, 300)).unwrap(); event["message"]["usage"].as_object_mut().unwrap().remove(absent); let want = 321 - [("input_tokens", 1), ("cache_read_input_tokens", 20), ("cache_creation_input_tokens", 300)].iter().find(|(name, _)| *name == absent).unwrap().1; - assert_eq!(read_usage(&event.to_string()), Usage::Request { id: "m".into(), prefix: want }, "{absent} absent"); + assert_eq!(read_usage(&event.to_string()), Usage::Request { id: "m".into(), prefix: want, stops: vec![] }, "{absent} absent"); } let mut no_usage: Value = serde_json::from_str(&request("m", 1, 1, 1)).unwrap(); no_usage["message"].as_object_mut().unwrap().remove("usage"); - assert_eq!(read_usage(&no_usage.to_string()), Usage::Request { id: "m".into(), prefix: 0 }); + assert_eq!(read_usage(&no_usage.to_string()), Usage::Request { id: "m".into(), prefix: 0, stops: vec![] }); } #[test] @@ -504,7 +504,7 @@ fn read_usage_classifies_the_owners_real_records() { let sums = [84210, 100751, 117174, 118336, 126746, 127250, 129853]; let mut seen = Vec::new(); for line in RECORDS.lines() { - let Usage::Request { id, prefix } = read_usage(line) else { panic!("an assistant record of the main thread is a request: {line}") }; + let Usage::Request { id, prefix, .. } = read_usage(line) else { panic!("an assistant record of the main thread is a request: {line}") }; if !seen.iter().any(|(known, _)| *known == id) { seen.push((id, prefix)); } @@ -525,19 +525,18 @@ fn read_usage_classifies_the_owners_real_records() { assert_eq!(read_usage(&init_line()), Usage::TurnBegan); let replayed = json!({"type": "user", "isReplay": true, "message": {"role": "user", "content": " t -toolu_7 "}}).to_string(); - assert_eq!(read_usage(&replayed), Usage::NotificationTaken { id: "toolu_7".into() }, "the CLI's replay of a notification is never our echo"); + assert_eq!(read_usage(&replayed), Usage::NotificationTaken { id: "t".into() }, "the CLI's replay of a notification is never our echo"); 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(); no_id["message"].as_object_mut().unwrap().remove("id"); assert_eq!(read_usage(&no_id.to_string()), Usage::Other); let mut null_parent: Value = serde_json::from_str(&at("m", 5)).unwrap(); null_parent["parent_tool_use_id"] = Value::Null; - assert_eq!(read_usage(&null_parent.to_string()), Usage::Request { id: "m".into(), prefix: 5 }); + assert_eq!(read_usage(&null_parent.to_string()), Usage::Request { id: "m".into(), prefix: 5, stops: vec![] }); let mut no_parent: Value = serde_json::from_str(&at("m", 5)).unwrap(); no_parent.as_object_mut().unwrap().remove("parent_tool_use_id"); - assert_eq!(read_usage(&no_parent.to_string()), Usage::Request { id: "m".into(), prefix: 5 }); + assert_eq!(read_usage(&no_parent.to_string()), Usage::Request { id: "m".into(), prefix: 5, stops: vec![] }); for unreadable in ["", "not json", "[1,2]", "\"assistant\"", "12", "null", "{\"type\":"] { assert_eq!(read_usage(unreadable), Usage::Unreadable, "{unreadable:?}"); } @@ -734,9 +733,9 @@ fn the_message_ids_of_a_long_run_are_all_remembered() { #[test] fn a_usage_sum_that_overflows_is_at_or_over_the_ceiling_not_a_panic_or_a_wrap() { let line = request("m", u64::MAX, 1, 0); - assert_eq!(read_usage(&line), Usage::Request { id: "m".into(), prefix: u64::MAX }); - assert_eq!(read_usage(&request("m", u64::MAX, u64::MAX, u64::MAX)), Usage::Request { id: "m".into(), prefix: u64::MAX }); - assert_eq!(read_usage(&request("m", 1, u64::MAX, 1)), Usage::Request { id: "m".into(), prefix: u64::MAX }); + assert_eq!(read_usage(&line), Usage::Request { id: "m".into(), prefix: u64::MAX, stops: vec![] }); + assert_eq!(read_usage(&request("m", u64::MAX, u64::MAX, u64::MAX)), Usage::Request { id: "m".into(), prefix: u64::MAX, stops: vec![] }); + assert_eq!(read_usage(&request("m", 1, u64::MAX, 1)), Usage::Request { id: "m".into(), prefix: u64::MAX, stops: vec![] }); // Through the actuator: the flush is asked for, and the headroom beyond a saturated prefix does not overflow // either, and the pass that wrote the flush does not interrupt. let ran = go(vec![Out(echo()), Out(request("m1", u64::MAX, 7, 0)), Out(request("m2", 0, 0, 0))]); @@ -1562,7 +1561,7 @@ mod live_session { let entry: Value = serde_json::from_str(entry).unwrap(); let at = crate::testkit::at((entry["t"].as_f64().unwrap() * 1000.0) as i64); if let Some(text) = entry.get("want").and_then(Value::as_str) { - let (label, fact) = if text.starts_with(FLUSH_MESSAGE) { ("flush-asked", Fact::Tick) } else { ("want", Fact::Inbox { pending: true, other: false }) }; + let (label, fact) = if text.starts_with(FLUSH_MESSAGE) { ("flush-asked", Fact::Tick) } else { ("want", Fact::Inbox { pending: true, other: false, other_first: false }) }; let (next, action) = decide(state, fact, at); state = next; trace.push((label.to_string(), action)); @@ -1573,11 +1572,12 @@ mod live_session { ttl_1h = hour; } let (label, fact) = match read_usage(line) { - Usage::Request { id, prefix } => (format!("request:{prefix}"), Fact::Request { id, prefix }), + Usage::Request { id, prefix, stops } => (format!("request:{prefix}"), Fact::Request { id, prefix, stops }), Usage::Echo => ("echo".to_string(), Fact::Echo), Usage::TurnBegan => ("turn".to_string(), Fact::TurnBegan), - Usage::TaskStarted { id } => ("task".to_string(), Fact::TaskStarted { id }), - Usage::TaskNotified { id } => ("notified".to_string(), Fact::TaskNotified { id }), + Usage::BackgroundTasks { live, main } => ("level".to_string(), Fact::BackgroundTasks { live, main }), + Usage::TaskBackgrounded { id } => ("backgrounded".to_string(), Fact::TaskBackgrounded { id }), + Usage::TaskNotified { id, stopped } => ("notified".to_string(), Fact::TaskNotified { id, stopped }), Usage::NotificationTaken { id } => ("taken".to_string(), Fact::NotificationTaken { id }), 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), @@ -1613,6 +1613,9 @@ mod live_session { const BG_RACE: &str = include_str!("claude-2.1.312-live-session-bg-race.jsonl"); const BG_AGENT: &str = include_str!("claude-2.1.312-live-session-bg-agent.jsonl"); const BG_LATE: &str = include_str!("claude-2.1.312-live-session-bg-late.jsonl"); + const BG_FGTO: &str = include_str!("claude-2.1.312-live-session-bg-fgto.jsonl"); + const BG_KILL: &str = include_str!("claude-2.1.312-live-session-bg-kill.jsonl"); + const BG_MONITOR: &str = include_str!("claude-2.1.312-live-session-bg-monitor.jsonl"); #[test] fn a_line_written_during_a_tool_call_is_folded_one_result_and_its_echo_before_the_request_that_carries_it() { @@ -1719,31 +1722,82 @@ mod live_session { kinds(trace).into_iter().filter(|kind| *kind != "other").collect() } + fn usages(jsonl: &str) -> Vec { + jsonl.lines().filter_map(|entry| serde_json::from_str::(entry).ok()?.get("line")?.as_str().map(read_usage)).collect() + } + #[test] - fn read_usage_reads_the_stream_lines_the_captures_hold() { - // A turn is begun by the init line, a backgrounded task by its start, its notification by its event, and the CLI's replay of that - // notification is taken, never our echo. - assert_eq!(significant(&replay(BG_FOLD)), ["turn", "echo", "request:10", "task", "notified", "request:10", "taken", "request:10", "result:ok"]); - assert_eq!(significant(&replay(BG_IDLE)), ["turn", "echo", "request:10", "task", "notified", "request:10", "result:ok", "turn", "request:10", "result:ok"]); - assert_eq!(significant(&replay(BG_RACE)), ["turn", "echo", "request:10", "task", "notified", "request:10", "result:ok", "turn", "request:10", "result:ok", "turn", "echo", "request:10", "result:ok"]); - assert_eq!(significant(&replay(BG_OTHER)), ["turn", "echo", "request:10", "task", "notified", "request:10", "result:ok", "turn", "request:10", "echo", "request:10", "result:ok"]); - let taken = BG_FOLD.lines().filter_map(|entry| serde_json::from_str::(entry).ok()?.get("line")?.as_str().map(read_usage)).find(|usage| matches!(usage, Usage::NotificationTaken { .. })); - assert_eq!(taken, Some(Usage::NotificationTaken { id: "toolu_0001".into() })); - // A task that was not started backgrounded is no task the session waits for. - let foreground = json!({"type": "system", "subtype": "task_started", "task_id": "t", "tool_use_id": "toolu_9", "is_backgrounded": false}).to_string(); - assert_eq!(read_usage(&foreground), Usage::Other); + fn read_usage_reads_the_level_the_backgrounded_lines_the_notifications_and_the_stops_the_captures_hold() { + // A turn is begun by the init line, a task joins the level and is backgrounded, its notification comes by its event and the level + // empties after it; the CLI's replay of that notification is taken, never our echo. + let level = ["level", "backgrounded", "notified", "level"]; + assert_eq!(significant(&replay(BG_FOLD)), [vec!["turn", "echo", "request:10"], level.to_vec(), vec!["request:10", "taken", "request:10", "result:ok"]].concat()); + assert_eq!(significant(&replay(BG_IDLE)), [vec!["turn", "echo", "request:10"], level.to_vec(), vec!["request:10", "result:ok", "turn", "request:10", "result:ok"]].concat()); + assert_eq!(significant(&replay(BG_RACE)), [vec!["turn", "echo", "request:10"], level.to_vec(), vec!["request:10", "result:ok", "turn", "request:10", "result:ok", "turn", "echo", "request:10", "result:ok"]].concat()); + assert_eq!(significant(&replay(BG_OTHER)), [vec!["turn", "echo", "request:10"], level.to_vec(), vec!["request:10", "result:ok", "turn", "request:10", "echo", "request:10", "result:ok"]].concat()); + // A persistent monitor: notice turns the CLI begins while its task stays in the level, then the notification and the level's end. + assert_eq!(significant(&replay(BG_MONITOR)).iter().filter(|kind| **kind == "level").count(), 2); + let task = |usage: &Usage| match usage { + Usage::TaskBackgrounded { id } | Usage::TaskNotified { id, .. } | Usage::NotificationTaken { id } => Some(id.clone()), + _ => None, + }; + let fold = usages(BG_FOLD); + let ids: Vec = fold.iter().filter_map(task).collect(); + assert_eq!(ids.len(), 3, "backgrounded, notified, taken"); + assert!(ids.iter().all(|id| *id == ids[0]) && !ids[0].starts_with("toolu_"), "one task_id names the task at its three lines: {ids:?}"); + assert!(fold.contains(&Usage::BackgroundTasks { live: vec![ids[0].clone()], main: vec![ids[0].clone()] }) && fold.contains(&Usage::BackgroundTasks { live: vec![], main: vec![] })); + // A task that went to the background at its timeout was announced foreground and is backgrounded by the update that moved it: the start is no reading. + let fgto = usages(BG_FGTO); + let moved: Vec<&Usage> = fgto.iter().filter(|usage| matches!(usage, Usage::TaskBackgrounded { .. })).collect(); + assert_eq!(moved.len(), 1, "{fgto:?}"); + assert!(fgto.iter().any(|usage| matches!(usage, Usage::TaskNotified { stopped: false, .. }))); + // A stop is a notification whose status is stopped, and the TaskStop call on the request line names the task. + let kill = usages(BG_KILL); + let Some(Usage::TaskBackgrounded { id: killed }) = kill.iter().find(|usage| matches!(usage, Usage::TaskBackgrounded { .. })) else { panic!("{kill:?}") }; + assert!(kill.contains(&Usage::TaskNotified { id: killed.clone(), stopped: true })); + assert!(kill.iter().any(|usage| matches!(usage, Usage::Request { stops, .. } if stops == std::slice::from_ref(killed))), "{kill:?}"); + assert_eq!(kill.iter().filter(|usage| matches!(usage, Usage::Request { stops, .. } if !stops.is_empty())).count(), 1); + // The level's `main` leaves out a subagent's and an awaited task; a start under a parent task, or awaited, or in the foreground, is no reading. + let changed = json!({"type": "system", "subtype": "background_tasks_changed", "tasks": [{"task_id": "m"}, {"task_id": "s", "parent_task_id": "p"}, {"task_id": "w", "awaited": true}, {"task_id": "n", "parent_task_id": null, "awaited": false}]}).to_string(); + assert_eq!(read_usage(&changed), Usage::BackgroundTasks { live: vec!["m".into(), "s".into(), "w".into(), "n".into()], main: vec!["m".into(), "n".into()] }); + let started = |extra: Value| { + let mut line = json!({"type": "system", "subtype": "task_started", "task_id": "t", "is_backgrounded": true}); + line.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone()); + read_usage(&line.to_string()) + }; + assert_eq!(started(json!({})), Usage::TaskBackgrounded { id: "t".into() }); + assert_eq!(started(json!({"is_backgrounded": false})), Usage::Other, "a foreground task is no reading"); + assert_eq!(started(json!({"parent_task_id": "p"})), Usage::Other); + assert_eq!(started(json!({"awaited": true})), Usage::Other); + // The notification's old key is no key: a line without a task_id is nothing. + let keyed = json!({"type": "system", "subtype": "task_notification", "tool_use_id": "toolu_9", "status": "completed"}).to_string(); + assert_eq!(read_usage(&keyed), Usage::Other); + // The stop is read under any name of the task's id the call carries, and only from a TaskStop call. + let call = |name: &str, input: Value| { + let mut event: Value = serde_json::from_str(RECORDS.lines().next().unwrap()).unwrap(); + event["message"]["content"] = json!([{"type": "text", "text": "stopping"}, {"type": "tool_use", "id": "toolu_5", "name": name, "input": input}]); + let Usage::Request { stops, .. } = read_usage(&event.to_string()) else { panic!("a request") }; + stops + }; + assert_eq!(call("TaskStop", json!({"task_id": "t"})), ["t"]); + assert_eq!(call("TaskStop", json!({"shell_id": "u"})), ["u"]); + for alias in ["KillShell", "KillBash"] { + assert_eq!(call(alias, json!({"shell_id": "u"})), ["u"], "{alias} is the same stop under another name"); + } + assert!(call("Bash", json!({"task_id": "t"})).is_empty()); } #[test] fn a_background_agent_announces_itself_as_a_backgrounded_task_and_the_session_is_not_idle_until_its_notification_turn_ends() { - let usages: Vec = BG_AGENT.lines().filter_map(|entry| serde_json::from_str::(entry).ok()?.get("line")?.as_str().map(read_usage)).collect(); - let at = |wanted: Usage| usages.iter().position(|usage| *usage == wanted).unwrap_or_else(|| panic!("{wanted:?} is not in the capture")); - // The Agent call's own tool_use_id names the task at its start and at its notification. - assert!(at(Usage::TaskStarted { id: "toolu_0001".into() }) < at(Usage::TaskNotified { id: "toolu_0001".into() })); + let usages = usages(BG_AGENT); + // One task_id names the task at its start and at its notification. + let ids: Vec<&Usage> = usages.iter().filter(|usage| matches!(usage, Usage::TaskBackgrounded { .. } | Usage::TaskNotified { .. })).collect(); + assert!(matches!(ids[..], [Usage::TaskBackgrounded { id: started }, Usage::TaskNotified { id: ended, stopped: false }] if started == ended), "{ids:?}"); + assert!(usages.iter().position(|usage| matches!(usage, Usage::TaskBackgrounded { .. })) < usages.iter().position(|usage| matches!(usage, Usage::TaskNotified { .. }))); let idle = idleness(BG_AGENT); let labels: Vec<&str> = idle.iter().map(|(label, _)| label.as_str()).collect(); let idle_after = |label: &str, nth: usize| idle.iter().zip(&labels).filter(|(_, l)| **l == label).nth(nth).unwrap().0.1; - assert!(!idle_after("task", 0), "the task runs"); + assert!(!idle_after("backgrounded", 0), "the task runs"); assert!(!idle_after("result:ok", 0), "the turn that started it ended, the task still runs"); assert!(!idle_after("notified", 0), "the notification is owed its turn"); assert!(idle_after("result:ok", 1), "the CLI's notification turn ended"); @@ -1753,7 +1807,7 @@ mod live_session { fn a_background_task_that_ends_while_no_turn_is_open_gets_a_turn_the_cli_begins_and_the_session_is_idle_after_its_result() { let idle = idleness(BG_LATE); let labels: Vec<&str> = idle.iter().map(|(label, _)| label.as_str()).collect(); - assert_eq!(significant(&replay(BG_LATE)), ["turn", "echo", "request:10", "task", "request:10", "result:ok", "notified", "turn", "request:10", "result:ok"]); + assert_eq!(significant(&replay(BG_LATE)), ["turn", "echo", "request:10", "level", "backgrounded", "request:10", "result:ok", "notified", "level", "turn", "request:10", "result:ok"]); let idle_after = |label: &str, nth: usize| idle.iter().zip(&labels).filter(|(_, l)| **l == label).nth(nth).unwrap().0.1; assert!(!idle_after("result:ok", 0), "idle but for the running task"); assert!(!idle_after("notified", 0), "owed its turn"); diff --git a/crates/mind-body/src/self_run/tests/conversation.rs b/crates/mind-body/src/self_run/tests/conversation.rs index a0873fd..ed9ed01 100644 --- a/crates/mind-body/src/self_run/tests/conversation.rs +++ b/crates/mind-body/src/self_run/tests/conversation.rs @@ -11,6 +11,9 @@ const BG_FOLD: &str = include_str!("../claude-2.1.312-live-session-bg-fold.jsonl const BG_OTHER: &str = include_str!("../claude-2.1.312-live-session-bg-other.jsonl"); const BG_RACE: &str = include_str!("../claude-2.1.312-live-session-bg-race.jsonl"); const BG_LATE: &str = include_str!("../claude-2.1.312-live-session-bg-late.jsonl"); +const BG_FGTO: &str = include_str!("../claude-2.1.312-live-session-bg-fgto.jsonl"); +const BG_KILL: &str = include_str!("../claude-2.1.312-live-session-bg-kill.jsonl"); +const BG_MONITOR: &str = include_str!("../claude-2.1.312-live-session-bg-monitor.jsonl"); fn saying(text: &str) -> String { json!({"type": "result", "subtype": "success", "is_error": false, "terminal_reason": "completed", "result": text}).to_string() @@ -78,7 +81,8 @@ fn label_of(line: &str) -> &'static str { Usage::Request { .. } => "request", Usage::Echo => "echo", Usage::Result { .. } => "result", - Usage::TaskStarted { .. } => "started", + Usage::BackgroundTasks { .. } => "level", + Usage::TaskBackgrounded { .. } => "backgrounded", Usage::TaskNotified { .. } => "notified", Usage::NotificationTaken { .. } => "taken", _ => "other", @@ -105,8 +109,8 @@ fn script(lines: &[String], splices: Vec<(usize, Vec)>) -> Vec { } /// A message of the CLI's that the stream gives for a task the session never saw start. -fn notification_of(tool_use: &str) -> String { - json!({"type": "system", "subtype": "task_notification", "task_id": "unseen", "tool_use_id": tool_use, "status": "completed", "summary": "synthetic task"}).to_string() +fn notification_of(task: &str) -> String { + json!({"type": "system", "subtype": "task_notification", "task_id": task, "status": "completed", "summary": "synthetic task"}).to_string() } // ---- who is served at the start -------------------------------------------- @@ -338,12 +342,13 @@ fn another_conversations_brief_waits_behind_a_notification_turn_with_no_tool() { /// A task that was never started backgrounded, and a notification of a task never seen, owe nothing: no wedge. #[test] fn a_notification_of_a_foreground_or_unknown_task_holds_nothing() { - let lines: Vec = stream(BG_IDLE).into_iter().map(|line| line.replace("\"is_backgrounded\":true", "\"is_backgrounded\":false")).collect(); - assert!(lines.iter().all(|line| !matches!(read_usage(line), Usage::TaskStarted { .. })), "the capture's task is foreground now"); + // The capture's task is foreground now: not announced backgrounded and in no level. + let lines: Vec = stream(BG_IDLE).into_iter().map(|line| line.replace("\"is_backgrounded\":true", "\"is_backgrounded\":false")).filter(|line| label_of(line) != "level").collect(); + assert!(lines.iter().all(|line| !matches!(read_usage(line), Usage::TaskBackgrounded { .. } | Usage::BackgroundTasks { .. })), "the capture's task is foreground now"); let (request1, result1) = (after(&lines, "request", 1), after(&lines, "result", 1)); let mind = serving_aqua(); let steps = script(&lines[..result1], vec![(request1, vec![briefed(&mind, 31, vec![libby("l1")])])]); - let ran = mind_run(&mind, [steps, vec![Out(notification_of("toolu_unseen")), Idle(31), Out(echo()), Out(saying("libby done"))]].concat()); + let ran = mind_run(&mind, [steps, vec![Out(notification_of("unseen")), Idle(31), Out(echo()), Out(saying("libby done"))]].concat()); assert!(ran.outcome.is_ok()); assert_eq!(tells(&ran).len(), 1, "the first idle tells Libby's brief: {:?}", ran.written()); assert_eq!(answered(&mind, 2), vec![id_of("l1")]); @@ -365,6 +370,113 @@ fn a_backgrounded_task_that_is_never_notified_holds_other_conversations_to_the_r assert_eq!(answered(&mind, 1), vec![id_of("a1")]); } +/// A task of the main session that the CLI moved to the background at its timeout is in the level from then on, with no `task_started` that says +/// so. Libby's brief, derived after Aqua's turn ended, is written into none of the CLI's turn for the task: not while the level holds it, and +/// not in the gap after the level empties and before the CLI's init line; only after that turn's result, which was Aqua's. +#[test] +fn a_task_moved_to_the_background_at_its_timeout_holds_other_conversations_until_the_clis_turn_for_it_ends() { + let lines = stream(BG_FGTO); + let (result1, level2, init2, request3, result2) = (after(&lines, "result", 1), after(&lines, "level", 2), after(&lines, "init", 2), after(&lines, "request", 3), after(&lines, "result", 2)); + let splices = |mind: &Rc| vec![(result1, vec![briefed(mind, 31, vec![libby("l1")]), Idle(31)]), (level2, vec![Idle(31)]), (init2, vec![Idle(31)]), (request3, vec![Idle(31)])]; + let mind = serving_aqua(); + let ran = mind_run(&mind, script(&lines[..result2 - 1], splices(&mind))); + assert!(ran.outcome.is_ok()); + assert!(!written_with(&ran, "@libby:chat") && !written_with(&ran, &id_of("l1")), "no line of Libby's reaches the CLI's turn for Aqua's task: {:?}", ran.written()); + let mind = serving_aqua(); + let mut steps = splices(&mind); + steps.push((result2, vec![Idle(31), Out(echo()), Out(saying("libby done"))])); + let ran = mind_run(&mind, script(&lines, steps)); + assert!(ran.outcome.is_ok()); + let told = tells(&ran); + assert_eq!(told.len(), 1, "{:?}", ran.written()); + assert!(told[0].contains(&id_of("l1")) && !told[0].contains(&id_of("a1")), "{}", told[0]); + assert_eq!((answered(&mind, 1), answered(&mind, 2), answered(&mind, 3)), (vec![id_of("a1")], vec![id_of("a1")], vec![id_of("l1")])); +} + +/// A task the session stops with TaskStop is notified inside the turn that stops it and the CLI gives it no turn: the next conversation is +/// told at the first idle after that turn's result. +#[test] +fn a_task_the_session_stops_with_taskstop_owes_no_turn() { + let lines = stream(BG_KILL); + let (request1, result1) = (after(&lines, "request", 1), after(&lines, "result", 1)); + let mind = serving_aqua(); + let ran = mind_run(&mind, script(&lines, vec![(request1, vec![briefed(&mind, 31, vec![libby("l1")])]), (result1, vec![Idle(31), Out(echo()), Out(saying("libby done"))])])); + assert!(ran.outcome.is_ok()); + let told = tells(&ran); + assert_eq!(told.len(), 1, "{:?}", ran.written()); + assert!(told[0].contains(&id_of("l1")) && !told[0].contains(&id_of("a1")), "{}", told[0]); + assert_eq!(ran.count(&Event::InputClosed), 0); + assert_eq!((answered(&mind, 1), answered(&mind, 2)), (vec![id_of("a1")], vec![id_of("l1")]), "no CLI turn was awaited"); +} + +/// The same stream with the stop made by something else: the TaskStop call is renamed, so the stop was not the session's and the CLI gives the +/// notification its turn, which follows the result. +#[test] +fn a_stop_the_session_did_not_ask_for_owes_its_turn() { + let lines: Vec = stream(BG_KILL).into_iter().map(|line| line.replace("\"name\":\"TaskStop\"", "\"name\":\"Other\"")).collect(); + assert!(lines.iter().all(|line| !matches!(read_usage(line), Usage::Request { stops, .. } if !stops.is_empty())), "no stop of ours is left in the capture"); + let (request1, result1) = (after(&lines, "request", 1), after(&lines, "result", 1)); + let cli_turn = |result: Vec| [vec![Idle(31), Out(init_line()), Out(at("c1", 10)), Idle(31)], result].concat(); + let splices = |mind: &Rc, rest: Vec| vec![(request1, vec![briefed(mind, 31, vec![libby("l1")])]), (result1, [vec![Idle(31)], cli_turn(rest)].concat())]; + let mind = serving_aqua(); + let ran = mind_run(&mind, script(&lines, splices(&mind, vec![]))); + assert!(ran.outcome.is_ok()); + assert!(!written_with(&ran, "@libby:chat") && !written_with(&ran, &id_of("l1")), "the CLI's turn for the stopped task is running: {:?}", ran.written()); + let mind = serving_aqua(); + let ran = mind_run(&mind, script(&lines, splices(&mind, vec![Out(saying("the task was stopped")), Idle(31), Out(echo()), Out(saying("libby done"))]))); + assert!(ran.outcome.is_ok()); + assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written()); + assert_eq!((answered(&mind, 1), answered(&mind, 2), answered(&mind, 3)), (vec![id_of("a1")], vec![id_of("a1")], vec![id_of("l1")]), "the CLI's turn answers Aqua's brief, never Libby's"); +} + +/// A persistent Monitor stays in the level through several notice turns the CLI begins; Libby's brief waits through all of them and the final turn +/// of the notification. +#[test] +fn a_persistent_monitor_holds_other_conversations_through_its_notice_turns() { + let lines = stream(BG_MONITOR); + let result1 = after(&lines, "result", 1); + let windows = |mind: &Rc, upto: usize| { + let mut splices: Vec<(usize, Vec)> = (result1..=upto).map(|at| (at, vec![Idle(31)])).collect(); + splices[0].1.insert(0, briefed(mind, 31, vec![libby("l1")])); + splices + }; + let mind = serving_aqua(); + let last_result = lines.len(); + let ran = mind_run(&mind, script(&lines[..last_result - 1], windows(&mind, last_result - 1))); + assert!(ran.outcome.is_ok()); + assert!(!written_with(&ran, "@libby:chat") && !written_with(&ran, &id_of("l1")), "no line of Libby's reaches a notice turn: {:?}", ran.written()); + let mind = serving_aqua(); + let mut splices = windows(&mind, last_result - 1); + splices.push((last_result, vec![Idle(31), Out(echo()), Out(saying("libby done"))])); + let ran = mind_run(&mind, script(&lines, splices)); + assert!(ran.outcome.is_ok()); + assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written()); + let results = lines.iter().filter(|line| label_of(line) == "result").count(); + for n in 1..=results { + assert_eq!(answered(&mind, u32::try_from(n).unwrap()), vec![id_of("a1")], "reply {n}: the turns of the monitor are Aqua's"); + } + assert_eq!(answered(&mind, u32::try_from(results + 1).unwrap()), vec![id_of("l1")]); +} + +/// A subagent's or an awaited task is in the CLI's level and holds the session while it runs; its notification is no one's turn. +#[test] +fn a_subagents_or_an_awaited_tasks_notification_owes_nothing() { + let level = |tasks: Value| json!({"type": "system", "subtype": "background_tasks_changed", "tasks": tasks}).to_string(); + let ended = |task: &str| json!({"type": "system", "subtype": "task_notification", "task_id": task, "status": "completed", "summary": "synthetic task"}).to_string(); + let both = || json!([{"task_id": "sub", "parent_task_id": "agent"}, {"task_id": "waited", "awaited": true}]); + let mind = serving_aqua(); + let steps = vec![Out(echo()), Out(level(both())), Out(saying("aqua done")), briefed(&mind, 31, vec![libby("l1")]), Idle(31), Out(ended("sub"))]; + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + assert!(!written_with(&ran, &id_of("l1")), "while the level holds them: {:?}", ran.written()); + let mind = serving_aqua(); + let steps = vec![Out(echo()), Out(level(both())), Out(saying("aqua done")), briefed(&mind, 31, vec![libby("l1")]), Idle(31), Out(ended("sub")), Out(ended("waited")), Out(level(json!([]))), Idle(31), Out(echo()), Out(saying("libby done"))]; + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + assert_eq!(tells(&ran).len(), 1, "idle once the level is empty: {:?}", ran.written()); + assert_eq!((answered(&mind, 1), answered(&mind, 2)), (vec![id_of("a1")], vec![id_of("l1")])); +} + // ---- who a result belongs to ------------------------------------------------ /// An error result makes no reply and leaves its briefs for the next result of the span, but never for another span. @@ -412,6 +524,21 @@ fn at_idle_the_oldest_waiting_conversation_is_served_first() { assert_eq!((answered(&mind, 1), answered(&mind, 2), answered(&mind, 3)), (vec![id_of("a1"), id_of("a2")], vec![id_of("l1")], vec![id_of("m1")])); } +/// Aqua is served and idle; Libby's brief is the older of the two now waiting, Aqua's own the newer: at idle the oldest goes first, so Aqua's +/// brief does not outrun Libby's again and again. +#[test] +fn at_idle_an_older_brief_of_another_conversation_goes_before_the_served_ones_newer_brief() { + let mind = serving_aqua(); + let steps = vec![Out(echo()), Out(saying("aqua done")), briefed(&mind, 0, vec![libby("l1")]), briefed(&mind, 31, vec![aqua("a2")]), Idle(31), Out(echo()), Out(saying("libby done")), Idle(31), Out(echo()), Out(saying("aqua again"))]; + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + let told = tells(&ran); + assert_eq!(told.len(), 2, "{:?}", ran.written()); + assert!(told[0].contains(&id_of("l1")) && !told[0].contains(&id_of("a2")), "the older brief first: {}", told[0]); + assert!(told[1].contains(&id_of("a2")) && !told[1].contains(&id_of("l1")), "{}", told[1]); + assert_eq!((answered(&mind, 1), answered(&mind, 2), answered(&mind, 3)), (vec![id_of("a1")], vec![id_of("l1")], vec![id_of("a2")])); +} + /// A result that took in no brief answers the briefs of the span being served: never the span before. #[test] fn every_result_of_a_span_answers_the_spans_briefs() { diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 644cc85..523c66d 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -221,7 +221,7 @@ 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 } , Echo , TurnBegan , TaskStarted { id : String } , TaskNotified { id : String } , NotificationTaken { id : String } , Result { ok : bool , blocked : bool } , Other , Unreadable , } +mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Usage { Request { id : String , prefix : u64 , stops : Vec < String > } , Echo , TurnBegan , BackgroundTasks { live : Vec < String > , main : Vec < String > } , TaskBackgrounded { id : String } , TaskNotified { id : String , stopped : bool } , NotificationTaken { id : String } , 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