diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index d5f7a81..ff3fe4a 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -805,6 +805,8 @@ pub fn carry(organs: &Organs, control: &impl ControlSource, swarm: &impl Control make_sandbox(organs.sandbox)?; let mut progress = Progress::default(); let started = organs.clock.now(); + // The replies a session opened alone for, this start: a reply that cannot be resolved never loops processes. + let mut opened_alone = BTreeSet::new(); for process in 0u32.. { if process > 0 && let Effective::Held { reason } = read(control, swarm, &instance) @@ -816,13 +818,15 @@ pub fn carry(organs: &Organs, control: &impl ControlSource, swarm: &impl Control progress.failure = Some(Failure::Chat); break; }; - let first = next_batch(config, pending, unix_ms(organs.clock.now())); + // A mind that cannot be asked for her Self's replies still serves her chat. + let routes = reply_routes(organs.mind, &BTreeSet::new()).unwrap_or_default(); + let Some(session) = next_session(config, pending, &routes, &mut opened_alone, process, unix_ms(organs.clock.now())) else { break }; let remaining_s = i64::try_from(organs.run_cap_s).unwrap_or(i64::MAX).saturating_sub((organs.clock.now() - started).num_seconds()); - if process > 0 && (first.is_empty() || remaining_s <= ROTATE_MARGIN_S) { + if process > 0 && remaining_s <= ROTATE_MARGIN_S { break; } let turns = progress.turn; - one_process(organs, control, swarm, config, &run, process, first, u64::try_from(remaining_s.max(1)).unwrap_or(1), &mut progress)?; + one_process(organs, control, swarm, config, &run, process, session, u64::try_from(remaining_s.max(1)).unwrap_or(1), &mut progress)?; // A process that gave no result answered nothing: another would only repeat it. if progress.failure.is_none() && progress.turn == turns { progress.failure = Some(Failure::Session); @@ -854,7 +858,7 @@ fn one_process( config: &Config, run: &PipelineRef, process: u32, - first: Vec, + session: Session, remaining_s: u64, progress: &mut Progress, ) -> Result<()> { @@ -870,7 +874,7 @@ fn one_process( } }; let campaigns = campaigns_in_force(organs.mind)?; - let rooms = reply_rooms(config, &first); + let rooms = reply_rooms(config, &session.conversation); let schema = reply_schema(&rooms, &campaigns.iter().map(|(slug, _)| slug.clone()).collect::>()); let prompt = organs.scratch.join(INSTRUCTIONS_FILE); if std::fs::write(&prompt, format!("{doctrine}{persona}\n\n{}", render_campaigns(&campaigns))).is_err() { @@ -885,12 +889,12 @@ fn one_process( return Ok(()); } }; - let session = Uuid::new_v5(&SESSION_NAMESPACE, format!("{}:p{process}", run.id.0).as_bytes()); - let Ok(child) = organs.spawner.spawn(&persona_args(session, &config.model, &schema, &prompt), lease.token()) else { + let session_id = Uuid::new_v5(&SESSION_NAMESPACE, format!("{}:p{process}", run.id.0).as_bytes()); + let Ok(child) = organs.spawner.spawn(&persona_args(session_id, &config.model, &schema, &prompt), lease.token()) else { progress.failure = Some(Failure::Session); return Ok(()); }; - let mut inbox = PersonaInbox::new(organs, control, swarm, config, run, progress, first, started_ms, render_plan(lease.plan_quota(), started_ms)); + let mut inbox = PersonaInbox::new(organs, control, swarm, config, run, progress, session, started_ms, render_plan(lease.plan_quota(), started_ms)); let driven = drive(organs.clock, child, &mut inbox, remaining_s, &mut std::io::stdout()); inbox.settle_replies(); driven?; @@ -937,7 +941,7 @@ fn persona_args(session: Uuid, model: &str, schema: &str, system_prompt_file: &P /// A conversation: the public rooms together, or one direct message room alone. #[derive(Clone, Debug, PartialEq, Eq)] -enum Conversation { +pub(crate) enum Conversation { Public, Direct(String), } @@ -946,6 +950,100 @@ impl Conversation { fn of(delivery: &ChatDelivery) -> Self { if delivery.is_direct { Self::Direct(delivery.room_id.clone()) } else { Self::Public } } + + /// The conversation an origin records: a direct message's room, else the public one. No origin (a brief + /// admitted before origins were recorded) is public. + pub(crate) fn of_origin(origin: Option<&MemoryOrigin>) -> Self { + match origin { + Some(MemoryOrigin::Direct { room, .. }) => Self::Direct(room.0.clone()), + _ => Self::Public, + } + } +} + +/// The conversation a reply is told in: the one conversation its briefs' origins name (no brief is the public one), +/// or none when they name two. A reply of two conversations 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 Some(first) = conversations.next() else { return Some(Conversation::Public) }; + conversations.all(|conversation| conversation == first).then_some(first) +} + +/// A reply of her Self in force and where it is told: the one derivation of a reply's conversation. Nothing stores it. +#[derive(Clone, Debug, PartialEq, Eq)] +struct Route { + id: String, + ordinal: u64, + at_unix_ms: u64, + 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, +} + +/// 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 whose briefs name two conversations is left out: no session tells it. +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 + .iter() + .flat_map(|header| match &header.facts { + PipelineFacts::Reply { answers, .. } => answers.iter().map(|brief| brief.id.0.clone()).collect(), + _ => Vec::new(), + }) + .collect(); + keys.sort(); + keys.dedup(); + let mut origins: BTreeMap> = BTreeMap::new(); + if !keys.is_empty() { + for header in headers(mind, &Selection { keys: Some(keys), ..of_kind(PipelineKind::FollowUp) })? { + if let PipelineFacts::FollowUp { origin, .. } = header.facts { + origins.insert(header.id.id.0, origin); + } + } + } + let mut routes = Vec::new(); + for header in replies { + let PipelineFacts::Reply { answers, .. } = &header.facts else { continue }; + let of_answers: Vec> = answers.iter().map(|brief| origins.get(&brief.id.0).cloned().flatten()).collect(); + 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), + conversation, + origin: of_answers.into_iter().flatten().next(), + }); + } + routes.sort_by_key(|route| route.ordinal); + Ok(routes) +} + +/// One Persona process: the conversation it serves, the origin a turn of it with no message records, and its first batch. +struct Session { + conversation: Conversation, + origin: MemoryOrigin, + first: Vec, +} + +/// The session the oldest waiting thing asks for. A delivery batch (`next_batch`) opens its conversation, and a reply of that +/// conversation rides it; a reply older than the batch, of another conversation, opens its own conversation alone, with no +/// message, at most once in a start (`opened_alone`). The first process with neither is the public wake; a later one ends the start. +fn next_session(config: &Config, pending: Vec, routes: &[Route], opened_alone: &mut BTreeSet, process: u32, now_ms: u64) -> Option { + let batch = next_batch(config, pending, now_ms); + let route = routes.iter().find(|route| !opened_alone.contains(&route.id)); + if let Some(first) = batch.first() + && route.is_none_or(|route| first.at_unix_ms <= route.at_unix_ms || Conversation::of(first) == route.conversation) + { + return Some(Session { conversation: Conversation::of(first), origin: origin_of(config, &batch, &MemoryOrigin::Wake).0, first: batch }); + } + match route { + Some(route) => { + opened_alone.insert(route.id.clone()); + Some(Session { conversation: route.conversation.clone(), origin: route.origin.clone().unwrap_or(MemoryOrigin::Wake), first: Vec::new() }) + } + None => (process == 0).then(|| Session { conversation: Conversation::Public, origin: MemoryOrigin::Wake, first: Vec::new() }), + } } /// What a batch's line showed her. Once the CLI has taken the line, it joins what a reply may name. @@ -1051,6 +1149,10 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> { progress: &'a mut Progress, rooms: Vec, conversation: Conversation, + /// Where a turn of this session with no message came from: the first batch's, else the reply's brief's, else a wake. + origin: MemoryOrigin, + /// An in-force reply of another conversation waits, as the last read saw it: this session yields to it at idle. + replies_elsewhere: bool, /// The plan windows the lease granted, as one line. plan: String, /// When this process began: a line of her own from then on she wrote in this session. @@ -1113,9 +1215,9 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> { impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { /// One process's session inbox over its first batch. The caller has read the gate and started the child. #[allow(clippy::too_many_arguments, reason = "one process's state is these few things, and a struct for them would be a second owner")] - fn new(organs: &'a Organs<'a>, control: &'a C, swarm: &'a S, config: &'a Config, run: &'a PipelineRef, progress: &'a mut Progress, first: Vec, started_ms: u64, plan: String) -> Self { - let rooms = reply_rooms(config, &first); - let conversation = first.first().map_or(Conversation::Public, Conversation::of); + fn new(organs: &'a Organs<'a>, control: &'a C, swarm: &'a S, config: &'a Config, run: &'a PipelineRef, progress: &'a mut Progress, session: Session, started_ms: u64, plan: String) -> Self { + let Session { conversation, origin, first } = session; + let rooms = reply_rooms(config, &conversation); Self { organs, control, @@ -1125,6 +1227,8 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { progress, rooms, conversation, + origin, + replies_elsewhere: false, plan, started_ms, told: BTreeSet::new(), @@ -1174,7 +1278,11 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { let senders = other_senders(self.config, deliveries); let cue = recall::Cue { senders, texts: deliveries.iter().rev().map(|delivery| delivery.text.as_str()).collect(), names: names.clone(), listed: &goals.listed, now: organs.clock.now() }; let recalled = recall::select(organs.mind, &cue)?; - let shown: Vec<&str> = goals.people.iter().map(String::as_str).chain(recalled.entries().flat_map(|entry| entry.people.iter().map(String::as_str))).collect(); + let session_with: &[Short] = match &self.origin { + MemoryOrigin::Direct { with, .. } => with, + _ => &[], + }; + let shown: Vec<&str> = goals.people.iter().map(String::as_str).chain(recalled.entries().flat_map(|entry| entry.people.iter().map(String::as_str))).chain(session_with.iter().map(|id| id.0.as_str())).collect(); let earlier = self.earlier(deliveries); let people = people_she_may_name(self.config, deliveries, &earlier, &shown); let started_self_today = started_self_today(organs.mind, &instance, organs.clock.now()).unwrap_or(true); @@ -1255,7 +1363,11 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> { return; } self.reply_read_due_ms = now + REPLY_READ_EVERY_S * 1000; - let Ok(unheard) = unheard_replies(self.organs.mind, &self.told_replies, &self.waiting_replies) else { return }; + let Ok(routes) = reply_routes(self.organs.mind, &self.told_replies) else { return }; + let (own, elsewhere): (Vec, Vec) = + routes.into_iter().filter(|route| !self.waiting_replies.iter().any(|reply| reply.id == route.id)).partition(|route| route.conversation == self.conversation); + self.replies_elsewhere = !elsewhere.is_empty(); + let Ok(unheard) = unheard_replies(self.organs.mind, &own) else { return }; self.waiting_replies.extend(unheard); self.waiting_replies.sort_by_key(|reply| reply.ordinal); } @@ -1302,6 +1414,12 @@ 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 } + } + fn held(&self) -> Option { match read(self.control, self.swarm, self.organs.mind.instance()) { Effective::Held { reason } => Some(reason), @@ -1342,7 +1460,7 @@ impl SessionInbox for PersonaInbox<'_, C, S> revision.ready |= revision.polls >= REVISION_POLLS || revision.awaiting.iter().all(|id| pending.iter().any(|delivery| delivery.event_id == *id)); } let revision_ready = self.revision_ready(); - let Some(oldest) = pending.first() else { return Ok(Polled { pending: self.retell.is_some() || revision_ready || !self.notices.is_empty() || !self.waiting_replies.is_empty(), yield_now: false }) }; + 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 }); } @@ -1354,9 +1472,10 @@ impl SessionInbox for PersonaInbox<'_, C, S> for delivery in &untold { self.first_seen.entry(delivery.seq).or_insert(now); } + let any_untold = !untold.is_empty(); let due = revision_ready || untold.iter().any(|delivery| now >= self.first_seen[&delivery.seq] + stagger_ms(self.config, delivery)); self.waiting = if due { one_line_of(untold, |batch| message_lines(self.config, batch, &self.earlier(batch)).len()) } else { Vec::new() }; - Ok(Polled { pending: !self.waiting.is_empty() || self.retell.is_some() || revision_ready || !self.notices.is_empty() || !self.waiting_replies.is_empty(), yield_now: false }) + 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 { @@ -1655,7 +1774,9 @@ fn act(inbox: &PersonaInbox<'_, C, S>, batch } outputs.push(TraceItem { provenance: "her own".into(), room: say.room.clone(), text }); } - let memories = build_memories(&instance, run, turn, &reply, &origin_of(config, batch)); + // The one origin of this turn: its memories and its brief record the same conversation. + let origin = origin_of(config, batch, &inbox.origin); + let memories = build_memories(&instance, run, turn, &reply, &origin); for remembered in &reply.remember { outputs.push(TraceItem { provenance: "her own".into(), room: String::new(), text: format!("remember {:?}: {}", remembered.sort, remembered.text) }); } @@ -1673,7 +1794,7 @@ fn act(inbox: &PersonaInbox<'_, C, S>, batch if let Some(ask) = &reply.start_self_run { outputs.push(TraceItem { provenance: "her own".into(), room: String::new(), text: format!("start_self_run {} {}: {}", ask.campaign, ask.repo, ask.brief) }); let taken = item_taken(ask, &memories)?; - match start_self_run(organs, control, swarm, run, &batch_id, ask, taken)? { + match start_self_run(organs, control, swarm, run, &batch_id, ask, taken, &origin.0)? { Ok(launched) => started = Some(launched), Err(why) => { organs.trace.record(&record(outputs, format!("failed: {why}")))?; @@ -1744,15 +1865,14 @@ fn one_line_of(mut waiting: Vec, rendered_bytes: impl Fn(&[ChatDel waiting } -/// The rooms a reply to these deliveries may name: hers, and the room each -/// direct message arrived under (a DM's alias is its room id, which the mouth -/// sends into). Never a room no delivery came from and she was not given. -fn reply_rooms(config: &Config, deliveries: &[ChatDelivery]) -> Vec { +/// The rooms a reply in this conversation may name: hers, and for a direct message its room id (a DM's alias is +/// its room id, which the mouth sends into). Never a room of another conversation. +fn reply_rooms(config: &Config, conversation: &Conversation) -> Vec { let mut rooms = config.rooms.clone(); - for delivery in deliveries.iter().filter(|delivery| delivery.is_direct) { - if !rooms.contains(&delivery.room_alias) { - rooms.push(delivery.room_alias.clone()); - } + if let Conversation::Direct(room) = conversation + && !rooms.contains(room) + { + rooms.push(room.clone()); } rooms } @@ -1827,17 +1947,12 @@ fn decode_reply(text: &str, rooms: &[String], people: &[String], open: &[String] Ok(reply) } -/// Her in-force replies that this process has neither told nor read already, oldest first: one selection of the kind, then the document of each new reply -/// and the repos of the briefs it answers. A reply whose document cannot be read is left for the next read. -fn unheard_replies(mind: &dyn MindPort, told: &BTreeSet, waiting: &[HeardReply]) -> Result> { - let mut fresh: Vec<(u64, String)> = headers(mind, &in_force(any_of(of_kind(PipelineKind::Reply), "root", &[&mind.instance().0])))? - .into_iter() - .filter(|header| !told.contains(&header.id.id.0) && !waiting.iter().any(|reply| reply.id == header.id.id.0)) - .map(|header| (header.admission.ordinal, header.id.id.0)) - .collect(); - fresh.sort(); +/// The replies of these routes, oldest first: the document of each and the repos of the briefs it answers. A reply whose +/// 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 (ordinal, id) in fresh { + for Route { id, ordinal, .. } in routes { + let (id, ordinal) = (id.clone(), *ordinal); 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 }; @@ -1892,6 +2007,7 @@ fn start_self_run( batch: &str, ask: &StartSelfRun, taken: Option, + origin: &MemoryOrigin, ) -> Result> { if !campaigns_in_force(organs.mind)?.iter().any(|(slug, _)| *slug == ask.campaign) { return Ok(Err(Failure::SelfRun)); @@ -1905,7 +2021,7 @@ fn start_self_run( item: ask.brief.as_str().into(), why_it_can_wait: "Started from chat; the Self run that claims it works it now.".into(), owner: "self-run".into(), - origin: None, + origin: Some(origin.clone()), }); let claim = PipelineRef { kind: PipelineKind::FollowUp, id: pipeline_key(&follow_up).map_err(|_| anyhow!("the follow-up could not be keyed"))?.as_str().into() }; let mut documents = vec![follow_up]; @@ -1974,10 +2090,11 @@ fn votes(reply: &PersonaReply) -> bool { } /// Where a batch came from and whom she heard it from, and the event ids it rests on: what the organ -/// knows, which no model writes. No deliveries is a wake; a direct first delivery is that DM and the -/// others who spoke in it; else the rooms' aliases and the batch's other senders. -fn origin_of(config: &Config, deliveries: &[ChatDelivery]) -> (MemoryOrigin, Vec) { - let Some(first) = deliveries.first() else { return (MemoryOrigin::Wake, vec![]) }; +/// knows, which no model writes. No deliveries is the session's origin (a wake, or the conversation a reply +/// alone opened); a direct first delivery is that DM and the others who spoke in it; else the rooms' aliases and +/// the batch's other senders. +fn origin_of(config: &Config, deliveries: &[ChatDelivery], session: &MemoryOrigin) -> (MemoryOrigin, Vec) { + let Some(first) = deliveries.first() else { return (session.clone(), vec![]) }; let heard: Vec = other_senders(config, deliveries).into_iter().take(SOURCES_MAX).map(|id| Short(id.to_string())).collect(); let origin = if first.is_direct { MemoryOrigin::Direct { room: Short(first.room_id.clone()), with: heard } @@ -2218,10 +2335,11 @@ fn render_responsibilities(mind: &dyn MindPort, instance: &Slug, names: &BTreeMa out.push_str("Open follow-ups, newest first:\n"); } for follow_up in &follow_ups { - let PipelineFacts::FollowUp { label, source, repo, item, .. } = &follow_up.facts else { continue }; + let PipelineFacts::FollowUp { label, source, repo, item, origin, .. } = &follow_up.facts else { continue }; + let asked = origin.as_ref().map_or_else(String::new, |origin| format!(", {}", recall::origin_phrase(origin, names))); let campaign = follow_up.id.id.0.split_once(":follow_up:").map_or("", |(campaign, _)| campaign); let held = if claimed.contains(&follow_up.id) { " [claimed by a run]" } else { "" }; - let line = format!("- {campaign}/{} ({}){held}: {}\n", label.0, repo.0, item.0.lines().next().unwrap_or("")); + let line = format!("- {campaign}/{} ({}{asked}){held}: {}\n", label.0, repo.0, item.0.lines().next().unwrap_or("")); if out.len() - open_len + line.len() > GOALS_MAX_BYTES - GOALS_RULINGS_RESERVE { dropped += 1; continue; diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index 718e214..dd3edbf 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/crates/mind-body/src/persona/tests.rs @@ -756,7 +756,7 @@ fn a_room_outside_hers_sends_nothing_and_acks_nothing() { } fn direct(seq: u64, room_id: &str, sender: &str, text: &str) -> ChatDelivery { - ChatDelivery { is_direct: true, ..delivery(seq, room_id, sender, text) } + ChatDelivery { is_direct: true, room_id: room_id.into(), ..delivery(seq, room_id, sender, text) } } #[test] @@ -1852,7 +1852,7 @@ fn where_and_from_whom_she_learned_a_memory_comes_from_the_batch_and_no_model_fi assert_eq!(memory.origin, MemoryOrigin::Room { rooms: "swarm, usage".into(), heard_from: vec!["@aqua:chat".into(), "@libby:chat".into()] }); assert_eq!(memory.sources, vec![Short(delivery(1, "swarm", "@aqua:chat", "hello").event_id), Short(delivery(2, "usage", "@libby:chat", "hi").event_id)]); // A direct message: the room id and the others who spoke in it, and its own event ids only. - let dm = Rig::new(vec![direct(1, "dm", "@aqua:chat", "psst")], &[reply(&[], json!([keeps("Learned", "a secret", &["@aqua:chat"])]), json!([]), json!(null))]); + let dm = Rig::new(vec![direct(1, "!dm:chat", "@aqua:chat", "psst")], &[reply(&[], json!([keeps("Learned", "a secret", &["@aqua:chat"])]), json!([]), json!(null))]); assert_eq!(failure_of(dm.go_with(&config())), None); let (_, memory) = memories(&dm).remove(0); assert_eq!((memory.origin, memory.sources.len()), (MemoryOrigin::Direct { room: "!dm:chat".into(), with: vec!["@aqua:chat".into()] }, 1)); @@ -1996,7 +1996,7 @@ fn resolution_sequences_count_up_per_subject() { #[test] fn a_memorys_rooms_are_cut_to_what_a_short_holds_and_whole_characters() { - let rooms = |alias: &str| match origin_of(&config(), &[delivery(1, alias, "@a:chat", "x")]).0 { + let rooms = |alias: &str| match origin_of(&config(), &[delivery(1, alias, "@a:chat", "x")], &MemoryOrigin::Wake).0 { MemoryOrigin::Room { rooms, .. } => rooms.0, other => panic!("{other:?}"), }; @@ -2004,7 +2004,7 @@ fn a_memorys_rooms_are_cut_to_what_a_short_holds_and_whole_characters() { assert_eq!(rooms(&"r".repeat(201)).len(), 200); let wide = format!("{}é", "r".repeat(199)); assert_eq!(rooms(&wide), "r".repeat(199), "a character that would straddle the bound goes whole"); - let joined = origin_of(&config(), &[delivery(1, &"a".repeat(150), "@a:chat", "x"), delivery(2, &"b".repeat(150), "@a:chat", "y")]).0; + let joined = origin_of(&config(), &[delivery(1, &"a".repeat(150), "@a:chat", "x"), delivery(2, &"b".repeat(150), "@a:chat", "y")], &MemoryOrigin::Wake).0; let MemoryOrigin::Room { rooms, .. } = joined else { panic!("a room origin") }; assert_eq!(rooms.0.len(), 200, "{rooms:?}"); } @@ -5936,7 +5936,7 @@ fn with_inbox(rig: &Rig, config: &Config, first: Vec, f: impl F let (control, swarm, run) = (FileSource::new(rig.store()), FileSource::new(rig.swarm_store()), rig.the_run()); let mut progress = Progress::default(); let started_ms = unix_ms(rig.clock.now()); - let mut inbox = PersonaInbox::new(&organs, &control, &swarm, config, &run, &mut progress, first, started_ms, String::new()); + let mut inbox = PersonaInbox::new(&organs, &control, &swarm, config, &run, &mut progress, session_of(config, first), started_ms, String::new()); f(&mut inbox) } @@ -6210,7 +6210,7 @@ fn eight_sessions_one_operator_question() { let mut inboxes: Vec> = progresses .iter_mut() .zip(&configs) - .map(|(progress, config)| PersonaInbox::new(&organs, &control, &swarm, config, &run, progress, vec![question.clone()], 0, String::new())) + .map(|(progress, config)| PersonaInbox::new(&organs, &control, &swarm, config, &run, progress, session_of(config, vec![question.clone()]), 0, String::new())) .collect(); for inbox in &mut inboxes { inbox.start_line().unwrap(); @@ -6362,3 +6362,8 @@ fn a_flush_written_beside_a_revision_line_still_makes_the_reply_final() { assert!(rig.chat.sent.borrow()[1].request_id.contains("-r1-")); }); } + +/// The session a start would open over `first` with no reply waiting: its conversation and origin are the batch's. +fn session_of(config: &Config, first: Vec) -> Session { + Session { conversation: first.first().map_or(Conversation::Public, Conversation::of), origin: origin_of(config, &first, &MemoryOrigin::Wake).0, first } +}