crates/mind-body/src/chat_client.rs

@@ -22,6 +22,8 @@ pub const PENDING: &str = "gamecult.persona.chat_pending.v2";
 /// The pending answer of a mouth that has no image field yet: its deliveries decode with `image: None`.
 pub const PENDING_V1: &str = "gamecult.persona.chat_pending.v1";
 pub const ACK: &str = "gamecult.persona.chat_ack.v1";
+/// Her read receipt: the newest event of a room that went into her model's input.
+pub const READ: &str = "gamecult.persona.chat_read.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";
@@ -84,6 +86,20 @@ pub struct ChatAck {
     pub through_seq: u64,
 }
 
+/// One room's read report: `[persona, room, event_id, at_unix_ms]`, positional like `ChatAck`.
+///
+/// `room` is the delivery's `room_id` (the Matrix room id), not its alias. The mouth posts `m.read` to the room it delivered
+/// the event for and accepts either spelling, but only the id names one room with no lookup, and a direct room has no alias
+/// (its alias is its id). `event_id` is the newest event of that room among those the model's input held, and `at_unix_ms`
+/// that event's own instant.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ChatRead {
+    pub persona: String,
+    pub room: String,
+    pub event_id: String,
+    pub at_unix_ms: u64,
+}
+
 /// Why the mouth refused a request, as its typed names (Bifrost
 /// `tools/persona-chat-documents.mjs` `refusals`).
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -187,6 +203,9 @@ pub trait Chat {
     /// advances its sync position only now, so a start that dies first has its
     /// deliveries again.
     fn ack(&self, through_seq: u64) -> Result<()>;
+    /// The model's input now holds `read.event_id` and every earlier event of its room. A display-only receipt: it
+    /// acknowledges nothing and the mouth answers nothing.
+    fn read(&self, read: &ChatRead) -> Result<()>;
 }
 
 #[cfg(test)]

crates/mind-body/src/chat_rudp.rs

@@ -29,7 +29,7 @@ use serde::de::DeserializeOwned;
 use sha2::{Digest, Sha256};
 
 use crate::chat_client::{
-    ACK, Chat, ChatAck, ChatDelivery, ChatPending, ChatPendingRequest, ChatRefusal, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_V1, REFUSAL, SEND_RECEIPT,
+    ACK, Chat, ChatAck, ChatDelivery, ChatPending, ChatPendingRequest, ChatRead, ChatRefusal, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_V1, READ, REFUSAL, SEND_RECEIPT,
     SEND_REQUEST, decode, encode,
 };
 
@@ -185,6 +185,10 @@ impl Chat for RudpChat {
         let ack = ChatAck { persona: self.persona.clone(), through_seq };
         self.exchange(ACK, &format!("{}:{through_seq}", self.persona), &ack, None).map(|_| ())
     }
