car-messaging 0.48.0

//! General outbound messaging — the host side of the runtime's
//! `messaging.send` tool.
//!
//! This crate's original scope was *approval* transports: an agent asks, a
//! human answers approve/deny. But the low-level iMessage seam it already owns
//! ([`crate::messaging_orchestrator::MessageSender`]) was never approval-
//! specific — it is `send(handle, body)`, fully general, with the approval
//! shape sitting above it. What was missing was a way for the runtime to reach
//! that seam for an ordinary message to a human, so every agent that wanted
//! one hand-rolled a transport and the policy engine never saw the send.
//!
//! [`OutboundRegistry`] implements [`car_engine::MessageSink`], so
//! `messaging.send` dispatches here after passing the runtime's validator,
//! policy engine and rate limiter. The registry does two things the individual
//! adapters must not each reinvent:
//!
//! 1. **Routes by channel name** to a registered [`OutboundAdapter`], falling
//!    back to a single optional catch-all adapter for channels nothing built
//!    in claims.
//! 2. **Honours the idempotency key**, so a retry after a
//!    delivered-but-unacknowledged send does not reach the human twice.
//!
//! Everything channel-specific — the wire, the consent model, which recipient
//! shapes even exist — belongs in the adapter.

use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};

use async_trait::async_trait;

use car_engine::messaging::{MessageReceipt, MessageSink, OutboundMessage, Recipient};
use car_server_types::channel::ChannelId;

use crate::messaging_config::MessagingConfigStore;
use crate::messaging_orchestrator::MessageSender;

/// The channel name the iMessage adapter answers to, as it appears in the
/// tool's `channel` parameter.
pub const IMESSAGE_CHANNEL: &str = "imessage";

/// How many delivered idempotency keys the registry remembers.
///
/// The ledger exists to make a *retry* safe, and a retry follows its original
/// within seconds or minutes — it is not a permanent delivery archive, and an
/// unbounded map on a long-lived daemon is a slow leak. 1024 keys is far more
/// than any plausible in-flight retry window while staying trivially small in
/// memory. Oldest-first eviction: the newest keys are the ones a retry can
/// still be chasing.
const LEDGER_CAPACITY: usize = 1024;

/// One channel's outbound transport.
///
/// Deliberately narrower than [`MessageSink`]: an adapter serves exactly one
/// channel and does not route, so it cannot accidentally become a second
/// dispatch layer. Fan-out and dedup live in [`OutboundRegistry`].
#[async_trait]
pub trait OutboundAdapter: Send + Sync {
    /// The channel name this adapter answers to (matched against
    /// [`OutboundMessage::channel`]).
    fn channel(&self) -> &str;

    /// Deliver one message, or explain why it was not delivered. The error
    /// text reaches the model as the tool's error, so it should say what would
    /// make the send work.
    async fn send(&self, msg: &OutboundMessage) -> Result<MessageReceipt, String>;
}

/// Bounded, insertion-ordered record of idempotency keys that have delivered.
struct Ledger {
    receipts: HashMap<String, MessageReceipt>,
    order: VecDeque<String>,
}

impl Ledger {
    fn new() -> Self {
        Self {
            receipts: HashMap::new(),
            order: VecDeque::new(),
        }
    }

    fn get(&self, key: &str) -> Option<MessageReceipt> {
        self.receipts.get(key).cloned()
    }

    fn record(&mut self, key: String, receipt: MessageReceipt) {
        if self.receipts.insert(key.clone(), receipt).is_none() {
            self.order.push_back(key);
        }
        while self.order.len() > LEDGER_CAPACITY {
            if let Some(oldest) = self.order.pop_front() {
                self.receipts.remove(&oldest);
            }
        }
    }
}

