diff --git a/crates/mind-body/src/chat_client.rs b/crates/mind-body/src/chat_client.rs index ad4a880..83e04c7 100644 --- a/crates/mind-body/src/chat_client.rs +++ b/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>(deserializer: D) -> Result { 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, + 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, } -/// 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, - pub refusal: Option, + pub refusal: Option, + pub newer: Vec, } /// A document as the mouth reads it: a positional tuple. diff --git a/crates/mind-body/src/chat_client/tests.rs b/crates/mind-body/src/chat_client/tests.rs index d926bde..aa3ce22 100644 --- a/crates/mind-body/src/chat_client/tests.rs +++ b/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(bytes: &[u8], expected: &T) { assert_eq!(&decode::(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| 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()] }); } diff --git a/crates/mind-body/src/chat_rudp.rs b/crates/mind-body/src/chat_rudp.rs index aef5165..e4b673b 100644 --- a/crates/mind-body/src/chat_rudp.rs +++ b/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(&self, kind: &str, record_key: &str, document: &T, answer: Option<(&[&str], &str)>) -> Result)>> { + 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(&self, kind: &str, record_key: &str, document: &T, answer: Option<(&[&str], &str)>) -> Result)>> { let mut slot = self.link.borrow_mut(); if slot.is_none() { *slot = Some(self.open()?); diff --git a/crates/mind-body/src/persona.rs b/crates/mind-body/src/persona.rs index fa8d575..d98424b 100644 --- a/crates/mind-body/src/persona.rs +++ b/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, String, Vec), /// 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, newer: Vec }, /// 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, 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, 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, + rooms: Vec, + sent: Vec, + awaiting: Vec, + 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, + /// 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, + /// The reply a stale send stopped, until it is told back. + revision: Option, + /// 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, /// 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, @@ -1045,7 +1067,48 @@ struct PersonaInbox<'a, C: ControlSource, S: ControlSource> { repo_read_due_ms: u64, } -impl 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, 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> { earlier_shown(deliveries, &self.told, &self.config.own_sender, self.started_ms) @@ -1088,7 +1151,7 @@ impl 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 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 = 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 { 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)), ¬ices); + 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)), ¬ices); - 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, ¬ices); @@ -1216,16 +1312,24 @@ impl 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 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 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 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 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 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 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 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 { } impl 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(inbox: &PersonaInbox<'_, C, S>, batch: &[ChatDelivery], turn: u32, text: &str) -> Result { +fn act(inbox: &PersonaInbox<'_, C, S>, batch: &[ChatDelivery], turn: u32, text: &str, finality: Finality) -> Result { 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(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(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. diff --git a/crates/mind-body/src/persona/tests.rs b/crates/mind-body/src/persona/tests.rs index 4011e89..ee03a4f 100644 --- a/crates/mind-body/src/persona/tests.rs +++ b/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, /// Runs after each delivered send, where the world changes between turns. on_send: RefCell>>, + /// 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>, + /// What landed, by request id, as Matrix keys a transaction: a repeated id returns the old event and writes nothing. + landed: RefCell>, + /// Who the fake mouth's sends are from; her own lines never move a room for her. + speaker: RefCell, + /// Acks are recorded and the deliveries stay, where several sessions share one pending. + keep_on_ack: Cell, +} + +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 { + 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`, `$e-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); @@ -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(rig: &Rig, config: &Config, first: Vec, 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::(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 { + 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::>(); + 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 = (0..8).map(|index| format!("@mind{index}:chat")).collect(); + let configs: Vec = ids.iter().map(|id| Config { own_sender: id.clone(), minds: ids.clone(), ..config() }).collect(); + let mut progresses: Vec = (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> = 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::(), 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 = 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 = 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-")); + }); +} diff --git a/crates/mind-body/tests/chat_rudp.rs b/crates/mind-body/tests/chat_rudp.rs index d16b3aa..7a5318a 100644 --- a/crates/mind-body/tests/chat_rudp.rs +++ b/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::(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::(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"); +} diff --git a/crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack deleted file mode 100644 index 9b082a3..0000000 --- a/crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack +++ /dev/null @@ -1 +0,0 @@ -“¥req-1¥$sentÀ \ No newline at end of file diff --git a/crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack deleted file mode 100644 index 44f6485..0000000 --- a/crates/mind-body/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack +++ /dev/null @@ -1 +0,0 @@ -”¥req-1¦eureka¶#operator:gamecult.org¢hi \ No newline at end of file diff --git a/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.room-moved.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.room-moved.msgpack new file mode 100644 index 0000000..663ca9d --- /dev/null +++ b/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 \ No newline at end of file diff --git a/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack new file mode 100644 index 0000000..3372b43 --- /dev/null +++ b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_receipt.v2.sent.msgpack @@ -0,0 +1 @@ +”¥req-1¥$sentÀ \ No newline at end of file diff --git a/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.nil-basis.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.nil-basis.msgpack new file mode 100644 index 0000000..99f0521 --- /dev/null +++ b/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À \ No newline at end of file diff --git a/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.seen-and-since.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.seen-and-since.msgpack new file mode 100644 index 0000000..37bfca4 Binary files /dev/null and b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.seen-and-since.msgpack differ diff --git a/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.since-only.msgpack b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.since-only.msgpack new file mode 100644 index 0000000..43aa319 Binary files /dev/null and b/crates/mind-body/tests/fixtures/persona-chat-v2/gamecult.persona.chat_send_request.v2.since-only.msgpack differ diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 7570af6..9589a0f 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/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 @@ -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 > ; }