use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use mobius::backend::checkpoint::CheckpointStore;
use mobius::middleware::swarm::SwarmBackend;
use serde::{Deserialize, Serialize};
use tokio::sync::{Mutex, MutexGuard, mpsc};
use uuid::Uuid;
use crate::wire::{SwarmMemberRecord, SwarmMessageRecord, SwarmRecord};
use crate::{Error, Result};
const STATE_SCOPE: &str = "gateway";
const STATE_KEY: &str = "swarms.v1";
const MAX_HANDLE_BYTES: usize = 64;
const MAX_ID_BYTES: usize = 512;
const MAX_TITLE_BYTES: usize = 256;
const MAX_MESSAGE_BYTES: usize = 24_000;
const MAX_ACKNOWLEDGED_ENTRIES: usize = 256;
const MAX_PENDING_DELIVERIES_PER_RECIPIENT: usize = 256;
const MAX_PAGE_ENTRIES: usize = 256;
const MAX_SWARM_MEMBERS: usize = 100;
const MAX_CATALOG_BYTES: usize = 8 * 1024 * 1024;
const MAX_TOOL_READ_BYTES: usize = 32_000;
const MAX_TOOL_READ_TEXT_BYTES: usize = 4_000;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SwarmMember {
pub session_id: String,
pub handle: String,
pub joined_at_ms: i64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SwarmSummary {
pub id: String,
pub title: String,
pub leader_session_id: String,
pub members: Vec<SwarmMember>,
pub latest_sequence: u64,
pub created_at_ms: i64,
pub updated_at_ms: i64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SwarmSnapshot {
pub swarm: SwarmSummary,
pub handle: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BoardEntry {
pub id: String,
pub sequence: u64,
pub created_at_ms: i64,
pub author: SwarmMember,
pub text: String,
pub mentioned_recipient_session_ids: Vec<String>,
pub pending_recipient_session_ids: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SwarmPost {
pub entry: BoardEntry,
pub resolved_recipient_session_ids: Vec<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BoardPage {
pub entries: Vec<BoardEntry>,
pub next_before_sequence: Option<u64>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PendingDelivery {
pub swarm_id: String,
pub swarm_title: String,
pub entry: BoardEntry,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum SwarmDelivery {
Changed,
RetryPending,
Pending { target_session_id: String },
Acknowledged {
target_session_id: String,
message_id: String,
},
Rejected {
target_session_id: String,
message_id: String,
},
CapacityAvailable { target_session_id: String },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum AcknowledgeOutcome {
Acknowledged,
AlreadyAcknowledged,
MessageGone,
}
#[derive(Clone)]
pub struct SwarmStore {
checkpoints: Arc<dyn CheckpointStore>,
state: Arc<Mutex<Option<Catalog>>>,
deliveries: mpsc::UnboundedSender<SwarmDelivery>,
}
#[derive(Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Catalog {
swarms: BTreeMap<String, StoredSwarm>,
}
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct StoredSwarm {
title: String,
leader_session_id: String,
members: BTreeMap<String, SwarmMember>,
latest_sequence: u64,
board: VecDeque<BoardEntry>,
created_at_ms: i64,
updated_at_ms: i64,
}
impl SwarmStore {
#[must_use]
pub(crate) fn new(
checkpoints: Arc<dyn CheckpointStore>,
) -> (Self, mpsc::UnboundedReceiver<SwarmDelivery>) {
let (deliveries, receiver) = mpsc::unbounded_channel();
(
Self {
checkpoints,
state: Arc::new(Mutex::new(None)),
deliveries,
},
receiver,
)
}
pub async fn summaries(&self) -> Result<Vec<SwarmSummary>> {
let state = self.lock_loaded().await?;
Ok(state
.as_ref()
.expect("swarm catalog loaded")
.swarms
.iter()
.map(|(id, swarm)| swarm.summary(id))
.collect())
}
pub(crate) async fn records(&self) -> Result<Vec<SwarmRecord>> {
let state = self.lock_loaded().await?;
Ok(state
.as_ref()
.expect("swarm catalog loaded")
.swarms
.iter()
.map(|(id, swarm)| {
let mut messages = swarm
.board
.iter()
.rev()
.take(MAX_PAGE_ENTRIES)
.map(|entry| SwarmMessageRecord {
id: entry.id.clone(),
sequence: entry.sequence,
author_session_id: entry.author.session_id.clone(),
author_handle: entry.author.handle.clone(),
text: entry.text.clone(),
created_at_ms: entry.created_at_ms,
})
.collect::<Vec<_>>();
messages.reverse();
SwarmRecord {
id: id.clone(),
title: swarm.title.clone(),
leader_session_id: swarm.leader_session_id.clone(),
members: swarm
.members
.values()
.map(|member| SwarmMemberRecord {
session_id: member.session_id.clone(),
handle: member.handle.clone(),
})
.collect(),
messages,
updated_at_ms: swarm.updated_at_ms,
}
})
.collect())
}
pub(crate) async fn create(
&self,
leader_session_id: String,
member_session_ids: Vec<String>,
) -> Result<SwarmSummary> {
validate_swarm_members(&leader_session_id, &member_session_ids)?;
self.mutate(move |catalog| {
for session_id in &member_session_ids {
ensure_session_available(catalog, session_id)?;
}
let id = Uuid::new_v4();
let title = format!(
"Swarm {}",
id.simple().to_string().chars().take(8).collect::<String>()
);
let mut members = BTreeMap::new();
for session_id in member_session_ids {
let member = generated_member(session_id, &members);
members.insert(member.handle.clone(), member);
}
let now = unix_ms();
let swarm = StoredSwarm {
title,
leader_session_id,
members,
latest_sequence: 0,
board: VecDeque::new(),
created_at_ms: now,
updated_at_ms: now,
};
let id = id.to_string();
let summary = swarm.summary(&id);
catalog.swarms.insert(id, swarm);
Ok(summary)
})
.await
}
pub(crate) async fn join(&self, swarm_id: &str, session_id: String) -> Result<SwarmSummary> {
validate_swarm_id(swarm_id)?;
validate_session_id(&session_id)?;
let swarm_id = swarm_id.to_owned();
self.mutate(move |catalog| {
ensure_session_available(catalog, &session_id)?;
let swarm = catalog
.swarms
.get_mut(&swarm_id)
.ok_or_else(|| config(format!("unknown swarm `{swarm_id}`")))?;
if swarm.members.len() >= MAX_SWARM_MEMBERS {
return Err(config(format!(
"a swarm supports at most {MAX_SWARM_MEMBERS} members"
)));
}
let member = generated_member(session_id, &swarm.members);
swarm.members.insert(member.handle.clone(), member);
swarm.updated_at_ms = unix_ms();
Ok(swarm.summary(&swarm_id))
})
.await
}
pub(crate) async fn leave(&self, swarm_id: &str, session_id: &str) -> Result<SwarmSummary> {
validate_swarm_id(swarm_id)?;
validate_session_id(session_id)?;
let swarm_id = swarm_id.to_owned();
let session_id = session_id.to_owned();
let acknowledged_target = session_id.clone();
let (summary, acknowledged_messages) = self
.mutate(move |catalog| {
let swarm = catalog
.swarms
.get_mut(&swarm_id)
.ok_or_else(|| config(format!("unknown swarm `{swarm_id}`")))?;
if swarm.leader_session_id == session_id {
return Err(config("swarm leader must disband instead of leaving"));
}
let handle = swarm
.members
.iter()
.find_map(|(handle, member)| {
(member.session_id == session_id).then(|| handle.clone())
})
.ok_or_else(|| {
config(format!("session `{session_id}` is not in this swarm"))
})?;
swarm.members.remove(&handle);
let mut acknowledged_messages = Vec::new();
for entry in &mut swarm.board {
if entry
.pending_recipient_session_ids
.iter()
.any(|pending| pending == &session_id)
{
acknowledged_messages.push(entry.id.clone());
}
entry
.pending_recipient_session_ids
.retain(|pending| pending != &session_id);
}
trim_acknowledged(&mut swarm.board);
swarm.updated_at_ms = unix_ms();
Ok((swarm.summary(&swarm_id), acknowledged_messages))
})
.await?;
for message_id in acknowledged_messages {
self.notify_acknowledged(&message_id, &acknowledged_target);
}
Ok(summary)
}
pub(crate) async fn disband(&self, swarm_id: &str) -> Result<SwarmSummary> {
validate_swarm_id(swarm_id)?;
let swarm_id = swarm_id.to_owned();
let (summary, pending_deliveries) = self
.mutate(move |catalog| {
let swarm = catalog
.swarms
.remove(&swarm_id)
.ok_or_else(|| config(format!("unknown swarm `{swarm_id}`")))?;
let pending_deliveries = swarm
.board
.iter()
.flat_map(|entry| {
entry
.pending_recipient_session_ids
.iter()
.map(|target| (entry.id.clone(), target.clone()))
})
.collect::<BTreeSet<_>>();
Ok((swarm.summary(&swarm_id), pending_deliveries))
})
.await?;
for (message_id, target_session_id) in pending_deliveries {
self.notify_acknowledged(&message_id, &target_session_id);
}
Ok(summary)
}
pub async fn snapshot_for_session(&self, session_id: &str) -> Result<Option<SwarmSnapshot>> {
validate_session_id(session_id)?;
let state = self.lock_loaded().await?;
let catalog = state.as_ref().expect("swarm catalog loaded");
Ok(catalog.swarms.iter().find_map(|(id, swarm)| {
swarm
.members
.values()
.find(|member| member.session_id == session_id)
.map(|member| SwarmSnapshot {
swarm: swarm.summary(id),
handle: member.handle.clone(),
})
}))
}
pub async fn post(&self, sender_session_id: &str, text: String) -> Result<SwarmPost> {
validate_session_id(sender_session_id)?;
validate_message(&text)?;
let sender_session_id = sender_session_id.to_owned();
let post = self
.mutate(move |catalog| {
let swarm_id =
swarm_id_for_session(catalog, &sender_session_id).ok_or_else(|| {
config(format!("session `{sender_session_id}` is not in a swarm"))
})?;
let swarm = catalog
.swarms
.get_mut(&swarm_id)
.expect("resolved swarm exists");
let author = swarm
.members
.values()
.find(|member| member.session_id == sender_session_id)
.cloned()
.expect("resolved swarm contains sender");
let handles = mentioned_handles(&text);
let unknown = handles
.iter()
.filter(|handle| !swarm.members.contains_key(*handle))
.cloned()
.collect::<Vec<_>>();
if !unknown.is_empty() {
return Err(config(format!(
"unknown swarm mention{}: {}",
if unknown.len() == 1 { "" } else { "s" },
unknown
.iter()
.map(|handle| format!("@{handle}"))
.collect::<Vec<_>>()
.join(", ")
)));
}
let recipients = handles
.iter()
.map(|handle| {
swarm
.members
.get(handle)
.expect("mentions validated against roster")
.session_id
.clone()
})
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
if recipients
.iter()
.any(|session| session == &sender_session_id)
{
return Err(config("a swarm message cannot mention its author"));
}
if let Some(recipient) = recipients.iter().find(|recipient| {
swarm
.board
.iter()
.filter(|entry| {
entry
.pending_recipient_session_ids
.iter()
.any(|pending| pending == *recipient)
})
.count()
>= MAX_PENDING_DELIVERIES_PER_RECIPIENT
}) {
return Err(config(format!(
"session `{recipient}` has {MAX_PENDING_DELIVERIES_PER_RECIPIENT} pending swarm messages"
)));
}
let sequence = swarm
.latest_sequence
.checked_add(1)
.ok_or_else(|| config("swarm board sequence exhausted"))?;
let entry = BoardEntry {
id: Uuid::new_v4().to_string(),
sequence,
created_at_ms: unix_ms(),
author,
text,
mentioned_recipient_session_ids: recipients.clone(),
pending_recipient_session_ids: recipients.clone(),
};
swarm.latest_sequence = sequence;
swarm.updated_at_ms = entry.created_at_ms;
swarm.board.push_back(entry.clone());
trim_acknowledged(&mut swarm.board);
Ok(SwarmPost {
entry,
resolved_recipient_session_ids: recipients,
})
})
.await?;
let _ = self.deliveries.send(SwarmDelivery::Changed);
for target_session_id in &post.resolved_recipient_session_ids {
self.notify_pending(target_session_id);
}
Ok(post)
}
pub async fn board_page(
&self,
swarm_id: &str,
before_sequence: Option<u64>,
limit: usize,
) -> Result<BoardPage> {
validate_swarm_id(swarm_id)?;
if limit == 0 || limit > MAX_PAGE_ENTRIES {
return Err(config(format!(
"board page limit must be 1-{MAX_PAGE_ENTRIES}"
)));
}
let state = self.lock_loaded().await?;
let swarm = state
.as_ref()
.expect("swarm catalog loaded")
.swarms
.get(swarm_id)
.ok_or_else(|| config(format!("unknown swarm `{swarm_id}`")))?;
let mut matches = swarm
.board
.iter()
.rev()
.filter(|entry| before_sequence.is_none_or(|before| entry.sequence < before));
let entries = matches.by_ref().take(limit).cloned().collect::<Vec<_>>();
let next_before_sequence = matches
.next()
.and_then(|_| entries.last().map(|entry| entry.sequence));
Ok(BoardPage {
entries,
next_before_sequence,
})
}
pub async fn pending_deliveries(
&self,
target_session_id: &str,
) -> Result<Vec<PendingDelivery>> {
validate_session_id(target_session_id)?;
let state = self.lock_loaded().await?;
let catalog = state.as_ref().expect("swarm catalog loaded");
Ok(catalog
.swarms
.iter()
.flat_map(|(swarm_id, swarm)| {
swarm
.board
.iter()
.filter(|entry| {
entry
.pending_recipient_session_ids
.iter()
.any(|pending| pending == target_session_id)
})
.map(|entry| PendingDelivery {
swarm_id: swarm_id.clone(),
swarm_title: swarm.title.clone(),
entry: entry.clone(),
})
})
.collect())
}
pub(crate) async fn pending_recipient_session_ids(&self) -> Result<Vec<String>> {
let state = self.lock_loaded().await?;
Ok(state
.as_ref()
.expect("swarm catalog loaded")
.swarms
.values()
.flat_map(|swarm| &swarm.board)
.flat_map(|entry| &entry.pending_recipient_session_ids)
.cloned()
.collect::<BTreeSet<_>>()
.into_iter()
.collect())
}
pub(crate) fn notify_pending(&self, target_session_id: &str) {
let _ = self.deliveries.send(SwarmDelivery::Pending {
target_session_id: target_session_id.to_owned(),
});
}
pub(crate) fn retry_pending(&self) {
let _ = self.deliveries.send(SwarmDelivery::RetryPending);
}
pub(crate) fn notify_acknowledged(&self, message_id: &str, target_session_id: &str) {
let _ = self.deliveries.send(SwarmDelivery::Acknowledged {
target_session_id: target_session_id.to_owned(),
message_id: message_id.to_owned(),
});
}
pub(crate) fn notify_rejected(&self, message_id: &str, target_session_id: &str) {
let _ = self.deliveries.send(SwarmDelivery::Rejected {
target_session_id: target_session_id.to_owned(),
message_id: message_id.to_owned(),
});
}
pub(crate) fn notify_capacity_available(&self, target_session_id: &str) {
let _ = self.deliveries.send(SwarmDelivery::CapacityAvailable {
target_session_id: target_session_id.to_owned(),
});
}
pub(crate) async fn acknowledge(
&self,
message_id: &str,
target_session_id: &str,
) -> Result<AcknowledgeOutcome> {
validate_message_id(message_id)?;
validate_session_id(target_session_id)?;
let message_id = message_id.to_owned();
let target_session_id = target_session_id.to_owned();
let acknowledged_message = message_id.clone();
let acknowledged_target = target_session_id.clone();
let outcome = self
.mutate_if_some(move |catalog| {
let Some(entry) = catalog
.swarms
.values_mut()
.find_map(|swarm| swarm.board.iter_mut().find(|entry| entry.id == message_id))
else {
return Ok(None);
};
if !entry
.mentioned_recipient_session_ids
.iter()
.any(|recipient| recipient == &target_session_id)
{
return Err(config(format!(
"session `{target_session_id}` is not a recipient of message `{message_id}`"
)));
}
let was_pending = entry
.pending_recipient_session_ids
.iter()
.any(|pending| pending == &target_session_id);
entry
.pending_recipient_session_ids
.retain(|pending| pending != &target_session_id);
for swarm in catalog.swarms.values_mut() {
trim_acknowledged(&mut swarm.board);
}
Ok(Some(if was_pending {
AcknowledgeOutcome::Acknowledged
} else {
AcknowledgeOutcome::AlreadyAcknowledged
}))
})
.await?
.unwrap_or(AcknowledgeOutcome::MessageGone);
self.notify_acknowledged(&acknowledged_message, &acknowledged_target);
Ok(outcome)
}
async fn mutate<T>(&self, mutation: impl FnOnce(&mut Catalog) -> Result<T>) -> Result<T> {
Ok(self
.mutate_if_some(|catalog| mutation(catalog).map(Some))
.await?
.expect("required mutation returns a value"))
}
async fn mutate_if_some<T>(
&self,
mutation: impl FnOnce(&mut Catalog) -> Result<Option<T>>,
) -> Result<Option<T>> {
let mut state = self.lock_loaded().await?;
let mut candidate = state.as_ref().expect("swarm catalog loaded").clone();
let Some(output) = mutation(&mut candidate)? else {
return Ok(None);
};
validate_catalog(&candidate)?;
self.checkpoints
.save_state(STATE_SCOPE, STATE_KEY, &serde_json::to_value(&candidate)?)
.await?;
*state = Some(candidate);
Ok(Some(output))
}
async fn lock_loaded(&self) -> Result<MutexGuard<'_, Option<Catalog>>> {
let mut state = self.state.lock().await;
if state.is_none() {
let catalog = self
.checkpoints
.load_state(STATE_SCOPE, STATE_KEY)
.await?
.map(serde_json::from_value)
.transpose()?
.unwrap_or_default();
validate_catalog(&catalog)?;
*state = Some(catalog);
}
Ok(state)
}
}
pub(crate) fn validate_swarm_members(
leader_session_id: &str,
member_session_ids: &[String],
) -> Result<()> {
validate_session_id(leader_session_id)?;
if member_session_ids.len() < 2 {
return Err(config("a swarm requires at least two members"));
}
if member_session_ids.len() > MAX_SWARM_MEMBERS {
return Err(config(format!(
"a swarm supports at most {MAX_SWARM_MEMBERS} members"
)));
}
let mut unique_sessions = BTreeSet::new();
for session_id in member_session_ids {
validate_session_id(session_id)?;
if !unique_sessions.insert(session_id.as_str()) {
return Err(config(format!(
"session `{session_id}` appears more than once"
)));
}
}
if !unique_sessions.contains(leader_session_id) {
return Err(config("swarm leader must be a member"));
}
Ok(())
}
impl SwarmBackend for SwarmStore {
fn active<'a>(&'a self, session_id: &'a str) -> mobius::BoxFuture<'a, mobius::Result<bool>> {
Box::pin(async move {
self.snapshot_for_session(session_id)
.await
.map(|snapshot| snapshot.is_some())
.map_err(mobius_error)
})
}
fn roster<'a>(&'a self, session_id: &'a str) -> mobius::BoxFuture<'a, mobius::Result<String>> {
Box::pin(async move {
let snapshot = self
.snapshot_for_session(session_id)
.await
.map_err(mobius_error)?
.ok_or_else(|| mobius::Error::Config("this chat is not in a swarm".into()))?;
serde_json::to_string(&snapshot).map_err(Into::into)
})
}
fn read<'a>(&'a self, session_id: &'a str) -> mobius::BoxFuture<'a, mobius::Result<String>> {
Box::pin(async move {
let snapshot = self
.snapshot_for_session(session_id)
.await
.map_err(mobius_error)?
.ok_or_else(|| mobius::Error::Config("this chat is not in a swarm".into()))?;
let page = self
.board_page(&snapshot.swarm.id, None, MAX_PAGE_ENTRIES)
.await
.map_err(mobius_error)?;
let mut entries = Vec::new();
let mut has_older = page.next_before_sequence.is_some();
for entry in page.entries {
let end = entry
.text
.floor_char_boundary(MAX_TOOL_READ_TEXT_BYTES.min(entry.text.len()));
let text = &entry.text[..end];
entries.push(serde_json::json!({
"id": entry.id,
"sequence": entry.sequence,
"created_at_ms": entry.created_at_ms,
"author_session_id": entry.author.session_id,
"author_handle": entry.author.handle,
"text": text,
"text_truncated": text.len() != entry.text.len(),
}));
if serde_json::to_vec(&serde_json::json!({
"entries": &entries,
"has_older": has_older,
}))?
.len()
> MAX_TOOL_READ_BYTES
{
entries.pop();
has_older = true;
break;
}
}
serde_json::to_string(&serde_json::json!({
"entries": entries,
"has_older": has_older,
}))
.map_err(Into::into)
})
}
fn post<'a>(
&'a self,
session_id: &'a str,
text: String,
) -> mobius::BoxFuture<'a, mobius::Result<String>> {
Box::pin(async move {
let post = SwarmStore::post(self, session_id, text)
.await
.map_err(mobius_error)?;
serde_json::to_string(&post).map_err(Into::into)
})
}
}
impl StoredSwarm {
fn summary(&self, id: &str) -> SwarmSummary {
SwarmSummary {
id: id.to_owned(),
title: self.title.clone(),
leader_session_id: self.leader_session_id.clone(),
members: self.members.values().cloned().collect(),
latest_sequence: self.latest_sequence,
created_at_ms: self.created_at_ms,
updated_at_ms: self.updated_at_ms,
}
}
}
fn generated_member(session_id: String, members: &BTreeMap<String, SwarmMember>) -> SwarmMember {
let handle = loop {
let suffix = Uuid::new_v4()
.simple()
.to_string()
.chars()
.take(8)
.collect::<String>();
let handle = format!("agent_{suffix}");
if !members.contains_key(&handle) {
break handle;
}
};
SwarmMember {
session_id,
handle,
joined_at_ms: unix_ms(),
}
}
fn ensure_session_available(catalog: &Catalog, session_id: &str) -> Result<()> {
if catalog.swarms.values().any(|swarm| {
swarm
.members
.values()
.any(|member| member.session_id == session_id)
}) {
return Err(config(format!(
"session `{session_id}` already belongs to a swarm"
)));
}
Ok(())
}
fn swarm_id_for_session(catalog: &Catalog, session_id: &str) -> Option<String> {
catalog.swarms.iter().find_map(|(id, swarm)| {
swarm
.members
.values()
.any(|member| member.session_id == session_id)
.then(|| id.clone())
})
}
fn trim_acknowledged(board: &mut VecDeque<BoardEntry>) {
while board
.iter()
.filter(|entry| entry.pending_recipient_session_ids.is_empty())
.count()
> MAX_ACKNOWLEDGED_ENTRIES
{
let index = board
.iter()
.position(|entry| entry.pending_recipient_session_ids.is_empty())
.expect("acknowledged entry count is positive");
board.remove(index);
}
}
fn mentioned_handles(text: &str) -> BTreeSet<String> {
let bytes = text.as_bytes();
let mut handles = BTreeSet::new();
let mut index = 0;
while index < bytes.len() {
if bytes[index] != b'@'
|| index
.checked_sub(1)
.is_some_and(|previous| is_mention_byte(bytes[previous]))
{
index += 1;
continue;
}
let start = index + 1;
let mut end = start;
while end < bytes.len() && is_mention_byte(bytes[end]) {
end += 1;
}
if start < end {
handles.insert(text[start..end].to_owned());
}
index = end.max(index + 1);
}
handles
}
const fn is_mention_byte(byte: u8) -> bool {
byte.is_ascii_alphanumeric() || byte == b'_'
}
fn validate_catalog(catalog: &Catalog) -> Result<()> {
let mut sessions = BTreeSet::new();
let mut message_ids = BTreeSet::new();
for (id, swarm) in &catalog.swarms {
let mut pending_counts = BTreeMap::<&str, usize>::new();
validate_swarm_id(id)?;
validate_title(&swarm.title)?;
validate_session_id(&swarm.leader_session_id)?;
if swarm.members.len() > MAX_SWARM_MEMBERS {
return Err(config(format!(
"swarm `{id}` has more than {MAX_SWARM_MEMBERS} members"
)));
}
if swarm.created_at_ms < 0 || swarm.updated_at_ms < swarm.created_at_ms {
return Err(config(format!("swarm `{id}` has invalid timestamps")));
}
if !swarm
.members
.values()
.any(|member| member.session_id == swarm.leader_session_id)
{
return Err(config(format!("swarm `{id}` leader is not a member")));
}
for (handle, member) in &swarm.members {
validate_member(member)?;
if handle != &member.handle {
return Err(config(format!(
"swarm `{id}` member handle does not match its key"
)));
}
if !sessions.insert(member.session_id.clone()) {
return Err(config(format!(
"session `{}` belongs to more than one swarm",
member.session_id
)));
}
}
let mut previous_sequence = 0;
for entry in &swarm.board {
validate_message_id(&entry.id)?;
validate_member(&entry.author)?;
validate_message(&entry.text)?;
if entry.created_at_ms < swarm.created_at_ms
|| entry.created_at_ms > swarm.updated_at_ms
{
return Err(config(format!(
"swarm message `{}` has an invalid timestamp",
entry.id
)));
}
if !message_ids.insert(entry.id.clone()) {
return Err(config(format!(
"swarm message `{}` appears more than once",
entry.id
)));
}
if entry.sequence <= previous_sequence || entry.sequence > swarm.latest_sequence {
return Err(config(format!("swarm `{id}` has invalid board sequences")));
}
previous_sequence = entry.sequence;
validate_recipient_ids(&entry.mentioned_recipient_session_ids)?;
validate_recipient_ids(&entry.pending_recipient_session_ids)?;
for recipient in &entry.pending_recipient_session_ids {
let count = pending_counts.entry(recipient).or_default();
*count += 1;
if *count > MAX_PENDING_DELIVERIES_PER_RECIPIENT {
return Err(config(format!(
"session `{recipient}` has too many pending swarm messages"
)));
}
}
if entry.pending_recipient_session_ids.iter().any(|pending| {
!entry
.mentioned_recipient_session_ids
.iter()
.any(|mentioned| mentioned == pending)
}) {
return Err(config(format!(
"swarm message `{}` has a pending non-recipient",
entry.id
)));
}
}
if swarm
.board
.back()
.is_some_and(|entry| entry.sequence != swarm.latest_sequence)
|| (swarm.board.is_empty() && swarm.latest_sequence != 0)
|| swarm
.board
.iter()
.filter(|entry| entry.pending_recipient_session_ids.is_empty())
.count()
> MAX_ACKNOWLEDGED_ENTRIES
{
return Err(config(format!("swarm `{id}` board state is invalid")));
}
}
if serde_json::to_vec(catalog)?.len() > MAX_CATALOG_BYTES {
return Err(config(format!(
"swarm catalog exceeds {MAX_CATALOG_BYTES} encoded bytes"
)));
}
Ok(())
}
fn validate_recipient_ids(recipients: &[String]) -> Result<()> {
let mut unique = BTreeSet::new();
for recipient in recipients {
validate_session_id(recipient)?;
if !unique.insert(recipient) {
return Err(config("swarm message recipient appears more than once"));
}
}
Ok(())
}
fn validate_member(member: &SwarmMember) -> Result<()> {
validate_session_id(&member.session_id)?;
validate_handle(&member.handle)?;
if member.joined_at_ms < 0 {
return Err(config("swarm member join time cannot be negative"));
}
Ok(())
}
fn validate_handle(handle: &str) -> Result<()> {
if handle.is_empty()
|| handle.len() > MAX_HANDLE_BYTES
|| !handle
.bytes()
.all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
{
return Err(config(
"swarm handle must contain 1-64 lowercase letters, digits, or underscores",
));
}
Ok(())
}
fn validate_title(title: &str) -> Result<()> {
if title.trim().is_empty() || title.len() > MAX_TITLE_BYTES {
return Err(config(format!(
"swarm title must contain 1-{MAX_TITLE_BYTES} UTF-8 bytes"
)));
}
Ok(())
}
fn validate_message(text: &str) -> Result<()> {
if text.trim().is_empty() || text.len() > MAX_MESSAGE_BYTES {
return Err(config(format!(
"swarm message must contain 1-{MAX_MESSAGE_BYTES} UTF-8 bytes"
)));
}
Ok(())
}
fn validate_session_id(session_id: &str) -> Result<()> {
if session_id.is_empty() || session_id.len() > MAX_ID_BYTES {
return Err(config(format!(
"session id must contain 1-{MAX_ID_BYTES} bytes"
)));
}
Ok(())
}
fn validate_swarm_id(id: &str) -> Result<()> {
Uuid::parse_str(id)
.map(|_| ())
.map_err(|_| config("invalid swarm id"))
}
fn validate_message_id(id: &str) -> Result<()> {
Uuid::parse_str(id)
.map(|_| ())
.map_err(|_| config("invalid swarm message id"))
}
fn config(message: impl Into<String>) -> Error {
Error::Config(message.into())
}
fn mobius_error(error: Error) -> mobius::Error {
mobius::Error::Config(error.to_string())
}
fn unix_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX))
.unwrap_or_default()
}
#[cfg(test)]
mod tests {
use mobius::backend::checkpoint::sqlite::SqliteCheckpoint;
use super::*;
fn store() -> (
tempfile::TempDir,
Arc<dyn CheckpointStore>,
SwarmStore,
mpsc::UnboundedReceiver<SwarmDelivery>,
) {
let directory = tempfile::tempdir().expect("workspace");
let checkpoints: Arc<dyn CheckpointStore> = Arc::new(
SqliteCheckpoint::new(directory.path().join("checkpoints.sqlite3"))
.expect("checkpoints"),
);
let (store, deliveries) = SwarmStore::new(Arc::clone(&checkpoints));
(directory, checkpoints, store, deliveries)
}
async fn create_swarm(store: &SwarmStore) -> SwarmSummary {
store
.create("leader".into(), vec!["leader".into(), "reviewer".into()])
.await
.expect("create swarm")
}
#[tokio::test]
async fn membership_is_unique_and_lazily_reloads() {
let (_directory, checkpoints, store, _deliveries) = store();
let created = create_swarm(&store).await;
let created_title = created.title.clone();
assert!(created.created_at_ms > 0);
assert_eq!(created.created_at_ms, created.updated_at_ms);
let duplicate = store
.create("leader".into(), vec!["leader".into(), "third".into()])
.await
.expect_err("session cannot join two swarms");
assert!(duplicate.to_string().contains("already belongs"));
let (reloaded, _deliveries) = SwarmStore::new(checkpoints);
assert_eq!(reloaded.summaries().await.expect("reload"), vec![created]);
let snapshot = reloaded
.snapshot_for_session("reviewer")
.await
.expect("snapshot")
.expect("membership");
assert!(snapshot.handle.starts_with("agent_"));
assert_eq!(snapshot.swarm.title, created_title);
}
#[tokio::test]
async fn joining_cannot_exceed_the_member_limit() {
let (_directory, _checkpoints, store, _deliveries) = store();
let members = (0..MAX_SWARM_MEMBERS)
.map(|index| format!("member-{index}"))
.collect::<Vec<_>>();
let swarm = store
.create(members[0].clone(), members)
.await
.expect("full swarm");
let error = store
.join(&swarm.id, "overflow".into())
.await
.expect_err("member limit");
assert!(error.to_string().contains("at most 100"));
}
#[tokio::test]
async fn mentions_are_resolved_delivered_and_acknowledged() {
let (_directory, checkpoints, store, mut deliveries) = store();
let swarm = create_swarm(&store).await;
let leader_handle = swarm
.members
.iter()
.find(|member| member.session_id == "leader")
.expect("leader")
.handle
.clone();
let reviewer_handle = swarm
.members
.iter()
.find(|member| member.session_id == "reviewer")
.expect("reviewer")
.handle
.clone();
let unknown = store
.post("leader", "Can @missing check this?".into())
.await
.expect_err("unknown mention");
assert!(unknown.to_string().contains("@missing"));
let text = format!("Can @{reviewer_handle} check this?");
let post = store.post("leader", text.clone()).await.expect("post");
assert_eq!(deliveries.recv().await, Some(SwarmDelivery::Changed));
assert_eq!(
deliveries.recv().await,
Some(SwarmDelivery::Pending {
target_session_id: "reviewer".into()
})
);
assert_eq!(post.entry.sequence, 1);
assert!(post.entry.created_at_ms >= swarm.created_at_ms);
assert_eq!(post.entry.author.handle, leader_handle);
assert_eq!(post.resolved_recipient_session_ids, ["reviewer"]);
let records = store.records().await.expect("wire records");
assert_eq!(records[0].messages[0].text, text);
assert!(
SwarmBackend::active(&store, "reviewer")
.await
.expect("active")
);
assert_eq!(
store
.pending_deliveries("reviewer")
.await
.expect("pending")
.len(),
1
);
assert_eq!(
store
.acknowledge(&post.entry.id, "reviewer")
.await
.expect("acknowledge"),
AcknowledgeOutcome::Acknowledged
);
assert_eq!(
deliveries.recv().await,
Some(SwarmDelivery::Acknowledged {
target_session_id: "reviewer".into(),
message_id: post.entry.id.clone(),
})
);
assert_eq!(
store
.acknowledge(&post.entry.id, "reviewer")
.await
.expect("idempotent acknowledge"),
AcknowledgeOutcome::AlreadyAcknowledged
);
assert_eq!(
deliveries.recv().await,
Some(SwarmDelivery::Acknowledged {
target_session_id: "reviewer".into(),
message_id: post.entry.id.clone(),
})
);
assert!(
store
.pending_deliveries("reviewer")
.await
.expect("pending")
.is_empty()
);
let (reloaded, _deliveries) = SwarmStore::new(checkpoints);
let page = reloaded
.board_page(&swarm.id, None, 10)
.await
.expect("board page");
assert_eq!(page.entries[0].id, post.entry.id);
assert!(page.entries[0].pending_recipient_session_ids.is_empty());
}
#[tokio::test]
async fn leave_and_disband_settle_removed_pending_deliveries() {
let (_directory, _checkpoints, store, mut deliveries) = store();
let swarm = store
.create(
"leader".into(),
vec!["leader".into(), "observer".into(), "reviewer".into()],
)
.await
.expect("create swarm");
let handle = |session_id: &str| {
swarm
.members
.iter()
.find(|member| member.session_id == session_id)
.expect("member")
.handle
.clone()
};
let post = store
.post(
"leader",
format!(
"@{} @{} please review",
handle("observer"),
handle("reviewer")
),
)
.await
.expect("post");
for _ in 0..3 {
deliveries.recv().await.expect("initial delivery signal");
}
store
.leave(&swarm.id, "reviewer")
.await
.expect("leave swarm");
store.disband(&swarm.id).await.expect("disband swarm");
assert_eq!(
[deliveries.recv().await, deliveries.recv().await],
[
Some(SwarmDelivery::Acknowledged {
target_session_id: "reviewer".into(),
message_id: post.entry.id.clone(),
}),
Some(SwarmDelivery::Acknowledged {
target_session_id: "observer".into(),
message_id: post.entry.id,
}),
]
);
}
#[tokio::test]
async fn acknowledgement_is_idempotent_after_its_board_is_gone() {
let (_directory, checkpoints, store, _deliveries) = store();
let swarm = create_swarm(&store).await;
let reviewer_handle = swarm
.members
.iter()
.find(|member| member.session_id == "reviewer")
.expect("reviewer")
.handle
.clone();
let post = store
.post("leader", format!("@{reviewer_handle} review this"))
.await
.expect("post");
store.disband(&swarm.id).await.expect("disband");
let (reloaded, mut deliveries) = SwarmStore::new(checkpoints);
let outcome = reloaded
.acknowledge(&post.entry.id, "reviewer")
.await
.expect("acknowledge removed board");
assert_eq!(outcome, AcknowledgeOutcome::MessageGone);
assert_eq!(
deliveries.recv().await,
Some(SwarmDelivery::Acknowledged {
target_session_id: "reviewer".into(),
message_id: post.entry.id,
})
);
}
#[tokio::test]
async fn acknowledgement_rejects_a_non_recipient_of_a_retained_message() {
let (_directory, _checkpoints, store, _deliveries) = store();
create_swarm(&store).await;
let post = store
.post("leader", "Board-only update".into())
.await
.expect("post");
let error = store
.acknowledge(&post.entry.id, "reviewer")
.await
.expect_err("non-recipient acknowledgement");
assert!(error.to_string().contains("is not a recipient"));
}
#[tokio::test]
async fn board_only_posts_signal_catalog_changes_without_peer_delivery() {
let (_directory, _checkpoints, store, mut deliveries) = store();
create_swarm(&store).await;
store
.post("leader", "Shared status update".into())
.await
.expect("board-only post");
assert_eq!(deliveries.recv().await, Some(SwarmDelivery::Changed));
assert!(matches!(
deliveries.try_recv(),
Err(mpsc::error::TryRecvError::Empty)
));
}
#[tokio::test]
async fn model_board_read_stays_valid_json_below_the_tool_output_cap() {
let (_directory, _checkpoints, store, _deliveries) = store();
create_swarm(&store).await;
store
.post("leader", "\0".repeat(MAX_MESSAGE_BYTES))
.await
.expect("escape-heavy post");
let output = SwarmBackend::read(&store, "leader")
.await
.expect("model board read");
let output: serde_json::Value = serde_json::from_str(&output).expect("valid JSON output");
assert!(serde_json::to_vec(&output).expect("encoded output").len() <= MAX_TOOL_READ_BYTES);
assert_eq!(output["entries"][0]["text_truncated"], true);
assert_eq!(output["has_older"], false);
}
#[tokio::test]
async fn retention_never_evicts_pending_entries() {
let (_directory, _checkpoints, store, _deliveries) = store();
let swarm = create_swarm(&store).await;
let reviewer_handle = swarm
.members
.iter()
.find(|member| member.session_id == "reviewer")
.expect("reviewer")
.handle
.clone();
let pending = store
.post("leader", format!("Please check @{reviewer_handle}"))
.await
.expect("pending post");
for sequence in 0..=MAX_ACKNOWLEDGED_ENTRIES {
store
.post("leader", format!("board update {sequence}"))
.await
.expect("board post");
}
let page = store
.board_page(&swarm.id, None, MAX_PAGE_ENTRIES)
.await
.expect("board page");
assert_eq!(page.entries.len(), MAX_ACKNOWLEDGED_ENTRIES);
assert!(page.next_before_sequence.is_some());
let pending_delivery = store
.pending_deliveries("reviewer")
.await
.expect("pending delivery");
assert_eq!(pending_delivery.len(), 1);
assert_eq!(pending_delivery[0].entry.id, pending.entry.id);
}
#[tokio::test]
async fn posting_backpressures_a_recipient_with_a_full_pending_queue() {
let (_directory, _checkpoints, store, _deliveries) = store();
let swarm = create_swarm(&store).await;
let reviewer_handle = swarm
.members
.iter()
.find(|member| member.session_id == "reviewer")
.expect("reviewer")
.handle
.clone();
for sequence in 0..MAX_PENDING_DELIVERIES_PER_RECIPIENT {
store
.post("leader", format!("@{reviewer_handle} review {sequence}"))
.await
.expect("pending post within limit");
}
let error = store
.post("leader", format!("@{reviewer_handle} one too many"))
.await
.expect_err("pending delivery limit");
assert!(error.to_string().contains("pending swarm messages"));
assert_eq!(
store
.pending_deliveries("reviewer")
.await
.expect("pending deliveries")
.len(),
MAX_PENDING_DELIVERIES_PER_RECIPIENT
);
}
#[test]
fn catalog_encoded_size_is_bounded_before_persistence_or_broadcast() {
let now = unix_ms();
let author = SwarmMember {
session_id: "leader".into(),
handle: "agent_leader".into(),
joined_at_ms: now,
};
let board = (1..=64)
.map(|sequence| BoardEntry {
id: Uuid::new_v4().to_string(),
sequence,
created_at_ms: now,
author: author.clone(),
text: "\0".repeat(MAX_MESSAGE_BYTES),
mentioned_recipient_session_ids: Vec::new(),
pending_recipient_session_ids: Vec::new(),
})
.collect();
let catalog = Catalog {
swarms: BTreeMap::from([(
Uuid::new_v4().to_string(),
StoredSwarm {
title: "Bounded swarm".into(),
leader_session_id: author.session_id.clone(),
members: BTreeMap::from([(author.handle.clone(), author)]),
latest_sequence: 64,
board,
created_at_ms: now,
updated_at_ms: now,
},
)]),
};
assert!(
validate_catalog(&catalog)
.expect_err("oversized catalog")
.to_string()
.contains("encoded bytes")
);
}
#[test]
fn mention_parser_ignores_email_boundaries_and_deduplicates() {
assert_eq!(
mentioned_handles("@one mail@two @one; @three"),
BTreeSet::from(["one".into(), "three".into()])
);
}
}