car_engine/

messaging.rs

//! Outbound human-directed messaging as a runtime capability.
//!
//! CAR owns tools, state, retry, idempotency, timeouts and rollback — but
//! historically NOT "send a message to a human". Every agent that needed to
//! reach a person hand-rolled its own transport, which put the single most
//! visible side effect an agent can have outside the runtime's sight: the
//! declarative policy engine could not see it, the rate limiter could not
//! bound it, and the event log did not record it.
//!
//! This module is the seam that closes that. The runtime does not own a
//! transport — same as it does not own tools — it owns the *verb*. A host
//! attaches a [`MessageSink`] (see `car-messaging`'s `OutboundRegistry` for
//! the production one), and `messaging.send` becomes a first-class tool that
//! flows through the same validator → policy → rate-limit → eventlog chain as
//! every other side effect.
//!
//! Two invariants are worth stating because they are easy to erode:
//!
//! - **A tool the runtime cannot execute is never advertised.** The schema is
//!   registered by [`crate::Runtime::with_message_sink`] and nowhere else.
//! - **No fall-through.** With no sink attached, `messaging.send` is an error,
//!   not a request handed to the configured host executor. A silent
//!   fall-through would re-open the exact ungoverned path this exists to close.

use serde::{Deserialize, Serialize};
use serde_json::Value;

/// Where a message is addressed.
///
/// Two verbs, not one: a direct message to a person and a post into a shared
/// channel are different acts with different blast radius, and collapsing them
/// into a single opaque "address" string loses one of the two real call shapes
/// — a sink could no longer tell "text Keenan" from "post to #general", and
/// neither could a policy rule written against the parameters.
///
/// Serializes adjacently-tagged as `{"kind": "direct", "to": "<handle>"}`,
/// which is deliberately the same `kind`/`to` pair the `messaging.send` tool
/// schema exposes — one vocabulary from the model's JSON down to the sink.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "to", rename_all = "lowercase")]
pub enum Recipient {
    /// A person, named by whatever handle the channel uses (phone number,
    /// email, workspace member id).
    Direct(String),
    /// A shared channel, named by the channel's own id.
    Channel(String),
}

/// One outbound message, as the runtime hands it to a sink.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OutboundMessage {
    /// Channel name — "imessage", or any name a host adapter answers to.
    /// The runtime keeps no list of valid channels; the sink is the authority.
    pub channel: String,
    /// Who the message is for.
    pub to: Recipient,
    /// The message text, as the human will read it.
    pub body: String,
    /// Owner-phone classification, separate from recipient kind. Omission is a question.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message_kind: Option<String>,
    /// Absolute Unix-millisecond deadline for an owner question.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub deadline_ms: Option<u64>,
    /// Caller-supplied dedup key. A retry after a delivered-but-timed-out send
    /// is the classic duplicate-message bug — the transport succeeded, the
    /// acknowledgement did not arrive, the caller retries, and the human gets
    /// the same message twice. A sink that honours this key makes the retry
    /// safe. `None` means "no dedup", which is the right answer for a message
    /// the caller genuinely wants sent again.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub idempotency_key: Option<String>,
}

/// What a sink reports back about a delivered (or suppressed) send.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MessageReceipt {
    /// The channel that carried it.
    pub channel: String,
    /// Channel-assigned id, when the channel exposes one. Several do not
    /// (iMessage via JXA gives back nothing addressable), hence `Option`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message_id: Option<String>,
    /// Owner-floor topic, readable by the calling session via conversation.get.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topic_id: Option<String>,
    /// True when this send was suppressed because its idempotency key had
    /// already been delivered. Surfaced rather than hidden: a caller that sees
    /// `deduplicated: true` learns its retry was unnecessary, and a caller
    /// that never expected a duplicate learns it has a resend bug.
    pub deduplicated: bool,
    /// Whether transport delivery is confirmed by this receipt.
    #[serde(default)]
    pub delivered: bool,
    /// Why an accepted send was not delivered, when known.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    /// Accepted owner question waiting for its turn on the floor.
    #[serde(default)]
    pub queued: bool,
}

impl MessageReceipt {
    /// A fresh delivery on `channel` with no channel-assigned id.
    pub fn delivered(channel: impl Into<String>) -> Self {
        Self {
            channel: channel.into(),
            message_id: None,
            topic_id: None,
            deduplicated: false,
            queued: false,
            delivered: true,
            reason: None,
        }
    }

    /// Attach a channel-assigned message id.
    pub fn with_message_id(mut self, id: impl Into<String>) -> Self {
        self.message_id = Some(id.into());
        self
    }
}

