car_server_core/

host_channel.rs

//! Host-backed outbound channel adapter — how a channel CAR has no built-in
//! transport for still gets delivered.
//!
//! The runtime owns message *semantics* (validation, policy, rate limiting,
//! idempotency, the event log); the host owns *channels*. Microsoft Teams is
//! the motivating case: a Teams post goes out through a host-side HTTP
//! service, and compiling a Teams client into the runtime would couple the
//! runtime to a platform for no gain (Parslee-ai/car#885 marks it explicitly
//! out of scope). So an unclaimed channel reaches the host instead.
//!
//! ## Why the tool-callback path rather than a new FFI surface
//!
//! Issue #885 left "what does adapter registration look like across the FFI
//! boundary?" open. This answers it by *not* opening a new boundary: calling
//! back out to the host through [`car_engine::ToolExecutor`] is exactly how
//! the runtime already reaches host-owned capability today — the daemon's
//! `WsToolExecutor` turns it into a `tools.execute` JSON-RPC request, and
//! every host (WS client, NAPI, PyO3, UniFFI) already implements it. A host
//! that wants Teams handles one more tool name; it registers nothing, links
//! nothing, and no binding surface has to grow a `register_message_adapter`
//! entry point that four FFI crates would then have to keep in sync. If a
//! dedicated registration API is ever warranted, it can be added later without
//! stranding the hosts that took this path.
//!
//! The callback is `messaging.channel_send` with
//! `{channel, kind, to, body}` — deliberately the same `kind`/`to` vocabulary
//! the `messaging.send` tool schema advertises to the model, so a host author
//! reads one shape from the model's JSON through to their handler.

use std::sync::Arc;

use async_trait::async_trait;
use serde_json::{json, Value};

use car_engine::messaging::{MessageReceipt, OutboundMessage, Recipient};
use car_engine::ToolExecutor;
use car_messaging::outbound::OutboundAdapter;

/// The host-side tool name this adapter calls back on.
pub const HOST_CHANNEL_SEND_TOOL: &str = "messaging.channel_send";

/// The name this adapter reports as "its" channel.
///
/// Only meaningful if someone `register`s it as an exact-match adapter; in its
/// intended role — `OutboundRegistry::set_fallback` — routing never consults
/// it, because the whole point is that it answers for channels it cannot
/// enumerate.
pub const HOST_CHANNEL: &str = "host";

/// Delivers a message by asking the host to do it.
pub struct HostChannelAdapter {
    executor: Arc<dyn ToolExecutor>,
}

impl HostChannelAdapter {
    /// Wrap the session's composed tool executor — the same one the runtime
    /// dispatches ordinary tool calls to, so the host sees this callback on
    /// the channel it is already serving.
    pub fn new(executor: Arc<dyn ToolExecutor>) -> Self {
        Self { executor }
    }
}

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

    async fn send(&self, msg: &OutboundMessage) -> Result<MessageReceipt, String> {
        let (kind, to) = match &msg.to {
            Recipient::Direct(handle) => ("direct", handle.as_str()),
            Recipient::Channel(id) => ("channel", id.as_str()),
        };
        let params = json!({
            "channel": msg.channel,
            "kind": kind,
            "to": to,
            "body": msg.body,
        });

        match self.executor.execute(HOST_CHANNEL_SEND_TOOL, &params).await {
            Ok(value) => Ok(receipt_from_host(&msg.channel, &value)),
            // `unknown tool` is this codebase's established sentinel for "this
            // executor does not handle that name" (see `SubstrateShadowExecutor`
            // and the assistant's `ChainedDelegate`, both of which fall through
            // on the same prefix). Reaching it here means the host simply has
            // not implemented the callback — an UNCONFIGURED host, not a
            // runtime fault. Letting the raw text through would tell the model
            // and the operator that CAR is missing a tool it advertises, which
            // sends them looking for a bug that does not exist.
            Err(e) if e.starts_with("unknown tool") => Err(format!(
                "no transport for messaging channel '{}': this host does not implement the \
                 '{}' tool callback. A host that can deliver on '{}' should handle \
                 '{}' with parameters {{channel, kind, to, body}} and return \
                 {{\"message_id\": \"…\"}} (message_id optional).",
                msg.channel, HOST_CHANNEL_SEND_TOOL, msg.channel, HOST_CHANNEL_SEND_TOOL
            )),
            // Any other error is the host's own account of why delivery failed
            // ("not a member of that team", "rate limited"). That is exactly
            // what the model needs to adapt, so it passes through verbatim.
            Err(e) => Err(e),
        }
    }
}

