soma_integrations/
gateway.rs

1//! Implements `soma-application`'s [`GatewayPort`] over `soma-gateway`'s
2//! `GatewayManager`, translating Soma product principal/scope types into the
3//! generic gateway's access and route-scope types (plan section 3.20,
4//! "GatewayControl" in section 5's illustrative flow).
5//!
6//! Moved out of `apps/soma` (formerly `application_ports.rs`), where it was a
7//! temporary bridge kept there only for mergeability — this crate is its
8//! permanent home per PR 11's acceptance criterion that `apps/soma`
9//! constructs adapters but contains none of their implementation logic.
10
11use std::sync::Arc;
12
13use async_trait::async_trait;
14use serde_json::Value;
15
16use soma_application::{
17    ExecutionContext, GatewayExecuteRequest, GatewayMcpOutcome, GatewayMcpRoundTrip, GatewayPort,
18    GatewayPromptRoute, GatewayReloadRequest, GatewayResourceRoute, GatewayRouteScope,
19    GatewayToolRoute, PortError,
20};
21use soma_domain::{
22    AuthorizationMode,
23    actions::{READ_SCOPE, scopes_satisfy},
24    scopes::has_admin_scope,
25};
26use soma_gateway::gateway::dispatch::{
27    GatewayAccess, GatewayDispatchError, dispatch_gateway_action,
28};
29use soma_gateway::{
30    gateway::{
31        manager::GatewayManager, manager::GatewayManagerError,
32        protected_routes::ProtectedRouteScope,
33    },
34    upstream::{McpRequestOutcome, McpRoundTrip},
35};
36
37/// `soma-gateway`'s `GatewayProductState` is `Arc<GatewayManager>` — this
38/// adapter takes the manager directly rather than depending on `soma-runtime`
39/// for the type alias, keeping this crate's dependency shape limited to
40/// `soma-application`, `soma-domain`, and `soma-gateway`.
41pub struct GatewayApplicationPort {
42    gateway: Arc<GatewayManager>,
43}
44
45impl GatewayApplicationPort {
46    pub fn new(gateway: Arc<GatewayManager>) -> Self {
47        Self { gateway }
48    }
49
50    async fn dispatch(
51        &self,
52        action: &str,
53        params: Value,
54        context: &ExecutionContext,
55    ) -> Result<Value, PortError> {
56        dispatch_gateway_action(&self.gateway, gateway_access(context), action, params)
57            .await
58            .map_err(|error| gateway_port_error(action, error))
59    }
60}
61
62#[async_trait]
63impl GatewayPort for GatewayApplicationPort {
64    async fn status(&self, context: &ExecutionContext) -> Result<Value, PortError> {
65        self.dispatch("gateway.list", serde_json::json!({}), context)
66            .await
67    }
68
69    async fn reload(
70        &self,
71        request: GatewayReloadRequest,
72        context: &ExecutionContext,
73    ) -> Result<Value, PortError> {
74        self.dispatch("gateway.reload", request.config, context)
75            .await
76    }
77
78    async fn execute(
79        &self,
80        request: GatewayExecuteRequest,
81        context: &ExecutionContext,
82    ) -> Result<Value, PortError> {
83        self.dispatch(&request.action, request.params, context)
84            .await
85    }
86
87    async fn list_mcp_tools(
88        &self,
89        scope: Option<&GatewayRouteScope>,
90        context: &ExecutionContext,
91    ) -> Result<Vec<GatewayToolRoute>, PortError> {
92        let scope = scope.map(protected_route_scope);
93        self.gateway
94            .tool_routes_for_subject_and_scope(Some(gateway_subject(context)), scope.as_ref())
95            .await
96            .map(|routes| {
97                routes
98                    .into_iter()
99                    .map(|route| GatewayToolRoute {
100                        name: route.name,
101                        description: route.descriptor.description,
102                        input_schema: route.descriptor.input_schema,
103                        output_schema: route.descriptor.output_schema,
104                        destructive: route.descriptor.destructive,
105                    })
106                    .collect()
107            })
108            .map_err(|error| gateway_manager_port_error("tools/list", error))
109    }
110
111    async fn call_mcp_tool(
112        &self,
113        name: &str,
114        params: Value,
115        scope: Option<&GatewayRouteScope>,
116        context: &ExecutionContext,
117    ) -> Result<Option<Value>, PortError> {
118        let scope = scope.map(protected_route_scope);
119        self.gateway
120            .call_mcp_tool_for_subject_and_scope(
121                name,
122                params,
123                Some(gateway_subject(context)),
124                scope.as_ref(),
125            )
126            .await
127            .map_err(|error| gateway_manager_port_error("tools/call", error))
128    }
129
130    async fn call_mcp_tool_once(
131        &self,
132        name: &str,
133        params: Value,
134        round_trip: GatewayMcpRoundTrip,
135        scope: Option<&GatewayRouteScope>,
136        context: &ExecutionContext,
137    ) -> Result<Option<GatewayMcpOutcome>, PortError> {
138        let scope = scope.map(protected_route_scope);
139        self.gateway
140            .call_mcp_tool_once_for_subject_and_scope(
141                name,
142                params,
143                mcp_round_trip(round_trip),
144                Some(gateway_subject(context)),
145                scope.as_ref(),
146            )
147            .await
148            .map(|outcome| outcome.map(gateway_mcp_outcome))
149            .map_err(|error| gateway_manager_port_error("tools/call", error))
150    }
151
152    async fn get_mcp_task(
153        &self,
154        task_id: &str,
155        context: &ExecutionContext,
156    ) -> Result<Value, PortError> {
157        self.gateway
158            .get_mcp_task_for_subject(task_id, Some(gateway_subject(context)))
159            .await
160            .map_err(|error| gateway_manager_port_error("tasks/get", error))
161    }
162
163    async fn update_mcp_task(
164        &self,
165        task_id: &str,
166        input_responses: std::collections::BTreeMap<String, Value>,
167        context: &ExecutionContext,
168    ) -> Result<(), PortError> {
169        self.gateway
170            .update_mcp_task_for_subject(task_id, input_responses, Some(gateway_subject(context)))
171            .await
172            .map_err(|error| gateway_manager_port_error("tasks/update", error))
173    }
174
175    async fn cancel_mcp_task(
176        &self,
177        task_id: &str,
178        context: &ExecutionContext,
179    ) -> Result<(), PortError> {
180        self.gateway
181            .cancel_mcp_task_for_subject(task_id, Some(gateway_subject(context)))
182            .await
183            .map_err(|error| gateway_manager_port_error("tasks/cancel", error))
184    }
185
186    async fn list_mcp_resources(
187        &self,
188        scope: Option<&GatewayRouteScope>,
189        context: &ExecutionContext,
190    ) -> Result<Vec<GatewayResourceRoute>, PortError> {
191        let scope = scope.map(protected_route_scope);
192        self.gateway
193            .resource_routes_for_subject_and_scope(Some(gateway_subject(context)), scope.as_ref())
194            .await
195            .map(|routes| {
196                routes
197                    .into_iter()
198                    .map(|route| GatewayResourceRoute {
199                        uri: route.uri,
200                        native_uri: route.native_uri,
201                        name: route.descriptor.name,
202                    })
203                    .collect()
204            })
205            .map_err(|error| gateway_manager_port_error("resources/list", error))
206    }
207
208    async fn read_mcp_resource(
209        &self,
210        uri: &str,
211        scope: Option<&GatewayRouteScope>,
212        context: &ExecutionContext,
213    ) -> Result<Option<Value>, PortError> {
214        let scope = scope.map(protected_route_scope);
215        self.gateway
216            .read_mcp_resource_for_subject_and_scope(
217                uri,
218                Some(gateway_subject(context)),
219                scope.as_ref(),
220            )
221            .await
222            .map_err(|error| gateway_manager_port_error("resources/read", error))
223    }
224
225    async fn read_mcp_resource_once(
226        &self,
227        uri: &str,
228        round_trip: GatewayMcpRoundTrip,
229        scope: Option<&GatewayRouteScope>,
230        context: &ExecutionContext,
231    ) -> Result<Option<GatewayMcpOutcome>, PortError> {
232        let scope = scope.map(protected_route_scope);
233        self.gateway
234            .read_mcp_resource_once_for_subject_and_scope(
235                uri,
236                mcp_round_trip(round_trip),
237                Some(gateway_subject(context)),
238                scope.as_ref(),
239            )
240            .await
241            .map(|outcome| outcome.map(gateway_mcp_outcome))
242            .map_err(|error| gateway_manager_port_error("resources/read", error))
243    }
244
245    async fn list_mcp_prompts(
246        &self,
247        scope: Option<&GatewayRouteScope>,
248        context: &ExecutionContext,
249    ) -> Result<Vec<GatewayPromptRoute>, PortError> {
250        let scope = scope.map(protected_route_scope);
251        self.gateway
252            .prompt_routes_for_subject_and_scope(Some(gateway_subject(context)), scope.as_ref())
253            .await
254            .map(|routes| {
255                routes
256                    .into_iter()
257                    .map(|route| GatewayPromptRoute {
258                        name: route.name,
259                        description: route.descriptor.description,
260                    })
261                    .collect()
262            })
263            .map_err(|error| gateway_manager_port_error("prompts/list", error))
264    }
265
266    async fn get_mcp_prompt(
267        &self,
268        name: &str,
269        arguments: Option<serde_json::Map<String, Value>>,
270        scope: Option<&GatewayRouteScope>,
271        context: &ExecutionContext,
272    ) -> Result<Option<Value>, PortError> {
273        let scope = scope.map(protected_route_scope);
274        self.gateway
275            .get_mcp_prompt_for_subject_and_scope(
276                name,
277                arguments,
278                Some(gateway_subject(context)),
279                scope.as_ref(),
280            )
281            .await
282            .map_err(|error| gateway_manager_port_error("prompts/get", error))
283    }
284    async fn get_mcp_prompt_once(
285        &self,
286        name: &str,
287        arguments: Option<serde_json::Map<String, Value>>,
288        round_trip: GatewayMcpRoundTrip,
289        scope: Option<&GatewayRouteScope>,
290        context: &ExecutionContext,
291    ) -> Result<Option<GatewayMcpOutcome>, PortError> {
292        let scope = scope.map(protected_route_scope);
293        self.gateway
294            .get_mcp_prompt_once_for_subject_and_scope(
295                name,
296                arguments,
297                mcp_round_trip(round_trip),
298                Some(gateway_subject(context)),
299                scope.as_ref(),
300            )
301            .await
302            .map(|outcome| outcome.map(gateway_mcp_outcome))
303            .map_err(|error| gateway_manager_port_error("prompts/get", error))
304    }
305}
306
307fn mcp_round_trip(round_trip: GatewayMcpRoundTrip) -> McpRoundTrip {
308    McpRoundTrip {
309        input_responses: round_trip.input_responses,
310        request_state: round_trip.request_state,
311        request_meta: round_trip.request_meta,
312    }
313}
314
315fn gateway_mcp_outcome(outcome: McpRequestOutcome) -> GatewayMcpOutcome {
316    match outcome {
317        McpRequestOutcome::Complete(value) => GatewayMcpOutcome::Complete(value),
318        McpRequestOutcome::InputRequired(value) => GatewayMcpOutcome::InputRequired(value),
319        McpRequestOutcome::Task(value) => GatewayMcpOutcome::Task(value),
320    }
321}
322
323fn gateway_subject(context: &ExecutionContext) -> &str {
324    const SHARED_GATEWAY_SUBJECT: &str = "gateway";
325    if !matches!(context.authorization_mode, AuthorizationMode::Mounted) {
326        return SHARED_GATEWAY_SUBJECT;
327    }
328    let Some(principal) = context.principal.as_ref() else {
329        return SHARED_GATEWAY_SUBJECT;
330    };
331    if principal.issuer.as_deref() == Some("local") || has_admin_scope(&principal.scopes.to_vec()) {
332        SHARED_GATEWAY_SUBJECT
333    } else {
334        &principal.subject
335    }
336}
337
338fn protected_route_scope(scope: &GatewayRouteScope) -> ProtectedRouteScope {
339    ProtectedRouteScope {
340        upstreams: scope.upstreams.clone(),
341        services: scope.services.clone(),
342        expose_code_mode: scope.expose_code_mode,
343    }
344}
345
346fn gateway_access(context: &ExecutionContext) -> GatewayAccess {
347    if !matches!(context.authorization_mode, AuthorizationMode::Mounted) {
348        return GatewayAccess {
349            read: true,
350            admin: true,
351        };
352    }
353    let scopes = context
354        .principal
355        .as_ref()
356        .map(|principal| principal.scopes.to_vec())
357        .unwrap_or_default();
358    let admin = has_admin_scope(&scopes);
359    GatewayAccess {
360        read: admin || scopes_satisfy(&scopes, READ_SCOPE),
361        admin,
362    }
363}
364
365fn gateway_port_error(action: &str, error: GatewayDispatchError) -> PortError {
366    let message = error.to_string();
367    port_error_from_structured(error.structured(action), message)
368}
369
370/// `soma-gateway`'s own `GatewayDispatchError::structured()` already gives an
371/// exhaustive, per-variant `code`/`kind`/`remediation` mapping for every
372/// `GatewayManagerError` (including every `UpstreamError` case) via its
373/// `Manager` variant — reuse it here instead of collapsing every MCP
374/// tools/resources/prompts proxy failure into one `gateway_proxy_failed` code
375/// with a blanket `retryable: true`, which previously made permanent
376/// failures (e.g. an unknown/misconfigured upstream) look identical to
377/// transient ones (e.g. a live connect/call failure).
378fn gateway_manager_port_error(operation: &str, error: GatewayManagerError) -> PortError {
379    let message = format!("{operation} failed: {error}");
380    port_error_from_structured(
381        GatewayDispatchError::from(error).structured(operation),
382        message,
383    )
384}
385
386fn port_error_from_structured(
387    structured: soma_gateway::dispatch_helpers::GatewayStructuredError,
388    message: String,
389) -> PortError {
390    PortError {
391        code: structured.code.to_owned(),
392        message,
393        retryable: matches!(structured.kind, "runtime"),
394        remediation: structured.remediation.to_owned(),
395    }
396}
397
398#[cfg(test)]
399#[path = "gateway_tests.rs"]
400mod tests;
soma_integrations/gateway.rs

soma_integrations/
gateway.rs