/// The host-supplied outbound transport the runtime dispatches
/// `messaging.send` to.
///
/// Mirrors the [`crate::ToolExecutor`] arrangement: the runtime owns the
/// governance (validation, policy, rate limiting, logging) and the caller owns
/// the wire. One sink fronts every channel — routing among channels is the
/// sink's job, because only the sink knows which adapters it has.
#[async_trait::async_trait]
pub trait MessageSink: Send + Sync {
    /// Channel names this sink can deliver to.
    ///
    /// Advisory — used to build actionable error messages ("unknown channel
    /// 'slak'; registered: imessage") and for host introspection. It is NOT a
    /// pre-flight gate: [`Self::send`] remains the authority, since a sink's
    /// set of channels can change between the two calls.
    async fn channels(&self) -> Vec<String>;

    /// Deliver one message, or explain why it was not delivered.
    ///
    /// `Err` carries a human-readable reason that reaches the model as the
    /// tool's error, so it should say what would make the send work (pair the
    /// handle, use a direct recipient, name a registered channel).
    async fn send(&self, msg: &OutboundMessage) -> Result<MessageReceipt, String>;
}

impl OutboundMessage {
    /// Parse the `messaging.send` tool parameters into a message.
    ///
    /// Errors are written for two readers at once: the model, which has to fix
    /// the call on its next turn, and the operator reading the event log. Both
    /// need to know which field was wrong and what a correct one looks like,
    /// so every message names the offending key and the accepted shape.
    ///
    /// There is deliberately exactly one accepted spelling of the address
    /// parameter: `to`, the name the schema advertises. Policy rules match the
    /// LITERAL parameter key, so an alias would be an ungoverned spelling of a
    /// governed field — a rule written as `allow_tool_param { param = "to" }`
    /// would simply not see a call that spelled it differently.
    pub fn from_tool_params(params: &Value) -> Result<Self, String> {
        let obj = params
            .as_object()
            .ok_or("messaging.send: parameters must be a JSON object")?;

        let channel = required_str(obj.get("channel"), "channel")?;

        let address = match obj.get("to") {
            Some(v) => required_str(Some(v), "to")?,
            None => {
                return Err("messaging.send: missing required parameter 'to' — \
                            pass 'to' naming the handle or channel id"
                    .to_string())
            }
        };

        // Absent `kind` means the narrower of the two acts. Defaulting to
        // `direct` keeps an under-specified call from broadcasting.
        let to = match obj.get("kind") {
            None | Some(Value::Null) => Recipient::Direct(address),
            Some(Value::String(k)) => match k.as_str() {
                "direct" => Recipient::Direct(address),
                "channel" => Recipient::Channel(address),
                other => {
                    return Err(format!(
                        "messaging.send: unknown kind '{other}' — expected \
                         'direct' (a person) or 'channel' (a shared channel)"
                    ))
                }
            },
            Some(other) => {
                return Err(format!(
                    "messaging.send: 'kind' must be the string 'direct' or \
                     'channel', got {other}"
                ))
            }
        };

        let body = required_str(obj.get("body"), "body")?;

        let idempotency_key = match obj.get("idempotency_key") {
            None | Some(Value::Null) => None,
            Some(Value::String(s)) if !s.trim().is_empty() => Some(s.clone()),
            Some(Value::String(_)) => {
                return Err("messaging.send: 'idempotency_key' must not be blank \
                            — omit it entirely to opt out of dedup"
                    .to_string())
            }
            Some(other) => {
                return Err(format!(
                    "messaging.send: 'idempotency_key' must be a string, got {other}"
                ))
            }
        };

        let deadline_ms = obj
            .get("deadline_ms")
            .filter(|v| !v.is_null())
            .map(|v| {
                v.as_u64().ok_or_else(|| {
                    "messaging.send: 'deadline_ms' must be a Unix-millisecond unsigned integer"
                        .to_string()
                })
            })
            .transpose()?;
        Ok(Self {
            channel,
            to,
            body,
            deadline_ms,
            message_kind: obj
                .get("message_kind")
                .filter(|v| !v.is_null())
                .map(|v| required_str(Some(v), "message_kind"))
                .transpose()?,
            idempotency_key,
        })
    }
}

