1use std::sync::Arc;
4
5use kcode_kweb_db::{Node, NodeId, ObjectId, Provenance};
6use kcode_kweb_manager::KwebManager;
7use kcode_server_object_envelopes::{StoredFile, decode_file, encode_file};
8use serde_json::Value;
9use uuid::Uuid;
10
11#[derive(Clone)]
12pub struct LocalServices {
13 pub kmap: KwebManager,
14 pub intelligence: kcode_intelligence_router::Intelligence,
15 pub history: kcode_session_history::SessionHistory,
16 pub speech_classifier: Arc<kcode_speech_classification::SpeechClassifier>,
17 pub dev_tools: kcode_dev_tools::Service,
18 pub agents: kcode_agent_runtime::AgentRuntime,
19 pub telegram: kcode_telegram_session_coordinator::Service,
20}
21
22#[derive(Debug, Clone)]
23pub(crate) struct ApiError {
24 pub(crate) message: String,
25 pub(crate) receipt: Option<Box<kcode_intelligence_router::UsageReceipt>>,
26}
27
28impl std::fmt::Display for ApiError {
29 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
30 formatter.write_str(&self.message)
31 }
32}
33
34impl std::error::Error for ApiError {}
35
36#[derive(Clone)]
37pub struct Api {
38 services: Arc<LocalServices>,
39}
40
41impl Api {
42 pub fn new(services: LocalServices) -> Self {
43 Self {
44 services: Arc::new(services),
45 }
46 }
47
48 pub(crate) fn telegram(&self) -> &kcode_telegram_session_coordinator::Service {
49 &self.services.telegram
50 }
51
52 pub(crate) fn create_history_session(
53 &self,
54 input: kcode_session_history::NewSession,
55 ) -> anyhow::Result<kcode_session_history::Session> {
56 self.services.history.create_session(input)
57 }
58
59 pub(crate) fn history_session(
60 &self,
61 metadata: kcode_session_history::chatend::SessionMetadata,
62 provider_model: &str,
63 ) -> anyhow::Result<kcode_session_history::Session> {
64 self.services
65 .history
66 .open_session_with_provider_model(metadata, Some(provider_model))
67 }
68
69 pub(crate) fn kmap_node(&self, node_id: &str) -> Result<Node, ApiError> {
70 let node_id = node_id.parse::<NodeId>().map_err(local_api_error)?;
71 self.services.kmap.get_node(node_id).map_err(kmap_error)
72 }
73
74 pub(crate) fn commit_kweb_session(
75 &self,
76 input: kcode_commit_session::CommitRequest,
77 ) -> Result<kcode_commit_session::CommitReceipt, ApiError> {
78 self.services.kmap.commit_session(input).map_err(kmap_error)
79 }
80
81 pub(crate) fn kmap_file(&self, object_id: &str) -> Result<StoredFile, ApiError> {
82 let object_id = object_id.parse::<ObjectId>().map_err(local_api_error)?;
83 let bytes = self
84 .services
85 .kmap
86 .get_object(object_id)
87 .map_err(kmap_error)?;
88 decode_file(object_id, bytes).map_err(local_api_error)
89 }
90
91 pub(crate) fn save_generated_image(
92 &self,
93 bytes: Vec<u8>,
94 file_name: &str,
95 media_type: &str,
96 model: &str,
97 ) -> Result<String, ApiError> {
98 let bytes = encode_file(
99 "generated-image",
100 Some(file_name),
101 media_type,
102 Some("image"),
103 bytes,
104 )
105 .map_err(local_api_error)?;
106 self.services
107 .kmap
108 .store_object(
109 Provenance {
110 author: model.into(),
111 source: "kennedy-generated-image".into(),
112 source_created_at: chrono::Utc::now(),
113 data: "Image generated or modified through Kennedy intelligence.".into(),
114 },
115 bytes,
116 )
117 .map(|id| id.to_string())
118 .map_err(kmap_error)
119 }
120
121 pub(crate) fn agent_runtime(&self) -> kcode_agent_runtime::AgentRuntime {
122 self.services.agents.clone()
123 }
124
125 pub(crate) async fn search(
126 &self,
127 user_id: &str,
128 request: kcode_intelligence_router::SearchRequest,
129 ) -> Result<
130 kcode_intelligence_router::Accounted<kcode_intelligence_router::SearchResponse>,
131 ApiError,
132 > {
133 self.services
134 .intelligence
135 .for_user(user_id)
136 .map_err(intelligence_error)?
137 .search(request)
138 .await
139 .map_err(intelligence_error)
140 }
141
142 pub(crate) async fn fetch(
143 &self,
144 user_id: &str,
145 request: kcode_intelligence_router::FetchRequest,
146 ) -> Result<kcode_intelligence_router::FetchResponse, ApiError> {
147 self.services
148 .intelligence
149 .for_user(user_id)
150 .map_err(intelligence_error)?
151 .fetch(request)
152 .await
153 .map_err(intelligence_error)
154 }
155
156 pub(crate) async fn managed_source_execute(
157 &self,
158 session_id: &str,
159 name: &str,
160 arguments: Value,
161 objects: Vec<Vec<u8>>,
162 ) -> Result<kcode_dev_tools::ToolExecution, ApiError> {
163 let mut execution = self
164 .services
165 .dev_tools
166 .execute(session_id.to_owned(), name.to_owned(), arguments, objects)
167 .await
168 .map_err(dev_tools_error)?;
169 let mut object_ids = Vec::with_capacity(execution.objects.len());
170 for bytes in std::mem::take(&mut execution.objects) {
171 object_ids.push(
172 self.services
173 .kmap
174 .store_object(
175 Provenance {
176 author: "Kennedy".into(),
177 source: "kennedy-rust-binary".into(),
178 source_created_at: chrono::Utc::now(),
179 data: "Output payload from a managed Rust-binary call.".into(),
180 },
181 bytes,
182 )
183 .map_err(kmap_error)?
184 .to_string(),
185 );
186 }
187 append_object_ids(&mut execution.text, &object_ids);
188 Ok(execution)
189 }
190
191 pub(crate) async fn execute_speech_classification_tool(
192 &self,
193 name: &str,
194 arguments: Value,
195 ) -> Result<String, ApiError> {
196 let call = kcode_speech_classification::decode_ktool(name, &arguments)
197 .map_err(speech_ktool_error)?;
198 let classifier = Arc::clone(&self.services.speech_classifier);
199 tokio::task::spawn_blocking(move || classifier.execute_ktool(call))
200 .await
201 .map_err(speech_task_error)?
202 .map_err(speech_ktool_error)
203 }
204
205 pub(crate) async fn release_managed_sources(&self, session_id: &str) {
206 if let Err(error) = self.services.dev_tools.release(session_id.to_owned()).await {
207 tracing::warn!(error=%error.message, "Managed-source session release failed");
208 }
209 }
210
211 #[allow(clippy::too_many_arguments)]
212 pub(crate) async fn transcribe_audio(
213 &self,
214 user_id: &str,
215 model: &str,
216 prompt: &str,
217 bytes: Vec<u8>,
218 filename: String,
219 mime: &str,
220 parent_operation_id: Uuid,
221 ) -> Result<
222 kcode_intelligence_router::Accounted<kcode_intelligence_router::TranscriptionResponse>,
223 ApiError,
224 > {
225 self.services
226 .intelligence
227 .for_user(user_id)
228 .map_err(intelligence_error)?
229 .transcribe(kcode_intelligence_router::TranscriptionRequest {
230 prompt: prompt.to_owned(),
231 model: model.to_owned(),
232 media: kcode_intelligence_router::Media::audio(bytes, filename, mime)
233 .map_err(intelligence_error)?,
234 operation_id: Uuid::new_v4(),
235 parent_operation_id: Some(parent_operation_id),
236 })
237 .await
238 .map_err(intelligence_error)
239 }
240
241 pub(crate) async fn extract_document(
242 &self,
243 bytes: Vec<u8>,
244 filename: String,
245 mime: &str,
246 ) -> Result<kcode_intelligence_router::DocumentExtraction, ApiError> {
247 self.services
248 .intelligence
249 .extract_document(kcode_intelligence_router::Document {
250 bytes,
251 file_name: filename,
252 content_type: mime.to_owned(),
253 })
254 .await
255 .map_err(intelligence_error)
256 }
257
258 #[allow(clippy::too_many_arguments)]
259 pub(crate) async fn annotate_media(
260 &self,
261 user_id: &str,
262 model: &str,
263 prompt: &str,
264 bytes: Vec<u8>,
265 filename: String,
266 mime: &str,
267 parent_operation_id: Uuid,
268 ) -> Result<
269 kcode_intelligence_router::Accounted<kcode_intelligence_router::AnnotationResponse>,
270 ApiError,
271 > {
272 self.services
273 .intelligence
274 .for_user(user_id)
275 .map_err(intelligence_error)?
276 .annotate(kcode_intelligence_router::AnnotationRequest {
277 prompt: prompt.to_owned(),
278 model: model.to_owned(),
279 media: media_for_annotation(bytes, filename, mime).map_err(intelligence_error)?,
280 operation_id: Uuid::new_v4(),
281 parent_operation_id: Some(parent_operation_id),
282 })
283 .await
284 .map_err(intelligence_error)
285 }
286
287 pub(crate) async fn generate_image(
288 &self,
289 user_id: &str,
290 model: &str,
291 prompt: &str,
292 references: Vec<(Vec<u8>, String, String)>,
293 parent_operation_id: Uuid,
294 ) -> Result<
295 kcode_intelligence_router::Accounted<kcode_intelligence_router::ImageResponse>,
296 ApiError,
297 > {
298 let references = references
299 .into_iter()
300 .map(|(bytes, filename, mime)| {
301 media_for_image(bytes, filename, &mime).map_err(intelligence_error)
302 })
303 .collect::<Result<Vec<_>, _>>()?;
304 self.services
305 .intelligence
306 .for_user(user_id)
307 .map_err(intelligence_error)?
308 .generate_image(kcode_intelligence_router::ImageRequest {
309 model: model.to_owned(),
310 prompt: prompt.to_owned(),
311 references,
312 operation_id: Uuid::new_v4(),
313 parent_operation_id: Some(parent_operation_id),
314 })
315 .await
316 .map_err(intelligence_error)
317 }
318}
319
320fn kmap_error(error: kcode_kweb_manager::Error) -> ApiError {
321 let message = match error.kind() {
322 kcode_kweb_manager::ErrorKind::InvalidInput
323 | kcode_kweb_manager::ErrorKind::NotFound
324 | kcode_kweb_manager::ErrorKind::Conflict => error.to_string(),
325 _ => "An unexpected Kmap database error occurred.".into(),
326 };
327 ApiError {
328 message,
329 receipt: None,
330 }
331}
332
333fn intelligence_error(error: kcode_intelligence_router::Error) -> ApiError {
334 ApiError {
335 message: error.message().into(),
336 receipt: error.receipt().cloned().map(Box::new),
337 }
338}
339
340fn append_object_ids(text: &mut String, object_ids: &[String]) {
341 if object_ids.is_empty() {
342 return;
343 }
344 if !text.is_empty() && !text.ends_with('\n') {
345 text.push('\n');
346 }
347 text.push_str(&object_ids.join("\n"));
348}
349
350fn dev_tools_error(error: kcode_dev_tools::ToolError) -> ApiError {
351 ApiError {
352 message: error.message,
353 receipt: None,
354 }
355}
356
357fn speech_task_error(error: tokio::task::JoinError) -> ApiError {
358 tracing::error!(%error, "In-process speaker-classification task stopped unexpectedly");
359 ApiError {
360 message: "An unexpected Kennedy speaker-classification error occurred.".into(),
361 receipt: None,
362 }
363}
364
365fn speech_ktool_error(error: kcode_speech_classification::KtoolError) -> ApiError {
366 let kcode_speech_classification::KtoolError::Classifier(error) = error else {
367 return ApiError {
368 message: error.to_string(),
369 receipt: None,
370 };
371 };
372 let internal = matches!(
373 error,
374 kcode_speech_classification::Error::UnsupportedSchema { .. }
375 | kcode_speech_classification::Error::Storage(_)
376 | kcode_speech_classification::Error::CorruptStorage(_)
377 );
378 ApiError {
379 message: if internal {
380 tracing::error!(%error, "Speaker-classification storage failed");
381 "An unexpected Kennedy speaker-classification error occurred.".into()
382 } else {
383 error.to_string()
384 },
385 receipt: None,
386 }
387}
388
389fn local_api_error(error: impl std::fmt::Display) -> ApiError {
390 ApiError {
391 message: error.to_string(),
392 receipt: None,
393 }
394}
395
396fn media_for_annotation(
397 bytes: Vec<u8>,
398 filename: String,
399 mime: &str,
400) -> kcode_intelligence_router::Result<kcode_intelligence_router::Media> {
401 let normalized = mime
402 .split(';')
403 .next()
404 .unwrap_or("application/octet-stream")
405 .trim()
406 .to_ascii_lowercase();
407 let kind = if normalized.starts_with("image/") {
408 kcode_intelligence_router::MediaKind::Image
409 } else if normalized.starts_with("audio/")
410 || matches!(normalized.as_str(), "application/ogg" | "video/ogg")
411 || filename.rsplit_once('.').is_some_and(|(_, extension)| {
412 matches!(
413 extension.to_ascii_lowercase().as_str(),
414 "ogg" | "oga" | "opus"
415 )
416 })
417 {
418 kcode_intelligence_router::MediaKind::Audio
419 } else if normalized.starts_with("video/") {
420 kcode_intelligence_router::MediaKind::Video
421 } else {
422 return Err(kcode_intelligence_router::Error::invalid(
423 "annotation requires image, audio, or video media",
424 ));
425 };
426 kcode_intelligence_router::Media::new(kind, bytes, filename, normalized)
427}
428
429fn media_for_image(
430 bytes: Vec<u8>,
431 filename: String,
432 mime: &str,
433) -> kcode_intelligence_router::Result<kcode_intelligence_router::Media> {
434 let normalized = mime
435 .split(';')
436 .next()
437 .unwrap_or("application/octet-stream")
438 .trim()
439 .to_ascii_lowercase();
440 if !normalized.starts_with("image/") {
441 return Err(kcode_intelligence_router::Error::invalid(
442 "image references must use an image content type",
443 ));
444 }
445 kcode_intelligence_router::Media::new(
446 kcode_intelligence_router::MediaKind::Image,
447 bytes,
448 filename,
449 normalized,
450 )
451}
452
453#[cfg(test)]
454mod tests {
455 use super::*;
456
457 #[test]
458 fn speech_ktool_errors_keep_the_existing_public_failure_boundary() {
459 let malformed =
460 speech_ktool_error(kcode_speech_classification::KtoolError::InvalidArguments {
461 tool: kcode_speech_classification::IDENTIFY_TOOL,
462 source: serde_json::from_str::<Value>("{").unwrap_err(),
463 });
464 assert_eq!(
465 malformed.message,
466 "decoding kcode-speech-classification/identify arguments"
467 );
468
469 let validation = speech_ktool_error(kcode_speech_classification::KtoolError::Classifier(
470 kcode_speech_classification::Error::Validation {
471 field: "row.perceived_age".into(),
472 message: "must be positive".into(),
473 },
474 ));
475 assert_eq!(validation.message, "row.perceived_age: must be positive");
476
477 let storage = speech_ktool_error(kcode_speech_classification::KtoolError::Classifier(
478 kcode_speech_classification::Error::Storage("private detail".into()),
479 ));
480 assert_eq!(
481 storage.message,
482 "An unexpected Kennedy speaker-classification error occurred."
483 );
484 }
485}