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 temperature: Option<f32>,
221 parent_operation_id: Uuid,
222 ) -> Result<
223 kcode_intelligence_router::Accounted<kcode_intelligence_router::TranscriptionResponse>,
224 ApiError,
225 > {
226 self.services
227 .intelligence
228 .for_user(user_id)
229 .map_err(intelligence_error)?
230 .transcribe(kcode_intelligence_router::TranscriptionRequest {
231 prompt: prompt.to_owned(),
232 model: model.to_owned(),
233 media: kcode_intelligence_router::Media::audio(bytes, filename, mime)
234 .map_err(intelligence_error)?,
235 temperature,
236 operation_id: Uuid::new_v4(),
237 parent_operation_id: Some(parent_operation_id),
238 })
239 .await
240 .map_err(intelligence_error)
241 }
242
243 pub(crate) async fn extract_document(
244 &self,
245 bytes: Vec<u8>,
246 filename: String,
247 mime: &str,
248 ) -> Result<kcode_intelligence_router::DocumentExtraction, ApiError> {
249 self.services
250 .intelligence
251 .extract_document(kcode_intelligence_router::Document {
252 bytes,
253 file_name: filename,
254 content_type: mime.to_owned(),
255 })
256 .await
257 .map_err(intelligence_error)
258 }
259
260 #[allow(clippy::too_many_arguments)]
261 pub(crate) async fn annotate_media(
262 &self,
263 user_id: &str,
264 model: &str,
265 prompt: &str,
266 bytes: Vec<u8>,
267 filename: String,
268 mime: &str,
269 parent_operation_id: Uuid,
270 ) -> Result<
271 kcode_intelligence_router::Accounted<kcode_intelligence_router::AnnotationResponse>,
272 ApiError,
273 > {
274 self.services
275 .intelligence
276 .for_user(user_id)
277 .map_err(intelligence_error)?
278 .annotate(kcode_intelligence_router::AnnotationRequest {
279 prompt: prompt.to_owned(),
280 model: model.to_owned(),
281 media: media_for_annotation(bytes, filename, mime).map_err(intelligence_error)?,
282 operation_id: Uuid::new_v4(),
283 parent_operation_id: Some(parent_operation_id),
284 })
285 .await
286 .map_err(intelligence_error)
287 }
288
289 pub(crate) async fn generate_image(
290 &self,
291 user_id: &str,
292 model: &str,
293 prompt: &str,
294 references: Vec<(Vec<u8>, String, String)>,
295 parent_operation_id: Uuid,
296 ) -> Result<
297 kcode_intelligence_router::Accounted<kcode_intelligence_router::ImageResponse>,
298 ApiError,
299 > {
300 let references = references
301 .into_iter()
302 .map(|(bytes, filename, mime)| {
303 media_for_image(bytes, filename, &mime).map_err(intelligence_error)
304 })
305 .collect::<Result<Vec<_>, _>>()?;
306 self.services
307 .intelligence
308 .for_user(user_id)
309 .map_err(intelligence_error)?
310 .generate_image(kcode_intelligence_router::ImageRequest {
311 model: model.to_owned(),
312 prompt: prompt.to_owned(),
313 references,
314 operation_id: Uuid::new_v4(),
315 parent_operation_id: Some(parent_operation_id),
316 })
317 .await
318 .map_err(intelligence_error)
319 }
320}
321
322fn kmap_error(error: kcode_kweb_manager::Error) -> ApiError {
323 let message = match error.kind() {
324 kcode_kweb_manager::ErrorKind::InvalidInput
325 | kcode_kweb_manager::ErrorKind::NotFound
326 | kcode_kweb_manager::ErrorKind::Conflict => error.to_string(),
327 _ => "An unexpected Kmap database error occurred.".into(),
328 };
329 ApiError {
330 message,
331 receipt: None,
332 }
333}
334
335fn intelligence_error(error: kcode_intelligence_router::Error) -> ApiError {
336 ApiError {
337 message: error.message().into(),
338 receipt: error.receipt().cloned().map(Box::new),
339 }
340}
341
342fn append_object_ids(text: &mut String, object_ids: &[String]) {
343 if object_ids.is_empty() {
344 return;
345 }
346 if !text.is_empty() && !text.ends_with('\n') {
347 text.push('\n');
348 }
349 text.push_str(&object_ids.join("\n"));
350}
351
352fn dev_tools_error(error: kcode_dev_tools::ToolError) -> ApiError {
353 ApiError {
354 message: error.message,
355 receipt: None,
356 }
357}
358
359fn speech_task_error(error: tokio::task::JoinError) -> ApiError {
360 tracing::error!(%error, "In-process speaker-classification task stopped unexpectedly");
361 ApiError {
362 message: "An unexpected Kennedy speaker-classification error occurred.".into(),
363 receipt: None,
364 }
365}
366
367fn speech_ktool_error(error: kcode_speech_classification::KtoolError) -> ApiError {
368 let kcode_speech_classification::KtoolError::Classifier(error) = error else {
369 return ApiError {
370 message: error.to_string(),
371 receipt: None,
372 };
373 };
374 let internal = matches!(
375 error,
376 kcode_speech_classification::Error::UnsupportedSchema { .. }
377 | kcode_speech_classification::Error::Storage(_)
378 | kcode_speech_classification::Error::CorruptStorage(_)
379 );
380 ApiError {
381 message: if internal {
382 tracing::error!(%error, "Speaker-classification storage failed");
383 "An unexpected Kennedy speaker-classification error occurred.".into()
384 } else {
385 error.to_string()
386 },
387 receipt: None,
388 }
389}
390
391fn local_api_error(error: impl std::fmt::Display) -> ApiError {
392 ApiError {
393 message: error.to_string(),
394 receipt: None,
395 }
396}
397
398fn media_for_annotation(
399 bytes: Vec<u8>,
400 filename: String,
401 mime: &str,
402) -> kcode_intelligence_router::Result<kcode_intelligence_router::Media> {
403 let normalized = mime
404 .split(';')
405 .next()
406 .unwrap_or("application/octet-stream")
407 .trim()
408 .to_ascii_lowercase();
409 let kind = if normalized.starts_with("image/") {
410 kcode_intelligence_router::MediaKind::Image
411 } else if normalized.starts_with("audio/")
412 || matches!(normalized.as_str(), "application/ogg" | "video/ogg")
413 || filename.rsplit_once('.').is_some_and(|(_, extension)| {
414 matches!(
415 extension.to_ascii_lowercase().as_str(),
416 "ogg" | "oga" | "opus"
417 )
418 })
419 {
420 kcode_intelligence_router::MediaKind::Audio
421 } else if normalized.starts_with("video/") {
422 kcode_intelligence_router::MediaKind::Video
423 } else {
424 return Err(kcode_intelligence_router::Error::invalid(
425 "annotation requires image, audio, or video media",
426 ));
427 };
428 kcode_intelligence_router::Media::new(kind, bytes, filename, normalized)
429}
430
431fn media_for_image(
432 bytes: Vec<u8>,
433 filename: String,
434 mime: &str,
435) -> kcode_intelligence_router::Result<kcode_intelligence_router::Media> {
436 let normalized = mime
437 .split(';')
438 .next()
439 .unwrap_or("application/octet-stream")
440 .trim()
441 .to_ascii_lowercase();
442 if !normalized.starts_with("image/") {
443 return Err(kcode_intelligence_router::Error::invalid(
444 "image references must use an image content type",
445 ));
446 }
447 kcode_intelligence_router::Media::new(
448 kcode_intelligence_router::MediaKind::Image,
449 bytes,
450 filename,
451 normalized,
452 )
453}
454
455#[cfg(test)]
456mod tests {
457 use super::*;
458
459 #[test]
460 fn speech_ktool_errors_keep_the_existing_public_failure_boundary() {
461 let malformed =
462 speech_ktool_error(kcode_speech_classification::KtoolError::InvalidArguments {
463 tool: kcode_speech_classification::IDENTIFY_TOOL,
464 source: serde_json::from_str::<Value>("{").unwrap_err(),
465 });
466 assert_eq!(
467 malformed.message,
468 "decoding kcode-speech-classification/identify arguments"
469 );
470
471 let validation = speech_ktool_error(kcode_speech_classification::KtoolError::Classifier(
472 kcode_speech_classification::Error::Validation {
473 field: "row.perceived_age".into(),
474 message: "must be positive".into(),
475 },
476 ));
477 assert_eq!(validation.message, "row.perceived_age: must be positive");
478
479 let storage = speech_ktool_error(kcode_speech_classification::KtoolError::Classifier(
480 kcode_speech_classification::Error::Storage("private detail".into()),
481 ));
482 assert_eq!(
483 storage.message,
484 "An unexpected Kennedy speaker-classification error occurred."
485 );
486 }
487}