/// A required string field: present, a string, and not blank. A blank channel
/// routes nowhere, a blank recipient addresses nobody, and a blank body pages
/// a human with silence — all three are bugs, not permissive inputs.
fn required_str(value: Option<&Value>, field: &str) -> Result<String, String> {
    match value {
        None | Some(Value::Null) => Err(format!(
            "messaging.send: missing required parameter '{field}'"
        )),
        Some(Value::String(s)) if !s.trim().is_empty() => Ok(s.clone()),
        Some(Value::String(_)) => Err(format!("messaging.send: '{field}' must not be empty")),
        Some(other) => Err(format!(
            "messaging.send: '{field}' must be a string, got {other}"
        )),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn parses_direct_by_default() {
        let msg = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "to": "+15551112222",
            "body": "build is green",
        }))
        .unwrap();
        assert_eq!(msg.channel, "imessage");
        assert_eq!(msg.to, Recipient::Direct("+15551112222".into()));
        assert_eq!(msg.body, "build is green");
        assert!(msg.idempotency_key.is_none());
    }

    #[test]
    fn parses_explicit_kinds() {
        let direct = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "to": "keenan@parslee.ai",
            "kind": "direct",
            "body": "hi",
            "idempotency_key": "run-42",
        }))
        .unwrap();
        assert_eq!(direct.to, Recipient::Direct("keenan@parslee.ai".into()));
        assert_eq!(direct.idempotency_key.as_deref(), Some("run-42"));

        let channel = OutboundMessage::from_tool_params(&json!({
            "channel": "slack",
            "to": "C012ABCDEF",
            "kind": "channel",
            "body": "deploy done",
        }))
        .unwrap();
        assert_eq!(channel.to, Recipient::Channel("C012ABCDEF".into()));
    }

    /// `to` is the only accepted spelling. Policy matches the literal parameter
    /// key, so an alias would be an ungoverned spelling of a governed field.
    #[test]
    fn rejects_recipient_as_an_alias_for_to() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "recipient": "+15551112222",
            "body": "hi",
        }))
        .unwrap_err();
        assert!(err.contains("missing required parameter 'to'"), "{err}");
    }

    #[test]
    fn rejects_non_object_params() {
        let err = OutboundMessage::from_tool_params(&json!("just a string")).unwrap_err();
        assert!(err.contains("must be a JSON object"), "{err}");
    }

    #[test]
    fn rejects_missing_channel() {
        let err = OutboundMessage::from_tool_params(&json!({
            "to": "+15551112222",
            "body": "hi",
        }))
        .unwrap_err();
        assert!(
            err.contains("missing required parameter 'channel'"),
            "{err}"
        );
    }

    #[test]
    fn rejects_blank_channel() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "   ",
            "to": "+15551112222",
            "body": "hi",
        }))
        .unwrap_err();
        assert!(err.contains("'channel' must not be empty"), "{err}");
    }

    #[test]
    fn rejects_missing_recipient() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "body": "hi",
        }))
        .unwrap_err();
        assert!(err.contains("missing required parameter 'to'"), "{err}");
    }

    #[test]
    fn rejects_missing_body() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "to": "+15551112222",
        }))
        .unwrap_err();
        assert!(err.contains("missing required parameter 'body'"), "{err}");
    }

    #[test]
    fn rejects_non_string_body() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "to": "+15551112222",
            "body": 42,
        }))
        .unwrap_err();
        assert!(err.contains("'body' must be a string"), "{err}");
    }

    #[test]
    fn rejects_unknown_kind() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "to": "+15551112222",
            "kind": "broadcast",
            "body": "hi",
        }))
        .unwrap_err();
        assert!(err.contains("unknown kind 'broadcast'"), "{err}");
        assert!(err.contains("'direct'"), "{err}");
    }

    #[test]
    fn rejects_non_string_kind() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "to": "+15551112222",
            "kind": true,
            "body": "hi",
        }))
        .unwrap_err();
        assert!(err.contains("'kind' must be the string"), "{err}");
    }

    #[test]
    fn rejects_blank_idempotency_key() {
        let err = OutboundMessage::from_tool_params(&json!({
            "channel": "imessage",
            "to": "+15551112222",
            "body": "hi",
            "idempotency_key": "",
        }))
        .unwrap_err();
        assert!(err.contains("'idempotency_key' must not be blank"), "{err}");
    }

    #[test]
    fn recipient_serializes_with_the_tool_vocabulary() {
        // The wire shape a sink (or a policy rule) sees must be the same
        // `kind`/`to` pair the tool schema advertises.
        let json = serde_json::to_value(Recipient::Direct("+15551112222".into())).unwrap();
        assert_eq!(json, json!({ "kind": "direct", "to": "+15551112222" }));
        let round: Recipient = serde_json::from_value(json).unwrap();
        assert_eq!(round, Recipient::Direct("+15551112222".into()));

        let json = serde_json::to_value(Recipient::Channel("C1".into())).unwrap();
        assert_eq!(json, json!({ "kind": "channel", "to": "C1" }));
    }

    #[test]
    fn receipt_round_trips() {
        let receipt = MessageReceipt::delivered("imessage").with_message_id("m-1");
        let json = serde_json::to_value(&receipt).unwrap();
        assert_eq!(
            json,
            json!({ "channel": "imessage", "message_id": "m-1", "deduplicated": false, "queued": false, "delivered": true })
        );
        let round: MessageReceipt = serde_json::from_value(json).unwrap();
        assert_eq!(round, receipt);
    }
}