systemprompt_api/routes/messaging/
mod.rs

1//! Platform-agnostic dispatch for chat-platform inbound messages.
2//!
3//! Slack and Teams differ only at their edges — request verification, payload
4//! shape, and reply rendering. Everything between (identity, authorization,
5//! deterministic conversation context, per-user A2A token minting, the blocking
6//! `message/send` through the proxy, and reply extraction) is identical and
7//! lives here once. A per-platform route normalizes its wire payload into a
8//! [`MessagingInbound`] and calls [`dispatch_messaging`]; the returned
9//! [`DispatchOutcome`] is rendered back into the platform's UI by the route.
10//!
11//! The pipeline is **synchronous, spawned**: the route acks the platform within
12//! its timeout, then a spawned task runs this blocking dispatch and posts the
13//! reply. There is no responder job and no dispatch-state table — a stable
14//! [`ContextId`] (derived from the conversation) ties multi-turn history
15//! together instead.
16//!
17//! Copyright (c) systemprompt.io — Business Source License 1.1.
18//! See <https://systemprompt.io> for licensing details.
19
20mod a2a;
21pub mod identity;
22
23use std::sync::LazyLock;
24
25use serde_json::json;
26use systemprompt_identifiers::{AgentName, ContextId, SessionId, TraceId};
27use systemprompt_runtime::AppContext;
28use systemprompt_security::authz::{AuthzContext, AuthzDecision, AuthzRequest, EntityRef};
29use systemprompt_traits::FederatedIdentityClaims;
30
31use a2a::{authenticated_user, build_a2a_request, mint_a2a_token, run_agent};
32use identity::resolve_or_link_user;
33
34static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(reqwest::Client::new);
35
36#[must_use]
37pub fn http_client() -> reqwest::Client {
38    CLIENT.clone()
39}
40
41#[derive(Debug, Clone)]
42pub enum ReplyTarget {
43    Channel { id: String },
44    Url { url: String },
45}
46
47/// A surface-agnostic inbound message ready for dispatch. Per-platform routes
48/// build this from their normalized payload; the dispatch core never sees a
49/// Slack- or Teams-specific type.
50#[derive(Debug, Clone)]
51pub struct MessagingInbound {
52    pub platform: &'static str,
53    pub issuer: String,
54    pub org_id: String,
55    pub channel_id: String,
56    pub external_user_id: String,
57    pub text: String,
58    pub agent_name: AgentName,
59    pub entity: EntityRef,
60    pub reply: ReplyTarget,
61    // Why: empty claims mean "unlinked" -- the sender resolves to a role-less
62    // first-touch user, which no rule grants anything to. A platform route that
63    // cannot read a profile must pass empty rather than guess.
64    pub claims: FederatedIdentityClaims,
65}
66
67#[derive(Debug, Clone)]
68pub enum DispatchOutcome {
69    Replied(String),
70    Denied(String),
71}
72
73/// Failures along the dispatch pipeline. This is an internal system surface;
74/// messages are deliberately descriptive for operator debugging.
75#[derive(Debug, thiserror::Error)]
76pub enum MessagingError {
77    #[error("identity resolution failed: {0}")]
78    Identity(String),
79    #[error("token minting failed: {0}")]
80    Token(String),
81    #[error("agent dispatch failed: {0}")]
82    Dispatch(String),
83    #[error("malformed agent response: {0}")]
84    Response(String),
85}
86
87impl MessagingError {
88    #[must_use]
89    pub fn user_message(&self) -> String {
90        let opaque = "Sorry — something went wrong handling that.";
91        if cfg!(feature = "test-api") {
92            format!("{opaque} ({self})")
93        } else {
94            opaque.to_owned()
95        }
96    }
97}
98
99pub async fn dispatch_messaging(
100    ctx: &AppContext,
101    inbound: MessagingInbound,
102) -> Result<DispatchOutcome, MessagingError> {
103    let user = resolve_or_link_user(
104        ctx,
105        &inbound.issuer,
106        &inbound.external_user_id,
107        &inbound.claims,
108    )
109    .await?;
110    let authed = authenticated_user(&user)?;
111
112    let context_id =
113        ContextId::derived_from_messaging(inbound.platform, &inbound.org_id, &inbound.channel_id);
114
115    let authz = AuthzRequest {
116        entity: inbound.entity.clone(),
117        user_id: user.id.clone(),
118        roles: user.roles.clone(),
119        attributes: std::collections::BTreeMap::new(),
120        trace_id: TraceId::generate(),
121        session_id: None,
122        context: AuthzContext::extension(
123            format!("{}.message", inbound.platform),
124            json!({ "channel": inbound.channel_id }),
125        ),
126        context_id: Some(context_id.clone()),
127        task_id: None,
128        act_chain: Vec::new(),
129    };
130    if let AuthzDecision::Deny { reason, policy } = ctx.authz_hook().evaluate(authz).await {
131        return Ok(DispatchOutcome::Denied(format!("{policy}: {reason}")));
132    }
133
134    let session_id = SessionId::new(uuid::Uuid::new_v4().to_string());
135    let token = mint_a2a_token(ctx, &authed, &session_id)?;
136
137    let request = build_a2a_request(&inbound, &authed, &session_id, &token, &context_id)?;
138    let reply = run_agent(ctx, inbound.agent_name.as_str(), request).await?;
139    Ok(DispatchOutcome::Replied(reply))
140}
141
142#[cfg(feature = "test-api")]
143pub mod test_api {
144    use systemprompt_agent::models::a2a::Task;
145    use systemprompt_models::auth::Permission;
146
147    #[must_use]
148    pub fn reply_text(task: Option<&Task>) -> String {
149        super::a2a::reply_text(task)
150    }
151
152    #[must_use]
153    pub fn permissions_for(roles: &[String]) -> Vec<Permission> {
154        super::a2a::permissions_for(roles)
155    }
156}
systemprompt_api/routes/messaging/mod.rs

systemprompt_api/routes/messaging/
mod.rs
