From cd3066dde9f70917915be6dc6b6662090f69cbf9 Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 16:28:12 +0000 Subject: [PATCH] chat_rudp: opt-in chat_pending.v4 request with step-down, warned set RudpChat::with_reply_link asks the v4 kind first and steps down one rung per UnknownKind refusal (v4 -> v2 -> v1); flag off sends today's bytes. The dropped-reply warning prints once per event id and reply_defects counts them. Spec: gamecult-chat:cut_spec:cut-chat-reply-link-decode.r4 Co-Authored-By: Claude Sonnet 5.5 --- crates/mind-body/src/chat_rudp.rs | 37 ++++++++++++--- crates/mind-body/tests/chat_rudp.rs | 71 ++++++++++++++++++++++++++++- 2 files changed, 100 insertions(+), 8 deletions(-) diff --git a/crates/mind-body/src/chat_rudp.rs b/crates/mind-body/src/chat_rudp.rs index 9726dfc..ba1e8ac 100644 --- a/crates/mind-body/src/chat_rudp.rs +++ b/crates/mind-body/src/chat_rudp.rs @@ -16,6 +16,7 @@ //! request id, which the mouth treats as the same send (`chat_client`). use std::cell::RefCell; +use std::collections::BTreeSet; use std::net::{SocketAddr, UdpSocket}; use std::time::{Duration, Instant}; @@ -29,7 +30,7 @@ use serde::de::DeserializeOwned; use sha2::{Digest, Sha256}; use crate::chat_client::{ - ACK, Chat, ChatAck, ChatDelivery, ChatOpenDirectReceipt, ChatOpenDirectRequest, ChatPending, ChatPendingRequest, ChatRead, ChatRefusal, MouthRefused, Refusal, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_REQUEST_V1, PENDING_V2, OPEN_DIRECT_RECEIPT, OPEN_DIRECT_REQUEST, PENDING_V1, READ, REFUSAL, SEND_RECEIPT, + ACK, Chat, ChatAck, ChatDelivery, ChatOpenDirectReceipt, ChatOpenDirectRequest, ChatPending, ChatPendingRequest, ChatRead, ChatRefusal, MouthRefused, Refusal, reply_warnings, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_REQUEST_V1, PENDING_REQUEST_V4, PENDING_V2, PENDING_V4, OPEN_DIRECT_RECEIPT, OPEN_DIRECT_REQUEST, PENDING_V1, READ, REFUSAL, SEND_RECEIPT, SEND_REQUEST, decode, encode, }; @@ -61,6 +62,10 @@ pub struct RudpChat { connection_id: u32, persona: String, reply_timeout: Duration, + /// Ask for the answer with the reply link first. Off by default: a mouth that has not been told to answer it refuses the kind. + reply_link: bool, + /// Event ids whose dropped reply link has been warned of: one warning per event however often the answer is polled. + warned: RefCell>, link: RefCell>, } @@ -69,7 +74,18 @@ impl RudpChat { /// (`connection_id`), `persona` the instance the mouth serves. Nothing is /// sent until the first call. pub fn new(remote: SocketAddr, connection_id: u32, persona: impl Into, reply_timeout: Duration) -> Self { - Self { remote, connection_id, persona: persona.into(), reply_timeout, link: RefCell::new(None) } + Self { remote, connection_id, persona: persona.into(), reply_timeout, reply_link: false, warned: RefCell::new(BTreeSet::new()), link: RefCell::new(None) } + } + + /// Ask the mouth for the reply link (`chat_pending.v4`) first. A mouth that refuses the kind is asked the older way. + pub fn with_reply_link(mut self, on: bool) -> Self { + self.reply_link = on; + self + } + + /// How many distinct events this process has seen arrive with a reply link out of the mouth's contract, dropped. + pub fn reply_defects(&self) -> usize { + self.warned.borrow().len() } fn open(&self) -> Result { @@ -161,11 +177,15 @@ impl Chat for RudpChat { fn pending(&self, after_seq: u64) -> Result { let request = ChatPendingRequest { persona: self.persona.clone(), after_seq }; let p = self.persona.as_str(); - let answers = [(PENDING, p), (PENDING_V2, p), (PENDING_V1, p), (REFUSAL, p)]; - let (mut kind, mut payload) = self.exchange(PENDING_REQUEST, p, &request, Some(&answers))?.ok_or_else(|| anyhow!("no answer payload"))?; - // A mouth that does not serve the v2 request answers `UnknownKind`: ask it the v1 way, and its deliveries carry no addressing facts. - if kind == REFUSAL && decode::(&payload)?.refusal == Refusal::UnknownKind { - (kind, payload) = self.exchange(PENDING_REQUEST_V1, p, &request, Some(&answers))?.ok_or_else(|| anyhow!("no answer payload"))?; + let answers = [(PENDING_V4, p), (PENDING, p), (PENDING_V2, p), (PENDING_V1, p), (REFUSAL, p)]; + // The newest kind first; a mouth that does not serve it answers `UnknownKind`, and is asked the next older way: v4 (reply link, + // only when on), v2 (addressing facts), v1 (none). + let asks: &[&str] = if self.reply_link { &[PENDING_REQUEST_V4, PENDING_REQUEST, PENDING_REQUEST_V1] } else { &[PENDING_REQUEST, PENDING_REQUEST_V1] }; + let mut asked = asks.iter(); + let (mut kind, mut payload) = self.exchange(asked.next().expect("at least one kind"), p, &request, Some(&answers))?.ok_or_else(|| anyhow!("no answer payload"))?; + while kind == REFUSAL && decode::(&payload)?.refusal == Refusal::UnknownKind { + let Some(older) = asked.next() else { break }; + (kind, payload) = self.exchange(older, p, &request, Some(&answers))?.ok_or_else(|| anyhow!("no answer payload"))?; } if kind == REFUSAL { let refusal: ChatRefusal = decode(&payload)?; @@ -175,6 +195,9 @@ impl Chat for RudpChat { if pending.persona != self.persona { bail!("the mouth answered for another Persona"); } + for line in reply_warnings(&pending, &mut self.warned.borrow_mut()) { + eprintln!("mind-chat: {line}"); + } Ok(pending) } diff --git a/crates/mind-body/tests/chat_rudp.rs b/crates/mind-body/tests/chat_rudp.rs index 05036fd..2ec7ae8 100644 --- a/crates/mind-body/tests/chat_rudp.rs +++ b/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, Basis, Form, ChatRefusal, MouthRefused, Refusal, REFUSAL, ACK, Chat, ChatAck, ChatOpenDirectReceipt, ChatOpenDirectRequest, DirectOutcome, OPEN_DIRECT_RECEIPT, OPEN_DIRECT_REQUEST, ChatPending, ChatPendingRequest, ChatRead, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_REQUEST_V1, PENDING_V1, PENDING_V2, READ, SEND_RECEIPT, SEND_REQUEST}; +use mind_body::chat_client::{self, ReplyLink, ReplyViolation, Basis, Form, ChatRefusal, MouthRefused, Refusal, REFUSAL, ACK, Chat, ChatAck, ChatOpenDirectReceipt, ChatOpenDirectRequest, DirectOutcome, OPEN_DIRECT_RECEIPT, OPEN_DIRECT_REQUEST, ChatPending, ChatPendingRequest, ChatRead, ChatSendReceipt, ChatSendRequest, PENDING, PENDING_REQUEST, PENDING_REQUEST_V1, PENDING_REQUEST_V4, PENDING_V1, PENDING_V2, PENDING_V4, 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"); @@ -26,6 +26,8 @@ const PENDING_GAPS_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v7/gamec const PENDING_V2_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack"); const PENDING_REQUEST_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v5/gamecult.persona.chat_pending_request.v2.msgpack"); const PENDING_REQUEST_V1_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack"); +// Bifrost dc35c4a2's own golden bytes for the reply link, copied unchanged as `persona-chat-v8/`. +const PENDING_V4_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v8/gamecult.persona.chat_pending.v4.reply-to.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 OPEN_REQUEST_BYTES: &[u8] = include_bytes!("fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_request.v1.msgpack"); @@ -89,6 +91,11 @@ fn imaged_mouth(key: &str) -> Mouth { serve(key, "127.0.0.1:0", PENDING_V2, PENDING_V2_BYTES, (OPEN_DIRECT_RECEIPT, None, OPENED_BYTES)) } +/// A mouth that answers the reply-link request (`chat_pending_request.v3`) with `chat_pending.v4`, and does not serve the older kinds. +fn linked_mouth(key: &str) -> Mouth { + serve(key, "127.0.0.1:0", PENDING_V4, PENDING_V4_BYTES, (OPEN_DIRECT_RECEIPT, None, OPENED_BYTES)) +} + /// The same mouth on a chosen address: a restart binds the address the old one held. fn mouth_at(key: &str, bind: &str) -> Mouth { serve(key, bind, PENDING, PENDING_BYTES, (OPEN_DIRECT_RECEIPT, None, OPENED_BYTES)) @@ -115,6 +122,9 @@ fn serve(key: &str, bind: &str, pending_kind: &'static str, pending_bytes: &'sta PENDING_REQUEST if for_eureka(&chat_client::decode::(&document.payload).unwrap().persona) => { Some(if pending_kind == PENDING { answer(pending_kind, "eureka", pending_bytes) } else { answer(REFUSAL, "eureka", UNKNOWN_KIND_BYTES) }) } + PENDING_REQUEST_V4 if for_eureka(&chat_client::decode::(&document.payload).unwrap().persona) => { + Some(if pending_kind == PENDING_V4 { answer(pending_kind, "eureka", pending_bytes) } else { answer(REFUSAL, "eureka", UNKNOWN_KIND_BYTES) }) + } PENDING_REQUEST_V1 if pending_kind != PENDING && for_eureka(&chat_client::decode::(&document.payload).unwrap().persona) => Some(answer(pending_kind, "eureka", pending_bytes)), SEND_REQUEST if for_eureka(&chat_client::decode::(&document.payload).unwrap().persona) => Some(answer(SEND_RECEIPT, &document.record_key, SEND_RECEIPT_BYTES)), OPEN_DIRECT_REQUEST if for_eureka(&chat_client::decode::(&document.payload).unwrap().persona) => Some(answer(open.0, open.1.unwrap_or(&document.record_key), open.2)), @@ -248,6 +258,65 @@ fn a_current_mouth_answers_the_v2_request_with_basis_and_form() { assert_eq!(kinds, [PENDING_REQUEST], "no v1 request follows an answered v2 request"); } +fn kinds_seen(mouth: &Mouth) -> Vec { + mouth.seen.lock().unwrap().iter().map(|(kind, _)| kind.clone()).collect() +} + +#[test] +fn a_flag_off_client_asks_only_the_v2_kind_and_a_flag_on_client_asks_v4_first() { + let current = mouth(KEY); + let off = client(¤t, KEY, "eureka", Duration::from_millis(1500)).with_reply_link(false).pending(3).unwrap(); + assert!(off.deliveries[0].reply_to.quote().is_none()); + assert_eq!(kinds_seen(¤t), [PENDING_REQUEST], "flag off sends exactly today's request"); + assert_eq!(current.seen.lock().unwrap()[0].1, PENDING_REQUEST_BYTES); + + let linked = linked_mouth(KEY); + let on = client(&linked, KEY, "eureka", Duration::from_millis(1500)).with_reply_link(true).pending(3).unwrap(); + assert_eq!(kinds_seen(&linked), [PENDING_REQUEST_V4]); + let quote = on.deliveries[0].reply_to.quote().cloned().expect("the reply link crossed the wire"); + assert_eq!((quote.event_id.as_str(), quote.sender.as_str(), quote.excerpt.as_deref()), ("$q1", "@alice:gamecult.org", Some("can someone look at the build?"))); +} + +#[test] +fn a_bad_reply_to_over_rudp_delivers_the_poll_warns_once_and_counts() { + let mut value: serde_json::Value = rmp_serde::from_slice(PENDING_V4_BYTES).unwrap(); + let valid = value[1][0].clone(); + let mut bad = valid.clone(); + bad[3] = "$bad".into(); + bad[14] = serde_json::json!(["$q1", "@alice:gamecult.org", 5]); + value[1] = serde_json::json!([valid.clone(), bad, valid]); + let bytes: &'static [u8] = Box::leak(rmp_serde::to_vec(&value).unwrap().into_boxed_slice()); + let check = |pending: chat_client::ChatPending| { + assert_eq!(pending.deliveries.len(), 3); + assert!(pending.deliveries[0].reply_to.quote().is_some() && pending.deliveries[2].reply_to.quote().is_some()); + assert_eq!(pending.deliveries[1].reply_to, ReplyLink::Dropped(ReplyViolation::ExcerptNotText)); + }; + let mouth = serve(KEY, "127.0.0.1:0", PENDING_V4, bytes, (OPEN_DIRECT_RECEIPT, None, OPENED_BYTES)); + let chat = client(&mouth, KEY, "eureka", Duration::from_millis(1500)).with_reply_link(true); + check(chat.pending(3).unwrap()); + assert_eq!(chat.reply_defects(), 1); + check(chat.pending(3).unwrap()); + assert_eq!(chat.reply_defects(), 1, "the same event is not counted again"); + // The lenient decode does not depend on the request kind: a flag-off client reads the same bytes. + let current = serve(KEY, "127.0.0.1:0", PENDING, bytes, (OPEN_DIRECT_RECEIPT, None, OPENED_BYTES)); + let off = client(¤t, KEY, "eureka", Duration::from_millis(1500)); + check(off.pending(3).unwrap()); + assert_eq!(off.reply_defects(), 1); +} + +#[test] +fn a_flag_on_client_steps_down_from_a_mouth_that_refuses_the_v4_kind() { + let current = mouth(KEY); + let pending = client(¤t, KEY, "eureka", Duration::from_millis(1500)).with_reply_link(true).pending(3).unwrap(); + assert_eq!(kinds_seen(¤t), [PENDING_REQUEST_V4, PENDING_REQUEST], "a flag-off mouth refuses v4 and answers v2"); + assert!(pending.deliveries[0].reply_to.quote().is_none() && pending.deliveries[0].basis == Some(Basis::Tag)); + for old in [legacy_mouth(KEY), imaged_mouth(KEY)] { + let pending = client(&old, KEY, "eureka", Duration::from_millis(1500)).with_reply_link(true).pending(3).unwrap(); + assert_eq!(kinds_seen(&old), [PENDING_REQUEST_V4, PENDING_REQUEST, PENDING_REQUEST_V1]); + assert!(pending.deliveries[0].reply_to.quote().is_none()); + } +} + #[test] fn an_open_crosses_a_real_rudp_session_in_the_mouths_bytes_and_a_receipt_for_another_open_is_refused() { let mouth = mouth(KEY);