From 7182ff107bca4278fae50972be1432987a4627a9 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 7 Nov 2026 03:16:26 +0100 Subject: [PATCH] self: a notification's result is its task's turn's; a briefless turn holds other conversations; a resolved claim's conversation is read by key The stream's task_started and task_notification events carry the tool_use_id of the call that started the task. The actuator records, per task, the briefs the starting turn was answering; a briefless result after a notification answers those (any notification it cannot place answers none). While a notification's turn is open, only the conversation it is attributed to may be written. A restart reads a resolved claim's conversation by key and opens the gate for it alone. The flush result is the note to the successor, in the journal: it is no reply. The start message states the rules. Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/self_run.rs | 120 ++++++++++++---- crates/mind-body/src/self_run/inbox.rs | 10 ++ .../src/self_run/tests/conversation.rs | 129 +++++++++++++++++- 3 files changed, 226 insertions(+), 33 deletions(-) diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index 31e7361..d607ae6 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::HashSet; +use std::collections::{HashMap, 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, and what you say ending a turn reaches your Persona only in the conversation of the briefs that turn answers; a turn that took in no brief continues the one before.\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. 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{}", instance.0, run.id.0, render_plan(plan_quota, now_ms) @@ -519,14 +519,22 @@ pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawne let Some((run, claims)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) }; let open = inbox::derive(mind, &HashSet::new())?; // One Self turn answers one conversation: the start set is the claimed briefs and the open briefs of the first claimed brief's - // conversation (with no claimed brief, of the oldest open brief's). The rest are derived by poll, one conversation at a time. - let conversation = claims.iter().find_map(|claim| open.iter().find(|brief| brief.id == claim.id.0)).or(open.first()).map(|brief| brief.conversation.clone()); - // A claim of another conversation, or one no longer open beside open briefs of a conversation, is not rendered or carried: - // the start line holds one conversation only. - let claims: Vec = claims - .into_iter() - .filter(|claim| claim.kind != PipelineKind::FollowUp || open.iter().find(|brief| brief.id == claim.id.0).map_or(conversation.is_none(), |brief| Some(&brief.conversation) == conversation.as_ref())) - .collect(); + // conversation (with no claimed brief, of the oldest open brief's). A claim no longer open (resolved) has its conversation read by + // key from the mind, as a reply's briefs are: the gate opens only for it. The rest are derived by poll, one conversation at a time. + let mut claimed: Vec<(PipelineRef, Option)> = Vec::new(); + for claim in claims { + let of = if claim.kind != PipelineKind::FollowUp { + None + } else if let Some(brief) = open.iter().find(|brief| brief.id == claim.id.0) { + Some(brief.conversation.clone()) + } else { + inbox::conversation_by_key(mind, &claim.id.0).with_context(|| format!("the claim `{}` could not be read from the mind", claim.id.0))? + }; + claimed.push((claim, of)); + } + let conversation = 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 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 also: Vec<&Brief> = told.iter().filter(|brief| !claims.iter().any(|claim| claim.id.0 == brief.id)).collect(); @@ -588,10 +596,15 @@ struct SelfInbox<'a> { /// 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 (a task notification) continues that turn, so it answers them, - /// only while every brief this run told is of one conversation: the actuator cannot tell whose work a notification is, so once two - /// conversations were told it answers nothing, is told in no session and stays in force. + /// 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. 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, /// 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. @@ -612,7 +625,7 @@ 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(), flush_written: false, flushing: false, replies: 0, unadmitted: Vec::new() } + 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(), 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. @@ -633,18 +646,13 @@ impl<'a> SelfInbox<'a> { 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); - if ok && !text.trim().is_empty() { + let flush = std::mem::take(&mut self.flushing); + // The flush result is the session's note to its successor, in the journal: it is no reply, so nothing ever points at it. + 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 flush = std::mem::take(&mut self.flushing); - let taken = if flush { Vec::new() } else { std::mem::take(&mut self.echoed) }; - let answered: Vec = if !taken.is_empty() { - taken.into_iter().take(64).collect() - } else if !flush && self.one_conversation_told() { - self.last_answers.clone() - } else { - Vec::new() - }; + 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 answers = answered.iter().map(|id| PipelineRef { kind: PipelineKind::FollowUp, id: id.as_str().into() }).collect(); if !answered.is_empty() { self.last_answers = answered; @@ -657,6 +665,10 @@ impl<'a> SelfInbox<'a> { text: cut_text(text, REPLY_TEXT_LIMIT).as_str().into(), }); } + // A turn that took in no brief has ended: it was the turn of the notifications pending, which are now answered. + if !flush && self.echoed.is_empty() { + self.notified.clear(); + } self.admit_unadmitted(); self.conversation = self.carried.first().or(self.echoed.first()).and_then(|id| self.conversation_of(id)); } @@ -666,14 +678,67 @@ impl<'a> SelfInbox<'a> { 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 pending notifications are attributed to; none when they name two, or one could not be attributed. + fn notified_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 { + // A notification with no id cannot be placed: the sentinel finds no task, so the turn it opens answers none. + if event.get("subtype").and_then(Value::as_str) == Some("task_notification") { + self.notified.push(String::new()); + } + 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") if !self.notified.iter().any(|pending| pending == tool_use) => 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 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.conversation.as_ref().is_none_or(|conversation| *conversation == brief.conversation) + (self.notified.is_empty() || self.notified_conversation().as_ref() == Some(&brief.conversation)) && self.conversation.as_ref().is_none_or(|conversation| *conversation == brief.conversation) } } @@ -693,8 +758,8 @@ impl SessionInbox for SelfInbox<'_> { } fn tell(&mut self) -> Result { - // Free, the oldest waiting brief's conversation is the line's; the other conversations' briefs stay untold for a later poll. - let conversation = self.conversation.clone().or_else(|| self.waiting.first().map(|brief| brief.conversation.clone())); + // 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()); self.waiting = rest; let text = inbox::render(&mine, &self.plan); @@ -711,6 +776,7 @@ impl SessionInbox for SelfInbox<'_> { fn line(&mut self, line: &str) { self.heard.hear(line); + self.note_task(line); } fn echoed(&mut self) { diff --git a/crates/mind-body/src/self_run/inbox.rs b/crates/mind-body/src/self_run/inbox.rs index ea6ffb7..dd367a9 100644 --- a/crates/mind-body/src/self_run/inbox.rs +++ b/crates/mind-body/src/self_run/inbox.rs @@ -14,6 +14,7 @@ use std::collections::{HashMap, HashSet}; use anyhow::Result; +use cultnet_rs::Selection; use eureka_pipeline::PipelineKind; use huginn_mind::{PipelineFacts, PipelineStatusSummary}; @@ -68,6 +69,15 @@ fn open_briefs(mind: &dyn MindPort) -> Result> { Ok(briefs) } +/// The conversation a follow-up was given in, read by its key whether or not it is still in force; none when the mind does not hold it. +pub(crate) fn conversation_by_key(mind: &dyn MindPort, id: &str) -> Result> { + let one = Selection { keys: Some(vec![id.to_string()]), ..of_kind(PipelineKind::FollowUp) }; + Ok(headers(mind, &one)?.into_iter().find_map(|header| match header.facts { + PipelineFacts::FollowUp { origin, .. } => Some(persona::Conversation::of_origin(origin.as_ref())), + _ => None, + })) +} + /// Her briefs in force that the session has not been told and no open question /// holds: a held brief is neither claimed nor told until its question is /// answered or withdrawn. Reads the hold once. diff --git a/crates/mind-body/src/self_run/tests/conversation.rs b/crates/mind-body/src/self_run/tests/conversation.rs index 5d4cca8..06e673b 100644 --- a/crates/mind-body/src/self_run/tests/conversation.rs +++ b/crates/mind-body/src/self_run/tests/conversation.rs @@ -160,23 +160,24 @@ fn the_result_of_the_flush_answers_no_brief() { let ran = mind_run(&mind, steps); assert!(ran.outcome.is_ok()); assert_eq!(answered(&mind, 1), vec![id_of("a1")]); - assert!(answered(&mind, 2).is_empty(), "the flush is never told as a reply"); + 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 start line holds the open brief only. +/// 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_puts_one_conversation_on_the_start_line() { +fn a_restart_with_a_resolved_claim_opens_the_gate_for_its_conversation_only() { let mind = TestMind::seeded(); mind.committed(vec![spec("x"), run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]); persona_commit(&mind, vec![aqua("a1")]); persona_commit(&mind, vec![libby("l1")]); mind.committed(vec![run("selfrun-a", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("a1")])]); mind.committed(vec![close(brief_ref("a1"), recorded())]); - let ran = carry_with(&Seen::of(&mind), vec![Out(echo()), Out(saying("libby done"))], Some(0), false); + let ran = carry_with(&Seen::of(&mind), vec![Out(echo()), Out(saying("aqua done")), Idle(31), Out(echo()), Out(saying("libby done"))], Some(0), false); assert!(ran.outcome.is_ok()); let start = &ran.written()[0]; - assert!(start.contains(&id_of("l1")) && !start.contains(&id_of("a1")), "{start}"); - assert_eq!(answered(&mind, 1), vec![id_of("l1")], "told in the right conversation, none stranded"); + assert!(start.contains(&id_of("a1")) && !start.contains(&id_of("l1")), "the resolved claim's conversation, read by key, is the start line's: {start}"); + assert_eq!(answered(&mind, 1), vec![id_of("a1")]); + assert_eq!(answered(&mind, 2), vec![id_of("l1")], "told in the right conversation, none stranded"); } /// The claim is the first open brief and another conversation's brief is next: the start set follows the claim. @@ -221,3 +222,119 @@ fn the_gate_holds_across_a_result_while_a_line_of_its_conversation_is_unechoed() 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] +fn a_restart_with_a_resolved_claim_and_no_open_brief_holds_the_gate_for_the_claims_conversation() { + let setup = || { + let mind = Rc::new(TestMind::seeded()); + mind.committed(vec![spec("x"), run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]); + persona_commit(&mind, vec![aqua("a1")]); + mind.committed(vec![run("selfrun-a", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("a1")])]); + mind.committed(vec![close(brief_ref("a1"), recorded())]); + mind + }; + let steps = |mind: &Rc| vec![Out(echo()), briefed(mind, 31, vec![libby("l1")]), Idle(31)]; + let mind = setup(); + let ran = mind_run(&mind, steps(&mind)); + assert!(ran.outcome.is_ok()); + assert!(ran.written()[0].contains(&id_of("a1")), "{:?}", ran.written()); + assert!(!written_with(&ran, &id_of("l1")), "{:?}", ran.written()); + + let mind = setup(); + let steps = [steps(&mind), vec![Out(saying("aqua done")), Idle(31), Out(echo()), Out(saying("libby done"))]].concat(); + let ran = mind_run(&mind, steps); + assert!(ran.outcome.is_ok()); + assert_eq!((answered(&mind, 1), answered(&mind, 2)), (vec![id_of("a1")], vec![id_of("l1")])); +}