From e10fbb191a7117781674917df6e646b53d2ebfe8 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 2 Nov 2026 15:09:57 +0100 Subject: [PATCH] A refused batch's count and reason survive a session restart; a refusal names the failing field; a line shows a room's earlier lines once, within a named bound Co-Authored-By: Claude Opus 5.5 --- crates/mind-body/src/persona.rs | 144 +++++++++++++--- crates/mind-body/src/persona/tests.rs | 236 ++++++++++++++++++++++++-- 2 files changed, 337 insertions(+), 43 deletions(-) diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index 78b9a78..d901949 100644 --- a/crates/mind-body/src/persona.rs +++ b/crates/mind-body/src/persona.rs @@ -43,7 +43,7 @@ //! she votes for and the urgency she declares) and, through `open_and_launch`, a //! Self run. It never writes the brakes or her persona state. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::fmt; use std::path::{Path, PathBuf}; @@ -61,7 +61,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use uuid::Uuid; -use crate::chat_client::{Chat, ChatDelivery, ChatSendRequest}; +use crate::chat_client::{Chat, ChatDelivery, ChatSendRequest, ContextLine}; use crate::connector::render_plan; use crate::control::{ControlSource, Effective, HeldReason, derive}; use crate::heartbeat::{PollTrace, TempoView}; @@ -80,8 +80,11 @@ pub const PERSONA_POLL_S: i64 = 5; pub const SANDBOX_DIR: &str = "sandbox"; /// The file inside the unit's runtime directory that holds one process's system prompt. const INSTRUCTIONS_FILE: &str = "persona-instructions.md"; -/// A refused batch is told back; its third refusal ends the session, however many answers lie between. +/// A refused batch is told back; its third refusal ends the start, however many answers or session restarts lie between. const REFUSALS_PER_BATCH: u32 = 3; +/// The most bytes of `earlier` text one line carries for one room: the room's tail, newest kept. Each delivery brings its own +/// context, so a batch repeats a room's tail per message; `earlier_shown` owns this bound because a line's size decides how soon the session hits its ceiling. +const EARLIER_MAX_BYTES_PER_ROOM: usize = 16 * 1024; /// The environment variables the unit's `persona.env` sets for the persona flags. pub const ENV_ROOMS: &str = "MIND_PERSONA_ROOMS"; pub const ENV_OPERATOR: &str = "MIND_PERSONA_OPERATOR"; @@ -106,6 +109,13 @@ const GOALS_MAX_BYTES: usize = 6 * 1024; const GOALS_RULINGS_RESERVE: usize = 1024; /// The most rulings a turn's input resolves. const GOALS_RULINGS_MAX: usize = 12; +/// What a refusal tells her about the fields she can be asked to correct most often. +const INTENT_FAULT: &str = "intent: blank or over its length bound"; +const START_FAULT: &str = "start_self_run: its brief, repo or campaign is out of bounds"; +const VOTE_FAULT: &str = "tempo_vote_beat_s or urgency_permille: outside its range"; +const ABOUT_FAULT: &str = "remember.about: every entry must be an id copied exactly from the People line"; +const SETTLE_ITEM_FAULT: &str = "settle.item: not an open item this conversation showed"; +const FOR_ITEM_FAULT: &str = "start_self_run.for_item: not an open item shown, or new: for an Asked or Committed entry of this reply"; /// The most memories, and the most settles, one reply carries. const REMEMBER_MAX: usize = 8; const SETTLE_MAX: usize = 8; @@ -612,6 +622,8 @@ pub enum Failure { /// Her child could not be started, or ended without answering. Session, Undecodable, + /// A reply that parsed and broke the bound of the field this names: the organ's words, never the reply's. + OutOfBounds(&'static str), RoomNotAllowed, Send, SelfRun, @@ -631,6 +643,7 @@ impl fmt::Display for Failure { Self::Lease => f.write_str("the credential lease was refused or unavailable"), Self::Session => f.write_str("her session could not be started or ended without answering"), Self::Undecodable => f.write_str("the reply is not a persona reply"), + Self::OutOfBounds(field) => f.write_str(field), Self::RoomNotAllowed => f.write_str("the reply speaks outside her rooms"), Self::Send => f.write_str("a send was refused or failed"), Self::SelfRun => f.write_str("the Self run could not be started"), @@ -693,6 +706,16 @@ struct Progress { last_intent: Option, failure: Option, self_run: Option, + /// The batch refused and not yet answered: a session restart does not reset its count or what she is told. + refused: Option, +} + +/// A refused batch as the start remembers it. The mouth still holds its deliveries, so this needs only their lowest seq. +#[derive(Clone, Copy)] +struct Refused { + lowest: u64, + why: Failure, + refusals: u32, } /// The instance brake and the swarm store read once: what they allow. @@ -1017,9 +1040,13 @@ impl PersonaInbox<'_, C, S> { } impl SessionInbox for PersonaInbox<'_, C, S> { + /// The first line of a process. A batch an earlier process left refused opens with that refusal and keeps its count. fn start_line(&mut self) -> Result { let batch = std::mem::take(&mut self.start); - self.write(batch, None, 0) + match self.progress.refused { + Some(Refused { why, refusals, .. }) => self.write(batch, Some(refused_note(why)), refusals), + None => self.write(batch, None, 0), + } } fn poll_every_s(&self) -> i64 { @@ -1047,7 +1074,7 @@ impl SessionInbox for PersonaInbox<'_, C, S> fn tell(&mut self) -> Result { if let Some(Retell { batch, why, refusals }) = self.retell.take() { - return self.write(batch, Some(format!("Your last reply was refused ({why}). Answer the same messages again, as one reply.")), refusals); + return self.write(batch, Some(refused_note(why)), refusals); } let batch = std::mem::take(&mut self.waiting); self.write(batch, None, 0) @@ -1104,6 +1131,9 @@ impl SessionInbox for PersonaInbox<'_, C, S> // told in seq order, so while one is unanswered below an answered seq, everything // answered lies above it and nothing is acked; the ack follows its answer. self.answered_through = self.answered_through.max(batch.iter().map(|delivery| delivery.seq).max().unwrap_or(0)); + if self.progress.refused.is_some_and(|refused| batch.iter().any(|delivery| delivery.seq == refused.lowest)) { + self.progress.refused = None; + } let unanswered = self.retell.iter().flat_map(|retell| &retell.batch).chain(self.written.iter().flat_map(|written| &written.batch)).map(|delivery| delivery.seq).min(); let past_unanswered = unanswered.is_some_and(|lowest| lowest <= self.answered_through); if !past_unanswered && self.progress.failure.is_none() && self.answered_through > self.progress.acked { @@ -1134,6 +1164,12 @@ impl SessionInbox for PersonaInbox<'_, C, S> } merged.sort_by_key(|delivery| delivery.seq); merged.dedup_by_key(|delivery| delivery.seq); + // A line in flight may hold a batch at a higher count than this result's, so the start's record only rises. + let held = self.progress.refused.take(); + self.progress.refused = merged + .first() + .map(|first| Refused { lowest: held.map_or(first.seq, |held| held.lowest.min(first.seq)), why, refusals: held.map_or(most, |held| held.refusals.max(most)) }) + .or(held); self.retell = Some(Retell { batch: merged, why, refusals: most }); Ok((true, false)) } @@ -1146,6 +1182,11 @@ impl SessionInbox for PersonaInbox<'_, C, S> } } +/// What a refused batch's next line opens with: which field to correct, and that the refused reply did nothing. +fn refused_note(why: Failure) -> String { + format!("Your last reply was refused ({why}), so nothing in it was said or kept: answer the same messages again, as one reply.") +} + /// The structured output of a result line, as the text `decode_reply` reads; none when it has none. fn structured_output(line: &str) -> Option { let event: Value = serde_json::from_str(line).ok()?; @@ -1271,23 +1312,40 @@ fn distinct(items: &[T]) -> bool { items.iter().enumerate().all(|(index, item)| !items[..index].contains(item)) } -/// What she keeps and settles is in the schema's bounds, names only people she may name, and -/// settles or takes on only an open item shown this turn (or a `new:` ask or commitment she writes now). -fn memory_in_bounds(reply: &PersonaReply, people: &[String], open: &[String]) -> bool { - let remembered = reply.remember.len() <= REMEMBER_MAX - && reply.remember.iter().all(|remember| { - (1..=REMEMBER_TEXT_MAX_BYTES).contains(&remember.text.len()) - && !remember.text.trim().is_empty() - && remember.salience_permille <= SALIENCE_MAX_PERMILLE - && remember.about.len() <= ABOUT_MAX - && remember.about.iter().all(|who| people.contains(who)) - && distinct(&remember.about) - }); - let settled = reply.settle.len() <= SETTLE_MAX - && reply.settle.iter().all(|settle| { - open.contains(&settle.item) && (1..=WHY_MAX_CHARS).contains(&settle.why.chars().count()) && settle.why.len() <= 1000 && !settle.why.trim().is_empty() - }) - && distinct(&reply.settle.iter().map(|settle| settle.item.as_str()).collect::>()); +/// The field a memory or settle breaks, if any: what she keeps and settles is in the schema's bounds, names only people she may +/// name, and settles or takes on only an open item shown this turn (or a `new:` ask or commitment she writes now). +fn memory_fault(reply: &PersonaReply, people: &[String], open: &[String]) -> Option<&'static str> { + if reply.remember.len() > REMEMBER_MAX { + return Some("remember: more entries than one reply may carry"); + } + for remember in &reply.remember { + if !(1..=REMEMBER_TEXT_MAX_BYTES).contains(&remember.text.len()) || remember.text.trim().is_empty() { + return Some("remember.text: blank or over its length bound"); + } + if remember.salience_permille > SALIENCE_MAX_PERMILLE { + return Some("remember.salience_permille: over its maximum"); + } + if remember.about.len() > ABOUT_MAX || !distinct(&remember.about) { + return Some("remember.about: too many entries or one repeated"); + } + if !remember.about.iter().all(|who| people.contains(who)) { + return Some(ABOUT_FAULT); + } + } + if reply.settle.len() > SETTLE_MAX { + return Some("settle: more entries than one reply may carry"); + } + for settle in &reply.settle { + if !open.contains(&settle.item) { + return Some(SETTLE_ITEM_FAULT); + } + if !(1..=WHY_MAX_CHARS).contains(&settle.why.chars().count()) || settle.why.len() > 1000 || settle.why.trim().is_empty() { + return Some("settle.why: blank or over its length bound"); + } + } + if !distinct(&reply.settle.iter().map(|settle| settle.item.as_str()).collect::>()) { + return Some("settle: one item settled twice"); + } let taken = reply.start_self_run.as_ref().and_then(|start| start.for_item.as_deref()).is_none_or(|item| match item.strip_prefix("new:") { Some(index) => { index.len() == 1 @@ -1295,15 +1353,24 @@ fn memory_in_bounds(reply: &PersonaReply, people: &[String], open: &[String]) -> } None => open.iter().any(|shown| shown == item) && !reply.settle.iter().any(|settle| settle.item == item), }); - remembered && settled && taken + (!taken).then_some(FOR_ITEM_FAULT) } /// The reply as the one type it must be, with every room one of `rooms`, every person one of /// `people` and every settled or taken-on item one of the `open` items shown this turn. fn decode_reply(text: &str, rooms: &[String], people: &[String], open: &[String]) -> Result { let reply: PersonaReply = serde_json::from_str(text).map_err(|_| Failure::Undecodable)?; - if reply.intent.trim().is_empty() || reply.intent.chars().count() > INTENT_MAX_CHARS || reply.start_self_run.as_ref().is_some_and(|ask| !start_in_bounds(ask)) || !vote_in_bounds(&reply) || !memory_in_bounds(&reply, people, open) { - return Err(Failure::Undecodable); + if reply.intent.trim().is_empty() || reply.intent.chars().count() > INTENT_MAX_CHARS { + return Err(Failure::OutOfBounds(INTENT_FAULT)); + } + if reply.start_self_run.as_ref().is_some_and(|ask| !start_in_bounds(ask)) { + return Err(Failure::OutOfBounds(START_FAULT)); + } + if !vote_in_bounds(&reply) { + return Err(Failure::OutOfBounds(VOTE_FAULT)); + } + if let Some(field) = memory_fault(&reply, people, open) { + return Err(Failure::OutOfBounds(field)); } if reply.say.iter().any(|say| !rooms.contains(&say.room) || say.text.trim().is_empty()) { return Err(Failure::RoomNotAllowed); @@ -1800,6 +1867,29 @@ const WAKE_QUESTION: &str = "No new messages. Where is your machine least perfec /// Last text of a line with deliveries, after the messages. const TRAFFIC_QUESTION: &str = "Answer what needs you. Then the question of every turn: where is your machine least perfect right now, by your values, and does anything here change your move? Put that in `intent` first."; +/// The `earlier` lines a batch's line shows, one list per delivery: each distinct line of a room once, where it first +/// comes in the batch, none the batch carries as a message, and of each room only the newest within `EARLIER_MAX_BYTES_PER_ROOM`. +fn earlier_shown(deliveries: &[ChatDelivery]) -> Vec> { + let mut seen: BTreeSet<(&str, &str, &str)> = deliveries.iter().map(|delivery| (delivery.room_alias.as_str(), delivery.sender.as_str(), delivery.text.as_str())).collect(); + let mut shown: Vec> = + deliveries.iter().map(|delivery| delivery.context.iter().filter(|line| seen.insert((delivery.room_alias.as_str(), line.sender.as_str(), line.text.as_str()))).collect()).collect(); + let mut left: BTreeMap<&str, usize> = BTreeMap::new(); + for (delivery, lines) in deliveries.iter().zip(&mut shown).rev() { + let room = left.entry(delivery.room_alias.as_str()).or_insert(EARLIER_MAX_BYTES_PER_ROOM); + let kept = lines + .iter() + .rev() + .take_while(|line| { + let fits = line.text.len() <= *room; + *room = if fits { *room - line.text.len() } else { 0 }; + fits + }) + .count(); + lines.drain(..lines.len() - kept); + } + shown +} + /// A batch's line: the core instruction (then the note a retell opens with), her latest runs, what she is responsible for, what she /// remembers, the usage and tempo, the rooms and people she may name, the messages (each a framed /// line whose role the organ derived from the sender id and not from anything the sender wrote), @@ -1831,8 +1921,8 @@ fn render_input(config: &Config, deliveries: &[ChatDelivery], note: Option<&str> out.push_str(MESSAGES_HEADER); out.push('\n'); let names = sender_names(deliveries); - for delivery in deliveries { - for earlier in &delivery.context { + for (delivery, shown) in deliveries.iter().zip(earlier_shown(deliveries)) { + for earlier in shown { out.push_str(&message_line(config, "earlier", &delivery.room_alias, &earlier.sender, None, &Addressees { to_you: false, others: vec![] }, &earlier.text)); } let to = addressees(config, &delivery.mentions, &names); diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index 8f1d5b2..15da141 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/crates/mind-body/src/persona/tests.rs @@ -694,7 +694,7 @@ fn a_start_outside_the_reply_bounds_fails_the_turn_without_an_admission_or_an_ac let reply = json!({ "intent": "Settle the open ask", "say": [], "start_self_run": { "brief": "x", "repo": "not a repo", "campaign": CAMPAIGN, "for_item": null }, "remember": [], "settle": [], "urgency_permille": null }).to_string(); let rig = Rig::new(vec![operator_says(1, "go")], &thrice(reply)); let Outcome::Served { failure, .. } = rig.go() else { panic!("served") }; - assert_eq!(failure, Some(Failure::Undecodable)); + assert_eq!(failure, Some(Failure::OutOfBounds(START_FAULT))); assert!(rig.chat.acked.borrow().is_empty() && rig.mind.ids(K::FollowUp).is_empty()); assert!(rig.launcher.started.borrow().is_empty()); } @@ -1228,7 +1228,7 @@ fn a_vote_or_urgency_outside_its_bounds_fails_the_turn_without_a_pulse_or_an_ack for (beat, urgency) in [(json!(59), json!(null)), (json!(86_401), json!(null)), (json!(null), json!(1001)), (json!(null), json!(-1)), (json!("fast"), json!(null))] { let rig = Rig::new(vec![operator_says(1, "go")], &thrice(voting(Some(beat.clone()), urgency.clone()))); let Outcome::Served { failure, .. } = rig.go() else { panic!("served") }; - assert_eq!(failure, Some(Failure::Undecodable), "{beat} {urgency}"); + assert!(matches!(failure, Some(Failure::Undecodable | Failure::OutOfBounds(VOTE_FAULT))), "{beat} {urgency}: {failure:?}"); assert!(rig.mind.ids(K::Pulse).is_empty() && rig.chat.acked.borrow().is_empty(), "{beat} {urgency}"); } } @@ -1375,6 +1375,7 @@ fn outcomes_and_failures_read_as_one_line_each_without_values() { Failure::Lease, Failure::Session, Failure::Undecodable, + Failure::OutOfBounds(ABOUT_FAULT), Failure::RoomNotAllowed, Failure::Send, Failure::SelfRun, @@ -1718,7 +1719,8 @@ fn an_ask_not_shown_this_turn_is_a_decode_failure_and_nothing_is_sent() { not_asked("new:00"), ] { let rig = Rig::new(vec![delivery(1, "swarm", "@op:chat", "hi")], &thrice(text.clone())); - assert_eq!(failure_of(rig.go()), Some(Failure::Undecodable), "{text}"); + let failure = failure_of(rig.go()); + assert!(matches!(failure, Some(Failure::Undecodable | Failure::OutOfBounds(_))), "{text}: {failure:?}"); assert!(rig.sent_texts().is_empty() && rig.chat.acked.borrow().is_empty() && rig.mind.ids(K::Memory).is_empty()); } } @@ -1731,7 +1733,7 @@ fn an_ask_cannot_be_settled_and_taken_on_in_one_reply() { let id = memories(&rig).remove(0).0; rig.model.replies.borrow_mut().extend(thrice(reply(&[], json!([]), json!([{ "item": id, "outcome": "done", "why": "x" }]), starting(json!(id))))); rig.chat.deliveries.borrow_mut().push(delivery(2, "swarm", "@op:chat", "go")); - assert_eq!(failure_of(rig.go()), Some(Failure::Undecodable)); + assert_eq!(failure_of(rig.go()), Some(Failure::OutOfBounds(FOR_ITEM_FAULT))); } #[test] @@ -1859,12 +1861,12 @@ fn the_open_items_shown_are_what_the_decoder_lets_her_settle_or_take_on() { let shown = ["yggdrasil:memory:a".to_string()]; let settling = |item: &str| reply(&[], json!([]), json!([{ "item": item, "outcome": "done", "why": "x" }]), json!(null)); assert!(decode_reply(&settling("yggdrasil:memory:a"), &rooms, &[], &shown).is_ok()); - assert_eq!(decode_reply(&settling("yggdrasil:memory:b"), &rooms, &[], &shown), Err(Failure::Undecodable), "an item not shown"); - assert_eq!(decode_reply(&settling("yggdrasil:memory:a"), &rooms, &[], &[]), Err(Failure::Undecodable), "with none shown she can settle nothing"); + assert_eq!(decode_reply(&settling("yggdrasil:memory:b"), &rooms, &[], &shown), Err(Failure::OutOfBounds(SETTLE_ITEM_FAULT)), "an item not shown"); + assert_eq!(decode_reply(&settling("yggdrasil:memory:a"), &rooms, &[], &[]), Err(Failure::OutOfBounds(SETTLE_ITEM_FAULT)), "with none shown she can settle nothing"); let taking = |item: &str| reply(&[], json!([]), json!([]), starting(json!(item))); assert!(decode_reply(&taking("yggdrasil:memory:a"), &rooms, &[], &shown).is_ok()); - assert_eq!(decode_reply(&taking("yggdrasil:memory:b"), &rooms, &[], &shown), Err(Failure::Undecodable)); - assert_eq!(decode_reply(&taking("new:0"), &rooms, &[], &shown), Err(Failure::Undecodable), "new: names a memory this reply writes"); + assert_eq!(decode_reply(&taking("yggdrasil:memory:b"), &rooms, &[], &shown), Err(Failure::OutOfBounds(FOR_ITEM_FAULT))); + assert_eq!(decode_reply(&taking("new:0"), &rooms, &[], &shown), Err(Failure::OutOfBounds(FOR_ITEM_FAULT)), "new: names a memory this reply writes"); // The schema leaves the ids to the decoder and the line, and still offers the one-reply form. let schema: serde_json::Value = serde_json::from_str(&reply_schema(&rooms, &[CAMPAIGN.to_string()])).unwrap(); assert_eq!(schema["properties"]["settle"]["items"]["properties"]["item"], json!({ "type": "string" })); @@ -2036,7 +2038,7 @@ fn settling_and_taking_on_a_commitment() { assert!(matches!(outcome_of(&rig, &id), Some(ResolutionOutcome::Deferred { .. }))); // new: on anything else is a decode failure with nothing sent. let rig = Rig::new(vec![delivery(1, "swarm", libby, "fork")], &thrice(reply(&[("swarm", "x")], json!([keeps("Rapport", "we get on", &[libby])]), json!([]), starting(json!("new:0"))))); - assert_eq!(failure_of(rig.go()), Some(Failure::Undecodable)); + assert_eq!(failure_of(rig.go()), Some(Failure::OutOfBounds(FOR_ITEM_FAULT))); assert!(rig.sent_texts().is_empty() && rig.mind.ids(K::Memory).is_empty()); } @@ -2092,7 +2094,7 @@ fn about_can_name_an_absent_mind() { // A stranger is not in it, and naming one is undecodable with nothing sent. let stranger = reply(&[("swarm", "x")], json!([keeps("Idea", "someone", &["@stranger:chat"])]), json!([]), json!(null)); let rig = Rig::new(vec![delivery(1, "swarm", libby, "hi")], &thrice(stranger)); - assert_eq!(failure_of(rig.go()), Some(Failure::Undecodable)); + assert_eq!(failure_of(rig.go()), Some(Failure::OutOfBounds(ABOUT_FAULT))); assert!(rig.sent_texts().is_empty() && rig.chat.acked.borrow().is_empty()); } @@ -2101,7 +2103,7 @@ fn a_long_sender_id_stays_out_of_the_schema_and_never_poisons_a_turn() { let long = format!("@{}:chat", "a".repeat(220)); let named = reply(&[("swarm", "said")], json!([keeps("Learned", "x", &[&long])]), json!([]), json!(null)); let rig = Rig::new(vec![delivery(1, "swarm", &long, "hi")], &thrice(named)); - assert_eq!(failure_of(rig.go()), Some(Failure::Undecodable), "naming it is refused before any say"); + assert_eq!(failure_of(rig.go()), Some(Failure::OutOfBounds(ABOUT_FAULT)), "naming it is refused before any say"); assert!(rig.sent_texts().is_empty() && rig.chat.acked.borrow().is_empty() && rig.mind.ids(K::Memory).is_empty()); assert!(!about_enum(&rig, 0).contains(&long)); // A reply that does not name it is admitted and acked; the member's message still reads as a message. @@ -2779,8 +2781,9 @@ fn a_reply_states_its_intent_first() { let rooms = ["operator".to_string()]; assert!(decode_reply(&intending(&"x".repeat(600)), &rooms, &[], &[]).is_ok()); let missing = json!({ "say": [], "start_self_run": null, "remember": [], "settle": [], "urgency_permille": null }).to_string(); - for refused in [intending(&"x".repeat(601)), intending(""), intending(" \n"), missing] { - assert_eq!(decode_reply(&refused, &rooms, &[], &[]), Err(Failure::Undecodable), "{refused}"); + assert_eq!(decode_reply(&missing, &rooms, &[], &[]), Err(Failure::Undecodable)); + for refused in [intending(&"x".repeat(601)), intending(""), intending(" \n")] { + assert_eq!(decode_reply(&refused, &rooms, &[], &[]), Err(Failure::OutOfBounds(INTENT_FAULT)), "{refused}"); } } @@ -3051,7 +3054,7 @@ fn over_the_ceiling(rig: &Rig, replies: [serde_json::Value; 3]) -> (Closing, Rc< (1, 6_000, echo_line()), (1, 0, result_line(first)), (2, 0, echo_line()), - (2, 0, request_at("m-over", CEILING_TOKENS_FOR_TESTS)), + (2, 0, request_at("m-over", crate::self_run::CEILING_TOKENS)), (3, 0, result_line(retold)), (3, 0, echo_line()), (3, 0, result_line(flush)), @@ -3064,7 +3067,23 @@ fn over_the_ceiling(rig: &Rig, replies: [serde_json::Value; 3]) -> (Closing, Rc< (Closing(child, false), writes) } -const CEILING_TOKENS_FOR_TESTS: u64 = crate::self_run::CEILING_TOKENS; +/// One process that says nothing more than a good reply to its first line, then ends when the mouth is empty. +fn answering(rig: &Rig, reply: serde_json::Value) -> (Closing, Rc>>) { + let writes: Rc>> = Rc::default(); + let child = Timed { writes: writes.clone(), script: VecDeque::from([(1, 6_000, echo_line()), (1, 0, result_line(reply))]), deliveries: rig.chat.deliveries.clone(), arriving: vec![], now: rig.clock.now.clone(), asked: 0 }; + (Closing(child, false), writes) +} + +fn empty_reply() -> serde_json::Value { + json!({"intent": "Nothing is stranded", "say": [], "remember": [], "settle": [], "start_self_run": null, "urgency_permille": null}) +} + +/// The text of the first line a process was told. +fn first_line(writes: &Rc>>) -> String { + let written = writes.borrow(); + let event: serde_json::Value = serde_json::from_str(&written[0]).unwrap(); + event["message"]["content"].as_str().unwrap().to_string() +} #[test] fn a_batch_refused_across_ceiling_restarts_ends_the_start_at_its_third_refusal() { @@ -3075,7 +3094,192 @@ fn a_batch_refused_across_ceiling_restarts_ends_the_start_at_its_third_refusal() let (outcome, acked) = carry_children(&rig, children); let processes = writes.iter().filter(|written| !written.borrow().is_empty()).count(); assert_eq!(processes, 2, "the first process refuses twice, the second's first refusal is the batch's third"); - assert!(failure_of(outcome).is_some()); + assert_eq!(failure_of(outcome), Some(Failure::OutOfBounds(ABOUT_FAULT))); assert!(rig.chat.sent.borrow().is_empty() && acked.is_empty(), "nothing was said and nothing passes the ack, so the mouth still holds both messages"); assert_eq!(rig.chat.deliveries.borrow().len(), 2); + assert!(rig.closing_reason().contains(ABOUT_FAULT), "the run's closing names the field: {}", rig.closing_reason()); + let restarted = first_line(&writes[1]); + assert!(restarted.contains(&format!("Your last reply was refused ({ABOUT_FAULT}), so nothing in it was said or kept")), "a fresh session is told what the last one was refused: {restarted}"); + assert!(restarted.contains(r#"text="three""#) && restarted.contains(r#"text="four""#)); +} + +/// The one reply that broke the loop on 2026-11-02 came from a fresh process, and it is acked as any answer is. +#[test] +fn a_batch_refused_before_a_restart_and_answered_after_it_is_acked() { + let rig = Rig::new(vec![operator_says(3, "three")], &[]); + rig.clock.step.set(0); + let refused = over_the_ceiling(&rig, [reply_about_bare_name(), reply_about_bare_name(), reply_about_bare_name()]); + let answered = answering(&rig, reply_saying("corrected")); + let writes = [refused.1.clone(), answered.1.clone()]; + let (outcome, acked) = carry_children(&rig, vec![refused.0, answered.0]); + assert_eq!(failure_of(outcome), None); + assert_eq!(rig.sent_texts(), ["corrected"]); + assert_eq!(acked, [3]); + assert!(first_line(&writes[1]).contains("was refused")); +} + +/// A flush alone carries no batch, so a good reply to it answers nothing: the batch's count stands. +#[test] +fn a_good_flush_after_a_refused_batch_leaves_the_batchs_count_standing() { + let rig = Rig::new(vec![operator_says(3, "three")], &[]); + rig.clock.step.set(0); + let bare = reply_about_bare_name; + let (children, writes): (Vec<_>, Vec<_>) = [over_the_ceiling(&rig, [bare(), bare(), empty_reply()]), over_the_ceiling(&rig, [bare(), bare(), bare()]), over_the_ceiling(&rig, [bare(), bare(), bare()])].into_iter().unzip(); + let (outcome, acked) = carry_children(&rig, children); + assert_eq!(failure_of(outcome), Some(Failure::OutOfBounds(ABOUT_FAULT))); + assert_eq!(writes.iter().filter(|written| !written.borrow().is_empty()).count(), 2); + assert!(acked.is_empty()); +} + +/// A new batch after an answered one starts at no refusals, across a restart too. +#[test] +fn an_answer_clears_the_count_so_a_later_batch_gets_its_own_three() { + let rig = Rig::new(vec![operator_says(3, "three")], &[]); + rig.clock.step.set(0); + let bare = reply_about_bare_name; + let first = over_the_ceiling(&rig, [bare(), bare(), bare()]); + let writes: Rc>> = Rc::default(); + // The retold 3 is answered, 5 arrives and is refused once, and the ceiling ends the process. + let second = Timed { + writes: writes.clone(), + script: VecDeque::from([ + (1, 6_000, echo_line()), + (1, 0, result_line(reply_saying("for three"))), + (2, 0, echo_line()), + (2, 0, request_at("m-over-2", crate::self_run::CEILING_TOKENS)), + (3, 0, result_line(bare())), + (3, 0, echo_line()), + (3, 0, result_line(empty_reply())), + ]), + deliveries: rig.chat.deliveries.clone(), + arriving: vec![(1, operator_says(5, "five"))], + now: rig.clock.now.clone(), + asked: 0, + }; + // 5 is refused once more here, its second: added to 3's two it would be a third and end the start. + let third_writes: Rc>> = Rc::default(); + let third = Timed { + writes: third_writes.clone(), + script: VecDeque::from([(1, 6_000, echo_line()), (1, 0, result_line(bare())), (2, 0, echo_line()), (2, 0, result_line(reply_saying("for five")))]), + deliveries: rig.chat.deliveries.clone(), + arriving: vec![], + now: rig.clock.now.clone(), + asked: 0, + }; + let (outcome, acked) = carry_children(&rig, vec![first.0, Closing(second, false), Closing(third, false)]); + assert_eq!(failure_of(outcome), None, "5's one refusal is not added to 3's two"); + assert_eq!(rig.sent_texts(), ["for three", "for five"]); + assert_eq!(acked, [3, 5]); + assert!(first_line(&third_writes).contains("was refused"), "5's refusal crosses the restart"); +} + +/// The reply itself never reaches the note or the closing: the field is named in the organ's words. +#[test] +fn a_refusal_names_the_field_and_echoes_nothing_she_wrote() { + let canary = "@canary-value:nowhere"; + let reply = json!({"intent": "x", "say": [], "remember": [{"sort": "Did", "text": "kept", "about": [canary], "salience_permille": 1}], "settle": [], "start_self_run": null, "urgency_permille": null}).to_string(); + let rig = Rig::new(vec![operator_says(1, "a")], &thrice(reply)); + assert_eq!(failure_of(rig.go()), Some(Failure::OutOfBounds(ABOUT_FAULT))); + let told = rig.model.requests.borrow()[1].input.clone(); + assert!(told.contains(ABOUT_FAULT), "{told}"); + assert!(!told.contains(canary) && !rig.closing_reason().contains(canary)); +} + +#[test] +fn every_memory_bound_names_its_own_field() { + let rooms = ["swarm".to_string()]; + let people = ["@aqua:chat".to_string()]; + let memory = |about: serde_json::Value, text: &str, salience: u16| json!({"sort": "Did", "text": text, "about": about, "salience_permille": salience}); + let keeping = |remember: serde_json::Value| reply(&[], remember, json!([]), json!(null)); + let settling = |settle: serde_json::Value| reply(&[], json!([]), settle, json!(null)); + let open = ["yggdrasil:memory:a".to_string()]; + let done = |why: &str| json!({ "item": "yggdrasil:memory:a", "outcome": "done", "why": why }); + let eight_more = json!((0..9).map(|_| memory(json!([]), "x", 1)).collect::>()); + for (text, field) in [ + (keeping(eight_more), "remember: more entries than one reply may carry"), + (keeping(json!([memory(json!([]), " ", 1)])), "remember.text: blank or over its length bound"), + (keeping(json!([memory(json!([]), &"x".repeat(REMEMBER_TEXT_MAX_BYTES + 1), 1)])), "remember.text: blank or over its length bound"), + (keeping(json!([memory(json!([]), "x", SALIENCE_MAX_PERMILLE + 1)])), "remember.salience_permille: over its maximum"), + (keeping(json!([memory(json!(["@aqua:chat", "@aqua:chat"]), "x", 1)])), "remember.about: too many entries or one repeated"), + (keeping(json!([memory(json!(["aqua"]), "x", 1)])), ABOUT_FAULT), + (settling(json!((0..SETTLE_MAX + 1).map(|_| done("x")).collect::>())), "settle: more entries than one reply may carry"), + (settling(json!([done(" ")])), "settle.why: blank or over its length bound"), + (settling(json!([done(&"x".repeat(WHY_MAX_CHARS + 1))])), "settle.why: blank or over its length bound"), + (settling(json!([done("x"), done("y")])), "settle: one item settled twice"), + ] { + assert_eq!(decode_reply(&text, &rooms, &people, &open), Err(Failure::OutOfBounds(field)), "{field}"); + } + assert!(decode_reply(&keeping(json!([memory(json!(["@aqua:chat"]), "x", SALIENCE_MAX_PERMILLE)])), &rooms, &people, &open).is_ok()); + assert!(decode_reply(&settling(json!([done(&"x".repeat(WHY_MAX_CHARS))])), &rooms, &people, &open).is_ok()); +} + +fn earlier(sender: &str, text: &str) -> ContextLine { + ContextLine { sender: sender.into(), text: text.into(), at_unix_ms: 1_700_000_000_000 } +} + +fn with_context(mut message: ChatDelivery, context: Vec) -> ChatDelivery { + message.context = context; + message +} + +fn earlier_lines(input: &str) -> Vec { + input.lines().filter(|line| line.starts_with("earlier ")).map(String::from).collect() +} + +/// Nibu's line of 2026-11-02 in small: every message carries the room's tail, which holds the batch's own earlier messages. +#[test] +fn a_batchs_line_shows_each_earlier_line_once_where_it_first_comes() { + let (old, older) = (earlier("@aqua:chat", "old"), earlier("@libby:chat", "older")); + let messages = vec![ + with_context(delivery(1, "swarm", "@op:chat", "one"), vec![older.clone(), old.clone()]), + with_context(delivery(2, "swarm", "@op:chat", "two"), vec![old.clone(), earlier("@op:chat", "one"), earlier("@nibu:chat", "her own between")]), + with_context(delivery(3, "swarm", "@op:chat", "three"), vec![earlier("@op:chat", "one"), earlier("@nibu:chat", "her own between"), earlier("@op:chat", "two")]), + ]; + let rig = Rig::new(messages, &[says(&[("swarm", "ok")])]); + assert_eq!(failure_of(rig.go()), None); + let input = rig.model.requests.borrow()[0].input.clone(); + let shown = earlier_lines(&input); + assert_eq!(shown.len(), 3, "older, old and her line, and no message of the batch again: {shown:?}"); + let at = |needle: &str| input.find(needle).unwrap_or_else(|| panic!("{needle} in {input}")); + assert!(at(r#"text="older""#) < at(r#"text="old""#) && at(r#"text="old""#) < at(r#"text="one""#), "a room's lines stay in order"); + assert!(at(r#"text="one""#) < at(r#"text="her own between""#) && at(r#"text="her own between""#) < at(r#"text="two""#), "a line shows where it first comes"); +} + +#[test] +fn the_same_text_in_another_room_is_another_line() { + let line = earlier("@aqua:chat", "same words"); + let messages = vec![with_context(delivery(1, "swarm", "@op:chat", "one"), vec![line.clone()]), with_context(delivery(2, "operator", "@op:chat", "two"), vec![line])]; + let rig = Rig::new(messages, &[says(&[("swarm", "ok")])]); + assert_eq!(failure_of(rig.go()), None); + assert_eq!(earlier_lines(&rig.model.requests.borrow()[0].input).len(), 2); +} + +/// The bound keeps the tail: the newest lines whole up to exactly the bound, nothing older past the first that does not fit, and each room its own. +#[test] +fn a_rooms_earlier_lines_keep_the_newest_within_the_bound() { + let half = EARLIER_MAX_BYTES_PER_ROOM / 2; + let line = |mark: char| earlier("@aqua:chat", &mark.to_string().repeat(half)); + let messages = vec![ + with_context(delivery(1, "swarm", "@op:chat", "one"), vec![line('a'), earlier("@aqua:chat", "tiny")]), + with_context(delivery(2, "swarm", "@op:chat", "two"), vec![line('b'), line('c')]), + with_context(delivery(3, "operator", "@op:chat", "three"), vec![line('d'), line('e')]), + ]; + let rig = Rig::new(messages, &[says(&[("swarm", "ok")])]); + assert_eq!(failure_of(rig.go()), None); + let shown = earlier_lines(&rig.model.requests.borrow()[0].input); + let marks: Vec = shown.iter().filter_map(|line| line.rsplit("text=\"").next().and_then(|text| text.chars().next())).collect(); + assert_eq!(marks, ['b', 'c', 'd', 'e'], "b and c fill swarm's bound exactly, so tiny and a are gone though tiny would fit; operator keeps its own"); +} + +/// The line that blew the prefix: 34 messages, each with its 20 of context, down to the room's distinct lines. +#[test] +fn thirty_four_messages_with_a_shared_tail_do_not_repeat_it() { + let tail: Vec = (0..40).map(|index| earlier("@aqua:chat", &format!("line {index}"))).collect(); + let messages: Vec = (0..34u64).map(|seq| { + let start = usize::try_from(seq).unwrap() / 2; + with_context(delivery(seq + 1, "swarm", "@op:chat", &format!("message {seq}")), tail[start..start + 20].to_vec()) + }).collect(); + let rig = Rig::new(messages, &[says(&[("swarm", "ok")])]); + assert_eq!(failure_of(rig.go()), None); + assert_eq!(earlier_lines(&rig.model.requests.borrow()[0].input).len(), 36, "16 starts and 20 of tail, not 680"); }