plan line: absolute instants, so a figure rendered once stays true for the session (squash of eureka-body/plan-read-honest f0566fd..7ac4098; Soul s5)

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

crates/mind-body/src/chat_client.rs

@@ -4,8 +4,8 @@
 //! The mouth is Bifrost's chat ear and mouth (cut `persona-mouth`). Five CultNet document kinds cross
 //! it, each a positional MessagePack tuple (fields in declaration order, `rmp_serde::to_vec`; the mouth refuses maps): the pending request and its answer, the
 //! ack, and the send request with its receipt. Their field order is the mouth's, pinned
-//! by its golden fixtures (`tests/fixtures/persona-chat-v1/`, Bifrost
-//! `eureka-body/persona-mouth` ab2bdc6) in both directions. `chat_rudp::RudpChat` is
+//! by its golden fixtures (`tests/fixtures/persona-chat-v1/` for the pending, ack and refusal kinds, `persona-chat-v2/` for the send
+//! and its receipt) in both directions. `chat_rudp::RudpChat` is
 //! the RUDP transport that carries them; `Chat` is its seam, and `encode` and
 //! `decode` are its codec. Room aliases are the mouth's
 //! full aliases (`#operator:gamecult.org`), which the caller's `--rooms` carries.
@@ -20,8 +20,8 @@ use serde::{Deserialize, Serialize, de::DeserializeOwned};
 pub const PENDING_REQUEST: &str = "gamecult.persona.chat_pending_request.v1";
 pub const PENDING: &str = "gamecult.persona.chat_pending.v1";
 pub const ACK: &str = "gamecult.persona.chat_ack.v1";
-pub const SEND_REQUEST: &str = "gamecult.persona.chat_send_request.v1";
-pub const SEND_RECEIPT: &str = "gamecult.persona.chat_send_receipt.v1";
+pub const SEND_REQUEST: &str = "gamecult.persona.chat_send_request.v2";
+pub const SEND_RECEIPT: &str = "gamecult.persona.chat_send_receipt.v2";
 pub const REFUSAL: &str = "gamecult.persona.chat_refusal.v1";
 
 /// One earlier line of a room, delivered with a message for its context.
@@ -78,6 +78,8 @@ pub enum Refusal {
     EmptyText,
     SendFailed,
     UnknownPersona,
+    /// A send with a basis found the room had moved past it: the receipt names what moved it.
+    RoomMoved,
 }
 
 impl Refusal {
@@ -90,6 +92,7 @@ impl Refusal {
             Self::EmptyText => "EmptyText",
             Self::SendFailed => "SendFailed",
             Self::UnknownPersona => "UnknownPersona",
+            Self::RoomMoved => "RoomMoved",
         }
     }
 }
@@ -103,7 +106,7 @@ impl Serialize for Refusal {
 impl<'de> Deserialize<'de> for Refusal {
     fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
         let name = String::deserialize(deserializer)?;
-        [Self::Brake, Self::PersonaMismatch, Self::UnknownRoom, Self::EmptyText, Self::SendFailed, Self::UnknownPersona]
+        [Self::Brake, Self::PersonaMismatch, Self::UnknownRoom, Self::EmptyText, Self::SendFailed, Self::UnknownPersona, Self::RoomMoved]
             .into_iter()
             .find(|refusal| refusal.name() == name)
             .ok_or_else(|| serde::de::Error::custom("the mouth named a refusal this side does not know"))
@@ -118,22 +121,34 @@ pub struct ChatRefusal {
     pub refusal: Refusal,
 }
 
+/// What a send's author had seen of its room: the newest event of the room her input held (none when it held
+/// none), and when her process began. The mouth refuses the send if another sender's message reached the room
+/// after that event, or after that time where there is none.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SendBasis {
+    pub seen_event_id: Option<String>,
+    pub since_unix_ms: u64,
+}
+
 /// A send. `request_id` is the mouth's transaction id, so a repeat of the same
-/// id is the same send.
+/// id is the same send. No `basis` is her choice standing: the mouth checks nothing.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ChatSendRequest {
     pub request_id: String,
     pub persona: String,
     pub room_alias: String,
     pub text: String,
+    pub basis: Option<SendBasis>,
 }
 
-/// The mouth's answer to a send: the event it made, or why it did not.
+/// The mouth's answer to a send: the event it made, or why it did not. `newer` names the events that moved the
+/// room, oldest first, exactly when the refusal is `RoomMoved`.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ChatSendReceipt {
     pub request_id: String,
     pub event_id: Option<String>,
-    pub refusal: Option<String>,
+    pub refusal: Option<Refusal>,
+    pub newer: Vec<String>,
 }
 
 /// A document as the mouth reads it: a positional tuple.

crates/mind-body/src/chat_client/tests.rs

@@ -1,5 +1,5 @@
-//! The chat documents against the mouth's own golden bytes (Bifrost
-//! `eureka-body/persona-mouth` ab2bdc6, `tests/fixtures/persona-chat-v1/`):
+//! The chat documents against the mouth's golden bytes (`tests/fixtures/persona-chat-v1/` for the pending, ack and refusal
+//! kinds, Bifrost `eureka-body/persona-mouth` ab2bdc6; `persona-chat-v2/` for the send and its receipt, copied from Bifrost `eureka-body/optimistic-send-mouth`):
 //! each decodes into the typed document and the encoder reproduces it byte for
 //! byte. The mouth reads positional tuples and refuses maps.
 
@@ -9,8 +9,11 @@ const REFUSAL_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v
 const ACK_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack");
 const PENDING_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack");
 const PENDING_REQUEST_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack");
