clawft_kernel/

a2a.rs

//! Agent-to-agent IPC protocol.
//!
//! The [`A2ARouter`] provides direct PID-to-PID messaging with
//! capability-checked routing, per-agent inboxes, and request-response
//! patterns with timeout support. It integrates with the
//! [`TopicRouter`] for pub/sub delivery.
//!
//! # Message Flow
//!
//! ```text
//! Agent A (PID 1)       A2ARouter          Agent B (PID 7)
//!      |                    |                    |
//!      |-- send(msg) ------>|                    |
//!      |                    |-- check_scope ---->|
//!      |                    |<-- Ok -------------|
//!      |                    |-- inbox.send(7) -->|
//!      |                    |                    |-- recv msg
//! ```

use std::sync::Arc;
use std::time::{Duration, Instant};

use dashmap::DashMap;
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, warn};

use crate::capability::CapabilityChecker;
use crate::error::{KernelError, KernelResult};
use crate::ipc::{KernelMessage, MessageTarget};
use crate::process::{Pid, ProcessState, ProcessTable};
use crate::service::ServiceRegistry;
use crate::topic::TopicRouter;

#[cfg(feature = "exochain")]
use crate::chain::ChainManager;

#[cfg(feature = "mesh")]
use crate::mesh_ipc::MeshIpcEnvelope;
#[cfg(feature = "mesh")]
use crate::mesh_runtime::MeshRuntime;

/// Default inbox channel capacity per agent.
const DEFAULT_INBOX_CAPACITY: usize = 1024;

/// Maximum serialized message size (16 MiB) -- prevents DoS via oversized payloads.
///
/// This mirrors [`crate::mesh::MAX_MESSAGE_SIZE`]. Enforcement happens at the
/// mesh boundary in `mesh_framing.rs` where raw bytes are received from remote
/// nodes. Within a single kernel, messages travel as typed `KernelMessage`
/// structs over `mpsc` channels, so the size limit is not checked on the
/// in-process hot path.
#[allow(dead_code)]
const MAX_A2A_MESSAGE_SIZE: usize = 16 * 1024 * 1024;

/// A pending request awaiting a correlated response.
struct PendingRequest {
    /// Sender to deliver the response on.
    response_tx: oneshot::Sender<KernelMessage>,
    /// When the request was sent (for timeout tracking / diagnostics).
    #[allow(dead_code)]
    sent_at: Instant,
}

/// Agent-to-agent message router.
///
/// Manages per-agent inboxes (bounded `mpsc` channels), validates
/// IPC scope through the capability checker, and routes messages
/// to their targets (direct PID, topic, broadcast, service).
pub struct A2ARouter {
    /// Process table for state validation.
    process_table: Arc<ProcessTable>,

    /// Capability checker for IPC scope enforcement.
    capability_checker: Arc<CapabilityChecker>,

    /// Topic router for pub/sub delivery.
    topic_router: Arc<TopicRouter>,

    /// Service registry for service-based routing (D1, D19, K2.1).
    service_registry: Option<Arc<ServiceRegistry>>,

    /// Per-agent inboxes: PID -> sender half of inbox channel.
    inboxes: DashMap<Pid, mpsc::Sender<KernelMessage>>,

    /// Pending request-response tracking: request_id -> PendingRequest.
    pending_requests: DashMap<String, PendingRequest>,

    /// Optional routing-time gate backend (C4 dual-layer governance).
    ///
    /// Uses `OnceLock` to support post-construction wiring from boot
    /// (the governance gate is created after the A2ARouter is wrapped
    /// in `Arc`). `with_gate()` and `set_gate()` both target this field.
    #[cfg(feature = "exochain")]
    gate: std::sync::OnceLock<Arc<dyn crate::gate::GateBackend>>,

    /// Dead letter queue for undeliverable messages (os-patterns).
    #[cfg(feature = "os-patterns")]
    dead_letter_queue: std::sync::OnceLock<Arc<crate::dead_letter::DeadLetterQueue>>,

    /// Optional mesh runtime for cross-node message delivery (K6).
    #[cfg(feature = "mesh")]
    mesh_runtime: std::sync::OnceLock<Arc<MeshRuntime>>,
}

impl A2ARouter {
    /// Create a new A2A router.
    pub fn new(
        process_table: Arc<ProcessTable>,
        capability_checker: Arc<CapabilityChecker>,
        topic_router: Arc<TopicRouter>,
    ) -> Self {
        Self {
            process_table,
            capability_checker,
            topic_router,
            service_registry: None,
            inboxes: DashMap::new(),
            pending_requests: DashMap::new(),
            #[cfg(feature = "exochain")]
            gate: std::sync::OnceLock::new(),
            #[cfg(feature = "os-patterns")]
            dead_letter_queue: std::sync::OnceLock::new(),
            #[cfg(feature = "mesh")]
            mesh_runtime: std::sync::OnceLock::new(),
        }
    }

    /// Attach a service registry for service-based routing (D1, D19).
    pub fn with_service_registry(mut self, registry: Arc<ServiceRegistry>) -> Self {
        self.service_registry = Some(registry);
        self
    }

    /// Attach a routing-time gate for dual-layer governance (C4).
    ///
    /// When set, every message routed through `send()` is checked against
    /// the gate *before* inbox delivery. A `Deny` decision blocks the
    /// message; `Defer` still delivers (the handler-time gate decides).
    #[cfg(feature = "exochain")]
    pub fn with_gate(self, gate: Arc<dyn crate::gate::GateBackend>) -> Self {
        let _ = self.gate.set(gate);
        self
    }

    /// Set the routing-time gate after construction (for boot wiring).
    ///
    /// This allows the governance gate to be attached after the router
    /// is already wrapped in an `Arc`, since the `OnceLock` provides
    /// interior mutability for the first (and only) write.
    #[cfg(feature = "exochain")]
    pub fn set_gate(&self, gate: Arc<dyn crate::gate::GateBackend>) {
        let _ = self.gate.set(gate);
    }

    /// Set the dead letter queue after construction (for boot wiring).
    ///
    /// Like `set_gate()`, uses `OnceLock` for interior mutability so
    /// the DLQ can be attached after the router is wrapped in `Arc`.
    #[cfg(feature = "os-patterns")]
    pub fn set_dead_letter_queue(&self, dlq: Arc<crate::dead_letter::DeadLetterQueue>) {
        let _ = self.dead_letter_queue.set(dlq);
    }

    /// Get the dead letter queue (if configured).
    #[cfg(feature = "os-patterns")]
    pub fn dead_letter_queue(&self) -> Option<&Arc<crate::dead_letter::DeadLetterQueue>> {
        self.dead_letter_queue.get()
    }

    /// Attach the mesh runtime for cross-node message delivery (K6).
    ///
    /// Uses `OnceLock` so the runtime can be attached after the router
    /// is already wrapped in `Arc`.
    #[cfg(feature = "mesh")]
    pub fn set_mesh_runtime(&self, runtime: Arc<MeshRuntime>) {
        let _ = self.mesh_runtime.set(runtime);
    }

