soma_application/
ports.rs

1use std::{collections::BTreeMap, path::Path, sync::Arc};
2
3use async_trait::async_trait;
4use serde_json::Value;
5
6use crate::{
7    CodeModeExecuteRequest, ExecutionContext, GatewayExecuteRequest, GatewayMcpOutcome,
8    GatewayMcpRoundTrip, GatewayPromptRoute, GatewayReloadRequest, GatewayResourceRoute,
9    GatewayRouteScope, GatewayToolRoute, OpenApiExecuteRequest,
10};
11use soma_provider_adapters::python::materializer::PythonEnvironmentUpdateReport;
12
13/// Error returned by an engine port when an operation cannot be completed.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct PortError {
16    /// Stable, machine-readable error code.
17    pub code: String,
18    /// Human-readable description of the failure.
19    pub message: String,
20    /// Whether retrying the operation might succeed.
21    pub retryable: bool,
22    /// Suggested remediation the caller can act on.
23    pub remediation: String,
24}
25
26impl PortError {
27    /// Builds a `PortError` from a code and message, defaulting to
28    /// non-retryable with a generic remediation hint.
29    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
30        Self {
31            code: code.into(),
32            message: message.into(),
33            retryable: false,
34            remediation: "Check the engine configuration and retry.".to_owned(),
35        }
36    }
37}
38
39/// Port to the MCP gateway engine: status, reload, execution, and
40/// tool/resource/prompt routing.
41#[async_trait]
42pub trait GatewayPort: Send + Sync {
43    /// Returns the gateway's current status snapshot.
44    async fn status(&self, context: &ExecutionContext) -> Result<Value, PortError>;
45    /// Reloads the gateway configuration.
46    async fn reload(
47        &self,
48        request: GatewayReloadRequest,
49        context: &ExecutionContext,
50    ) -> Result<Value, PortError>;
51    /// Executes a gateway operation.
52    async fn execute(
53        &self,
54        request: GatewayExecuteRequest,
55        context: &ExecutionContext,
56    ) -> Result<Value, PortError>;
57
58    /// Lists MCP tool routes exposed through the gateway, optionally scoped.
59    async fn list_mcp_tools(
60        &self,
61        scope: Option<&GatewayRouteScope>,
62        context: &ExecutionContext,
63    ) -> Result<Vec<GatewayToolRoute>, PortError>;
64
65    /// Calls an MCP tool by name through the gateway.
66    async fn call_mcp_tool(
67        &self,
68        name: &str,
69        params: Value,
70        scope: Option<&GatewayRouteScope>,
71        context: &ExecutionContext,
72    ) -> Result<Option<Value>, PortError>;
73
74    /// Calls an MCP tool once, preserving modern multi-round-trip and task outcomes.
75    async fn call_mcp_tool_once(
76        &self,
77        name: &str,
78        params: Value,
79        round_trip: GatewayMcpRoundTrip,
80        scope: Option<&GatewayRouteScope>,
81        context: &ExecutionContext,
82    ) -> Result<Option<GatewayMcpOutcome>, PortError> {
83        let _ = round_trip;
84        self.call_mcp_tool(name, params, scope, context)
85            .await
86            .map(|result| {
87                result.map(|value| {
88                    GatewayMcpOutcome::Complete(serde_json::json!({
89                        "resultType": "complete",
90                        "content": [],
91                        "structuredContent": value
92                    }))
93                })
94            })
95    }
96
97    /// Gets the latest state of a routed MCP task.
98    async fn get_mcp_task(
99        &self,
100        task_id: &str,
101        context: &ExecutionContext,
102    ) -> Result<Value, PortError> {
103        let _ = (task_id, context);
104        Err(PortError::new(
105            "tasks_unsupported",
106            "the configured gateway does not support MCP tasks",
107        ))
108    }
109
110    /// Supplies follow-up input to a routed MCP task.
111    async fn update_mcp_task(
112        &self,
113        task_id: &str,
114        input_responses: BTreeMap<String, Value>,
115        context: &ExecutionContext,
116    ) -> Result<(), PortError> {
117        let _ = (task_id, input_responses, context);
118        Err(PortError::new(
119            "tasks_unsupported",
120            "the configured gateway does not support MCP tasks",
121        ))
122    }
123
124    /// Requests cancellation of a routed MCP task.
125    async fn cancel_mcp_task(
126        &self,
127        task_id: &str,
128        context: &ExecutionContext,
129    ) -> Result<(), PortError> {
130        let _ = (task_id, context);
131        Err(PortError::new(
132            "tasks_unsupported",
133            "the configured gateway does not support MCP tasks",
134        ))
135    }
136
137    /// Lists MCP resource routes exposed through the gateway, optionally scoped.
138    async fn list_mcp_resources(
139        &self,
140        scope: Option<&GatewayRouteScope>,
141        context: &ExecutionContext,
142    ) -> Result<Vec<GatewayResourceRoute>, PortError>;
143
144    /// Reads an MCP resource by URI through the gateway.
145    async fn read_mcp_resource(
146        &self,
147        uri: &str,
148        scope: Option<&GatewayRouteScope>,
149        context: &ExecutionContext,
150    ) -> Result<Option<Value>, PortError>;
151
152    /// Reads an MCP resource once, preserving a possible input-required outcome.
153    async fn read_mcp_resource_once(
154        &self,
155        uri: &str,
156        round_trip: GatewayMcpRoundTrip,
157        scope: Option<&GatewayRouteScope>,
158        context: &ExecutionContext,
159    ) -> Result<Option<GatewayMcpOutcome>, PortError> {
160        let _ = round_trip;
161        self.read_mcp_resource(uri, scope, context)
162            .await
163            .map(|result| result.map(GatewayMcpOutcome::Complete))
164    }
165
166    /// Lists MCP prompt routes exposed through the gateway, optionally scoped.
167    async fn list_mcp_prompts(
168        &self,
169        scope: Option<&GatewayRouteScope>,
170        context: &ExecutionContext,
171    ) -> Result<Vec<GatewayPromptRoute>, PortError>;
172
173    /// Gets an MCP prompt by name, with optional arguments, through the gateway.
174    async fn get_mcp_prompt(
175        &self,
176        name: &str,
177        arguments: Option<serde_json::Map<String, Value>>,
178        scope: Option<&GatewayRouteScope>,
179        context: &ExecutionContext,
180    ) -> Result<Option<Value>, PortError>;
181    /// Gets an MCP prompt once, preserving a possible input-required outcome.
182    async fn get_mcp_prompt_once(
183        &self,
184        name: &str,
185        arguments: Option<serde_json::Map<String, Value>>,
186        round_trip: GatewayMcpRoundTrip,
187        scope: Option<&GatewayRouteScope>,
188        context: &ExecutionContext,
189    ) -> Result<Option<GatewayMcpOutcome>, PortError> {
190        let _ = round_trip;
191        self.get_mcp_prompt(name, arguments, scope, context)
192            .await
193            .map(|result| result.map(GatewayMcpOutcome::Complete))
194    }
195}
196
197/// Port to the Code Mode engine that runs sandboxed JavaScript against the
198/// tool catalog.
199#[async_trait]
200pub trait CodeModePort: Send + Sync {
201    /// Executes a Code Mode request.
202    async fn execute(
203        &self,
204        request: CodeModeExecuteRequest,
205        context: &ExecutionContext,
206    ) -> Result<Value, PortError>;
207}
208
209/// Port to the OpenAPI engine that dispatches requests against described APIs.
210#[async_trait]
211pub trait OpenApiPort: Send + Sync {
212    /// Executes an OpenAPI request.
213    async fn execute(
214        &self,
215        request: OpenApiExecuteRequest,
216        context: &ExecutionContext,
217    ) -> Result<Value, PortError>;
218}
219
220/// Result of preparing an immutable environment update before registry activation.
221pub struct PythonEnvironmentUpdateCandidate(PythonEnvironmentUpdateReport);
222
223impl PythonEnvironmentUpdateCandidate {
224    /// Wraps the materializer's typed report so the activated candidate and
225    /// operator-facing report cannot be constructed independently.
226    #[must_use]
227    pub fn from_report(report: PythonEnvironmentUpdateReport) -> Self {
228        Self(report)
229    }
230
231    pub(crate) fn into_report(self) -> PythonEnvironmentUpdateReport {
232        self.0
233    }
234}
235
236/// Operator control plane for production-managed Python environments.
237#[async_trait]
238pub trait PythonEnvironmentPort: Send + Sync {
239    /// Inventories every managed cache entry without executing provider code.
240    async fn status(&self) -> Result<Value, PortError>;
241    /// Plans or applies a bounded prune of stale non-ready entries.
242    async fn prune(
243        &self,
244        stale_before_unix_seconds: u64,
245        max_entries: usize,
246        apply: bool,
247    ) -> Result<Value, PortError>;
248    /// Repairs the exact environment planned for one managed provider.
249    async fn repair(&self, provider_path: &Path) -> Result<Value, PortError>;
250    /// Prepares an immutable candidate for later atomic registry activation.
251    async fn update(
252        &self,
253        provider_path: &Path,
254    ) -> Result<PythonEnvironmentUpdateCandidate, PortError>;
255}
256
257/// Bundle of the engine ports the application depends on.
258pub struct ApplicationPorts {
259    /// MCP gateway engine port.
260    pub gateway: Arc<dyn GatewayPort>,
261    /// Code Mode engine port.
262    pub codemode: Arc<dyn CodeModePort>,
263    /// OpenAPI engine port.
264    pub openapi: Arc<dyn OpenApiPort>,
265    /// Immutable Python environment operator port.
266    pub python_environment: Arc<dyn PythonEnvironmentPort>,
267}
268
269impl ApplicationPorts {
270    /// Builds a port bundle where every engine reports itself as unavailable.
271    pub fn unavailable() -> Self {
272        let port = Arc::new(UnavailableEnginePort);
273        Self {
274            gateway: port.clone(),
275            codemode: port.clone(),
276            openapi: port.clone(),
277            python_environment: port,
278        }
279    }
280
281    /// Replaces the gateway port and returns the updated bundle.
282    pub fn with_gateway(mut self, gateway: Arc<dyn GatewayPort>) -> Self {
283        self.gateway = gateway;
284        self
285    }
286
287    /// Replaces the Code Mode port and returns the updated bundle.
288    pub fn with_codemode(mut self, codemode: Arc<dyn CodeModePort>) -> Self {
289        self.codemode = codemode;
290        self
291    }
292
293    /// Replaces the OpenAPI port and returns the updated bundle.
294    pub fn with_openapi(mut self, openapi: Arc<dyn OpenApiPort>) -> Self {
295        self.openapi = openapi;
296        self
297    }
298
299    /// Replaces the Python environment operator port and returns the updated bundle.
300    pub fn with_python_environment(
301        mut self,
302        python_environment: Arc<dyn PythonEnvironmentPort>,
303    ) -> Self {
304        self.python_environment = python_environment;
305        self
306    }
307}
308
309struct UnavailableEnginePort;
310
311impl UnavailableEnginePort {
312    fn error(engine: &str) -> PortError {
313        PortError::new(
314            "engine_unavailable",
315            format!("{engine} is not configured for this application instance"),
316        )
317    }
318}
319
320#[async_trait]
321impl GatewayPort for UnavailableEnginePort {
322    async fn status(&self, _context: &ExecutionContext) -> Result<Value, PortError> {
323        Err(Self::error("gateway"))
324    }
325
326    async fn reload(
327        &self,
328        _request: GatewayReloadRequest,
329        _context: &ExecutionContext,
330    ) -> Result<Value, PortError> {
331        Err(Self::error("gateway"))
332    }
333
334    async fn execute(
335        &self,
336        _request: GatewayExecuteRequest,
337        _context: &ExecutionContext,
338    ) -> Result<Value, PortError> {
339        Err(Self::error("gateway"))
340    }
341
342    async fn list_mcp_tools(
343        &self,
344        _scope: Option<&GatewayRouteScope>,
345        _context: &ExecutionContext,
346    ) -> Result<Vec<GatewayToolRoute>, PortError> {
347        Err(Self::error("gateway"))
348    }
349
350    async fn call_mcp_tool(
351        &self,
352        _name: &str,
353        _params: Value,
354        _scope: Option<&GatewayRouteScope>,
355        _context: &ExecutionContext,
356    ) -> Result<Option<Value>, PortError> {
357        Err(Self::error("gateway"))
358    }
359
360    async fn list_mcp_resources(
361        &self,
362        _scope: Option<&GatewayRouteScope>,
363        _context: &ExecutionContext,
364    ) -> Result<Vec<GatewayResourceRoute>, PortError> {
365        Err(Self::error("gateway"))
366    }
367
368    async fn read_mcp_resource(
369        &self,
370        _uri: &str,
371        _scope: Option<&GatewayRouteScope>,
372        _context: &ExecutionContext,
373    ) -> Result<Option<Value>, PortError> {
374        Err(Self::error("gateway"))
375    }
376
377    async fn list_mcp_prompts(
378        &self,
379        _scope: Option<&GatewayRouteScope>,
380        _context: &ExecutionContext,
381    ) -> Result<Vec<GatewayPromptRoute>, PortError> {
382        Err(Self::error("gateway"))
383    }
384
385    async fn get_mcp_prompt(
386        &self,
387        _name: &str,
388        _arguments: Option<serde_json::Map<String, Value>>,
389        _scope: Option<&GatewayRouteScope>,
390        _context: &ExecutionContext,
391    ) -> Result<Option<Value>, PortError> {
392        Err(Self::error("gateway"))
393    }
394}
395
396#[async_trait]
397impl CodeModePort for UnavailableEnginePort {
398    async fn execute(
399        &self,
400        _request: CodeModeExecuteRequest,
401        _context: &ExecutionContext,
402    ) -> Result<Value, PortError> {
403        Err(Self::error("Code Mode"))
404    }
405}
406
407#[async_trait]
408impl OpenApiPort for UnavailableEnginePort {
409    async fn execute(
410        &self,
411        _request: OpenApiExecuteRequest,
412        _context: &ExecutionContext,
413    ) -> Result<Value, PortError> {
414        Err(Self::error("OpenAPI"))
415    }
416}
417
418#[async_trait]
419impl PythonEnvironmentPort for UnavailableEnginePort {
420    async fn status(&self) -> Result<Value, PortError> {
421        Err(Self::error("Python environment lifecycle"))
422    }
423
424    async fn prune(
425        &self,
426        _stale_before_unix_seconds: u64,
427        _max_entries: usize,
428        _apply: bool,
429    ) -> Result<Value, PortError> {
430        Err(Self::error("Python environment lifecycle"))
431    }
432
433    async fn repair(&self, _provider_path: &Path) -> Result<Value, PortError> {
434        Err(Self::error("Python environment lifecycle"))
435    }
436
437    async fn update(
438        &self,
439        _provider_path: &Path,
440    ) -> Result<PythonEnvironmentUpdateCandidate, PortError> {
441        Err(Self::error("Python environment lifecycle"))
442    }
443}
soma_application/ports.rs

soma_application/
ports.rs
