GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

chat_rudp: opt-in chat_pending.v4 request with step-down, warned set

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Eureka 2026-11-11 16:28:12 +00:00
commit cd3066dde9
2 changed files with 100 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

37
crates/mind-body/src/chat_rudp.rs
View file

@ -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<BTreeSet<String>>,
link: RefCell<Option<Link>>,
}
@ -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<String>, 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<Link> {
@ -161,11 +177,15 @@ impl Chat for RudpChat {
fn pending(&self, after_seq: u64) -> Result<ChatPending> {
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::<ChatRefusal>(&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::<ChatRefusal>(&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)
}

71
crates/mind-body/tests/chat_rudp.rs
View file

@ -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::<ChatPendingRequest>(&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::<ChatPendingRequest>(&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::<ChatPendingRequest>(&document.payload).unwrap().persona) => Some(answer(pending_kind, "eureka", pending_bytes)),
SEND_REQUEST if for_eureka(&chat_client::decode::<ChatSendRequest>(&document.payload).unwrap().persona) => Some(answer(SEND_RECEIPT, &document.record_key, SEND_RECEIPT_BYTES)),
OPEN_DIRECT_REQUEST if for_eureka(&chat_client::decode::<ChatOpenDirectRequest>(&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<String> {
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(&current, 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(&current), [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(&current, 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(&current, KEY, "eureka", Duration::from_millis(1500)).with_reply_link(true).pending(3).unwrap();
assert_eq!(kinds_seen(&current), [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);