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, GatewayPort, GatewayPromptRoute, GatewayReloadRequest,
18    GatewayResourceRoute, GatewayRouteScope, GatewayToolRoute, PortError,
19};
20use soma_domain::{
21    actions::{scopes_satisfy, READ_SCOPE},
22    scopes::has_admin_scope,
23    AuthorizationMode,
24};
25use soma_gateway::gateway::dispatch::{
26    dispatch_gateway_action, GatewayAccess, GatewayDispatchError,
27};
28use soma_gateway::gateway::{
29    manager::GatewayManager, manager::GatewayManagerError, protected_routes::ProtectedRouteScope,
30};
31
32/// `soma-gateway`'s `GatewayProductState` is `Arc<GatewayManager>` — this
33/// adapter takes the manager directly rather than depending on `soma-runtime`
34/// for the type alias, keeping this crate's dependency shape limited to
35/// `soma-application`, `soma-domain`, and `soma-gateway`.
36pub struct GatewayApplicationPort {
37    gateway: Arc<GatewayManager>,
38}
39
40impl GatewayApplicationPort {
41    pub fn new(gateway: Arc<GatewayManager>) -> Self {
42        Self { gateway }
43    }
44
45    async fn dispatch(
46        &self,
47        action: &str,
48        params: Value,
49        context: &ExecutionContext,
50    ) -> Result<Value, PortError> {
51        dispatch_gateway_action(&self.gateway, gateway_access(context), action, params)
52            .await
53            .map_err(|error| gateway_port_error(action, error))
54    }
55}
56
57#[async_trait]
58impl GatewayPort for GatewayApplicationPort {
59    async fn status(&self, context: &ExecutionContext) -> Result<Value, PortError> {
60        self.dispatch("gateway.list", serde_json::json!({}), context)
61            .await
62    }
63
64    async fn reload(
65        &self,
66        request: GatewayReloadRequest,
67        context: &ExecutionContext,
68    ) -> Result<Value, PortError> {
69        self.dispatch("gateway.reload", request.config, context)
70            .await
71    }
72
73    async fn execute(
74        &self,
75        request: GatewayExecuteRequest,
76        context: &ExecutionContext,
77    ) -> Result<Value, PortError> {
78        self.dispatch(&request.action, request.params, context)
79            .await
80    }
81
82    async fn list_mcp_tools(
83        &self,
84        scope: Option<&GatewayRouteScope>,
85        context: &ExecutionContext,
86    ) -> Result<Vec<GatewayToolRoute>, PortError> {
87        let scope = scope.map(protected_route_scope);
88        self.gateway
89            .tool_routes_for_subject_and_scope(Some(gateway_subject(context)), scope.as_ref())
90            .await
91            .map(|routes| {
92                routes
93                    .into_iter()
94                    .map(|route| GatewayToolRoute {
95                        name: route.name,
96                        description: route.descriptor.description,
97                        input_schema: route.descriptor.input_schema,
98                        output_schema: route.descriptor.output_schema,
99                        destructive: route.descriptor.destructive,
100                    })
101                    .collect()
102            })
103            .map_err(|error| gateway_manager_port_error("tools/list", error))
104    }
105
106    async fn call_mcp_tool(
107        &self,
108        name: &str,
109        params: Value,
110        scope: Option<&GatewayRouteScope>,
111        context: &ExecutionContext,
112    ) -> Result<Option<Value>, PortError> {
113        let scope = scope.map(protected_route_scope);
114        self.gateway
115            .call_mcp_tool_for_subject_and_scope(
116                name,
117                params,
118                Some(gateway_subject(context)),
119                scope.as_ref(),
120            )
121            .await
122            .map_err(|error| gateway_manager_port_error("tools/call", error))
123    }
124
125    async fn list_mcp_resources(
126        &self,
127        scope: Option<&GatewayRouteScope>,
128        context: &ExecutionContext,
129    ) -> Result<Vec<GatewayResourceRoute>, PortError> {
130        let scope = scope.map(protected_route_scope);
131        self.gateway
132            .resource_routes_for_subject_and_scope(Some(gateway_subject(context)), scope.as_ref())
133            .await
134            .map(|routes| {
135                routes
136                    .into_iter()
137                    .map(|route| GatewayResourceRoute {
138                        uri: route.uri,
139                        native_uri: route.native_uri,
140                        name: route.descriptor.name,
141                    })
142                    .collect()
143            })
144            .map_err(|error| gateway_manager_port_error("resources/list", error))
145    }
146
147    async fn read_mcp_resource(
148        &self,
149        uri: &str,
150        scope: Option<&GatewayRouteScope>,
151        context: &ExecutionContext,
152    ) -> Result<Option<Value>, PortError> {
153        let scope = scope.map(protected_route_scope);
154        self.gateway
155            .read_mcp_resource_for_subject_and_scope(
156                uri,
157                Some(gateway_subject(context)),
158                scope.as_ref(),
159            )
160            .await
161            .map_err(|error| gateway_manager_port_error("resources/read", error))
162    }
163
164    async fn list_mcp_prompts(
165        &self,
166        scope: Option<&GatewayRouteScope>,
167        context: &ExecutionContext,
168    ) -> Result<Vec<GatewayPromptRoute>, PortError> {
169        let scope = scope.map(protected_route_scope);
170        self.gateway
171            .prompt_routes_for_subject_and_scope(Some(gateway_subject(context)), scope.as_ref())
172            .await
173            .map(|routes| {
174                routes
175                    .into_iter()
176                    .map(|route| GatewayPromptRoute {
177                        name: route.name,
178                        description: route.descriptor.description,
179                    })
180                    .collect()
181            })
182            .map_err(|error| gateway_manager_port_error("prompts/list", error))
183    }
184
185    async fn get_mcp_prompt(
186        &self,
187        name: &str,
188        arguments: Option<serde_json::Map<String, Value>>,
189        scope: Option<&GatewayRouteScope>,
190        context: &ExecutionContext,
191    ) -> Result<Option<Value>, PortError> {
192        let scope = scope.map(protected_route_scope);
193        self.gateway
194            .get_mcp_prompt_for_subject_and_scope(
195                name,
196                arguments,
197                Some(gateway_subject(context)),
198                scope.as_ref(),
199            )
200            .await
201            .map_err(|error| gateway_manager_port_error("prompts/get", error))
202    }
203}
204
205fn gateway_subject(context: &ExecutionContext) -> &str {
206    const SHARED_GATEWAY_SUBJECT: &str = "gateway";
207    if !matches!(context.authorization_mode, AuthorizationMode::Mounted) {
208        return SHARED_GATEWAY_SUBJECT;
209    }
210    let Some(principal) = context.principal.as_ref() else {
211        return SHARED_GATEWAY_SUBJECT;
212    };
213    if principal.issuer.as_deref() == Some("local") || has_admin_scope(&principal.scopes.to_vec()) {
214        SHARED_GATEWAY_SUBJECT
215    } else {
216        &principal.subject
217    }
218}
219
220fn protected_route_scope(scope: &GatewayRouteScope) -> ProtectedRouteScope {
221    ProtectedRouteScope {
222        upstreams: scope.upstreams.clone(),
223        services: scope.services.clone(),
224        expose_code_mode: scope.expose_code_mode,
225    }
226}
227
228fn gateway_access(context: &ExecutionContext) -> GatewayAccess {
229    if !matches!(context.authorization_mode, AuthorizationMode::Mounted) {
230        return GatewayAccess {
231            read: true,
232            admin: true,
233        };
234    }
235    let scopes = context
236        .principal
237        .as_ref()
238        .map(|principal| principal.scopes.to_vec())
239        .unwrap_or_default();
240    let admin = has_admin_scope(&scopes);
241    GatewayAccess {
242        read: admin || scopes_satisfy(&scopes, READ_SCOPE),
243        admin,
244    }
245}
246
247fn gateway_port_error(action: &str, error: GatewayDispatchError) -> PortError {
248    let message = error.to_string();
249    port_error_from_structured(error.structured(action), message)
250}
251
252/// `soma-gateway`'s own `GatewayDispatchError::structured()` already gives an
253/// exhaustive, per-variant `code`/`kind`/`remediation` mapping for every
254/// `GatewayManagerError` (including every `UpstreamError` case) via its
255/// `Manager` variant — reuse it here instead of collapsing every MCP
256/// tools/resources/prompts proxy failure into one `gateway_proxy_failed` code
257/// with a blanket `retryable: true`, which previously made permanent
258/// failures (e.g. an unknown/misconfigured upstream) look identical to
259/// transient ones (e.g. a live connect/call failure).
260fn gateway_manager_port_error(operation: &str, error: GatewayManagerError) -> PortError {
261    let message = format!("{operation} failed: {error}");
262    port_error_from_structured(
263        GatewayDispatchError::from(error).structured(operation),
264        message,
265    )
266}
267
268fn port_error_from_structured(
269    structured: soma_gateway::dispatch_helpers::GatewayStructuredError,
270    message: String,
271) -> PortError {
272    PortError {
273        code: structured.code.to_owned(),
274        message,
275        retryable: matches!(structured.kind, "runtime"),
276        remediation: structured.remediation.to_owned(),
277    }
278}
279
280#[cfg(test)]
281#[path = "gateway_tests.rs"]
282mod tests;
soma_integrations/gateway.rs

soma_integrations/
gateway.rs
