use std::sync::Arc;
use kcode_kweb_db::{Node, NodeId, ObjectId, Provenance};
use kcode_kweb_manager::KwebManager;
use kcode_server_object_envelopes::{StoredFile, decode_file, encode_file};
use serde_json::Value;
use uuid::Uuid;
#[derive(Clone)]
pub struct LocalServices {
pub kmap: KwebManager,
pub intelligence: kcode_intelligence_router::Intelligence,
pub history: kcode_session_history::SessionHistory,
pub speech_classifier: Arc<kcode_speech_classification::SpeechClassifier>,
pub dev_tools: kcode_dev_tools::Service,
pub agents: kcode_agent_runtime::AgentRuntime,
pub telegram: kcode_telegram_session_coordinator::Service,
}
#[derive(Debug, Clone)]
pub(crate) struct ApiError {
pub(crate) message: String,
pub(crate) receipt: Option<Box<kcode_intelligence_router::UsageReceipt>>,
}
impl std::fmt::Display for ApiError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for ApiError {}
#[derive(Clone)]
pub struct Api {
services: Arc<LocalServices>,
}
impl Api {
pub fn new(services: LocalServices) -> Self {
Self {
services: Arc::new(services),
}
}
pub(crate) fn telegram(&self) -> &kcode_telegram_session_coordinator::Service {
&self.services.telegram
}
pub(crate) fn create_history_session(
&self,
input: kcode_session_history::NewSession,
) -> anyhow::Result<kcode_session_history::Session> {
self.services.history.create_session(input)
}
pub(crate) fn history_session(
&self,
metadata: kcode_session_history::chatend::SessionMetadata,
provider_model: &str,
) -> anyhow::Result<kcode_session_history::Session> {
self.services
.history
.open_session_with_provider_model(metadata, Some(provider_model))
}
pub(crate) fn kmap_node(&self, node_id: &str) -> Result<Node, ApiError> {
let node_id = node_id.parse::<NodeId>().map_err(local_api_error)?;
self.services.kmap.get_node(node_id).map_err(kmap_error)
}
pub(crate) fn commit_kweb_session(
&self,
input: kcode_commit_session::CommitRequest,
) -> Result<kcode_commit_session::CommitReceipt, ApiError> {
self.services.kmap.commit_session(input).map_err(kmap_error)
}
pub(crate) fn kmap_file(&self, object_id: &str) -> Result<StoredFile, ApiError> {
let object_id = object_id.parse::<ObjectId>().map_err(local_api_error)?;
let bytes = self
.services
.kmap
.get_object(object_id)
.map_err(kmap_error)?;
decode_file(object_id, bytes).map_err(local_api_error)
}
pub(crate) fn save_generated_image(
&self,
bytes: Vec<u8>,
file_name: &str,
media_type: &str,
model: &str,
) -> Result<String, ApiError> {
let bytes = encode_file(
"generated-image",
Some(file_name),
media_type,
Some("image"),
bytes,
)
.map_err(local_api_error)?;
self.services
.kmap
.store_object(
Provenance {
author: model.into(),
source: "kennedy-generated-image".into(),
source_created_at: chrono::Utc::now(),
data: "Image generated or modified through Kennedy intelligence.".into(),
},
bytes,
)
.map(|id| id.to_string())
.map_err(kmap_error)
}
pub(crate) fn agent_runtime(&self) -> kcode_agent_runtime::AgentRuntime {
self.services.agents.clone()
}
pub(crate) async fn search(
&self,
user_id: &str,
request: kcode_intelligence_router::SearchRequest,
) -> Result<
kcode_intelligence_router::Accounted<kcode_intelligence_router::SearchResponse>,
ApiError,
> {
self.services
.intelligence
.for_user(user_id)
.map_err(intelligence_error)?
.search(request)
.await
.map_err(intelligence_error)
}
pub(crate) async fn fetch(
&self,
user_id: &str,
request: kcode_intelligence_router::FetchRequest,
) -> Result<kcode_intelligence_router::FetchResponse, ApiError> {
self.services
.intelligence
.for_user(user_id)
.map_err(intelligence_error)?
.fetch(request)
.await
.map_err(intelligence_error)
}
pub(crate) async fn managed_source_execute(
&self,
session_id: &str,
name: &str,
arguments: Value,
objects: Vec<Vec<u8>>,
) -> Result<kcode_dev_tools::ToolExecution, ApiError> {
let mut execution = self
.services
.dev_tools
.execute(session_id.to_owned(), name.to_owned(), arguments, objects)
.await
.map_err(dev_tools_error)?;
let mut object_ids = Vec::with_capacity(execution.objects.len());
for bytes in std::mem::take(&mut execution.objects) {
object_ids.push(
self.services
.kmap
.store_object(
Provenance {
author: "Kennedy".into(),
source: "kennedy-rust-binary".into(),
source_created_at: chrono::Utc::now(),
data: "Output payload from a managed Rust-binary call.".into(),
},
bytes,
)
.map_err(kmap_error)?
.to_string(),
);
}
append_object_ids(&mut execution.text, &object_ids);
Ok(execution)
}
pub(crate) async fn execute_speech_classification_tool(
&self,
name: &str,
arguments: Value,
) -> Result<String, ApiError> {
let call = kcode_speech_classification::decode_ktool(name, &arguments)
.map_err(speech_ktool_error)?;
let classifier = Arc::clone(&self.services.speech_classifier);
tokio::task::spawn_blocking(move || classifier.execute_ktool(call))
.await
.map_err(speech_task_error)?
.map_err(speech_ktool_error)
}
pub(crate) async fn release_managed_sources(&self, session_id: &str) {
if let Err(error) = self.services.dev_tools.release(session_id.to_owned()).await {
tracing::warn!(error=%error.message, "Managed-source session release failed");
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn transcribe_audio(
&self,
user_id: &str,
model: &str,
prompt: &str,
bytes: Vec<u8>,
filename: String,
mime: &str,
parent_operation_id: Uuid,
) -> Result<
kcode_intelligence_router::Accounted<kcode_intelligence_router::TranscriptionResponse>,
ApiError,
> {
self.services
.intelligence
.for_user(user_id)
.map_err(intelligence_error)?
.transcribe(kcode_intelligence_router::TranscriptionRequest {
prompt: prompt.to_owned(),
model: model.to_owned(),
media: kcode_intelligence_router::Media::audio(bytes, filename, mime)
.map_err(intelligence_error)?,
operation_id: Uuid::new_v4(),
parent_operation_id: Some(parent_operation_id),
})
.await
.map_err(intelligence_error)
}
pub(crate) async fn extract_document(
&self,
bytes: Vec<u8>,
filename: String,
mime: &str,
) -> Result<kcode_intelligence_router::DocumentExtraction, ApiError> {
self.services
.intelligence
.extract_document(kcode_intelligence_router::Document {
bytes,
file_name: filename,
content_type: mime.to_owned(),
})
.await
.map_err(intelligence_error)
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn annotate_media(
&self,
user_id: &str,
model: &str,
prompt: &str,
bytes: Vec<u8>,
filename: String,
mime: &str,
parent_operation_id: Uuid,
) -> Result<
kcode_intelligence_router::Accounted<kcode_intelligence_router::AnnotationResponse>,
ApiError,
> {
self.services
.intelligence
.for_user(user_id)
.map_err(intelligence_error)?
.annotate(kcode_intelligence_router::AnnotationRequest {
prompt: prompt.to_owned(),
model: model.to_owned(),
media: media_for_annotation(bytes, filename, mime).map_err(intelligence_error)?,
operation_id: Uuid::new_v4(),
parent_operation_id: Some(parent_operation_id),
})
.await
.map_err(intelligence_error)
}
pub(crate) async fn generate_image(
&self,
user_id: &str,
model: &str,
prompt: &str,
references: Vec<(Vec<u8>, String, String)>,
parent_operation_id: Uuid,
) -> Result<
kcode_intelligence_router::Accounted<kcode_intelligence_router::ImageResponse>,
ApiError,
> {
let references = references
.into_iter()
.map(|(bytes, filename, mime)| {
media_for_image(bytes, filename, &mime).map_err(intelligence_error)
})
.collect::<Result<Vec<_>, _>>()?;
self.services
.intelligence
.for_user(user_id)
.map_err(intelligence_error)?
.generate_image(kcode_intelligence_router::ImageRequest {
model: model.to_owned(),
prompt: prompt.to_owned(),
references,
operation_id: Uuid::new_v4(),
parent_operation_id: Some(parent_operation_id),
})
.await
.map_err(intelligence_error)
}
}
fn kmap_error(error: kcode_kweb_manager::Error) -> ApiError {
let message = match error.kind() {
kcode_kweb_manager::ErrorKind::InvalidInput
| kcode_kweb_manager::ErrorKind::NotFound
| kcode_kweb_manager::ErrorKind::Conflict => error.to_string(),
_ => "An unexpected Kmap database error occurred.".into(),
};
ApiError {
message,
receipt: None,
}
}
fn intelligence_error(error: kcode_intelligence_router::Error) -> ApiError {
ApiError {
message: error.message().into(),
receipt: error.receipt().cloned().map(Box::new),
}
}
fn append_object_ids(text: &mut String, object_ids: &[String]) {
if object_ids.is_empty() {
return;
}
if !text.is_empty() && !text.ends_with('\n') {
text.push('\n');
}
text.push_str(&object_ids.join("\n"));
}
fn dev_tools_error(error: kcode_dev_tools::ToolError) -> ApiError {
ApiError {
message: error.message,
receipt: None,
}
}
fn speech_task_error(error: tokio::task::JoinError) -> ApiError {
tracing::error!(%error, "In-process speaker-classification task stopped unexpectedly");
ApiError {
message: "An unexpected Kennedy speaker-classification error occurred.".into(),
receipt: None,
}
}
fn speech_ktool_error(error: kcode_speech_classification::KtoolError) -> ApiError {
let kcode_speech_classification::KtoolError::Classifier(error) = error else {
return ApiError {
message: error.to_string(),
receipt: None,
};
};
let internal = matches!(
error,
kcode_speech_classification::Error::UnsupportedSchema { .. }
| kcode_speech_classification::Error::Storage(_)
| kcode_speech_classification::Error::CorruptStorage(_)
);
ApiError {
message: if internal {
tracing::error!(%error, "Speaker-classification storage failed");
"An unexpected Kennedy speaker-classification error occurred.".into()
} else {
error.to_string()
},
receipt: None,
}
}
fn local_api_error(error: impl std::fmt::Display) -> ApiError {
ApiError {
message: error.to_string(),
receipt: None,
}
}
fn media_for_annotation(
bytes: Vec<u8>,
filename: String,
mime: &str,
) -> kcode_intelligence_router::Result<kcode_intelligence_router::Media> {
let normalized = mime
.split(';')
.next()
.unwrap_or("application/octet-stream")
.trim()
.to_ascii_lowercase();
let kind = if normalized.starts_with("image/") {
kcode_intelligence_router::MediaKind::Image
} else if normalized.starts_with("audio/")
|| matches!(normalized.as_str(), "application/ogg" | "video/ogg")
|| filename.rsplit_once('.').is_some_and(|(_, extension)| {
matches!(
extension.to_ascii_lowercase().as_str(),
"ogg" | "oga" | "opus"
)
})
{
kcode_intelligence_router::MediaKind::Audio
} else if normalized.starts_with("video/") {
kcode_intelligence_router::MediaKind::Video
} else {
return Err(kcode_intelligence_router::Error::invalid(
"annotation requires image, audio, or video media",
));
};
kcode_intelligence_router::Media::new(kind, bytes, filename, normalized)
}
fn media_for_image(
bytes: Vec<u8>,
filename: String,
mime: &str,
) -> kcode_intelligence_router::Result<kcode_intelligence_router::Media> {
let normalized = mime
.split(';')
.next()
.unwrap_or("application/octet-stream")
.trim()
.to_ascii_lowercase();
if !normalized.starts_with("image/") {
return Err(kcode_intelligence_router::Error::invalid(
"image references must use an image content type",
));
}
kcode_intelligence_router::Media::new(
kcode_intelligence_router::MediaKind::Image,
bytes,
filename,
normalized,
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn speech_ktool_errors_keep_the_existing_public_failure_boundary() {
let malformed =
speech_ktool_error(kcode_speech_classification::KtoolError::InvalidArguments {
tool: kcode_speech_classification::IDENTIFY_TOOL,
source: serde_json::from_str::<Value>("{").unwrap_err(),
});
assert_eq!(
malformed.message,
"decoding kcode-speech-classification/identify arguments"
);
let validation = speech_ktool_error(kcode_speech_classification::KtoolError::Classifier(
kcode_speech_classification::Error::Validation {
field: "row.perceived_age".into(),
message: "must be positive".into(),
},
));
assert_eq!(validation.message, "row.perceived_age: must be positive");
let storage = speech_ktool_error(kcode_speech_classification::KtoolError::Classifier(
kcode_speech_classification::Error::Storage("private detail".into()),
));
assert_eq!(
storage.message,
"An unexpected Kennedy speaker-classification error occurred."
);
}
}