diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index b70407d..5aadc30 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -74,6 +74,7 @@ use uuid::Uuid; use crate::launch::{Clock, closure, holders}; use crate::connector::render_plan; +use crate::persona::Conversation; use crate::lease_client::{LeaseRefusalKind, LeaseRequest, Lender, Lent, TokenFd}; use model_connector::PlanQuota; use crate::queue::MindPort; @@ -151,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.\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, 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{}", instance.0, run.id.0, render_plan(plan_quota, now_ms) @@ -517,7 +518,11 @@ fn withdraw_misconfigured(mind: &dyn MindPort, clock: &dyn Clock, run: &Pipeline pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write) -> Result { let Some((run, claims)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) }; let mut task = task_section(mind, &claims)?; - let mut told: Vec = inbox::derive(mind, &HashSet::new())?; + 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()); + 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(); if !also.is_empty() { task.push_str(" @@ -529,7 +534,7 @@ Also open for you:"); } } let mut heard = Heard::default(); - let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &claims, &task), &mut told, &mut heard); + let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &claims, &task, conversation), &mut told, &mut heard); let closed = close_if_open(mind, clock, &run, &ran, &heard); if closed.is_err() { let _ = writeln!(out, "the run could not be closed: the next launch closes it"); @@ -542,7 +547,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): (&PipelineRef, &[PipelineRef], &str), 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, Option), 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)? { @@ -554,7 +559,7 @@ fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, s Lent::Refused(refused) => return close_refused(mind, clock, run, refused.kind), }; let child = spawner.spawn(&claude_args(session_id(run)), lease.token())?; - let mut inbox = SelfInbox::new(mind, (run, claims), task, (lease.plan_quota(), started_ms), told, heard); + let mut inbox = SelfInbox::new(mind, (run, claims), task, (lease.plan_quota(), started_ms), (told, conversation), heard); let ran = drive(clock, child, &mut inbox, run_cap_s, out); inbox.admit_unadmitted(); ran @@ -574,6 +579,11 @@ 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 (a task notification, a flush) continues that turn, so it answers them. + last_answers: Vec, /// Replies made so far; the next is `-r`. replies: u32, /// Replies the mind did not take: process memory, retried at the next result and once at the end. The journal keeps their text either way. @@ -582,7 +592,7 @@ 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: &'a mut Vec, heard: &'a mut Heard) -> Self { + 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 { 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() { @@ -590,7 +600,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(), 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(), 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. @@ -614,7 +624,10 @@ impl<'a> SelfInbox<'a> { if ok && !text.trim().is_empty() { self.replies += 1; let run_label = self.run.id.0.rsplit(':').next().unwrap_or(&self.run.id.0); - let answers = std::mem::take(&mut self.echoed).into_iter().take(64).map(|id| PipelineRef { kind: PipelineKind::FollowUp, id: id.as_str().into() }).collect(); + let taken = std::mem::take(&mut self.echoed); + 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(); + self.last_answers = answered; self.unadmitted.push(PipelineReply { instance: self.mind.instance().clone(), label: format!("{run_label}-r{}", self.replies).as_str().into(), @@ -624,6 +637,17 @@ impl<'a> SelfInbox<'a> { }); } self.admit_unadmitted(); + self.conversation = self.carried.first().or(self.echoed.first()).and_then(|id| self.conversation_of(id)); + } + + /// 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. + fn may_tell(&self, brief: &Brief) -> bool { + self.conversation.as_ref().is_none_or(|conversation| *conversation == brief.conversation) } } @@ -639,13 +663,18 @@ 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.is_empty(), yield_now: false }) + Ok(Polled { pending: self.waiting.iter().any(|brief| self.may_tell(brief)), yield_now: false }) } fn tell(&mut self) -> Result { - let text = inbox::render(&self.waiting, &self.plan); - self.carried.extend(self.waiting.iter().map(|brief| brief.id.clone())); - self.told.append(&mut self.waiting); + // 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())); + 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); + self.carried.extend(mine.iter().map(|brief| brief.id.clone())); + self.told.append(&mut mine); + self.conversation = conversation; Ok(text) } diff --git a/crates/mind-body/src/self_run/inbox.rs b/crates/mind-body/src/self_run/inbox.rs index 8390b58..979ef59 100644 --- a/crates/mind-body/src/self_run/inbox.rs +++ b/crates/mind-body/src/self_run/inbox.rs @@ -33,6 +33,9 @@ pub struct Brief { pub item: String, /// The admission ordinal: the order briefs were given in. pub ordinal: u64, + /// The conversation her Persona was in when it gave the brief, from the origin the follow-up records + /// (none recorded, a brief admitted before origins existed, is the public one). + pub conversation: persona::Conversation, } /// The ids of the follow-ups a closed run claims, with the run each was claimed by. @@ -54,9 +57,9 @@ fn open_briefs(mind: &dyn MindPort) -> Result> { let mut briefs: Vec = headers(mind, &in_force(of_kind(PipelineKind::FollowUp)))? .into_iter() .filter_map(|summary| match summary.facts { - PipelineFacts::FollowUp { repo, owner, item, .. } if owner.0 == OWNER && summary.admission.provenance.agent.0 == persona::AGENT => { + PipelineFacts::FollowUp { repo, owner, item, origin, .. } if owner.0 == OWNER && summary.admission.provenance.agent.0 == persona::AGENT => { let campaign = summary.id.id.0.split(':').next().unwrap_or("").to_string(); - Some(Brief { id: summary.id.id.0, campaign, repo: repo.0, item: item.0, ordinal: summary.admission.ordinal }) + Some(Brief { id: summary.id.id.0, campaign, repo: repo.0, item: item.0, ordinal: summary.admission.ordinal, conversation: persona::Conversation::of_origin(origin.as_ref()) }) } _ => None, }) diff --git a/crates/mind-body/src/self_run/inbox/tests.rs b/crates/mind-body/src/self_run/inbox/tests.rs index 81a19af..052f3f3 100644 --- a/crates/mind-body/src/self_run/inbox/tests.rs +++ b/crates/mind-body/src/self_run/inbox/tests.rs @@ -101,7 +101,7 @@ fn the_backlog_lists_in_force_persona_briefs_claimed_by_closed_runs_and_admits_n #[test] fn a_tell_names_each_brief_the_plan_and_how_she_resolves_it() { - let brief = Brief { id: "eureka-body:follow_up:b1".into(), campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: "Wire the thing.".into(), ordinal: 1 }; + let brief = Brief { id: "eureka-body:follow_up:b1".into(), campaign: "eureka-body".into(), repo: "GameCult/Huginn".into(), item: "Wire the thing.".into(), ordinal: 1, conversation: persona::Conversation::Public }; let text = render(&[brief], "Plan: 5h at 40%."); assert!(text.contains("eureka-body:follow_up:b1") && text.contains("Wire the thing."), "{text}"); assert!(text.contains("resolve follow-up eureka-body:follow_up:b1 Recorded") && text.contains("Withdrawn"), "she is told to resolve it: {text}"); diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index d49f344..2b204b7 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -1192,6 +1192,7 @@ result subtype=success is_error=false: The answer. mod lease; mod closure; +mod conversation; mod reply; // ---- the task in the start message --------------------------------------- diff --git a/crates/mind-body/src/self_run/tests/conversation.rs b/crates/mind-body/src/self_run/tests/conversation.rs new file mode 100644 index 0000000..e375a29 --- /dev/null +++ b/crates/mind-body/src/self_run/tests/conversation.rs @@ -0,0 +1,129 @@ +//! 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. + +use eureka_pipeline::{MemoryOrigin, PipelineReply}; + +use super::*; + +fn saying(text: &str) -> String { + json!({"type": "result", "subtype": "success", "is_error": false, "terminal_reason": "completed", "result": text}).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), + _ => None, + } +} + +fn answered(mind: &TestMind, n: u32) -> Vec { + reply_of(mind, n).unwrap_or_else(|| panic!("reply {n} was not admitted")).answers.iter().map(|brief| brief.id.0.clone()).collect() +} + +/// A brief her Persona gave from a DM with `member` in `room`. +fn brief_in(label: &str, room: &str, member: &str) -> PipelineDocument { + let PipelineDocument::FollowUp(mut document) = brief(label, &format!("Walnut work for {member}.")) else { unreachable!() }; + document.origin = Some(MemoryOrigin::Direct { room: room.into(), with: vec![member.into()] }); + PipelineDocument::FollowUp(document) +} + +fn aqua(label: &str) -> PipelineDocument { + brief_in(label, "!aqua:chat", "@aqua:chat") +} + +fn libby(label: &str) -> PipelineDocument { + brief_in(label, "!libby:chat", "@libby:chat") +} + +fn id_of(label: &str) -> String { + brief_ref(label).id.0 +} + +#[test] +fn the_start_line_carries_one_conversation_and_the_other_is_told_on_its_own_line() { + let mind = TestMind::seeded(); + mind.committed(vec![spec("x"), run("persona", RunTurn::PersonaTurn, RunOperator::Mind, &[])]); + persona_commit(&mind, vec![aqua("a1"), libby("l1"), aqua("a2")]); + mind.committed(vec![run("selfrun-a", RunTurn::SelfRun, RunOperator::Mind, &[brief_ref("a1")])]); + let steps = vec![Out(echo()), Out(saying("aqua done")), Idle(31), Out(echo()), Out(saying("libby done"))]; + let ran = carry_with(&Seen::of(&mind), steps, Some(0), false); + assert!(ran.outcome.is_ok()); + let start = &ran.written()[0]; + assert!(start.contains(&id_of("a1")) && start.contains(&id_of("a2")), "the claimed brief and the open one of its conversation: {start}"); + assert!(!start.contains(&id_of("l1")), "the other conversation's brief is not on the start line: {start}"); + 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].contains(&id_of("a2")), "{}", told[0]); + assert_eq!(answered(&mind, 1).len(), 2, "the start line's briefs"); + assert!(!answered(&mind, 1).contains(&id_of("l1"))); + assert_eq!(answered(&mind, 2), vec![id_of("l1")]); +} + +#[test] +fn with_no_claimed_brief_the_start_set_is_the_oldest_open_briefs_conversation() { + let mind = live_mind(); + for brief in [libby("l1"), aqua("a1"), libby("l2")] { + persona_commit(&mind, vec![brief]); + } + let ran = mind_run(&mind, vec![Out(echo()), Out(saying("libby done")), Idle(31), Out(echo()), Out(saying("aqua done"))]); + assert!(ran.outcome.is_ok()); + let start = &ran.written()[0]; + assert!(start.contains(&id_of("l1")) && start.contains(&id_of("l2")) && !start.contains(&id_of("a1")), "{start}"); + 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 or a flush is a turn that took in no brief: it continues the turn before, so its words are that conversation's. +#[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"); +} diff --git a/crates/mind-body/src/self_run/tests/reply.rs b/crates/mind-body/src/self_run/tests/reply.rs index 29322ca..0a9244a 100644 --- a/crates/mind-body/src/self_run/tests/reply.rs +++ b/crates/mind-body/src/self_run/tests/reply.rs @@ -44,7 +44,7 @@ fn each_result_with_text_is_one_reply_in_order_answering_the_briefs_echoed_since let (r1, r2, r3) = (reply_of(&mind, 1).unwrap(), reply_of(&mind, 2).unwrap(), reply_of(&mind, 3).unwrap()); assert_eq!((r1.text.0.as_str(), r2.text.0.as_str(), r3.text.0.as_str()), ("first", "second", "third")); assert_eq!(answered(&r1), vec![brief_ref("b1").id.0], "the brief whose line the CLI took before the first result"); - assert!(answered(&r2).is_empty(), "no line was taken between the first result and the second"); + assert_eq!(answered(&r2), vec![brief_ref("b1").id.0], "no line was taken between the first result and the second: it continues the turn before"); assert_eq!(answered(&r3), vec![brief_ref("b2").id.0]); assert!(r1.run == run_ref("selfrun-a") && r1.instance.0 == INSTANCE); assert!(resolutions_of(&mind, &brief_ref("b1")).is_empty() && resolutions_of(&mind, &brief_ref("b2")).is_empty(), "the actuator never resolves a brief");