IdeA/crates/infrastructure/tests/agent_inbox.rs

//! 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<_>>(),
        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<dyn AgentInbox>);

    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();
}