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};
29
30use a2a::{authenticated_user, build_a2a_request, mint_a2a_token, run_agent};
31use identity::resolve_or_link_user;
32
33/// Process-wide HTTP client for every outbound platform call. Cloning shares
34/// the underlying connection pool; a fresh `reqwest::Client` per reply would
35/// discard keep-alive connections and TLS sessions.
36static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(reqwest::Client::new);
37
38#[must_use]
39pub fn http_client() -> reqwest::Client {
40    CLIENT.clone()
41}
42
43#[derive(Debug, Clone)]
44pub enum ReplyTarget {
45    Channel { id: String },
46    Url { url: String },
47}
48
49/// A surface-agnostic inbound message ready for dispatch. Per-platform routes
50/// build this from their normalized payload; the dispatch core never sees a
51/// Slack- or Teams-specific type.
52#[derive(Debug, Clone)]
53pub struct MessagingInbound {
54    /// Stable platform tag (`"slack"` / `"teams"`), used in the derived
55    /// `ContextId` and the authz context kind.
56    pub platform: &'static str,
57    pub issuer: String,
58    pub org_id: String,
59    pub channel_id: String,
60    pub external_user_id: String,
61    pub text: String,
62    pub agent_name: AgentName,
63    /// The authz target — the workspace/tenant entity, config-seedable from
64    /// `allowed_roles`.
65    pub entity: EntityRef,
66    pub reply: ReplyTarget,
67}
68
69#[derive(Debug, Clone)]
70pub enum DispatchOutcome {
71    /// Authorized; the agent's reply text (may be empty if the agent produced
72    /// no message).
73    Replied(String),
74    /// Authorization denied; the reason is surfaced as an ephemeral refusal.
75    Denied(String),
76}
77
78/// Failures along the dispatch pipeline. This is an internal system surface;
79/// messages are deliberately descriptive for operator debugging.
80#[derive(Debug, thiserror::Error)]
81pub enum MessagingError {
82    #[error("identity resolution failed: {0}")]
83    Identity(String),
84    #[error("token minting failed: {0}")]
85    Token(String),
86    #[error("agent dispatch failed: {0}")]
87    Dispatch(String),
88    #[error("malformed agent response: {0}")]
89    Response(String),
90}
91
92/// Run the shared inbound pipeline.
93///
94/// Links the platform sender to a governed identity, authorizes against the
95/// workspace/tenant entity, mints a per-user A2A token, dispatches a blocking
96/// `message/send` through the proxy, and returns the agent's reply.
97pub async fn dispatch_messaging(
98    ctx: &AppContext,
99    inbound: MessagingInbound,
100) -> Result<DispatchOutcome, MessagingError> {
101    let user = resolve_or_link_user(ctx, &inbound.issuer, &inbound.external_user_id).await?;
102    let authed = authenticated_user(&user)?;
103
104    let context_id =
105        ContextId::derived_from_messaging(inbound.platform, &inbound.org_id, &inbound.channel_id);
106
107    let authz = AuthzRequest {
108        entity: inbound.entity.clone(),
109        user_id: user.id.clone(),
110        roles: user.roles.clone(),
111        attributes: std::collections::BTreeMap::new(),
112        trace_id: TraceId::generate(),
113        session_id: None,
114        context: AuthzContext::extension(
115            format!("{}.message", inbound.platform),
116            json!({ "channel": inbound.channel_id }),
117        ),
118        context_id: Some(context_id.clone()),
119        task_id: None,
120        act_chain: Vec::new(),
121    };
122    if let AuthzDecision::Deny { reason, policy } = ctx.authz_hook().evaluate(authz).await {
123        return Ok(DispatchOutcome::Denied(format!("{policy}: {reason}")));
124    }
125
126    let session_id = SessionId::new(uuid::Uuid::new_v4().to_string());
127    let token = mint_a2a_token(ctx, &authed, &session_id)?;
128
129    let request = build_a2a_request(&inbound, &authed, &session_id, &token, &context_id)?;
130    let reply = run_agent(ctx, inbound.agent_name.as_str(), request).await?;
131    Ok(DispatchOutcome::Replied(reply))
132}
133
134#[cfg(feature = "test-api")]
135pub mod test_api {
136    use systemprompt_agent::models::a2a::Task;
137
138    #[must_use]
139    pub fn reply_text(task: Option<&Task>) -> String {
140        super::a2a::reply_text(task)
141    }
142}
systemprompt_api/routes/messaging/mod.rs

systemprompt_api/routes/messaging/
mod.rs
