GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

self: idleness reads the CLI's own background-task level and its notification-to-turn rule; at idle the oldest waiting conversation goes first

Browse source 
Cut: State.tasks, Event::TaskStarted, the tool_use_id keying of task lines and the notification replay, and TaskNotified's tasks.remove gate. running is a copy of the CLI's last background_tasks_changed, replaced whole; a notification owes a turn when the task was backgrounded and the session did not stop it with TaskStop (or KillShell, KillBash: the stream shows the alias as the model named it); Polled and Inbox carry other_first, so a served conversation's newer brief no longer outruns an older brief of another. Fixtures bg-fgto, bg-kill and bg-monitor pin the three falsified cases; the probe's fake API gains a generic tool step and a timed scenario.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-07 06:48:03 +01:00
commit 62ce2ba12d
10 changed files with 446 additions and 137 deletions
Download patch file Download diff file Expand all files Collapse all files

6
crates/mind-body/src/persona.rs
View file

@ -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<HeldReason> {
@ -1448,7 +1448,7 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
fn poll(&mut self) -> Result<Polled> {
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<C: ControlSource, S: ControlSource> 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<ChatDelivery> = 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

60
crates/mind-body/src/self_run.rs
View file

@ -259,25 +259,33 @@ pub fn claude_args(session: Uuid) -> Vec<String> {
/// What a user line starts with when the CLI replays a task notification to the model.
const NOTIFICATION_TAG: &str = "<task-notification>";
/// 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 = "<synthetic>";
/// 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<String> },
/// 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<String>, main: Vec<String> },
/// 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 `<task-id>` 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<String, Value>| 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<String, Value>| 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<String, Value>> = 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("<tool-use-id>")
.and_then(|(_, rest)| rest.split_once("</tool-use-id>"))
.split_once("<task-id>")
.and_then(|(_, rest)| rest.split_once("</task-id>"))
.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

76
crates/mind-body/src/self_run/decide.rs
View file

@ -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<String> },
/// 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<String>, main: Vec<String> },
/// 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<i32>),
@ -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<String>,
/// 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<String>,
/// 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<String>,
/// 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<String>,
/// Notifications of backgrounded tasks that have not yet been given their turn, by `task_id`.
owed: BTreeMap<String, Owed>,
/// 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<DateTime<Utc>>,
@ -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<Utc>) -> (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<Utc>) -> (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<Utc>) -> (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) => {

143
crates/mind-body/src/self_run/decide/tests.rs
View file

@ -23,7 +23,7 @@ fn t(seconds: i64) -> DateTime<Utc> {
}
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<String> = 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::<Vec<_>>();
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::<Vec<_>>(), ["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");

8
crates/mind-body/src/self_run/probe/driver.py
View file

@ -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)

12
crates/mind-body/src/self_run/probe/fake.py
View file

@ -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:<s> | fg:<s> | agent | monitor | hold:<s> (text after s seconds) |
t:<key> (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"

11
crates/mind-body/src/self_run/session.rs
View file

@ -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<dyn Child>, 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<dyn Child>, 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);
}

126
crates/mind-body/src/self_run/tests.rs
View file

@ -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": "<task-notification>
<task-id>t</task-id>
<tool-use-id>toolu_7</tool-use-id>
</task-notification>"}}).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<Usage> {
jsonl.lines().filter_map(|entry| serde_json::from_str::<Value>(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::<Value>(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<String> = 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<Usage> = BG_AGENT.lines().filter_map(|entry| serde_json::from_str::<Value>(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");

139
crates/mind-body/src/self_run/tests/conversation.rs
View file

@ -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<Step>)>) -> Vec<Step> {
}
/// 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<String> = 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<String> = 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<TestMind>| 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<String> = 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<Step>| [vec![Idle(31), Out(init_line()), Out(at("c1", 10)), Idle(31)], result].concat();
let splices = |mind: &Rc<TestMind>, rest: Vec<Step>| 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<TestMind>, upto: usize| {
let mut splices: Vec<(usize, Vec<Step>)> = (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() {

2
crates/mind-body/tests/public_surface.txt
View file

@ -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