//! B4 tests for the bounded AgentInbox facade over MediatedInbox. use std::sync::Arc; use std::time::Duration; use domain::{ AgentId, AgentInbox, InboxError, InboxItem, InboxItemKind, InboxReceiptStatus, InboxSource, InputMediator, Ticket, TicketId, }; use infrastructure::{ start_background_ready_inbox_bridge, BackgroundTaskReadyToDeliver, InMemoryMailbox, MediatedInbox, MillisClock, }; use tokio::sync::mpsc::unbounded_channel; use uuid::Uuid; struct FixedClock(u64); impl MillisClock for FixedClock { fn now_ms(&self) -> u64 { self.0 } } fn agent(n: u128) -> AgentId { AgentId::from_uuid(Uuid::from_u128(n)) } fn ticket_id(n: u128) -> TicketId { TicketId::from_uuid(Uuid::from_u128(n)) } fn task_id(n: u128) -> domain::TaskId { domain::TaskId::from_uuid(Uuid::from_u128(n)) } fn inbox(capacity: usize) -> MediatedInbox { MediatedInbox::new( Arc::new(InMemoryMailbox::new()), Arc::new(FixedClock(1_000)), ) .with_inbox_capacity(capacity) } fn user_item(id: u128, target: AgentId, body: &str) -> InboxItem { InboxItem { id: ticket_id(id), agent_id: target, source: InboxSource::Human, kind: InboxItemKind::UserMessage, body: body.into(), created_at_ms: 1_000, correlation_id: None, } } fn completion_item(id: u128, target: AgentId, task: u128) -> InboxItem { InboxItem { id: ticket_id(id), agent_id: target, source: InboxSource::BackgroundTask { task_id: task_id(task), }, kind: InboxItemKind::BackgroundCompletion, body: "done".into(), created_at_ms: 1_000, correlation_id: Some(task_id(task).to_string()), } } fn make_busy(inbox: &MediatedInbox, target: AgentId) { let ticket = Ticket::new(ticket_id(900), "active", "active turn"); let _pending = inbox.enqueue(target, ticket); } #[test] fn enqueue_message_while_busy_is_queued_not_rejected() { let inbox = inbox(10); let target = agent(1); make_busy(&inbox, target); let receipt = inbox .enqueue_message(target, user_item(1, target, "queued")) .unwrap(); assert_eq!(receipt.status, InboxReceiptStatus::Queued); assert_eq!(receipt.depth, 2); let snapshot = inbox.snapshot(target); assert_eq!(snapshot.depth, 2); assert_eq!(snapshot.items.len(), 1); assert_eq!(snapshot.items[0].body, "queued"); } #[test] fn inbox_fifo_drains_two_entrants_in_order() { let inbox = inbox(10); let target = agent(1); let first = user_item(1, target, "first"); let second = user_item(2, target, "second"); inbox.enqueue_message(target, first.clone()).unwrap(); inbox.enqueue_message(target, second.clone()).unwrap(); assert_eq!(inbox.dequeue_next(target), Some(first)); assert_eq!(inbox.dequeue_next(target), Some(second)); assert_eq!(inbox.dequeue_next(target), None); } #[test] fn overflow_normal_message_returns_inbox_full() { let inbox = inbox(1); let target = agent(1); make_busy(&inbox, target); let err = inbox .enqueue_message(target, user_item(1, target, "overflow")) .unwrap_err(); assert_eq!( err, InboxError::InboxFull { agent_id: target, capacity: 1, } ); } #[test] fn overflow_completion_is_deferred_not_lost_or_enqueued() { let inbox = inbox(1); let target = agent(1); make_busy(&inbox, target); let receipt = inbox .enqueue_message(target, completion_item(1, target, 42)) .unwrap(); assert_eq!(receipt.status, InboxReceiptStatus::Deferred); assert_eq!(receipt.depth, 1); assert!(inbox.snapshot(target).items.is_empty()); } #[test] fn snapshot_exposes_queue_depth_and_items() { let inbox = inbox(10); let target = agent(1); inbox .enqueue_message(target, user_item(1, target, "first")) .unwrap(); inbox .enqueue_message(target, user_item(2, target, "second")) .unwrap(); let snapshot = inbox.snapshot(target); assert_eq!(snapshot.agent_id, target); assert_eq!(snapshot.depth, 2); assert_eq!( snapshot .items .iter() .map(|item| item.body.as_str()) .collect::>(), vec!["first", "second"] ); } #[test] fn drain_removes_only_one_item_at_a_time() { let inbox = inbox(10); let target = agent(1); inbox .enqueue_message(target, user_item(1, target, "first")) .unwrap(); inbox .enqueue_message(target, user_item(2, target, "second")) .unwrap(); assert_eq!(inbox.dequeue_next(target).unwrap().body, "first"); let snapshot = inbox.snapshot(target); assert_eq!(snapshot.depth, 1); assert_eq!(snapshot.items[0].body, "second"); } #[tokio::test] async fn ready_bridge_enqueues_background_completion_item() { let inbox = Arc::new(inbox(10)); let target = agent(1); let (tx, rx) = unbounded_channel(); let handle = start_background_ready_inbox_bridge(rx, inbox.clone() as Arc); tx.send(BackgroundTaskReadyToDeliver { task_id: task_id(42), project_id: domain::ProjectId::from_uuid(Uuid::from_u128(7)), owner_agent_id: target, }) .unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; let snapshot = inbox.snapshot(target); assert_eq!(snapshot.depth, 1); assert_eq!(snapshot.items[0].kind, InboxItemKind::BackgroundCompletion); assert_eq!( snapshot.items[0].source, InboxSource::BackgroundTask { task_id: task_id(42) } ); drop(tx); handle.await.unwrap(); }