//! Inter-agent **mailbox** port (Option 1 « Terminal + MCP », lot B-1). //! //! A delegating agent calls `idea_ask_agent(target, task)` and **blocks** until the //! target renders a result via `idea_reply(result)`. Between those two MCP calls, //! IdeA must (1) hand the task to the target's single FIFO input and (2) hold the //! caller's await on a reply slot that the target's later `idea_reply` resolves. //! //! This module owns the **pure** contract of that rendezvous: the [`AgentMailbox`] //! port plus its value objects ([`Ticket`], [`TicketId`], [`MailboxError`]) and the //! [`PendingReply`] handle the caller awaits. It is **I/O-free**: the actual queue, //! its mutex and the one-shot reply channel are an infrastructure concern (the //! `InMemoryMailbox` adapter). Keeping the contract here means the FIFO/resolution //! invariants are expressed against a port the application layer depends on, never //! against a concrete channel type (hexagonal + DIP). //! //! ## Model //! //! - **One queue per target agent** (`AgentId`). `enqueue` appends a [`Ticket`] and //! returns a [`PendingReply`] the caller awaits. //! - **`resolve(agent, result)` corrèle implicitement** the result with the ticket //! at the **head** of that agent's queue — the agent is processing one task at a //! time (1 agent = 1 employee, FIFO input), so its current `idea_reply` answers //! the task it is currently working on. No ticket id is exposed to the model. //! - **Timeout** is the caller's concern: it drops its [`PendingReply`] and calls //! [`AgentMailbox::cancel_head`] to retire the stuck head ticket so the queue //! advances. use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; use crate::conversation::ConversationId; use crate::ids::AgentId; use crate::input::InputSource; /// A read-only, cloned view of one queued [`Ticket`] in a target agent's FIFO. /// /// Carries **only the data** of a ticket (never the one-shot reply sender, which /// stays inside the adapter), plus its `position` in the FIFO at snapshot time /// (`0` = head). Pure value object used by the [`AgentQueueSnapshot`] read port so /// the work-state read model can list pending delegations without touching the /// mutating [`AgentMailbox`] surface (ISP). #[derive(Debug, Clone, PartialEq, Eq)] pub struct QueuedTicketSnapshot { /// Stable id of the queued ticket. pub id: TicketId, /// The origin of the input (Human or a delegating Agent). pub source: InputSource, /// The conversation thread this task enters. pub conversation: ConversationId, /// Display name (or id) of the requester. pub requester: String, /// The full task/message text (preview/truncation is the caller's concern). pub task: String, /// Position in the FIFO at snapshot time (`0` = head, recomputed each call). pub position: u32, } /// Read-only inspection port over the per-agent delegation FIFO. /// /// Segregated from the mutating [`AgentMailbox`] (Interface Segregation): the /// work-state read model depends on this port to *observe* the queue, never on /// `enqueue`/`resolve`/`cancel`. Implementations return **cloned** snapshots in /// FIFO order with recomputed positions; observing the queue must never mutate it. /// /// Object-safe (`&self`) so the application layer holds it as /// `Arc`; one concrete `InMemoryMailbox` can be shared as /// both an [`AgentMailbox`] (mutation) and an [`AgentQueueSnapshot`] (read) view. pub trait AgentQueueSnapshot: Send + Sync { /// Returns a cloned, FIFO-ordered snapshot of `agent`'s queued tickets. /// /// An agent with no queue yields an empty `Vec`. Positions are recomputed from /// the current order (`0` = head). Pure read: the queue is left untouched. fn queue_for(&self, agent: AgentId) -> Vec; } /// Identifies one queued [`Ticket`] within a target agent's mailbox. /// /// Newtype around [`uuid::Uuid`]; minted by the adapter on `enqueue`. It is **never /// exposed to the model** (the MCP `idea_reply` schema carries only `result`): /// correlation is positional (head of the FIFO), the id only lets the caller name /// the exact ticket it wants retired on timeout ([`AgentMailbox::cancel_head`]). #[derive( Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, serde::Serialize, serde::Deserialize, )] #[serde(transparent)] pub struct TicketId(pub uuid::Uuid); impl TicketId { /// Wraps an existing [`uuid::Uuid`]. #[must_use] pub const fn from_uuid(id: uuid::Uuid) -> Self { Self(id) } /// Mints a fresh random ticket id. #[must_use] pub fn new_random() -> Self { Self(uuid::Uuid::new_v4()) } /// Returns the inner [`uuid::Uuid`]. #[must_use] pub const fn as_uuid(&self) -> uuid::Uuid { self.0 } } impl std::fmt::Display for TicketId { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}", self.0) } } /// A delegation request queued for a target agent: who asked, and what for. /// /// The `id` lets the caller retire **this** ticket on timeout; `requester` and /// `task` are carried so the application layer can prefix the task with the asking /// agent's identity when it writes to the target's input (`[IdeA · tâche de A · /// ticket …]`). Pure value object (no behaviour, no I/O). /// /// **Origin & target** (cadrage C1): a ticket also carries its [`InputSource`] /// (Human or Agent — the *source of truth* for the requester, the `requester` /// string being a derived display label) and the [`ConversationId`] of the thread /// it enters. These are added through **additive** constructors ([`Ticket::from_human`], /// [`Ticket::from_agent`]); [`Ticket::new`] keeps its signature (Open/Closed) and /// defaults to a `Human` source with a nil conversation for back-compat. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Ticket { /// Stable id of this queued request (minted by the adapter on `enqueue`). pub id: TicketId, /// The origin of this input (Human or a delegating Agent). pub source: InputSource, /// The conversation thread this task enters. pub conversation: ConversationId, /// Display name (or id) of the agent that issued the `idea_ask_agent`. pub requester: String, /// The task/message to deliver to the target agent. pub task: String, } impl Ticket { /// Builds a ticket from its parts (back-compat: `Human` source, nil conversation). /// /// Preserved for existing call sites. Prefer [`Ticket::from_human`] / /// [`Ticket::from_agent`] when the source and conversation are known. #[must_use] pub fn new(id: TicketId, requester: impl Into, task: impl Into) -> Self { Self { id, source: InputSource::Human, conversation: ConversationId::from_uuid(uuid::Uuid::nil()), requester: requester.into(), task: task.into(), } } /// Builds a ticket originating from the **human** operator, for a given thread. #[must_use] pub fn from_human( id: TicketId, conversation: ConversationId, requester: impl Into, task: impl Into, ) -> Self { Self { id, source: InputSource::Human, conversation, requester: requester.into(), task: task.into(), } } /// Builds a ticket originating from a delegating **agent**, for a given thread. #[must_use] pub fn from_agent( id: TicketId, source: AgentId, conversation: ConversationId, requester: impl Into, task: impl Into, ) -> Self { Self { id, source: InputSource::agent(source), conversation, requester: requester.into(), task: task.into(), } } } /// Errors raised by the [`AgentMailbox`] port. #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] pub enum MailboxError { /// `resolve` was called for an agent whose queue is **empty** — an /// `idea_reply` with no matching `idea_ask_agent` in flight. Surfaced as a typed /// error (never a panic) so the offending agent can read it and adjust. #[error("no pending request to reply to for agent {0}")] NoPendingRequest(AgentId), /// The awaited reply channel closed before a result arrived — the target's /// session ended (or was killed) without rendering an `idea_reply`. The caller's /// await resolves to this instead of hanging forever. #[error("reply channel closed before a result was rendered")] Cancelled, } /// A handle the caller awaits to receive a target agent's reply. /// /// Returned by [`AgentMailbox::enqueue`]. Awaiting it yields the `String` result a /// later [`AgentMailbox::resolve`] feeds into this ticket's slot, or /// [`MailboxError::Cancelled`] if the reply channel closes first. The future is /// **opaque** (the adapter builds it from its one-shot receiver): the domain stays /// free of any concrete channel type, and the await point lives in the application /// layer's `ask_agent`. pub struct PendingReply { inner: Pin> + Send>>, } impl PendingReply { /// Wraps a reply future built by the adapter (e.g. over a one-shot receiver). #[must_use] pub fn new(inner: Pin> + Send>>) -> Self { Self { inner } } } impl Future for PendingReply { type Output = Result; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { self.inner.as_mut().poll(cx) } } impl std::fmt::Debug for PendingReply { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("PendingReply").finish_non_exhaustive() } } /// The inter-agent rendezvous port: a per-agent FIFO of delegation tickets, each /// awaiting a one-shot reply (Option 1, lot B-1). /// /// **Per-agent FIFO** (1 agent = 1 employee): tickets for the **same** target are /// served head-first; tickets for **different** targets never block one another. /// `enqueue` returns the [`PendingReply`] the caller awaits; the target's later /// `idea_reply` lands in [`AgentMailbox::resolve`], which corrèle it with the head /// of that agent's queue. /// /// Kept object-safe (`&self`, owned/`Copy` args) so the application layer holds it /// as `Arc`; the implementation owns all interior mutability. pub trait AgentMailbox: Send + Sync { /// Appends `ticket` to `agent`'s FIFO and returns the handle to await its reply. /// /// The returned [`PendingReply`] resolves when a later [`AgentMailbox::resolve`] /// feeds a result to this ticket (once it reaches the head and is answered), or /// to [`MailboxError::Cancelled`] if the reply channel closes first. fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply; /// Resolves the request at the **head** of `agent`'s FIFO with `result`, /// waking its awaiting [`PendingReply`] and removing it from the queue. /// /// Positional correlation: the agent processes one task at a time, so its /// current `idea_reply` answers the head ticket. /// /// # Errors /// [`MailboxError::NoPendingRequest`] when `agent` has no queued ticket (an /// `idea_reply` with no matching ask in flight). fn resolve(&self, agent: AgentId, result: String) -> Result<(), MailboxError>; /// Resolves the request identified by `ticket_id` **anywhere** in `agent`'s FIFO /// with `result`, waking its awaiting [`PendingReply`] and removing it from the /// queue (cadrage C3 §3.3 — deterministic, multi-thread correlation). /// /// This is the **by-ticket** counterpart of [`AgentMailbox::resolve`]: when an /// agent answers a delegation it echoes the ticket id from the `[IdeA · … · /// ticket ]` prefix, so IdeA correlates exactly that request even when the /// agent has several threads in flight. /// /// The default implementation falls back to [`AgentMailbox::resolve`] (head of /// queue), so a mailbox that does not track ids stays correct for mono-thread /// agents. The in-memory adapter overrides it with true id-keyed resolution. /// /// # Errors /// [`MailboxError::NoPendingRequest`] when no queued ticket matches `ticket_id` /// (and, for the default, when `agent`'s queue is empty). fn resolve_ticket( &self, agent: AgentId, _ticket_id: TicketId, result: String, ) -> Result<(), MailboxError> { self.resolve(agent, result) } /// Retires the ticket `ticket_id` **iff** it is currently at the head of /// `agent`'s FIFO (the caller timed out waiting for it), so the queue advances. /// /// A no-op when the head is a different ticket (the timed-out one was already /// resolved, or another caller's ticket is now in front) — idempotent and safe. fn cancel_head(&self, agent: AgentId, ticket_id: TicketId); } #[cfg(test)] mod tests { use super::*; #[test] fn ticket_carries_its_parts() { let id = TicketId::new_random(); let t = Ticket::new(id, "Main", "do X"); assert_eq!(t.id, id); assert_eq!(t.requester, "Main"); assert_eq!(t.task, "do X"); // Back-compat default: human source, nil conversation. assert_eq!(t.source, InputSource::Human); assert_eq!(t.conversation, ConversationId::from_uuid(uuid::Uuid::nil())); } #[test] fn from_human_carries_source_and_conversation() { let conv = ConversationId::from_uuid(uuid::Uuid::from_u128(5)); let t = Ticket::from_human(TicketId::new_random(), conv, "User", "do X"); assert_eq!(t.source, InputSource::Human); assert_eq!(t.conversation, conv); assert_eq!(t.task, "do X"); } #[test] fn from_agent_carries_agent_source_and_conversation() { let conv = ConversationId::from_uuid(uuid::Uuid::from_u128(6)); let from = AgentId::from_uuid(uuid::Uuid::from_u128(2)); let t = Ticket::from_agent(TicketId::new_random(), from, conv, "A", "delegate"); assert_eq!(t.source, InputSource::agent(from)); assert_eq!(t.source.as_agent(), Some(from)); assert_eq!(t.conversation, conv); } #[test] fn ticket_id_roundtrips_through_uuid() { let u = uuid::Uuid::from_u128(7); let id = TicketId::from_uuid(u); assert_eq!(id.as_uuid(), u); assert_eq!(id.to_string(), u.to_string()); } #[test] fn mailbox_errors_are_distinct_and_typed() { let a = AgentId::from_uuid(uuid::Uuid::from_u128(1)); assert_ne!(MailboxError::NoPendingRequest(a), MailboxError::Cancelled); } #[test] fn queued_ticket_snapshot_carries_ticket_data_and_position() { let conv = ConversationId::from_uuid(uuid::Uuid::from_u128(3)); let from = AgentId::from_uuid(uuid::Uuid::from_u128(4)); let snap = QueuedTicketSnapshot { id: TicketId::from_uuid(uuid::Uuid::from_u128(8)), source: InputSource::agent(from), conversation: conv, requester: "Main".to_owned(), task: "delegate".to_owned(), position: 2, }; assert_eq!(snap.source.as_agent(), Some(from)); assert_eq!(snap.conversation, conv); assert_eq!(snap.position, 2); assert_eq!(snap.task, "delegate"); } }