-const SEND_RECEIPT_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack");
-const SEND_REQUEST_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack");
+const SENT_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack");
+const ROOM_MOVED_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.room-moved.msgpack");
+const NIL_BASIS_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.nil-basis.msgpack");
+const SEEN_AND_SINCE_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.seen-and-since.msgpack");
+const SINCE_ONLY_BYTES: &[u8] = include_bytes!("../../tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.since-only.msgpack");
 
 fn golden<T: Serialize + DeserializeOwned + PartialEq + std::fmt::Debug>(bytes: &[u8], expected: &T) {
     assert_eq!(&decode::<T>(bytes).unwrap(), expected);
@@ -49,9 +52,10 @@ fn a_refusal_is_the_mouths_bytes_and_a_typed_name() {
 
 #[test]
 fn a_send_and_its_receipt_are_the_mouths_bytes() {
-    golden(
-        SEND_REQUEST_BYTES,
-        &ChatSendRequest { request_id: "req-1".into(), persona: "eureka".into(), room_alias: "#operator:gamecult.org".into(), text: "hi".into() },
-    );
-    golden(SEND_RECEIPT_BYTES, &ChatSendReceipt { request_id: "req-1".into(), event_id: Some("$sent".into()), refusal: None });
+    let request = |basis: Option<SendBasis>| ChatSendRequest { request_id: "req-1".into(), persona: "eureka".into(), room_alias: "#swarm:gamecult.org".into(), text: "hi".into(), basis };
+    golden(NIL_BASIS_BYTES, &request(None));
+    golden(SEEN_AND_SINCE_BYTES, &request(Some(SendBasis { seen_event_id: Some("$seen".into()), since_unix_ms: 1_793_698_425_998 })));
+    golden(SINCE_ONLY_BYTES, &request(Some(SendBasis { seen_event_id: None, since_unix_ms: 1_793_698_425_998 })));
+    golden(SENT_BYTES, &ChatSendReceipt { request_id: "req-1".into(), event_id: Some("$sent".into()), refusal: None, newer: vec![] });
+    golden(ROOM_MOVED_BYTES, &ChatSendReceipt { request_id: "req-1".into(), event_id: None, refusal: Some(Refusal::RoomMoved), newer: vec!["$a".into(), "$b".into()] });
 }

crates/mind-body/src/chat_rudp.rs

@@ -95,7 +95,21 @@ impl RudpChat {
     /// Puts one document and, when `answer` names a kind and record key, waits for
     /// the document the mouth answers it with. Documents of other kinds or keys
     /// (a late receipt of an earlier send) are skipped.
+    ///
+    /// A failed exchange drops the link: a restarted mouth silently ignores packets from a sender it has not connected, so a held link
+    /// would time out on every call after it. The next call opens a fresh one. Nothing is repeated on it: the caller's request id is
+    /// what makes a repeat of a send the same send.
     fn exchange<T: Serialize>(&self, kind: &str, record_key: &str, document: &T, answer: Option<(&[&str], &str)>) -> Result<Option<(String, Vec<u8>)>> {
+        let result = self.exchange_on_link(kind, record_key, document, answer);
+        if result.is_err()
+            && let Some(mut link) = self.link.borrow_mut().take()
+        {
+            let _ = link.transport.disconnect(Vec::new());
+        }
+        result
+    }
+
+    fn exchange_on_link<T: Serialize>(&self, kind: &str, record_key: &str, document: &T, answer: Option<(&[&str], &str)>) -> Result<Option<(String, Vec<u8>)>> {
         let mut slot = self.link.borrow_mut();
         if slot.is_none() {
             *slot = Some(self.open()?);

crates/mind-body/src/connector.rs

@@ -4,28 +4,24 @@
 use chrono::{DateTime, SecondsFormat, Utc};
 use model_connector::PlanQuota;
 
-/// A span in the two largest units that fit: `45s`, `12m`, `3h 20m`, `4d 11h`.
-fn span(ms: u64) -> String {
-    let seconds = ms / 1000;
-    match seconds {
-        0..=59 => format!("{seconds}s"),
-        60..=3599 => format!("{}m", seconds / 60),
-        3600..=86399 => format!("{}h {}m", seconds / 3600, seconds % 3600 / 60),
-        _ => format!("{}d {}h", seconds / 86400, seconds % 86400 / 3600),
-    }
+/// When a window resets: the UTC time alone. An instant is true when read again hours later; "in 4d 11h"
+/// or "already past" is not.
+fn resets(at_unix_ms: u64) -> String {
+    let at = i64::try_from(at_unix_ms).ok().and_then(DateTime::<Utc>::from_timestamp_millis).map_or_else(|| at_unix_ms.to_string(), |at| at.to_rfc3339_opts(SecondsFormat::Secs, true));
+    format!("resets at {at}")
 }
 
-/// When a window resets: the UTC time, and how far off it is from `now_ms`.
-fn resets(at_unix_ms: u64, now_ms: u64) -> String {
-    let at = i64::try_from(at_unix_ms).ok().and_then(DateTime::<Utc>::from_timestamp_millis).map_or_else(|| at_unix_ms.to_string(), |at| at.to_rfc3339_opts(SecondsFormat::Secs, true));
-    if at_unix_ms > now_ms { format!("resets at {at} (in {})", span(at_unix_ms - now_ms)) } else { format!("resets at {at} (already past)") }
+/// The minute a reading was taken, as a UTC time. An instant, not an age: the line stays true when
+/// it is read again hours later in a session that cannot refresh it.
+fn read_at(at_unix_ms: u64) -> String {
+    i64::try_from(at_unix_ms).ok().and_then(DateTime::<Utc>::from_timestamp_millis).map_or_else(|| at_unix_ms.to_string(), |at| at.format("%Y-%m-%dT%H:%MZ").to_string())
 }
 
 /// The plan windows as one line for a Mind: each window's percent used, its
-/// status, when it resets and how old the reading is. Rendered from the readout
+/// status, when it resets and the minute the reading was taken. Rendered from the readout
 /// and nothing else, so no number in it is ever a paraphrase; the guidance that
 /// follows it states the swarm's aim and that nothing enforces it.
-pub fn render_plan(windows: &[PlanQuota], now_ms: u64) -> String {
+pub fn render_plan(windows: &[PlanQuota], _now_ms: u64) -> String {
     let guidance = "The swarm aims to have used about 90% of the weekly allowance at reset; nothing enforces it; you and the other Minds set the tempo.";
     if windows.is_empty() {
         return format!("Plan usage: unobserved. {guidance}");
@@ -34,15 +30,15 @@ pub fn render_plan(windows: &[PlanQuota], now_ms: u64) -> String {
         .iter()
         .map(|plan| match plan.utilization_permille {
             Some(permille) => format!(
-                "{} {}.{}% {}, {}, read {} ago",
+                "{} {}.{}% {}, {}, read at {}",
                 plan.window,
                 permille / 10,
                 permille % 10,
                 plan.status,
-                resets(plan.resets_at_unix_ms, now_ms),
-                span(now_ms.saturating_sub(plan.observed_at_unix_ms))
+                resets(plan.resets_at_unix_ms),
+                read_at(plan.observed_at_unix_ms)
             ),
-            None => format!("{} expired, {}", plan.window, resets(plan.resets_at_unix_ms, now_ms)),
+            None => format!("{} expired, {}", plan.window, resets(plan.resets_at_unix_ms)),
         })
         .collect();
     format!("Plan usage: {}. {guidance}", rendered.join("; "))
@@ -57,20 +53,25 @@ mod tests {
     }
 
     #[test]
-    fn a_window_reads_as_a_time_and_spans_not_unix_milliseconds() {
+    fn a_window_reads_as_a_time_not_unix_milliseconds() {
         let now = 1_793_307_000_000 - (4 * 86_400 + 11 * 3600 + 30 * 60) * 1000;
         let line = render_plan(&[window(Some(423), 1_793_307_000_000, now - 385_639_000)], now);
-        assert!(line.starts_with("Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z (in 4d 11h), read 4d 11h ago. "), "{line}");
+        assert!(line.starts_with("Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z, read at 2026-10-20T22:12Z. "), "{line}");
         assert!(!line.contains("unix ms"), "{line}");
     }
 
     #[test]
-    fn spans_use_the_two_largest_units_and_a_passed_reset_says_so() {
+fn the_reset_is_an_instant_whatever_the_moment_of_rendering() {        let reset = 1_793_307_000_000;        let at = |now| render_plan(&[window(Some(10), reset, reset - 5000)], now);        let before = at(reset - 5000);        assert!(before.contains("resets at 2026-10-29T20:50:00Z, read at 2026-10-29T20:49Z"), "{before}");        for now in [reset - 12_000_000, reset, reset + 1000] {            assert_eq!(at(now), before, "the line does not depend on when it is rendered");        }        assert!(!before.contains("(in ") && !before.contains("already past"), "{before}");        assert_eq!(render_plan(&[window(None, reset, reset)], reset - 12_000_000).split(". ").next().unwrap(), "Plan usage: five_hour expired, resets at 2026-10-29T20:50:00Z");    }
+    #[test]
+    fn the_reading_is_dated_by_the_minute_it_was_taken_and_the_line_does_not_age() {
         let reset = 1_793_307_000_000;
-        assert!(render_plan(&[window(Some(10), reset, reset - 5000)], reset - 5000).contains("(in 5s), read 0s ago"));
-        assert!(render_plan(&[window(Some(10), reset, reset - 720_000)], reset - 600_000).contains("(in 10m), read 2m ago"));
-        assert!(render_plan(&[window(Some(10), reset, reset)], reset).contains("(already past), read 0s ago"), "a reset at this very moment is past");
-        assert!(render_plan(&[window(Some(10), reset, reset)], reset + 1000).contains("(already past), read 1s ago"));
-        assert!(render_plan(&[window(None, reset, reset)], reset - 12_000_000).contains("five_hour expired, resets at 2026-10-29T20:50:00Z (in 3h 20m)"));
+        let taken = 1_793_306_999_000; // 20:49:59Z: the minute is cut, not rounded
+        let first = render_plan(&[window(Some(10), reset, taken)], taken);
+        assert!(first.contains(", read at 2026-10-29T20:49Z. "), "{first}");
+        let hours_later = render_plan(&[window(Some(10), reset, taken)], taken + 3 * 3_600_000);
+        assert!(hours_later.contains(", read at 2026-10-29T20:49Z. "), "{hours_later}");
+        assert!(!first.contains(" ago") && !hours_later.contains(" ago"), "{first} {hours_later}");
+        assert!(render_plan(&[window(Some(10), reset, 0)], reset).contains(", read at 1970-01-01T00:00Z. "));
+        assert!(render_plan(&[window(Some(10), reset, u64::MAX)], reset).contains(&format!(", read at {}. ", u64::MAX)), "an instant out of range is printed as the number it is");
     }
 }

crates/mind-body/src/persona.rs

@@ -35,7 +35,18 @@
 //! third in a row ends the session. Any other failure sends nothing further, acks nothing and
 //! ends the start; the run is closed Recorded with the failure, so the breaker and the next wake
 //! see it. Nothing is acked that was not answered, and a repeat of a send carries the same
-//! request id, so a start that dies between a send and its ack repeats no message.
+//! request id, so a start that dies between a send and its ack does not repeat it; an answer to a revision line carries a new id,
+//! so it can repeat a send.
+//!
+//! Each send names what her input held of its room (ruling `eureka-body:ruling:send-refused-if-room-moved`):
+//! the newest message of that room a line the CLI took in showed her. The mouth decides whether the room moved;
+//! the organ owns only what her input held and whether a reply is her final answer. A send the mouth refuses as
+//! `RoomMoved` stops that reply at once (the says before it stand, nothing after it is done) and is not a
+//! refused result: she is handed the messages that interrupted her as her next input, once they have reached
+//! the organ, to send, change or drop. Her answer to that line is final: it goes out unchecked, as does the reply
+//! to a flush, which has no next turn to revise in. A message reaches a live session after her rank among the Minds for that message (the Minds it names first), at one
+//! step per rank
+//! (ruling `eureka-body:ruling:unaddressed-tell-stagger-ruled`), so the Minds read each other before they answer.
 //!
 //! The reply is decoded, never scanned: text that is not one `PersonaReply` is a refused
 //! result. Chat, persona state and trace are ports; the chat transport is `chat_rudp::RudpChat`,
@@ -66,7 +77,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 use uuid::Uuid;
 
-use crate::chat_client::{Chat, ChatDelivery, ChatSendRequest, ContextLine};
+use crate::chat_client::{Chat, ChatDelivery, ChatSendRequest, ContextLine, Refusal, SendBasis};
 use crate::connector::render_plan;
 use crate::control::{ControlSource, Effective, HeldReason, derive};
 use crate::heartbeat::{PollTrace, TempoView};
@@ -96,6 +107,13 @@ const REFUSALS_PER_BATCH: u32 = 3;
 /// The most bytes of `earlier` text, as rendered into the line (JSON-quoted), 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;
+/// A live session is told a message this many seconds per rank, her rank among the Minds being a stable hash of the Mind and the
+/// event, the Minds the message names first: the first Mind by that order is told at once, the next after one step. The step is
+/// calibration, about the time a Mind takes to answer (7 to 13 s observed on 2026-11-03).
+const STAGGER_STEP_S: u64 = 8;
+/// Polls a revision waits for the interrupting messages to reach the organ's pending before it is told without them (at most
+/// `REVISION_POLLS * PERSONA_POLL_S` seconds).
+const REVISION_POLLS: u32 = 2;
 /// 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";
@@ -701,6 +719,9 @@ enum Step {
     Answered(Option<PipelineRef>, String, Vec<String>),
     /// The reply could not be decoded: it is told back to her, and sends and acks nothing.
     Refused(Failure),
+    /// The mouth refused a send because the room moved past what her input held: the room, the rooms sent before it in the
+    /// reply, and the events that moved it. Nothing after the send was done.
+    Stale { room: String, sent: Vec<String>, newer: Vec<String> },
     /// The start cannot go on; the intent her decoded reply stated still closes the run beside the failure.
     Failed(Failure, String),
 }
@@ -863,35 +884,7 @@ fn one_process(
         progress.failure = Some(Failure::Session);
         return Ok(());
     };
-    let conversation = first.first().map_or(Conversation::Public, Conversation::of);
-    let mut inbox = PersonaInbox {
-        organs,
-        control,
-        swarm,
-        config,
-        run,
-        progress,
-        rooms,
-        conversation,
-        plan: render_plan(lease.plan_quota(), started_ms),
-        started_ms,
-        told: BTreeSet::new(),
-        start: first,
-        waiting: Vec::new(),
-        told_through: 0,
-        answered_through: 0,
-        written: None,
-        echoed: Vec::new(),
-        echoed_refusals: 0,
-        seen: Shown::default(),
-        retell: None,
-        slot_seen: Vec::new(),
-        slot_read_due_ms: 0,
-        notices: Vec::new(),
-        repos: Vec::new(),
-        repo_head: None,
-        repo_read_due_ms: 0,
-    };
+    let mut inbox = PersonaInbox::new(organs, control, swarm, config, run, progress, first, started_ms, render_plan(lease.plan_quota(), started_ms));
     drive(organs.clock, child, &mut inbox, remaining_s, &mut std::io::stdout())?;
     Ok(())
 }
@@ -975,6 +968,8 @@ struct Written {
     batch: Vec<ChatDelivery>,
     shown: Shown,
     refusals: u32,
+    /// The line is a revision (or the retell of one): the reply to it is her final answer.
+    revision: bool,
 }
 
 /// Every refused batch not yet told again, merged into one, with the refusal to tell and the most
@@ -983,6 +978,21 @@ struct Retell {
     batch: Vec<ChatDelivery>,
     why: Failure,
     refusals: u32,
+    /// An undecodable revision stays a revision: its answer is final.
+    revision: bool,
+}
+
+/// A reply stopped by a send the mouth refused as `RoomMoved`, waiting to be handed back to her as her next input: the
+/// deliveries it answered, where it stopped and what it had sent, and the events that moved the room, which the line waits to
+/// have reached the organ.
+#[derive(Default)]
+struct Revision {
+    batch: Vec<ChatDelivery>,
+    rooms: Vec<String>,
+    sent: Vec<String>,
+    awaiting: Vec<String>,
+    polls: u32,
+    ready: bool,
 }
 
 /// A rendered line: its text, what it lets a reply name, and the lines it tells.
@@ -1030,6 +1040,18 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> {
     seen: Shown,
     /// What refused results left to be told again.
     retell: Option<Retell>,
+    /// What each room's newest message was among the lines the CLI took in this process: the seq and the event. Filled by `echoed` only:
+    /// a line written and not echoed she has not seen.
+    held: BTreeMap<String, (u64, String)>,
+    /// The reply a stale send stopped, until it is told back.
+    revision: Option<Revision>,
+    /// A revision line, or a flush, was echoed since the last result: that result is her final answer, sent without a basis.
+    echoed_revision: bool,
+    echoed_flush: bool,
+    /// The next line written carries a flush.
+    flush_asked: bool,
+    /// When each untold delivery was first seen by a poll: its stagger runs from there and nothing later moves it.
+    first_seen: BTreeMap<u64, u64>,
     /// Her Self runs last seen holding the slot, each until a read shows it gone and how it closed: whoever started it, her, the
     /// heartbeat or an earlier process. Her own start joins it unread. A failed read leaves this as it stood.
     slot_seen: Vec<PipelineRef>,
@@ -1045,7 +1067,48 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> {
     repo_read_due_ms: u64,
 }
 
-impl<C: ControlSource, S: ControlSource> PersonaInbox<'_, C, S> {
+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<ChatDelivery>, started_ms: u64, plan: String) -> Self {
+        let rooms = reply_rooms(config, &first);
+        let conversation = first.first().map_or(Conversation::Public, Conversation::of);
+        Self {
+            organs,
+            control,
+            swarm,
+            config,
+            run,
+            progress,
+            rooms,
+            conversation,
+            plan,
+            started_ms,
+            told: BTreeSet::new(),
+            start: first,
+            waiting: Vec::new(),
+            told_through: 0,
+            answered_through: 0,
+            written: None,
+            echoed: Vec::new(),
+            echoed_refusals: 0,
+            seen: Shown::default(),
+            retell: None,
+            held: BTreeMap::new(),
+            revision: None,
+            echoed_revision: false,
+            echoed_flush: false,
+            flush_asked: false,
+            first_seen: BTreeMap::new(),
+            slot_seen: Vec::new(),
+            slot_read_due_ms: 0,
+            notices: Vec::new(),
+            repos: Vec::new(),
+            repo_head: None,
+            repo_read_due_ms: 0,
+        }
+    }
+
     /// The `earlier` lines a batch brings now.
     fn earlier<'d>(&self, deliveries: &'d [ChatDelivery]) -> Vec<Vec<&'d ContextLine>> {
         earlier_shown(deliveries, &self.told, &self.config.own_sender, self.started_ms)
@@ -1088,7 +1151,7 @@ impl<C: ControlSource, S: ControlSource> PersonaInbox<'_, C, S> {
             self.told_through = self.told_through.max(through);
         }
         self.told.extend(told);
-        self.written = Some(Written { batch, shown, refusals });
+        self.written = Some(Written { batch, shown, refusals, revision: false });
         input
     }
 
@@ -1194,20 +1257,53 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
         self.note_repo_changes();
         let mut pending = self.organs.chat.pending(self.progress.acked)?;
         pending.sort_by_key(|delivery| delivery.seq);
-        let Some(oldest) = pending.first() else { return Ok(Polled { pending: self.retell.is_some() || !self.notices.is_empty(), yield_now: false }) };
+        if let Some(revision) = &mut self.revision {
+            revision.polls += 1;
+            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(), yield_now: false }) };
         if Conversation::of(oldest) != self.conversation {
             return Ok(Polled { pending: false, yield_now: true });
         }
-        let untold = pending.into_iter().filter(|delivery| delivery.seq > self.told_through).take_while(|delivery| Conversation::of(delivery) == self.conversation).collect();
-        self.waiting = one_line_of(untold, |batch| message_lines(self.config, batch, &self.earlier(batch)).len());
-        Ok(Polled { pending: !self.waiting.is_empty() || self.retell.is_some() || !self.notices.is_empty(), yield_now: false })
+        let untold: Vec<ChatDelivery> = pending.into_iter().filter(|delivery| delivery.seq > self.told_through).take_while(|delivery| Conversation::of(delivery) == self.conversation).collect();
+        // A delivery's stagger runs from the poll that first saw it, so a stream of newer messages moves nobody's clock. The line
+        // is told once any delivery in it is due, whole: she is never shown a message and kept from the ones before it.
+        let now = unix_ms(self.organs.clock.now());
+        self.first_seen.retain(|seq, _| *seq > self.told_through);
+        for delivery in &untold {
+            self.first_seen.entry(delivery.seq).or_insert(now);
+        }
+        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(), yield_now: false })
     }
 
     fn tell(&mut self) -> Result<String> {
         let notices = std::mem::take(&mut self.notices).concat();
-        if let Some(Retell { batch, why, refusals }) = self.retell.take() {
+        if self.revision_ready()
+            && let Some(revision) = self.revision.take()
+        {
+            // Her refused words are not repeated: she has them. The messages that interrupted her are the line.
+            let waiting = std::mem::take(&mut self.waiting);
+            let line = self.tell_line(&waiting, Some(&stale_note(&revision)), &notices);
+            let mut batch = revision.batch;
+            batch.extend(waiting);
+            batch.sort_by_key(|delivery| delivery.seq);
+            batch.dedup_by_key(|delivery| delivery.seq);
+            let input = self.write(batch, line, 0);
+            if let Some(written) = &mut self.written {
+                written.revision = true;
+            }
+            return Ok(input);
+        }
+        if let Some(Retell { batch, why, refusals, revision }) = self.retell.take() {
             let line = self.tell_line(&batch, Some(&refused_note(why)), &notices);
-            return Ok(self.write(batch, line, refusals));
+            let input = self.write(batch, line, refusals);
+            if let Some(written) = &mut self.written {
+                written.revision = revision;
+            }
+            return Ok(input);
         }
         let batch = std::mem::take(&mut self.waiting);
         let line = self.tell_line(&batch, None, &notices);
@@ -1216,16 +1312,24 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
 
     fn flush_text(&mut self, why: Why) -> String {
         if self.written.is_none() {
-            self.written = Some(Written { batch: Vec::new(), shown: Shown::default(), refusals: 0 });
+            self.written = Some(Written { batch: Vec::new(), shown: Shown::default(), refusals: 0, revision: false });
         }
+        self.flush_asked = true;
         format!("{TURN_REMINDER}\n\n{}", flush_ask(why))
     }
 
     fn echoed(&mut self) {
-        if let Some(Written { batch, shown, refusals }) = self.written.take() {
+        if let Some(Written { batch, shown, refusals, revision }) = self.written.take() {
             self.echoed_refusals = self.echoed_refusals.max(refusals);
+            for delivery in &batch {
+                if self.held.get(&delivery.room_alias).is_none_or(|(seq, _)| *seq < delivery.seq) {
+                    self.held.insert(delivery.room_alias.clone(), (delivery.seq, delivery.event_id.clone()));
+                }
+            }
             self.echoed.extend(batch);
             self.seen.add(shown);
+            self.echoed_revision |= revision;
+            self.echoed_flush |= std::mem::take(&mut self.flush_asked);
         }
     }
 
@@ -1236,6 +1340,8 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
     fn result(&mut self, line: &str, ok: bool, blocked: bool) -> Result<(bool, bool)> {
         let batch = std::mem::take(&mut self.echoed);
         let refused_before = std::mem::take(&mut self.echoed_refusals);
+        let revised = std::mem::take(&mut self.echoed_revision);
+        let final_reply = revised | std::mem::take(&mut self.echoed_flush);
         if blocked || !ok {
             self.progress.failure.get_or_insert(Failure::Session);
             return Ok((false, true));
@@ -1243,7 +1349,7 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
         let turn = self.progress.turn;
         self.progress.turn += 1;
         let step = match structured_output(line) {
-            Some(text) => act(self, &batch, turn, &text)?,
+            Some(text) => act(self, &batch, turn, &text, Finality { final_reply, revised })?,
             None => {
                 self.record_refusal(&batch, turn)?;
                 Step::Refused(Failure::Undecodable)
@@ -1258,11 +1364,20 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
                 // failed (a batch's third refusal leaves no retell to hold the ack back). Deliveries are
                 // 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));
+                // Her final answer lands after she took in every line, so a revision still waiting is answered by it and its note, that a send was not sent, would be false.
+                let spent = if final_reply { self.revision.take().map(|revision| revision.batch) } else { None };
+                self.answered_through = self.answered_through.max(batch.iter().chain(spent.iter().flatten()).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 unanswered = self
+                    .retell
+                    .iter()
+                    .flat_map(|retell| &retell.batch)
+                    .chain(self.written.iter().flat_map(|written| &written.batch))
+                    .chain(self.revision.iter().flat_map(|revision| &revision.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 {
                     if self.organs.chat.ack(self.answered_through).is_err() {
@@ -1282,6 +1397,30 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
                 self.progress.answered += 1;
                 Ok((true, false))
             }
+            Step::Stale { room, sent, newer } => {
+                // Not a refused result: no count, no ack, nothing for the retell. The reply answered nothing; its deliveries wait for the revision's answer.
+                let mut revision = self.revision.take().unwrap_or_default();
+                revision.batch.extend(batch);
+                revision.batch.sort_by_key(|delivery| delivery.seq);
+                revision.batch.dedup_by_key(|delivery| delivery.seq);
+                if !revision.rooms.contains(&room) {
+                    revision.rooms.push(room);
+                }
+                for sent in sent {
+                    if !revision.sent.contains(&sent) {
+                        revision.sent.push(sent);
+                    }
+                }
+                for id in newer {
+                    if !revision.awaiting.contains(&id) {
+                        revision.awaiting.push(id);
+                    }
+                }
+                revision.polls = 0;
+                revision.ready = false;
+                self.revision = Some(revision);
+                Ok((true, false))
+            }
             Step::Refused(why) => {
                 let refusals = refused_before + 1;
                 if refusals >= REFUSALS_PER_BATCH {
@@ -1290,8 +1429,10 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
                 }
                 let mut merged = batch;
                 let mut most = refusals;
+                let mut revision = revised;
                 if let Some(waiting) = self.retell.take() {
                     most = most.max(waiting.refusals);
+                    revision |= waiting.revision;
                     merged.extend(waiting.batch);
                 }
                 merged.sort_by_key(|delivery| delivery.seq);
@@ -1301,7 +1442,7 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
                 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)) });
-                self.retell = Some(Retell { batch: merged, why, refusals: most });
+                self.retell = Some(Retell { batch: merged, why, refusals: most, revision });
                 Ok((true, false))
             }
             Step::Failed(why, intent) => {
@@ -1313,6 +1454,14 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
     }
 }
 
+/// What a stale send's revision line opens with. Her refused words are not in it: she has them. Everything else in the reply was
+/// dropped with it.
+fn stale_note(revision: &Revision) -> String {
+    let rooms = revision.rooms.join(", ");
+    let sent = if revision.sent.is_empty() { String::new() } else { format!(" Already sent: {}.", revision.sent.join(", ")) };
+    format!("Your message to {rooms} was not sent: newer messages reached that room first, below. Send it, change it, or drop it; your answer goes out as you give it. If someone has already said it, drop it; saying nothing costs nothing.{sent} Nothing else in that reply was kept: repeat any remember, settle or start_self_run you still want.")
+}
+
 /// 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.")
@@ -1325,16 +1474,30 @@ fn structured_output(line: &str) -> Option<String> {
 }
 
 impl<C: ControlSource, S: ControlSource> PersonaInbox<'_, C, S> {
+    /// A revision is told once its interrupters have reached the organ and no revision line is in flight: her answer to that line is final and may have said all, so a second note would be told after it landed.
+    fn revision_ready(&self) -> bool {
+        self.revision.as_ref().is_some_and(|revision| revision.ready) && !self.echoed_revision && !self.written.as_ref().is_some_and(|written| written.revision)
+    }
+
     /// The trace of a result that gave nothing to decode.
     fn record_refusal(&self, batch: &[ChatDelivery], turn: u32) -> Result<()> {
         self.organs.trace.record(&TurnTrace { run: label_of(self.run).into(), turn, at: self.organs.clock.now(), inputs: trace_inputs(self.config, batch), outputs: vec![], outcome: format!("failed: {}", Failure::Undecodable) })
     }
 }
 
+/// Whether a reply is her final answer, which goes out with no basis, and whether it answers a revision line, which gives its
+/// sends their own request ids: the mouth's transaction id is idempotent, so a revised say that reused an id of the same batch
+/// would return the old event and lose her new words.
+#[derive(Clone, Copy)]
+struct Finality {
+    final_reply: bool,
+    revised: bool,
+}
+
 /// What one result does in the world: decode, say, remember, settle, start a Self run, and trace.
 /// It acks nothing: the caller acks after it, once, from what is still unanswered. `batch` is what
 /// the result answers.
-fn act<C: ControlSource, S: ControlSource>(inbox: &PersonaInbox<'_, C, S>, batch: &[ChatDelivery], turn: u32, text: &str) -> Result<Step> {
+fn act<C: ControlSource, S: ControlSource>(inbox: &PersonaInbox<'_, C, S>, batch: &[ChatDelivery], turn: u32, text: &str, finality: Finality) -> Result<Step> {
     let (organs, config, run) = (inbox.organs, inbox.config, inbox.run);
     let instance = organs.mind.instance().clone();
     let (control, swarm) = (inbox.control, inbox.swarm);
@@ -1384,11 +1547,22 @@ fn act<C: ControlSource, S: ControlSource>(inbox: &PersonaInbox<'_, C, S>, batch
             return Ok(Step::Failed(Failure::Held(reason), reply.intent));
         }
         let text = say.text.clone();
-        let sent = ChatSendRequest { request_id: format!("{batch_id}-{index}"), persona: instance.0.clone(), room_alias: say.room.clone(), text: text.clone() };
-        let delivered = organs.chat.send(&sent).is_ok_and(|receipt| receipt.refusal.is_none() && receipt.event_id.is_some());
-        if !delivered {
-            organs.trace.record(&record(outputs, "failed: send".into()))?;
-            return Ok(Step::Failed(Failure::Send, reply.intent));
+        let request_id = if finality.revised { format!("{batch_id}-r{turn}-{index}") } else { format!("{batch_id}-{index}") };
+        let basis = (!finality.final_reply).then(|| SendBasis { seen_event_id: inbox.held.get(&say.room).map(|(_, event)| event.clone()), since_unix_ms: inbox.started_ms });
+        let sent = ChatSendRequest { request_id, persona: instance.0.clone(), room_alias: say.room.clone(), text: text.clone(), basis };
+        match organs.chat.send(&sent) {
+            Ok(receipt) if receipt.refusal.is_none() && receipt.event_id.is_some() => {}
+            Ok(receipt) if receipt.refusal == Some(Refusal::RoomMoved) && !finality.final_reply => {
+                for unsent in &reply.say[index..] {
+                    outputs.push(TraceItem { provenance: "her own".into(), room: unsent.room.clone(), text: format!("not sent: {}", unsent.text) });
+                }
+                organs.trace.record(&record(outputs, format!("stale: {}", say.room)))?;
+                return Ok(Step::Stale { room: say.room.clone(), sent: reply.say[..index].iter().map(|sent| sent.room.clone()).collect(), newer: receipt.newer });
+            }
+            _ => {
+                organs.trace.record(&record(outputs, "failed: send".into()))?;
+                return Ok(Step::Failed(Failure::Send, reply.intent));
+            }
         }
         outputs.push(TraceItem { provenance: "her own".into(), room: say.room.clone(), text });
     }
@@ -1423,6 +1597,38 @@ fn act<C: ControlSource, S: ControlSource>(inbox: &PersonaInbox<'_, C, S>, batch
     Ok(Step::Answered(started, reply.intent, opened))
 }
 
+/// How long after a poll first sees `delivery` a live session is told it: none in a direct message, otherwise her rank times the
+/// step. The Minds the message names rank first, by a hash of the Mind and the event, and the rest follow them, ranked the same
+/// way: the first named Mind hears at once, the others one step apart, so tagging several does not make a chorus. A rank is the
+/// number of Minds in her set whose hash sorts before hers, from the message and the configured Minds alone (the sender, who will
+/// not be told its own message, is not counted), so every session computes the same order with no word between them and nobody
+/// is always last.
+fn stagger_ms(config: &Config, delivery: &ChatDelivery) -> u64 {
+    if delivery.is_direct {
+        return 0;
+    }
+    let mut minds: Vec<&str> = config.minds.iter().map(String::as_str).chain([config.own_sender.as_str()]).filter(|mind| *mind != delivery.sender).collect();
+    minds.sort_unstable();
+    minds.dedup();
+    let named = |mind: &str| delivery.mentions.iter().any(|mention| mention == mind);
+    let (named_minds, others): (Vec<&str>, Vec<&str>) = minds.into_iter().partition(|mind| named(mind));
+    let (set, before) = if named(&config.own_sender) { (&named_minds, 0) } else { (&others, named_minds.len()) };
+    let mine = (stagger_hash(&config.own_sender, &delivery.event_id), config.own_sender.as_str());
+    let rank = before + set.iter().filter(|mind| (stagger_hash(mind, &delivery.event_id), **mind) < mine).count();
+    rank as u64 * STAGGER_STEP_S * 1000
+}
+
+/// FNV-1a over the Mind, a separator and the event, then murmur3's finalizer so a Mind's order against another changes from event to
+/// event (FNV alone leaves it nearly fixed by the Minds' names): fixed by this definition, so every Mind's process computes the same order.
+fn stagger_hash(mind: &str, event_id: &str) -> u64 {
+    let mut hash = mind.bytes().chain(std::iter::once(0xff)).chain(event_id.bytes()).fold(0xcbf2_9ce4_8422_2325, |hash, byte| (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3));
+    hash ^= hash >> 33;
+    hash = hash.wrapping_mul(0xff51_afd7_ed55_8ccd);
+    hash ^= hash >> 33;
+    hash = hash.wrapping_mul(0xc4ce_b9fe_1a85_ec53);
+    hash ^ (hash >> 33)
+}
+
 /// The oldest deliveries that share a conversation: the public rooms together, or one direct
 /// message room alone. A prefix by seq, so an ack through its last never passes an unanswered one,
 /// and one member's DM text never enters another room's turn.

crates/mind-body/src/persona/tests.rs

@@ -8,7 +8,7 @@
 //! the next canned reply, and it ends when the canned replies are spent.
 
 use std::cell::{Cell, RefCell};
-use std::collections::VecDeque;
+use std::collections::{BTreeMap, VecDeque};
 use std::path::PathBuf;
 use std::rc::Rc;
 use std::time::Duration as StdDuration;
@@ -20,7 +20,7 @@ use serde_json::json;
 use tempfile::TempDir;
 
 use super::*;
-use crate::chat_client::{ChatSendReceipt, ContextLine};
+use crate::chat_client::{ChatSendReceipt, ContextLine, Refusal};
 use crate::control::{ControlWriter, FileSource, SwarmWriter};
 use crate::lease_client::{LeaseRefusalKind, TokenFd};
 use crate::self_run::{Child, Read, Stop};
@@ -43,6 +43,15 @@ impl Clock for Ticking {
     }
 }
 
+/// A message in a room of the fake homeserver, as the mouth reads the head of a room.
+#[derive(Clone)]
+struct RoomEvent {
+    room: String,
+    event_id: String,
+    sender: String,
+    at_unix_ms: u64,
+}
+
 struct FakeChat {
     log: Log,
     deliveries: Deliveries,
@@ -54,6 +63,43 @@ struct FakeChat {
     receipt: Cell<u8>,
     /// Runs after each delivered send, where the world changes between turns.
     on_send: RefCell<Option<Box<dyn Fn()>>>,
+    /// The rooms of the fake homeserver, oldest first. A send with a basis is answered as the mouth answers it: refused
+    /// `RoomMoved` with the ids of the other senders' messages after what the basis names, or sent.
+    room_log: RefCell<Vec<RoomEvent>>,
+    /// What landed, by request id, as Matrix keys a transaction: a repeated id returns the old event and writes nothing.
+    landed: RefCell<BTreeMap<String, (String, String)>>,
+    /// Who the fake mouth's sends are from; her own lines never move a room for her.
+    speaker: RefCell<String>,
+    /// Acks are recorded and the deliveries stay, where several sessions share one pending.
+    keep_on_ack: Cell<bool>,
+}
+
+impl FakeChat {
+    /// The mouth's `newerThan`: from the newest event back, stopping at the one the basis saw; where it saw none, at its floor;
+    /// where it saw one the room does not hold, the whole room counts. Her own lines never move a room.
+    fn newer_than(&self, room: &str, basis: &SendBasis) -> Vec<String> {
+        let log = self.room_log.borrow();
+        let mut newer = Vec::new();
+        for event in log.iter().rev().filter(|event| event.room == room) {
+            let reached = match &basis.seen_event_id {
+                Some(seen) => &event.event_id == seen,
+                None => event.at_unix_ms <= basis.since_unix_ms,
+            };
+            if reached {
+                break;
+            }
+            if event.sender != *self.speaker.borrow() {
+                newer.push(event.event_id.clone());
+            }
+        }
+        newer.reverse();
+        newer
+    }
+
+    /// Another sender's message reaches a room.
+    fn room_hears(&self, room: &str, event_id: &str, at_unix_ms: u64) {
+        self.room_log.borrow_mut().push(RoomEvent { room: room.into(), event_id: event_id.into(), sender: "@other:chat".into(), at_unix_ms });
+    }
 }
 
 impl Chat for FakeChat {
@@ -71,18 +117,34 @@ impl Chat for FakeChat {
         if let Some(on_send) = &*self.on_send.borrow() {
             on_send();
         }
+        let newer = match (&request.basis, self.receipt.get()) {
+            (Some(basis), 0) => self.newer_than(&request.room_alias, basis),
+            _ => Vec::new(),
+        };
+        if !newer.is_empty() {
+            return Ok(ChatSendReceipt { request_id: request.request_id.clone(), event_id: None, refusal: Some(Refusal::RoomMoved), newer });
+        }
         let (event_id, refusal) = match self.receipt.get() {
-            0 => (Some("$e".to_string()), None),
-            1 => (None, Some("Brake".to_string())),
+            0 => {
+                let mut landed = self.landed.borrow_mut();
+                let (event_id, _) = landed.entry(request.request_id.clone()).or_insert_with(|| {
+                    self.room_log.borrow_mut().push(RoomEvent { room: request.room_alias.clone(), event_id: format!("$sent-{}", request.request_id), sender: self.speaker.borrow().clone(), at_unix_ms: 0 });
+                    (format!("$sent-{}", request.request_id), request.text.clone())
+                });
+                (Some(event_id.clone()), None)
+            }
+            1 => (None, Some(Refusal::Brake)),
             _ => (None, None),
         };
-        Ok(ChatSendReceipt { request_id: request.request_id.clone(), event_id, refusal })
+        Ok(ChatSendReceipt { request_id: request.request_id.clone(), event_id, refusal, newer: Vec::new() })
     }
 
     fn ack(&self, through_seq: u64) -> Result<()> {
         self.log.borrow_mut().push("ack".into());
         self.acked.borrow_mut().push(through_seq);
-        self.deliveries.borrow_mut().retain(|delivery| delivery.seq > through_seq);
+        if !self.keep_on_ack.get() {
+            self.deliveries.borrow_mut().retain(|delivery| delivery.seq > through_seq);
+        }
         Ok(())
     }
 }
@@ -278,12 +340,20 @@ fn config() -> Config {
     }
 }
 
+/// A delivery whose event the stagger tells her at once, whatever the clock does: the first of `$e<seq>`, `$e<seq>-1`... where she
+/// ranks first among the Minds. The tests that are about the stagger name their events.
 fn delivery(seq: u64, room: &str, sender: &str, text: &str) -> ChatDelivery {
+    let ranked_first = |event_id: &str| stagger_ms(&config(), &ChatDelivery { event_id: event_id.into(), ..delivery_of(seq, room, sender, text, event_id) }) == 0;
+    let event_id = (0..).map(|n| if n == 0 { format!("$e{seq}") } else { format!("$e{seq}-{n}") }).find(|id| ranked_first(id)).unwrap();
+    delivery_of(seq, room, sender, text, &event_id)
+}
+
+fn delivery_of(seq: u64, room: &str, sender: &str, text: &str, event_id: &str) -> ChatDelivery {
     ChatDelivery {
         seq,
         room_alias: room.into(),
         room_id: format!("!{room}:chat"),
-        event_id: format!("$e{seq}"),
+        event_id: event_id.into(),
         sender: sender.into(),
         sender_display: sender.trim_start_matches('@').split(':').next().unwrap_or_default().to_string(),
         text: text.into(),
@@ -319,6 +389,10 @@ impl Rig {
                 fail_send_from: Cell::new(None),
                 receipt: Cell::new(0),
                 on_send: RefCell::new(None),
+                room_log: RefCell::default(),
+                landed: RefCell::default(),
+                speaker: RefCell::new("@eureka:chat".into()),
+                keep_on_ack: Cell::new(false),
             },
             model: FakeSession {
                 log: log.clone(),
@@ -1773,7 +1847,7 @@ fn where_and_from_whom_she_learned_a_memory_comes_from_the_batch_and_no_model_fi
     assert_eq!(failure_of(rig.go()), None);
     let (_, memory) = memories(&rig).remove(0);
     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("$e1".into()), Short("$e2".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))]);
     assert_eq!(failure_of(dm.go_with(&config())), None);
@@ -3511,7 +3585,7 @@ Near what was said:
 - (2026-10-09, in #swarm, Did): "Closed the old bridge"
 
 
-Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z (in 20d 7h), read 0s ago. The swarm aims to have used about 90% of the weekly allowance at reset; nothing enforces it; you and the other Minds set the tempo.
+Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z, read at 2026-10-29T20:33Z. The swarm aims to have used about 90% of the weekly allowance at reset; nothing enforces it; you and the other Minds set the tempo.
 
 Tempo: not yet read. The swarm beat is the lower median of the votes cast: a Mind with no standing vote abstains and is not counted, and with no vote cast the beat is 900 s. A vote only needs changing when conditions change; leave it standing otherwise. You vote with tempo_vote_beat_s (60 to 86400 seconds per beat; omit the key to leave your standing vote as it is, and send null to withdraw it and abstain) and declare your own urgency with urgency_permille (0 to 1000, null for none); an urgency lasts until your next turn, so use it as etiquette, not as a habit.
 
@@ -3538,7 +3612,7 @@ What you remember (from your own mind; `huginn view <id>` reads a record):
 - nothing is remembered yet.
 
 
-Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z (in 20d 7h), read 0s ago. The swarm aims to have used about 90% of the weekly allowance at reset; nothing enforces it; you and the other Minds set the tempo.
+Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z, read at 2026-10-29T20:33Z. The swarm aims to have used about 90% of the weekly allowance at reset; nothing enforces it; you and the other Minds set the tempo.
 
 Tempo: not yet read. The swarm beat is the lower median of the votes cast: a Mind with no standing vote abstains and is not counted, and with no vote cast the beat is 900 s. A vote only needs changing when conditions change; leave it standing otherwise. You vote with tempo_vote_beat_s (60 to 86400 seconds per beat; omit the key to leave your standing vote as it is, and send null to withdraw it and abstain) and declare your own urgency with urgency_permille (0 to 1000, null for none); an urgency lasts until your next turn, so use it as etiquette, not as a habit.
 
@@ -5576,3 +5650,466 @@ fn an_undated_record_makes_its_read_missing() {
         assert_eq!(facts.len(), 2, "{spoiled}: the missing line and the other read's facts: {facts:?}");
     }
 }
+
+// The send gate: a stale send is her next input, and a message that does not address her waits her rank.
+
+impl Rig {
+    fn organs<'a>(&'a self, sandbox: &'a Path, scratch: &'a Path) -> Organs<'a> {
+        Organs {
+            mind: &self.mind,
+            launcher: &self.launcher,
+            clock: &self.clock,
+            host: "yggdrasil-host",
+            chat: &self.chat,
+            persona: &self.persona,
+            trace: &self.trace,
+            tempo: None,
+            lender: &self.lender,
+            spawner: &self.model,
+            run_cap_s: self.cap.get(),
+            sandbox,
+            scratch,
+        }
+    }
+
+    fn the_run(&self) -> PipelineRef {
+        if holders(&self.mind, self.mind.instance(), RunTurn::PersonaTurn).unwrap().is_empty() {
+            self.open_the_run();
+        }
+        holders(&self.mind, self.mind.instance(), RunTurn::PersonaTurn).unwrap().remove(0).id
+    }
+}
+
+/// One process's inbox over `first`, driven by hand: the order of a line's echo and a result is the test's.
+fn with_inbox<R>(rig: &Rig, config: &Config, first: Vec<ChatDelivery>, f: impl FnOnce(&mut PersonaInbox<'_, FileSource, FileSource>) -> R) -> R {
+    let (sandbox, scratch) = (rig.sandbox(), rig.scratch());
+    let organs = rig.organs(&sandbox, &scratch);
+    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());
+    f(&mut inbox)
+}
+
+fn reply_result(reply: &str) -> String {
+    json!({"type": "result", "subtype": "success", "is_error": false, "structured_output": serde_json::from_str::<serde_json::Value>(reply).unwrap()}).to_string()
+}
+
+fn hear(rig: &Rig, delivery: &ChatDelivery) {
+    rig.chat.room_hears(&delivery.room_alias, &delivery.event_id, delivery.at_unix_ms);
+}
+
+/// A message reaches the room and, a moment later, the organ's pending.
+fn arrives(rig: &Rig, delivery: &ChatDelivery) {
+    hear(rig, delivery);
+    rig.chat.deliveries.borrow_mut().push(delivery.clone());
+}
+
+fn trace_outcomes(rig: &Rig) -> Vec<String> {
+    rig.trace.0.borrow().iter().map(|trace| trace.outcome.clone()).collect()
+}
+
+#[test]
+fn a_stale_refusal_is_her_next_input_and_costs_no_refusal() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (ask, answer) = (delivery(1, "swarm", "@op:chat", "who owns the stairs?"), delivery(2, "swarm", "@libby:chat", "the stairs are mine"));
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &answer);
+        assert_eq!(inbox.result(&reply_result(&says(&[("swarm", "I own the stairs")])), true, false).unwrap(), (true, false));
+
+        let basis = rig.chat.sent.borrow()[0].basis.clone();
+        assert_eq!(basis, Some(SendBasis { seen_event_id: Some(ask.event_id.clone()), since_unix_ms: inbox.started_ms }));
+        assert!(inbox.progress.refused.is_none() && inbox.progress.failure.is_none() && inbox.retell.is_none(), "a stale send is not a refused result");
+        assert_eq!((inbox.progress.answered, inbox.progress.turn), (0, 1));
+        assert!(rig.chat.acked.borrow().is_empty(), "nothing is acked until the revision answers");
+        assert_eq!(trace_outcomes(&rig), ["stale: swarm"]);
+
+        assert!(inbox.poll().unwrap().pending, "the interrupting message has reached the organ");
+        let expected = inbox.tell_line(std::slice::from_ref(&answer), None, "").input;
+        let line = inbox.tell().unwrap();
+        assert!(line.starts_with("Your message to swarm was not sent: newer messages reached that room first, below. Send it, change it, or drop it; your answer goes out as you give it."), "{line}");
+        assert!(line.contains("Nothing else in that reply was kept: repeat any remember, settle or start_self_run you still want."));
+        assert!(!line.contains("Already sent"), "{line}");
+        assert!(!line.contains("I own the stairs"), "her refused words are not repeated: {line}");
+        assert!(line.ends_with(&expected), "the interrupting message is told as any message is, its envelope bare: {line}");
+
+        inbox.echoed();
+        arrives(&rig, &delivery(3, "swarm", "@aqua:chat", "and I the railing"));
+        inbox.result(&reply_result(&says(&[("swarm", "then the cellar is mine")])), true, false).unwrap();
+        let sent = rig.chat.sent.borrow().clone();
+        assert_eq!(sent[1].basis, None, "one revision, then her choice stands, though the room moved again");
+        assert!(sent[1].request_id.contains("-r1-"), "{}", sent[1].request_id);
+        assert!(rig.chat.landed.borrow().values().any(|(_, text)| text == "then the cellar is mine"));
+        assert_eq!((inbox.progress.answered, inbox.progress.failure), (1, None));
+        assert_eq!(rig.chat.acked.borrow().as_slice(), &[2], "the batch is acked once the revision is answered");
+    });
+}
+
+#[test]
+fn a_second_revision_waits_for_the_first_and_her_final_answer_spends_it() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (ask, second, newest) = (delivery(1, "swarm", "@op:chat", "who owns the stairs?"), delivery(2, "swarm", "@op:chat", "and the cellar?"), delivery(3, "swarm", "@aqua:chat", "the stairs are mine"));
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &second);
+        assert!(inbox.poll().unwrap().pending);
+        inbox.tell().unwrap();
+        inbox.echoed();
+        arrives(&rig, &newest);
+        inbox.result(&reply_result(&says(&[("swarm", "the stairs are mine")])), true, false).unwrap();
+        assert!(inbox.poll().unwrap().pending);
+        inbox.tell().unwrap();
+        assert!(inbox.written.as_ref().is_some_and(|written| written.revision), "a revision line is in flight");
+        inbox.result(&reply_result(&says(&[("swarm", "the cellar too")])), true, false).unwrap();
+        inbox.poll().unwrap();
+        assert!(inbox.revision.as_ref().is_some_and(|revision| revision.ready), "its interrupters have reached the organ");
+        assert!(!inbox.revision_ready(), "a second note is not told while the first revision line is in flight");
+        inbox.echoed();
+        assert!(!inbox.revision_ready(), "nor once it is echoed and unanswered");
+        inbox.result(&reply_result(&says(&[("swarm", "then the stairs, final")])), true, false).unwrap();
+        assert!(inbox.revision.is_none(), "her final answer lands after she took in every line: no note that a send was not sent follows it");
+        assert!(rig.chat.acked.borrow().last().is_some_and(|last| *last >= 2), "the revision batch is acked with her final answer");
+    });
+}
+
+#[test]
+fn a_stale_note_tells_her_to_drop_what_someone_has_already_said() {
+    let note = stale_note(&Revision { rooms: vec!["swarm".into()], ..Revision::default() });
+    assert!(note.contains("Send it, change it, or drop it; your answer goes out as you give it. If someone has already said it, drop it; saying nothing costs nothing."), "{note}");
+}
+
+#[test]
+fn eight_stale_refusals_in_a_row_in_different_batches_never_end_the_session() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    with_inbox(&rig, &config(), vec![], |inbox| {
+        for round in 0..8u64 {
+            arrives(&rig, &delivery(round * 10 + 1, "swarm", "@op:chat", &format!("question {round}")));
+            assert!(inbox.poll().unwrap().pending);
+            inbox.tell().unwrap();
+            inbox.echoed();
+            arrives(&rig, &delivery(round * 10 + 2, "swarm", "@libby:chat", &format!("answer {round}")));
+            inbox.result(&reply_result(&says(&[("swarm", "mine")])), true, false).unwrap();
+            assert!(inbox.poll().unwrap().pending);
+            inbox.tell().unwrap();
+            inbox.echoed();
+            inbox.result(&reply_result(&says(&[("swarm", "revised")])), true, false).unwrap();
+            assert_eq!((inbox.progress.failure, inbox.progress.refused.is_none()), (None, true), "round {round}");
+        }
+        assert_eq!((inbox.progress.answered, inbox.progress.turn), (8, 16));
+    });
+    assert_eq!(trace_outcomes(&rig).iter().filter(|outcome| outcome.starts_with("stale:")).count(), 8);
+}
+
+#[test]
+fn a_result_answering_a_line_in_flight_is_not_the_revision() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (ask, answer) = (delivery(1, "swarm", "@op:chat", "question"), delivery(2, "swarm", "@libby:chat", "answer"));
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &answer);
+        inbox.result(&reply_result(&says(&[("swarm", "mine")])), true, false).unwrap();
+        assert!(inbox.poll().unwrap().pending);
+        inbox.tell().unwrap();
+        // The revision line is written; a result for the line before it arrives first, its echo not yet.
+        inbox.result(&reply_result(&says(&[("usage", "an aside")])), true, false).unwrap();
+        assert!(rig.chat.sent.borrow()[1].basis.is_some(), "only the echo of the revision line makes a reply final");
+        inbox.echoed();
+        inbox.result(&reply_result(&says(&[("swarm", "revised")])), true, false).unwrap();
+        assert_eq!(rig.chat.sent.borrow()[2].basis, None);
+    });
+}
+
+#[test]
+fn held_comes_from_echoes_not_writes() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (first, second) = (delivery(1, "swarm", "@op:chat", "one"), delivery(3, "swarm", "@op:chat", "two"));
+    arrives(&rig, &first);
+    with_inbox(&rig, &config(), vec![first.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &second);
+        assert!(inbox.poll().unwrap().pending);
+        inbox.tell().unwrap(); // written, never echoed: she has not seen it
+        inbox.result(&reply_result(&says(&[("swarm", "hi")])), true, false).unwrap();
+        let basis = rig.chat.sent.borrow()[0].basis.clone().unwrap();
+        assert_eq!(basis.seen_event_id, Some(first.event_id.clone()), "a line written and not echoed is not in the basis");
+    });
+}
+
+#[test]
+fn a_flush_reply_is_final_even_when_the_room_moved() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let ask = delivery(1, "swarm", "@op:chat", "question");
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &delivery(2, "swarm", "@libby:chat", "answer"));
+        inbox.flush_text(Why::Idle);
+        inbox.echoed();
+        inbox.result(&reply_result(&says(&[("swarm", "last words")])), true, false).unwrap();
+        assert_eq!(rig.chat.sent.borrow()[0].basis, None);
+        assert!(inbox.revision.is_none() && inbox.progress.answered == 1, "the flush has no next turn to revise in");
+    });
+}
+
+#[test]
+fn a_stale_say_leaves_earlier_says_standing_and_a_revision_never_reuses_a_landed_id() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let ask = delivery(1, "swarm", "@op:chat", "question");
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        // The room moved, but the message has not reached the organ's pending: the ear is late.
+        rig.chat.room_hears("swarm", "$ghost", 0);
+        let memory = reply(&[("usage", "an aside"), ("swarm", "refused words")], json!([keeps("Learned", "a thing", &[])]), json!([]), json!(null));
+        inbox.result(&reply_result(&memory), true, false).unwrap();
+        assert_eq!(rig.chat.sent.borrow().len(), 2);
+        assert!(memories(&rig).is_empty(), "nothing after the stale say is done: no memory");
+        assert_eq!(trace_outcomes(&rig), ["stale: swarm"]);
+        let traced = rig.trace.0.borrow()[0].outputs.iter().map(|item| item.text.clone()).collect::<Vec<_>>();
+        assert!(traced.contains(&"not sent: refused words".to_string()), "{traced:?}");
+
+        assert!(!inbox.poll().unwrap().pending, "the awaited message has not arrived");
+        assert!(inbox.poll().unwrap().pending, "a silent ear delays her two polls at most and never strands her");
+        let line = inbox.tell().unwrap();
+        assert!(line.contains("Already sent: usage."), "{line}");
+        inbox.echoed();
+        inbox.result(&reply_result(&says(&[("swarm", "new words")])), true, false).unwrap();
+        let sent = rig.chat.sent.borrow().clone();
+        assert_ne!(sent[2].request_id, sent[1].request_id);
+        assert!(sent.iter().take(2).all(|request| request.request_id != sent[2].request_id));
+        let landed = rig.chat.landed.borrow();
+        assert!(landed.values().any(|(_, text)| text == "new words") && landed.values().any(|(_, text)| text == "an aside"), "{landed:?}");
+    });
+}
+
+#[test]
+fn a_process_that_ends_with_a_revision_waiting_leaves_the_batch_unacked() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (ask, answer) = (delivery(1, "swarm", "@op:chat", "question"), delivery(2, "swarm", "@libby:chat", "answer"));
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &answer);
+        inbox.result(&reply_result(&says(&[("swarm", "mine")])), true, false).unwrap();
+        assert!(inbox.revision.is_some());
+    });
+    assert!(rig.chat.acked.borrow().is_empty());
+    let next = next_batch(&config(), rig.chat.pending(0).unwrap(), 0);
+    assert_eq!(next.len(), 2);
+    with_inbox(&rig, &config(), next, |inbox| {
+        let line = inbox.start_line().unwrap();
+        assert!(line.contains("question") && line.contains("answer") && !line.contains("was not sent"), "{line}");
+    });
+}
+
+#[test]
+fn an_undecodable_revision_stays_final() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (ask, answer) = (delivery(1, "swarm", "@op:chat", "question"), delivery(2, "swarm", "@libby:chat", "answer"));
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &answer);
+        inbox.result(&reply_result(&says(&[("swarm", "mine")])), true, false).unwrap();
+        assert!(inbox.poll().unwrap().pending);
+        inbox.tell().unwrap();
+        inbox.echoed();
+        inbox.result(&json!({"type": "result", "is_error": false, "structured_output": {"nonsense": 1}}).to_string(), true, false).unwrap();
+        assert_eq!(inbox.progress.refused.map(|refused| refused.refusals), Some(1), "an undecodable reply is a refusal, as ever");
+        assert!(inbox.poll().unwrap().pending);
+        inbox.tell().unwrap();
+        inbox.echoed();
+        inbox.result(&reply_result(&says(&[("swarm", "second try")])), true, false).unwrap();
+        assert_eq!(rig.chat.sent.borrow().last().unwrap().basis, None);
+    });
+}
+
+#[test]
+fn eight_sessions_one_operator_question() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    rig.chat.keep_on_ack.set(true);
+    let (sandbox, scratch) = (rig.sandbox(), rig.scratch());
+    let organs = rig.organs(&sandbox, &scratch);
+    let (control, swarm, run) = (FileSource::new(rig.store()), FileSource::new(rig.swarm_store()), rig.the_run());
+    let ids: Vec<String> = (0..8).map(|index| format!("@mind{index}:chat")).collect();
+    let configs: Vec<Config> = ids.iter().map(|id| Config { own_sender: id.clone(), minds: ids.clone(), ..config() }).collect();
+    let mut progresses: Vec<Progress> = (0..8).map(|_| Progress::default()).collect();
+    let question = delivery_of(1, "swarm", "@op:chat", "which of us takes the cellar?", "$question");
+    arrives(&rig, &question);
+    let mut inboxes: Vec<PersonaInbox<'_, FileSource, FileSource>> = progresses
+        .iter_mut()
+        .zip(&configs)
+        .map(|(progress, config)| PersonaInbox::new(&organs, &control, &swarm, config, &run, progress, vec![question.clone()], 0, String::new()))
+        .collect();
+    for inbox in &mut inboxes {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+    }
+    // Every Mind answers the same question at once; the homeserver takes the first and refuses the rest.
+    for (index, inbox) in inboxes.iter_mut().enumerate() {
+        *rig.chat.speaker.borrow_mut() = ids[index].clone();
+        inbox.result(&reply_result(&says(&[("swarm", &format!("mind {index} answers"))])), true, false).unwrap();
+        if index == 0 {
+            let landed = rig.chat.landed.borrow().values().next().unwrap().0.clone();
+            arrives(&rig, &delivery_of(2, "swarm", &ids[0], "mind 0 answers", &landed));
+        }
+    }
+    assert_eq!((rig.chat.sent.borrow().len(), rig.chat.landed.borrow().len()), (8, 1), "one message went out in the first round");
+    // The seven who were refused read it and drop their own.
+    for (index, inbox) in inboxes.iter_mut().enumerate().skip(1) {
+        *rig.chat.speaker.borrow_mut() = ids[index].clone();
+        assert!(inbox.poll().unwrap().pending, "mind {index}");
+        assert!(inbox.tell().unwrap().contains("mind 0 answers"));
+        inbox.echoed();
+        inbox.result(&reply_result(&says(&[])), true, false).unwrap();
+    }
+    assert_eq!(rig.chat.sent.borrow().len(), 8, "no revision was sent: each dropped");
+    assert_eq!(inboxes.iter().map(|inbox| inbox.progress.turn).sum::<u32>(), 15, "eight first-round results and seven revision turns");
+    assert!(inboxes.iter().all(|inbox| inbox.progress.failure.is_none() && inbox.progress.refused.is_none()));
+}
+
+fn three_minds(own: &str) -> Config {
+    Config { own_sender: own.into(), minds: ["@eureka:chat", "@libby:chat", "@aqua:chat"].map(String::from).to_vec(), ..config() }
+}
+
+/// An unaddressed swarm message whose event ranks `eureka` at `rank` of three.
+fn unaddressed_at_rank(seq: u64, rank: u64, text: &str) -> ChatDelivery {
+    (0..).map(|n| delivery_of(seq, "swarm", "@op:chat", text, &format!("$r{seq}-{n}"))).find(|candidate| stagger_ms(&three_minds("@eureka:chat"), candidate) == rank * STAGGER_STEP_S * 1000).unwrap()
+}
+
+#[test]
+fn an_addressed_message_is_told_at_once_and_an_unaddressed_one_waits_her_rank() {
+    let config = three_minds("@eureka:chat");
+    let plain = unaddressed_at_rank(1, 2, "who takes the cellar?");
+    let addressed = ChatDelivery { mentions: vec!["@eureka:chat".into()], ..plain.clone() };
+    let in_dm = ChatDelivery { is_direct: true, ..plain.clone() };
+    assert_eq!(STAGGER_STEP_S, 8, "the step is the calibration the ruling names");
+    for (name, message, wait_ms) in [("addressed", addressed, 0), ("direct", in_dm, 0), ("unaddressed", plain, 16_000)] {
+        let rig = Rig::new(vec![], &[]);
+        rig.clock.step.set(0);
+        let t0 = rig.clock.now.get();
+        arrives(&rig, &message);
+        with_inbox(&rig, &config, vec![message.clone()], |inbox| {
+            if wait_ms > 0 {
+                for waited in [0, 8_000, 15_999] {
+                    rig.clock.now.set(t0 + waited);
+                    assert!(!inbox.poll().unwrap().pending, "{name}: told after {waited} ms of {wait_ms}");
+                }
+            }
+            rig.clock.now.set(t0 + wait_ms);
+            assert!(inbox.poll().unwrap().pending, "{name}");
+            let expected = inbox.tell_line(std::slice::from_ref(&message), None, "").input;
+            assert_eq!(inbox.tell().unwrap(), expected, "{name}: the tell is the message and its envelope, unchanged by the wait");
+        });
+    }
+}
+
+#[test]
+fn the_wait_is_bounded_under_a_stream_of_messages() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let config = three_minds("@eureka:chat");
+    let t0 = rig.clock.now.get();
+    arrives(&rig, &unaddressed_at_rank(1, 2, "the first"));
+    with_inbox(&rig, &config, vec![], |inbox| {
+        let mut told_at = None;
+        for second in 0..=40i64 {
+            rig.clock.now.set(t0 + second * 1000);
+            if second > 0 && second % 3 == 0 {
+                arrives(&rig, &unaddressed_at_rank(u64::try_from(second).unwrap() + 1, 2, &format!("more {second}")));
+            }
+            if inbox.poll().unwrap().pending {
+                told_at = Some(second);
+                break;
+            }
+        }
+        assert_eq!(told_at, Some(16), "the first message's clock is its own: newer messages do not move it");
+        let line = inbox.tell().unwrap();
+        assert!(line.contains("the first") && line.contains("more 15"), "{line}");
+    });
+}
+
+#[test]
+fn the_stagger_ranks_the_minds_by_a_hash_of_mind_and_event_so_nobody_is_always_last() {
+    let minds = ["@eureka:chat", "@libby:chat", "@aqua:chat"];
+    let mut last = BTreeSet::new();
+    let mut first = BTreeSet::new();
+    for n in 0..64 {
+        let event = delivery_of(1, "swarm", "@op:chat", "q", &format!("$event{n}"));
+        let mut ranks: Vec<u64> = minds.iter().map(|mind| stagger_ms(&three_minds(mind), &event) / (STAGGER_STEP_S * 1000)).collect();
+        for (mind, rank) in minds.iter().zip(&ranks) {
+            if *rank == 2 {
+                last.insert(*mind);
+            }
+            if *rank == 0 {
+                first.insert(*mind);
+            }
+        }
+        ranks.sort_unstable();
+        assert_eq!(ranks, [0, 1, 2], "event {n}: one Mind at each rank");
+    }
+    assert_eq!((last.len(), first.len()), (3, 3), "over many events each Mind is first and each is last");
+}
+
+#[test]
+fn naming_several_minds_does_not_make_a_chorus_and_the_unnamed_follow_the_named() {
+    let minds = ["@eureka:chat", "@libby:chat", "@aqua:chat"];
+    let step = STAGGER_STEP_S * 1000;
+    let mut named_first = BTreeSet::new();
+    for n in 0..64 {
+        let tagged_two = ChatDelivery { mentions: vec!["@eureka:chat".into(), "@libby:chat".into()], ..delivery_of(1, "swarm", "@op:chat", "q", &format!("$event{n}")) };
+        let waits: Vec<u64> = minds.iter().map(|mind| stagger_ms(&three_minds(mind), &tagged_two)).collect();
+        assert_ne!(waits[0], waits[1], "event {n}: two named Minds never share a rank");
+        assert_eq!(waits[0].min(waits[1]), 0, "event {n}: the first named Mind hears at once");
+        assert_eq!(waits[0].max(waits[1]), step, "event {n}: the second one step later");
+        assert_eq!(waits[2], 2 * step, "event {n}: the unnamed Mind comes after every named one");
+        named_first.insert(if waits[0] == 0 { 0 } else { 1 });
+        let tagged_one = ChatDelivery { mentions: vec!["@aqua:chat".into()], ..tagged_two };
+        assert_eq!(stagger_ms(&three_minds("@aqua:chat"), &tagged_one), 0, "event {n}: a message naming one Mind reaches her at once");
+        assert!(minds[..2].iter().all(|mind| stagger_ms(&three_minds(mind), &tagged_one) >= step), "event {n}");
+    }
+    assert_eq!(named_first.len(), 2, "which named Mind goes first differs per event");
+}
+
+#[test]
+fn a_flush_written_beside_a_revision_line_still_makes_the_reply_final() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (ask, answer) = (delivery(1, "swarm", "@op:chat", "question"), delivery(2, "swarm", "@libby:chat", "answer"));
+    arrives(&rig, &ask);
+    with_inbox(&rig, &config(), vec![ask.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        arrives(&rig, &answer);
+        inbox.result(&reply_result(&says(&[("swarm", "mine")])), true, false).unwrap();
+        assert!(inbox.poll().unwrap().pending);
+        inbox.tell().unwrap();
+        inbox.flush_text(Why::Idle);
+        inbox.echoed();
+        inbox.result(&reply_result(&says(&[("swarm", "revised")])), true, false).unwrap();
+        assert_eq!(rig.chat.sent.borrow()[1].basis, None);
+        assert!(rig.chat.sent.borrow()[1].request_id.contains("-r1-"));
+    });
+}