/// The host's outbound message sink: channel adapters plus the idempotency
/// ledger.
///
/// One per daemon, held behind an `Arc` and handed to
/// `Runtime::with_message_sink`.
pub struct OutboundRegistry {
    adapters: Mutex<HashMap<String, Arc<dyn OutboundAdapter>>>,
    fallback: Mutex<Option<Arc<dyn OutboundAdapter>>>,
    ledger: Mutex<Ledger>,
}

impl OutboundRegistry {
    pub fn new() -> Self {
        Self {
            adapters: Mutex::new(HashMap::new()),
            fallback: Mutex::new(None),
            ledger: Mutex::new(Ledger::new()),
        }
    }

    /// Register (or replace) the adapter for `adapter.channel()`.
    ///
    /// Takes `&self` so a host can add channels after the registry is already
    /// shared with the runtime — channels come and go as an operator pairs a
    /// device or connects a workspace, and re-plumbing the runtime for that
    /// would be absurd.
    pub fn register(&self, adapter: Arc<dyn OutboundAdapter>) {
        let name = adapter.channel().to_string();
        self.adapters
            .lock()
            .expect("outbound adapter registry poisoned")
            .insert(name, adapter);
    }

    /// Install (or replace) the catch-all adapter used when no registered
    /// adapter claims a channel.
    ///
    /// The point is that the runtime must NOT have to compile in an adapter
    /// for every channel a deployment cares about. Teams, Discord, a bespoke
    /// internal bus — those are host concerns, and a host that can deliver on
    /// them registers one fallback rather than teaching this crate a new
    /// transport per platform. Routing is deliberately *exact match first*: a
    /// built-in adapter always wins for its own channel, so installing a
    /// fallback can never silently divert `imessage` away from the transport
    /// that has the pairing state for it.
    ///
    /// Takes `&self` for the same reason [`Self::register`] does — the host
    /// may only learn it can reach a channel after the registry is already
    /// shared with the runtime.
    pub fn set_fallback(&self, adapter: Arc<dyn OutboundAdapter>) {
        *self
            .fallback
            .lock()
            .expect("outbound adapter registry poisoned") = Some(adapter);
    }

    /// Registered channel names, sorted for stable error text and listings.
    ///
    /// Only the *exact-match* adapters appear here. A fallback claims channels
    /// it cannot enumerate — that is what makes it a fallback — so listing it
    /// would mean either advertising nothing or advertising a lie.
    pub fn channel_names(&self) -> Vec<String> {
        let mut names: Vec<String> = self
            .adapters
            .lock()
            .expect("outbound adapter registry poisoned")
            .keys()
            .cloned()
            .collect();
        names.sort();
        names
    }

    /// Exact channel match first, then the fallback. Nothing else: an unclaimed
    /// channel with no fallback installed is an error, never a silent drop.
    fn adapter_for(&self, channel: &str) -> Option<Arc<dyn OutboundAdapter>> {
        let exact = self
            .adapters
            .lock()
            .expect("outbound adapter registry poisoned")
            .get(channel)
            .cloned();
        exact.or_else(|| {
            self.fallback
                .lock()
                .expect("outbound adapter registry poisoned")
                .clone()
        })
    }
}

impl Default for OutboundRegistry {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl MessageSink for OutboundRegistry {
    async fn channels(&self) -> Vec<String> {
        self.channel_names()
    }