+
+    fn read(&self, read: &ChatRead) -> Result<()> {
+        self.exchange(READ, &format!("{}:{}", self.persona, read.event_id), read, None).map(|_| ())
+    }
 }
 
 impl Drop for RudpChat {

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

@@ -99,6 +99,10 @@ impl Chat for FakeChat {
     fn ack(&self, _: u64) -> Result<()> {
         panic!("the heartbeat never acks a delivery")
     }
+
+    fn read(&self, _: &crate::chat_client::ChatRead) -> Result<()> {
+        panic!("the heartbeat reads nothing")
+    }
 }
 
 #[derive(Default)]

crates/mind-body/src/persona.rs

@@ -84,7 +84,7 @@ use uuid::Uuid;
 
 use crate::address::{Addressee, Name, addresses, localpart, project_names};
 use crate::chat_image::{self, Image, MAX_PER_LINE};
-use crate::chat_client::{Chat, ChatDelivery, ChatSendRequest, ContextLine, Refusal, SendBasis};
+use crate::chat_client::{Chat, ChatDelivery, ChatRead, ChatSendRequest, ContextLine, Refusal, SendBasis};
 use crate::connector::render_plan;
 use crate::control::{ControlSource, Effective, HeldReason, derive};
 use crate::heartbeat::{HeartbeatView, PollTrace};
@@ -1228,6 +1228,9 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> {
     /// 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 seq of the newest event reported read in each room (by room id), so a retold or revised batch never reports an event
+    /// at or behind one already reported. Advanced only by a report the mouth took.
+    read_through: BTreeMap<String, u64>,
     /// 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.
@@ -1295,6 +1298,7 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> {
             seen: Shown::default(),
             retell: None,
             held: BTreeMap::new(),
+            read_through: BTreeMap::new(),
             revision: None,
             echoed_revision: false,
             echoed_flush: false,
@@ -1482,6 +1486,32 @@ impl<'a, C: ControlSource, S: ControlSource> PersonaInbox<'a, C, S> {
         Polled { pending, other: false, other_first: false, yield_now: !pending && !own_untold && self.replies_elsewhere }
     }
 
+    /// Her read receipts for a batch the CLI just took in: one `chat_read` per room, naming only the newest event of that room in
+    /// the batch. Every delivery of an echoed batch is a `message` line of her input, and the `earlier` lines before it are part
+    /// of the input too, so the newest message is the newest event she saw; nothing queued and not yet told is in `batch`, so
+    /// nothing unread is ever reported. Event ids are deliberately NOT added to the `earlier` or older message lines: a receipt
+    /// needs only the latest event the model saw, and an id on every line would grow the prompt for no receipt benefit.
+    /// Display only: a refused or unreachable report is dropped, never ends the session, and the ack stays the delivery
+    /// guarantee. A room already reported at or past a delivery's seq (a retold batch) is not reported again.
+    fn report_read(&mut self, batch: &[ChatDelivery]) {
+        let mut newest: BTreeMap<&str, &ChatDelivery> = BTreeMap::new();
+        for delivery in batch {
+            if newest.get(delivery.room_id.as_str()).is_none_or(|kept| kept.seq < delivery.seq) {
+                newest.insert(&delivery.room_id, delivery);
+            }
+        }
+        let persona = self.organs.mind.instance().0.clone();
+        for (room, delivery) in newest {
+            if self.read_through.get(room).is_some_and(|through| *through >= delivery.seq) {
+                continue;
+            }
+            let read = ChatRead { persona: persona.clone(), room: room.to_owned(), event_id: delivery.event_id.clone(), at_unix_ms: delivery.at_unix_ms };
+            if self.organs.chat.read(&read).is_ok() {
+                self.read_through.insert(room.to_owned(), delivery.seq);
+            }
+        }
+    }
+
     fn held(&self) -> Option<HeldReason> {
         match read(self.control, self.swarm, self.organs.mind.instance()) {
             Effective::Held { reason } => Some(reason),
@@ -1599,6 +1629,7 @@ impl<C: ControlSource, S: ControlSource> SessionInbox for PersonaInbox<'_, C, S>
                     self.held.insert(delivery.room_alias.clone(), (delivery.seq, delivery.event_id.clone()));
                 }
             }
+            self.report_read(&batch);
             self.echoed.extend(batch);
             self.seen.add(shown);
             self.echoed_revision |= revision;

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

@@ -20,7 +20,7 @@ use serde_json::json;
 use tempfile::TempDir;
 
 use super::*;
-use crate::chat_client::{ChatImage, ChatSendReceipt, ContextLine, Refusal};
+use crate::chat_client::{ChatImage, ChatRead, ChatSendReceipt, ContextLine, Refusal};
 use base64::Engine;
 use eureka_pipeline::OrgRepo;
 use crate::control::{ControlWriter, FileSource, SwarmWriter};
@@ -59,6 +59,8 @@ struct FakeChat {
     deliveries: Deliveries,
     sent: RefCell<Vec<ChatSendRequest>>,
     acked: RefCell<Vec<u64>>,
+    reads: RefCell<Vec<ChatRead>>,
+    fail_read: Cell<bool>,
     pending_calls: Cell<usize>,
     fail_send_from: Cell<Option<usize>>,
     /// 0 a delivered receipt, 1 a refused one, 2 one with neither an event nor a refusal.
@@ -141,6 +143,14 @@ impl Chat for FakeChat {
         Ok(ChatSendReceipt { request_id: request.request_id.clone(), event_id, refusal, newer: Vec::new() })
     }
 
+    fn read(&self, read: &ChatRead) -> Result<()> {
+        if self.fail_read.get() {
+            return Err(anyhow!("the mouth is down"));
+        }
+        self.reads.borrow_mut().push(read.clone());
+        Ok(())
+    }
+
     fn ack(&self, through_seq: u64) -> Result<()> {
         self.log.borrow_mut().push("ack".into());
         self.acked.borrow_mut().push(through_seq);
@@ -414,6 +424,8 @@ impl Rig {
                 deliveries,
                 sent: RefCell::default(),
                 acked: RefCell::default(),
+                reads: RefCell::default(),
+                fail_read: Cell::new(false),
                 pending_calls: Cell::new(0),
                 fail_send_from: Cell::new(None),
                 receipt: Cell::new(0),
@@ -7516,3 +7528,80 @@ fn the_dropped_note_names_six_campaigns_and_counts_the_rest_together() {
     assert_eq!(dropped_note(&counts(6)), "(6 more open follow-ups not shown: c0 1, c1 1, c2 1, c3 1, c4 1, c5 1; `huginn open <campaign>` lists them.)\n");
     assert_eq!(dropped_note(&counts(7)), "(7 more open follow-ups not shown: c0 1, c1 1, c2 1, c3 1, c4 1, c5 1, 1 more campaigns 1; `huginn open <campaign>` lists them.)\n");
 }
+
+fn read_of(rig: &Rig) -> Vec<(String, String)> {
+    rig.chat.reads.borrow().iter().map(|read| (read.room.clone(), read.event_id.clone())).collect()
+}
+
+#[test]
+fn a_read_is_sent_only_at_the_echo_one_per_room_for_the_newest_event_that_went_into_her_input() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    let (a1, a2, b1) = (delivery(1, "swarm", "@op:chat", "swarm-one"), delivery(2, "swarm", "@op:chat", "swarm-two"), delivery(3, "ops", "@op:chat", "ops-one"));
+    let unread = delivery(4, "swarm", "@op:chat", "swarm-queued-unread");
+    for each in [&a1, &a2, &b1] {
+        arrives(&rig, each);
+    }
+    with_inbox(&rig, &config(), vec![a1.clone(), a2.clone(), b1.clone()], |inbox| {
+        let line = inbox.start_line().unwrap();
+        assert!(rig.chat.reads.borrow().is_empty(), "a line written and not echoed is not read");
+        arrives(&rig, &unread); // in the mouth's queue, never in her input
+        assert!(inbox.poll().unwrap().pending);
+        inbox.echoed();
+        let reads = rig.chat.reads.borrow().clone();
+        assert_eq!(reads.len(), 2, "one per room");
+        for read in &reads {
+            assert_eq!(read.persona, "yggdrasil");
+            // The report names an event whose message is in the model's input, and no queued-unread one.
+            let reported = [&a1, &a2, &b1, &unread].into_iter().find(|delivery| delivery.event_id == read.event_id).expect("a delivered event");
+            assert!(line.contains(&reported.text), "the reported event {} is not in her input", read.event_id);
+            assert_ne!(read.event_id, unread.event_id);
+            assert_eq!(read.room, reported.room_id, "the room id of the delivery, not its alias");
+            assert_eq!(read.at_unix_ms, reported.at_unix_ms);
+        }
+        assert_eq!(read_of(&rig), vec![("!ops:chat".to_string(), b1.event_id.clone()), ("!swarm:chat".to_string(), a2.event_id.clone())], "the newest of each room");
+        // Each line of the prompt still carries no event id of its own.
+        assert!(!line.contains(&a1.event_id) && !line.contains(&a2.event_id));
+    });
+}
+
+#[test]
+fn a_line_told_and_not_echoed_reports_nothing_and_a_report_is_never_repeated_or_moved_back() {
+    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(false).unwrap();
+        assert_eq!(read_of(&rig).len(), 1, "written, never echoed: not read");
+        inbox.report_read(&[first.clone()]);
+        assert_eq!(read_of(&rig).len(), 1, "the same event is not reported twice");
+        inbox.echoed();
+        assert_eq!(read_of(&rig), vec![("!swarm:chat".to_string(), first.event_id.clone()), ("!swarm:chat".to_string(), second.event_id.clone())]);
+        inbox.report_read(&[first.clone()]);
+        assert_eq!(read_of(&rig).len(), 2, "an older event after a newer one is not reported");
+    });
+}
+
+#[test]
+fn a_read_the_mouth_does_not_take_is_dropped_without_ending_the_session_and_is_tried_again_with_the_next_batch() {
+    let rig = Rig::new(vec![], &[]);
+    rig.clock.step.set(0);
+    rig.chat.fail_read.set(true);
+    let first = delivery(1, "swarm", "@op:chat", "one");
+    arrives(&rig, &first);
+    with_inbox(&rig, &config(), vec![first.clone()], |inbox| {
+        inbox.start_line().unwrap();
+        inbox.echoed();
+        inbox.result(&reply_result(&says(&[("swarm", "hi")])), true, false).unwrap();
+        assert!(inbox.progress.failure.is_none() && inbox.progress.answered == 1, "a failed receipt is cosmetic");
+        assert_eq!(rig.chat.acked.borrow().as_slice(), &[1], "the ack is the delivery guarantee and still goes");
+        rig.chat.fail_read.set(false);
+        inbox.report_read(&[first.clone()]);
+        assert_eq!(read_of(&rig), vec![("!swarm:chat".to_string(), first.event_id.clone())], "a report that failed did not count as reported");
+    });
+}

crates/mind-body/tests/chat_rudp.rs

@@ -13,7 +13,7 @@ use std::thread::JoinHandle;
 use std::time::Duration;
 
 use cultnet_rs::{CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions};
-use mind_body::chat_client::{self, ACK, Chat, ChatAck, ChatPending, ChatPendingRequest, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_V1, SEND_RECEIPT, SEND_REQUEST};
+use mind_body::chat_client::{self, ACK, Chat, ChatAck, ChatPending, ChatPendingRequest, ChatRead, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_V1, READ, SEND_RECEIPT, SEND_REQUEST};
 use mind_body::chat_rudp::{RudpChat, connection_id};
 
 const ACK_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack");
@@ -145,6 +145,20 @@ fn pending_send_and_ack_cross_a_real_rudp_session_in_the_mouths_bytes() {
     );
 }
 
+#[test]
+fn a_read_reaches_the_mouth_as_the_positional_tuple_persona_room_event_instant() {
+    let mouth = mouth(KEY);
+    let chat = client(&mouth, KEY, "eureka", Duration::from_secs(5));
+    let read = ChatRead { persona: "eureka".into(), room: "!abc:gamecult.org".into(), event_id: "$e1".into(), at_unix_ms: 1_761_700_000_000 };
+    chat.read(&read).unwrap();
+    let seen = mouth.seen.lock().unwrap().clone();
+    assert_eq!(seen.len(), 1);
+    assert_eq!(seen[0].0, READ);
+    assert_eq!(chat_client::decode::<ChatRead>(&seen[0].1).unwrap(), read);
+    let tuple: serde_json::Value = rmp_serde::from_slice(&seen[0].1).unwrap();
+    assert_eq!(tuple, serde_json::json!(["eureka", "!abc:gamecult.org", "$e1", 1_761_700_000_000u64]), "a positional tuple, not a map");
+}
+
 #[test]
 fn a_peer_without_the_credential_cannot_open_the_session() {
     let mouth = mouth(KEY);

crates/mind-body/tests/public_surface.txt

@@ -19,6 +19,7 @@ mind_body::chat_client: const ACK : & str
 mind_body::chat_client: const PENDING : & str
 mind_body::chat_client: const PENDING_REQUEST : & str
 mind_body::chat_client: const PENDING_V1 : & str
+mind_body::chat_client: const READ : & str
 mind_body::chat_client: const REFUSAL : & str
 mind_body::chat_client: const SEND_RECEIPT : & str
 mind_body::chat_client: const SEND_REQUEST : & str
@@ -28,12 +29,13 @@ mind_body::chat_client: impl Refusal
 mind_body::chat_client: impl Refusal: fn name (self) -> & 'static str
 mind_body::chat_client: impl Serialize for Refusal
 mind_body::chat_client: impl< 'de > Deserialize < 'de > for Refusal
-mind_body::chat_client: pub trait Chat { fn pending (& self , after_seq : u64) -> Result < Vec < ChatDelivery > > ; fn send (& self , request : & ChatSendRequest) -> Result < ChatSendReceipt > ; fn ack (& self , through_seq : u64) -> Result < () > ; }
+mind_body::chat_client: pub trait Chat { fn pending (& self , after_seq : u64) -> Result < Vec < ChatDelivery > > ; fn send (& self , request : & ChatSendRequest) -> Result < ChatSendReceipt > ; fn ack (& self , through_seq : u64) -> Result < () > ; fn read (& self , read : & ChatRead) -> Result < () > ; }
 mind_body::chat_client: struct ChatAck # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, through_seq: u64]
 mind_body::chat_client: struct ChatDelivery # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[seq: u64, room_alias: String, room_id: String, event_id: String, sender: String, sender_display: String, text: String, at_unix_ms: u64, context: Vec < ContextLine >, mentions: Vec < String >, is_direct: bool, image: Option < ChatImage >]
 mind_body::chat_client: struct ChatImage # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[path: String, mime: String, size: u64, sha256: String]
 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 ChatRead # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[persona: String, room: String, event_id: String, at_unix_ms: 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 < 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 >]