#![forbid(unsafe_code)]
use std::{
path::{Path, PathBuf},
sync::Arc,
};
use anyhow::Context;
use kcode_kennedy_cli::{Args, Command};
use kcode_speaker_system::SpeechClassifier;
pub use kcode_kennedy_maintenance_guard::maintenance_guard;
const SPEECH_CLASSIFICATION_DATABASE_PATH: &str = "./data/kennedy-speech-classification.sqlite3";
#[tokio::main]
pub async fn main() -> anyhow::Result<()> {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {
"kennedy_server=info,kcode_kennedy_app=info,kcode_kennedy_orchestration=info,kcode_kennedy_telegram_runtime=info,kcode_kennedy_roots=info,kcode_kweb_db=info,kcode_codex_runtime=info,kcode_session_history=info,kcode_tg_kennedy_bot=info,tower_http=info".into()
}),
)
.init();
rustls::crypto::ring::default_provider()
.install_default()
.map_err(|_| anyhow::anyhow!("installing TLS crypto provider"))?;
let mut args = kcode_kennedy_cli::parse();
let vault_path = args.vault_path.clone();
match args.command.take() {
Some(Command::Secrets { command }) => {
let _maintenance_guard = tokio::net::TcpListener::bind(&args.kweb_bind)
.await
.with_context(|| {
format!(
"binding maintenance lock {}; stop the running Kennedy server before changing its credential vault",
args.kweb_bind
)
})?;
kcode_kennedy_bootstrap_secrets::manage(command, &vault_path)
}
Some(Command::KmapSize) => {
let _maintenance_guard =
maintenance_guard(&args.kweb_bind, "measuring the Kweb").await?;
let kweb_config = kcode_kennedy_bootstrap_secrets::unlock_kweb(&vault_path)?;
let size = kcode_kmap_size::measure(&args.kweb_root, kweb_config)?;
println!("{}", kcode_kmap_size::render(&size));
Ok(())
}
None => run_server(args, vault_path).await,
}
}
async fn run_server(args: Args, vault_path: PathBuf) -> anyhow::Result<()> {
let kweb_listener = tokio::net::TcpListener::bind(&args.kweb_bind)
.await
.with_context(|| format!("binding Kweb listener {}", args.kweb_bind))?;
ensure_runtime_parent_directories(&args, &vault_path)?;
let kcode_kennedy_bootstrap_secrets::ServerSecrets {
openai_api_key,
gemini_api_key,
telegram_bot_token,
crates_io_key,
kweb_config,
} = kcode_kennedy_bootstrap_secrets::unlock_server(&vault_path)?;
let telegram_bot_token = telegram_bot_token
.map(kcode_tg_kennedy_bot::BotToken::new)
.transpose()?;
let codex_catalog_cache =
kcode_codex_runtime::CatalogCache::new(kcode_codex_runtime::DEFAULT_CODEX_EXECUTABLE);
let (kmap, system_roots) =
kcode_kennedy_roots::open(&args.kweb_root, kweb_config, &args.user_database)?;
let (kmap_commands, kmap_command_runtime) =
kcode_kmap_command_lane::open(&args.user_database, kmap.clone())?;
let credits = kcode_credits::Credits::open(&args.credits_database)?;
let task_board = kcode_task_board::TaskBoard::open(&args.task_board_database, credits.clone())?;
let speech_classifier = SpeechClassifier::open(SPEECH_CLASSIFICATION_DATABASE_PATH)
.with_context(|| {
format!("opening speaker-classification database {SPEECH_CLASSIFICATION_DATABASE_PATH}")
})?;
let speech_classifier = Arc::new(speech_classifier);
let dev_tools = kcode_dev_tools::Service::open(kcode_dev_tools::Config {
rust_libraries_root: args.rust_libs_root.clone(),
web_libraries_root: args.web_libs_root.clone(),
web_publications_root: args.web_libs_published_root.clone(),
rust_binaries_root: args.rust_bins_root.clone(),
rust_binary_publications_root: args.rust_bin_artifacts_root.clone(),
crates_io_registry_token: crates_io_key,
})
.map_err(anyhow::Error::new)
.with_context(|| {
format!(
"opening managed Kcode development roots under {}",
args.rust_libs_root
.parent()
.unwrap_or(Path::new("."))
.display()
)
})?;
let web_publications_root = dev_tools.web_publications_root().to_path_buf();
let telegram_identity = std::sync::Arc::new(kcode_telegram_identity::Directory::open(
&args.user_database,
&args.telegram_bootstrap_username,
)?);
let history_service =
kcode_session_history::SessionHistory::open(kcode_session_history::Config {
directory: args.session_directory,
completed_list: args.session_history_file,
provider_cost_compatibility: Some(
kcode_intelligence_chatend::provider_cost_compatibility(),
),
})?;
let (intelligence_service, intelligence_runtime) =
kcode_intelligence_router::open(kcode_intelligence_router::Config {
openai_api_key,
gemini_api_key,
codex_catalog_cache,
receipt_directory: args.intelligence_usage_directory,
})
.await?;
let agent_runtime = kcode_agent_runtime::AgentRuntime::new(intelligence_service.clone());
let telegram_runtime = kcode_tg_kennedy_bot::open(kcode_tg_kennedy_bot::Config {
database: args.telegram_database,
bot_token: telegram_bot_token,
identity_sink: telegram_identity.clone(),
max_voice_bytes: args.telegram_max_voice_bytes,
})
.await?;
let telegram_service = telegram_runtime.service();
let chunk_intelligence = intelligence_service.clone();
let transcribe_chunk: kcode_audio_ingress::AudioChunkCall = Arc::new(move |request| {
let intelligence = chunk_intelligence.clone();
Box::pin(async move {
let user = intelligence
.for_user(request.user_id)
.map_err(audio_intelligence_error)?;
let media = kcode_intelligence_router::Media::audio(
request.audio_ogg,
"audio-chunk.ogg",
"audio/ogg",
)
.map_err(audio_intelligence_error)?;
user.analyze_audio(kcode_intelligence_router::AudioAnalysisRequest {
operation: "transcribe_chunk".into(),
prompt: request.prompt,
model: request.model,
media,
schema: request.schema,
max_output_tokens: request.max_output_tokens,
temperature: None,
operation_id: uuid::Uuid::new_v4(),
parent_operation_id: None,
})
.await
.map(|response| response.value.text)
.map_err(audio_intelligence_error)
})
});
let text_intelligence = intelligence_service.clone();
let generate_text: kcode_audio_ingress::TextGenerationCall = Arc::new(move |request| {
let intelligence = text_intelligence.clone();
Box::pin(async move {
let reasoning_effort = match request.reasoning_effort.as_str() {
"xhigh" => kcode_intelligence_router::ReasoningEffort::XHigh,
_ => {
return Err(kcode_audio_ingress::IntelligenceError::new(
"AudioIngress requested an unsupported reasoning effort.",
false,
));
}
};
let user = intelligence
.for_user(request.user_id)
.map_err(audio_intelligence_error)?;
user.generate_text(kcode_intelligence_router::TextGenerationRequest {
operation: request.operation,
prompt: request.prompt,
model: request.model,
reasoning_effort,
timeout: request.timeout,
operation_id: uuid::Uuid::new_v4(),
parent_operation_id: None,
})
.await
.map(|response| response.value.text)
.map_err(audio_intelligence_error)
})
});
let audio_transcriber =
kcode_audio_ingress::AudioTranscriber::new(transcribe_chunk, generate_text);
let audio = kcode_audio_ingress::AudioIngress::open(
&args.audio_ingress_directory,
audio_transcriber,
Arc::clone(&speech_classifier),
)
.await?;
let audio_coordinator = kcode_audio_session_ingress::Coordinator::new(
audio,
history_service.clone(),
kcode_audio_session_ingress::Config {
user_id: system_roots.user.to_string(),
effective_context_tokens: intelligence_runtime.context_window_tokens,
},
)?;
let http_router = kcode_http_api::router(kcode_http_api::Config {
kmap: kmap.clone(),
kmap_commands,
user_root_node_id: system_roots.user,
kennedy_root_node_id: system_roots.kennedy,
telegram: telegram_service.clone(),
session_history: history_service.clone(),
audio_ingress: audio_coordinator.clone(),
audio_max_upload_bytes: args.audio_ingress_max_upload_bytes,
task_board: task_board.clone(),
credits,
web_publications_root,
})?;
let orchestration_config = kcode_kennedy_orchestration::Config {
user_root_node_id: system_roots.user.to_string(),
kennedy_root_node_id: system_roots.kennedy.to_string(),
telegram_max_media_bytes: args.telegram_max_voice_bytes,
runtime_model: kcode_kennedy_orchestration::RuntimeModel::from_intelligence(
intelligence_runtime,
),
};
let telegram_sessions = kcode_telegram_session_coordinator::Service::new(
telegram_service.clone(),
telegram_identity.clone(),
);
let session_service =
kcode_kennedy_sessions::Service::new(kcode_kennedy_sessions::Capabilities {
load_fixed_connections: args.fixed,
kmap: kmap.clone(),
intelligence: intelligence_service.clone(),
agents: agent_runtime,
history: history_service.clone(),
speech_classifier,
dev_tools: dev_tools.clone(),
telegram: telegram_sessions,
})
.with_task_board(task_board);
let orchestration_api = kcode_kennedy_orchestration::Api::new(
&orchestration_config,
kcode_kennedy_orchestration::LocalServices {
kmap: kmap.clone(),
intelligence: intelligence_service,
history: history_service.clone(),
audio: audio_coordinator,
directory: telegram_identity.clone(),
dev_tools,
telegram: telegram_service,
},
);
let orchestration_worker = kcode_kennedy_orchestration::build(
orchestration_config,
orchestration_api,
session_service,
);
let directory_roots = kcode_kennedy_roots::DirectoryRoots::new(
kmap,
telegram_identity,
args.telegram_bootstrap_username.clone(),
system_roots.user,
orchestration_worker.writer().clone(),
);
let telegram_session_runtime = Arc::new(kcode_kennedy_telegram_runtime::Runtime::new(
kcode_kennedy_telegram_runtime::Config {
telegram_max_media_bytes: args.telegram_max_voice_bytes,
telegram_web_user_handle: args.telegram_bootstrap_username,
},
orchestration_worker.clone(),
directory_roots,
));
tokio::try_join!(
async {
kcode_http_api::serve(kweb_listener, http_router)
.await
.map_err(anyhow::Error::new)
},
telegram_runtime.run(),
kcode_kennedy_orchestration::run(orchestration_worker),
telegram_session_runtime.run(),
async { kmap_command_runtime.await.map_err(anyhow::Error::new) },
)?;
Ok(())
}
fn audio_intelligence_error(
error: kcode_intelligence_router::Error,
) -> kcode_audio_ingress::IntelligenceError {
let retryable = error.retryable();
kcode_audio_ingress::IntelligenceError::new(error.message(), retryable)
}
fn ensure_runtime_parent_directories(args: &Args, vault_path: &Path) -> anyhow::Result<()> {
for path in [
vault_path,
&args.kweb_root,
&args.conversation_history_database,
&args.session_directory,
&args.session_history_file,
&args.telegram_database,
&args.user_database,
&args.task_board_database,
&args.credits_database,
Path::new(SPEECH_CLASSIFICATION_DATABASE_PATH),
&args.audio_ingress_directory,
&args.intelligence_usage_directory,
&args.rust_libs_root,
&args.web_libs_root,
&args.web_libs_published_root,
&args.rust_bins_root,
&args.rust_bin_artifacts_root,
] {
let Some(parent) = path.parent().filter(|value| !value.as_os_str().is_empty()) else {
continue;
};
if parent.exists() {
continue;
}
let mut builder = std::fs::DirBuilder::new();
builder.recursive(true);
#[cfg(unix)]
{
use std::os::unix::fs::DirBuilderExt;
builder.mode(0o700);
}
builder
.create(parent)
.with_context(|| format!("creating runtime data directory {}", parent.display()))?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn native_orchestration_remains_a_rust_backend_concern() {
assert_eq!(
std::any::type_name::<kcode_kennedy_orchestration::Session>(),
"kcode_kennedy_sessions::Session"
);
}
#[test]
fn dependency_closure_selects_cache_safe_chain() {
let manifest = include_str!("../Cargo.toml");
assert!(manifest.contains("version = \"0.5.14\""));
assert!(manifest.contains("kcode-http-api = \"0.4.2\""));
assert!(manifest.contains("kcode-audio-ingress = \"0.7.6\""));
assert!(manifest.contains("kcode-dev-tools-chatend = \"0.1.3\""));
assert!(manifest.contains("kcode-kennedy-bootstrap-secrets = \"0.1.0\""));
assert!(manifest.contains("kcode-kennedy-maintenance-guard = \"0.1.0\""));
assert!(manifest.contains("kcode-kennedy-orchestration = \"0.3.3\""));
assert!(!manifest.contains("kcode-kennedy-orchestration = \"=0.3.3\""));
assert!(manifest.contains("kcode-kennedy-sessions = \"0.2.8\""));
assert!(!manifest.contains("kcode-kennedy-sessions = \"=0.2.8\""));
assert!(manifest.contains("kcode-kennedy-telegram-runtime = \"0.3.3\""));
assert!(manifest.contains("kcode-kweb-context = \"0.2.9\""));
assert!(manifest.contains("kcode-session-history = \"0.1.15\""));
assert!(manifest.contains("kcode-telegram-identity = \"0.1.8\""));
}
#[tokio::test]
async fn unified_dev_tools_service_opens_all_roots_and_routes_three_source_kinds() {
let directory = std::env::temp_dir().join(format!(
"kennedy-dev-tools-open-test-{}",
uuid::Uuid::new_v4()
));
let rust_libraries = directory.join("kcode-rust-libs");
let web_libraries = directory.join("kcode-web-libs");
let web_publications = directory.join("kcode-web-libs-published");
let rust_binaries = directory.join("kcode-rust-bins");
let rust_binary_artifacts = directory.join("kcode-rust-bin-artifacts");
let service = kcode_dev_tools::Service::open(kcode_dev_tools::Config {
rust_libraries_root: rust_libraries.clone(),
web_libraries_root: web_libraries.clone(),
web_publications_root: web_publications.clone(),
rust_binaries_root: rust_binaries.clone(),
rust_binary_publications_root: rust_binary_artifacts.clone(),
crates_io_registry_token: "test-token".into(),
})
.unwrap();
assert_eq!(
service.web_libraries_root(),
std::fs::canonicalize(&web_libraries).unwrap()
);
assert_eq!(
service.web_publications_root(),
std::fs::canonicalize(&web_publications).unwrap()
);
for path in [
rust_libraries,
web_libraries,
web_publications,
rust_binaries,
rust_binary_artifacts,
] {
assert!(
path.is_dir(),
"managed root was not created: {}",
path.display()
);
}
for (create, open, write, name, path, kind) in [
(
kcode_dev_tools::CREATE_RUST_LIB_TOOL,
kcode_dev_tools::OPEN_RUST_LIB_TOOL,
kcode_dev_tools::WRITE_FILE_FREEFORM_RUST_LIB_TOOL,
"kennedy-test-lib",
"src/extra.rs",
kcode_dev_tools::ManagedSourceKind::RustLibrary,
),
(
kcode_dev_tools::CREATE_WEB_LIB_TOOL,
kcode_dev_tools::OPEN_WEB_LIB_TOOL,
kcode_dev_tools::WRITE_FILE_FREEFORM_WEB_LIB_TOOL,
"kennedy-test-web",
"extra.js",
kcode_dev_tools::ManagedSourceKind::WebLibrary,
),
(
kcode_dev_tools::CREATE_RUST_BIN_TOOL,
kcode_dev_tools::OPEN_RUST_BIN_TOOL,
kcode_dev_tools::WRITE_FILE_FREEFORM_RUST_BIN_TOOL,
"kennedy-test-bin",
"src/extra.rs",
kcode_dev_tools::ManagedSourceKind::RustBinary,
),
] {
let created = service
.execute(
"create-session",
create,
serde_json::json!({"name":name}),
Vec::new(),
)
.await
.unwrap();
assert_eq!(created.snapshot.unwrap().kind, kind);
let written = service
.execute(
"create-session",
write,
serde_json::json!({
"name":name,
"path":path,
"contents":"// Kennedy managed source\n",
}),
Vec::new(),
)
.await
.unwrap();
assert_eq!(written.snapshot.unwrap().kind, kind);
let open_result = service
.execute(
"open-session",
open,
serde_json::json!({"name":name}),
Vec::new(),
)
.await
.unwrap();
assert_eq!(open_result.snapshot.unwrap().kind, kind);
}
let asset = service
.execute(
"create-session",
kcode_dev_tools::ATTACH_OBJECT_WEB_LIB_TOOL,
serde_json::json!({
"name":"kennedy-test-web",
"path":"assets/fonts/display.woff2",
"objectId":"pending:1",
}),
vec![vec![0, 159, 146, 150, 255]],
)
.await
.unwrap();
let snapshot = asset.snapshot.unwrap();
assert_eq!(
snapshot.kind,
kcode_dev_tools::ManagedSourceKind::WebLibrary
);
assert!(snapshot.text.contains("Asset: assets/fonts/display.woff2"));
assert!(snapshot.text.contains("Bytes: 5"));
assert!(!snapshot.text.contains("SHA-256:"));
assert_eq!(service.release("create-session").await.unwrap(), 3);
assert_eq!(service.release("open-session").await.unwrap(), 3);
drop(service);
std::fs::remove_dir_all(directory).unwrap();
}
#[tokio::test]
async fn occupied_kweb_address_prevents_server_from_opening_persistent_state() {
let directory =
std::env::temp_dir().join(format!("kennedy-server-lock-test-{}", uuid::Uuid::new_v4()));
std::fs::create_dir(&directory).unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let bind = listener.local_addr().unwrap().to_string();
let vault = directory.join("vault.age");
let kmap = directory.join("kweb");
let conversations = directory.join("conversations.sqlite3");
let telegram = directory.join("telegram.sqlite3");
let users = directory.join("users.sqlite3");
let tasks = directory.join("tasks.sqlite3");
let credits = directory.join("credits.sqlite3");
let audio_media = directory.join("audio-media");
let args = Args {
vault_path: vault.clone(),
command: None,
kweb_bind: bind,
kweb_root: kmap.clone(),
conversation_history_database: conversations.clone(),
session_directory: directory.join("sessions"),
session_history_file: directory.join("session-history.txt"),
telegram_database: telegram.clone(),
user_database: users.clone(),
task_board_database: tasks.clone(),
credits_database: credits.clone(),
audio_ingress_directory: audio_media.clone(),
intelligence_usage_directory: directory.join("intelligence-usage"),
rust_libs_root: directory.join("rust-libs"),
web_libs_root: directory.join("kcode-web-libs"),
web_libs_published_root: directory.join("kcode-web-libs-published"),
rust_bins_root: directory.join("kcode-rust-bins"),
rust_bin_artifacts_root: directory.join("kcode-rust-bin-artifacts"),
telegram_bootstrap_username: "@test".to_owned(),
telegram_max_voice_bytes: 1024,
audio_ingress_max_upload_bytes: 1024,
fixed: false,
};
let error = run_server(args, vault.clone()).await.unwrap_err();
assert!(error.to_string().contains("binding Kweb listener"));
assert!(!vault.exists());
assert!(!kmap.exists());
assert!(!conversations.exists());
assert!(!telegram.exists());
assert!(!users.exists());
assert!(!tasks.exists());
assert!(!credits.exists());
assert!(!audio_media.exists());
std::fs::remove_dir_all(directory).unwrap();
}
}