    /// Get the mesh runtime (if configured).
    #[cfg(feature = "mesh")]
    pub fn mesh_runtime(&self) -> Option<&Arc<MeshRuntime>> {
        self.mesh_runtime.get()
    }

    /// Get the service registry (if configured).
    pub fn service_registry(&self) -> Option<&Arc<ServiceRegistry>> {
        self.service_registry.as_ref()
    }

    /// Create an inbox for a process.
    ///
    /// Returns the receiver half that the agent should poll for
    /// incoming messages. The sender half is stored internally
    /// for routing.
    ///
    /// If an inbox already exists for this PID, the old one is
    /// replaced (existing messages are lost).
    pub fn create_inbox(&self, pid: Pid) -> mpsc::Receiver<KernelMessage> {
        let (tx, rx) = mpsc::channel(DEFAULT_INBOX_CAPACITY);
        self.inboxes.insert(pid, tx);
        debug!(pid, "created inbox");
        rx
    }

    /// Remove an inbox (used during process cleanup).
    pub fn remove_inbox(&self, pid: Pid) {
        self.inboxes.remove(&pid);
        debug!(pid, "removed inbox");
    }

    /// Send a message, routing it to the appropriate target.
    ///
    /// Validates that the sender exists and is running, checks
    /// IPC scope via the capability checker, then delivers the
    /// message to the target.
    ///
    /// # Routing
    ///
    /// - `Process(pid)`: delivers directly to the target's inbox
    /// - `Topic(name)`: publishes to all topic subscribers
    /// - `Broadcast`: delivers to all inboxes except the sender
    /// - `Service(name)`: logs a warning (service routing is a
    ///   future extension)
    /// - `Kernel`: logs a warning (kernel messages are internal)
    ///
    /// # Errors
    ///
    /// Returns `KernelError::ProcessNotFound` if the sender PID
    /// is not in the process table, or `KernelError::CapabilityDenied`
    /// if the sender's IPC scope does not permit the target.
    pub async fn send(&self, msg: KernelMessage) -> KernelResult<()> {
        let from = msg.from;

        // Validate sender exists and is running
        let sender = self
            .process_table
            .get(from)
            .ok_or(KernelError::ProcessNotFound { pid: from })?;

        if !matches!(sender.state, ProcessState::Running | ProcessState::Suspended) {
            return Err(KernelError::Ipc(format!(
                "sender PID {from} is not running (state: {})",
                sender.state
            )));
        }

        // C4: Routing-time gate check (first layer of dual-layer governance).
        // A Deny blocks the message before it reaches any inbox. Defer
        // still delivers — the handler-time gate makes the final call.
        #[cfg(feature = "exochain")]
        if let Some(gate) = self.gate.get() {
            let action = match &msg.payload {
                crate::ipc::MessagePayload::ToolCall { name, .. } => format!("tool.{name}"),
                crate::ipc::MessagePayload::Signal(_) => "ipc.signal".to_string(),
                _ => "ipc.send".to_string(),
            };
            let context = serde_json::json!({
                "from": from,
                "target": format!("{:?}", msg.target),
                "layer": "routing",
            });
            match gate.check(&from.to_string(), &action, &context) {
                crate::gate::GateDecision::Deny { reason, .. } => {
                    return Err(KernelError::CapabilityDenied {
                        pid: from,
                        action,
                        reason: format!("routing gate denied: {reason}"),
                    });
                }
                crate::gate::GateDecision::Defer { .. }
                | crate::gate::GateDecision::Permit { .. } => {
                    // Permitted or deferred — continue to delivery.
                }
            }
        }

        // Route based on target
        match &msg.target {
            MessageTarget::Process(target_pid) => {
                // Check IPC scope
                self.capability_checker
                    .check_ipc_target(from, *target_pid)?;

                self.deliver_to_inbox(*target_pid, msg).await
            }
            MessageTarget::Topic(topic) => {
                let subscribers = self.topic_router.live_subscribers(topic);
                let mut delivered = 0u32;
                for &sub_pid in &subscribers {
                    if sub_pid != from {
                        let msg_clone = msg.clone();
                        if self.deliver_to_inbox(sub_pid, msg_clone).await.is_ok() {
                            delivered += 1;
                        }
                    }
                }
                debug!(from, topic, delivered, "published to topic");
                Ok(())
            }
            MessageTarget::Broadcast => {
                let mut delivered = 0u32;
                let pids: Vec<Pid> = self.inboxes.iter().map(|entry| *entry.key()).collect();

                for pid in pids {
                    if pid != from {
                        // Check IPC scope for each target
                        if self.capability_checker.check_ipc_target(from, pid).is_ok() {
                            let msg_clone = msg.clone();
                            if self.deliver_to_inbox(pid, msg_clone).await.is_ok() {
                                delivered += 1;
                            }
                        }
                    }
                }
                debug!(from, delivered, "broadcast sent");
                Ok(())
            }
            MessageTarget::Service(name) => {
                let name = name.clone();
                self.route_to_service(from, &name, msg).await
            }
            MessageTarget::ServiceMethod { service, .. } => {
                let service_name = service.clone();
                self.route_to_service(from, &service_name, msg).await
            }
            MessageTarget::Kernel => {
                debug!(from, "kernel message routing not yet implemented");
                Ok(())
            }
            MessageTarget::RemoteNode { node_id, .. } => {
                #[cfg(feature = "mesh")]
                {
                    if let Some(runtime) = self.mesh_runtime.get() {
                        let node_id = node_id.clone();
                        debug!(from, %node_id, "routing message to remote node via mesh");
                        let envelope = MeshIpcEnvelope::new(
                            runtime.node_id().to_string(),
                            node_id.clone(),
                            msg,
                        );
                        runtime
                            .send_to_peer(&node_id, envelope)
                            .await
                            .map_err(|e| KernelError::Mesh(format!(
                                "failed to send to remote node '{node_id}': {e}"
                            )))
                    } else {
                        debug!(from, %node_id, "remote node routing: no mesh runtime attached");
                        Err(KernelError::Mesh(format!(
                            "remote routing to node '{node_id}' not yet implemented"
                        )))
                    }
                }
                #[cfg(not(feature = "mesh"))]
                {
                    debug!(from, %node_id, "remote node routing not available (mesh feature disabled)");
                    Err(KernelError::Mesh(format!(
                        "remote routing to node '{node_id}' not yet implemented"
                    )))
                }
            }
        }
    }

