self: one Self turn answers one conversation

The start set is one conversation's; poll and tell gate briefs by the conversation of those unanswered;
a result that took in no brief continues the turn before; the start message says so.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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 <its id> 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 <its id> 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<SelfRunOutcome> {
     let Some((run, claims)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) };
     let mut task = task_section(mind, &claims)?;
-    let mut told: Vec<Brief> = 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<Brief> = 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<Brief>, heard: &mut Heard) -> Result<SelfRunOutcome> {
+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<Conversation>), told: &mut Vec<Brief>, heard: &mut Heard) -> Result<SelfRunOutcome> {
     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<String>,
     /// The briefs of the lines the CLI took, which the next reply answers.
     echoed: Vec<String>,
+    /// 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<Conversation>,
+    /// 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<String>,
     /// Replies made so far; the next is `<run label>-r<n+1>`.
     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<Brief>, 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<Brief>, Option<Conversation>), heard: &'a mut Heard) -> Self {
         let start = format!("{}{task}", start_message(mind.instance(), run, plan_quota, now_ms));
         let mut carried: Vec<String> = 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<String> = 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<Conversation> {
+        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<Polled> {
         let held: HashSet<String> = 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<String> {
-        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<Brief>, Vec<Brief>) = 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)
     }
 

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<Vec<Brief>> {
     let mut briefs: Vec<Brief> = 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,
         })

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}");

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

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<PipelineReply> {
+    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<String> {
+    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");
+}

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");