crates/mind-body/src/self_run/tests.rs

@@ -357,7 +357,7 @@ fn the_start_message_is_the_first_thing_written_and_names_the_skill_the_instance
     let ran = go(vec![Out(result_line())]);
     assert!(matches!(&ran.events[0], Event::Wrote(_)), "the start precedes the first read: {:?}", ran.events[0]);
     let written = ran.written();
-    for needle in ["Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z (in 20d 7h), read 0s ago", "about 90% of the weekly allowance", "nothing enforces it"] {
+    for needle in ["Plan usage: five_hour 42.3% allowed, resets at 2026-10-29T20:50:00Z, read at 2026-10-29T20:33Z", "about 90% of the weekly allowance", "nothing enforces it"] {
         assert!(written[0].contains(needle), "the lease grant plan windows reach the child: {needle}: {}", written[0]);
     }
     for needle in ["Use the eureka skill.", "yggdrasil", "yggdrasil:run:selfrun-a", AGENT] {
@@ -1376,6 +1376,8 @@ fn a_brief_derived_while_a_turn_runs_is_written_at_once_counted_at_its_echo_and_
     assert_eq!(told.len(), 1, "{:?}", ran.written());
     assert!(told[0].contains(&brief_ref("b1").id.0) && told[0].contains("Survey the noise functions."), "{}", told[0]);
     assert!(told[0].contains("resolve follow-up") && told[0].contains("Recorded"), "she is told how to close it: {}", told[0]);
+    // The plan line is restated under every brief for the run's life, so it names the instant of the reading, never an age that stops being true.
+    assert!(told[0].contains("read at 2026-10-29T20:33Z") && !told[0].contains(" ago"), "{}", told[0]);
     assert_eq!(ran.written().len(), 2, "the start message and one tell: {:?}", ran.written());
     assert!(ran.out.contains("decision Write { tells: true, flush: None } prefix 10 ttl_s 0 idle_s working inbox true written 2 consumed 1\n"), "{}", ran.out);
     assert_eq!(ran.count(&Event::InputClosed), 0);

crates/mind-body/tests/chat_rudp.rs

@@ -19,8 +19,8 @@ use mind_body::chat_rudp::{RudpChat, connection_id};
 const ACK_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack");
 const PENDING_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack");
 const PENDING_REQUEST_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack");
-const SEND_RECEIPT_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack");
-const SEND_REQUEST_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack");
+const SEND_RECEIPT_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack");
+const SEND_REQUEST_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.nil-basis.msgpack");
 
 const KEY: &str = "connection-key-for-test";
 const TAG: &str = "cultnet.transport.rudp.v0";
@@ -62,7 +62,12 @@ fn answer(kind: &str, record_key: &str, payload: &[u8]) -> CultNetMessage {
 
 /// The mouth of the Persona `eureka`, whose connection id derives from `key`.
 fn mouth(key: &str) -> Mouth {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    mouth_at(key, "127.0.0.1:0")
+}
+
+/// The same mouth on a chosen address: a restart binds the address the old one held.
+fn mouth_at(key: &str, bind: &str) -> Mouth {
+    let socket = UdpSocket::bind(bind).unwrap();
     socket.set_read_timeout(Some(Duration::from_millis(20))).unwrap();
     let address = socket.local_addr().unwrap();
     let mut server = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions::server("mouth-test", socket, connection_id(key))).unwrap();
@@ -143,6 +148,32 @@ fn a_request_the_mouth_does_not_answer_ends_in_an_error_not_a_hang() {
     let mouth = mouth(KEY);
     let chat = client(&mouth, KEY, "someone-else", Duration::from_millis(600));
     assert!(chat.pending(0).is_err(), "the mouth answers only its own Persona");
-    let sent = ChatSendRequest { request_id: "r".into(), persona: "someone-else".into(), room_alias: "#a:b".into(), text: "x".into() };
+    let sent = ChatSendRequest { request_id: "r".into(), persona: "someone-else".into(), room_alias: "#a:b".into(), text: "x".into(), basis: None };
     assert!(chat.send(&sent).is_err());
 }
+
+#[test]
+fn a_client_recovers_from_a_restarted_mouth_on_its_next_call() {
+    let first = mouth(KEY);
+    let address = first.address;
+    let chat = client(&first, KEY, "eureka", Duration::from_millis(1500));
+    assert!(chat.pending(0).is_ok());
+    drop(first);
+    let second = mouth_at(KEY, &address.to_string());
+    assert!(chat.pending(0).is_err(), "the restarted mouth ignores a link it never saw: the call times out");
+    assert_eq!(chat.pending(0).unwrap(), chat_client::decode::<ChatPending>(PENDING_BYTES).unwrap().deliveries, "the next call opens a fresh link");
+    assert_eq!(chat.pending(0).unwrap().len(), 1, "and keeps it");
+    let reads = second.seen.lock().unwrap().iter().filter(|(kind, _)| kind == PENDING_REQUEST).count();
+    assert_eq!(reads, 2, "the restarted mouth read the two requests that reached it on a fresh link");
+}
+
+#[test]
+fn a_send_that_timed_out_is_not_repeated_by_the_client() {
+    let mouth = mouth(KEY);
+    let chat = client(&mouth, KEY, "someone-else", Duration::from_millis(600));
+    let sent = ChatSendRequest { request_id: "r-1".into(), persona: "someone-else".into(), room_alias: "#a:b".into(), text: "x".into(), basis: None };
+    assert!(chat.send(&sent).is_err());
+    chat.ack(1).unwrap();
+    let sends: Vec<_> = mouth.seen.lock().unwrap().iter().filter(|(kind, _)| kind == SEND_REQUEST).map(|(_, payload)| chat_client::decode::<ChatSendRequest>(payload).unwrap().request_id).collect();
+    assert_eq!(sends, ["r-1"], "the send reached the mouth once; a repeat is the caller's, under the same request id");
+}

crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack

@@ -1 +0,0 @@
-“¥req-1¥$sentÀ

crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack

@@ -1 +0,0 @@
-”¥req-1¦eureka¶#operator:gamecult.org¢hi

crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.room-moved.msgpack

@@ -0,0 +1 @@
+”¥req-1À©RoomMoved’¢$a¢$b

crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack

@@ -0,0 +1 @@
+”¥req-1¥$sentÀ<74>

crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.nil-basis.msgpack

@@ -0,0 +1 @@
+払req-1ｦeurekaｳ#swarm:gamecult.org｢hiﾀ

crates/mind-body/tests/public_surface.txt

@@ -10,7 +10,7 @@ mind_body: mod persona
 mind_body: mod queue
 mind_body: mod self_run
 mind_body: mod tempo
-mind_body::chat_client: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Refusal { Brake , PersonaMismatch , UnknownRoom , EmptyText , SendFailed , UnknownPersona , }
+mind_body::chat_client: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Refusal { Brake , PersonaMismatch , UnknownRoom , EmptyText , SendFailed , UnknownPersona , RoomMoved , }
 mind_body::chat_client: const ACK : & str
 mind_body::chat_client: const PENDING : & str
 mind_body::chat_client: const PENDING_REQUEST : & str
@@ -29,9 +29,10 @@ mind_body::chat_client: struct ChatDelivery # [derive (Clone , Debug , PartialEq
 mind_body::chat_client: struct ChatPending # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, deliveries: Vec < ChatDelivery >]
 mind_body::chat_client: struct ChatPendingRequest # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, after_seq: u64]
 mind_body::chat_client: struct ChatRefusal # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, refusal: Refusal]
-mind_body::chat_client: struct ChatSendReceipt # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[request_id: String, event_id: Option < String >, refusal: Option < String >]
-mind_body::chat_client: struct ChatSendRequest # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[request_id: String, persona: String, room_alias: String, text: String]
+mind_body::chat_client: struct ChatSendReceipt # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[request_id: String, event_id: Option < String >, refusal: Option < Refusal >, newer: Vec < String >]
+mind_body::chat_client: struct ChatSendRequest # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[request_id: String, persona: String, room_alias: String, text: String, basis: Option < SendBasis >]
 mind_body::chat_client: struct ContextLine # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[sender: String, text: String, at_unix_ms: u64]
+mind_body::chat_client: struct SendBasis # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[seen_event_id: Option < String >, since_unix_ms: u64]
 mind_body::chat_rudp: fn connection_id (key_file_text : & str) -> u32
 mind_body::chat_rudp: impl Chat for RudpChat
 mind_body::chat_rudp: impl Drop for RudpChat
@@ -41,7 +42,7 @@ mind_body::chat_rudp: struct RudpChat  pub-fields[]
 mind_body::cli: fn instance_and_endpoint (args : & [String] , env_endpoint : Option < String > , usage : & str) -> Result < (Slug , SocketAddr) >
 mind_body::cli: fn operator_name (sudo_user : Option < String > , user : Option < String >) -> String
 mind_body::cli: fn run (args : & [String] , root : & Path , set_by : & str , now : DateTime < Utc > , out : & mut impl Write , stop_units : & mut dyn FnMut (StopScope < '_ >) -> Result < () > ,) -> Result < () >
-mind_body::connector: fn render_plan (windows : & [PlanQuota] , now_ms : u64) -> String
+mind_body::connector: fn render_plan (windows : & [PlanQuota] , _now_ms : u64) -> String
 mind_body::control: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum HeldReason { BrakeAbsent , BrakeHeld , Undecodable , }
 mind_body::control: # [derive (Debug , PartialEq , Eq)] pub enum Effective { Held { reason : HeldReason } , Released (Grant) , }
 mind_body::control: # [derive (Debug , PartialEq , Eq)] pub enum SwarmEffective { Held (HeldReason) , Released , }
@@ -106,8 +107,8 @@ mind_body::heartbeat: struct Organs< 'a >  pub-fields[seats: & 'a [Seat < 'a >],
 mind_body::heartbeat: struct PersonaId # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[slug: Slug, public_name: String, handle: String, matrix_id: String, huginn: SocketAddr]
 mind_body::heartbeat: struct PollTrace # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[at: DateTime < Utc >, tempo_s: u32, instances: Vec < InstanceTrace >, launches: Vec < String >]
 mind_body::heartbeat: struct Seat< 'a >  pub-fields[who: PersonaId, mind: & 'a dyn MindPort, chat: & 'a dyn Chat]
-mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Declined { Admission (MindRefusal) , Conflict , AlreadyAdmitted , StartFailed , GrantForOtherInstance , PersonaRunTakesNoClaims , }
 mind_body::launch: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum StopScope < 'a > { Mind (& 'a Slug) , Swarm , }
+mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Declined { Admission (MindRefusal) , Conflict , AlreadyAdmitted , StartFailed , GrantForOtherInstance , PersonaRunTakesNoClaims , }
 mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum LaunchOutcome { Launched { run : PipelineRef , } , NotQueued { item : PipelineRef , } , Tripped { run : PipelineRef , question : PipelineRef , } , Busy { run : PipelineRef , } , Refused (Declined) , }
 mind_body::launch: const START_GRACE_S : i64
 mind_body::launch: fn open_and_launch (ports : & Ports , request : LaunchRequest) -> Result < LaunchOutcome >
@@ -180,6 +181,7 @@ mind_body::persona: impl Trace for Noted < '_ >
 mind_body::persona: impl WearFact
 mind_body::persona: impl fmt :: Display for Failure
 mind_body::persona: impl fmt :: Display for Outcome
+mind_body::persona: impl< 'a , C : ControlSource , S : ControlSource > PersonaInbox < 'a , C , S >
 mind_body::persona: impl< C : ControlSource , S : ControlSource > PersonaInbox < '_ , C , S >
 mind_body::persona: impl< C : ControlSource , S : ControlSource > SessionInbox for PersonaInbox < '_ , C , S >
 mind_body::persona: pub trait PersonaState { fn instructions (& self) -> Result < String > ; }