    /// Deliver a message to a specific PID's inbox.
    ///
    /// If the inbox does not exist or is full, the message is routed to
    /// the dead letter queue (when os-patterns is enabled) and an error
    /// is returned.
    async fn deliver_to_inbox(&self, pid: Pid, msg: KernelMessage) -> KernelResult<()> {
        // Clone the sender so we release the DashMap read lock before
        // any potential remove() call (which needs a write lock on the
        // same shard — holding both would deadlock).
        let tx = match self.inboxes.get(&pid) {
            Some(tx) => tx.clone(),
            None => {
                warn!(pid, "no inbox for PID, dead-lettering");
                #[cfg(feature = "os-patterns")]
                if let Some(dlq) = self.dead_letter_queue.get() {
                    dlq.intake(
                        msg,
                        crate::dead_letter::DeadLetterReason::TargetNotFound { pid },
                    );
                }
                return Err(KernelError::Ipc(format!("no inbox for PID {pid}")));
            }
        };

        match tx.try_send(msg) {
            Ok(()) => {
                debug!(pid, "message delivered to inbox");
                Ok(())
            }
            Err(mpsc::error::TrySendError::Full(rejected_msg)) => {
                warn!(pid, "inbox full, dead-lettering");
                #[cfg(feature = "os-patterns")]
                if let Some(dlq) = self.dead_letter_queue.get() {
                    dlq.intake(
                        rejected_msg,
                        crate::dead_letter::DeadLetterReason::InboxFull { pid },
                    );
                }
                Err(KernelError::Ipc(format!("inbox full for PID {pid}")))
            }
            Err(mpsc::error::TrySendError::Closed(rejected_msg)) => {
                warn!(pid, "inbox closed, removing and dead-lettering");
                self.inboxes.remove(&pid);
                #[cfg(feature = "os-patterns")]
                if let Some(dlq) = self.dead_letter_queue.get() {
                    dlq.intake(
                        rejected_msg,
                        crate::dead_letter::DeadLetterReason::AgentExited { pid },
                    );
                }
                Err(KernelError::Ipc(format!("inbox closed for PID {pid}")))
            }
        }
    }

    /// Route a message to a named service via the ServiceRegistry.
    ///
    /// Resolves the service name to an owning agent PID, then delivers
    /// the message to that agent's inbox.
    async fn route_to_service(
        &self,
        from: Pid,
        service_name: &str,
        msg: KernelMessage,
    ) -> KernelResult<()> {
        let registry = self.service_registry.as_ref().ok_or_else(|| {
            KernelError::Ipc(format!(
                "no service registry configured; cannot route to service '{service_name}'"
            ))
        })?;

        let target_pid = registry.resolve_target(service_name).ok_or_else(|| {
            KernelError::Ipc(format!("service not found: '{service_name}'"))
        })?;

        // Check IPC scope
        self.capability_checker
            .check_ipc_target(from, target_pid)?;

        self.deliver_to_inbox(target_pid, msg).await
    }

    /// Send a message with chain-event logging.
    ///
    /// This mirrors `KernelIpc::send_checked` but for the A2ARouter:
    /// every routed message is logged as an `ipc.send` chain event with
    /// sender, target, payload type, and message ID — forming a
    /// tamper-evident IPC audit trail in the exochain.
    ///
    /// When the `exochain` feature is disabled this is equivalent to
    /// a plain `send()`.
    #[cfg(feature = "exochain")]
    pub async fn send_checked(
        &self,
        msg: KernelMessage,
        chain: Option<&ChainManager>,
    ) -> KernelResult<()> {
        // Log the IPC event before delivery so the chain records intent
        // even if the inbox is full or closed.
        if let Some(cm) = chain {
            cm.append(
                "ipc",
                "ipc.send",
                Some(serde_json::json!({
                    "from": msg.from,
                    "target": format!("{:?}", msg.target),
                    "payload_type": msg.payload.type_name(),
                    "msg_id": msg.id,
                })),
            );
        }
        self.send(msg).await
    }

    /// Get the topic router.
    pub fn topic_router(&self) -> &Arc<TopicRouter> {
        &self.topic_router
    }

    /// Get the number of active inboxes.
    pub fn inbox_count(&self) -> usize {
        self.inboxes.len()
    }

    /// Check whether a PID has an inbox.
    pub fn has_inbox(&self, pid: Pid) -> bool {
        self.inboxes.contains_key(&pid)
    }

    /// Send a request and wait for a correlated response with timeout.
    ///
    /// The request message is sent normally, but its `id` is registered
    /// as a pending request. When a response arrives with a matching
    /// `correlation_id`, it is delivered to the returned future instead
    /// of the sender's inbox.
    ///
    /// # Errors
    ///
    /// Returns `KernelError::Timeout` if no response arrives within the
    /// specified duration, or `KernelError::Ipc` if the response channel
    /// is closed before a response arrives.
    pub async fn request(
        &self,
        msg: KernelMessage,
        timeout: Duration,
    ) -> KernelResult<KernelMessage> {
        let request_id = msg.id.clone();
        let (tx, rx) = oneshot::channel();

        // Register pending request before sending so there is no race.
        self.pending_requests.insert(
            request_id.clone(),
            PendingRequest {
                response_tx: tx,
                sent_at: Instant::now(),
            },
        );

        // Send the message; clean up on failure.
        if let Err(e) = self.send(msg).await {
            self.pending_requests.remove(&request_id);
            return Err(e);
        }

        // Wait for response with timeout.
        match tokio::time::timeout(timeout, rx).await {
            Ok(Ok(response)) => Ok(response),
            Ok(Err(_)) => {
                self.pending_requests.remove(&request_id);
                Err(KernelError::Ipc("response channel closed".into()))
            }
            Err(_) => {
                self.pending_requests.remove(&request_id);
                Err(KernelError::Timeout {
                    operation: format!("request {request_id}"),
                    duration_ms: timeout.as_millis() as u64,
                })
            }
        }
    }

    /// Try to complete a pending request with a correlated response.
    ///
    /// If the message has a `correlation_id` that matches a pending
    /// request, the response is delivered to the waiting future and
    /// `true` is returned. Otherwise returns `false`.
    pub fn try_complete_request(&self, msg: KernelMessage) -> bool {
        if let Some(ref corr_id) = msg.correlation_id
            && let Some((_, pending)) = self.pending_requests.remove(corr_id)
        {
            let _ = pending.response_tx.send(msg);
            return true;
        }
        false
    }