/// Build the receipt from the host's reply.
///
/// `deduplicated` is always `false`: dedup belongs to `OutboundRegistry`, which
/// consults its ledger BEFORE routing, so by the time an adapter runs the send
/// has already been established as one the host has never been asked to make.
/// A host claiming otherwise would be reporting on a ledger it does not own.
fn receipt_from_host(channel: &str, value: &Value) -> MessageReceipt {
    let receipt = MessageReceipt::delivered(channel);
    match value
        .get("message_id")
        .and_then(|v| v.as_str())
        .filter(|s| !s.trim().is_empty())
    {
        Some(id) => receipt.with_message_id(id),
        // Plenty of channels hand back nothing addressable, and a host that
        // returns `true` or `{}` has still delivered. An absent id is not a
        // failure.
        None => receipt,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    /// Records the last callback and replays a canned result.
    struct FakeExecutor {
        seen: Mutex<Vec<(String, Value)>>,
        result: Result<Value, String>,
    }

    impl FakeExecutor {
        fn new(result: Result<Value, String>) -> Arc<Self> {
            Arc::new(Self {
                seen: Mutex::new(Vec::new()),
                result,
            })
        }

        fn last(&self) -> (String, Value) {
            self.seen.lock().unwrap().last().cloned().expect("no call")
        }

        fn calls(&self) -> usize {
            self.seen.lock().unwrap().len()
        }
    }

    #[async_trait]
    impl ToolExecutor for FakeExecutor {
        async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
            self.seen
                .lock()
                .unwrap()
                .push((tool.to_string(), params.clone()));
            self.result.clone()
        }
    }

    fn msg(kind_channel: bool) -> OutboundMessage {
        OutboundMessage {
            channel: "teams".to_string(),
            to: if kind_channel {
                Recipient::Channel("19:meeting@thread.v2".to_string())
            } else {
                Recipient::Direct("keenan@parslee.ai".to_string())
            },
            body: "deploy is green".to_string(),
            idempotency_key: Some("run-42".to_string()),
        }
    }

    #[tokio::test]
    async fn direct_send_builds_the_expected_callback() {
        let exec = FakeExecutor::new(Ok(json!({ "message_id": "1700000000.1" })));
        let adapter = HostChannelAdapter::new(exec.clone());

        let receipt = adapter.send(&msg(false)).await.unwrap();
        assert_eq!(receipt.channel, "teams");
        assert_eq!(receipt.message_id.as_deref(), Some("1700000000.1"));
        assert!(!receipt.deduplicated);

        let (tool, params) = exec.last();
        assert_eq!(tool, HOST_CHANNEL_SEND_TOOL);
        assert_eq!(
            params,
            json!({
                "channel": "teams",
                "kind": "direct",
                "to": "keenan@parslee.ai",
                "body": "deploy is green",
            })
        );
    }

    #[tokio::test]
    async fn channel_send_uses_the_channel_kind() {
        let exec = FakeExecutor::new(Ok(json!({})));
        let adapter = HostChannelAdapter::new(exec.clone());

        let receipt = adapter.send(&msg(true)).await.unwrap();
        // No id in the reply is fine — many channels have none to give.
        assert_eq!(receipt.message_id, None);

        let (_, params) = exec.last();
        assert_eq!(params["kind"], "channel");
        assert_eq!(params["to"], "19:meeting@thread.v2");
    }

    #[tokio::test]
    async fn unknown_tool_becomes_an_actionable_message() {
        let exec = FakeExecutor::new(Err("unknown tool: 'messaging.channel_send'".to_string()));
        let adapter = HostChannelAdapter::new(exec.clone());

        let err = adapter.send(&msg(false)).await.unwrap_err();
        assert!(
            !err.contains("unknown tool"),
            "the raw sentinel must not surface: {err}"
        );
        assert!(
            err.contains("no transport for messaging channel 'teams'"),
            "{err}"
        );
        assert!(err.contains(HOST_CHANNEL_SEND_TOOL), "{err}");
        assert_eq!(exec.calls(), 1);
    }

    #[tokio::test]
    async fn a_host_error_passes_through() {
        let exec = FakeExecutor::new(Err("not a member of that team".to_string()));
        let adapter = HostChannelAdapter::new(exec.clone());

        let err = adapter.send(&msg(false)).await.unwrap_err();
        assert_eq!(err, "not a member of that team");
    }

    #[tokio::test]
    async fn a_blank_message_id_is_treated_as_absent() {
        let exec = FakeExecutor::new(Ok(json!({ "message_id": "   " })));
        let adapter = HostChannelAdapter::new(exec.clone());

        let receipt = adapter.send(&msg(false)).await.unwrap();
        assert_eq!(receipt.message_id, None);
    }
}