    async fn send(&self, msg: &OutboundMessage) -> Result<MessageReceipt, String> {
        // Dedup FIRST: a repeat key must not reach the adapter at all. Checking
        // after the send would still deliver the duplicate, which is the whole
        // failure being prevented.
        if let Some(key) = msg.idempotency_key.as_deref() {
            let seen = self
                .ledger
                .lock()
                .expect("outbound idempotency ledger poisoned")
                .get(key);
            if let Some(mut receipt) = seen {
                receipt.deduplicated = true;
                return Ok(receipt);
            }
        }

        let adapter = self.adapter_for(&msg.channel).ok_or_else(|| {
            let registered = self.channel_names();
            if registered.is_empty() {
                format!(
                    "unknown messaging channel '{}': no channels are registered on this host",
                    msg.channel
                )
            } else {
                format!(
                    "unknown messaging channel '{}': registered channels are {}",
                    msg.channel,
                    registered.join(", ")
                )
            }
        })?;

        // A FAILED send must not record the key. Recording it would turn one
        // transient failure into a permanent refusal to ever deliver that
        // message — the retry the key exists to make safe would be swallowed
        // as a duplicate of something that never arrived.
        let receipt = adapter.send(msg).await?;

        if let Some(key) = msg.idempotency_key.as_deref() {
            self.ledger
                .lock()
                .expect("outbound idempotency ledger poisoned")
                .record(key.to_string(), receipt.clone());
        }
        Ok(receipt)
    }
}

/// iMessage outbound adapter — the general form of the send the approval
/// transport already performs.
///
/// Reuses [`MessageSender`], the same injectable seam
/// [`crate::messaging_orchestrator::MessagingOrchestrator`] sends approval
/// prompts through, so outbound messaging inherits the transport's tested
/// hard-vs-soft failure mapping instead of growing a parallel one.
pub struct ImessageOutboundAdapter {
    sender: Arc<dyn MessageSender>,
    config: MessagingConfigStore,
}

impl ImessageOutboundAdapter {
    pub fn new(sender: Arc<dyn MessageSender>, config: MessagingConfigStore) -> Self {
        Self { sender, config }
    }
}

#[async_trait]
impl OutboundAdapter for ImessageOutboundAdapter {
    fn channel(&self) -> &str {
        IMESSAGE_CHANNEL
    }

