traverse_runtime/router/
mod.rs

1//! Governed by spec 016-runtime-placement-router
2//!
3//! `PlacementRouter` is the single public entry point for all capability execution
4//! in `traverse-runtime`.  It wires together:
5//!
6//! 1. Placement evaluation ([`PlacementConstraintEvaluator`])
7//! 2. Executor selection ([`CapabilityExecutorRegistry`])
8//! 3. Execution ([`CapabilityExecutor`])
9//! 4. Trace recording ([`TraceStore`])
10//! 5. Conditional event publishing ([`EventBroker`])
11
12use std::{
13    collections::HashMap,
14    sync::{Arc, Mutex},
15    time::Instant,
16};
17
18use chrono::Utc;
19use serde_json::Value;
20use traverse_contracts::{CapabilityContract, ServiceType, ViolationRecord};
21
22use crate::{
23    events::types::{EventBroker, TraverseEvent},
24    executor::{ArtifactType, CapabilityExecutor, ExecutorCapability},
25    placement::{
26        PlacementConstraintEvaluator, PlacementDecision, PlacementError, PlacementRequest,
27        RuntimeSnapshot,
28    },
29    trace::{PrivateTraceEntry, PublicTraceEntry, TraceOutcome, TraceStore, new_trace_id_and_time},
30};
31
32use traverse_contracts::ExecutionTarget;
33
34// ---------------------------------------------------------------------------
35// Public types
36// ---------------------------------------------------------------------------
37
38/// Maps [`ArtifactType`] to the appropriate [`CapabilityExecutor`] implementation.
39pub type CapabilityExecutorRegistry = HashMap<ArtifactType, Box<dyn CapabilityExecutor>>;
40
41/// Input to [`PlacementRouter::execute`].
42pub struct RouterRequest {
43    /// Unique capability identifier.
44    pub capability_id: String,
45    /// How the capability is packaged.
46    pub artifact_type: ArtifactType,
47    /// The validated contract for this capability (used for placement evaluation).
48    pub contract: CapabilityContract,
49    /// Optional caller hint for target placement.
50    pub target_hint: Option<ExecutionTarget>,
51    /// Current runtime load snapshot used by the placement evaluator.
52    pub runtime_snapshot: RuntimeSnapshot,
53    /// JSON input payload for the capability.
54    pub input: Value,
55    /// Resolved capability descriptor passed to the executor.
56    pub executor_capability: ExecutorCapability,
57    /// When set, used as the public/private [`TraceStore`] id instead of minting a new UUID.
58    pub trace_id_override: Option<String>,
59}
60
61/// Errors returned by [`PlacementRouter::execute`].
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub enum RouterError {
64    /// The placement constraint evaluator rejected the request.
65    PlacementFailed(PlacementError),
66    /// No executor is registered for the requested [`ArtifactType`].
67    ExecutorNotFound(String),
68    /// The selected executor returned an error.
69    ExecutionFailed(String),
70    /// Execution violated a governed contract (aggregate violations).
71    ContractViolation(Vec<ViolationRecord>),
72    /// The trace store lock was poisoned.
73    TraceLockPoisoned,
74}
75
76impl std::fmt::Display for RouterError {
77    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        match self {
79            Self::PlacementFailed(e) => write!(f, "placement failed: {e:?}"),
80            Self::ExecutorNotFound(t) => write!(f, "no executor registered for artifact type: {t}"),
81            Self::ExecutionFailed(msg) => write!(f, "execution failed: {msg}"),
82            Self::ContractViolation(violations) => {
83                write!(f, "contract violation: {} violation(s)", violations.len())
84            }
85            Self::TraceLockPoisoned => write!(f, "trace store lock is poisoned"),
86        }
87    }
88}
89
90impl std::error::Error for RouterError {}
91
92/// Result of a successful [`PlacementRouter::execute`] call.
93#[derive(Debug)]
94pub struct RouterResponse {
95    /// The JSON output produced by the executor.
96    pub output: Value,
97    /// Events the executor emitted and validated during this call (spec
98    /// 098-capability-event-host-abi), already published to `EventBroker`
99    /// by Step 5 for `Subscribable` capabilities.
100    pub emitted_events: Vec<TraverseEvent>,
101    /// The public trace entry written to the store.
102    pub trace_id: String,
103    /// The placement decision that was made.
104    pub placement_decision: PlacementDecision,
105}
106
107// ---------------------------------------------------------------------------
108// PlacementRouter
109// ---------------------------------------------------------------------------
110
111/// Single orchestrating entry point for all capability execution in Traverse.
112///
113/// Wires together placement evaluation → executor selection → execution →
114/// trace recording → event publishing.
115pub struct PlacementRouter {
116    evaluator: PlacementConstraintEvaluator,
117    executor_registry: CapabilityExecutorRegistry,
118    trace_store: Arc<Mutex<TraceStore>>,
119    event_broker: Arc<dyn EventBroker>,
120}
121
122impl PlacementRouter {
123    /// Construct a new [`PlacementRouter`] from injected dependencies.
124    #[must_use]
125    pub fn new(
126        evaluator: PlacementConstraintEvaluator,
127        executor_registry: CapabilityExecutorRegistry,
128        trace_store: Arc<Mutex<TraceStore>>,
129        event_broker: Arc<dyn EventBroker>,
130    ) -> Self {
131        Self {
132            evaluator,
133            executor_registry,
134            trace_store,
135            event_broker,
136        }
137    }
138
139    /// Execute a capability end-to-end.
140    ///
141    /// Steps:
142    /// 1. Evaluate placement constraints — returns [`RouterError::PlacementFailed`] with no trace on failure.
143    /// 2. Select executor by `artifact_type`.
144    /// 3. Run the executor.
145    /// 4. Write public + private trace entries to the store.
146    /// 5. If `service_type == Subscribable`, publish emitted events.
147    ///
148    /// # Errors
149    ///
150    /// Returns [`RouterError`] when any step cannot complete.
151    pub fn execute(&self, request: RouterRequest) -> Result<RouterResponse, RouterError> {
152        let executor = self
153            .executor_registry
154            .get(&request.artifact_type)
155            .ok_or_else(|| RouterError::ExecutorNotFound(format!("{:?}", request.artifact_type)))?;
156        self.execute_with_executor(request, executor.as_ref())
157    }
158
159    /// Execute a capability with an explicitly provided executor.
160    ///
161    /// Used by the live `Runtime::execute` path to bridge a host
162    /// `LocalExecutor` without requiring a `'static` registry entry.
163    ///
164    /// # Errors
165    ///
166    /// Returns [`RouterError`] when any step cannot complete.
167    pub fn execute_with_executor(
168        &self,
169        request: RouterRequest,
170        executor: &dyn CapabilityExecutor,
171    ) -> Result<RouterResponse, RouterError> {
172        // --- Step 1: Placement evaluation ---
173        let placement_req = PlacementRequest {
174            capability_id: request.capability_id.clone(),
175            target_hint: request.target_hint,
176            runtime_snapshot: request.runtime_snapshot,
177        };
178
179        let decision = self
180            .evaluator
181            .evaluate(&placement_req, &request.contract)
182            .map_err(RouterError::PlacementFailed)?;
183
184        let placement_target_str = format!("{:?}", decision.target);
185
186        // --- Step 3: Execute capability ---
187        // Events emitted via `traverse_host::emit_event` (spec
188        // 098-capability-event-host-abi) are already validated
189        // synchronously, at call time, against `request.contract.emits` and
190        // `service_type` by the host function itself (FR-002/FR-003) — no
191        // post-hoc enforcement gate is needed here.
192        let start = Instant::now();
193        let exec_result = executor.execute(&request.executor_capability, &request.input);
194        let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
195
196        let (output, emitted_events, outcome) = match exec_result {
197            Ok(exec_output) => (
198                exec_output.value,
199                exec_output.emitted_events,
200                TraceOutcome::Success,
201            ),
202            Err(e) => return Err(RouterError::ExecutionFailed(format!("{e}"))),
203        };
204
205        // --- Step 4: Write trace ---
206        let (trace_id, time) = match request.trace_id_override {
207            Some(override_id) => (override_id, Utc::now().to_rfc3339()),
208            None => new_trace_id_and_time(),
209        };
210
211        let public_entry = PublicTraceEntry::new(
212            trace_id.clone(),
213            request.capability_id.clone(),
214            placement_target_str,
215            outcome,
216            duration_ms,
217            time,
218        );
219
220        let input_str = serde_json::to_string(&request.input).unwrap_or_default();
221        let output_str = serde_json::to_string(&output).unwrap_or_default();
222        let private_entry =
223            PrivateTraceEntry::new(trace_id.clone(), &input_str, &output_str, duration_ms);
224
225        {
226            let mut store = self
227                .trace_store
228                .lock()
229                .map_err(|_| RouterError::TraceLockPoisoned)?;
230            store.insert(public_entry, Some(private_entry));
231        }
232
233        // --- Step 5: Publish events for Subscribable capabilities ---
234        let published_events = if request.contract.service_type == ServiceType::Subscribable {
235            for event in &emitted_events {
236                // Best-effort: publish errors are logged but do not fail the response.
237                let _ = self.event_broker.publish(event.clone());
238            }
239            emitted_events
240        } else {
241            Vec::new()
242        };
243
244        Ok(RouterResponse {
245            output,
246            emitted_events: published_events,
247            trace_id,
248            placement_decision: decision,
249        })
250    }
251}
traverse_runtime/router/mod.rs

traverse_runtime/router/
mod.rs
