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
17mod a2a;
18pub mod identity;
19
20use std::sync::LazyLock;
21
22use serde_json::json;
23use systemprompt_identifiers::{AgentName, ContextId, SessionId, TraceId};
24use systemprompt_runtime::AppContext;
25use systemprompt_security::authz::{AuthzContext, AuthzDecision, AuthzRequest, EntityRef};
26
27use a2a::{authenticated_user, build_a2a_request, mint_a2a_token, run_agent};
28use identity::resolve_or_link_user;
29
30/// Process-wide HTTP client for every outbound platform call. Cloning shares
31/// the underlying connection pool; a fresh `reqwest::Client` per reply would
32/// discard keep-alive connections and TLS sessions.
33static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(reqwest::Client::new);
34
35#[must_use]
36pub fn http_client() -> reqwest::Client {
37 CLIENT.clone()
38}
39
40/// Where a rendered reply is posted back to.
41#[derive(Debug, Clone)]
42pub enum ReplyTarget {
43 /// Post into a channel/conversation by id (Slack `chat.postMessage`, Teams
44 /// Bot Connector).
45 Channel { id: String },
46 /// Reply to a captured callback URL (Slack `response_url`).
47 Url { url: String },
48}
49
50/// A surface-agnostic inbound message ready for dispatch. Per-platform routes
51/// build this from their normalized payload; the dispatch core never sees a
52/// Slack- or Teams-specific type.
53#[derive(Debug, Clone)]
54pub struct MessagingInbound {
55 /// Stable platform tag (`"slack"` / `"teams"`), used in the derived
56 /// `ContextId` and the authz context kind.
57 pub platform: &'static str,
58 /// The federated issuer that namespaces `external_user_id`.
59 pub issuer: String,
60 /// Workspace/tenant id — the org the conversation belongs to.
61 pub org_id: String,
62 /// Channel/conversation id within the org.
63 pub channel_id: String,
64 /// The platform's user id (already verified by the route).
65 pub external_user_id: String,
66 pub text: String,
67 /// The agent resolved from app config for this conversation.
68 pub agent_name: AgentName,
69 /// The authz target — the workspace/tenant entity, config-seedable from
70 /// `allowed_roles`.
71 pub entity: EntityRef,
72 /// Where the rendered reply is posted (owned by the route).
73 pub reply: ReplyTarget,
74}
75
76/// The result of a dispatch, rendered back into the platform by the route.
77#[derive(Debug, Clone)]
78pub enum DispatchOutcome {
79 /// Authorized; the agent's reply text (may be empty if the agent produced
80 /// no message).
81 Replied(String),
82 /// Authorization denied; the reason is surfaced as an ephemeral refusal.
83 Denied(String),
84}
85
86/// Failures along the dispatch pipeline. This is an internal system surface;
87/// messages are deliberately descriptive for operator debugging.
88#[derive(Debug, thiserror::Error)]
89pub enum MessagingError {
90 #[error("identity resolution failed: {0}")]
91 Identity(String),
92 #[error("token minting failed: {0}")]
93 Token(String),
94 #[error("agent dispatch failed: {0}")]
95 Dispatch(String),
96 #[error("malformed agent response: {0}")]
97 Response(String),
98}
99
100/// Run the shared inbound pipeline.
101///
102/// Links the platform sender to a governed identity, authorizes against the
103/// workspace/tenant entity, mints a per-user A2A token, dispatches a blocking
104/// `message/send` through the proxy, and returns the agent's reply.
105pub async fn dispatch_messaging(
106 ctx: &AppContext,
107 inbound: MessagingInbound,
108) -> Result<DispatchOutcome, MessagingError> {
109 let user = resolve_or_link_user(ctx, &inbound.issuer, &inbound.external_user_id).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/// Test-only re-export of the agent-reply extraction, so it can be unit-tested
143/// without driving the full dispatch pipeline.
144#[cfg(feature = "test-api")]
145pub mod test_api {
146 use systemprompt_agent::models::a2a::Task;
147
148 #[must_use]
149 pub fn reply_text(task: Option<&Task>) -> String {
150 super::a2a::reply_text(task)
151 }
152}