    /// Get the number of pending requests.
    pub fn pending_request_count(&self) -> usize {
        self.pending_requests.len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::capability::AgentCapabilities;
    use crate::ipc::MessagePayload;
    use crate::process::{ProcessEntry, ResourceUsage};
    use tokio_util::sync::CancellationToken;

    fn setup_router(
        agent_count: usize,
    ) -> (A2ARouter, Vec<Pid>, Vec<mpsc::Receiver<KernelMessage>>) {
        let table = Arc::new(ProcessTable::new(64));
        let mut pids = Vec::new();

        for i in 0..agent_count {
            let entry = ProcessEntry {
                pid: 0,
                agent_id: format!("agent-{i}"),
                state: ProcessState::Running,
                capabilities: AgentCapabilities::default(),
                resource_usage: ResourceUsage::default(),
                cancel_token: CancellationToken::new(),
                parent_pid: None,
            };
            let pid = table.insert(entry).unwrap();
            pids.push(pid);
        }

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        let mut receivers = Vec::new();
        for &pid in &pids {
            let rx = router.create_inbox(pid);
            receivers.push(rx);
        }

        (router, pids, receivers)
    }

    #[tokio::test]
    async fn direct_message_delivery() {
        let (router, pids, mut receivers) = setup_router(2);

        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "hello");
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert_eq!(received.from, pids[0]);
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "hello"
        ));
    }

    #[tokio::test]
    async fn message_to_self_works() {
        let (router, pids, mut receivers) = setup_router(1);

        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[0]), "self-msg");
        router.send(msg).await.unwrap();

        let received = receivers[0].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "self-msg"
        ));
    }

    #[tokio::test]
    async fn broadcast_delivers_to_all_except_sender() {
        let (router, pids, mut receivers) = setup_router(3);

        let msg = KernelMessage::text(pids[0], MessageTarget::Broadcast, "broadcast");
        router.send(msg).await.unwrap();

        // Sender should not receive
        assert!(receivers[0].try_recv().is_err());

        // Others should receive
        let r1 = receivers[1].try_recv().unwrap();
        assert!(matches!(
            r1.payload,
            MessagePayload::Text(ref t) if t == "broadcast"
        ));
        let r2 = receivers[2].try_recv().unwrap();
        assert!(matches!(
            r2.payload,
            MessagePayload::Text(ref t) if t == "broadcast"
        ));
    }

    #[tokio::test]
    async fn topic_publish_delivers_to_subscribers() {
        let (router, pids, mut receivers) = setup_router(3);

        // Subscribe pids[1] and pids[2] to "build"
        router.topic_router().subscribe(pids[1], "build");
        router.topic_router().subscribe(pids[2], "build");

        let msg = KernelMessage::text(pids[0], MessageTarget::Topic("build".into()), "build done");
        router.send(msg).await.unwrap();

        // Sender not subscribed, should not receive
        assert!(receivers[0].try_recv().is_err());

        // Subscribers should receive
        assert!(receivers[1].try_recv().is_ok());
        assert!(receivers[2].try_recv().is_ok());
    }

    #[tokio::test]
    async fn topic_publish_excludes_sender_if_subscribed() {
        let (router, pids, mut receivers) = setup_router(2);

        // Both subscribe
        router.topic_router().subscribe(pids[0], "build");
        router.topic_router().subscribe(pids[1], "build");

        let msg = KernelMessage::text(pids[0], MessageTarget::Topic("build".into()), "done");
        router.send(msg).await.unwrap();

        // Sender should not receive their own publish
        assert!(receivers[0].try_recv().is_err());
        // Other subscriber should receive
        assert!(receivers[1].try_recv().is_ok());
    }

    #[tokio::test]
    async fn send_from_nonexistent_pid_fails() {
        let (router, _pids, _receivers) = setup_router(1);

        let msg = KernelMessage::text(999, MessageTarget::Process(1), "hello");
        let result = router.send(msg).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn send_to_pid_without_inbox_fails() {
        let table = Arc::new(ProcessTable::new(64));

        // Create sender (running)
        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "sender".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        // Create target (running) but don't create inbox
        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        // Create inbox only for sender
        let _rx = router.create_inbox(sender_pid);

        let msg = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "hello");
        let result = router.send(msg).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn ipc_scope_restricts_messaging() {
        let table = Arc::new(ProcessTable::new(64));

        // Create sender with restricted IPC scope
        use crate::capability::IpcScope;
        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "restricted".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities {
                ipc_scope: IpcScope::Restricted(vec![]), // No allowed PIDs
                ..Default::default()
            },
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        // Create target
        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx1 = router.create_inbox(sender_pid);
        let _rx2 = router.create_inbox(target_pid);

        let msg = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "blocked");
        let result = router.send(msg).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn ipc_scope_none_blocks_all() {
        let table = Arc::new(ProcessTable::new(64));

        use crate::capability::IpcScope;
        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "no-ipc".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities {
                can_ipc: false,
                ipc_scope: IpcScope::None,
                ..Default::default()
            },
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx1 = router.create_inbox(sender_pid);
        let _rx2 = router.create_inbox(target_pid);

        let msg = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "blocked");
        let result = router.send(msg).await;
        assert!(result.is_err());
    }

    #[test]
    fn create_and_remove_inbox() {
        let table = Arc::new(ProcessTable::new(64));
        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        let _rx = router.create_inbox(42);
        assert!(router.has_inbox(42));
        assert_eq!(router.inbox_count(), 1);

        router.remove_inbox(42);
        assert!(!router.has_inbox(42));
        assert_eq!(router.inbox_count(), 0);
    }

    #[tokio::test]
    async fn tool_call_message_routes() {
        let (router, pids, mut receivers) = setup_router(2);

        let msg = KernelMessage::tool_call(
            pids[0],
            MessageTarget::Process(pids[1]),
            "read_file",
            serde_json::json!({"path": "/test"}),
        );
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::ToolCall { ref name, .. } if name == "read_file"
        ));
    }

    #[tokio::test]
    async fn tool_result_message_routes() {
        let (router, pids, mut receivers) = setup_router(2);

        let msg = KernelMessage::tool_result(
            pids[1],
            MessageTarget::Process(pids[0]),
            "call-1",
            serde_json::json!({"content": "data"}),
        );
        router.send(msg).await.unwrap();

        let received = receivers[0].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::ToolResult { ref call_id, .. } if call_id == "call-1"
        ));
    }

    #[cfg(feature = "exochain")]
    #[tokio::test]
    async fn send_checked_logs_chain_event() {
        let (router, pids, mut receivers) = setup_router(2);

        let chain = crate::chain::ChainManager::new(0, 1000);
        let initial_seq = chain.sequence();

        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "audited");
        router.send_checked(msg, Some(&chain)).await.unwrap();

        // Message should still be delivered
        let received = receivers[1].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "audited"
        ));

        // Chain should have a new ipc.send event
        assert_eq!(chain.sequence(), initial_seq + 1);
        let events = chain.tail(1);
        assert_eq!(events[0].kind, "ipc.send");
        assert_eq!(events[0].source, "ipc");
        let payload = events[0].payload.as_ref().unwrap();
        assert_eq!(payload["from"], pids[0]);
        assert_eq!(payload["payload_type"], "text");
    }

    #[cfg(feature = "exochain")]
    #[tokio::test]
    async fn send_checked_without_chain_still_delivers() {
        let (router, pids, mut receivers) = setup_router(2);

        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "no-chain");
        router.send_checked(msg, None).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "no-chain"
        ));
    }

    #[tokio::test]
    async fn rvf_payload_routes() {
        let (router, pids, mut receivers) = setup_router(2);

        let msg = KernelMessage::new(
            pids[0],
            MessageTarget::Process(pids[1]),
            MessagePayload::Rvf {
                segment_type: 0x40,
                data: vec![0xCA, 0xFE],
            },
        );
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Rvf { segment_type: 0x40, .. }
        ));
    }

    #[tokio::test]
    async fn closed_inbox_auto_removed() {
        let (router, pids, receivers) = setup_router(2);

        // Drop receiver for pids[1] — closes the inbox channel
        drop(receivers);

        assert!(router.has_inbox(pids[1]));

        // Sending should fail with "inbox closed" and auto-remove the entry
        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "gone");
        let result = router.send(msg).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(err_msg.contains("inbox closed"), "expected 'inbox closed', got: {err_msg}");

        // Inbox should have been removed automatically
        assert!(!router.has_inbox(pids[1]));
    }

    #[tokio::test]
    async fn inbox_overflow_returns_error() {
        // Create a router with 2 agents, but use a small inbox
        // We can't change DEFAULT_INBOX_CAPACITY, so we fill a normal inbox.
        // Instead, create a custom channel with capacity 2 for a targeted test.
        let table = Arc::new(ProcessTable::new(64));

        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "sender".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        // Create sender inbox normally
        let _rx_sender = router.create_inbox(sender_pid);

        // Manually insert a tiny-capacity channel for target (capacity=2)
        let (tx, _rx_target) = mpsc::channel(2);
        router.inboxes.insert(target_pid, tx);

        // Fill it up
        let m1 = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "msg1");
        let m2 = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "msg2");
        router.send(m1).await.unwrap();
        router.send(m2).await.unwrap();

        // Third message should fail — inbox full
        let m3 = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "overflow");
        let result = router.send(m3).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(err_msg.contains("inbox full"), "expected 'inbox full', got: {err_msg}");

        // Inbox should still exist (not removed on full, only on closed)
        assert!(router.has_inbox(target_pid));
    }

    #[tokio::test]
    async fn concurrent_sends_to_same_pid() {
        let (router, pids, mut receivers) = setup_router(4);
        let router = Arc::new(router);
        let target = pids[0];

        // 3 senders each send a message to the same target concurrently
        let mut handles = Vec::new();
        for &sender_pid in &pids[1..] {
            let r = Arc::clone(&router);
            let msg = KernelMessage::text(
                sender_pid,
                MessageTarget::Process(target),
                &format!("from-{sender_pid}"),
            );
            handles.push(tokio::spawn(async move { r.send(msg).await }));
        }

        // All sends should succeed
        for h in handles {
            h.await.unwrap().unwrap();
        }

        // Target should have received all 3 messages
        let mut received = Vec::new();
        while let Ok(msg) = receivers[0].try_recv() {
            if let MessagePayload::Text(t) = &msg.payload {
                received.push(t.clone());
            }
        }
        assert_eq!(received.len(), 3);
        // All senders represented (order may vary)
        for &sender_pid in &pids[1..] {
            assert!(
                received.iter().any(|t| t == &format!("from-{sender_pid}")),
                "missing message from PID {sender_pid}"
            );
        }
    }

    #[tokio::test]
    async fn send_from_non_running_process_fails() {
        let table = Arc::new(ProcessTable::new(64));

        // Create sender in Exited state
        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "exited-sender".to_owned(),
            state: ProcessState::Exited(0),
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        // Create target in Running state
        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx1 = router.create_inbox(sender_pid);
        let _rx2 = router.create_inbox(target_pid);

        let msg = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "from-dead");
        let result = router.send(msg).await;
        assert!(result.is_err(), "send from non-Running process should fail");
    }

    // ── Service routing tests (K2.1 T3: D19 + D1) ──────────────

    fn setup_router_with_registry(
        agent_count: usize,
    ) -> (
        A2ARouter,
        Vec<Pid>,
        Vec<mpsc::Receiver<KernelMessage>>,
        Arc<crate::service::ServiceRegistry>,
    ) {
        let table = Arc::new(ProcessTable::new(64));
        let mut pids = Vec::new();

        for i in 0..agent_count {
            let entry = ProcessEntry {
                pid: 0,
                agent_id: format!("agent-{i}"),
                state: ProcessState::Running,
                capabilities: AgentCapabilities::default(),
                resource_usage: ResourceUsage::default(),
                cancel_token: CancellationToken::new(),
                parent_pid: None,
            };
            let pid = table.insert(entry).unwrap();
            pids.push(pid);
        }

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let registry = Arc::new(crate::service::ServiceRegistry::new());
        let router = A2ARouter::new(table, checker, topic_router)
            .with_service_registry(registry.clone());

        let mut receivers = Vec::new();
        for &pid in &pids {
            let rx = router.create_inbox(pid);
            receivers.push(rx);
        }

        (router, pids, receivers, registry)
    }

    #[tokio::test]
    async fn route_to_service_by_name() {
        let (router, pids, mut receivers, registry) = setup_router_with_registry(2);

        // Register a service owned by pids[1]
        registry
            .register_entry(crate::service::ServiceEntry {
                name: "auth".into(),
                owner_pid: Some(pids[1]),
                endpoint: crate::service::ServiceEndpoint::AgentInbox(pids[1]),
                audit_level: crate::service::ServiceAuditLevel::Full,
                registered_at: chrono::Utc::now(),
            })
            .unwrap();

        let msg = KernelMessage::text(pids[0], MessageTarget::Service("auth".into()), "validate");
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert_eq!(received.from, pids[0]);
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "validate"
        ));
    }

    #[tokio::test]
    async fn route_service_method() {
        let (router, pids, mut receivers, registry) = setup_router_with_registry(2);

        registry
            .register_entry(crate::service::ServiceEntry {
                name: "auth".into(),
                owner_pid: Some(pids[1]),
                endpoint: crate::service::ServiceEndpoint::AgentInbox(pids[1]),
                audit_level: crate::service::ServiceAuditLevel::Full,
                registered_at: chrono::Utc::now(),
            })
            .unwrap();

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::ServiceMethod {
                service: "auth".into(),
                method: "validate_token".into(),
            },
            "token-123",
        );
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert_eq!(received.from, pids[0]);
        assert!(matches!(
            received.target,
            MessageTarget::ServiceMethod { ref service, ref method }
            if service == "auth" && method == "validate_token"
        ));
    }

    #[tokio::test]
    async fn service_not_found_returns_error() {
        let (router, pids, _receivers, _registry) = setup_router_with_registry(2);

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::Service("nonexistent".into()),
            "hello",
        );
        let result = router.send(msg).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(
            err_msg.contains("service not found"),
            "expected 'service not found', got: {err_msg}"
        );
    }

    #[tokio::test]
    async fn service_entry_registration() {
        let registry = crate::service::ServiceRegistry::new();

        let entry = crate::service::ServiceEntry {
            name: "cache".into(),
            owner_pid: Some(42),
            endpoint: crate::service::ServiceEndpoint::AgentInbox(42),
            audit_level: crate::service::ServiceAuditLevel::Full,
            registered_at: chrono::Utc::now(),
        };
        registry.register_entry(entry).unwrap();

        let retrieved = registry.get_entry("cache").unwrap();
        assert_eq!(retrieved.name, "cache");
        assert_eq!(retrieved.owner_pid, Some(42));
    }

    #[tokio::test]
    async fn service_entry_with_audit_level() {
        let registry = crate::service::ServiceRegistry::new();

        let entry = crate::service::ServiceEntry {
            name: "metrics".into(),
            owner_pid: Some(10),
            endpoint: crate::service::ServiceEndpoint::AgentInbox(10),
            audit_level: crate::service::ServiceAuditLevel::GateOnly,
            registered_at: chrono::Utc::now(),
        };
        registry.register_entry(entry).unwrap();

        let retrieved = registry.get_entry("metrics").unwrap();
        assert_eq!(retrieved.audit_level, crate::service::ServiceAuditLevel::GateOnly);
    }

    #[tokio::test]
    async fn resolve_target_finds_owner_pid() {
        let registry = crate::service::ServiceRegistry::new();

        let entry = crate::service::ServiceEntry {
            name: "search".into(),
            owner_pid: Some(77),
            endpoint: crate::service::ServiceEndpoint::AgentInbox(77),
            audit_level: crate::service::ServiceAuditLevel::Full,
            registered_at: chrono::Utc::now(),
        };
        registry.register_entry(entry).unwrap();

        assert_eq!(registry.resolve_target("search"), Some(77));
        assert_eq!(registry.resolve_target("nonexistent"), None);
    }

    #[tokio::test]
    async fn service_without_registry_returns_error() {
        // Router without service registry should fail on Service target
        let (router_no_reg, pids, _receivers) = setup_router(2);

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::Service("missing".into()),
            "hello",
        );
        let result = router_no_reg.send(msg).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(
            err_msg.contains("no service registry"),
            "expected 'no service registry', got: {err_msg}"
        );
    }

    #[tokio::test]
    async fn external_service_no_pid_returns_error() {
        let (router, pids, _receivers, registry) = setup_router_with_registry(2);

        // Register external service with no owner_pid
        registry
            .register_entry(crate::service::ServiceEntry {
                name: "redis".into(),
                owner_pid: None,
                endpoint: crate::service::ServiceEndpoint::External {
                    url: "redis://localhost:6379".into(),
                },
                audit_level: crate::service::ServiceAuditLevel::GateOnly,
                registered_at: chrono::Utc::now(),
            })
            .unwrap();

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::Service("redis".into()),
            "ping",
        );
        let result = router.send(msg).await;
        assert!(result.is_err(), "external service with no PID should fail routing");
    }

    // ── Request-response tests ──────────────────────────────────

    #[tokio::test]
    async fn request_response_completes() {
        let (router, pids, mut receivers) = setup_router(2);
        let router = Arc::new(router);
        let router2 = Arc::clone(&router);

        // Spawn a responder that reads from pids[1]'s inbox and replies.
        let from_pid = pids[0];
        let to_pid = pids[1];
        tokio::spawn(async move {
            let msg = receivers[1].recv().await.unwrap();
            // Build a correlated response back to the sender.
            let reply = KernelMessage::with_correlation(
                to_pid,
                MessageTarget::Process(from_pid),
                MessagePayload::Text("pong".into()),
                msg.id.clone(),
            );
            router2.try_complete_request(reply);
        });

        let request = KernelMessage::text(from_pid, MessageTarget::Process(to_pid), "ping");
        let response = router
            .request(request, Duration::from_secs(5))
            .await
            .unwrap();
        assert!(matches!(
            response.payload,
            MessagePayload::Text(ref t) if t == "pong"
        ));
        assert_eq!(router.pending_request_count(), 0);
    }

    #[tokio::test]
    async fn request_response_timeout() {
        let (router, pids, _receivers) = setup_router(2);

        let request =
            KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "no-reply");
        let result = router.request(request, Duration::from_millis(50)).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(
            err_msg.contains("timeout"),
            "expected timeout error, got: {err_msg}"
        );
        assert_eq!(router.pending_request_count(), 0);
    }

    #[tokio::test]
    async fn try_complete_request_matching() {
        let (router, pids, _receivers) = setup_router(2);

        // Manually register a pending request.
        let (tx, rx) = oneshot::channel();
        let request_id = "req-42".to_string();
        router.pending_requests.insert(
            request_id.clone(),
            PendingRequest {
                response_tx: tx,
                sent_at: Instant::now(),
            },
        );

        // Build a response with matching correlation_id.
        let reply = KernelMessage::with_correlation(
            pids[1],
            MessageTarget::Process(pids[0]),
            MessagePayload::Text("reply".into()),
            request_id,
        );
        let completed = router.try_complete_request(reply);
        assert!(completed, "try_complete_request should return true for matching id");

        let response = rx.await.unwrap();
        assert!(matches!(
            response.payload,
            MessagePayload::Text(ref t) if t == "reply"
        ));
    }

    #[tokio::test]
    async fn try_complete_request_no_match() {
        let (router, pids, _receivers) = setup_router(2);

        // No pending requests registered.
        let reply = KernelMessage::with_correlation(
            pids[1],
            MessageTarget::Process(pids[0]),
            MessagePayload::Text("orphan".into()),
            "nonexistent-id".into(),
        );
        let completed = router.try_complete_request(reply);
        assert!(!completed, "try_complete_request should return false for non-matching id");

        // Also test message with no correlation_id.
        let plain = KernelMessage::text(pids[1], MessageTarget::Process(pids[0]), "plain");
        assert!(!router.try_complete_request(plain));
    }

    #[tokio::test]
    async fn pending_request_count_tracks_correctly() {
        let (router, _pids, _receivers) = setup_router(2);
        assert_eq!(router.pending_request_count(), 0);

        // Insert two pending requests.
        let (tx1, _rx1) = oneshot::channel();
        let (tx2, _rx2) = oneshot::channel();
        router.pending_requests.insert(
            "req-a".into(),
            PendingRequest {
                response_tx: tx1,
                sent_at: Instant::now(),
            },
        );
        assert_eq!(router.pending_request_count(), 1);

        router.pending_requests.insert(
            "req-b".into(),
            PendingRequest {
                response_tx: tx2,
                sent_at: Instant::now(),
            },
        );
        assert_eq!(router.pending_request_count(), 2);

        // Complete one via try_complete_request.
        let reply = KernelMessage::with_correlation(
            1,
            MessageTarget::Process(2),
            MessagePayload::Text("done".into()),
            "req-a".into(),
        );
        router.try_complete_request(reply);
        assert_eq!(router.pending_request_count(), 1);

        // Remove the other manually.
        router.pending_requests.remove("req-b");
        assert_eq!(router.pending_request_count(), 0);
    }

    // ── C4: Routing-time gate tests ────────────────────────────────

    #[cfg(feature = "exochain")]
    #[tokio::test]
    async fn routing_gate_denies_message() {
        struct DenyGate;
        impl crate::gate::GateBackend for DenyGate {
            fn check(
                &self,
                _agent: &str,
                _action: &str,
                _ctx: &serde_json::Value,
            ) -> crate::gate::GateDecision {
                crate::gate::GateDecision::Deny {
                    reason: "blocked by policy".into(),
                    receipt: None,
                }
            }
        }

        let (router, pids, _receivers) = setup_router(2);
        let router = router.with_gate(Arc::new(DenyGate));

        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "hello");
        let result = router.send(msg).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(
            err_msg.contains("routing gate denied"),
            "expected 'routing gate denied', got: {err_msg}"
        );
    }

    #[cfg(feature = "exochain")]
    #[tokio::test]
    async fn routing_gate_permits_message() {
        struct PermitGate;
        impl crate::gate::GateBackend for PermitGate {
            fn check(
                &self,
                _agent: &str,
                _action: &str,
                _ctx: &serde_json::Value,
            ) -> crate::gate::GateDecision {
                crate::gate::GateDecision::Permit { token: None }
            }
        }

        let (router, pids, mut receivers) = setup_router(2);
        let router = router.with_gate(Arc::new(PermitGate));

        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "hello");
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert_eq!(received.from, pids[0]);
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "hello"
        ));
    }

    #[cfg(feature = "exochain")]
    #[tokio::test]
    async fn routing_gate_defer_still_delivers() {
        struct DeferGate;
        impl crate::gate::GateBackend for DeferGate {
            fn check(
                &self,
                _agent: &str,
                _action: &str,
                _ctx: &serde_json::Value,
            ) -> crate::gate::GateDecision {
                crate::gate::GateDecision::Defer {
                    reason: "pending review".into(),
                }
            }
        }

        let (router, pids, mut receivers) = setup_router(2);
        let router = router.with_gate(Arc::new(DeferGate));

        let msg = KernelMessage::text(pids[0], MessageTarget::Process(pids[1]), "deferred");
        router.send(msg).await.unwrap(); // should succeed (defer = deliver)
        assert!(receivers[1].try_recv().is_ok());
    }

    // ── W5: Test hardening — additional coverage ─────────────────

    #[tokio::test]
    async fn multiple_messages_queued_in_order() {
        let (router, pids, mut receivers) = setup_router(2);

        for i in 0..5 {
            let msg = KernelMessage::text(
                pids[0],
                MessageTarget::Process(pids[1]),
                &format!("msg-{i}"),
            );
            router.send(msg).await.unwrap();
        }

        // Messages should arrive in FIFO order
        for i in 0..5 {
            let received = receivers[1].try_recv().unwrap();
            let expected = format!("msg-{i}");
            assert!(
                matches!(received.payload, MessagePayload::Text(ref t) if t == &expected),
                "expected '{expected}' but got {:?}",
                received.payload,
            );
        }
    }

    #[tokio::test]
    async fn unsubscribe_stops_topic_delivery() {
        let (router, pids, mut receivers) = setup_router(3);

        router.topic_router().subscribe(pids[1], "events");
        router.topic_router().subscribe(pids[2], "events");

        // Unsubscribe pids[1]
        router.topic_router().unsubscribe(pids[1], "events");

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::Topic("events".into()),
            "after-unsub",
        );
        router.send(msg).await.unwrap();

        // pids[1] should NOT receive (unsubscribed)
        assert!(receivers[1].try_recv().is_err(), "unsubscribed agent should not receive");

        // pids[2] should still receive
        let received = receivers[2].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "after-unsub"
        ));
    }

    #[tokio::test]
    async fn publish_to_empty_topic_succeeds() {
        let (router, pids, _receivers) = setup_router(2);

        // Publish to a topic with no subscribers — should succeed with no error
        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::Topic("empty-topic".into()),
            "nobody-listening",
        );
        let result = router.send(msg).await;
        assert!(result.is_ok(), "publish to empty topic should succeed");
    }

    #[tokio::test]
    async fn broadcast_with_zero_other_agents() {
        let (router, pids, mut receivers) = setup_router(1);

        let msg = KernelMessage::text(pids[0], MessageTarget::Broadcast, "alone");
        let result = router.send(msg).await;
        assert!(result.is_ok(), "broadcast with only sender should succeed");

        // Sender should not receive their own broadcast
        assert!(receivers[0].try_recv().is_err());
    }

    #[tokio::test]
    async fn ipc_scope_restricted_allows_listed_pids() {
        let table = Arc::new(ProcessTable::new(64));

        use crate::capability::IpcScope;
        // Create target first so we know its PID for the restricted list
        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        // Create sender with restricted IPC scope that includes the target
        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "restricted-sender".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities {
                ipc_scope: IpcScope::Restricted(vec![target_pid]),
                ..Default::default()
            },
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx_sender = router.create_inbox(sender_pid);
        let mut rx_target = router.create_inbox(target_pid);

        let msg = KernelMessage::text(
            sender_pid,
            MessageTarget::Process(target_pid),
            "allowed",
        );
        let result = router.send(msg).await;
        assert!(result.is_ok(), "restricted sender should reach allowed PID");

        let received = rx_target.try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "allowed"
        ));
    }

    #[tokio::test]
    async fn ipc_scope_restricted_blocks_unlisted_pids() {
        let table = Arc::new(ProcessTable::new(64));

        use crate::capability::IpcScope;
        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        // Restricted to PID 999 (not the target)
        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "restricted-sender".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities {
                ipc_scope: IpcScope::Restricted(vec![999]),
                ..Default::default()
            },
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx_sender = router.create_inbox(sender_pid);
        let _rx_target = router.create_inbox(target_pid);

        let msg = KernelMessage::text(
            sender_pid,
            MessageTarget::Process(target_pid),
            "blocked",
        );
        let result = router.send(msg).await;
        assert!(result.is_err(), "restricted sender should be blocked from unlisted PID");
    }

    #[test]
    fn create_inbox_returns_receiver() {
        let table = Arc::new(ProcessTable::new(64));
        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        let rx = router.create_inbox(10);
        assert!(router.has_inbox(10));
        assert_eq!(router.inbox_count(), 1);
        drop(rx);
    }

    #[test]
    fn create_inbox_replaces_existing() {
        let table = Arc::new(ProcessTable::new(64));
        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        let _rx1 = router.create_inbox(10);
        assert_eq!(router.inbox_count(), 1);

        // Replace — old receiver is invalidated
        let _rx2 = router.create_inbox(10);
        assert_eq!(router.inbox_count(), 1);
        assert!(router.has_inbox(10));
    }

    #[test]
    fn remove_nonexistent_inbox_is_noop() {
        let table = Arc::new(ProcessTable::new(64));
        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        // Should not panic
        router.remove_inbox(999);
        assert_eq!(router.inbox_count(), 0);
    }

    #[test]
    fn inbox_count_tracks_multiple_inboxes() {
        let table = Arc::new(ProcessTable::new(64));
        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);

        let _rx1 = router.create_inbox(1);
        let _rx2 = router.create_inbox(2);
        let _rx3 = router.create_inbox(3);
        assert_eq!(router.inbox_count(), 3);

        router.remove_inbox(2);
        assert_eq!(router.inbox_count(), 2);
        assert!(!router.has_inbox(2));
        assert!(router.has_inbox(1));
        assert!(router.has_inbox(3));
    }

    #[tokio::test]
    async fn send_to_kernel_target_succeeds() {
        let (router, pids, _receivers) = setup_router(1);

        let msg = KernelMessage::text(pids[0], MessageTarget::Kernel, "kernel-msg");
        let result = router.send(msg).await;
        assert!(result.is_ok(), "kernel target routing should succeed (even if no-op)");
    }

    #[tokio::test]
    async fn send_to_remote_node_fails() {
        let (router, pids, _receivers) = setup_router(1);

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::RemoteNode {
                node_id: "remote-1".into(),
                target: Box::new(MessageTarget::Process(42)),
            },
            "remote-msg",
        );
        let result = router.send(msg).await;
        assert!(result.is_err(), "remote node routing should fail (not yet implemented)");
        let err_msg = format!("{}", result.unwrap_err());
        assert!(err_msg.contains("remote routing"), "expected remote routing error, got: {err_msg}");
    }

    #[tokio::test]
    async fn suspended_process_can_still_send() {
        let table = Arc::new(ProcessTable::new(64));

        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "suspended-sender".to_owned(),
            state: ProcessState::Suspended,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx_sender = router.create_inbox(sender_pid);
        let mut rx_target = router.create_inbox(target_pid);

        let msg = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "from-suspended");
        let result = router.send(msg).await;
        assert!(result.is_ok(), "suspended process should be allowed to send");

        let received = rx_target.try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Text(ref t) if t == "from-suspended"
        ));
    }

    #[tokio::test]
    async fn topic_multiple_subscribers_receive_same_message() {
        let (router, pids, mut receivers) = setup_router(5);

        // Subscribe pids[1..5] to the topic
        for &pid in &pids[1..] {
            router.topic_router().subscribe(pid, "news");
        }

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::Topic("news".into()),
            "breaking",
        );
        router.send(msg).await.unwrap();

        // All subscribers should receive the same payload
        for (i, rx) in receivers[1..].iter_mut().enumerate() {
            let received = rx.try_recv().unwrap_or_else(|_| {
                panic!("subscriber {} (pid {}) should have received message", i + 1, pids[i + 1])
            });
            assert!(matches!(
                received.payload,
                MessagePayload::Text(ref t) if t == "breaking"
            ));
            assert_eq!(received.from, pids[0]);
        }
    }

    #[tokio::test]
    async fn broadcast_skips_scope_restricted_targets() {
        let table = Arc::new(ProcessTable::new(64));

        use crate::capability::IpcScope;
        // Sender with restricted scope (empty list = nobody allowed)
        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "restricted-broadcaster".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities {
                ipc_scope: IpcScope::Restricted(vec![]),
                ..Default::default()
            },
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx_sender = router.create_inbox(sender_pid);
        let mut rx_target = router.create_inbox(target_pid);

        let msg = KernelMessage::text(sender_pid, MessageTarget::Broadcast, "restricted-broadcast");
        let result = router.send(msg).await;
        assert!(result.is_ok(), "broadcast itself should succeed even if all targets are blocked");

        // Target should NOT receive (scope blocks it)
        assert!(rx_target.try_recv().is_err(), "restricted broadcast should not reach any target");
    }

    #[test]
    fn service_registry_accessor() {
        let table = Arc::new(ProcessTable::new(64));
        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));

        let router = A2ARouter::new(table.clone(), checker.clone(), topic_router.clone());
        assert!(router.service_registry().is_none(), "no registry by default");

        let registry = Arc::new(crate::service::ServiceRegistry::new());
        let router = router.with_service_registry(registry);
        assert!(router.service_registry().is_some(), "registry should be present after with_service_registry");
    }

    #[test]
    fn topic_router_accessor() {
        let table = Arc::new(ProcessTable::new(64));
        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let expected = Arc::clone(&topic_router);

        let router = A2ARouter::new(table, checker, topic_router);
        assert!(Arc::ptr_eq(router.topic_router(), &expected));
    }

    #[tokio::test]
    async fn json_payload_routes() {
        let (router, pids, mut receivers) = setup_router(2);

        let msg = KernelMessage::new(
            pids[0],
            MessageTarget::Process(pids[1]),
            MessagePayload::Json(serde_json::json!({"key": "value"})),
        );
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Json(ref v) if v["key"] == "value"
        ));
    }

    #[tokio::test]
    async fn signal_payload_routes() {
        use crate::ipc::KernelSignal;
        let (router, pids, mut receivers) = setup_router(2);

        let msg = KernelMessage::signal(
            pids[0],
            MessageTarget::Process(pids[1]),
            KernelSignal::Shutdown,
        );
        router.send(msg).await.unwrap();

        let received = receivers[1].try_recv().unwrap();
        assert!(matches!(
            received.payload,
            MessagePayload::Signal(KernelSignal::Shutdown)
        ));
    }

    #[tokio::test]
    async fn request_to_nonexistent_target_cleans_up_pending() {
        let table = Arc::new(ProcessTable::new(64));

        let sender_entry = ProcessEntry {
            pid: 0,
            agent_id: "sender".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let sender_pid = table.insert(sender_entry).unwrap();

        // Create target in process table but no inbox
        let target_entry = ProcessEntry {
            pid: 0,
            agent_id: "no-inbox-target".to_owned(),
            state: ProcessState::Running,
            capabilities: AgentCapabilities::default(),
            resource_usage: ResourceUsage::default(),
            cancel_token: CancellationToken::new(),
            parent_pid: None,
        };
        let target_pid = table.insert(target_entry).unwrap();

        let checker = Arc::new(CapabilityChecker::new(table.clone()));
        let topic_router = Arc::new(TopicRouter::new(table.clone()));
        let router = A2ARouter::new(table, checker, topic_router);
        let _rx_sender = router.create_inbox(sender_pid);
        // Intentionally no inbox for target

        let msg = KernelMessage::text(sender_pid, MessageTarget::Process(target_pid), "request");
        let result = router.request(msg, Duration::from_millis(100)).await;
        assert!(result.is_err(), "request to PID without inbox should fail");
        assert_eq!(router.pending_request_count(), 0, "pending request should be cleaned up on send failure");
    }

    #[tokio::test]
    async fn service_method_not_found_returns_error() {
        let (router, pids, _receivers, _registry) = setup_router_with_registry(2);

        let msg = KernelMessage::text(
            pids[0],
            MessageTarget::ServiceMethod {
                service: "nonexistent-svc".into(),
                method: "do_thing".into(),
            },
            "call",
        );
        let result = router.send(msg).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(
            err_msg.contains("service not found"),
            "expected 'service not found' for missing ServiceMethod target, got: {err_msg}"
        );
    }
}