From d655b59ce80b6d6f121de266a9948dff7c5b0e8c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 7 Nov 2026 05:14:04 +0100 Subject: [PATCH] self: a session serves one conversation at a time; another conversation's brief is written only at idle, and every result belongs to the conversation being served decide owns idleness (no turn open, nothing outstanding, no backgrounded task running, no notification owed its turn) and the switch of conversation; read_usage reads the task lines for liveness only; the CLI's replay of a task notification is no echo of ours. SelfInbox serves one Conversation and attributes a result by the span, not by the order of the stream. Deleted: briefless_open, briefless_answers, turn_conversation, turn_allows, may_tell, one_conversation_told, note_task, tasks, notified, the Persona pointer line and its plumbing. ScheduleWakeup and CronCreate are denied to the Self session. Tests replay the real-order captures with briefs spliced in. Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/persona.rs | 40 +- crates/mind-body/src/persona/tests.rs | 70 +- crates/mind-body/src/self_run.rs | 202 ++---- crates/mind-body/src/self_run/decide.rs | 109 ++- crates/mind-body/src/self_run/decide/tests.rs | 145 +++- crates/mind-body/src/self_run/session.rs | 20 +- crates/mind-body/src/self_run/tests.rs | 104 ++- .../src/self_run/tests/conversation.rs | 673 +++++++++--------- crates/mind-body/tests/public_surface.txt | 2 +- 9 files changed, 774 insertions(+), 591 deletions(-) diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index ae070a2..dfe6d50 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -961,12 +961,11 @@ impl Conversation { } } -/// The conversation a reply is told in: the one conversation its briefs' origins name, or none when they name two or -/// it answers no brief (a result the Self could not attribute). Such a reply has no conversation: the public session is told a -/// pointer to it without its text, so no member's words reach another's room. +/// The conversation a reply is told in: the one conversation its briefs' origins name, the public one when it answers no brief, or none +/// when they name two. Such a reply is told in no session, so no member's words reach another's room, and it stays in force. fn reply_conversation(answers: &[Option]) -> Option { let mut conversations = answers.iter().map(|origin| Conversation::of_origin(origin.as_ref())); - let first = conversations.next()?; + let first = conversations.next().unwrap_or(Conversation::Public); conversations.all(|conversation| conversation == first).then_some(first) } @@ -979,12 +978,10 @@ struct Route { conversation: Conversation, /// The origin of the first brief the reply answers that has one: what a session opened for the reply alone records. origin: Option, - /// The reply could not be placed in one conversation (it answers none, or briefs of two): the public session is told a pointer to it, never its text. - pointer: bool, } /// Her in-force replies this process has not told, oldest first, each with its conversation: one selection of the replies and -/// one keyed selection of the briefs they answer. A reply with no one conversation is a pointer route in the public one. +/// one keyed selection of the briefs they answer. A reply with no one conversation (briefs of two) is dropped: told in no session, and in force. fn reply_routes(mind: &dyn MindPort, told: &BTreeSet) -> Result> { let replies: Vec<_> = headers(mind, &in_force(any_of(of_kind(PipelineKind::Reply), "root", &[&mind.instance().0])))?.into_iter().filter(|header| !told.contains(&header.id.id.0)).collect(); let mut keys: Vec = replies @@ -1008,13 +1005,12 @@ fn reply_routes(mind: &dyn MindPort, told: &BTreeSet) -> Result> = answers.iter().map(|brief| origins.get(&brief.id.0).cloned().flatten()).collect(); - let placed = reply_conversation(&of_answers); + let Some(conversation) = reply_conversation(&of_answers) else { continue }; routes.push(Route { id: header.id.id.0.clone(), ordinal: header.admission.ordinal, at_unix_ms: instant_ms(&header.admission.admitted_at).unwrap_or(0), - pointer: placed.is_none(), - conversation: placed.unwrap_or(Conversation::Public), + conversation, origin: of_answers.into_iter().flatten().next(), }); } @@ -1126,19 +1122,11 @@ struct HeardReply { /// The briefs it answers, each with its repo (empty when the brief's repo could not be read). answers: Vec<(String, String)>, text: String, - /// A pointer to a reply that could not be placed: it holds no answers and no text, only the id. - pointer: bool, } -/// What a pointer line says, in the organ's own words: the reply is in the mind under its id, and nothing of it is in this line. -const UNPLACED_NOTE: &str = "Your Self finished something it could not place in any one conversation. Its words are not here: read the reply by this id, privately, and decide where it belongs."; - /// A reply as one line: `role=self` is the organ's word that it is her Self's, and every field a Self wrote is quoted, so nothing in /// it can end the line or pass for a message, a notice or a field. fn reply_line(reply: &HeardReply) -> String { - if reply.pointer { - return format!("reply role=self at={} id={} unplaced=true note={}\n", at_text(reply.at_unix_ms), quoted(&reply.id), quoted(UNPLACED_NOTE)); - } let answers = reply .answers .iter() @@ -1428,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, yield_now: !pending && !own_untold && self.replies_elsewhere } + Polled { pending, other: false, yield_now: !pending && !own_untold && self.replies_elsewhere } } fn held(&self) -> Option { @@ -1460,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, yield_now: true }); + return Ok(Polled { pending: false, other: false, yield_now: true }); } self.read_replies(); self.note_repo_changes(); @@ -1473,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, yield_now: true }); + return Ok(Polled { pending: false, other: 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 @@ -1489,7 +1477,7 @@ impl SessionInbox for PersonaInbox<'_, C, S> Ok(self.polled(!self.waiting.is_empty() || self.retell.is_some() || revision_ready || !self.notices.is_empty() || !self.waiting_replies.is_empty(), any_untold)) } - fn tell(&mut self) -> Result { + fn tell(&mut self, _switch: bool) -> Result { let (heard, replies) = self.take_reply_lines(); let notices = format!("{heard}{}", std::mem::take(&mut self.notices).concat()); if self.revision_ready() @@ -1962,12 +1950,8 @@ fn decode_reply(text: &str, rooms: &[String], people: &[String], open: &[String] /// document cannot be read is left for the next read. Only the routes of the session's own conversation are given. fn unheard_replies(mind: &dyn MindPort, routes: &[Route]) -> Result> { let mut found = Vec::new(); - for Route { id, ordinal, at_unix_ms, pointer, .. } in routes { + for Route { id, ordinal, .. } in routes { let (id, ordinal) = (id.clone(), *ordinal); - if *pointer { - found.push(HeardReply { id, ordinal, at_unix_ms: *at_unix_ms, answers: Vec::new(), text: String::new(), pointer: true }); - continue; - } let one = Selection { keys: Some(vec![id.clone()]), projection: "document".into(), ..Selection::default() }; let PipelinePageItems::Documents(views) = mind.query(&one)?.items else { bail!("the mind answered headers to a document selection") }; let Some(view) = views.into_iter().next() else { continue }; @@ -1982,7 +1966,7 @@ fn unheard_replies(mind: &dyn MindPort, routes: &[Route]) -> Result Route { - Route { id: id.into(), ordinal: at_unix_ms, at_unix_ms, conversation, origin: None, pointer: false } + Route { id: id.into(), ordinal: at_unix_ms, at_unix_ms, conversation, origin: None } } fn at_ms(delivery: ChatDelivery, at_unix_ms: u64) -> ChatDelivery { @@ -6523,8 +6523,8 @@ fn the_oldest_waiting_thing_decides_the_next_session() { } #[test] -fn a_reply_to_a_public_brief_or_a_brief_with_no_origin_is_told_in_the_public_session_and_a_reply_to_no_brief_in_none() { - assert_eq!(reply_conversation(&[]), None); +fn a_reply_to_a_public_brief_a_brief_with_no_origin_or_no_brief_is_told_in_the_public_session() { + assert_eq!(reply_conversation(&[]), Some(Conversation::Public)); for origin in [None, Some(MemoryOrigin::Wake), Some(MemoryOrigin::SelfRun { run: "r".into() }), Some(MemoryOrigin::Room { rooms: "swarm".into(), heard_from: vec![] })] { assert_eq!(reply_conversation(&[origin]), Some(Conversation::Public)); } @@ -6539,21 +6539,14 @@ fn a_reply_to_a_public_brief_or_a_brief_with_no_origin_is_told_in_the_public_ses for process in processes { let all = process.concat(); let public = all.contains("public asks about oak"); - for words in ["old-brief words", "public-brief words"] { + for words in ["old-brief words", "public-brief words", "no-brief words"] { assert_eq!(all.contains(words), public, "{words} is told in the public session only: {all}"); } - assert!(!all.contains("no-brief words"), "a reply the Self could not attribute puts no text in any session: {all}"); } } -/// The inputs of the sessions that were told a pointer, and every input the model was given. -fn pointed_inputs(rig: &Rig) -> (Vec, Vec) { - let all: Vec = rig.model.requests.borrow().iter().map(|told| told.input.clone()).collect(); - (all.iter().filter(|input| input.contains("unplaced=true")).cloned().collect(), all) -} - #[test] -fn a_reply_whose_briefs_name_two_conversations_is_a_text_free_pointer_in_the_public_session_only() { +fn a_reply_whose_briefs_name_two_conversations_is_told_in_no_session_and_stays_in_force() { assert_eq!(reply_conversation(&[Some(dm_origin(AQUA_DM, "@a:chat")), Some(dm_origin(LIBBY_DM, "@l:chat"))]), None); assert_eq!(reply_conversation(&[Some(dm_origin(AQUA_DM, "@a:chat")), None]), None); let pending = vec![direct(1, AQUA_DM, "@aqua:chat", AQUA_ASK), direct(2, LIBBY_DM, "@libby:chat", "libby asks about cedar"), delivery(3, "swarm", "@m:chat", "public asks about oak")]; @@ -6561,28 +6554,9 @@ fn a_reply_whose_briefs_name_two_conversations_is_a_text_free_pointer_in_the_pub rig.mind.committed(briefs_in_every_room()); rig.mind.committed(vec![self_reply_to("r-both", &["ask-aqua", "ask-libby"], "both-rooms words")]); assert_eq!(failure_of(rig.go()), None); - let (pointed, all) = pointed_inputs(&rig); - assert_eq!(pointed.len(), 1, "one pointer, in one session: {all:?}"); - assert!(pointed[0].contains(&reply_id("r-both")), "{}", pointed[0]); - assert!(!pointed[0].contains(AQUA_ASK) && !pointed[0].contains("libby asks about cedar"), "no direct message in the pointer's session: {}", pointed[0]); - assert!(!all.iter().any(|input| input.contains("both-rooms words")), "no reply text reaches any line"); - // The Persona reads it by id: the reply document holds the words the line does not. - let PipelineDocument::Reply(reply) = rig.mind.view(K::Reply, &reply_id("r-both")).unwrap().document else { panic!("a reply") }; - assert_eq!(reply.text.0, "both-rooms words"); - assert!(!reply_in_force(&rig.mind, &reply_id("r-both")).unwrap(), "heard once, it no longer re-pages on every read"); -} - -#[test] -fn a_reply_that_answers_no_brief_is_a_text_free_pointer_with_its_id() { - let rig = Rig::new(vec![delivery(1, "swarm", "@m:chat", "public asks about oak")], &[says(&[]), says(&[])]); - rig.mind.committed(briefs_in_every_room()); - rig.mind.committed(vec![self_reply_to("r-none", &[], "stranded words")]); - assert_eq!(failure_of(rig.go()), None); - let (pointed, all) = pointed_inputs(&rig); - assert_eq!(pointed.len(), 1, "{all:?}"); - assert!(pointed[0].contains(&reply_id("r-none")) && pointed[0].contains("could not place"), "{}", pointed[0]); - assert!(!all.iter().any(|input| input.contains("stranded words"))); - assert!(!reply_in_force(&rig.mind, &reply_id("r-none")).unwrap()); + let all: Vec = rig.model.requests.borrow().iter().map(|told| told.input.clone()).collect(); + assert!(!all.iter().any(|input| input.contains("both-rooms words") || input.contains("unplaced")), "no session is told a reply of two conversations, nor a pointer to it: {all:?}"); + assert!(reply_in_force(&rig.mind, &reply_id("r-both")).unwrap(), "told in no session, it stays in force"); } #[test] diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index d868d6e..8dafe0e 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -55,7 +55,7 @@ //! foreign owner's JSON: it is parsed into `serde_json::Value` here and nothing //! parsed is stored but a result's text, as a reply. -use std::collections::{HashMap, HashSet}; +use std::collections::HashSet; use std::ffi::{OsStr, OsString}; use std::fmt; use std::io::{BufRead, BufReader, Write}; @@ -152,7 +152,7 @@ pub const SESSION_NAMESPACE: Uuid = Uuid::NAMESPACE_OID; /// skill: it names the skill, the instance and the run id. pub fn start_message(instance: &Slug, run: &PipelineRef, plan_quota: &[PlanQuota], now_ms: u64) -> String { format!( - "Use the eureka skill. You are Self for the instance `{}`; your run is `{}`. Admit as faculty SelfFaculty, agent {AGENT}.\n\nThis session stays open after you answer: briefs from your Persona arrive while you work and between tasks. When you answer a brief, resolve follow-up Recorded with your answer as rationale; if you decline it, resolve it Withdrawn with why. Work you take on beyond an answer is a follow-up of your own. What you say when you end a turn reaches your Persona as a reply, and your Persona's answers come as briefs. Briefs reach you one conversation at a time. What you say ending a turn reaches your Persona in the conversation of the briefs that turn took in. When a background task you started finishes, its result goes to the conversation of the turn that started it, and no other conversation's brief is told into that turn. A result that cannot be placed reaches your Persona only as a pointer to it by id, without its text.\n\n{}", + "Use the eureka skill. You are Self for the instance `{}`; your run is `{}`. Admit as faculty SelfFaculty, agent {AGENT}.\n\nThis session stays open after you answer: briefs from your Persona arrive while you work and between tasks. When you answer a brief, resolve follow-up Recorded with your answer as rationale; if you decline it, resolve it Withdrawn with why. Work you take on beyond an answer is a follow-up of your own. What you say when you end a turn reaches your Persona as a reply, and your Persona's answers come as briefs. Briefs reach you one conversation at a time. More briefs of the conversation you are serving may arrive while you work; another conversation's brief is told only once you are idle, with no turn running and no background task of yours still running or waiting for its turn. What you say ending any turn, the turns that report your background tasks included, reaches your Persona in the conversation you are serving, so a long background task keeps other members waiting.\n\n{}", instance.0, run.id.0, render_plan(plan_quota, now_ms) @@ -228,6 +228,9 @@ pub fn session_id(run: &PipelineRef) -> Uuid { /// the seed fills); the working directory's project and local settings, which a /// run could have written, do not. pub fn claude_args(session: Uuid) -> Vec { + // `--disallowedTools` (probe, claude 2.1.312): the init line's tools lose these two. A turn the CLI begins on its own must be a + // notification turn the actuator tracks (a task's), so no wake of the Self's own is scheduled in the CLI: her heartbeat owns her wakes. + const DENIED_TOOLS: &str = "ScheduleWakeup,CronCreate"; [ "-p", "--input-format", @@ -246,11 +249,16 @@ pub fn claude_args(session: Uuid) -> Vec { "--setting-sources", "user", "--replay-user-messages", + "--disallowedTools", + DENIED_TOOLS, ] .map(String::from) .to_vec() } +/// What a user line starts with when the CLI replays a task notification to the model. +const NOTIFICATION_TAG: &str = ""; + /// The model name on the CLI's own assistant lines, which no request produced. const SYNTHETIC_MODEL: &str = ""; @@ -260,8 +268,17 @@ 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 user line of the main thread that the CLI took into a request (a replay echo): ours, never a tool result. + /// 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. + NotificationTaken { id: String }, /// A turn ended. `ok`: not an error. `blocked`: the CLI refused the context (`blocking_limit`). Result { ok: bool, blocked: bool }, /// Valid JSON that is none of the above, or a request of a subagent. @@ -290,12 +307,26 @@ 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"), }, - Some("user") - if event.get("isReplay").and_then(Value::as_bool) == Some(true) - && event.get("parent_tool_use_id").is_none_or(Value::is_null) - && event.get("message").and_then(|message| message.get("content")).is_some_and(Value::is_string) => - { - Usage::Echo + // Task lines are read for liveness only: they name a task by its `tool_use_id` and never a conversation. + Some("system") => { + let id = || event.get("tool_use_id").and_then(Value::as_str).map(String::from); + 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 }), + _ => Usage::Other, + } + } + Some("user") if event.get("isReplay").and_then(Value::as_bool) == Some(true) && event.get("parent_tool_use_id").is_none_or(Value::is_null) => { + 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("")) + .map_or(Usage::Other, |(id, _)| Usage::NotificationTaken { id: id.to_string() }), + Some(_) => Usage::Echo, + None => Usage::Other, + } } Some("assistant") if event.get("parent_tool_use_id").is_none_or(Value::is_null) => { let message = event.get("message"); @@ -532,11 +563,13 @@ pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawne }; claimed.push((claim, of)); } - let conversation = claimed.iter().find_map(|(_, of)| of.clone()).or_else(|| open.first().map(|brief| brief.conversation.clone())); + let first = claimed.iter().find_map(|(_, of)| of.clone()).or_else(|| open.first().map(|brief| brief.conversation.clone())); // A claim of another conversation is not rendered or carried: the start line holds one conversation only. - let claims: Vec = claimed.into_iter().filter(|(claim, of)| claim.kind != PipelineKind::FollowUp || *of == conversation).map(|(claim, _)| claim).collect(); + let claims: Vec = claimed.into_iter().filter(|(claim, of)| claim.kind != PipelineKind::FollowUp || *of == first).map(|(claim, _)| claim).collect(); + // The conversation the session serves from its start: the claim's, else the oldest open brief's, else the public one. + let conversation = first.clone().unwrap_or(Conversation::Public); let mut task = task_section(mind, &claims)?; - let mut told: Vec = open.into_iter().filter(|brief| claims.iter().any(|claim| claim.id.0 == brief.id) || Some(&brief.conversation) == conversation.as_ref()).collect(); + let mut told: Vec = open.into_iter().filter(|brief| claims.iter().any(|claim| claim.id.0 == brief.id) || Some(&brief.conversation) == first.as_ref()).collect(); let also: Vec<&Brief> = told.iter().filter(|brief| !claims.iter().any(|claim| claim.id.0 == brief.id)).collect(); if !also.is_empty() { task.push_str(" @@ -561,7 +594,7 @@ Also open for you:"); } #[allow(clippy::too_many_arguments, reason = "carry's ports, passed on whole")] -fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write, (run, claims, task, conversation): (&PipelineRef, &[PipelineRef], &str, Option), told: &mut Vec, heard: &mut Heard) -> Result { +fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write, (run, claims, task, conversation): (&PipelineRef, &[PipelineRef], &str, Conversation), told: &mut Vec, heard: &mut Heard) -> Result { let started_ms = unix_ms(clock.now()); let request = LeaseRequest { expires_at_unix_ms: started_ms.saturating_add(run_cap_s.saturating_mul(1000)) }; let lease = match lender.lend(&request)? { @@ -580,6 +613,10 @@ fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, s } /// The Self's side of the session: her briefs, derived from the mind, are what she is told. +/// +/// One conversation at a time (ruling `eureka-body:ruling:self-serves-one-conversation-at-a-time`): the session serves one conversation, and +/// every result belongs to it. The actuator changes it only in a write `decide` marks `switch`, which `decide` issues only when the session is +/// idle; a result is never attributed from the order of the stream. struct SelfInbox<'a> { mind: &'a dyn MindPort, start: String, @@ -593,21 +630,12 @@ struct SelfInbox<'a> { carried: Vec, /// The briefs of the lines the CLI took, which the next reply answers. echoed: Vec, - /// The conversation of the briefs carried or echoed and not yet answered by a reply: while it is set, only that conversation's briefs - /// are told, so what she says ending a turn reaches her Persona in one conversation. None when every brief told has been answered. - conversation: Option, - /// The briefs the last reply answered. A result that took in no brief and follows no task notification continues that turn, so it - /// answers them, only while every brief this run told is of one conversation; once two were told it answers nothing and is pointed at. + /// The conversation being served: the start line's, then the one each switch writes. Its new briefs may be written at any time; another + /// conversation's only at the switch. + serving: Conversation, + /// The briefs the last reply answered, or the briefs of the line that began this span. A result that took in no brief (a notification + /// turn, an error, an empty answer) answers them: it belongs to the span being served. last_answers: Vec, - /// What each background task answers, by the `tool_use_id` the stream carries on its start: the briefs of the turn that started it. - /// A notification's result goes to that conversation. - tasks: HashMap>, - /// The `tool_use_id`s of task notifications whose turn has not yet ended in a result that took in no brief. While any is pending, a - /// brief is written only when it is of the conversation the notification is attributed to. - notified: Vec, - /// A turn that took in no brief is open: a request came with nothing echoed, and no line of ours has joined it. Until its result, a brief - /// is written only when it is of the conversation that turn is attributed to (the notification'"'"'s task, else the turn before). - briefless_open: bool, /// The flush was written and the CLI has not yet taken it. flush_written: bool, /// The CLI took the flush and its result has not come: that result is the session's note to its successor, never a reply to a brief. @@ -619,8 +647,8 @@ struct SelfInbox<'a> { } impl<'a> SelfInbox<'a> { - /// The start line carries the claimed follow-ups and every brief already open (`told`). - fn new(mind: &'a dyn MindPort, (run, claims): (&PipelineRef, &[PipelineRef]), task: &str, (plan_quota, now_ms): (&[PlanQuota], u64), (told, conversation): (&'a mut Vec, Option), heard: &'a mut Heard) -> Self { + /// The start line carries the claimed follow-ups and every brief already open (`told`), all of `serving`'s conversation. + fn new(mind: &'a dyn MindPort, (run, claims): (&PipelineRef, &[PipelineRef]), task: &str, (plan_quota, now_ms): (&[PlanQuota], u64), (told, serving): (&'a mut Vec, Conversation), heard: &'a mut Heard) -> Self { let start = format!("{}{task}", start_message(mind.instance(), run, plan_quota, now_ms)); let mut carried: Vec = claims.iter().filter(|claim| claim.kind == PipelineKind::FollowUp).map(|claim| claim.id.0.clone()).collect(); for brief in told.iter() { @@ -628,7 +656,8 @@ impl<'a> SelfInbox<'a> { carried.push(brief.id.clone()); } } - Self { mind, start, plan: render_plan(plan_quota, now_ms), told, waiting: Vec::new(), heard, run: run.clone(), carried, echoed: Vec::new(), conversation, last_answers: Vec::new(), tasks: HashMap::new(), notified: Vec::new(), briefless_open: false, flush_written: false, flushing: false, replies: 0, unadmitted: Vec::new() } + let last_answers = carried.clone(); + Self { mind, start, plan: render_plan(plan_quota, now_ms), told, waiting: Vec::new(), heard, run: run.clone(), carried, echoed: Vec::new(), serving, last_answers, flush_written: false, flushing: false, replies: 0, unadmitted: Vec::new() } } /// Admits each reply the mind has not taken, in order; one try each. A refusal or an unreachable mind keeps the reply and never ends the session. @@ -644,19 +673,18 @@ impl<'a> SelfInbox<'a> { self.unadmitted = kept; } - /// One result of the main thread: with words, one reply answering the briefs the CLI has taken since the last reply (the first 64, in tell order). - /// An error or an empty result makes none and leaves those briefs for the next reply. + /// One result of the main thread: with words, one reply answering the briefs the CLI has taken since the last reply (the first 64, in tell + /// order), else the briefs of the span being served. An error or an empty result makes none and leaves those briefs for the next reply. fn reply_to(&mut self, line: &str, ok: bool) { let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return }; let text = result_text(&event); let flush = std::mem::take(&mut self.flushing); - let took = self.echoed.first().and_then(|id| self.conversation_of(id)); - // The flush result is the session's note to its successor, in the journal: it is no reply, so nothing ever points at it. + // The flush result is the session's note to its successor, in the journal: it is no reply. if ok && !flush && !text.trim().is_empty() { self.replies += 1; let run_label = self.run.id.0.rsplit(':').next().unwrap_or(&self.run.id.0); let taken = std::mem::take(&mut self.echoed); - let answered: Vec = if taken.is_empty() { self.briefless_answers() } else { taken.into_iter().take(64).collect() }; + let answered: Vec = if taken.is_empty() { self.last_answers.clone() } else { taken.into_iter().take(64).collect() }; let answers = answered.iter().map(|id| PipelineRef { kind: PipelineKind::FollowUp, id: id.as_str().into() }).collect(); if !answered.is_empty() { self.last_answers = answered; @@ -669,82 +697,7 @@ impl<'a> SelfInbox<'a> { text: cut_text(text, REPLY_TEXT_LIMIT).as_str().into(), }); } - // The turn has ended. A turn that took in no brief answered the notifications pending; one that took in the notification's own conversation's briefs - // took the notification in with them. A notification of another conversation stays pending: its turn is still to come. - if took.is_none() || self.turn_conversation() == took { - self.notified.clear(); - } - self.briefless_open = false; self.admit_unadmitted(); - self.conversation = self.carried.first().or(self.echoed.first()).and_then(|id| self.conversation_of(id)); - } - - /// Whether every brief this run told is of one conversation. - fn one_conversation_told(&self) -> bool { - self.told.iter().all(|brief| brief.conversation == self.told[0].conversation) - } - - /// The briefs a turn that took in no brief answers: those of the turns that started the tasks whose notifications are pending - /// (any notification the stream did not attribute makes the result unplaceable), else, with no notification, the turn before - /// while the run has told one conversation only. - fn briefless_answers(&self) -> Vec { - let continued = if self.one_conversation_told() { self.last_answers.clone() } else { Vec::new() }; - if self.notified.is_empty() { - return continued; - } - let mut answers: Vec = Vec::new(); - for id in &self.notified { - let own = self.tasks.get(id).filter(|briefs| !briefs.is_empty()); - let Some(own) = own else { return Vec::new() }; - answers.extend(own.iter().filter(|brief| !answers.contains(brief)).cloned().collect::>()); - } - answers - } - - /// The one conversation the open turn is attributed to: its notifications' tasks', else the turn before. None when they name two, or one could not be attributed. - fn turn_conversation(&self) -> Option { - let answers = self.briefless_answers(); - let mut conversations = answers.iter().map(|id| self.conversation_of(id)); - let first = conversations.next()??; - conversations.all(|conversation| conversation.as_ref() == Some(&first)).then_some(first) - } - - /// What a task's start or a notification in the stream says: a task is answered by the turn that started it. - fn note_task(&mut self, line: &str) { - if !line.contains("task_started") && !line.contains("task_notification") { - return; - } - let Ok(Value::Object(event)) = serde_json::from_str::(line) else { return }; - if event.get("type").and_then(Value::as_str) != Some("system") { - return; - } - let Some(tool_use) = event.get("tool_use_id").and_then(Value::as_str) else { return }; - match event.get("subtype").and_then(Value::as_str) { - Some("task_started") => { - // The turn that starts it is answering the briefs echoed, else what a briefless turn answers now. - let answers = if self.echoed.is_empty() { self.briefless_answers() } else { self.echoed.iter().take(64).cloned().collect() }; - self.tasks.entry(tool_use.to_string()).or_insert(answers); - } - Some("task_notification") => self.notified.push(tool_use.to_string()), - _ => {} - } - } - - /// The conversation of a brief told in this run. - fn conversation_of(&self, id: &str) -> Option { - self.told.iter().find(|brief| brief.id == id).map(|brief| brief.conversation.clone()) - } - - /// Whether the open turn lets this brief in: while a task notification or another turn that took in no brief is open, only the conversation - /// it is attributed to (none when it cannot be placed), unless nothing has been told yet and no other conversation exists. - fn turn_allows(&self, brief: &Brief) -> bool { - (self.notified.is_empty() && (!self.briefless_open || self.told.is_empty())) || self.turn_conversation().as_ref() == Some(&brief.conversation) - } - - /// Whether a waiting brief may be told now: any when every brief told has been answered, else only the unanswered ones' conversation. - /// While a task notification's turn is open, only the conversation the notification is attributed to may be told (none when it cannot be placed). - fn may_tell(&self, brief: &Brief) -> bool { - self.turn_allows(brief) && self.conversation.as_ref().is_none_or(|conversation| *conversation == brief.conversation) } } @@ -760,18 +713,26 @@ impl SessionInbox for SelfInbox<'_> { fn poll(&mut self) -> Result { let held: HashSet = self.told.iter().map(|brief| brief.id.clone()).collect(); self.waiting = inbox::derive(self.mind, &held)?; - Ok(Polled { pending: self.waiting.iter().any(|brief| self.may_tell(brief)), yield_now: false }) + 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 }) } - fn tell(&mut self) -> Result { - // The oldest waiting brief the gate lets through decides the line's conversation; the other conversations' briefs stay untold for a later poll. - let conversation = self.waiting.iter().find(|brief| self.may_tell(brief)).map(|brief| brief.conversation.clone()); - let (mut mine, rest): (Vec, Vec) = std::mem::take(&mut self.waiting).into_iter().partition(|brief| Some(&brief.conversation) == conversation.as_ref()); + /// The served conversation's waiting briefs; at a switch, those of the oldest waiting brief's conversation, which then is served and starts + /// its own span: no turn of it inherits the old one's briefs. + fn tell(&mut self, switch: bool) -> Result { + let conversation = if switch { self.waiting.first().map_or_else(|| self.serving.clone(), |brief| brief.conversation.clone()) } else { self.serving.clone() }; + let (mut mine, rest): (Vec, Vec) = std::mem::take(&mut self.waiting).into_iter().partition(|brief| brief.conversation == conversation); self.waiting = rest; let text = inbox::render(&mine, &self.plan); - self.carried.extend(mine.iter().map(|brief| brief.id.clone())); + let ids: Vec = mine.iter().map(|brief| brief.id.clone()).collect(); + if conversation != self.serving { + self.serving = conversation; + self.echoed.clear(); + self.last_answers = ids.clone(); + } + self.carried.extend(ids); self.told.append(&mut mine); - self.conversation = conversation; Ok(text) } @@ -782,16 +743,9 @@ impl SessionInbox for SelfInbox<'_> { fn line(&mut self, line: &str) { self.heard.hear(line); - self.note_task(line); - if matches!(read_usage(line), Usage::Request { .. }) && self.echoed.is_empty() { - self.briefless_open = true; - } } fn echoed(&mut self) { - if !self.carried.is_empty() { - self.briefless_open = false; - } self.echoed.append(&mut self.carried); self.flushing |= std::mem::take(&mut self.flush_written); } diff --git a/crates/mind-body/src/self_run/decide.rs b/crates/mind-body/src/self_run/decide.rs index ffa3769..96f1fce 100644 --- a/crates/mind-body/src/self_run/decide.rs +++ b/crates/mind-body/src/self_run/decide.rs @@ -15,7 +15,13 @@ //! replay echo increments `consumed`. The echo is the point the CLI takes a line //! into a request, folded into the running turn or queued for the next. What //! waits (the briefs, the flush ask) is coalesced into the next single write. -//! Idle is derived: no turn open and no write outstanding. +//! 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 +//! 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: //! (1) a result that is blocked, or an error before any answer, closes stdin: no @@ -36,7 +42,7 @@ //! of an event; a deadline that is already due yields its action at the next //! event, and `next_deadline` says it is now. -use std::collections::HashSet; +use std::collections::{BTreeMap, BTreeSet, HashSet}; use chrono::{DateTime, Duration, Utc}; @@ -64,8 +70,16 @@ pub(super) enum Event { 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 waits to be told. - Inbox { pending: 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 }, + /// 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. + NotificationTaken { id: String }, /// `waitpid` returned: the child's exit code, none when a signal ended it. ChildExited(Option), StdoutEof, @@ -80,8 +94,8 @@ pub(super) enum Event { #[derive(Clone, Debug, PartialEq, Eq)] pub(super) enum Action { None, - /// The one write: the waiting briefs and the flush ask, coalesced into one user line. - Write { tells: bool, flush: Option }, + /// The one write: the waiting briefs and the flush ask, coalesced into one user line. `switch`: the briefs are another conversation's, which is written only at idle and never with the flush. + Write { tells: bool, switch: bool, flush: Option }, CloseStdin, /// To the child's whole process group. Signal(Stop), @@ -102,6 +116,24 @@ struct Flush { why: Why, } +/// Who began the open turn: decided by the first of an echo (our line) and a request (the CLI's own). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum TurnBy { + #[default] + Unknown, + Us, + Cli, +} + +/// A task notification that has come and has not yet been given its turn. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct Owed { + /// The turn count when it came: the turn open then (or the last one) is not the turn that settles it. + since: u32, + /// The CLI replayed it to the model: the result of the turn that did settles it. + taken: bool, +} + #[derive(Clone, Debug, Default, PartialEq, Eq)] pub(super) struct State { seen: HashSet, @@ -111,6 +143,15 @@ pub(super) struct State { pub(super) written: u32, pub(super) consumed: u32, turn_open: bool, + /// 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. + owed: BTreeMap, + /// A brief of another conversation than the one served waits. + other: bool, /// An ok result has come: the session holds an answer worth its cache. answered_once: bool, last_result_at: Option>, @@ -143,8 +184,23 @@ 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. fn idle(&self) -> bool { - !self.turn_open && !self.outstanding() + !self.turn_open && !self.outstanding() && self.tasks.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. + fn open_turn(&mut self, by: Option) { + if !self.turn_open { + self.turn_open = true; + self.turns += 1; + self.turn_by = TurnBy::Unknown; + } + if self.turn_by == TurnBy::Unknown + && let Some(by) = by + { + self.turn_by = by; + } } /// Seconds idle since the last result, none while a line or a turn is outstanding. @@ -183,7 +239,7 @@ fn due(state: &State) -> Option<(DateTime, Why)> { return None; } let rotation = state.cap.map(|(started, cap)| (started + Duration::seconds(cap.saturating_sub(ROTATE_MARGIN_S)), Why::Rotation)); - let idle = (state.idle() && !state.inbox).then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Why::Idle)); + let idle = (state.idle() && !state.inbox && !state.other).then_some(state.last_result_at).flatten().map(|last| (last + Duration::seconds(state.ttl_s - TTL_MARGIN_S), Why::Idle)); [rotation, idle].into_iter().flatten().min_by_key(|(at, _)| *at) } @@ -237,7 +293,9 @@ fn send(state: &mut State) -> Action { return Action::None; } let flush = state.flush.as_ref().map(|flush| flush.why); - if !state.inbox && flush.is_none() { + // 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(); + if !state.inbox && !switch && flush.is_none() { return Action::None; } state.written += 1; @@ -245,7 +303,10 @@ fn send(state: &mut State) -> Action { if let Some(flush) = state.flush.as_mut() { flush.seq = Some(seq); } - Action::Write { tells: std::mem::take(&mut state.inbox), flush } + if switch { + state.other = false; + } + Action::Write { tells: std::mem::take(&mut state.inbox) || switch, switch, flush } } pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (State, Action) { @@ -254,7 +315,24 @@ 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); + } + // 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) { + let since = state.turns; + state.owed.insert(id, Owed { since, taken: false }); + } + } + Event::NotificationTaken { id } => { + if let Some(owed) = state.owed.get_mut(&id) { + owed.taken = true; + } + } Event::Request { id, prefix } => { + state.open_turn(Some(TurnBy::Cli)); if state.seen.insert(id) { state.requests += 1; state.prefix = prefix; @@ -271,7 +349,7 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta } Event::Echo => { state.consumed += 1; - state.turn_open = true; + state.open_turn(Some(TurnBy::Us)); let consumed = state.consumed; if let Some(flush) = state.flush.as_mut() && flush.seq.is_some_and(|seq| consumed >= seq) @@ -281,6 +359,10 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta } Event::Result { ok, blocked, ttl_1h } => { state.turn_open = false; + // A notification is settled by the result of a turn that took it in, or of a turn the CLI began after it came. A turn our line + // began settles none it did not replay: if the CLI merged it, the hold lasts to the rotation flush. + let (turns, by) = (state.turns, state.turn_by); + state.owed.retain(|_, owed| !(owed.taken || (by != TurnBy::Us && owed.since < turns))); state.last_result_at = Some(now); state.ttl_s = if ttl_1h { TTL_1H_S } else { TTL_5M_S }; let no_answer = !ok && !state.answered_once; @@ -291,7 +373,10 @@ pub(super) fn decide(mut state: State, event: Event, now: DateTime) -> (Sta action = Action::CloseStdin; } } - Event::Inbox { pending } => state.inbox = pending, + Event::Inbox { pending, other } => { + state.inbox = pending; + state.other = other; + } Event::Yield => state.yielding = true, Event::ChildExited(code) => { state.exited.get_or_insert((code, now)); diff --git a/crates/mind-body/src/self_run/decide/tests.rs b/crates/mind-body/src/self_run/decide/tests.rs index 7f06b6c..8647155 100644 --- a/crates/mind-body/src/self_run/decide/tests.rs +++ b/crates/mind-body/src/self_run/decide/tests.rs @@ -51,11 +51,11 @@ fn last(events: Vec<(Event, i64)>) -> Action { /// The one write with only the waiting briefs, and with only the flush. fn tells() -> Action { - Action::Write { tells: true, flush: None } + Action::Write { tells: true, switch: false, flush: None } } fn flush(why: Why) -> Action { - Action::Write { tells: false, flush: Some(why) } + Action::Write { tells: false, switch: false, flush: Some(why) } } fn delivered(how: Stop) -> Event { @@ -87,16 +87,16 @@ 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 }, 2), (Event::Tick, 3)]); + let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 1), (Event::Inbox { pending: true, other: 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. let (state, action) = feed(state, Event::Echo, 4); - assert_eq!(action, Action::Write { tells: true, flush: Some(Why::Ceiling) }); + assert_eq!(action, Action::Write { tells: true, switch: false, flush: Some(Why::Ceiling) }); assert_eq!((state.written, state.consumed), (2, 1)); assert_eq!(next_deadline(&state), Some(t(1 + GRACE)), "the grace runs from the ask"); // Briefs alone: the first poll writes, the second finds that write outstanding and waits for its echo. - let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true }, 1), (Event::Inbox { pending: true }, 2)]); + let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true, other: false }, 1), (Event::Inbox { pending: true, other: 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 }, 8)]); + let (state, actions) = run(vec![(Event::Echo, 0), (OK, 7), (Event::Inbox { pending: true, other: 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 }, 1), (OK, 2)]); + let (state, _) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true, other: 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 }, 2), (Event::Echo, 3), (FAILED, 4)]); + let (state, actions) = run(vec![(Event::Echo, 0), (OK, 1), (Event::Inbox { pending: true, other: 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 }, 1)]); + let (state, actions) = run(vec![(Event::Echo, 0), (Event::Inbox { pending: true, other: 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 }, 2).1, Action::None); + assert_eq!(feed(state.clone(), Event::Inbox { pending: false, other: false }, 2).1, Action::None); assert_eq!(feed(state, Event::Tick, 3).1, Action::None); // Idle: the same. let (_, actions) = idle_at(5); assert!(actions.iter().all(|action| *action == Action::None)); - assert_eq!(feed(idle_at(5).0, Event::Inbox { pending: true }, 6).1, tells()); + assert_eq!(feed(idle_at(5).0, Event::Inbox { pending: true, other: 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 }, 1).1, Action::None); + assert_eq!(feed(state.clone(), Event::Inbox { pending: true, other: 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 }, 11)]); + let (waiting, _) = run_from(idle_at(10).0, vec![(Event::Inbox { pending: true, other: 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)]); @@ -397,10 +397,11 @@ enum Sym { Failed, Waiting, Nothing, + Other, Late, } -const SYMBOLS: [Sym; 8] = [Sym::Request, Sym::Ceiling, Sym::Echo, Sym::Ok, Sym::Failed, Sym::Waiting, Sym::Nothing, Sym::Late]; +const SYMBOLS: [Sym; 9] = [Sym::Request, Sym::Ceiling, Sym::Echo, Sym::Ok, Sym::Failed, Sym::Waiting, Sym::Nothing, Sym::Other, Sym::Late]; /// Plays the symbols and asserts, from the actions alone, the write accounting /// and the single-writer invariants. @@ -425,15 +426,17 @@ fn play(symbols: &[Sym]) { } Sym::Ok => OK, Sym::Failed => FAILED, - Sym::Waiting => Event::Inbox { pending: true }, - Sym::Nothing => Event::Inbox { pending: false }, + Sym::Waiting => Event::Inbox { pending: true, other: false }, + Sym::Nothing => Event::Inbox { pending: false, other: false }, + Sym::Other => other_waits(), Sym::Late => Event::Tick, }; let (next, action) = feed(state, event, now); state = next; match action { - Action::Write { tells: told, flush } => { + Action::Write { tells: told, switch, flush } => { assert!(!flushed && !closed, "a write after the flush or the close: {symbols:?}"); + assert!(!(switch && flush.is_some()), "a switch carried the flush: {symbols:?}"); writes += 1; tells += u32::from(told); if flush.is_some() { @@ -500,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 }, 6).1, Action::None, "closed: text that waits is told to the next session"); + assert_eq!(feed(state, Event::Inbox { pending: true, other: false }, 6).1, Action::None, "closed: text that waits is told to the next session"); } #[test] @@ -510,3 +513,109 @@ fn a_flush_already_written_ends_the_session_by_itself_and_a_yield_does_not_cut_i let (_, action) = feed(state, Event::Yield, 3); assert_eq!(action, Action::None, "the flush's own answer closes stdin"); } + +fn started(id: &str) -> Event { + Event::TaskStarted { id: id.to_string() } +} + +fn notified(id: &str) -> Event { + Event::TaskNotified { id: id.to_string() } +} + +fn taken(id: &str) -> Event { + Event::NotificationTaken { id: id.to_string() } +} + +fn switches() -> Action { + Action::Write { tells: true, switch: true, flush: None } +} + +fn other_waits() -> Event { + Event::Inbox { pending: false, other: true } +} + +/// The start message echoed and answered: a Self turn of ours is over, at second 2. +fn answered_start() -> Vec<(Event, i64)> { + vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (OK, 2)] +} + +#[test] +fn a_turn_opens_at_the_first_turn_line_echo_or_request_and_the_first_echo_or_request_says_who_began_it() { + let by = |events: Vec<(Event, i64)>| { + let (state, _) = run(events); + (state.turns, state.turn_by) + }; + assert_eq!(by(vec![(Event::TurnBegan, 0)]), (1, TurnBy::Unknown), "an init line opens a turn nobody has claimed"); + assert_eq!(by(vec![(Event::TurnBegan, 0), (Event::Echo, 0)]), (1, TurnBy::Us), "an echo first: our line began it"); + assert_eq!(by(vec![(Event::TurnBegan, 0), (req("a", 10), 0)]), (1, TurnBy::Cli), "a request first: the CLI began it"); + assert_eq!(by(vec![(Event::TurnBegan, 0), (req("a", 10), 0), (Event::Echo, 1)]), (1, TurnBy::Cli), "a folded echo does not change who began it"); + assert_eq!(by(vec![(Event::TurnBegan, 0), (Event::Echo, 0), (req("a", 10), 1), (Event::Echo, 2)]), (1, TurnBy::Us)); + assert_eq!(by(vec![(Event::Echo, 0), (req("a", 10), 1)]), (1, TurnBy::Us), "no init line: the echo opens it"); + assert_eq!(by([answered_start(), vec![(Event::TurnBegan, 3), (req("b", 10), 3)]].concat()), (2, TurnBy::Cli), "the CLI's own turn after a result"); + assert_eq!(by([answered_start(), vec![(Event::TurnBegan, 3), (Event::Echo, 3)]].concat()), (2, TurnBy::Us)); +} + +#[test] +fn a_session_is_idle_only_with_no_turn_no_write_no_running_task_and_no_notification_owed_its_turn() { + let idle = |events: Vec<(Event, i64)>| run(events).0.idle_s(t(1_000)).is_some(); + 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(); + 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. + 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)]; + 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)]; + 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. + assert!(!idle([inside.clone(), vec![(Event::TurnBegan, 3), (Event::Echo, 3), (req("b", 10), 3), (OK, 4)]].concat())); + // The replay of the notification settles it with the result of the turn that took it, ours or the CLI's. + assert!(idle([inside.clone(), vec![(Event::TurnBegan, 3), (Event::Echo, 3), (taken("t"), 3), (req("b", 10), 3), (OK, 4)]].concat())); + assert!(idle([inside, vec![(Event::TurnBegan, 3), (req("b", 10), 3), (OK, 4)]].concat()), "a turn the CLI began after it came"); +} + +#[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)]), 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. + 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()); + // 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); +} + +#[test] +fn the_cache_end_flush_waits_for_a_running_background_task_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)]; + 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 (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"); + assert_eq!(last([running, notified_turn, vec![(Event::Tick, 4000 + ttl)]].concat()), flush(Why::Idle)); + // Another conversation's waiting brief is no reason to end the session at its cache's end either: it is switched to at idle. + assert_eq!(last([answered_start(), vec![(other_waits(), 2 + ttl)]].concat()), switches()); +} diff --git a/crates/mind-body/src/self_run/session.rs b/crates/mind-body/src/self_run/session.rs index 0a34884..0947019 100644 --- a/crates/mind-body/src/self_run/session.rs +++ b/crates/mind-body/src/self_run/session.rs @@ -20,8 +20,10 @@ use crate::launch::Clock; /// What one poll of an inbox found. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) struct Polled { - /// Something waits to be told. + /// A brief of the conversation being served waits to be told. pub(crate) pending: bool, + /// A brief of another conversation waits: told only when the session is idle (the switch). + pub(crate) other: bool, /// The session should end at its next idle: what is oldest and unanswered is not its business. pub(crate) yield_now: bool, } @@ -33,8 +35,8 @@ pub(crate) trait SessionInbox { /// Seconds between polls. fn poll_every_s(&self) -> i64; fn poll(&mut self) -> Result; - /// The text of what waits, which is thereby told. - fn tell(&mut self) -> Result; + /// The text of what waits, which is thereby told. `switch`: the session is idle and the text is another conversation's, the oldest waiting; else it is the served conversation's. + fn tell(&mut self, switch: bool) -> Result; fn flush_text(&mut self, why: Why) -> String; /// Every line of the child's stream, before it is read. fn line(&mut self, _line: &str) {} @@ -75,6 +77,10 @@ pub(crate) fn drive(clock: &dyn Clock, mut child: Box, inbox: &mut dy inbox.echoed(); Event::Echo } + Usage::TurnBegan => Event::TurnBegan, + Usage::TaskStarted { id } => Event::TaskStarted { id }, + Usage::TaskNotified { id } => Event::TaskNotified { id }, + Usage::NotificationTaken { id } => Event::NotificationTaken { id }, Usage::Result { ok, blocked } => { let (ok, blocked) = inbox.result(&line, ok, blocked)?; Event::Result { ok, blocked, ttl_1h } @@ -92,7 +98,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 }); + events.push_back(Event::Inbox { pending: found.pending, other: found.other }); if found.yield_now { events.push_back(Event::Yield); } @@ -103,7 +109,7 @@ pub(crate) fn drive(clock: &dyn Clock, mut child: Box, inbox: &mut dy while let Some(heard_event) = events.pop_front() { let counted = state.requests; // The decision's inputs, as they stood when the event came. - let (ttl_s, inbox_pending, idle_s) = (state.ttl_s, state.inbox || matches!(&heard_event, Event::Inbox { pending: true }), state.idle_s(now)); + let (ttl_s, inbox_pending, idle_s) = (state.ttl_s, state.inbox || matches!(&heard_event, Event::Inbox { pending: true, .. }), state.idle_s(now)); let (next, action) = decide(state, heard_event, now); state = next; if state.requests != counted { @@ -111,10 +117,10 @@ pub(crate) fn drive(clock: &dyn Clock, mut child: Box, inbox: &mut dy } match &action { Action::None => {} - Action::Write { tells, flush } => { + Action::Write { tells, switch, flush } => { let mut parts = Vec::new(); if *tells { - parts.push(inbox.tell()?); + parts.push(inbox.tell(*switch)?); } if let Some(why) = *flush { parts.push(inbox.flush_text(why)); diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index c47e3e7..462d6be 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -324,6 +324,8 @@ fn claude_args_is_exactly_the_fixed_list_for_a_session() { "--setting-sources", "user", "--replay-user-messages", + "--disallowedTools", + "ScheduleWakeup,CronCreate", ] ); // Those belong to the run measure or to the unit's managed settings. @@ -520,7 +522,12 @@ fn read_usage_classifies_the_owners_real_records() { assert_eq!(read_usage(¬_replayed), Usage::Other); let subagent_echo = json!({"type": "user", "isReplay": true, "parent_tool_use_id": "toolu_1", "message": {"role": "user", "content": "a line"}}).to_string(); assert_eq!(read_usage(&subagent_echo), Usage::Other, "a subagent's line is not ours"); - assert_eq!(read_usage(&init_line()), Usage::Other); + assert_eq!(read_usage(&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(&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"); @@ -1389,7 +1396,7 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_ // The plan line is restated under every brief for the run's life, so it names the instant of the reading, never an age that stops being true. assert!(told[0].contains("read at 2026-10-29T20:33Z") && !told[0].contains(" ago"), "{}", told[0]); assert_eq!(ran.written().len(), 2, "the start message and one tell: {:?}", ran.written()); - assert!(ran.out.contains("decision Write { tells: true, flush: None } prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out); + assert!(ran.out.contains("decision Write { tells: true, switch: false, flush: None } prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out); assert_eq!(ran.count(&Event::InputClosed), 0); // A result is not an answer: nothing closes the brief by its timing. assert!(resolutions_of(&mind, &brief_ref("b1")).is_empty()); @@ -1479,7 +1486,7 @@ fn the_session_ends_where_the_cache_would_be_evicted_with_a_flush_and_a_waiting_ let flush_at = hour.events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains(FLUSH_MESSAGE))).unwrap(); let closed_at = hour.events.iter().position(|event| *event == Event::InputClosed).unwrap(); assert!(flush_at < closed_at && hour.count(&Event::InputClosed) == 1, "{:?}", hour.events); - assert!(hour.out.contains("decision Write { tells: false, flush: Some(Idle) } prefix 10 ttl_s 3600 idle_s 3540 inbox false written 2 consumed 1\n"), "{}", hour.out); + assert!(hour.out.contains("decision Write { tells: false, switch: false, flush: Some(Idle) } prefix 10 ttl_s 3600 idle_s 3540 inbox false written 2 consumed 1\n"), "{}", hour.out); assert!(hour.written().iter().any(|text| text.contains("nothing is waiting and its cache would lapse") && text.contains("Resolve each brief you answered")), "{:?}", hour.written()); // Five minutes' cache: the flush at 240 s. let short = go(vec![Out(echo()), Out(five_minute_request("m1", 10)), Out(ok_result()), Idle(239), Idle(1), Out(echo()), Out(ok_result()), Idle(1)]); @@ -1538,7 +1545,7 @@ mod live_session { use super::*; fn flush(why: Why) -> Do { - Do::Write { tells: false, flush: Some(why) } + Do::Write { tells: false, switch: false, flush: Some(why) } } /// One recorded line as the loop would hear it: a label for the stream, the decision's action. @@ -1555,7 +1562,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 }) }; + let (label, fact) = if text.starts_with(FLUSH_MESSAGE) { ("flush-asked", Fact::Tick) } else { ("want", Fact::Inbox { pending: true, other: false }) }; let (next, action) = decide(state, fact, at); state = next; trace.push((label.to_string(), action)); @@ -1568,6 +1575,10 @@ mod live_session { let (label, fact) = match read_usage(line) { Usage::Request { id, prefix } => (format!("request:{prefix}"), Fact::Request { id, prefix }), Usage::Echo => ("echo".to_string(), Fact::Echo), + Usage::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::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), }; @@ -1596,12 +1607,18 @@ mod live_session { const API_ERROR: [&str; 2] = [include_str!("claude-2.1.312-live-session-api-error-first.jsonl"), include_str!("claude-2.1.312-live-session-api-error-first-scrub.jsonl")]; const BLOCKING: [&str; 2] = [include_str!("claude-2.1.312-live-session-blocking-limit.jsonl"), include_str!("claude-2.1.312-live-session-blocking-limit-scrub.jsonl")]; const COMPACTION_OFF: [&str; 2] = [include_str!("claude-2.1.312-live-session-compaction-off-360k.jsonl"), include_str!("claude-2.1.312-live-session-compaction-off-360k-scrub.jsonl")]; + const BG_IDLE: &str = include_str!("claude-2.1.312-live-session-bg-idle.jsonl"); + 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_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"); #[test] fn a_line_written_during_a_tool_call_is_folded_one_result_and_its_echo_before_the_request_that_carries_it() { let trace = replay(FOLDED); // The tool_result's own user line is no echo: it is a foreign owner's array content. - assert_eq!(kinds(&trace), ["other", "echo", "request:10", "other", "other", "other", "echo", "request:10", "result:ok"]); + assert_eq!(kinds(&trace), ["turn", "echo", "request:10", "other", "notified", "other", "echo", "request:10", "result:ok"]); assert!(acted(&trace).is_empty(), "{:?}", acted(&trace)); } @@ -1609,7 +1626,7 @@ mod live_session { fn a_line_written_while_the_last_request_is_in_flight_is_queued_two_results_and_its_echo_after_the_first() { for stream in QUEUED { let trace = replay(stream); - assert_eq!(kinds(&trace), ["other", "echo", "request:10", "result:ok", "other", "echo", "request:10", "result:ok"]); + assert_eq!(kinds(&trace), ["turn", "echo", "request:10", "result:ok", "turn", "echo", "request:10", "result:ok"]); assert!(acted(&trace).is_empty(), "{:?}", acted(&trace)); } } @@ -1633,10 +1650,10 @@ mod live_session { let (trace, state) = replay_state(stream); assert_eq!( kinds(&trace), - ["other", "want", "want", "echo", "request:10", "result:ok", "other", "echo", "request:300000", "result:ok", "flush-asked", "other", "echo", "request:10", "result:ok"] + ["turn", "want", "want", "echo", "request:10", "result:ok", "turn", "echo", "request:300000", "result:ok", "flush-asked", "turn", "echo", "request:10", "result:ok"] ); // Both briefs wait for the start message's echo and go as one line; the flush is written at the ceiling request; the result after its echo closes stdin. - assert_eq!(acted(&trace), vec![(3, Do::Write { tells: true, flush: None }), (8, flush(Why::Ceiling)), (14, Do::CloseStdin)]); + assert_eq!(acted(&trace), vec![(3, Do::Write { tells: true, switch: false, flush: None }), (8, flush(Why::Ceiling)), (14, Do::CloseStdin)]); assert_eq!((sent(stream), state.written, state.consumed), (3, 3, 3), "every line written has its own echo"); } } @@ -1644,7 +1661,7 @@ mod live_session { #[test] fn the_flush_folded_into_a_running_turn_is_answered_at_its_one_result() { let trace = replay(CEILING_FOLDED); - assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "other", "other", "other", "echo", "request:10", "result:ok"]); + assert_eq!(kinds(&trace), ["turn", "echo", "request:300000", "other", "notified", "other", "echo", "request:10", "result:ok"]); assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (8, Do::CloseStdin)], "never a grace interrupt"); } @@ -1652,7 +1669,7 @@ mod live_session { fn the_flush_queued_behind_the_turn_that_reached_the_ceiling_is_answered_at_its_own_second_result() { for stream in CEILING_QUEUED { let trace = replay(stream); - assert_eq!(kinds(&trace), ["other", "echo", "request:300000", "result:ok", "other", "echo", "request:10", "result:ok"]); + assert_eq!(kinds(&trace), ["turn", "echo", "request:300000", "result:ok", "turn", "echo", "request:10", "result:ok"]); // The first result ends the turn the flush was queued behind: it closes nothing. assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)], "never a grace interrupt"); } @@ -1662,7 +1679,7 @@ mod live_session { fn an_api_error_before_any_answer_closes_the_session_at_once_with_no_flush() { for stream in API_ERROR { let trace = replay(stream); - assert_eq!(kinds(&trace), ["other", "echo", "other", "result:error"]); + assert_eq!(kinds(&trace), ["turn", "echo", "other", "result:error"]); assert_eq!(acted(&trace), vec![(3, Do::CloseStdin)]); } } @@ -1671,7 +1688,7 @@ mod live_session { fn a_context_over_the_blocking_limit_is_a_blocked_result_that_closes_the_session() { for stream in BLOCKING { let trace = replay(stream); - assert_eq!(kinds(&trace), ["other", "echo", "request:990000", "result:ok", "other", "echo", "other", "result:blocked"]); + assert_eq!(kinds(&trace), ["turn", "echo", "request:990000", "result:ok", "turn", "echo", "other", "result:blocked"]); assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)]); } } @@ -1680,10 +1697,69 @@ mod live_session { fn with_compaction_off_a_request_at_360k_is_sent_and_answered_not_compacted_or_blocked() { for stream in COMPACTION_OFF { let trace = replay(stream); - assert_eq!(kinds(&trace), ["other", "echo", "request:360000", "result:ok", "other", "echo", "request:10", "result:ok"]); + assert_eq!(kinds(&trace), ["turn", "echo", "request:360000", "result:ok", "turn", "echo", "request:10", "result:ok"]); assert_eq!(acted(&trace), vec![(2, flush(Why::Ceiling)), (7, Do::CloseStdin)]); } } + + /// Whether the session is idle after each stream line, by the line's label. + fn idleness(jsonl: &str) -> Vec<(String, bool)> { + let lines: Vec<&str> = jsonl.lines().collect(); + (1..=lines.len()) + .filter_map(|n| { + let entry: Value = serde_json::from_str(lines[n - 1]).unwrap(); + entry.get("line")?.as_str()?; + let (trace, state) = replay_state(&lines[..n].join("\n")); + Some((trace.last().unwrap().0.clone(), state.idle_s(crate::testkit::at(1_000_000)).is_some())) + }) + .collect() + } + + fn significant(trace: &[(String, Do)]) -> Vec<&str> { + kinds(trace).into_iter().filter(|kind| *kind != "other").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); + } + + #[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 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("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"); + } + + #[test] + 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"]); + 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"); + assert!(!idle_after("request:10", 2), "the CLI-begun turn runs"); + assert!(idle_after("result:ok", 1)); + } } #[test] diff --git a/crates/mind-body/src/self_run/tests/conversation.rs b/crates/mind-body/src/self_run/tests/conversation.rs index b65029d..a0873fd 100644 --- a/crates/mind-body/src/self_run/tests/conversation.rs +++ b/crates/mind-body/src/self_run/tests/conversation.rs @@ -1,14 +1,25 @@ -//! One Self turn answers one conversation: her Persona tells a reply only in the conversation of the briefs it answers, so the actuator -//! never writes a line holding two conversations' briefs while the first's are unanswered. +//! A Self session serves one conversation at a time (ruling `eureka-body:ruling:self-serves-one-conversation-at-a-time`): another +//! conversation's brief is written only when the session is idle, and every result belongs to the conversation being served. The real-order +//! tests replay the claude 2.1.312 captures with briefs spliced in at named points of the stream, and carry synthetic words only. use eureka_pipeline::{MemoryOrigin, PipelineReply}; use super::*; +const BG_IDLE: &str = include_str!("../claude-2.1.312-live-session-bg-idle.jsonl"); +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"); + fn saying(text: &str) -> String { json!({"type": "result", "subtype": "success", "is_error": false, "terminal_reason": "completed", "result": text}).to_string() } +fn failing() -> String { + json!({"type": "result", "subtype": "error_during_execution", "is_error": true, "terminal_reason": "model_error", "result": "the API said no"}).to_string() +} + fn reply_of(mind: &TestMind, n: u32) -> Option { match mind.view(K::Reply, &format!("{INSTANCE}:reply:selfrun-a-r{n}"))?.document { PipelineDocument::Reply(reply) => Some(reply), @@ -35,10 +46,71 @@ fn libby(label: &str) -> PipelineDocument { brief_in(label, "!libby:chat", "@libby:chat") } +fn mira(label: &str) -> PipelineDocument { + brief_in(label, "!mira:chat", "@mira:chat") +} + fn id_of(label: &str) -> String { brief_ref(label).id.0 } +fn written_with(ran: &Ran, text: &str) -> bool { + ran.written().iter().any(|line| line.contains(text)) +} + +/// A mind whose Self run has one open brief of Aqua's: Aqua's conversation is the one served from the start. +fn serving_aqua() -> Rc { + let mind = live_mind(); + persona_commit(&mind, vec![aqua("a1")]); + mind +} + +// ---- the captures, with briefs spliced in at named points ----------------- + +/// The lines of a capture, in order: what the CLI said. The probe's own writes are dropped. +fn stream(jsonl: &str) -> Vec { + jsonl.lines().filter_map(|entry| serde_json::from_str::(entry).ok()?.get("line")?.as_str().map(String::from)).collect() +} + +fn label_of(line: &str) -> &'static str { + match read_usage(line) { + Usage::TurnBegan => "init", + Usage::Request { .. } => "request", + Usage::Echo => "echo", + Usage::Result { .. } => "result", + Usage::TaskStarted { .. } => "started", + Usage::TaskNotified { .. } => "notified", + Usage::NotificationTaken { .. } => "taken", + _ => "other", + } +} + +/// The index just after the `nth` (from 1) line of this label. +fn after(lines: &[String], label: &str, nth: usize) -> usize { + lines.iter().enumerate().filter(|(_, line)| label_of(line) == label).nth(nth - 1).map(|(at, _)| at + 1).unwrap_or_else(|| panic!("no {label} number {nth}")) +} + +/// The capture's lines as steps, with each splice's steps put in after the line at its index. +fn script(lines: &[String], splices: Vec<(usize, Vec)>) -> Vec { + let mut steps = Vec::new(); + for (at, line) in lines.iter().enumerate() { + steps.push(Out(line.clone())); + for (point, extra) in &splices { + if *point == at + 1 { + steps.extend(extra.iter().cloned()); + } + } + } + steps +} + +/// 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() +} + +// ---- who is served at the start -------------------------------------------- + #[test] fn the_start_line_carries_one_conversation_and_the_other_is_told_on_its_own_line() { let mind = TestMind::seeded(); @@ -72,97 +144,6 @@ fn with_no_claimed_brief_the_start_set_is_the_oldest_open_briefs_conversation() assert_eq!(answered(&mind, 2), vec![id_of("a1")]); } -/// Two conversations' briefs derived at one poll, and across polls while the first's are unanswered: the gate tells one conversation per -/// line and holds the other until a reply has answered. -#[test] -fn a_live_self_never_writes_a_line_holding_a_second_conversations_briefs_while_the_first_are_unanswered() { - let mind = live_mind(); - let steps = vec![ - Out(echo()), - Out(at("m1", 10)), - // The folded shape: both derived at one poll. - briefed(&mind, 31, vec![aqua("a1"), libby("l1")]), - Out(echo()), - // Across polls, while the turn that took a1 runs: a third conversation's brief arrives too. - briefed(&mind, 31, vec![brief_in("m1", "!mira:chat", "@mira:chat")]), - Idle(31), - Idle(31), - Out(saying("aqua done")), - Idle(31), - Out(echo()), - Out(saying("libby done")), - Idle(31), - Out(echo()), - Out(saying("mira done")), - ]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - let told = tells(&ran); - assert_eq!(told.len(), 3, "{:?}", ran.written()); - for (line, (own, others)) in told.iter().zip([("a1", ["l1", "m1"]), ("l1", ["a1", "m1"]), ("m1", ["a1", "l1"])]) { - assert!(line.contains(&id_of(own)), "{line}"); - assert!(others.iter().all(|other| !line.contains(&id_of(other))), "one conversation per line: {line}"); - } - assert_eq!((answered(&mind, 1), answered(&mind, 2), answered(&mind, 3)), (vec![id_of("a1")], vec![id_of("l1")], vec![id_of("m1")])); -} - -#[test] -fn the_same_conversations_briefs_are_still_told_while_its_turn_is_unanswered() { - let mind = live_mind(); - let steps = vec![Out(echo()), Out(at("m1", 10)), briefed(&mind, 31, vec![aqua("a1")]), Out(echo()), briefed(&mind, 31, vec![aqua("a2")]), Out(echo()), Out(saying("both"))]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert_eq!(tells(&ran).len(), 2, "{:?}", ran.written()); - assert_eq!(answered(&mind, 1), vec![id_of("a1"), id_of("a2")]); -} - -/// A task notification is a turn that took in no brief: while the run has told one conversation only, it continues the turn before. -#[test] -fn a_result_that_took_in_no_brief_answers_the_previous_replys_briefs() { - let mind = live_mind(); - let steps = vec![Out(echo()), Out(saying("nothing yet")), briefed(&mind, 31, vec![aqua("a1")]), Out(echo()), Out(saying("first")), Out(saying("after a notification"))]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert!(answered(&mind, 1).is_empty(), "a run's first reply with nothing echoed answers none"); - assert_eq!(answered(&mind, 2), vec![id_of("a1")]); - assert_eq!(answered(&mind, 3), vec![id_of("a1")], "continues the turn before"); -} - -/// The actuator cannot tell whose work a notification is, so once the run has told two conversations a result that took in no brief -/// answers nothing (her Persona tells such a reply in no session). -#[test] -fn a_result_that_took_in_no_brief_answers_nothing_once_two_conversations_were_told() { - let mind = live_mind(); - let steps = vec![ - Out(echo()), - Out(saying("nothing yet")), - briefed(&mind, 31, vec![aqua("a1")]), - Out(echo()), - Out(saying("aqua turn: started a background task")), - briefed(&mind, 31, vec![libby("l1")]), - Out(echo()), - Out(saying("libby turn")), - Out(saying("the background task for aqua finished")), - ]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 2), vec![id_of("a1")]); - assert_eq!(answered(&mind, 3), vec![id_of("l1")]); - assert!(answered(&mind, 4).is_empty(), "the notification is told in neither conversation"); -} - -/// The flush result is the session's note to its successor: it answers no brief, though one conversation was told and a notification would continue it. -#[test] -fn the_result_of_the_flush_answers_no_brief() { - let mind = live_mind(); - persona_commit(&mind, vec![aqua("a1")]); - let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(saying("aqua work")), Out(echo()), Out(saying("note to my successor"))]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 1), vec![id_of("a1")]); - assert!(reply_of(&mind, 2).is_none(), "the flush is the note to the successor, kept in the journal: it is no reply, so nothing points at it"); -} - /// A restart whose claimed brief was resolved while another conversation's brief is open: the claim's conversation, read by key, is the start line's. #[test] fn a_restart_with_a_resolved_claim_opens_the_gate_for_its_conversation_only() { @@ -194,125 +175,6 @@ fn the_start_set_follows_the_claim_not_the_next_open_brief() { assert!(start.contains(&id_of("a1")) && !start.contains(&id_of("l1")), "{start}"); } -/// A result arrives while a line of its conversation is written and not yet echoed, after another conversation's brief was told first: -/// the gate holds on the written line's conversation, so the other conversation's brief waits until that line is answered. -#[test] -fn the_gate_holds_across_a_result_while_a_line_of_its_conversation_is_unechoed() { - use crate::self_run::session::SessionInbox; - let mind = live_mind(); - persona_commit(&mind, vec![libby("l0")]); - let mut told = inbox::derive(&*mind, &HashSet::new()).unwrap(); - let conversation = told.first().map(|brief| brief.conversation.clone()); - let (run, claims) = her_run(&*mind).unwrap().unwrap(); - let mut heard = Heard::default(); - let mut live = SelfInbox::new(&*mind, (&run, &claims), "", (&[], 0), (&mut told, conversation), &mut heard); - live.echoed(); - live.result(&saying("l0 done"), true, false).unwrap(); - persona_commit(&mind, vec![aqua("a1")]); - assert!(live.poll().unwrap().pending); - live.tell().unwrap(); - live.echoed(); - persona_commit(&mind, vec![aqua("a2")]); - assert!(live.poll().unwrap().pending); - live.tell().unwrap(); - live.result(&saying("a1 done"), true, false).unwrap(); - persona_commit(&mind, vec![libby("l2")]); - assert!(!live.poll().unwrap().pending, "a2 is written and unanswered: another conversation's brief waits"); - live.echoed(); - live.result(&saying("a2 done"), true, false).unwrap(); - assert!(live.poll().unwrap().pending, "a2 is answered: the gate is free"); -} - -fn task_started(tool_use: &str) -> String { - json!({"type": "system", "subtype": "task_started", "task_id": "bg1", "tool_use_id": tool_use, "description": "synthetic task", "task_type": "local_bash"}).to_string() -} - -fn task_notification(tool_use: &str) -> String { - json!({"type": "system", "subtype": "task_notification", "task_id": "bg1", "tool_use_id": tool_use, "status": "completed", "summary": "synthetic task"}).to_string() -} - -fn written_with(ran: &Ran, id: &str) -> bool { - ran.written().iter().any(|line| line.contains(id)) -} - -/// A task notification's result is attributed by the `tool_use_id` the stream carries to the turn that started the task, not to the -/// conversation answered last (stream shape: the `task_started` and `task_notification` system events of the 2.1.312 captures). -#[test] -fn a_notifications_result_answers_the_briefs_of_the_turn_that_started_its_task() { - let mind = live_mind(); - let steps = vec![ - Out(echo()), - Out(saying("nothing yet")), - briefed(&mind, 31, vec![aqua("a1")]), - Out(echo()), - Out(task_started("toolu_aqua")), - Out(saying("aqua turn: started a task")), - briefed(&mind, 31, vec![libby("l1")]), - Out(echo()), - Out(task_started("toolu_libby")), - Out(saying("libby turn: started a task")), - Out(task_notification("toolu_aqua")), - Out(saying("the aqua task finished")), - Out(task_notification("toolu_libby")), - Out(saying("the libby task finished")), - ]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert_eq!((answered(&mind, 2), answered(&mind, 3)), (vec![id_of("a1")], vec![id_of("l1")])); - assert_eq!(answered(&mind, 4), vec![id_of("a1")], "the aqua task's result is aqua's, though libby was answered last"); - assert_eq!(answered(&mind, 5), vec![id_of("l1")]); -} - -/// A notification the stream gives no start for, in a run that told two conversations, cannot be placed: it answers nothing. -#[test] -fn a_notification_with_no_known_task_answers_nothing_once_two_conversations_were_told() { - let mind = live_mind(); - let steps = vec![ - Out(echo()), - Out(saying("nothing yet")), - briefed(&mind, 31, vec![aqua("a1")]), - Out(echo()), - Out(saying("aqua turn")), - briefed(&mind, 31, vec![libby("l1")]), - Out(echo()), - Out(saying("libby turn")), - Out(task_notification("toolu_unknown")), - Out(saying("some task finished")), - ]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert!(answered(&mind, 4).is_empty()); -} - -/// While a notification's turn is open no other conversation's brief is written into it: it waits for the turn's result. -#[test] -fn another_conversations_brief_waits_for_an_open_notification_turn() { - let before = |mind: &Rc| { - vec![ - Out(echo()), - Out(saying("nothing yet")), - briefed(mind, 31, vec![aqua("a1")]), - Out(echo()), - Out(task_started("toolu_aqua")), - Out(saying("aqua turn: started a task")), - Out(task_notification("toolu_aqua")), - briefed(mind, 31, vec![libby("l1")]), - Idle(31), - ] - }; - let mind = live_mind(); - let ran = mind_run(&mind, before(&mind)); - assert!(ran.outcome.is_ok()); - assert!(!written_with(&ran, &id_of("l1")), "the open notification turn is aqua's: {:?}", ran.written()); - - let mind = live_mind(); - let steps = [before(&mind), vec![Out(saying("the aqua task finished")), Idle(31), Out(echo()), Out(saying("libby done"))]].concat(); - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 3), vec![id_of("a1")], "the outcome is aqua's"); - assert_eq!(answered(&mind, 4), vec![id_of("l1")], "libby's brief is told after, on its own line"); -} - /// A restart whose claim is resolved and which has no open brief: a brief of another conversation arriving is not written while the /// claim's start line is unanswered. #[test] @@ -339,133 +201,6 @@ fn a_restart_with_a_resolved_claim_and_no_open_brief_holds_the_gate_for_the_clai assert_eq!((answered(&mind, 1), answered(&mind, 2)), (vec![id_of("a1")], vec![id_of("l1")])); } -/// A turn that took in no brief is running (a request with nothing echoed) and the stream gave no notification event: it continues the turn -/// before, so another conversation's brief is not written into it. -#[test] -fn another_conversations_brief_waits_for_any_turn_that_took_in_no_brief() { - let before = |mind: &Rc| { - vec![ - Out(echo()), - Out(saying("nothing yet")), - briefed(mind, 31, vec![aqua("a1")]), - Out(echo()), - Out(saying("aqua turn: started a background task")), - Out(at("n1", 10)), - briefed(mind, 31, vec![libby("l1")]), - Idle(31), - ] - }; - let mind = live_mind(); - let ran = mind_run(&mind, before(&mind)); - assert!(ran.outcome.is_ok()); - assert!(!written_with(&ran, &id_of("l1")), "{:?}", ran.written()); - - let mind = live_mind(); - let ran = mind_run(&mind, [before(&mind), vec![Out(saying("SYNTHETIC aqua task outcome")), Idle(31), Out(echo()), Out(saying("libby done"))]].concat()); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 3), vec![id_of("a1")], "the outcome continues aqua's turn, not libby's"); - assert_eq!(answered(&mind, 4), vec![id_of("l1")]); -} - -/// The start message states the routing rules her Self works under, in place of the old continuation sentence. -#[test] -fn the_start_message_states_the_routing_rules() { - let text = start_message(&Slug("yggdrasil".into()), &run_ref("selfrun-a"), &[], 0); - for rule in ["one conversation at a time", "the conversation of the turn that started it", "no other conversation's brief is told into that turn", "pointer to it by id, without its text"] { - assert!(text.contains(rule), "{rule}: {text}"); - } - assert!(!text.contains("continues the one before")); -} - -/// The brief steps up to a notification of an aqua task: aqua's turn started the task and ended, then the notification arrived. -fn after_aqua_started_a_task(mind: &Rc) -> Vec { - vec![Out(echo()), Out(saying("nothing yet")), briefed(mind, 31, vec![aqua("a1")]), Out(echo()), Out(task_started("toolu_aqua")), Out(saying("aqua turn: started a task"))] -} - -/// A brief of the notification's own conversation joins its turn, and takes the notification in with it: the result answers that brief, the -/// notification is no longer pending, and another conversation's brief is not starved. -#[test] -fn the_notified_conversations_brief_joins_its_turn_and_takes_the_notification_in() { - let steps = |mind: &Rc| [after_aqua_started_a_task(mind), vec![Out(task_notification("toolu_aqua")), briefed(mind, 31, vec![aqua("a2")]), Idle(31)]].concat(); - let mind = live_mind(); - let ran = mind_run(&mind, steps(&mind)); - assert!(written_with(&ran, &id_of("a2")), "the notification's own conversation may be told into its turn: {:?}", ran.written()); - - let mind = live_mind(); - let ran = mind_run(&mind, [steps(&mind), vec![Out(echo()), Out(saying("outcome and a2")), briefed(&mind, 31, vec![libby("l1")]), Idle(31), Out(echo()), Out(saying("libby done"))]].concat()); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 3), vec![id_of("a2")]); - assert_eq!(answered(&mind, 4), vec![id_of("l1")], "the notification was taken in, so libby's brief is told"); -} - -/// A notification of aqua's task arrives while libby's turn runs: libby's result does not answer it, so the turn that follows is still aqua's. -#[test] -fn a_notification_of_another_conversation_stays_pending_through_a_turn_that_took_in_briefs() { - let mind = live_mind(); - let steps = [ - after_aqua_started_a_task(&mind), - vec![briefed(&mind, 31, vec![libby("l1")]), Out(echo()), Out(task_notification("toolu_aqua")), Out(saying("libby turn")), Out(at("n1", 10)), Out(saying("the aqua task finished"))], - ] - .concat(); - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 3), vec![id_of("l1")]); - assert_eq!(answered(&mind, 4), vec![id_of("a1")], "the notification turn is aqua's, though libby was answered last"); -} - -/// Notifications of two conversations' tasks pending together cannot be placed: no brief is written into their turn and the result answers both, which her Persona places in no conversation. -#[test] -fn notifications_of_two_conversations_open_a_turn_no_brief_may_join() { - let steps = |mind: &Rc| { - vec![ - Out(echo()), - Out(saying("nothing yet")), - briefed(mind, 31, vec![aqua("a1")]), - Out(echo()), - Out(task_started("toolu_aqua")), - Out(saying("aqua turn")), - briefed(mind, 31, vec![libby("l1")]), - Out(echo()), - Out(task_started("toolu_libby")), - Out(saying("libby turn")), - Out(task_notification("toolu_aqua")), - Out(task_notification("toolu_libby")), - briefed(mind, 31, vec![aqua("a2")]), - Idle(31), - ] - }; - let mind = live_mind(); - let ran = mind_run(&mind, steps(&mind)); - assert!(!written_with(&ran, &id_of("a2")), "{:?}", ran.written()); - - let mind = live_mind(); - let ran = mind_run(&mind, [steps(&mind), vec![Out(saying("both tasks finished")), Idle(31), Out(echo()), Out(saying("a2 done"))]].concat()); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 4), vec![id_of("a1"), id_of("l1")], "two conversations' work in one turn answers both: her Persona places it in neither and points at it"); - assert_eq!(answered(&mind, 5), vec![id_of("a2")]); -} - -/// A line that is no request, between turns, opens no turn: the next conversation's brief is told at once. -#[test] -fn a_line_between_turns_that_is_no_request_opens_no_turn() { - let mind = live_mind(); - let steps = vec![ - Out(echo()), - Out(saying("nothing yet")), - briefed(&mind, 31, vec![aqua("a1")]), - Out(echo()), - Out(saying("aqua done")), - Out(json!({"type": "system", "subtype": "status", "status": "idle"}).to_string()), - briefed(&mind, 31, vec![libby("l1")]), - Idle(31), - Out(echo()), - Out(saying("libby done")), - ]; - let ran = mind_run(&mind, steps); - assert!(ran.outcome.is_ok()); - assert_eq!(answered(&mind, 3), vec![id_of("l1")]); -} - /// The resolved claim is libby's and an aqua brief that was admitted before it is open: the claim's own conversation, read by its key, is the start line's. #[test] fn a_resolved_claims_conversation_is_read_by_its_own_key() { @@ -494,3 +229,263 @@ fn a_claimed_cut_spec_stays_on_the_start_line_beside_open_briefs() { let start = &ran.written()[0]; assert!(start.contains(&format!("claimed cut spec `{}`", spec_ref("x").id.0)) && start.contains(&id_of("a1")), "{start}"); } + +// ---- the real order of a background task ----------------------------------- + +/// The notification arrives inside the starting turn; after its result the CLI begins a turn of its own for it (init, a request, no echo). +/// Libby's brief, derived inside Aqua's turn and in every window after, is written into none of it: only after the notification turn's result. +#[test] +fn a_notified_task_holds_other_conversations_until_the_turn_the_cli_begins_for_it_ends() { + let lines = stream(BG_IDLE); + let (request1, result1, init2, request3, result2) = (after(&lines, "request", 1), after(&lines, "result", 1), after(&lines, "init", 2), after(&lines, "request", 3), after(&lines, "result", 2)); + let splices = |mind: &Rc| vec![(request1, vec![briefed(mind, 31, vec![libby("l1")])]), (result1, vec![Idle(31)]), (init2, vec![Idle(31)]), (request3, vec![Idle(31)])]; + // Held: every window up to the notification turn's last line, with the turn still open. + 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 Aqua's task turns: {:?}", ran.written()); + // The notification turn ends: the session is idle and Libby's brief is told on its own line; the task's result was Aqua's. + 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")])); +} + +/// Libby is served and idle; Aqua's brief switches the session; Aqua's turn starts a task. Its notification turn belongs to Aqua, not to the +/// conversation answered before the switch. +#[test] +fn a_background_tasks_result_reaches_the_conversation_that_started_it_after_another_conversation_was_served() { + let mind = live_mind(); + persona_commit(&mind, vec![libby("l1")]); + let steps = [vec![Out(echo()), Out(saying("libby done")), briefed(&mind, 31, vec![aqua("a1")])], script(&stream(BG_IDLE), vec![])].concat(); + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written()); + assert_eq!(answered(&mind, 1), vec![id_of("l1")]); + assert_eq!(answered(&mind, 2), vec![id_of("a1")], "the turn that started the task"); + assert_eq!(answered(&mind, 3), vec![id_of("a1")], "the notification turn: never none, never Libby's"); +} + +/// The CLI replays the notification to the model in the running turn (a user line starting ) after the ceiling flush was +/// written: the replay is no echo of ours, so the flush is not taken by it and its result is not mistaken for the flush's. +#[test] +fn a_replayed_task_notification_is_not_our_echo() { + let lines = stream(BG_FOLD); + let ceiling = after(&lines, "request", 2); + let mind = serving_aqua(); + let ran = mind_run(&mind, script(&lines, vec![(ceiling, vec![Out(at("ceiling", CEILING_TOKENS))])])); + assert!(ran.outcome.is_ok()); + assert_eq!(ran.written().len(), 2, "the start line and the flush: {:?}", ran.written()); + assert_eq!(flushes(&ran), 1); + assert_eq!(ran.count(&Event::InputClosed), 0, "the flush has not been taken: the replay is the CLI's line, not its echo"); + assert_eq!(answered(&mind, 1), vec![id_of("a1")], "Aqua's answer is a reply, not the flush's note"); + assert!(reply_of(&mind, 2).is_none()); + // The flush's own echo takes it; its result is the note to the successor and is no reply. + let mind = serving_aqua(); + let ran = mind_run(&mind, [script(&lines, vec![(ceiling, vec![Out(at("ceiling", CEILING_TOKENS))])]), vec![Out(echo()), Out(saying("note to my successor"))]].concat()); + assert!(ran.outcome.is_ok()); + assert_eq!(ran.count(&Event::InputClosed), 1); + assert_eq!(answered(&mind, 1), vec![id_of("a1")]); + assert!(reply_of(&mind, 2).is_none(), "the note is in the journal only"); +} + +/// The notification turn runs a tool, so a line written into it is folded: the served conversation's brief joins it and its result answers +/// that brief; another conversation's brief, briefed at the same point, waits for idle. +#[test] +fn the_served_conversations_brief_joins_the_clis_notification_turn_and_another_waits() { + let lines = stream(BG_OTHER); + let request3 = after(&lines, "request", 3); + let mind = serving_aqua(); + let held = script(&lines[..lines.len() - 1], vec![(request3, vec![briefed(&mind, 31, vec![aqua("a2"), libby("l1")])])]); + let ran = mind_run(&mind, held); + assert!(ran.outcome.is_ok()); + assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written()); + assert!(tells(&ran)[0].contains(&id_of("a2")) && !written_with(&ran, &id_of("l1")), "{:?}", ran.written()); + assert_eq!(answered(&mind, 1), vec![id_of("a1")]); + let mind = serving_aqua(); + let steps = script(&lines, vec![(request3, vec![briefed(&mind, 31, vec![aqua("a2"), libby("l1")])]), (lines.len(), vec![Idle(31), Out(echo()), Out(saying("libby done"))])]); + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + let told = tells(&ran); + assert_eq!(told.len(), 2, "{:?}", ran.written()); + assert!(told[1].contains(&id_of("l1")) && !told[1].contains(&id_of("a2")), "{}", told[1]); + assert_eq!(answered(&mind, 2), vec![id_of("a2")], "the folded brief's result answers it"); + assert_eq!(answered(&mind, 3), vec![id_of("l1")]); +} + +/// A brief arrives while the CLI's notification turn (no tool) runs: written into it, it would be queued behind and answered by the wrong +/// result. It is written after the turn's result. +#[test] +fn another_conversations_brief_waits_behind_a_notification_turn_with_no_tool() { + let lines = stream(BG_RACE); + let (init2, request3, result2) = (after(&lines, "init", 2), after(&lines, "request", 3), after(&lines, "result", 2)); + let mind = serving_aqua(); + let ran = mind_run(&mind, script(&lines[..result2 - 1], vec![(init2, vec![briefed(&mind, 31, vec![libby("l1")])]), (request3, vec![Idle(31)])])); + assert!(ran.outcome.is_ok()); + assert!(!written_with(&ran, &id_of("l1")) && !written_with(&ran, "@libby:chat"), "{:?}", ran.written()); + let mind = serving_aqua(); + let ran = mind_run(&mind, script(&lines, vec![(init2, vec![briefed(&mind, 31, vec![libby("l1")])]), (request3, vec![Idle(31)]), (result2, vec![Idle(31)])])); + assert!(ran.outcome.is_ok()); + assert_eq!(tells(&ran).len(), 1, "{:?}", ran.written()); + assert_eq!((answered(&mind, 2), answered(&mind, 3)), (vec![id_of("a1")], vec![id_of("l1")]), "the notification turn is Aqua's; Libby's brief is answered by its own result"); +} + +/// 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"); + 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()); + 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")]); +} + +/// A task that never ends holds other conversations; the rotation flush is the bound and never switches. +#[test] +fn a_backgrounded_task_that_is_never_notified_holds_other_conversations_to_the_rotation_flush_only() { + let lines = stream(BG_LATE); + 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 rotation = i64::try_from(TEST_CAP_S).unwrap() - ROTATE_MARGIN_S; + let ran = mind_run(&mind, [steps, vec![Idle(60), Idle(rotation), Out(echo()), Out(saying("note to my successor"))]].concat()); + assert!(ran.outcome.is_ok()); + assert_eq!(flushes(&ran), 1, "{:?}", ran.written()); + assert!(!written_with(&ran, &id_of("l1")), "the flush never carries another conversation's brief: {:?}", ran.written()); + assert_eq!(ran.count(&Event::InputClosed), 1, "the flush's result closes stdin"); + assert_eq!(answered(&mind, 1), vec![id_of("a1")]); +} + +// ---- 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. +#[test] +fn an_error_results_briefs_stay_in_their_own_span() { + let common = |mind: &Rc| vec![Out(echo()), Out(saying("first")), briefed(mind, 31, vec![aqua("a2")]), Out(echo()), Out(failing())]; + let mind = serving_aqua(); + let ran = mind_run(&mind, [common(&mind), vec![Out(saying("second"))]].concat()); + assert!(ran.outcome.is_ok()); + assert_eq!((answered(&mind, 1), answered(&mind, 2)), (vec![id_of("a1")], vec![id_of("a2")]), "the failed turn's brief is answered by the next result of its span"); + let mind = serving_aqua(); + let ran = mind_run(&mind, [common(&mind), vec![briefed(&mind, 31, vec![libby("l1")]), Out(echo()), Out(saying("libby done"))]].concat()); + assert!(ran.outcome.is_ok()); + assert_eq!(tells(&ran).len(), 2, "{:?}", ran.written()); + assert_eq!(answered(&mind, 2), vec![id_of("l1")], "no turn of Libby's span answers Aqua's failed brief"); +} + +/// Libby, Mira and Aqua briefs waiting behind an Aqua turn: Aqua's own brief drops in at once, the others go oldest first at successive idles. +#[test] +fn at_idle_the_oldest_waiting_conversation_is_served_first() { + let mind = serving_aqua(); + let steps = vec![ + Out(echo()), + Out(at("m1", 10)), + briefed(&mind, 31, vec![libby("l1")]), + briefed(&mind, 31, vec![mira("m1")]), + briefed(&mind, 31, vec![aqua("a2")]), + Out(echo()), + Out(saying("aqua done")), + Idle(31), + Out(echo()), + Out(saying("libby done")), + Idle(31), + Out(echo()), + Out(saying("mira done")), + ]; + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + let told = tells(&ran); + assert_eq!(told.len(), 3, "{:?}", ran.written()); + for (line, (own, others)) in told.iter().zip([("a2", ["l1", "m1"]), ("l1", ["a1", "m1"]), ("m1", ["a1", "l1"])]) { + assert!(line.contains(&id_of(own)), "{line}"); + assert!(others.iter().all(|other| !line.contains(&id_of(other))), "one conversation per line: {line}"); + } + 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")])); +} + +/// 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() { + let mind = live_mind(); + persona_commit(&mind, vec![libby("l1")]); + let ran = mind_run(&mind, vec![Out(echo()), Out(saying("libby done")), briefed(&mind, 31, vec![aqua("a1")]), Out(echo()), Out(saying("aqua done")), Out(saying("a task of aqua's finished")), Out(saying("another task of aqua's"))]); + assert!(ran.outcome.is_ok()); + assert_eq!(answered(&mind, 1), vec![id_of("l1")]); + for n in 2..=4 { + assert_eq!(answered(&mind, n), vec![id_of("a1")], "reply {n}"); + } + // A start that serves no one: its results answer none, and her Persona tells them in the public session. + let mind = live_mind(); + let ran = mind_run(&mind, vec![Out(echo()), Out(saying("a wake of its own")), Out(saying("a task finished"))]); + assert!(ran.outcome.is_ok()); + assert!(answered(&mind, 1).is_empty() && answered(&mind, 2).is_empty()); +} + +/// The same conversation's new brief is written at once, whatever runs; another conversation's waits for the turn to end. +#[test] +fn the_served_conversations_briefs_drop_in_mid_turn_and_another_conversations_wait() { + let mind = serving_aqua(); + let steps = vec![Out(echo()), Out(at("m1", 10)), briefed(&mind, 31, vec![aqua("a2")]), Out(echo()), briefed(&mind, 31, vec![libby("l1")]), Idle(31), Out(saying("both")), Idle(31), Out(echo()), Out(saying("libby done"))]; + 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("a2")) && told[1].contains(&id_of("l1")), "{told:?}"); + assert_eq!((answered(&mind, 1), answered(&mind, 2)), (vec![id_of("a1"), id_of("a2")], vec![id_of("l1")])); +} + +/// The flush result is the session's note to its successor: it answers no brief, though one conversation was told and a notification would continue it. +#[test] +fn the_result_of_the_flush_answers_no_brief() { + let mind = live_mind(); + persona_commit(&mind, vec![aqua("a1")]); + let steps = vec![Out(echo()), Out(at("m1", CEILING_TOKENS)), Out(saying("aqua work")), Out(echo()), Out(saying("note to my successor"))]; + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + assert_eq!(answered(&mind, 1), vec![id_of("a1")]); + assert!(reply_of(&mind, 2).is_none(), "the flush is the note to the successor, kept in the journal: it is no reply, so nothing points at it"); +} + +/// The start message states the serialization her Self works under, in place of the routing rules of the earlier rulings. +#[test] +fn the_start_message_states_one_conversation_at_a_time_and_what_holds_it() { + let text = start_message(&Slug("yggdrasil".into()), &run_ref("selfrun-a"), &[], 0); + for rule in ["one conversation at a time", "only once you are idle", "in the conversation you are serving", "a long background task keeps other members waiting"] { + assert!(text.contains(rule), "{rule}: {text}"); + } + for gone in ["pointer", "continues the one before", "the conversation of the turn that started it"] { + assert!(!text.contains(gone), "{gone}: {text}"); + } +} + +/// A switch begins a span of its own: a result that took in no brief, before any reply of the span, answers the briefs of the line that began +/// it and never the span before. +#[test] +fn a_switch_seeds_its_span_with_its_own_briefs_so_no_result_inherits_the_span_before() { + use crate::self_run::session::SessionInbox; + let mind = live_mind(); + persona_commit(&mind, vec![libby("l1")]); + let mut told = inbox::derive(&*mind, &HashSet::new()).unwrap(); + let conversation = told.first().map(|brief| brief.conversation.clone()).unwrap(); + let (run, claims) = her_run(&*mind).unwrap().unwrap(); + let mut heard = Heard::default(); + let mut live = SelfInbox::new(&*mind, (&run, &claims), "", (&[], 0), (&mut told, conversation), &mut heard); + live.echoed(); + live.result(&saying("libby done"), true, false).unwrap(); + persona_commit(&mind, vec![aqua("a1")]); + let polled = live.poll().unwrap(); + assert!(!polled.pending && polled.other, "Aqua's brief is another conversation's"); + live.tell(true).unwrap(); + live.result(&saying("a turn that took in no brief of Aqua's line"), true, false).unwrap(); + assert_eq!(answered(&mind, 1), vec![id_of("l1")]); + assert_eq!(answered(&mind, 2), vec![id_of("a1")], "the span's own briefs, not Libby's"); +} diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 254996f..644cc85 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 , Result { ok : bool , blocked : bool } , Other , Unreadable , } +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: const AGENT : & str mind_body::self_run: const CEILING_TOKENS : u64 mind_body::self_run: const ESCALATE_AFTER_S : i64