    async fn send(&self, msg: &OutboundMessage) -> Result<MessageReceipt, String> {
        let handle = match &msg.to {
            Recipient::Direct(handle) => handle,
            Recipient::Channel(id) => {
                return Err(format!(
                    "imessage has no channel-post form (asked to post to '{id}') — \
                     address a person with kind 'direct' instead"
                ))
            }
        };

        // Channel-level consent. This is the device pairing the operator
        // performed on THIS channel (`messaging.json`'s per-channel
        // allowlist), and it is INDEPENDENT of project policy: policy decides
        // whether the agent may use the messaging tool at all and under what
        // conditions, while this decides whether a given human ever agreed to
        // receive iMessages from this host. Neither subsumes the other — do
        // not delete one as redundant with the other.
        //
        // `is_allowlisted_for` errs only on a malformed config file; treat an
        // error as "not allowlisted" (fail closed), exactly as the inbound
        // path does at `messaging_orchestrator.rs`'s allowlist wall. A
        // corrupt config must not become an open door.
        let allowlisted = self
            .config
            .is_allowlisted_for(ChannelId::IMessage, handle)
            .unwrap_or(false);
        if !allowlisted {
            return Err(format!(
                "imessage recipient '{handle}' is not paired with this host — \
                 pair the handle (messaging.pairing.start) before messaging it"
            ));
        }

        // A HARD failure comes back as `Err` unchanged; a SOFT failure
        // (`sent: false`) is a failure too — pre-U3 code swallowed those as
        // success, and a silently-undelivered message to a human is the worst
        // possible outcome of this tool.
        match self.sender.send(handle, &msg.body)? {
            outcome if outcome.sent => Ok(MessageReceipt::delivered(IMESSAGE_CHANNEL)),
            outcome => Err(outcome
                .reason
                .unwrap_or_else(|| "Messages reported the message was not sent".to_string())),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::messaging_orchestrator::SendOutcome;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// Counting adapter: proves whether the transport was reached at all.
    struct SpyAdapter {
        channel: String,
        calls: AtomicUsize,
        /// When false, every send fails — the retryability tests need a send
        /// that does NOT deliver.
        succeed: bool,
    }

    impl SpyAdapter {
        fn new(channel: &str, succeed: bool) -> Self {
            Self {
                channel: channel.to_string(),
                calls: AtomicUsize::new(0),
                succeed,
            }
        }

        fn calls(&self) -> usize {
            self.calls.load(Ordering::SeqCst)
        }
    }

    #[async_trait]
    impl OutboundAdapter for SpyAdapter {
        fn channel(&self) -> &str {
            &self.channel
        }

        async fn send(&self, _msg: &OutboundMessage) -> Result<MessageReceipt, String> {
            let n = self.calls.fetch_add(1, Ordering::SeqCst) + 1;
            if self.succeed {
                Ok(MessageReceipt::delivered(self.channel.as_str())
                    .with_message_id(format!("{}-{n}", self.channel)))
            } else {
                Err("transport down".to_string())
            }
        }
    }

    /// Counting `MessageSender`: the iMessage tests assert the sender is never
    /// reached for a refused recipient.
    struct SpySender {
        calls: AtomicUsize,
        outcome: Result<SendOutcome, String>,
    }

    impl SpySender {
        fn ok() -> Self {
            Self {
                calls: AtomicUsize::new(0),
                outcome: Ok(SendOutcome::ok()),
            }
        }

        fn with_outcome(outcome: Result<SendOutcome, String>) -> Self {
            Self {
                calls: AtomicUsize::new(0),
                outcome,
            }
        }

        fn calls(&self) -> usize {
            self.calls.load(Ordering::SeqCst)
        }
    }

    impl MessageSender for SpySender {
        fn send(&self, _handle: &str, _body: &str) -> Result<SendOutcome, String> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            self.outcome.clone()
        }
    }

    fn msg(channel: &str, key: Option<&str>) -> OutboundMessage {
        OutboundMessage {
            channel: channel.to_string(),
            to: Recipient::Direct("+15551112222".to_string()),
            body: "build is green".to_string(),
            idempotency_key: key.map(|k| k.to_string()),
        }
    }

    #[tokio::test]
    async fn routes_to_the_matching_adapter() {
        let registry = OutboundRegistry::new();
        let imessage = Arc::new(SpyAdapter::new("imessage", true));
        let slack = Arc::new(SpyAdapter::new("slack", true));
        registry.register(imessage.clone());
        registry.register(slack.clone());

        let receipt = registry.send(&msg("slack", None)).await.unwrap();
        assert_eq!(receipt.channel, "slack");
        assert_eq!(receipt.message_id.as_deref(), Some("slack-1"));
        assert_eq!(slack.calls(), 1);
        assert_eq!(imessage.calls(), 0);

        assert_eq!(registry.channels().await, vec!["imessage", "slack"]);
    }

    #[tokio::test]
    async fn unknown_channel_names_the_registered_ones() {
        let registry = OutboundRegistry::new();
        registry.register(Arc::new(SpyAdapter::new("imessage", true)));
        registry.register(Arc::new(SpyAdapter::new("slack", true)));

        let err = registry.send(&msg("slak", None)).await.unwrap_err();
        assert!(err.contains("unknown messaging channel 'slak'"), "{err}");
        assert!(err.contains("imessage, slack"), "{err}");
    }

    #[tokio::test]
    async fn unknown_channel_with_nothing_registered_says_so() {
        let registry = OutboundRegistry::new();
        let err = registry.send(&msg("imessage", None)).await.unwrap_err();
        assert!(err.contains("no channels are registered"), "{err}");
    }

    #[tokio::test]
    async fn repeat_idempotency_key_dedups_without_resending() {
        let registry = OutboundRegistry::new();
        let adapter = Arc::new(SpyAdapter::new("imessage", true));
        registry.register(adapter.clone());

        let first = registry
            .send(&msg("imessage", Some("run-42")))
            .await
            .unwrap();
        assert!(!first.deduplicated);
        assert_eq!(adapter.calls(), 1);

        let second = registry
            .send(&msg("imessage", Some("run-42")))
            .await
            .unwrap();
        assert!(second.deduplicated, "repeat key must report deduplicated");
        assert_eq!(second.message_id, first.message_id);
        assert_eq!(adapter.calls(), 1, "the adapter must NOT be called again");

        // A different key is a different message and does send.
        registry
            .send(&msg("imessage", Some("run-43")))
            .await
            .unwrap();
        assert_eq!(adapter.calls(), 2);
    }

    #[tokio::test]
    async fn fallback_takes_an_unclaimed_channel() {
        let registry = OutboundRegistry::new();
        let imessage = Arc::new(SpyAdapter::new("imessage", true));
        let host = Arc::new(SpyAdapter::new("host", true));
        registry.register(imessage.clone());
        registry.set_fallback(host.clone());

        // No adapter claims "teams" — the fallback delivers it.
        let receipt = registry.send(&msg("teams", None)).await.unwrap();
        assert_eq!(receipt.message_id.as_deref(), Some("host-1"));
        assert_eq!(host.calls(), 1);
        assert_eq!(imessage.calls(), 0);

        // The fallback does NOT appear in the advertised channel list.
        assert_eq!(registry.channels().await, vec!["imessage"]);
    }

    #[tokio::test]
    async fn a_registered_adapter_beats_the_fallback() {
        let registry = OutboundRegistry::new();
        let imessage = Arc::new(SpyAdapter::new("imessage", true));
        let host = Arc::new(SpyAdapter::new("host", true));
        registry.register(imessage.clone());
        registry.set_fallback(host.clone());

        let receipt = registry.send(&msg("imessage", None)).await.unwrap();
        assert_eq!(receipt.message_id.as_deref(), Some("imessage-1"));
        assert_eq!(imessage.calls(), 1);
        assert_eq!(host.calls(), 0, "the fallback must not shadow a channel");
    }

    #[tokio::test]
    async fn without_a_fallback_the_unknown_channel_error_is_unchanged() {
        let registry = OutboundRegistry::new();
        registry.register(Arc::new(SpyAdapter::new("imessage", true)));

        let err = registry.send(&msg("teams", None)).await.unwrap_err();
        assert!(err.contains("unknown messaging channel 'teams'"), "{err}");
        assert!(err.contains("registered channels are imessage"), "{err}");
    }

    #[tokio::test]
    async fn the_ledger_applies_to_the_fallback_too() {
        let registry = OutboundRegistry::new();
        let host = Arc::new(SpyAdapter::new("host", true));
        registry.set_fallback(host.clone());

        let first = registry.send(&msg("teams", Some("run-42"))).await.unwrap();
        assert!(!first.deduplicated);
        let second = registry.send(&msg("teams", Some("run-42"))).await.unwrap();
        assert!(second.deduplicated, "dedup must not be bypassed by routing");
        assert_eq!(second.message_id, first.message_id);
        assert_eq!(host.calls(), 1, "the fallback must NOT be called again");
    }

    #[tokio::test]
    async fn a_failed_fallback_send_stays_retryable() {
        let registry = OutboundRegistry::new();
        let host = Arc::new(SpyAdapter::new("host", false));
        registry.set_fallback(host.clone());

        assert!(registry.send(&msg("teams", Some("k"))).await.is_err());
        assert!(registry.send(&msg("teams", Some("k"))).await.is_err());
        assert_eq!(host.calls(), 2);
    }

    #[tokio::test]
    async fn no_key_means_no_dedup() {
        let registry = OutboundRegistry::new();
        let adapter = Arc::new(SpyAdapter::new("imessage", true));
        registry.register(adapter.clone());

        registry.send(&msg("imessage", None)).await.unwrap();
        registry.send(&msg("imessage", None)).await.unwrap();
        assert_eq!(adapter.calls(), 2);
    }

    #[tokio::test]
    async fn failed_send_stays_retryable() {
        let registry = OutboundRegistry::new();
        let failing = Arc::new(SpyAdapter::new("imessage", false));
        registry.register(failing.clone());

        let err = registry
            .send(&msg("imessage", Some("run-42")))
            .await
            .unwrap_err();
        assert_eq!(err, "transport down");

        // Same key again: the failure must NOT have been recorded, so the
        // retry genuinely re-attempts rather than reporting a phantom dedup.
        let err = registry
            .send(&msg("imessage", Some("run-42")))
            .await
            .unwrap_err();
        assert_eq!(err, "transport down");
        assert_eq!(failing.calls(), 2, "a failed send must remain retryable");
    }

    #[test]
    fn ledger_evicts_oldest_first() {
        let mut ledger = Ledger::new();
        for i in 0..(LEDGER_CAPACITY + 10) {
            ledger.record(format!("key-{i}"), MessageReceipt::delivered("imessage"));
        }
        assert_eq!(ledger.order.len(), LEDGER_CAPACITY);
        assert!(ledger.get("key-0").is_none(), "oldest key must be evicted");
        assert!(ledger
            .get(&format!("key-{}", LEDGER_CAPACITY + 9))
            .is_some());
    }

    fn imessage_adapter(
        sender: Arc<SpySender>,
        allowlisted: &[&str],
    ) -> (ImessageOutboundAdapter, tempfile::TempDir) {
        let dir = tempfile::tempdir().unwrap();
        let config = MessagingConfigStore::with_base_dir(dir.path());
        for handle in allowlisted {
            config.add_handle_for(ChannelId::IMessage, handle).unwrap();
        }
        (ImessageOutboundAdapter::new(sender, config), dir)
    }

    #[tokio::test]
    async fn imessage_sends_to_an_allowlisted_handle() {
        let sender = Arc::new(SpySender::ok());
        let (adapter, _dir) = imessage_adapter(sender.clone(), &["+15551112222"]);

        let receipt = adapter.send(&msg("imessage", None)).await.unwrap();
        assert_eq!(receipt.channel, "imessage");
        assert!(!receipt.deduplicated);
        assert_eq!(sender.calls(), 1);
    }

    #[tokio::test]
    async fn imessage_refuses_an_unpaired_handle_without_sending() {
        let sender = Arc::new(SpySender::ok());
        let (adapter, _dir) = imessage_adapter(sender.clone(), &[]);

        let err = adapter.send(&msg("imessage", None)).await.unwrap_err();
        assert!(err.contains("not paired with this host"), "{err}");
        assert_eq!(
            sender.calls(),
            0,
            "a refused send must never reach the wire"
        );
    }

    #[tokio::test]
    async fn imessage_rejects_a_channel_recipient() {
        let sender = Arc::new(SpySender::ok());
        let (adapter, _dir) = imessage_adapter(sender.clone(), &["+15551112222"]);

        let mut m = msg("imessage", None);
        m.to = Recipient::Channel("C012ABCDEF".to_string());
        let err = adapter.send(&m).await.unwrap_err();
        assert!(err.contains("no channel-post form"), "{err}");
        assert_eq!(sender.calls(), 0);
    }

    #[tokio::test]
    async fn imessage_maps_a_soft_failure_to_an_error() {
        let sender = Arc::new(SpySender::with_outcome(Ok(SendOutcome::soft_fail(
            "recipient not found",
        ))));
        let (adapter, _dir) = imessage_adapter(sender.clone(), &["+15551112222"]);

        let err = adapter.send(&msg("imessage", None)).await.unwrap_err();
        assert_eq!(err, "recipient not found");
        assert_eq!(sender.calls(), 1);
    }

    #[tokio::test]
    async fn imessage_passes_a_hard_failure_through() {
        let sender = Arc::new(SpySender::with_outcome(Err("osascript exploded".into())));
        let (adapter, _dir) = imessage_adapter(sender.clone(), &["+15551112222"]);

        let err = adapter.send(&msg("imessage", None)).await.unwrap_err();
        assert_eq!(err, "osascript exploded");
    }
}