1use std::collections::HashMap;
2use std::path::{Path, PathBuf};
3use std::sync::Arc;
4
5use async_trait::async_trait;
6use serde_json::json;
7use tokio::sync::{broadcast, RwLock};
8use tracing::debug;
9use uuid::Uuid;
10
11use orchestral_core::config::load_config;
12use orchestral_core::store::Event;
13
14use crate::bootstrap::RuntimeApp;
15use crate::orchestrator::OrchestratorResult;
16
17use super::dto::{
18 HistoryEventView, InteractionSubmitRequest, InteractionSubmitResponse, SubmitStatus, ThreadView,
19};
20use super::{ApiError, ApiService};
21
22#[async_trait]
23pub trait RuntimeAppBuilder: Send + Sync {
24 async fn build(&self, config_path: PathBuf) -> Result<RuntimeApp, ApiError>;
25}
26
27pub struct DefaultRuntimeAppBuilder;
28
29#[async_trait]
30impl RuntimeAppBuilder for DefaultRuntimeAppBuilder {
31 async fn build(&self, config_path: PathBuf) -> Result<RuntimeApp, ApiError> {
32 if config_has_enabled_runtime_extensions(&config_path)? {
33 tracing::warn!(
34 "config enables runtime extensions but running with minimal runtime; \
35 extensions will be ignored in the slim 1.0 workspace."
36 );
37 }
38 RuntimeApp::from_config_path(config_path)
39 .await
40 .map_err(|err| ApiError::Internal(format!("build runtime app failed: {}", err)))
41 }
42}
43
44fn config_has_enabled_runtime_extensions(config_path: &Path) -> Result<bool, ApiError> {
45 let config = load_config(config_path).map_err(|err| {
46 ApiError::InvalidArgument(format!(
47 "load config '{}' failed: {}",
48 config_path.display(),
49 err
50 ))
51 })?;
52 Ok(config.extensions.runtime.iter().any(|spec| spec.enabled))
53}
54
55#[derive(Clone)]
56pub struct RuntimeApi {
57 config_path: PathBuf,
58 app_builder: Arc<dyn RuntimeAppBuilder>,
59 apps: Arc<RwLock<HashMap<String, Arc<RuntimeApp>>>>,
60}
61
62impl RuntimeApi {
63 pub async fn from_config_path(config: PathBuf) -> Result<Self, ApiError> {
64 Self::from_config_path_with_builder(config, Arc::new(DefaultRuntimeAppBuilder)).await
65 }
66
67 pub async fn from_config_path_with_builder(
68 config: PathBuf,
69 app_builder: Arc<dyn RuntimeAppBuilder>,
70 ) -> Result<Self, ApiError> {
71 Ok(Self {
72 config_path: config,
73 app_builder,
74 apps: Arc::new(RwLock::new(HashMap::new())),
75 })
76 }
77
78 pub async fn runtime_app_for_thread(
79 &self,
80 thread_id: &str,
81 ) -> Result<Arc<RuntimeApp>, ApiError> {
82 self.get_app(thread_id).await
83 }
84
85 async fn create_app_for_thread(&self, thread_id: &str) -> Result<Arc<RuntimeApp>, ApiError> {
86 let app = self.app_builder.build(self.config_path.clone()).await?;
87 {
88 let mut thread = app.orchestrator.thread_runtime.thread.write().await;
89 thread.id = thread_id.into();
90 }
91 Ok(Arc::new(app))
92 }
93
94 async fn get_app(&self, thread_id: &str) -> Result<Arc<RuntimeApp>, ApiError> {
95 if thread_id.trim().is_empty() {
96 return Err(ApiError::InvalidArgument(
97 "thread_id must not be empty".to_string(),
98 ));
99 }
100
101 if let Some(existing) = self.apps.read().await.get(thread_id).cloned() {
102 return Ok(existing);
103 }
104
105 Err(ApiError::NotFound(format!(
106 "thread '{}' not found",
107 thread_id
108 )))
109 }
110
111 async fn get_or_create_app(&self, thread_id: &str) -> Result<Arc<RuntimeApp>, ApiError> {
112 if thread_id.trim().is_empty() {
113 return Err(ApiError::InvalidArgument(
114 "thread_id must not be empty".to_string(),
115 ));
116 }
117
118 if let Some(existing) = self.apps.read().await.get(thread_id).cloned() {
119 return Ok(existing);
120 }
121
122 let created = self.create_app_for_thread(thread_id).await?;
123 let mut apps = self.apps.write().await;
124 if let Some(existing) = apps.get(thread_id).cloned() {
125 return Ok(existing);
126 }
127 apps.insert(thread_id.to_string(), created.clone());
128 Ok(created)
129 }
130}
131
132#[async_trait]
133impl ApiService for RuntimeApi {
134 async fn create_thread(&self, preferred_id: Option<String>) -> Result<ThreadView, ApiError> {
135 let thread_id = preferred_id
136 .filter(|id| !id.trim().is_empty())
137 .unwrap_or_else(|| Uuid::new_v4().to_string());
138
139 let app = self.get_or_create_app(&thread_id).await?;
140 let thread = app.orchestrator.thread_runtime.thread.read().await;
141 Ok(ThreadView {
142 id: thread.id.to_string(),
143 created_at: thread.created_at,
144 updated_at: thread.updated_at,
145 })
146 }
147
148 async fn get_thread(&self, thread_id: &str) -> Result<ThreadView, ApiError> {
149 let app = self.get_app(thread_id).await?;
150 let thread = app.orchestrator.thread_runtime.thread.read().await;
151 Ok(ThreadView {
152 id: thread.id.to_string(),
153 created_at: thread.created_at,
154 updated_at: thread.updated_at,
155 })
156 }
157
158 async fn submit_interaction(
159 &self,
160 thread_id: &str,
161 request: InteractionSubmitRequest,
162 ) -> Result<InteractionSubmitResponse, ApiError> {
163 debug!(
164 thread_id = %thread_id,
165 request_id = ?request.request_id,
166 input_len = request.input.len(),
167 input_preview = %log_preview(&request.input, 80),
168 "submit_chain: runtime_api submit_interaction received"
169 );
170 if request.input.trim().is_empty() {
171 return Err(ApiError::InvalidArgument(
172 "input must not be empty".to_string(),
173 ));
174 }
175
176 let app = self.get_app(thread_id).await?;
177 debug!(
178 thread_id = %thread_id,
179 "submit_chain: runtime_api resolved runtime app"
180 );
181 let interaction_id = request.request_id.unwrap_or_else(|| "api".to_string());
182 let event = Event::user_input(thread_id.to_string(), interaction_id, json!(request.input));
183
184 let result = app.orchestrator.handle_event(event).await.map_err(|err| {
185 ApiError::Internal(format!("orchestrator handle_event failed: {}", err))
186 })?;
187 debug!(
188 result = %orchestrator_result_label(&result),
189 "submit_chain: runtime_api orchestrator.handle_event finished"
190 );
191
192 let response = match result {
193 OrchestratorResult::Started {
194 interaction_id,
195 task_id,
196 ..
197 } => InteractionSubmitResponse {
198 status: SubmitStatus::Started,
199 interaction_id: Some(interaction_id.to_string()),
200 task_id: Some(task_id.to_string()),
201 message: None,
202 },
203 OrchestratorResult::Merged {
204 interaction_id,
205 task_id,
206 ..
207 } => InteractionSubmitResponse {
208 status: SubmitStatus::Merged,
209 interaction_id: Some(interaction_id.to_string()),
210 task_id: Some(task_id.to_string()),
211 message: None,
212 },
213 OrchestratorResult::Rejected { reason } => InteractionSubmitResponse {
214 status: SubmitStatus::Rejected,
215 interaction_id: None,
216 task_id: None,
217 message: Some(reason),
218 },
219 OrchestratorResult::Queued => InteractionSubmitResponse {
220 status: SubmitStatus::Queued,
221 interaction_id: None,
222 task_id: None,
223 message: Some("interaction queued".to_string()),
224 },
225 };
226 debug!(
227 status = ?response.status,
228 interaction_id = ?response.interaction_id,
229 task_id = ?response.task_id,
230 "submit_chain: runtime_api submit_interaction response"
231 );
232
233 Ok(response)
234 }
235
236 async fn query_history(
237 &self,
238 thread_id: &str,
239 limit: usize,
240 ) -> Result<Vec<HistoryEventView>, ApiError> {
241 let app = self.get_app(thread_id).await?;
242 let mut events = app
243 .orchestrator
244 .thread_runtime
245 .query_history(limit)
246 .await
247 .map_err(|err| ApiError::Internal(format!("query history failed: {}", err)))?;
248 events.sort_by_key(|a| a.timestamp());
249 Ok(events.iter().map(event_to_view).collect())
250 }
251
252 async fn subscribe_events(
253 &self,
254 thread_id: &str,
255 ) -> Result<broadcast::Receiver<Event>, ApiError> {
256 let app = self.get_app(thread_id).await?;
257 Ok(app.orchestrator.thread_runtime.subscribe_events())
258 }
259}
260
261fn event_to_view(event: &Event) -> HistoryEventView {
262 match event {
263 Event::UserInput {
264 payload, timestamp, ..
265 } => HistoryEventView {
266 event_type: "user_input".to_string(),
267 timestamp: *timestamp,
268 role: "user".to_string(),
269 content: payload_to_string(payload),
270 },
271 Event::AssistantOutput {
272 payload, timestamp, ..
273 } => HistoryEventView {
274 event_type: "assistant_output".to_string(),
275 timestamp: *timestamp,
276 role: "assistant".to_string(),
277 content: payload_to_string(payload),
278 },
279 Event::Artifact {
280 reference_id,
281 timestamp,
282 ..
283 } => HistoryEventView {
284 event_type: "artifact".to_string(),
285 timestamp: *timestamp,
286 role: "system".to_string(),
287 content: format!("artifact:{}", reference_id),
288 },
289 Event::ExternalEvent {
290 kind,
291 payload,
292 timestamp,
293 ..
294 } => HistoryEventView {
295 event_type: "external_event".to_string(),
296 timestamp: *timestamp,
297 role: "system".to_string(),
298 content: format!("external:{} {}", kind, payload_to_string(payload)),
299 },
300 Event::SystemTrace {
301 level,
302 payload,
303 timestamp,
304 ..
305 } => HistoryEventView {
306 event_type: "system_trace".to_string(),
307 timestamp: *timestamp,
308 role: "system".to_string(),
309 content: format!("trace:{} {}", level, payload_to_string(payload)),
310 },
311 }
312}
313
314fn payload_to_string(payload: &serde_json::Value) -> String {
315 if let Some(s) = payload.as_str() {
316 return s.to_string();
317 }
318 for key in ["content", "message", "text"] {
319 if let Some(s) = payload.get(key).and_then(|v| v.as_str()) {
320 return s.to_string();
321 }
322 }
323 payload.to_string()
324}
325
326fn log_preview(text: &str, max_chars: usize) -> String {
327 text.chars().take(max_chars).collect()
328}
329
330fn orchestrator_result_label(result: &OrchestratorResult) -> &'static str {
331 match result {
332 OrchestratorResult::Started { .. } => "Started",
333 OrchestratorResult::Merged { .. } => "Merged",
334 OrchestratorResult::Rejected { .. } => "Rejected",
335 OrchestratorResult::Queued => "Queued",
336 }
337}
338
339#[cfg(test)]
340mod tests {
341 use super::*;
342 use std::path::Path;
343 use std::sync::atomic::{AtomicUsize, Ordering};
344
345 struct CountingBuilder {
346 calls: Arc<AtomicUsize>,
347 }
348
349 #[async_trait]
350 impl RuntimeAppBuilder for CountingBuilder {
351 async fn build(&self, config_path: PathBuf) -> Result<RuntimeApp, ApiError> {
352 self.calls.fetch_add(1, Ordering::SeqCst);
353 RuntimeApp::from_config_path(config_path)
354 .await
355 .map_err(|err| ApiError::Internal(format!("build runtime app failed: {}", err)))
356 }
357 }
358
359 fn sample_config_path() -> PathBuf {
360 let manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"));
361 manifest_dir
362 .join("../../configs/orchestral.yaml")
363 .canonicalize()
364 .expect("config path")
365 }
366
367 fn sample_cli_config_path() -> PathBuf {
368 let manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"));
369 manifest_dir
370 .join("../../configs/orchestral.cli.yaml")
371 .canonicalize()
372 .expect("cli config path")
373 }
374
375 #[tokio::test]
376 async fn test_runtime_api_uses_injected_builder() {
377 let calls = Arc::new(AtomicUsize::new(0));
378 let builder = Arc::new(CountingBuilder {
379 calls: calls.clone(),
380 });
381 let api = RuntimeApi::from_config_path_with_builder(sample_config_path(), builder)
382 .await
383 .expect("api");
384
385 let thread = api
386 .create_thread(Some("thread-with-builder".to_string()))
387 .await
388 .expect("create thread");
389 assert_eq!(thread.id, "thread-with-builder");
390 assert_eq!(calls.load(Ordering::SeqCst), 1);
391 }
392
393 #[tokio::test]
394 async fn test_default_builder_accepts_extension_config() {
395 let api = RuntimeApi::from_config_path(sample_cli_config_path())
396 .await
397 .expect("should construct api even with extensions in config");
398 assert_eq!(api.config_path, sample_cli_config_path(),);
399 }
400}